wombatu-kun commented on code in PR #18961:
URL: https://github.com/apache/hudi/pull/18961#discussion_r3849079789
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java:
##########
@@ -298,6 +298,24 @@ 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)
+ .sinceVersion("1.3.0")
+ .withDocumentation("When enabled, the shredding schema for variant
columns without an explicit "
+ + "typed_value in the write schema is inferred automatically per
parquet file from a sample of "
+ + "the records written to that file, mirroring Spark 4.1's "
+ + "spark.sql.variant.inferShreddingSchema. Requires Spark 4.1+ on
the writer classpath; "
+ + "writes stay unshredded otherwise (Spark 4.0, Flink, Java
engines). Applies to every "
+ + "parquet file the writer produces: base files and, on table
version 10+, the native "
+ + "parquet log files of MOR tables (each infers its own schema);
legacy Avro log blocks "
Review Comment:
Parquet log data blocks stay unshredded too:
`ParquetUtils.serializeRecordsToLogBlock` goes through the `OutputStream`
overload of `newParquetFileWriter`, which neither factory wraps. Widening the
clause past Avro would cover a table on
`hoodie.logfile.data.block.format=parquet`.
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/VariantShreddingInferenceInternalRowFileWriter.java:
##########
@@ -0,0 +1,260 @@
+/*
+ * 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.SparkAdapterSupport$;
+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.util.CloseableUtils;
+import org.apache.hudi.common.util.DefaultSizeEstimator;
+import org.apache.hudi.core.io.storage.VariantShreddingInferenceFileWriter;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A {@link HoodieInternalRowFileWriter} decorator that infers a per-file
variant shredding
+ * schema from the first rows before opening the real parquet writer; the
row-writer-path
+ * sibling of {@link VariantShreddingInferenceFileWriter}, sharing its
buffering thresholds and
+ * failure semantics.
+ *
+ * <p>Meta columns including the commit seqno are composed into the row by the
handle BEFORE
+ * {@code writeRow}, so ordered replay is value-exact here by construction.
Rows and keys are
+ * copied because Spark iterators reuse their instances.</p>
+ */
+@Slf4j
+public class VariantShreddingInferenceInternalRowFileWriter implements
HoodieInternalRowFileWriter {
+
+ private static final int SIZE_ESTIMATE_INTERVAL = 100;
+
+ /** Creates the real row file writer once the inferred typed_value schemas
are known. */
+ @FunctionalInterface
+ public interface InferredRowWriterFactory {
+ HoodieInternalRowFileWriter create(Map<String, HoodieSchema>
inferredTypedValues) throws IOException;
+ }
+
+ private final List<String> variantColumns;
+ private final int[] ordinals;
+ private final VariantShreddingSchemaInferrer inferrer;
+ private final InferredRowWriterFactory writerFactory;
+ private final long maxBufferedBytes;
+ private final DefaultSizeEstimator<InternalRow> sizeEstimator = new
DefaultSizeEstimator<>();
+
+ private final List<BufferedRow> buffer = new ArrayList<>();
+ private final List<VariantSample[]> samples = new ArrayList<>();
+ private long bufferedBytes = 0;
+ private long estimatedRowSize = 0;
+ private long estimatedRowCount = 0;
+ private HoodieInternalRowFileWriter delegate;
+ private IOException fatalFailure;
+ private boolean closed = false;
+
+ public VariantShreddingInferenceInternalRowFileWriter(List<String>
variantColumns,
+ int[] ordinals,
+
VariantShreddingSchemaInferrer inferrer,
+
InferredRowWriterFactory writerFactory,
+ long maxFileSize) {
+ this.variantColumns = variantColumns;
+ this.ordinals = ordinals;
+ this.inferrer = inferrer;
+ this.writerFactory = writerFactory;
+ this.maxBufferedBytes =
Math.min(VariantShreddingInferenceFileWriter.MAX_BUFFERED_BYTES, Math.max(1,
maxFileSize));
Review Comment:
Every `maxFileSize` the tests pass is either far below 64MB or
`Long.MAX_VALUE` with a buffer that never approaches it, so dropping
`Math.min(MAX_BUFFERED_BYTES, ...)` from either decorator keeps every test
green. One `UnsafeRow` reporting a size at the cap, with `maxFileSize` at
`Long.MAX_VALUE`, pins it here without allocating.
##########
hudi-common/src/main/java/org/apache/hudi/core/io/storage/VariantShreddingInferenceFileWriter.java:
##########
@@ -0,0 +1,331 @@
+/*
+ * 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.CloseableUtils;
+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}, which
serves two purposes:
+ * an extractor that has to materialize the record to sample it (the Avro one
deserializes
+ * payload-backed records) returns the materialized form for buffering so the
replay does not
+ * repeat that work, and (since copy() returns the caller's own wrapper for
every record type
+ * today) both shipped extractors return a wrapper the caller does not hold,
so a handle that
+ * deflates its record right after the write call cannot blank a buffered one.
Records with
+ * nothing to materialize (delete payloads) are buffered as they are, so
replay still 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. No production caller reaches this today: both
writeRow callers (the
+ // native log-format delete writer and the CDC writer) pass schemas
without a top-level variant,
+ // so the factories never wrap them.
+ 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 | Error e) {
+ // Error included: materialize() rethrows the Error it latches, and the
delegate it
+ // created must still be closed.
+ if (delegate != null && !delegateClosed) {
+ CloseableUtils.closeSuppressing(delegate, e);
Review Comment:
No test sets `failCloseWith` on a delegate that also fails the replay, so
only the close itself is pinned and swapping `closeSuppressing` for a bare
swallow keeps both decorator test classes green. Setting it in
`testReplayFailureIsLatchedAndRethrown` and asserting `getSuppressed()[0]`
would cover the suppression.
##########
hudi-common/src/test/java/org/apache/hudi/common/avro/TestAvroVariantSampleExtractor.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.avro.VariantShreddingSchemaInferrer.VariantSample;
+import org.apache.hudi.common.model.HoodieAvroIndexedRecord;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieAvroRecord;
+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.schema.HoodieSchemaField;
+import org.apache.hudi.common.schema.HoodieSchemaType;
+import org.apache.hudi.common.util.ObjectSizeCalculator;
+import org.apache.hudi.common.util.Option;
+
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Properties;
+
+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.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class TestAvroVariantSampleExtractor {
+
+ private static final HoodieSchema SCHEMA = HoodieSchema.createRecord("rec",
"ns", null, Arrays.asList(
+ HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaField.of("v",
HoodieSchema.createNullable(HoodieSchema.createVariant()))));
+ private static final Properties PROPS = new Properties();
+ private static final HoodieKey KEY = new HoodieKey("r1", "p");
+
+ private static GenericRecord variant(Object value, Object metadata) {
+ HoodieSchema variantSchema =
SCHEMA.getField("v").get().schema().getNonNullType();
+ GenericRecord record = new
GenericData.Record(variantSchema.toAvroSchema());
+ record.put(HoodieSchema.Variant.VARIANT_VALUE_FIELD, value);
+ record.put(HoodieSchema.Variant.VARIANT_METADATA_FIELD, metadata);
+ return record;
+ }
+
+ private static GenericRecord row(GenericRecord variantValue) {
+ GenericRecord record = new GenericData.Record(SCHEMA.toAvroSchema());
+ record.put("id", "r1");
+ record.put("v", variantValue);
+ return record;
+ }
+
+ @Test
+ public void testExtractsDefensiveCopiesOfByteBufferAndByteArrayPayloads()
throws IOException {
+ AvroVariantSampleExtractor extractor = new
AvroVariantSampleExtractor(singletonList("v"));
+
+ // ByteBuffer payloads, the usual Avro representation of bytes.
+ byte[] valueBytes = {1, 2};
+ byte[] metadataBytes = {3};
+ VariantSample[] fromBuffers = extractor.extract(
+ new HoodieAvroIndexedRecord(KEY,
row(variant(ByteBuffer.wrap(valueBytes), ByteBuffer.wrap(metadataBytes)))),
+ SCHEMA, PROPS);
+ assertEquals(1, fromBuffers.length);
+ assertArrayEquals(new byte[] {1, 2}, fromBuffers[0].getValue());
+ assertArrayEquals(new byte[] {3}, fromBuffers[0].getMetadata());
+ // Samples must not alias the record's backing arrays: they outlive the
buffered record.
+ valueBytes[0] = 9;
+ metadataBytes[0] = 9;
+ assertArrayEquals(new byte[] {1, 2}, fromBuffers[0].getValue());
+ assertArrayEquals(new byte[] {3}, fromBuffers[0].getMetadata());
+
+ // Raw byte[] payloads are accepted too.
+ byte[] rawValue = {4};
+ byte[] rawMetadata = {5};
+ VariantSample[] fromArrays = extractor.extract(
+ new HoodieAvroIndexedRecord(KEY, row(variant(rawValue, rawMetadata))),
SCHEMA, PROPS);
+ rawValue[0] = 9;
+ rawMetadata[0] = 9;
+ assertArrayEquals(new byte[] {4}, fromArrays[0].getValue());
+ assertArrayEquals(new byte[] {5}, fromArrays[0].getMetadata());
+ }
+
+ @Test
+ public void testAbsentColumnNullVariantAndMalformedVariantYieldNullSamples()
throws IOException {
+ // "w" is not in the record's own schema (per-call schemas can differ from
the writer schema);
+ // it is skipped rather than failing the write.
+ AvroVariantSampleExtractor extractor = new
AvroVariantSampleExtractor(Arrays.asList("v", "w"));
+
+ VariantSample[] nullVariant = extractor.extract(new
HoodieAvroIndexedRecord(KEY, row(null)), SCHEMA, PROPS);
+ assertEquals(2, nullVariant.length);
+ assertNull(nullVariant[0]);
+ assertNull(nullVariant[1]);
+
+ // A variant record with a null value payload contributes nothing either...
+ VariantSample[] nullValue = extractor.extract(
+ new HoodieAvroIndexedRecord(KEY, row(variant(null, ByteBuffer.wrap(new
byte[] {1})))), SCHEMA, PROPS);
+ assertNull(nullValue[0]);
+
+ // ...nor does a variant-positioned record whose own schema lacks the
value member, nor a
+ // non-record value in the variant position: malformed data declines
sampling, never fails it.
+ GenericRecord metadataOnly = new
GenericData.Record(HoodieSchema.createRecord("metadata_only", null, null,
+ singletonList(HoodieSchemaField.of("metadata",
HoodieSchema.create(HoodieSchemaType.BYTES)))).toAvroSchema());
+ metadataOnly.put("metadata", ByteBuffer.wrap(new byte[] {1}));
+ assertNull(extractor.extract(new HoodieAvroIndexedRecord(KEY,
row(metadataOnly)), SCHEMA, PROPS)[0]);
+ GenericRecord notARecord = new GenericData.Record(SCHEMA.toAvroSchema());
+ notARecord.put("id", "r1");
+ notARecord.put("v", "not a variant record");
+ assertNull(extractor.extract(new HoodieAvroIndexedRecord(KEY, notARecord),
SCHEMA, PROPS)[0]);
+ }
+
+ @Test
+ public void testSharedSizeIsTheSchemaGraphMemoizedPerSchema() {
+ // The Avro Schema every materialized record references is charged once,
not per record.
+ AvroVariantSampleExtractor extractor = new
AvroVariantSampleExtractor(singletonList("v"));
+ long shared = extractor.sharedSizeEstimate(SCHEMA);
+ assertEquals(ObjectSizeCalculator.getObjectSize(SCHEMA.toAvroSchema()),
shared);
+ assertTrue(shared > 0);
+ assertEquals(shared, extractor.sharedSizeEstimate(SCHEMA));
Review Comment:
Re-asking with the same `SCHEMA` instance returns the same number whether
the size is memoized, recomputed on every call, or memoized once and never
invalidated, so nothing here pins the per-schema half of the name. A second,
differently-shaped `HoodieSchema` followed by `SCHEMA` again would pin the
identity-keyed invalidation.
##########
hudi-common/src/main/java/org/apache/hudi/core/io/storage/VariantShreddingInferenceFileWriter.java:
##########
@@ -0,0 +1,331 @@
+/*
+ * 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.CloseableUtils;
+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}, which
serves two purposes:
+ * an extractor that has to materialize the record to sample it (the Avro one
deserializes
+ * payload-backed records) returns the materialized form for buffering so the
replay does not
+ * repeat that work, and (since copy() returns the caller's own wrapper for
every record type
+ * today) both shipped extractors return a wrapper the caller does not hold,
so a handle that
+ * deflates its record right after the write call cannot blank a buffered one.
Records with
+ * nothing to materialize (delete payloads) are buffered as they are, so
replay still 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. No production caller reaches this today: both
writeRow callers (the
+ // native log-format delete writer and the CDC writer) pass schemas
without a top-level variant,
Review Comment:
`HoodieRowDataCreateHandle` is a third caller, through
`HoodieRowDataFileWriter` which extends `HoodieFileWriter`, and it is excluded
because no Flink factory builds this decorator rather than by its schema.
Pointing at the two factories that do build it would keep the claim true as
callers come and go.
--
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]
