dongjoon-hyun commented on code in PR #58370:
URL: https://github.com/apache/spark/pull/58370#discussion_r3884760509
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala:
##########
@@ -143,13 +144,26 @@ class
RowLevelOperationRuntimeGroupFiltering(optimizeSubqueries: Rule[LogicalPla
private def buildDynamicPruningCond(
matchingRowsPlan: LogicalPlan,
- buildKeys: Seq[Attribute],
- pruningKeys: Seq[Attribute]): Expression = {
+ buildKeys: Seq[NamedExpression],
+ pruningKeys: Seq[NamedExpression]): Expression = {
assert(buildKeys.nonEmpty && pruningKeys.nonEmpty)
- val buildQuery = Aggregate(buildKeys, buildKeys, matchingRowsPlan)
+ // Nested references resolve to aliases over field-extraction expressions.
Materialize those
+ // aliases before aggregation so grouping-expression cleanup preserves the
subquery schema.
+ val buildKeyAliases = buildKeys.collect { case alias: Alias => alias }
+ val buildPlan = if (buildKeyAliases.nonEmpty) {
+ Project(matchingRowsPlan.output ++ buildKeyAliases, matchingRowsPlan)
+ } else {
+ matchingRowsPlan
+ }
+ val buildKeyAttrs = buildKeys.map(_.toAttribute)
+ val buildQuery = Aggregate(buildKeyAttrs, buildKeyAttrs, buildPlan)
Review Comment:
I gave this a try without the manual `Project`, using the shape the analyzer
itself produces for `GROUP BY s.a`:
```scala
def unalias(e: NamedExpression): Expression = e match {
case alias: Alias => alias.child
case other => other
}
val buildQuery = Aggregate(buildKeys.map(unalias), buildKeys,
matchingRowsPlan)
DynamicPruningExpression(
InSubquery(pruningKeys.map(unalias), ListQuery(buildQuery, numCols =
buildQuery.output.length)))
```
The row-level suites all still pass with that:
`GroupBasedRowLevelOperationCatalystRuntimeFilterSuite`,
`DeltaBasedRowLevelOperationCatalystRuntimeFilterSuite`,
`GroupBasedDeleteFromTableSuite`, `DeltaBasedDeleteFromTableSuite`,
`GroupBasedUpdateTableSuite`, `GroupBasedMergeIntoTableSuite` -- 236/236,
including your new nested DELETE test.
My reading is that `PullOutGroupingExpressions` already covers this: it is
in the `Finish Analysis` batch, `OptimizeSubqueries` re-runs the whole
optimizer on the subquery, and it inserts the equivalent `Project` while
preserving the alias expr IDs. If the hand-rolled version is guarding against
something my run does not reach, could the comment name it? Otherwise the
simpler form seems nicer.
Minor either way: if it stays, `Project(buildKeys, matchingRowsPlan)` is
enough -- carrying all of `matchingRowsPlan.output` is not needed to build
`buildKeyAttrs`.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala:
##########
@@ -49,6 +51,35 @@ class GroupBasedRowLevelOperationCatalystRuntimeFilterSuite
checkDeleteMetrics(numDeletedRows = 1, numCopiedRows = 1)
}
+ test("delete runtime group filtering by a nested attribute") {
+ val schema = "pk INT NOT NULL, id INT, salary INT, " +
+ "dep STRUCT<name: STRING, region: STRING>"
+ createTable(schema, Array[Transform](identity(reference(Seq("dep",
"name")))))
+ append(schema,
+ """{"pk":1,"id":1,"salary":300,"dep":{"name":"hr","region":"west"}}
+ |{"pk":2,"id":2,"salary":150,"dep":{"name":"software","region":"west"}}
+ |{"pk":3,"id":3,"salary":120,"dep":{"name":"hr","region":"east"}}
+ |""".stripMargin)
+
+ val executedPlan = executeAndKeepPlan {
+ sql(s"DELETE FROM $tableNameAsString WHERE salary IN (300, 400, 500)")
+ }
+ assertCatalystGroupFilter(
+ executedPlan,
+ expectedFilterAttrs = Seq("dep.name"),
+ expectedFilter = GroupFilter(
+ scanSchema = "salary INT, dep STRUCT<name: STRING>", groups =
Seq("hr")),
+ expectedFilterRefs = Seq("dep"))
Review Comment:
Nice test. One coverage thought: this is the only nested row-level case, and
it only exercises the group-based DELETE path. Delta-based and MERGE reach
`buildDynamicPruningCond` with different `matchingRowsPlan` shapes -- MERGE
goes through `RewritePredicateSubquery`, so the new `Project` would land over a
join. Since that is the genuinely new code in this PR, one more nested case
there would be reassuring.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationCatalystRuntimeFilterSuiteBase.scala:
##########
@@ -137,7 +137,8 @@ abstract class
RowLevelOperationCatalystRuntimeFilterSuiteBase
protected def assertCatalystGroupFilter(
executedPlan: SparkPlan,
expectedFilterAttrs: Seq[String],
- expectedFilter: GroupFilter): Unit = {
+ expectedFilter: GroupFilter,
+ expectedFilterRefs: Seq[String] = Seq.empty): Unit = {
Review Comment:
Nit: `Seq.empty` meaning "fall back to `expectedFilterAttrs`" took me a
second -- `Option[Seq[String]] = None` would say that more directly.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -290,12 +286,22 @@ case class DataSourceV2ScanRelation(
Statistics(sizeInBytes = conf.defaultSizeInBytes)
}
- private def checkRuntimeFilteringInterfaces(): Unit = scan match {
- case _: SupportsRuntimeV2Filtering with SupportsRuntimeCatalystFiltering =>
- throw SparkException.internalError(
- "A scan must not implement both SupportsRuntimeV2Filtering and " +
- s"SupportsRuntimeCatalystFiltering, but ${scan.getClass.getName}
implements both.")
- case _ =>
+ private def checkRuntimeFilteringInterfaces(): Unit = {
+ scan match {
+ case _: SupportsRuntimeV2Filtering with SupportsRuntimeCatalystFiltering
=>
+ throw SparkException.internalError(
+ "A scan must not implement both SupportsRuntimeV2Filtering and " +
+ s"SupportsRuntimeCatalystFiltering, but ${scan.getClass.getName}
implements both.")
+ case _: SupportsRuntimeCatalystFiltering =>
+ declaredFullyPushedRuntimeFilterAttrs.find(_.fieldNames.length >
1).foreach { ref =>
+ throw SparkException.internalError(
+ "Fully pushed runtime filter attribute " +
+ s"'${ref.fieldNames.mkString(".")}' declared by " +
+ s"${scan.getClass.getName} must be a top-level attribute of the
scan read schema, " +
+ "but it is a nested reference.")
+ }
Review Comment:
Small readability thought: this method now does two fairly different jobs --
validating the interface combination, and validating the shape of the
fully-pushed references. Since `DataSourceV2Strategy` always touches
`runtimeFilterAttrs`, it also means `fullyPushedFilterAttributes()` is now
called on every Catalyst-filtering scan at planning time, even when nothing is
being filtered.
The behavior is fine (and your new test depends on it), but would a separate
`checkFullyPushedFilterAttrs()` called from both lazy vals read better? The
current name no longer quite describes what happens inside.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala:
##########
@@ -90,29 +91,21 @@ class InMemoryCatalystRuntimeFilterTable(
.map(_.split(",").map(_.trim).toSet)
.getOrElse(Set.empty)
- /**
- * The partition columns, each named by the top level read schema column
it lives under, the
- * form both interface methods require. Columns pruned out of the read
schema are dropped,
- * since neither method may name one. Examples:
- * - `PARTITIONED BY (part)` -> `"part"`
- * - `PARTITIONED BY (s.nested)` -> `"s"`, the struct column holding the
partition field
- */
- private def partitionAttrNames: Array[String] = {
- val scanFields = readSchema.fields.map(_.name).toSet
- partitioning.flatMap(_.references()).map(_.fieldNames.head).distinct
- .filter(scanFields.contains)
+ /** 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] = {
- partitionAttrNames
- .filter(name => restrictedFilterAttrs.forall(_.contains(name)))
- .map(FieldReference.column)
+ partitionAttrs.filter(ref =>
restrictedFilterAttrs.forall(_.contains(ref.toString)))
Review Comment:
Tiny inconsistency: this fixture matches the table properties against
`ref.toString`, which back-quotes any name needing quoting, while
`InMemoryBaseTable`, `InMemoryTableWithV2Filter`,
`InMemoryRowLevelOperationTable` and `assertCatalystGroupFilter` all use
`fieldNames.mkString(".")`. A column such as `a b` would silently stop matching
here (same on line 108). Probably worth picking one spelling.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala:
##########
@@ -116,6 +116,57 @@ class DataSourceV2CatalystRuntimeFilterSuite extends
SharedSparkSession {
}
}
+ test("nested fully pushed filter attribute -> rejected without a runtime
filter") {
+ val tbl = s"$catalogName.tbl_nested_fully_pushed"
+ withTable(tbl) {
+ sql(s"CREATE TABLE $tbl (id INT, s STRUCT<part: INT, other: INT>) USING
$v2Source " +
+ "PARTITIONED BY (s.part) " +
+ "TBLPROPERTIES('fully-pushed-filter-attributes' = 's.part')")
+
+ val e = intercept[SparkException] {
+ sql(s"SELECT * FROM $tbl").queryExecution.executedPlan
+ }
+ assert(e.getMessage.contains("must be a top-level attribute"),
+ s"expected the nested fully pushed reference to be rejected, got
${e.getMessage}")
+ }
+ }
+
+ test("nested filter attribute under a non-struct column -> rejected during
resolution") {
+ val tbl = s"$catalogName.tbl_malformed_nested_filter_attr"
+ withTable(tbl) {
+ sql(s"CREATE TABLE $tbl (id INT, part INT) USING $v2Source PARTITIONED
BY (part)")
+
+ val scanRelation = sql(s"SELECT * FROM
$tbl").queryExecution.optimizedPlan.collectFirst {
+ case r: DataSourceV2ScanRelation => r
+ }.getOrElse(fail("Expected a DataSourceV2ScanRelation"))
+ val e = intercept[AnalysisException] {
+ scanRelation.copy(scan = new
NestedFilterAttributeScan).runtimeFilterAttrs
+ }
+ checkError(
+ exception = e,
+ condition = "INVALID_EXTRACT_BASE_FIELD_TYPE",
+ parameters = Map("base" -> "\"part\"", "other" -> "\"INT\""))
Review Comment:
Not blocking, but this pins a diagnosability regression that might be worth
a second look: #58296 rejected this case with an internal error naming the
offending scan class, and now a connector author gets
`INVALID_EXTRACT_BASE_FIELD_TYPE` with `base` = `"part"` / `other` = `"INT"`,
with no hint that runtime filtering or their `filterAttributes()` is involved.
Could `resolveFilterAttrs` catch the resolution failure and re-throw with
the scan class name attached? Then this test could assert on that instead.
--
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]
