From af84338177f4efac1d675123c198d889e7b7cd19 Mon Sep 17 00:00:00 2001 From: Y Ethan Guo Date: Tue, 28 Jul 2026 23:07:54 -0700 Subject: [PATCH 1/2] test(spark): extend Spark 4 variant shredding coverage --- .../TestSpark4VariantShreddingProvider.java | 282 ++++++++++++++++-- 1 file changed, 261 insertions(+), 21 deletions(-) diff --git a/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java b/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java index 37800e98dc023..88d16730d6d79 100644 --- a/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java +++ b/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java @@ -22,7 +22,9 @@ import org.apache.hudi.common.schema.HoodieSchema; import org.apache.hudi.common.schema.HoodieSchemaField; import org.apache.hudi.common.schema.HoodieSchemaType; +import org.apache.hudi.exception.HoodieException; +import org.apache.avro.Conversions; import org.apache.avro.Schema; import org.apache.avro.generic.GenericData; import org.apache.avro.generic.GenericRecord; @@ -30,53 +32,138 @@ import org.apache.spark.types.variant.VariantBuilder; import org.junit.jupiter.api.Test; +import java.math.BigDecimal; import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; import java.time.ZoneOffset; import java.util.Arrays; import java.util.LinkedHashMap; import java.util.Map; +import java.util.UUID; +import static org.apache.hudi.common.schema.HoodieSchema.Variant.VARIANT_METADATA_FIELD; +import static org.apache.hudi.common.schema.HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD; +import static org.apache.hudi.common.schema.HoodieSchema.Variant.VARIANT_VALUE_FIELD; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; /** - * Round-trip coverage for {@link Spark4VariantShreddingProvider}: shred an unshredded variant, then - * reconstruct it, and assert it round-trips. This exercises {@code rebuildVariantRecord} and the - * {@code AvroVariantRow}/{@code AvroObjectRow}/{@code AvroArrayRow} accessors across scalar, object, - * and array shapes - the AVRO read-path reconstruction (#18931) that the Spark MOR SQL test cannot - * reach (Spark compaction reads base files via the InternalRow reader, not HoodieAvroParquetReader). + * Round-trip and behavior-pinning coverage for {@link Spark4VariantShreddingProvider}: shred an + * unshredded variant, then reconstruct it, asserting both the intermediate shredded schema/values and + * the reconstructed variant. This exercises {@code shredVariantRecord}, {@code rebuildVariantRecord}, + * {@code avroTypeToScalarType}, {@code convertScalarToAvro}, and the + * {@code AvroVariantRow}/{@code AvroObjectRow}/{@code AvroArrayRow} accessors across every scalar leaf + * type, object/array shapes, partial (residual) shredding, and the null/error guards - the AVRO + * read-path reconstruction that the Spark MOR SQL test cannot reach (Spark compaction reads base + * files via the InternalRow reader, not HoodieAvroParquetReader). */ class TestSpark4VariantShreddingProvider { private final Spark4VariantShreddingProvider provider = new Spark4VariantShreddingProvider(); private final Schema unshreddedSchema = HoodieSchema.createVariant().getAvroSchema(); - /** Parse json to a variant, shred it to {@code shredded}, rebuild it, assert the json round-trips. */ - private void assertRoundTrips(String json, HoodieSchema.Variant shredded) throws Exception { - Variant variant = VariantBuilder.parseJson(json, false); - GenericRecord unshreddedRecord = new GenericData.Record(unshreddedSchema); - unshreddedRecord.put(HoodieSchema.Variant.VARIANT_METADATA_FIELD, ByteBuffer.wrap(variant.getMetadata())); - unshreddedRecord.put(HoodieSchema.Variant.VARIANT_VALUE_FIELD, ByteBuffer.wrap(variant.getValue())); + /** Wrap a fully built {@link Variant} into the unshredded {metadata, value} Avro record. */ + private GenericRecord unshredded(Variant variant) { + GenericRecord record = new GenericData.Record(unshreddedSchema); + record.put(VARIANT_METADATA_FIELD, ByteBuffer.wrap(variant.getMetadata())); + record.put(VARIANT_VALUE_FIELD, ByteBuffer.wrap(variant.getValue())); + return record; + } + + private GenericRecord shred(Variant variant, HoodieSchema.Variant shredded) { + return provider.shredVariantRecord(unshredded(variant), shredded.getAvroSchema(), shredded); + } - Schema shreddedSchema = shredded.getAvroSchema(); - GenericRecord shreddedRecord = provider.shredVariantRecord(unshreddedRecord, shreddedSchema, shredded); - GenericRecord rebuilt = provider.rebuildVariantRecord(shreddedRecord, shreddedSchema, unshreddedSchema); + private Variant rebuild(GenericRecord shreddedRecord, HoodieSchema.Variant shredded) { + GenericRecord rebuilt = + provider.rebuildVariantRecord(shreddedRecord, shredded.getAvroSchema(), unshreddedSchema); + return new Variant(toBytes(rebuilt.get(VARIANT_VALUE_FIELD)), toBytes(rebuilt.get(VARIANT_METADATA_FIELD))); + } - Variant rebuiltVariant = new Variant( - toBytes(rebuilt.get(HoodieSchema.Variant.VARIANT_VALUE_FIELD)), - toBytes(rebuilt.get(HoodieSchema.Variant.VARIANT_METADATA_FIELD))); - assertEquals(variant.toJson(ZoneOffset.UTC), rebuiltVariant.toJson(ZoneOffset.UTC), - "variant did not round-trip through shred/rebuild for: " + json); + private void assertRoundTrips(Variant variant, HoodieSchema.Variant shredded) { + Variant rebuilt = rebuild(shred(variant, shredded), shredded); + assertEquals(variant.toJson(ZoneOffset.UTC), rebuilt.toJson(ZoneOffset.UTC), + "variant did not round-trip through shred/rebuild"); + } + + /** Parse json to a variant, shred it, rebuild it, assert the json round-trips. */ + private void assertRoundTrips(String json, HoodieSchema.Variant shredded) throws Exception { + assertRoundTrips(VariantBuilder.parseJson(json, false), shredded); } private void assertScalarRoundTrips(String json, HoodieSchema typedValue) throws Exception { assertRoundTrips(json, HoodieSchema.createVariantShredded(typedValue)); } + /** + * Shred {@code variant} into a scalar {@code typedValue} schema, assert the value fully shredded + * (into {@code typed_value}, exactly {@code expectedTypedValue}, no residual {@code value}) and + * round-trips back to the original variant. + */ + private void assertScalarShredsTo(Variant variant, HoodieSchema typedValue, Object expectedTypedValue) { + HoodieSchema.Variant shredded = HoodieSchema.createVariantShredded(typedValue); + GenericRecord shreddedRecord = shred(variant, shredded); + assertNotNull(shreddedRecord.get(VARIANT_METADATA_FIELD), "shredded record must carry metadata"); + assertNull(shreddedRecord.get(VARIANT_VALUE_FIELD), "a matching scalar leaves no residual value"); + assertEquals(expectedTypedValue, shreddedRecord.get(VARIANT_TYPED_VALUE_FIELD), + "scalar was not shredded into typed_value as expected"); + assertEquals(variant.toJson(ZoneOffset.UTC), rebuild(shreddedRecord, shredded).toJson(ZoneOffset.UTC), + "scalar did not round-trip through shred/rebuild"); + } + + /** + * Shred {@code variant} against a scalar {@code typedValue} it does not match: assert it is NOT + * shredded (typed_value stays null, the value lands in the residual {@code value}) yet still + * round-trips. Exercises the "decline to shred, keep in residual" fallbacks. + */ + private void assertStaysInResidual(Variant variant, HoodieSchema typedValue) { + HoodieSchema.Variant shredded = HoodieSchema.createVariantShredded(typedValue); + GenericRecord shreddedRecord = shred(variant, shredded); + assertNull(shreddedRecord.get(VARIANT_TYPED_VALUE_FIELD), "value should not have been shredded"); + assertNotNull(shreddedRecord.get(VARIANT_VALUE_FIELD), "unshredded value must survive in residual"); + assertEquals(variant.toJson(ZoneOffset.UTC), rebuild(shreddedRecord, shredded).toJson(ZoneOffset.UTC), + "residual value did not round-trip through shred/rebuild"); + } + + private static Variant scalar(java.util.function.Consumer append) { + VariantBuilder builder = new VariantBuilder(false); + append.accept(builder); + return builder.result(); + } + + // --------------------------------------------------------------------------- + // Scalar leaf types: one per branch of avroTypeToScalarType / convertScalarToAvro. + // --------------------------------------------------------------------------- + @Test void numericRoundTrips() throws Exception { assertScalarRoundTrips("42", HoodieSchema.create(HoodieSchemaType.LONG)); } + @Test + void intLeafShredsToInteger() { + // An integral variant shreds into an INT typed_value as a boxed Integer (Avro has no byte/short, + // so avroTypeToScalarType only ever emits IntegralSize.INT here). + assertScalarShredsTo(scalar(b -> b.appendLong(1000)), HoodieSchema.create(HoodieSchemaType.INT), 1000); + } + + @Test + void longLeafShredsToLong() { + assertScalarShredsTo(scalar(b -> b.appendLong(100000)), HoodieSchema.create(HoodieSchemaType.LONG), 100000L); + } + + @Test + void floatShredsAndRoundTrips() { + assertScalarShredsTo(scalar(b -> b.appendFloat(1.5f)), HoodieSchema.create(HoodieSchemaType.FLOAT), 1.5f); + } + + @Test + void doubleShredsAndRoundTrips() { + assertScalarShredsTo(scalar(b -> b.appendDouble(2.5d)), HoodieSchema.create(HoodieSchemaType.DOUBLE), 2.5d); + } + @Test void stringRoundTrips() throws Exception { assertScalarRoundTrips("\"hello world\"", HoodieSchema.create(HoodieSchemaType.STRING)); @@ -87,11 +174,68 @@ void booleanRoundTrips() throws Exception { assertScalarRoundTrips("true", HoodieSchema.create(HoodieSchemaType.BOOLEAN)); } + @Test + void binaryShredsToByteBuffer() { + byte[] payload = "not-utf8-ÿ".getBytes(StandardCharsets.ISO_8859_1); + assertScalarShredsTo(scalar(b -> b.appendBinary(payload)), + HoodieSchema.create(HoodieSchemaType.BYTES), ByteBuffer.wrap(payload)); + } + + @Test + void uuidShredsToString() { + UUID uuid = UUID.fromString("12345678-1234-1234-1234-123456789abc"); + assertScalarShredsTo(scalar(b -> b.appendUuid(uuid)), HoodieSchema.createUUID(), uuid.toString()); + } + + @Test + void dateShredsToDaysSinceEpoch() { + assertScalarShredsTo(scalar(b -> b.appendDate(19000)), HoodieSchema.createDate(), 19000); + } + + @Test + void timestampMicrosShredsToMicros() { + long micros = 1_700_000_000_000_000L; + assertScalarShredsTo(scalar(b -> b.appendTimestamp(micros)), HoodieSchema.createTimestampMicros(), micros); + } + + @Test + void localTimestampMicrosShredsToMicros() { + long micros = 1_700_000_000_000_000L; + assertScalarShredsTo(scalar(b -> b.appendTimestampNtz(micros)), HoodieSchema.createLocalTimestampMicros(), micros); + } + @Test void decimalRoundTrips() throws Exception { assertScalarRoundTrips("123.45", HoodieSchema.createDecimal(10, 2)); } + // --------------------------------------------------------------------------- + // "Decline to shred" fallbacks: value stays in the residual binary. + // --------------------------------------------------------------------------- + + @Test + void millisTimestampIsNotShreddedIntoMicrosLeaf() { + // A millisecond-precision typed_value cannot represent a micros variant timestamp, so + // avroTypeToScalarType returns null and the value is left unshredded in the residual. + assertStaysInResidual(scalar(b -> b.appendTimestamp(1_700_000_000_000_000L)), + HoodieSchema.createTimestampMillis()); + assertStaysInResidual(scalar(b -> b.appendTimestampNtz(1_700_000_000_000_000L)), + HoodieSchema.createLocalTimestampMillis()); + } + + @Test + void fixedLeafShredsBinaryToByteBuffer() { + // A FIXED typed_value maps to BinaryType, so a binary variant is shredded into typed_value (had + // avroTypeToScalarType returned null for FIXED, the value would fall to the residual instead). + byte[] payload = {1, 2, 3, 4}; + assertScalarShredsTo(scalar(b -> b.appendBinary(payload)), + HoodieSchema.createFixed("fx", "org.apache.hudi.test", null, 4), ByteBuffer.wrap(payload)); + } + + // --------------------------------------------------------------------------- + // Object and array shapes, including partial (residual) object shredding. + // --------------------------------------------------------------------------- + @Test void objectRoundTrips() throws Exception { Map shreddedFields = new LinkedHashMap<>(); @@ -100,15 +244,111 @@ void objectRoundTrips() throws Exception { assertRoundTrips("{\"a\":\"x\",\"b\":5}", HoodieSchema.createVariantShreddedObject(shreddedFields)); } + @Test + void partialObjectShreddingKeepsExtraFieldsInResidual() throws Exception { + // Shredded schema declares {a, b} but the variant provides {a, c}: "a" shreds into typed_value, + // "b" is absent (null value + null typed_value), and the extra "c" lands in the residual value. + Map shreddedFields = new LinkedHashMap<>(); + shreddedFields.put("a", HoodieSchema.create(HoodieSchemaType.STRING)); + shreddedFields.put("b", HoodieSchema.create(HoodieSchemaType.LONG)); + HoodieSchema.Variant shredded = HoodieSchema.createVariantShreddedObject(shreddedFields); + + Variant variant = VariantBuilder.parseJson("{\"a\":\"x\",\"c\":99}", false); + GenericRecord shreddedRecord = shred(variant, shredded); + + // The unmatched field forces a non-null residual value at the top level. + assertNotNull(shreddedRecord.get(VARIANT_VALUE_FIELD), "extra field must be captured in residual value"); + GenericRecord typedValue = (GenericRecord) shreddedRecord.get(VARIANT_TYPED_VALUE_FIELD); + GenericRecord bField = (GenericRecord) typedValue.get("b"); + assertNull(bField.get(VARIANT_VALUE_FIELD), "absent field b carries no residual value"); + assertNull(bField.get(VARIANT_TYPED_VALUE_FIELD), "absent field b carries no typed_value"); + + assertEquals(variant.toJson(ZoneOffset.UTC), rebuild(shreddedRecord, shredded).toJson(ZoneOffset.UTC)); + } + @Test void arrayRoundTrips() throws Exception { // typed_value for an array is array<{value, typed_value}>: each element is itself a shredded struct. HoodieSchema element = HoodieSchema.createRecord("v_array_element", "org.apache.hudi.test", null, Arrays.asList( - HoodieSchemaField.of(HoodieSchema.Variant.VARIANT_VALUE_FIELD, HoodieSchema.createNullable(HoodieSchemaType.BYTES)), - HoodieSchemaField.of(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD, HoodieSchema.create(HoodieSchemaType.LONG)))); + HoodieSchemaField.of(VARIANT_VALUE_FIELD, HoodieSchema.createNullable(HoodieSchemaType.BYTES)), + HoodieSchemaField.of(VARIANT_TYPED_VALUE_FIELD, HoodieSchema.create(HoodieSchemaType.LONG)))); assertScalarRoundTrips("[1,2,3]", HoodieSchema.createArray(element)); } + // --------------------------------------------------------------------------- + // Decimal reconstruction from the on-disk (avro-decoded) encodings a parquet reader produces: + // the shred path emits a BigDecimal, but a base file feeds rebuild a ByteBuffer / GenericFixed. + // --------------------------------------------------------------------------- + + @Test + void rebuildDecimalFromBytesEncoding() { + assertDecimalRebuildsFromEncoding(HoodieSchema.createDecimal(10, 2), false); + } + + @Test + void rebuildDecimalFromFixedEncoding() { + assertDecimalRebuildsFromEncoding( + HoodieSchema.createDecimal("dec_fixed", "org.apache.hudi.test", null, 10, 2, 8), true); + } + + private void assertDecimalRebuildsFromEncoding(HoodieSchema decimalType, boolean fixed) { + BigDecimal value = new BigDecimal("123.45"); + HoodieSchema.Variant shredded = HoodieSchema.createVariantShredded(decimalType); + GenericRecord shreddedRecord = shred(scalar(b -> b.appendDecimal(value)), shredded); + + Schema tvSchema = shredded.getAvroSchema().getField(VARIANT_TYPED_VALUE_FIELD).schema(); + Conversions.DecimalConversion conversion = new Conversions.DecimalConversion(); + Object encoded = fixed + ? conversion.toFixed(value, tvSchema, tvSchema.getLogicalType()) + : conversion.toBytes(value, tvSchema, tvSchema.getLogicalType()); + shreddedRecord.put(VARIANT_TYPED_VALUE_FIELD, encoded); + + Variant original = scalar(b -> b.appendDecimal(value)); + assertEquals(original.toJson(ZoneOffset.UTC), rebuild(shreddedRecord, shredded).toJson(ZoneOffset.UTC)); + } + + @Test + void rebuildDecimalRejectsUnexpectedEncoding() { + HoodieSchema.Variant shredded = HoodieSchema.createVariantShredded(HoodieSchema.createDecimal(10, 2)); + GenericRecord shreddedRecord = shred(scalar(b -> b.appendDecimal(new BigDecimal("1.00"))), shredded); + shreddedRecord.put(VARIANT_TYPED_VALUE_FIELD, "not-a-decimal"); + assertThrows(IllegalStateException.class, + () -> provider.rebuildVariantRecord(shreddedRecord, shredded.getAvroSchema(), unshreddedSchema)); + } + + // --------------------------------------------------------------------------- + // Null / error guards. + // --------------------------------------------------------------------------- + + @Test + void shredReturnsNullWhenValueOrMetadataMissing() { + HoodieSchema.Variant shredded = HoodieSchema.createVariantShredded(HoodieSchema.create(HoodieSchemaType.LONG)); + Variant variant = scalar(b -> b.appendLong(1)); + + GenericRecord missingValue = unshredded(variant); + missingValue.put(VARIANT_VALUE_FIELD, null); + assertNull(provider.shredVariantRecord(missingValue, shredded.getAvroSchema(), shredded)); + + GenericRecord missingMetadata = unshredded(variant); + missingMetadata.put(VARIANT_METADATA_FIELD, null); + assertNull(provider.shredVariantRecord(missingMetadata, shredded.getAvroSchema(), shredded)); + } + + @Test + void rebuildReturnsNullForNullRecord() { + HoodieSchema.Variant shredded = HoodieSchema.createVariantShredded(HoodieSchema.create(HoodieSchemaType.LONG)); + assertNull(provider.rebuildVariantRecord(null, shredded.getAvroSchema(), unshreddedSchema)); + } + + @Test + void rebuildThrowsWhenMetadataMissing() { + HoodieSchema.Variant shredded = HoodieSchema.createVariantShredded(HoodieSchema.create(HoodieSchemaType.LONG)); + GenericRecord shreddedRecord = shred(scalar(b -> b.appendLong(7)), shredded); + shreddedRecord.put(VARIANT_METADATA_FIELD, null); + assertThrows(HoodieException.class, + () -> provider.rebuildVariantRecord(shreddedRecord, shredded.getAvroSchema(), unshreddedSchema)); + } + private static byte[] toBytes(Object byteBuffer) { ByteBuffer buf = ((ByteBuffer) byteBuffer).duplicate(); byte[] out = new byte[buf.remaining()]; From caba485a3f5bdb3659108b3a74df623c14896afd Mon Sep 17 00:00:00 2001 From: voon Date: Fri, 31 Jul 2026 16:14:16 +0800 Subject: [PATCH 2/2] test(spark): address review comments on variant shredding coverage - Drop fixedLeafShredsBinaryToByteBuffer: it pinned a typed_value that cannot survive a real Avro/parquet write or read (FIXED handling bug to be filed separately) - Build every scalar leaf with a nullable typed_value via shreddedLeaf, matching the shape production emits and the shredding spec requires - Validate every shredded record against ConvertingGenericData (the write-path data model) inside the shred helper - Pin allowNumericScaleChanges: integral-into-decimal, decimal-into-long, and a lossy decimal declining to the residual - Assert the positive half of partial object shredding and add the field-level type-mismatch residual case - Replace the raw NUL-byte binary literal with an explicit byte array - Correct the scalar section comment (Byte/Short widening unreachable), rename the micros-into-millis decline test, fold numericRoundTrips into longLeafShredsToLong, upgrade string/boolean/decimal tests to assertScalarShredsTo, add ENUM and no-typed_value decline cases - Derive the decimal encoding from the schema, restore json context in round-trip failure messages, note the null-record guard is defensive --- .../TestSpark4VariantShreddingProvider.java | 149 +++++++++++++----- 1 file changed, 108 insertions(+), 41 deletions(-) diff --git a/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java b/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java index 88d16730d6d79..8abb9bf992fe8 100644 --- a/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java +++ b/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java @@ -19,6 +19,7 @@ package org.apache.hudi.variant; +import org.apache.hudi.common.avro.ConvertingGenericData; import org.apache.hudi.common.schema.HoodieSchema; import org.apache.hudi.common.schema.HoodieSchemaField; import org.apache.hudi.common.schema.HoodieSchemaType; @@ -34,7 +35,6 @@ import java.math.BigDecimal; import java.nio.ByteBuffer; -import java.nio.charset.StandardCharsets; import java.time.ZoneOffset; import java.util.Arrays; import java.util.LinkedHashMap; @@ -48,15 +48,16 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; /** * Round-trip and behavior-pinning coverage for {@link Spark4VariantShreddingProvider}: shred an * unshredded variant, then reconstruct it, asserting both the intermediate shredded schema/values and * the reconstructed variant. This exercises {@code shredVariantRecord}, {@code rebuildVariantRecord}, * {@code avroTypeToScalarType}, {@code convertScalarToAvro}, and the - * {@code AvroVariantRow}/{@code AvroObjectRow}/{@code AvroArrayRow} accessors across every scalar leaf - * type, object/array shapes, partial (residual) shredding, and the null/error guards - the AVRO - * read-path reconstruction that the Spark MOR SQL test cannot reach (Spark compaction reads base + * {@code AvroVariantRow}/{@code AvroObjectRow}/{@code AvroArrayRow} accessors across every reachable + * scalar leaf type, object/array shapes, partial (residual) shredding, and the null/error guards - the + * AVRO read-path reconstruction that the Spark MOR SQL test cannot reach (Spark compaction reads base * files via the InternalRow reader, not HoodieAvroParquetReader). */ class TestSpark4VariantShreddingProvider { @@ -73,7 +74,12 @@ private GenericRecord unshredded(Variant variant) { } private GenericRecord shred(Variant variant, HoodieSchema.Variant shredded) { - return provider.shredVariantRecord(unshredded(variant), shredded.getAvroSchema(), shredded); + GenericRecord record = provider.shredVariantRecord(unshredded(variant), shredded.getAvroSchema(), shredded); + // ConvertingGenericData.INSTANCE is the data model HoodieAvroWriteSupport hands to parquet-avro, + // so this pins that every shredded record produced here is writable at its declared schema. + assertTrue(ConvertingGenericData.INSTANCE.validate(shredded.getAvroSchema(), record), + "shredded record is not writable at its declared schema"); + return record; } private Variant rebuild(GenericRecord shreddedRecord, HoodieSchema.Variant shredded) { @@ -83,18 +89,31 @@ private Variant rebuild(GenericRecord shreddedRecord, HoodieSchema.Variant shred } private void assertRoundTrips(Variant variant, HoodieSchema.Variant shredded) { + assertRoundTrips(variant, shredded, ""); + } + + private void assertRoundTrips(Variant variant, HoodieSchema.Variant shredded, String context) { Variant rebuilt = rebuild(shred(variant, shredded), shredded); assertEquals(variant.toJson(ZoneOffset.UTC), rebuilt.toJson(ZoneOffset.UTC), - "variant did not round-trip through shred/rebuild"); + "variant did not round-trip through shred/rebuild" + context); } /** Parse json to a variant, shred it, rebuild it, assert the json round-trips. */ private void assertRoundTrips(String json, HoodieSchema.Variant shredded) throws Exception { - assertRoundTrips(VariantBuilder.parseJson(json, false), shredded); + assertRoundTrips(VariantBuilder.parseJson(json, false), shredded, " for: " + json); + } + + /** + * Wrap a scalar leaf into a shredded variant schema with a nullable {@code typed_value}, the shape + * production always emits ({@code createShreddedFieldStruct} wraps every leaf in + * {@code createNullable}, and the parquet shredding spec requires {@code typed_value} optional). + */ + private static HoodieSchema.Variant shreddedLeaf(HoodieSchema typedValue) { + return HoodieSchema.createVariantShredded(HoodieSchema.createNullable(typedValue)); } private void assertScalarRoundTrips(String json, HoodieSchema typedValue) throws Exception { - assertRoundTrips(json, HoodieSchema.createVariantShredded(typedValue)); + assertRoundTrips(json, shreddedLeaf(typedValue)); } /** @@ -103,7 +122,7 @@ private void assertScalarRoundTrips(String json, HoodieSchema typedValue) throws * round-trips back to the original variant. */ private void assertScalarShredsTo(Variant variant, HoodieSchema typedValue, Object expectedTypedValue) { - HoodieSchema.Variant shredded = HoodieSchema.createVariantShredded(typedValue); + HoodieSchema.Variant shredded = shreddedLeaf(typedValue); GenericRecord shreddedRecord = shred(variant, shredded); assertNotNull(shreddedRecord.get(VARIANT_METADATA_FIELD), "shredded record must carry metadata"); assertNull(shreddedRecord.get(VARIANT_VALUE_FIELD), "a matching scalar leaves no residual value"); @@ -113,13 +132,17 @@ private void assertScalarShredsTo(Variant variant, HoodieSchema typedValue, Obje "scalar did not round-trip through shred/rebuild"); } + private void assertScalarShredsTo(String json, HoodieSchema typedValue, Object expectedTypedValue) throws Exception { + assertScalarShredsTo(VariantBuilder.parseJson(json, false), typedValue, expectedTypedValue); + } + /** * Shred {@code variant} against a scalar {@code typedValue} it does not match: assert it is NOT * shredded (typed_value stays null, the value lands in the residual {@code value}) yet still * round-trips. Exercises the "decline to shred, keep in residual" fallbacks. */ private void assertStaysInResidual(Variant variant, HoodieSchema typedValue) { - HoodieSchema.Variant shredded = HoodieSchema.createVariantShredded(typedValue); + HoodieSchema.Variant shredded = shreddedLeaf(typedValue); GenericRecord shreddedRecord = shred(variant, shredded); assertNull(shreddedRecord.get(VARIANT_TYPED_VALUE_FIELD), "value should not have been shredded"); assertNotNull(shreddedRecord.get(VARIANT_VALUE_FIELD), "unshredded value must survive in residual"); @@ -134,14 +157,10 @@ private static Variant scalar(java.util.function.Consumer append } // --------------------------------------------------------------------------- - // Scalar leaf types: one per branch of avroTypeToScalarType / convertScalarToAvro. + // Scalar leaf types: one per reachable branch of avroTypeToScalarType / convertScalarToAvro + // (the Byte/Short widening arms in convertScalarToAvro are unreachable: Avro has no byte/short). // --------------------------------------------------------------------------- - @Test - void numericRoundTrips() throws Exception { - assertScalarRoundTrips("42", HoodieSchema.create(HoodieSchemaType.LONG)); - } - @Test void intLeafShredsToInteger() { // An integral variant shreds into an INT typed_value as a boxed Integer (Avro has no byte/short, @@ -150,8 +169,10 @@ void intLeafShredsToInteger() { } @Test - void longLeafShredsToLong() { + void longLeafShredsToLong() throws Exception { assertScalarShredsTo(scalar(b -> b.appendLong(100000)), HoodieSchema.create(HoodieSchemaType.LONG), 100000L); + // Same branch through the json entry point. + assertScalarShredsTo("42", HoodieSchema.create(HoodieSchemaType.LONG), 42L); } @Test @@ -165,18 +186,18 @@ void doubleShredsAndRoundTrips() { } @Test - void stringRoundTrips() throws Exception { - assertScalarRoundTrips("\"hello world\"", HoodieSchema.create(HoodieSchemaType.STRING)); + void stringLeafShredsToString() throws Exception { + assertScalarShredsTo("\"hello world\"", HoodieSchema.create(HoodieSchemaType.STRING), "hello world"); } @Test - void booleanRoundTrips() throws Exception { - assertScalarRoundTrips("true", HoodieSchema.create(HoodieSchemaType.BOOLEAN)); + void booleanLeafShredsToBoolean() throws Exception { + assertScalarShredsTo("true", HoodieSchema.create(HoodieSchemaType.BOOLEAN), true); } @Test void binaryShredsToByteBuffer() { - byte[] payload = "not-utf8-ÿ".getBytes(StandardCharsets.ISO_8859_1); + byte[] payload = {'n', 'o', 't', '-', 'u', 't', 'f', '8', (byte) 0xC0, (byte) 0xFF}; assertScalarShredsTo(scalar(b -> b.appendBinary(payload)), HoodieSchema.create(HoodieSchemaType.BYTES), ByteBuffer.wrap(payload)); } @@ -205,8 +226,17 @@ void localTimestampMicrosShredsToMicros() { } @Test - void decimalRoundTrips() throws Exception { - assertScalarRoundTrips("123.45", HoodieSchema.createDecimal(10, 2)); + void decimalLeafShredsToBigDecimal() throws Exception { + assertScalarShredsTo("123.45", HoodieSchema.createDecimal(10, 2), new BigDecimal("123.45")); + } + + @Test + void numericScaleChangesShredWhenExact() { + // allowNumericScaleChanges() lets Spark shred an integral variant into a decimal leaf (and a + // decimal variant into an integral leaf) when the value is exactly representable. + assertScalarShredsTo(scalar(b -> b.appendLong(5)), HoodieSchema.createDecimal(10, 2), new BigDecimal("5.00")); + assertScalarShredsTo(scalar(b -> b.appendDecimal(new BigDecimal("5.00"))), + HoodieSchema.create(HoodieSchemaType.LONG), 5L); } // --------------------------------------------------------------------------- @@ -214,7 +244,7 @@ void decimalRoundTrips() throws Exception { // --------------------------------------------------------------------------- @Test - void millisTimestampIsNotShreddedIntoMicrosLeaf() { + void microsTimestampIsNotShreddedIntoMillisLeaf() { // A millisecond-precision typed_value cannot represent a micros variant timestamp, so // avroTypeToScalarType returns null and the value is left unshredded in the residual. assertStaysInResidual(scalar(b -> b.appendTimestamp(1_700_000_000_000_000L)), @@ -224,12 +254,24 @@ void millisTimestampIsNotShreddedIntoMicrosLeaf() { } @Test - void fixedLeafShredsBinaryToByteBuffer() { - // A FIXED typed_value maps to BinaryType, so a binary variant is shredded into typed_value (had - // avroTypeToScalarType returned null for FIXED, the value would fall to the residual instead). - byte[] payload = {1, 2, 3, 4}; - assertScalarShredsTo(scalar(b -> b.appendBinary(payload)), - HoodieSchema.createFixed("fx", "org.apache.hudi.test", null, 4), ByteBuffer.wrap(payload)); + void lossyDecimalIsNotShredded() { + // decimal(10,2) cannot represent 123.456 exactly, so the value must fall to the residual. + assertStaysInResidual(scalar(b -> b.appendDecimal(new BigDecimal("123.456"))), + HoodieSchema.createDecimal(10, 2)); + } + + @Test + void enumLeafIsNotShredded() { + // avroTypeToScalarType has no mapping for ENUM (the default: branch), so the value declines + // to the residual. + assertStaysInResidual(scalar(b -> b.appendString("A")), + HoodieSchema.createEnum("e", "org.apache.hudi.test", null, Arrays.asList("A", "B"))); + } + + @Test + void noTypedValueSchemaRoundTrips() throws Exception { + // A shredded schema without typed_value keeps the whole variant in the residual value. + assertRoundTrips("{\"a\":1}", HoodieSchema.createVariantShredded(null)); } // --------------------------------------------------------------------------- @@ -259,6 +301,9 @@ void partialObjectShreddingKeepsExtraFieldsInResidual() throws Exception { // The unmatched field forces a non-null residual value at the top level. assertNotNull(shreddedRecord.get(VARIANT_VALUE_FIELD), "extra field must be captured in residual value"); GenericRecord typedValue = (GenericRecord) shreddedRecord.get(VARIANT_TYPED_VALUE_FIELD); + GenericRecord aField = (GenericRecord) typedValue.get("a"); + assertEquals("x", aField.get(VARIANT_TYPED_VALUE_FIELD), "declared field a must shred into typed_value"); + assertNull(aField.get(VARIANT_VALUE_FIELD), "matched field a carries no residual"); GenericRecord bField = (GenericRecord) typedValue.get("b"); assertNull(bField.get(VARIANT_VALUE_FIELD), "absent field b carries no residual value"); assertNull(bField.get(VARIANT_TYPED_VALUE_FIELD), "absent field b carries no typed_value"); @@ -266,6 +311,25 @@ void partialObjectShreddingKeepsExtraFieldsInResidual() throws Exception { assertEquals(variant.toJson(ZoneOffset.UTC), rebuild(shreddedRecord, shredded).toJson(ZoneOffset.UTC)); } + @Test + void typeMismatchedObjectFieldFallsToFieldResidual() throws Exception { + // The field is declared LONG but the variant supplies a string: the field's own value carries + // the residual, its typed_value stays null, and the top-level value stays null. + Map shreddedFields = new LinkedHashMap<>(); + shreddedFields.put("a", HoodieSchema.create(HoodieSchemaType.LONG)); + HoodieSchema.Variant shredded = HoodieSchema.createVariantShreddedObject(shreddedFields); + + Variant variant = VariantBuilder.parseJson("{\"a\":\"str\"}", false); + GenericRecord shreddedRecord = shred(variant, shredded); + + assertNull(shreddedRecord.get(VARIANT_VALUE_FIELD), "no unmatched field, so the top-level residual stays null"); + GenericRecord aField = (GenericRecord) ((GenericRecord) shreddedRecord.get(VARIANT_TYPED_VALUE_FIELD)).get("a"); + assertNotNull(aField.get(VARIANT_VALUE_FIELD), "mismatched field a keeps its value in the field residual"); + assertNull(aField.get(VARIANT_TYPED_VALUE_FIELD), "mismatched field a must not shred"); + + assertEquals(variant.toJson(ZoneOffset.UTC), rebuild(shreddedRecord, shredded).toJson(ZoneOffset.UTC)); + } + @Test void arrayRoundTrips() throws Exception { // typed_value for an array is array<{value, typed_value}>: each element is itself a shredded struct. @@ -282,23 +346,24 @@ void arrayRoundTrips() throws Exception { @Test void rebuildDecimalFromBytesEncoding() { - assertDecimalRebuildsFromEncoding(HoodieSchema.createDecimal(10, 2), false); + assertDecimalRebuildsFromEncoding(HoodieSchema.createDecimal(10, 2)); } @Test void rebuildDecimalFromFixedEncoding() { assertDecimalRebuildsFromEncoding( - HoodieSchema.createDecimal("dec_fixed", "org.apache.hudi.test", null, 10, 2, 8), true); + HoodieSchema.createDecimal("dec_fixed", "org.apache.hudi.test", null, 10, 2, 8)); } - private void assertDecimalRebuildsFromEncoding(HoodieSchema decimalType, boolean fixed) { + private void assertDecimalRebuildsFromEncoding(HoodieSchema decimalType) { BigDecimal value = new BigDecimal("123.45"); - HoodieSchema.Variant shredded = HoodieSchema.createVariantShredded(decimalType); + HoodieSchema.Variant shredded = shreddedLeaf(decimalType); GenericRecord shreddedRecord = shred(scalar(b -> b.appendDecimal(value)), shredded); - Schema tvSchema = shredded.getAvroSchema().getField(VARIANT_TYPED_VALUE_FIELD).schema(); + // Encode against the concrete decimal branch (typed_value itself is a nullable union). + Schema tvSchema = decimalType.getAvroSchema(); Conversions.DecimalConversion conversion = new Conversions.DecimalConversion(); - Object encoded = fixed + Object encoded = tvSchema.getType() == Schema.Type.FIXED ? conversion.toFixed(value, tvSchema, tvSchema.getLogicalType()) : conversion.toBytes(value, tvSchema, tvSchema.getLogicalType()); shreddedRecord.put(VARIANT_TYPED_VALUE_FIELD, encoded); @@ -309,7 +374,7 @@ private void assertDecimalRebuildsFromEncoding(HoodieSchema decimalType, boolean @Test void rebuildDecimalRejectsUnexpectedEncoding() { - HoodieSchema.Variant shredded = HoodieSchema.createVariantShredded(HoodieSchema.createDecimal(10, 2)); + HoodieSchema.Variant shredded = shreddedLeaf(HoodieSchema.createDecimal(10, 2)); GenericRecord shreddedRecord = shred(scalar(b -> b.appendDecimal(new BigDecimal("1.00"))), shredded); shreddedRecord.put(VARIANT_TYPED_VALUE_FIELD, "not-a-decimal"); assertThrows(IllegalStateException.class, @@ -322,7 +387,7 @@ void rebuildDecimalRejectsUnexpectedEncoding() { @Test void shredReturnsNullWhenValueOrMetadataMissing() { - HoodieSchema.Variant shredded = HoodieSchema.createVariantShredded(HoodieSchema.create(HoodieSchemaType.LONG)); + HoodieSchema.Variant shredded = shreddedLeaf(HoodieSchema.create(HoodieSchemaType.LONG)); Variant variant = scalar(b -> b.appendLong(1)); GenericRecord missingValue = unshredded(variant); @@ -336,13 +401,15 @@ void shredReturnsNullWhenValueOrMetadataMissing() { @Test void rebuildReturnsNullForNullRecord() { - HoodieSchema.Variant shredded = HoodieSchema.createVariantShredded(HoodieSchema.create(HoodieSchemaType.LONG)); + // Coverage of a defensive guard only: the sole production caller (HoodieVariantReconstruction) + // already checks instanceof GenericRecord before calling rebuildVariantRecord. + HoodieSchema.Variant shredded = shreddedLeaf(HoodieSchema.create(HoodieSchemaType.LONG)); assertNull(provider.rebuildVariantRecord(null, shredded.getAvroSchema(), unshreddedSchema)); } @Test void rebuildThrowsWhenMetadataMissing() { - HoodieSchema.Variant shredded = HoodieSchema.createVariantShredded(HoodieSchema.create(HoodieSchemaType.LONG)); + HoodieSchema.Variant shredded = shreddedLeaf(HoodieSchema.create(HoodieSchemaType.LONG)); GenericRecord shreddedRecord = shred(scalar(b -> b.appendLong(7)), shredded); shreddedRecord.put(VARIANT_METADATA_FIELD, null); assertThrows(HoodieException.class,