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 14b019853df..d7f0afd7967 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 @@ -1008,8 +1008,9 @@ public class ConfigOptions { .booleanType() .defaultValue(false) .withDescription( - "Whether to roll a non-empty active log segment when it has expired " - + "according to the table log TTL. Disabled by default."); + "Whether to roll a non-empty active log segment after it has expired " + + "according to the effective local cleanup TTL and the high " + + "watermark has reached the log end offset. Disabled by default."); public static final ConfigOption LOG_REPLICA_HIGH_WATERMARK_CHECKPOINT_INTERVAL = key("log.replica.high-watermark.checkpoint-interval") diff --git a/website/docs/assets/kv-retention.png b/website/docs/assets/kv-retention.png new file mode 100644 index 00000000000..32f4269face Binary files /dev/null and b/website/docs/assets/kv-retention.png differ diff --git a/website/docs/assets/log-retention.png b/website/docs/assets/log-retention.png new file mode 100644 index 00000000000..59912e4ddc0 Binary files /dev/null and b/website/docs/assets/log-retention.png differ diff --git a/website/docs/assets/partition-retention.png b/website/docs/assets/partition-retention.png new file mode 100644 index 00000000000..f69e011631f Binary files /dev/null and b/website/docs/assets/partition-retention.png differ diff --git a/website/docs/maintenance/configuration.md b/website/docs/maintenance/configuration.md index 49d29db7f6a..eedb1b821bb 100644 --- a/website/docs/maintenance/configuration.md +++ b/website/docs/maintenance/configuration.md @@ -7,8 +7,9 @@ sidebar_position: 1 All configurations can be set in Fluss configuration file `conf/server.yaml` -The configuration is parsed and evaluated when the Fluss processes are started. -Changes to the configuration file require restarting the relevant processes. +The configuration is parsed and evaluated when the Fluss processes are started. Most changes to +the configuration file require restarting the relevant processes. Options that support dynamic +updates are listed in [Updating Configs](operations/updating-configs.md#updating-cluster-configs). Users can organize config in format `key: value`, such as: @@ -142,6 +143,8 @@ The logging-related environment options (`env.log.dir`, `env.log.level`, `env.lo | log.file-preallocate | Boolean | false | True if we should preallocate the file on disk when creating a new log segment. | | log.flush.interval-messages | Long | Long.MAX_VALUE | This setting allows specifying an interval at which we will force a fsync of data written to the log. For example if this was set to 1, we would fsync after every message; if it were 5 we would fsync after every five messages. | | log.flush.offset.checkpoint-interval | Duration | 1min | The frequency with which we update the persistent record of the last flush which acts as the log recovery point. The default setting is 60 seconds. | +| log.retention.check-interval | Duration | 5min | The frequency with which the log manager checks whether local log segments are eligible for TTL cleanup. The value must be greater than 0. | +| log.retention.roll-active-segment.enabled | Boolean | false | Whether to roll a non-empty active log segment after the effective local cleanup TTL expires and the high watermark reaches the log end offset. Rolling makes the segment eligible for remote upload and subsequent local cleanup. This option is disabled by default and supports [dynamic updates](operations/updating-configs.md#updating-cluster-configs). | | log.replica.high-watermark.checkpoint-interval | Duration | 5s | The frequency with which the high watermark is saved out to disk. The default setting is 5 seconds. | | log.replica.max-lag-time | Duration | 30s | If a follower replica hasn't sent any fetch log requests or hasn't consumed up the leaders log end offset for at least this time, the leader will remove the follower replica from isr | | log.replica.write-operation-purge-number | Integer | 1000 | The purge number (in number of requests) of the write operation manager, the default value is 1000. | diff --git a/website/docs/maintenance/operations/updating-configs.md b/website/docs/maintenance/operations/updating-configs.md index d950e2848e3..c6f6620a279 100644 --- a/website/docs/maintenance/operations/updating-configs.md +++ b/website/docs/maintenance/operations/updating-configs.md @@ -20,6 +20,9 @@ Currently, the supported dynamically updatable server configurations include: - `datalake.enabled`: Control whether the cluster is ready to create and manage lakehouse tables. When this option is explicitly configured to true, `datalake.format` must also be configured. - `datalake.format`: Specify the lakehouse format, e.g., `paimon`, `iceberg`. When enabling lakehouse storage explicitly, use it together with `datalake.enabled = true`. - Options with prefix `datalake.${datalake.format}` +- `log.retention.roll-active-segment.enabled`: Control whether a non-empty active segment can be + rolled after the effective local cleanup TTL expires and the high watermark reaches the log end + offset. See [TTL](../../table-design/data-distribution/ttl.md#active-segment-rolling). - `kv.rocksdb.shared-rate-limiter.bytes-per-sec`: Control RocksDB flush and compaction write rate shared across all RocksDB instances on the TabletServer. The rate limiter is always enabled. Set to a lower value (e.g., 100MB) to limit the rate, or a very high value to effectively disable rate limiting. - `security.sasl.plain.credentials`: Add, change, or remove the users that can authenticate with SASL/PLAIN, see [Authentication](security/authentication.md#managing-multiple-users). - `server.historical-partition.thread-pool.max-size`: Change the maximum number of threads used for historical partition operations. @@ -75,6 +78,11 @@ CALL sys.set_cluster_configs( CALL sys.set_cluster_configs( config_pairs => 'kv.rocksdb.shared-rate-limiter.bytes-per-sec', '200MB' ); + +-- Allow expired active log segments to be rolled +CALL sys.set_cluster_configs( + config_pairs => 'log.retention.roll-active-segment.enabled', 'true' +); ``` See [Procedures](engine-flink/procedures.md#cluster-configuration-procedures) for detailed documentation on `get_cluster_configs`, `set_cluster_configs`, `append_cluster_configs`, `subtract_cluster_configs`, and `reset_cluster_configs` procedures. diff --git a/website/docs/maintenance/operations/upgrade-notes-1.0.md b/website/docs/maintenance/operations/upgrade-notes-1.0.md index 4417d12c091..b6e34049901 100644 --- a/website/docs/maintenance/operations/upgrade-notes-1.0.md +++ b/website/docs/maintenance/operations/upgrade-notes-1.0.md @@ -33,6 +33,31 @@ Custom credentials providers must now implement `software.amazon.awssdk.auth.cre Deployments using static access keys or the default AWS credentials provider chain do not require configuration changes. +### Active Segment Retention Rollout + +When upgrading a cluster from v0.9, keep +`log.retention.roll-active-segment.enabled` disabled for the entire upgrade. This is the default, so +no configuration change is required before or during the rolling upgrade. + +After every CoordinatorServer and TabletServer has been upgraded to v1.0 and the upgrade is +complete, enable the option with a dynamic cluster configuration update: + +```sql +CALL sys.set_cluster_configs( + config_pairs => 'log.retention.roll-active-segment.enabled', 'true' +); +``` + +Enabling the option allows a non-empty active local log segment to be rolled after its effective +local cleanup TTL expires and all records are committed. For tiered logs, the effective TTL is +`table.log.local-ttl`, or `table.log.ttl` when the local option is not configured. The rolled +segment can then be uploaded to remote storage and cleaned up locally. This avoids indefinitely +retaining an expired active segment on low-traffic tables. + +See [TTL](../../table-design/data-distribution/ttl.md) for the segment lifecycle, +[remote storage](../tiered-storage/remote-storage.md) for tiered-log retention, and +[updating configs](updating-configs.md#updating-cluster-configs) for other dynamic update methods. + ### New `datalake.enabled` Cluster Configuration Starting in v1.0, Fluss introduces the cluster-level configuration `datalake.enabled` to control whether the cluster is ready to create and manage lakehouse tables. diff --git a/website/docs/maintenance/tiered-storage/remote-storage.md b/website/docs/maintenance/tiered-storage/remote-storage.md index 4e903c7e67a..72ac4454da6 100644 --- a/website/docs/maintenance/tiered-storage/remote-storage.md +++ b/website/docs/maintenance/tiered-storage/remote-storage.md @@ -33,13 +33,25 @@ Below is the list for all configurations to control the log segments tiered beha ### Table configurations about remote log -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`. When `table.log.local-ttl` is not configured, it falls back to -`table.log.ttl`. A non-positive local TTL disables TTL-based local cleanup. When both TTLs are -positive, the local TTL must be less than or equal to `table.log.ttl`. An expired local segment is -deleted only after it has been copied to remote storage. +After a rolled local log segment is copied to remote storage, it can be removed to reduce local disk +usage. Uncopied segments are never eligible for local TTL cleanup. + +Use the following table options to control local retention: + +- `table.log.local-ttl` controls TTL-based cleanup. It inherits `table.log.ttl` when it is not + configured. Setting it to `0ms` disables TTL-based local cleanup. When both TTLs are positive, + the local TTL must be less than or equal to `table.log.ttl`. +- `table.log.tiered.local-segments` keeps the configured number of recent local segments from + count-based cleanup (default: 2). Copied segments beyond that count can be removed even before + their local TTL expires. + +The two cleanup policies are independent: a copied local segment can be removed when it exceeds the +configured segment count or when its local TTL expires. + +`table.log.ttl` independently controls the retention of table log data, including its remote copy. +See [TTL](../../table-design/data-distribution/ttl.md) for the complete lifecycle from an active +local segment through rolling, upload, local cleanup, and remote expiration. The server-side +remote-log settings are listed in [server configuration](../configuration.md#log-tiered-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 06eeee5c060..2915f4de99f 100644 --- a/website/docs/table-design/data-distribution/ttl.md +++ b/website/docs/table-design/data-distribution/ttl.md @@ -1,43 +1,106 @@ --- -title: TTL +title: Data Retention and TTL +sidebar_label: TTL sidebar_position: 3 --- -# TTL +# Data Retention and TTL -Fluss supports TTL for data by setting the TTL attribute for tables with `'table.log.ttl' = ''` (default is 7 days). Fluss can periodically and automatically check for and clean up expired data in the table. +Fluss provides three retention mechanisms, which can be configured together: -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). +| Mechanism | What expires | Main configuration | +|---|---|---| +| [Log Retention](#log-retention) | Data logs in log tables; changelogs in primary key tables | `table.log.ttl` | +| [KV Retention](#kv-retention) | Individual rows in a primary key table's KV store | `table.kv.ttl` | +| [Partition Retention](#partition-retention) | Entire time partitions in log tables or primary key tables | `table.auto-partition.*` | -When tiered storage is enabled, `table.log.local-ttl` can be used to control how long copied local log segments are retained. If it is not configured, it falls back to `table.log.ttl` for backward compatibility. A non-positive local TTL disables TTL-based local cleanup. When both TTLs are positive, the local TTL must be less than or equal to `table.log.ttl`. +For example, a primary key table can keep its current rows while retaining only one day of changelog. +Expiration is asynchronous, so a TTL is not an exact deadline for data removal or disk-space reclamation. -## Row TTL for Primary Key Tables +## Log Retention -Primary key tables can configure row-level TTL with `'table.kv.ttl' = ''`. The duration must be at least 1 millisecond. The option has no default value; if it is not configured, row-level TTL is disabled. +### Table Log TTL -```sql title="Flink SQL" -CREATE TABLE pk_table -( - id BIGINT, - name STRING, - PRIMARY KEY (id) NOT ENFORCED -) WITH ( - 'bucket.num' = '4', - 'table.kv.ttl' = '7 d' -); -``` +Set `table.log.ttl` to control how long logs are retained. The default is **7 days**; `0ms` disables +expiration based on this TTL. + +- For **log tables**, this controls the retention of table data. +- For **primary key tables**, this controls changelog history available to consumers. Current KV + rows remain governed by [KV Retention](#kv-retention). + +### Local Log TTL + +With [remote log storage](../../maintenance/tiered-storage/remote-storage.md#remote-log) enabled +(the default), local and remote storage serve different purposes: + +- **Local storage is the hot tier** for recent data. It serves low-latency tail reads and, with the + default Apache Arrow log format, supports [column pruning](../../engine-flink/reads.md#column-pruning) + so streaming consumers read only the columns they need. +- **Remote storage is the cold tier** that keeps older logs readable within the configured log + retention period, even after their local copies are removed. It supports historical replay and + consumers catching up beyond the local window. Remote log reads typically have more network + bandwidth available and do not affect the Fluss cluster's online read and write traffic. + +Size local retention to cover the data your tailing consumers need: their expected maximum lag, +plus time for restarts and catching up after backpressure, with some headroom. Size remote retention +for how far back consumers need to replay data. This keeps the hot working set local while using +remote storage for longer history. + +![Local hot storage and remote cold storage with independent cleanup policies](../../assets/log-retention.png) + +The following options control the two tiers: + +| Option | Default | Effect | +|---|---|---| +| `table.log.ttl` | `7 d` | Controls remote log retention. | +| `table.log.local-ttl` | Inherits `table.log.ttl` | Controls TTL-based cleanup of local copies. `0ms` disables this cleanup policy. | +| `table.log.tiered.local-segments` | `2` | Controls how many recent local log segments the count-based cleanup policy retains. Must be greater than 0. | -Row TTL is best-effort cleanup. A row becomes eligible for cleanup after the configured duration, but expired rows may still be visible until RocksDB compaction removes them. Fluss stores the TTL timestamp in the primary-key table value and uses a compaction filter to remove expired rows during RocksDB compaction. +When both TTLs are positive, `table.log.local-ttl` must not exceed `table.log.ttl`. Local copies +can be removed when **either** the local TTL or segment-count limit is reached, after they have +been copied remotely. Setting local TTL to `0ms` leaves count-based cleanup enabled. -Auto partitioning is the recommended expiration mechanism when data can be partitioned by time, because expiring whole partitions preserves changelog completeness. Row TTL is intended for cases that require per-row cleanup and can accept weaker changelog semantics. +For example, if consumers normally lag by at most 30 minutes and need an additional hour for +restarts and catching up, `table.log.local-ttl = '2 h'` is a starting point with some headroom. +Set `table.log.ttl = '7 d'` if consumers need seven days of replay history. Also size +`table.log.tiered.local-segments` to cover that two-hour hot window at peak write rates, accounting +for how frequently segments roll in each bucket. The default of two segments does **not** guarantee +two hours of local data: the segment-count policy may remove local copies before the local TTL +expires. Reads beyond the remaining local window access remote storage. -Row TTL cleanup does not emit delete records. Downstream consumers of `$changelog` or `$binlog` will not receive delete changes when rows expire. After compaction removes an expired row, the next write for the same primary key is treated as an insert and does not emit an `UPDATE_BEFORE` for the expired value. +All three table options support `ALTER TABLE ... SET` and `ALTER TABLE ... RESET`. +See [Updating Configs](../../maintenance/operations/updating-configs.md#updating-table-configs). -By default, row TTL uses processing time. To use event time, configure `table.kv.ttl.time-column` when creating the table: + +### Active-segment Rolling + +For low-traffic tables, enable the server option `log.retention.roll-active-segment.enabled` to +allow expired active logs to become eligible for upload and cleanup even when no new data arrives. +It is **disabled by default** and supports +[dynamic cluster updates](../../maintenance/operations/updating-configs.md#updating-cluster-configs). +The expiration time follows the effective local log TTL described above. + +When upgrading from v0.9, keep this option disabled until every CoordinatorServer and TabletServer +has been upgraded to v1.0 and the upgrade is complete. Then enable it dynamically as described in +the [1.0 Upgrade Notes](../../maintenance/operations/upgrade-notes-1.0.md#active-segment-retention-rollout). + +## KV Retention + +### Row TTL for Primary Key Tables + +Configure row TTL when creating a primary key table: + +| Option | Default | Effect | +|---|---|---| +| `table.kv.ttl` | Disabled when unset | Expires individual KV rows. Must be at least **1 millisecond** when configured. | +| `table.kv.ttl.time-column` | Unset; uses processing time | Uses the specified column's event time to determine row expiration. Requires `table.kv.ttl`. | + +![Rows older than a seven-day KV TTL are removed while other rows remain](../../assets/kv-retention.png) + +For example, this table expires rows after seven days by event time and retains one day of changelog: ```sql title="Flink SQL" -CREATE TABLE pk_table_with_event_time +CREATE TABLE pk_table ( id BIGINT, event_time BIGINT, @@ -46,14 +109,54 @@ CREATE TABLE pk_table_with_event_time ) WITH ( 'bucket.num' = '4', 'table.kv.ttl' = '7 d', - 'table.kv.ttl.time-column' = 'event_time' + 'table.kv.ttl.time-column' = 'event_time', + 'table.log.ttl' = '1 d' ); ``` -The event-time column must be `BIGINT` epoch milliseconds, `TIMESTAMP`, or `TIMESTAMP_LTZ`. `TIMESTAMP` values are interpreted in the TabletServer's system time zone, so all TabletServers must use the same system time zone. Rows with null event-time values do not expire through row TTL. +Omit `table.kv.ttl.time-column` to use processing time. For event time, the column must be `BIGINT` +epoch milliseconds, `TIMESTAMP`, or `TIMESTAMP_LTZ`. `TIMESTAMP` uses the TabletServer's system +time zone, which must be consistent across servers. Rows with null event-time values do not expire +through row TTL. + +Keep these effects and restrictions in mind: + +- Expired rows may remain queryable until background cleanup runs. +- KV TTL cleanup **does not emit DELETE records** to changelogs. So `$changelog`/`$binlog` virtual tables and changelog consumers can't observe the `DELETE` event for the TTL cleaned rows. After a row is cleaned up, writing the same key is treated as an insert. +- Both options must be configured at table creation; changing or resetting them is unsupported yet. +- Upgrade all servers to **Fluss 1.0+** before creating a KV-TTL table. Downgrading below 1.0 after + such a table exists is unsupported. + +Existing remote snapshot files have separate retention, controlled by +[`kv.snapshot.num-retained`](../../maintenance/tiered-storage/remote-storage.md#remote-snapshot-of-primary-key-table). + +## Partition Retention + +Use [auto partitioning](partitioning.md#auto-partitioning) when data should expire in whole time +partitions. It applies to both log tables and primary key tables. + +Enable it with `table.auto-partition.enabled = 'true'` (default: `false`), then configure: + +| Option | Default | Effect | +|---|---|---| +| `table.auto-partition.time-unit` | `DAY` | Sets the calendar unit for partition creation and expiration. | +| `table.auto-partition.num-retention` | `7` | Retains this many historical time units before the current unit. Older partitions expire. | +| `table.auto-partition.num-precreate` | `2` | Pre-creates this many partitions, including the current one. Does not extend retention. | +| `table.auto-partition.time-zone` | System time zone | Sets the time zone for calendar boundaries. | + +![Three historical daily partitions are retained, with the current and next partitions pre-created](../../assets/partition-retention.png) + +For daily partitions with `num-retention = '3'`, on November 11 the retained historical partitions +are November 8–10, in addition to the current partition on November 11. The default pre-creation +count also creates November 12. Writing new rows into an older partition does not extend its retention. -Row TTL must be configured when creating the primary key table. Changing or disabling `table.kv.ttl`, or changing `table.kv.ttl.time-column`, with `ALTER TABLE ... SET` or `ALTER TABLE ... RESET` is not supported in this version. +Log and KV TTLs can expire data inside a retained partition. Longer log or KV TTLs do not prevent +the partition itself from expiring. The retention and pre-creation counts support +`ALTER TABLE ... SET` and `ALTER TABLE ... RESET`. See [Partitioning](partitioning.md) for complete +examples and multi-field partition rules. -Before creating a row-TTL primary key table, upgrade every server to Fluss 1.0+ so that all servers understand the tagged value layout used for row TTL. After a row-TTL table exists, downgrading to a version < 1.0 that does not understand that layout is not supported. +For lakehouse tables, data already tiered into the lake can remain accessible through Union Read +after Fluss logs or partitions expire, see [Lakehouse Data Retention](../../maintenance/tiered-storage/lakehouse-storage.md#data-retention). -See [Flink Connector Options](/engine-flink/options.md#storage-options) for the complete row TTL option definitions. +See [Flink Connector Options](../../engine-flink/options.md#storage-options) and +[Server Configuration](../../maintenance/configuration.md#log) for complete configuration references.