prashantwason commented on code in PR #20029:
URL: https://github.com/apache/hudi/pull/20029#discussion_r4197639422
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/HoodieSparkSessionExtension.scala:
##########
@@ -57,3 +63,26 @@ class HoodieSparkSessionExtension extends
(SparkSessionExtensions => Unit)
sparkAdapter.injectPlannerStrategies(extensions)
}
}
+
+/**
+ * Ensures that Spark's nested predicate pushdown allowlist includes
"hoodiefilegroup"
+ * so that nested struct field predicates (e.g., msg.field = 'value') are
pushed down
+ * to the Parquet reader when using HoodieFileGroupReaderBasedFileFormat.
+ */
+case class HoodieNestedPredicatePushdownConfig(spark: SparkSession) extends
Rule[LogicalPlan] with Logging {
Review Comment:
Outdated: the session-extension approach is gone. `shortName()` now returns
the base file format's name (`parquet` / `orc`), so the default allowlist
matches on every read path, including plain `spark.read.format("hudi")`, with
no config mutation (0c24ffd).
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/HoodieSparkSessionExtension.scala:
##########
@@ -57,3 +63,26 @@ class HoodieSparkSessionExtension extends
(SparkSessionExtensions => Unit)
sparkAdapter.injectPlannerStrategies(extensions)
}
}
+
+/**
+ * Ensures that Spark's nested predicate pushdown allowlist includes
"hoodiefilegroup"
+ * so that nested struct field predicates (e.g., msg.field = 'value') are
pushed down
+ * to the Parquet reader when using HoodieFileGroupReaderBasedFileFormat.
+ */
+case class HoodieNestedPredicatePushdownConfig(spark: SparkSession) extends
Rule[LogicalPlan] with Logging {
+ @volatile private var applied = false
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ if (!applied) {
+ val key =
"spark.sql.optimizer.nestedPredicatePushdown.supportedFileSources"
+ val current = spark.conf.get(key, "parquet,orc")
+ if (!current.toLowerCase.contains("hoodiefilegroup")) {
Review Comment:
Outdated: nothing touches
`spark.sql.optimizer.nestedPredicatePushdown.supportedFileSources` any more
(0c24ffd). A user who empties it to turn nested pushdown off now gets exactly
that, and the negative leg in `TestNestedSchemaPruningOptimization` covers it.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/HoodieSparkSessionExtension.scala:
##########
@@ -57,3 +63,26 @@ class HoodieSparkSessionExtension extends
(SparkSessionExtensions => Unit)
sparkAdapter.injectPlannerStrategies(extensions)
}
}
+
+/**
+ * Ensures that Spark's nested predicate pushdown allowlist includes
"hoodiefilegroup"
+ * so that nested struct field predicates (e.g., msg.field = 'value') are
pushed down
+ * to the Parquet reader when using HoodieFileGroupReaderBasedFileFormat.
+ */
+case class HoodieNestedPredicatePushdownConfig(spark: SparkSession) extends
Rule[LogicalPlan] with Logging {
+ @volatile private var applied = false
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
Review Comment:
Outdated: the rule and the `@volatile applied` flag were removed.
`shortName()` returning the base file format replaces them (0c24ffd).
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -299,6 +299,39 @@ 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 selectDF = spark.sql(s"SELECT id FROM $tableName WHERE item.name =
'a1' AND item.price > 5")
+
+ val fileScan = findFileScan(selectDF.queryExecution.executedPlan)
+ val planString = fileScan.toString()
+
+ assertTrue(fileScan.dataFilters.nonEmpty,
+ s"[$tableType] dataFilters should contain nested predicates but was
empty")
+ assertFalse(planString.contains("PushedFilters: []"),
+ s"[$tableType] PushedFilters should not be empty — nested predicate
pushdown is not working.\n" +
+ s"Verify that shortName() returns a lowercase value present in " +
+
s"spark.sql.optimizer.nestedPredicatePushdown.supportedFileSources.\nPlan:\n$planString")
+ assertTrue(planString.contains("PushedFilters: ["),
Review Comment:
Good catch, thanks. The test now asserts on the nested predicates themselves
(`EqualTo(item.name,a1)`, `GreaterThan(item.price,5)`) in `PushedFilters`, with
a negative leg (empty allowlist) showing they're absent. The MOR leg also
merges a log file now (08101e3).
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/HoodieSparkSessionExtension.scala:
##########
@@ -45,6 +49,8 @@ class HoodieSparkSessionExtension extends
(SparkSessionExtensions => Unit)
extensions.injectOptimizerRule(ruleBuilder(_))
}
+ extensions.injectOptimizerRule(HoodieNestedPredicatePushdownConfig(_))
Review Comment:
Agreed, done in 0c24ffd: `shortName()` returns the base file format
(`parquet` / `orc`; multi-format tables report `parquet`), restoring the 0.x /
1.0.x behaviour on every read path. Lance and Vortex report their own names, so
they stay off the allowlist (f75f8b4).
--
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]
