dongjoon-hyun commented on code in PR #58503:
URL: https://github.com/apache/spark/pull/58503#discussion_r3927706529
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -841,14 +837,14 @@ abstract class InMemoryBaseTable(
var pushedFilters: Array[Filter] = Array.empty
override def filterAttributes(): Array[NamedReference] = {
- partitioning.flatMap(_.references)
+ identityPartitionReferences
Review Comment:
Nit: this
`identityPartitionReferences.filter(readSchema.findNestedField(...).isDefined)`
expression is now copied in four scans (here, `InMemoryTableWithV2Filter`,
`InMemoryRowLevelOperationTable`, and
`InMemoryCatalystRuntimeFilterTable.identityPartitionAttrs`), and only the last
one carries `.distinct`. A single protected helper on `BatchScanBaseClass` next
to `identityPartitionReferences` would remove the drift.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -212,8 +212,11 @@ object PushDownUtils extends Logging {
// filters whose translation was not already accepted in the first
pass. (See SPARK-55596)
// Only candidates whose referenced columns are declared in
filterAttributes() are eligible.
val partPredicatesPushed = filterableScan.supportsIterativePushdown()
&& {
- val filterAttrs = V2ExpressionUtils.resolveAttributeRefs(
- filterableScan.filterAttributes(), output)
+ val filterAttrs = DataSourceV2ScanRelation.resolveRuntimeFilterAttrs(
Review Comment:
With this and the two `PartitionPruning` call sites rewired,
`V2ExpressionUtils.resolveAttributeRefs` has no remaining callers, and the new
companion helper re-implements its body plus the error wrapping. Leaving both
around means a future call site can pick the public `V2ExpressionUtils` one and
surface `_LEGACY_ERROR_TEMP_1137` for the same misdeclaration every other path
now reports as `DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.CANNOT_RESOLVE`.
Either remove it (it shipped in 4.2.0 on a public object, so worth a note in
the PR description) or make it the home of the error-wrapping version and drop
the new `object DataSourceV2ScanRelation`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -326,6 +315,17 @@ case class DataSourceV2ScanRelation(
scanClass = scan.getClass.getName,
relationOutput = fromAttributes(output))
}
+ declaredFullyPushedRuntimeFilterAttrs.find { fullyPushedRef =>
Review Comment:
This name-only membership check runs before `fullyPushedFilterAttributes()`
is resolved against the output. A fully pushed reference that does not exist at
all is therefore reported as `NOT_IN_FILTER_ATTRIBUTES` ("must also be returned
by `filterAttributes()`") rather than `CANNOT_RESOLVE`. That is exactly the
`MissingFullyPushedFilterAttributeScan` case, whose expectation was flipped in
this PR and whose `getCause` assertion was dropped.
A connector author following that message would add `missing` to
`filterAttributes()` and only then get `CANNOT_RESOLVE` on the next run.
Checking resolvability of the fully pushed refs before the membership check (or
resolving both lists first) would surface the root cause in one round and let
the original test expectation stand.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala:
##########
@@ -55,32 +55,32 @@ class
RowLevelOperationRuntimeGroupFiltering(optimizeSubqueries: Rule[LogicalPla
override def apply(plan: LogicalPlan): LogicalPlan =
plan.transformDownWithPruning(
_.containsAnyPattern(REPLACE_DATA, WRITE_DELTA)) {
case GroupBasedRowLevelOperation(replaceData, _, Some(cond),
- ExtractV2Scan(scan: SupportsRuntimeV2Filtering))
- if canInjectGroupFilters(cond, scan.filterAttributes) =>
+ r @ ExtractV2Scan(scan: SupportsRuntimeV2Filtering))
Review Comment:
Nit: now that eligibility comes from `r.runtimeFilterAttrs`, the four arms
here differ only by the interface type test and `filterAttributes` vs
`filterAttributes()`, and the two V2 arms in
`PartitionPruning.getFilterableTableScan` are identical with `scan` unused.
`declaredRuntimeFilterAttrs` already folds both interfaces (and the
neither-interface case) into one array.
`PartitionPruning` could collapse to a single `case (resExp, r:
DataSourceV2ScanRelation) if resExp.references.subsetOf(r.runtimeFilterAttrs)`,
and this rule to two arms (Group/Delta) taking `r` and using `r.scan` plus a
`private[sql]` `declaredRuntimeFilterAttrs` for the raw `NamedReference` array
that `injectGroupFilters` still needs for nested paths. A scan implementing
neither interface yields an empty set without throwing, so this is
behavior-preserving.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -841,14 +837,14 @@ abstract class InMemoryBaseTable(
var pushedFilters: Array[Filter] = Array.empty
override def filterAttributes(): Array[NamedReference] = {
- partitioning.flatMap(_.references)
+ identityPartitionReferences
.filter(ref => readSchema.findNestedField(
ref.fieldNames.toImmutableArraySeq, resolver =
SQLConf.get.resolver).isDefined)
}
override def filter(filters: Array[Filter]): Unit = {
- if (partitioning.length == 1 && partitioning.head.references().length ==
1) {
- val ref = partitioning.head.references().head
+ if (partitioning.length == 1 && identityPartitionReferences.length == 1)
{
Review Comment:
`InMemoryScanBuilder.canEvaluate` (line 519) still uses `partitioning.length
== 1 && partitioning.head.references.length == 1`, so an `In` on a single
non-identity transform (e.g. `PARTITIONED BY (days(part))`) is still classified
as fully evaluable and removed from `postScanFilters` by `pushFilters`. With
this guard now identity-only, `build()` hands that `In` to `filter()` and it is
skipped, so the static filter is evaluated nowhere.
Example: `CREATE TABLE t (id INT, part DATE) PARTITIONED BY (days(part))`,
two dates inserted, `SELECT * FROM t WHERE part IN (DATE '2026-08-01')` now
returns both rows.
Could we make `canEvaluate` use the same identity-only guard
(`identityPartitionReferences.length == 1`) so the two sides agree?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -347,6 +347,30 @@ case class DataSourceV2ScanRelation(
}
}
+object DataSourceV2ScanRelation {
+ private[sql] def resolveRuntimeFilterAttrs(
+ filterAttrs: Array[NamedReference],
+ method: String,
+ scanClass: String,
+ output: Seq[AttributeReference]): AttributeSet = {
+ val plan = LocalRelation(output)
Review Comment:
Nit: `LocalRelation(output)` is built even when `filterAttrs` is empty, and
the instance path now forwards here instead of resolving against `this` as
before. `DataSourceV2Strategy` reads `runtimeFilterAttrs` for every V2 scan
relation, including scans with no runtime-filter interface, so each of those
now allocates two `LocalRelation`s for nothing. Small cost, but a one-line `if
(filterAttrs.isEmpty) return AttributeSet.empty` short-circuit (or resolving
against `this` in the instance path) avoids it.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala:
##########
@@ -432,30 +475,28 @@ class DataSourceV2CatalystRuntimeFilterSuite extends
SharedSparkSession {
}
}
- test("filter on column outside filterAttributes -> not pushed, even if
declared fully pushed") {
+ test("fully pushed attribute outside filterAttributes -> rejected") {
Review Comment:
Turning the old tbl4 test into a pure rejection test drops the suite's only
coverage of the valid case: a scalar-subquery filter on a partition column
outside a restricted `filterAttributes()` must not be routed as a runtime
filter, `filter()` must not be called, and the post-scan `FilterExec` must be
kept. `'filter-attributes'` now only appears in this invalid-declaration test.
Could we keep a sibling test with `'filter-attributes' = 'p1'` and no
fully-pushed property, asserting `runtimeFilters.isEmpty`,
`assertPushedCatalystPredicates(df, 0)`,
`assertScalarSubqueryEvaluatedAfterScan(df, expected = true)` and all 5
partitions retained?
--
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]
