Refreshable tables with a TTL watermark panic in the frontend or the compute node #27509

Problem

Creating a refreshable table (refresh_mode = 'FULL_RELOAD') with WATERMARK ... WITH TTL panics. What panics depends on whether the TTL column is in the primary key.

TTL column not in the primary key: the frontend panics. StreamMaterialize::derive_staging_table_catalog maps the stream key to the pk-only staging table with mapping.try_map_all(stream_key).unwrap(). The stream key of a TTL table includes the TTL column, which the staging table does not have.

Reproduced on main (a73da86):

CREATE TABLE refresh_ttl (id INT PRIMARY KEY, ts TIMESTAMP, WATERMARK FOR ts AS ts - INTERVAL '1' HOUR WITH TTL) WITH (
    connector = '__for_testing_only_batch_posix_fs',
    batch_posix_fs.root = '/tmp/refresh_root',
    refresh_mode = 'FULL_RELOAD',
    match_pattern = '*.csv'
) FORMAT PLAIN ENCODE CSV (without_header = 'true', delimiter = ',');

ERROR:  Panicked when handling the request: called `Option::unwrap()` on a `None` value

The frontend log shows panicked at src/frontend/src/optimizer/plan_node/stream_materialize.rs:506:57.

TTL column in the primary key: the compute node panics. The staging catalog copies the main table's clean_watermark_indices without mapping them to its pk-only columns, so the index can be out of bounds.

CREATE TABLE refresh_ttl_pk (id INT, v INT, ts TIMESTAMP, PRIMARY KEY (id, ts), WATERMARK FOR ts AS ts - INTERVAL '1' HOUR WITH TTL) WITH (
    -- the same options as above
) FORMAT PLAIN ENCODE CSV (without_header = 'true', delimiter = ',');

ERROR:  Failed to run the query

Caused by these errors (recent errors listed first):
  1: gRPC request to meta service (call `/ddl_service.DdlService/CreateTable`) failed: Internal error
  2: completion notifier dropped

The compute node log shows panicked at src/stream/src/common/table/state_table.rs:1019:40: index out of bounds: the len is 2 but the index is 2. The index is that of ts in the table; the staging table only has the two pk columns.

Notes

Read from the code, not verified: supporting this combination would also require two changes in the MaterializeExecutor's MergingData stage:

• feed TTL watermarks into the table before forwarding them (now it only forwards them);
• skip expired rows during the merge instead of emitting deletes for them.

Until then, the frontend could reject the combination with a clear error instead of panicking.