Data Lake Formation：Paimon主鍵表：配置與特性

資料的分發

預設情況下，DLF將根據每條資料主鍵的雜湊值，確定該資料屬於哪個分桶。

資料可見度

延遲分桶表的資料可見度取決於寫入方式。滿足以下條件時，DLF 將啟用直寫固定分桶最佳化（即跳過 bucket-postpone 臨時目錄，將資料直接寫入固定分桶），資料寫入並 commit 後即刻可見：

• 主鍵表未啟用 deletion vectors，且通過 Flink 批作業或 Spark 作業執行 INSERT INTO 或 INSERT OVERWRITE 操作。

• 主鍵表已啟用 deletion vectors，且通過 Flink 批作業或 Spark 作業執行 INSERT OVERWRITE 操作。

其它情況下，資料需等待 DLF 處理後可見，通常在 1 ~ 3 分鐘內。若 DLF 進行中資源或分桶數調整，可能延長至 5 ~ 10 分鐘。可通過選擇特定的儲存最佳化模式減少調整頻率，詳見儲存最佳化模式。

資料一致性

• 同一作業寫入：對於同一使用者作業寫入的資料，DLF 最佳化作業將按寫入順序處理資料，即保證 sequential 層級的一致性。

  說明

  從 checkpoint 或 savepoint 重啟，且並發數不變的作業，也認為與重啟前是同一使用者作業。

• 不同作業寫入：對於不同使用者作業寫入的資料，Paimon 表的最終狀態可能是兩個作業結果資料的混合，但不會有資料丟失，即保證 snapshot isolation 層級的一致性。如果對資料的合并順序有需求，可以設定 sequence field 以指定資料的合并順序，詳見處理亂序資料。

資料的分發

與延遲分桶一致，詳見資料的分發。

資料可見度

固定分桶表的資料可見度與 deletion vectors 有關：

• 未啟用 deletion vectors 的表，資料寫入並 commit 之後馬上可見。

• 啟用 deletion vectors 的表，寫入的資料需要等待 DLF 最佳化作業處理後才能讀取。一般情況下，資料將在 1 ~ 3 分鐘後可見。

資料的分發

動態分桶表會先將資料寫入已有的分桶中，當分桶的資料量超過限制時，再自動建立新的分桶。以下表參數將會影響動態分桶的行為：

動態分桶表的更新行為和資源消耗取決於主鍵與分區鍵的關係：

• 主鍵完全包含分區鍵 → 非跨分區更新（僅消耗堆記憶體）

• 主鍵不完全包含分區鍵 → 跨分區更新（需要 RocksDB，效能開銷更大）

資料可見度

與固定分桶表一致，詳見資料可見度。

支援的彙總函式

• sum：求和函數。支援的資料類型為 DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE。

• product：求乘積函數。支援的資料類型為 DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE。

• max：求最大值的函數。支援的資料類型為 CHAR、VARCHAR、DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP、TIMESTAMP_LTZ。

• min：求最小值的函數。支援的資料類型為 CHAR、VARCHAR、DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP、TIMESTAMP_LTZ。

• first_value：返回第一次輸入的值，包括 null。支援所有資料類型。

• first_not_null_value：返回第一次輸入的非 null 值。支援所有資料類型。

• last_value：返回最新輸入的值，包括 null。支援所有資料類型。

• last_non_null_value（預設）：返回最新輸入的非 null 值。支援所有資料類型。

• listagg：將輸入的字串依次用英文逗號串連。例如，輸入的字串為 apple、banana、cherry，將返回 apple,banana,cherry。支援 STRING 類型。

• bool_and：對所有輸入的值求 and。支援 BOOLEAN 類型。

• bool_or：對所有輸入的值求 or。支援 BOOLEAN 類型。

• rbm32：合并多個 32 位 RoaringBitmap。支援 VARBINARY 類型。

• rbm64：合并多個 64 位元 RoaringBitmap。支援 VARBINARY 類型。

• nested_update：將多行彙總到一個 ARRAY 中。通過設定表參數 'fields.<field-name>.nested-key' = 'pk0,pk1,...'，還可以將 ARRAY 中的資料根據 nested key 進行去重（保留最後一條）。若未設定 nested key，則每條新資料將追加在 ARRAY 末尾。

• nested_partial_update：與 nested_update 類似，但會根據 nested key 對 ARRAY 中的資料進行部分列更新。必須設定表參數 'fields.<field-name>.nested-key' = 'pk0,pk1,...'。

• collect：將多個 ARRAY 合并成一個。通過設定表參數 'fields.<field-name>.distinct' = 'true'，可以對 ARRAY 中的資料進行去重。

• merge_map：將多個 MAP 合并成一個。相同 key 的資料將保留最後一個 value。

彙總函式使用舉例

例如，使用nested_update彙總函式合并多個子訂單記錄：

-- 主訂單表
CREATE TABLE orders (
  order_id BIGINT PRIMARY KEY NOT ENFORCED,
  user_name STRING,
  address STRING
);

-- 子訂單表
CREATE TABLE sub_orders (
  order_id BIGINT,
  sub_order_id INT,
  product_name STRING,
  price BIGINT,
  PRIMARY KEY (order_id, sub_order_id) NOT ENFORCED
);

-- 寬表
-- 將同一主訂單的所有子訂單彙總到 sub_orders 中
-- 同時，sub_orders 中的結果將根據 sub_order_id 進行去重
CREATE TABLE order_wide (
  order_id BIGINT PRIMARY KEY NOT ENFORCED,
  user_name STRING,
  address STRING,
  sub_orders ARRAY<ROW<sub_order_id BIGINT, product_name STRING, price BIGINT>>
) WITH (
  'merge-engine' = 'aggregation',
  'fields.sub_orders.aggregate-function' = 'nested_update',
  'fields.sub_orders.nested-key' = 'sub_order_id'
);

INSERT INTO order_wide

SELECT
  order_id,
  user_name,
  address,
  CAST (NULL AS ARRAY<ROW<sub_order_id BIGINT, product_name STRING, price BIGINT>>)
FROM orders

UNION ALL

SELECT
  order_id,
  CAST (NULL AS STRING),
  CAST (NULL AS STRING),
  ARRAY[ROW(sub_order_id, product_name, price)]
FROM sub_orders;

假設主訂單表中的資料為：

• (1, 'Alice', 'add1'),

• (2, 'Bob', 'add2')

子訂單表中的資料為：

• (1, 11, 'apple', 10),

• (1, 12, 'banana', 20),

• (2, 21, 'cherry', 30),

• (2, 22, 'peach', 40),

• (2, 21, 'cherry', 50)

那麼寬表的最終資料為：

• (1, 'Alice', 'add1', [(11, 'apple', 10), (12, 'banana', 20)])

• (2, 'Bob', 'add2', [(21, 'cherry', 50), (22, 'peach', 40)])

為不同列分別指定合并順序

在 partial-update 合并機制中，除了 sequence field 之外，您也可以通過 sequence group 為不同列分別指定合并順序。該功能可用於在打寬情境下處理亂序資料，為來自不同源表的列分別指定合并順序。

例如，考慮以下建立 Paimon 表的 Flink SQL 陳述式。

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'
);

a、b 兩列以 g_1 列的值作為合并順序依據（值越大表示資料越新），c、d 兩列以 g_2 列的值作為合并順序依據。

同時進行資料的打寬與彙總

在 partial-update 合并機制中，可以在表參數中設定 fields.<field-name>.aggregate-function，為 <field-name> 列指定彙總函式，對該列的值進行彙總。

例如，考慮以下建立 Paimon 表的 Flink SQL 陳述式。

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'
);

a、b 兩列以 g_1 列的值作為合并順序依據（值越大表示資料越新），其中 a 列保留最新的非 null 值（未指定，預設使用last_non_null_value彙總函式），b 列保留輸入的最大值。c、d 兩列以 g_2 列的值作為合并順序依據，其中 c 列保留最新的非 null 值，d 列求出輸入資料的和。