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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
