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 5dee63857193 fix(spark): reject variant leaves Spark 4 cannot shred
(#20326)
5dee63857193 is described below
commit 5dee63857193d19e5a61110ed6bbbbffe7955094
Author: voonhous <[email protected]>
AuthorDate: Thu Oct 15 19:29:53 2026 +0800
fix(spark): reject variant leaves Spark 4 cannot shred (#20326)
* fix(spark): decline shredding into fixed leaves
Spark4VariantShreddingProvider mapped an Avro fixed typed_value leaf
to Spark's BinaryType, so shredding handed parquet-avro a ByteBuffer
where a GenericFixed is required (ClassCastException at write) and the
fixed size was never checked. A uuid logical type on fixed(16) mapped
to UuidType, which converts to a String that a fixed cannot hold
(UnresolvedUnionException).
Spark's VariantSchema has no fixed-length binary type, so decline to
shred both leaves: avroTypeToScalarType returns null and the value
stays in the residual, as it already does for millisecond timestamps.
Decimal on fixed is unaffected because the decimal logical-type check
runs first.
Addresses the write side of #19442; the read-side plain FLBA case
is left open.
* fix(spark): reject unshreddable variant leaves
Match Spark 4's own variant shredding on the write path. Spark's
SparkShreddingUtils.buildVariantSchema only accepts Boolean, Byte,
Short, Int, Long, Float, Double, String, Binary, Date, Timestamp,
TimestampNTZ and Decimal scalars and throws
invalidVariantShreddingSchema for anything else; its inference keeps
UUIDs in the residual.
Spark4VariantShreddingProvider now throws a HoodieException when a
scalar typed_value is not the Avro equivalent of one of those types:
plain fixed, uuid on string or fixed, millisecond and nanosecond
timestamps, time types, enum and map. Previously these broke the write,
were silently declined, mapped time to an integer, or (uuid on string)
read back as a plain string.
The check is write-only. avroTypeToScalarType goes back to its master
mapping because the read path shares it, so read behavior is unchanged.
Hudi's shredding DDL and inference never produce these leaf types.
* refactor(spark): hoist shreddable scalar type list
Keep the error-message list of supported typed_value types in a
constant next to isSparkShreddableScalar so the two do not drift.
---
.../variant/Spark4VariantShreddingProvider.java | 51 ++++++++++++--
.../TestSpark4VariantShreddingProvider.java | 77 ++++++++++++++++------
2 files changed, 104 insertions(+), 24 deletions(-)
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 2c1e5f36df63..63a6e9d17588 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
@@ -176,7 +176,15 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
break;
default:
- // Scalar shredding
+ // 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)) {
+ LogicalType logicalType = tvSchema.getLogicalType();
+ String typeName = tvSchema.getType().getName()
+ + (logicalType == null ? "" : " (" + logicalType.getName() +
")");
+ throw new HoodieException("Variant shredding schema has a
typed_value of type " + typeName
+ + ", which Spark 4 cannot shred. Supported typed_value types:
" + SPARK_SHREDDABLE_SCALAR_TYPES + ".");
+ }
scalarSchema = avroTypeToScalarType(tvSchema);
break;
}
@@ -195,6 +203,41 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
return result;
}
+ /**
+ * The scalar {@code typed_value} types {@link #isSparkShreddableScalar}
accepts, for error messages.
+ * Keep in sync with that switch.
+ */
+ private static final String SPARK_SHREDDABLE_SCALAR_TYPES = "boolean, int,
long, float, double, string, bytes, "
+ + "decimal, date, timestamp-micros, local-timestamp-micros";
+
+ /**
+ * Whether a scalar {@code typed_value} is the Avro equivalent of a scalar
type Spark 4's
+ * {@code SparkShreddingUtils.buildVariantSchema} accepts.
+ */
+ private static boolean isSparkShreddableScalar(Schema schema) {
+ LogicalType logicalType = schema.getLogicalType();
+ String logicalName = logicalType == null ? null : logicalType.getName();
+ switch (schema.getType()) {
+ case BOOLEAN:
+ case FLOAT:
+ case DOUBLE:
+ case STRING: // a uuid string is not allowed
+ return logicalName == null;
+ case INT:
+ return logicalName == null || "date".equals(logicalName);
+ case LONG:
+ return logicalName == null
+ || "timestamp-micros".equals(logicalName)
+ || "local-timestamp-micros".equals(logicalName);
+ case BYTES:
+ return logicalName == null || "decimal".equals(logicalName);
+ case FIXED:
+ return "decimal".equals(logicalName);
+ default:
+ return false;
+ }
+ }
+
/**
* Maps an Avro {@link Schema} type (potentially with logical type
annotations)
* to a {@link VariantSchema.ScalarType}.
@@ -219,9 +262,9 @@ public class Spark4VariantShreddingProvider implements
VariantShreddingProvider
return new VariantSchema.TimestampNTZType();
}
// The Variant binary spec stores timestamps in microseconds, so a
millisecond-precision
- // typed_value cannot represent a variant timestamp. Decline to shred it
as a scalar (the value
- // stays in the residual unshredded binary) rather than mapping it to
the micros TimestampType,
- // which would silently scale the value by 1000x.
+ // 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;
}
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 edbf9f270420..70f893d3c06d 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
@@ -26,12 +26,16 @@ import org.apache.hudi.common.schema.HoodieSchemaType;
import org.apache.hudi.exception.HoodieException;
import org.apache.avro.Conversions;
+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.spark.types.variant.Variant;
import org.apache.spark.types.variant.VariantBuilder;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
@@ -40,6 +44,7 @@ import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.UUID;
+import java.util.stream.Stream;
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;
@@ -203,9 +208,16 @@ class TestSpark4VariantShreddingProvider {
}
@Test
- void uuidShredsToString() {
+ 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");
- assertScalarShredsTo(scalar(b -> b.appendUuid(uuid)),
HoodieSchema.createUUID(), uuid.toString());
+ 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
@@ -228,6 +240,9 @@ class TestSpark4VariantShreddingProvider {
@Test
void decimalLeafShredsToBigDecimal() throws Exception {
assertScalarShredsTo("123.45", HoodieSchema.createDecimal(10, 2), new
BigDecimal("123.45"));
+ // Decimal-on-fixed still shreds: a fixed typed_value is rejected unless
it carries a decimal.
+ assertScalarShredsTo("123.45", HoodieSchema.createDecimal("dec_fixed",
"org.apache.hudi.test", null, 10, 2, 8),
+ new BigDecimal("123.45"));
}
@Test
@@ -243,16 +258,6 @@ class TestSpark4VariantShreddingProvider {
// "Decline to shred" fallbacks: value stays in the residual binary.
//
---------------------------------------------------------------------------
- @Test
- void microsTimestampIsNotShreddedIntoMillisLeaf() {
- // A millisecond-precision typed_value cannot represent a micros variant
timestamp, so
- // avroTypeToScalarType returns null and the value is left unshredded in
the residual.
- assertStaysInResidual(scalar(b ->
b.appendTimestamp(1_700_000_000_000_000L)),
- HoodieSchema.createTimestampMillis());
- assertStaysInResidual(scalar(b ->
b.appendTimestampNtz(1_700_000_000_000_000L)),
- HoodieSchema.createLocalTimestampMillis());
- }
-
@Test
void lossyDecimalIsNotShredded() {
// decimal(10,2) cannot represent 123.456 exactly, so the value must fall
to the residual.
@@ -260,20 +265,52 @@ class TestSpark4VariantShreddingProvider {
HoodieSchema.createDecimal(10, 2));
}
- @Test
- void enumLeafIsNotShredded() {
- // avroTypeToScalarType has no mapping for ENUM (the default: branch), so
the value declines
- // to the residual.
- assertStaysInResidual(scalar(b -> b.appendString("A")),
- HoodieSchema.createEnum("e", "org.apache.hudi.test", null,
Arrays.asList("A", "B")));
- }
-
@Test
void noTypedValueSchemaRoundTrips() throws Exception {
// A shredded schema without typed_value keeps the whole variant in the
residual value.
assertRoundTrips("{\"a\":1}", HoodieSchema.createVariantShredded(null));
}
+ //
---------------------------------------------------------------------------
+ // Leaf types Spark 4 cannot shred: rejected on write, as Spark's own
+ // SparkShreddingUtils.buildVariantSchema rejects them (GH 19442).
+ //
---------------------------------------------------------------------------
+
+ private static Stream<Arguments> typedValueLeavesSparkCannotShred() {
+ UUID uuid = UUID.fromString("12345678-1234-1234-1234-123456789abc");
+ long micros = 1_700_000_000_000_000L;
+ return Stream.of(
+ Arguments.of("fixed", HoodieSchema.createFixed("fixed4",
"org.apache.hudi.test", null, 4),
+ scalar(b -> b.appendBinary(new byte[] {1, 2, 3, 4}))),
+ Arguments.of("uuid on fixed", HoodieSchema.fromAvroSchema(
+ LogicalTypes.uuid().addToSchema(Schema.createFixed("uuid_fixed",
null, "org.apache.hudi.test", 16))),
+ scalar(b -> b.appendUuid(uuid))),
+ Arguments.of("uuid on string", HoodieSchema.createUUID(), scalar(b ->
b.appendUuid(uuid))),
+ Arguments.of("timestamp-millis", HoodieSchema.createTimestampMillis(),
+ scalar(b -> b.appendTimestamp(micros))),
+ Arguments.of("local-timestamp-millis",
HoodieSchema.createLocalTimestampMillis(),
+ scalar(b -> b.appendTimestampNtz(micros))),
+ Arguments.of("timestamp-nanos", HoodieSchema.fromAvroSchema(
+
LogicalTypes.timestampNanos().addToSchema(Schema.create(Schema.Type.LONG))),
+ scalar(b -> b.appendTimestamp(micros))),
+ Arguments.of("local-timestamp-nanos", HoodieSchema.fromAvroSchema(
+
LogicalTypes.localTimestampNanos().addToSchema(Schema.create(Schema.Type.LONG))),
+ scalar(b -> b.appendTimestampNtz(micros))),
+ Arguments.of("time-millis", HoodieSchema.createTimeMillis(), scalar(b
-> b.appendLong(1000))),
+ Arguments.of("time-micros", HoodieSchema.createTimeMicros(), scalar(b
-> b.appendLong(1000))),
+ Arguments.of("enum", HoodieSchema.createEnum("e",
"org.apache.hudi.test", null, Arrays.asList("A", "B")),
+ scalar(b -> b.appendString("A"))),
+ Arguments.of("map",
HoodieSchema.createMap(HoodieSchema.create(HoodieSchemaType.LONG)),
+ scalar(b -> b.appendLong(1))));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("typedValueLeavesSparkCannotShred")
+ void rejectsTypedValueLeafSparkCannotShred(String leaf, HoodieSchema
typedValue, Variant variant) {
+ HoodieException e = assertThrows(HoodieException.class, () ->
shred(variant, shreddedLeaf(typedValue)));
+ assertTrue(e.getMessage().contains("Spark 4 cannot shred"), "unexpected
message: " + e.getMessage());
+ }
+
//
---------------------------------------------------------------------------
// Object and array shapes, including partial (residual) object shredding.
//
---------------------------------------------------------------------------
