@@ -137,8 +137,13 @@ derive the column and partition by it, for example `withColumn("year", year(col(
 |---------------------------|---------|-------------|
 | `batch.size`              | `8192`  | Rows buffered per write batch; range 1–65536. Each string or binary field of a batch must fit in 2 GiB, so lower it for values over about 256 KB. |
 | `vortex.write.batch.size` | —       | Rows buffered per write batch for Vortex alone, overriding `batch.size`. |
+| `vortex.write.batch.bytes` | —      | Positive byte threshold for the open batch's Arrow allocations; flush after a row reaches it or the row limit, whichever comes first. Unset disables byte-based flushing. |
 | `vortex.listLayout`       | `false` | Store array columns in the unstable list layout; see below. |
 
+The byte threshold includes nested buffers and Arrow's allocation rounding. It is checked after
+each complete row, so one row or a buffer growth can exceed it. It bounds batching rather than
+total process memory: native code may retain earlier batches and allocate compression buffers.
+
 By default an array column is stored as whole chunks, so reading one field of the structs in an
 array reads every field of them. With `vortex.listLayout` set to `true`, the writer stores each
 array's offsets, nulls, and elements apart, and splits struct elements into one column per field,
 
@@ -139,6 +139,24 @@ public int getInt(String key, int fallback) {
         }
     }
 
+    /**
+     * The value set for {@code key} as a long, or {@code fallback} when it is not set.
+     *
+     * @throws IllegalArgumentException if the value is not a long
+     */
+    public long getLong(String key, long fallback) {
+        Optional<String> value = get(key);
+        if (value.isEmpty()) {
+            return fallback;
+        }
+        try {
+            return Long.parseLong(value.get().trim());
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException(
+                    String.format(Locale.ROOT, "%s must be a long, got '%s'", key, value.get()), e);
+        }
+    }
+
     /**
      * The value set for {@code key} as a boolean, or {@code fallback} when it is not set.
      *
 
@@ -0,0 +1,77 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+package dev.vortex.spark.write;
+
+import dev.vortex.relocated.org.apache.arrow.memory.AllocationListener;
+import dev.vortex.relocated.org.apache.arrow.memory.ArrowBuf;
+import dev.vortex.relocated.org.apache.arrow.memory.ReferenceManager;
+import dev.vortex.relocated.org.apache.arrow.vector.BaseFixedWidthVector;
+import dev.vortex.relocated.org.apache.arrow.vector.BaseVariableWidthVector;
+import dev.vortex.relocated.org.apache.arrow.vector.FieldVector;
+import dev.vortex.relocated.org.apache.arrow.vector.VectorSchemaRoot;
+import dev.vortex.relocated.org.apache.arrow.vector.complex.ListVector;
+import dev.vortex.relocated.org.apache.arrow.vector.complex.StructVector;
+import java.util.IdentityHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+/** Measures the open batch independently of older batches retained or released by native code. */
+final class ArrowBatchMemory implements AllocationListener {
+    private final AtomicLong allocations = new AtomicLong();
+    private final IdentityHashMap<ReferenceManager, Boolean> seen = new IdentityHashMap<>();
+    private long measuredAllocations = -1;
+    private long bytes;
+
+    @Override
+    public void onAllocation(long size) {
+        allocations.incrementAndGet();
+    }
+
+    long bytes(VectorSchemaRoot root) {
+        long currentAllocations = allocations.get();
+        // Between buffer growths the batch owns the same allocations; do not walk a wide schema for every row.
+        if (currentAllocations != measuredAllocations) {
+            bytes = 0;
+            try {
+                for (FieldVector vector : root.getFieldVectors()) {
+                    bytes += bytes(vector);
+                }
+            } finally {
+                seen.clear();
+            }
+            measuredAllocations = currentAllocations;
+        }
+        return bytes;
+    }
+
+    void reset() {
+        measuredAllocations = -1;
+        bytes = 0;
+    }
+
+    private long bytes(FieldVector vector) {
+        // getFieldBuffers() finalizes variable-width offsets using valueCount, which is still zero mid-batch.
+        long total = bytes(vector.getValidityBuffer());
+        if (vector instanceof BaseFixedWidthVector) {
+            total += bytes(vector.getDataBuffer());
+        } else if (vector instanceof BaseVariableWidthVector) {
+            total += bytes(vector.getDataBuffer());
+            total += bytes(vector.getOffsetBuffer());
+        } else if (vector instanceof ListVector) {
+            total += bytes(vector.getOffsetBuffer());
+        } else if (!(vector instanceof StructVector)) {
+            throw new IllegalArgumentException(
+                    "Unsupported writer vector: " + vector.getClass().getName());
+        }
+        for (FieldVector child : vector.getChildrenFromFields()) {
+            total += bytes(child);
+        }
+        return total;
+    }
+
+    private long bytes(ArrowBuf buffer) {
+        ReferenceManager manager = buffer.getReferenceManager();
+        // Fixed-width vectors can slice validity and data out of one allocation.
+        return seen.put(manager, Boolean.TRUE) == null ? manager.getSize() : 0;
+    }
+}
@@ -40,6 +40,9 @@ public final class VortexOutputWriter extends OutputWriter {
     /** Option sizing the write batch for Vortex alone, overriding {@value #BATCH_SIZE_OPTION}. */
     public static final String WRITE_BATCH_SIZE_OPTION = "vortex.write.batch.size";
 
+    /** Optional positive byte threshold for the open batch's Arrow allocations, checked after each row. */
+    public static final String WRITE_BATCH_BYTES_OPTION = "vortex.write.batch.bytes";
+
     /**
      * Option storing list columns in the unstable list layout, so a read of some fields of array elements skips the
      * others. Off by default; future versions may not read files written with it.
@@ -54,6 +57,8 @@ public final class VortexOutputWriter extends OutputWriter {
     private final String filePath;
     private final VortexOptions options;
     private final int batchSize;
+    private final long batchBytes;
+    private final ArrowBatchMemory batchMemory;
     private NativeWritable writable;
 
     private Session session;
@@ -82,19 +87,16 @@ public VortexOutputWriter(String filePath, StructType schema, VortexOptions opti
         this.writable = writable;
 
         this.batchSize = configuredBatchSize(options);
+        this.batchBytes = configuredBatchBytes(options);
+        this.batchMemory = batchBytes == 0 ? null : new ArrowBatchMemory();
 
         try {
-            this.allocator = new RootAllocator();
+            this.allocator = batchMemory == null ? new RootAllocator() : new RootAllocator(batchMemory, Long.MAX_VALUE);
             var arrowSchema = SparkToArrowSchema.convert(schema);
 
             this.vectorSchemaRoot = VectorSchemaRoot.create(arrowSchema, allocator);
             // Small start so writers that see few rows hold little memory; Arrow grows on demand.
-            // It also keeps the largest capacity a batch needed, so one batch of huge values makes
-            // later batches allocate as much.
-            int initialCapacity = Math.min(batchSize, INITIAL_VECTOR_CAPACITY);
-            for (FieldVector vector : vectorSchemaRoot.getFieldVectors()) {
-                vector.setInitialCapacity(initialCapacity);
-            }
+            resetInitialCapacity();
             StructField[] fields = schema.fields();
             this.fieldWriters = new ArrowFieldWriter[fields.length];
             for (int i = 0; i < fields.length; i++) {
@@ -144,6 +146,19 @@ static int configuredBatchSize(VortexOptions options) {
         return batchSize;
     }
 
+    /** Returns the positive batch byte threshold, or zero when unset. */
+    static long configuredBatchBytes(VortexOptions options) {
+        if (options.get(WRITE_BATCH_BYTES_OPTION).isEmpty()) {
+            return 0;
+        }
+        long bytes = options.getLong(WRITE_BATCH_BYTES_OPTION, 0);
+        if (bytes <= 0) {
+            throw new IllegalArgumentException(
+                    WRITE_BATCH_BYTES_OPTION + " must be a positive number of bytes, got " + bytes);
+        }
+        return bytes;
+    }
+
     /**
      * Writes a single row to the Vortex file.
      *
@@ -171,8 +186,13 @@ public void write(InternalRow row) {
                 fieldWriters[i].write(row, i);
             }
             batchRowCount++;
-            if (batchRowCount >= batchSize) {
+            boolean reachedBytes = batchMemory != null && batchMemory.bytes(vectorSchemaRoot) >= batchBytes;
+            if (batchRowCount >= batchSize || reachedBytes) {
                 writeBatch();
+                if (reachedBytes) {
+                    // An oversized row must not make every later small batch allocate its remembered capacity.
+                    resetInitialCapacity();
+                }
             }
         } catch (IOException e) {
             writeFailure = e;
@@ -185,6 +205,13 @@ public void write(InternalRow row) {
         }
     }
 
+    private void resetInitialCapacity() {
+        int initialCapacity = Math.min(batchSize, INITIAL_VECTOR_CAPACITY);
+        for (FieldVector vector : vectorSchemaRoot.getFieldVectors()) {
+            vector.setInitialCapacity(initialCapacity);
+        }
+    }
+
     /** Hands the rows appended since the last batch to the Vortex writer. */
     private void writeBatch() throws IOException {
         if (batchRowCount == 0) {
@@ -202,6 +229,9 @@ private void writeBatch() throws IOException {
             writer.reset();
         }
         batchRowCount = 0;
+        if (batchMemory != null) {
+            batchMemory.reset();
+        }
     }
 
     private static long arrowBytes(List<FieldVector> vectors) {
@@ -296,6 +326,11 @@ BufferAllocator allocator() {
         return allocator;
     }
 
+    /** Package-private so tests can observe batch boundaries without depending on native scheduling. */
+    int batchRowCount() {
+        return batchRowCount;
+    }
+
     private IOException closeArrowResources(IOException failure) {
         if (vectorSchemaRoot != null) {
             try {
 
@@ -32,6 +32,7 @@ public VortexOutputWriterFactory(StructType schema, VortexOptions options) {
      */
     public static void validate(VortexOptions options) {
         VortexOutputWriter.configuredBatchSize(options);
+        VortexOutputWriter.configuredBatchBytes(options);
         VortexOutputWriter.listLayout(options);
         VortexSparkSession.validate(options);
     }
 
@@ -42,6 +42,7 @@ void anUnsetOptionFallsBack() {
         assertEquals(Optional.empty(), options.get("vortex.absent"));
         assertEquals("fallback", options.get("vortex.absent", "fallback"));
         assertEquals(7, options.getInt("vortex.absent", 7));
+        assertEquals(7, options.getLong("vortex.absent", 7));
         assertTrue(options.getBoolean("vortex.absent", true));
     }
 
@@ -53,6 +54,19 @@ void booleansAreReadWithoutRegardToCase() {
         assertFalse(options.getBoolean("off", true));
     }
 
+    @Test
+    void longOptionsAcceptValuesBeyondIntegerRangeAndTrimWhitespace() {
+        assertEquals(
+                3_000_000_000L,
+                VortexOptions.of(Map.of("BYTES", " 3000000000 ")).getLong("bytes", 0));
+        for (String value : new String[] {"many", "9223372036854775808"}) {
+            VortexOptions options = VortexOptions.of(Map.of("bytes", value));
+            assertTrue(assertThrows(IllegalArgumentException.class, () -> options.getLong("bytes", 0))
+                    .getMessage()
+                    .contains("bytes"));
+        }
+    }
+
     @Test
     void aValueOfTheWrongShapeNamesItsOption() {
         VortexOptions options = VortexOptions.of(Map.of("vortex.count", "many", "vortex.flag", "yes"));
 
@@ -53,6 +53,10 @@ public void tearDown() {
         "batch.size, lots",
         "batch.size, 0",
         "vortex.write.batch.size, 65537",
+        "vortex.write.batch.bytes, lots",
+        "vortex.write.batch.bytes, 0",
+        "vortex.write.batch.bytes, -1",
+        "vortex.write.batch.bytes, 9223372036854775808",
         "vortex.listLayout, yes",
         "vortex.session.provider, dev.vortex.spark.NoSuchProvider",
         "vortex.session.provider, java.lang.String",
 
@@ -0,0 +1,88 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+package dev.vortex.spark.write;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import dev.vortex.relocated.org.apache.arrow.memory.ArrowBuf;
+import dev.vortex.relocated.org.apache.arrow.memory.RootAllocator;
+import dev.vortex.relocated.org.apache.arrow.vector.BigIntVector;
+import dev.vortex.relocated.org.apache.arrow.vector.VarBinaryVector;
+import dev.vortex.relocated.org.apache.arrow.vector.VarCharVector;
+import dev.vortex.relocated.org.apache.arrow.vector.VectorSchemaRoot;
+import dev.vortex.relocated.org.apache.arrow.vector.complex.ListVector;
+import dev.vortex.relocated.org.apache.arrow.vector.complex.StructVector;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.Test;
+
+final class ArrowBatchMemoryTest {
+    @Test
+    void measuringAnUnfinishedBatchPreservesStringsAndBinaryAcrossBufferGrowth() {
+        ArrowBatchMemory memory = new ArrowBatchMemory();
+        StructType schema = new StructType().add("s", DataTypes.StringType).add("b", DataTypes.BinaryType);
+        try (RootAllocator allocator = new RootAllocator(memory, Long.MAX_VALUE);
+                VectorSchemaRoot root = VectorSchemaRoot.create(SparkToArrowSchema.convert(schema), allocator)) {
+            root.allocateNew();
+            VarCharVector strings = (VarCharVector) root.getVector("s");
+            VarBinaryVector binary = (VarBinaryVector) root.getVector("b");
+            strings.setSafe(0, "first".getBytes(StandardCharsets.UTF_8));
+            binary.setSafe(0, new byte[] {1, 2, 3});
+            memory.bytes(root);
+
+            String largeString = "x".repeat(100_000);
+            byte[] largeBinary = new byte[100_000];
+            Arrays.fill(largeBinary, (byte) 7);
+            strings.setSafe(1, largeString.getBytes(StandardCharsets.UTF_8));
+            binary.setSafe(1, largeBinary);
+            memory.bytes(root);
+            root.setRowCount(2);
+
+            assertEquals("first", strings.getObject(0).toString());
+            assertEquals(largeString, strings.getObject(1).toString());
+            assertArrayEquals(new byte[] {1, 2, 3}, binary.get(0));
+            assertArrayEquals(largeBinary, binary.get(1));
+        }
+    }
+
+    @Test
+    void measuresNestedAllocationsOnceAndIgnoresOlderBatchReleases() {
+        ArrowBatchMemory memory = new ArrowBatchMemory();
+        StructType schema = new StructType()
+                .add("id", DataTypes.LongType, false)
+                .add("nested", new StructType().add("items", DataTypes.createArrayType(DataTypes.LongType, false)));
+        try (RootAllocator allocator = new RootAllocator(memory, Long.MAX_VALUE);
+                VectorSchemaRoot root = VectorSchemaRoot.create(SparkToArrowSchema.convert(schema), allocator)) {
+            long batchBytes;
+            try (ArrowBuf previousBatch = allocator.buffer(1_000_000)) {
+                root.allocateNew();
+                batchBytes = memory.bytes(root);
+                assertEquals(
+                        allocator.getAllocatedMemory()
+                                - previousBatch.getReferenceManager().getSize(),
+                        batchBytes);
+                assertTrue(batchBytes > 0);
+            }
+            // A baseline subtraction would shrink here, although the open batch has not changed.
+            assertEquals(batchBytes, memory.bytes(root));
+            assertEquals(allocator.getAllocatedMemory(), memory.bytes(root));
+
+            StructVector nested = (StructVector) root.getVector("nested");
+            ListVector items = (ListVector) nested.getChild("items");
+            ((BigIntVector) items.getDataVector()).setSafe(100_000, 1L);
+            assertTrue(memory.bytes(root) > batchBytes);
+            assertEquals(allocator.getAllocatedMemory(), memory.bytes(root));
+
+            root.clear();
+            memory.reset();
+            assertEquals(0, memory.bytes(root));
+            root.allocateNew();
+            assertEquals(allocator.getAllocatedMemory(), memory.bytes(root));
+        }
+    }
+}
Original file line number	Diff line number	Diff line change
`@@ -32,6 +32,7 @@ public VortexOutputWriterFactory(StructType schema, VortexOptions options) {`
`32`	`32`	`*/`
`33`	`33`	`public static void validate(VortexOptions options) {`
`34`	`34`	`VortexOutputWriter.configuredBatchSize(options);`
	`35`	`+ VortexOutputWriter.configuredBatchBytes(options);`
`35`	`36`	`VortexOutputWriter.listLayout(options);`
`36`	`37`	`VortexSparkSession.validate(options);`
`37`	`38`	`}`