From 4b83e9ee5e015f9c7a8f150fd3119e1aafe263e2 Mon Sep 17 00:00:00 2001 From: Junbo Wang Date: Mon, 19 Oct 2026 11:22:39 +0800 Subject: [PATCH 1/4] [lake/paimon] Optimize historical split initialization (#4656) --- .../lake/paimon/utils/PaimonConversions.java | 11 ++ .../lake/paimon/PaimonLakeCatalogTest.java | 75 ++++++++-- .../paimon/tiering/PaimonTieringTest.java | 140 +++++++++++++++--- website/docs/engine-flink/options.md | 2 +- .../data-distribution/partitioning.md | 9 ++ 5 files changed, 205 insertions(+), 32 deletions(-) diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java index 96c9027839d..e101585cc3b 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java @@ -20,6 +20,8 @@ import org.apache.fluss.annotation.VisibleForTesting; import org.apache.fluss.config.AutoPartitionTimeUnit; import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.TableConfig; import org.apache.fluss.exception.InvalidConfigException; import org.apache.fluss.exception.InvalidTableException; import org.apache.fluss.lake.paimon.source.FlussRowAsPaimonRow; @@ -397,6 +399,15 @@ public static Schema toPaimonSchema(TableDescriptor tableDescriptor) { options.set( CoreOptions.CHANGELOG_PRODUCER.key(), CoreOptions.ChangelogProducer.INPUT.toString()); + TableConfig tableConfig = + new TableConfig(Configuration.fromMap(tableDescriptor.getProperties())); + if (tableConfig.isHistoricalPartitionEnabled() + && !tableConfig.isDataLakeAutoCompaction()) { + // Use the snapshot's maximum sequence number to avoid repeated manifest scans + // when historical splits write to multiple partitions and buckets. + // Use the literal key for compatibility with Paimon 1.4. + options.set("write.sequence-number-init-mode", "snapshot"); + } } // set partition keys diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java index 083b73ee7c4..e82c1c42117 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java @@ -44,6 +44,10 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import javax.annotation.Nullable; import java.io.File; import java.util.Arrays; @@ -51,12 +55,15 @@ import java.util.List; import java.util.Optional; +import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION; import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_ENABLED; import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_FORMAT; +import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_ENABLED; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.LAKESTREAM_ENABLED_OPTION_KEY; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon; import static org.apache.fluss.lake.paimon.utils.PaimonTableValidation.isPaimonSchemaCompatible; +import static org.apache.paimon.CoreOptions.WRITE_SEQUENCE_NUMBER_INIT_MODE; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -836,20 +843,64 @@ void testAlterTableAddColumnWhenPaimonSchemaNotMatch() throws Exception { changes)); } - @Test - void testCreateTableSetsLakeStreamEnabledForCleanTable() throws Exception { - String database = "test_create_lakestream_db"; - String tableName = "test_create_lakestream_table"; - TablePath tablePath = TablePath.of(database, tableName); - Identifier identifier = Identifier.create(database, tableName); - - // getTableDescriptor sets table.datalake.enabled=true, so the clean table advertises its - // LakeStream state to Paimon - flussPaimonCatalog.createTable( - tablePath, getTableDescriptor(FLUSS_SCHEMA), LAKE_CATALOG_CONTEXT); + @ParameterizedTest + @CsvSource({ + "true,true,,,snapshot", + "true,true,false,,snapshot", + "true,true,true,,", + "true,false,false,,", + "false,true,false,,", + "false,true,true,,", + "true,true,false,scan,scan", + "true,true,true,snapshot,snapshot" + }) + void testCreateTableOptions( + boolean primaryKey, + boolean historicalPartition, + @Nullable Boolean autoCompaction, + @Nullable String sequenceMode, + @Nullable String expectedSequenceMode) + throws Exception { + TablePath tablePath = TablePath.of("test_db", "test_create_options"); + + Schema.Builder schema = Schema.newBuilder().fromSchema(FLUSS_SCHEMA); + if (primaryKey) { + schema.primaryKey("id", "address"); + } + TableDescriptor.Builder builder = + TableDescriptor.builder() + .schema(schema.build()) + .partitionedBy("address") + .distributedBy(3) + .property(TABLE_DATALAKE_ENABLED, true) + .property(TABLE_DATALAKE_FORMAT.key(), "paimon") + .property(TABLE_DATALAKE_HISTORICAL_PARTITION_ENABLED, historicalPartition); + if (autoCompaction != null) { + builder.property(TABLE_DATALAKE_AUTO_COMPACTION, autoCompaction); + } + String option = "paimon." + WRITE_SEQUENCE_NUMBER_INIT_MODE.key(); + if (sequenceMode != null) { + builder.customProperty(option, sequenceMode); + } + TableDescriptor descriptor = builder.build(); + TestingLakeCatalogContext context = new TestingLakeCatalogContext(descriptor); + flussPaimonCatalog.createTable(tablePath, descriptor, context); - Table table = flussPaimonCatalog.getPaimonCatalog().getTable(identifier); + Table table = flussPaimonCatalog.getPaimonCatalog().getTable(toPaimon(tablePath)); assertThat(table.options()).containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true"); + assertThat(table.options().get(WRITE_SEQUENCE_NUMBER_INIT_MODE.key())) + .isEqualTo(expectedSequenceMode); + + if ("snapshot".equals(expectedSequenceMode)) { + flussPaimonCatalog.alterTable( + tablePath, Collections.singletonList(TableChange.set(option, "scan")), context); + assertThat( + flussPaimonCatalog + .getPaimonCatalog() + .getTable(toPaimon(tablePath)) + .options()) + .containsEntry(WRITE_SEQUENCE_NUMBER_INIT_MODE.key(), "scan"); + } } @Test diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java index ae798236f18..e5005254a8c 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java @@ -52,12 +52,16 @@ import org.apache.paimon.catalog.CatalogFactory; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.fs.FileIO; import org.apache.paimon.manifest.ManifestCommittable; import org.apache.paimon.options.Options; import org.apache.paimon.partition.Partition; import org.apache.paimon.schema.Schema; import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FileStoreTableFactory; +import org.apache.paimon.table.sink.CommitMessageImpl; import org.apache.paimon.table.sink.TableCommitImpl; +import org.apache.paimon.table.sink.TableWriteImpl; import org.apache.paimon.table.source.DataSplit; import org.apache.paimon.table.source.ReadBuilder; import org.apache.paimon.table.source.Split; @@ -69,6 +73,7 @@ import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.MethodSource; import javax.annotation.Nullable; @@ -88,6 +93,7 @@ import static org.apache.fluss.lake.committer.LakeCommitter.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon; +import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimonSchema; import static org.apache.fluss.lake.writer.LakeTieringFactory.FLUSS_LAKE_TIERING_COMMIT_USER; import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME; import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME; @@ -99,8 +105,17 @@ import static org.apache.fluss.record.ChangeType.UPDATE_BEFORE; import static org.apache.fluss.record.TestData.DEFAULT_REMOTE_DATA_DIR; import static org.apache.fluss.utils.PartitionUtils.HISTORICAL_PARTITION_VALUE; +import static org.apache.paimon.CoreOptions.WRITE_SEQUENCE_NUMBER_INIT_MODE; import static org.apache.paimon.table.sink.BatchWriteBuilder.COMMIT_IDENTIFIER; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.argThat; +import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.clearInvocations; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; /** The UT for tiering to Paimon via {@link PaimonLakeTieringFactory}. */ class PaimonTieringTest { @@ -230,15 +245,24 @@ void testTieringWriteTable(boolean isPrimaryKeyTable, boolean isPartitioned) thr } } - @Test - void testHistoricalPrimaryKeyTiering() throws Exception { + @ParameterizedTest + @CsvSource({"false,", "true,", "false,scan"}) + void testHistoricalPrimaryKeyTiering(boolean autoCompaction, @Nullable String sequenceMode) + throws Exception { TablePath tablePath = TablePath.of("paimon", "test_historical_primary_key"); - TableInfo tableInfo = createHistoricalTable(tablePath, true); + Configuration properties = new Configuration(); + properties.set(ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION, autoCompaction); + if (sequenceMode != null) { + properties.setString("paimon." + WRITE_SEQUENCE_NUMBER_INIT_MODE.key(), sequenceMode); + } + TableInfo tableInfo = createHistoricalTable(tablePath, true, properties.toMap()); long timestamp = 1_000L; List records = Arrays.asList( historicalRecord(0L, timestamp, 1, "20240101", INSERT), - historicalRecord(1L, timestamp, 1, "20240102", INSERT)); + historicalRecord(1L, timestamp, 1, "20240101", INSERT), + historicalRecord(2L, timestamp, 1, "20240101", INSERT), + historicalRecord(3L, timestamp, 1, "20240102", INSERT)); PaimonWriteResult writeResult; try (LakeWriter lakeWriter = @@ -255,21 +279,77 @@ void testHistoricalPrimaryKeyTiering() throws Exception { serializer.deserialize(serializer.getVersion(), serializer.serialize(writeResult)); assertHistoricalPartitions(writeResult); - try (LakeCommitter committer = - createLakeCommitter(tablePath, tableInfo, new Configuration())) { - committer.commit( - committer.toCommittable(Collections.singletonList(writeResult)), - new CommitContext(Collections.emptyMap(), Collections.emptyMap())); - } + commitWriteResult(tablePath, tableInfo, writeResult); assertThat(paimonCatalog.listPartitions(toPaimon(tablePath))) .extracting(partition -> partition.spec().get("c3")) .containsExactlyInAnyOrder("20240101", "20240102"); + boolean snapshotSequence = !autoCompaction && sequenceMode == null; + assertMaxSequenceNumber(tablePath, 1L, snapshotSequence ? 2L : null); + FileStoreTable table = (FileStoreTable) paimonCatalog.getTable(toPaimon(tablePath)); + FileIO fileIO = spy(table.fileIO()); + FileStoreTable uncachedTable = FileStoreTableFactory.create(fileIO, table.location()); + Catalog catalog = mock(Catalog.class); + when(catalog.getTable(toPaimon(tablePath))).thenReturn(uncachedTable); + PaimonCatalogProvider provider = mock(PaimonCatalogProvider.class); + when(provider.get()).thenReturn(catalog); + WriterInitContext context = mock(WriterInitContext.class); + when(context.tableInfo()).thenReturn(tableInfo); + when(context.tableBucket()).thenReturn(new TableBucket(0, 1L, 0)); + when(context.bucketCount()).thenReturn(1); + + long snapshotId = 1; + // The persisted option applies to both historical and normal partition writers. + for (String partition : Arrays.asList("20240102", "20240101")) { + when(context.partition()) + .thenReturn(snapshotId == 1 ? HISTORICAL_PARTITION_VALUE : partition); + clearInvocations(fileIO); + try (PaimonLakeWriter writer = new PaimonLakeWriter(provider, context)) { + writer.write(historicalRecord(4L, timestamp + 1_000L, 1, partition, INSERT)); + verify(fileIO, snapshotSequence ? never() : atLeastOnce()) + .newInputStream(argThat(path -> path.getName().startsWith("manifest-"))); + writeResult = writer.complete(); + } + commitWriteResult(tablePath, tableInfo, writeResult); + snapshotId++; + assertMaxSequenceNumber( + tablePath, snapshotId, snapshotSequence ? snapshotId + 1 : null); + } + + // An external compactor inherits the option without Fluss runtime overrides. + try (TableWriteImpl compactor = table.newWrite("external-compactor"); + TableCommitImpl commit = table.newCommit("external-compactor")) { + compactor.compact( + ((CommitMessageImpl) writeResult.commitMessages().get(0)).partition(), 0, true); + commit.commit(compactor.prepareCommit()); + } + Snapshot compactSnapshot = table.snapshotManager().latestSnapshot(); + assertThat(compactSnapshot.commitKind()).isEqualTo(Snapshot.CommitKind.COMPACT); + assertMaxSequenceNumber(tablePath, compactSnapshot.id(), snapshotSequence ? 4L : null); + + LogRecord updated = historicalRecord(6L, timestamp + 2_000L, 1, "20240102", INSERT); + when(context.partition()).thenReturn(HISTORICAL_PARTITION_VALUE); + clearInvocations(fileIO); + try (PaimonLakeWriter writer = new PaimonLakeWriter(provider, context)) { + writer.write(historicalRecord(5L, timestamp + 2_000L, 1, "20240101", DELETE)); + writer.write(updated); + verify(fileIO, snapshotSequence ? never() : atLeastOnce()) + .newInputStream(argThat(path -> path.getName().startsWith("manifest-"))); + writeResult = writer.complete(); + } + commitWriteResult(tablePath, tableInfo, writeResult); + try (CloseableIterator deleted = + getPaimonRows(tablePath, "20240101", true, 0); + CloseableIterator remaining = + getPaimonRows(tablePath, "20240102", true, 0)) { + assertThat(deleted.hasNext()).isFalse(); + verifyTableRecords(remaining, Collections.singletonList(updated), 0, "20240102"); + } } @Test void testHistoricalArrowBatchTiering() throws Exception { TablePath tablePath = TablePath.of("paimon", "test_historical_arrow"); - TableInfo tableInfo = createHistoricalTable(tablePath, false); + TableInfo tableInfo = createHistoricalTable(tablePath, false, Collections.emptyMap()); long baseOffset = 10L; long timestamp = 1_000L; List records = @@ -1181,6 +1261,16 @@ public Configuration flussClientConfig() { }); } + private void commitWriteResult( + TablePath tablePath, TableInfo tableInfo, PaimonWriteResult result) throws Exception { + try (LakeCommitter committer = + createLakeCommitter(tablePath, tableInfo, new Configuration())) { + committer.commit( + committer.toCommittable(Collections.singletonList(result)), + new CommitContext(Collections.emptyMap(), Collections.emptyMap())); + } + } + private long commitEmptySnapshot(TablePath tablePath, TableInfo tableInfo, String offsets) throws Exception { try (LakeCommitter lakeCommitter = @@ -1196,6 +1286,16 @@ private long commitEmptySnapshot(TablePath tablePath, TableInfo tableInfo, Strin } } + private void assertMaxSequenceNumber( + TablePath tablePath, long snapshotId, @Nullable Long expected) throws Exception { + Map properties = getSnapshotProperties(tablePath, snapshotId); + assertThat( + properties == null + ? null + : properties.get("sequence.generation.max-sequence-number")) + .isEqualTo(expected == null ? null : expected.toString()); + } + private void createTable( TablePath tablePath, boolean isPrimaryTable, @@ -1244,15 +1344,9 @@ private TableInfo createNonPartitionedLogTable(TablePath tablePath) throws Excep return TableInfo.of(tablePath, 0, 1, descriptor, DEFAULT_REMOTE_DATA_DIR, 1L, 1L); } - private TableInfo createHistoricalTable(TablePath tablePath, boolean isPrimaryKeyTable) + private TableInfo createHistoricalTable( + TablePath tablePath, boolean isPrimaryKeyTable, Map properties) throws Exception { - createTable( - tablePath, - isPrimaryKeyTable, - true, - isPrimaryKeyTable ? 1 : null, - Collections.emptyMap()); - org.apache.fluss.metadata.Schema.Builder schemaBuilder = org.apache.fluss.metadata.Schema.newBuilder() .column("c1", org.apache.fluss.types.DataTypes.INT()) @@ -1267,7 +1361,15 @@ private TableInfo createHistoricalTable(TablePath tablePath, boolean isPrimaryKe .partitionedBy("c3") .distributedBy(1) .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true) + .property(ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_ENABLED, true) + .properties(properties) .build(); + createTable( + tablePath, + isPrimaryKeyTable, + true, + isPrimaryKeyTable ? 1 : null, + toPaimonSchema(descriptor).options()); return TableInfo.of(tablePath, 0, 1, descriptor, DEFAULT_REMOTE_DATA_DIR, 1L, 1L); } diff --git a/website/docs/engine-flink/options.md b/website/docs/engine-flink/options.md index e6286a2009d..f1e4e1f807b 100644 --- a/website/docs/engine-flink/options.md +++ b/website/docs/engine-flink/options.md @@ -88,7 +88,7 @@ See more details about [ALTER TABLE ... SET](engine-flink/ddl.md#set-properties) | table.log.tiered.local-segments | Integer | 2 | The number of log segments to retain in local for each table when log tiered storage is enabled. It must be greater that 0. The default is 2. | | table.datalake.enabled | Boolean | false | Whether enable lakehouse storage for the table. Disabled by default. When this option is set to ture and the datalake tiering service is up, the table will be tiered and compacted into datalake format stored on lakehouse storage. | | table.datalake.format | Enum | (None) | The data lake format of the table specifies the tiered Lakehouse storage format. Currently, supported formats are `paimon`, `iceberg`, and `lance`. In the future, more kinds of data lake format will be supported, such as DeltaLake or Hudi. Once the `table.datalake.format` property is configured, Fluss adopts the key encoding and bucketing strategy used by the corresponding data lake format. This ensures consistency in key encoding and bucketing, enabling seamless **Union Read** functionality across Fluss and Lakehouse. The `table.datalake.format` can be pre-defined before enabling `table.datalake.enabled`. This allows the data lake feature to be dynamically enabled on the table without requiring table recreation. If `table.datalake.format` is not explicitly set during table creation, the table will default to the format specified by the `datalake.format` configuration in the Fluss cluster. | -| table.datalake.historical-partition.enabled | Boolean | false | Whether to enable [historical partition access](../table-design/data-distribution/partitioning.md#historical-partition-access) for writes to expired partitions and, for primary-key tables, lookups of expired partition data. | +| table.datalake.historical-partition.enabled | Boolean | false | Whether to enable [historical partition access](../table-design/data-distribution/partitioning.md#historical-partition-access) for writes to expired partitions and, for primary-key tables, lookups of expired partition data. When enabled, newly created Paimon primary-key tables with automatic compaction disabled default `write.sequence-number-init-mode` to `snapshot`. Set `paimon.write.sequence-number-init-mode` to `scan` to disable this optimization. | | table.datalake.database-name | String | (None) | Specifies the database name of the datalake table. This option is currently supported only for Paimon. If not set, the Fluss database name is used. The option may be configured before the Paimon table is created and cannot be changed after creation. | | table.datalake.table-name | String | (None) | Specifies the table name of the datalake table. This option is currently supported only for Paimon. If not set, the Fluss table name is used. The option may be configured before the Paimon table is created and cannot be changed after creation. | | table.datalake.historical-partition.lookup-mode | Enum | SST | The mode used to look up historical partitions in Paimon. `SST` creates and caches local lookup files. `SCAN` scans the requested partition and bucket with primary-key filters without creating local lookup files. This option can be changed with `ALTER TABLE SET` or reset to `SST` with `ALTER TABLE RESET`. | diff --git a/website/docs/table-design/data-distribution/partitioning.md b/website/docs/table-design/data-distribution/partitioning.md index e525478eb03..ce4a84519f1 100644 --- a/website/docs/table-design/data-distribution/partitioning.md +++ b/website/docs/table-design/data-distribution/partitioning.md @@ -102,6 +102,15 @@ When enabled, the Coordinator creates and retains an internal `__historical__` s - **Reads:** Primary-key point lookups continue to resolve rows after the original Fluss partition no longer exists. Historical reads currently support only primary-key point lookups. +For primary-key tables with historical partition access enabled and +`table.datalake.auto-compaction=false` (the default), Fluss defaults +`write.sequence-number-init-mode` to `snapshot` when creating the Paimon table. This uses the +snapshot's maximum sequence number to avoid repeated file-metadata scans during write-only +writer initialization (requires Paimon 2.1 or later). Explicit +`paimon.write.sequence-number-init-mode` settings take precedence; set it to `scan` in +`CREATE TABLE` or `ALTER TABLE ... SET` to disable this optimization. +Existing Paimon tables are not changed automatically. + For primary-key tables, `table.datalake.historical-partition.lookup-mode` selects how historical partition lookups read Paimon when `table.datalake.historical-partition.enabled` is `true`. `SST`, the default, creates and caches local lookup files. `SCAN` applies primary-key filters while From 752df799a9666dbb4bb935a3f3e947aca2e5bb88 Mon Sep 17 00:00:00 2001 From: Junbo Wang Date: Mon, 19 Oct 2026 16:38:27 +0800 Subject: [PATCH 2/4] [test] Split historical Paimon tiering tests --- .../paimon/tiering/PaimonTieringTest.java | 146 ++++++++++++------ 1 file changed, 96 insertions(+), 50 deletions(-) diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java index e5005254a8c..1b482fab6ff 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java @@ -73,7 +73,6 @@ import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; -import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.MethodSource; import javax.annotation.Nullable; @@ -154,6 +153,19 @@ private static Stream snapshotExpireArgs() { Arguments.of(false, false, false)); } + private static Stream historicalWriterOptions() { + return Stream.of( + Arguments.of(Collections.emptyMap(), true), + Arguments.of( + Collections.singletonMap( + ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION.key(), "true"), + false), + Arguments.of( + Collections.singletonMap( + "paimon." + WRITE_SEQUENCE_NUMBER_INIT_MODE.key(), "scan"), + false)); + } + private static Stream partitionExpireArgs() { return Stream.of( Arguments.of(true, true), @@ -245,34 +257,15 @@ void testTieringWriteTable(boolean isPrimaryKeyTable, boolean isPartitioned) thr } } - @ParameterizedTest - @CsvSource({"false,", "true,", "false,scan"}) - void testHistoricalPrimaryKeyTiering(boolean autoCompaction, @Nullable String sequenceMode) - throws Exception { + @Test + void testHistoricalPrimaryKeyTiering() throws Exception { TablePath tablePath = TablePath.of("paimon", "test_historical_primary_key"); - Configuration properties = new Configuration(); - properties.set(ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION, autoCompaction); - if (sequenceMode != null) { - properties.setString("paimon." + WRITE_SEQUENCE_NUMBER_INIT_MODE.key(), sequenceMode); - } - TableInfo tableInfo = createHistoricalTable(tablePath, true, properties.toMap()); - long timestamp = 1_000L; - List records = - Arrays.asList( - historicalRecord(0L, timestamp, 1, "20240101", INSERT), - historicalRecord(1L, timestamp, 1, "20240101", INSERT), - historicalRecord(2L, timestamp, 1, "20240101", INSERT), - historicalRecord(3L, timestamp, 1, "20240102", INSERT)); - - PaimonWriteResult writeResult; - try (LakeWriter lakeWriter = - createLakeWriter(tablePath, 0, HISTORICAL_PARTITION_VALUE, 1L, tableInfo)) { - for (LogRecord record : records) { - lakeWriter.write(record); - } - writeResult = lakeWriter.complete(); - } - + TableInfo tableInfo = createHistoricalTable(tablePath, true, Collections.emptyMap()); + PaimonWriteResult writeResult = + writeHistoricalRecords( + tableInfo, + historicalRecord(0L, 1_000L, 1, "20240101", INSERT), + historicalRecord(1L, 1_000L, 1, "20240102", INSERT)); SimpleVersionedSerializer serializer = paimonLakeTieringFactory.getWriteResultSerializer(); writeResult = @@ -283,28 +276,33 @@ void testHistoricalPrimaryKeyTiering(boolean autoCompaction, @Nullable String se assertThat(paimonCatalog.listPartitions(toPaimon(tablePath))) .extracting(partition -> partition.spec().get("c3")) .containsExactlyInAnyOrder("20240101", "20240102"); - boolean snapshotSequence = !autoCompaction && sequenceMode == null; + } + + @ParameterizedTest + @MethodSource("historicalWriterOptions") + void testHistoricalWriterInitialization( + Map properties, boolean snapshotSequence) throws Exception { + TablePath tablePath = TablePath.of("paimon", "test_historical_writer_initialization"); + TableInfo tableInfo = createHistoricalTable(tablePath, true, properties); + PaimonWriteResult writeResult = + writeHistoricalRecords( + tableInfo, + historicalRecord(0L, 1_000L, 1, "20240101", INSERT), + historicalRecord(1L, 1_000L, 1, "20240101", INSERT), + historicalRecord(2L, 1_000L, 1, "20240101", INSERT), + historicalRecord(3L, 1_000L, 1, "20240102", INSERT)); + commitWriteResult(tablePath, tableInfo, writeResult); assertMaxSequenceNumber(tablePath, 1L, snapshotSequence ? 2L : null); + FileStoreTable table = (FileStoreTable) paimonCatalog.getTable(toPaimon(tablePath)); FileIO fileIO = spy(table.fileIO()); - FileStoreTable uncachedTable = FileStoreTableFactory.create(fileIO, table.location()); - Catalog catalog = mock(Catalog.class); - when(catalog.getTable(toPaimon(tablePath))).thenReturn(uncachedTable); - PaimonCatalogProvider provider = mock(PaimonCatalogProvider.class); - when(provider.get()).thenReturn(catalog); - WriterInitContext context = mock(WriterInitContext.class); - when(context.tableInfo()).thenReturn(tableInfo); - when(context.tableBucket()).thenReturn(new TableBucket(0, 1L, 0)); - when(context.bucketCount()).thenReturn(1); - long snapshotId = 1; - // The persisted option applies to both historical and normal partition writers. for (String partition : Arrays.asList("20240102", "20240101")) { - when(context.partition()) - .thenReturn(snapshotId == 1 ? HISTORICAL_PARTITION_VALUE : partition); + String writerPartition = snapshotId == 1 ? HISTORICAL_PARTITION_VALUE : partition; clearInvocations(fileIO); - try (PaimonLakeWriter writer = new PaimonLakeWriter(provider, context)) { - writer.write(historicalRecord(4L, timestamp + 1_000L, 1, partition, INSERT)); + try (PaimonLakeWriter writer = + createLakeWriter(tableInfo, table, writerPartition, fileIO)) { + writer.write(historicalRecord(4L, 2_000L, 1, partition, INSERT)); verify(fileIO, snapshotSequence ? never() : atLeastOnce()) .newInputStream(argThat(path -> path.getName().startsWith("manifest-"))); writeResult = writer.complete(); @@ -314,8 +312,28 @@ void testHistoricalPrimaryKeyTiering(boolean autoCompaction, @Nullable String se assertMaxSequenceNumber( tablePath, snapshotId, snapshotSequence ? snapshotId + 1 : null); } + } + + @ParameterizedTest + @MethodSource("historicalWriterOptions") + void testHistoricalTieringAfterExternalCompaction( + Map properties, boolean snapshotSequence) throws Exception { + TablePath tablePath = TablePath.of("paimon", "test_historical_external_compaction"); + TableInfo tableInfo = createHistoricalTable(tablePath, true, properties); + commitWriteResult( + tablePath, + tableInfo, + writeHistoricalRecords( + tableInfo, + historicalRecord(0L, 1_000L, 1, "20240101", INSERT), + historicalRecord(1L, 1_000L, 1, "20240102", INSERT))); + PaimonWriteResult writeResult = + writeHistoricalRecords( + tableInfo, historicalRecord(2L, 1_000L, 1, "20240101", INSERT)); + commitWriteResult(tablePath, tableInfo, writeResult); // An external compactor inherits the option without Fluss runtime overrides. + FileStoreTable table = (FileStoreTable) paimonCatalog.getTable(toPaimon(tablePath)); try (TableWriteImpl compactor = table.newWrite("external-compactor"); TableCommitImpl commit = table.newCommit("external-compactor")) { compactor.compact( @@ -324,13 +342,13 @@ void testHistoricalPrimaryKeyTiering(boolean autoCompaction, @Nullable String se } Snapshot compactSnapshot = table.snapshotManager().latestSnapshot(); assertThat(compactSnapshot.commitKind()).isEqualTo(Snapshot.CommitKind.COMPACT); - assertMaxSequenceNumber(tablePath, compactSnapshot.id(), snapshotSequence ? 4L : null); + assertMaxSequenceNumber(tablePath, compactSnapshot.id(), snapshotSequence ? 1L : null); - LogRecord updated = historicalRecord(6L, timestamp + 2_000L, 1, "20240102", INSERT); - when(context.partition()).thenReturn(HISTORICAL_PARTITION_VALUE); - clearInvocations(fileIO); - try (PaimonLakeWriter writer = new PaimonLakeWriter(provider, context)) { - writer.write(historicalRecord(5L, timestamp + 2_000L, 1, "20240101", DELETE)); + FileIO fileIO = spy(table.fileIO()); + LogRecord updated = historicalRecord(4L, 2_000L, 1, "20240102", INSERT); + try (PaimonLakeWriter writer = + createLakeWriter(tableInfo, table, HISTORICAL_PARTITION_VALUE, fileIO)) { + writer.write(historicalRecord(3L, 2_000L, 1, "20240101", DELETE)); writer.write(updated); verify(fileIO, snapshotSequence ? never() : atLeastOnce()) .newInputStream(argThat(path -> path.getName().startsWith("manifest-"))); @@ -1182,6 +1200,34 @@ private CloseableIterator getPaimonRowsThreePartition( return readBuilder.newRead().createReader(splits).toCloseableIterator(); } + private PaimonWriteResult writeHistoricalRecords(TableInfo tableInfo, LogRecord... records) + throws Exception { + try (LakeWriter writer = + createLakeWriter( + tableInfo.getTablePath(), 0, HISTORICAL_PARTITION_VALUE, 1L, tableInfo)) { + for (LogRecord record : records) { + writer.write(record); + } + return writer.complete(); + } + } + + private PaimonLakeWriter createLakeWriter( + TableInfo tableInfo, FileStoreTable table, String partition, FileIO fileIO) + throws Exception { + FileStoreTable uncachedTable = FileStoreTableFactory.create(fileIO, table.location()); + Catalog catalog = mock(Catalog.class); + when(catalog.getTable(toPaimon(tableInfo.getLakeTablePath()))).thenReturn(uncachedTable); + PaimonCatalogProvider provider = mock(PaimonCatalogProvider.class); + when(provider.get()).thenReturn(catalog); + WriterInitContext context = mock(WriterInitContext.class); + when(context.tableInfo()).thenReturn(tableInfo); + when(context.tableBucket()).thenReturn(new TableBucket(0, 1L, 0)); + when(context.bucketCount()).thenReturn(tableInfo.getNumBuckets()); + when(context.partition()).thenReturn(partition); + return new PaimonLakeWriter(provider, context); + } + private LakeWriter createLakeWriter( TablePath tablePath, int bucket, From a5ce5df0fbeab17e775a2ac1b9a505caccf444bb Mon Sep 17 00:00:00 2001 From: Junbo Wang Date: Mon, 19 Oct 2026 16:57:08 +0800 Subject: [PATCH 3/4] [test] Verify historical tiering scan counts without mocks --- .../paimon/tiering/PaimonTieringTest.java | 57 +++++++------------ 1 file changed, 21 insertions(+), 36 deletions(-) diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java index 1b482fab6ff..a00de90e562 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java @@ -45,6 +45,7 @@ import org.apache.arrow.vector.IntVector; import org.apache.arrow.vector.VarCharVector; import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.flink.testutils.logging.LoggerAuditingExtension; import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; import org.apache.paimon.catalog.Catalog; @@ -52,13 +53,12 @@ import org.apache.paimon.catalog.CatalogFactory; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.data.InternalRow; -import org.apache.paimon.fs.FileIO; import org.apache.paimon.manifest.ManifestCommittable; +import org.apache.paimon.operation.AbstractFileStoreScan; import org.apache.paimon.options.Options; import org.apache.paimon.partition.Partition; import org.apache.paimon.schema.Schema; import org.apache.paimon.table.FileStoreTable; -import org.apache.paimon.table.FileStoreTableFactory; import org.apache.paimon.table.sink.CommitMessageImpl; import org.apache.paimon.table.sink.TableCommitImpl; import org.apache.paimon.table.sink.TableWriteImpl; @@ -70,10 +70,12 @@ import org.apache.paimon.utils.SnapshotManager; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; +import org.slf4j.event.Level; import javax.annotation.Nullable; @@ -107,18 +109,14 @@ import static org.apache.paimon.CoreOptions.WRITE_SEQUENCE_NUMBER_INIT_MODE; import static org.apache.paimon.table.sink.BatchWriteBuilder.COMMIT_IDENTIFIER; import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.ArgumentMatchers.argThat; -import static org.mockito.Mockito.atLeastOnce; -import static org.mockito.Mockito.clearInvocations; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.spy; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; /** The UT for tiering to Paimon via {@link PaimonLakeTieringFactory}. */ class PaimonTieringTest { + @RegisterExtension + final LoggerAuditingExtension scanLogs = + new LoggerAuditingExtension(AbstractFileStoreScan.class, Level.INFO); + private @TempDir File tempWarehouseDir; private PaimonLakeTieringFactory paimonLakeTieringFactory; private Catalog paimonCatalog; @@ -294,17 +292,14 @@ void testHistoricalWriterInitialization( commitWriteResult(tablePath, tableInfo, writeResult); assertMaxSequenceNumber(tablePath, 1L, snapshotSequence ? 2L : null); - FileStoreTable table = (FileStoreTable) paimonCatalog.getTable(toPaimon(tablePath)); - FileIO fileIO = spy(table.fileIO()); long snapshotId = 1; for (String partition : Arrays.asList("20240102", "20240101")) { String writerPartition = snapshotId == 1 ? HISTORICAL_PARTITION_VALUE : partition; - clearInvocations(fileIO); - try (PaimonLakeWriter writer = - createLakeWriter(tableInfo, table, writerPartition, fileIO)) { + long scansBefore = scanCount(); + try (LakeWriter writer = + createLakeWriter(tablePath, 0, writerPartition, 1L, tableInfo)) { writer.write(historicalRecord(4L, 2_000L, 1, partition, INSERT)); - verify(fileIO, snapshotSequence ? never() : atLeastOnce()) - .newInputStream(argThat(path -> path.getName().startsWith("manifest-"))); + assertThat(scanCount() - scansBefore).isEqualTo(snapshotSequence ? 0 : 1); writeResult = writer.complete(); } commitWriteResult(tablePath, tableInfo, writeResult); @@ -344,14 +339,14 @@ void testHistoricalTieringAfterExternalCompaction( assertThat(compactSnapshot.commitKind()).isEqualTo(Snapshot.CommitKind.COMPACT); assertMaxSequenceNumber(tablePath, compactSnapshot.id(), snapshotSequence ? 1L : null); - FileIO fileIO = spy(table.fileIO()); + long scansBefore = scanCount(); LogRecord updated = historicalRecord(4L, 2_000L, 1, "20240102", INSERT); - try (PaimonLakeWriter writer = - createLakeWriter(tableInfo, table, HISTORICAL_PARTITION_VALUE, fileIO)) { + try (LakeWriter writer = + createLakeWriter(tablePath, 0, HISTORICAL_PARTITION_VALUE, 1L, tableInfo)) { writer.write(historicalRecord(3L, 2_000L, 1, "20240101", DELETE)); + assertThat(scanCount() - scansBefore).isEqualTo(snapshotSequence ? 0 : 1); writer.write(updated); - verify(fileIO, snapshotSequence ? never() : atLeastOnce()) - .newInputStream(argThat(path -> path.getName().startsWith("manifest-"))); + assertThat(scanCount() - scansBefore).isEqualTo(snapshotSequence ? 0 : 2); writeResult = writer.complete(); } commitWriteResult(tablePath, tableInfo, writeResult); @@ -1212,20 +1207,10 @@ private PaimonWriteResult writeHistoricalRecords(TableInfo tableInfo, LogRecord. } } - private PaimonLakeWriter createLakeWriter( - TableInfo tableInfo, FileStoreTable table, String partition, FileIO fileIO) - throws Exception { - FileStoreTable uncachedTable = FileStoreTableFactory.create(fileIO, table.location()); - Catalog catalog = mock(Catalog.class); - when(catalog.getTable(toPaimon(tableInfo.getLakeTablePath()))).thenReturn(uncachedTable); - PaimonCatalogProvider provider = mock(PaimonCatalogProvider.class); - when(provider.get()).thenReturn(catalog); - WriterInitContext context = mock(WriterInitContext.class); - when(context.tableInfo()).thenReturn(tableInfo); - when(context.tableBucket()).thenReturn(new TableBucket(0, 1L, 0)); - when(context.bucketCount()).thenReturn(tableInfo.getNumBuckets()); - when(context.partition()).thenReturn(partition); - return new PaimonLakeWriter(provider, context); + private long scanCount() { + return scanLogs.getMessages().stream() + .filter(message -> message.startsWith("File store scan plan completed")) + .count(); } private LakeWriter createLakeWriter( From 1b16131af73fe8148db809724cacbef5a45fd930 Mon Sep 17 00:00:00 2001 From: Junbo Wang Date: Mon, 19 Oct 2026 18:22:29 +0800 Subject: [PATCH 4/4] [docs] Clarify Paimon version for historical tiering optimization --- .../org/apache/fluss/lake/paimon/utils/PaimonConversions.java | 2 +- website/docs/table-design/data-distribution/partitioning.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java index e101585cc3b..949e23a6c0b 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java @@ -405,7 +405,7 @@ public static Schema toPaimonSchema(TableDescriptor tableDescriptor) { && !tableConfig.isDataLakeAutoCompaction()) { // Use the snapshot's maximum sequence number to avoid repeated manifest scans // when historical splits write to multiple partitions and buckets. - // Use the literal key for compatibility with Paimon 1.4. + // Use a literal key because Paimon 1.4 does not define this option. options.set("write.sequence-number-init-mode", "snapshot"); } } diff --git a/website/docs/table-design/data-distribution/partitioning.md b/website/docs/table-design/data-distribution/partitioning.md index ce4a84519f1..5f2d1f260d4 100644 --- a/website/docs/table-design/data-distribution/partitioning.md +++ b/website/docs/table-design/data-distribution/partitioning.md @@ -106,7 +106,7 @@ For primary-key tables with historical partition access enabled and `table.datalake.auto-compaction=false` (the default), Fluss defaults `write.sequence-number-init-mode` to `snapshot` when creating the Paimon table. This uses the snapshot's maximum sequence number to avoid repeated file-metadata scans during write-only -writer initialization (requires Paimon 2.1 or later). Explicit +writer initialization (requires Paimon 2.0 or later). Explicit `paimon.write.sequence-number-init-mode` settings take precedence; set it to `scan` in `CREATE TABLE` or `ALTER TABLE ... SET` to disable this optimization. Existing Paimon tables are not changed automatically.