szehon-ho commented on code in PR #58503:
URL: https://github.com/apache/spark/pull/58503#discussion_r3928268897
##########
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:
Fixed in 5cd473027a7. `canEvaluate` now uses the same identity-transform
condition as the scan evaluator. I also added a shared V1/V2 regression for
`days(part)` that verifies the `IN` filter remains residual, returns only the
matching row, and retains both source partitions.
##########
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:
Fixed in 5cd473027a7. Fully-pushed references are now checked for the
top-level constraint, resolved against the output, and only then checked for
exact membership after ordinary filter attributes are also resolved. The
missing-reference test again expects `CANNOT_RESOLVE` and checks its underlying
resolution cause.
##########
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:
Restored in 5cd473027a7 as a sibling valid-case test. It restricts
`filterAttributes()` to `p1` without a fully-pushed declaration and verifies
that the scalar-subquery predicate on `p2` creates no runtime filter, calls no
Catalyst pushdown, remains post-scan, and retains all five partitions.
##########
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:
Addressed in 5cd473027a7. The contextual error-wrapping resolver now lives
in `V2ExpressionUtils`, and the new `DataSourceV2ScanRelation` companion was
removed. I retained the existing public `resolveAttributeRefs` method because
it shipped in 4.2; the runtime-filter-specific helper is `private[sql]` and is
used by both relation and pushdown paths.
##########
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:
Simplified in 5cd473027a7. `PartitionPruning` now has one
`DataSourceV2ScanRelation` arm, and row-level group filtering has one arm each
for Group and Delta operations. The exact declared references are exposed
`private[sql]` for building nested pruning keys.
##########
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:
Fixed in 5cd473027a7. `BatchScanBaseClass` now provides one protected
`identityPartitionAttributes` helper, including deduplication and read-schema
resolution, and the V1, V2, Catalyst fully-pushed, and row-level scans all use
it.
##########
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:
Fixed in 5cd473027a7. The centralized runtime-filter resolver returns
`AttributeSet.empty` before constructing a `LocalRelation` (or deriving the
relation output type), so scans with no declared attributes avoid those
allocations.
--
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]
