voonhous commented on code in PR #19406:
URL: https://github.com/apache/hudi/pull/19406#discussion_r3690581432
##########
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:
Rewrote it as `Test nested schema pruning with DefaultHoodieRecordPayload`
-- asserts the pruned schema it actually produces on the FGR path.
Commented-out asserts and the dead `cow` branch are gone.
##########
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:
No longer a duplicate -- the table is now partitioned, gets a log file via
`UPDATE`, and the test projects through `tags`/`props` and filters on
`item.name` with value assertions.
##########
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:
Right -- the comments and test titles now describe the end-to-end pruning
behavior on the `HadoopFsRelation` read path and no longer mention
`countLeaves` or the rule. Will update the PR description to match and file a
separate issue for the dead `HoodieNestedSchemaPruning` rule.
##########
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:
Added `SELECT tags.k` and `SELECT props['m0'].a` with exact pruned read
schema and value assertions.
##########
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:
Added `SELECT id, item.price ... WHERE item.name = 'a1'` with value
assertion.
##########
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:
Added an `UPDATE` before the reads so every query merges base + log, and
swapped `count` for `checkAnswer`. Side note: had to keep meta fields enabled
for this table -- `UPDATE` on MOR with `hoodie.populate.meta.fields=false`
writes a log file with an empty fileId and fails on
`HoodieLogFile.parseFieldsFromPath`. Looks like a separate bug, will file.
##########
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:
Dropped it.
##########
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:
Removed.
##########
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:
Removed the `RowDataSourceScanExec` arms, `expectedReadSchemaClause`,
`explain`/`executePlan` and the `SparkAdapterSupport` mixin. Both helpers now
go through a single `fileScanOf` that fails with the plan dump on an unexpected
shape.
--
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]
