From 86cf34507480efb028ba700d4513bb92fdcde897 Mon Sep 17 00:00:00 2001
From: Eunbin Son <thswlsqls@naver.com>
Date: Thu, 3 Sep 2026 12:27:20 +0900
Subject: [PATCH] [vortex] Honor write.batch-memory in the writer factory

VortexFileFormat.createWriterFactory built ArrowFormatCWriter with the
three-argument constructor, so the write.batch-memory limit was dropped
and only write.batch-size bounded a batch. Pass the limit through, as
orc, lance and mosaic already do.

Generated-by: Claude Code
---
 .../format/vortex/VortexFileFormat.java       |  8 ++++-
 .../format/vortex/VortexWriterFactory.java    |  6 ++++
 .../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 7f13b14a6de5..d32c1669ad75 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 FormatReaderFactory createReaderFactory(
     @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 e3840acd366c..84798b2cea7b 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 VortexWriterFactory(Supplier<ArrowFormatCWriter> cWriterSupplier) {
         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 30f1c90db0eb..9b7965a3c8a6 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 void testCreateWriterFactory() {
         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 =