This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 379f566530d0 fix(Spark): Fix nested struct predicate pushdown for
HoodieFileGroupReaderBasedFileFormat (#20029)
379f566530d0 is described below
commit 379f566530d0a73e865864a2a2878c2b5b94da42
Author: Prashant Wason <[email protected]>
AuthorDate: Tue Oct 6 16:51:39 2026 -0700
fix(Spark): Fix nested struct predicate pushdown for
HoodieFileGroupReaderBasedFileFormat (#20029)
---
.../HoodieFileGroupReaderBasedFileFormat.scala | 17 ++-
.../parquet/ParquetSchemaEvolutionUtils.scala | 55 +++++--
.../parquet/TestParquetSchemaEvolutionUtils.scala | 77 +++++++++-
.../TestNestedSchemaPruningOptimization.scala | 158 ++++++++++++++++++++-
.../apache/spark/sql/hudi/ddl/TestSpark3DDL.scala | 6 +
5 files changed, 296 insertions(+), 17 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
index f1285f95ca0e..16f9a7554756 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
@@ -52,6 +52,8 @@ import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType}
import org.apache.spark.util.SerializableConfiguration
+import java.util.Locale
+
import scala.collection.JavaConverters.mapAsJavaMapConverter
trait HoodieFormatTrait {
@@ -90,7 +92,20 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath: String,
private lazy val hasTimestampMillisFieldInTableSchema =
HoodieSchemaRepair.hasTimestampMillisField(schema)
private lazy val supportBatchWithTableSchema = HoodieSparkUtils.gteqSpark3_5
|| !hasTimestampMillisFieldInTableSchema
- override def shortName(): String = "HudiFileGroup"
+ /**
+ * Returns the base file format's short name so that Spark's nested
predicate pushdown allowlist
+ * (spark.sql.optimizer.nestedPredicatePushdown.supportedFileSources,
default "parquet,orc")
+ * matches on every read path. Tables with multiple base file formats report
"parquet", since
+ * both the Parquet and ORC filter translation resolve nested column names.
Other formats (Lance,
+ * Vortex) report their own name, which keeps them off the default allowlist.
+ */
+ override def shortName(): String = {
+ if (isMultipleBaseFileFormatsEnabled || hoodieFileFormat == null) {
+ "parquet"
+ } else {
+ hoodieFileFormat.name().toLowerCase(Locale.ROOT)
+ }
+ }
override def toString: String = "HoodieFileGroupReaderBasedFileFormat"
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
index b9470141aac1..b99ba83850e7 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
@@ -44,6 +44,8 @@ import org.apache.parquet.schema.{GroupType, MessageType,
Type => ParquetType}
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
import org.apache.spark.sql.HoodieSchemaUtils
import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
UnsafeProjection}
+import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
+import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
import org.apache.spark.sql.execution.datasources.SparkSchemaTransformUtils
import
org.apache.spark.sql.execution.datasources.parquet.ParquetSchemaEvolutionUtils.pruneInternalSchema
import org.apache.spark.sql.sources._
@@ -52,6 +54,7 @@ import org.apache.spark.sql.types.{ArrayType, AtomicType,
BinaryType, DataType,
import java.time.ZoneId
import scala.collection.convert.ImplicitConversions.`collection
AsScalaIterable`
+import scala.util.Try
class ParquetSchemaEvolutionUtils(readConf: Configuration,
filePath: Path,
@@ -192,31 +195,31 @@ object ParquetSchemaEvolutionUtils {
} else {
oldFilter match {
case eq: EqualTo =>
- val newAttribute =
InternalSchemaUtils.reBuildFilterName(eq.attribute, fileSchema, querySchema)
+ val newAttribute = rebuildFilterName(eq.attribute, fileSchema,
querySchema)
if (newAttribute.isEmpty) AlwaysTrue else eq.copy(attribute =
newAttribute)
case eqs: EqualNullSafe =>
- val newAttribute =
InternalSchemaUtils.reBuildFilterName(eqs.attribute, fileSchema, querySchema)
+ val newAttribute = rebuildFilterName(eqs.attribute, fileSchema,
querySchema)
if (newAttribute.isEmpty) AlwaysTrue else eqs.copy(attribute =
newAttribute)
case gt: GreaterThan =>
- val newAttribute =
InternalSchemaUtils.reBuildFilterName(gt.attribute, fileSchema, querySchema)
+ val newAttribute = rebuildFilterName(gt.attribute, fileSchema,
querySchema)
if (newAttribute.isEmpty) AlwaysTrue else gt.copy(attribute =
newAttribute)
case gtr: GreaterThanOrEqual =>
- val newAttribute =
InternalSchemaUtils.reBuildFilterName(gtr.attribute, fileSchema, querySchema)
+ val newAttribute = rebuildFilterName(gtr.attribute, fileSchema,
querySchema)
if (newAttribute.isEmpty) AlwaysTrue else gtr.copy(attribute =
newAttribute)
case lt: LessThan =>
- val newAttribute =
InternalSchemaUtils.reBuildFilterName(lt.attribute, fileSchema, querySchema)
+ val newAttribute = rebuildFilterName(lt.attribute, fileSchema,
querySchema)
if (newAttribute.isEmpty) AlwaysTrue else lt.copy(attribute =
newAttribute)
case lte: LessThanOrEqual =>
- val newAttribute =
InternalSchemaUtils.reBuildFilterName(lte.attribute, fileSchema, querySchema)
+ val newAttribute = rebuildFilterName(lte.attribute, fileSchema,
querySchema)
if (newAttribute.isEmpty) AlwaysTrue else lte.copy(attribute =
newAttribute)
case i: In =>
- val newAttribute =
InternalSchemaUtils.reBuildFilterName(i.attribute, fileSchema, querySchema)
+ val newAttribute = rebuildFilterName(i.attribute, fileSchema,
querySchema)
if (newAttribute.isEmpty) AlwaysTrue else i.copy(attribute =
newAttribute)
case isn: IsNull =>
- val newAttribute =
InternalSchemaUtils.reBuildFilterName(isn.attribute, fileSchema, querySchema)
+ val newAttribute = rebuildFilterName(isn.attribute, fileSchema,
querySchema)
if (newAttribute.isEmpty) AlwaysTrue else isn.copy(attribute =
newAttribute)
case isnn: IsNotNull =>
- val newAttribute =
InternalSchemaUtils.reBuildFilterName(isnn.attribute, fileSchema, querySchema)
+ val newAttribute = rebuildFilterName(isnn.attribute, fileSchema,
querySchema)
if (newAttribute.isEmpty) AlwaysTrue else isnn.copy(attribute =
newAttribute)
case And(left, right) =>
And(rebuildFilterFromParquet(left, fileSchema, querySchema),
rebuildFilterFromParquet(right, fileSchema, querySchema))
@@ -225,13 +228,13 @@ object ParquetSchemaEvolutionUtils {
case Not(child) =>
Not(rebuildFilterFromParquet(child, fileSchema, querySchema))
case ssw: StringStartsWith =>
- val newAttribute =
InternalSchemaUtils.reBuildFilterName(ssw.attribute, fileSchema, querySchema)
+ val newAttribute = rebuildFilterName(ssw.attribute, fileSchema,
querySchema)
if (newAttribute.isEmpty) AlwaysTrue else ssw.copy(attribute =
newAttribute)
case ses: StringEndsWith =>
- val newAttribute =
InternalSchemaUtils.reBuildFilterName(ses.attribute, fileSchema, querySchema)
+ val newAttribute = rebuildFilterName(ses.attribute, fileSchema,
querySchema)
if (newAttribute.isEmpty) AlwaysTrue else ses.copy(attribute =
newAttribute)
case sc: StringContains =>
- val newAttribute =
InternalSchemaUtils.reBuildFilterName(sc.attribute, fileSchema, querySchema)
+ val newAttribute = rebuildFilterName(sc.attribute, fileSchema,
querySchema)
if (newAttribute.isEmpty) AlwaysTrue else sc.copy(attribute =
newAttribute)
case AlwaysTrue =>
AlwaysTrue
@@ -243,6 +246,34 @@ object ParquetSchemaEvolutionUtils {
}
}
+ /**
+ * Re-spells one filter column name onto the file schema. With nested
predicate pushdown, Spark
+ * names a column by its multipart path with every part that is not a plain
identifier quoted
+ * (`名字`, item.`my-name`, `a.b`), while the internal schema keys fields by
their dot-joined
+ * parts. The name is parsed into parts for the lookup, and the file's name
is quoted back the
+ * same way, which is how ParquetFilters keys its columns. A name that does
not parse is one
+ * Spark passed through raw (nested pushdown off) and is looked up as it is.
+ */
+ private def rebuildFilterName(name: String, fileSchema: InternalSchema,
querySchema: InternalSchema): String = {
+ Try(CatalystSqlParser.parseMultipartIdentifier(name)).toOption match {
+ case Some(parts) =>
+ val fullName = parts.mkString(".")
+ val fileFullName = InternalSchemaUtils.reBuildFilterName(fullName,
fileSchema, querySchema)
+ if (fileFullName.isEmpty) {
+ ""
+ } else if (fileFullName == fullName) {
+ name
+ } else {
+ // Any level of the path may have been renamed, so resolve each
prefix by field id.
+ parts.indices.map { i =>
+ fileSchema.findField(querySchema.findIdByName(parts.take(i +
1).mkString("."))).name()
+ }.quoted
+ }
+ case None =>
+ InternalSchemaUtils.reBuildFilterName(name, fileSchema, querySchema)
+ }
+ }
+
/**
* Fails fast when schema-on-read meets a shredded variant file. The
internal schema models a
* variant as a two-field {metadata, value} record (with sentinel negative
field ids, see
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
index 463c1993b888..686288fb8a23 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
@@ -20,8 +20,10 @@
package org.apache.spark.sql.execution.datasources.parquet
import org.apache.hudi.common.schema.{HoodieSchema, HoodieSchemaField,
HoodieSchemaType}
-import org.apache.hudi.common.schema.internal.InternalSchema
+import org.apache.hudi.common.schema.internal.{InternalSchema, Type =>
InternalType, Types => InternalTypes}
+import org.apache.hudi.common.schema.internal.action.TableChanges
import org.apache.hudi.common.schema.internal.convert.InternalSchemaConverter
+import org.apache.hudi.common.schema.internal.utils.SchemaChangeUtils
import org.apache.hudi.common.util
import org.apache.hudi.exception.HoodieException
@@ -394,6 +396,79 @@ class TestParquetSchemaEvolutionUtils {
ParquetSchemaEvolutionUtils.rebuildFilterFromParquet(untouched, null,
null))
}
+ /**
+ * A nested field renamed under schema-on-read is re-spelled onto the file's
name, as a dotted
+ * path, rather than collapsing to AlwaysTrue (which would read correctly
but skip nothing).
+ */
+ @Test
+ def testRebuildFilterFromParquetRespellsNestedFilters(): Unit = {
+ val intType = HoodieSchema.create(HoodieSchemaType.INT)
+ val stringType = HoodieSchema.create(HoodieSchemaType.STRING)
+ val fileSchema = internalSchemaOf(("id", intType),
+ ("item", recordOf("item", ("name", stringType), ("price", intType))))
+ val querySchema = evolve(fileSchema, renames = Seq("item.name" ->
"title"), adds = Seq(("item", "added")))
+
+ def rebuild(filter: Filter): Filter =
+ ParquetSchemaEvolutionUtils.rebuildFilterFromParquet(filter, fileSchema,
querySchema)
+
+ Assertions.assertEquals(EqualTo("item.name", "a1"),
rebuild(EqualTo("item.title", "a1")))
+ Assertions.assertEquals(GreaterThan("item.price", 5),
rebuild(GreaterThan("item.price", 5)))
+ Assertions.assertEquals(AlwaysTrue, rebuild(IsNotNull("item.added")))
+ }
+
+ /**
+ * With nested predicate pushdown, Spark quotes every name part that is not
a plain identifier
+ * (non-ASCII, hyphens, dots), so the filter carries e.g. `名字` or
item.`my-title`. The rebuild
+ * must resolve those against the internal schema and hand ParquetFilters
the file's name quoted
+ * the same way, since that is how ParquetFilters keys its columns.
+ */
+ @Test
+ def testRebuildFilterFromParquetHandlesQuotedNames(): Unit = {
+ // Built from internal types directly, since these names are not valid
Avro names.
+ def field(id: Int, name: String, fieldType: InternalType):
InternalTypes.Field =
+ InternalTypes.Field.get(id, true, name, fieldType)
+ val fileSchema = new InternalSchema(InternalTypes.RecordType.get(
+ field(0, "名字", InternalTypes.StringType.get()),
+ field(1, "old col", InternalTypes.IntType.get()),
+ field(2, "a.b", InternalTypes.IntType.get()),
+ field(3, "item", InternalTypes.RecordType.get(
+ field(4, "my-name", InternalTypes.StringType.get()),
+ field(5, "price", InternalTypes.IntType.get())))))
+ val querySchema = evolve(fileSchema,
+ renames = Seq("old col" -> "new-col", "item.my-name" -> "my-title"),
adds = Seq(("", "新增")))
+
+ def rebuild(filter: Filter): Filter =
+ ParquetSchemaEvolutionUtils.rebuildFilterFromParquet(filter, fileSchema,
querySchema)
+
+ // Untouched columns keep their quoted spelling.
+ Assertions.assertEquals(EqualTo("`名字`", "x"), rebuild(EqualTo("`名字`",
"x")))
+ Assertions.assertEquals(EqualTo("`a.b`", 1), rebuild(EqualTo("`a.b`", 1)))
+ // Renamed columns are re-spelled onto the file's name, quoted where it
needs to be.
+ Assertions.assertEquals(GreaterThan("`old col`", 1),
rebuild(GreaterThan("`new-col`", 1)))
+ Assertions.assertEquals(EqualTo("item.`my-name`", "x"),
rebuild(EqualTo("item.`my-title`", "x")))
+ // A quoted column the file does not hold collapses to AlwaysTrue.
+ Assertions.assertEquals(AlwaysTrue, rebuild(IsNotNull("`新增`")))
+ // Without nested pushdown Spark passes top-level names through raw, which
must keep working.
+ Assertions.assertEquals(EqualTo("名字", "x"), rebuild(EqualTo("名字", "x")))
+ Assertions.assertEquals(GreaterThan("old col", 1),
rebuild(GreaterThan("new-col", 1)))
+ }
+
+ /** Applies schema-on-read renames, then adds int columns under the given
parent ("" for top level). */
+ private def evolve(schema: InternalSchema,
+ renames: Seq[(String, String)],
+ adds: Seq[(String, String)]): InternalSchema = {
+ val updates = TableChanges.ColumnUpdateChange.get(schema)
+ renames.foreach { case (name, newName) => updates.renameColumn(name,
newName) }
+ val renamed = SchemaChangeUtils.applyTableChanges2Schema(schema, updates)
+ val addChange = TableChanges.ColumnAddChange.get(renamed)
+ adds.foreach { case (parent, name) => addChange.addColumns(parent, name,
InternalTypes.IntType.get(), null) }
+ SchemaChangeUtils.applyTableChanges2Schema(renamed, addChange)
+ }
+
+ private def recordOf(name: String, fields: (String, HoodieSchema)*):
HoodieSchema =
+ HoodieSchema.createRecord(name, "org.apache.hudi.test", null,
+ Arrays.asList(fields.map { case (n, schema) => HoodieSchemaField.of(n,
schema) }: _*))
+
/**
* The read configuration may be shared by other readers: a file that needs
its own requested schema or a
* pushed filter gets a copy, and any other file reads with the read
configuration itself.
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala
index 3ca08f7eec80..f4b539de252d 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala
@@ -17,9 +17,10 @@
package org.apache.spark.sql.hudi.common
+import org.apache.hudi.{HoodieSchemaConversionUtils, HoodieTableSchema}
import org.apache.hudi.client.SparkRDDWriteClient
import org.apache.hudi.client.common.HoodieSparkEngineContext
-import org.apache.hudi.common.model.{HoodieAvroPayload, HoodieAvroRecord,
HoodieKey, HoodieRecord, HoodieTableType}
+import org.apache.hudi.common.model.{HoodieAvroPayload, HoodieAvroRecord,
HoodieFileFormat, HoodieKey, HoodieRecord, HoodieTableType}
import org.apache.hudi.common.schema.{HoodieSchema, HoodieSchemaType}
import org.apache.hudi.common.table.{HoodieTableMetaClient,
TableSchemaResolver}
import org.apache.hudi.common.table.read.CustomPayloadForTesting
@@ -31,10 +32,11 @@ import
org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
import org.apache.avro.generic.{GenericData, GenericRecord}
import org.apache.spark.api.java.JavaSparkContext
import org.apache.spark.sql.{DataFrame, Row}
-import org.apache.spark.sql.execution.{FileSourceScanExec, ProjectExec}
+import org.apache.spark.sql.execution.{FileSourceScanExec, ProjectExec,
SparkPlan}
+import
org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{ArrayType, IntegerType, MapType,
StringType, StructField, StructType}
-import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
import scala.collection.JavaConverters._
@@ -299,6 +301,156 @@ 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"
+
+ // NOTE: Meta fields stay enabled since the SQL UPDATE on MOR needs
them to locate the file group
+ createTableWithNestedStructSchema(tableType, tableName, tablePath,
populateMetaFields = true)
+ // Move id 1 off the filtered value and add id 2 on it; for MOR the
update lands in a log
+ // file, so the read has to merge a base record that matches with a
log record that does not.
+ spark.sql(s"INSERT INTO $tableName SELECT 2 AS id,
named_struct('name', 'a1', 'price', 20) AS item, 123457 AS ts")
+ spark.sql(s"UPDATE $tableName SET item = named_struct('name', 'zz',
'price', 10) WHERE id = 1")
+ if (tableType == "mor") {
+ assertHasLogFiles(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(2))
+ checkAnswer(s"SELECT id FROM $tableName WHERE item.name =
'zz'")(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(2))
+ }
+ }
+ }
+ }
+
+ 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,
populateMetaFields = true)
+ spark.sql(s"ALTER TABLE $tableName RENAME COLUMN item.name TO title")
+ // For MOR, also merge in a log file written under the new name. COW
is left alone so its
+ // base file keeps the old name (an update would rewrite it under
the new one).
+ if (tableType == "mor") {
+ spark.sql(s"UPDATE $tableName SET ts = ts + 1 WHERE id = 1")
+ assertHasLogFiles(tablePath)
+ }
+
+ // 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("Nested struct predicate pushdown with schema on read and quoted column
names") {
+ withTempDir { tmp =>
+ Seq("cow", "mor").foreach { tableType =>
+ // NOTE: A top-level rename fails the incoming-schema check unless
column drops are allowed
+ withSQLConf("hoodie.schema.on.read.enable" -> "true",
+ "hoodie.datasource.write.schema.allow.auto.evolution.column.drop" ->
"true") {
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ // Names that are not plain identifiers reach the reader
backtick-quoted (`名字`,
+ // item.`名称`) once nested predicate pushdown is enabled for the
source. Only names
+ // Avro accepts can be written, so hyphens and dots are covered by
the unit tests.
+ spark.sql(
+ s"""
+ |CREATE TABLE $tableName USING HUDI
+ |TBLPROPERTIES (type = '$tableType', primaryKey = 'id',
orderingFields = 'ts')
+ |LOCATION '$tablePath'
+ |AS SELECT
+ | 1 AS id,
+ | '李明' AS `名字`,
+ | 10 AS `旧分数`,
+ | named_struct('名称', 'a1', 'price', 10) AS item,
+ | 123456 AS ts
+ """.stripMargin)
+ spark.sql(s"ALTER TABLE $tableName RENAME COLUMN `旧分数` TO `新分数`")
+ spark.sql(s"ALTER TABLE $tableName RENAME COLUMN item.`名称` TO `标题`")
+ if (tableType == "mor") {
+ spark.sql(s"UPDATE $tableName SET ts = ts + 1 WHERE id = 1")
+ assertHasLogFiles(tablePath)
+ }
+
+ val nestedQuery = s"SELECT id FROM $tableName WHERE item.`标题` = 'a1'"
+ val pushedFilters = pushedFiltersOf(spark.sql(nestedQuery))
+ assertTrue(pushedFilters.contains("EqualTo(item.`标题`,a1)"),
+ s"[$tableType] EqualTo(item.`标题`,a1) should be pushed down, got
$pushedFilters")
+ checkAnswer(nestedQuery)(Seq(1))
+ checkAnswer(s"SELECT id FROM $tableName WHERE item.`标题` = 'a2'")()
+ checkAnswer(s"SELECT id FROM $tableName WHERE `名字` = '李明'")(Seq(1))
+ checkAnswer(s"SELECT id FROM $tableName WHERE `新分数` > 5")(Seq(1))
+ checkAnswer(s"SELECT id FROM $tableName WHERE `新分数` > 50")()
+ }
+ }
+ }
+ }
+
+ 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())
+ assertEquals("lance", formatFor(HoodieFileFormat.LANCE).shortName())
+ assertEquals("vortex", formatFor(HoodieFileFormat.VORTEX).shortName())
+ assertEquals("parquet", formatFor(HoodieFileFormat.ORC,
multipleBaseFileFormats = true).shortName())
+ }
+
+ private def pushedFiltersOf(selectDF: DataFrame): String =
+
findFileScan(selectDF.queryExecution.executedPlan).metadata.getOrElse("PushedFilters",
"")
+
+ private def assertHasLogFiles(tablePath: String): Unit = {
+ val stream = java.nio.file.Files.walk(java.nio.file.Paths.get(tablePath))
+ try {
+ val hasLogFile = stream.iterator().asScala.exists { p =>
+ !p.toString.contains("/.hoodie/") &&
p.getFileName.toString.contains(".log.")
+ }
+ assertTrue(hasLogFile, s"Expected log files under $tablePath")
+ } finally {
+ stream.close()
+ }
+ }
+
+ private def findFileScan(plan: SparkPlan): FileSourceScanExec = {
+ val scans = plan.collect { case f: FileSourceScanExec => f }
+ assert(scans.nonEmpty, s"No FileSourceScanExec found in plan:\n$plan")
+ scans.head
+ }
+
private def assertPrunedReadSchema(selectDF: DataFrame,
tableName: String,
expectedSchema: StructType): Unit = {
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestSpark3DDL.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestSpark3DDL.scala
index 2219fefc87c3..d8c1d00849c6 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestSpark3DDL.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestSpark3DDL.scala
@@ -563,6 +563,12 @@ class TestSpark3DDL extends HoodieSparkSqlTestBase {
| (1,3,'李明', '读书', 100,180.0001,99.0001,DATE'2021-12-26',
DATE'2021-12-26')
|""".stripMargin)
+ // Filters on non-ASCII columns, including one renamed after the
base file was written,
+ // are pushed down with backtick-quoted names and must resolve
against the file schema.
+ checkAnswer(spark.sql(s"select id from $tableName where `名字` =
'李明'").collect())(Seq(1))
+ checkAnswer(spark.sql(s"select id from $tableName where `爱好_Best` =
'读书'").collect())(Seq(1))
+ checkAnswer(spark.sql(s"select id from $tableName where `爱好_Best` =
'跑步'").collect())()
+
// alter date to string
spark.sql(s"alter table $tableName alter column `上次更新时间` type string
")
checkAnswer(spark.sql(s"select `上次更新时间` from $tableName").collect())(
