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 c815c3f06ef2 test(spark): extend Spark 4 variant shredding coverage
(#19403)
c815c3f06ef2 is described below
commit c815c3f06ef2fd3c6399af9f198ffaaf47916d98
Author: Y Ethan Guo <[email protected]>
AuthorDate: Fri Jul 31 04:57:41 2026 -0700
test(spark): extend Spark 4 variant shredding coverage (#19403)
* test(spark): extend Spark 4 variant shredding coverage
* 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
---------
Co-authored-by: voon <[email protected]>
---
.../TestSpark4VariantShreddingProvider.java | 367 +++++++++++++++++++--
1 file changed, 337 insertions(+), 30 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 37800e98dc02..8abb9bf992fe 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.Variant;
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<VariantBuilder>
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<String, HoodieSchema> shreddedFields = new LinkedHashMap<>();
@@ -100,15 +286,136 @@ class TestSpark4VariantShreddingProvider {
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<String, HoodieSchema> 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<String, HoodieSchema> 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()];
