szehon-ho commented on code in PR #58412:
URL: https://github.com/apache/spark/pull/58412#discussion_r3897440150
##########
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:
Thanks. `fullyPushedFilterAttributes()` is now restricted to identity
transforms, matching what the fixture evaluator can enforce directly.
Non-identity sources may still receive ordinary Catalyst runtime filters, but
Spark retains the post-scan residual filter. I added a `days(part)` regression
verifying the runtime filter is pushed, both partitions remain at the source,
and the residual returns only the matching date.
##########
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:
Thanks, good catch. I restricted both public V1/V2 fixtures to advertise
only identity-transform source attributes, and added the same identity guard in
their evaluators. The shared regression uses `truncate(derives.toStr, 1)` and
runs against both fixtures; it verifies no DPP filter is pushed on the
transformed source and that all partitions remain available for the residual
join.
--
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]
