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 576e5b6e70e1 fix(avro): detect the two-field shredded variant shape
and harden the variant merge path (#19620)
576e5b6e70e1 is described below
commit 576e5b6e70e151f4feed7922acc07a3f3c23d88e
Author: voonhous <[email protected]>
AuthorDate: Mon Aug 17 14:21:43 2026 +0800
fix(avro): detect the two-field shredded variant shape and harden the
variant merge path (#19620)
* review(19582): guard the strip rebuild and correct the nested
reachability note
Two things the self-review turned up, both in code the round-3 commits
touched but no test or comment covered.
stripVariantShredding used to re-add untouched fields by reference and
then hand them to Schema.setFields, which throws "Field already used" for
any record of {other column + shredded variant}. That is the same defect
#18938 fixed in the sibling HoodieVariantReconstruction.create; the twin
here was missed. The recursive rewrite copies every field via withSchema
and so fixes it incidentally - now pinned by a test at the production
call site, HoodieAvroWriteSupport.generateEffectiveSchema with shredding
disabled, which is the clustering/compaction path that reaches it.
The nested comments claimed the row writer's depth recursion as the
producer of nested shredded files. That overstates reachability: the
forced-shredding hook is top-level only in BOTH write supports, so no DDL
or table property reaches depth and only a hand-authored write schema
can. Say that plainly instead, and note why the nested coverage is unit
level rather than end to end.
* test(19582): sweep the small-file merge over record type and layout
The shredded and unshredded small-file merge tests were near-identical
copies, and both were pinned to AVRO. Fold them into one sweep over
(record type, layout) and pull the two repeated pins into helpers.
What each leg is for:
- AVRO + shredded is the #19567 bug; it goes red without the
HoodieMergeHelper alignment (verified by reverting that hunk: rows 1-2
come back [1,null,1000], [2,null,1000]).
- AVRO + unshredded is the no-op guard, kept because a reviewer asked for
the twin, now without the 65-line copy.
- SPARK is new coverage. HoodieSparkParquetReader.getSchema returns a
nullable UNION rather than a RECORD, so alignShreddedVariants bails at
its RECORD/RECORD guard and that reader is untouched by the fix. The
leg passes identically with and without the fix; it is swept so that
the guard which scopes the fix to the AVRO reader is pinned rather
than merely assumed.
assertVariantLayout and assertSingleFileGroup also replace the inline
pins in the evolving-schema test; the layout pin was copy-pasted 11 times
across this file. Net -56 lines.
Verified on spark4.1/scala-2.13: 15 succeeded, 0 failed, 2 canceled (the
Spark 3.x-only tests).
* fix(avro): detect the two-field shredded variant shape, and stop
rebuilding schemas that do not change
Two things found self-reviewing #19582 after it merged.
isShreddedVariantShape demanded exactly three fields with a bytes `value`,
but HoodieSchema.Variant.determineIfShredded - the answer used whenever the
logical type survives - calls anything carrying typed_value shredded. The
shredding spec lets a writer omit `value` when every row is typed, and
because the parquet footer always strips the logical type, the shape check
is the only detector that runs on real files. So a two-field group read at
the unshredded schema and lost its payload silently: #19567 again by
another shape. Accept {metadata, typed_value} with `value` optional; the
requested-side variant anchor is what keeps plain user structs out, so this
costs no false positives, and a struct with a fourth field is still
rejected.
Second, both recursion loops copied every field into a new list and then
threw the list away when nothing matched. That is the path every
non-variant table takes, on every runMerge and every avro parquet read, and
it costs an Avro Field plus a defaultVal() lookup per field per level.
Build the list lazily, backfilling only once a replacement actually
appears. Measured on a 500-column nested schema: alignShreddedVariants
730us -> 335us per runMerge, create() 834us -> 365us per read.
Tests: the two-field shape engages and rebuilds, and a four-field struct
that merely carries typed_value stays untouched.
* review(19620): scope the nested-shredding claim to the avro path, cover
the value-less group
- The "forced shredding is top-level only in both write supports" note was
wrong for the
row path: processNestedDataType recurses into structs and
generateShreddedSchema re-reads
the DDL on every entry, so struct<v variant> plus the force-shredding
property does shred
at depth. Scoped the claim to
HoodieAvroWriteSupport.applyForcedShreddingSchema in all
three places that carried it.
- Added a value-less round-trip case in
TestHoodieVariantReconstructionRoundTrip. The
detection test in hudi-hadoop-common uses a stub provider that ignores
the shredded
schema, so nothing exercised
Spark4VariantShreddingProvider.buildVariantSchema with the
variantIdx = -1 that a {metadata, typed_value} group produces.
- Moved the stranded listDataParquetFiles javadoc back onto its method.
* review(19620): address the round-2 comments
- Scope the write-support strip test's javadoc the same way the
VariantSchemaUtils and TestHoodieVariantReconstruction comments now
are: applyForcedShreddingSchema is top-level only, so the nested
shredded schema this path strips comes from a file the ROW path (or
another engine) wrote, not from this path's own hook.
- Add the {metadata, typed_value, extra} rejection case. The four-field
struct dies at the field-count guard, so nothing reached the
no-value arm of isShreddedVariantShape; mutating its fieldCount == 2
to true left every test green. The new case survives the count guard
and pins that arm.
- Fold the CDC test's verbatim layout block into assertVariantLayout;
tablePath, shredded and leg were already in scope.
- Run hudi-spark4-common in the spark4.2 java-test part1 lane. The
module's two test classes were compiled (spark4.2.x depends on it,
and the build step uses -am) but never executed: part1 is the only
lane whose java UT filter is an exclusion list, so listing the module
there is what makes them run.
---
.github/workflows/bot.yml | 5 +-
.../hudi/common/avro/VariantSchemaUtils.java | 122 +++++++---
.../hadoop/HoodieVariantReconstruction.java | 6 +-
.../avro/TestHoodieAvroWriteSupportShredding.java | 47 ++++
.../hadoop/TestHoodieVariantReconstruction.java | 71 +++++-
.../sql/hudi/dml/schema/TestVariantDataType.scala | 260 +++++++++------------
.../TestHoodieVariantReconstructionRoundTrip.java | 65 ++++++
7 files changed, 383 insertions(+), 193 deletions(-)
diff --git a/.github/workflows/bot.yml b/.github/workflows/bot.yml
index 751e460ad297..08db07cd2c12 100644
--- a/.github/workflows/bot.yml
+++ b/.github/workflows/bot.yml
@@ -732,9 +732,12 @@ jobs:
# - scalaProfile: "scala-2.13"
# sparkProfile: "spark4.1"
# sparkModules: "hudi-spark-datasource/hudi-spark4.1.x"
+ # hudi-spark4-common runs here (and only here): the other spark
lanes either pin a
+ # fixed test-class list or run scala/functional profiles, so without
this entry the
+ # module's java unit tests are compiled but never executed in CI.
- scalaProfile: "scala-2.13"
sparkProfile: "spark4.2"
- sparkModules: "hudi-spark-datasource/hudi-spark4.2.x"
+ sparkModules:
"hudi-spark-datasource/hudi-spark4-common,hudi-spark-datasource/hudi-spark4.2.x"
steps:
- if: needs.changes.outputs.relevant == 'true'
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/avro/VariantSchemaUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/avro/VariantSchemaUtils.java
index b2155d35eb05..a656fc5de875 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/avro/VariantSchemaUtils.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/avro/VariantSchemaUtils.java
@@ -40,10 +40,14 @@ public class VariantSchemaUtils {
/**
* Strips shredding from the variant fields in {@code schema}, replacing
each shredded variant
* with its unshredded form (dropping {@code typed_value}). Variants nested
inside records, array
- * elements and map values are stripped too, since the row writer shreds at
any depth
- * ({@code HoodieRowParquetWriteSupport.processNestedDataType}). Non-variant
fields and
- * already-unshredded variants pass through unchanged; returns {@code
schema} as-is when nothing
- * changes.
+ * elements and map values are stripped too; see {@link
#swapShreddedVariantFields} for when such
+ * a schema can arise. Non-variant fields and already-unshredded variants
pass through unchanged;
+ * returns {@code schema} as-is when nothing changes.
+ *
+ * <p>Every field of a rebuilt record is copied via {@code withSchema},
including the untouched
+ * ones: reusing an Avro {@code Field} still bound to the source record makes
+ * {@code Schema.setFields} throw "Field already used" (the defect #18938
fixed in the sibling
+ * HoodieVariantReconstruction).
*/
public static HoodieSchema stripVariantShredding(HoodieSchema schema) {
if (schema.getType() != HoodieSchemaType.RECORD) {
@@ -53,19 +57,24 @@ public class VariantSchemaUtils {
}
private static HoodieSchema stripRecordVariantShredding(HoodieSchema record)
{
- List<HoodieSchemaField> newFields = new ArrayList<>();
- boolean changed = false;
- for (HoodieSchemaField field : record.getFields()) {
+ List<HoodieSchemaField> fields = record.getFields();
+ // Built lazily: every schema without a shredded variant walks this
method, and copying fields
+ // only to discard them costs an Avro Field plus a defaultVal() lookup per
field per level.
+ List<HoodieSchemaField> newFields = null;
+ for (int i = 0; i < fields.size(); i++) {
+ HoodieSchemaField field = fields.get(i);
HoodieSchema fieldSchema = field.schema();
HoodieSchema replacement = stripVariantShreddingAt(fieldSchema);
- if (replacement != fieldSchema) {
- changed = true;
+ if (replacement != fieldSchema && newFields == null) {
+ newFields = copyFieldsBefore(fields, i);
+ }
+ if (newFields != null) {
+ // withSchema makes a fresh Avro Field: reusing one already bound to
this record would fail
+ // Schema.setFields with "Field already used" when building the
replacement record below.
+ newFields.add(field.withSchema(replacement));
}
- // withSchema makes a fresh Avro Field: reusing one already bound to
this record would fail
- // Schema.setFields with "Field already used" when building the
replacement record below.
- newFields.add(field.withSchema(replacement));
}
- if (!changed) {
+ if (newFields == null) {
return record;
}
return HoodieSchema.createRecord(
@@ -166,30 +175,44 @@ public class VariantSchemaUtils {
/**
* Walks {@code base} against its matching {@code other} fields by name,
replacing every shredded
- * variant position with the other side's schema. Recurses through records,
array elements and map
- * values, since the row writer shreds variants at any depth
- * ({@code HoodieRowParquetWriteSupport.processNestedDataType}). {@code
baseIsFile} says which of
- * the two is the file side, which is what {@link #isShreddedVariantTarget}
needs to anchor
- * detection. Returns {@code base} when nothing matches.
+ * variant position with the other side's schema. {@code baseIsFile} says
which of the two is the
+ * file side, which is what {@link #isShreddedVariantTarget} needs to anchor
detection. Returns
+ * {@code base} when nothing matches.
+ *
+ * <p>The walk recurses through records, array elements and map values
because the row writer
+ * shreds at any depth its write schema asks it to
+ * ({@code HoodieRowParquetWriteSupport.processNestedDataType} recurses into
structs, array
+ * elements and map values, and {@code generateShreddedSchema} re-reads the
forced-shredding DDL
+ * on every entry, so {@code struct<v variant>} plus
+ * {@code hoodie.parquet.variant.force.shredding.schema.for.test} does shred
at depth on the ROW
+ * path). The AVRO path is the narrower one: {@code
HoodieAvroWriteSupport.applyForcedShreddingSchema}
+ * walks top-level fields only, so on that path a nested shredded column
needs a hand-authored
+ * write schema that declares {@code typed_value} below the top level.
*/
private static HoodieSchema swapShreddedVariantFields(HoodieSchema base,
HoodieSchema other, boolean baseIsFile) {
- List<HoodieSchemaField> newFields = new ArrayList<>();
- boolean changed = false;
- for (HoodieSchemaField baseField : base.getFields()) {
+ List<HoodieSchemaField> baseFields = base.getFields();
+ // Built lazily, as in stripRecordVariantShredding: alignShreddedVariants
runs on every
+ // HoodieMergeHelper.runMerge, so the overwhelmingly common case is a
schema that matches
+ // nothing and must not pay for a full field copy it will throw away.
+ List<HoodieSchemaField> newFields = null;
+ for (int i = 0; i < baseFields.size(); i++) {
+ HoodieSchemaField baseField = baseFields.get(i);
HoodieSchema baseFieldSchema = baseField.schema();
Option<HoodieSchemaField> otherField = other.getField(baseField.name());
HoodieSchema replacement = otherField.isPresent()
? swapShreddedVariantsAt(baseFieldSchema, otherField.get().schema(),
baseIsFile)
: baseFieldSchema;
- if (replacement != baseFieldSchema) {
- changed = true;
+ if (replacement != baseFieldSchema && newFields == null) {
+ newFields = copyFieldsBefore(baseFields, i);
+ }
+ if (newFields != null) {
+ // Copy untouched fields too (withSchema makes a fresh Avro Field):
reusing a field already
+ // bound to the base record would fail Schema.setFields with "Field
already used" when
+ // building the swapped record below.
+ newFields.add(baseField.withSchema(replacement));
}
- // Copy untouched fields too (withSchema makes a fresh Avro Field):
reusing a field already
- // bound to the base record would fail Schema.setFields with "Field
already used" when
- // building the swapped record below.
- newFields.add(baseField.withSchema(replacement));
}
- if (!changed) {
+ if (newFields == null) {
return base;
}
return HoodieSchema.createRecord(
@@ -239,16 +262,51 @@ public class VariantSchemaUtils {
return wasNullable ? HoodieSchema.createNullable(replacement) :
replacement;
}
- /** The on-disk shredded variant shape: a record of exactly {metadata:
bytes, value: [nullable] bytes, typed_value}. */
+ /**
+ * The on-disk shredded variant shape: a record of {metadata: bytes,
typed_value} plus an optional
+ * {value: [nullable] bytes}, and nothing else.
+ *
+ * <p>{@code value} is deliberately optional. The shredding spec lets a
writer omit it when every
+ * row is typed, and {@link HoodieSchema.Variant#determineIfShredded} - the
answer used whenever
+ * the logical type survives - calls anything with a {@code typed_value}
shredded regardless. Since
+ * the footer always strips the logical type, this shape check is the only
detector that runs on
+ * real files, so demanding {@code value} here made a two-field group read
at the unshredded schema
+ * and silently drop its payload: #19567 again by another shape. The
requested-side variant anchor
+ * in {@link #isShreddedVariantTarget} is what keeps plain user structs out,
so accepting the
+ * two-field form costs no false positives.
+ */
private static boolean isShreddedVariantShape(HoodieSchema schema) {
- if (schema.getType() != HoodieSchemaType.RECORD ||
schema.getFields().size() != 3) {
+ if (schema.getType() != HoodieSchemaType.RECORD) {
+ return false;
+ }
+ int fieldCount = schema.getFields().size();
+ if (fieldCount < 2 || fieldCount > 3) {
return false;
}
if
(!schema.getField(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD).isPresent()) {
return false;
}
- return isBytesField(schema, HoodieSchema.Variant.VARIANT_METADATA_FIELD)
- && isBytesField(schema, HoodieSchema.Variant.VARIANT_VALUE_FIELD);
+ if (!isBytesField(schema, HoodieSchema.Variant.VARIANT_METADATA_FIELD)) {
+ return false;
+ }
+ boolean hasValue =
schema.getField(HoodieSchema.Variant.VARIANT_VALUE_FIELD).isPresent();
+ // A third field that is not `value` is some other user struct, not a
variant group.
+ return hasValue
+ ? isBytesField(schema, HoodieSchema.Variant.VARIANT_VALUE_FIELD)
+ : fieldCount == 2;
+ }
+
+ /**
+ * Fresh copies of {@code fields[0, end)}, for the point a rebuild first
turns out to be needed.
+ * Copies rather than reuses because the originals are still bound to their
source record.
+ */
+ private static List<HoodieSchemaField>
copyFieldsBefore(List<HoodieSchemaField> fields, int end) {
+ List<HoodieSchemaField> copied = new ArrayList<>(fields.size());
+ for (int i = 0; i < end; i++) {
+ HoodieSchemaField field = fields.get(i);
+ copied.add(field.withSchema(field.schema()));
+ }
+ return copied;
}
private static boolean isBytesField(HoodieSchema schema, String fieldName) {
diff --git
a/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieVariantReconstruction.java
b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieVariantReconstruction.java
index a3cf0e5bd237..cfc0d680d626 100644
---
a/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieVariantReconstruction.java
+++
b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieVariantReconstruction.java
@@ -129,8 +129,10 @@ final class HoodieVariantReconstruction {
* side is matched into it by field name. Detection is anchored by the
requested side because the
* file schema usually comes from converting the parquet footer MessageType,
which loses the
* variant logical type; see VariantSchemaUtils.isShreddedVariantTarget
(#19567). Records, array
- * elements and map values are all descended into, since the row writer
shreds variants at any
- * depth ({@code HoodieRowParquetWriteSupport.processNestedDataType}).
+ * elements and map values are all descended into, matching what the row
writer can emit; see
+ * {@code VariantSchemaUtils.swapShreddedVariantFields} for what actually
produces a nested
+ * shredded file today (the row path shreds at depth off the
forced-shredding property; the AVRO
+ * path needs a hand-authored write schema).
*/
private static Rebuilder buildRebuilder(HoodieSchema outputSchema,
HoodieSchema fileSchema) {
if (VariantSchemaUtils.isShreddedVariantTarget(fileSchema, outputSchema)) {
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroWriteSupportShredding.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroWriteSupportShredding.java
index eb936ac512e4..9ec32c9da3fd 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroWriteSupportShredding.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroWriteSupportShredding.java
@@ -22,6 +22,7 @@ package org.apache.hudi.avro;
import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaField;
+import org.apache.hudi.common.schema.HoodieSchemaType;
import org.junit.jupiter.api.Test;
@@ -31,7 +32,10 @@ import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
class TestHoodieAvroWriteSupportShredding {
@@ -68,4 +72,47 @@ class TestHoodieAvroWriteSupportShredding {
.collect(Collectors.toList());
assertEquals(Arrays.asList("a", "b", "c"), shreddedFieldNames);
}
+
+ /**
+ * Disabling shredding over an already-shredded schema - the
clustering/compaction case
+ * {@link HoodieAvroWriteSupport#generateEffectiveSchema} calls out - has to
strip typed_value
+ * without tripping Avro's "Field already used". Rebuilding a record while
reusing a
+ * {@code Schema.Field} still bound to the source record throws, so any
table with a shredded
+ * variant AND at least one other column failed here. #18938 fixed exactly
that defect in the
+ * sibling HoodieVariantReconstruction and left this twin behind. Nested
variants must be
+ * stripped too, but this AVRO path's own hook is not what puts one there -
+ * {@code applyForcedShreddingSchema} walks top-level fields only. The ROW
write path shreds at
+ * any depth its forced-shredding DDL asks
(HoodieRowParquetWriteSupport.processNestedDataType),
+ * so a clustering/compaction schema read back from such a file can carry
typed_value below the
+ * top level, which is why the nested leg is pinned here at unit level.
+ */
+ @Test
+ void disablingShreddingStripsTypedValueAtEveryDepth() {
+ HoodieSchema record = HoodieSchema.createRecord(
+ "test_record", "org.apache.hudi.test", null, Arrays.asList(
+ HoodieSchemaField.of("id",
HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaField.of("v", HoodieSchema.createVariantShredded(
+ "v", "org.apache.hudi.test", null,
HoodieSchema.create(HoodieSchemaType.INT))),
+ HoodieSchemaField.of("nested", HoodieSchema.createRecord(
+ "nested_record", "org.apache.hudi.test", null,
+ Collections.singletonList(HoodieSchemaField.of("nv",
HoodieSchema.createVariantShredded(
+ "nv", "org.apache.hudi.test", null,
HoodieSchema.create(HoodieSchemaType.INT))))))));
+
+ Properties props = new Properties();
+
props.setProperty(HoodieStorageConfig.PARQUET_VARIANT_WRITE_SHREDDING_ENABLED.key(),
"false");
+
+ HoodieSchema effective = assertDoesNotThrow(
+ () -> HoodieAvroWriteSupport.generateEffectiveSchema(record, props),
+ "stripping shredding must rebuild the record with fresh Avro fields");
+
+ assertEquals("id", effective.getFields().get(0).name(), "non-variant
fields must survive the rebuild");
+ assertUnshredded(effective.getField("v").get().schema(), "top-level
variant");
+
assertUnshredded(effective.getField("nested").get().schema().getField("nv").get().schema(),
"nested variant");
+ }
+
+ private static void assertUnshredded(HoodieSchema fieldSchema, String label)
{
+ HoodieSchema unwrapped = fieldSchema.isNullable() ?
fieldSchema.getNonNullType() : fieldSchema;
+ assertInstanceOf(HoodieSchema.Variant.class, unwrapped, label + " should
still be a variant");
+ assertFalse(((HoodieSchema.Variant) unwrapped).isShredded(), label + "
should no longer be shredded");
+ }
}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstruction.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstruction.java
index 99399e361fea..571379740f4f 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstruction.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstruction.java
@@ -186,6 +186,68 @@ class TestHoodieVariantReconstruction {
storageWithReadingShredded(tmp, false)));
}
+ @Test
+ void engagesOnTwoFieldShreddedShapeWithNoValueColumn(@TempDir Path tmp) {
+ // The shredding spec lets a writer omit `value` when every row is typed,
and
+ // HoodieSchema.Variant.determineIfShredded calls anything with a
typed_value shredded. Shape
+ // detection is the only detector that runs on real files, since the
footer strips the logical
+ // type, so requiring `value` here dropped such a column's payload
silently - #19567 again by
+ // another shape. Hudi's own writer always emits three fields, so this is
about files written
+ // elsewhere.
+ HoodieSchema twoFieldShredded = HoodieSchema.createRecord("v",
"org.apache.hudi.test", null, Arrays.asList(
+ HoodieSchemaField.of("metadata",
HoodieSchema.create(HoodieSchemaType.BYTES)),
+ HoodieSchemaField.of("typed_value",
HoodieSchema.createNullable(HoodieSchemaType.INT))));
+ HoodieSchema fileSchema = recordWithIdAndVariant(twoFieldShredded);
+ HoodieSchema requestedSchema =
recordWithIdAndVariant(HoodieSchema.createVariant());
+ HoodieStorage storage = storageWithReadingShredded(tmp, true);
+
storage.getConf().set(HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_PROVIDER_CLASS.key(),
+ TestVariantShreddingProvider.class.getName());
+
+ HoodieVariantReconstruction reconstruction =
HoodieVariantReconstruction.create(
+ fileSchema, requestedSchema, storage);
+ assertNotNull(reconstruction, "A shredded group with no value column must
still engage");
+
+ GenericRecord shredded = new GenericData.Record(
+
reconstruction.intermediateSchema().getField("v").get().schema().getNonNullType().toAvroSchema());
+ shredded.put("metadata", ByteBuffer.wrap(new byte[] {1}));
+ shredded.put("typed_value", 42);
+ GenericRecord input = new
GenericData.Record(reconstruction.intermediateSchema().toAvroSchema());
+ input.put("id", "record-1");
+ input.put("v", shredded);
+
+ GenericRecord variant = (GenericRecord)
reconstruction.reconstruct(input).get(1);
+ assertEquals(ByteBuffer.wrap(new byte[] {42}), variant.get("value"));
+ }
+
+ @Test
+ void ignoresFourFieldStructThatMerelyCarriesTypedValue(@TempDir Path tmp) {
+ // The other side of relaxing the field count: a user struct that happens
to hold metadata,
+ // value and typed_value plus anything else is not a variant group and
must stay untouched.
+ // The field-count guard rejects this one before any field is inspected.
+ HoodieSchema fourField = HoodieSchema.createRecord("v",
"org.apache.hudi.test", null, Arrays.asList(
+ HoodieSchemaField.of("metadata",
HoodieSchema.create(HoodieSchemaType.BYTES)),
+ HoodieSchemaField.of("value",
HoodieSchema.createNullable(HoodieSchemaType.BYTES)),
+ HoodieSchemaField.of("typed_value",
HoodieSchema.createNullable(HoodieSchemaType.INT)),
+ HoodieSchemaField.of("extra",
HoodieSchema.createNullable(HoodieSchemaType.STRING))));
+ assertNull(HoodieVariantReconstruction.create(
+ recordWithIdAndVariant(fourField),
recordWithIdAndVariant(HoodieSchema.createVariant()),
+ storageWithReadingShredded(tmp, false)));
+ }
+
+ @Test
+ void ignoresThreeFieldStructWhoseThirdFieldIsNotValue(@TempDir Path tmp) {
+ // Unlike the four-field case above, this one survives the field-count
guard and reaches the
+ // no-`value` arm of the shape check: three fields carrying typed_value
but no `value` must be
+ // read as a user struct, because a shredded group's only optional third
field is `value`.
+ HoodieSchema threeField = HoodieSchema.createRecord("v",
"org.apache.hudi.test", null, Arrays.asList(
+ HoodieSchemaField.of("metadata",
HoodieSchema.create(HoodieSchemaType.BYTES)),
+ HoodieSchemaField.of("typed_value",
HoodieSchema.createNullable(HoodieSchemaType.INT)),
+ HoodieSchemaField.of("extra",
HoodieSchema.createNullable(HoodieSchemaType.STRING))));
+ assertNull(HoodieVariantReconstruction.create(
+ recordWithIdAndVariant(threeField),
recordWithIdAndVariant(HoodieSchema.createVariant()),
+ storageWithReadingShredded(tmp, false)));
+ }
+
@Test
void ignoresShreddedShapeWhenRequestedColumnIsNotVariant(@TempDir Path tmp) {
// A user struct that merely has the {metadata, value, typed_value} shape
must not be
@@ -199,9 +261,12 @@ class TestHoodieVariantReconstruction {
@Test
void reconstructsShreddedVariantsNestedInRecordsArraysAndMaps(@TempDir Path
tmp) {
- // HoodieRowParquetWriteSupport.processNestedDataType shreds variants at
any depth, so a nested
- // variant reaches this reader as a plain {metadata, value, typed_value}
record too. Detection
- // and rebuild must descend into records, array elements and map values,
or the nested payload
+ // The row writer shreds at any depth its write schema asks it to
+ // (HoodieRowParquetWriteSupport.processNestedDataType), so a nested
variant can reach this
+ // reader as a plain {metadata, value, typed_value} record too. On the
AVRO write path such a
+ // file needs a hand-authored write schema -
HoodieAvroWriteSupport.applyForcedShreddingSchema
+ // walks top-level fields only - which is why this is a unit test and not
an end-to-end one.
+ // Detection and rebuild must descend into records, array elements and map
values, or the nested payload
// is dropped the way #19567 dropped the top-level one - and silently,
since nothing at the top
// level looks shredded and neither fail-fast branch is reached.
HoodieSchema fileSchema = recordWithNestedVariants(
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
index b942e5a234ce..007fbd26ca79 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
@@ -504,19 +504,7 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
// Pin the layout the CDC reads: without this, one leg silently
degrades into a
// second copy of the other and its half of the rewrite branch goes
uncovered.
- val baseFiles = listDataParquetFiles(tablePath)
- assert(baseFiles.nonEmpty, s"[$leg] should have a base parquet file
after the insert")
- baseFiles.foreach { filePath =>
- val parquetSchema = readParquetSchema(filePath)
- val variantGroup = getFieldAsGroup(parquetSchema, "v")
- if (shredded) {
- assert(variantGroup.containsField("typed_value"),
- s"[$leg] base file should carry typed_value.
Schema:\n$variantGroup")
- } else {
- assert(!variantGroup.containsField("typed_value"),
- s"[$leg] base file must not carry typed_value.
Schema:\n$variantGroup")
- }
- }
+ assertVariantLayout(tablePath, shredded, leg)
spark.sql(s"""update $tableName set v =
parse_json('{"key":"value2"}'), ts = 1001 where id = 1""")
@@ -551,75 +539,84 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
}
}
- test("Test COW small-file merge preserves shredded VARIANT values") {
- // The #19567 repro: the second insert bin-packs into the existing small
file group, and the
- // small-file merge (HoodieConcatHandle -> HoodieMergeHelper) rewrites the
old base file
- // through the AVRO read path (HoodieAvroParquetReader +
HoodieVariantReconstruction), not the
- // Spark reader context the clustering tests above exercise. Before the
fix, the footer-derived
- // reader schema lost the variant logical type, so the merge's
strict-projection check failed
- // and degenerated the requested schema to the footer schema itself;
reconstruction had no
- // variant column to anchor on, and the writer-schema rewrite silently
dropped typed_value,
- // nulling rows 1-2 after the second commit. Pinned to the AVRO record
type: that is the leg
- // this fix covers, and the record type picks the reader the merge uses.
Unlike the clustering
- // tests, small.file.limit stays at its default on purpose: the bin-pack
is the trigger.
+ test("Test COW small-file merge preserves VARIANT values") {
+ // The #19567 repro. The second insert bin-packs into the existing small
file group, and the
+ // small-file merge (HoodieConcatHandle -> HoodieMergeHelper) rewrites the
old base file.
+ // Before the fix the footer-derived reader schema lost the variant
logical type, so the
+ // strict-projection check failed and degenerated the requested schema to
the footer schema
+ // itself; reconstruction had no variant column to anchor on, and the
writer-schema rewrite
+ // silently dropped typed_value, nulling rows 1-2 after the second commit.
+ //
+ // Swept over both dimensions because each leg pins something different:
+ // - AVRO + shredded: the bug itself. Goes red without the
HoodieMergeHelper alignment.
+ // - AVRO + unshredded: the other direction. alignShreddedVariants runs on
every runMerge, so
+ // an ordinary variant table must round-trip exactly as it did before.
Red here while the
+ // shredded leg stays green means the alignment is reaching columns it
should not.
+ // - SPARK: HoodieSparkParquetReader.getSchema returns a nullable UNION
rather than a RECORD,
+ // so alignShreddedVariants bails at its RECORD/RECORD guard and is a
strict no-op on that
+ // reader - this leg is byte-for-byte unchanged by the fix, and Spark's
own parquet read
+ // handles the shredded column. Swept anyway because nothing else covers
it, and because
+ // that guard is what scopes the fix to the AVRO reader; if it ever
changes, this catches it.
+ //
+ // Unlike the clustering tests above, small.file.limit stays at its
default on purpose: the
+ // bin-pack is the trigger.
assume(HoodieSparkUtils.gteqSpark4_1, "Shredded variant base-file read
requires Spark 4.1 or higher")
- withRecordType(Seq(HoodieRecordType.AVRO))(withTempDir { tmp =>
- val tableName = generateTableName
- val tablePath = tmp.getCanonicalPath
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | v variant,
- | ts long
- |) using hudi
- | location '$tablePath'
- | tblproperties (
- | primaryKey = 'id',
- | type = 'cow',
- | preCombineField = 'ts',
- | hoodie.parquet.variant.write.shredding.enabled = 'true',
- | hoodie.parquet.variant.force.shredding.schema.for.test = 'key
string',
- | hoodie.index.type = 'INMEMORY'
- | )
- """.stripMargin)
-
- spark.sql(s"insert into $tableName values " +
- "(1, parse_json('{\"key\":\"value1\"}'), 1000), " +
- "(2, parse_json('{\"key\":\"value2\"}'), 1000)")
+ Seq(HoodieRecordType.AVRO, HoodieRecordType.SPARK).foreach { recordType =>
+ Seq(true, false).foreach { shredded =>
+ withRecordType(Seq(recordType))(withTempDir { tmp =>
+ val leg = s"$recordType, shredded=$shredded"
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ // The forced shredding schema belongs to the shredded leg only; the
unshredded leg must
+ // reach the writer with shredding off and no forced schema.
+ val forceShreddingProp =
+ if (shredded)
"hoodie.parquet.variant.force.shredding.schema.for.test = 'key string'," else ""
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | v variant,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts',
+ | hoodie.parquet.variant.write.shredding.enabled = '$shredded',
+ | $forceShreddingProp
+ | hoodie.index.type = 'INMEMORY'
+ | )
+ """.stripMargin)
- // Pin the trigger: the first base file must be shredded, or this
degenerates into a
- // plain unshredded merge.
- val firstFiles = listDataParquetFiles(tablePath)
- assert(firstFiles.nonEmpty, "Should have at least one data parquet file
after the first insert")
- firstFiles.foreach { filePath =>
- val parquetSchema = readParquetSchema(filePath)
- val variantGroup = getFieldAsGroup(parquetSchema, "v")
- assert(variantGroup.containsField("typed_value"),
- s"First base file should carry typed_value. Schema:\n$variantGroup")
+ spark.sql(s"insert into $tableName values " +
+ "(1, parse_json('{\"key\":\"value1\"}'), 1000), " +
+ "(2, parse_json('{\"key\":\"value2\"}'), 1000)")
+
+ // Pin the trigger: without this the shredded leg can silently
degenerate into a plain
+ // unshredded merge, or the unshredded leg into a copy of the
shredded one.
+ assertVariantLayout(tablePath, shredded, leg)
+
+ spark.sql(s"insert into $tableName values " +
+ "(3, parse_json('{\"key\":\"value3\"}'), 1000), " +
+ "(4, parse_json('{\"key\":\"value4\"}'), 1000)")
+
+ // Pin that the second commit went through the small-file merge:
both parquet versions
+ // belong to one file group, no second group was created.
+ assertSingleFileGroup(tablePath, leg)
+
+ // Rows 1 and 2 survive only if the merge carried them out of the
base file; nulls here
+ // mean the read path dropped typed_value (#19567).
+ checkAnswer(s"select id, cast(v as string), ts from $tableName order
by id")(
+ Seq(1, "{\"key\":\"value1\"}", 1000),
+ Seq(2, "{\"key\":\"value2\"}", 1000),
+ Seq(3, "{\"key\":\"value3\"}", 1000),
+ Seq(4, "{\"key\":\"value4\"}", 1000)
+ )
+ })
}
-
- spark.sql(s"insert into $tableName values " +
- "(3, parse_json('{\"key\":\"value3\"}'), 1000), " +
- "(4, parse_json('{\"key\":\"value4\"}'), 1000)")
-
- // Pin that the second commit went through the small-file merge: both
parquet versions
- // belong to one file group, no second group was created.
- val fileGroupIds = listDataParquetFiles(tablePath)
- .map(f => FSUtils.getFileId(new HadoopPath(f).getName)).distinct
- assert(fileGroupIds.size == 1,
- s"Second insert should bin-pack into the first file group via the
small-file merge, got: $fileGroupIds")
-
- // Rows 1 and 2 survive only if the merge carried them out of the
shredded base file;
- // nulls here mean the AVRO read path dropped typed_value (#19567).
- checkAnswer(s"select id, cast(v as string), ts from $tableName order by
id")(
- Seq(1, "{\"key\":\"value1\"}", 1000),
- Seq(2, "{\"key\":\"value2\"}", 1000),
- Seq(3, "{\"key\":\"value3\"}", 1000),
- Seq(4, "{\"key\":\"value4\"}", 1000)
- )
- })
+ }
}
test("Test COW small-file merge preserves shredded VARIANT values when the
schema evolves") {
@@ -657,14 +654,7 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
"(1, parse_json('{\"key\":\"value1\"}'), 1000), " +
"(2, parse_json('{\"key\":\"value2\"}'), 1000)")
- val firstFiles = listDataParquetFiles(tablePath)
- assert(firstFiles.nonEmpty, "Should have at least one data parquet file
after the first insert")
- firstFiles.foreach { filePath =>
- val parquetSchema = readParquetSchema(filePath)
- val variantGroup = getFieldAsGroup(parquetSchema, "v")
- assert(variantGroup.containsField("typed_value"),
- s"First base file should carry typed_value. Schema:\n$variantGroup")
- }
+ assertVariantLayout(tablePath, shredded = true, "schema evolves")
// The added column is what breaks the strict projection: the writer
schema now carries a
// field the base file does not have.
@@ -674,10 +664,7 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
"(3, parse_json('{\"key\":\"value3\"}'), 1000, 'n3'), " +
"(4, parse_json('{\"key\":\"value4\"}'), 1000, 'n4')")
- val fileGroupIds = listDataParquetFiles(tablePath)
- .map(f => FSUtils.getFileId(new HadoopPath(f).getName)).distinct
- assert(fileGroupIds.size == 1,
- s"Second insert should bin-pack into the first file group via the
small-file merge, got: $fileGroupIds")
+ assertSingleFileGroup(tablePath, "schema evolves")
// Rows 1-2 keep their variants and pick up a null for the new column;
rows 3-4 carry it.
checkAnswer(s"select id, cast(v as string), ts, note from $tableName
order by id")(
@@ -689,71 +676,6 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
})
}
- test("Test COW small-file merge preserves unshredded VARIANT values") {
- // Companion to the shredded small-file merge test above, with shredding
disabled, split the
- // same way as the clustering pair. The shredded twin only proves the
merge stopped losing
- // typed_value; this one guards the other direction, because
alignShreddedVariants now runs on
- // every HoodieMergeHelper.runMerge. An ordinary variant table writes the
plain {metadata,
- // value} layout, nothing should match the shredded shape, the alignment
must be a no-op, and
- // the bin-pack must keep round-tripping exactly as it did before the fix.
If this goes red
- // while the shredded twin stays green, the alignment is reaching columns
it should not.
- assume(HoodieSparkUtils.gteqSpark4_1, "Variant small-file merge read-back
requires Spark 4.1 or higher")
-
- withRecordType(Seq(HoodieRecordType.AVRO))(withTempDir { tmp =>
- val tableName = generateTableName
- val tablePath = tmp.getCanonicalPath
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | v variant,
- | ts long
- |) using hudi
- | location '$tablePath'
- | tblproperties (
- | primaryKey = 'id',
- | type = 'cow',
- | preCombineField = 'ts',
- | hoodie.parquet.variant.write.shredding.enabled = 'false',
- | hoodie.index.type = 'INMEMORY'
- | )
- """.stripMargin)
-
- spark.sql(s"insert into $tableName values " +
- "(1, parse_json('{\"key\":\"value1\"}'), 1000), " +
- "(2, parse_json('{\"key\":\"value2\"}'), 1000)")
-
- // Pin the layout: these files must NOT carry typed_value, or this twin
silently becomes a
- // copy of the shredded test above and the unshredded leg of the merge
goes uncovered.
- val firstFiles = listDataParquetFiles(tablePath)
- assert(firstFiles.nonEmpty, "Should have at least one data parquet file
after the first insert")
- firstFiles.foreach { filePath =>
- val parquetSchema = readParquetSchema(filePath)
- val variantGroup = getFieldAsGroup(parquetSchema, "v")
- assert(!variantGroup.containsField("typed_value"),
- s"Unshredded base file must not carry typed_value.
Schema:\n$variantGroup")
- }
-
- spark.sql(s"insert into $tableName values " +
- "(3, parse_json('{\"key\":\"value3\"}'), 1000), " +
- "(4, parse_json('{\"key\":\"value4\"}'), 1000)")
-
- // Same bin-pack pin as the shredded twin: one file group, so this
really went through the
- // small-file merge and not a fresh insert.
- val fileGroupIds = listDataParquetFiles(tablePath)
- .map(f => FSUtils.getFileId(new HadoopPath(f).getName)).distinct
- assert(fileGroupIds.size == 1,
- s"Second insert should bin-pack into the first file group via the
small-file merge, got: $fileGroupIds")
-
- checkAnswer(s"select id, cast(v as string), ts from $tableName order by
id")(
- Seq(1, "{\"key\":\"value1\"}", 1000),
- Seq(2, "{\"key\":\"value2\"}", 1000),
- Seq(3, "{\"key\":\"value3\"}", 1000),
- Seq(4, "{\"key\":\"value4\"}", 1000)
- )
- })
- }
-
test("Test bulk_insert row-writer round-trips VARIANT") {
assume(HoodieSparkUtils.gteqSpark4_0, "Variant type requires Spark 4.0 or
higher")
@@ -1250,6 +1172,34 @@ class TestVariantDataType extends HoodieSparkSqlTestBase
{
})
}
+ /**
+ * Pins the on-disk layout of the `v` column across every base file. Without
it a leg meant to
+ * exercise the shredded path can silently degenerate into the unshredded
one, or the reverse,
+ * and the branch it was written for goes uncovered.
+ */
+ private def assertVariantLayout(tablePath: String, shredded: Boolean, leg:
String): Unit = {
+ val files = listDataParquetFiles(tablePath)
+ assert(files.nonEmpty, s"[$leg] should have at least one data parquet
file")
+ files.foreach { filePath =>
+ val variantGroup = getFieldAsGroup(readParquetSchema(filePath), "v")
+ if (shredded) {
+ assert(variantGroup.containsField("typed_value"),
+ s"[$leg] base file should carry typed_value. Schema:\n$variantGroup")
+ } else {
+ assert(!variantGroup.containsField("typed_value"),
+ s"[$leg] base file must not carry typed_value.
Schema:\n$variantGroup")
+ }
+ }
+ }
+
+ /** Pins that a write bin-packed into the existing file group rather than
creating a new one. */
+ private def assertSingleFileGroup(tablePath: String, leg: String): Unit = {
+ val fileGroupIds = listDataParquetFiles(tablePath)
+ .map(f => FSUtils.getFileId(new HadoopPath(f).getName)).distinct
+ assert(fileGroupIds.size == 1,
+ s"[$leg] insert should bin-pack into the first file group via the
small-file merge, got: $fileGroupIds")
+ }
+
/**
* Lists data parquet files in the table directory, excluding Hudi metadata
files.
*/
diff --git
a/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstructionRoundTrip.java
b/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstructionRoundTrip.java
index 0f851e98fa6a..412f79541663 100644
---
a/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstructionRoundTrip.java
+++
b/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieVariantReconstructionRoundTrip.java
@@ -43,6 +43,7 @@ import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
/**
* Round-trip coverage for the successful {@code create} -> {@code
reconstruct} path of
@@ -104,6 +105,70 @@ class TestHoodieVariantReconstructionRoundTrip {
assertEquals(original.toJson(ZoneOffset.UTC),
rebuilt.toJson(ZoneOffset.UTC));
}
+ @Test
+ void createThenReconstructRebuildsAValueLessShreddedGroup(@TempDir Path tmp)
throws Exception {
+ // The shredding spec lets a writer omit `value` when nothing is left
over. TestHoodieVariantReconstruction
+ // pins that Hudi detects the resulting two-field group, but its stub
provider ignores the shredded
+ // schema, so this is the only place the real provider sees the shape:
buildVariantSchema assigns
+ // variantIdx = -1 there, which shifts every other ordinal and is what
ShreddingUtils.rebuild reads to
+ // decide there is no residual. Get the mapping wrong and the rebuild
reads the wrong Avro field.
+ Map<String, HoodieSchema> shreddedFields = new LinkedHashMap<>();
+ shreddedFields.put("a", HoodieSchema.create(HoodieSchemaType.STRING));
+ shreddedFields.put("b", HoodieSchema.create(HoodieSchemaType.LONG));
+ HoodieSchema.Variant shreddedVariant =
HoodieSchema.createVariantShreddedObject(shreddedFields);
+ HoodieSchema.Variant unshreddedVariant = HoodieSchema.createVariant();
+
+ // Shred with the real provider first, then drop the (null) top-level
residual, which yields exactly
+ // what a writer that omits `value` would have put on disk.
+ Spark4VariantShreddingProvider provider = new
Spark4VariantShreddingProvider();
+ Variant original = VariantBuilder.parseJson("{\"a\":\"x\",\"b\":5}",
false);
+ GenericRecord unshreddedV = new
GenericData.Record(unshreddedVariant.getAvroSchema());
+ unshreddedV.put(HoodieSchema.Variant.VARIANT_METADATA_FIELD,
ByteBuffer.wrap(original.getMetadata()));
+ unshreddedV.put(HoodieSchema.Variant.VARIANT_VALUE_FIELD,
ByteBuffer.wrap(original.getValue()));
+ GenericRecord shreddedV =
+ provider.shredVariantRecord(unshreddedV,
shreddedVariant.getAvroSchema(), shreddedVariant);
+ assertNull(shreddedV.get(HoodieSchema.Variant.VARIANT_VALUE_FIELD),
+ "both fields matched the shredding schema, so there is no residual and
dropping `value` loses nothing");
+
+ // The file side is a plain {metadata, typed_value} record: a real footer
conversion loses the variant
+ // logical type, so this is also what shape detection has to recognize.
+ HoodieSchema valueLessShredded = HoodieSchema.createRecord("v",
"org.apache.hudi.test", null, Arrays.asList(
+ HoodieSchemaField.of(HoodieSchema.Variant.VARIANT_METADATA_FIELD,
HoodieSchema.create(HoodieSchemaType.BYTES)),
+ HoodieSchemaField.of(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD,
+ shreddedVariant.getTypedValueField().get())));
+ HoodieSchema fileSchema = HoodieSchema.createRecord("r",
"org.apache.hudi.test", null, Arrays.asList(
+ HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.LONG)),
+ HoodieSchemaField.of("v", valueLessShredded)));
+ HoodieSchema requestedSchema = HoodieSchema.createRecord("r",
"org.apache.hudi.test", null, Arrays.asList(
+ HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.LONG)),
+ HoodieSchemaField.of("v", unshreddedVariant)));
+
+ HoodieStorage storage = HoodieTestUtils.getStorage(tmp.toString()); //
allow.reading.shredded defaults true
+ HoodieVariantReconstruction reconstruction =
+ HoodieVariantReconstruction.create(fileSchema, requestedSchema,
storage);
+ assertNotNull(reconstruction, "a shredded group with no value column must
still engage");
+
+ GenericRecord valueLessV = new GenericData.Record(
+
reconstruction.intermediateSchema().getAvroSchema().getField("v").schema());
+ valueLessV.put(HoodieSchema.Variant.VARIANT_METADATA_FIELD,
+ shreddedV.get(HoodieSchema.Variant.VARIANT_METADATA_FIELD));
+ valueLessV.put(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD,
+ shreddedV.get(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD));
+
+ GenericRecord input = new
GenericData.Record(reconstruction.intermediateSchema().getAvroSchema());
+ input.put("id", 7L);
+ input.put("v", valueLessV);
+
+ IndexedRecord out = reconstruction.reconstruct(input);
+
+ assertEquals(7L, out.get(0));
+ GenericRecord rebuiltV = (GenericRecord) out.get(1);
+ Variant rebuilt = new Variant(
+ toBytes(rebuiltV.get(HoodieSchema.Variant.VARIANT_VALUE_FIELD)),
+ toBytes(rebuiltV.get(HoodieSchema.Variant.VARIANT_METADATA_FIELD)));
+ assertEquals(original.toJson(ZoneOffset.UTC),
rebuilt.toJson(ZoneOffset.UTC));
+ }
+
private static byte[] toBytes(Object byteBuffer) {
ByteBuffer buf = ((ByteBuffer) byteBuffer).duplicate();
byte[] out = new byte[buf.remaining()];
