实时计算Flink版：Paimon主键表和Append Only表

设置'merge-engine' = 'first-row'后，Paimon只会保留相同主键数据中的第一条。与deduplicate合并机制相比，first-row只会产生insert类型的变更数据，且变更数据的产出效率更高。

例如，创建Paimon表的DDL语句。

CREATE TABLE T (
  k INT,
  v1 DOUBLE,
  v2 STRING,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'first-row'
);

如果写入Paimon表的数据依次为+I(1, 2.0, 'apple')，+I(1, 4.0, 'banana'), +I(1, 8.0, 'cherry')，则SELECT * FROM T WHERE k = 1将查询到(1, 2.0, 'apple')这条数据。

对于多条具有相同主键的数据，Paimon主键表将会根据您指定的聚合函数进行聚合。对于不属于主键的每一列，都需要通过fields.<field-name>.aggregate-function指定一个聚合函数，否则该列将默认使用last_non_null_value聚合函数。

例如，price列将会使用max函数进行聚合，而sales列将会使用sum函数进行聚合。

CREATE TABLE T (
  product_id BIGINT,
  price DOUBLE,
  sales BIGINT,
  PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
  'merge-engine' = 'aggregation',
  'fields.price.aggregate-function' = 'max',
  'fields.sales.aggregate-function' = 'sum'
);

如果写入Paimon表的数据依次为+I(1, 23.0, 15)和+I(1, 30.2, 20)，SELECT * FROM T WHERE product_id = 1将查询到(1, 30.2, 35)这条数据。

支持的聚合函数与对应的数据类型如下：

• sum（求和）：支持DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT和DOUBLE。

• product（求乘积）：支持DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT和DOUBLE。

• count（统计非null值总数）：支持INTEGER和BIGINT。

• max（最大值）和min（最小值）：CHAR、VARCHAR、DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP和TIMESTAMP_LTZ。

• first_value（返回第一次输入的值）和last_value（返回最新输入的值）：支持所有数据类型，包括null。

• first_not_null_value（返回第一次输入的非null值）和last_non_null_value（返回最新输入的非 null 值）：支持所有数据类型。

• listagg（将输入的字符串依次用英文逗号连接）：支持STRING。

• bool_and和bool_or：支持BOOLEAN。

设置'merge-engine' = 'partial-update'后，您可以通过多条消息对数据进行逐步更新，并最终得到完整的数据。即具有相同主键的新数据将会覆盖原来的数据，但值为null的列不会进行覆盖。

例如，考虑以下创建 Paimon 表的 DDL 语句。

CREATE TABLE T (
  k INT,
  v1 DOUBLE,
  v2 BIGINT,
  v3 STRING,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'partial-update'
);

如果写入Paimon表的数据依次为+I(1, 23.0, 10, NULL)，+I(1, NULL, NULL, 'This is a book')，+I(1, 25.2, NULL, NULL)，则SELECT * FROM T WHERE k = 1将查询到(1, 25.2, 10, 'This is a book')这条数据。

在partial-update合并机制中，您还可以设置指定WITH参数指定合并顺序或对数据进行打宽与聚合，详情如下：

• 通过Sequence Group为不同列分别指定合并顺序

  除了sequence field之外，您也可以通过sequence group为不同列分别指定合并顺序。您可以为来自不同源表的列分别指定合并顺序，帮您在打宽场景下处理乱序数据。

  例如，a、b 两列根据 g_1 列的值从小到大进行合并，而 c、d 两列将根据g_2 列的值从小到大进行合并。

  CREATE TABLE T (
    k INT,
    a STRING,
    b STRING,
    g_1 INT,
    c STRING,
    d STRING,
    g_2 INT,
    PRIMARY KEY (k) NOT ENFORCED
  ) WITH (
    'merge-engine' = 'partial-update',
    'fields.g_1.sequence-group' = 'a,b',
    'fields.g_2.sequence-group' = 'c,d'
  );

• 同时进行数据的打宽与聚合

  您还可以在WITH参数中设置fields.<field-name>.aggregate-function，为<field-name>这一列指定聚合函数，对该列的值进行聚合。<field-name>这一列需要属于某个 sequence group。aggregation合并机制支持的聚合函数均可使用。

  例如，a、b两列将根据g_1 列的值从小到大进行合并，其中a列将会保留最新的非null值，而 b列将会保留输入的最大值。c、d两列将根据g_2列的值从小到大进行合并，其中c列将会保留最新的非null值，而d列将会求出输入数据的和。

  CREATE TABLE T (
    k INT,
    a STRING,
    b INT,
    g_1 INT,
    c STRING,
    d INT,
    g_2 INT,
    PRIMARY KEY (k) NOT ENFORCED
  ) WITH (
    'merge-engine' = 'partial-update',
    'fields.g_1.sequence-group' = 'a,b',
    'fields.b.aggregate-function' = 'max',
    'fields.g_2.sequence-group' = 'c,d',
    'fields.d.aggregate-function' = 'sum'
  );