use a wrapper

Author: cloud-fan

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

@@ -25,7 +25,6 @@ import org.apache.spark.sql.catalyst.expressions.{And, Expression, NamedExpressi
 import org.apache.spark.sql.catalyst.planning.PhysicalOperation
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.connector.catalog.{StagingTableCatalog, SupportsNamespaces, TableCapability, TableCatalog, TableChange}
-import org.apache.spark.sql.connector.read.V1Scan
 import org.apache.spark.sql.connector.read.streaming.{ContinuousStream, MicroBatchStream}
 import org.apache.spark.sql.execution.{FilterExec, LeafExecNode, ProjectExec, RowDataSourceScanExec, SparkPlan}
 import org.apache.spark.sql.execution.datasources.DataSourceStrategy
@@ -55,14 +54,12 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
 
   override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
     case PhysicalOperation(project, filters,
-        relation @ DataSourceV2ScanRelation(_, v1Scan: V1Scan, output)) =>
-      val pushedFilters = relation.getTagValue(V2ScanRelationPushDown.PUSHED_FILTERS_TAG)
-        .getOrElse(Seq.empty)
-      val v1Relation = v1Scan.toV1TableScan[BaseRelation with TableScan](session.sqlContext)
-      if (v1Relation.schema != v1Scan.readSchema()) {
+        relation @ DataSourceV2ScanRelation(_, V1ScanWrapper(scan, translated, pushed), output)) =>
+      val v1Relation = scan.toV1TableScan[BaseRelation with TableScan](session.sqlContext)
+      if (v1Relation.schema != scan.readSchema()) {
         throw new IllegalArgumentException(
           "The fallback v1 relation reports inconsistent schema:\n" +
-            "Schema of v2 scan:     " + v1Scan.readSchema() + "\n" +
+            "Schema of v2 scan:     " + scan.readSchema() + "\n" +
             "Schema of v1 relation: " + v1Relation.schema)
       }
       val rdd = v1Relation.buildScan()
@@ -72,8 +69,8 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
       val dsScan = RowDataSourceScanExec(
         output,
         requiredColumnsIndex,
-        pushedFilters.toSet,
-        pushedFilters.toSet,
+        translated.toSet,
+        pushed.toSet,
         unsafeRowRDD,
         v1Relation,
         tableIdentifier = None)

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

@@ -21,14 +21,14 @@ import org.apache.spark.sql.catalyst.expressions.{And, Expression, NamedExpressi
 import org.apache.spark.sql.catalyst.planning.ScanOperation
 import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan, Project}
 import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.catalyst.trees.TreeNodeTag
+import org.apache.spark.sql.connector.read.{Scan, V1Scan}
 import org.apache.spark.sql.execution.datasources.DataSourceStrategy
+import org.apache.spark.sql.sources
+import org.apache.spark.sql.types.StructType
 
 object V2ScanRelationPushDown extends Rule[LogicalPlan] {
   import DataSourceV2Implicits._
 
-  val PUSHED_FILTERS_TAG = TreeNodeTag[Seq[org.apache.spark.sql.sources.Filter]]("pushed_filters")
-
   override def apply(plan: LogicalPlan): LogicalPlan = plan transformDown {
     case ScanOperation(project, filters, relation: DataSourceV2Relation) =>
       val scanBuilder = relation.table.asReadable.newScanBuilder(relation.options)
@@ -57,8 +57,14 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] {
            |Output: ${output.mkString(", ")}
          """.stripMargin)
 
-      val scanRelation = DataSourceV2ScanRelation(relation.table, scan, output)
-      scanRelation.setTagValue(PUSHED_FILTERS_TAG, pushedFilters)
+      val wrappedScan = scan match {
+        case v1: V1Scan =>
+          val translated = filters.flatMap(DataSourceStrategy.translateFilter)
+          V1ScanWrapper(v1, translated, pushedFilters)
+        case _ => scan
+      }
+
+      val scanRelation = DataSourceV2ScanRelation(relation.table, wrappedScan, output)
 
       val projectionOverSchema = ProjectionOverSchema(output.toStructType)
       val projectionFunc = (expr: Expression) => expr transformDown {
@@ -81,3 +87,12 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] {
       withProjection
   }
 }
+
+// A wrapper for v1 scan to carry the translated filters and the handled ones. This is required by
+// the physical v1 scan node.
+case class V1ScanWrapper(
+    v1Scan: V1Scan,
+    translatedFilters: Seq[sources.Filter],
+    handledFilters: Seq[sources.Filter]) extends Scan {
+  override def readSchema(): StructType = v1Scan.readSchema()
+}