This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 79fee924cd [vortex] Honor write.batch-memory in the writer factory
(#9605)
79fee924cd is described below
commit 79fee924cd9f8a34abc24c0721c51e03c998a980
Author: Eunbin Son <[email protected]>
AuthorDate: Mon Sep 7 23:44:07 2026 +0900
[vortex] Honor write.batch-memory in the writer factory (#9605)
---
.../paimon/format/vortex/VortexFileFormat.java | 8 +++++-
.../paimon/format/vortex/VortexWriterFactory.java | 6 ++++
.../paimon/format/vortex/VortexFileFormatTest.java | 33 ++++++++++++++++++++++
3 files changed, 46 insertions(+), 1 deletion(-)
diff --git
a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileFormat.java
b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileFormat.java
index 7f13b14a6d..d32c1669ad 100644
---
a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileFormat.java
+++
b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexFileFormat.java
@@ -79,7 +79,13 @@ public class VortexFileFormat extends FileFormat {
@Override
public FormatWriterFactory createWriterFactory(RowType type) {
return new VortexWriterFactory(
- () -> new ArrowFormatCWriter(type,
formatContext.writeBatchSize(), true));
+ () ->
+ new ArrowFormatCWriter(
+ type,
+ formatContext.writeBatchSize(),
+ true,
+ formatContext.writeBatchMemory().getBytes(),
+ null));
}
@Override
diff --git
a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexWriterFactory.java
b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexWriterFactory.java
index e3840acd36..84798b2cea 100644
---
a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexWriterFactory.java
+++
b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexWriterFactory.java
@@ -18,6 +18,7 @@
package org.apache.paimon.format.vortex;
+import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.arrow.vector.ArrowFormatCWriter;
import org.apache.paimon.format.FormatWriter;
import org.apache.paimon.format.FormatWriterFactory;
@@ -42,6 +43,11 @@ public class VortexWriterFactory implements
FormatWriterFactory, SupportsDirectW
this.cWriterSupplier = cWriterSupplier;
}
+ @VisibleForTesting
+ Supplier<ArrowFormatCWriter> cWriterSupplier() {
+ return cWriterSupplier;
+ }
+
@Override
public FormatWriter create(PositionOutputStream positionOutputStream,
String compression) {
throw new UnsupportedOperationException(
diff --git
a/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexFileFormatTest.java
b/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexFileFormatTest.java
index 30f1c90db0..9b7965a3c8 100644
---
a/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexFileFormatTest.java
+++
b/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexFileFormatTest.java
@@ -18,13 +18,19 @@
package org.apache.paimon.format.vortex;
+import org.apache.paimon.arrow.vector.ArrowFormatCWriter;
+import org.apache.paimon.data.GenericRow;
import org.apache.paimon.format.FileFormatFactory;
+import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -64,6 +70,33 @@ public class VortexFileFormatTest {
assertDoesNotThrow(() -> format.createWriterFactory(rowType));
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testWriterFactoryHonorsWriteBatchMemory(boolean limitMemory) {
+ int writeBatchSize = 1024;
+ FileFormatFactory.FormatContext formatContext =
+ limitMemory
+ ? new FileFormatFactory.FormatContext(
+ new Options(), 1024, writeBatchSize,
MemorySize.parse("1 kb"))
+ : new FileFormatFactory.FormatContext(new Options(),
1024, writeBatchSize);
+ VortexFileFormat format = new VortexFileFormat(formatContext);
+ RowType rowType = RowType.of(DataTypes.BYTES(), DataTypes.BYTES());
+ VortexWriterFactory factory = (VortexWriterFactory)
format.createWriterFactory(rowType);
+
+ GenericRow row = new GenericRow(2);
+ row.setField(0, new byte[1024]);
+ row.setField(1, new byte[1024]);
+
+ try (ArrowFormatCWriter writer = factory.cWriterSupplier().get()) {
+ // the memory limit is only re-checked every 32 rows
+ for (int i = 0; i < 32; i++) {
+ assertThat(writer.write(row)).isTrue();
+ }
+ // this row is still far below write-batch-size, so only the
memory limit can reject it
+ assertThat(writer.write(row)).isEqualTo(!limitMemory);
+ }
+ }
+
@Test
public void testValidateDataFields_UnsupportedMapType() {
VortexFileFormat format =
