cloud-fan commented on code in PR #58145:
URL: https://github.com/apache/spark/pull/58145#discussion_r3860154683
##########
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:
**Blocking:**
Collect advisories only when the Catalyst callback actually handled a
non-empty eligible filter list. `PushDownUtils` prefers
`SupportsPushDownV2Filters`, but this independent call still matches
`SupportsPushDownCatalystFilters`, so a builder mixing both traits receives
`advisoryFilters` without the promised Catalyst `pushFilters` call. The
subquery-only path likewise has no eligible Catalyst filter. Carry the actual
dispatch and eligibility state into this gate, and cover both cases.
##########
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:
**Blocking:**
Treat non-empty advisory metadata as a scan-merge blocker until `PlanMerger`
preserves its provenance. `MergeSubplans` feeds the enclosing advisory `Filter`
to a fresh builder as an ordinary best-effort condition; if that builder does
not re-report it as advisory, the rebuilt relation loses the marker and
`DataSourceV2Strategy` can execute a predicate the contract says Spark never
evaluates. Please include a scan-rebuild regression test.
##########
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:
**Non-blocking:**
Return `filters` immediately when both exclusion sequences are empty. In the
ordinary V2 path this otherwise visits every predicate, and
`ExpressionSet.contains` canonicalizes each one even though nothing can be
removed.
##########
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:
**Nit:**
Narrow this claim: variant pushdown accepts a non-empty filter sequence and
rewrites it after building the scan.
```suggestion
// interactions with Spark-side filters vary, but advisory filters do
not need to be
```
##########
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:
**Nit:**
`Negated predicate` conventionally means the Boolean complement, but this
helper substitutes `j = -i` and reverses the inequality. Please name the
arithmetic transformation directly.
```suggestion
// pushed predicate on i implies the corresponding predicate on j after
negating both sides.
```
##########
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:
**Non-blocking:**
Use an advisory predicate implied by `WHERE id >= 1`. `id = 1L` rejects rows
2 through 5 even though the expected result keeps them, so this fixture
violates the advisory contract and does not test a valid source scenario.
```suggestion
val advisory = "id > 0L"
```
--
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]
