sunchao commented on code in PR #58412:
URL: https://github.com/apache/spark/pull/58412#discussion_r3889865198
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala:
##########
@@ -66,9 +67,9 @@ class InMemoryTableWithV2Filter(
extends BatchScanBaseClass(_data, readSchema, tableSchema) with
SupportsRuntimeV2Filtering {
override def filterAttributes(): Array[NamedReference] = {
- val scanFields = readSchema.fields.map(_.name).toSet
partitioning.flatMap(_.references)
- .filter(ref => scanFields.contains(ref.fieldNames.mkString(".")))
+ .filter(ref => readSchema.findNestedField(
+ ref.fieldNames.toImmutableArraySeq, resolver =
SQLConf.get.resolver).isDefined)
Review Comment:
[P2] Apply transform safety to the public V1/V2 fixtures
Both public fixtures still advertise nested source references from every
partition transform, but their evaluators compare source predicate values
directly against transformed partition keys. With `PARTITIONED BY
(truncate(s.part, 1))`, a row containing `s.part = 'AB'` is stored under key
`'A'`. A V2 predicate such as `s.part = (SELECT max(value) FROM dim)`, with
`dim.value = 'AB'`, removes that matching partition. V1/V2 DPP has the same
problem when it pushes `IN ('AB')`. The equivalent V1 declaration is in
`InMemoryBaseTable.filterAttributes()`.
The base did not advertise nested `s.part`, so this pruning path was not
enabled. The new identity-transform guard applies only to
`CatalystRuntimeFilteringScan`; it does not protect these public-filter
evaluators. Please restrict them to supported identity transforms or translate
source predicates through the partition transform. Retained residual filters
cannot recover a partition already removed.
This finding is limited to the in-memory test connectors and is established
from source, without runtime reproduction.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala:
##########
@@ -74,23 +75,28 @@ class InMemoryCatalystRuntimeFilterTable(
Option(InMemoryCatalystRuntimeFilterTable.this.properties.get(FilterAttributesKey))
.map(_.split(",").map(_.trim).toSet)
+ private val fullyPushedFilterAttrs: Set[String] = Option(
+
InMemoryCatalystRuntimeFilterTable.this.properties.get(FullyPushedFilterAttributesKey))
+ .map(_.split(",").map(_.trim).toSet)
+ .getOrElse(Set.empty)
+
+ /** Partition source columns that are present in the scan read schema. */
+ private def partitionAttrs: Array[NamedReference] = {
+ partitioning.flatMap(_.references()).distinct
+ .filter(ref => readSchema.findNestedField(
+ ref.fieldNames.toImmutableArraySeq, resolver =
SQLConf.get.resolver).isDefined)
+ }
+
override def filterAttributes(): Array[NamedReference] = {
- val scanFields = readSchema.fields.map(_.name).toSet
- partitioning.flatMap(_.references()).filter { ref =>
- val name = ref.fieldNames.mkString(".")
- scanFields.contains(name) &&
- restrictedFilterAttrs.forall(_.contains(name))
+ partitionAttrs.filter { ref =>
+ restrictedFilterAttrs.forall(_.contains(ref.fieldNames.mkString(".")))
}
}
+ // Not intersected with `filterAttributes()`, so a table can declare a
fully pushed attribute
+ // that is not a filter attribute, a combination the interface forbids.
override def fullyPushedFilterAttributes(): Array[NamedReference] = {
- val fullyPushedFilterAttrs = Option(
-
InMemoryCatalystRuntimeFilterTable.this.properties.get(FullyPushedFilterAttributesKey))
- .map(_.split(",").map(_.trim).toSet)
- .getOrElse(Set.empty)
- filterAttributes().filter { ref =>
- fullyPushedFilterAttrs.contains(ref.fieldNames.mkString("."))
- }
+ partitionAttrs.filter(ref =>
fullyPushedFilterAttrs.contains(ref.fieldNames.mkString(".")))
Review Comment:
[P2] Align fully pushed declarations with the new evaluator
`fullyPushedFilterAttributes()` still advertises source columns of
non-identity transforms, but the new evaluator skips those transforms. For a
`DATE part` column partitioned by `days(part)` with
`TBLPROPERTIES('fully-pushed-filter-attributes' = 'part')`, a predicate such as
`part = (SELECT max(value) FROM dim)` selecting one of two dates has its
residual filter removed by `DataSourceV2Strategy`. The fixture then returns
without filtering because no identity slots exist, so both dates are returned.
This worked on the base: `getKey` preserves the original DATE day value for
`DaysTransform`, so the old evaluator correctly evaluated the source predicate.
Please restrict fully pushed declarations to sources the evaluator actually
enforces, or retain their correct evaluation.
This is a regression in the test connector introduced by the follow-up
commit and is established from source, without runtime reproduction.
--
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]
