prashantwason commented on code in PR #20029:
URL: https://github.com/apache/hudi/pull/20029#discussion_r4197642891
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -299,6 +301,84 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase {
}
}
Review Comment:
You were right: with nested pushdown on, Spark backtick-quotes
non-identifier names (```名字```, ``item.`my-name```), and
`rebuildFilterFromParquet` threw `cannot find filter col name`. Fixed in
08101e3: the name is parsed into its parts for the internal-schema lookup, and
the file-side name is quoted back the way `ParquetFilters` keys its columns. I
added `WHERE` filters to `TestSpark3DDL` "Test Chinese table", plus a COW/MOR
test with renamed non-ASCII top-level and nested columns. Both failed with that
error before the fix.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -299,6 +301,84 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase {
}
}
+ test("Nested struct predicate pushdown") {
+ withTempDir { tmp =>
+ Seq("cow", "mor").foreach { tableType =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithNestedStructSchema(tableType, tableName, tablePath)
+
+ val query = s"SELECT id FROM $tableName WHERE item.name = 'a1' AND
item.price > 5"
+ // NOTE: Spark infers a top-level IsNotNull(item) from the nested
predicates and pushes it
+ // regardless of the allowlist, so assert on the nested
predicates themselves.
+ val nestedFilters = Seq("EqualTo(item.name,a1)",
"GreaterThan(item.price,5)")
+
+ val pushedFilters = pushedFiltersOf(spark.sql(query))
+ nestedFilters.foreach { f =>
+ assertTrue(pushedFilters.contains(f), s"[$tableType] $f should be
pushed down, got $pushedFilters")
+ }
+ checkAnswer(query)(Seq(1))
+
+ // Negative leg: with an empty allowlist the nested predicates must
not be pushed.
+ withSQLConf(SQLConf.NESTED_PREDICATE_PUSHDOWN_FILE_SOURCE_LIST.key ->
"") {
+ val prunedFilters = pushedFiltersOf(spark.sql(query))
+ nestedFilters.foreach { f =>
+ assertFalse(prunedFilters.contains(f), s"[$tableType] $f should
not be pushed down, got $prunedFilters")
+ }
+ checkAnswer(query)(Seq(1))
+ }
+ }
+ }
+ }
+
+ test("Nested struct predicate pushdown with schema on read and a renamed
nested column") {
+ withTempDir { tmp =>
+ Seq("cow", "mor").foreach { tableType =>
+ withSQLConf("hoodie.schema.on.read.enable" -> "true") {
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithNestedStructSchema(tableType, tableName, tablePath)
+ spark.sql(s"ALTER TABLE $tableName RENAME COLUMN item.name TO title")
+
+ // The base file still stores item.name, so the pushed filter must
be rewritten to the
+ // file-side name rather than dropped or failing the read.
+ val query = s"SELECT id FROM $tableName WHERE item.title = 'a1'"
+ val pushedFilters = pushedFiltersOf(spark.sql(query))
Review Comment:
Agreed, the e2e test couldn't tell those apart. Added
`testRebuildFilterFromParquetRespellsNestedFilters` to
`TestParquetSchemaEvolutionUtils`: it asserts `EqualTo("item.title")` becomes
`EqualTo("item.name")` and a nested column the file lacks becomes `AlwaysTrue`.
A sibling test covers quoted, dotted and hyphenated names (08101e3).
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -299,6 +301,84 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase {
}
}
+ test("Nested struct predicate pushdown") {
+ withTempDir { tmp =>
+ Seq("cow", "mor").foreach { tableType =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithNestedStructSchema(tableType, tableName, tablePath)
+
+ val query = s"SELECT id FROM $tableName WHERE item.name = 'a1' AND
item.price > 5"
+ // NOTE: Spark infers a top-level IsNotNull(item) from the nested
predicates and pushes it
+ // regardless of the allowlist, so assert on the nested
predicates themselves.
+ val nestedFilters = Seq("EqualTo(item.name,a1)",
"GreaterThan(item.price,5)")
+
+ val pushedFilters = pushedFiltersOf(spark.sql(query))
+ nestedFilters.foreach { f =>
+ assertTrue(pushedFilters.contains(f), s"[$tableType] $f should be
pushed down, got $pushedFilters")
+ }
+ checkAnswer(query)(Seq(1))
+
+ // Negative leg: with an empty allowlist the nested predicates must
not be pushed.
+ withSQLConf(SQLConf.NESTED_PREDICATE_PUSHDOWN_FILE_SOURCE_LIST.key ->
"") {
+ val prunedFilters = pushedFiltersOf(spark.sql(query))
+ nestedFilters.foreach { f =>
+ assertFalse(prunedFilters.contains(f), s"[$tableType] $f should
not be pushed down, got $prunedFilters")
+ }
+ checkAnswer(query)(Seq(1))
+ }
+ }
+ }
+ }
+
+ test("Nested struct predicate pushdown with schema on read and a renamed
nested column") {
+ withTempDir { tmp =>
+ Seq("cow", "mor").foreach { tableType =>
Review Comment:
Done (08101e3). The MOR legs now assert log files exist. The pushdown test
does an update that moves id 1 off the filtered value (base matches, log
doesn't); the rename tests merge a log written under the new name over a base
file under the old one.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -299,6 +301,84 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase {
}
}
+ test("Nested struct predicate pushdown") {
+ withTempDir { tmp =>
+ Seq("cow", "mor").foreach { tableType =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithNestedStructSchema(tableType, tableName, tablePath)
+
+ val query = s"SELECT id FROM $tableName WHERE item.name = 'a1' AND
item.price > 5"
+ // NOTE: Spark infers a top-level IsNotNull(item) from the nested
predicates and pushes it
+ // regardless of the allowlist, so assert on the nested
predicates themselves.
+ val nestedFilters = Seq("EqualTo(item.name,a1)",
"GreaterThan(item.price,5)")
+
+ val pushedFilters = pushedFiltersOf(spark.sql(query))
+ nestedFilters.foreach { f =>
+ assertTrue(pushedFilters.contains(f), s"[$tableType] $f should be
pushed down, got $pushedFilters")
+ }
+ checkAnswer(query)(Seq(1))
+
+ // Negative leg: with an empty allowlist the nested predicates must
not be pushed.
+ withSQLConf(SQLConf.NESTED_PREDICATE_PUSHDOWN_FILE_SOURCE_LIST.key ->
"") {
+ val prunedFilters = pushedFiltersOf(spark.sql(query))
+ nestedFilters.foreach { f =>
+ assertFalse(prunedFilters.contains(f), s"[$tableType] $f should
not be pushed down, got $prunedFilters")
+ }
+ checkAnswer(query)(Seq(1))
+ }
+ }
+ }
+ }
+
+ test("Nested struct predicate pushdown with schema on read and a renamed
nested column") {
+ withTempDir { tmp =>
+ Seq("cow", "mor").foreach { tableType =>
+ withSQLConf("hoodie.schema.on.read.enable" -> "true") {
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ createTableWithNestedStructSchema(tableType, tableName, tablePath)
+ spark.sql(s"ALTER TABLE $tableName RENAME COLUMN item.name TO title")
+
+ // The base file still stores item.name, so the pushed filter must
be rewritten to the
+ // file-side name rather than dropped or failing the read.
+ val query = s"SELECT id FROM $tableName WHERE item.title = 'a1'"
+ val pushedFilters = pushedFiltersOf(spark.sql(query))
+ assertTrue(pushedFilters.contains("EqualTo(item.title,a1)"),
+ s"[$tableType] EqualTo(item.title,a1) should be pushed down, got
$pushedFilters")
+ checkAnswer(query)(Seq(1))
+ checkAnswer(s"SELECT id FROM $tableName WHERE item.title = 'a2'")()
+ }
+ }
+ }
+ }
+
+ test("shortName matches the base file format") {
+ val unusedSchema = StructType.fromDDL("id int, ts long")
+ def formatFor(fileFormat: HoodieFileFormat, multipleBaseFileFormats:
Boolean = false) =
+ new HoodieFileGroupReaderBasedFileFormat(
+ "unused", HoodieTableSchema(unusedSchema,
+
HoodieSchemaConversionUtils.convertStructTypeToHoodieSchema(unusedSchema,
"unused", "hoodie.test")),
+ "unused", "000", Seq.empty, isMOR = false, isBootstrap = false,
isIncremental = false,
+ "", shouldUseRecordPosition = false, Seq.empty,
multipleBaseFileFormats, fileFormat)
+
+ assertEquals("parquet", formatFor(HoodieFileFormat.PARQUET).shortName())
+ assertEquals("orc", formatFor(HoodieFileFormat.ORC).shortName())
Review Comment:
Leaving ORC e2e out of this PR as you suggested. The `shortName()` unit test
pins `orc` for ORC tables, and `OrcFilters` resolves nested names natively.
--
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]
