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. now uses the same identity-transform condition as the
scan evaluator. I also added a shared V1/V2 regression for that verifies the
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 and checks its underlying resolution
cause.
##########
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 , and the new companion was removed. I retained the existing public method
because it shipped in 4.2; the runtime-filter-specific helper is 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. now has one arm, and row-level group filtering
has one arm each for Group and Delta operations. The exact declared references
are exposed 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. now provides one protected 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
before constructing a (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]
