voonhous commented on code in PR #19403:
URL: https://github.com/apache/hudi/pull/19403#discussion_r3688987997
##########
hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java:
##########
@@ -22,61 +22,148 @@
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;
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.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) {
Review Comment:
**correctness (major):** every write-side assertion in this class is
in-memory; nothing checks the shredded record is writable at its declared
schema, and no test in the repo pushes this provider's output across the
Avro/parquet boundary. One line in this helper would have caught both the FIXED
and the required-`typed_value` issues flagged in the other threads:
```java
assertTrue(ConvertingGenericData.INSTANCE.validate(shredded.getAvroSchema(),
record));
```
(`ConvertingGenericData.INSTANCE` is the data model `HoodieAvroWriteSupport`
uses, so `validate` is the cheapest proxy for "this record can be written".)
Please add it here before returning. A follow-up functional test writing
`binary` + `decimal(10,2)` leaves via `HoodieAvroWriteSupport` and reading back
through `HoodieAvroParquetReader` would close this bug class entirely -- those
are the two DDL-expressible leaves no test currently writes.
##########
hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java:
##########
@@ -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() {
Review Comment:
**correctness (major):** this pins behavior that cannot survive a real write
or read, and it blocks the fix.
`avroTypeToScalarType` maps FIXED to `BinaryType` (latent since #18065,
`23fe7bfcd7e6`), but:
- **write**: `convertScalarToAvro` returns a `ByteBuffer`;
`ConvertingGenericData.INSTANCE.validate(...)` is `false` for it, and
`AvroParquetWriter` with that data model (the one `HoodieAvroWriteSupport`
passes to parquet-avro) throws `ClassCastException: HeapByteBuffer cannot be
cast to GenericFixed`.
- **read**: parquet-avro decodes a FIXED column to `GenericData$Fixed`,
which the `(ByteBuffer)` cast in `AvroVariantRow.getBinary` rejects.
- there is no `getFixedSize()` check: a 7-byte payload shreds into
`FIXED(4)` without complaint (the 4-byte payload here hides that).
The round-trip half never crosses the serde boundary (the same in-memory
`ByteBuffer` is fed straight back into rebuild), so it proves nothing about
real files. Worse, the cheap correct fix -- `avroTypeToScalarType` returning
null for FIXED (decline to shred) -- would fail exactly this test.
Please invert this test to `assertStaysInResidual(scalar(b ->
b.appendBinary(payload)), HoodieSchema.createFixed("fx",
"org.apache.hudi.test", null, 4))` (or drop it), and file a follow-up issue for
the FIXED handling in `Spark4VariantShreddingProvider`.
##########
hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java:
##########
@@ -22,61 +22,148 @@
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;
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.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);
Review Comment:
**correctness (major):** `createVariantShredded(<plain scalar>)` declares
`typed_value` as **required**, a shape production never emits:
`createShreddedFieldStruct` (`HoodieSchema.java:987`) wraps every leaf in
`createNullable`, and the parquet shredding spec requires `typed_value`
optional. Two consequences:
- the decline-to-shred test below (`millisTimestamp...`) builds a record
with null in a required field -- unwritable, so it does not prove the fallback
works end to end;
- `unwrapNullable` in the provider is never exercised at the leaf (only one
level down, via the two object tests).
Please build
`HoodieSchema.createVariantShredded(HoodieSchema.createNullable(typedValue))`
here and in `assertScalarRoundTrips` / `assertStaysInResidual`. Note
`assertDecimalRebuildsFromEncoding` then needs to unwrap the union before
calling `tvSchema.getLogicalType()`, otherwise `DecimalConversion.toBytes` NPEs
-- which is exactly the realism the current fixture hides.
##########
hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java:
##########
@@ -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 {
Review Comment:
**coverage gap (major):** `allowNumericScaleChanges()` in
`Spark4VariantShreddingProvider` is live policy with zero coverage -- flipping
it to `false` still passes all 24 tests in this class (mutation-verified).
Spark's `VariantShreddingWriter.tryTypedShred` consults it to shred a JSON
integer into a decimal `typed_value` and a decimal into an integral leaf;
`{"amount": 5}` into a `decimal(10,2)` column is the common real mix. Current
behavior is correct (probed: long 5 -> `5.00`, decimal `5.00` -> `5`) but
nothing pins it.
Please add three cases next to this test:
```java
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);
assertStaysInResidual(scalar(b -> b.appendDecimal(new
BigDecimal("123.456"))), HoodieSchema.createDecimal(10, 2));
```
The third pins the lossy-rejection side (a value the leaf cannot represent
must fall to the residual).
##########
hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java:
##########
@@ -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<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");
Review Comment:
**correctness (minor):** the name and comment claim `"a" shreds into
typed_value`, but nothing asserts it -- only `b`'s absence and the residual's
presence are checked. Verified actual behavior: `a.typed_value == "x"`
(String), `a.value == null`. Separately, the complementary canonical case -- a
declared field whose type does NOT match (`{"a":"str"}` against `{a: long}`),
which must set the field-level `value` and leave the top-level residual null --
is asserted nowhere in the repo (probe-verified it works today).
Please add here:
```java
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");
```
plus one field-level type-mismatch sibling test asserting `a.value != null`,
`a.typed_value == null`, top-level `value == null`.
##########
hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java:
##########
@@ -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);
Review Comment:
**cleanliness (minor):** this literal contains a raw NUL byte (the bytes on
disk are `not-utf8-\x00\xc3\xbf`, NUL at file offset 8412). Plain `grep` now
reports the whole file as `Binary file ... matches` with no line output, and
`rg` skips it entirely; only `git grep` still finds anything in this file. It
is also only ~400 bytes short of git rendering diffs of the whole file as
binary. The "not-utf8" property is compiler-encoding-dependent besides (read as
non-UTF-8 source, the trailing character becomes the valid-UTF-8 pair `C3 BF`).
```suggestion
byte[] payload = {'n', 'o', 't', '-', 'u', 't', 'f', '8', (byte) 0xC0,
(byte) 0xFF};
```
`C0 FF` is unambiguously invalid UTF-8, with no control characters and no
source-encoding dependency.
##########
hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java:
##########
@@ -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() {
Review Comment:
**cleanliness (nit):** name is backwards -- the variant carries a *micros*
timestamp and the leaf is *millis* precision (the comment above has it right).
```suggestion
void microsTimestampIsNotShreddedIntoMillisLeaf() {
```
##########
hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java:
##########
@@ -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<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 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) {
Review Comment:
**cleanliness (nit, optional):** the `fixed` flag duplicates information
already in `decimalType`, so the two arguments can silently disagree. Suggest
deriving it and dropping the parameter:
```java
boolean fixed = decimalType.getAvroSchema().getType() == Schema.Type.FIXED;
```
##########
hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java:
##########
@@ -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<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 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() {
Review Comment:
**cleanliness (nit, optional):** this guards an unreachable branch -- the
only production caller (`HoodieVariantReconstruction.java:167`) invokes
`rebuildVariantRecord` inside `if (isTarget[i] && value instanceof
GenericRecord)`, so a null record cannot reach it. Fine to keep as a guard for
future callers, but add a one-line comment saying it is coverage of a defensive
guard, so it does not read as behavior coverage. Feel free to ignore.
##########
hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java:
##########
@@ -22,61 +22,148 @@
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;
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.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<VariantBuilder>
append) {
+ VariantBuilder builder = new VariantBuilder(false);
+ append.accept(builder);
+ return builder.result();
+ }
+
+ //
---------------------------------------------------------------------------
+ // Scalar leaf types: one per branch of avroTypeToScalarType /
convertScalarToAvro.
Review Comment:
**cleanliness (minor):** "one per branch" (and the PR body's "every scalar
leaf type") overclaims: the `Byte`/`Short`/`Integer` widening arms in
`convertScalarToAvro` (provider lines 379-398) and `getByte`/`getShort` are
unreachable -- `avroTypeToScalarType` only ever emits `IntegralSize.INT`/`LONG`
since Avro has no byte/short -- and the ENUM `default:`, the no-`typed_value`
schema, and the non-zero-position `toByteArray` path stay uncovered
(jacoco-verified: lines 251, 261-263, 340, 382-396 missed).
```suggestion
// Scalar leaf types: one per reachable branch of avroTypeToScalarType /
convertScalarToAvro
// (the Byte/Short widening arms in convertScalarToAvro are unreachable:
Avro has no byte/short).
```
Optional: an ENUM `assertStaysInResidual` case and a
`createVariantShredded(null)` round-trip are one-liners with the existing
helpers. Deleting the dead widening arms belongs in a follow-up, not this
test-only PR.
##########
hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java:
##########
@@ -22,61 +22,148 @@
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;
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.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<VariantBuilder>
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 {
Review Comment:
**cleanliness (nit, optional):** `numericRoundTrips` is now strictly
subsumed by `longLeafShredsToLong` -- same LONG-leaf branch, weaker assertion;
`parseJson("42")` vs `appendLong` differ only in encoded width, which is
decoded inside Spark's `Variant`, not Hudi code. Meanwhile `stringRoundTrips` /
`booleanRoundTrips` / `decimalRoundTrips` still use the weak json-round-trip
tier while every new scalar pins the typed_value.
Suggest: delete this test (fold `parseJson("42", false)` into
`longLeafShredsToLong` if the json entry point matters) and upgrade the other
three to `assertScalarShredsTo` (`"hello world"` -> String, `true` -> Boolean,
`123.45` -> `new BigDecimal("123.45")`; residual null in all three,
probe-verified) so the class has one assertion standard.
##########
hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java:
##########
@@ -22,61 +22,148 @@
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;
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.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");
Review Comment:
**cleanliness (nit, optional):** master's failure message included the
offending input (`"... for: " + json`); the refactor drops it, so a failing
json-driven test no longer says which value broke. Suggest threading a label
through: have the `String` overload pass `"for: " + json` and append it to this
message.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
