voonhous commented on code in PR #19403:
URL: https://github.com/apache/hudi/pull/19403#discussion_r3689100894
##########
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:
Addressed in caba485a3f5b: added the `a.typed_value == "x"` / `a.value ==
null` asserts, plus a new `typeMismatchedObjectFieldFallsToFieldResidual` test
pinning the field-level residual with a null top-level `value`.
##########
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:
Addressed in caba485a3f5b: replaced with an explicit byte array ending `0xC0
0xFF` (unambiguously invalid UTF-8). `grep`/`rg` see the file as text again.
##########
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:
Addressed in caba485a3f5b: comment now reads "one per reachable branch" and
names the unreachable Byte/Short widening arms; added `enumLeafIsNotShredded`
and `noTypedValueSchemaRoundTrips`. Dead-code deletion deferred to #19442.
##########
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:
Addressed in caba485a3f5b: `numericRoundTrips` folded into
`longLeafShredsToLong` via a json overload of `assertScalarShredsTo`;
string/boolean/decimal tests upgraded to pin their `typed_value` and renamed to
match the sibling convention.
--
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]
