[SPARK-32708] Query optimization fails to reuse exchange with DataSourceV2

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExec.scala

@@ -22,6 +22,7 @@ import scala.collection.JavaConverters._
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.QueryPlan
 import org.apache.spark.sql.catalyst.plans.physical
 import org.apache.spark.sql.catalyst.plans.physical.SinglePartition
 import org.apache.spark.sql.execution.{ColumnarBatchScan, LeafExecNode, WholeStageCodegenExec}
@@ -52,6 +53,17 @@ case class DataSourceV2ScanExec(
     case _ => false
   }
 
+  override def doCanonicalize(): DataSourceV2ScanExec = {
+    DataSourceV2ScanExec(
+      output.map(QueryPlan.normalizeExprId(_, output)),
+      source,
+      options,
+      QueryPlan.normalizePredicates(
+        pushedFilters,
+        AttributeSeq(pushedFilters.flatMap(_.references).distinct)),
+      reader)
+  }
+
   override def hashCode(): Int = {
     Seq(output, source, options).hashCode()
   }