You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by vh...@apache.org on 2026/08/28 14:01:24 UTC
(hudi) branch master updated: test(spark): cover schema-on-read evolution reads through the file-group reader (#19165)
This is an automated email from the ASF dual-hosted git repository.
voonhous 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 816bd96a583c test(spark): cover schema-on-read evolution reads through the file-group reader (#19165)
816bd96a583c is described below
commit 816bd96a583c29aafa18982ab8d30187eb48490a
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Fri Aug 28 07:01:16 2026 -0700
test(spark): cover schema-on-read evolution reads through the file-group reader (#19165)
Schema-on-read evolution had no DataFrame-level test that reads an
old-schema base file through the file-group reader: the existing
add-column and type-promotion tests either rewrite every file group
before reading or drive the reader API directly, and no test asserts an
incremental query after an evolution.
Two cases are added to TestBasicSchemaEvolution, which carried a TODO for
upcasting; no new suite.
- testSchemaOnReadAddColumnSnapshotAndIncrementalRead (COW and MOR):
commit 2 adds a nullable `bonus` column and touches partition p1 only,
so p2's base file keeps the old schema (on MOR the p1 base file does
too and gains a log file). Asserts the null fill for rows served from
old-schema files, pushed-down filters over the added column against a
file that lacks it, and the exact record set plus `bonus` values of
the incremental read after the evolution.
- testSchemaOnReadTypePromotionOnMorBaseAndLogMerge (MOR, parameterized
on promoteNested): commit 2 promotes `age` from int to long in p1 only,
so the read merges an int base file with a long log file and widens
the base-only int file in p2. The second arm also promotes `nested.a`:
the vectorized read must fail fast with the ParquetSchemaEvolutionUtils
non-atomic type-change rejection, which only engages when `nested` is
projected, and with spark.sql.parquet.enableVectorizedReader=false the
row-based read must widen `nested.a` across both file shapes. COW
promotion is already covered by TestLegacyParquetReadPath.
TestHoodieFileIndex's multi-column pruning test with complex expressions
additionally asserts the parsed int partition value with getInt; the
existing toSeq(...).mkString rendering cannot tell an Integer from a
UTF8String.
Test-only change; no production code is modified.
---------
Co-authored-by: voon <vo...@gmail.com>
---
.../org/apache/hudi/TestHoodieFileIndex.scala | 3 +
.../hudi/functional/TestBasicSchemaEvolution.scala | 189 +++++++++++++++++++--
2 files changed, 182 insertions(+), 10 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieFileIndex.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieFileIndex.scala
index 8d06e257d178..3dd498564f6d 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieFileIndex.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieFileIndex.scala
@@ -530,6 +530,9 @@ class TestHoodieFileIndex extends HoodieSparkClientTestBase with ScalaAssertionS
val PartitionDirectory(partitionActualValues, filesAfterPrune) = partitionAndFilesAfterPrune.head
val partitionExpectValues = Seq("default", "2021-03-01", "5", "CN")
assertEquals(partitionExpectValues.mkString(","), partitionActualValues.toSeq(Seq(StringType)).mkString(","))
+ // `hh` is an int partition column, so the value parsed from the path must be typed: the rendered
+ // string above cannot tell Integer(5) from UTF8String("5"), getInt can
+ assertEquals(5, partitionActualValues.getInt(2))
assertEquals(getFileCountInPartitionPath(makePartitionPath(partitionNames, partitionExpectValues, complexExpressionPushDown)),
filesAfterPrune.size)
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestBasicSchemaEvolution.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestBasicSchemaEvolution.scala
index 328d969763f2..fb78d2927e58 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestBasicSchemaEvolution.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestBasicSchemaEvolution.scala
@@ -17,26 +17,26 @@
package org.apache.hudi.functional
-import org.apache.hudi.{DataSourceWriteOptions, HoodieSchemaConversionUtils, ScalaAssertionSupport, SparkAdapterSupport}
+import org.apache.hudi.{DataSourceReadOptions, DataSourceWriteOptions, HoodieSchemaConversionUtils, ScalaAssertionSupport, SparkAdapterSupport}
import org.apache.hudi.HoodieConversionUtils.toJavaOption
-import org.apache.hudi.common.config.RecordMergeMode
+import org.apache.hudi.common.config.{HoodieCommonConfig, RecordMergeMode}
import org.apache.hudi.common.model.{HoodieRecord, HoodieTableType}
import org.apache.hudi.common.table.{HoodieTableConfig, TableSchemaResolver}
import org.apache.hudi.common.util.Option
-import org.apache.hudi.config.HoodieWriteConfig
+import org.apache.hudi.config.{HoodieCompactionConfig, HoodieWriteConfig}
import org.apache.hudi.exception.SchemaCompatibilityException
import org.apache.hudi.functional.TestBasicSchemaEvolution.{dropColumn, injectColumnAt}
-import org.apache.hudi.testutils.HoodieSparkClientTestBase
+import org.apache.hudi.testutils.{DataSourceTestUtils, HoodieSparkClientTestBase}
import org.apache.hudi.util.JFunction
import org.apache.hadoop.fs.FileSystem
-import org.apache.spark.sql.{functions, Row, SaveMode, SparkSession, SparkSessionExtensions}
+import org.apache.spark.sql.{functions, DataFrame, Row, SaveMode, SparkSession, SparkSessionExtensions}
import org.apache.spark.sql.hudi.HoodieSparkSessionExtension
-import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType}
+import org.apache.spark.sql.types.{DoubleType, IntegerType, LongType, StringType, StructField, StructType}
import org.junit.jupiter.api.{AfterEach, BeforeEach}
-import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.CsvSource
+import org.junit.jupiter.params.provider.{CsvSource, EnumSource, ValueSource}
import java.util.function.Consumer
@@ -75,8 +75,6 @@ class TestBasicSchemaEvolution extends HoodieSparkClientTestBase with ScalaAsser
initHoodieStorage()
}
- // TODO add test-case for upcasting
-
@ParameterizedTest
@CsvSource(value = Array(
"COPY_ON_WRITE,bulk_insert,true",
@@ -382,6 +380,177 @@ class TestBasicSchemaEvolution extends HoodieSparkClientTestBase with ScalaAsser
// TODO add test w/ overlapping updates
}
+
+ private def schemaOnReadOpts(tableType: HoodieTableType): Map[String, String] = commonOpts ++ Map(
+ DataSourceWriteOptions.TABLE_TYPE.key -> tableType.name,
+ HoodieCommonConfig.SCHEMA_EVOLUTION_ENABLE.key -> "true",
+ // HoodieSparkSqlWriter turns inline compaction on for batch MOR writes; keep commit 2 in a log file
+ HoodieCompactionConfig.INLINE_COMPACT.key -> "false")
+
+ /**
+ * Add-column evolution under schema-on-read, read through the file-group reader. Commit 2 only
+ * touches partition p1: on COW that rewrites the p1 file group with the new column while the p2
+ * base file keeps the old schema; on MOR p1 gains a log file and the base files of both partitions
+ * keep the old schema. The snapshot read must therefore fill `bonus` with null for rows served from
+ * an old-schema base file, and a pushed-down filter over `bonus` must be dropped for a file that
+ * lacks the column (InternalSchemaUtils.reBuildFilterName's "added column" branch). The incremental
+ * read after the evolution must return exactly the commit-2 records with their `bonus` values.
+ */
+ @ParameterizedTest
+ @EnumSource(classOf[HoodieTableType])
+ def testSchemaOnReadAddColumnSnapshotAndIncrementalRead(tableType: HoodieTableType): Unit = {
+ val _spark = spark
+ import _spark.implicits._
+ val opts = schemaOnReadOpts(tableType)
+
+ // commit 1: ages 10..17, even ids in p1, odd ids in p2
+ val v1 = (0 until 8).map(i => (s"id$i", s"n$i", 10 + i, 1L, if (i % 2 == 0) "p1" else "p2"))
+ .toDF("_row_key", "name", "age", "timestamp", "partition")
+ v1.write.format("hudi")
+ .options(opts)
+ .option(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Overwrite)
+ .save(basePath)
+ val firstCompletion = DataSourceTestUtils.latestCommitCompletionTime(storage, basePath)
+
+ // commit 2, p1 only: update id2 and insert id8, both carrying the new nullable `bonus` column
+ val v2 = Seq[(String, String, Int, Long, String, scala.Option[Double])](
+ ("id2", "n2u", 12, 2L, "p1", Some(100.0d)),
+ ("id8", "n8", 20, 2L, "p1", Some(300.0d)))
+ .toDF("_row_key", "name", "age", "timestamp", "partition", "bonus")
+ v2.write.format("hudi")
+ .options(opts)
+ .option(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+
+ val snapshot = spark.read.format("hudi")
+ .option(HoodieCommonConfig.SCHEMA_EVOLUTION_ENABLE.key, "true")
+ .load(basePath)
+ assertEquals(DoubleType, snapshot.schema("bonus").dataType)
+ assertEquals(9, snapshot.count())
+ assertEquals(2, snapshot.filter("bonus is not null").count())
+ // ages are now {10..17, 20}
+ assertEquals(7, snapshot.filter("age >= 12").count())
+ assertEquals(2, snapshot.filter("age >= 12 AND bonus is not null").count())
+ // the `bonus` predicate is evaluated against p2's file, which does not contain the column
+ assertEquals(0, snapshot.filter("partition = 'p2' AND bonus is not null").count())
+ assertEquals(4, snapshot.filter("partition = 'p2' AND bonus is null").count())
+
+ val byId = snapshot.select("_row_key", "name", "age", "bonus").collect().map(r => r.getString(0) -> r).toMap
+ assertEquals("n2u", byId("id2").getString(1))
+ assertEquals(100.0d, byId("id2").getDouble(3))
+ // id1 is served from the untouched p2 base file that lacks `bonus`; id0 from the rewritten p1
+ // file group (COW) or the merged p1 base+log slice (MOR)
+ assertEquals(11, byId("id1").getInt(2))
+ assertTrue(byId("id1").isNullAt(3))
+ assertTrue(byId("id0").isNullAt(3))
+
+ val incremental = spark.read.format("hudi")
+ .option(HoodieCommonConfig.SCHEMA_EVOLUTION_ENABLE.key, "true")
+ .option(DataSourceReadOptions.QUERY_TYPE.key, DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
+ .option(DataSourceReadOptions.START_COMMIT.key, firstCompletion)
+ .load(basePath)
+ val incRows = incremental.select("_row_key", "bonus").collect().map(r => r.getString(0) -> r.getDouble(1)).toMap
+ assertEquals(Map("id2" -> 100.0d, "id8" -> 300.0d), incRows)
+ }
+
+ /**
+ * int->long promotion under schema-on-read on a MOR table. Commit 2 only touches p1, so the read
+ * spans a base-only int file (p2) and an int base file merged with a long log file (p1). The
+ * top-level `age` promotion is atomic and is read with the vectorized parquet reader on. When the
+ * same promotion is applied inside the `nested` struct the changed top-level column is no longer
+ * atomic: ParquetSchemaEvolutionUtils.getHadoopConfClone must reject it fast on the base slice
+ * instead of returning corrupt columns, and the workaround it advertises (disabling the vectorized
+ * reader) must actually widen `nested.a` across both shapes. COW is covered by
+ * TestLegacyParquetReadPath#testCowSnapshotReadWithNestedTypeChange.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = Array(false, true))
+ def testSchemaOnReadTypePromotionOnMorBaseAndLogMerge(promoteNested: Boolean): Unit = {
+ val _spark = spark
+ import _spark.implicits._
+ val opts = schemaOnReadOpts(HoodieTableType.MERGE_ON_READ)
+ val widenedBase = 10000000000L
+ // id4's `nested.a` only leaves the int range in the promoting arm
+ val expectedNestedA4 = if (promoteNested) widenedBase + 4 else 4L
+
+ // commit 1: `age` and `nested.a` are int; even ids in p1, odd ids in p2
+ val v1 = (0 until 6).map(i => (s"id$i", s"n$i", 10 + i, i, s"v$i", 1L, if (i % 2 == 0) "p1" else "p2"))
+ .toDF("_row_key", "name", "age", "a", "b", "timestamp", "partition")
+ .withColumn("nested", functions.struct(functions.col("a"), functions.col("b")))
+ .drop("a", "b")
+ v1.write.format("hudi")
+ .options(opts)
+ .option(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Overwrite)
+ .save(basePath)
+
+ // commit 2, p1 only: `age` is long now; `nested.a` is promoted to long only in the second arm
+ val v2Raw = Seq(
+ ("id2", "n2u", 12L, 2L, "v2u", 2L, "p1"),
+ ("id4", "n4u", widenedBase + 14, expectedNestedA4, "v4u", 2L, "p1"),
+ ("id6", "n6", 42L, 6L, "v6", 2L, "p1"))
+ .toDF("_row_key", "name", "age", "a", "b", "timestamp", "partition")
+ val v2 = (if (promoteNested) v2Raw else v2Raw.withColumn("a", functions.col("a").cast(IntegerType)))
+ .withColumn("nested", functions.struct(functions.col("a"), functions.col("b")))
+ .drop("a", "b")
+ v2.write.format("hudi")
+ .options(opts)
+ .option(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+
+ def loadSnapshot(): DataFrame = spark.read.format("hudi")
+ .option(HoodieCommonConfig.SCHEMA_EVOLUTION_ENABLE.key, "true")
+ .load(basePath)
+
+ def assertPromotedRows(df: DataFrame): Unit = {
+ assertEquals(7, df.count())
+ val byId = df.selectExpr("_row_key", "name", "age", "cast(nested.a as long)", "nested.b").collect()
+ .map(r => r.getString(0) -> (r.getString(1), r.getLong(2), r.getLong(3), r.getString(4))).toMap
+ // id1: untouched p2 base file, int on disk, widened on read
+ assertEquals(("n1", 11L, 1L, "v1"), byId("id1"))
+ // id0: p1 int base file merged with the p1 log file, record itself untouched
+ assertEquals(("n0", 10L, 0L, "v0"), byId("id0"))
+ // id4: updated in the log with values outside the int range
+ assertEquals(("n4u", widenedBase + 14, expectedNestedA4, "v4u"), byId("id4"))
+ assertEquals(("n6", 42L, 6L, "v6"), byId("id6"))
+ // filters over the promoted columns across int and long files: ages {10, 11, 12, 13, widened+14, 15, 42}
+ assertEquals(2, df.filter("age > 40").count())
+ assertEquals(1, df.filter("age > 1000000000").count())
+ assertEquals(if (promoteNested) 1 else 0, df.filter("nested.a > 1000000000").count())
+ }
+
+ if (!promoteNested) {
+ val snapshot = loadSnapshot()
+ assertEquals(LongType, snapshot.schema("age").dataType)
+ assertEquals(IntegerType, snapshot.schema("nested").dataType.asInstanceOf[StructType]("a").dataType)
+ assertPromotedRows(snapshot)
+ } else {
+ // the non-atomic type change must fail fast in vectorized mode rather than return corrupt columns.
+ // `nested` has to be projected for the guard to engage: a bare count() prunes it away and passes.
+ val thrown = assertThrows(classOf[Throwable]) {
+ loadSnapshot().select("_row_key", "nested").collect()
+ }
+ val causes = Iterator.iterate(thrown: Throwable)(_.getCause).takeWhile(_ != null).take(10).toSeq
+ assertTrue(causes.exists(c => c.isInstanceOf[IllegalArgumentException]
+ && String.valueOf(c.getMessage).contains("cannot be read in vectorized mode")),
+ s"Expected the non-atomic type-change rejection but got: $thrown")
+
+ val vectorizedKey = "spark.sql.parquet.enableVectorizedReader"
+ val previous = spark.conf.get(vectorizedKey, "true")
+ spark.conf.set(vectorizedKey, "false")
+ try {
+ val snapshot = loadSnapshot()
+ assertEquals(LongType, snapshot.schema("age").dataType)
+ assertEquals(LongType, snapshot.schema("nested").dataType.asInstanceOf[StructType]("a").dataType)
+ assertPromotedRows(snapshot)
+ } finally {
+ spark.conf.set(vectorizedKey, previous)
+ }
+ }
+ }
}
object TestBasicSchemaEvolution {