wombatu-kun commented on code in PR #18961:
URL: https://github.com/apache/hudi/pull/18961#discussion_r3818615757
##########
hudi-common/src/main/java/org/apache/hudi/common/avro/VariantSchemaUtils.java:
##########
@@ -123,6 +143,228 @@ private static HoodieSchema
stripVariantShreddingAt(HoodieSchema schema) {
return wasNullable ? HoodieSchema.createNullable(replacement) :
replacement;
}
+ /**
+ * Strips {@code typed_value} from top-level fields that have the variant
SHAPE but lost the
+ * variant logical type, i.e. plain records of {@code {metadata: bytes,
value: [nullable]
+ * bytes, typed_value}} (see {@link #isShreddedVariantShape}).
Parquet-footer-derived schemas
+ * come back this way (the converter does not attach the variant logical
type), so
+ * {@link #stripVariantShredding} alone cannot see them. Used by the
table-schema footer
+ * fallback only; returns {@code schema} as-is when nothing matches.
+ *
+ * <p>Unlike {@link #isShreddedVariantTarget}, the match here has NO
requested-side anchor:
+ * the footer fallback runs precisely when no table schema is available to
anchor on, so a
+ * plain user struct that happens to have exactly this shape is stripped too
(a documented,
+ * accepted false positive: {@code metadata} plus {@code typed_value} is the
variant spec's
+ * vocabulary). Top-level fields only, matching the scope of
+ * {@link #getInferableVariantColumns}: inference never shreds a nested
variant, and the
+ * forced-shredding hooks of both write supports are top-level too.
+ *
+ * <p>The shape check also admits the spec's two-field {@code {metadata,
typed_value}} form (a
+ * writer may omit {@code value} when every row is typed). Stripping that
would leave a
+ * one-field record, so {@code value} is restored as nullable bytes: the
result is always the
+ * unshredded {@code {metadata, value}} shape.
+ */
+ public static HoodieSchema stripVariantShreddingByShape(HoodieSchema schema)
{
+ if (schema.getType() != HoodieSchemaType.RECORD) {
+ return schema;
+ }
+
+ List<HoodieSchemaField> newFields = new ArrayList<>();
+ boolean changed = false;
+
+ for (HoodieSchemaField field : schema.getFields()) {
+ HoodieSchema fieldSchema = field.schema();
+ boolean wasNullable = fieldSchema.isNullable();
+ HoodieSchema unwrapped = wasNullable ? fieldSchema.getNonNullType() :
fieldSchema;
+
+ if (isShreddedVariantShape(unwrapped)) {
+ List<HoodieSchemaField> strippedFields = new ArrayList<>();
+ for (HoodieSchemaField member : unwrapped.getFields()) {
+ if
(!HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD.equals(member.name())) {
+ strippedFields.add(HoodieSchemaUtils.createNewSchemaField(member));
+ }
+ }
+ if
(!unwrapped.getField(HoodieSchema.Variant.VARIANT_VALUE_FIELD).isPresent()) {
+ strippedFields.add(HoodieSchemaField.of(
+ HoodieSchema.Variant.VARIANT_VALUE_FIELD,
HoodieSchema.createNullable(HoodieSchemaType.BYTES)));
Review Comment:
The synthesized `value` uses the two-arg `HoodieSchemaField.of` and so
carries no default, while every other nullable field in a footer-derived schema
is built with `NULL_VALUE` by
`AvroSchemaConverterWithTimestampNTZ.convertFields`. Passing
`HoodieSchema.NULL_VALUE` here would keep the resolved schema internally
consistent.
##########
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/storage/TestSparkVariantSampleExtractor.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io.storage;
+
+import
org.apache.hudi.common.avro.VariantShreddingSchemaInferrer.VariantSample;
+import org.apache.hudi.common.model.HoodieAvroIndexedRecord;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieSparkRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaField;
+import org.apache.hudi.common.schema.HoodieSchemaType;
+
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Properties;
+
+import static java.util.Collections.singletonList;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The branches of {@link SparkVariantSampleExtractor} that need no Spark
adapter: ordinal
+ * resolution, the absent-column and non-row (delete payload) legs, and the
shared-size estimate.
+ * Extraction of a present column goes through {@code
SparkAdapter.extractVariantBinary} and is
+ * covered by the functional inference tests.
+ */
+public class TestSparkVariantSampleExtractor {
+
+ private static final HoodieSchema SCHEMA = HoodieSchema.createRecord("rec",
"ns", null, Arrays.asList(
+ HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.LONG)),
+ HoodieSchemaField.of("v",
HoodieSchema.createNullable(HoodieSchema.createVariant()))));
+ private static final StructType STRUCT_TYPE = new StructType()
+ .add("id", DataTypes.LongType)
+ .add("v", DataTypes.BinaryType);
+ private static final Properties PROPS = new Properties();
+ private static final HoodieKey KEY = new HoodieKey("r1", "p");
+
+ @Test
+ public void testAbsentColumnYieldsNullSample() throws Exception {
+ // "w" is not in the row's StructType: sampled as null (ordinal -1), never
through the adapter.
+ SparkVariantSampleExtractor extractor = new
SparkVariantSampleExtractor(singletonList("w"), STRUCT_TYPE);
+ GenericInternalRow row = new GenericInternalRow(new Object[] {1L, null});
+ VariantSample[] samples = extractor.extract(new HoodieSparkRecord(KEY,
row, STRUCT_TYPE, false), SCHEMA, PROPS);
+ assertEquals(1, samples.length);
+ assertNull(samples[0]);
+ }
+
+ @Test
+ public void testNonRowDataYieldsNullSamples() throws Exception {
+ // A record whose data is not an InternalRow (a delete payload)
contributes one empty slot per column.
+ SparkVariantSampleExtractor extractor = new
SparkVariantSampleExtractor(Arrays.asList("v", "w"), STRUCT_TYPE);
+ GenericRecord avroData = new GenericData.Record(SCHEMA.toAvroSchema());
+ avroData.put("id", 1L);
+ VariantSample[] samples = extractor.extract(new
HoodieAvroIndexedRecord(KEY, avroData), SCHEMA, PROPS);
+ assertEquals(2, samples.length);
+ assertNull(samples[0]);
+ assertNull(samples[1]);
+ }
+
+ @Test
+ public void testSharedSizeIsTheStructType() {
+ // Every Spark record of the file references the (cached) StructType; the
decorator subtracts
+ // it from each record's size estimate so the byte cap is not consumed by
the schema.
+ SparkVariantSampleExtractor extractor = new
SparkVariantSampleExtractor(singletonList("v"), STRUCT_TYPE);
+ assertTrue(extractor.sharedSizeEstimate(SCHEMA) > 0);
+ assertEquals(extractor.sharedSizeEstimate(SCHEMA),
extractor.sharedSizeEstimate(SCHEMA));
Review Comment:
This compares `sharedSizeEstimate` with itself, so
`testSharedSizeIsTheStructType` passes even when the extractor measures the
wrong object. `TestAvroVariantSampleExtractor` asserts against
`ObjectSizeCalculator.getObjectSize(...)` directly - the same shape works here
with `STRUCT_TYPE`.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/marker/MarkerCreationStatus.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.table.marker;
+
+/**
+ * Outcome of a marker creation request served by the timeline server.
+ *
+ * <p>The timeline server used to answer marker creation with a bare boolean,
which collapsed three
+ * distinct outcomes into one {@code false}. Callers have to tell them apart:
an already existing
+ * marker is a documented no-op for {@code WriteMarkers#createIfNotExists},
while a detected
+ * conflict has to abort the write.
+ */
+public enum MarkerCreationStatus {
Review Comment:
`MarkerCreationStatus` is referenced by nothing in this PR and nothing on
master, and its javadoc describes a timeline-server marker refactor unrelated
to variant shredding. Did it get carried over from another branch by accident?
##########
hudi-common/src/main/java/org/apache/hudi/core/io/storage/VariantShreddingInferenceFileWriter.java:
##########
@@ -0,0 +1,323 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.core.io.storage;
+
+import org.apache.hudi.common.avro.VariantShreddingSchemaInferrer;
+import
org.apache.hudi.common.avro.VariantShreddingSchemaInferrer.VariantSample;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.util.DefaultSizeEstimator;
+import org.apache.hudi.common.util.SizeEstimator;
+import org.apache.hudi.exception.HoodieIOException;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * A {@link HoodieFileWriter} decorator that infers a per-file variant
shredding schema from the
+ * first records before opening the real parquet writer.
+ *
+ * <p>Records are buffered (and their variant binaries sampled) until a
threshold is reached or
+ * the writer closes, the sampled binaries are fed to a {@link
VariantShreddingSchemaInferrer},
+ * the real writer is created against the schema with the inferred typed_value
spliced in, and
+ * the buffer is replayed in arrival order. Replay reproduces each call
exactly (write vs
+ * writeWithMetadata), so commit seqnos, bloom filters and min/max record keys
come out
+ * identical to a non-buffered write. Buffering thresholds mirror Spark's
+ * {@code ParquetOutputWriterWithVariantShredding} (4096 rows / 64MB).
+ *
+ * <p>Buffered records are {@link HoodieRecord#copy() copied} because Spark
iterators reuse row
+ * instances, then handed to {@link VariantSampleExtractor#prepare} so an
extractor that has to
+ * materialize the record to sample it (the Avro one deserializes
payload-backed records) can
+ * return the materialized form for buffering and the replay does not repeat
that work. For
+ * record types where copy() is identity (Avro), replay additionally relies on
writer-level
+ * records being freshly allocated per record, which holds today; variant
samples are extracted
+ * eagerly into immutable byte arrays so inference itself never depends on it.
+ *
+ * <p>Inference failures never fail the write: the file falls back to
unshredded variants. This
+ * deliberately diverges from Spark (which propagates inference failures)
because a throwing
+ * inference would fail compaction. Writer-creation or replay failures,
however, are latched and
+ * rethrown from every subsequent call including {@link #close()}, so a task
cannot silently
+ * drop buffered records that the handle already counted as written.
+ *
+ * <p>{@link #writeRow} carries neither a {@link HoodieRecord} nor a schema to
sample from, so the
+ * first such call materializes the real writer with whatever has been sampled
so far (unshredded
+ * when nothing has) and passes the row straight through. Footer metadata
added before
+ * materialization is queued and handed to the real writer once it exists;
parquet only consumes
+ * it at close, so nothing is lost.
+ *
+ * <p>Single-threaded by contract, same as the writers it wraps.
+ *
+ * <p>See https://github.com/apache/hudi/issues/18937.</p>
+ *
+ * @param <T> the engine-native record type of the wrapped writer
+ */
+@Slf4j
+public class VariantShreddingInferenceFileWriter<T> implements
HoodieFileWriter<T> {
+
+ /** Buffer caps mirroring Spark's ParquetOutputWriterWithVariantShredding. */
+ public static final int MAX_BUFFERED_RECORDS = 4096;
+ public static final long MAX_BUFFERED_BYTES = 64L * 1024 * 1024;
+ private static final int SIZE_ESTIMATE_INTERVAL = 100;
+
+ /**
+ * Extracts the variant binaries of the inferable columns from a record.
Bound to the writer
+ * schema and column set by the creating factory; must defensively copy the
bytes.
+ */
+ @FunctionalInterface
+ public interface VariantSampleExtractor {
+ VariantSample[] extract(HoodieRecord record, HoodieSchema schema,
Properties props) throws IOException;
+
+ /**
+ * Returns the record to buffer for replay; {@link #extract} is then
called with that record.
+ * An extractor that must materialize the record to sample it returns the
materialized form,
+ * so the replay does not redo the work. Defaults to the record itself.
+ */
+ default HoodieRecord prepare(HoodieRecord record, HoodieSchema schema,
Properties props) throws IOException {
+ return record;
+ }
+
+ /**
+ * Bytes of state that every buffered record references but that is shared
across them, so a
+ * deep object-size walk of one record counts it in full: the Avro {@code
Schema} graph of an
+ * Avro record, the {@code StructType} of a Spark row. Subtracted from
each record's size
+ * estimate so the byte cap budgets record payload rather than the schema
times the record
+ * count (the HUDI-9499 class of over-estimate, which would shrink the
inference sample to a
+ * fraction of the intended 4096 rows). Defaults to 0.
+ */
+ default long sharedSizeEstimate(HoodieSchema schema) {
+ return 0;
+ }
+ }
+
+ /** Creates the real file writer once the inferred typed_value schemas are
known. */
+ @FunctionalInterface
+ public interface InferredWriterFactory<T> {
+ HoodieFileWriter<T> create(Map<String, HoodieSchema> inferredTypedValues)
throws IOException;
+ }
+
+ private final List<String> variantColumns;
+ private final VariantSampleExtractor extractor;
+ private final VariantShreddingSchemaInferrer inferrer;
+ private final InferredWriterFactory<T> writerFactory;
+ private final long maxBufferedBytes;
+ private final SizeEstimator<HoodieRecord> sizeEstimator = new
DefaultSizeEstimator<>();
+
+ private final List<BufferedWrite> buffer = new ArrayList<>();
+ private final List<VariantSample[]> samples = new ArrayList<>();
+ private final Map<String, String> pendingFooterMetadata = new
LinkedHashMap<>();
+ private long estimatedRecordSize = 0;
+ private long bufferedBytes = 0;
+ private HoodieFileWriter<T> delegate;
+ private IOException fatalFailure;
+ private boolean closed = false;
+
+ public VariantShreddingInferenceFileWriter(List<String> variantColumns,
+ VariantSampleExtractor extractor,
+ VariantShreddingSchemaInferrer
inferrer,
+ InferredWriterFactory<T>
writerFactory,
+ long maxFileSize) {
+ this.variantColumns = variantColumns;
+ this.extractor = extractor;
+ this.inferrer = inferrer;
+ this.writerFactory = writerFactory;
+ this.maxBufferedBytes = Math.min(MAX_BUFFERED_BYTES, Math.max(1,
maxFileSize));
+ }
+
+ @Override
+ public boolean canWrite() {
+ // Nothing has been physically written while buffering, so size-based
rollover cannot apply yet.
+ return delegate == null || delegate.canWrite();
+ }
+
+ @Override
+ public void writeWithMetadata(HoodieKey key, HoodieRecord record,
HoodieSchema schema, Properties props) throws IOException {
+ rethrowIfFailed();
+ if (delegate != null) {
+ delegate.writeWithMetadata(key, record, schema, props);
+ } else {
+ buffer(true, key, null, record, schema, props);
+ }
+ }
+
+ @Override
+ public void write(String recordKey, HoodieRecord record, HoodieSchema
schema, Properties props) throws IOException {
+ rethrowIfFailed();
+ if (delegate != null) {
+ delegate.write(recordKey, record, schema, props);
+ } else {
+ buffer(false, null, recordKey, record, schema, props);
+ }
+ }
+
+ @Override
+ public void writeRow(String recordKey, T record) throws IOException {
+ rethrowIfFailed();
+ // No HoodieRecord or schema to sample from: materialize with what has
been sampled so far and
+ // pass the row through. Today only the native log-format delete writer
takes this path, and its
+ // delete schema has no variant column to infer for.
+ materialize();
+ delegate.writeRow(recordKey, record);
+ }
+
+ @Override
+ public void addFooterMetadata(Map<String, String> footerMetadata) {
+ if (delegate != null) {
+ delegate.addFooterMetadata(footerMetadata);
+ } else {
+ // Footer metadata is only consumed at close, so it can wait for the
real writer.
+ pendingFooterMetadata.putAll(footerMetadata);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ boolean delegateClosed = false;
+ try {
+ rethrowIfFailed();
+ // Materialize even with an empty buffer: handles expect the file to
exist at close.
+ materialize();
+ // Mark before close() so a throwing delegate.close() surfaces, not
retried in the catch.
+ delegateClosed = true;
+ delegate.close();
+ } catch (IOException | RuntimeException e) {
+ if (delegate != null && !delegateClosed) {
+ try {
+ delegate.close();
+ } catch (Exception suppressed) {
+ // Best-effort cleanup; surface the original failure.
Review Comment:
The cleanup-close failure is dropped entirely here - neither logged nor
suppressed onto the primary - in a `@Slf4j` class. `hudi-common` already has
`CloseableUtils.closeSuppressing(AutoCloseable, Throwable)` for exactly this.
##########
hudi-common/src/main/java/org/apache/hudi/core/io/storage/VariantShreddingInferenceFileWriter.java:
##########
@@ -0,0 +1,323 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.core.io.storage;
+
+import org.apache.hudi.common.avro.VariantShreddingSchemaInferrer;
+import
org.apache.hudi.common.avro.VariantShreddingSchemaInferrer.VariantSample;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.util.DefaultSizeEstimator;
+import org.apache.hudi.common.util.SizeEstimator;
+import org.apache.hudi.exception.HoodieIOException;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * A {@link HoodieFileWriter} decorator that infers a per-file variant
shredding schema from the
+ * first records before opening the real parquet writer.
+ *
+ * <p>Records are buffered (and their variant binaries sampled) until a
threshold is reached or
+ * the writer closes, the sampled binaries are fed to a {@link
VariantShreddingSchemaInferrer},
+ * the real writer is created against the schema with the inferred typed_value
spliced in, and
+ * the buffer is replayed in arrival order. Replay reproduces each call
exactly (write vs
+ * writeWithMetadata), so commit seqnos, bloom filters and min/max record keys
come out
+ * identical to a non-buffered write. Buffering thresholds mirror Spark's
+ * {@code ParquetOutputWriterWithVariantShredding} (4096 rows / 64MB).
+ *
+ * <p>Buffered records are {@link HoodieRecord#copy() copied} because Spark
iterators reuse row
+ * instances, then handed to {@link VariantSampleExtractor#prepare} so an
extractor that has to
+ * materialize the record to sample it (the Avro one deserializes
payload-backed records) can
+ * return the materialized form for buffering and the replay does not repeat
that work. For
+ * record types where copy() is identity (Avro), replay additionally relies on
writer-level
+ * records being freshly allocated per record, which holds today; variant
samples are extracted
+ * eagerly into immutable byte arrays so inference itself never depends on it.
+ *
+ * <p>Inference failures never fail the write: the file falls back to
unshredded variants. This
+ * deliberately diverges from Spark (which propagates inference failures)
because a throwing
+ * inference would fail compaction. Writer-creation or replay failures,
however, are latched and
+ * rethrown from every subsequent call including {@link #close()}, so a task
cannot silently
+ * drop buffered records that the handle already counted as written.
+ *
+ * <p>{@link #writeRow} carries neither a {@link HoodieRecord} nor a schema to
sample from, so the
+ * first such call materializes the real writer with whatever has been sampled
so far (unshredded
+ * when nothing has) and passes the row straight through. Footer metadata
added before
+ * materialization is queued and handed to the real writer once it exists;
parquet only consumes
+ * it at close, so nothing is lost.
+ *
+ * <p>Single-threaded by contract, same as the writers it wraps.
+ *
+ * <p>See https://github.com/apache/hudi/issues/18937.</p>
+ *
+ * @param <T> the engine-native record type of the wrapped writer
+ */
+@Slf4j
+public class VariantShreddingInferenceFileWriter<T> implements
HoodieFileWriter<T> {
+
+ /** Buffer caps mirroring Spark's ParquetOutputWriterWithVariantShredding. */
+ public static final int MAX_BUFFERED_RECORDS = 4096;
+ public static final long MAX_BUFFERED_BYTES = 64L * 1024 * 1024;
+ private static final int SIZE_ESTIMATE_INTERVAL = 100;
+
+ /**
+ * Extracts the variant binaries of the inferable columns from a record.
Bound to the writer
+ * schema and column set by the creating factory; must defensively copy the
bytes.
+ */
+ @FunctionalInterface
+ public interface VariantSampleExtractor {
+ VariantSample[] extract(HoodieRecord record, HoodieSchema schema,
Properties props) throws IOException;
+
+ /**
+ * Returns the record to buffer for replay; {@link #extract} is then
called with that record.
+ * An extractor that must materialize the record to sample it returns the
materialized form,
+ * so the replay does not redo the work. Defaults to the record itself.
+ */
+ default HoodieRecord prepare(HoodieRecord record, HoodieSchema schema,
Properties props) throws IOException {
+ return record;
+ }
+
+ /**
+ * Bytes of state that every buffered record references but that is shared
across them, so a
+ * deep object-size walk of one record counts it in full: the Avro {@code
Schema} graph of an
+ * Avro record, the {@code StructType} of a Spark row. Subtracted from
each record's size
+ * estimate so the byte cap budgets record payload rather than the schema
times the record
+ * count (the HUDI-9499 class of over-estimate, which would shrink the
inference sample to a
+ * fraction of the intended 4096 rows). Defaults to 0.
+ */
+ default long sharedSizeEstimate(HoodieSchema schema) {
+ return 0;
+ }
+ }
+
+ /** Creates the real file writer once the inferred typed_value schemas are
known. */
+ @FunctionalInterface
+ public interface InferredWriterFactory<T> {
+ HoodieFileWriter<T> create(Map<String, HoodieSchema> inferredTypedValues)
throws IOException;
+ }
+
+ private final List<String> variantColumns;
+ private final VariantSampleExtractor extractor;
+ private final VariantShreddingSchemaInferrer inferrer;
+ private final InferredWriterFactory<T> writerFactory;
+ private final long maxBufferedBytes;
+ private final SizeEstimator<HoodieRecord> sizeEstimator = new
DefaultSizeEstimator<>();
+
+ private final List<BufferedWrite> buffer = new ArrayList<>();
+ private final List<VariantSample[]> samples = new ArrayList<>();
+ private final Map<String, String> pendingFooterMetadata = new
LinkedHashMap<>();
+ private long estimatedRecordSize = 0;
+ private long bufferedBytes = 0;
+ private HoodieFileWriter<T> delegate;
+ private IOException fatalFailure;
+ private boolean closed = false;
+
+ public VariantShreddingInferenceFileWriter(List<String> variantColumns,
+ VariantSampleExtractor extractor,
+ VariantShreddingSchemaInferrer
inferrer,
+ InferredWriterFactory<T>
writerFactory,
+ long maxFileSize) {
+ this.variantColumns = variantColumns;
+ this.extractor = extractor;
+ this.inferrer = inferrer;
+ this.writerFactory = writerFactory;
+ this.maxBufferedBytes = Math.min(MAX_BUFFERED_BYTES, Math.max(1,
maxFileSize));
+ }
+
+ @Override
+ public boolean canWrite() {
+ // Nothing has been physically written while buffering, so size-based
rollover cannot apply yet.
+ return delegate == null || delegate.canWrite();
+ }
+
+ @Override
+ public void writeWithMetadata(HoodieKey key, HoodieRecord record,
HoodieSchema schema, Properties props) throws IOException {
+ rethrowIfFailed();
+ if (delegate != null) {
+ delegate.writeWithMetadata(key, record, schema, props);
+ } else {
+ buffer(true, key, null, record, schema, props);
+ }
+ }
+
+ @Override
+ public void write(String recordKey, HoodieRecord record, HoodieSchema
schema, Properties props) throws IOException {
+ rethrowIfFailed();
+ if (delegate != null) {
+ delegate.write(recordKey, record, schema, props);
+ } else {
+ buffer(false, null, recordKey, record, schema, props);
+ }
+ }
+
+ @Override
+ public void writeRow(String recordKey, T record) throws IOException {
+ rethrowIfFailed();
+ // No HoodieRecord or schema to sample from: materialize with what has
been sampled so far and
+ // pass the row through. Today only the native log-format delete writer
takes this path, and its
+ // delete schema has no variant column to infer for.
+ materialize();
+ delegate.writeRow(recordKey, record);
+ }
+
+ @Override
+ public void addFooterMetadata(Map<String, String> footerMetadata) {
+ if (delegate != null) {
+ delegate.addFooterMetadata(footerMetadata);
+ } else {
+ // Footer metadata is only consumed at close, so it can wait for the
real writer.
+ pendingFooterMetadata.putAll(footerMetadata);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ boolean delegateClosed = false;
+ try {
+ rethrowIfFailed();
+ // Materialize even with an empty buffer: handles expect the file to
exist at close.
+ materialize();
+ // Mark before close() so a throwing delegate.close() surfaces, not
retried in the catch.
+ delegateClosed = true;
+ delegate.close();
+ } catch (IOException | RuntimeException e) {
+ if (delegate != null && !delegateClosed) {
+ try {
+ delegate.close();
+ } catch (Exception suppressed) {
+ // Best-effort cleanup; surface the original failure.
+ }
+ }
+ throw e;
+ }
+ }
+
+ @Override
+ public Object getFileFormatMetadata() {
+ try {
+ rethrowIfFailed();
+ materialize();
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to materialize the parquet writer
for format metadata", e);
+ }
+ return delegate.getFileFormatMetadata();
+ }
+
+ private void buffer(boolean withMetadata, HoodieKey key, String recordKey,
HoodieRecord record,
+ HoodieSchema schema, Properties props) throws
IOException {
+ rethrowIfFailed();
+ HoodieRecord buffered = extractor.prepare(record.copy(), schema, props);
Review Comment:
`copy()` returns `this` for both `HoodieAvroRecord` and `HoodieSparkRecord`,
so on the SPARK path (no `prepare()` override) the buffer holds the handle's
own record, which `HoodieWriteMergeHandle` deflates right after this call
whenever the file-name meta field is not populated. Should
`SparkVariantSampleExtractor` override `prepare()` to hand back a detached
record the way the Avro one does?
##########
hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/hudi/variant/Spark42VariantShreddingSchemaInferrer.scala:
##########
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.variant
+
+import org.apache.hudi.HoodieSchemaConversionUtils
+import org.apache.hudi.common.avro.VariantShreddingSchemaInferrer
+import org.apache.hudi.common.avro.VariantShreddingSchemaInferrer.VariantSample
+import org.apache.hudi.common.schema.HoodieSchema
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
+import
org.apache.spark.sql.execution.datasources.parquet.InferVariantShreddingSchema
+import org.apache.spark.sql.types.{ArrayType, DataType, StructField,
StructType, VariantType}
+import org.apache.spark.unsafe.types.VariantVal
+
+import java.{util => ju}
+
+import scala.jdk.CollectionConverters.ListHasAsScala
+
+/**
+ * Infers per-file variant shredding schemas by delegating to Spark's
+ * [[InferVariantShreddingSchema]] (SPARK-53659), so Hudi inherits Spark's
merge and
+ * finalization heuristics verbatim (field-frequency dropping, type widening,
width/depth caps).
+ *
+ * Loaded reflectively from hudi-common via classpath detection; the Spark 4.2
twin of the
+ * spark4.1 module's inferrer (each spark4.x profile builds only its own
version module, and the
+ * Spark class does not exist before 4.1).
+ */
+class Spark42VariantShreddingSchemaInferrer extends
VariantShreddingSchemaInferrer {
+
+ private val typedValueField = HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD
+ // Avro identifier shape; the Hudi schema the result splices into is
Avro-backed.
+ private val avroNamePattern = "[A-Za-z_][A-Za-z0-9_]*".r.pattern
+
+ override def inferTypedValueSchemas(columnNames: ju.List[String],
+ rowSamples:
ju.List[Array[VariantSample]]): ju.Map[String, HoodieSchema] = {
+ val names = columnNames.asScala.toSeq
+ val inputSchema = StructType(names.map(name => StructField(name,
VariantType, nullable = true)))
+ val rows: Seq[InternalRow] = rowSamples.asScala.map { row =>
+ val values = new Array[Any](row.length)
+ var i = 0
+ while (i < row.length) {
+ if (row(i) != null) {
+ values(i) = new VariantVal(row(i).getValue, row(i).getMetadata)
+ }
+ i += 1
+ }
+ new GenericInternalRow(values): InternalRow
+ }.toSeq
+
+ // One call covers all variant columns of the file: Spark's max-width
budget is global
+ // across the schema, and per-column calls would skew it.
+ val inferred = new
InferVariantShreddingSchema(inputSchema).inferSchema(rows)
Review Comment:
`InferVariantShreddingSchema` resolves
`spark.sql.variant.shredding.maxSchemaWidth` and `maxSchemaDepth` from
`SQLConf.get()` in its constructor, which on an executor task thread outside a
SQL execution context is a fresh fallback conf, so a user override is honored
under bulk-insert but silently dropped during compaction.
`HoodieRowParquetWriteSupport.resolveSessionLocalTimeZone` already has the
SparkConf-fallback pattern for this - worth reusing with
`SQLConf.withExistingConf`?
##########
hudi-common/src/main/java/org/apache/hudi/common/avro/VariantShreddingRuntime.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.common.avro;
+
+import org.apache.hudi.common.util.Option;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Classpath detection of engine-specific variant shredding components.
+ *
+ * <p>Engine modules (currently the Spark 4.x bundles) ship implementations of
+ * {@link VariantShreddingProvider} and {@link
VariantShreddingSchemaInferrer}; hudi-common
+ * discovers them by probing well-known class names so that it stays free of
engine
+ * dependencies. Probes are memoized: classpath content does not change within
a JVM.</p>
+ */
+@Slf4j
+public final class VariantShreddingRuntime {
+
+ /** Provider candidates, most specific first. Mirrors what each Spark bundle
ships. */
+ private static final String[] PROVIDER_CANDIDATES = {
+ "org.apache.hudi.variant.Spark4VariantShreddingProvider"
+ };
+
+ /**
+ * Inferrer candidates, one per Spark version module that ships one
(inference exists only in
+ * Spark 4.1+, SPARK-53659), most recent first. Each spark4.x profile builds
only its own
+ * version module, so every version that should infer needs its own entry
here: a runtime whose
+ * module is missing from this list silently writes unshredded.
+ */
+ private static final String[] INFERRER_CANDIDATES = {
+ "org.apache.hudi.variant.Spark42VariantShreddingSchemaInferrer",
+ "org.apache.hudi.variant.Spark41VariantShreddingSchemaInferrer"
+ };
+
+ private static final Option<String> PROVIDER_CLASS =
probe(PROVIDER_CANDIDATES);
+ private static final Option<VariantShreddingSchemaInferrer> INFERRER =
loadInferrer();
+
+ private VariantShreddingRuntime() {
+ }
+
+ /**
+ * The fully-qualified name of the first {@link VariantShreddingProvider}
implementation
+ * found on the classpath, if any.
+ */
+ public static Option<String> getProviderClass() {
+ return PROVIDER_CLASS;
+ }
+
+ /**
+ * A shared {@link VariantShreddingSchemaInferrer} instance from the
classpath, if any.
+ * Implementations are stateless and thread-safe by contract, so one
instance is shared.
+ * Tests also use this as the capability probe to filter inference tests to
classpaths
+ * that ship an inferrer.
+ */
+ public static Option<VariantShreddingSchemaInferrer> lookupInferrer() {
+ return INFERRER;
+ }
+
+ /**
+ * Both probes run from this class's static initializer, so nothing here may
let an error
+ * escape: an escaping {@link LinkageError} would fail {@code <clinit>} and
every later use
+ * (including {@link #getProviderClass()} on the main Avro write path) would
see a bare
+ * "Could not initialize class" with the original cause lost. Candidates are
therefore loaded
+ * WITHOUT initialization (no static initializer of theirs runs here), and
every
Review Comment:
"Both probes ... loaded WITHOUT initialization" does not hold for
`loadInferrer`, which calls `getDeclaredConstructor().newInstance()` and so
runs the candidate's static initializer. Either scope that sentence to `probe`
and note that `loadInferrer` initializes but latches, or widen `loadInferrer`'s
catch to `Throwable`.
##########
hudi-common/src/main/java/org/apache/hudi/common/avro/VariantSchemaUtils.java:
##########
@@ -123,6 +143,228 @@ private static HoodieSchema
stripVariantShreddingAt(HoodieSchema schema) {
return wasNullable ? HoodieSchema.createNullable(replacement) :
replacement;
}
+ /**
+ * Strips {@code typed_value} from top-level fields that have the variant
SHAPE but lost the
+ * variant logical type, i.e. plain records of {@code {metadata: bytes,
value: [nullable]
+ * bytes, typed_value}} (see {@link #isShreddedVariantShape}).
Parquet-footer-derived schemas
+ * come back this way (the converter does not attach the variant logical
type), so
+ * {@link #stripVariantShredding} alone cannot see them. Used by the
table-schema footer
+ * fallback only; returns {@code schema} as-is when nothing matches.
+ *
+ * <p>Unlike {@link #isShreddedVariantTarget}, the match here has NO
requested-side anchor:
+ * the footer fallback runs precisely when no table schema is available to
anchor on, so a
+ * plain user struct that happens to have exactly this shape is stripped too
(a documented,
+ * accepted false positive: {@code metadata} plus {@code typed_value} is the
variant spec's
+ * vocabulary). Top-level fields only, matching the scope of
+ * {@link #getInferableVariantColumns}: inference never shreds a nested
variant, and the
+ * forced-shredding hooks of both write supports are top-level too.
Review Comment:
"the forced-shredding hooks of both write supports are top-level too" does
not hold for the row writer:
`HoodieRowParquetWriteSupport.processNestedDataType` recurses into structs,
arrays and maps and `generateShreddedSchema` re-reads the forced DDL on every
entry, which the `swapShreddedVariantFields` javadoc in this same file spells
out. Could the claim be scoped to
`HoodieAvroWriteSupport.applyForcedShreddingSchema`?
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java:
##########
@@ -298,6 +298,23 @@ public class HoodieStorageConfig extends HoodieConfig {
+ "The provider parses variant binary data and populates typed_value
columns. "
+ "When not set, the provider is auto-detected from the classpath.");
+ public static final ConfigProperty<Boolean>
PARQUET_VARIANT_SHREDDING_SCHEMA_INFERENCE_ENABLED = ConfigProperty
+ .key("hoodie.parquet.variant.shredding.schema.inference.enabled")
+ .defaultValue(false)
Review Comment:
The body carries `Closes #18937` and also says the default flip is a
follow-up tracked in #18937, whose task list still holds that unticked box.
Should #18937 become a plain reference so merging does not close it, leaving
only `Closes #18038`?
##########
hudi-common/src/main/java/org/apache/hudi/core/io/storage/VariantShreddingInferenceFileWriter.java:
##########
@@ -0,0 +1,323 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.core.io.storage;
+
+import org.apache.hudi.common.avro.VariantShreddingSchemaInferrer;
+import
org.apache.hudi.common.avro.VariantShreddingSchemaInferrer.VariantSample;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.util.DefaultSizeEstimator;
+import org.apache.hudi.common.util.SizeEstimator;
+import org.apache.hudi.exception.HoodieIOException;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * A {@link HoodieFileWriter} decorator that infers a per-file variant
shredding schema from the
+ * first records before opening the real parquet writer.
+ *
+ * <p>Records are buffered (and their variant binaries sampled) until a
threshold is reached or
+ * the writer closes, the sampled binaries are fed to a {@link
VariantShreddingSchemaInferrer},
+ * the real writer is created against the schema with the inferred typed_value
spliced in, and
+ * the buffer is replayed in arrival order. Replay reproduces each call
exactly (write vs
+ * writeWithMetadata), so commit seqnos, bloom filters and min/max record keys
come out
+ * identical to a non-buffered write. Buffering thresholds mirror Spark's
+ * {@code ParquetOutputWriterWithVariantShredding} (4096 rows / 64MB).
+ *
+ * <p>Buffered records are {@link HoodieRecord#copy() copied} because Spark
iterators reuse row
+ * instances, then handed to {@link VariantSampleExtractor#prepare} so an
extractor that has to
+ * materialize the record to sample it (the Avro one deserializes
payload-backed records) can
+ * return the materialized form for buffering and the replay does not repeat
that work. For
+ * record types where copy() is identity (Avro), replay additionally relies on
writer-level
+ * records being freshly allocated per record, which holds today; variant
samples are extracted
+ * eagerly into immutable byte arrays so inference itself never depends on it.
+ *
+ * <p>Inference failures never fail the write: the file falls back to
unshredded variants. This
+ * deliberately diverges from Spark (which propagates inference failures)
because a throwing
+ * inference would fail compaction. Writer-creation or replay failures,
however, are latched and
+ * rethrown from every subsequent call including {@link #close()}, so a task
cannot silently
+ * drop buffered records that the handle already counted as written.
+ *
+ * <p>{@link #writeRow} carries neither a {@link HoodieRecord} nor a schema to
sample from, so the
+ * first such call materializes the real writer with whatever has been sampled
so far (unshredded
+ * when nothing has) and passes the row straight through. Footer metadata
added before
+ * materialization is queued and handed to the real writer once it exists;
parquet only consumes
+ * it at close, so nothing is lost.
+ *
+ * <p>Single-threaded by contract, same as the writers it wraps.
+ *
+ * <p>See https://github.com/apache/hudi/issues/18937.</p>
+ *
+ * @param <T> the engine-native record type of the wrapped writer
+ */
+@Slf4j
+public class VariantShreddingInferenceFileWriter<T> implements
HoodieFileWriter<T> {
+
+ /** Buffer caps mirroring Spark's ParquetOutputWriterWithVariantShredding. */
+ public static final int MAX_BUFFERED_RECORDS = 4096;
+ public static final long MAX_BUFFERED_BYTES = 64L * 1024 * 1024;
+ private static final int SIZE_ESTIMATE_INTERVAL = 100;
+
+ /**
+ * Extracts the variant binaries of the inferable columns from a record.
Bound to the writer
+ * schema and column set by the creating factory; must defensively copy the
bytes.
+ */
+ @FunctionalInterface
+ public interface VariantSampleExtractor {
+ VariantSample[] extract(HoodieRecord record, HoodieSchema schema,
Properties props) throws IOException;
+
+ /**
+ * Returns the record to buffer for replay; {@link #extract} is then
called with that record.
+ * An extractor that must materialize the record to sample it returns the
materialized form,
+ * so the replay does not redo the work. Defaults to the record itself.
+ */
+ default HoodieRecord prepare(HoodieRecord record, HoodieSchema schema,
Properties props) throws IOException {
+ return record;
+ }
+
+ /**
+ * Bytes of state that every buffered record references but that is shared
across them, so a
+ * deep object-size walk of one record counts it in full: the Avro {@code
Schema} graph of an
+ * Avro record, the {@code StructType} of a Spark row. Subtracted from
each record's size
+ * estimate so the byte cap budgets record payload rather than the schema
times the record
+ * count (the HUDI-9499 class of over-estimate, which would shrink the
inference sample to a
+ * fraction of the intended 4096 rows). Defaults to 0.
+ */
+ default long sharedSizeEstimate(HoodieSchema schema) {
+ return 0;
+ }
+ }
+
+ /** Creates the real file writer once the inferred typed_value schemas are
known. */
+ @FunctionalInterface
+ public interface InferredWriterFactory<T> {
+ HoodieFileWriter<T> create(Map<String, HoodieSchema> inferredTypedValues)
throws IOException;
+ }
+
+ private final List<String> variantColumns;
+ private final VariantSampleExtractor extractor;
+ private final VariantShreddingSchemaInferrer inferrer;
+ private final InferredWriterFactory<T> writerFactory;
+ private final long maxBufferedBytes;
+ private final SizeEstimator<HoodieRecord> sizeEstimator = new
DefaultSizeEstimator<>();
+
+ private final List<BufferedWrite> buffer = new ArrayList<>();
+ private final List<VariantSample[]> samples = new ArrayList<>();
+ private final Map<String, String> pendingFooterMetadata = new
LinkedHashMap<>();
+ private long estimatedRecordSize = 0;
+ private long bufferedBytes = 0;
+ private HoodieFileWriter<T> delegate;
+ private IOException fatalFailure;
+ private boolean closed = false;
+
+ public VariantShreddingInferenceFileWriter(List<String> variantColumns,
+ VariantSampleExtractor extractor,
+ VariantShreddingSchemaInferrer
inferrer,
+ InferredWriterFactory<T>
writerFactory,
+ long maxFileSize) {
+ this.variantColumns = variantColumns;
+ this.extractor = extractor;
+ this.inferrer = inferrer;
+ this.writerFactory = writerFactory;
+ this.maxBufferedBytes = Math.min(MAX_BUFFERED_BYTES, Math.max(1,
maxFileSize));
+ }
+
+ @Override
+ public boolean canWrite() {
+ // Nothing has been physically written while buffering, so size-based
rollover cannot apply yet.
+ return delegate == null || delegate.canWrite();
+ }
+
+ @Override
+ public void writeWithMetadata(HoodieKey key, HoodieRecord record,
HoodieSchema schema, Properties props) throws IOException {
+ rethrowIfFailed();
+ if (delegate != null) {
+ delegate.writeWithMetadata(key, record, schema, props);
+ } else {
+ buffer(true, key, null, record, schema, props);
+ }
+ }
+
+ @Override
+ public void write(String recordKey, HoodieRecord record, HoodieSchema
schema, Properties props) throws IOException {
+ rethrowIfFailed();
+ if (delegate != null) {
+ delegate.write(recordKey, record, schema, props);
+ } else {
+ buffer(false, null, recordKey, record, schema, props);
+ }
+ }
+
+ @Override
+ public void writeRow(String recordKey, T record) throws IOException {
+ rethrowIfFailed();
+ // No HoodieRecord or schema to sample from: materialize with what has
been sampled so far and
+ // pass the row through. Today only the native log-format delete writer
takes this path, and its
+ // delete schema has no variant column to infer for.
+ materialize();
+ delegate.writeRow(recordKey, record);
+ }
+
+ @Override
+ public void addFooterMetadata(Map<String, String> footerMetadata) {
+ if (delegate != null) {
+ delegate.addFooterMetadata(footerMetadata);
+ } else {
+ // Footer metadata is only consumed at close, so it can wait for the
real writer.
+ pendingFooterMetadata.putAll(footerMetadata);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ boolean delegateClosed = false;
+ try {
+ rethrowIfFailed();
+ // Materialize even with an empty buffer: handles expect the file to
exist at close.
+ materialize();
+ // Mark before close() so a throwing delegate.close() surfaces, not
retried in the catch.
+ delegateClosed = true;
+ delegate.close();
+ } catch (IOException | RuntimeException e) {
+ if (delegate != null && !delegateClosed) {
+ try {
+ delegate.close();
+ } catch (Exception suppressed) {
+ // Best-effort cleanup; surface the original failure.
+ }
+ }
+ throw e;
+ }
+ }
+
+ @Override
+ public Object getFileFormatMetadata() {
+ try {
+ rethrowIfFailed();
+ materialize();
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to materialize the parquet writer
for format metadata", e);
+ }
+ return delegate.getFileFormatMetadata();
+ }
+
+ private void buffer(boolean withMetadata, HoodieKey key, String recordKey,
HoodieRecord record,
+ HoodieSchema schema, Properties props) throws
IOException {
+ rethrowIfFailed();
+ HoodieRecord buffered = extractor.prepare(record.copy(), schema, props);
+ // Eager extraction: immutable byte copies decouple inference from
buffered-record identity,
+ // and per-record extraction failures (corrupt binaries) surface exactly
like an eager write.
+ samples.add(extractor.extract(buffered, schema, props));
+ buffer.add(new BufferedWrite(withMetadata, key, recordKey, buffered,
schema, props));
+ // Re-estimate periodically so a small first record cannot defeat the byte
cap
+ // (same moving-average idiom as ExternalSpillableMap).
+ if (estimatedRecordSize == 0 || buffer.size() % SIZE_ESTIMATE_INTERVAL ==
0) {
+ long sampled = Math.max(1, sizeEstimator.sizeEstimate(buffered) -
extractor.sharedSizeEstimate(schema));
+ estimatedRecordSize = estimatedRecordSize == 0
+ ? sampled : (long) (estimatedRecordSize * 0.9 + sampled * 0.1);
+ }
+ bufferedBytes += estimatedRecordSize;
+ if (buffer.size() >= MAX_BUFFERED_RECORDS || bufferedBytes >=
maxBufferedBytes) {
+ materialize();
+ }
+ }
+
+ private void materialize() throws IOException {
+ if (delegate != null) {
+ return;
+ }
+ try {
+ delegate = writerFactory.create(inferTypedValues());
+ if (!pendingFooterMetadata.isEmpty()) {
+ delegate.addFooterMetadata(pendingFooterMetadata);
+ pendingFooterMetadata.clear();
+ }
+ for (BufferedWrite write : buffer) {
+ if (write.withMetadata) {
+ delegate.writeWithMetadata(write.key, write.record, write.schema,
write.props);
+ } else {
+ delegate.write(write.recordKey, write.record, write.schema,
write.props);
+ }
+ }
+ buffer.clear();
+ samples.clear();
+ } catch (IOException e) {
+ fatalFailure = e;
+ throw e;
+ } catch (RuntimeException e) {
Review Comment:
`materialize()` latches `IOException` and `RuntimeException` but not
`Error`, so an `Error` thrown mid-replay leaves `delegate` non-null with
nothing latched and the next call returns early - `close()` then finishes the
file without the un-replayed records. Since `inferTypedValues` already treats
`LinkageError` as reachable here, should this catch widen to `Throwable`?
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieVariantReconstruction.java:
##########
@@ -268,7 +269,7 @@ private static VariantShreddingProvider
loadProvider(HoodieStorage storage) {
String providerClass = storage.getConf()
Review Comment:
The changelog credits this PR with fixing the Avro "Field already used" in
`stripVariantShredding` and `VariantReconstruction`, but that method's body is
unchanged from master and this file's entire net diff is the `getProviderClass`
rename. Dropping that clause would keep the focus on the two changes that are
here.
##########
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/storage/row/TestVariantShreddingInferenceInternalRowFileWriter.java:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io.storage.row;
+
+import org.apache.hudi.common.avro.VariantShreddingSchemaInferrer;
+import
org.apache.hudi.common.avro.VariantShreddingSchemaInferrer.VariantSample;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaType;
+import org.apache.hudi.common.util.DefaultSizeEstimator;
+import org.apache.hudi.core.io.storage.VariantShreddingInferenceFileWriter;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static java.util.Collections.singletonList;
+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.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The row-writer sibling of {@code TestVariantShreddingInferenceFileWriter}.
The decorator is
+ * driven with ordinals of -1 (the column absent from the row's StructType),
which is the
+ * sampling branch that needs no Spark adapter on the classpath; sample
extraction itself goes
+ * through {@code SparkAdapter.extractVariantBinary} and is covered by the
functional tests.
+ */
+public class TestVariantShreddingInferenceInternalRowFileWriter {
+
+ private static final int[] ABSENT_COLUMN = {-1};
+
+ private static InternalRow row(long id) {
+ return new GenericInternalRow(new Object[] {id});
+ }
+
+ /** A row of about a kilobyte, so byte-cap arithmetic on the estimator is
not dominated by rounding. */
+ private static InternalRow wideRow(long id) {
+ return new GenericInternalRow(new Object[] {id, UTF8String.fromString(new
String(new char[1024]).replace('\0', 'x'))});
+ }
+
+ /** A 16-byte UnsafeRow (8-byte null bitset + one long), so byte-cap
arithmetic is exact. */
+ private static UnsafeRow unsafeRow(long id) {
+ UnsafeRow row = new UnsafeRow(1);
+ byte[] buffer = new byte[16];
+ row.pointTo(buffer, 16);
+ row.setLong(0, id);
+ return row;
+ }
+
+ /** Records every call so replay order and call kinds can be asserted. */
+ private static class RecordingRowWriter implements
HoodieInternalRowFileWriter {
+ private final List<String> calls = new ArrayList<>();
+ private final List<InternalRow> rows = new ArrayList<>();
+ private int closeCount = 0;
+ private IOException failWriteWith;
+
+ @Override
+ public boolean canWrite() {
+ return true;
+ }
+
+ @Override
+ public void writeRow(UTF8String key, InternalRow row) throws IOException {
+ failIfConfigured();
+ calls.add("keyed:" + key);
+ rows.add(row);
+ }
+
+ @Override
+ public void writeRow(InternalRow row) throws IOException {
+ failIfConfigured();
+ calls.add("plain:" + row.getLong(0));
+ rows.add(row);
+ }
+
+ @Override
+ public void close() {
Review Comment:
`RecordingRowWriter.close()` does not declare `throws IOException`, so no
test here can drive `delegate.close()` to throw and the `delegateClosed` guard
is untestable - dropping the flag leaves every test in the class green. Could
`failCloseWith` and `testThrowingDelegateCloseSurfacesAndIsNotRetried` be
ported from `TestVariantShreddingInferenceFileWriter`?
##########
hudi-hadoop-common/src/main/java/org/apache/parquet/avro/AvroSchemaConverterWithTimestampNTZ.java:
##########
@@ -266,9 +289,21 @@ private Type convertField(String fieldName, HoodieSchema
schema, Type.Repetition
case UNION:
return convertUnion(fieldName, schema, repetition, schemaPath);
case VARIANT:
Review Comment:
This stamps the parquet VARIANT logical type on every variant group the AVRO
path writes with no gate on the new config, so on parquet 1.16+ the footer of
existing variant tables changes with inference still off. The Impact section
says "No behavior change unless the new config is enabled" - worth calling the
annotation out there?
##########
hudi-hadoop-common/src/test/java/org/apache/parquet/avro/TestAvroSchemaConverter.java:
##########
@@ -119,6 +119,22 @@ private void testAvroToParquetConversion(Configuration
conf, HoodieSchema schema
assertEquals(expectedMT.toString(), messageType.toString());
}
+ /**
+ * The VARIANT logical type annotation the converter stamps on variant
groups on parquet 1.16+
+ * (Spark 4.1+ profiles), as {@code GroupType.toString()} renders it; empty
on older parquet,
+ * where the annotation type does not exist and groups stay plain. {@code
MessageTypeParser}
+ * cannot parse the annotation, so variant expectations are compared against
the converter's
+ * {@code toString()} directly rather than through {@link
#testAvroToParquetConversion}.
+ */
+ private static String variantAnnotation() {
+ return AvroSchemaConverterWithTimestampNTZ.isVariantLogicalTypeSupported()
? " (VARIANT(1))" : "";
Review Comment:
The expected annotation comes from `isVariantLogicalTypeSupported`, the same
reflective probe the converter uses to decide whether to emit it, so a lookup
that stops resolving flips production and the expectation together and every
assertion stays green. Could this go back to the independent
`LogicalTypeAnnotation.class.getMethod("variantType", byte.class)` probe an
earlier commit here used?
##########
hudi-common/src/main/java/org/apache/hudi/core/io/storage/VariantShreddingInferenceFileWriter.java:
##########
@@ -0,0 +1,323 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.core.io.storage;
+
+import org.apache.hudi.common.avro.VariantShreddingSchemaInferrer;
+import
org.apache.hudi.common.avro.VariantShreddingSchemaInferrer.VariantSample;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.util.DefaultSizeEstimator;
+import org.apache.hudi.common.util.SizeEstimator;
+import org.apache.hudi.exception.HoodieIOException;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * A {@link HoodieFileWriter} decorator that infers a per-file variant
shredding schema from the
+ * first records before opening the real parquet writer.
+ *
+ * <p>Records are buffered (and their variant binaries sampled) until a
threshold is reached or
+ * the writer closes, the sampled binaries are fed to a {@link
VariantShreddingSchemaInferrer},
+ * the real writer is created against the schema with the inferred typed_value
spliced in, and
+ * the buffer is replayed in arrival order. Replay reproduces each call
exactly (write vs
+ * writeWithMetadata), so commit seqnos, bloom filters and min/max record keys
come out
+ * identical to a non-buffered write. Buffering thresholds mirror Spark's
+ * {@code ParquetOutputWriterWithVariantShredding} (4096 rows / 64MB).
+ *
+ * <p>Buffered records are {@link HoodieRecord#copy() copied} because Spark
iterators reuse row
+ * instances, then handed to {@link VariantSampleExtractor#prepare} so an
extractor that has to
+ * materialize the record to sample it (the Avro one deserializes
payload-backed records) can
+ * return the materialized form for buffering and the replay does not repeat
that work. For
+ * record types where copy() is identity (Avro), replay additionally relies on
writer-level
+ * records being freshly allocated per record, which holds today; variant
samples are extracted
+ * eagerly into immutable byte arrays so inference itself never depends on it.
+ *
+ * <p>Inference failures never fail the write: the file falls back to
unshredded variants. This
+ * deliberately diverges from Spark (which propagates inference failures)
because a throwing
+ * inference would fail compaction. Writer-creation or replay failures,
however, are latched and
+ * rethrown from every subsequent call including {@link #close()}, so a task
cannot silently
+ * drop buffered records that the handle already counted as written.
+ *
+ * <p>{@link #writeRow} carries neither a {@link HoodieRecord} nor a schema to
sample from, so the
+ * first such call materializes the real writer with whatever has been sampled
so far (unshredded
+ * when nothing has) and passes the row straight through. Footer metadata
added before
+ * materialization is queued and handed to the real writer once it exists;
parquet only consumes
+ * it at close, so nothing is lost.
+ *
+ * <p>Single-threaded by contract, same as the writers it wraps.
+ *
+ * <p>See https://github.com/apache/hudi/issues/18937.</p>
+ *
+ * @param <T> the engine-native record type of the wrapped writer
+ */
+@Slf4j
+public class VariantShreddingInferenceFileWriter<T> implements
HoodieFileWriter<T> {
+
+ /** Buffer caps mirroring Spark's ParquetOutputWriterWithVariantShredding. */
+ public static final int MAX_BUFFERED_RECORDS = 4096;
+ public static final long MAX_BUFFERED_BYTES = 64L * 1024 * 1024;
+ private static final int SIZE_ESTIMATE_INTERVAL = 100;
+
+ /**
+ * Extracts the variant binaries of the inferable columns from a record.
Bound to the writer
+ * schema and column set by the creating factory; must defensively copy the
bytes.
+ */
+ @FunctionalInterface
+ public interface VariantSampleExtractor {
+ VariantSample[] extract(HoodieRecord record, HoodieSchema schema,
Properties props) throws IOException;
+
+ /**
+ * Returns the record to buffer for replay; {@link #extract} is then
called with that record.
+ * An extractor that must materialize the record to sample it returns the
materialized form,
+ * so the replay does not redo the work. Defaults to the record itself.
+ */
+ default HoodieRecord prepare(HoodieRecord record, HoodieSchema schema,
Properties props) throws IOException {
+ return record;
+ }
+
+ /**
+ * Bytes of state that every buffered record references but that is shared
across them, so a
+ * deep object-size walk of one record counts it in full: the Avro {@code
Schema} graph of an
+ * Avro record, the {@code StructType} of a Spark row. Subtracted from
each record's size
+ * estimate so the byte cap budgets record payload rather than the schema
times the record
+ * count (the HUDI-9499 class of over-estimate, which would shrink the
inference sample to a
+ * fraction of the intended 4096 rows). Defaults to 0.
+ */
+ default long sharedSizeEstimate(HoodieSchema schema) {
+ return 0;
+ }
+ }
+
+ /** Creates the real file writer once the inferred typed_value schemas are
known. */
+ @FunctionalInterface
+ public interface InferredWriterFactory<T> {
+ HoodieFileWriter<T> create(Map<String, HoodieSchema> inferredTypedValues)
throws IOException;
+ }
+
+ private final List<String> variantColumns;
+ private final VariantSampleExtractor extractor;
+ private final VariantShreddingSchemaInferrer inferrer;
+ private final InferredWriterFactory<T> writerFactory;
+ private final long maxBufferedBytes;
+ private final SizeEstimator<HoodieRecord> sizeEstimator = new
DefaultSizeEstimator<>();
+
+ private final List<BufferedWrite> buffer = new ArrayList<>();
+ private final List<VariantSample[]> samples = new ArrayList<>();
+ private final Map<String, String> pendingFooterMetadata = new
LinkedHashMap<>();
+ private long estimatedRecordSize = 0;
+ private long bufferedBytes = 0;
+ private HoodieFileWriter<T> delegate;
+ private IOException fatalFailure;
+ private boolean closed = false;
+
+ public VariantShreddingInferenceFileWriter(List<String> variantColumns,
+ VariantSampleExtractor extractor,
+ VariantShreddingSchemaInferrer
inferrer,
+ InferredWriterFactory<T>
writerFactory,
+ long maxFileSize) {
+ this.variantColumns = variantColumns;
+ this.extractor = extractor;
+ this.inferrer = inferrer;
+ this.writerFactory = writerFactory;
+ this.maxBufferedBytes = Math.min(MAX_BUFFERED_BYTES, Math.max(1,
maxFileSize));
+ }
+
+ @Override
+ public boolean canWrite() {
+ // Nothing has been physically written while buffering, so size-based
rollover cannot apply yet.
+ return delegate == null || delegate.canWrite();
+ }
+
+ @Override
+ public void writeWithMetadata(HoodieKey key, HoodieRecord record,
HoodieSchema schema, Properties props) throws IOException {
+ rethrowIfFailed();
+ if (delegate != null) {
+ delegate.writeWithMetadata(key, record, schema, props);
+ } else {
+ buffer(true, key, null, record, schema, props);
+ }
+ }
+
+ @Override
+ public void write(String recordKey, HoodieRecord record, HoodieSchema
schema, Properties props) throws IOException {
+ rethrowIfFailed();
+ if (delegate != null) {
+ delegate.write(recordKey, record, schema, props);
+ } else {
+ buffer(false, null, recordKey, record, schema, props);
+ }
+ }
+
+ @Override
+ public void writeRow(String recordKey, T record) throws IOException {
+ rethrowIfFailed();
+ // No HoodieRecord or schema to sample from: materialize with what has
been sampled so far and
+ // pass the row through. Today only the native log-format delete writer
takes this path, and its
+ // delete schema has no variant column to infer for.
+ materialize();
+ delegate.writeRow(recordKey, record);
+ }
+
+ @Override
+ public void addFooterMetadata(Map<String, String> footerMetadata) {
+ if (delegate != null) {
+ delegate.addFooterMetadata(footerMetadata);
+ } else {
+ // Footer metadata is only consumed at close, so it can wait for the
real writer.
+ pendingFooterMetadata.putAll(footerMetadata);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ boolean delegateClosed = false;
+ try {
+ rethrowIfFailed();
+ // Materialize even with an empty buffer: handles expect the file to
exist at close.
+ materialize();
+ // Mark before close() so a throwing delegate.close() surfaces, not
retried in the catch.
+ delegateClosed = true;
+ delegate.close();
+ } catch (IOException | RuntimeException e) {
+ if (delegate != null && !delegateClosed) {
+ try {
+ delegate.close();
+ } catch (Exception suppressed) {
+ // Best-effort cleanup; surface the original failure.
+ }
+ }
+ throw e;
+ }
+ }
+
+ @Override
+ public Object getFileFormatMetadata() {
+ try {
+ rethrowIfFailed();
+ materialize();
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to materialize the parquet writer
for format metadata", e);
+ }
+ return delegate.getFileFormatMetadata();
+ }
+
+ private void buffer(boolean withMetadata, HoodieKey key, String recordKey,
HoodieRecord record,
+ HoodieSchema schema, Properties props) throws
IOException {
+ rethrowIfFailed();
+ HoodieRecord buffered = extractor.prepare(record.copy(), schema, props);
+ // Eager extraction: immutable byte copies decouple inference from
buffered-record identity,
+ // and per-record extraction failures (corrupt binaries) surface exactly
like an eager write.
+ samples.add(extractor.extract(buffered, schema, props));
+ buffer.add(new BufferedWrite(withMetadata, key, recordKey, buffered,
schema, props));
+ // Re-estimate periodically so a small first record cannot defeat the byte
cap
+ // (same moving-average idiom as ExternalSpillableMap).
+ if (estimatedRecordSize == 0 || buffer.size() % SIZE_ESTIMATE_INTERVAL ==
0) {
+ long sampled = Math.max(1, sizeEstimator.sizeEstimate(buffered) -
extractor.sharedSizeEstimate(schema));
+ estimatedRecordSize = estimatedRecordSize == 0
+ ? sampled : (long) (estimatedRecordSize * 0.9 + sampled * 0.1);
+ }
+ bufferedBytes += estimatedRecordSize;
Review Comment:
`bufferedBytes` is never rescaled when `estimatedRecordSize` is revised, so
the first hundred records keep contributing the first record's estimate and the
64MB cap trips late whenever early records are small. `ExternalSpillableMap`
sets `currentInMemoryMapSize = size * estimatedPayloadSize` right after the
re-estimate - the same line here restores the bound.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
