TTL for append-stream (#467)

Author: chenziliang

docs/append-stream-tiered-storage.md

@@ -1,4 +1,4 @@
-# Tiered Storage
+# Tier Storage
 
 Since Timeplus Enterprise 2.8, we have introduced a new feature called Tiered Storage. This feature allows users to store data in a mix of local and remote storage. For example, users can store hot data in a local high-performance storage(e.g. NVMe SSD) for quick access and move the data to object storage(e.g. S3) for long-term retention.
 

docs/append-stream-ttl.md

@@ -0,0 +1,96 @@
+# TTL
+
+TTL expression for Append Stream can specify the logic of automatic moving data between disks and volumes, or recompressing parts where all the data has been expired or just discarding the expired data.
+
+Timeplus evalutes TTL against the data in the historical store in background asychronously during background data merge. The frequence can be controlled by **`merge_with_ttl_timeout`** setting.
+
+## TTL Expression
+
+```sql
+TTL expr
+    [DELETE | RECOMPRESS codec_name1 | TO DISK 'xxx' | TO VOLUME 'xxx'][, DELETE | RECOMPRESS codec_name2 | TO DISK 'aaa' | TO VOLUME 'bbb'] ...
+    [WHERE conditions]
+```
+
+Type of TTL rule may follow each TTL expression. It affects an action which is to be done once the expression is satisfied (reaches current time):
+
+- **`DELETE`** - delete expired rows (default action);
+- **`RECOMPRESS codec_name`** - recompress data part with the `codec_name`;
+- **`TO DISK 'aaa'`** - move part to the disk `aaa`;
+- **`TO VOLUME 'bbb'`** - move part to the disk `bbb`;
+
+**`DELETE`** action can be used together with **`WHERE`** clause to delete only some of the expired rows based on a filtering condition:
+
+```sql
+TTL time_column + INTERVAL 1 MONTH DELETE WHERE column = 'value'
+```
+
+## Alter TTL
+
+```sql
+ALTER STREAM <db.stream> MODIFY TTL <expr>;
+```
+
+**Example:**
+```sql
+ALTER STREAM test MODIFY TTL d + INTERVAL 1 DAY;
+```
+
+## Trigger Data Delete with Merge
+
+Expired data is removed only during the merge process, which runs asynchronously in the background.
+
+To accelerate this cleanup, you can manually trigger a merge by running the `OPTIMIZE` command. This attempts to start an unscheduled merge of data parts for a stream:
+
+```sql
+OPTIMIZE STREAM <db.stream-name>;
+```
+
+## Examples
+
+### Delete After Expired 
+
+Creating a stream, where the rows are expired after one month. The expired rows where dates are Mondays are deleted:
+```sql
+CREATE STREAM stream_with_where
+(
+    d datetime,
+    a int
+)
+PARTITION BY to_YYYYMM(d)
+ORDER BY d
+TTL d + INTERVAL 1 MONTH DELETE WHERE to_day_of_week(d) = 1;
+```
+
+### Recompress After Expired 
+
+Creating a stream, where expired rows are recompressed:
+```sql
+CREATE TABLE stream_for_recompression
+(
+    d datetime,
+    key uint64,
+    value string
+) 
+ORDER BY tuple()
+PARTITION BY key
+TTL d + INTERVAL 1 MONTH RECOMPRESS CODEC(ZSTD(17)), d + INTERVAL 1 YEAR RECOMPRESS CODEC(LZ4HC(10))
+SETTINGS 
+    min_rows_for_wide_part = 0, 
+    min_bytes_for_wide_part = 0;
+```
+
+### Move After Expired
+
+```sql
+CREATE STREAM stream_for_move
+(
+    d datetime,
+    a int
+)
+PARTITION BY to_YYYYMM(d)
+ORDER BY d
+TTL d + INTERVAL 1 MONTH DELETE,
+    d + INTERVAL 1 WEEK TO VOLUME 'aaa',
+    d + INTERVAL 2 WEEK TO DISK 'bbb';
+```

docs/append-stream.md

@@ -40,7 +40,8 @@ SETTINGS
     fetch_threads=<remote-fetch-threads>,
     flush_threshold_count=<batch-flush-rows>,
     flush_threshold_ms=<batch-flush-timeout>,
-    flush_threshold_bytes=<batch-flush-size>;
+    flush_threshold_bytes=<batch-flush-size>,
+    merge_with_ttl_timeout=<timeout-in-seconds>;
 ```
 
 ### Storage Architecture
@@ -164,6 +165,10 @@ In most cases, you don't need a partition key, and if you do need to partition,
 
 For partitioning by month, use the `to_YYYYMM(date_column)` expression, where `date_column` is a column with a date of the type `date`. The partition names here have the `YYYYMM` format.
 
+### `TTL expr`
+
+See [TTL](/append-stream-ttl) for details 
+
 ### Settings
 
 #### `shards`
@@ -292,14 +297,26 @@ Controls the parallelism when fetching data from remote shared storage.
 
 Flushes data to the backend columnar store when this row threshold is reached.
 
+**Default**: `100000`
+
 #### `flush_threshold_ms`
 
 Flushes data to the backend columnar store when this time threshold is reached.
 
+**Default**: `2000`
+
 #### `flush_threshold_bytes`
 
 Flushes data to the backend columnar store when this bytes threshold is reached.
 
+**Default**: `16777216`
+
+#### `merge_with_ttl_timeout`
+
+Minimum delay in seconds before repeating a merge with delete TTL.
+
+**Default**: `14400`
+
 ## Enable Zero-Replication WAL
 
 You can store WAL (NativeLog) data in S3-compatible cloud storage. To enable this, configure a disk and then create an append stream using that disk.

docs/mutable-stream.md

@@ -220,10 +220,14 @@ Controls the parallelism when fetching data from remote shared storage.
 
 Flushes data to the backend key-value store (RocksDB) when this row threshold is reached.
 
+**Default**: `100000`
+
 #### `flush_ms`
 
 Flushes data to the backend key-value store (RocksDB) when this time threshold is reached.
 
+**Default**: `30000`
+
 #### `log_kvstore`
 
 If `true`, logs internal RocksDB activity for debugging.

docs/working-with-streams.md

@@ -9,29 +9,28 @@ Timeplus `streams` are conceptually similar to `tables` in traditional SQL datab
 
 To support a variety of use cases efficiently, Timeplus offers multiple types of streams:
 
-1. [Append Stream](/append-stream)
+- [Append Stream](/append-stream)
 
    The default stream type in Timeplus. It uses columnar encoding and is optimized for **range scans** via a **sorting key**. It suits workloads with infrequent data mutations (e.g., `UPDATE` or `DELETE`).
 
-2. [Mutable Stream](/mutable-stream)
+- [Mutable Stream](/mutable-stream)
 
    Row-encoded and similar in behavior to a **MySQL table**, where each primary key corresponds to a single row. It is optimized for **frequent data mutations** (`UPDATE`, `UPSERT`, `DELETE`) and supports **point and range queries** via **primary or secondary indexes**.
 
-3. [Versioned Key-Value Stream](/versioned-stream)
+- [Versioned Key-Value Stream](/versioned-stream)
 
    Similar to the mutable stream but uses **columnar encoding**. It offers better **compression** but lower performance for updates and point queries, especially when cardinality is high. Best suited for scenarios where **data mutations are less frequent**.
 
-4. [Changelog Key-Value Stream](/changelog-stream)
+- [Changelog Key-Value Stream](/changelog-stream)
 
    Designed to model **change data capture (CDC) events**, with **columnar encoding** for efficient downstream processing.
 
-5. [External Stream](/external-stream)
+- [External Stream](/external-stream)
 
    As the name implies, the data resides outside of Timeplus. Timeplus can reference external sources (e.g., a **Kafka topic**) and execute **streaming SQL** queries against them in real time.
 
 > Note: Timeplus also supports [External Tables](/sql-create-external-table), which allow **historical queries and inserts** only (e.g., against ClickHouse, MySQL, PostgreSQL, MongoDB, etc.).
 
-
 ## Stream Internals
 
 When users [create a stream](/sql-create-stream) in Timeplus, they can specify the number of **shards** for the stream. Each shard consists of two core components at the storage layer: