From 6d3641b4abe756b6f92804775c884caad97a75dd Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Mon, 27 Jul 2026 10:26:11 +0800 Subject: [PATCH 1/2] feat: support table.log.local-ttl --- .../fluss/client/admin/FlussAdminITCase.java | 41 ++++++++++++ .../apache/fluss/config/ConfigOptions.java | 10 +++ .../org/apache/fluss/config/TableConfig.java | 10 +++ .../apache/fluss/config/TableConfigTest.java | 14 +++++ .../apache/fluss/server/log/LogManager.java | 10 +-- .../apache/fluss/server/log/LogTablet.java | 31 +++++----- .../server/log/remote/LogTieringTask.java | 7 ++- .../server/log/remote/RemoteLogManager.java | 33 +++++++++- .../server/log/remote/RemoteLogTablet.java | 34 +++++++++- .../apache/fluss/server/replica/Replica.java | 2 +- .../fluss/server/replica/ReplicaManager.java | 4 ++ .../utils/TableDescriptorValidation.java | 21 +++++++ .../zk/data/RemoteLogManifestHandle.java | 8 +++ .../fluss/server/log/LocalSegmentTTLTest.java | 62 +++++++++++++++++++ .../server/log/remote/RemoteLogTTLTest.java | 17 +++++ .../log/remote/RemoteLogTabletTest.java | 2 + .../log/remote/TieredLocalSegmentTTLTest.java | 56 ++++++++++++++--- website/docs/engine-flink/options.md | 1 + .../tiered-storage/remote-storage.md | 5 +- .../table-design/data-distribution/ttl.md | 2 + 20 files changed, 336 insertions(+), 34 deletions(-) create mode 100644 fluss-server/src/test/java/org/apache/fluss/server/log/LocalSegmentTTLTest.java diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java index 150b84e5679..74f02b7282d 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java @@ -865,6 +865,29 @@ void testCreateTableWithInvalidProperty() { .isInstanceOf(InvalidConfigException.class) .hasMessage("'table.log.tiered.local-segments' must be greater than 0."); + TableDescriptor invalidLocalTtl = + TableDescriptor.builder() + .schema(DEFAULT_SCHEMA) + .comment("test table") + .property(ConfigOptions.TABLE_LOG_TTL.key(), "1h") + .property(ConfigOptions.TABLE_LOG_LOCAL_TTL.key(), "2h") + .build(); + assertThatThrownBy(() -> admin.createTable(tablePath, invalidLocalTtl, false).get()) + .cause() + .isInstanceOf(InvalidConfigException.class) + .hasMessage("'table.log.local-ttl' must be less than or equal to 'table.log.ttl'."); + + TableDescriptor nonPositiveLocalTtl = + TableDescriptor.builder() + .schema(DEFAULT_SCHEMA) + .comment("test table") + .property(ConfigOptions.TABLE_LOG_LOCAL_TTL.key(), "0ms") + .build(); + assertThatThrownBy(() -> admin.createTable(tablePath, nonPositiveLocalTtl, false).get()) + .cause() + .isInstanceOf(InvalidConfigException.class) + .hasMessage("'table.log.local-ttl' must be greater than 0."); + TableDescriptor t4 = TableDescriptor.builder() .schema(DEFAULT_SCHEMA) // no pk @@ -928,6 +951,24 @@ void testCreateTableWithInvalidProperty() { "Currently, Primary Key Table supports ARROW or COMPACTED log format when kv format is COMPACTED."); } + @Test + void testCreateTableWithLocalTtlAndInfiniteRemoteTtl() throws Exception { + TablePath tablePath = + TablePath.of( + DEFAULT_TABLE_PATH.getDatabaseName(), + "test_local_ttl_with_infinite_remote_ttl"); + TableDescriptor tableDescriptor = + TableDescriptor.builder() + .schema(DEFAULT_SCHEMA) + .property(ConfigOptions.TABLE_LOG_TTL.key(), "0ms") + .property(ConfigOptions.TABLE_LOG_LOCAL_TTL.key(), "1h") + .build(); + + admin.createTable(tablePath, tableDescriptor, false).get(); + assertThat(admin.tableExists(tablePath).get()).isTrue(); + admin.dropTable(tablePath, false).get(); + } + @Test void testCreateTableWithInvalidReplicationFactor() throws Exception { TablePath tablePath = TablePath.of(DEFAULT_TABLE_PATH.getDatabaseName(), "t1"); diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java index 233902cb391..1b3aa57f453 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java @@ -1792,6 +1792,16 @@ public class ConfigOptions { "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."); + public static final ConfigOption TABLE_LOG_LOCAL_TTL = + key("table.log.local-ttl") + .durationType() + .noDefaultValue() + .withDescription( + "The time to retain local log segments after they have been copied to remote storage. " + + "If not configured, the value of `table.log.ttl` is used. When configured, " + + "the value must be greater than 0 and, if `table.log.ttl` is positive, less " + + "than or equal to `table.log.ttl`."); + public static final ConfigOption TABLE_DATALAKE_ENABLED = key("table.datalake.enabled") .booleanType() diff --git a/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java b/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java index fbf8c772644..6a6ccd50c43 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java @@ -95,6 +95,16 @@ public int getTieredLogLocalSegments() { return config.get(ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS); } + /** + * Gets the TTL of local segments for tiered log. The table log TTL is used when no local TTL is + * configured. + */ + public long getLocalLogTTLMs() { + return config.getOptional(ConfigOptions.TABLE_LOG_LOCAL_TTL) + .orElseGet(() -> config.get(ConfigOptions.TABLE_LOG_TTL)) + .toMillis(); + } + /** Whether the data lake is enabled. */ public boolean isDataLakeEnabled() { return config.get(ConfigOptions.TABLE_DATALAKE_ENABLED); diff --git a/fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java b/fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java index 5d18fcd1c97..6f214f559e3 100644 --- a/fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java @@ -21,6 +21,8 @@ import org.junit.jupiter.api.Test; +import java.time.Duration; + import static org.assertj.core.api.Assertions.assertThat; /** Tests for {@link TableConfig}. */ @@ -44,4 +46,16 @@ void testDeleteBehavior() { TableConfig tableConfig3 = new TableConfig(conf); assertThat(tableConfig3.getDeleteBehavior()).hasValue(DeleteBehavior.IGNORE); } + + @Test + void testTieredLogLocalTtl() { + Configuration conf = new Configuration(); + conf.set(ConfigOptions.TABLE_LOG_TTL, Duration.ofDays(3)); + TableConfig tableConfig = new TableConfig(conf); + + assertThat(tableConfig.getLocalLogTTLMs()).isEqualTo(Duration.ofDays(3).toMillis()); + + conf.set(ConfigOptions.TABLE_LOG_LOCAL_TTL, Duration.ofHours(6)); + assertThat(tableConfig.getLocalLogTTLMs()).isEqualTo(Duration.ofHours(6).toMillis()); + } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java b/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java index f18411cbfa6..8b94360f0c7 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java @@ -292,7 +292,7 @@ private void waitForLoadLogsInDir(LogRecoveryTask recoveryTask) throws Throwable * @param tableBucket the table bucket * @param logFormat the log format * @param tieredLogLocalSegments the number of segments to retain in local for tiered log - * @param logTtlMs the log TTL in milliseconds from table configuration + * @param localLogTtlMs the local log TTL in milliseconds from table configuration * @param isChangelog whether the log is a changelog of primary key table */ public LogTablet getOrCreateLog( @@ -301,7 +301,7 @@ public LogTablet getOrCreateLog( TableBucket tableBucket, LogFormat logFormat, int tieredLogLocalSegments, - long logTtlMs, + long localLogTtlMs, boolean isChangelog) throws Exception { return inLock( @@ -324,7 +324,7 @@ public LogTablet getOrCreateLog( scheduler, logFormat, tieredLogLocalSegments, - logTtlMs, + localLogTtlMs, isChangelog, clock, true); @@ -355,7 +355,7 @@ public LogTablet getOrCreateLog( tableBucket, logFormat, tieredLogLocalSegments, - tableConfig.getLogTTLMs(), + tableConfig.getLocalLogTTLMs(), isChangelog); } @@ -469,7 +469,7 @@ private LogTablet loadLog( scheduler, tableInfo.getTableConfig().getLogFormat(), tableInfo.getTableConfig().getTieredLogLocalSegments(), - tableInfo.getTableConfig().getLogTTLMs(), + tableInfo.getTableConfig().getLocalLogTTLMs(), tableInfo.hasPrimaryKey(), clock, isCleanShutdown); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java index 875bef312a5..b5b9315efd0 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java @@ -108,7 +108,8 @@ public final class LogTablet { private volatile int tieredLogLocalSegments; private final Clock clock; private final boolean isChangeLog; - private final long logTtlMs; + private final long localLogTtlMs; + private final boolean remoteLogEnabled; @GuardedBy("lock") private volatile LogOffsetMetadata highWatermarkMetadata; @@ -147,7 +148,7 @@ private LogTablet( WriterStateManager writerStateManager, LogFormat logFormat, int tieredLogLocalSegments, - long logTtlMs, + long localLogTtlMs, boolean isChangelog, Clock clock) { this.dataDir = dataDir; @@ -159,7 +160,9 @@ private LogTablet( (int) conf.get(ConfigOptions.WRITER_ID_EXPIRATION_CHECK_INTERVAL).toMillis(); this.writerStateManager = writerStateManager; this.highWatermarkMetadata = new LogOffsetMetadata(0L); - this.logTtlMs = logTtlMs; + this.localLogTtlMs = localLogTtlMs; + this.remoteLogEnabled = + conf.get(ConfigOptions.REMOTE_LOG_TASK_INTERVAL_DURATION).toMillis() > 0L; this.scheduler = scheduler; // scheduler the writer expiration interval check. @@ -347,7 +350,7 @@ public static LogTablet create( Scheduler scheduler, LogFormat logFormat, int tieredLogLocalSegments, - long logTtlMs, + long localLogTtlMs, boolean isChangelog, Clock clock, boolean isCleanShutdown) @@ -396,7 +399,7 @@ public static LogTablet create( writerStateManager, logFormat, tieredLogLocalSegments, - logTtlMs, + localLogTtlMs, isChangelog, clock); } @@ -427,7 +430,7 @@ public static LogTablet create( scheduler, logFormat, tieredLogLocalSegments, - tableConfig.getLogTTLMs(), + tableConfig.getLocalLogTTLMs(), isChangelog, clock, isCleanShutdown); @@ -742,15 +745,15 @@ public void deleteSegmentsAlreadyExistsInRemote() { this::deletableRemoteSegments); } - /** Deletes inactive local segments that have expired according to the table log TTL. */ + /** + * Deletes inactive local segments that have expired according to the local log TTL. + * + *

When remote log tiering is enabled, cleanup is bounded by the highest offset copied to + * remote storage. Otherwise, cleanup is bounded by the high watermark. + */ public void deleteExpiredSegments() { - // A missing remote end offset can mean either that no segment has been uploaded or that - // all remote segments have expired. In both cases, table.log.ttl remains authoritative for - // local retention, while the high watermark and minRetainOffset still protect data that - // cannot be deleted yet. - long cleanupToOffset = remoteLogEndOffset == -1L ? getHighWatermark() : remoteLogEndOffset; deleteSegments( - cleanupToOffset, + remoteLogEnabled ? remoteLogEndOffset : getHighWatermark(), SegmentDeletionReason.LOG_RETENTION, this::deletableExpiredSegments); } @@ -1362,7 +1365,7 @@ private List deletableExpiredSegments(long endOffset) throws IOExcep for (int i = 0; i < logSegments.size() - 1; i++) { if (logSegments.get(i + 1).getBaseOffset() > endOffset - || !isSegmentExpired(now, logSegments.get(i), logTtlMs)) { + || !isSegmentExpired(now, logSegments.get(i), localLogTtlMs)) { break; } deletableSegments.add(logSegments.get(i)); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java index c84aff7188c..91fbb0c550d 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java @@ -357,7 +357,10 @@ public boolean tryToCommitRemoteLogManifest( // 2. sending the CommitRemoteLogManifestRequest to coordinator server // to try to commit this snapshot. long newRemoteLogStartOffset = newRemoteLogManifest.getRemoteLogStartOffset(); - long newRemoteLogEndOffset = newRemoteLogManifest.getRemoteLogEndOffset(); + long newRemoteLogEndOffset = + Math.max( + newRemoteLogManifest.getRemoteLogEndOffset(), + remoteLogTablet.getHighestRemoteLogEndOffset().orElse(-1L)); long newRemoteLogSize = newRemoteLogManifest.getRemoteLogSize(); int retrySendCommitTimes = 1; while (retrySendCommitTimes <= 10) { @@ -436,7 +439,7 @@ private void maybeUpdateCopiedOffset(LogTablet logTablet) { } private long findRemoteLogEndOffset(LogTablet logTablet) { - OptionalLong remoteLogEndOffsetOpt = remoteLog.getRemoteLogEndOffset(); + OptionalLong remoteLogEndOffsetOpt = remoteLog.getHighestRemoteLogEndOffset(); long newRemoteLogEndOffset; if (remoteLogEndOffsetOpt.isPresent()) { long remoteLogEndOffset = remoteLogEndOffsetOpt.getAsLong(); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java index 4f27c2872a5..c3092673f0a 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java @@ -157,14 +157,17 @@ public void registerReplica(Replica replica) throws Exception { Optional remoteLogManifestHandleOpt = zkClient.getRemoteLogManifestHandle(tableBucket); if (remoteLogManifestHandleOpt.isPresent()) { + RemoteLogManifestHandle remoteLogManifestHandle = remoteLogManifestHandleOpt.get(); // If there is remote log manifest handle in remote, we will download // the manifest snapshot from remote storage and write to cache. RemoteLogManifest manifest = remoteLogStorage.readRemoteLogManifestSnapshot( - remoteLogManifestHandleOpt.get().getRemoteLogManifestPath()); + remoteLogManifestHandle.getRemoteLogManifestPath()); remoteLog.loadRemoteLogManifest(manifest); + remoteLog.updateHighestRemoteLogEndOffset( + remoteLogManifestHandle.getRemoteLogEndOffset()); + log.updateRemoteLogEndOffset(remoteLogManifestHandle.getRemoteLogEndOffset()); } - remoteLog.getRemoteLogEndOffset().ifPresent(log::updateRemoteLogEndOffset); log.updateRemoteLogStartOffset(remoteLog.getRemoteLogStartOffset()); log.updateRemoteLogSize(remoteLog.getRemoteSizeInBytes()); // leader needs to register the remote log metrics @@ -172,6 +175,32 @@ public void registerReplica(Replica replica) throws Exception { remoteLogs.put(tableBucket, remoteLog); } + /** + * Best-effort restores the highest copied remote log end offset from persistent metadata. + * + *

A failure only delays local log cleanup until a later remote offset notification and must + * not prevent the replica from becoming a follower. + */ + public void restoreRemoteLogEndOffset(Replica replica) { + if (remoteDisabled()) { + return; + } + try { + Optional remoteLogManifestHandle = + zkClient.getRemoteLogManifestHandle(replica.getTableBucket()); + if (remoteLogManifestHandle.isPresent()) { + replica.getLogTablet() + .updateRemoteLogEndOffset( + remoteLogManifestHandle.get().getRemoteLogEndOffset()); + } + } catch (Exception e) { + LOG.warn( + "Failed to restore remote log end offset for follower replica {}.", + replica.getTableBucket(), + e); + } + } + /** Start the log tiering task for the given replica. */ public void startLogTiering(Replica replica) { TableBucket tableBucket = replica.getTableBucket(); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogTablet.java index 9f0ae6e949c..8f8e3a53408 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogTablet.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogTablet.java @@ -95,10 +95,14 @@ public class RemoteLogTablet { */ private volatile long remoteLogStartOffset; + /** The end offset of the segments in the current remote log manifest. */ + private volatile long remoteLogEndOffset; + /** - * It represents the remote log end offset of the segments that have copied to remote storage. + * The highest end offset ever committed to remote storage. Unlike {@link #remoteLogEndOffset}, + * this value is not reset when all remote segments expire. */ - private volatile long remoteLogEndOffset; + private volatile long highestRemoteLogEndOffset = INIT_REMOTE_LOG_END_OFFSET; private volatile boolean closed = false; @@ -267,6 +271,31 @@ public OptionalLong getRemoteLogEndOffset() { : OptionalLong.of(remoteLogEndOffset); } + /** + * Returns the highest end offset ever committed to remote storage, or empty if no segment has + * been committed. + */ + public OptionalLong getHighestRemoteLogEndOffset() { + return highestRemoteLogEndOffset == INIT_REMOTE_LOG_END_OFFSET + ? OptionalLong.empty() + : OptionalLong.of(highestRemoteLogEndOffset); + } + + /** + * Restores the highest end offset from the persisted remote log manifest handle. + * + *

The handle keeps this offset even when its manifest no longer contains segments. + */ + public void updateHighestRemoteLogEndOffset(long remoteLogEndOffset) { + inWriteLock( + lock, + () -> { + if (remoteLogEndOffset > highestRemoteLogEndOffset) { + highestRemoteLogEndOffset = remoteLogEndOffset; + } + }); + } + /** * Gets the snapshot of current remote log segment manifest. The snapshot including the exists * remoteLogSegment already committed. @@ -313,6 +342,7 @@ public void addAndDeleteLogSegments( if (remoteLogSegment.remoteLogEndOffset() > remoteLogEndOffset) { remoteLogEndOffset = remoteLogSegment.remoteLogEndOffset(); } + updateHighestRemoteLogEndOffset(remoteLogSegment.remoteLogEndOffset()); newSizeInBytes += remoteLogSegment.segmentSizeInBytes(); } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java index f3d1ff76af1..d5563a87bb0 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java @@ -2162,7 +2162,7 @@ private LogTablet createLog( tableBucket, tableConfig.getLogFormat(), tableConfig.getTieredLogLocalSegments(), - tableConfig.getLogTTLMs(), + tableConfig.getLocalLogTTLMs(), isKvTable()); // update high watermark. Optional watermarkOpt = lazyHighWatermarkCheckpoint.fetch(tableBucket); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java index 88d5840e5a7..31acc582704 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java @@ -1216,6 +1216,10 @@ private void makeFollowers( if (replica.makeFollower(data)) { replicasBecomeFollower.add(replica); scannerManager.closeScannersForBucket(tb); + // A follower may not receive another remote offset notification for an idle + // bucket, so best-effort restore the persisted highest copied offset once when + // the replica transitions to follower. + remoteLogManager.restoreRemoteLogEndOffset(replica); } // stop the remote log tiering tasks for followers remoteLogManager.stopLogTiering(replica); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java index 8a6e9fa2c9e..a538e1497fc 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java @@ -46,6 +46,7 @@ import javax.annotation.Nullable; +import java.time.Duration; import java.util.Arrays; import java.util.Collections; import java.util.EnumSet; @@ -457,6 +458,26 @@ private static void checkTieredLog(Configuration tableConf) { "'%s' must be greater than 0.", ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS.key())); } + + Optional localTtl = tableConf.getOptional(ConfigOptions.TABLE_LOG_LOCAL_TTL); + if (!localTtl.isPresent()) { + return; + } + if (localTtl.get().isZero() || localTtl.get().isNegative()) { + throw new InvalidConfigException( + String.format( + "'%s' must be greater than 0.", + ConfigOptions.TABLE_LOG_LOCAL_TTL.key())); + } + + Duration logTtl = tableConf.get(ConfigOptions.TABLE_LOG_TTL); + if (!logTtl.isZero() && !logTtl.isNegative() && localTtl.get().compareTo(logTtl) > 0) { + throw new InvalidConfigException( + String.format( + "'%s' must be less than or equal to '%s'.", + ConfigOptions.TABLE_LOG_LOCAL_TTL.key(), + ConfigOptions.TABLE_LOG_TTL.key())); + } } private static void checkPartition( diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/RemoteLogManifestHandle.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/RemoteLogManifestHandle.java index a3a6e2cee4d..fa2b2c34a02 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/RemoteLogManifestHandle.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/RemoteLogManifestHandle.java @@ -26,6 +26,8 @@ */ public class RemoteLogManifestHandle { private final FsPath remoteLogManifestPath; + // The highest end offset ever copied to remote storage. It is monotonic and may be greater + // than the end offset in the current manifest after remote segments expire. private final long remoteLogEndOffset; public RemoteLogManifestHandle(FsPath remoteLogManifestPath, long remoteLogEndOffset) { @@ -41,6 +43,12 @@ public FsPath getRemoteLogManifestPath() { return remoteLogManifestPath; } + /** + * Returns the highest end offset ever copied to remote storage. + * + *

The value is monotonic and may be greater than the end offset represented by the current + * remote manifest after expired segments are removed. + */ public long getRemoteLogEndOffset() { return remoteLogEndOffset; } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/LocalSegmentTTLTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/LocalSegmentTTLTest.java new file mode 100644 index 00000000000..9b52a24c334 --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/LocalSegmentTTLTest.java @@ -0,0 +1,62 @@ +/* + * 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.fluss.server.log; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.server.replica.ReplicaTestBase; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.Collections; + +import static org.apache.fluss.record.TestData.DATA1_SCHEMA; +import static org.apache.fluss.record.TestData.DATA1_TABLE_ID; +import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH; +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests local log TTL cleanup when remote log tiering is disabled. */ +final class LocalSegmentTTLTest extends ReplicaTestBase { + + @BeforeEach + public void setup() throws Exception { + super.setup(); + registerTableInZkClient( + DATA1_TABLE_PATH, + DATA1_SCHEMA, + DATA1_TABLE_ID, + Collections.emptyList(), + Collections.singletonMap(ConfigOptions.TABLE_LOG_TTL.key(), "1h")); + } + + @Test + void testExpiredSegmentsDeletedUsingHighWatermarkWhenRemoteLogDisabled() throws Exception { + TableBucket tableBucket = new TableBucket(DATA1_TABLE_ID, 0); + makeLogTableAsLeader(tableBucket, false); + LogTablet logTablet = replicaManager.getReplicaOrException(tableBucket).getLogTablet(); + + addMultiSegmentsToLogTablet(logTablet, 5); + manualClock.advanceTime(Duration.ofMinutes(90)); + logManager.cleanupExpiredLocalLogSegments(); + + assertThat(logTablet.getSegments()).hasSize(1); + assertThat(logTablet.localLogStartOffset()).isEqualTo(40L); + } +} diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTTLTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTTLTest.java index 6206a2310b0..c9530090fb8 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTTLTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTTLTest.java @@ -25,6 +25,7 @@ import org.apache.fluss.server.log.FetchParams; import org.apache.fluss.server.log.LogTablet; import org.apache.fluss.server.log.remote.RemoteLogIndexCache.Entry; +import org.apache.fluss.server.replica.Replica; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.params.ParameterizedTest; @@ -138,6 +139,22 @@ void testRemoteLogTTL(boolean partitionTable) throws Exception { remoteLogTaskScheduler.triggerPeriodicScheduledTasks(); assertThat(remoteLog.allRemoteLogSegments()).isEmpty(); assertThat(remoteLog.getRemoteLogStartOffset()).isEqualTo(Long.MAX_VALUE); + assertThat(remoteLog.getRemoteLogEndOffset()).isEmpty(); + assertThat(remoteLog.getHighestRemoteLogEndOffset()).hasValue(40L); + assertThat( + zkClient.getRemoteLogManifestHandle(tb) + .orElseThrow(AssertionError::new) + .getRemoteLogEndOffset()) + .isEqualTo(40L); + + // Reload the empty manifest and verify the persisted highest copied offset still + // distinguishes it from a bucket that has never copied a segment. + Replica replica = replicaManager.getReplicaOrException(tb); + remoteLogManager.stopLogTiering(replica); + remoteLogManager.registerReplica(replica); + RemoteLogTablet restoredRemoteLog = remoteLogManager.remoteLogTablet(tb); + assertThat(restoredRemoteLog.getRemoteLogEndOffset()).isEmpty(); + assertThat(restoredRemoteLog.getHighestRemoteLogEndOffset()).hasValue(40L); // Fetch records from remote. // mock to update remote log end offset and remote log start offset as diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletTest.java index 0d571e1ea55..59a6af0e012 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletTest.java @@ -64,6 +64,7 @@ void testDelete(boolean partitionTable) throws Exception { remoteLogTablet.addAndDeleteLogSegments(remoteLogSegmentList, Collections.emptyList()); assertThat(remoteLogTablet.getIdToRemoteLogSegmentMap()).hasSize(5); assertThat(remoteLogTablet.getRemoteLogStartOffset()).isEqualTo(0); + assertThat(remoteLogTablet.getHighestRemoteLogEndOffset()).hasValue(50L); // try to delete two segments. RemoteLogSegment firstSegment = remoteLogSegmentList.get(0); @@ -79,6 +80,7 @@ void testDelete(boolean partitionTable) throws Exception { assertThat(remoteLogTablet.getIdToRemoteLogSegmentMap()).isEmpty(); assertThat(remoteLogTablet.getRemoteLogStartOffset()).isEqualTo(Long.MAX_VALUE); assertThat(remoteLogTablet.getRemoteLogEndOffset()).isEqualTo(OptionalLong.empty()); + assertThat(remoteLogTablet.getHighestRemoteLogEndOffset()).hasValue(50L); } @ParameterizedTest diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/TieredLocalSegmentTTLTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/TieredLocalSegmentTTLTest.java index 1ea904ef9a4..32f7c079c16 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/TieredLocalSegmentTTLTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/TieredLocalSegmentTTLTest.java @@ -18,15 +18,24 @@ package org.apache.fluss.server.log.remote; import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.fs.FsPath; +import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.server.entity.NotifyLeaderAndIsrData; import org.apache.fluss.server.log.LogTablet; +import org.apache.fluss.server.zk.data.LeaderAndIsr; +import org.apache.fluss.server.zk.data.RemoteLogManifestHandle; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; import java.time.Duration; +import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; +import java.util.Map; import static org.apache.fluss.record.TestData.DATA1_SCHEMA; import static org.apache.fluss.record.TestData.DATA1_TABLE_ID; @@ -39,12 +48,15 @@ final class TieredLocalSegmentTTLTest extends RemoteLogTestBase { @BeforeEach public void setup() throws Exception { super.setup(); + Map properties = new HashMap<>(); + properties.put(ConfigOptions.TABLE_LOG_TTL.key(), "2h"); + properties.put(ConfigOptions.TABLE_LOG_LOCAL_TTL.key(), "1h"); registerTableInZkClient( DATA1_TABLE_PATH, DATA1_SCHEMA, DATA1_TABLE_ID, Collections.emptyList(), - Collections.singletonMap(ConfigOptions.TABLE_LOG_TTL.key(), "1h")); + properties); } @ParameterizedTest @@ -67,7 +79,7 @@ void testInactiveTieredLocalSegmentRemovedAfterTtl(boolean partitionTable) throw assertThat(logTablet.localLogStartOffset()).isEqualTo(30L); assertThat(remoteLog.allRemoteLogSegments()).hasSize(4); - manualClock.advanceTime(Duration.ofHours(2)); + manualClock.advanceTime(Duration.ofMinutes(90)); // Run local retention without updating the remote manifest or uploading another segment. logManager.cleanupExpiredLocalLogSegments(); @@ -80,7 +92,7 @@ void testInactiveTieredLocalSegmentRemovedAfterTtl(boolean partitionTable) throw @ParameterizedTest @ValueSource(booleans = {true, false}) - void testExpiredLocalSegmentsRemovedWithoutRemoteLogEndOffset(boolean partitionTable) + void testExpiredLocalSegmentsRetainedWithoutRemoteLogEndOffset(boolean partitionTable) throws Exception { TableBucket tb = partitionTable @@ -92,11 +104,11 @@ void testExpiredLocalSegmentsRemovedWithoutRemoteLogEndOffset(boolean partitionT addMultiSegmentsToLogTablet(logTablet, 5); assertThat(remoteLogManager.remoteLogTablet(tb).getRemoteLogEndOffset()).isEmpty(); - manualClock.advanceTime(Duration.ofHours(2)); + manualClock.advanceTime(Duration.ofMinutes(90)); logManager.cleanupExpiredLocalLogSegments(); - assertThat(logTablet.getSegments()).hasSize(1); - assertThat(logTablet.localLogStartOffset()).isEqualTo(40L); + assertThat(logTablet.getSegments()).hasSize(5); + assertThat(logTablet.localLogStartOffset()).isZero(); assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(40L); } @@ -114,11 +126,41 @@ void testTtlCleanupBoundedByRemoteLogEndOffset(boolean partitionTable) throws Ex logTablet.updateTieredLogLocalSegments(5); logTablet.updateRemoteLogEndOffset(20L); - manualClock.advanceTime(Duration.ofHours(2)); + manualClock.advanceTime(Duration.ofMinutes(90)); logManager.cleanupExpiredLocalLogSegments(); assertThat(logTablet.getSegments()).hasSize(3); assertThat(logTablet.localLogStartOffset()).isEqualTo(20L); assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(40L); } + + @Test + void testFollowerRestoresHighestCopiedOffset() throws Exception { + TableBucket tableBucket = new TableBucket(DATA1_TABLE_ID, 0); + zkClient.upsertRemoteLogManifestHandle( + tableBucket, + new RemoteLogManifestHandle(new FsPath("test:///remote-log-manifest"), 40L)); + makeLeaderAndFollower( + Collections.singletonList( + new NotifyLeaderAndIsrData( + PhysicalTablePath.of(DATA1_TABLE_PATH), + tableBucket, + Arrays.asList(TABLET_SERVER_ID, 2), + new LeaderAndIsr( + LeaderAndIsr.NO_LEADER, + 0, + Collections.singletonList(TABLET_SERVER_ID), + Collections.emptyList(), + 0, + 0)))); + LogTablet logTablet = replicaManager.getReplicaOrException(tableBucket).getLogTablet(); + + addMultiSegmentsToLogTablet(logTablet, 5); + logTablet.updateHighWatermark(logTablet.localLogEndOffset()); + manualClock.advanceTime(Duration.ofMinutes(90)); + logManager.cleanupExpiredLocalLogSegments(); + + assertThat(logTablet.getSegments()).hasSize(1); + assertThat(logTablet.localLogStartOffset()).isEqualTo(40L); + } } diff --git a/website/docs/engine-flink/options.md b/website/docs/engine-flink/options.md index 4b131252d25..8b69df43f7a 100644 --- a/website/docs/engine-flink/options.md +++ b/website/docs/engine-flink/options.md @@ -67,6 +67,7 @@ See more details about [ALTER TABLE ... SET](engine-flink/ddl.md#set-properties) | bucket.num | int | The bucket number of Fluss cluster. | The number of buckets of a Fluss table. | | bucket.key | String | (None) | Specific the distribution policy of the Fluss table. Data will be distributed to each bucket according to the hash value of bucket-key (It must be a subset of the primary keys excluding partition keys of the primary key table). If you specify multiple fields, delimiter is `,`. If the table has a primary key and a bucket key is not specified, the bucket key will be used as primary key(excluding the partition key). If the table has no primary key and the bucket key is not specified, the data will be distributed to each bucket randomly. | | table.log.ttl | Duration | 7 days | The time to live for log segments. The configuration controls the maximum time we will retain a log before we will delete old segments to free up space. If set to -1, the log will not be deleted. | +| table.log.local-ttl | Duration | table.log.ttl | The time to retain local log segments after they have been copied to remote storage. If not configured, the value of `table.log.ttl` is used. When configured, the value must be greater than 0 and, if `table.log.ttl` is positive, less than or equal to `table.log.ttl`. | | table.auto-partition.enabled | Boolean | false | Whether enable auto partition for the table. Disable by default. When auto partition is enabled, the partitions of the table will be created automatically. | | table.auto-partition.key | String | (None) | This configuration defines the time-based partition key to be used for auto-partitioning when a table is partitioned with multiple keys. Auto-partitioning utilizes a time-based partition key to handle partitions automatically, including creating new ones and removing outdated ones, by comparing the time value of the partition with the current system time. In the case of a table using multiple partition keys (such as a composite partitioning strategy), this feature determines which key should serve as the primary time dimension for making auto-partitioning decisions. And If the table has only one partition key, this config is not necessary. Otherwise, it must be specified. | | table.auto-partition.time-unit | ENUM | DAY | The time granularity for auto created partitions. The default value is `DAY`. Valid values are `HOUR`, `DAY`, `MONTH`, `QUARTER`, `YEAR`. If the value is `HOUR`, the partition format for auto created is yyyyMMddHH. If the value is `DAY`, the partition format for auto created is yyyyMMdd. If the value is `MONTH`, the partition format for auto created is yyyyMM. If the value is `QUARTER`, the partition format for auto created is yyyyQ. If the value is `YEAR`, the partition format for auto created is yyyy. | diff --git a/website/docs/maintenance/tiered-storage/remote-storage.md b/website/docs/maintenance/tiered-storage/remote-storage.md index 1d00a537d34..fbf4bdf5586 100644 --- a/website/docs/maintenance/tiered-storage/remote-storage.md +++ b/website/docs/maintenance/tiered-storage/remote-storage.md @@ -35,7 +35,10 @@ Below is the list for all configurations to control the log segments tiered beha When local log segments are copied to remote storage, the local log segments will be deleted to reduce local disk cost. But sometimes, we want to keep the several latest log segments retain in local, although they have been coped to remote storage for better read performance. -You can control how many log segments to retain in local by setting the configuration `table.log.tiered.local-segments`(default is 2) per table. +You can control local retention per table with `table.log.tiered.local-segments` (default is 2) +and `table.log.local-ttl` (defaults to `table.log.ttl`). A configured local TTL must be greater +than zero and, when `table.log.ttl` is positive, less than or equal to `table.log.ttl`. A local +segment is deleted only after it has been copied to remote storage. ## Remote snapshot of primary key table diff --git a/website/docs/table-design/data-distribution/ttl.md b/website/docs/table-design/data-distribution/ttl.md index d356c1e334d..a424efe7338 100644 --- a/website/docs/table-design/data-distribution/ttl.md +++ b/website/docs/table-design/data-distribution/ttl.md @@ -9,3 +9,5 @@ Fluss supports TTL for data by setting the TTL attribute for tables with `'table For log tables, this attribute indicates the expiration time of the log table data. For primary key tables, this attribute indicates the expiration time of the changelog and does not represent the expiration time of the primary key table data. If you also want the data in the primary key table to expire automatically, please use [auto partitioning](partitioning.md#auto-partitioning). + +When tiered storage is enabled, `table.log.local-ttl` can be used to retain copied local log segments for a shorter period than remote log segments. It defaults to `table.log.ttl`. A configured local TTL must be greater than zero and, when `table.log.ttl` is positive, less than or equal to `table.log.ttl`. From a204f50a44c7d4da538d6ea63d649f0f69df85c9 Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Tue, 28 Jul 2026 15:32:19 +0800 Subject: [PATCH 2/2] feat: set default value --- .../org/apache/fluss/config/ConfigOptions.java | 8 ++++---- .../java/org/apache/fluss/config/TableConfig.java | 9 ++------- .../org/apache/fluss/config/TableConfigTest.java | 2 +- .../fluss/flink/utils/FlinkConversionsTest.java | 9 +++++++++ .../org/apache/fluss/server/log/LogManager.java | 15 +++++++++++++-- .../org/apache/fluss/server/replica/Replica.java | 1 + website/docs/engine-flink/options.md | 2 +- .../maintenance/tiered-storage/remote-storage.md | 6 +++--- .../docs/table-design/data-distribution/ttl.md | 2 +- 9 files changed, 35 insertions(+), 19 deletions(-) diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java index 1b3aa57f453..31f260ebaac 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java @@ -1795,12 +1795,12 @@ public class ConfigOptions { public static final ConfigOption TABLE_LOG_LOCAL_TTL = key("table.log.local-ttl") .durationType() - .noDefaultValue() + .defaultValue(Duration.ofDays(1)) .withDescription( "The time to retain local log segments after they have been copied to remote storage. " - + "If not configured, the value of `table.log.ttl` is used. When configured, " - + "the value must be greater than 0 and, if `table.log.ttl` is positive, less " - + "than or equal to `table.log.ttl`."); + + "The default value is 1 day. When configured, the value must be greater " + + "than 0 and, if `table.log.ttl` is positive, less than or equal to " + + "`table.log.ttl`."); public static final ConfigOption TABLE_DATALAKE_ENABLED = key("table.datalake.enabled") diff --git a/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java b/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java index 6a6ccd50c43..3e896bdcd8b 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java @@ -95,14 +95,9 @@ public int getTieredLogLocalSegments() { return config.get(ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS); } - /** - * Gets the TTL of local segments for tiered log. The table log TTL is used when no local TTL is - * configured. - */ + /** Gets the TTL of local segments for tiered log. */ public long getLocalLogTTLMs() { - return config.getOptional(ConfigOptions.TABLE_LOG_LOCAL_TTL) - .orElseGet(() -> config.get(ConfigOptions.TABLE_LOG_TTL)) - .toMillis(); + return config.get(ConfigOptions.TABLE_LOG_LOCAL_TTL).toMillis(); } /** Whether the data lake is enabled. */ diff --git a/fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java b/fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java index 6f214f559e3..c0b008cac8c 100644 --- a/fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java @@ -53,7 +53,7 @@ void testTieredLogLocalTtl() { conf.set(ConfigOptions.TABLE_LOG_TTL, Duration.ofDays(3)); TableConfig tableConfig = new TableConfig(conf); - assertThat(tableConfig.getLocalLogTTLMs()).isEqualTo(Duration.ofDays(3).toMillis()); + assertThat(tableConfig.getLocalLogTTLMs()).isEqualTo(Duration.ofDays(1).toMillis()); conf.set(ConfigOptions.TABLE_LOG_LOCAL_TTL, Duration.ofHours(6)); assertThat(tableConfig.getLocalLogTTLMs()).isEqualTo(Duration.ofHours(6).toMillis()); diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java index efc386eb648..3167b1c38fe 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java @@ -372,6 +372,15 @@ void testOptionConversions() { .withDescription( ConfigOptions.CLIENT_REQUEST_TIMEOUT.description())); + flinkOption = FlinkConversions.toFlinkOption(ConfigOptions.TABLE_LOG_LOCAL_TTL); + assertThat(flinkOption) + .isEqualTo( + org.apache.flink.configuration.ConfigOptions.key( + ConfigOptions.TABLE_LOG_LOCAL_TTL.key()) + .stringType() + .defaultValue("86400000 ms") + .withDescription(ConfigOptions.TABLE_LOG_LOCAL_TTL.description())); + flinkOption = FlinkConversions.toFlinkOption(ConfigOptions.CLIENT_WRITER_BUFFER_MEMORY_SIZE); assertThat(flinkOption) diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java b/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java index 8b94360f0c7..ae9c885c0f4 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java @@ -292,6 +292,7 @@ private void waitForLoadLogsInDir(LogRecoveryTask recoveryTask) throws Throwable * @param tableBucket the table bucket * @param logFormat the log format * @param tieredLogLocalSegments the number of segments to retain in local for tiered log + * @param logTtlMs the table log TTL in milliseconds * @param localLogTtlMs the local log TTL in milliseconds from table configuration * @param isChangelog whether the log is a changelog of primary key table */ @@ -301,6 +302,7 @@ public LogTablet getOrCreateLog( TableBucket tableBucket, LogFormat logFormat, int tieredLogLocalSegments, + long logTtlMs, long localLogTtlMs, boolean isChangelog) throws Exception { @@ -324,7 +326,7 @@ public LogTablet getOrCreateLog( scheduler, logFormat, tieredLogLocalSegments, - localLogTtlMs, + effectiveLocalLogTtlMs(logTtlMs, localLogTtlMs), isChangelog, clock, true); @@ -355,6 +357,7 @@ public LogTablet getOrCreateLog( tableBucket, logFormat, tieredLogLocalSegments, + tableConfig.getLogTTLMs(), tableConfig.getLocalLogTTLMs(), isChangelog); } @@ -469,7 +472,9 @@ private LogTablet loadLog( scheduler, tableInfo.getTableConfig().getLogFormat(), tableInfo.getTableConfig().getTieredLogLocalSegments(), - tableInfo.getTableConfig().getLocalLogTTLMs(), + effectiveLocalLogTtlMs( + tableInfo.getTableConfig().getLogTTLMs(), + tableInfo.getTableConfig().getLocalLogTTLMs()), tableInfo.hasPrimaryKey(), clock, isCleanShutdown); @@ -494,6 +499,12 @@ private LogTablet loadLog( return logTablet; } + private long effectiveLocalLogTtlMs(long logTtlMs, long localLogTtlMs) { + return conf.get(ConfigOptions.REMOTE_LOG_TASK_INTERVAL_DURATION).toMillis() > 0L + ? localLogTtlMs + : logTtlMs; + } + /** Close all the logs. */ public void shutdown() { LOG.info("Shutting down LogManager."); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java index d5563a87bb0..ddcd83d7ec3 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java @@ -2162,6 +2162,7 @@ private LogTablet createLog( tableBucket, tableConfig.getLogFormat(), tableConfig.getTieredLogLocalSegments(), + tableConfig.getLogTTLMs(), tableConfig.getLocalLogTTLMs(), isKvTable()); // update high watermark. diff --git a/website/docs/engine-flink/options.md b/website/docs/engine-flink/options.md index 8b69df43f7a..47c8fa34213 100644 --- a/website/docs/engine-flink/options.md +++ b/website/docs/engine-flink/options.md @@ -67,7 +67,7 @@ See more details about [ALTER TABLE ... SET](engine-flink/ddl.md#set-properties) | bucket.num | int | The bucket number of Fluss cluster. | The number of buckets of a Fluss table. | | bucket.key | String | (None) | Specific the distribution policy of the Fluss table. Data will be distributed to each bucket according to the hash value of bucket-key (It must be a subset of the primary keys excluding partition keys of the primary key table). If you specify multiple fields, delimiter is `,`. If the table has a primary key and a bucket key is not specified, the bucket key will be used as primary key(excluding the partition key). If the table has no primary key and the bucket key is not specified, the data will be distributed to each bucket randomly. | | table.log.ttl | Duration | 7 days | The time to live for log segments. The configuration controls the maximum time we will retain a log before we will delete old segments to free up space. If set to -1, the log will not be deleted. | -| table.log.local-ttl | Duration | table.log.ttl | The time to retain local log segments after they have been copied to remote storage. If not configured, the value of `table.log.ttl` is used. When configured, the value must be greater than 0 and, if `table.log.ttl` is positive, less than or equal to `table.log.ttl`. | +| table.log.local-ttl | Duration | 1 day | The time to retain local log segments after they have been copied to remote storage. The default value is 1 day. When configured, the value must be greater than 0 and, if `table.log.ttl` is positive, less than or equal to `table.log.ttl`. | | table.auto-partition.enabled | Boolean | false | Whether enable auto partition for the table. Disable by default. When auto partition is enabled, the partitions of the table will be created automatically. | | table.auto-partition.key | String | (None) | This configuration defines the time-based partition key to be used for auto-partitioning when a table is partitioned with multiple keys. Auto-partitioning utilizes a time-based partition key to handle partitions automatically, including creating new ones and removing outdated ones, by comparing the time value of the partition with the current system time. In the case of a table using multiple partition keys (such as a composite partitioning strategy), this feature determines which key should serve as the primary time dimension for making auto-partitioning decisions. And If the table has only one partition key, this config is not necessary. Otherwise, it must be specified. | | table.auto-partition.time-unit | ENUM | DAY | The time granularity for auto created partitions. The default value is `DAY`. Valid values are `HOUR`, `DAY`, `MONTH`, `QUARTER`, `YEAR`. If the value is `HOUR`, the partition format for auto created is yyyyMMddHH. If the value is `DAY`, the partition format for auto created is yyyyMMdd. If the value is `MONTH`, the partition format for auto created is yyyyMM. If the value is `QUARTER`, the partition format for auto created is yyyyQ. If the value is `YEAR`, the partition format for auto created is yyyy. | diff --git a/website/docs/maintenance/tiered-storage/remote-storage.md b/website/docs/maintenance/tiered-storage/remote-storage.md index fbf4bdf5586..03fae04c7e4 100644 --- a/website/docs/maintenance/tiered-storage/remote-storage.md +++ b/website/docs/maintenance/tiered-storage/remote-storage.md @@ -36,9 +36,9 @@ Below is the list for all configurations to control the log segments tiered beha When local log segments are copied to remote storage, the local log segments will be deleted to reduce local disk cost. But sometimes, we want to keep the several latest log segments retain in local, although they have been coped to remote storage for better read performance. You can control local retention per table with `table.log.tiered.local-segments` (default is 2) -and `table.log.local-ttl` (defaults to `table.log.ttl`). A configured local TTL must be greater -than zero and, when `table.log.ttl` is positive, less than or equal to `table.log.ttl`. A local -segment is deleted only after it has been copied to remote storage. +and `table.log.local-ttl` (default is 1 day). A configured local TTL must be greater than zero +and, when `table.log.ttl` is positive, less than or equal to `table.log.ttl`. A local segment is +deleted only after it has been copied to remote storage. ## Remote snapshot of primary key table diff --git a/website/docs/table-design/data-distribution/ttl.md b/website/docs/table-design/data-distribution/ttl.md index a424efe7338..dd7af3d6dbc 100644 --- a/website/docs/table-design/data-distribution/ttl.md +++ b/website/docs/table-design/data-distribution/ttl.md @@ -10,4 +10,4 @@ Fluss supports TTL for data by setting the TTL attribute for tables with `'table For log tables, this attribute indicates the expiration time of the log table data. For primary key tables, this attribute indicates the expiration time of the changelog and does not represent the expiration time of the primary key table data. If you also want the data in the primary key table to expire automatically, please use [auto partitioning](partitioning.md#auto-partitioning). -When tiered storage is enabled, `table.log.local-ttl` can be used to retain copied local log segments for a shorter period than remote log segments. It defaults to `table.log.ttl`. A configured local TTL must be greater than zero and, when `table.log.ttl` is positive, less than or equal to `table.log.ttl`. +When tiered storage is enabled, `table.log.local-ttl` can be used to control how long copied local log segments are retained. Its default value is 1 day. A configured local TTL must be greater than zero and, when `table.log.ttl` is positive, less than or equal to `table.log.ttl`.