diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 0776eaf9e369..2841618e7c34 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-row-num
+ 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. PyPaimon file-store writers do not support this option and fail fast when it is enabled. 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..e0c7fd4004c7 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 InlineElement getDescription() { text("append table: the default value is 256 MB.")) .build()); + public static final ConfigOption 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 COMPACTION_SMALL_FILE_RATIO = key("compaction.small-file-ratio") .doubleType() @@ -3325,6 +3340,10 @@ public long targetFileSize(boolean hasPrimaryKey) { .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 ea81dea50f60..5f47789a2619 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 writeCols; private final DataFilePathFactory pathFactory; @@ -112,6 +113,7 @@ public AppendOnlyWriter( long targetFileSize, long blobTargetFileSize, long vectorTargetFileSize, + long targetFileRowNum, RowType writeSchema, @Nullable List writeCols, long maxSequenceNumber, @@ -139,6 +141,7 @@ public AppendOnlyWriter( 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 @@ private RollingFileWriter createRollingRowWriter() { targetFileSize, blobTargetFileSize, vectorTargetFileSize, + targetFileRowNum, writeSchema, pathFactory, seqNumCounterProvider, @@ -350,7 +354,8 @@ private RollingFileWriter createRollingRowWriter() { 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 9b8897862a8b..3cf08fc2639c 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 targetFileRowNum; // State management private final List closedWriters; @@ -121,6 +122,7 @@ public class DedicatedFormatRollingFileWriter RollingFileWriterImpl, List> vectorStoreWriter; private long recordCount = 0; + private long currentFileRecordCount = 0; private boolean closed = false; public DedicatedFormatRollingFileWriter( @@ -131,6 +133,7 @@ public DedicatedFormatRollingFileWriter( long targetFileSize, long blobTargetFileSize, long vectorTargetFileSize, + long targetFileRowNum, RowType writeSchema, DataFilePathFactory pathFactory, Supplier seqNumCounterSupplier, @@ -141,7 +144,12 @@ public 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 DedicatedFormatRollingFileWriter( statsDenseStore, pathFactory.isExternalPath(), vectorStoreColumnNames), - targetFileSize), + targetFileSize, + Long.MAX_VALUE), vectorStoreProjection); } @@ -352,8 +361,9 @@ public void write(InternalRow row) throws IOException { currentWriter.write(row); } recordCount++; + currentFileRecordCount++; - if (currentWriter != null && rollingFile()) { + if (rollingFile()) { closeCurrentWriter(); } } catch (Throwable e) { @@ -416,11 +426,19 @@ public void abort() { } } - /** Checks if the current file should be rolled based on size and record count. */ + /** + * Checks if the current file should be rolled. The row cap applies even when there is no main + * writer (all fields dedicated), so blob/vector writers roll together. + */ private boolean rollingFile() throws IOException { - return currentWriter - .writer() - .reachTargetSize(recordCount % CHECK_ROLLING_RECORD_CNT == 0, targetFileSize); + if (currentFileRecordCount >= targetFileRowNum) { + return true; + } + return currentWriter != null + && currentWriter + .writer() + .reachTargetSize( + recordCount % CHECK_ROLLING_RECORD_CNT == 0, targetFileSize); } /** @@ -455,6 +473,7 @@ private void closeCurrentWriter() throws IOException { // Reset current writer currentWriter = null; + currentFileRecordCount = 0; } /** Closes the main writer and returns its metadata. */ @@ -471,6 +490,7 @@ private List closeBlobWriter() throws IOException { } blobWriter.close(); List results = blobWriter.result(); + closedWriters.addAll(blobWriter.drainAbortExecutors()); blobWriter = null; return results; } @@ -482,6 +502,7 @@ private List closeVectorStoreWriter() throws IOException { } vectorStoreWriter.close(); List 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 39d6c6eac833..7a0901a75f4c 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.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 List result() throws IOException { return results; } + List drainAbortExecutors() { + List abortExecutors = new ArrayList<>(); + for (BlobProjectedFileWriter blobWriter : blobWriters) { + abortExecutors.addAll(blobWriter.writer().drainAbortExecutors()); + } + return abortExecutors; + } + private static class BlobProjectedFileWriter extends ProjectedFileWriter< RollingFileWriterImpl, List> { @@ -138,7 +147,9 @@ public BlobProjectedFileWriter( Supplier> 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 77c3eaa91528..0019c4e17f92 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_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 6c97bb980880..67992a755a8e 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 void createTable(Identifier identifier, Schema schema, boolean ignoreIfEx } copyTableDefaultOptions(schema.options()); + validateCreateTable(schema, false); switch (Options.fromMap(schema.options()).get(TYPE)) { case TABLE: @@ -516,6 +517,7 @@ public void replaceTable(Identifier identifier, Schema newSchema, boolean ignore validateCreateTable(newSchema, false); validateCustomTablePath(newSchema.options()); copyTableDefaultOptions(newSchema.options()); + validateCreateTable(newSchema, false); Table existing; try { 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 ee793980d392..0ee552d887f6 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 static void validateCreateTable(Schema schema, boolean dataTokenEnabled) "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 46fd14390672..d5abb480a587 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 List rollingWrite( Iterator entries, long sequenceNumber, Content content) { RollingFileWriterImpl 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 f081abf310b9..96e297c327f9 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.fs.FileIO; import org.apache.paimon.fs.TwoPhaseOutputStream; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.Preconditions; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -32,31 +33,38 @@ import java.util.List; import java.util.function.Supplier; -/** - * Format table's writer to roll over to a new file if the current size exceed the target file size. - */ +/** Format table's writer which rolls files by target size or target row count. */ public class FormatTableRollingFileWriter implements AutoCloseable { private static final Logger LOG = LoggerFactory.getLogger(FormatTableRollingFileWriter.class); private static final int CHECK_ROLLING_RECORD_CNT = 1000; + private final FileIO fileIO; private final Supplier writerFactory; private final long targetFileSize; + private final long targetFileRowNum; private final List closedWriters; private final List 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 FormatTableRollingFileWriter( pathFactory.newPath(), fileCompression); this.targetFileSize = targetFileSize; + this.targetFileRowNum = targetFileRowNum; this.closedWriters = new ArrayList<>(); this.committers = new ArrayList<>(); } @@ -81,9 +90,11 @@ public void write(InternalRow row) throws IOException { currentWriter.write(row); recordCount += 1; + currentFileRecordCount += 1; boolean needRolling = - currentWriter.reachTargetSize( - recordCount % CHECK_ROLLING_RECORD_CNT == 0, targetFileSize); + currentFileRecordCount >= targetFileRowNum + || currentWriter.reachTargetSize( + recordCount % CHECK_ROLLING_RECORD_CNT == 0, targetFileSize); if (needRolling) { closeCurrentWriter(); } @@ -109,15 +120,26 @@ private void closeCurrentWriter() throws IOException { } currentWriter = null; + currentFileRecordCount = 0; } public void abort() { if (currentWriter != null) { currentWriter.abort(); + currentWriter = null; + } + for (TwoPhaseOutputStream.Committer committer : committers) { + try { + committer.discard(fileIO); + } catch (Throwable e) { + LOG.warn("Exception occurs when discarding file {}.", committer.targetPath(), e); + } } + committers.clear(); for (FileWriterAbortExecutor abortExecutor : closedWriters) { abortExecutor.abort(); } + closedWriters.clear(); } public List 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 dfe875120dcf..5644aab37512 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 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 createRollingChangelogFileWriter(int level) { @@ -156,7 +160,8 @@ public RollingFileWriter createRollingChangelogFileWrite FileSource.APPEND, pathFactory.isExternalPath()); }, - suggestedFileSize); + suggestedFileSize, + Long.MAX_VALUE); } public RollingFileWriter createRollingClusteringFileWriter() { @@ -167,7 +172,8 @@ public RollingFileWriter createRollingClusteringFileWrit 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 299a4f97421e..0ebdc5feb790 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 implements RollingFileWriter { private final Supplier> writerFactory; private final long targetFileSize; + private final long targetFileRowNum; 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) { + Supplier> 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 long targetFileSize() { } 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 void write(T row) throws IOException { currentWriter.write(row); recordCount += 1; + currentFileRecordCount += 1; if (rollingFile(false)) { closeCurrentWriter(); @@ -101,6 +112,7 @@ public void writeBundle(BundleRecords bundle) throws IOException { currentWriter.writeBundle(bundle); recordCount += bundle.rowCount(); + currentFileRecordCount += bundle.rowCount(); if (rollingFile(true)) { closeCurrentWriter(); @@ -132,6 +144,7 @@ protected void closeCurrentWriter() throws IOException { currentWriter.abortExecutor().ifPresent(closedWriters::add); results.add(currentWriter.result()); currentWriter = null; + currentFileRecordCount = 0; } @Override @@ -155,6 +168,14 @@ public List result() { return results; } + /** Transfers ownership of abort executors for closed files to the caller. */ + public List drainAbortExecutors() { + Preconditions.checkState(closed, "Cannot drain abort executors unless close all writers."); + List 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 0da1badf6229..e9874ef9b9e2 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 RowDataRollingFileWriter( boolean asyncFileWrite, boolean statsDenseStore, @Nullable List writeCols, - @Nullable FileFormat rowSidecarFormat) { + @Nullable FileFormat rowSidecarFormat, + long targetFileRowNum) { super( () -> { Path dataPath = pathFactory.newPath(); @@ -76,6 +77,7 @@ public RowDataRollingFileWriter( 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 79687c9368e7..08ebe634a6d5 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 @@ -379,6 +379,7 @@ public void createTable(Identifier identifier, Schema schema, boolean ignoreIfEx 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 527568f7f088..71b8020e3b7b 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 List write(List entries) { public RollingFileWriter 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 7dceb12e5c8b..c3ad22b65c7b 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.targetFileRowNum(), writeType, writeCols, restoredMaxSeqNumber, @@ -312,7 +313,8 @@ private RowDataRollingFileWriter createRollingFileWriter( 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 86256a93425f..3db1052c4c1a 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 void createTable(Identifier identifier, Schema schema, boolean ignoreIfEx 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 void replaceTable(Identifier identifier, Schema newSchema, boolean ignore 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 4914ef1b40f5..1c6ba3a0fb07 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.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 23061e0d1437..9f18ee5758a4 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.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 void write(BinaryRow partition, InternalRow data) throws Exception { } public void close() throws Exception { - writers.clear(); + try { + IOUtils.closeAll(writers.values()); + } finally { + writers.clear(); + } } public List prepareCommit() throws Exception { - List commitMessages = new ArrayList<>(); - for (FormatTableRecordWriter writer : writers.values()) { - List commiters = writer.closeAndGetCommitters(); - for (TwoPhaseOutputStream.Committer committer : commiters) { - TwoPhaseCommitMessage twoPhaseCommitMessage = new TwoPhaseCommitMessage(committer); - commitMessages.add(twoPhaseCommitMessage); + List 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 commitMessages = new ArrayList<>(); + for (TwoPhaseOutputStream.Committer committer : committers) { + commitMessages.add(new TwoPhaseCommitMessage(committer)); } return commitMessages; } @@ -118,6 +140,7 @@ private FormatTableRecordWriter createWriter(BinaryRow partition) { 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 b611f2805de4..83677d9448c2 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 FormatTableRecordWriter( this.writeSchema = writeSchema; this.fileFormat = fileFormat; this.targetFileSize = targetFileSize; + this.targetFileRowNum = targetFileRowNum; } public void write(InternalRow data) throws Exception { @@ -82,6 +85,12 @@ public void close() throws Exception { 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 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..f03e562f12bc 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 void setUp() throws IOException { TARGET_FILE_SIZE, TARGET_FILE_SIZE, TARGET_FILE_SIZE, + Long.MAX_VALUE, SCHEMA, pathFactory, () -> seqNumCounter, @@ -147,6 +148,73 @@ 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 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 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 void testBundleWritingPreservesMainFileIndexSideEffects() throws IOExcept TARGET_FILE_SIZE, TARGET_FILE_SIZE, TARGET_FILE_SIZE, + Long.MAX_VALUE, SCHEMA, pathFactory, () -> seqNumCounter, @@ -252,6 +321,7 @@ public void testBlobTargetFileSize() throws IOException { 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 void testBundleWritingRespectsBlobTargetFileSize() throws IOException { 128 * 1024 * 1024, blobTargetFileSize, 128 * 1024 * 1024, + Long.MAX_VALUE, SCHEMA, new DataFilePathFactory( new Path(tempDir + "/bundle-blob-size-test"), @@ -396,6 +467,7 @@ void testBlobFileNameFormatWithSharedUuid() throws IOException { 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 @@ void testBlobFileNameFormatWithSharedUuidNonDescriptorMode() throws IOException 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 @@ void testBlobStatsSchemaWithCustomColumnName() throws IOException { 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 156c79672ec4..bed8ef7b936f 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.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 void setUp() throws IOException { TARGET_FILE_SIZE, TARGET_FILE_SIZE, VECTOR_TARGET_FILE_SIZE, + Long.MAX_VALUE, SCHEMA, pathFactory, () -> seqNumCounter, @@ -147,6 +150,42 @@ public void testMultipleWrites() throws Exception { 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 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 void testBundleWritingRespectsVectorTargetFileSize() throws Exception { 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 void testBundleWritingPreservesMainFileIndexSideEffects() throws Exceptio TARGET_FILE_SIZE, TARGET_FILE_SIZE, VECTOR_TARGET_FILE_SIZE, + Long.MAX_VALUE, SCHEMA, pathFactory, () -> seqNumCounter, @@ -367,6 +408,7 @@ public void testVectorStoreNoBlob() throws Exception { 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 3ffb6a1ecbde..417ccacdddb2 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.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 @@ /** 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 094ede3847ab..5e186cdf10df 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.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 void testCreateTableCaseSensitive() throws Exception { 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 4b5c73038664..ad8bfb512a44 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_ROW_NUM, 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..bbfb221fbee3 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 targetFileRowNum) { 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, + targetFileRowNum); } @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()); @@ -158,7 +206,8 @@ public void testWriteRowSidecar() throws IOException { true, false, null, - rowFormat); + rowFormat, + Long.MAX_VALUE); writer.write(GenericRow.of(1)); writer.close(); @@ -208,7 +257,8 @@ public void testWriteRowSidecarWithBundle() throws IOException { 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 36b8668bcc1f..ab94f33b3a63 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 testTargetFileRowNumMustBePositive() { + Map 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 7766da98de24..4f080f96675d 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_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 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_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 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-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 000000000000..3864f39ad6cf --- /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 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 dataFiles = + messages.stream() + .map( + message -> + ((TwoPhaseCommitMessage) message) + .getCommitter() + .targetPath()) + .collect(Collectors.toList()); + try (BatchTableCommit commit = writeBuilder.newCommit()) { + commit.commit(messages); + } + + List 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 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 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 f57393b7ca27..0ae551d0330b 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_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 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 3b890cc0d62d..20ee9ddb8c25 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 @@ def target_file_size(self, has_primary_key, default=None): 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 c9a38c532fd6..928e177a8b8b 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 @@ def create_table(self, schema: Schema) -> TableSchema: 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 997594d92ac8..75b22d7186ad 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 @@ 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 @@ def __init__( ) 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 @@ def _write_single_batch( 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 @@ def _write_single_batch( ) ) 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 67bbfbedbdec..0aa898195480 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 @@ 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_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 ccf6dd4e412f..16f7630dd53a 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 @@ def test_format_table_read_write(self, file_format): 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 d8718680e5dc..f58d8876c548 100644 --- a/paimon-python/pypaimon/tests/schema_manager_test.py +++ b/paimon-python/pypaimon/tests/schema_manager_test.py @@ -22,10 +22,12 @@ 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 @@ def test_create_table_still_rejects_primary_key_map_blob(self): 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 793871e11a24..afbb18b469db 100644 --- a/paimon-python/pypaimon/tests/table_update_test.py +++ b/paimon-python/pypaimon/tests/table_update_test.py @@ -945,8 +945,7 @@ def test_large_table_partial_column_updates(self): ) 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 @@ def test_update_with_large_file(self): })) 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 0895c47bf17e..92fa5259d7ae 100644 --- a/paimon-python/pypaimon/write/file_store_write.py +++ b/paimon-python/pypaimon/write/file_store_write.py @@ -56,9 +56,12 @@ def __init__(self, table, commit_user): 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 @@ def write_row(self, partition: Tuple, bucket: int, row, values_by_name: dict): 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 e99afc95b6e0..5536d4363d82 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,