voonhous commented on code in PR #19406:
URL: https://github.com/apache/hudi/pull/19406#discussion_r3690432981
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -134,9 +118,6 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase with Sp
val expectedReadSchemaClause = "ReadSchema:
struct<id:int,item:struct<name:string,price:int>>"
Review Comment:
Pre-existing, but worth fixing while this PR reworks the suite: on master
this test goes through the **COW** arm below and asserts the **pruned** schema
`{id, item{name}}` -- the opposite of the test's name -- while this unpruned
`expectedReadSchemaClause` is never checked (the `RowDataSourceScanExec` arm is
unreachable now that MOR reads resolve to `HadoopFsRelation`, see #17457). The
commented-out asserts at lines 139/142 have been dead since #10245.
**Action:** rename/rewrite this test around behavior that is actually
asserted, or delete it; at minimum drop the commented-out asserts.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -167,6 +148,128 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase with Sp
}
}
+ test("Test NestedSchemaPruning prunes nested struct when array and map
columns are present") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithComplexNestedSchema(tableName, tablePath)
+
+ // Only a single nested sub-field is projected, so "item" is pruned down
to just "name" and the
+ // unreferenced "tags" (array<struct>) and "props" (map<string,struct>)
columns are dropped.
+ // Traversing the full data schema exercises the array and map branches
of countLeaves.
Review Comment:
**This comment does not hold: `countLeaves` is never executed -- by these
tests or by any non-metadata read.** The pruning asserted in this suite is done
by Spark's built-in `SchemaPruning`.
Evidence:
- All 5 tests in this suite pass unchanged with
`spark.sql.optimizer.excludedRules=org.apache.spark.sql.execution.datasources.HoodieNestedSchemaPruning`;
excluding Spark's `SchemaPruning` instead makes the pruning disappear.
- jacoco over the suite: the only line of `HoodieNestedSchemaPruning` this
PR newly covers is the `else plan` arm of `apply` (via the disabled-flag test);
`countLeaves`, `prunePhysicalColumns`, `buildNewProjection`,
`buildPrunedRelation` all stay at 0.
- Cause: since #17457 every non-metadata read returns a `HadoopFsRelation`
(`HoodieFileGroupReaderBasedFileFormat extends ParquetFileFormat`, so Spark
prunes natively). The rule only matches `HoodieBaseRelation if
canPruneRelationSchema`, and `canPruneRelationSchema` requires
`!isMetadataTable` -- but `DefaultSource` only builds a `HoodieBaseRelation`
**for** the metadata table. The prune branch is unreachable from SQL.
**Action:** reword the rule-internals comments here and at the two no-op
tests, and retitle the tests and PR description to what is actually pinned:
end-to-end nested schema pruning over Hudi's `HadoopFsRelation` read path.
Separately, file a follow-up issue that `HoodieNestedSchemaPruning` has been
dead code since #17457 -- delete it, or cover it with a direct unit test
against a `HoodieBaseRelation`.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -167,6 +148,128 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase with Sp
}
}
+ test("Test NestedSchemaPruning prunes nested struct when array and map
columns are present") {
Review Comment:
This test is assertion-identical to the `mor` leg of `Test
NestedSchemaPruning optimization successful`: same `SELECT id, item.name`, same
expected schema, same `ReadSchema` clause. The added `tags`/`props` columns
change nothing that is asserted, because dropping unreferenced top-level
columns is done by Spark's `ColumnPruning`, not nested schema pruning.
**Action:** fold the array/map columns into
`createTableWithNestedStructSchema` so the existing test covers this shape and
drop this test, or differentiate it by projecting through the complex columns
(see the comment on the `SELECT` below).
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -167,6 +148,128 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase with Sp
}
}
+ test("Test NestedSchemaPruning prunes nested struct when array and map
columns are present") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithComplexNestedSchema(tableName, tablePath)
+
+ // Only a single nested sub-field is projected, so "item" is pruned down
to just "name" and the
+ // unreferenced "tags" (array<struct>) and "props" (map<string,struct>)
columns are dropped.
+ // Traversing the full data schema exercises the array and map branches
of countLeaves.
+ val selectDF = spark.sql(s"SELECT id, item.name FROM $tableName")
Review Comment:
No query in this suite has a `WHERE` on a nested field, so the pushed-filter
plus pruned-schema interaction is untested (it is also the only input that
would reach the `filters.nonEmpty` branch of `buildNewProjection`, if the rule
were ever revived).
**Action:** add `WHERE item.name = 'a1'` to one of the pruning cases and
assert the returned values.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -167,6 +148,128 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase with Sp
}
}
+ test("Test NestedSchemaPruning prunes nested struct when array and map
columns are present") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithComplexNestedSchema(tableName, tablePath)
+
+ // Only a single nested sub-field is projected, so "item" is pruned down
to just "name" and the
+ // unreferenced "tags" (array<struct>) and "props" (map<string,struct>)
columns are dropped.
+ // Traversing the full data schema exercises the array and map branches
of countLeaves.
+ val selectDF = spark.sql(s"SELECT id, item.name FROM $tableName")
Review Comment:
`tags` and `props` are only ever **dropped** here, never projected through,
so the pruned schema never contains a complex type and the array/map recursion
in `HoodieSchemaUtils.pruneDataSchema` (the path #18566 fixed) stays untested.
**Action:** add projections through the complex columns and assert the
values:
- `SELECT tags.k FROM t` gives `ReadSchema:
struct<tags:array<struct<k:string>>>`
- `SELECT props['m0'].a FROM t` gives `ReadSchema:
struct<props:map<string,struct<a:int>>>`
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -167,6 +148,128 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase with Sp
}
}
+ test("Test NestedSchemaPruning prunes nested struct when array and map
columns are present") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithComplexNestedSchema(tableName, tablePath)
+
+ // Only a single nested sub-field is projected, so "item" is pruned down
to just "name" and the
+ // unreferenced "tags" (array<struct>) and "props" (map<string,struct>)
columns are dropped.
+ // Traversing the full data schema exercises the array and map branches
of countLeaves.
+ val selectDF = spark.sql(s"SELECT id, item.name FROM $tableName")
+
+ val expectedSchema = StructType(Seq(
+ StructField("id", IntegerType, nullable = true),
+ StructField("item", StructType(Seq(StructField("name", StringType,
nullable = false))), nullable = true)
+ ))
+ val expectedReadSchemaClause = "ReadSchema:
struct<id:int,item:struct<name:string>>"
+
+ assertPrunedReadSchema(selectDF, tableName, expectedSchema,
expectedReadSchemaClause)
+
+ // Execute the query to make sure it's working as expected (smoke test)
+ selectDF.count
Review Comment:
No test in this suite ever creates a log file (all tables are CTAS-only), so
this `count` reads base parquet only and never merges a pruned schema against
delta records -- exactly the HUDI-5443 / #7528 bug class (exception reading a
MOR table after `NestedSchemaPruning` when delta-log merging kicks in), also
the case called out in #18570.
**Action:** run `UPDATE $tableName SET ... WHERE id = 1` before the
projection so the file slice has a log file, and assert the returned rows with
`checkAnswer(...)` instead of `selectDF.count`.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -167,6 +148,128 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase with Sp
}
}
+ test("Test NestedSchemaPruning prunes nested struct when array and map
columns are present") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithComplexNestedSchema(tableName, tablePath)
+
+ // Only a single nested sub-field is projected, so "item" is pruned down
to just "name" and the
+ // unreferenced "tags" (array<struct>) and "props" (map<string,struct>)
columns are dropped.
+ // Traversing the full data schema exercises the array and map branches
of countLeaves.
+ val selectDF = spark.sql(s"SELECT id, item.name FROM $tableName")
+
+ val expectedSchema = StructType(Seq(
+ StructField("id", IntegerType, nullable = true),
+ StructField("item", StructType(Seq(StructField("name", StringType,
nullable = false))), nullable = true)
+ ))
+ val expectedReadSchemaClause = "ReadSchema:
struct<id:int,item:struct<name:string>>"
+
+ assertPrunedReadSchema(selectDF, tableName, expectedSchema,
expectedReadSchemaClause)
+
+ // Execute the query to make sure it's working as expected (smoke test)
+ selectDF.count
+ }
+ }
+
+ test("Test NestedSchemaPruning is a no-op when all nested sub-fields are
selected") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithNestedStructSchema("mor", tableName, tablePath)
+
+ // Every leaf is projected, so the pruned schema has the same leaf count
as the data schema and
+ // the rule leaves "item" untouched (the countLeaves comparison is an
equality, not a >).
+ val selectDF = spark.sql(s"SELECT id, item.name, item.price, ts FROM
$tableName")
+
+ val expectedItemStruct = StructType(Seq(
+ StructField("name", StringType, nullable = false),
+ StructField("price", IntegerType, nullable = false)
+ ))
+ assertEquals(expectedItemStruct, prunedStructTypeOf(selectDF, "item"))
+
+ selectDF.count
+ }
+ }
+
+ test("Test NestedSchemaPruning is a no-op when nested schema pruning is
disabled") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithNestedStructSchema("mor", tableName, tablePath)
+
+ // With the optimizer flag off the rule short-circuits, so "item" keeps
"price" even though only
+ // "item.name" is projected.
+ val expectedItemStruct = StructType(Seq(
+ StructField("name", StringType, nullable = false),
+ StructField("price", IntegerType, nullable = false)
+ ))
+
+ withSQLConf(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> "false") {
+ val selectDF = spark.sql(s"SELECT id, item.name FROM $tableName")
+ assertEquals(expectedItemStruct, prunedStructTypeOf(selectDF, "item"))
+ selectDF.count
+ }
+ }
+ }
+
+ private def assertPrunedReadSchema(selectDF: DataFrame,
+ tableName: String,
+ expectedSchema: StructType,
+ expectedReadSchemaClause: String,
+ hint: String = ""): Unit = {
+ // NOTE: Unfortunately, we can't use pattern-matching to extract required
fields, due to a need to maintain
+ // compatibility w/ Spark 2.4
+ selectDF.queryExecution.executedPlan match {
+ // COW
+ case ProjectExec(_, fileScan: FileSourceScanExec) =>
+ assertEquals(tableName, fileScan.tableIdentifier.get.table)
+ assertEquals(expectedSchema, fileScan.requiredSchema, hint)
+
+ // MOR
+ case ProjectExec(_, dataScan: RowDataSourceScanExec) =>
+ // NOTE: This is temporary solution to assert for Spark 2.4, until
it's deprecated
+ val explainedPlan = explain(selectDF.queryExecution.logical)
+ assertTrue(explainedPlan.contains(expectedReadSchemaClause))
+
+ assertEquals(tableName, dataScan.tableIdentifier.get.table)
+ assertEquals(expectedSchema, dataScan.requiredSchema, hint)
+ }
+ }
+
+ private def prunedStructTypeOf(selectDF: DataFrame, fieldName: String):
StructType = {
+ val requiredSchema = selectDF.queryExecution.executedPlan match {
+ case ProjectExec(_, fileScan: FileSourceScanExec) =>
fileScan.requiredSchema
+ case ProjectExec(_, dataScan: RowDataSourceScanExec) =>
dataScan.requiredSchema
+ case fileScan: FileSourceScanExec => fileScan.requiredSchema
+ case dataScan: RowDataSourceScanExec => dataScan.requiredSchema
+ }
+ requiredSchema(fieldName).dataType.asInstanceOf[StructType]
+ }
+
+ private def createTableWithComplexNestedSchema(tableName: String,
Review Comment:
nit (feel free to ignore): 17-line copy of
`createTableWithNestedStructSchema` differing only in the select list and the
hardcoded `type = 'mor'`. An `extraColumns: String = ""` parameter on the
existing helper would serve both.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -167,6 +148,128 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase with Sp
}
}
+ test("Test NestedSchemaPruning prunes nested struct when array and map
columns are present") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithComplexNestedSchema(tableName, tablePath)
+
+ // Only a single nested sub-field is projected, so "item" is pruned down
to just "name" and the
+ // unreferenced "tags" (array<struct>) and "props" (map<string,struct>)
columns are dropped.
+ // Traversing the full data schema exercises the array and map branches
of countLeaves.
+ val selectDF = spark.sql(s"SELECT id, item.name FROM $tableName")
+
+ val expectedSchema = StructType(Seq(
+ StructField("id", IntegerType, nullable = true),
+ StructField("item", StructType(Seq(StructField("name", StringType,
nullable = false))), nullable = true)
+ ))
+ val expectedReadSchemaClause = "ReadSchema:
struct<id:int,item:struct<name:string>>"
+
+ assertPrunedReadSchema(selectDF, tableName, expectedSchema,
expectedReadSchemaClause)
+
+ // Execute the query to make sure it's working as expected (smoke test)
+ selectDF.count
+ }
+ }
+
+ test("Test NestedSchemaPruning is a no-op when all nested sub-fields are
selected") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithNestedStructSchema("mor", tableName, tablePath)
+
+ // Every leaf is projected, so the pruned schema has the same leaf count
as the data schema and
+ // the rule leaves "item" untouched (the countLeaves comparison is an
equality, not a >).
+ val selectDF = spark.sql(s"SELECT id, item.name, item.price, ts FROM
$tableName")
+
+ val expectedItemStruct = StructType(Seq(
+ StructField("name", StringType, nullable = false),
+ StructField("price", IntegerType, nullable = false)
+ ))
+ assertEquals(expectedItemStruct, prunedStructTypeOf(selectDF, "item"))
+
+ selectDF.count
+ }
+ }
+
+ test("Test NestedSchemaPruning is a no-op when nested schema pruning is
disabled") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithNestedStructSchema("mor", tableName, tablePath)
+
+ // With the optimizer flag off the rule short-circuits, so "item" keeps
"price" even though only
+ // "item.name" is projected.
+ val expectedItemStruct = StructType(Seq(
+ StructField("name", StringType, nullable = false),
+ StructField("price", IntegerType, nullable = false)
+ ))
+
+ withSQLConf(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> "false") {
+ val selectDF = spark.sql(s"SELECT id, item.name FROM $tableName")
+ assertEquals(expectedItemStruct, prunedStructTypeOf(selectDF, "item"))
+ selectDF.count
+ }
+ }
+ }
+
+ private def assertPrunedReadSchema(selectDF: DataFrame,
+ tableName: String,
+ expectedSchema: StructType,
+ expectedReadSchemaClause: String,
+ hint: String = ""): Unit = {
+ // NOTE: Unfortunately, we can't use pattern-matching to extract required
fields, due to a need to maintain
+ // compatibility w/ Spark 2.4
Review Comment:
nit: stale note copied along -- Spark 2.4 support is long gone (oldest
supported profile is 3.3, Spark 3.0-3.2 removed in ef8a7ad4f17a).
```suggestion
```
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -167,6 +148,128 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase with Sp
}
}
+ test("Test NestedSchemaPruning prunes nested struct when array and map
columns are present") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithComplexNestedSchema(tableName, tablePath)
+
+ // Only a single nested sub-field is projected, so "item" is pruned down
to just "name" and the
+ // unreferenced "tags" (array<struct>) and "props" (map<string,struct>)
columns are dropped.
+ // Traversing the full data schema exercises the array and map branches
of countLeaves.
+ val selectDF = spark.sql(s"SELECT id, item.name FROM $tableName")
+
+ val expectedSchema = StructType(Seq(
+ StructField("id", IntegerType, nullable = true),
+ StructField("item", StructType(Seq(StructField("name", StringType,
nullable = false))), nullable = true)
+ ))
+ val expectedReadSchemaClause = "ReadSchema:
struct<id:int,item:struct<name:string>>"
+
+ assertPrunedReadSchema(selectDF, tableName, expectedSchema,
expectedReadSchemaClause)
+
+ // Execute the query to make sure it's working as expected (smoke test)
+ selectDF.count
+ }
+ }
+
+ test("Test NestedSchemaPruning is a no-op when all nested sub-fields are
selected") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithNestedStructSchema("mor", tableName, tablePath)
+
+ // Every leaf is projected, so the pruned schema has the same leaf count
as the data schema and
+ // the rule leaves "item" untouched (the countLeaves comparison is an
equality, not a >).
+ val selectDF = spark.sql(s"SELECT id, item.name, item.price, ts FROM
$tableName")
+
+ val expectedItemStruct = StructType(Seq(
+ StructField("name", StringType, nullable = false),
+ StructField("price", IntegerType, nullable = false)
+ ))
+ assertEquals(expectedItemStruct, prunedStructTypeOf(selectDF, "item"))
+
+ selectDF.count
+ }
+ }
+
+ test("Test NestedSchemaPruning is a no-op when nested schema pruning is
disabled") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithNestedStructSchema("mor", tableName, tablePath)
+
+ // With the optimizer flag off the rule short-circuits, so "item" keeps
"price" even though only
+ // "item.name" is projected.
+ val expectedItemStruct = StructType(Seq(
+ StructField("name", StringType, nullable = false),
+ StructField("price", IntegerType, nullable = false)
+ ))
+
+ withSQLConf(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> "false") {
+ val selectDF = spark.sql(s"SELECT id, item.name FROM $tableName")
+ assertEquals(expectedItemStruct, prunedStructTypeOf(selectDF, "item"))
+ selectDF.count
+ }
+ }
+ }
+
+ private def assertPrunedReadSchema(selectDF: DataFrame,
+ tableName: String,
+ expectedSchema: StructType,
+ expectedReadSchemaClause: String,
+ hint: String = ""): Unit = {
+ // NOTE: Unfortunately, we can't use pattern-matching to extract required
fields, due to a need to maintain
+ // compatibility w/ Spark 2.4
+ selectDF.queryExecution.executedPlan match {
+ // COW
+ case ProjectExec(_, fileScan: FileSourceScanExec) =>
+ assertEquals(tableName, fileScan.tableIdentifier.get.table)
+ assertEquals(expectedSchema, fileScan.requiredSchema, hint)
+
+ // MOR
+ case ProjectExec(_, dataScan: RowDataSourceScanExec) =>
Review Comment:
Dead code carried into the new helper: MOR reads resolve to
`ProjectExec(FileSourceScanExec)` on master (FGR-based `HadoopFsRelation`), so
this `RowDataSourceScanExec` arm never runs (jacoco: lines 233-239 unexecuted
across the whole suite). Consequently `expectedReadSchemaClause` is never
asserted (the literal at line 167 is dead) and `explain`/`executePlan` exist
only for this arm. Same story for 3 of the 4 cases in `prunedStructTypeOf`
(lines 246-248).
**Action:** drop the `RowDataSourceScanExec` arms, the
`expectedReadSchemaClause` parameter, and the `explain`/`executePlan` helpers;
add `case other => fail(s"unexpected plan shape: $other")` so a future
plan-shape change fails with a message instead of a `MatchError`.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -167,6 +148,128 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase with Sp
}
}
+ test("Test NestedSchemaPruning prunes nested struct when array and map
columns are present") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithComplexNestedSchema(tableName, tablePath)
+
+ // Only a single nested sub-field is projected, so "item" is pruned down
to just "name" and the
+ // unreferenced "tags" (array<struct>) and "props" (map<string,struct>)
columns are dropped.
+ // Traversing the full data schema exercises the array and map branches
of countLeaves.
+ val selectDF = spark.sql(s"SELECT id, item.name FROM $tableName")
+
+ val expectedSchema = StructType(Seq(
+ StructField("id", IntegerType, nullable = true),
+ StructField("item", StructType(Seq(StructField("name", StringType,
nullable = false))), nullable = true)
+ ))
+ val expectedReadSchemaClause = "ReadSchema:
struct<id:int,item:struct<name:string>>"
+
+ assertPrunedReadSchema(selectDF, tableName, expectedSchema,
expectedReadSchemaClause)
+
+ // Execute the query to make sure it's working as expected (smoke test)
+ selectDF.count
+ }
+ }
+
+ test("Test NestedSchemaPruning is a no-op when all nested sub-fields are
selected") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithNestedStructSchema("mor", tableName, tablePath)
+
+ // Every leaf is projected, so the pruned schema has the same leaf count
as the data schema and
+ // the rule leaves "item" untouched (the countLeaves comparison is an
equality, not a >).
+ val selectDF = spark.sql(s"SELECT id, item.name, item.price, ts FROM
$tableName")
+
+ val expectedItemStruct = StructType(Seq(
+ StructField("name", StringType, nullable = false),
+ StructField("price", IntegerType, nullable = false)
+ ))
+ assertEquals(expectedItemStruct, prunedStructTypeOf(selectDF, "item"))
+
+ selectDF.count
+ }
+ }
+
+ test("Test NestedSchemaPruning is a no-op when nested schema pruning is
disabled") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithNestedStructSchema("mor", tableName, tablePath)
+
+ // With the optimizer flag off the rule short-circuits, so "item" keeps
"price" even though only
+ // "item.name" is projected.
+ val expectedItemStruct = StructType(Seq(
+ StructField("name", StringType, nullable = false),
+ StructField("price", IntegerType, nullable = false)
+ ))
+
+ withSQLConf(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> "false") {
+ val selectDF = spark.sql(s"SELECT id, item.name FROM $tableName")
+ assertEquals(expectedItemStruct, prunedStructTypeOf(selectDF, "item"))
+ selectDF.count
+ }
+ }
+ }
+
+ private def assertPrunedReadSchema(selectDF: DataFrame,
+ tableName: String,
+ expectedSchema: StructType,
+ expectedReadSchemaClause: String,
+ hint: String = ""): Unit = {
+ // NOTE: Unfortunately, we can't use pattern-matching to extract required
fields, due to a need to maintain
+ // compatibility w/ Spark 2.4
+ selectDF.queryExecution.executedPlan match {
+ // COW
+ case ProjectExec(_, fileScan: FileSourceScanExec) =>
+ assertEquals(tableName, fileScan.tableIdentifier.get.table)
+ assertEquals(expectedSchema, fileScan.requiredSchema, hint)
+
+ // MOR
+ case ProjectExec(_, dataScan: RowDataSourceScanExec) =>
+ // NOTE: This is temporary solution to assert for Spark 2.4, until
it's deprecated
+ val explainedPlan = explain(selectDF.queryExecution.logical)
+ assertTrue(explainedPlan.contains(expectedReadSchemaClause))
+
+ assertEquals(tableName, dataScan.tableIdentifier.get.table)
+ assertEquals(expectedSchema, dataScan.requiredSchema, hint)
+ }
+ }
+
+ private def prunedStructTypeOf(selectDF: DataFrame, fieldName: String):
StructType = {
+ val requiredSchema = selectDF.queryExecution.executedPlan match {
+ case ProjectExec(_, fileScan: FileSourceScanExec) =>
fileScan.requiredSchema
+ case ProjectExec(_, dataScan: RowDataSourceScanExec) =>
dataScan.requiredSchema
+ case fileScan: FileSourceScanExec => fileScan.requiredSchema
+ case dataScan: RowDataSourceScanExec => dataScan.requiredSchema
+ }
+ requiredSchema(fieldName).dataType.asInstanceOf[StructType]
+ }
+
+ private def createTableWithComplexNestedSchema(tableName: String,
+ tablePath: String): Unit = {
+ spark.sql(
+ s"""
+ |CREATE TABLE $tableName USING HUDI TBLPROPERTIES (
Review Comment:
All tables in this suite are non-partitioned, so pruning over a partitioned
schema is untested. #18570 was exactly this: `FileGroupReader` dropped
mandatory partition columns from the pruned `dataSchema`, returning silent
nulls -- which a `count` smoke test cannot catch.
**Action:** add one `PARTITIONED BY` variant and assert returned values
including the partition column.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -167,6 +148,128 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase with Sp
}
}
+ test("Test NestedSchemaPruning prunes nested struct when array and map
columns are present") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithComplexNestedSchema(tableName, tablePath)
+
+ // Only a single nested sub-field is projected, so "item" is pruned down
to just "name" and the
+ // unreferenced "tags" (array<struct>) and "props" (map<string,struct>)
columns are dropped.
+ // Traversing the full data schema exercises the array and map branches
of countLeaves.
+ val selectDF = spark.sql(s"SELECT id, item.name FROM $tableName")
+
+ val expectedSchema = StructType(Seq(
+ StructField("id", IntegerType, nullable = true),
+ StructField("item", StructType(Seq(StructField("name", StringType,
nullable = false))), nullable = true)
+ ))
+ val expectedReadSchemaClause = "ReadSchema:
struct<id:int,item:struct<name:string>>"
+
+ assertPrunedReadSchema(selectDF, tableName, expectedSchema,
expectedReadSchemaClause)
+
+ // Execute the query to make sure it's working as expected (smoke test)
+ selectDF.count
+ }
+ }
+
+ test("Test NestedSchemaPruning is a no-op when all nested sub-fields are
selected") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithNestedStructSchema("mor", tableName, tablePath)
+
+ // Every leaf is projected, so the pruned schema has the same leaf count
as the data schema and
+ // the rule leaves "item" untouched (the countLeaves comparison is an
equality, not a >).
+ val selectDF = spark.sql(s"SELECT id, item.name, item.price, ts FROM
$tableName")
+
+ val expectedItemStruct = StructType(Seq(
+ StructField("name", StringType, nullable = false),
+ StructField("price", IntegerType, nullable = false)
+ ))
+ assertEquals(expectedItemStruct, prunedStructTypeOf(selectDF, "item"))
Review Comment:
This assertion cannot fail: with every leaf projected,
`requiredSchema("item")` is `struct<name,price>` whether the rule no-ops on the
leaf-count equality, fires and rebuilds an identical relation, or is removed
from the optimizer entirely (verified: byte-identical plan with the rule in
`spark.sql.optimizer.excludedRules`). It pins no behavior.
**Action:** drop this test, or make it discriminating, e.g. assert
`queryExecution.optimizedPlan` equals the plan of the same query executed with
the pruning rule excluded.
--
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]
