diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
index 0776eaf9e369..907f7512002f 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1681,6 +1681,12 @@
| Boolean |
Whether to enable tag expiration by retained time. |
+
+ target-file-num-rows |
+ 9223372036854775807 |
+ Long |
+ 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. Disabled by default. |
+
target-file-size |
(none) |
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index ec6a2296bf9a..2a566a591610 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -818,6 +818,20 @@ public InlineElement getDescription() {
text("append table: the default value is 256 MB."))
.build());
+ public static final ConfigOption TARGET_FILE_NUM_ROWS =
+ key("target-file-num-rows")
+ .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. "
+ + "Disabled by default.");
+
public static final ConfigOption COMPACTION_SMALL_FILE_RATIO =
key("compaction.small-file-ratio")
.doubleType()
@@ -3325,6 +3339,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)
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java b/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java
index ea81dea50f60..9ad51c552418 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java
+++ b/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 writeCols;
private final DataFilePathFactory pathFactory;
@@ -103,6 +104,7 @@ public class AppendOnlyWriter implements BatchRecordWriter, MemoryOwner {
private SinkWriter sinkWriter;
private MemorySegmentPool memorySegmentPool;
+ /** Overload without a row limit, kept for backward compatibility. */
public AppendOnlyWriter(
FileIO fileIO,
@Nullable IOManager ioManager,
@@ -132,6 +134,68 @@ 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 writeCols,
+ long maxSequenceNumber,
+ CompactManager compactManager,
+ IOFunction, RecordReaderIterator> 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;
@@ -139,6 +203,7 @@ public AppendOnlyWriter(
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 createRollingRowWriter() {
targetFileSize,
blobTargetFileSize,
vectorTargetFileSize,
+ targetFileNumRows,
writeSchema,
pathFactory,
seqNumCounterProvider,
@@ -350,7 +416,8 @@ private RollingFileWriter createRollingRowWriter() {
asyncFileWrite,
statsDenseStore,
writeCols,
- rowSidecarFileFormat);
+ rowSidecarFileFormat,
+ targetFileNumRows);
}
private void trySyncLatestCompaction(boolean blocking)
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java
index 9b8897862a8b..ac8d68fb00ae 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java
@@ -109,6 +109,7 @@ public class DedicatedFormatRollingFileWriter
RollingFileWriterImpl, List>>
vectorStoreWriterFactory;
private final long targetFileSize;
+ private final long targetFileNumRows;
// State management
private final List closedWriters;
@@ -121,8 +122,10 @@ public class DedicatedFormatRollingFileWriter
RollingFileWriterImpl, List>
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 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. */
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
index 77c3eaa91528..5826d7b8e0cf 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
@@ -61,6 +61,8 @@ private static Map dynamicWriteOptions() {
Map 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);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java
index dfe875120dcf..bc8701bf9377 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java
@@ -136,13 +136,17 @@ public DataFilePathFactory pathFactory(int level) {
public RollingFileWriter createRollingMergeTreeFileWriter(
int level, FileSource fileSource) {
WriteFormatKey key = new WriteFormatKey(level, false);
+ // Row limit applies to writes only; compaction output stays size-only.
+ long targetFileNumRows =
+ fileSource == FileSource.COMPACT ? Long.MAX_VALUE : options.targetFileNumRows();
return new RollingFileWriterImpl<>(
() -> {
DataFilePathFactory pathFactory = formatContext.pathFactory(key);
return createDataFileWriter(
pathFactory.newPath(), key, fileSource, pathFactory.isExternalPath());
},
- suggestedFileSize);
+ suggestedFileSize,
+ targetFileNumRows);
}
public RollingFileWriter createRollingChangelogFileWriter(int level) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java b/paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java
index 299a4f97421e..31b16d50f570 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java
@@ -41,17 +41,31 @@ public class RollingFileWriterImpl implements RollingFileWriter {
private final Supplier> writerFactory;
private final long targetFileSize;
+ private final long targetFileNumRows;
private final List closedWriters;
protected final List results;
private SingleFileWriter currentWriter = null;
private long recordCount = 0;
+ private long currentFileRecordCount = 0;
private boolean closed = false;
public RollingFileWriterImpl(
Supplier> writerFactory, long targetFileSize) {
+ this(writerFactory, targetFileSize, Long.MAX_VALUE);
+ }
+
+ public RollingFileWriterImpl(
+ Supplier> 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
diff --git a/paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java
index 0da1badf6229..bbaf7790974b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java
@@ -52,6 +52,42 @@ public RowDataRollingFileWriter(
boolean statsDenseStore,
@Nullable List writeCols,
@Nullable FileFormat rowSidecarFormat) {
+ this(
+ fileIO,
+ schemaId,
+ fileFormat,
+ targetFileSize,
+ writeSchema,
+ pathFactory,
+ seqNumCounterSupplier,
+ fileCompression,
+ statsCollectors,
+ fileIndexOptions,
+ fileSource,
+ asyncFileWrite,
+ statsDenseStore,
+ writeCols,
+ rowSidecarFormat,
+ Long.MAX_VALUE);
+ }
+
+ public RowDataRollingFileWriter(
+ FileIO fileIO,
+ long schemaId,
+ FileFormat fileFormat,
+ long targetFileSize,
+ RowType writeSchema,
+ DataFilePathFactory pathFactory,
+ Supplier seqNumCounterSupplier,
+ String fileCompression,
+ SimpleColStatsCollector.Factory[] statsCollectors,
+ FileIndexOptions fileIndexOptions,
+ FileSource fileSource,
+ boolean asyncFileWrite,
+ boolean statsDenseStore,
+ @Nullable List writeCols,
+ @Nullable FileFormat rowSidecarFormat,
+ long targetFileNumRows) {
super(
() -> {
Path dataPath = pathFactory.newPath();
@@ -76,6 +112,7 @@ public RowDataRollingFileWriter(
rowSidecarFormat,
rowSidecarPath);
},
- targetFileSize);
+ targetFileSize,
+ targetFileNumRows);
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
index 7dceb12e5c8b..45cdf5039ddc 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
@@ -160,6 +160,7 @@ protected RecordWriter createWriter(
options.targetFileSize(false),
options.blobTargetFileSize(),
options.vectorTargetFileSize(),
+ options.targetFileNumRows(),
writeType,
writeCols,
restoredMaxSeqNumber,
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index 4914ef1b40f5..4383771dc62e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -191,6 +191,10 @@ public static void validateTableSchema(TableSchema schema, Set dynamicOp
FULL_COMPACTION_DELTA_COMMITS.key()));
}
+ checkArgument(
+ options.targetFileNumRows() > 0,
+ CoreOptions.TARGET_FILE_NUM_ROWS.key() + " should be at least 1");
+
checkArgument(
options.snapshotNumRetainMin() > 0,
SNAPSHOT_NUM_RETAINED_MIN.key() + " should be at least 1");
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
index bd5d1d2cf54f..45f2a8a4b15f 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
@@ -1028,6 +1028,7 @@ private AppendOnlyWriter createSharedShreddingAppendWriter(
1024 * 1024L,
1024 * 1024L,
1024 * 1024L,
+ Long.MAX_VALUE,
writeType,
null,
maxSequenceNumber,
@@ -1238,6 +1239,7 @@ private Pair> createWriterBase(
targetFileSize,
targetFileSize,
targetFileSize,
+ Long.MAX_VALUE,
writeSchema,
null,
getMaxSequenceNumber(toCompact),
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterTest.java b/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterTest.java
index dbe9413804c6..b826800a889b 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterTest.java
@@ -147,6 +147,40 @@ public void testMultipleWrites() throws IOException {
metasResult.subList(1, 4).stream().mapToLong(DataFileMeta::rowCount).sum());
}
+ @Test
+ public void testRollingByRows() throws IOException {
+ CoreOptions options = new CoreOptions(new Options());
+ long hugeSize = 1024L * 1024 * 1024;
+ DedicatedFormatRollingFileWriter cappedWriter =
+ new DedicatedFormatRollingFileWriter(
+ LocalFileIO.create(),
+ SCHEMA_ID,
+ FileFormat.fromIdentifier("parquet", new Options()),
+ null,
+ hugeSize,
+ hugeSize,
+ hugeSize,
+ 5L,
+ SCHEMA,
+ pathFactory,
+ LongCounter::new,
+ COMPRESSION,
+ new StatsCollectorFactories(options),
+ new FileIndexOptions(),
+ FileSource.APPEND,
+ false,
+ BlobFileContext.create(SCHEMA, options));
+ for (int i = 0; i < 11; i++) {
+ cappedWriter.write(
+ GenericRow.of(i, BinaryString.fromString("t" + i), new BlobData(testBlobData)));
+ }
+ cappedWriter.close();
+ assertThat(cappedWriter.result())
+ .filteredOn(f -> "parquet".equals(f.fileFormat()))
+ .extracting(DataFileMeta::rowCount)
+ .containsExactly(5L, 5L, 1L);
+ }
+
@Test
public void testDoesNotWriteRowSidecar() throws IOException {
// Tests that: dedicated blob files do not create row-store sidecars.
diff --git a/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java b/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java
index 4b5c73038664..fa51b5732882 100644
--- a/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java
@@ -178,6 +178,29 @@ public void testWriteAndReadDataFileWithFileExtractingRollingFile() throws Excep
testWriteAndReadDataFileImpl("avro-extract");
}
+ @Test
+ public void testMergeTreeCompactionIgnoresRowLimit() throws Exception {
+ Options options = new Options();
+ options.set(CoreOptions.TARGET_FILE_NUM_ROWS, 2L);
+ KeyValueFileWriterFactory writerFactory =
+ createWriterFactory(tempDir.toString(), "avro", options);
+ List content = gen.next().content;
+
+ RollingFileWriter appendWriter =
+ writerFactory.createRollingMergeTreeFileWriter(0, FileSource.APPEND);
+ appendWriter.write(CloseableIterator.fromList(content, kv -> {}));
+ appendWriter.close();
+
+ RollingFileWriter compactWriter =
+ writerFactory.createRollingMergeTreeFileWriter(0, FileSource.COMPACT);
+ compactWriter.write(CloseableIterator.fromList(content, kv -> {}));
+ compactWriter.close();
+
+ // Writes roll by rows (cap=2); compaction output is size-only, so it is not row-split.
+ assertThat(appendWriter.result().size()).isGreaterThan(1);
+ assertThat(compactWriter.result()).hasSize(1);
+ }
+
private void testWriteAndReadDataFileImpl(String format) throws Exception {
DataFileTestDataGenerator.Data data = gen.next();
KeyValueFileWriterFactory writerFactory = createWriterFactory(tempDir.toString(), format);
@@ -341,6 +364,7 @@ public void testFileSuffix(@TempDir java.nio.file.Path tempDir) throws Exception
10,
10,
10,
+ Long.MAX_VALUE,
schema,
null,
0,
diff --git a/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java b/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java
index ec334007e507..b7ac286b08fd 100644
--- a/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java
@@ -72,6 +72,14 @@ public void initialize(String identifier) {
}
public void initialize(String identifier, boolean statsDenseStore) {
+ initialize(identifier, statsDenseStore, TARGET_FILE_SIZE, Long.MAX_VALUE);
+ }
+
+ public void initialize(
+ String identifier,
+ boolean statsDenseStore,
+ long targetFileSize,
+ long targetFileNumRows) {
FileFormat fileFormat = FileFormat.fromIdentifier(identifier, new Options());
rollingFileWriter =
new RollingFileWriterImpl<>(
@@ -106,7 +114,8 @@ public void initialize(String identifier, boolean statsDenseStore) {
statsDenseStore,
false,
null),
- TARGET_FILE_SIZE);
+ targetFileSize,
+ targetFileNumRows);
}
@ParameterizedTest
@@ -129,6 +138,45 @@ public void testRolling(String formatType) throws IOException {
}
}
+ @Test
+ public void testRollingByRows() throws IOException {
+ // Huge byte target so only the row-count limit triggers rolling.
+ initialize("avro", false, 1024L * 1024 * 1024, 100L);
+ for (int i = 0; i < 350; i++) {
+ rollingFileWriter.write(GenericRow.of(i));
+ }
+ rollingFileWriter.close();
+ List files = rollingFileWriter.result();
+ assertThat(files).hasSize(4);
+ assertThat(files.get(0).rowCount()).isEqualTo(100);
+ assertThat(files.get(1).rowCount()).isEqualTo(100);
+ assertThat(files.get(2).rowCount()).isEqualTo(100);
+ assertThat(files.get(3).rowCount()).isEqualTo(50);
+ }
+
+ @Test
+ public void testRollingByRowsWithBundle() throws IOException {
+ // Bundle-granular cap: a file may overshoot by up to one bundle.
+ initialize("avro", false, 1024L * 1024 * 1024, 100L);
+ rollingFileWriter.writeBundle(bundle(150));
+ rollingFileWriter.writeBundle(bundle(120));
+ rollingFileWriter.writeBundle(bundle(30));
+ rollingFileWriter.close();
+ List files = rollingFileWriter.result();
+ assertThat(files).hasSize(3);
+ assertThat(files.get(0).rowCount()).isEqualTo(150);
+ assertThat(files.get(1).rowCount()).isEqualTo(120);
+ assertThat(files.get(2).rowCount()).isEqualTo(30);
+ }
+
+ private static BundleRecords bundle(int rowCount) {
+ List rows = new ArrayList<>();
+ for (int i = 0; i < rowCount; i++) {
+ rows.add(GenericRow.of(i));
+ }
+ return new SingleUseBundleRecords(rows);
+ }
+
@Test
public void testWriteRowSidecar() throws IOException {
FileFormat fileFormat = FileFormat.fromIdentifier("parquet", new Options());
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index 36b8668bcc1f..78ccf0c5c69c 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -112,6 +112,14 @@ public void testFromTimestampConflict() {
"must set only one key in [scan.timestamp-millis,scan.timestamp] when you use from-timestamp for scan.mode");
}
+ @Test
+ public void testTargetFileNumRowsMustBePositive() {
+ Map options = new HashMap<>();
+ options.put(CoreOptions.TARGET_FILE_NUM_ROWS.key(), "0");
+ assertThatThrownBy(() -> validateTableSchemaExec(options))
+ .hasMessageContaining("target-file-num-rows should be at least 1");
+ }
+
@Test
public void testFromSnapshotConflict() {
String timestampString =
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
index 7766da98de24..b7db5b80e864 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
@@ -1126,6 +1126,90 @@ public void testCompactCoordinator() throws Exception {
assertThat(task.compactBefore().size()).isEqualTo(20);
}
+ @Test
+ public void testDataEvolutionReadWithRolledColumns() throws Exception {
+ createTableDefault();
+ FileStoreTable table =
+ getTableDefault()
+ .copy(
+ Collections.singletonMap(
+ CoreOptions.TARGET_FILE_NUM_ROWS.key(), "100"));
+ int count = 350;
+ RowType wt0 = table.schema().logicalRowType().project(Arrays.asList("f0", "f1"));
+ BatchWriteBuilder b0 = table.newBatchWriteBuilder();
+ try (BatchTableWrite w0 = b0.newWrite().withWriteType(wt0)) {
+ for (int i = 0; i < count; i++) {
+ w0.write(GenericRow.of(i, BinaryString.fromString("a" + i)));
+ }
+ b0.newCommit().commit(w0.prepareCommit());
+ }
+ long firstRowId = table.snapshotManager().latestSnapshot().nextRowId() - count;
+
+ RowType wt1 = table.schema().logicalRowType().project(Collections.singletonList("f2"));
+ BatchWriteBuilder b1 = table.newBatchWriteBuilder();
+ try (BatchTableWrite w1 = b1.newWrite().withWriteType(wt1)) {
+ for (int i = 0; i < count; i++) {
+ w1.write(GenericRow.of(BinaryString.fromString("b" + i)));
+ }
+ List msgs = w1.prepareCommit();
+ assignCumulativeFirstRowId(msgs, firstRowId);
+ b1.newCommit().commit(msgs);
+ }
+
+ ReadBuilder rb = table.newReadBuilder();
+ RecordReader reader = rb.newRead().createReader(rb.newScan().plan());
+ AtomicInteger cnt = new AtomicInteger(0);
+ reader.forEachRemaining(
+ r -> {
+ int i = r.getInt(0);
+ assertThat(r.getString(1).toString()).isEqualTo("a" + i);
+ assertThat(r.getString(2).toString()).isEqualTo("b" + i);
+ cnt.incrementAndGet();
+ });
+ assertThat(cnt.get()).isEqualTo(count);
+ }
+
+ private void assignCumulativeFirstRowId(List msgs, long firstRowId) {
+ long cur = firstRowId;
+ for (CommitMessage c : msgs) {
+ CommitMessageImpl m = (CommitMessageImpl) c;
+ List files = new ArrayList<>(m.newFilesIncrement().newFiles());
+ m.newFilesIncrement().newFiles().clear();
+ List assigned = new ArrayList<>();
+ for (DataFileMeta f : files) {
+ assigned.add(f.assignFirstRowId(cur));
+ cur += f.rowCount();
+ }
+ m.newFilesIncrement().newFiles().addAll(assigned);
+ }
+ }
+
+ @Test
+ public void testDataEvolutionWriteRollsByRows() throws Exception {
+ createTableDefault();
+ FileStoreTable table =
+ getTableDefault()
+ .copy(
+ Collections.singletonMap(
+ CoreOptions.TARGET_FILE_NUM_ROWS.key(), "100"));
+ RowType writeType = table.schema().logicalRowType().project(Arrays.asList("f0", "f1"));
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite().withWriteType(writeType)) {
+ for (int i = 0; i < 350; i++) {
+ write.write(GenericRow.of(i, BinaryString.fromString("a" + i)));
+ }
+ builder.newCommit().commit(write.prepareCommit());
+ }
+
+ List rowCounts = new ArrayList<>();
+ Iterator files = table.newSnapshotReader().readFileIterator();
+ while (files.hasNext()) {
+ rowCounts.add(files.next().file().rowCount());
+ }
+ assertThat(rowCounts.stream().mapToLong(Long::longValue).sum()).isEqualTo(350L);
+ assertThat(Collections.max(rowCounts)).isLessThanOrEqualTo(100L);
+ }
+
@Test
public void testCompact() throws Exception {
for (int i = 0; i < 5; i++) {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionPartialWriteOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionPartialWriteOperator.java
index f57393b7ca27..f3cd4c6f9b9c 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionPartialWriteOperator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionPartialWriteOperator.java
@@ -92,10 +92,17 @@ public class DataEvolutionPartialWriteOperator
private transient AbstractFileStoreWrite tableWrite;
private transient Writer writer;
+ private static Map dataEvolutionWriteOptions() {
+ // Data evolution requires a single output file: disable both size and row rolling.
+ Map options = new HashMap<>();
+ options.put(CoreOptions.TARGET_FILE_SIZE.key(), "99999 G");
+ options.put(CoreOptions.TARGET_FILE_NUM_ROWS.key(), String.valueOf(Long.MAX_VALUE));
+ return options;
+ }
+
public DataEvolutionPartialWriteOperator(
FileStoreTable table, RowType dataType, Long baseSnapshotId) {
- this.table =
- table.copy(Collections.singletonMap(CoreOptions.TARGET_FILE_SIZE.key(), "99999 G"));
+ this.table = table.copy(dataEvolutionWriteOptions());
this.baseSnapshotId = baseSnapshotId;
List fieldNames =
dataType.getFieldNames().stream()
diff --git a/paimon-python/pypaimon/tests/reader_append_only_test.py b/paimon-python/pypaimon/tests/reader_append_only_test.py
index 67bbfbedbdec..181a828f9801 100644
--- a/paimon-python/pypaimon/tests/reader_append_only_test.py
+++ b/paimon-python/pypaimon/tests/reader_append_only_test.py
@@ -71,6 +71,17 @@ def test_parquet_ao_reader(self):
actual = self._read_test_table(read_builder).sort_by('user_id')
self.assertEqual(actual, self.expected)
+ def test_target_file_num_rows_fails_fast(self):
+ schema = Schema.from_pyarrow_schema(
+ self.pa_schema, partition_keys=['dt'],
+ options={'target-file-num-rows': '5'})
+ self.catalog.create_table('default.test_target_file_num_rows', schema, False)
+ table = self.catalog.get_table('default.test_target_file_num_rows')
+ with self.assertRaises(NotImplementedError):
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_write.write_arrow(self.expected)
+
def test_orc_ao_reader(self):
schema = Schema.from_pyarrow_schema(self.pa_schema, partition_keys=['dt'], options={'file.format': 'orc'})
self.catalog.create_table('default.test_append_only_orc', schema, False)
diff --git a/paimon-python/pypaimon/write/file_store_write.py b/paimon-python/pypaimon/write/file_store_write.py
index 0895c47bf17e..edaadf7560cb 100644
--- a/paimon-python/pypaimon/write/file_store_write.py
+++ b/paimon-python/pypaimon/write/file_store_write.py
@@ -49,6 +49,12 @@ def __init__(self, table, commit_user):
self.blob_consumer = None
self.commit_identifier = 0
self.options = CoreOptions.copy(table.options)
+ # pypaimon has no row-count rolling yet; fail fast instead of silently ignoring the option.
+ row_cap = table.table_schema.options.get("target-file-num-rows")
+ if row_cap is not None and str(row_cap).strip() not in ("", str(2 ** 63 - 1)):
+ raise NotImplementedError(
+ "target-file-num-rows is set on this table but pypaimon writers do not support "
+ "row-count based file rolling yet; unset it or write with Java/Flink/Spark.")
self.changelog_producer = self.options.changelog_producer()
if self.table.bucket_mode() == BucketMode.POSTPONE_MODE:
self.options.set(CoreOptions.DATA_FILE_PREFIX,
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPaimonWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPaimonWriter.scala
index e99afc95b6e0..01e787a7d19b 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPaimonWriter.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPaimonWriter.scala
@@ -31,17 +31,19 @@ import org.apache.paimon.utils.SerializationUtils
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.analysis.SimpleAnalyzer.resolver
-import java.util.Collections
-
import scala.collection.JavaConverters._
import scala.collection.mutable
case class DataEvolutionPaimonWriter(paimonTable: FileStoreTable, dataSplits: Seq[DataSplit])
extends WriteHelper {
- // File rolling will never be performed
- override val table: FileStoreTable =
- paimonTable.copy(Collections.singletonMap(CoreOptions.TARGET_FILE_SIZE.key(), "99999 G"))
+ // File rolling will never be performed: disable both size and row rolling.
+ override val table: FileStoreTable = {
+ val writeOptions = Map(
+ CoreOptions.TARGET_FILE_SIZE.key() -> "99999 G",
+ CoreOptions.TARGET_FILE_NUM_ROWS.key() -> Long.MaxValue.toString)
+ paimonTable.copy(writeOptions.asJava)
+ }
def writePartialFields(
data: DataFrame,