Copilot commented on code in PR #3621:
URL: https://github.com/apache/fluss/pull/3621#discussion_r3563040617
##########
website/docs/table-design/data-distribution/ttl.md:
##########
@@ -8,4 +8,46 @@ sidebar_position: 3
Fluss supports TTL for data by setting the TTL attribute for tables with
`'table.log.ttl' = '<duration>'` (default is 7 days). Fluss can periodically
and automatically check for and clean up expired data in the 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).
+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.
+
+## Row TTL for Primary Key Tables
+
+Primary key tables can configure row-level TTL with `'table.row.ttl' =
'<duration>'`. The option has no default value; if it is not configured,
row-level TTL is disabled.
+
+```sql title="Flink SQL"
+CREATE TABLE pk_table
+(
+ id BIGINT,
+ name STRING,
+ PRIMARY KEY (id) NOT ENFORCED
+) WITH (
+ 'bucket.num' = '4',
+ 'table.row.ttl' = '7 d'
+);
+```
+
+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.
+
+TTL cleanup does not emit delete records. Downstream consumers of `$changelog`
or `$binlog` will not receive delete changes when rows expire due to TTL.
+
+By default, row TTL uses processing time. To use event time, configure
`table.row.ttl.time-column` when creating the table:
+
+```sql title="Flink SQL"
+CREATE TABLE pk_table_with_event_time
+(
+ id BIGINT,
+ event_time BIGINT,
+ name STRING,
+ PRIMARY KEY (id) NOT ENFORCED
+) WITH (
+ 'bucket.num' = '4',
+ 'table.row.ttl' = '7 d',
+ 'table.row.ttl.time-column' = 'event_time'
+);
+```
+
+The event-time column must be `BIGINT` epoch milliseconds or `TIMESTAMP_LTZ`.
Rows with null event-time values do not expire through row TTL.
+
+Row TTL must be configured when creating the primary key table. Changing or
disabling `table.row.ttl`, or changing `table.row.ttl.time-column`, with `ALTER
TABLE ... SET` or `ALTER TABLE ... RESET` is not supported in this version.
Review Comment:
This doc section describes event-time row TTL as supported via
`table.row.ttl.time-column`. However, the linked issue (#3620) explicitly
states phase 1 should be processing-time TTL only and that the time-column
option is reserved but must be rejected until a future Fluss-managed expiration
mode is implemented. Please clarify the intended contract: either remove/mark
this event-time section as unsupported for this version, or update the linked
requirements to match the implemented behavior.
##########
fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java:
##########
@@ -242,6 +252,154 @@ private static void checkSystemColumns(RowType schema) {
}
}
+ private static void checkRowTTL(Configuration tableConf, Schema schema,
boolean hasPrimaryKey) {
+ Optional<Duration> rowTTL =
tableConf.getOptional(ConfigOptions.TABLE_ROW_TTL);
+ Optional<String> timeColumn =
+ tableConf.getOptional(ConfigOptions.TABLE_ROW_TTL_TIME_COLUMN);
+ Optional<String> timeColumnId =
+
Optional.ofNullable(tableConf.toMap().get(TableConfig.ROW_TTL_TIME_COLUMN_ID_KEY));
+
+ if (timeColumn.isPresent() && !rowTTL.isPresent()) {
+ throw new InvalidConfigException(
+ String.format(
+ "'%s' requires '%s' to be set.",
+ ConfigOptions.TABLE_ROW_TTL_TIME_COLUMN.key(),
+ ConfigOptions.TABLE_ROW_TTL.key()));
+ }
+
+ if (timeColumnId.isPresent() && !timeColumn.isPresent()) {
+ throw new InvalidConfigException(
+ String.format(
+ "'%s' requires '%s' to be set.",
+ TableConfig.ROW_TTL_TIME_COLUMN_ID_KEY,
+ ConfigOptions.TABLE_ROW_TTL_TIME_COLUMN.key()));
+ }
+
+ if (!rowTTL.isPresent()) {
+ return;
+ }
+
+ if (!hasPrimaryKey) {
+ throw new InvalidTableException(
+ String.format(
+ "'%s' is only supported for primary key tables.",
+ ConfigOptions.TABLE_ROW_TTL.key()));
+ }
+
+ validateRowTTLDuration(rowTTL.get());
+
+ if (timeColumn.isPresent()) {
+ Schema.Column column = getRowTTLTimeColumn(schema,
timeColumn.get());
+ validateRowTTLTimeColumnType(column.getDataType());
+ if (timeColumnId.isPresent()) {
+ validateRowTTLTimeColumnId(timeColumnId.get(),
column.getColumnId());
+ }
+ }
Review Comment:
The linked issue (#3620) specifies that phase 1 row TTL is processing-time
only and that `table.row.ttl.time-column` should be reserved but *rejected*
until Fluss owns the expiration logic/value layout. This validation currently
accepts `table.row.ttl.time-column` (and its internal id) and enables
event-time TTL semantics, which contradicts the issue’s stated requirements.
Please either (a) reject `table.row.ttl.time-column` in this version (throw
InvalidConfigException when it is set), or (b) explicitly update the linked
issue/requirements to reflect that event-time TTL is now supported and stable
enough to commit to.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
