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 c0ed52483fb2 fix(spark): read shredded variant leaves like Spark 4
(#20332)
c0ed52483fb2 is described below
commit c0ed52483fb22e57c57a9c7b211b9bcb4919cf5b
Author: voonhous <[email protected]>
AuthorDate: Tue Oct 27 23:26:34 2026 +0800
fix(spark): read shredded variant leaves like Spark 4 (#20332)
* fix(spark): match Spark 4 shredded variant reads
Hudi rebuilt shredded variants from the footer's Avro conversion, which
drops DECIMAL on INT32/INT64, the unsigned flag, JSON and NANOS. Files
written by other engines therefore read DECIMAL on INT32/INT64 as the
unscaled integer, failed on plain FLBA and TIMESTAMP(MILLIS), and read
TIME and TIMESTAMP(NANOS) as integers where Spark 4 rejects the file.
Read each typed_value leaf from its Parquet type instead, as Spark 4's
ParquetToSparkSchemaConverter and SparkShreddingUtils.buildVariantSchema
do:
- HoodieAvroParquetReader passes the file's MessageType to
HoodieVariantReconstruction, which prepares one rebuilder per shredded
variant column at file open (records, LIST encodings, map values).
- VariantShreddingProvider.rebuildVariantRecord becomes prepareRebuild,
so the layout is validated once per column instead of per record.
- Spark4VariantShreddingProvider maps leaves like Spark (unsigned ints,
JSON as string, millis scaled to micros, unscaled decimals) and
rejects TIME, TIMESTAMP(NANOS), FLOAT16, UUID, INTERVAL and INT96 at
open. Code made dead by the new path is removed.
Addresses the read side of #19442.
* docs(spark): cross-reference variant list-element rules
The planner and the Spark 4 provider each resolve a parquet LIST
element under a different converter's rules and on different
groups. Say so on both so a change to one points at the other.
---
.../hudi/common/avro/VariantShreddingProvider.java | 52 +-
.../io/storage/hadoop/HoodieAvroParquetReader.java | 16 +-
.../hadoop/HoodieVariantReconstruction.java | 211 +++++---
.../avro/TestHoodieAvroWriteSupportShredding.java | 3 +-
.../hadoop/TestHoodieVariantReconstruction.java | 103 ++--
.../variant/Spark4VariantShreddingProvider.java | 529 +++++++++++++++++----
.../TestHoodieVariantReconstructionRoundTrip.java | 129 ++++-
.../TestSpark4VariantShreddingProvider.java | 264 +++++++++-
8 files changed, 1066 insertions(+), 241 deletions(-)
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/avro/VariantShreddingProvider.java
b/hudi-common/src/main/java/org/apache/hudi/common/avro/VariantShreddingProvider.java
index c77c62ed92e0..65ff26d15187 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/avro/VariantShreddingProvider.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/avro/VariantShreddingProvider.java
@@ -23,9 +23,10 @@ import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
+import org.apache.parquet.schema.GroupType;
/**
- * Interface for shredding variant values at write time.
+ * Interface for shredding variant values at write time and reconstructing
them at read time.
* <p>
* Implementations parse variant binary data (value + metadata bytes) and
produce
* a shredded {@link GenericRecord} with typed_value columns populated
according
@@ -65,24 +66,39 @@ public interface VariantShreddingProvider {
HoodieSchema.Variant variantSchema);
/**
- * Reconstruct an unshredded variant GenericRecord from a shredded one (the
inverse of
- * {@link #shredVariantRecord}).
+ * Prepares the reconstruction of one shredded variant column of a parquet
file (the inverse of
+ * {@link #shredVariantRecord}), validating its shredding layout once, when
the file is opened.
* <p>
* Used on the read path: records read from an already-shredded base file
(compaction/clustering)
- * arrive with {@code typed_value} populated. This rebuilds the full variant
binary so the record
- * presents the standard unshredded {@code {metadata, value}} shape before
it reaches the
- * merger/writer.
+ * arrive with {@code typed_value} populated. The returned rebuilder turns
each such record back
+ * into the standard unshredded {@code {metadata, value}} shape before it
reaches the merger/writer.
+ * <p>
+ * The layout comes from the file's parquet type rather than from the Avro
schema the record is read
+ * at: converting the parquet footer to Avro loses encodings (unsigned and
decimal INT32/INT64, JSON,
+ * nanosecond timestamps) that decide how a {@code typed_value} must be read.
*
- * @param shreddedVariant GenericRecord with {value, metadata, typed_value}
read from a shredded base file
- * @param shreddedSchema the Avro schema of {@code shreddedVariant}
(carries typed_value)
- * @param unshreddedSchema target Avro schema with {value: ByteBuffer,
metadata: ByteBuffer}
- * @return a GenericRecord conforming to {@code unshreddedSchema} with the
full reconstructed
- * variant binary in {@code value}, or {@code null} when {@code
shreddedVariant} is null
- * @throws org.apache.hudi.exception.HoodieException if {@code
shreddedVariant} is missing its
- * required {@code metadata} field (a malformed shredded base file)
+ * @param columnPath dotted path of the variant column, for error
messages
+ * @param parquetVariantType the variant column's parquet group, as it sits
in the file
+ * @param unshreddedSchema target Avro schema with {value: ByteBuffer,
metadata: ByteBuffer}
+ * @return a rebuilder for records of this column
+ * @throws org.apache.hudi.exception.HoodieException if the column's
shredding layout cannot be read
*/
- GenericRecord rebuildVariantRecord(
- GenericRecord shreddedVariant,
- Schema shreddedSchema,
- Schema unshreddedSchema);
-}
\ No newline at end of file
+ VariantRebuilder prepareRebuild(String columnPath, GroupType
parquetVariantType, Schema unshreddedSchema);
+
+ /**
+ * Rebuilds the records of one shredded variant column, as prepared by
{@link #prepareRebuild}.
+ */
+ interface VariantRebuilder {
+
+ /**
+ * Rebuilds the full variant binary of one record of the column.
+ *
+ * @param shreddedVariant GenericRecord with {value, metadata,
typed_value} read from a shredded base file
+ * @return a GenericRecord conforming to the unshredded schema with the
full reconstructed variant
+ * binary in {@code value}, or {@code null} when {@code
shreddedVariant} is null
+ * @throws org.apache.hudi.exception.HoodieException if {@code
shreddedVariant} is missing its
+ * required {@code metadata} field (a malformed shredded base file)
+ */
+ GenericRecord rebuild(GenericRecord shreddedVariant);
+ }
+}
diff --git
a/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieAvroParquetReader.java
b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieAvroParquetReader.java
index 118e7c63856b..257801c13db0 100644
---
a/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieAvroParquetReader.java
+++
b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieAvroParquetReader.java
@@ -52,6 +52,7 @@ import org.apache.parquet.avro.HoodieAvroParquetReaderBuilder;
import org.apache.parquet.hadoop.ParquetInputFormat;
import org.apache.parquet.hadoop.ParquetReader;
import org.apache.parquet.hadoop.util.HadoopInputFile;
+import org.apache.parquet.schema.MessageType;
import java.io.IOException;
import java.util.ArrayList;
@@ -72,6 +73,7 @@ public class HoodieAvroParquetReader extends
HoodieAvroFileReader {
private final HoodieStorage storage;
private final FileFormatUtils parquetUtils;
private final List<ParquetReaderIterator> readerIterators = new
ArrayList<>();
+ private Option<MessageType> fileParquetSchema = Option.empty();
private Option<HoodieSchema> fileSchema = Option.empty();
public HoodieAvroParquetReader(HoodieStorage storage, StoragePath path) {
@@ -127,12 +129,22 @@ public class HoodieAvroParquetReader extends
HoodieAvroFileReader {
// Lazy initialization with caching: read schema from parquet file footer
on first call,
// then cache it in fileSchema to avoid repeated I/O on subsequent calls
return fileSchema.orElseGet(() -> {
- HoodieSchema schema = parquetUtils.readSchema(storage, path);
+ // Same conversion as ParquetUtils.readSchema, from the footer cached
for the variant reconstruction.
+ HoodieSchema schema =
getAvroSchemaConverter(storage.getConf().unwrapAs(Configuration.class))
+ .convert(getParquetSchema());
fileSchema = Option.ofNullable(schema);
return schema;
});
}
+ private MessageType getParquetSchema() {
+ return fileParquetSchema.orElseGet(() -> {
+ MessageType messageType = ParquetUtils.readFileMetadataOnly(storage,
path).getFileMetaData().getSchema();
+ fileParquetSchema = Option.of(messageType);
+ return messageType;
+ });
+ }
+
@Override
public void close() {
readerIterators.forEach(ParquetReaderIterator::close);
@@ -190,7 +202,7 @@ public class HoodieAvroParquetReader extends
HoodieAvroFileReader {
Configuration hadoopConf =
storage.getConf().unwrapCopyAs(Configuration.class);
// If the on-disk file shreds variant columns, read them in their shredded
(typed_value-bearing)
// shape and reconstruct the unshredded variant per-record below (#18931).
null when not needed.
- HoodieVariantReconstruction reconstruction =
HoodieVariantReconstruction.create(getSchema(), schema, storage);
+ HoodieVariantReconstruction reconstruction =
HoodieVariantReconstruction.create(getSchema(), getParquetSchema(), schema,
storage);
HoodieSchema readSchema = reconstruction == null ? schema :
reconstruction.intermediateSchema();
HoodieSchema repairedFileSchema =
HoodieSchemaRepair.repairLogicalTypes(getSchema(), readSchema);
Option<HoodieSchema> promotedSchema = Option.empty();
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 75b64a8afa56..64adb787624c 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
@@ -35,7 +35,12 @@ import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.IndexedRecord;
+import org.apache.parquet.avro.AvroSchemaConverter;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.Type;
+import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -47,7 +52,7 @@ import java.util.Map;
* <p>parquet-avro does not understand variant shredding, so a shredded
variant column comes back as
* a raw {@code {metadata, value, typed_value}} record. This reads such
columns at their shredded
* (typed_value-bearing) schema, then rebuilds the standard unshredded {@code
{metadata, value}}
- * variant via {@link VariantShreddingProvider#rebuildVariantRecord} before
records reach the
+ * variant via {@link VariantShreddingProvider#prepareRebuild} before records
reach the
* merger/writer. The Spark/InternalRow read path reconstructs natively and
does not use this.
*
* <p>See https://github.com/apache/hudi/issues/18931.
@@ -55,13 +60,10 @@ import java.util.Map;
final class HoodieVariantReconstruction {
private final HoodieSchema intermediateSchema;
- private final VariantShreddingProvider provider;
private final Rebuilder rootRebuilder;
- private HoodieVariantReconstruction(HoodieSchema intermediateSchema,
- VariantShreddingProvider provider,
Rebuilder rootRebuilder) {
+ private HoodieVariantReconstruction(HoodieSchema intermediateSchema,
Rebuilder rootRebuilder) {
this.intermediateSchema = intermediateSchema;
- this.provider = provider;
this.rootRebuilder = rootRebuilder;
}
@@ -78,15 +80,27 @@ final class HoodieVariantReconstruction {
* none is needed (the file has no shredded variant columns). Throws when
the file has shredded
* variant columns to reconstruct but reading shredded variants is disabled,
or no provider is
* available: either way, reading at the unshredded schema would silently
drop the typed_value payload.
+ * Also throws when a shredded variant column uses a typed_value encoding
that cannot be read, so the
+ * failure happens when the file is opened rather than on the first affected
record.
+ *
+ * @param fileSchema the file schema, converted from {@code
fileParquetSchema}
+ * @param fileParquetSchema the file's parquet schema, which carries the
typed_value encodings the
+ * Avro conversion loses
+ * @param requestedSchema the schema the records are requested at
+ * @param storage the storage the file is read with
*/
- static HoodieVariantReconstruction create(HoodieSchema fileSchema,
HoodieSchema requestedSchema, HoodieStorage storage) {
+ static HoodieVariantReconstruction create(HoodieSchema fileSchema,
MessageType fileParquetSchema,
+ HoodieSchema requestedSchema,
HoodieStorage storage) {
if (requestedSchema.getType() != HoodieSchemaType.RECORD ||
fileSchema.getType() != HoodieSchemaType.RECORD) {
return null;
}
// Records leave this reader unshredded; output field order matches the
requested/intermediate order.
HoodieSchema outputSchema =
VariantSchemaUtils.stripVariantShredding(requestedSchema);
- Rebuilder rootRebuilder = buildRebuilder(outputSchema, fileSchema);
+ // The list layout rule must match the one the file schema was converted
with (see
+ // HoodieAvroParquetReader.tryOverrideDefaultConfigs).
+ Planner planner = new
Planner(storage.getConf().getBoolean(AvroSchemaConverter.ADD_LIST_ELEMENT_RECORDS,
true));
+ Rebuilder rootRebuilder = planner.plan(outputSchema, fileSchema,
fileParquetSchema, null);
if (rootRebuilder == null) {
// No shredded variant columns in the file: nothing to reconstruct,
regardless of the flag.
return null;
@@ -110,9 +124,12 @@ final class HoodieVariantReconstruction {
+ HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_PROVIDER_CLASS.key()
+ " or add a provider implementation (e.g. the Spark variant module)
to the classpath.");
}
+ for (VariantColumnRebuilder variantColumn : planner.variantColumns) {
+ variantColumn.prepare(provider);
+ }
return new HoodieVariantReconstruction(
- VariantSchemaUtils.toShreddedReadSchema(requestedSchema, fileSchema),
provider, rootRebuilder);
+ VariantSchemaUtils.toShreddedReadSchema(requestedSchema, fileSchema),
rootRebuilder);
}
/**
@@ -120,80 +137,142 @@ final class HoodieVariantReconstruction {
* a record conforming to the unshredded output schema.
*/
IndexedRecord reconstruct(IndexedRecord in) {
- return (IndexedRecord) rootRebuilder.rebuild(in, provider);
+ return (IndexedRecord) rootRebuilder.rebuild(in);
}
- /**
- * Plans the rebuild for one schema position, or returns {@code null} when
nothing below it is a
- * shredded variant and the value can be passed through untouched. The walk
is driven by the
- * output (requested) side because that is the shape the record read from
parquet has; the file
- * 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, matching what the writers
can emit; see
- * {@code VariantSchemaUtils.swapShreddedVariantFields} for what actually
produces a nested
- * shredded file today (both the row and the AVRO write path shred variant
record members at any
- * depth off the forced-shredding property; a variant that is directly an
array element or a map
- * value shreds on neither path unless the write schema itself declares
typed_value there).
- */
- private static Rebuilder buildRebuilder(HoodieSchema outputSchema,
HoodieSchema fileSchema) {
- if (VariantSchemaUtils.isShreddedVariantTarget(fileSchema, outputSchema)) {
- return new VariantRebuilder(fileSchema.getNonNullType().getAvroSchema(),
- outputSchema.getNonNullType().getAvroSchema());
- }
- HoodieSchema output = outputSchema.getNonNullType();
- HoodieSchema file = fileSchema.getNonNullType();
- if (output.getType() != file.getType()) {
- return null;
+ /** Plans the rebuild of a file, collecting its shredded variant columns for
preparation. */
+ private static final class Planner {
+ // Mirrors the file schema conversion's ADD_LIST_ELEMENT_RECORDS, which
decides the list element.
+ private final boolean assumeRepeatedIsListElement;
+ private final List<VariantColumnRebuilder> variantColumns = new
ArrayList<>();
+
+ private Planner(boolean assumeRepeatedIsListElement) {
+ this.assumeRepeatedIsListElement = assumeRepeatedIsListElement;
}
- switch (output.getType()) {
- case RECORD: {
- List<HoodieSchemaField> outputFields = output.getFields();
- Rebuilder[] fieldRebuilders = new Rebuilder[outputFields.size()];
- boolean anyTarget = false;
- for (int i = 0; i < outputFields.size(); i++) {
- HoodieSchemaField outputField = outputFields.get(i);
- Option<HoodieSchemaField> fileField =
file.getField(outputField.name());
- fieldRebuilders[i] = fileField.isPresent()
- ? buildRebuilder(outputField.schema(), fileField.get().schema())
- : null;
- anyTarget |= fieldRebuilders[i] != null;
+
+ /**
+ * Plans the rebuild for one schema position, or returns {@code null} when
nothing below it is a
+ * shredded variant and the value can be passed through untouched. The
walk is driven by the
+ * output (requested) side because that is the shape the record read from
parquet has; the file
+ * side, and the parquet type it was converted from, are 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, matching what the writers can emit; see
+ * {@code VariantSchemaUtils.swapShreddedVariantFields} for what actually
produces a nested
+ * shredded file today (both the row and the AVRO write path shred variant
record members at any
+ * depth off the forced-shredding property; a variant that is directly an
array element or a map
+ * value shreds on neither path unless the write schema itself declares
typed_value there).
+ */
+ private Rebuilder plan(HoodieSchema outputSchema, HoodieSchema fileSchema,
Type parquetType, String path) {
+ if (VariantSchemaUtils.isShreddedVariantTarget(fileSchema,
outputSchema)) {
+ if (parquetType == null || parquetType.isPrimitive()) {
+ throw new IllegalStateException("Shredded variant column '" + path
+ + "' has no matching parquet group in the file schema: " +
parquetType);
}
- return anyTarget ? new RecordRebuilder(output.getAvroSchema(),
fieldRebuilders) : null;
+ VariantColumnRebuilder variantColumn = new VariantColumnRebuilder(
+ path, parquetType.asGroupType(),
outputSchema.getNonNullType().getAvroSchema());
+ variantColumns.add(variantColumn);
+ return variantColumn;
+ }
+ HoodieSchema output = outputSchema.getNonNullType();
+ HoodieSchema file = fileSchema.getNonNullType();
+ if (output.getType() != file.getType() || parquetType == null ||
parquetType.isPrimitive()) {
+ return null;
}
- case ARRAY: {
- Rebuilder elementRebuilder = buildRebuilder(output.getElementType(),
file.getElementType());
- return elementRebuilder == null ? null : new
ArrayRebuilder(output.getAvroSchema(), elementRebuilder);
+ GroupType group = parquetType.asGroupType();
+ switch (output.getType()) {
+ case RECORD: {
+ List<HoodieSchemaField> outputFields = output.getFields();
+ Rebuilder[] fieldRebuilders = new Rebuilder[outputFields.size()];
+ boolean anyTarget = false;
+ for (int i = 0; i < outputFields.size(); i++) {
+ HoodieSchemaField outputField = outputFields.get(i);
+ Option<HoodieSchemaField> fileField =
file.getField(outputField.name());
+ fieldRebuilders[i] = fileField.isPresent()
+ ? plan(outputField.schema(), fileField.get().schema(),
childType(group, outputField.name()),
+ path == null ? outputField.name() : path + "." +
outputField.name())
+ : null;
+ anyTarget |= fieldRebuilders[i] != null;
+ }
+ return anyTarget ? new RecordRebuilder(output.getAvroSchema(),
fieldRebuilders) : null;
+ }
+ case ARRAY: {
+ Rebuilder elementRebuilder = plan(output.getElementType(),
file.getElementType(),
+ listElementType(group), path + ".element");
+ return elementRebuilder == null ? null : new
ArrayRebuilder(output.getAvroSchema(), elementRebuilder);
+ }
+ case MAP: {
+ Rebuilder valueRebuilder = plan(output.getValueType(),
file.getValueType(),
+ mapValueType(group), path + ".value");
+ return valueRebuilder == null ? null : new
MapRebuilder(valueRebuilder);
+ }
+ default:
+ return null;
}
- case MAP: {
- Rebuilder valueRebuilder = buildRebuilder(output.getValueType(),
file.getValueType());
- return valueRebuilder == null ? null : new
MapRebuilder(valueRebuilder);
+ }
+
+ /**
+ * The element type of a parquet LIST group, by the same rules the file
schema's Avro conversion
+ * applied (AvroSchemaConverter.isElementType): the repeated field itself
in the legacy 2-level
+ * layouts, otherwise the single field of the repeated group (3-level
layout). Applies only to
+ * lists around a variant column;
Spark4VariantShreddingProvider.listElementType reads the
+ * typed_value list inside one under Spark's rules, so the two are not
meant to agree.
+ */
+ private Type listElementType(GroupType list) {
+ if (list.getFieldCount() != 1) {
+ return null;
}
- default:
+ Type repeated = list.getType(0);
+ boolean repeatedIsElement = repeated.isPrimitive()
+ || repeated.asGroupType().getFieldCount() > 1
+ ||
repeated.asGroupType().getType(0).isRepetition(Type.Repetition.REPEATED)
+ || repeated.getName().equals("array")
+ || repeated.getName().equals(list.getName() + "_tuple")
+ || assumeRepeatedIsListElement;
+ return repeatedIsElement ? repeated : repeated.asGroupType().getType(0);
+ }
+
+ /** The value type of a parquet MAP group: the second field of its
repeated key_value group. */
+ private static Type mapValueType(GroupType map) {
+ if (map.getFieldCount() != 1 || map.getType(0).isPrimitive()) {
return null;
+ }
+ GroupType keyValue = map.getType(0).asGroupType();
+ return keyValue.getFieldCount() == 2 ? keyValue.getType(1) : null;
+ }
+
+ private static Type childType(GroupType group, String name) {
+ return group.containsField(name) ? group.getType(name) : null;
}
}
/** Rebuilds one value read at the file's shape into its unshredded output
shape. */
private interface Rebuilder {
- Object rebuild(Object value, VariantShreddingProvider provider);
+ Object rebuild(Object value);
}
- private static final class VariantRebuilder implements Rebuilder {
- private final Schema shreddedSchema;
+ private static final class VariantColumnRebuilder implements Rebuilder {
+ private final String path;
+ private final GroupType parquetType;
private final Schema unshreddedSchema;
+ // Set by prepare() once the provider is loaded, before any record is
rebuilt.
+ private VariantShreddingProvider.VariantRebuilder rebuilder;
- private VariantRebuilder(Schema shreddedSchema, Schema unshreddedSchema) {
- this.shreddedSchema = shreddedSchema;
+ private VariantColumnRebuilder(String path, GroupType parquetType, Schema
unshreddedSchema) {
+ this.path = path;
+ this.parquetType = parquetType;
this.unshreddedSchema = unshreddedSchema;
}
+ private void prepare(VariantShreddingProvider provider) {
+ rebuilder = provider.prepareRebuild(path, parquetType, unshreddedSchema);
+ }
+
@Override
- public Object rebuild(Object value, VariantShreddingProvider provider) {
+ public Object rebuild(Object value) {
// A null variant passes through unchanged.
- return value instanceof GenericRecord
- ? provider.rebuildVariantRecord((GenericRecord) value,
shreddedSchema, unshreddedSchema)
- : value;
+ return value instanceof GenericRecord ?
rebuilder.rebuild((GenericRecord) value) : value;
}
}
@@ -208,7 +287,7 @@ final class HoodieVariantReconstruction {
}
@Override
- public Object rebuild(Object value, VariantShreddingProvider provider) {
+ public Object rebuild(Object value) {
if (!(value instanceof IndexedRecord)) {
return value;
}
@@ -216,7 +295,7 @@ final class HoodieVariantReconstruction {
GenericRecord out = new GenericData.Record(outputSchema);
for (int i = 0; i < fieldRebuilders.length; i++) {
Object fieldValue = in.get(i);
- out.put(i, fieldRebuilders[i] == null ? fieldValue :
fieldRebuilders[i].rebuild(fieldValue, provider));
+ out.put(i, fieldRebuilders[i] == null ? fieldValue :
fieldRebuilders[i].rebuild(fieldValue));
}
return out;
}
@@ -232,14 +311,14 @@ final class HoodieVariantReconstruction {
}
@Override
- public Object rebuild(Object value, VariantShreddingProvider provider) {
+ public Object rebuild(Object value) {
if (!(value instanceof List)) {
return value;
}
List<?> in = (List<?>) value;
GenericData.Array<Object> out = new GenericData.Array<>(in.size(),
outputSchema);
for (Object element : in) {
- out.add(elementRebuilder.rebuild(element, provider));
+ out.add(elementRebuilder.rebuild(element));
}
return out;
}
@@ -253,14 +332,14 @@ final class HoodieVariantReconstruction {
}
@Override
- public Object rebuild(Object value, VariantShreddingProvider provider) {
+ public Object rebuild(Object value) {
if (!(value instanceof Map)) {
return value;
}
Map<?, ?> in = (Map<?, ?>) value;
Map<Object, Object> out = new LinkedHashMap<>(in.size());
for (Map.Entry<?, ?> entry : in.entrySet()) {
- out.put(entry.getKey(), valueRebuilder.rebuild(entry.getValue(),
provider));
+ out.put(entry.getKey(), valueRebuilder.rebuild(entry.getValue()));
}
return out;
}
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 4c35a5300d6d..9e31357aa264 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
@@ -677,8 +677,7 @@ class TestHoodieAvroWriteSupportShredding {
}
@Override
- public GenericRecord rebuildVariantRecord(GenericRecord shreddedVariant,
Schema shreddedSchema,
- Schema unshreddedSchema) {
+ public VariantRebuilder prepareRebuild(String columnPath, GroupType
parquetVariantType, Schema unshreddedSchema) {
throw new UnsupportedOperationException(
"the round trip reads raw parquet; the read half is
TestHoodieVariantReconstruction's");
}
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 99d1cf45f2fa..4cd1af7d9397 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
@@ -32,17 +32,26 @@ import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.IndexedRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.avro.AvroSchemaConverter;
+import org.apache.parquet.avro.AvroWriteSupport;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.MessageType;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.nio.ByteBuffer;
import java.nio.file.Path;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import static
org.apache.parquet.avro.HoodieAvroParquetSchemaConverter.getAvroSchemaConverter;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -56,6 +65,15 @@ class TestHoodieVariantReconstruction {
Collections.singletonList(HoodieSchemaField.of("v", variantSchema)));
}
+ /** Creates the reconstruction with the parquet schema the file schema would
have been read from. */
+ private static HoodieVariantReconstruction create(HoodieSchema fileSchema,
HoodieSchema requestedSchema,
+ HoodieStorage storage) {
+ MessageType fileParquetSchema = fileSchema.getType() ==
HoodieSchemaType.RECORD
+ ?
getAvroSchemaConverter(storage.getConf().unwrapAs(Configuration.class)).convert(fileSchema)
+ : null;
+ return HoodieVariantReconstruction.create(fileSchema, fileParquetSchema,
requestedSchema, storage);
+ }
+
private static HoodieStorage storageWithReadingShredded(Path tmp, boolean
enabled) {
HoodieStorage storage = HoodieTestUtils.getStorage(tmp.toString());
storage.getConf().set(HoodieStorageConfig.PARQUET_VARIANT_ALLOW_READING_SHREDDED.key(),
@@ -68,29 +86,33 @@ class TestHoodieVariantReconstruction {
HoodieSchema fileSchema = recordWithVariant(
HoodieSchema.createVariantShredded(HoodieSchema.create(HoodieSchemaType.INT)));
HoodieSchema requestedSchema =
recordWithVariant(HoodieSchema.createVariant());
+ HoodieStorage storage = storageWithReadingShredded(tmp, false);
+
storage.getConf().set(HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_PROVIDER_CLASS.key(),
+ TestVariantShreddingProvider.class.getName());
+ TestVariantShreddingProvider.PREPARED.clear();
- HoodieException ex = assertThrows(HoodieException.class, () ->
- HoodieVariantReconstruction.create(fileSchema, requestedSchema,
- storageWithReadingShredded(tmp, false)));
+ HoodieException ex = assertThrows(HoodieException.class, () ->
create(fileSchema, requestedSchema, storage));
// The flag check must run after shredded-column detection: dropping
typed_value silently would
// corrupt rows whose payload lives there, so a disabled read of a
shredded file fails fast.
assertTrue(ex.getMessage().contains("reading shredded variants is
disabled"), ex.getMessage());
+ // And before any column is prepared, so a provider's own layout errors
cannot mask it.
+ assertTrue(TestVariantShreddingProvider.PREPARED.isEmpty(),
TestVariantShreddingProvider.PREPARED::toString);
}
@Test
void returnsNullWhenNoShreddedColumnEvenIfReadingDisabled(@TempDir Path tmp)
{
HoodieSchema schema = recordWithVariant(HoodieSchema.createVariant());
// No shredded variant column to reconstruct: nothing to do regardless of
the flag.
- assertNull(HoodieVariantReconstruction.create(schema, schema,
+ assertNull(create(schema, schema,
storageWithReadingShredded(tmp, false)));
}
@Test
void returnsNullForNonRecordSchemas(@TempDir Path tmp) {
HoodieStorage storage = storageWithReadingShredded(tmp, true);
- assertNull(HoodieVariantReconstruction.create(
+ assertNull(create(
HoodieSchema.create(HoodieSchemaType.STRING),
recordWithVariant(HoodieSchema.createVariant()), storage));
- assertNull(HoodieVariantReconstruction.create(
+ assertNull(create(
recordWithVariant(HoodieSchema.createVariant()),
HoodieSchema.create(HoodieSchemaType.STRING), storage));
}
@@ -101,7 +123,7 @@ class TestHoodieVariantReconstruction {
HoodieSchema requestedSchema =
recordWithVariant(HoodieSchema.createVariant());
HoodieException exception = assertThrows(HoodieException.class, () ->
- HoodieVariantReconstruction.create(fileSchema, requestedSchema,
+ create(fileSchema, requestedSchema,
storageWithReadingShredded(tmp, true)));
assertTrue(exception.getMessage().contains("no VariantShreddingProvider is
available"));
}
@@ -115,7 +137,7 @@ class TestHoodieVariantReconstruction {
storage.getConf().set(HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_PROVIDER_CLASS.key(),
TestVariantShreddingProvider.class.getName());
- HoodieVariantReconstruction reconstruction =
HoodieVariantReconstruction.create(
+ HoodieVariantReconstruction reconstruction = create(
fileSchema, requestedSchema, storage);
assertNotNull(reconstruction);
assertTrue(reconstruction.intermediateSchema().getField("v").get().schema().getNonNullType()
@@ -154,7 +176,7 @@ class TestHoodieVariantReconstruction {
storage.getConf().set(HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_PROVIDER_CLASS.key(),
TestVariantShreddingProvider.class.getName());
- HoodieVariantReconstruction reconstruction =
HoodieVariantReconstruction.create(
+ HoodieVariantReconstruction reconstruction = create(
fileSchema, requestedSchema, storage);
assertNotNull(reconstruction, "Plain-record shredded shape with a variant
requested column must engage");
@@ -182,7 +204,7 @@ class TestHoodieVariantReconstruction {
// shredded reading disabled.
HoodieSchema fileSchema =
recordWithIdAndVariant(footerStylePlainUnshreddedSchema());
HoodieSchema requestedSchema =
recordWithIdAndVariant(HoodieSchema.createVariant());
- assertNull(HoodieVariantReconstruction.create(fileSchema, requestedSchema,
+ assertNull(create(fileSchema, requestedSchema,
storageWithReadingShredded(tmp, false)));
}
@@ -203,7 +225,7 @@ class TestHoodieVariantReconstruction {
storage.getConf().set(HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_PROVIDER_CLASS.key(),
TestVariantShreddingProvider.class.getName());
- HoodieVariantReconstruction reconstruction =
HoodieVariantReconstruction.create(
+ HoodieVariantReconstruction reconstruction = create(
fileSchema, requestedSchema, storage);
assertNotNull(reconstruction, "A shredded group with no value column must
still engage");
@@ -229,7 +251,7 @@ class TestHoodieVariantReconstruction {
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(
+ assertNull(create(
recordWithIdAndVariant(fourField),
recordWithIdAndVariant(HoodieSchema.createVariant()),
storageWithReadingShredded(tmp, false)));
}
@@ -243,7 +265,7 @@ class TestHoodieVariantReconstruction {
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(
+ assertNull(create(
recordWithIdAndVariant(threeField),
recordWithIdAndVariant(HoodieSchema.createVariant()),
storageWithReadingShredded(tmp, false)));
}
@@ -255,12 +277,13 @@ class TestHoodieVariantReconstruction {
// nothing to reconstruct, so create() returns null even with shredded
reading disabled.
HoodieSchema fileSchema =
recordWithIdAndVariant(footerStylePlainShreddedSchema());
HoodieSchema requestedSchema =
recordWithIdAndVariant(footerStylePlainShreddedSchema());
- assertNull(HoodieVariantReconstruction.create(fileSchema, requestedSchema,
+ assertNull(create(fileSchema, requestedSchema,
storageWithReadingShredded(tmp, false)));
}
- @Test
- void reconstructsShreddedVariantsNestedInRecordsArraysAndMaps(@TempDir Path
tmp) {
+ @ParameterizedTest(name = "writeOldListStructure={0}")
+ @ValueSource(booleans = {true, false})
+ void reconstructsShreddedVariantsNestedInRecordsArraysAndMaps(boolean
writeOldListStructure, @TempDir Path tmp) {
// Both write paths shred at any depth their write schema asks them to
// (HoodieRowParquetWriteSupport.processNestedDataType,
VariantSchemaUtils.applyForcedShredding),
// so a nested variant can reach this reader as a plain {metadata, value,
typed_value} record too.
@@ -281,11 +304,32 @@ class TestHoodieVariantReconstruction {
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);
+ // The file's lists are either the legacy 2-level `repeated group array`
or the 3-level
+ // `repeated group list { element }`; HoodieAvroParquetReader converts the
footer of both with
+ // ADD_LIST_ELEMENT_RECORDS=false (tryOverrideDefaultConfigs).
+ storage.getConf().set(AvroWriteSupport.WRITE_OLD_LIST_STRUCTURE,
Boolean.toString(writeOldListStructure));
+ storage.getConf().set(AvroSchemaConverter.ADD_LIST_ELEMENT_RECORDS,
"false");
+ MessageType fileParquetSchema =
+
getAvroSchemaConverter(storage.getConf().unwrapAs(Configuration.class)).convert(fileSchema);
+ GroupType itemsRepeated =
fileParquetSchema.getType("items").asGroupType().getType(0).asGroupType();
+ assertEquals(writeOldListStructure ? "array" : "list",
itemsRepeated.getName());
+ TestVariantShreddingProvider.PREPARED.clear();
+
+ HoodieVariantReconstruction reconstruction =
+ HoodieVariantReconstruction.create(fileSchema, fileParquetSchema,
requestedSchema, storage);
assertNotNull(reconstruction, "A variant nested below the top level must
engage reconstruction");
+ // Each variant column is prepared once, from its own parquet group: the
struct member, the list
+ // element (the repeated group itself in the 2-level layout, its only
field in the 3-level one) and
+ // the map value.
+ Map<String, GroupType> expectedGroups = new LinkedHashMap<>();
+ expectedGroups.put("nested.v",
fileParquetSchema.getType("nested").asGroupType().getType("v").asGroupType());
+ expectedGroups.put("items.element",
+ writeOldListStructure ? itemsRepeated :
itemsRepeated.getType(0).asGroupType());
+ expectedGroups.put("tags.value",
+
fileParquetSchema.getType("tags").asGroupType().getType(0).asGroupType().getType(1).asGroupType());
+ assertEquals(expectedGroups, TestVariantShreddingProvider.PREPARED);
+
// The file's shredded shape must reach every nested position of the read
schema, otherwise
// parquet never materializes typed_value there.
Schema intermediateAvro =
reconstruction.intermediateSchema().toAvroSchema();
@@ -342,7 +386,7 @@ class TestHoodieVariantReconstruction {
HoodieSchema.createVariant("nested_v", "org.apache.hudi.test", null),
HoodieSchema.createVariant("element_v", "org.apache.hudi.test", null),
HoodieSchema.createVariant("map_v", "org.apache.hudi.test", null));
- assertNull(HoodieVariantReconstruction.create(fileSchema, requestedSchema,
+ assertNull(create(fileSchema, requestedSchema,
storageWithReadingShredded(tmp, false)));
}
@@ -404,6 +448,9 @@ class TestHoodieVariantReconstruction {
}
public static class TestVariantShreddingProvider implements
VariantShreddingProvider {
+ // The provider is instantiated by class name, so what it is prepared with
is recorded statically.
+ static final Map<String, GroupType> PREPARED = new LinkedHashMap<>();
+
@Override
public GenericRecord shredVariantRecord(
GenericRecord unshreddedVariant, Schema shreddedSchema,
HoodieSchema.Variant variantSchema) {
@@ -412,13 +459,15 @@ class TestHoodieVariantReconstruction {
}
@Override
- public GenericRecord rebuildVariantRecord(
- GenericRecord shreddedVariant, Schema shreddedSchema, Schema
unshreddedSchema) {
- GenericRecord rebuilt = new GenericData.Record(unshreddedSchema);
- rebuilt.put("metadata", shreddedVariant.get("metadata"));
- rebuilt.put("value", ByteBuffer.wrap(new byte[] {
- ((Number) shreddedVariant.get("typed_value")).byteValue()}));
- return rebuilt;
+ public VariantRebuilder prepareRebuild(String columnPath, GroupType
parquetVariantType, Schema unshreddedSchema) {
+ PREPARED.put(columnPath, parquetVariantType);
+ return shreddedVariant -> {
+ GenericRecord rebuilt = new GenericData.Record(unshreddedSchema);
+ rebuilt.put("metadata", shreddedVariant.get("metadata"));
+ rebuilt.put("value", ByteBuffer.wrap(new byte[] {
+ ((Number) shreddedVariant.get("typed_value")).byteValue()}));
+ return rebuilt;
+ };
}
}
}
diff --git
a/hudi-spark-datasource/hudi-spark4-common/src/main/java/org/apache/hudi/variant/Spark4VariantShreddingProvider.java
b/hudi-spark-datasource/hudi-spark4-common/src/main/java/org/apache/hudi/variant/Spark4VariantShreddingProvider.java
index 63a6e9d17588..9df38cd39e44 100644
---
a/hudi-spark-datasource/hudi-spark4-common/src/main/java/org/apache/hudi/variant/Spark4VariantShreddingProvider.java
+++
b/hudi-spark-datasource/hudi-spark4-common/src/main/java/org/apache/hudi/variant/Spark4VariantShreddingProvider.java
@@ -23,13 +23,16 @@ import org.apache.hudi.common.avro.VariantShreddingProvider;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.exception.HoodieException;
-import org.apache.avro.Conversions;
import org.apache.avro.LogicalType;
import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericFixed;
import org.apache.avro.generic.GenericRecord;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Type;
import org.apache.spark.types.variant.ShreddingUtils;
import org.apache.spark.types.variant.Variant;
import org.apache.spark.types.variant.VariantSchema;
@@ -38,11 +41,15 @@ import
org.apache.spark.types.variant.VariantShreddingWriter.ShreddedResult;
import
org.apache.spark.types.variant.VariantShreddingWriter.ShreddedResultBuilder;
import java.math.BigDecimal;
+import java.math.BigInteger;
import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.UUID;
import static org.apache.hudi.common.avro.HoodieAvroUtils.unwrapNullable;
@@ -63,6 +70,11 @@ import static
org.apache.hudi.common.avro.HoodieAvroUtils.unwrapNullable;
*/
public class Spark4VariantShreddingProvider implements
VariantShreddingProvider {
+ // Spark's Decimal.MAX_INT_DIGITS, Decimal.MAX_LONG_DIGITS and
DecimalType.MAX_PRECISION.
+ private static final int MAX_INT_DIGITS = 9;
+ private static final int MAX_LONG_DIGITS = 18;
+ private static final int MAX_DECIMAL_PRECISION = 38;
+
@Override
public GenericRecord shredVariantRecord(
GenericRecord unshreddedVariant,
@@ -94,27 +106,27 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
}
@Override
- public GenericRecord rebuildVariantRecord(
- GenericRecord shreddedVariant,
- Schema shreddedSchema,
- Schema unshreddedSchema) {
+ public VariantRebuilder prepareRebuild(String columnPath, GroupType
parquetVariantType, Schema unshreddedSchema) {
+ Map<VariantSchema, LeafDecoding> decodings = new IdentityHashMap<>();
+ VariantSchema sparkSchema = buildReadVariantSchema(parquetVariantType,
true, columnPath, columnPath, decodings);
+ return shreddedVariant -> rebuildVariantRecord(shreddedVariant,
sparkSchema, decodings, unshreddedSchema);
+ }
+ private static GenericRecord rebuildVariantRecord(GenericRecord
shreddedVariant, VariantSchema sparkSchema,
+ Map<VariantSchema,
LeafDecoding> decodings,
+ Schema unshreddedSchema) {
if (shreddedVariant == null) {
return null;
}
- ByteBuffer metadataBuf = (ByteBuffer)
shreddedVariant.get(HoodieSchema.Variant.VARIANT_METADATA_FIELD);
- if (metadataBuf == null) {
+ if (shreddedVariant.get(HoodieSchema.Variant.VARIANT_METADATA_FIELD) ==
null) {
// metadata is REQUIRED in the shredded schema, so a null here means the
base file is malformed;
// fail fast rather than silently nulling out the variant column.
throw new HoodieException("Cannot reconstruct variant: shredded record
is missing the required '"
+ HoodieSchema.Variant.VARIANT_METADATA_FIELD + "' field; the base
file is malformed.");
}
- // Reuse the same VariantSchema index assignment as the write path (no
builder needed on read).
- VariantSchema sparkSchema = buildVariantSchema(shreddedSchema, true, null);
-
// Delegate to Spark's reconstruction algorithm (inverse of castShredded).
- Variant variant = ShreddingUtils.rebuild(new
AvroVariantRow(shreddedVariant, sparkSchema), sparkSchema);
+ Variant variant = ShreddingUtils.rebuild(new
AvroVariantRow(shreddedVariant, sparkSchema, decodings), sparkSchema);
GenericRecord out = new GenericData.Record(unshreddedSchema);
out.put(HoodieSchema.Variant.VARIANT_METADATA_FIELD,
ByteBuffer.wrap(variant.getMetadata()));
@@ -123,7 +135,7 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
}
/**
- * Builds a {@link VariantSchema} from an Avro {@link Schema} representing a
+ * Builds the write-side {@link VariantSchema} from an Avro {@link Schema}
representing a
* shredded variant structure ({@code value}, {@code metadata}, {@code
typed_value}).
*
* <p>This method also registers the Avro schema mapping in the builder so
that
@@ -176,9 +188,9 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
break;
default:
- // Scalar shredding. On write (non-null builder), reject leaf types
Spark 4 cannot shred,
- // as SparkShreddingUtils.buildVariantSchema does with
invalidVariantShreddingSchema.
- if (builder != null && !isSparkShreddableScalar(tvSchema)) {
+ // Scalar shredding. Reject leaf types Spark 4 cannot shred, as
+ // SparkShreddingUtils.buildVariantSchema does with
invalidVariantShreddingSchema.
+ if (!isSparkShreddableScalar(tvSchema)) {
LogicalType logicalType = tvSchema.getLogicalType();
String typeName = tvSchema.getType().getName()
+ (logicalType == null ? "" : " (" + logicalType.getName() +
")");
@@ -193,13 +205,7 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
VariantSchema result = new VariantSchema(
typedIdx, variantIdx, topLevelMetadataIdx, numFields,
scalarSchema, objectSchema, arraySchema);
-
- // The read (rebuild) path passes a null builder: it needs the
VariantSchema indices but no
- // Avro-schema registration (registration only feeds write-side result
construction).
- if (builder != null) {
- builder.registerSchema(result, avroSchema);
- }
-
+ builder.registerSchema(result, avroSchema);
return result;
}
@@ -239,8 +245,8 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
}
/**
- * Maps an Avro {@link Schema} type (potentially with logical type
annotations)
- * to a {@link VariantSchema.ScalarType}.
+ * Maps a scalar {@code typed_value} Avro {@link Schema} that passed
+ * {@link #isSparkShreddableScalar} to a {@link VariantSchema.ScalarType}.
*/
private VariantSchema.ScalarType avroTypeToScalarType(Schema schema) {
LogicalType logicalType = schema.getLogicalType();
@@ -261,16 +267,6 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
if ("local-timestamp-micros".equals(name)) {
return new VariantSchema.TimestampNTZType();
}
- // The Variant binary spec stores timestamps in microseconds, so a
millisecond-precision
- // typed_value cannot represent a variant timestamp. Map it to no scalar
type rather than the
- // micros TimestampType, which would silently scale the value by 1000x.
(The write path
- // rejects such a leaf before reaching here.)
- if ("timestamp-millis".equals(name) ||
"local-timestamp-millis".equals(name)) {
- return null;
- }
- if ("uuid".equals(name)) {
- return new VariantSchema.UuidType();
- }
}
switch (schema.getType()) {
@@ -288,13 +284,346 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
return new VariantSchema.StringType();
case BYTES:
return new VariantSchema.BinaryType();
- case FIXED:
- return new VariantSchema.BinaryType();
default:
- return null;
+ throw new IllegalStateException("Unexpected shreddable typed_value
type: " + schema);
+ }
+ }
+
+ /**
+ * How the Avro value parquet-avro reads for a scalar {@code typed_value}
leaf turns into the value
+ * Spark's {@link ShreddingUtils#rebuild} expects. The file schema's Avro
conversion drops these
+ * encodings, so they are taken from the leaf's parquet type, as
ParquetRowConverter does.
+ */
+ enum LeafDecoding {
+ /** Boolean, Integer, Long, Float, Double, CharSequence, or bytes
(ByteBuffer / GenericFixed), as is. */
+ DIRECT,
+ /** INT(32, false): the Integer is unsigned, widened to long. */
+ UNSIGNED_INT32,
+ /** INT(64, false): the Long is unsigned, read as DECIMAL(20, 0). */
+ UNSIGNED_INT64,
+ /** TIMESTAMP(MILLIS): the Long is in milliseconds, scaled to micros. */
+ TIMESTAMP_MILLIS,
+ /** DECIMAL on INT32 / INT64: the Integer / Long is the unscaled value. */
+ DECIMAL_UNSCALED,
+ /** DECIMAL on BINARY / FIXED_LEN_BYTE_ARRAY: big-endian two's-complement
unscaled bytes. */
+ DECIMAL_BYTES,
+ /** BINARY (JSON): UTF-8 bytes, read as a string. */
+ JSON
+ }
+
+ /** A scalar {@code typed_value} leaf: its Spark scalar type and how its
Avro value is decoded. */
+ private static final class ScalarLeaf {
+ private final VariantSchema.ScalarType type;
+ private final LeafDecoding decoding;
+
+ private ScalarLeaf(VariantSchema.ScalarType type, LeafDecoding decoding) {
+ this.type = type;
+ this.decoding = decoding;
+ }
+ }
+
+ /**
+ * Builds the read-side {@link VariantSchema} of a shredded variant group
from its parquet type, the
+ * way Spark 4 does when it opens a file:
SparkShreddingUtils.buildVariantSchema over the type that
+ * ParquetToSparkSchemaConverter derives with ignoreVariantAnnotation =
true. Ordinals are the parquet
+ * field positions, as in Spark. Records each scalar leaf's {@link
LeafDecoding} in {@code decodings},
+ * and throws for layouts and leaf encodings Spark 4 rejects.
+ */
+ private static VariantSchema buildReadVariantSchema(GroupType group, boolean
topLevel, String columnPath,
+ String groupPath,
Map<VariantSchema, LeafDecoding> decodings) {
+ if (group.getFieldCount() == 0) {
+ throw invalidShreddingLayout(columnPath, groupPath, group);
+ }
+ int typedIdx = -1;
+ int variantIdx = -1;
+ int topLevelMetadataIdx = -1;
+ ScalarLeaf scalarLeaf = null;
+ VariantSchema.ObjectField[] objectSchema = null;
+ VariantSchema arraySchema = null;
+
+ for (int i = 0; i < group.getFieldCount(); i++) {
+ Type field = group.getType(i);
+ String fieldPath = groupPath + "." + field.getName();
+ switch (field.getName()) {
+ case HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD:
+ if (typedIdx != -1) {
+ throw invalidShreddingLayout(columnPath, groupPath, group);
+ }
+ typedIdx = i;
+ if (field.isPrimitive()) {
+ scalarLeaf = toScalarLeaf(field.asPrimitiveType(), columnPath,
fieldPath);
+ } else if (field.getLogicalTypeAnnotation() instanceof
LogicalTypeAnnotation.ListLogicalTypeAnnotation) {
+ Type element = listElementType(field.asGroupType());
+ if (!isStructGroup(element)) {
+ throw invalidShreddingLayout(columnPath, fieldPath, field);
+ }
+ arraySchema = buildReadVariantSchema(
+ element.asGroupType(), false, columnPath, fieldPath + "." +
element.getName(), decodings);
+ } else if (isStructGroup(field)) {
+ objectSchema = buildReadObjectSchema(field.asGroupType(),
columnPath, fieldPath, decodings);
+ } else {
+ throw invalidShreddingLayout(columnPath, fieldPath, field);
+ }
+ break;
+ case HoodieSchema.Variant.VARIANT_VALUE_FIELD:
+ if (variantIdx != -1 || !isSparkBinary(field)) {
+ throw invalidShreddingLayout(columnPath, groupPath, group);
+ }
+ variantIdx = i;
+ break;
+ case HoodieSchema.Variant.VARIANT_METADATA_FIELD:
+ if (topLevelMetadataIdx != -1 || !isSparkBinary(field)) {
+ throw invalidShreddingLayout(columnPath, groupPath, group);
+ }
+ topLevelMetadataIdx = i;
+ break;
+ default:
+ throw invalidShreddingLayout(columnPath, groupPath, group);
+ }
+ }
+ if (topLevel != (topLevelMetadataIdx >= 0)) {
+ throw invalidShreddingLayout(columnPath, groupPath, group);
+ }
+
+ VariantSchema result = new VariantSchema(typedIdx, variantIdx,
topLevelMetadataIdx, group.getFieldCount(),
+ scalarLeaf == null ? null : scalarLeaf.type, objectSchema,
arraySchema);
+ if (scalarLeaf != null) {
+ decodings.put(result, scalarLeaf.decoding);
+ }
+ return result;
+ }
+
+ /** An object {@code typed_value}: a non-empty group of uniquely named
shredded field groups. */
+ private static VariantSchema.ObjectField[] buildReadObjectSchema(GroupType
typedValue, String columnPath,
+ String
typedValuePath,
+
Map<VariantSchema, LeafDecoding> decodings) {
+ if (typedValue.getFieldCount() == 0) {
+ throw invalidShreddingLayout(columnPath, typedValuePath, typedValue);
+ }
+ VariantSchema.ObjectField[] fields = new
VariantSchema.ObjectField[typedValue.getFieldCount()];
+ Set<String> names = new HashSet<>();
+ for (int i = 0; i < fields.length; i++) {
+ Type field = typedValue.getType(i);
+ if (!names.add(field.getName()) || !isStructGroup(field)) {
+ throw invalidShreddingLayout(columnPath, typedValuePath, typedValue);
+ }
+ fields[i] = new VariantSchema.ObjectField(field.getName(),
buildReadVariantSchema(
+ field.asGroupType(), false, columnPath, typedValuePath + "." +
field.getName(), decodings));
+ }
+ return fields;
+ }
+
+ /**
+ * Maps a scalar {@code typed_value} leaf's parquet type to the Spark scalar
type, mirroring
+ * ParquetToSparkSchemaConverter.convertPrimitiveField at its defaults
(binaryAsString = false,
+ * inferTimestampNTZ = true, nanosAsLong = false) followed by
SparkShreddingUtils.buildVariantSchema,
+ * and to the matching {@link LeafDecoding}. INT96 is rejected although
Spark reads it as a timestamp:
+ * Hudi's Avro reader does not support it.
+ */
+ private static ScalarLeaf toScalarLeaf(PrimitiveType leaf, String
columnPath, String leafPath) {
+ // Leaf annotations are matched against a whitelist, so the UNKNOWN
annotation (Spark's NullType,
+ // not a shreddable scalar) is rejected without naming its class, which
parquet < 1.16 lacks.
+ LogicalTypeAnnotation annotation = leaf.getLogicalTypeAnnotation();
+ switch (leaf.getPrimitiveTypeName()) {
+ case BOOLEAN:
+ if (annotation != null) {
+ throw unsupportedLeaf(columnPath, leafPath, leaf);
+ }
+ return new ScalarLeaf(new VariantSchema.BooleanType(),
LeafDecoding.DIRECT);
+ case FLOAT:
+ if (annotation != null) {
+ throw unsupportedLeaf(columnPath, leafPath, leaf);
+ }
+ return new ScalarLeaf(new VariantSchema.FloatType(),
LeafDecoding.DIRECT);
+ case DOUBLE:
+ if (annotation != null) {
+ throw unsupportedLeaf(columnPath, leafPath, leaf);
+ }
+ return new ScalarLeaf(new VariantSchema.DoubleType(),
LeafDecoding.DIRECT);
+ case INT32:
+ return toInt32Leaf(leaf, annotation, columnPath, leafPath);
+ case INT64:
+ return toInt64Leaf(leaf, annotation, columnPath, leafPath);
+ case BINARY:
+ if (annotation == null || annotation instanceof
LogicalTypeAnnotation.BsonLogicalTypeAnnotation) {
+ return new ScalarLeaf(new VariantSchema.BinaryType(),
LeafDecoding.DIRECT);
+ }
+ if (annotation instanceof
LogicalTypeAnnotation.StringLogicalTypeAnnotation
+ || annotation instanceof
LogicalTypeAnnotation.EnumLogicalTypeAnnotation) {
+ return new ScalarLeaf(new VariantSchema.StringType(),
LeafDecoding.DIRECT);
+ }
+ if (annotation instanceof
LogicalTypeAnnotation.JsonLogicalTypeAnnotation) {
+ return new ScalarLeaf(new VariantSchema.StringType(),
LeafDecoding.JSON);
+ }
+ if (annotation instanceof
LogicalTypeAnnotation.DecimalLogicalTypeAnnotation) {
+ return toDecimalLeaf(leaf, MAX_DECIMAL_PRECISION,
LeafDecoding.DECIMAL_BYTES, columnPath, leafPath);
+ }
+ throw unsupportedLeaf(columnPath, leafPath, leaf);
+ case FIXED_LEN_BYTE_ARRAY:
+ if (annotation == null) {
+ return new ScalarLeaf(new VariantSchema.BinaryType(),
LeafDecoding.DIRECT);
+ }
+ if (annotation instanceof
LogicalTypeAnnotation.DecimalLogicalTypeAnnotation) {
+ return toDecimalLeaf(leaf,
maxPrecisionForBytes(leaf.getTypeLength()), LeafDecoding.DECIMAL_BYTES,
+ columnPath, leafPath);
+ }
+ // UUID, FLOAT16, INTERVAL: Spark rejects them.
+ throw unsupportedLeaf(columnPath, leafPath, leaf);
+ case INT96:
+ throw new HoodieException("Cannot read shredded variant column '" +
columnPath + "': typed_value '"
+ + leafPath + "' has parquet type '" + leaf + "'. INT96 typed_value
is not supported by Hudi's Avro reader.");
+ default:
+ throw unsupportedLeaf(columnPath, leafPath, leaf);
+ }
+ }
+
+ private static ScalarLeaf toInt32Leaf(PrimitiveType leaf,
LogicalTypeAnnotation annotation,
+ String columnPath, String leafPath) {
+ if (annotation == null) {
+ return new ScalarLeaf(new
VariantSchema.IntegralType(VariantSchema.IntegralSize.INT),
LeafDecoding.DIRECT);
+ }
+ if (annotation instanceof LogicalTypeAnnotation.IntLogicalTypeAnnotation) {
+ LogicalTypeAnnotation.IntLogicalTypeAnnotation intType =
(LogicalTypeAnnotation.IntLogicalTypeAnnotation) annotation;
+ // Unsigned ints widen to the next signed size, as in Spark.
+ switch (intType.getBitWidth()) {
+ case 8:
+ return new ScalarLeaf(new
VariantSchema.IntegralType(intType.isSigned()
+ ? VariantSchema.IntegralSize.BYTE :
VariantSchema.IntegralSize.SHORT), LeafDecoding.DIRECT);
+ case 16:
+ return new ScalarLeaf(new
VariantSchema.IntegralType(intType.isSigned()
+ ? VariantSchema.IntegralSize.SHORT :
VariantSchema.IntegralSize.INT), LeafDecoding.DIRECT);
+ case 32:
+ return intType.isSigned()
+ ? new ScalarLeaf(new
VariantSchema.IntegralType(VariantSchema.IntegralSize.INT), LeafDecoding.DIRECT)
+ : new ScalarLeaf(new
VariantSchema.IntegralType(VariantSchema.IntegralSize.LONG),
LeafDecoding.UNSIGNED_INT32);
+ default:
+ throw unsupportedLeaf(columnPath, leafPath, leaf);
+ }
+ }
+ if (annotation instanceof LogicalTypeAnnotation.DateLogicalTypeAnnotation)
{
+ return new ScalarLeaf(new VariantSchema.DateType(), LeafDecoding.DIRECT);
+ }
+ if (annotation instanceof
LogicalTypeAnnotation.DecimalLogicalTypeAnnotation) {
+ return toDecimalLeaf(leaf, MAX_INT_DIGITS,
LeafDecoding.DECIMAL_UNSCALED, columnPath, leafPath);
+ }
+ throw unsupportedLeaf(columnPath, leafPath, leaf);
+ }
+
+ private static ScalarLeaf toInt64Leaf(PrimitiveType leaf,
LogicalTypeAnnotation annotation,
+ String columnPath, String leafPath) {
+ if (annotation == null) {
+ return new ScalarLeaf(new
VariantSchema.IntegralType(VariantSchema.IntegralSize.LONG),
LeafDecoding.DIRECT);
+ }
+ if (annotation instanceof LogicalTypeAnnotation.IntLogicalTypeAnnotation) {
+ LogicalTypeAnnotation.IntLogicalTypeAnnotation intType =
(LogicalTypeAnnotation.IntLogicalTypeAnnotation) annotation;
+ if (intType.getBitWidth() != 64) {
+ throw unsupportedLeaf(columnPath, leafPath, leaf);
+ }
+ // DECIMAL(20, 0) holds the largest unsigned long, as in Spark.
+ return intType.isSigned()
+ ? new ScalarLeaf(new
VariantSchema.IntegralType(VariantSchema.IntegralSize.LONG),
LeafDecoding.DIRECT)
+ : new ScalarLeaf(new VariantSchema.DecimalType(20, 0),
LeafDecoding.UNSIGNED_INT64);
+ }
+ if (annotation instanceof
LogicalTypeAnnotation.DecimalLogicalTypeAnnotation) {
+ return toDecimalLeaf(leaf, MAX_LONG_DIGITS,
LeafDecoding.DECIMAL_UNSCALED, columnPath, leafPath);
+ }
+ if (annotation instanceof
LogicalTypeAnnotation.TimestampLogicalTypeAnnotation) {
+ LogicalTypeAnnotation.TimestampLogicalTypeAnnotation timestamp =
+ (LogicalTypeAnnotation.TimestampLogicalTypeAnnotation) annotation;
+ LeafDecoding decoding;
+ switch (timestamp.getUnit()) {
+ case MICROS:
+ decoding = LeafDecoding.DIRECT;
+ break;
+ case MILLIS:
+ decoding = LeafDecoding.TIMESTAMP_MILLIS;
+ break;
+ default:
+ // NANOS: Spark reads it only as a plain long under its legacy
nanosAsLong flag.
+ throw unsupportedLeaf(columnPath, leafPath, leaf);
+ }
+ return new ScalarLeaf(timestamp.isAdjustedToUTC()
+ ? new VariantSchema.TimestampType() : new
VariantSchema.TimestampNTZType(), decoding);
+ }
+ // TIME included: Spark reads TIME(MICROS, local) as TimeType, which is
not a shreddable scalar.
+ throw unsupportedLeaf(columnPath, leafPath, leaf);
+ }
+
+ /** A decimal leaf whose precision must fit its physical type, as
ParquetToSparkSchemaConverter checks. */
+ private static ScalarLeaf toDecimalLeaf(PrimitiveType leaf, int
maxPrecision, LeafDecoding decoding,
+ String columnPath, String leafPath) {
+ LogicalTypeAnnotation.DecimalLogicalTypeAnnotation decimal =
+ (LogicalTypeAnnotation.DecimalLogicalTypeAnnotation)
leaf.getLogicalTypeAnnotation();
+ int precision = decimal.getPrecision();
+ int scale = decimal.getScale();
+ if (precision < 1 || precision > maxPrecision || precision >
MAX_DECIMAL_PRECISION
+ || scale < 0 || scale > precision) {
+ throw unsupportedLeaf(columnPath, leafPath, leaf);
+ }
+ return new ScalarLeaf(new VariantSchema.DecimalType(precision, scale),
decoding);
+ }
+
+ /** Spark's Decimal.maxPrecisionForBytes: the digits a signed {@code
numBytes}-byte integer always holds. */
+ private static int maxPrecisionForBytes(int numBytes) {
+ return (int) Math.round(Math.floor(Math.log10(Math.pow(2, 8 * numBytes -
1) - 1)));
+ }
+
+ /**
+ * The element type of a parquet LIST group under Spark's list rules
+ * (ParquetToSparkSchemaConverter.isElementType), or null for a malformed
list. Applies only to the
+ * typed_value list inside a variant; HoodieVariantReconstruction's planner
walks the lists around
+ * a variant column under parquet-avro's rules instead, so the two are not
meant to agree.
+ */
+ private static Type listElementType(GroupType list) {
+ if (list.getFieldCount() != 1 ||
!list.getType(0).isRepetition(Type.Repetition.REPEATED)) {
+ return null;
+ }
+ Type repeated = list.getType(0);
+ boolean repeatedIsElement = repeated.isPrimitive()
+ || repeated.asGroupType().getFieldCount() > 1
+ || repeated.getName().equals("array")
+ || repeated.getName().equals(list.getName() + "_tuple");
+ return repeatedIsElement ? repeated : repeated.asGroupType().getType(0);
+ }
+
+ /** Whether Spark reads {@code type} as a struct: a group that is neither a
LIST nor a MAP. */
+ private static boolean isStructGroup(Type type) {
+ if (type == null || type.isPrimitive()) {
+ return false;
+ }
+ LogicalTypeAnnotation annotation = type.getLogicalTypeAnnotation();
+ return !(annotation instanceof
LogicalTypeAnnotation.ListLogicalTypeAnnotation
+ || annotation instanceof LogicalTypeAnnotation.MapLogicalTypeAnnotation
+ || annotation instanceof
LogicalTypeAnnotation.MapKeyValueTypeAnnotation);
+ }
+
+ /** Whether Spark reads {@code type} as BinaryType, which the value and
metadata fields must be. */
+ private static boolean isSparkBinary(Type type) {
+ if (!type.isPrimitive()) {
+ return false;
+ }
+ LogicalTypeAnnotation annotation = type.getLogicalTypeAnnotation();
+ switch (type.asPrimitiveType().getPrimitiveTypeName()) {
+ case BINARY:
+ return annotation == null || annotation instanceof
LogicalTypeAnnotation.BsonLogicalTypeAnnotation;
+ case FIXED_LEN_BYTE_ARRAY:
+ return annotation == null;
+ default:
+ return false;
}
}
+ private static HoodieException unsupportedLeaf(String columnPath, String
leafPath, PrimitiveType leaf) {
+ return new HoodieException("Cannot read shredded variant column '" +
columnPath + "': typed_value '"
+ + leafPath + "' has parquet type '" + leaf + "', which Spark 4 does
not support reading as a shredded "
+ + "variant either.");
+ }
+
+ private static HoodieException invalidShreddingLayout(String columnPath,
String path, Type type) {
+ return new HoodieException("Cannot read shredded variant column '" +
columnPath + "': '" + path
+ + "' is not a valid shredded variant layout (" + type + "); Spark 4
rejects it too.");
+ }
+
private static byte[] toByteArray(ByteBuffer buffer) {
if (buffer.hasArray() && buffer.position() == 0
&& buffer.arrayOffset() == 0
@@ -384,8 +713,7 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
}
if (scalarValue != null) {
- Schema tvSchema = unwrapNullable(tvField.schema());
- record.put(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD,
convertScalarToAvro(scalarValue, tvSchema));
+ record.put(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD,
convertScalarToAvro(scalarValue));
} else if (objectFields != null) {
Schema tvSchema = unwrapNullable(tvField.schema());
GenericRecord tvRecord = new GenericData.Record(tvSchema);
@@ -410,38 +738,11 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
/**
* Converts a scalar value from Spark's variant representation to an
Avro-compatible type.
- * Handles type widening (Byte/Short to Int/Long) and binary wrapping.
+ * avroTypeToScalarType only emits scalar types whose Spark values
(Boolean, Integer, Long, Float,
+ * Double, String, BigDecimal) Avro takes as is, except binary, which Avro
wants as a ByteBuffer.
*/
- private static Object convertScalarToAvro(Object value, Schema avroSchema)
{
- if (value instanceof byte[]) {
- return ByteBuffer.wrap((byte[]) value);
- }
- if (value instanceof UUID) {
- return value.toString();
- }
- // Widen integer types to match Avro schema expectations
- if (avroSchema.getType() == Schema.Type.INT) {
- if (value instanceof Byte) {
- return ((Byte) value).intValue();
- }
- if (value instanceof Short) {
- return ((Short) value).intValue();
- }
- }
- if (avroSchema.getType() == Schema.Type.LONG) {
- if (value instanceof Byte) {
- return ((Byte) value).longValue();
- }
- if (value instanceof Short) {
- return ((Short) value).longValue();
- }
- if (value instanceof Integer) {
- return ((Integer) value).longValue();
- }
- }
- // BigDecimal, Boolean, String, Integer, Long, Float, Double
- // are directly compatible with Avro's type system
- return value;
+ private static Object convertScalarToAvro(Object value) {
+ return value instanceof byte[] ? ByteBuffer.wrap((byte[]) value) : value;
}
}
@@ -562,15 +863,18 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
/**
* A shredded variant struct {@code {value, [metadata], typed_value}}. Maps
the Spark
* {@link VariantSchema} ordinals (variantIdx / topLevelMetadataIdx /
typedIdx) back to the named
- * Avro fields, and reads {@code typed_value} for scalar/object/array
reconstruction.
+ * Avro fields, and reads {@code typed_value} for scalar/object/array
reconstruction, decoding a
+ * scalar per its {@link LeafDecoding}.
*/
static final class AvroVariantRow extends BaseAvroShreddedRow {
private final GenericRecord record;
private final VariantSchema schema;
+ private final Map<VariantSchema, LeafDecoding> decodings;
- AvroVariantRow(GenericRecord record, VariantSchema schema) {
+ AvroVariantRow(GenericRecord record, VariantSchema schema,
Map<VariantSchema, LeafDecoding> decodings) {
this.record = record;
this.schema = schema;
+ this.decodings = decodings;
}
private String fieldNameFor(int ordinal) {
@@ -591,7 +895,7 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
}
@Override public byte[] getBinary(int ordinal) {
- return toByteArray((ByteBuffer) record.get(fieldNameFor(ordinal)));
+ return bytesOf(record.get(fieldNameFor(ordinal)));
}
// The scalar/struct/array getters below ignore `ordinal` and read
typed_value directly via
@@ -602,6 +906,10 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
return record.get(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD);
}
+ private LeafDecoding decoding() {
+ return decodings.get(schema);
+ }
+
@Override public boolean getBoolean(int ordinal) {
return (Boolean) typedValue();
}
@@ -619,7 +927,16 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
}
@Override public long getLong(int ordinal) {
- return ((Number) typedValue()).longValue();
+ Object value = typedValue();
+ switch (decoding()) {
+ case UNSIGNED_INT32:
+ return Integer.toUnsignedLong((Integer) value);
+ case TIMESTAMP_MILLIS:
+ // Spark's DateTimeUtils.millisToMicros.
+ return Math.multiplyExact((Long) value, 1000L);
+ default:
+ return ((Number) value).longValue();
+ }
}
@Override public float getFloat(int ordinal) {
@@ -631,37 +948,50 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
}
@Override public String getString(int ordinal) {
- return typedValue().toString();
- }
-
- @Override public UUID getUuid(int ordinal) {
- return UUID.fromString(typedValue().toString());
+ Object value = typedValue();
+ return decoding() == LeafDecoding.JSON
+ ? new String(bytesOf(value), StandardCharsets.UTF_8)
+ : value.toString();
}
@Override public BigDecimal getDecimal(int ordinal, int precision, int
scale) {
Object value = typedValue();
- if (value instanceof BigDecimal) {
- return (BigDecimal) value;
- }
- Schema tvSchema =
unwrapNullable(record.getSchema().getField(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD).schema());
- Conversions.DecimalConversion conversion = new
Conversions.DecimalConversion();
- if (value instanceof ByteBuffer) {
- return conversion.fromBytes((ByteBuffer) value, tvSchema,
tvSchema.getLogicalType());
- }
- if (value instanceof GenericFixed) {
- return conversion.fromFixed((GenericFixed) value, tvSchema,
tvSchema.getLogicalType());
+ switch (decoding()) {
+ case UNSIGNED_INT64:
+ return new BigDecimal(Long.toUnsignedString((Long) value));
+ case DECIMAL_UNSCALED:
+ return BigDecimal.valueOf(((Number) value).longValue(), scale);
+ case DECIMAL_BYTES:
+ return value instanceof BigDecimal
+ ? (BigDecimal) value
+ : new BigDecimal(new BigInteger(bytesOf(value)), scale);
+ default:
+ throw new IllegalStateException("Unexpected decimal decoding: " +
decoding());
}
- throw new IllegalStateException("Unexpected decimal representation: " +
value);
}
@Override public ShreddingUtils.ShreddedRow getStruct(int ordinal, int
numFields) {
// Object shredding: typed_value is a record whose fields are the
shredded object fields.
- return new AvroObjectRow((GenericRecord) typedValue(), schema);
+ return new AvroObjectRow((GenericRecord) typedValue(), schema,
decodings);
}
@Override public ShreddingUtils.ShreddedRow getArray(int ordinal) {
- return new AvroArrayRow((List<?>) typedValue(), schema.arraySchema);
+ return new AvroArrayRow((List<?>) typedValue(), schema.arraySchema,
decodings);
+ }
+ }
+
+ /**
+ * The bytes of a binary Avro value: a ByteBuffer for parquet BINARY, a
GenericFixed for
+ * FIXED_LEN_BYTE_ARRAY.
+ */
+ private static byte[] bytesOf(Object value) {
+ if (value instanceof ByteBuffer) {
+ return toByteArray((ByteBuffer) value);
+ }
+ if (value instanceof GenericFixed) {
+ return ((GenericFixed) value).bytes();
}
+ throw new IllegalStateException("Unexpected binary representation: " +
value);
}
/**
@@ -671,10 +1001,13 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
static final class AvroObjectRow extends BaseAvroShreddedRow {
private final GenericRecord typedValueRecord;
private final VariantSchema parentSchema;
+ private final Map<VariantSchema, LeafDecoding> decodings;
- AvroObjectRow(GenericRecord typedValueRecord, VariantSchema parentSchema) {
+ AvroObjectRow(GenericRecord typedValueRecord, VariantSchema parentSchema,
+ Map<VariantSchema, LeafDecoding> decodings) {
this.typedValueRecord = typedValueRecord;
this.parentSchema = parentSchema;
+ this.decodings = decodings;
}
@Override public boolean isNullAt(int ordinal) {
@@ -683,7 +1016,7 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
@Override public ShreddingUtils.ShreddedRow getStruct(int ordinal, int
numFields) {
VariantSchema.ObjectField field = parentSchema.objectSchema[ordinal];
- return new AvroVariantRow((GenericRecord)
typedValueRecord.get(field.fieldName), field.schema);
+ return new AvroVariantRow((GenericRecord)
typedValueRecord.get(field.fieldName), field.schema, decodings);
}
}
@@ -694,10 +1027,12 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
static final class AvroArrayRow extends BaseAvroShreddedRow {
private final List<?> elements;
private final VariantSchema elementSchema;
+ private final Map<VariantSchema, LeafDecoding> decodings;
- AvroArrayRow(List<?> elements, VariantSchema elementSchema) {
+ AvroArrayRow(List<?> elements, VariantSchema elementSchema,
Map<VariantSchema, LeafDecoding> decodings) {
this.elements = elements;
this.elementSchema = elementSchema;
+ this.decodings = decodings;
}
@Override public int numElements() {
@@ -709,7 +1044,7 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
}
@Override public ShreddingUtils.ShreddedRow getStruct(int ordinal, int
numFields) {
- return new AvroVariantRow((GenericRecord) elements.get(ordinal),
elementSchema);
+ return new AvroVariantRow((GenericRecord) elements.get(ordinal),
elementSchema, decodings);
}
}
-}
\ No newline at end of file
+}
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 3f86fe893524..b8b1d2d177de 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
@@ -30,6 +30,7 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.core.io.storage.HoodieAvroBootstrapFileReader;
import org.apache.hudi.core.io.storage.HoodieFileReader;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.variant.Spark4VariantShreddingProvider;
@@ -37,19 +38,34 @@ import
org.apache.hudi.variant.Spark4VariantShreddingProvider;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.IndexedRecord;
+import org.apache.hadoop.conf.Configuration;
import org.apache.parquet.avro.AvroParquetWriter;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
import org.apache.spark.types.variant.Variant;
import org.apache.spark.types.variant.VariantBuilder;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import java.io.IOException;
+import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.nio.file.Path;
import java.time.ZoneOffset;
import java.util.Arrays;
+import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import static
org.apache.parquet.avro.HoodieAvroParquetSchemaConverter.getAvroSchemaConverter;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -68,6 +84,8 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
* InternalRow reader), so this is the only place the alignment is exercised
end to end.
* The null/throw guards of {@code create} are covered by {@code
TestHoodieVariantReconstruction}
* (hudi-hadoop-common); the provider's shred/rebuild in isolation by {@code
TestSpark4VariantShreddingProvider}.
+ * Files with typed_value encodings Hudi's writer never produces are read
through {@link HoodieAvroParquetReader},
+ * which hands the reconstruction the footer's parquet type.
*/
class TestHoodieVariantReconstructionRoundTrip {
@@ -89,7 +107,9 @@ class TestHoodieVariantReconstructionRoundTrip {
HoodieStorage storage = HoodieTestUtils.getStorage(tmp.toString()); //
allow.reading.shredded defaults true
HoodieVariantReconstruction reconstruction =
- HoodieVariantReconstruction.create(fileSchema, requestedSchema,
storage);
+ HoodieVariantReconstruction.create(fileSchema,
+
getAvroSchemaConverter(storage.getConf().unwrapAs(Configuration.class)).convert(fileSchema),
+ requestedSchema, storage);
assertNotNull(reconstruction, "create should build a reconstruction for a
shredded variant column");
// Build the input the parquet-avro reader would produce: a record at the
intermediate (shredded)
@@ -120,8 +140,8 @@ class TestHoodieVariantReconstructionRoundTrip {
@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
+ // pins that Hudi detects the resulting two-field group, but its stub
provider ignores the parquet
+ // group, so this is the only place the real provider sees the shape:
buildReadVariantSchema 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<>();
@@ -157,7 +177,9 @@ class TestHoodieVariantReconstructionRoundTrip {
HoodieStorage storage = HoodieTestUtils.getStorage(tmp.toString()); //
allow.reading.shredded defaults true
HoodieVariantReconstruction reconstruction =
- HoodieVariantReconstruction.create(fileSchema, requestedSchema,
storage);
+ HoodieVariantReconstruction.create(fileSchema,
+
getAvroSchemaConverter(storage.getConf().unwrapAs(Configuration.class)).convert(fileSchema),
+ requestedSchema, storage);
assertNotNull(reconstruction, "a shredded group with no value column must
still engage");
GenericRecord valueLessV = new GenericData.Record(
@@ -237,6 +259,105 @@ class TestHoodieVariantReconstructionRoundTrip {
reader.close();
}
+ private static final String VARIANT_TEST_NAMESPACE = "org.apache.hudi.test";
+
+ private static Variant scalar(Consumer<VariantBuilder> append) {
+ VariantBuilder builder = new VariantBuilder(false);
+ append.accept(builder);
+ return builder.result();
+ }
+
+ /** A required shredded variant group {@code column} with the given optional
typed_value declaration. */
+ private static String shreddedGroup(String column, String typedValue) {
+ return " required group " + column + " { required binary metadata;
optional binary value; optional "
+ + typedValue + "; }";
+ }
+
+ private static StoragePath writeParquet(Path tmp, MessageType schema, Group
row) throws IOException {
+ Path file = tmp.resolve("shredded.parquet");
+ try (ParquetWriter<Group> writer =
+ ExampleParquetWriter.builder(new
org.apache.hadoop.fs.Path(file.toString())).withType(schema).build()) {
+ writer.write(row);
+ }
+ return new StoragePath(file.toUri().toString());
+ }
+
+ @Test
+ void readerRebuildsTypedValueEncodingsTheFooterConversionLoses(@TempDir Path
tmp) throws Exception {
+ // A file from another writer whose typed_value leaves the footer's Avro
conversion reads as a plain
+ // int, long, fixed and bytes. The reader decodes each leaf by its parquet
type, as Spark 4 does.
+ String json = "{\"a\":[1,2]}";
+ byte[] bytes = {1, 2, 3, (byte) 0xFF};
+ long millis = 1_700_000_000_123L;
+ MessageType fileSchema = MessageTypeParser.parseMessageType("message r {"
+ + shreddedGroup("v_dec", "int32 typed_value (DECIMAL(9,2))")
+ + shreddedGroup("v_ts", "int64 typed_value (TIMESTAMP(MILLIS,true))")
+ + shreddedGroup("v_fixed", "fixed_len_byte_array(4) typed_value")
+ + shreddedGroup("v_u64", "int64 typed_value (INTEGER(64,false))")
+ + shreddedGroup("v_json", "binary typed_value (JSON)") + " }");
+ Binary metadata =
Binary.fromConstantByteArray(scalar(VariantBuilder::appendNull).getMetadata());
+ Group row = new SimpleGroupFactory(fileSchema).newGroup();
+ row.addGroup("v_dec").append(HoodieSchema.Variant.VARIANT_METADATA_FIELD,
metadata)
+ .append(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD, 150);
+ row.addGroup("v_ts").append(HoodieSchema.Variant.VARIANT_METADATA_FIELD,
metadata)
+ .append(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD, millis);
+
row.addGroup("v_fixed").append(HoodieSchema.Variant.VARIANT_METADATA_FIELD,
metadata)
+ .append(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD,
Binary.fromConstantByteArray(bytes));
+ row.addGroup("v_u64").append(HoodieSchema.Variant.VARIANT_METADATA_FIELD,
metadata)
+ .append(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD, -1L);
+ row.addGroup("v_json").append(HoodieSchema.Variant.VARIANT_METADATA_FIELD,
metadata)
+ .append(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD, json);
+
+ Map<String, Variant> expected = new LinkedHashMap<>();
+ expected.put("v_dec", scalar(b -> b.appendDecimal(new
BigDecimal("1.50"))));
+ expected.put("v_ts", scalar(b -> b.appendTimestamp(millis * 1000)));
+ expected.put("v_fixed", scalar(b -> b.appendBinary(bytes)));
+ expected.put("v_u64", scalar(b -> b.appendDecimal(new
BigDecimal("18446744073709551615"))));
+ expected.put("v_json", scalar(b -> b.appendString(json)));
+ HoodieSchema requestedSchema = HoodieSchema.createRecord("r",
VARIANT_TEST_NAMESPACE, null,
+ expected.keySet().stream()
+ .map(name -> HoodieSchemaField.of(name,
HoodieSchema.createVariant(name, VARIANT_TEST_NAMESPACE, null)))
+ .collect(Collectors.toList()));
+
+ HoodieStorage storage = HoodieTestUtils.getStorage(tmp.toString());
+ try (HoodieAvroParquetReader reader = new HoodieAvroParquetReader(storage,
writeParquet(tmp, fileSchema, row));
+ ClosableIterator<IndexedRecord> records =
reader.getIndexedRecordIterator(requestedSchema, requestedSchema)) {
+ GenericRecord out = (GenericRecord) records.next();
+ for (Map.Entry<String, Variant> column : expected.entrySet()) {
+ GenericRecord rebuiltV = (GenericRecord) out.get(column.getKey());
+ Variant rebuilt = new Variant(
+ toBytes(rebuiltV.get(HoodieSchema.Variant.VARIANT_VALUE_FIELD)),
+
toBytes(rebuiltV.get(HoodieSchema.Variant.VARIANT_METADATA_FIELD)));
+ assertEquals(column.getValue().toJson(ZoneOffset.UTC),
rebuilt.toJson(ZoneOffset.UTC), column.getKey());
+ // The binary also pins the variant type (e.g. decimal vs long), which
the json can hide.
+ assertArrayEquals(column.getValue().getValue(), rebuilt.getValue(),
column.getKey());
+ }
+ }
+ }
+
+ @Test
+ void readerFailsOnOpenForTypedValueEncodingSparkRejects(@TempDir Path tmp)
throws Exception {
+ // Spark 4 rejects a TIMESTAMP(NANOS) typed_value when it opens the file,
whatever the rows hold. So
+ // does the reader, here with every typed_value null and the variant kept
in the residual value.
+ MessageType fileSchema = MessageTypeParser.parseMessageType("message r {"
+ + shreddedGroup("v", "int64 typed_value (TIMESTAMP(NANOS,true))") + "
}");
+ Variant variant = scalar(b -> b.appendLong(7));
+ Group row = new SimpleGroupFactory(fileSchema).newGroup();
+ row.addGroup("v")
+ .append(HoodieSchema.Variant.VARIANT_METADATA_FIELD,
Binary.fromConstantByteArray(variant.getMetadata()))
+ .append(HoodieSchema.Variant.VARIANT_VALUE_FIELD,
Binary.fromConstantByteArray(variant.getValue()));
+ HoodieSchema requestedSchema = HoodieSchema.createRecord("r",
VARIANT_TEST_NAMESPACE, null,
+ Collections.singletonList(HoodieSchemaField.of("v",
HoodieSchema.createVariant())));
+
+ HoodieStorage storage = HoodieTestUtils.getStorage(tmp.toString());
+ try (HoodieAvroParquetReader reader = new HoodieAvroParquetReader(storage,
writeParquet(tmp, fileSchema, row))) {
+ HoodieException e = assertThrows(HoodieException.class,
+ () -> reader.getIndexedRecordIterator(requestedSchema,
requestedSchema));
+ assertTrue(e.getMessage().contains("column 'v'") &&
e.getMessage().contains("TIMESTAMP(NANOS,true)"),
+ e.getMessage());
+ }
+ }
+
private static byte[] toBytes(Object byteBuffer) {
ByteBuffer buf = ((ByteBuffer) byteBuffer).duplicate();
byte[] out = new byte[buf.remaining()];
diff --git
a/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java
b/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java
index 70f893d3c06d..2a4cfe787299 100644
---
a/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java
+++
b/hudi-spark-datasource/hudi-spark4-common/src/test/java/org/apache/hudi/variant/TestSpark4VariantShreddingProvider.java
@@ -20,6 +20,7 @@
package org.apache.hudi.variant;
import org.apache.hudi.common.avro.ConvertingGenericData;
+import org.apache.hudi.common.avro.VariantShreddingProvider;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.schema.HoodieSchemaType;
@@ -30,6 +31,16 @@ import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.util.Utf8;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.avro.AvroSchemaConverterWithTimestampNTZ;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+import org.apache.parquet.schema.Type;
import org.apache.spark.types.variant.Variant;
import org.apache.spark.types.variant.VariantBuilder;
import org.junit.jupiter.api.Test;
@@ -41,14 +52,36 @@ import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.time.ZoneOffset;
import java.util.Arrays;
+import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.UUID;
import java.util.stream.Stream;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.hudi.common.avro.HoodieAvroUtils.unwrapNullable;
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.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit.MICROS;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit.MILLIS;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit.NANOS;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.bsonType;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.decimalType;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.enumType;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.float16Type;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.intType;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.intervalType;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.jsonType;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.timeType;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.timestampType;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.uuidType;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY;
+import static
org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT96;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -58,12 +91,14 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* 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
+ * the reconstructed variant. This exercises {@code shredVariantRecord},
{@code avroTypeToScalarType},
+ * {@code convertScalarToAvro}, {@code prepareRebuild} (the read schema built
from the file's parquet
+ * type, with each scalar leaf's {@code LeafDecoding}), and the
* {@code AvroVariantRow}/{@code AvroObjectRow}/{@code AvroArrayRow} accessors
across every reachable
* scalar leaf type, object/array shapes, partial (residual) shredding, and
the null/error guards - the
* AVRO read-path reconstruction that the Spark MOR SQL test cannot reach
(Spark compaction reads base
- * files via the InternalRow reader, not HoodieAvroParquetReader).
+ * files via the InternalRow reader, not HoodieAvroParquetReader). Parquet
encodings Hudi's writer never
+ * produces are read straight from a parquet group and pinned against what
Spark 4's reader returns.
*/
class TestSpark4VariantShreddingProvider {
@@ -87,9 +122,19 @@ class TestSpark4VariantShreddingProvider {
return record;
}
+ /** Prepares the rebuild of {@code shredded} from the parquet group it is
written as. */
+ private VariantShreddingProvider.VariantRebuilder
rebuilder(HoodieSchema.Variant shredded) {
+ HoodieSchema record = HoodieSchema.createRecord("r",
"org.apache.hudi.test", null,
+ Collections.singletonList(HoodieSchemaField.of("v", shredded)));
+ GroupType parquetType = new
AvroSchemaConverterWithTimestampNTZ().convert(record).getType("v").asGroupType();
+ return provider.prepareRebuild("v", parquetType, unshreddedSchema);
+ }
+
private Variant rebuild(GenericRecord shreddedRecord, HoodieSchema.Variant
shredded) {
- GenericRecord rebuilt =
- provider.rebuildVariantRecord(shreddedRecord,
shredded.getAvroSchema(), unshreddedSchema);
+ return toVariant(rebuilder(shredded).rebuild(shreddedRecord));
+ }
+
+ private static Variant toVariant(GenericRecord rebuilt) {
return new Variant(toBytes(rebuilt.get(VARIANT_VALUE_FIELD)),
toBytes(rebuilt.get(VARIANT_METADATA_FIELD)));
}
@@ -162,14 +207,15 @@ class TestSpark4VariantShreddingProvider {
}
//
---------------------------------------------------------------------------
- // Scalar leaf types: one per reachable branch of avroTypeToScalarType /
convertScalarToAvro
- // (the Byte/Short widening arms in convertScalarToAvro are unreachable:
Avro has no byte/short).
+ // Scalar leaf types: one per branch of avroTypeToScalarType /
convertScalarToAvro, each read back
+ // from the parquet type Hudi's writer gives it.
//
---------------------------------------------------------------------------
@Test
void intLeafShredsToInteger() {
// An integral variant shreds into an INT typed_value as a boxed Integer
(Avro has no byte/short,
- // so avroTypeToScalarType only ever emits IntegralSize.INT here).
+ // so the write side only ever emits IntegralSize.INT here; the narrower
parquet INT(8) / INT(16)
+ // leaves only come from other writers, see
rebuildsLeafFromItsParquetEncodingLikeSpark).
assertScalarShredsTo(scalar(b -> b.appendLong(1000)),
HoodieSchema.create(HoodieSchemaType.INT), 1000);
}
@@ -207,19 +253,6 @@ class TestSpark4VariantShreddingProvider {
HoodieSchema.create(HoodieSchemaType.BYTES), ByteBuffer.wrap(payload));
}
- @Test
- void rebuildUuidFromStringLeaf() {
- // A uuid typed_value is rejected on write (see
rejectsTypedValueLeafSparkCannotShred) but stays
- // readable, so the shredded record is built directly rather than through
shred().
- UUID uuid = UUID.fromString("12345678-1234-1234-1234-123456789abc");
- Variant original = scalar(b -> b.appendUuid(uuid));
- HoodieSchema.Variant shredded = shreddedLeaf(HoodieSchema.createUUID());
- GenericRecord shreddedRecord = new
GenericData.Record(shredded.getAvroSchema());
- shreddedRecord.put(VARIANT_METADATA_FIELD,
ByteBuffer.wrap(original.getMetadata()));
- shreddedRecord.put(VARIANT_TYPED_VALUE_FIELD, uuid.toString());
- assertEquals(original.toJson(ZoneOffset.UTC), rebuild(shreddedRecord,
shredded).toJson(ZoneOffset.UTC));
- }
-
@Test
void dateShredsToDaysSinceEpoch() {
assertScalarShredsTo(scalar(b -> b.appendDate(19000)),
HoodieSchema.createDate(), 19000);
@@ -415,7 +448,188 @@ class TestSpark4VariantShreddingProvider {
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));
+ () -> rebuilder(shredded).rebuild(shreddedRecord));
+ }
+
+ //
---------------------------------------------------------------------------
+ // Reading a shredded file: each typed_value leaf is decoded by its parquet
type, as Spark 4's own
+ // reader does. The footer's Avro conversion drops the encodings that matter
here (unsigned and
+ // decimal INT32 / INT64, JSON, millisecond timestamps), so these leaves are
built straight from a
+ // parquet group, fed the value parquet-avro reads for them, and compared
with Spark's variant.
+ //
---------------------------------------------------------------------------
+
+ private static final byte[] EMPTY_METADATA =
scalar(VariantBuilder::appendNull).getMetadata();
+
+ /** A typed_value leaf, built directly because Types refuses some encodings
under test (e.g. DECIMAL(10, 0) on INT32). */
+ private static PrimitiveType leaf(PrimitiveTypeName type, int length,
LogicalTypeAnnotation annotation) {
+ return new PrimitiveType(Type.Repetition.OPTIONAL, type, length,
VARIANT_TYPED_VALUE_FIELD)
+ .withLogicalTypeAnnotation(annotation);
+ }
+
+ private static PrimitiveType leaf(PrimitiveTypeName type,
LogicalTypeAnnotation annotation) {
+ return leaf(type, 0, annotation);
+ }
+
+ /** A top-level shredded variant column {@code v} with a scalar typed_value.
*/
+ private static GroupType shreddedColumn(PrimitiveType typedValue) {
+ return new GroupType(Type.Repetition.OPTIONAL, "v",
+ new PrimitiveType(Type.Repetition.REQUIRED, BINARY,
VARIANT_METADATA_FIELD),
+ new PrimitiveType(Type.Repetition.OPTIONAL, BINARY,
VARIANT_VALUE_FIELD),
+ typedValue);
+ }
+
+ /** The variant column {@code v} declared by a parquet schema snippet. */
+ private static GroupType column(String group) {
+ return MessageTypeParser.parseMessageType("message r { " + group + "
}").getType("v").asGroupType();
+ }
+
+ /** The Avro schema HoodieAvroParquetReader reads {@code column} at: its
footer conversion, lists unwrapped. */
+ private static Schema avroReadSchema(GroupType column) {
+ Configuration conf = new Configuration(false);
+
conf.setBoolean(AvroSchemaConverterWithTimestampNTZ.ADD_LIST_ELEMENT_RECORDS,
false);
+ Schema record = new AvroSchemaConverterWithTimestampNTZ(conf).convert(new
MessageType("r", column)).toAvroSchema();
+ return unwrapNullable(record.getField(column.getName()).schema());
+ }
+
+ private static Schema fieldSchema(Schema record, String field) {
+ return unwrapNullable(record.getField(field).schema());
+ }
+
+ private static GenericRecord record(Schema schema, String field, Object
value) {
+ GenericRecord record = new GenericData.Record(schema);
+ record.put(field, value);
+ return record;
+ }
+
+ private static void assertSameVariant(Variant expected, Variant actual) {
+ assertEquals(expected.toJson(ZoneOffset.UTC),
actual.toJson(ZoneOffset.UTC));
+ // The binary also pins the variant type, which the json can hide.
+ assertArrayEquals(expected.getValue(), actual.getValue(), "variant type or
encoding differs");
+ }
+
+ private static Stream<Arguments> typedValueEncodingsSparkReads() {
+ byte[] bytes = {1, 2, 3, (byte) 0xFF};
+ String json = "{\"a\":[1,2]}";
+ long millis = 1_700_000_000_123L;
+ BigDecimal negative = new BigDecimal("-123.45");
+ Schema fixed5 = Schema.createFixed(VARIANT_TYPED_VALUE_FIELD, null, null,
5);
+ return Stream.of(
+ // Narrow and unsigned ints widen to the next signed size; INT(64,
false) becomes DECIMAL(20, 0).
+ Arguments.of(leaf(INT32, intType(8, true)), -5, scalar(b ->
b.appendLong(-5))),
+ Arguments.of(leaf(INT32, intType(16, false)), 60000, scalar(b ->
b.appendLong(60000))),
+ Arguments.of(leaf(INT32, intType(32, false)), -1, scalar(b ->
b.appendLong(4294967295L))),
+ Arguments.of(leaf(INT64, intType(64, false)), -1L,
+ scalar(b -> b.appendDecimal(new
BigDecimal("18446744073709551615")))),
+ // A decimal on INT32 / INT64 arrives as its unscaled value.
+ Arguments.of(leaf(INT32, decimalType(2, 9)), 150, scalar(b ->
b.appendDecimal(new BigDecimal("1.50")))),
+ Arguments.of(leaf(INT32, decimalType(2, 9)), -12345, scalar(b ->
b.appendDecimal(negative))),
+ Arguments.of(leaf(INT64, decimalType(4, 18)), 123456789012345678L,
+ scalar(b -> b.appendDecimal(new
BigDecimal("12345678901234.5678")))),
+ Arguments.of(leaf(FIXED_LEN_BYTE_ARRAY, 5, decimalType(2, 10)),
+ new Conversions.DecimalConversion().toFixed(negative, fixed5,
LogicalTypes.decimal(10, 2)),
+ scalar(b -> b.appendDecimal(negative))),
+ // Millisecond timestamps scale to micros and keep the UTC / local
distinction.
+ Arguments.of(leaf(INT64, timestampType(true, MILLIS)), millis,
scalar(b -> b.appendTimestamp(millis * 1000))),
+ Arguments.of(leaf(INT64, timestampType(false, MILLIS)), millis,
scalar(b -> b.appendTimestampNtz(millis * 1000))),
+ // JSON and ENUM read as strings; BSON and a plain
FIXED_LEN_BYTE_ARRAY as binary.
+ Arguments.of(leaf(BINARY, jsonType()),
ByteBuffer.wrap(json.getBytes(UTF_8)), scalar(b -> b.appendString(json))),
+ Arguments.of(leaf(BINARY, enumType()), new Utf8("A"), scalar(b ->
b.appendString("A"))),
+ Arguments.of(leaf(BINARY, bsonType()), ByteBuffer.wrap(bytes),
scalar(b -> b.appendBinary(bytes))),
+ Arguments.of(leaf(FIXED_LEN_BYTE_ARRAY, 4, null),
+ new
GenericData.Fixed(Schema.createFixed(VARIANT_TYPED_VALUE_FIELD, null, null, 4),
bytes),
+ scalar(b -> b.appendBinary(bytes))));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("typedValueEncodingsSparkReads")
+ void rebuildsLeafFromItsParquetEncodingLikeSpark(PrimitiveType leaf, Object
typedValue, Variant expected) {
+ GroupType column = shreddedColumn(leaf);
+ GenericRecord shreddedRecord = record(avroReadSchema(column),
VARIANT_TYPED_VALUE_FIELD, typedValue);
+ shreddedRecord.put(VARIANT_METADATA_FIELD,
ByteBuffer.wrap(EMPTY_METADATA));
+ assertSameVariant(expected, toVariant(provider.prepareRebuild("v", column,
unshreddedSchema).rebuild(shreddedRecord)));
+ }
+
+ private static Stream<Arguments> typedValueEncodingsSparkRejects() {
+ String sparkRejects = "which Spark 4 does not support reading";
+ return Stream.of(
+ Arguments.of(leaf(INT32, timeType(true, MILLIS)), sparkRejects),
+ Arguments.of(leaf(INT64, timeType(true, MICROS)), sparkRejects),
+ Arguments.of(leaf(INT64, timeType(false, MICROS)), sparkRejects),
+ Arguments.of(leaf(INT64, timeType(true, NANOS)), sparkRejects),
+ Arguments.of(leaf(INT64, timestampType(true, NANOS)), sparkRejects),
+ Arguments.of(leaf(INT64, timestampType(false, NANOS)), sparkRejects),
+ Arguments.of(leaf(FIXED_LEN_BYTE_ARRAY, 2, float16Type()),
sparkRejects),
+ Arguments.of(leaf(FIXED_LEN_BYTE_ARRAY, 16, uuidType()), sparkRejects),
+ Arguments.of(leaf(FIXED_LEN_BYTE_ARRAY, 12, intervalType()),
sparkRejects),
+ Arguments.of(leaf(INT32, decimalType(0, 10)), sparkRejects),
+ Arguments.of(leaf(INT64, decimalType(0, 19)), sparkRejects),
+ Arguments.of(leaf(INT96, null), "INT96 typed_value is not supported by
Hudi's Avro reader"));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("typedValueEncodingsSparkRejects")
+ void rejectsLeafSparkCannotReadWhenPrepared(PrimitiveType leaf, String
reason) {
+ // Fails when the file is opened, before any record, naming the column and
the parquet type.
+ HoodieException e = assertThrows(HoodieException.class,
+ () -> provider.prepareRebuild("v", shreddedColumn(leaf),
unshreddedSchema));
+ String message = e.getMessage();
+ assertTrue(message.contains("column 'v'") && message.contains("'" + leaf +
"'") && message.contains(reason), message);
+ }
+
+ @Test
+ void rebuildsAndRejectsNestedLeavesByTheirParquetEncoding() throws Exception
{
+ // DECIMAL(9, 2) on INT32 as an object field and as the element of a
3-level list.
+ GroupType column = column("optional group v { required binary metadata;
optional binary value;"
+ + " optional group typed_value {"
+ + " required group price { optional binary value; optional int32
typed_value (DECIMAL(9,2)); }"
+ + " required group history { optional binary value; optional group
typed_value (LIST) {"
+ + " repeated group list { required group element {"
+ + " optional binary value; optional int32 typed_value
(DECIMAL(9,2)); } } } } } }");
+ Variant expected =
VariantBuilder.parseJson("{\"price\":1.50,\"history\":[1.25,-0.75]}", false);
+
+ Schema v = avroReadSchema(column);
+ Schema typedValue = fieldSchema(v, VARIANT_TYPED_VALUE_FIELD);
+ Schema history = fieldSchema(typedValue, "history");
+ Schema element = unwrapNullable(fieldSchema(history,
VARIANT_TYPED_VALUE_FIELD).getElementType());
+ GenericRecord typedValueRecord = new GenericData.Record(typedValue);
+ typedValueRecord.put("price", record(fieldSchema(typedValue, "price"),
VARIANT_TYPED_VALUE_FIELD, 150));
+ typedValueRecord.put("history", record(history, VARIANT_TYPED_VALUE_FIELD,
Arrays.asList(
+ record(element, VARIANT_TYPED_VALUE_FIELD, 125), record(element,
VARIANT_TYPED_VALUE_FIELD, -75))));
+ GenericRecord shreddedRecord = record(v, VARIANT_TYPED_VALUE_FIELD,
typedValueRecord);
+ shreddedRecord.put(VARIANT_METADATA_FIELD,
ByteBuffer.wrap(expected.getMetadata()));
+ assertSameVariant(expected, toVariant(provider.prepareRebuild("v", column,
unshreddedSchema).rebuild(shreddedRecord)));
+
+ // A leaf Spark rejects below an object field: the error carries the
column's and the leaf's paths.
+ GroupType rejected = column("optional group v { required binary metadata;
optional binary value;"
+ + " optional group typed_value {"
+ + " required group t { optional binary value; optional int64
typed_value (TIME(MICROS,true)); } } }");
+ HoodieException e = assertThrows(HoodieException.class,
+ () -> provider.prepareRebuild("s.v", rejected, unshreddedSchema));
+ assertTrue(e.getMessage().contains("column 's.v'")
+ && e.getMessage().contains("typed_value
's.v.typed_value.t.typed_value'"), e.getMessage());
+ }
+
+ private static Stream<Arguments> invalidShreddingLayouts() {
+ String header = "optional group v { required binary metadata; optional
binary value;";
+ return Stream.of(
+ Arguments.of("unknown field in an object field group", header
+ + " optional group typed_value { required group a {"
+ + " optional binary value; optional int32 typed_value; optional
binary extra; } } }", "v.typed_value.a"),
+ Arguments.of("MAP typed_value", header
+ + " optional group typed_value (MAP) { repeated group key_value {
required binary key (STRING);"
+ + " required group value { optional binary value; optional int32
typed_value; } } } }", "v.typed_value"),
+ Arguments.of("LIST typed_value of non-struct elements", header
+ + " optional group typed_value (LIST) { repeated group list {
optional int32 element; } } }", "v.typed_value"),
+ Arguments.of("no top-level metadata",
+ "optional group v { optional binary value; optional int32
typed_value; }", "v"));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("invalidShreddingLayouts")
+ void rejectsInvalidShreddingLayoutWhenPrepared(String layout, String group,
String offendingPath) {
+ HoodieException e = assertThrows(HoodieException.class,
+ () -> provider.prepareRebuild("v", column(group), unshreddedSchema));
+ assertTrue(e.getMessage().contains("'" + offendingPath + "' is not a valid
shredded variant layout"), e.getMessage());
}
//
---------------------------------------------------------------------------
@@ -439,9 +653,9 @@ class TestSpark4VariantShreddingProvider {
@Test
void rebuildReturnsNullForNullRecord() {
// Coverage of a defensive guard only: the sole production caller
(HoodieVariantReconstruction)
- // already checks instanceof GenericRecord before calling
rebuildVariantRecord.
+ // already checks instanceof GenericRecord before calling the rebuilder.
HoodieSchema.Variant shredded =
shreddedLeaf(HoodieSchema.create(HoodieSchemaType.LONG));
- assertNull(provider.rebuildVariantRecord(null, shredded.getAvroSchema(),
unshreddedSchema));
+ assertNull(rebuilder(shredded).rebuild(null));
}
@Test
@@ -450,7 +664,7 @@ class TestSpark4VariantShreddingProvider {
GenericRecord shreddedRecord = shred(scalar(b -> b.appendLong(7)),
shredded);
shreddedRecord.put(VARIANT_METADATA_FIELD, null);
assertThrows(HoodieException.class,
- () -> provider.rebuildVariantRecord(shreddedRecord,
shredded.getAvroSchema(), unshreddedSchema));
+ () -> rebuilder(shredded).rebuild(shreddedRecord));
}
private static byte[] toBytes(Object byteBuffer) {
