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..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 @@ -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 a literal key because Paimon 1.4 does not define this option. + 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..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; @@ -53,11 +54,14 @@ import org.apache.paimon.catalog.Identifier; import org.apache.paimon.data.InternalRow; 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.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; @@ -66,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; @@ -88,6 +94,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,12 +106,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; /** 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; @@ -139,6 +151,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), @@ -233,43 +258,111 @@ void testTieringWriteTable(boolean isPrimaryKeyTable, boolean isPartitioned) thr @Test void testHistoricalPrimaryKeyTiering() throws Exception { TablePath tablePath = TablePath.of("paimon", "test_historical_primary_key"); - TableInfo tableInfo = createHistoricalTable(tablePath, true); - long timestamp = 1_000L; - List records = - Arrays.asList( - historicalRecord(0L, timestamp, 1, "20240101", INSERT), - historicalRecord(1L, 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 = 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"); } + @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); + + long snapshotId = 1; + for (String partition : Arrays.asList("20240102", "20240101")) { + String writerPartition = snapshotId == 1 ? HISTORICAL_PARTITION_VALUE : partition; + long scansBefore = scanCount(); + try (LakeWriter writer = + createLakeWriter(tablePath, 0, writerPartition, 1L, tableInfo)) { + writer.write(historicalRecord(4L, 2_000L, 1, partition, INSERT)); + assertThat(scanCount() - scansBefore).isEqualTo(snapshotSequence ? 0 : 1); + writeResult = writer.complete(); + } + commitWriteResult(tablePath, tableInfo, writeResult); + snapshotId++; + 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( + ((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 ? 1L : null); + + long scansBefore = scanCount(); + LogRecord updated = historicalRecord(4L, 2_000L, 1, "20240102", INSERT); + 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); + assertThat(scanCount() - scansBefore).isEqualTo(snapshotSequence ? 0 : 2); + 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 = @@ -1102,6 +1195,24 @@ 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 long scanCount() { + return scanLogs.getMessages().stream() + .filter(message -> message.startsWith("File store scan plan completed")) + .count(); + } + private LakeWriter createLakeWriter( TablePath tablePath, int bucket, @@ -1181,6 +1292,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 +1317,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 +1375,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 +1392,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..5f2d1f260d4 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.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. + 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