You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "voonhous (via GitHub)" <gi...@apache.org> on 2026/10/05 12:59:38 UTC
Re: [PR] fix(Spark): Fix nested struct predicate pushdown for HoodieFileGroupReaderBasedFileFormat [hudi]
voonhous commented on code in PR #20029:
URL: https://github.com/apache/hudi/pull/20029#discussion_r4184207126
##########
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:
Can you please check, i think the SoR rename operation cannot tell `rewritten` from `dropped`.
Maybe we can add a nested case to `testRebuildFilterFromParquetRespellsFiltersOntoFileNames`, for example:
`EqualTo("item.title")` becomes `EqualTo("item.name")`.
##########
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:
Can't find a place to inline this, so will add this here.
`schema-on-read` filtering on a non-ASCII column will now throw. IIRC, `DataSourceStrategy` will backtick-quote non-ascii haracters.
We currently have tests for Mandarin characters, but it does not trigger the path that we need because the test below has no `WHERE` predicate filter.
`TestSpark3DDL "Test Chinese table"`
Let's add one and try to test it.
##########
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:
We are testing MOR here but no LOG files are generated.
Can you please do an assertion check to ensure that there are log files for the MOR case?
##########
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:
Nit: I don't think we have a ORC end-to-end test (feel free to ignore). If it's not high priority, we can ignore it and focus on parquet only.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org