[core] Support row-count based data file rolling (write.target-row-num-per-file)

docs/generated/core_configuration.html

@@ -1837,6 +1837,12 @@
             <td><p>Enum</p></td>
             <td>Specify how to initialize the next sequence number for primary key table writers.<br /><br />Possible values:<ul><li>"scan": initialize by scanning existing file metadata.</li><li>"snapshot": initialize from the maximum sequence number recorded in snapshot properties, which can avoid scanning existing file metadata in write-only mode.</li></ul></td>
         </tr>
+        <tr>
+            <td><h5>write.target-row-num-per-file</h5></td>
+            <td style="word-wrap: break-word;">9223372036854775807</td>
+            <td>Long</td>
+            <td>Target number of rows per newly written 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. Only constrains files at write time: compaction is size-based and may merge into larger files, and data-evolution compaction still produces a single file. Bounds per-file rows for wide columns to avoid data-evolution OOM. PyPaimon file-store writers do not support this option and fail fast when it is enabled. Disabled by default.</td>
+        </tr>
         <tr>
             <td><h5>zorder.var-length-contribution</h5></td>
             <td style="word-wrap: break-word;">8</td>

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

@@ -818,6 +818,21 @@ public InlineElement getDescription() {
                                             text("append table: the default value is 256 MB."))
                                     .build());
 
+    public static final ConfigOption<Long> WRITE_TARGET_ROW_NUM_PER_FILE =
+            key("write.target-row-num-per-file")
+                    .longType()
+                    .defaultValue(Long.MAX_VALUE)
+                    .withDescription(
+                            "Target number of rows per newly written 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. Only constrains files at write time: "
+                                    + "compaction is size-based and may merge into larger files, and "
+                                    + "data-evolution compaction still produces a single file. Bounds "
+                                    + "per-file rows for wide columns to avoid data-evolution OOM. "
+                                    + "PyPaimon file-store writers do not support this option and "
+                                    + "fail fast when it is enabled. Disabled by default.");
+
     public static final ConfigOption<Double> COMPACTION_SMALL_FILE_RATIO =
             key("compaction.small-file-ratio")
                     .doubleType()
@@ -3325,6 +3340,10 @@ public long targetFileSize(boolean hasPrimaryKey) {
                 .getBytes();
     }
 
+    public long writeTargetRowNumPerFile() {
+        return options.get(WRITE_TARGET_ROW_NUM_PER_FILE);
+    }
+
     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 targetRowNumPerFile;
     private final RowType writeSchema;
     @Nullable private final List<String> writeCols;
     private final DataFilePathFactory pathFactory;
@@ -112,6 +113,7 @@ public AppendOnlyWriter(
             long targetFileSize,
             long blobTargetFileSize,
             long vectorTargetFileSize,
+            long targetRowNumPerFile,
             RowType writeSchema,
             @Nullable List<String> writeCols,
             long maxSequenceNumber,
@@ -139,6 +141,7 @@ public AppendOnlyWriter(
         this.targetFileSize = targetFileSize;
         this.blobTargetFileSize = blobTargetFileSize;
         this.vectorTargetFileSize = vectorTargetFileSize;
+        this.targetRowNumPerFile = targetRowNumPerFile;
         this.writeSchema = writeSchema;
         this.writeCols = writeCols;
         this.pathFactory = pathFactory;
@@ -325,6 +328,7 @@ private RollingFileWriter<InternalRow, DataFileMeta> createRollingRowWriter() {
                     targetFileSize,
                     blobTargetFileSize,
                     vectorTargetFileSize,
+                    targetRowNumPerFile,
                     writeSchema,
                     pathFactory,
                     seqNumCounterProvider,
@@ -350,7 +354,8 @@ private RollingFileWriter<InternalRow, DataFileMeta> createRollingRowWriter() {
                 asyncFileWrite,
                 statsDenseStore,
                 writeCols,
-                rowSidecarFileFormat);
+                rowSidecarFileFormat,
+                targetRowNumPerFile);
     }
 
     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 targetRowNumPerFile;
 
     // State management
     private final List<FileWriterAbortExecutor> closedWriters;
@@ -121,6 +122,7 @@ public class DedicatedFormatRollingFileWriter
                     RollingFileWriterImpl<InternalRow, DataFileMeta>, List<DataFileMeta>>
             vectorStoreWriter;
     private long recordCount = 0;
+    private long currentFileRecordCount = 0;
     private boolean closed = false;
 
     public DedicatedFormatRollingFileWriter(
@@ -131,6 +133,7 @@ public DedicatedFormatRollingFileWriter(
             long targetFileSize,
             long blobTargetFileSize,
             long vectorTargetFileSize,
+            long targetRowNumPerFile,
             RowType writeSchema,
             DataFilePathFactory pathFactory,
             Supplier<LongCounter> seqNumCounterSupplier,
@@ -141,7 +144,12 @@ public DedicatedFormatRollingFileWriter(
             boolean statsDenseStore,
             @Nullable BlobFileContext context) {
         // Initialize basic fields
+        Preconditions.checkArgument(
+                targetRowNumPerFile > 0,
+                "targetRowNumPerFile must be positive, but is %s",
+                targetRowNumPerFile);
         this.targetFileSize = targetFileSize;
+        this.targetRowNumPerFile = targetRowNumPerFile;
         this.results = new ArrayList<>();
         this.closedWriters = new ArrayList<>();
 
@@ -319,7 +327,8 @@ public DedicatedFormatRollingFileWriter(
                                         statsDenseStore,
                                         pathFactory.isExternalPath(),
                                         vectorStoreColumnNames),
-                        targetFileSize),
+                        targetFileSize,
+                        Long.MAX_VALUE),
                 vectorStoreProjection);
     }
 
@@ -352,8 +361,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 +426,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 >= targetRowNumPerFile) {
+            return true;
+        }
+        return currentWriter != null
+                && currentWriter
+                        .writer()
+                        .reachTargetSize(
+                                recordCount % CHECK_ROLLING_RECORD_CNT == 0, targetFileSize);
     }
 
     /**
@@ -455,6 +473,7 @@ private void closeCurrentWriter() throws IOException {
 
         // Reset current writer
         currentWriter = null;
+        currentFileRecordCount = 0;
     }
 
     /** Closes the main writer and returns its metadata. */
@@ -471,6 +490,7 @@ private List<DataFileMeta> closeBlobWriter() throws IOException {
         }
         blobWriter.close();
         List<DataFileMeta> results = blobWriter.result();
+        closedWriters.addAll(blobWriter.drainAbortExecutors());
         blobWriter = null;
         return results;
     }
@@ -482,6 +502,7 @@ private List<DataFileMeta> closeVectorStoreWriter() throws IOException {
         }
         vectorStoreWriter.close();
         List<DataFileMeta> results = vectorStoreWriter.result();
+        closedWriters.addAll(vectorStoreWriter.writer().drainAbortExecutors());
         vectorStoreWriter = null;
         return results;
     }

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

@@ -26,6 +26,7 @@
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.FileWriterAbortExecutor;
 import org.apache.paimon.io.RollingFileWriter;
 import org.apache.paimon.io.RollingFileWriterImpl;
 import org.apache.paimon.io.RowDataFileWriter;
@@ -131,14 +132,24 @@ public List<DataFileMeta> result() throws IOException {
         return results;
     }
 
+    List<FileWriterAbortExecutor> drainAbortExecutors() {
+        List<FileWriterAbortExecutor> abortExecutors = new ArrayList<>();
+        for (BlobProjectedFileWriter blobWriter : blobWriters) {
+            abortExecutors.addAll(blobWriter.writer().drainAbortExecutors());
+        }
+        return abortExecutors;
+    }
+
     private static class BlobProjectedFileWriter
             extends ProjectedFileWriter<
                     RollingFileWriterImpl<InternalRow, DataFileMeta>, List<DataFileMeta>> {
         public BlobProjectedFileWriter(
                 Supplier<? extends SingleFileWriter<InternalRow, DataFileMeta>> writerFactory,
                 long targetFileSize,
                 int[] projection) {
-            super(new RollingFileWriterImpl<>(writerFactory, targetFileSize), projection);
+            super(
+                    new RollingFileWriterImpl<>(writerFactory, targetFileSize, Long.MAX_VALUE),
+                    projection);
         }
     }
 }

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

@@ -61,6 +61,9 @@ 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.WRITE_TARGET_ROW_NUM_PER_FILE.key(), String.valueOf(Long.MAX_VALUE));
         return Collections.unmodifiableMap(options);
     }
 

paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java

@@ -442,6 +442,7 @@ public void createTable(Identifier identifier, Schema schema, boolean ignoreIfEx
         }
 
         copyTableDefaultOptions(schema.options());
+        validateCreateTable(schema, false);
 
         switch (Options.fromMap(schema.options()).get(TYPE)) {
             case TABLE:
@@ -516,6 +517,7 @@ public void replaceTable(Identifier identifier, Schema newSchema, boolean ignore
         validateCreateTable(newSchema, false);
         validateCustomTablePath(newSchema.options());
         copyTableDefaultOptions(newSchema.options());
+        validateCreateTable(newSchema, false);
 
         Table existing;
         try {

paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java

@@ -159,6 +159,10 @@ public static void validateCreateTable(Schema schema, boolean dataTokenEnabled)
                 "The value of %s property should be %s.",
                 AUTO_CREATE.key(),
                 Boolean.FALSE);
+        checkArgument(
+                options.get(CoreOptions.WRITE_TARGET_ROW_NUM_PER_FILE) > 0,
+                "%s should be at least 1.",
+                CoreOptions.WRITE_TARGET_ROW_NUM_PER_FILE.key());
 
         TableType tableType = options.get(CoreOptions.TYPE);
         if (tableType.equals(TableType.FORMAT_TABLE)) {

paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java

@@ -172,7 +172,9 @@ public List<IcebergManifestFileMeta> rollingWrite(
             Iterator<IcebergManifestEntry> entries, long sequenceNumber, Content content) {
         RollingFileWriterImpl<IcebergManifestEntry, IcebergManifestFileMeta> writer =
                 new RollingFileWriterImpl<>(
-                        () -> createWriter(sequenceNumber, content), targetFileSize.getBytes());
+                        () -> createWriter(sequenceNumber, content),
+                        targetFileSize.getBytes(),
+                        Long.MAX_VALUE);
         try {
             writer.write(entries);
             writer.close();

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

@@ -23,6 +23,7 @@
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.TwoPhaseOutputStream;
 import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Preconditions;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -32,31 +33,38 @@
 import java.util.List;
 import java.util.function.Supplier;
 
-/**
- * Format table's writer to roll over to a new file if the current size exceed the target file size.
- */
+/** Format table's writer which rolls files by target size or target row count. */
 public class FormatTableRollingFileWriter implements AutoCloseable {
 
     private static final Logger LOG = LoggerFactory.getLogger(FormatTableRollingFileWriter.class);
 
     private static final int CHECK_ROLLING_RECORD_CNT = 1000;
 
+    private final FileIO fileIO;
     private final Supplier<FormatTableSingleFileWriter> writerFactory;
     private final long targetFileSize;
+    private final long targetRowNumPerFile;
     private final List<FileWriterAbortExecutor> closedWriters;
     private final List<TwoPhaseOutputStream.Committer> committers;
 
     private FormatTableSingleFileWriter currentWriter = null;
     private long recordCount = 0;
+    private long currentFileRecordCount = 0;
     private boolean closed = false;
 
     public FormatTableRollingFileWriter(
             FileIO fileIO,
             FileFormat fileFormat,
             long targetFileSize,
+            long targetRowNumPerFile,
             RowType writeSchema,
             DataFilePathFactory pathFactory,
             String fileCompression) {
+        Preconditions.checkArgument(
+                targetRowNumPerFile > 0,
+                "targetRowNumPerFile must be positive, but is %s",
+                targetRowNumPerFile);
+        this.fileIO = fileIO;
         this.writerFactory =
                 () ->
                         new FormatTableSingleFileWriter(
@@ -65,6 +73,7 @@ public FormatTableRollingFileWriter(
                                 pathFactory.newPath(),
                                 fileCompression);
         this.targetFileSize = targetFileSize;
+        this.targetRowNumPerFile = targetRowNumPerFile;
         this.closedWriters = new ArrayList<>();
         this.committers = new ArrayList<>();
     }
@@ -81,9 +90,11 @@ public void write(InternalRow row) throws IOException {
 
             currentWriter.write(row);
             recordCount += 1;
+            currentFileRecordCount += 1;
             boolean needRolling =
-                    currentWriter.reachTargetSize(
-                            recordCount % CHECK_ROLLING_RECORD_CNT == 0, targetFileSize);
+                    currentFileRecordCount >= targetRowNumPerFile
+                            || currentWriter.reachTargetSize(
+                                    recordCount % CHECK_ROLLING_RECORD_CNT == 0, targetFileSize);
             if (needRolling) {
                 closeCurrentWriter();
             }
@@ -109,15 +120,26 @@ private void closeCurrentWriter() throws IOException {
         }
 
         currentWriter = null;
+        currentFileRecordCount = 0;
     }
 
     public void abort() {
         if (currentWriter != null) {
             currentWriter.abort();
+            currentWriter = null;
+        }
+        for (TwoPhaseOutputStream.Committer committer : committers) {
+            try {
+                committer.discard(fileIO);
+            } catch (Throwable e) {
+                LOG.warn("Exception occurs when discarding file {}.", committer.targetPath(), e);
+            }
         }
+        committers.clear();
         for (FileWriterAbortExecutor abortExecutor : closedWriters) {
             abortExecutor.abort();
         }
+        closedWriters.clear();
     }
 
     public List<TwoPhaseOutputStream.Committer> committers() {