You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "yihua (via GitHub)" <gi...@apache.org> on 2026/07/12 05:07:24 UTC
[PR] refactor(io): extract shared Arrow writer base for Lance and Vortex [hudi]
yihua opened a new pull request, #19254:
URL: https://github.com/apache/hudi/pull/19254
### Describe the issue this Pull Request addresses
The Lance and Vortex base writers duplicate ~85-90% of their structure (Arrow buffering, dual flush triggers, allocator lifecycle, bloom filter support, empty-file close handling): 362 and 345 lines that must be maintained in lockstep. This PR extracts the shared boilerplate so the next Arrow-native format only implements its FFI-specific hooks. Writer-side only; reader-side extraction is a follow-up.
### Summary and Changelog
- Add abstract `HoodieBaseArrowWriter` (`org.apache.hudi.io.arrow`, hudi-hadoop-common, ungated): owns the Arrow child-allocator lifecycle, `VectorSchemaRoot` management, row buffering, the dual flush trigger (row-count batch size or byte watermark), bloom-filter and min/max key support, `getDataSize`, empty-file-on-close handling, and close ordering with exception suppression. Imports only Arrow and Hudi types (no `dev.vortex`, no Lance native types), so it compiles on the Java 8 target.
- `HoodieBaseLanceWriter` shrinks 362 to 161 lines and `HoodieBaseVortexWriter` 345 to 184, each keeping only format-specific hooks (Lance native writer calls; Vortex C Data Interface export and the dedicated native allocator with its tolerated one-time schema-export retention). The Vortex writer stays in the JDK-17-gated source root.
- Format hooks: `initializeFormatWriter`, `writeBatch`, `closeFormatWriter`, `getFormatName`, `isFormatWriterInitialized`, plus optional `finalizeFooterMetadata` (Lance footer), `closeFormatResources`, and `closeAfterAllocator` (Vortex native allocator).
- No public or protected API changes: constructors, `addFooterMetadata`, `getWrittenRecordCount`, `getDataSize` are intact, the Spark/Flink subclasses are unmodified, and the reflective `HoodieSparkVortexWriter.newWriter` factory contract is untouched.
### Impact
Behavior-preserving deduplication of the Arrow writer path. Validated: Lance tests (TestLanceDataSource 61, TestHoodieSparkLanceWriter 23, TestHoodieSparkLanceReader 16) and the Vortex round-trip (2) all pass on JDK 17; the Java 11 Spark chain and Flink client compile; checkstyle clean.
### Risk Level
medium: touches the write path of two formats, mitigated by the full Lance and Vortex test runs above and an adversarial equivalence review of flush triggers, close ordering, and allocator lifecycle.
### Documentation Update
none
### Contributor's checklist
- [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] Adequate tests were added if applicable
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(io): extract shared Arrow writer base for Lance and Vortex [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19254:
URL: https://github.com/apache/hudi/pull/19254#discussion_r3695395085
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/io/arrow/HoodieBaseArrowWriter.java:
##########
@@ -0,0 +1,370 @@
+/*
+ * 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.arrow;
+
+import org.apache.hudi.common.avro.HoodieBloomFilterWriteSupport;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.io.memory.HoodieArrowAllocator;
+import org.apache.hudi.storage.StoragePath;
+
+import lombok.AccessLevel;
+import lombok.Getter;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import javax.annotation.concurrent.NotThreadSafe;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ * Base class for Hudi file writers of Arrow-native formats (e.g. Lance, Vortex) supporting
+ * different record types.
+ *
+ * This class handles the format-independent parts of writing an Arrow-batched file:
+ * - BufferAllocator management
+ * - Record buffering into a {@link VectorSchemaRoot} and batch flushing
+ * - File size checks
+ * - Record key tracking for bloom filter / min-max metadata
+ * - Empty-file handling and close ordering
+ *
+ * Format subclasses (Lance, Vortex, ...) implement the hooks that create the native format
+ * writer, hand each flushed batch to it, and close it; engine subclasses (Spark, Flink, ...)
+ * implement the type-specific conversion to Arrow format and provide the Arrow schema.
+ *
+ * @param <R> The record type (e.g., GenericRecord, InternalRow)
+ * @param <K> The record key type used for bloom filter tracking
+ */
+@NotThreadSafe
+public abstract class HoodieBaseArrowWriter<R, K extends Comparable<K>> implements Closeable {
+ protected static final int DEFAULT_BATCH_SIZE = 1000;
+ @Getter(value = AccessLevel.PROTECTED)
+ private final StoragePath path;
+ @Getter(value = AccessLevel.PROTECTED)
+ private final BufferAllocator allocator;
+ private final int batchSize;
+ private final long flushByteWatermark;
+ @Getter(value = AccessLevel.PROTECTED)
+ private long writtenRecordCount = 0;
+ private long totalFlushedDataSize = 0;
+ private int currentBatchSize = 0;
+ private VectorSchemaRoot root;
+ private ArrowWriter<R> arrowWriter;
+ protected final Option<HoodieBloomFilterWriteSupport<K>> bloomFilterWriteSupportOpt;
+
+ /**
+ * Constructor for base Arrow-format writer.
+ *
+ * @param path Path where the file will be written
+ * @param batchSize Row-count threshold; the current batch is flushed when this many records have been buffered
+ * @param allocatorSize Maximum bytes the per-writer Arrow child allocator may hold at once; sized so Arrow's
+ * power-of-2 buffer doubling never requests a chunk above this cap
+ * @param flushByteWatermark Byte-size threshold; the current batch is flushed when the sum of in-flight
+ * FieldVector buffer sizes reaches this value. Must be small enough that the next
+ * doubling step stays below {@code allocatorSize}.
+ * @param bloomFilterWriteSupportOpt Optional bloom filter write support for record key tracking
+ */
+ protected HoodieBaseArrowWriter(StoragePath path, int batchSize, long allocatorSize, long flushByteWatermark,
+ Option<HoodieBloomFilterWriteSupport<K>> bloomFilterWriteSupportOpt) {
+ this.path = path;
+ this.allocator = HoodieArrowAllocator.newChildAllocator(
+ getClass().getSimpleName() + "-data-" + path.getName(), allocatorSize);
+ this.batchSize = batchSize;
+ this.flushByteWatermark = flushByteWatermark;
+ this.bloomFilterWriteSupportOpt = bloomFilterWriteSupportOpt;
+ }
+
+ /**
+ * Create and initialize the arrow writer for writing records to VectorSchemaRoot.
+ * Called once during lazy initialization when the first record is written.
+ *
+ * @param root The VectorSchemaRoot to write into
+ * @return An arrow writer implementation that writes records of type R to the root
+ */
+ protected abstract ArrowWriter<R> createArrowWriter(VectorSchemaRoot root);
+
+ /**
+ * Get the Arrow schema for this writer.
+ * Subclasses must provide the Arrow schema corresponding to their record type; each written
+ * batch must conform to it.
+ *
+ * @return Arrow schema
+ */
+ protected abstract Schema getArrowSchema();
+
+ /**
+ * The format name (e.g. "Lance", "Vortex") used in error messages.
+ */
+ protected abstract String getFormatName();
+
+ /**
+ * Whether the native format writer has been initialized. Used as the lazy-initialization
+ * sentinel: {@link #initializeFormatWriter()} is invoked on the first write (or on close for
+ * an empty file) only while this returns false.
+ */
+ protected abstract boolean isFormatWriterInitialized();
+
+ /**
+ * Create the native format writer (lazy initialization).
+ */
+ protected abstract void initializeFormatWriter() throws IOException;
+
+ /**
+ * Write one finished batch to the native format writer.
+ *
+ * @param batch The VectorSchemaRoot holding the batch; row count is already set
+ */
+ protected abstract void writeBatch(VectorSchemaRoot batch) throws IOException;
+
+ /**
+ * Close the native format writer, finalizing the file. Called unconditionally during
+ * {@link #close()}; implementations must no-op when the writer was never initialized.
+ */
+ protected abstract void closeFormatWriter() throws Exception;
+
+ /**
+ * Subclass hook invoked once during {@link #close()}, after remaining records are flushed and
+ * empty-file handling, before the format writer closes. Formats that support file-level
+ * key-value metadata persist footer entries (bloom filter, min/max keys, ...) here.
+ * Default implementation does nothing.
+ */
+ protected void finalizeFooterMetadata() throws IOException {
+ }
+
+ /**
+ * Subclass hook for closing format-specific resources that must be released after the format
+ * writer closes but before the Arrow data buffers are freed. Called unconditionally during
+ * {@link #close()} in its own try/catch, so failures here are recorded alongside (not instead
+ * of) other close failures. Default implementation does nothing.
+ */
+ protected void closeFormatResources() throws Exception {
+ }
+
+ /**
+ * Subclass hook for cleanup that must run after the data allocator has closed (e.g. a
+ * dedicated native-side allocator). Called last during {@link #close()}; exceptions thrown
+ * here propagate directly, so implementations should catch what they intend to tolerate.
+ * Default implementation does nothing.
+ */
+ protected void closeAfterAllocator() {
+ }
+
+ /**
+ * Write a single record. Records are buffered and flushed in batches.
+ *
+ * @param record Record to write
+ * @throws IOException if write fails
+ */
+ public void write(R record) throws IOException {
+ // Lazy initialization on first write
+ if (!isFormatWriterInitialized()) {
+ initializeFormatWriter();
+ }
+ if (root == null) {
+ root = VectorSchemaRoot.create(getArrowSchema(), allocator);
+ }
+ if (arrowWriter == null) {
+ arrowWriter = createArrowWriter(root);
+ }
+
+ // Reset arrow writer at the start of each new batch
+ if (currentBatchSize == 0) {
+ arrowWriter.reset();
+ }
+
+ arrowWriter.write(record);
+ currentBatchSize++;
+ writtenRecordCount++;
+
+ // Flush when row-count batch is full OR in-flight Arrow buffers cross the byte watermark.
+ // The byte-based check bounds in-flight memory so Arrow's power-of-2 vector reallocation
+ // can't escalate to a chunk size above the allocator cap regardless of per-row payload.
+ if (currentBatchSize >= batchSize || currentBufferBytes() >= flushByteWatermark) {
+ flushBatch();
+ }
+ }
+
+ /**
+ * Bytes currently held by the writer's Arrow child allocator. Used to drive the byte-aware
+ * flush.
+ *
+ * <p>Note: we deliberately do <em>not</em> use {@code FieldVector.getBufferSize()} here.
+ * For variable-width vectors (e.g. {@code BaseLargeVariableWidthVector} backing BLOB columns)
+ * that method short-circuits to 0 when {@code valueCount == 0}, and {@code valueCount} is only
+ * set during {@link ArrowWriter#finishBatch()} — i.e. at flush time. Mid-batch it always
+ * reports zero, so a watermark driven by it never fires. {@link BufferAllocator#getAllocatedMemory()}
+ * tracks the underlying ArrowBuf capacities directly and is exactly the quantity the allocator
+ * cap is enforced against.
+ */
+ private long currentBufferBytes() {
+ return allocator.getAllocatedMemory();
+ }
+
+ /**
+ * Close the writer, flushing any remaining buffered records.
+ *
+ * @throws IOException if close fails
+ */
+ @Override
+ public void close() throws IOException {
+ Exception primaryException = null;
+
+ // 1. Flush remaining records
+ try {
+ // Flush any remaining records in current batch
+ if (currentBatchSize > 0) {
+ flushBatch();
+ }
+
+ // Ensure writer is initialized even if no data was written
+ // This creates an empty file with just schema metadata
+ if (!isFormatWriterInitialized() && root == null) {
+ initializeFormatWriter();
Review Comment:
+1
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(io): extract shared Arrow writer base for Lance and Vortex [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19254:
URL: https://github.com/apache/hudi/pull/19254#issuecomment-5026961970
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "eafc52e44f39561e8c0c4d5375af7d7c12f111a6",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15449",
"triggerID" : "eafc52e44f39561e8c0c4d5375af7d7c12f111a6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* eafc52e44f39561e8c0c4d5375af7d7c12f111a6 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15449)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(io): extract shared Arrow writer base for Lance and Vortex [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19254:
URL: https://github.com/apache/hudi/pull/19254#issuecomment-4950059726
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "eafc52e44f39561e8c0c4d5375af7d7c12f111a6",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "eafc52e44f39561e8c0c4d5375af7d7c12f111a6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* eafc52e44f39561e8c0c4d5375af7d7c12f111a6 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(io): extract shared Arrow writer base for Lance and Vortex [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #19254:
URL: https://github.com/apache/hudi/pull/19254#discussion_r3685898515
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/io/arrow/HoodieBaseArrowWriter.java:
##########
@@ -0,0 +1,370 @@
+/*
+ * 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.arrow;
+
+import org.apache.hudi.common.avro.HoodieBloomFilterWriteSupport;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.io.memory.HoodieArrowAllocator;
+import org.apache.hudi.storage.StoragePath;
+
+import lombok.AccessLevel;
+import lombok.Getter;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import javax.annotation.concurrent.NotThreadSafe;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ * Base class for Hudi file writers of Arrow-native formats (e.g. Lance, Vortex) supporting
+ * different record types.
+ *
+ * This class handles the format-independent parts of writing an Arrow-batched file:
+ * - BufferAllocator management
+ * - Record buffering into a {@link VectorSchemaRoot} and batch flushing
+ * - File size checks
+ * - Record key tracking for bloom filter / min-max metadata
+ * - Empty-file handling and close ordering
+ *
+ * Format subclasses (Lance, Vortex, ...) implement the hooks that create the native format
+ * writer, hand each flushed batch to it, and close it; engine subclasses (Spark, Flink, ...)
+ * implement the type-specific conversion to Arrow format and provide the Arrow schema.
+ *
+ * @param <R> The record type (e.g., GenericRecord, InternalRow)
+ * @param <K> The record key type used for bloom filter tracking
+ */
+@NotThreadSafe
+public abstract class HoodieBaseArrowWriter<R, K extends Comparable<K>> implements Closeable {
+ protected static final int DEFAULT_BATCH_SIZE = 1000;
+ @Getter(value = AccessLevel.PROTECTED)
+ private final StoragePath path;
+ @Getter(value = AccessLevel.PROTECTED)
+ private final BufferAllocator allocator;
+ private final int batchSize;
+ private final long flushByteWatermark;
+ @Getter(value = AccessLevel.PROTECTED)
+ private long writtenRecordCount = 0;
+ private long totalFlushedDataSize = 0;
+ private int currentBatchSize = 0;
+ private VectorSchemaRoot root;
+ private ArrowWriter<R> arrowWriter;
+ protected final Option<HoodieBloomFilterWriteSupport<K>> bloomFilterWriteSupportOpt;
+
+ /**
+ * Constructor for base Arrow-format writer.
+ *
+ * @param path Path where the file will be written
+ * @param batchSize Row-count threshold; the current batch is flushed when this many records have been buffered
+ * @param allocatorSize Maximum bytes the per-writer Arrow child allocator may hold at once; sized so Arrow's
+ * power-of-2 buffer doubling never requests a chunk above this cap
+ * @param flushByteWatermark Byte-size threshold; the current batch is flushed when the sum of in-flight
+ * FieldVector buffer sizes reaches this value. Must be small enough that the next
+ * doubling step stays below {@code allocatorSize}.
+ * @param bloomFilterWriteSupportOpt Optional bloom filter write support for record key tracking
+ */
+ protected HoodieBaseArrowWriter(StoragePath path, int batchSize, long allocatorSize, long flushByteWatermark,
+ Option<HoodieBloomFilterWriteSupport<K>> bloomFilterWriteSupportOpt) {
+ this.path = path;
+ this.allocator = HoodieArrowAllocator.newChildAllocator(
+ getClass().getSimpleName() + "-data-" + path.getName(), allocatorSize);
+ this.batchSize = batchSize;
+ this.flushByteWatermark = flushByteWatermark;
+ this.bloomFilterWriteSupportOpt = bloomFilterWriteSupportOpt;
+ }
+
+ /**
+ * Create and initialize the arrow writer for writing records to VectorSchemaRoot.
+ * Called once during lazy initialization when the first record is written.
+ *
+ * @param root The VectorSchemaRoot to write into
+ * @return An arrow writer implementation that writes records of type R to the root
+ */
+ protected abstract ArrowWriter<R> createArrowWriter(VectorSchemaRoot root);
+
+ /**
+ * Get the Arrow schema for this writer.
+ * Subclasses must provide the Arrow schema corresponding to their record type; each written
+ * batch must conform to it.
+ *
+ * @return Arrow schema
+ */
+ protected abstract Schema getArrowSchema();
+
+ /**
+ * The format name (e.g. "Lance", "Vortex") used in error messages.
+ */
+ protected abstract String getFormatName();
+
+ /**
+ * Whether the native format writer has been initialized. Used as the lazy-initialization
+ * sentinel: {@link #initializeFormatWriter()} is invoked on the first write (or on close for
+ * an empty file) only while this returns false.
+ */
+ protected abstract boolean isFormatWriterInitialized();
+
+ /**
+ * Create the native format writer (lazy initialization).
+ */
+ protected abstract void initializeFormatWriter() throws IOException;
+
+ /**
+ * Write one finished batch to the native format writer.
+ *
+ * @param batch The VectorSchemaRoot holding the batch; row count is already set
+ */
+ protected abstract void writeBatch(VectorSchemaRoot batch) throws IOException;
+
+ /**
+ * Close the native format writer, finalizing the file. Called unconditionally during
+ * {@link #close()}; implementations must no-op when the writer was never initialized.
+ */
+ protected abstract void closeFormatWriter() throws Exception;
+
+ /**
+ * Subclass hook invoked once during {@link #close()}, after remaining records are flushed and
+ * empty-file handling, before the format writer closes. Formats that support file-level
+ * key-value metadata persist footer entries (bloom filter, min/max keys, ...) here.
+ * Default implementation does nothing.
+ */
+ protected void finalizeFooterMetadata() throws IOException {
+ }
+
+ /**
+ * Subclass hook for closing format-specific resources that must be released after the format
+ * writer closes but before the Arrow data buffers are freed. Called unconditionally during
+ * {@link #close()} in its own try/catch, so failures here are recorded alongside (not instead
+ * of) other close failures. Default implementation does nothing.
+ */
+ protected void closeFormatResources() throws Exception {
+ }
+
+ /**
+ * Subclass hook for cleanup that must run after the data allocator has closed (e.g. a
+ * dedicated native-side allocator). Called last during {@link #close()}; exceptions thrown
+ * here propagate directly, so implementations should catch what they intend to tolerate.
+ * Default implementation does nothing.
+ */
+ protected void closeAfterAllocator() {
+ }
+
+ /**
+ * Write a single record. Records are buffered and flushed in batches.
+ *
+ * @param record Record to write
+ * @throws IOException if write fails
+ */
+ public void write(R record) throws IOException {
+ // Lazy initialization on first write
+ if (!isFormatWriterInitialized()) {
+ initializeFormatWriter();
+ }
+ if (root == null) {
+ root = VectorSchemaRoot.create(getArrowSchema(), allocator);
+ }
+ if (arrowWriter == null) {
+ arrowWriter = createArrowWriter(root);
+ }
+
+ // Reset arrow writer at the start of each new batch
+ if (currentBatchSize == 0) {
+ arrowWriter.reset();
+ }
+
+ arrowWriter.write(record);
+ currentBatchSize++;
+ writtenRecordCount++;
+
+ // Flush when row-count batch is full OR in-flight Arrow buffers cross the byte watermark.
+ // The byte-based check bounds in-flight memory so Arrow's power-of-2 vector reallocation
+ // can't escalate to a chunk size above the allocator cap regardless of per-row payload.
+ if (currentBatchSize >= batchSize || currentBufferBytes() >= flushByteWatermark) {
+ flushBatch();
+ }
+ }
+
+ /**
+ * Bytes currently held by the writer's Arrow child allocator. Used to drive the byte-aware
+ * flush.
+ *
+ * <p>Note: we deliberately do <em>not</em> use {@code FieldVector.getBufferSize()} here.
+ * For variable-width vectors (e.g. {@code BaseLargeVariableWidthVector} backing BLOB columns)
+ * that method short-circuits to 0 when {@code valueCount == 0}, and {@code valueCount} is only
+ * set during {@link ArrowWriter#finishBatch()} — i.e. at flush time. Mid-batch it always
+ * reports zero, so a watermark driven by it never fires. {@link BufferAllocator#getAllocatedMemory()}
+ * tracks the underlying ArrowBuf capacities directly and is exactly the quantity the allocator
+ * cap is enforced against.
+ */
+ private long currentBufferBytes() {
+ return allocator.getAllocatedMemory();
+ }
+
+ /**
+ * Close the writer, flushing any remaining buffered records.
+ *
+ * @throws IOException if close fails
+ */
+ @Override
+ public void close() throws IOException {
+ Exception primaryException = null;
+
+ // 1. Flush remaining records
+ try {
+ // Flush any remaining records in current batch
+ if (currentBatchSize > 0) {
+ flushBatch();
+ }
+
+ // Ensure writer is initialized even if no data was written
+ // This creates an empty file with just schema metadata
+ if (!isFormatWriterInitialized() && root == null) {
+ initializeFormatWriter();
Review Comment:
🤖 nit: the `&& root == null` clause here is always true when `!isFormatWriterInitialized()` — `root` is only ever created inside `write()`, which always calls `initializeFormatWriter()` first. Could you drop the second condition and leave just `if (!isFormatWriterInitialized())`? As written, a reader may wonder what scenario has the format writer uninitialized but `root` already set, and find no answer.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(io): extract shared Arrow writer base for Lance and Vortex [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19254:
URL: https://github.com/apache/hudi/pull/19254#discussion_r3695394418
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/io/arrow/HoodieBaseArrowWriter.java:
##########
@@ -0,0 +1,370 @@
+/*
+ * 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.arrow;
+
+import org.apache.hudi.common.avro.HoodieBloomFilterWriteSupport;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.io.memory.HoodieArrowAllocator;
+import org.apache.hudi.storage.StoragePath;
+
+import lombok.AccessLevel;
+import lombok.Getter;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import javax.annotation.concurrent.NotThreadSafe;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ * Base class for Hudi file writers of Arrow-native formats (e.g. Lance, Vortex) supporting
+ * different record types.
+ *
+ * This class handles the format-independent parts of writing an Arrow-batched file:
+ * - BufferAllocator management
+ * - Record buffering into a {@link VectorSchemaRoot} and batch flushing
+ * - File size checks
+ * - Record key tracking for bloom filter / min-max metadata
+ * - Empty-file handling and close ordering
+ *
+ * Format subclasses (Lance, Vortex, ...) implement the hooks that create the native format
+ * writer, hand each flushed batch to it, and close it; engine subclasses (Spark, Flink, ...)
+ * implement the type-specific conversion to Arrow format and provide the Arrow schema.
+ *
+ * @param <R> The record type (e.g., GenericRecord, InternalRow)
+ * @param <K> The record key type used for bloom filter tracking
+ */
+@NotThreadSafe
+public abstract class HoodieBaseArrowWriter<R, K extends Comparable<K>> implements Closeable {
+ protected static final int DEFAULT_BATCH_SIZE = 1000;
+ @Getter(value = AccessLevel.PROTECTED)
+ private final StoragePath path;
+ @Getter(value = AccessLevel.PROTECTED)
+ private final BufferAllocator allocator;
+ private final int batchSize;
+ private final long flushByteWatermark;
+ @Getter(value = AccessLevel.PROTECTED)
+ private long writtenRecordCount = 0;
+ private long totalFlushedDataSize = 0;
+ private int currentBatchSize = 0;
+ private VectorSchemaRoot root;
+ private ArrowWriter<R> arrowWriter;
+ protected final Option<HoodieBloomFilterWriteSupport<K>> bloomFilterWriteSupportOpt;
+
+ /**
+ * Constructor for base Arrow-format writer.
+ *
+ * @param path Path where the file will be written
+ * @param batchSize Row-count threshold; the current batch is flushed when this many records have been buffered
+ * @param allocatorSize Maximum bytes the per-writer Arrow child allocator may hold at once; sized so Arrow's
+ * power-of-2 buffer doubling never requests a chunk above this cap
+ * @param flushByteWatermark Byte-size threshold; the current batch is flushed when the sum of in-flight
+ * FieldVector buffer sizes reaches this value. Must be small enough that the next
+ * doubling step stays below {@code allocatorSize}.
+ * @param bloomFilterWriteSupportOpt Optional bloom filter write support for record key tracking
+ */
+ protected HoodieBaseArrowWriter(StoragePath path, int batchSize, long allocatorSize, long flushByteWatermark,
+ Option<HoodieBloomFilterWriteSupport<K>> bloomFilterWriteSupportOpt) {
+ this.path = path;
+ this.allocator = HoodieArrowAllocator.newChildAllocator(
+ getClass().getSimpleName() + "-data-" + path.getName(), allocatorSize);
+ this.batchSize = batchSize;
+ this.flushByteWatermark = flushByteWatermark;
+ this.bloomFilterWriteSupportOpt = bloomFilterWriteSupportOpt;
+ }
+
+ /**
+ * Create and initialize the arrow writer for writing records to VectorSchemaRoot.
+ * Called once during lazy initialization when the first record is written.
+ *
+ * @param root The VectorSchemaRoot to write into
+ * @return An arrow writer implementation that writes records of type R to the root
+ */
+ protected abstract ArrowWriter<R> createArrowWriter(VectorSchemaRoot root);
+
+ /**
+ * Get the Arrow schema for this writer.
+ * Subclasses must provide the Arrow schema corresponding to their record type; each written
+ * batch must conform to it.
+ *
+ * @return Arrow schema
+ */
+ protected abstract Schema getArrowSchema();
+
+ /**
+ * The format name (e.g. "Lance", "Vortex") used in error messages.
+ */
+ protected abstract String getFormatName();
+
+ /**
+ * Whether the native format writer has been initialized. Used as the lazy-initialization
+ * sentinel: {@link #initializeFormatWriter()} is invoked on the first write (or on close for
+ * an empty file) only while this returns false.
+ */
+ protected abstract boolean isFormatWriterInitialized();
+
+ /**
+ * Create the native format writer (lazy initialization).
+ */
+ protected abstract void initializeFormatWriter() throws IOException;
+
+ /**
+ * Write one finished batch to the native format writer.
+ *
+ * @param batch The VectorSchemaRoot holding the batch; row count is already set
+ */
+ protected abstract void writeBatch(VectorSchemaRoot batch) throws IOException;
+
+ /**
+ * Close the native format writer, finalizing the file. Called unconditionally during
+ * {@link #close()}; implementations must no-op when the writer was never initialized.
+ */
+ protected abstract void closeFormatWriter() throws Exception;
+
+ /**
+ * Subclass hook invoked once during {@link #close()}, after remaining records are flushed and
+ * empty-file handling, before the format writer closes. Formats that support file-level
+ * key-value metadata persist footer entries (bloom filter, min/max keys, ...) here.
+ * Default implementation does nothing.
+ */
+ protected void finalizeFooterMetadata() throws IOException {
+ }
+
+ /**
+ * Subclass hook for closing format-specific resources that must be released after the format
+ * writer closes but before the Arrow data buffers are freed. Called unconditionally during
+ * {@link #close()} in its own try/catch, so failures here are recorded alongside (not instead
+ * of) other close failures. Default implementation does nothing.
+ */
+ protected void closeFormatResources() throws Exception {
+ }
+
+ /**
+ * Subclass hook for cleanup that must run after the data allocator has closed (e.g. a
+ * dedicated native-side allocator). Called last during {@link #close()}; exceptions thrown
+ * here propagate directly, so implementations should catch what they intend to tolerate.
+ * Default implementation does nothing.
+ */
+ protected void closeAfterAllocator() {
+ }
+
+ /**
+ * Write a single record. Records are buffered and flushed in batches.
+ *
+ * @param record Record to write
+ * @throws IOException if write fails
+ */
+ public void write(R record) throws IOException {
+ // Lazy initialization on first write
+ if (!isFormatWriterInitialized()) {
+ initializeFormatWriter();
+ }
+ if (root == null) {
+ root = VectorSchemaRoot.create(getArrowSchema(), allocator);
+ }
+ if (arrowWriter == null) {
+ arrowWriter = createArrowWriter(root);
+ }
+
+ // Reset arrow writer at the start of each new batch
+ if (currentBatchSize == 0) {
+ arrowWriter.reset();
+ }
+
+ arrowWriter.write(record);
+ currentBatchSize++;
+ writtenRecordCount++;
+
+ // Flush when row-count batch is full OR in-flight Arrow buffers cross the byte watermark.
+ // The byte-based check bounds in-flight memory so Arrow's power-of-2 vector reallocation
+ // can't escalate to a chunk size above the allocator cap regardless of per-row payload.
+ if (currentBatchSize >= batchSize || currentBufferBytes() >= flushByteWatermark) {
+ flushBatch();
+ }
+ }
+
+ /**
+ * Bytes currently held by the writer's Arrow child allocator. Used to drive the byte-aware
+ * flush.
+ *
+ * <p>Note: we deliberately do <em>not</em> use {@code FieldVector.getBufferSize()} here.
+ * For variable-width vectors (e.g. {@code BaseLargeVariableWidthVector} backing BLOB columns)
+ * that method short-circuits to 0 when {@code valueCount == 0}, and {@code valueCount} is only
+ * set during {@link ArrowWriter#finishBatch()} — i.e. at flush time. Mid-batch it always
+ * reports zero, so a watermark driven by it never fires. {@link BufferAllocator#getAllocatedMemory()}
+ * tracks the underlying ArrowBuf capacities directly and is exactly the quantity the allocator
+ * cap is enforced against.
+ */
+ private long currentBufferBytes() {
+ return allocator.getAllocatedMemory();
+ }
+
+ /**
+ * Close the writer, flushing any remaining buffered records.
+ *
+ * @throws IOException if close fails
+ */
+ @Override
+ public void close() throws IOException {
+ Exception primaryException = null;
+
+ // 1. Flush remaining records
+ try {
+ // Flush any remaining records in current batch
+ if (currentBatchSize > 0) {
+ flushBatch();
+ }
+
+ // Ensure writer is initialized even if no data was written
+ // This creates an empty file with just schema metadata
+ if (!isFormatWriterInitialized() && root == null) {
+ initializeFormatWriter();
+ root = VectorSchemaRoot.create(getArrowSchema(), allocator);
+ root.setRowCount(0);
+ writeBatch(root);
+ }
+
+ // Persist footer key-value metadata (bloom filter, min/max keys, ...) for formats
+ // that support it, before the format writer closes.
+ finalizeFooterMetadata();
+ } catch (Exception e) {
+ primaryException = e;
+ }
+
+ // Close the format writer (finalizes the file)
+ try {
+ closeFormatWriter();
+ } catch (Exception e) {
+ primaryException = addSuppressed(primaryException, e);
+ }
+
+ // Close format-specific resources that must be released before the Arrow buffers
+ try {
+ closeFormatResources();
+ } catch (Exception e) {
+ primaryException = addSuppressed(primaryException, e);
+ }
+
+ // Close VectorSchemaRoot
+ if (root != null) {
+ try {
+ root.close();
+ } catch (Exception e) {
+ primaryException = addSuppressed(primaryException, e);
+ }
+ }
+
+ // Always close allocator
+ try {
+ allocator.close();
+ } catch (Exception e) {
+ primaryException = addSuppressed(primaryException, e);
+ }
+
+ // Format-specific cleanup that must run after the data allocator closes
+ closeAfterAllocator();
Review Comment:
`closeAfterAllocator()` is the one close step outside the `addSuppressed` ladder, so a throw there drops `primaryException` entirely instead of suppressing into it. Wrapping it like the other steps keeps the original write failure as the reported cause.
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/io/arrow/HoodieBaseArrowWriter.java:
##########
@@ -0,0 +1,370 @@
+/*
+ * 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.arrow;
+
+import org.apache.hudi.common.avro.HoodieBloomFilterWriteSupport;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.io.memory.HoodieArrowAllocator;
+import org.apache.hudi.storage.StoragePath;
+
+import lombok.AccessLevel;
+import lombok.Getter;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import javax.annotation.concurrent.NotThreadSafe;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ * Base class for Hudi file writers of Arrow-native formats (e.g. Lance, Vortex) supporting
+ * different record types.
+ *
+ * This class handles the format-independent parts of writing an Arrow-batched file:
+ * - BufferAllocator management
+ * - Record buffering into a {@link VectorSchemaRoot} and batch flushing
+ * - File size checks
+ * - Record key tracking for bloom filter / min-max metadata
+ * - Empty-file handling and close ordering
+ *
+ * Format subclasses (Lance, Vortex, ...) implement the hooks that create the native format
+ * writer, hand each flushed batch to it, and close it; engine subclasses (Spark, Flink, ...)
+ * implement the type-specific conversion to Arrow format and provide the Arrow schema.
+ *
+ * @param <R> The record type (e.g., GenericRecord, InternalRow)
+ * @param <K> The record key type used for bloom filter tracking
+ */
+@NotThreadSafe
+public abstract class HoodieBaseArrowWriter<R, K extends Comparable<K>> implements Closeable {
+ protected static final int DEFAULT_BATCH_SIZE = 1000;
+ @Getter(value = AccessLevel.PROTECTED)
+ private final StoragePath path;
+ @Getter(value = AccessLevel.PROTECTED)
+ private final BufferAllocator allocator;
+ private final int batchSize;
+ private final long flushByteWatermark;
+ @Getter(value = AccessLevel.PROTECTED)
+ private long writtenRecordCount = 0;
+ private long totalFlushedDataSize = 0;
+ private int currentBatchSize = 0;
+ private VectorSchemaRoot root;
+ private ArrowWriter<R> arrowWriter;
+ protected final Option<HoodieBloomFilterWriteSupport<K>> bloomFilterWriteSupportOpt;
+
+ /**
+ * Constructor for base Arrow-format writer.
+ *
+ * @param path Path where the file will be written
+ * @param batchSize Row-count threshold; the current batch is flushed when this many records have been buffered
+ * @param allocatorSize Maximum bytes the per-writer Arrow child allocator may hold at once; sized so Arrow's
+ * power-of-2 buffer doubling never requests a chunk above this cap
+ * @param flushByteWatermark Byte-size threshold; the current batch is flushed when the sum of in-flight
+ * FieldVector buffer sizes reaches this value. Must be small enough that the next
+ * doubling step stays below {@code allocatorSize}.
+ * @param bloomFilterWriteSupportOpt Optional bloom filter write support for record key tracking
+ */
+ protected HoodieBaseArrowWriter(StoragePath path, int batchSize, long allocatorSize, long flushByteWatermark,
+ Option<HoodieBloomFilterWriteSupport<K>> bloomFilterWriteSupportOpt) {
+ this.path = path;
+ this.allocator = HoodieArrowAllocator.newChildAllocator(
+ getClass().getSimpleName() + "-data-" + path.getName(), allocatorSize);
+ this.batchSize = batchSize;
+ this.flushByteWatermark = flushByteWatermark;
+ this.bloomFilterWriteSupportOpt = bloomFilterWriteSupportOpt;
+ }
+
+ /**
+ * Create and initialize the arrow writer for writing records to VectorSchemaRoot.
+ * Called once during lazy initialization when the first record is written.
+ *
+ * @param root The VectorSchemaRoot to write into
+ * @return An arrow writer implementation that writes records of type R to the root
+ */
+ protected abstract ArrowWriter<R> createArrowWriter(VectorSchemaRoot root);
+
+ /**
+ * Get the Arrow schema for this writer.
+ * Subclasses must provide the Arrow schema corresponding to their record type; each written
+ * batch must conform to it.
+ *
+ * @return Arrow schema
+ */
+ protected abstract Schema getArrowSchema();
+
+ /**
+ * The format name (e.g. "Lance", "Vortex") used in error messages.
+ */
+ protected abstract String getFormatName();
+
+ /**
+ * Whether the native format writer has been initialized. Used as the lazy-initialization
+ * sentinel: {@link #initializeFormatWriter()} is invoked on the first write (or on close for
+ * an empty file) only while this returns false.
+ */
+ protected abstract boolean isFormatWriterInitialized();
+
+ /**
+ * Create the native format writer (lazy initialization).
+ */
+ protected abstract void initializeFormatWriter() throws IOException;
+
+ /**
+ * Write one finished batch to the native format writer.
+ *
+ * @param batch The VectorSchemaRoot holding the batch; row count is already set
+ */
+ protected abstract void writeBatch(VectorSchemaRoot batch) throws IOException;
+
+ /**
+ * Close the native format writer, finalizing the file. Called unconditionally during
+ * {@link #close()}; implementations must no-op when the writer was never initialized.
+ */
+ protected abstract void closeFormatWriter() throws Exception;
+
+ /**
+ * Subclass hook invoked once during {@link #close()}, after remaining records are flushed and
+ * empty-file handling, before the format writer closes. Formats that support file-level
+ * key-value metadata persist footer entries (bloom filter, min/max keys, ...) here.
+ * Default implementation does nothing.
+ */
+ protected void finalizeFooterMetadata() throws IOException {
+ }
+
+ /**
+ * Subclass hook for closing format-specific resources that must be released after the format
+ * writer closes but before the Arrow data buffers are freed. Called unconditionally during
+ * {@link #close()} in its own try/catch, so failures here are recorded alongside (not instead
+ * of) other close failures. Default implementation does nothing.
+ */
+ protected void closeFormatResources() throws Exception {
+ }
+
+ /**
+ * Subclass hook for cleanup that must run after the data allocator has closed (e.g. a
+ * dedicated native-side allocator). Called last during {@link #close()}; exceptions thrown
+ * here propagate directly, so implementations should catch what they intend to tolerate.
+ * Default implementation does nothing.
+ */
+ protected void closeAfterAllocator() {
+ }
+
+ /**
+ * Write a single record. Records are buffered and flushed in batches.
+ *
+ * @param record Record to write
+ * @throws IOException if write fails
+ */
+ public void write(R record) throws IOException {
+ // Lazy initialization on first write
+ if (!isFormatWriterInitialized()) {
+ initializeFormatWriter();
+ }
+ if (root == null) {
+ root = VectorSchemaRoot.create(getArrowSchema(), allocator);
+ }
+ if (arrowWriter == null) {
+ arrowWriter = createArrowWriter(root);
+ }
+
+ // Reset arrow writer at the start of each new batch
+ if (currentBatchSize == 0) {
+ arrowWriter.reset();
+ }
+
+ arrowWriter.write(record);
+ currentBatchSize++;
+ writtenRecordCount++;
+
+ // Flush when row-count batch is full OR in-flight Arrow buffers cross the byte watermark.
+ // The byte-based check bounds in-flight memory so Arrow's power-of-2 vector reallocation
+ // can't escalate to a chunk size above the allocator cap regardless of per-row payload.
+ if (currentBatchSize >= batchSize || currentBufferBytes() >= flushByteWatermark) {
+ flushBatch();
+ }
+ }
+
+ /**
+ * Bytes currently held by the writer's Arrow child allocator. Used to drive the byte-aware
+ * flush.
+ *
+ * <p>Note: we deliberately do <em>not</em> use {@code FieldVector.getBufferSize()} here.
+ * For variable-width vectors (e.g. {@code BaseLargeVariableWidthVector} backing BLOB columns)
+ * that method short-circuits to 0 when {@code valueCount == 0}, and {@code valueCount} is only
+ * set during {@link ArrowWriter#finishBatch()} — i.e. at flush time. Mid-batch it always
+ * reports zero, so a watermark driven by it never fires. {@link BufferAllocator#getAllocatedMemory()}
+ * tracks the underlying ArrowBuf capacities directly and is exactly the quantity the allocator
+ * cap is enforced against.
+ */
+ private long currentBufferBytes() {
+ return allocator.getAllocatedMemory();
+ }
+
+ /**
+ * Close the writer, flushing any remaining buffered records.
+ *
+ * @throws IOException if close fails
+ */
+ @Override
+ public void close() throws IOException {
+ Exception primaryException = null;
+
+ // 1. Flush remaining records
+ try {
+ // Flush any remaining records in current batch
+ if (currentBatchSize > 0) {
+ flushBatch();
+ }
+
+ // Ensure writer is initialized even if no data was written
+ // This creates an empty file with just schema metadata
+ if (!isFormatWriterInitialized() && root == null) {
+ initializeFormatWriter();
+ root = VectorSchemaRoot.create(getArrowSchema(), allocator);
+ root.setRowCount(0);
+ writeBatch(root);
Review Comment:
This is the one path the extraction reshapes for Vortex: the empty root now lives until after the native writer closes, where the old try-with-resources released it first. TestVortexReaderWriterRoundTrip has no zero-record case - could you add a Vortex empty-write test mirroring TestHoodieSparkLanceWriter.testWriteEmptyDataset?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org