This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new a998338ec6 [core] Support row-count based data file rolling (#8739)
a998338ec6 is described below
commit a998338ec62f2cae5c0649e797eda4d02be5de36
Author: XiaoHongbo <[email protected]>
AuthorDate: Fri Jul 24 11:52:29 2026 +0800
[core] Support row-count based data file rolling (#8739)
---
docs/generated/core_configuration.html | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 19 +++
.../org/apache/paimon/append/AppendOnlyWriter.java | 7 +-
.../append/DedicatedFormatRollingFileWriter.java | 33 +++-
.../paimon/append/MultipleBlobFileWriter.java | 13 +-
.../dataevolution/DataEvolutionCompactTask.java | 2 +
.../org/apache/paimon/catalog/AbstractCatalog.java | 2 +
.../org/apache/paimon/catalog/CatalogUtils.java | 4 +
.../iceberg/manifest/IcebergManifestFile.java | 4 +-
.../paimon/io/FormatTableRollingFileWriter.java | 32 +++-
.../paimon/io/KeyValueFileWriterFactory.java | 12 +-
.../apache/paimon/io/RollingFileWriterImpl.java | 27 ++-
.../apache/paimon/io/RowDataRollingFileWriter.java | 6 +-
.../java/org/apache/paimon/jdbc/JdbcCatalog.java | 1 +
.../org/apache/paimon/manifest/ManifestFile.java | 3 +-
.../paimon/operation/BaseAppendFileStoreWrite.java | 4 +-
.../java/org/apache/paimon/rest/RESTCatalog.java | 2 +
.../org/apache/paimon/schema/SchemaValidation.java | 4 +
.../paimon/table/format/FormatTableFileWriter.java | 37 +++-
.../table/format/FormatTableRecordWriter.java | 11 +-
.../apache/paimon/append/AppendOnlyWriterTest.java | 2 +
.../DedicatedFormatRollingFileWriterTest.java | 74 ++++++++
...DedicatedFormatRollingFileWriterVectorTest.java | 42 +++++
.../apache/paimon/catalog/CatalogUtilsTest.java | 14 ++
.../paimon/catalog/FileSystemCatalogTest.java | 42 +++++
.../paimon/io/KeyValueFileReadWriteTest.java | 24 +++
.../apache/paimon/io/RollingFileWriterTest.java | 56 +++++-
.../apache/paimon/schema/SchemaValidationTest.java | 8 +
.../paimon/table/DataEvolutionTableTest.java | 84 +++++++++
.../paimon/table/format/FormatTableWriteTest.java | 189 +++++++++++++++++++++
.../DataEvolutionPartialWriteOperator.java | 11 +-
.../pypaimon/common/options/core_options.py | 14 ++
paimon-python/pypaimon/schema/schema_manager.py | 16 ++
.../pypaimon/table/format/format_table_write.py | 44 ++++-
.../pypaimon/tests/reader_append_only_test.py | 12 ++
.../pypaimon/tests/rest/rest_format_table_test.py | 72 ++++++++
.../pypaimon/tests/schema_manager_test.py | 53 ++++++
paimon-python/pypaimon/tests/table_update_test.py | 8 +-
paimon-python/pypaimon/write/file_store_write.py | 20 ++-
.../spark/commands/DataEvolutionPaimonWriter.scala | 12 +-
40 files changed, 974 insertions(+), 52 deletions(-)
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 0776eaf9e3..2841618e7c 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1681,6 +1681,12 @@ If the data size allocated for the sorting task is
uneven,which may lead to perf
<td>Boolean</td>
<td>Whether to enable tag expiration by retained time.</td>
</tr>
+ <tr>
+ <td><h5>target-file-row-num</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 supp [...]
+ </tr>
<tr>
<td><h5>target-file-size</h5></td>
<td style="word-wrap: break-word;">(none)</td>
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 ec6a2296bf..e0c7fd4004 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,21 @@ public class CoreOptions implements Serializable {
text("append table: the default
value is 256 MB."))
.build());
+ public static final ConfigOption<Long> TARGET_FILE_ROW_NUM =
+ key("target-file-row-num")
+ .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 class CoreOptions implements Serializable {
.getBytes();
}
+ public long targetFileRowNum() {
+ return options.get(TARGET_FILE_ROW_NUM);
+ }
+
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 ea81dea50f..5f47789a26 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 targetFileRowNum;
private final RowType writeSchema;
@Nullable private final List<String> writeCols;
private final DataFilePathFactory pathFactory;
@@ -112,6 +113,7 @@ public class AppendOnlyWriter implements BatchRecordWriter,
MemoryOwner {
long targetFileSize,
long blobTargetFileSize,
long vectorTargetFileSize,
+ long targetFileRowNum,
RowType writeSchema,
@Nullable List<String> writeCols,
long maxSequenceNumber,
@@ -139,6 +141,7 @@ public class AppendOnlyWriter implements BatchRecordWriter,
MemoryOwner {
this.targetFileSize = targetFileSize;
this.blobTargetFileSize = blobTargetFileSize;
this.vectorTargetFileSize = vectorTargetFileSize;
+ this.targetFileRowNum = targetFileRowNum;
this.writeSchema = writeSchema;
this.writeCols = writeCols;
this.pathFactory = pathFactory;
@@ -325,6 +328,7 @@ public class AppendOnlyWriter implements BatchRecordWriter,
MemoryOwner {
targetFileSize,
blobTargetFileSize,
vectorTargetFileSize,
+ targetFileRowNum,
writeSchema,
pathFactory,
seqNumCounterProvider,
@@ -350,7 +354,8 @@ public class AppendOnlyWriter implements BatchRecordWriter,
MemoryOwner {
asyncFileWrite,
statsDenseStore,
writeCols,
- rowSidecarFileFormat);
+ rowSidecarFileFormat,
+ targetFileRowNum);
}
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 9b8897862a..3cf08fc263 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<InternalRow, DataFileMeta>,
List<DataFileMeta>>>
vectorStoreWriterFactory;
private final long targetFileSize;
+ private final long targetFileRowNum;
// 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 class DedicatedFormatRollingFileWriter
long targetFileSize,
long blobTargetFileSize,
long vectorTargetFileSize,
+ long targetFileRowNum,
RowType writeSchema,
DataFilePathFactory pathFactory,
Supplier<LongCounter> seqNumCounterSupplier,
@@ -141,7 +144,12 @@ public class DedicatedFormatRollingFileWriter
boolean statsDenseStore,
@Nullable BlobFileContext context) {
// Initialize basic fields
+ Preconditions.checkArgument(
+ targetFileRowNum > 0,
+ "targetFileRowNum must be positive, but is %s",
+ targetFileRowNum);
this.targetFileSize = targetFileSize;
+ this.targetFileRowNum = targetFileRowNum;
this.results = new ArrayList<>();
this.closedWriters = new ArrayList<>();
@@ -319,7 +327,8 @@ public class DedicatedFormatRollingFileWriter
statsDenseStore,
pathFactory.isExternalPath(),
vectorStoreColumnNames),
- targetFileSize),
+ targetFileSize,
+ Long.MAX_VALUE),
vectorStoreProjection);
}
@@ -352,8 +361,9 @@ public class DedicatedFormatRollingFileWriter
currentWriter.write(row);
}
recordCount++;
+ currentFileRecordCount++;
- if (currentWriter != null && rollingFile()) {
+ if (rollingFile()) {
closeCurrentWriter();
}
} catch (Throwable e) {
@@ -416,11 +426,19 @@ public class DedicatedFormatRollingFileWriter
}
}
- /** 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 >= targetFileRowNum) {
+ return true;
+ }
+ return currentWriter != null
+ && currentWriter
+ .writer()
+ .reachTargetSize(
+ recordCount % CHECK_ROLLING_RECORD_CNT == 0,
targetFileSize);
}
/**
@@ -455,6 +473,7 @@ public class DedicatedFormatRollingFileWriter
// Reset current writer
currentWriter = null;
+ currentFileRecordCount = 0;
}
/** Closes the main writer and returns its metadata. */
@@ -471,6 +490,7 @@ public class DedicatedFormatRollingFileWriter
}
blobWriter.close();
List<DataFileMeta> results = blobWriter.result();
+ closedWriters.addAll(blobWriter.drainAbortExecutors());
blobWriter = null;
return results;
}
@@ -482,6 +502,7 @@ public class DedicatedFormatRollingFileWriter
}
vectorStoreWriter.close();
List<DataFileMeta> results = vectorStoreWriter.result();
+ closedWriters.addAll(vectorStoreWriter.writer().drainAbortExecutors());
vectorStoreWriter = null;
return results;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java
b/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java
index 39d6c6eac8..7a0901a75f 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java
@@ -26,6 +26,7 @@ import org.apache.paimon.format.blob.BlobFileFormat;
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,6 +132,14 @@ public class MultipleBlobFileWriter implements Closeable {
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>> {
@@ -138,7 +147,9 @@ public class MultipleBlobFileWriter implements Closeable {
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);
}
}
}
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 77c3eaa915..0019c4e17f 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 @@ public abstract class DataEvolutionCompactTask {
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_ROW_NUM.key(),
String.valueOf(Long.MAX_VALUE));
return Collections.unmodifiableMap(options);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
index 6c97bb9808..67992a755a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
@@ -442,6 +442,7 @@ public abstract class AbstractCatalog implements Catalog {
}
copyTableDefaultOptions(schema.options());
+ validateCreateTable(schema, false);
switch (Options.fromMap(schema.options()).get(TYPE)) {
case TABLE:
@@ -516,6 +517,7 @@ public abstract class AbstractCatalog implements Catalog {
validateCreateTable(newSchema, false);
validateCustomTablePath(newSchema.options());
copyTableDefaultOptions(newSchema.options());
+ validateCreateTable(newSchema, false);
Table existing;
try {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java
b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java
index ee793980d3..0ee552d887 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java
@@ -159,6 +159,10 @@ public class CatalogUtils {
"The value of %s property should be %s.",
AUTO_CREATE.key(),
Boolean.FALSE);
+ checkArgument(
+ options.get(CoreOptions.TARGET_FILE_ROW_NUM) > 0,
+ "%s should be at least 1.",
+ CoreOptions.TARGET_FILE_ROW_NUM.key());
TableType tableType = options.get(CoreOptions.TYPE);
if (tableType.equals(TableType.FORMAT_TABLE)) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java
index 46fd143906..d5abb480a5 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java
@@ -172,7 +172,9 @@ public class IcebergManifestFile extends
ObjectsFile<IcebergManifestEntry> {
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();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/FormatTableRollingFileWriter.java
b/paimon-core/src/main/java/org/apache/paimon/io/FormatTableRollingFileWriter.java
index f081abf310..96e297c327 100644
---
a/paimon-core/src/main/java/org/apache/paimon/io/FormatTableRollingFileWriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/io/FormatTableRollingFileWriter.java
@@ -23,6 +23,7 @@ import org.apache.paimon.format.FileFormat;
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.ArrayList;
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 targetFileRowNum;
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 targetFileRowNum,
RowType writeSchema,
DataFilePathFactory pathFactory,
String fileCompression) {
+ Preconditions.checkArgument(
+ targetFileRowNum > 0,
+ "targetFileRowNum must be positive, but is %s",
+ targetFileRowNum);
+ this.fileIO = fileIO;
this.writerFactory =
() ->
new FormatTableSingleFileWriter(
@@ -65,6 +73,7 @@ public class FormatTableRollingFileWriter implements
AutoCloseable {
pathFactory.newPath(),
fileCompression);
this.targetFileSize = targetFileSize;
+ this.targetFileRowNum = targetFileRowNum;
this.closedWriters = new ArrayList<>();
this.committers = new ArrayList<>();
}
@@ -81,9 +90,11 @@ public class FormatTableRollingFileWriter implements
AutoCloseable {
currentWriter.write(row);
recordCount += 1;
+ currentFileRecordCount += 1;
boolean needRolling =
- currentWriter.reachTargetSize(
- recordCount % CHECK_ROLLING_RECORD_CNT == 0,
targetFileSize);
+ currentFileRecordCount >= targetFileRowNum
+ || currentWriter.reachTargetSize(
+ recordCount % CHECK_ROLLING_RECORD_CNT ==
0, targetFileSize);
if (needRolling) {
closeCurrentWriter();
}
@@ -109,15 +120,26 @@ public class FormatTableRollingFileWriter implements
AutoCloseable {
}
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() {
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 dfe875120d..5644aab375 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 class KeyValueFileWriterFactory {
public RollingFileWriter<KeyValue, DataFileMeta>
createRollingMergeTreeFileWriter(
int level, FileSource fileSource) {
WriteFormatKey key = new WriteFormatKey(level, false);
+ // Row limit applies to writes only; compaction output stays size-only.
+ long targetFileRowNum =
+ fileSource == FileSource.COMPACT ? Long.MAX_VALUE :
options.targetFileRowNum();
return new RollingFileWriterImpl<>(
() -> {
DataFilePathFactory pathFactory =
formatContext.pathFactory(key);
return createDataFileWriter(
pathFactory.newPath(), key, fileSource,
pathFactory.isExternalPath());
},
- suggestedFileSize);
+ suggestedFileSize,
+ targetFileRowNum);
}
public RollingFileWriter<KeyValue, DataFileMeta>
createRollingChangelogFileWriter(int level) {
@@ -156,7 +160,8 @@ public class KeyValueFileWriterFactory {
FileSource.APPEND,
pathFactory.isExternalPath());
},
- suggestedFileSize);
+ suggestedFileSize,
+ Long.MAX_VALUE);
}
public RollingFileWriter<KeyValue, DataFileMeta>
createRollingClusteringFileWriter() {
@@ -167,7 +172,8 @@ public class KeyValueFileWriterFactory {
return createKvSeparatedFileWriter(
pathFactory.newPath(), key,
pathFactory.isExternalPath());
},
- suggestedFileSize);
+ suggestedFileSize,
+ Long.MAX_VALUE);
}
private KeyValueClusteringFileWriter createKvSeparatedFileWriter(
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 299a4f9742..0ebdc5feb7 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,26 @@ public class RollingFileWriterImpl<T, R> implements
RollingFileWriter<T, R> {
private final Supplier<? extends SingleFileWriter<T, R>> writerFactory;
private final long targetFileSize;
+ private final long targetFileRowNum;
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) {
+ Supplier<? extends SingleFileWriter<T, R>> writerFactory,
+ long targetFileSize,
+ long targetFileRowNum) {
+ Preconditions.checkArgument(
+ targetFileRowNum > 0,
+ "targetFileRowNum must be positive, but is %s",
+ targetFileRowNum);
this.writerFactory = writerFactory;
this.targetFileSize = targetFileSize;
+ this.targetFileRowNum = targetFileRowNum;
this.results = new ArrayList<>();
this.closedWriters = new ArrayList<>();
}
@@ -62,8 +71,9 @@ public class RollingFileWriterImpl<T, R> implements
RollingFileWriter<T, R> {
}
private boolean rollingFile(boolean forceCheck) throws IOException {
- return currentWriter.reachTargetSize(
- forceCheck || recordCount % CHECK_ROLLING_RECORD_CNT == 0,
targetFileSize);
+ return currentFileRecordCount >= targetFileRowNum
+ || currentWriter.reachTargetSize(
+ forceCheck || recordCount % CHECK_ROLLING_RECORD_CNT
== 0, targetFileSize);
}
@Override
@@ -76,6 +86,7 @@ public class RollingFileWriterImpl<T, R> implements
RollingFileWriter<T, R> {
currentWriter.write(row);
recordCount += 1;
+ currentFileRecordCount += 1;
if (rollingFile(false)) {
closeCurrentWriter();
@@ -101,6 +112,7 @@ public class RollingFileWriterImpl<T, R> implements
RollingFileWriter<T, R> {
currentWriter.writeBundle(bundle);
recordCount += bundle.rowCount();
+ currentFileRecordCount += bundle.rowCount();
if (rollingFile(true)) {
closeCurrentWriter();
@@ -132,6 +144,7 @@ public class RollingFileWriterImpl<T, R> implements
RollingFileWriter<T, R> {
currentWriter.abortExecutor().ifPresent(closedWriters::add);
results.add(currentWriter.result());
currentWriter = null;
+ currentFileRecordCount = 0;
}
@Override
@@ -155,6 +168,14 @@ public class RollingFileWriterImpl<T, R> implements
RollingFileWriter<T, R> {
return results;
}
+ /** Transfers ownership of abort executors for closed files to the caller.
*/
+ public List<FileWriterAbortExecutor> drainAbortExecutors() {
+ Preconditions.checkState(closed, "Cannot drain abort executors unless
close all writers.");
+ List<FileWriterAbortExecutor> abortExecutors = new
ArrayList<>(closedWriters);
+ closedWriters.clear();
+ return abortExecutors;
+ }
+
@Override
public void close() throws IOException {
if (closed) {
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 0da1badf62..e9874ef9b9 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
@@ -51,7 +51,8 @@ public class RowDataRollingFileWriter extends
RollingFileWriterImpl<InternalRow,
boolean asyncFileWrite,
boolean statsDenseStore,
@Nullable List<String> writeCols,
- @Nullable FileFormat rowSidecarFormat) {
+ @Nullable FileFormat rowSidecarFormat,
+ long targetFileRowNum) {
super(
() -> {
Path dataPath = pathFactory.newPath();
@@ -76,6 +77,7 @@ public class RowDataRollingFileWriter extends
RollingFileWriterImpl<InternalRow,
rowSidecarFormat,
rowSidecarPath);
},
- targetFileSize);
+ targetFileSize,
+ targetFileRowNum);
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
index c2fc853866..a1fbaa6f39 100644
--- a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
@@ -380,6 +380,7 @@ public class JdbcCatalog extends AbstractCatalog {
getDatabase(identifier.getDatabaseName());
copyTableDefaultOptions(schema.options());
+ validateCreateTable(schema, false);
TableType tableType = Options.fromMap(schema.options()).get(TYPE);
switch (tableType) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
index 527568f7f0..71b8020e3b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
@@ -174,7 +174,8 @@ public class ManifestFile extends
ObjectsFile<ManifestEntry> {
public RollingFileWriter<ManifestEntry, ManifestFileMeta>
createRollingWriter() {
return new RollingFileWriterImpl<>(
() -> new ManifestEntryWriter(writerFactory,
pathFactory.newPath(), compression),
- suggestedFileSize);
+ suggestedFileSize,
+ Long.MAX_VALUE);
}
public ManifestEntryWriter createManifestEntryWriter(Path manifestPath) {
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 7dceb12e5c..c3ad22b65c 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 @@ public abstract class BaseAppendFileStoreWrite extends
MemoryFileStoreWrite<Inte
options.targetFileSize(false),
options.blobTargetFileSize(),
options.vectorTargetFileSize(),
+ options.targetFileRowNum(),
writeType,
writeCols,
restoredMaxSeqNumber,
@@ -312,7 +313,8 @@ public abstract class BaseAppendFileStoreWrite extends
MemoryFileStoreWrite<Inte
options.asyncFileWrite(),
options.statsDenseStore(),
rowType.equals(writeType) ? null : writeType.getFieldNames(),
- rowSidecarFileFormat());
+ rowSidecarFileFormat(),
+ Long.MAX_VALUE);
}
@Nullable
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
index 86256a9342..3db1052c4c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
@@ -572,6 +572,7 @@ public class RESTCatalog implements Catalog {
checkNotSystemTable(identifier, "createTable");
validateCreateTable(schema, dataTokenEnabled);
tableDefaultOptions.forEach(schema.options()::putIfAbsent);
+ validateCreateTable(schema, dataTokenEnabled);
// Defaults participate in the catalog-managed partition
combination, so validate
// the effective options rather than only the explicit ones.
validateCatalogManagedFormatTablePartitions(
@@ -654,6 +655,7 @@ public class RESTCatalog implements Catalog {
checkNotSystemTable(identifier, "replaceTable");
validateCreateTable(newSchema, dataTokenEnabled);
tableDefaultOptions.forEach(newSchema.options()::putIfAbsent);
+ validateCreateTable(newSchema, dataTokenEnabled);
// Defaults participate in the catalog-managed partition combination,
so validate the
// effective options rather than only the explicit ones. Externality
is not validated
// client-side here: a round-tripped schema of an internal table may
carry the synthetic
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 4914ef1b40..1c6ba3a0fb 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 class SchemaValidation {
FULL_COMPACTION_DELTA_COMMITS.key()));
}
+ checkArgument(
+ options.targetFileRowNum() > 0,
+ CoreOptions.TARGET_FILE_ROW_NUM.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/main/java/org/apache/paimon/table/format/FormatTableFileWriter.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableFileWriter.java
index 23061e0d14..9f18ee5758 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableFileWriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableFileWriter.java
@@ -28,6 +28,7 @@ import org.apache.paimon.fs.TwoPhaseOutputStream;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.IOUtils;
import java.util.ArrayList;
import java.util.HashMap;
@@ -88,17 +89,38 @@ public class FormatTableFileWriter {
}
public void close() throws Exception {
- writers.clear();
+ try {
+ IOUtils.closeAll(writers.values());
+ } finally {
+ writers.clear();
+ }
}
public List<CommitMessage> prepareCommit() throws Exception {
- List<CommitMessage> commitMessages = new ArrayList<>();
- for (FormatTableRecordWriter writer : writers.values()) {
- List<TwoPhaseOutputStream.Committer> commiters =
writer.closeAndGetCommitters();
- for (TwoPhaseOutputStream.Committer committer : commiters) {
- TwoPhaseCommitMessage twoPhaseCommitMessage = new
TwoPhaseCommitMessage(committer);
- commitMessages.add(twoPhaseCommitMessage);
+ List<TwoPhaseOutputStream.Committer> committers = new ArrayList<>();
+ try {
+ for (FormatTableRecordWriter writer : writers.values()) {
+ committers.addAll(writer.closeAndGetCommitters());
+ }
+ } catch (Exception e) {
+ for (TwoPhaseOutputStream.Committer committer : committers) {
+ try {
+ committer.discard(fileIO);
+ } catch (Exception cleanupException) {
+ e.addSuppressed(cleanupException);
+ }
+ }
+ try {
+ close();
+ } catch (Exception cleanupException) {
+ e.addSuppressed(cleanupException);
}
+ throw e;
+ }
+
+ List<CommitMessage> commitMessages = new ArrayList<>();
+ for (TwoPhaseOutputStream.Committer committer : committers) {
+ commitMessages.add(new TwoPhaseCommitMessage(committer));
}
return commitMessages;
}
@@ -118,6 +140,7 @@ public class FormatTableFileWriter {
fileIO,
fileFormat,
options.targetFileSize(false),
+ options.targetFileRowNum(),
pathFactory.createDataFilePathFactory(parent, null),
writeRowType,
options.formatTableFileCompression());
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRecordWriter.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRecordWriter.java
index b611f2805d..83677d9448 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRecordWriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRecordWriter.java
@@ -38,12 +38,14 @@ public class FormatTableRecordWriter implements
AutoCloseable {
private final String fileCompression;
private final FileFormat fileFormat;
private final long targetFileSize;
+ private final long targetFileRowNum;
private FormatTableRollingFileWriter writer;
public FormatTableRecordWriter(
FileIO fileIO,
FileFormat fileFormat,
long targetFileSize,
+ long targetFileRowNum,
DataFilePathFactory pathFactory,
RowType writeSchema,
String fileCompression) {
@@ -53,6 +55,7 @@ public class FormatTableRecordWriter implements AutoCloseable
{
this.writeSchema = writeSchema;
this.fileFormat = fileFormat;
this.targetFileSize = targetFileSize;
+ this.targetFileRowNum = targetFileRowNum;
}
public void write(InternalRow data) throws Exception {
@@ -82,6 +85,12 @@ public class FormatTableRecordWriter implements
AutoCloseable {
private FormatTableRollingFileWriter createRollingRowWriter() {
return new FormatTableRollingFileWriter(
- fileIO, fileFormat, targetFileSize, writeSchema, pathFactory,
fileCompression);
+ fileIO,
+ fileFormat,
+ targetFileSize,
+ targetFileRowNum,
+ writeSchema,
+ pathFactory,
+ fileCompression);
}
}
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 bd5d1d2cf5..45f2a8a4b1 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 @@ public class AppendOnlyWriterTest {
1024 * 1024L,
1024 * 1024L,
1024 * 1024L,
+ Long.MAX_VALUE,
writeType,
null,
maxSequenceNumber,
@@ -1238,6 +1239,7 @@ public class AppendOnlyWriterTest {
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 dbe9413804..f03e562f12 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
@@ -103,6 +103,7 @@ public class DedicatedFormatRollingFileWriterTest {
TARGET_FILE_SIZE,
TARGET_FILE_SIZE,
TARGET_FILE_SIZE,
+ Long.MAX_VALUE,
SCHEMA,
pathFactory,
() -> seqNumCounter,
@@ -147,6 +148,73 @@ public class DedicatedFormatRollingFileWriterTest {
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 testAbortAfterRollingByRowsDeletesBlobFiles() 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,
+ 1L,
+ SCHEMA,
+ pathFactory,
+ LongCounter::new,
+ COMPRESSION,
+ new StatsCollectorFactories(options),
+ new FileIndexOptions(),
+ FileSource.APPEND,
+ false,
+ BlobFileContext.create(SCHEMA, options));
+
+ cappedWriter.write(
+ GenericRow.of(1, BinaryString.fromString("test"), new
BlobData(testBlobData)));
+ cappedWriter.abort();
+
+ try (Stream<java.nio.file.Path> files = Files.walk(tempDir)) {
+ assertThat(files.filter(Files::isRegularFile).count()).isZero();
+ }
+ }
+
@Test
public void testDoesNotWriteRowSidecar() throws IOException {
// Tests that: dedicated blob files do not create row-store sidecars.
@@ -187,6 +255,7 @@ public class DedicatedFormatRollingFileWriterTest {
TARGET_FILE_SIZE,
TARGET_FILE_SIZE,
TARGET_FILE_SIZE,
+ Long.MAX_VALUE,
SCHEMA,
pathFactory,
() -> seqNumCounter,
@@ -252,6 +321,7 @@ public class DedicatedFormatRollingFileWriterTest {
128 * 1024 * 1024,
blobTargetFileSize, // Different blob target size
128 * 1024 * 1024,
+ Long.MAX_VALUE,
SCHEMA,
new DataFilePathFactory(
new Path(tempDir + "/blob-size-test"),
@@ -323,6 +393,7 @@ public class DedicatedFormatRollingFileWriterTest {
128 * 1024 * 1024,
blobTargetFileSize,
128 * 1024 * 1024,
+ Long.MAX_VALUE,
SCHEMA,
new DataFilePathFactory(
new Path(tempDir + "/bundle-blob-size-test"),
@@ -396,6 +467,7 @@ public class DedicatedFormatRollingFileWriterTest {
128 * 1024 * 1024,
blobTargetFileSize,
128 * 1024 * 1024,
+ Long.MAX_VALUE,
SCHEMA,
pathFactory, // Use the same pathFactory to ensure
shared UUID
() -> new LongCounter(),
@@ -476,6 +548,7 @@ public class DedicatedFormatRollingFileWriterTest {
128 * 1024 * 1024,
blobTargetFileSize,
128 * 1024 * 1024,
+ Long.MAX_VALUE,
SCHEMA,
pathFactory, // Use the same pathFactory to ensure
shared UUID
() -> new LongCounter(),
@@ -697,6 +770,7 @@ public class DedicatedFormatRollingFileWriterTest {
TARGET_FILE_SIZE,
TARGET_FILE_SIZE,
TARGET_FILE_SIZE,
+ Long.MAX_VALUE,
customSchema, // Use custom schema
pathFactory,
() -> seqNumCounter,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterVectorTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterVectorTest.java
index 156c79672e..bed8ef7b93 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterVectorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterVectorTest.java
@@ -50,12 +50,14 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
+import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Random;
+import java.util.stream.Stream;
import static org.apache.paimon.types.VectorType.isVectorStoreFile;
import static org.assertj.core.api.Assertions.assertThat;
@@ -111,6 +113,7 @@ public class DedicatedFormatRollingFileWriterVectorTest {
TARGET_FILE_SIZE,
TARGET_FILE_SIZE,
VECTOR_TARGET_FILE_SIZE,
+ Long.MAX_VALUE,
SCHEMA,
pathFactory,
() -> seqNumCounter,
@@ -147,6 +150,42 @@ public class DedicatedFormatRollingFileWriterVectorTest {
assertThat(metasResult.get(0).rowCount()).isEqualTo(metasResult.get(2).rowCount());
}
+ @Test
+ public void testAbortAfterRollingByRowsDeletesVectorFiles() throws
IOException {
+ RowType schema =
+ RowType.builder()
+ .field("f0", DataTypes.INT())
+ .field("f1", DataTypes.VECTOR(VECTOR_DIM,
DataTypes.FLOAT()))
+ .build();
+ long hugeSize = 1024L * 1024 * 1024;
+ writer =
+ new DedicatedFormatRollingFileWriter(
+ LocalFileIO.create(),
+ SCHEMA_ID,
+ FileFormat.fromIdentifier("parquet", new Options()),
+ FileFormat.fromIdentifier("json", new Options()),
+ hugeSize,
+ hugeSize,
+ hugeSize,
+ 1L,
+ schema,
+ pathFactory,
+ LongCounter::new,
+ COMPRESSION,
+ new StatsCollectorFactories(new CoreOptions(new
Options())),
+ new FileIndexOptions(),
+ FileSource.APPEND,
+ false,
+ null);
+
+ writer.write(GenericRow.of(1, BinaryVector.fromPrimitiveArray(new
float[VECTOR_DIM])));
+ writer.abort();
+
+ try (Stream<java.nio.file.Path> files = Files.walk(tempDir)) {
+ assertThat(files.filter(Files::isRegularFile).count()).isZero();
+ }
+ }
+
@Test
public void testVectorTargetFileSize() throws Exception {
// 100k vector-store data would create 1 normal, 1 blob, and 3
vector-store files
@@ -206,6 +245,7 @@ public class DedicatedFormatRollingFileWriterVectorTest {
128 * 1024 * 1024,
128 * 1024 * 1024,
vectorTargetFileSize,
+ Long.MAX_VALUE,
SCHEMA,
new DataFilePathFactory(
new Path(tempDir + "/bundle-vector-size-test"),
@@ -255,6 +295,7 @@ public class DedicatedFormatRollingFileWriterVectorTest {
TARGET_FILE_SIZE,
TARGET_FILE_SIZE,
VECTOR_TARGET_FILE_SIZE,
+ Long.MAX_VALUE,
SCHEMA,
pathFactory,
() -> seqNumCounter,
@@ -367,6 +408,7 @@ public class DedicatedFormatRollingFileWriterVectorTest {
TARGET_FILE_SIZE,
TARGET_FILE_SIZE,
VECTOR_TARGET_FILE_SIZE,
+ Long.MAX_VALUE,
schema,
pathFactory,
() -> seqNumCounter,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogUtilsTest.java
b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogUtilsTest.java
index 3ffb6a1ecb..417ccacddd 100644
--- a/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogUtilsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogUtilsTest.java
@@ -31,6 +31,7 @@ import static org.apache.paimon.CoreOptions.FILE_FORMAT;
import static org.apache.paimon.CoreOptions.FORMAT_TABLE_IMPLEMENTATION;
import static org.apache.paimon.CoreOptions.METASTORE_PARTITIONED_TABLE;
import static org.apache.paimon.CoreOptions.PATH;
+import static org.apache.paimon.CoreOptions.TARGET_FILE_ROW_NUM;
import static org.apache.paimon.CoreOptions.TYPE;
import static org.apache.paimon.TableType.FORMAT_TABLE;
import static org.apache.paimon.catalog.CatalogUtils.loadTable;
@@ -43,6 +44,19 @@ import static org.mockito.Mockito.when;
/** Tests for {@link CatalogUtils}. */
class CatalogUtilsTest {
+ @Test
+ void testRejectNonPositiveTargetRows() {
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .option(TARGET_FILE_ROW_NUM.key(), "0")
+ .build();
+
+ assertThatThrownBy(() -> validateCreateTable(schema, false))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("target-file-row-num should be at least 1.");
+ }
+
@Test
void testRejectEngineImplementationForCatalogManagedPartitionsOnCreate() {
Schema schema =
diff --git
a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
index 094ede3847..5e186cdf10 100644
---
a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
@@ -31,6 +31,7 @@ import
org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link FileSystemCatalog}. */
@@ -68,6 +69,47 @@ public class FileSystemCatalogTest extends CatalogTestBase {
catalog.createTable(identifier, schema, false);
}
+ @Test
+ public void testValidateFormatTableDefaultOptions() throws Exception {
+ String database = "format_table_default_validation_db";
+ catalog.createDatabase(database, false);
+ Identifier existing = Identifier.create(database, "existing_table");
+ catalog.createTable(
+ existing, Schema.newBuilder().column("id",
DataTypes.INT()).build(), false);
+ ((AbstractCatalog) DelegateCatalog.rootCatalog(catalog))
+
.tableDefaultOptions.put(CoreOptions.TARGET_FILE_ROW_NUM.key(), "0");
+ Schema createSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .option(CoreOptions.TYPE.key(),
TableType.FORMAT_TABLE.toString())
+ .build();
+
+ assertThatThrownBy(
+ () ->
+ catalog.createTable(
+ Identifier.create(database,
"format_table"),
+ createSchema,
+ false))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("target-file-row-num should be at least 1.");
+
+ Schema tableReplaceSchema = Schema.newBuilder().column("id",
DataTypes.INT()).build();
+ assertThatThrownBy(() -> catalog.replaceTable(existing,
tableReplaceSchema, false))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("target-file-row-num should be at least 1.");
+ assertThat(catalog.getTable(existing)).isNotNull();
+
+ Schema replaceSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .option(CoreOptions.TYPE.key(),
TableType.FORMAT_TABLE.toString())
+ .build();
+ assertThatThrownBy(() -> catalog.replaceTable(existing, replaceSchema,
false))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("target-file-row-num should be at least 1.");
+ assertThat(catalog.getTable(existing)).isNotNull();
+ }
+
@Test
public void testAlterDatabase() throws Exception {
String databaseName = "test_alter_db";
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 4b5c730386..ad8bfb512a 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 class KeyValueFileReadWriteTest {
testWriteAndReadDataFileImpl("avro-extract");
}
+ @Test
+ public void testMergeTreeCompactionIgnoresRowLimit() throws Exception {
+ Options options = new Options();
+ options.set(CoreOptions.TARGET_FILE_ROW_NUM, 2L);
+ KeyValueFileWriterFactory writerFactory =
+ createWriterFactory(tempDir.toString(), "avro", options);
+ List<KeyValue> content = gen.next().content;
+
+ RollingFileWriter<KeyValue, DataFileMeta> appendWriter =
+ writerFactory.createRollingMergeTreeFileWriter(0,
FileSource.APPEND);
+ appendWriter.write(CloseableIterator.fromList(content, kv -> {}));
+ appendWriter.close();
+
+ RollingFileWriter<KeyValue, DataFileMeta> 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 class KeyValueFileReadWriteTest {
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 ec334007e5..bbfb221fbe 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 class RollingFileWriterTest {
}
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 targetFileRowNum) {
FileFormat fileFormat = FileFormat.fromIdentifier(identifier, new
Options());
rollingFileWriter =
new RollingFileWriterImpl<>(
@@ -106,7 +114,8 @@ public class RollingFileWriterTest {
statsDenseStore,
false,
null),
- TARGET_FILE_SIZE);
+ targetFileSize,
+ targetFileRowNum);
}
@ParameterizedTest
@@ -129,6 +138,45 @@ public class RollingFileWriterTest {
}
}
+ @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<DataFileMeta> 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<DataFileMeta> 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<InternalRow> 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());
@@ -158,7 +206,8 @@ public class RollingFileWriterTest {
true,
false,
null,
- rowFormat);
+ rowFormat,
+ Long.MAX_VALUE);
writer.write(GenericRow.of(1));
writer.close();
@@ -208,7 +257,8 @@ public class RollingFileWriterTest {
true,
false,
null,
- rowFormat);
+ rowFormat,
+ Long.MAX_VALUE);
writer.writeBundle(
new SingleUseBundleRecords(Arrays.asList(GenericRow.of(1),
GenericRow.of(2))));
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 36b8668bcc..ab94f33b3a 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 @@ class SchemaValidationTest {
"must set only one key in
[scan.timestamp-millis,scan.timestamp] when you use from-timestamp for
scan.mode");
}
+ @Test
+ public void testTargetFileRowNumMustBePositive() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.TARGET_FILE_ROW_NUM.key(), "0");
+ assertThatThrownBy(() -> validateTableSchemaExec(options))
+ .hasMessageContaining("target-file-row-num 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 7766da98de..4f080f9667 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 class DataEvolutionTableTest extends
DataEvolutionTestBase {
assertThat(task.compactBefore().size()).isEqualTo(20);
}
+ @Test
+ public void testDataEvolutionReadWithRolledColumns() throws Exception {
+ createTableDefault();
+ FileStoreTable table =
+ getTableDefault()
+ .copy(
+ Collections.singletonMap(
+ CoreOptions.TARGET_FILE_ROW_NUM.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<CommitMessage> msgs = w1.prepareCommit();
+ assignCumulativeFirstRowId(msgs, firstRowId);
+ b1.newCommit().commit(msgs);
+ }
+
+ ReadBuilder rb = table.newReadBuilder();
+ RecordReader<InternalRow> 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<CommitMessage> msgs, long
firstRowId) {
+ long cur = firstRowId;
+ for (CommitMessage c : msgs) {
+ CommitMessageImpl m = (CommitMessageImpl) c;
+ List<DataFileMeta> files = new
ArrayList<>(m.newFilesIncrement().newFiles());
+ m.newFilesIncrement().newFiles().clear();
+ List<DataFileMeta> 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_ROW_NUM.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<Long> rowCounts = new ArrayList<>();
+ Iterator<ManifestEntry> 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-core/src/test/java/org/apache/paimon/table/format/FormatTableWriteTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableWriteTest.java
new file mode 100644
index 0000000000..3864f39ad6
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableWriteTest.java
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.format;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.TwoPhaseOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.paimon.CoreOptions.FILE_FORMAT;
+import static org.apache.paimon.CoreOptions.PATH;
+import static org.apache.paimon.CoreOptions.TARGET_FILE_ROW_NUM;
+import static org.apache.paimon.CoreOptions.TARGET_FILE_SIZE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests for writing a {@link FormatTable}. */
+class FormatTableWriteTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testRollsByTargetRowNumber() throws Exception {
+ Path tablePath = new Path(tempDir.toUri());
+ LocalFileIO fileIO = LocalFileIO.create();
+ FormatTable table = formatTable(tablePath, fileIO, 2);
+
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ List<CommitMessage> messages;
+ try (BatchTableWrite write = writeBuilder.newWrite()) {
+ for (int i = 0; i < 5; i++) {
+ write.write(GenericRow.of(i));
+ }
+ messages = write.prepareCommit();
+ }
+
+ assertThat(messages).hasSize(3);
+ List<Path> dataFiles =
+ messages.stream()
+ .map(
+ message ->
+ ((TwoPhaseCommitMessage) message)
+ .getCommitter()
+ .targetPath())
+ .collect(Collectors.toList());
+ try (BatchTableCommit commit = writeBuilder.newCommit()) {
+ commit.commit(messages);
+ }
+
+ List<Long> rowCounts =
+ dataFiles.stream()
+ .map(
+ path -> {
+ try (BufferedReader reader =
+ new BufferedReader(
+ new InputStreamReader(
+
fileIO.newInputStream(path),
+
StandardCharsets.UTF_8))) {
+ return reader.lines().count();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ })
+ .collect(Collectors.toList());
+ assertThat(rowCounts).containsExactlyInAnyOrder(2L, 2L, 1L);
+ }
+
+ @Test
+ void testRejectsNonPositiveTargetRowNumber() throws Exception {
+ Path tablePath = new Path(tempDir.toUri());
+ FormatTable table = formatTable(tablePath, LocalFileIO.create(), 0);
+
+ try (BatchTableWrite write = table.newBatchWriteBuilder().newWrite()) {
+ assertThatThrownBy(() -> write.write(GenericRow.of(1)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("targetFileRowNum must be positive, but is 0");
+ }
+ }
+
+ @Test
+ void testAbortAfterRollingDeletesTemporaryFiles() throws Exception {
+ FormatTable table = formatTable(new Path(tempDir.toUri()),
LocalFileIO.create(), 1);
+
+ try (BatchTableWrite write = table.newBatchWriteBuilder().newWrite()) {
+ write.write(GenericRow.of(1));
+ }
+
+ try (Stream<java.nio.file.Path> files = Files.walk(tempDir)) {
+ assertThat(files.filter(Files::isRegularFile).count()).isZero();
+ }
+ }
+
+ @Test
+ void testPrepareCommitFailureDiscardsPreparedFiles() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Options options = new Options();
+ options.set(PATH, new Path(tempDir.toUri()).toString());
+ options.set(FILE_FORMAT, "csv");
+ FormatTableFileWriter fileWriter =
+ new FormatTableFileWriter(
+ fileIO,
+ RowType.of(DataTypes.INT()),
+ new org.apache.paimon.CoreOptions(options),
+ RowType.of(DataTypes.INT()));
+ FormatTableRecordWriter recordWriter =
mock(FormatTableRecordWriter.class);
+ TwoPhaseOutputStream.Committer committer =
mock(TwoPhaseOutputStream.Committer.class);
+ java.util.concurrent.atomic.AtomicInteger closeCount =
+ new java.util.concurrent.atomic.AtomicInteger();
+ when(recordWriter.closeAndGetCommitters())
+ .thenAnswer(
+ ignored -> {
+ if (closeCount.getAndIncrement() == 0) {
+ return Collections.singletonList(committer);
+ }
+ throw new IOException("expected close failure");
+ });
+ fileWriter.writers.put(BinaryRow.singleColumn(1), recordWriter);
+ fileWriter.writers.put(BinaryRow.singleColumn(2), recordWriter);
+
+ assertThatThrownBy(fileWriter::prepareCommit)
+ .isInstanceOf(IOException.class)
+ .hasMessage("expected close failure");
+ verify(committer).discard(fileIO);
+ verify(recordWriter, atLeastOnce()).close();
+ }
+
+ private static FormatTable formatTable(
+ Path tablePath, LocalFileIO fileIO, long targetFileRowNum) {
+ Map<String, String> options = new HashMap<>();
+ options.put(PATH.key(), tablePath.toString());
+ options.put(FILE_FORMAT.key(), "csv");
+ options.put(TARGET_FILE_SIZE.key(), "1 gb");
+ options.put(TARGET_FILE_ROW_NUM.key(),
Long.toString(targetFileRowNum));
+ return FormatTable.builder()
+ .fileIO(fileIO)
+ .identifier(Identifier.create("test_db", "test_table"))
+ .rowType(RowType.of(DataTypes.INT()))
+ .partitionKeys(Collections.emptyList())
+ .location(tablePath.toString())
+ .format(FormatTable.Format.CSV)
+ .options(options)
+ .build();
+ }
+}
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 f57393b7ca..0ae551d033 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<InternalRow> tableWrite;
private transient Writer writer;
+ private static Map<String, String> dataEvolutionWriteOptions() {
+ // Data evolution requires a single output file: disable both size and
row rolling.
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.TARGET_FILE_SIZE.key(), "99999 G");
+ options.put(CoreOptions.TARGET_FILE_ROW_NUM.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<String> fieldNames =
dataType.getFieldNames().stream()
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 3b890cc0d6..20ee9ddb8c 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -390,6 +390,17 @@ class CoreOptions:
.with_description("The target file size for data files.")
)
+ TARGET_FILE_ROW_NUM: ConfigOption[int] = (
+ ConfigOptions.key("target-file-row-num")
+ .long_type()
+ .default_value((1 << 63) - 1)
+ .with_description(
+ "Target number of rows per newly written data file. PyPaimon
format-table "
+ "writers split files at this limit; file-store writers fail fast
when "
+ "this option is enabled."
+ )
+ )
+
BLOB_TARGET_FILE_SIZE: ConfigOption[MemorySize] = (
ConfigOptions.key("blob.target-file-size")
.memory_type()
@@ -1117,6 +1128,9 @@ class CoreOptions:
128 if has_primary_key else 256) if
default is None else MemorySize.parse(
default)).get_bytes()
+ def target_file_row_num(self):
+ return self.options.get(CoreOptions.TARGET_FILE_ROW_NUM)
+
def blob_target_file_size(self, default=None):
"""
Args:
diff --git a/paimon-python/pypaimon/schema/schema_manager.py
b/paimon-python/pypaimon/schema/schema_manager.py
index c9a38c532f..928e177a8b 100644
--- a/paimon-python/pypaimon/schema/schema_manager.py
+++ b/paimon-python/pypaimon/schema/schema_manager.py
@@ -452,6 +452,21 @@ def _validate_blob_fields(
)
+def _validate_options(options: dict):
+ core_options = CoreOptions(Options(options))
+ target_file_row_num = core_options.target_file_row_num()
+ max_value = CoreOptions.TARGET_FILE_ROW_NUM.default_value()
+ if target_file_row_num < 1:
+ raise ValueError(
+ f"{CoreOptions.TARGET_FILE_ROW_NUM.key()} should be at least 1"
+ )
+ if target_file_row_num > max_value:
+ raise ValueError(
+ f"{CoreOptions.TARGET_FILE_ROW_NUM.key()} "
+ f"should be at most {max_value}"
+ )
+
+
def _contains_blob_type(data_type: DataType) -> bool:
if is_blob_type(data_type):
return True
@@ -642,6 +657,7 @@ class SchemaManager:
return table_schema
def commit(self, new_schema: TableSchema) -> bool:
+ _validate_options(new_schema.options)
_validate_blob_fields(
new_schema.fields,
new_schema.options,
diff --git a/paimon-python/pypaimon/table/format/format_table_write.py
b/paimon-python/pypaimon/table/format/format_table_write.py
index 997594d92a..75b22d7186 100644
--- a/paimon-python/pypaimon/table/format/format_table_write.py
+++ b/paimon-python/pypaimon/table/format/format_table_write.py
@@ -22,6 +22,7 @@ from typing import Dict, List, Optional
import pyarrow
+from pypaimon.common.options.core_options import CoreOptions
from pypaimon.schema.data_types import PyarrowFieldParser
from pypaimon.table.format.format_commit_message import (
FormatTableCommitMessage,
@@ -99,6 +100,22 @@ class FormatTableWrite:
)
self._partition_only_value = opt.lower() == "true"
self._file_format = table.format()
+ self._target_file_row_num = CoreOptions.from_dict(
+ table.options()
+ ).target_file_row_num()
+ max_target_file_row_num = (
+ CoreOptions.TARGET_FILE_ROW_NUM.default_value()
+ )
+ if self._target_file_row_num < 1:
+ raise ValueError(
+ f"{CoreOptions.TARGET_FILE_ROW_NUM.key()} "
+ "should be at least 1"
+ )
+ if self._target_file_row_num > max_target_file_row_num:
+ raise ValueError(
+ f"{CoreOptions.TARGET_FILE_ROW_NUM.key()} "
+ f"should be at most {max_target_file_row_num}"
+ )
self._data_file_prefix = "data-"
self._suffix = {
"parquet": ".parquet",
@@ -156,7 +173,6 @@ class FormatTableWrite:
overwrite_this = (
self._overwrite
and dir_path not in self._overwritten_dirs
- and self.table.file_io.exists(dir_path)
)
if overwrite_this:
should_delete = (
@@ -167,12 +183,30 @@ class FormatTableWrite:
)
)
if should_delete:
- from pypaimon.table.format.format_table_commit import (
- _delete_data_files_in_path,
- )
- _delete_data_files_in_path(self.table.file_io, dir_path)
+ if self.table.file_io.exists(dir_path):
+ from pypaimon.table.format.format_table_commit import (
+ _delete_data_files_in_path,
+ )
+ _delete_data_files_in_path(self.table.file_io, dir_path)
self._overwritten_dirs.add(dir_path)
self.table.file_io.check_or_mkdirs(dir_path)
+
+ if data.num_rows <= self._target_file_row_num:
+ self._write_file(data, dir_path)
+ return
+
+ for offset in range(
+ 0, data.num_rows, self._target_file_row_num):
+ self._write_file(
+ data.slice(offset, self._target_file_row_num),
+ dir_path,
+ )
+
+ def _write_file(
+ self,
+ data: pyarrow.RecordBatch,
+ dir_path: str,
+ ) -> None:
file_name = f"{self._data_file_prefix}{uuid.uuid4().hex}{self._suffix}"
path = f"{dir_path}/{file_name}"
diff --git a/paimon-python/pypaimon/tests/reader_append_only_test.py
b/paimon-python/pypaimon/tests/reader_append_only_test.py
index 67bbfbedbd..0aa8981954 100644
--- a/paimon-python/pypaimon/tests/reader_append_only_test.py
+++ b/paimon-python/pypaimon/tests/reader_append_only_test.py
@@ -71,6 +71,18 @@ class AoReaderTest(unittest.TestCase):
actual = self._read_test_table(read_builder).sort_by('user_id')
self.assertEqual(actual, self.expected)
+ def test_target_file_row_num_fails_fast(self):
+ schema = Schema.from_pyarrow_schema(
+ self.pa_schema, partition_keys=['dt'],
+ options={'target-file-row-num': '5'})
+ self.catalog.create_table('default.test_target_file_row_num', schema,
False)
+ table = self.catalog.get_table('default.test_target_file_row_num')
+ with self.assertRaisesRegex(
+ NotImplementedError, 'row-count based file rolling'):
+ 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/tests/rest/rest_format_table_test.py
b/paimon-python/pypaimon/tests/rest/rest_format_table_test.py
index ccf6dd4e41..16f7630dd5 100644
--- a/paimon-python/pypaimon/tests/rest/rest_format_table_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_format_table_test.py
@@ -86,6 +86,78 @@ class RESTFormatTableTest(RESTBaseTest):
actual[col] = actual[col].astype(expected[col].dtype)
pd.testing.assert_frame_equal(actual, expected)
+ def test_format_table_write_rolls_by_row_count(self):
+ pa_schema = pa.schema([("id", pa.int32())])
+ schema = Schema.from_pyarrow_schema(
+ pa_schema,
+ options={
+ "type": "format-table",
+ "file.format": "parquet",
+ "target-file-row-num": "2",
+ },
+ )
+ table_name = "default.format_table_row_count_rolling"
+ self.rest_catalog.create_table(table_name, schema, False)
+ table = self.rest_catalog.get_table(table_name)
+
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ table_write.write_arrow(pa.table({"id": [1, 2, 3, 4, 5]}))
+ commit_messages = table_write.prepare_commit()
+ table_commit.commit(commit_messages)
+ table_write.close()
+ table_commit.close()
+
+ read_builder = table.new_read_builder()
+ splits = read_builder.new_scan().plan().splits()
+ self.assertEqual(3, len(splits))
+ splits_by_name = {
+ split.data_path().rsplit("/", 1)[-1]: split
+ for split in splits
+ }
+ row_counts = [
+ read_builder.new_read().to_arrow([
+ splits_by_name[path.rsplit("/", 1)[-1]]
+ ]).num_rows
+ for path in commit_messages[0].written_paths
+ ]
+ self.assertEqual([2, 2, 1], row_counts)
+
+ def test_format_table_overwrite_new_partition_in_multiple_batches(self):
+ pa_schema = pa.schema([
+ ("id", pa.int32()),
+ ("pt", pa.int32()),
+ ])
+ schema = Schema.from_pyarrow_schema(
+ pa_schema,
+ partition_keys=["pt"],
+ options={
+ "type": "format-table",
+ "file.format": "parquet",
+ "target-file-row-num": "2",
+ },
+ )
+ table_name = "default.format_table_overwrite_new_partition_batches"
+ self.rest_catalog.drop_table(table_name, True)
+ self.rest_catalog.create_table(table_name, schema, False)
+ table = self.rest_catalog.get_table(table_name)
+
+ write_builder = table.new_batch_write_builder().overwrite({"pt": 1})
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ table_write.write_arrow(pa.table({"id": [1, 2], "pt": [1, 1]}))
+ table_write.write_arrow(pa.table({"id": [3, 4], "pt": [1, 1]}))
+ table_commit.commit(table_write.prepare_commit())
+ table_write.close()
+ table_commit.close()
+
+ read_builder = table.new_read_builder()
+ actual = read_builder.new_read().to_pandas(
+ read_builder.new_scan().plan().splits()
+ ).sort_values(by="id")
+ self.assertEqual([1, 2, 3, 4], actual["id"].tolist())
+
def test_format_table_text_read_write(self):
pa_schema = pa.schema([("value", pa.string())])
schema = Schema.from_pyarrow_schema(
diff --git a/paimon-python/pypaimon/tests/schema_manager_test.py
b/paimon-python/pypaimon/tests/schema_manager_test.py
index d8718680e5..f58d8876c5 100644
--- a/paimon-python/pypaimon/tests/schema_manager_test.py
+++ b/paimon-python/pypaimon/tests/schema_manager_test.py
@@ -22,10 +22,12 @@ import unittest
from pypaimon.branch.branch_manager import BranchManager
from pypaimon.common.identifier import DEFAULT_MAIN_BRANCH
from pypaimon.common.file_io import FileIO
+from pypaimon.common.options.core_options import CoreOptions
from pypaimon.common.options.options import Options
from pypaimon.schema.data_types import (ArrayType, AtomicType, DataField,
MapType, RowType)
from pypaimon.schema.schema import Schema
+from pypaimon.schema.schema_change import SetOption
from pypaimon.schema.schema_manager import SchemaManager
from pypaimon.schema.table_schema import TableSchema
@@ -242,5 +244,56 @@ class TestBlobPartitionValidation(unittest.TestCase):
manager.create_table(schema)
+class TestOptionValidation(unittest.TestCase):
+
+ def setUp(self):
+ self.temp_dir = tempfile.mkdtemp()
+ self.file_io = FileIO.get(self.temp_dir, Options({}))
+ self.table_path = f"{self.temp_dir}/test_db.db/test_table"
+
+ def tearDown(self):
+ import shutil
+ shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+ @staticmethod
+ def _schema(options=None):
+ return Schema(
+ fields=[DataField(0, "id", AtomicType("INT"))],
+ options=options or {},
+ )
+
+ def test_create_rejects_invalid_target_file_row_num(self):
+ key = CoreOptions.TARGET_FILE_ROW_NUM.key()
+ max_value = CoreOptions.TARGET_FILE_ROW_NUM.default_value()
+ manager = SchemaManager(self.file_io, self.table_path)
+
+ invalid_values = [
+ ("0", "should be at least 1"),
+ (str(max_value + 1), f"should be at most {max_value}"),
+ ]
+ for value, message in invalid_values:
+ with self.subTest(value=value):
+ with self.assertRaisesRegex(ValueError, f"{key} {message}"):
+ manager.create_table(self._schema({key: value}))
+
+ def test_alter_rejects_invalid_target_file_row_num(self):
+ key = CoreOptions.TARGET_FILE_ROW_NUM.key()
+ max_value = CoreOptions.TARGET_FILE_ROW_NUM.default_value()
+ manager = SchemaManager(self.file_io, self.table_path)
+ manager.create_table(self._schema())
+
+ invalid_values = [
+ ("-1", "should be at least 1"),
+ (str(max_value + 1), f"should be at most {max_value}"),
+ ]
+ for value, message in invalid_values:
+ with self.subTest(value=value):
+ with self.assertRaisesRegex(RuntimeError, f"{key} {message}"):
+ manager.commit_changes([SetOption(key, value)])
+
+ self.assertEqual(manager.latest().id, 0)
+ self.assertNotIn(key, manager.latest().options)
+
+
if __name__ == '__main__':
unittest.main()
diff --git a/paimon-python/pypaimon/tests/table_update_test.py
b/paimon-python/pypaimon/tests/table_update_test.py
index 793871e11a..afbb18b469 100644
--- a/paimon-python/pypaimon/tests/table_update_test.py
+++ b/paimon-python/pypaimon/tests/table_update_test.py
@@ -945,8 +945,7 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
)
def test_update_with_large_file(self):
- """Even with a tiny ``target-file-size`` the update produces one
- output file per first_row_id group (rolling is disabled internally)."""
+ """Updates disable both size- and row-based rolling."""
from pypaimon.schema.schema_change import SetOption
N = 5000
@@ -966,7 +965,10 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
}))
self.catalog.alter_table(
- table_identifier, [SetOption('target-file-size', '10kb')]
+ table_identifier, [
+ SetOption('target-file-size', '10kb'),
+ SetOption('target-file-row-num', '1'),
+ ]
)
table = self.catalog.get_table(table_identifier)
diff --git a/paimon-python/pypaimon/write/file_store_write.py
b/paimon-python/pypaimon/write/file_store_write.py
index 0895c47bf1..92fa5259d7 100644
--- a/paimon-python/pypaimon/write/file_store_write.py
+++ b/paimon-python/pypaimon/write/file_store_write.py
@@ -56,9 +56,12 @@ class FileStoreWrite:
f"-s-{random.randint(0, 2 ** 31 - 2)}-w-"))
def disable_rolling(self):
- """Disable file rolling by setting target_file_size to max."""
+ """Disable size- and row-based file rolling."""
+ max_value = CoreOptions.TARGET_FILE_ROW_NUM.default_value()
self.options.set(
- CoreOptions.TARGET_FILE_SIZE, str(2 ** 63 - 1))
+ CoreOptions.TARGET_FILE_SIZE, str(max_value))
+ self.options.set(
+ CoreOptions.TARGET_FILE_ROW_NUM, str(max_value))
def write(self, partition: Tuple, bucket: int, data: pa.RecordBatch):
key = (partition, bucket)
@@ -89,6 +92,19 @@ class FileStoreWrite:
writer.write(data.to_batches()[0])
def _create_data_writer(self, partition: Tuple, bucket: int, options:
CoreOptions) -> DataWriter:
+ row_limit = options.target_file_row_num()
+ max_value = CoreOptions.TARGET_FILE_ROW_NUM.default_value()
+ if row_limit < 1:
+ raise ValueError(
+ "target-file-row-num should be at least 1")
+ if row_limit > max_value:
+ raise ValueError(
+ f"target-file-row-num should be at most {max_value}")
+ if row_limit != max_value:
+ raise NotImplementedError(
+ "target-file-row-num is set on this table but pypaimon
file-store writers do not support "
+ "row-count based file rolling yet; unset it or write with
Java/Flink/Spark.")
+
def max_seq_number():
return self._seq_number_stats(partition).get(bucket, 1)
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 e99afc95b6..5536d4363d 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_ROW_NUM.key() -> Long.MaxValue.toString)
+ paimonTable.copy(writeOptions.asJava)
+ }
def writePartialFields(
data: DataFrame,
