Realtime Compute for Apache 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'
  );