[core] Support row-count based data file rolling (target-file-num-rows)

docs/generated/core_configuration.html

@@ -1681,6 +1681,12 @@
             <td>Boolean</td>
             <td>Whether to enable tag expiration by retained time.</td>
         </tr>
+        <tr>
+            <td><h5>target-file-num-rows</h5></td>
+            <td style="word-wrap: break-word;">9223372036854775807</td>
+            <td>Long</td>
+            <td>Target number of rows per data file; a file rolls when this or target-file-size is reached, whichever comes first. Enforced at bundle granularity, so a bundled write may exceed it by up to one bundle. Bounds per-file rows for wide columns to avoid data-evolution OOM. Disabled by default.</td>
+        </tr>
         <tr>
             <td><h5>target-file-size</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -818,6 +818,17 @@ public InlineElement getDescription() {
                                             text("append table: the default value is 256 MB."))
                                     .build());
 
+    public static final ConfigOption<Long> TARGET_FILE_NUM_ROWS =
+            key("target-file-num-rows")
+                    .longType()
+                    .defaultValue(Long.MAX_VALUE)
+                    .withDescription(
+                            "Target number of rows per data file; a file rolls when this or "
+                                    + "target-file-size is reached, whichever comes first. Enforced at "
+                                    + "bundle granularity, so a bundled write may exceed it by up to one "
+                                    + "bundle. Bounds per-file rows for wide columns to avoid "
+                                    + "data-evolution OOM. Disabled by default.");
+
     public static final ConfigOption<Double> COMPACTION_SMALL_FILE_RATIO =
             key("compaction.small-file-ratio")
                     .doubleType()
@@ -3325,6 +3336,10 @@ public long targetFileSize(boolean hasPrimaryKey) {
                 .getBytes();
     }
 
+    public long targetFileNumRows() {
+        return options.get(TARGET_FILE_NUM_ROWS);
+    }
+
     public long blobTargetFileSize() {
         return options.getOptional(BLOB_TARGET_FILE_SIZE)
                 .map(MemorySize::getBytes)

paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java

@@ -77,6 +77,7 @@ public class AppendOnlyWriter implements BatchRecordWriter, MemoryOwner {
     private final long targetFileSize;
     private final long blobTargetFileSize;
     private final long vectorTargetFileSize;
+    private final long targetFileNumRows;
     private final RowType writeSchema;
     @Nullable private final List<String> writeCols;
     private final DataFilePathFactory pathFactory;
@@ -103,6 +104,7 @@ public class AppendOnlyWriter implements BatchRecordWriter, MemoryOwner {
     private SinkWriter<InternalRow> sinkWriter;
     private MemorySegmentPool memorySegmentPool;
 
+    /** Overload without a row limit, kept for backward compatibility. */
     public AppendOnlyWriter(
             FileIO fileIO,
             @Nullable IOManager ioManager,
@@ -132,13 +134,76 @@ public AppendOnlyWriter(
             boolean dataEvolutionEnabled,
             @Nullable FileFormat rowSidecarFileFormat,
             @Nullable BlobFileContext blobContext) {
+        this(
+                fileIO,
+                ioManager,
+                schemaId,
+                fileFormat,
+                vectorFileFormat,
+                targetFileSize,
+                blobTargetFileSize,
+                vectorTargetFileSize,
+                Long.MAX_VALUE,
+                writeSchema,
+                writeCols,
+                maxSequenceNumber,
+                compactManager,
+                dataFileRead,
+                forceCompact,
+                pathFactory,
+                increment,
+                useWriteBuffer,
+                spillable,
+                fileCompression,
+                spillCompression,
+                statsCollectorFactories,
+                maxDiskSize,
+                fileIndexOptions,
+                asyncFileWrite,
+                statsDenseStore,
+                dataEvolutionEnabled,
+                rowSidecarFileFormat,
+                blobContext);
+    }
+
+    public AppendOnlyWriter(
+            FileIO fileIO,
+            @Nullable IOManager ioManager,
+            long schemaId,
+            FileFormat fileFormat,
+            @Nullable FileFormat vectorFileFormat,
+            long targetFileSize,
+            long blobTargetFileSize,
+            long vectorTargetFileSize,
+            long targetFileNumRows,
+            RowType writeSchema,
+            @Nullable List<String> writeCols,
+            long maxSequenceNumber,
+            CompactManager compactManager,
+            IOFunction<List<DataFileMeta>, RecordReaderIterator<InternalRow>> dataFileRead,
+            boolean forceCompact,
+            DataFilePathFactory pathFactory,
+            @Nullable CommitIncrement increment,
+            boolean useWriteBuffer,
+            boolean spillable,
+            String fileCompression,
+            CompressOptions spillCompression,
+            StatsCollectorFactories statsCollectorFactories,
+            MemorySize maxDiskSize,
+            FileIndexOptions fileIndexOptions,
+            boolean asyncFileWrite,
+            boolean statsDenseStore,
+            boolean dataEvolutionEnabled,
+            @Nullable FileFormat rowSidecarFileFormat,
+            @Nullable BlobFileContext blobContext) {
         this.fileIO = fileIO;
         this.schemaId = schemaId;
         this.fileFormat = fileFormat;
         this.vectorFileFormat = vectorFileFormat;
         this.targetFileSize = targetFileSize;
         this.blobTargetFileSize = blobTargetFileSize;
         this.vectorTargetFileSize = vectorTargetFileSize;
+        this.targetFileNumRows = targetFileNumRows;
         this.writeSchema = writeSchema;
         this.writeCols = writeCols;
         this.pathFactory = pathFactory;
@@ -325,6 +390,7 @@ private RollingFileWriter<InternalRow, DataFileMeta> createRollingRowWriter() {
                     targetFileSize,
                     blobTargetFileSize,
                     vectorTargetFileSize,
+                    targetFileNumRows,
                     writeSchema,
                     pathFactory,
                     seqNumCounterProvider,
@@ -350,7 +416,8 @@ private RollingFileWriter<InternalRow, DataFileMeta> createRollingRowWriter() {
                 asyncFileWrite,
                 statsDenseStore,
                 writeCols,
-                rowSidecarFileFormat);
+                rowSidecarFileFormat,
+                targetFileNumRows);
     }
 
     private void trySyncLatestCompaction(boolean blocking)

paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java

@@ -109,6 +109,7 @@ public class DedicatedFormatRollingFileWriter
                             RollingFileWriterImpl<InternalRow, DataFileMeta>, List<DataFileMeta>>>
             vectorStoreWriterFactory;
     private final long targetFileSize;
+    private final long targetFileNumRows;
 
     // State management
     private final List<FileWriterAbortExecutor> closedWriters;
@@ -121,8 +122,10 @@ public class DedicatedFormatRollingFileWriter
                     RollingFileWriterImpl<InternalRow, DataFileMeta>, List<DataFileMeta>>
             vectorStoreWriter;
     private long recordCount = 0;
+    private long currentFileRecordCount = 0;
     private boolean closed = false;
 
+    /** Overload without a row limit, kept for backward compatibility. */
     public DedicatedFormatRollingFileWriter(
             FileIO fileIO,
             long schemaId,
@@ -140,8 +143,51 @@ public DedicatedFormatRollingFileWriter(
             FileSource fileSource,
             boolean statsDenseStore,
             @Nullable BlobFileContext context) {
+        this(
+                fileIO,
+                schemaId,
+                fileFormat,
+                vectorFileFormat,
+                targetFileSize,
+                blobTargetFileSize,
+                vectorTargetFileSize,
+                Long.MAX_VALUE,
+                writeSchema,
+                pathFactory,
+                seqNumCounterSupplier,
+                fileCompression,
+                statsCollectorFactories,
+                fileIndexOptions,
+                fileSource,
+                statsDenseStore,
+                context);
+    }
+
+    public DedicatedFormatRollingFileWriter(
+            FileIO fileIO,
+            long schemaId,
+            FileFormat fileFormat,
+            @Nullable FileFormat vectorFileFormat,
+            long targetFileSize,
+            long blobTargetFileSize,
+            long vectorTargetFileSize,
+            long targetFileNumRows,
+            RowType writeSchema,
+            DataFilePathFactory pathFactory,
+            Supplier<LongCounter> seqNumCounterSupplier,
+            String fileCompression,
+            StatsCollectorFactories statsCollectorFactories,
+            FileIndexOptions fileIndexOptions,
+            FileSource fileSource,
+            boolean statsDenseStore,
+            @Nullable BlobFileContext context) {
         // Initialize basic fields
+        Preconditions.checkArgument(
+                targetFileNumRows > 0,
+                "targetFileNumRows must be positive, but is %s",
+                targetFileNumRows);
         this.targetFileSize = targetFileSize;
+        this.targetFileNumRows = targetFileNumRows;
         this.results = new ArrayList<>();
         this.closedWriters = new ArrayList<>();
 
@@ -352,8 +398,9 @@ public void write(InternalRow row) throws IOException {
                 currentWriter.write(row);
             }
             recordCount++;
+            currentFileRecordCount++;
 
-            if (currentWriter != null && rollingFile()) {
+            if (rollingFile()) {
                 closeCurrentWriter();
             }
         } catch (Throwable e) {
@@ -416,11 +463,19 @@ public void abort() {
         }
     }
 
-    /** Checks if the current file should be rolled based on size and record count. */
+    /**
+     * Checks if the current file should be rolled. The row cap applies even when there is no main
+     * writer (all fields dedicated), so blob/vector writers roll together.
+     */
     private boolean rollingFile() throws IOException {
-        return currentWriter
-                .writer()
-                .reachTargetSize(recordCount % CHECK_ROLLING_RECORD_CNT == 0, targetFileSize);
+        if (currentFileRecordCount >= targetFileNumRows) {
+            return true;
+        }
+        return currentWriter != null
+                && currentWriter
+                        .writer()
+                        .reachTargetSize(
+                                recordCount % CHECK_ROLLING_RECORD_CNT == 0, targetFileSize);
     }
 
     /**
@@ -455,6 +510,7 @@ private void closeCurrentWriter() throws IOException {
 
         // Reset current writer
         currentWriter = null;
+        currentFileRecordCount = 0;
     }
 
     /** Closes the main writer and returns its metadata. */

paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java

@@ -61,6 +61,8 @@ private static Map<String, String> dynamicWriteOptions() {
         Map<String, String> options = new HashMap<>();
         options.put(CoreOptions.TARGET_FILE_SIZE.key(), "99999 G");
         options.put(CoreOptions.BLOB_TARGET_FILE_SIZE.key(), "99999 G");
+        // Data evolution requires a single output file, so the row limit must not roll it either.
+        options.put(CoreOptions.TARGET_FILE_NUM_ROWS.key(), String.valueOf(Long.MAX_VALUE));
         return Collections.unmodifiableMap(options);
     }
 

paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java

@@ -142,7 +142,8 @@ public RollingFileWriter<KeyValue, DataFileMeta> createRollingMergeTreeFileWrite
                     return createDataFileWriter(
                             pathFactory.newPath(), key, fileSource, pathFactory.isExternalPath());
                 },
-                suggestedFileSize);
+                suggestedFileSize,
+                options.targetFileNumRows());
     }
 
     public RollingFileWriter<KeyValue, DataFileMeta> createRollingChangelogFileWriter(int level) {
@@ -156,7 +157,8 @@ public RollingFileWriter<KeyValue, DataFileMeta> createRollingChangelogFileWrite
                             FileSource.APPEND,
                             pathFactory.isExternalPath());
                 },
-                suggestedFileSize);
+                suggestedFileSize,
+                options.targetFileNumRows());
     }
 
     public RollingFileWriter<KeyValue, DataFileMeta> createRollingClusteringFileWriter() {

paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java

@@ -41,17 +41,31 @@ public class RollingFileWriterImpl<T, R> implements RollingFileWriter<T, R> {
 
     private final Supplier<? extends SingleFileWriter<T, R>> writerFactory;
     private final long targetFileSize;
+    private final long targetFileNumRows;
     private final List<FileWriterAbortExecutor> closedWriters;
     protected final List<R> results;
 
     private SingleFileWriter<T, R> currentWriter = null;
     private long recordCount = 0;
+    private long currentFileRecordCount = 0;
     private boolean closed = false;
 
     public RollingFileWriterImpl(
             Supplier<? extends SingleFileWriter<T, R>> writerFactory, long targetFileSize) {
+        this(writerFactory, targetFileSize, Long.MAX_VALUE);
+    }
+
+    public RollingFileWriterImpl(
+            Supplier<? extends SingleFileWriter<T, R>> writerFactory,
+            long targetFileSize,
+            long targetFileNumRows) {
+        Preconditions.checkArgument(
+                targetFileNumRows > 0,
+                "targetFileNumRows must be positive, but is %s",
+                targetFileNumRows);
         this.writerFactory = writerFactory;
         this.targetFileSize = targetFileSize;
+        this.targetFileNumRows = targetFileNumRows;
         this.results = new ArrayList<>();
         this.closedWriters = new ArrayList<>();
     }
@@ -62,8 +76,9 @@ public long targetFileSize() {
     }
 
     private boolean rollingFile(boolean forceCheck) throws IOException {
-        return currentWriter.reachTargetSize(
-                forceCheck || recordCount % CHECK_ROLLING_RECORD_CNT == 0, targetFileSize);
+        return currentFileRecordCount >= targetFileNumRows
+                || currentWriter.reachTargetSize(
+                        forceCheck || recordCount % CHECK_ROLLING_RECORD_CNT == 0, targetFileSize);
     }
 
     @Override
@@ -76,6 +91,7 @@ public void write(T row) throws IOException {
 
             currentWriter.write(row);
             recordCount += 1;
+            currentFileRecordCount += 1;
 
             if (rollingFile(false)) {
                 closeCurrentWriter();
@@ -101,6 +117,7 @@ public void writeBundle(BundleRecords bundle) throws IOException {
 
             currentWriter.writeBundle(bundle);
             recordCount += bundle.rowCount();
+            currentFileRecordCount += bundle.rowCount();
 
             if (rollingFile(true)) {
                 closeCurrentWriter();
@@ -132,6 +149,7 @@ protected void closeCurrentWriter() throws IOException {
         currentWriter.abortExecutor().ifPresent(closedWriters::add);
         results.add(currentWriter.result());
         currentWriter = null;
+        currentFileRecordCount = 0;
     }
 
     @Override