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..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,10 +19,13 @@ 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; +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,68 +33,251 @@ import org.apache.spark.types.variant.VariantBuilder; import org.junit.jupiter.api.Test; +import java.math.BigDecimal; import java.nio.ByteBuffer; 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; +import static org.junit.jupiter.api.Assertions.assertTrue; /** - * 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 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 { 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) { + 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; + } - 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))); + } + + private void assertRoundTrips(Variant variant, HoodieSchema.Variant shredded) { + assertRoundTrips(variant, shredded, ""); + } - 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, 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" + 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, " 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)); + } + + /** + * 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 = 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"); + 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"); + } + + 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 = 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"); + 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 reachable branch of avroTypeToScalarType / convertScalarToAvro + // (the Byte/Short widening arms in convertScalarToAvro are unreachable: Avro has no byte/short). + // --------------------------------------------------------------------------- + + @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() 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 + 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 stringLeafShredsToString() throws Exception { + assertScalarShredsTo("\"hello world\"", HoodieSchema.create(HoodieSchemaType.STRING), "hello world"); + } + + @Test + void booleanLeafShredsToBoolean() throws Exception { + assertScalarShredsTo("true", HoodieSchema.create(HoodieSchemaType.BOOLEAN), true); + } + + @Test + void binaryShredsToByteBuffer() { + 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)); + } + + @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 numericRoundTrips() throws Exception { - assertScalarRoundTrips("42", HoodieSchema.create(HoodieSchemaType.LONG)); + void decimalLeafShredsToBigDecimal() throws Exception { + assertScalarShredsTo("123.45", HoodieSchema.createDecimal(10, 2), new BigDecimal("123.45")); } @Test - void stringRoundTrips() throws Exception { - assertScalarRoundTrips("\"hello world\"", HoodieSchema.create(HoodieSchemaType.STRING)); + 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); } + // --------------------------------------------------------------------------- + // "Decline to shred" fallbacks: value stays in the residual binary. + // --------------------------------------------------------------------------- + @Test - void booleanRoundTrips() throws Exception { - assertScalarRoundTrips("true", HoodieSchema.create(HoodieSchemaType.BOOLEAN)); + 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)), + HoodieSchema.createTimestampMillis()); + assertStaysInResidual(scalar(b -> b.appendTimestampNtz(1_700_000_000_000_000L)), + HoodieSchema.createLocalTimestampMillis()); } @Test - void decimalRoundTrips() throws Exception { - assertScalarRoundTrips("123.45", HoodieSchema.createDecimal(10, 2)); + 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)); + } + + // --------------------------------------------------------------------------- + // Object and array shapes, including partial (residual) object shredding. + // --------------------------------------------------------------------------- + @Test void objectRoundTrips() throws Exception { Map shreddedFields = new LinkedHashMap<>(); @@ -100,15 +286,136 @@ 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 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"); + + 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. 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)); + } + + @Test + void rebuildDecimalFromFixedEncoding() { + assertDecimalRebuildsFromEncoding( + HoodieSchema.createDecimal("dec_fixed", "org.apache.hudi.test", null, 10, 2, 8)); + } + + private void assertDecimalRebuildsFromEncoding(HoodieSchema decimalType) { + BigDecimal value = new BigDecimal("123.45"); + HoodieSchema.Variant shredded = shreddedLeaf(decimalType); + GenericRecord shreddedRecord = shred(scalar(b -> b.appendDecimal(value)), shredded); + + // 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 = tvSchema.getType() == Schema.Type.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 = 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, + () -> provider.rebuildVariantRecord(shreddedRecord, shredded.getAvroSchema(), unshreddedSchema)); + } + + // --------------------------------------------------------------------------- + // Null / error guards. + // --------------------------------------------------------------------------- + + @Test + void shredReturnsNullWhenValueOrMetadataMissing() { + HoodieSchema.Variant shredded = shreddedLeaf(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() { + // 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 = shreddedLeaf(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()];