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/10/16 09:58:14 UTC
(hudi) branch master updated: fix(common): widen REQUIRED shredded variant reads (#20351)
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 1ba1edce0a3f fix(common): widen REQUIRED shredded variant reads (#20351)
1ba1edce0a3f is described below
commit 1ba1edce0a3f40d4e251328dca181581a76488ef
Author: voonhous <vo...@gmail.com>
AuthorDate: Fri Oct 16 17:58:07 2026 +0800
fix(common): widen REQUIRED shredded variant reads (#20351)
A COW upsert on the default AVRO record type failed with "Invalid
default for field v: null not a {record ...}" when the base file
stored a shredded variant column as REQUIRED and the table schema had
it nullable. Spark 4.1 writes such files with default configs: a
DataFrame insert of a non-null parse_json column, or a METADATA_ONLY
bootstrap of a vanilla Spark parquet file.
toShreddedReadSchema swapped in the file's shredded schema wholesale,
nullability included, and rebuilt the requested field around it with
its null default, which Avro rejects on a non-null record.
At a shredded position the swap now stays nullable when the base side
is nullable, at every depth it walks. The read schema keeps the
requested nullability (reading a REQUIRED column as nullable is a safe
widening), and alignShreddedVariants keeps a nullable file column
nullable under a non-null request, which hit the same invalid default.
A nullable side is never narrowed.
Closes #20349.
---
.../hudi/common/avro/VariantSchemaUtils.java | 6 +-
.../hudi/common/avro/TestVariantSchemaUtils.java | 56 +++++++++++++++++
.../schema/TestVariantShreddingMixedLayouts.scala | 71 ++++++++++++++++++++--
.../TestHoodieVariantReconstructionRoundTrip.java | 57 +++++++++++++++++
4 files changed, 184 insertions(+), 6 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/avro/VariantSchemaUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/avro/VariantSchemaUtils.java
index eef5d07401a7..9bceb8c313d9 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/avro/VariantSchemaUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/avro/VariantSchemaUtils.java
@@ -729,8 +729,10 @@ public class VariantSchemaUtils {
/** Swaps at one schema position, returning the {@code base} instance when nothing changes. */
private static HoodieSchema swapShreddedVariantsAt(HoodieSchema base, HoodieSchema other, boolean baseIsFile) {
if (baseIsFile ? isShreddedVariantTarget(base, other) : isShreddedVariantTarget(other, base)) {
- // Take the other side's schema wholesale, nullability included.
- return other;
+ // Take the other side's schema, staying nullable when base is: the rebuilt field keeps base's
+ // null default, which a non-null schema cannot carry (#20349), and reading a REQUIRED column
+ // as nullable is a safe widening. A nullable other side is never narrowed.
+ return base.isNullable() && !other.isNullable() ? HoodieSchema.createNullable(other) : other;
}
boolean wasNullable = base.isNullable();
HoodieSchema baseInner = wasNullable ? base.getNonNullType() : base;
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/avro/TestVariantSchemaUtils.java b/hudi-common/src/test/java/org/apache/hudi/common/avro/TestVariantSchemaUtils.java
index ecb8d379e934..80701a507f2b 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/avro/TestVariantSchemaUtils.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/avro/TestVariantSchemaUtils.java
@@ -440,6 +440,62 @@ public class TestVariantSchemaUtils {
assertEquals(HoodieSchema.NULL_VALUE, v.getField("value").get().defaultVal().get());
}
+ @Test
+ public void testToShreddedReadSchemaKeepsRequestedNullability() {
+ // A base file can hold a shredded variant REQUIRED while the table schema has it nullable (a
+ // Spark 4.1 non-null parse_json column, or a bootstrapped vanilla Spark file; #20349). The swap
+ // keeps the requested field and its null default, so the read schema must stay nullable at
+ // every shredded position; reading a REQUIRED column as nullable is a safe widening.
+ HoodieSchema typedValue = HoodieSchema.createNullable(HoodieSchemaType.LONG);
+ HoodieSchema nullableVariant = HoodieSchema.createNullable(HoodieSchema.createVariant());
+ HoodieSchema requested = HoodieSchema.createRecord("rec", null, null, Arrays.asList(
+ HoodieSchemaField.of("v", nullableVariant, null, HoodieSchema.NULL_VALUE),
+ HoodieSchemaField.of("s", HoodieSchemaTestUtils.createRecord("s_rec",
+ HoodieSchemaField.of("inner", nullableVariant, null, HoodieSchema.NULL_VALUE))),
+ HoodieSchemaTestUtils.createArrayField("items", nullableVariant),
+ HoodieSchemaTestUtils.createMapField("m", nullableVariant),
+ HoodieSchemaField.of("w", HoodieSchema.createVariant("w_variant", null, null))));
+ HoodieSchema file = HoodieSchema.createRecord("rec", null, null, Arrays.asList(
+ HoodieSchemaField.of("v", HoodieSchemaTestUtils.createPlainShreddedVariantRecord("v", typedValue)),
+ HoodieSchemaField.of("s", HoodieSchemaTestUtils.createRecord("s_rec",
+ HoodieSchemaField.of("inner", HoodieSchemaTestUtils.createPlainShreddedVariantRecord("inner_v", typedValue)))),
+ HoodieSchemaTestUtils.createArrayField("items",
+ HoodieSchemaTestUtils.createPlainShreddedVariantRecord("item_v", typedValue)),
+ HoodieSchemaTestUtils.createMapField("m",
+ HoodieSchemaTestUtils.createPlainShreddedVariantRecord("map_v", typedValue)),
+ HoodieSchemaField.of("w", HoodieSchema.createNullable(
+ HoodieSchemaTestUtils.createPlainShreddedVariantRecord("w", typedValue)), null, HoodieSchema.NULL_VALUE)));
+
+ HoodieSchema read = VariantSchemaUtils.toShreddedReadSchema(requested, file);
+ for (HoodieSchema position : Arrays.asList(
+ read.getField("v").get().schema(),
+ read.getField("s").get().schema().getField("inner").get().schema(),
+ read.getField("items").get().schema().getElementType(),
+ read.getField("m").get().schema().getValueType(),
+ // The opposite mismatch keeps the file's nullable group: reading an OPTIONAL column as
+ // REQUIRED would be a narrowing parquet refuses.
+ read.getField("w").get().schema())) {
+ assertTrue(position.isNullable(), position.toString());
+ assertTrue(position.getNonNullType().getField("typed_value").isPresent(), position.toString());
+ }
+ assertEquals(read.getAvroSchema(), new Schema.Parser().parse(read.toString()));
+ }
+
+ @Test
+ public void testAlignShreddedVariantsKeepsNullableFileColumn() {
+ // The dual direction: the file field (with its null default) is kept and the requested variant
+ // swapped in, so a non-null requested variant must come back nullable as well.
+ HoodieSchema file = HoodieSchema.createRecord("rec", null, null, Collections.singletonList(
+ HoodieSchemaField.of("v", HoodieSchema.createNullable(HoodieSchemaTestUtils.createPlainShreddedVariantRecord(
+ "v", HoodieSchema.createNullable(HoodieSchemaType.LONG))), null, HoodieSchema.NULL_VALUE)));
+ HoodieSchema requested = HoodieSchema.createRecord("rec", null, null, Collections.singletonList(
+ HoodieSchemaField.of("v", HoodieSchema.createVariant())));
+
+ HoodieSchema v = VariantSchemaUtils.alignShreddedVariants(file, requested).getField("v").get().schema();
+ assertTrue(v.isNullable());
+ assertEquals(HoodieSchemaType.VARIANT, v.getNonNullType().getType());
+ }
+
@Test
public void testApplyInferredShreddingToConfig() {
HoodieSchema schema = schemaWithVariants();
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
index aa54553799ec..45521b4ca3c9 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
@@ -34,6 +34,7 @@ import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
import org.apache.hadoop.fs.{Path => HadoopPath}
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
+import org.apache.parquet.schema.Type
import org.apache.spark.sql.Row
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.execution.FileSourceScanExec
@@ -841,8 +842,8 @@ class TestVariantShreddingMixedLayouts extends HoodieSparkSqlTestBase with Varia
val srcPath = s"${tmp.getCanonicalPath}/source"
val tablePath = s"${tmp.getCanonicalPath}/hudi"
- // The source is written by Spark's own parquet writer: a plain unshredded variant
- // column that no Hudi write path ever touched.
+ // The source is written by Spark's own parquet writer: a variant column that no Hudi
+ // write path ever touched, shredded by Spark 4.1's default inference and REQUIRED.
spark.sql("""select cast(id as int) as id, parse_json(concat('{"a":', id, '}')) as v, """ +
"1000L as ts from range(0, 10, 1, 1)")
.write.parquet(srcPath)
@@ -876,8 +877,8 @@ class TestVariantShreddingMixedLayouts extends HoodieSparkSqlTestBase with Varia
if (tableType == "mor") {
// A log over the bootstrapped file group, so the merged read joins skeleton, source
- // file and log. COW stays read-only: an upsert there would rewrite the file group
- // into a regular base file and the bootstrap read path would be gone.
+ // file and log. COW stays read-only until its last step: an upsert rewrites the file
+ // group into a regular base file and the bootstrap read path would be gone.
spark.sql("""select cast(id as int) as id, case when id = 9 then null """ +
"""else parse_json(concat('{"a":', 100 + id, '}')) end as v, """ +
"1001L as ts from range(5, 10, 1, 1)")
@@ -925,6 +926,21 @@ class TestVariantShreddingMixedLayouts extends HoodieSparkSqlTestBase with Varia
Seq("2", 2L, null, null, null, null, null, null, null))
assert(variantProjectionPushedIntoScan(projectedSql) == pushed,
s"[$leg] PushVariantIntoScan $verdict rewritten v into a projection struct")
+
+ if (tableType == "cow") {
+ // Spark 4.1 writes the non-null parse_json source column REQUIRED, while the
+ // bootstrapped table schema has it nullable, so the COW merge reads the source file
+ // through a nullable request (#20349).
+ spark.sql("""select 2 as id, parse_json('{"a":102}') as v, 1001L as ts""")
+ .write.format("hudi")
+ .options(writeOpts)
+ .option(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(tablePath)
+ spark.read.format("hudi").load(tablePath).createOrReplaceTempView(view)
+ checkAnswer(s"select id, variant_get(v, '$$.a', 'bigint') from $view where id in (1, 2) order by id")(
+ Seq(1, 1L), Seq(2, 102L))
+ }
spark.catalog.dropTempView(view)
}
}
@@ -932,6 +948,53 @@ class TestVariantShreddingMixedLayouts extends HoodieSparkSqlTestBase with Varia
}
}
+ test("An upsert merges against a base file that holds the shredded variant REQUIRED") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // In Spark 4.1 parse_json over a non-null expression is a non-nullable column and Hudi keeps
+ // that nullability, so a DataFrame insert writes `v` REQUIRED. An upsert whose `v` is nullable
+ // then merges against that file through a nullable requested schema (#20349). Default layout
+ // configs; a DataFrame write takes the merger as an option, not as a session conf.
+ Seq(HoodieRecordType.AVRO, HoodieRecordType.SPARK).foreach { recordType =>
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ val leg = s"$recordType, $tableName"
+ val writeOpts = Map(
+ HoodieWriteConfig.TBL_NAME.key -> tableName,
+ DataSourceWriteOptions.RECORDKEY_FIELD.key -> "id",
+ DataSourceWriteOptions.ORDERING_FIELDS.key -> "ts",
+ DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME.key -> classOf[NonpartitionedKeyGenerator].getName) ++
+ (if (recordType == HoodieRecordType.SPARK) {
+ Map(HoodieWriteConfig.RECORD_MERGE_IMPL_CLASSES.key -> classOf[DefaultSparkRecordMerger].getName)
+ } else {
+ Map.empty[String, String]
+ })
+ def write(sql: String, operation: String, mode: SaveMode): Unit =
+ spark.sql(sql).write.format("hudi").options(writeOpts)
+ .option(DataSourceWriteOptions.OPERATION.key, operation).mode(mode).save(tablePath)
+
+ write("""select cast(id as int) as id, parse_json(concat('{"a":', id, '}')) as v, 1000L as ts """ +
+ "from range(0, 10, 1, 1)", DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL, SaveMode.Overwrite)
+ val baseFiles = listDataParquetFiles(tablePath)
+ assert(baseFiles.size == 1, s"[$leg] expected one base file: $baseFiles")
+ val v = variantGroupOf(baseFiles.head, "v")
+ assert(v.isRepetition(Type.Repetition.REQUIRED) && v.containsField("typed_value"),
+ s"[$leg] expected a REQUIRED shredded v: $v")
+
+ write("""select cast(id as int) as id, case when id = 6 then null """ +
+ """else parse_json(concat('{"a":', 100 + id, '}')) end as v, 1001L as ts from range(5, 7, 1, 1)""",
+ DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL, SaveMode.Append)
+
+ val view = s"${tableName}_view"
+ spark.read.format("hudi").load(tablePath).createOrReplaceTempView(view)
+ checkAnswer(s"select id, variant_get(v, '$$.a', 'bigint') from $view order by id")(
+ (0 until 10).map(id => Seq(id, if (id == 5) 105L else if (id == 6) null else id.toLong)): _*)
+ spark.catalog.dropTempView(view)
+ }
+ }
+ }
+
test("Schema-on-read reads of shredded variant files fail fast") {
assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
diff --git a/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstructionRoundTrip.java b/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstructionRoundTrip.java
index 870c8501ace0..3f86fe893524 100644
--- a/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstructionRoundTrip.java
+++ b/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstructionRoundTrip.java
@@ -51,6 +51,7 @@ import java.util.LinkedHashMap;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -180,6 +181,62 @@ class TestHoodieVariantReconstructionRoundTrip {
assertEquals(original.toJson(ZoneOffset.UTC), rebuilt.toJson(ZoneOffset.UTC));
}
+ @Test
+ void requiredShreddedColumnReadsThroughANullableRequest(@TempDir Path tmp) throws Exception {
+ // Spark 4.1 writes a non-null variant column REQUIRED (a DataFrame insert of a non-null
+ // parse_json, or a vanilla Spark file brought in by bootstrap) while the table schema the merge
+ // requests has it nullable (#20349). parquet-avro must read the REQUIRED group through the
+ // nullable shredded read schema, and every row must rebuild into the requested schema.
+ Map<String, HoodieSchema> shreddedFields = new LinkedHashMap<>();
+ shreddedFields.put("a", HoodieSchema.create(HoodieSchemaType.STRING));
+ shreddedFields.put("b", HoodieSchema.create(HoodieSchemaType.LONG));
+ HoodieSchema.Variant shreddedVariant = HoodieSchema.createVariantShreddedObject(null, null, null, shreddedFields);
+ HoodieSchema.Variant unshreddedVariant = HoodieSchema.createVariant();
+ HoodieSchema fileSchema = HoodieSchema.createRecord("r", "org.apache.hudi.test", null, Arrays.asList(
+ HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.LONG)),
+ HoodieSchemaField.of("v", shreddedVariant)));
+
+ // Fully typed, partly residual (b missing, c extra), a scalar and a variant null.
+ Spark4VariantShreddingProvider provider = new Spark4VariantShreddingProvider();
+ String[] jsons = {"{\"a\":\"x\",\"b\":5}", "{\"a\":\"y\",\"c\":true}", "42", "null"};
+ Path file = tmp.resolve("required.parquet");
+ try (AvroParquetWriter<GenericRecord> writer =
+ new AvroParquetWriter<>(new org.apache.hadoop.fs.Path(file.toString()), fileSchema.getAvroSchema())) {
+ for (int i = 0; i < jsons.length; i++) {
+ Variant variant = VariantBuilder.parseJson(jsons[i], false);
+ GenericRecord unshreddedV = new GenericData.Record(unshreddedVariant.getAvroSchema());
+ unshreddedV.put(HoodieSchema.Variant.VARIANT_METADATA_FIELD, ByteBuffer.wrap(variant.getMetadata()));
+ unshreddedV.put(HoodieSchema.Variant.VARIANT_VALUE_FIELD, ByteBuffer.wrap(variant.getValue()));
+ GenericRecord record = new GenericData.Record(fileSchema.getAvroSchema());
+ record.put("id", (long) i);
+ record.put("v", provider.shredVariantRecord(unshreddedV, shreddedVariant.getAvroSchema(), shreddedVariant));
+ writer.write(record);
+ }
+ }
+
+ HoodieStorage storage = HoodieTestUtils.getStorage(tmp.toString());
+ HoodieAvroParquetReader reader = new HoodieAvroParquetReader(storage, new StoragePath(file.toUri().toString()));
+ assertFalse(reader.getSchema().getField("v").get().schema().isNullable(), "the file must hold v REQUIRED");
+ HoodieSchema requestedSchema = HoodieSchema.createRecord("r", "org.apache.hudi.test", null, Arrays.asList(
+ HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.LONG)),
+ HoodieSchemaField.of("v", HoodieSchema.createNullable(unshreddedVariant), null, HoodieSchema.NULL_VALUE)));
+ try (ClosableIterator<IndexedRecord> iterator = reader.getIndexedRecordIterator(requestedSchema, requestedSchema)) {
+ for (int i = 0; i < jsons.length; i++) {
+ assertTrue(iterator.hasNext());
+ IndexedRecord out = iterator.next();
+ assertEquals(requestedSchema.getAvroSchema(), out.getSchema());
+ assertEquals((long) i, out.get(0));
+ GenericRecord rebuiltV = (GenericRecord) out.get(1);
+ Variant rebuilt = new Variant(
+ toBytes(rebuiltV.get(HoodieSchema.Variant.VARIANT_VALUE_FIELD)),
+ toBytes(rebuiltV.get(HoodieSchema.Variant.VARIANT_METADATA_FIELD)));
+ assertEquals(VariantBuilder.parseJson(jsons[i], false).toJson(ZoneOffset.UTC), rebuilt.toJson(ZoneOffset.UTC));
+ }
+ assertFalse(iterator.hasNext());
+ }
+ reader.close();
+ }
+
private static byte[] toBytes(Object byteBuffer) {
ByteBuffer buf = ((ByteBuffer) byteBuffer).duplicate();
byte[] out = new byte[buf.remaining()];