szehon-ho commented on code in PR #58145:
URL: https://github.com/apache/spark/pull/58145#discussion_r3867318645
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -187,6 +193,7 @@ case class DataSourceV2ScanRelation(
keyGroupedPartitioning: Option[Seq[Expression]] = None,
ordering: Option[Seq[SortOrder]] = None,
pushedFilters: Seq[Expression] = Seq.empty,
+ advisoryFilters: Seq[Expression] = Seq.empty,
Review Comment:
Done in 319bffae37f. PlanMerger now preserves advisory provenance across a
scan merge: it requires both scans to carry equal advisory sets, offers them to
the fresh builder, and re-validates that the rebuilt scan fully pushes the
strict and advisory filters (otherwise the merge is declined). The inherited
strict/advisory metadata is restored so DataSourceV2Strategy still removes the
logical Filter and a later merge round accepts an equivalent scan. Added
scan-rebuild regression tests in MergeSubplansSuite (SPARK-58892).
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -126,6 +127,10 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan]
with PredicateHelper {
sHolder.pushedPredicates.mkString(", ")
}
+ sHolder.advisoryFilterExpressions = getAdvisoryFilters(sHolder)
Review Comment:
Done in 319bffae37f. Advisory filters are now collected only when the
Catalyst pushFilters callback actually dispatched (added
PushDownUtils.dispatchesToCatalystFilters, which encodes the
SupportsPushDownFilters / SupportsPushDownV2Filters precedence) and an eligible
non-subquery filter was pushed (normalizedFiltersWithoutSubquery.nonEmpty).
Covered both cases in DataSourceV2Suite: a builder mixing
SupportsPushDownV2Filters + SupportsPushDownCatalystFilters, and a
subquery-only filter.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2TableSampleSuite.scala:
##########
@@ -326,4 +333,49 @@ class DataSourceV2TableSampleSuite extends
DatasourceV2SQLBase
sql(s"DROP TABLE IF EXISTS $tableName")
}
}
+
+ test("advisory filters do not block TABLESAMPLE pushdown") {
+ registerCatalog("testsampleadvisory",
classOf[InMemoryTableWithTableSampleAndAdvisoryCatalog])
+ val table = "testsampleadvisory.ns.sample_tbl"
+ val advisory = "id = 1L"
Review Comment:
Done, changed the fixture advisory to `id > 0L`, which is implied by `WHERE
id >= 1`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -126,8 +126,17 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
tableSpec.withNewLocation(newLoc)
}
+ private def removeNotEvaluatedFilters(
+ filters: Seq[Expression],
+ relation: DataSourceV2ScanRelation,
+ otherNotEvaluatedFilters: Seq[Expression] = Nil): Seq[Expression] = {
+ val notEvaluatedFilterSet =
+ ExpressionSet(otherNotEvaluatedFilters ++ relation.advisoryFilters)
+ filters.filterNot(notEvaluatedFilterSet.contains)
Review Comment:
Done. removeNotEvaluatedFilters now returns the filters unchanged when both
exclusion sequences are empty, avoiding ExpressionSet canonicalization on the
ordinary V2 path.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -126,6 +127,10 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan]
with PredicateHelper {
sHolder.pushedPredicates.mkString(", ")
}
+ sHolder.advisoryFilterExpressions = getAdvisoryFilters(sHolder)
+ // Keep advisory filters off the plan until the other source pushdowns
have run. Their
+ // matchers require that no Spark-side filters remain, but advisory
filters do not need to be
Review Comment:
Done, reworded to "interactions with Spark-side filters vary, but advisory
filters do not need to be evaluated ...".
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala:
##########
@@ -1880,26 +2037,168 @@ class CatalystFilterDataSourceV2 extends
TestingV2Source {
override def getTable(options: CaseInsensitiveStringMap): Table = new
SimpleBatchTable {
override def newScanBuilder(options: CaseInsensitiveStringMap):
ScanBuilder = {
- new CatalystFilterScanBuilder()
+ new CatalystFilterScanBuilder(options)
+ }
+ }
+}
+
+class CatalystFilterJoinDataSourceV2 extends TestingV2Source {
+
+ override def getTable(options: CaseInsensitiveStringMap): Table = new
SimpleBatchTable {
+ override def newScanBuilder(options: CaseInsensitiveStringMap):
ScanBuilder = {
+ new CatalystFilterJoinScanBuilder(options)
+ }
+ }
+}
+
+class CatalystFilterJoinScanBuilder(options: CaseInsensitiveStringMap)
+ extends CatalystFilterScanBuilder(options) with SupportsPushDownJoin {
+
+ private var joinedSchema: Option[StructType] = None
+
+ override def isOtherSideCompatibleForJoin(other: SupportsPushDownJoin):
Boolean =
+ other.isInstanceOf[CatalystFilterJoinScanBuilder]
+
+ override def pushDownJoin(
+ other: SupportsPushDownJoin,
+ joinType: V2JoinType,
+ leftColumns: Array[ColumnWithAlias],
+ rightColumns: Array[ColumnWithAlias],
+ condition: Predicate): Boolean = {
+ def fields(columns: Array[ColumnWithAlias]): Array[StructField] =
columns.map { col =>
+ val name = if (col.alias() != null) col.alias() else col.colName()
+ TestingV2Source.schema(col.colName()).copy(name = name)
+ }
+ joinedSchema = Some(StructType(fields(leftColumns) ++
fields(rightColumns)))
+ true
+ }
+
+ override def readSchema(): StructType =
joinedSchema.getOrElse(super.readSchema())
+}
+
+class CatalystFilterLimitDataSourceV2 extends TestingV2Source {
+
+ override def getTable(options: CaseInsensitiveStringMap): Table = new
SimpleBatchTable {
+ override def newScanBuilder(options: CaseInsensitiveStringMap):
ScanBuilder = {
+ new CatalystFilterLimitScanBuilder(options)
+ }
+ }
+}
+
+class CatalystFilterLimitScanBuilder(options: CaseInsensitiveStringMap)
+ extends CatalystFilterScanBuilder(options) with SupportsPushDownLimit {
+
+ private var pushedLimit: Option[Int] = None
+
+ override def pushLimit(limit: Int): Boolean = {
+ pushedLimit = Some(limit)
+ true
+ }
+
+ override def isPartiallyPushed: Boolean = false
+
+ override def planInputPartitions(): Array[InputPartition] = {
+ val partitions = super.planInputPartitions()
+ pushedLimit match {
+ case Some(n) =>
+ val values = partitions.flatMap {
+ case ValuesInputPartition(vs) => vs
+ case RangeInputPartition(start, end) => start until end
+ case other =>
+ throw new IllegalArgumentException(s"Unexpected partition: $other")
+ }.take(n)
+ Array(ValuesInputPartition(values.toSeq))
+ case None =>
+ partitions
}
}
+
+ override def createReaderFactory(): PartitionReaderFactory = {
+ if (pushedLimit.isDefined) ValuesReaderFactory else
super.createReaderFactory()
+ }
}
-class CatalystFilterScanBuilder extends SimpleScanBuilder
+class CatalystFilterScanBuilder(options: CaseInsensitiveStringMap) extends
SimpleScanBuilder
with SupportsPushDownCatalystFilters {
+ private var pushedCatalystFilters = Seq.empty[CatalystExpression]
+ private val deriveAdvisory = CatalystFilterScanBuilder.derivation(options)
+
override def pushFilters(filters: Seq[CatalystExpression]):
Seq[CatalystExpression] = {
if (filters.exists(!_.deterministic)) {
throw new IllegalArgumentException(
s"Non-deterministic filters should not be pushed:
${filters.mkString(", ")}")
}
+ pushedCatalystFilters = filters
Nil
}
+ override def advisoryFilters: Seq[CatalystExpression] =
deriveAdvisory(pushedCatalystFilters)
+
override def pushedFilters: Array[Predicate] = Array.empty
override def planInputPartitions(): Array[InputPartition] = {
- throw new IllegalArgumentException("planInputPartitions must not be
called")
+ // Spark never evaluates an advisory filter, so the source has to apply it
itself.
+ val enforcedFilters = pushedCatalystFilters ++ advisoryFilters
+ enforcedFilters.reduceLeftOption(CatalystAnd) match {
+ case Some(filter) =>
+ val attrs = DataTypeUtils.toAttributes(readSchema())
+ val bound = filter.transformUp {
+ case attr: AttributeReference =>
+ attrs.find(_.name == attr.name).getOrElse(
+ throw new IllegalArgumentException(s"Unknown column:
${attr.name}"))
+ }
+ val predicate = CatalystPredicate.createInterpreted(
+ BindReferences.bindReference(bound, attrs))
+ Array(ValuesInputPartition((0 until 10).filter { i =>
+ predicate.eval(InternalRow(i, -i))
+ }))
+ case None =>
+ Array(RangeInputPartition(0, 5), RangeInputPartition(5, 10))
+ }
+ }
+
+ override def createReaderFactory(): PartitionReaderFactory = {
+ if (pushedCatalystFilters.nonEmpty) ValuesReaderFactory else
SimpleReaderFactory
+ }
+}
+
+object CatalystFilterScanBuilder {
+ val ADVISORY_DERIVATION: String = "advisoryDerivation"
+ val NEGATE_I_TO_J: String = "negate-i-to-j"
+
+ // Common derivation used by advisory-filter tests. Rows are (i, j) with j =
-i, so a
+ // pushed predicate on i implies the negated predicate on j.
Review Comment:
Done, renamed to "pushed predicate on i implies the corresponding predicate
on j after negating both sides."
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
