[kv] Support log tables with deduplication keys #5102

Search before asking

• I searched the existing issues and found no matching proposal.

Motivation

Some append-only workloads need to filter repeated business events while retaining the first row for each event key. Today, users can use a primary-key table with the first_row merge engine, even when they only consume its log and do not need upserts or keyed lookups. This also retains complete row values in KV state and snapshots.

A log-table API with a deduplication key would express this use case directly and allow KV state to retain only key-presence markers.

Solution

Allow a log table to declare a deduplication key without declaring a public primary key:

CREATE TABLE events (
    tenant_id STRING NOT NULL,
    event_id STRING NOT NULL,
    payload STRING
) WITH (
    'bucket.num' = '16',
    'deduplication.key' = 'tenant_id,event_id'
);

• Keep the first accepted row for each key and filter later rows, including different payloads for the same key.
• Expose append and log reads. Append success includes filtered rows; no per-row inserted/duplicate result is required. Upserts, deletes, partial updates and keyed lookups are not exposed.
• Internally reuse the primary-key table's first-row write, idempotence, replication and recovery paths. Store key-presence markers in KV and complete accepted rows in the log.
• Require a nonempty bucket key drawn from the non-partition deduplication columns, allowing equality. Default to all non-partition deduplication columns. Equal keys route to one bucket without cross-bucket coordination.
• Keep keys for the table lifetime, independently of log retention. For partitioned tables, require partition columns in the logical key and use partition-instance scope; dropping and recreating a partition resets that scope.
• Support the Java schema builder and insert-only Flink SQL sinks. Non-key schema evolution follows existing log-table rules.

Anything else?

The expected storage difference from a first_row primary-key table is the removal of complete row values from KV state and snapshots. Key storage, exact lookups, RocksDB maintenance and complete-row logging remain necessary.

Willingness to contribute

• I'm willing to submit a PR!