主键表

基本概念

Fluss中的主键表需要确保指定主键的唯一性，并支持INSERT、UPDATE和DELETE操作。

例如，以下 Flink SQL 语句创建了一个主键表，其中shop_id、user_id和ds被定义为主键，并将数据分布到4个分桶中。

CREATE TABLE pk_table (
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  ds STRING,
  PRIMARY KEY (shop_id,user_id,ds) NOT ENFORCED
) WITH (
  'bucket.num' = '4'
);

• 当多条具有相同主键的数据被写入时，会根据处理时间，默认保留最后一条写入的数据。更多写入策略请参见Merge Engines。

• 如果是分区表 ，主键必须包含所有分区键字段（例如分区键是ds，则ds必须为主键之一）。因为数据是按分区键划分，且保证每个分区内数据的主键唯一性可以维护，避免出现不同分区内有相同主键的数据。

数据分桶

对于主键表，Fluss 根据每条记录的分桶键（bucket key）的哈希值决定其归属的分桶。

• 分桶键必须是主键的一部分（不包括分区键）。

  例如主键为shop_id，user_id，ds。分区键为ds。则分桶键可以为'bucket.key' = 'user_id,shop_id'、'bucket.key' = 'user_id'或'bucket.key' = 'shop_id'。

• 若未显式指定分桶键，则默认使用主键中除去分区键的部分作为分桶键。

  例如主键为shop_id、user_id，ds。分区键为ds。未指定分桶键，则分桶键为'bucket.key' = 'user_id,shop_id'。

• 相同哈希值的数据会被分配到同一个分桶中，便于并行处理和负载均衡。

更多详情请参见数据分桶。

部分列更新

对于主键表，Fluss 支持部分列更新（Partial Update），您可以只写入部分列来对数据进行逐步更新，并最终得到完整的数据。需要注意的是，要写入的列必须包含主键列。

CREATE TABLE T (
  k INT,
  v1 DOUBLE,
  v2 STRING,
  PRIMARY KEY (k) NOT ENFORCED
);

可以分别写入部分列数据，但必须包含主键列，如下图所示

Merge Engines

Fluss中的合并引擎（Merge Engine）是一个核心组件，旨在高效处理和整合主键表的数据更新操作。它为用户提供了灵活性，允许定义如何将新进数据记录与具有相同主键的现有记录进行合并。此外，用户还可以指定不同的合并引擎，以根据具体使用场景自定义合并行为。

以下是支持的合并引擎：

• LastRow Merge Engine

  当新记录与现有记录具有相同主键时，保留最新的记录（依据处理时间，最后写入的记录）。

• FirstRow Merge Engine

  • 与默认策略相反，该引擎会保留最早的记录（依据处理时间，即最先写入的记录）。

  • 适用于需要保留初始数据状态的场景。

• Versioned Merge Engine

  • 支持基于版本号的合并逻辑。

  • 用户可以根据记录中的版本号字段来决定保留哪个版本的数据。

  • 适用于需要管理数据版本化或增量更新的场景。

• Aggregation Merge Engine

  • 对于相同主键进行分组，并根据指定的聚合函数，对每个非主键字段逐一进行计算。

  • 涵盖大量常用的聚合函数，简化了数据聚合的场景。

Changelog 生成

Fluss 会记录主键表中数据增删与更新的变更数据，即 changelog，下游消费者可以直接消费 changelog 来得到表的变更。

如果写入 Fluss 主键表的数据依次为插入两条数据(1, 2.0, 'apple')，(1, 4.0, 'banana'), 再删除一条数据(1, 4.0, 'banana')，则将产生如下的变更数据：

+I(1, 2.0, 'apple')
-U(1, 2.0, 'apple')
+U(1, 4.0, 'banana')
-D(1, 4.0, 'banana')

本地存储与分区生命周期

主键表的存储结构包含用于回撤流的 LogTablet 和用于状态查询的 KvTablet。由于两者机制不同，仅依赖常规 TTL 无法完全释放本地磁盘空间。

• LogTablet 管理： table.log.ttl参数仅控制 Changelog 的保留时间，不影响 KvTablet 的数据存储大小。

• KvTablet 管理（推荐）： 若需彻底清理过期数据并释放 KvTablet 占用的本地磁盘空间，必须启用分区生命周期管理（Partition TTL）。

  • 开启自动分区： 设置 table.auto-partition.enabled = true。

  • 设置保留策略： 配置 table.auto-partition.num-retention 指定保留的分区数量。

  • 执行逻辑： 系统将定期轮询并物理删除超出保留数量的旧分区，从而实现本地存储空间的有效回收。

自增列

自增列可自动为每行数据生成唯一的递增数值，无需手动指定。典型应用场景包括：

• 加速去重计数： 将 STRING 类型的标识符映射为自增整数，对整数值执行去重计数，查询速度可提升数倍至数十倍。

• RoaringBitmap 聚合： 配合聚合合并引擎中内置的 rbm32（32 位）和 rbm64（64 位）函数，将自增整数 ID 聚合为 RoaringBitmap，实现高效的去重计数。

CREATE TABLE uid_mapping (
  user_id STRING,
  uid BIGINT,
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'auto-increment.fields' = 'uid',
  'bucket.num' = '2'
);

基本特性

• 唯一性： 自增列生成的值在整张表范围内唯一。

• 单调性： 每个表分桶会在 TabletServer 上缓存一批自增 ID，因此全局范围内无法保证严格单调递增，仅保证大致按时间顺序递增。

  缓存数量由表属性 table.auto-increment.cache-size 控制，默认值为 100,000。较大的缓存可提升分配性能，但会降低单调性。该属性在表创建后不可修改。

示例说明

• table.auto-increment.cache-size = 100000（默认值）

• 表有 2 个分桶，系统启动时向每个分桶预分配一段 ID 范围：

  • 分桶 0 → [1, 100000]

  • 分桶 1 → [100001, 200000]

如果数据落到分桶 1，uid 就从 100001 开始。分桶 0 用完 100,000 个 ID 后，会再申请下一批 [200001, 300000]，以此类推。每个分桶可以分配的 ID 总量没有限制。100000不是上限，是缓存批次大小。

场景应用

自增列可与 Flink Lookup Join 的 lookup.insert-if-not-exists 选项配合使用，实现流处理过程中自动构建字典表：当查找键未命中时，Fluss 自动插入新行并分配唯一的自增 ID。

典型场景是将高基数的字符串标识符映射为紧凑的整数 ID，以便后续使用 RoaringBitmap 进行高效去重聚合。详情请参见维表关联。

使用限制

• 仅支持主键表。

• 不允许显式指定自增列的值，只能由系统隐式分配。

• 每张表只能有一个自增列。

• 自增列的类型必须为 INT 或 BIGINT。

• 不支持指定自增列的起始值和步长。

结合 Lookup Join 构建字典表

通过将自增列与 Flink Lookup Join 的 `lookup.insert-if-not-exists` 选项结合，可以在流处理过程中自动构建字典表——当查找键未命中时，Fluss 自动插入新行并分配自增 ID。这对于将高基数字符串标识符映射为紧凑整数 ID、用于高效的 RoaringBitmap 去重聚合尤为实用。详情和示例请参见维表关联。

主键表的读取与查询

• 数据清理：Fluss 服务仅管理自身的存储资源，不执行 Paimon 存储层的数据清理或垃圾回收操作。

• 实时查询限制：Fluss 仅提供生命周期内（未过期）数据的查询服务。一旦数据超过保留期限，系统将其从 Fluss 层移除，用户无法通过 Fluss 接口检索到该部分数据。

• 历史数据访问：对于已过期的分区数据，系统后续将支持通过 Paimon 的 Union Read 机制进行查询，以实现对全量历史数据的访问。