Aggregation Merge Engine - Realtime Compute for Apache Flink

本文介紹合并引擎（Merge Engine ）中的Aggregation Merge Engine用法。

Aggregation Merge Engine說明

合并策略

對於相同主鍵進行分組，並根據指定的彙總函式，對每個非主鍵欄位逐一進行計算
配置方式

在建立表時，通過設定表屬性'table.merge-engine' = 'aggregation'來啟用。
使用方式
- 使用控制台建立表時，選擇對應的合并策略後添加進階配置 'fields.<欄位名>.agg' = '<函數名>' 為各非主鍵欄位指定具體的彙總函式。
- 對於需要附加參數的彙總函式（如 listagg 的自訂分隔字元）。
```
'fields.<欄位名>.agg' = '<函數名>',
'fields.<欄位名>.<函數名>.<參數名>' = '<參數值>'
```
  若未為某欄位指定彙總函式，系統將預設使用 last_value_ignore_nulls 作為彙總行為。
適用情境
- 計算累計總和與統計指標
- 維護計數器與度量值
- 跟蹤最大/最小值變化
- 構建即時儀錶盤與分析
使用限制
- 不支援回撤（Retraction）語義。引擎採用單向彙總邏輯，無法對已彙總的結果執行反向操作（例如，無法從 sum 結果中減去被刪除的值，也無法撤銷 max 以恢複次大值）。
- 對 DELETE 操作支援受限。引擎預設靜默忽略刪除操作（'table.delete.behavior' = 'ignore'）。
- 若需更改刪除行為，可通過配置將其設定為拒絕並報錯（disable），或允許直接物理移除整行記錄（allow）。
```
'table.delete.behavior' = 'ignore'   -- 預設值，靜默忽略刪除操作
'table.delete.behavior' = 'disable'  -- 拒絕刪除操作並返回錯誤
'table.delete.behavior' = 'allow'    -- 允許刪除操作
```

使用樣本

CREATE TABLE product_stats (
    product_id BIGINT,
    price DOUBLE,
    sales BIGINT,
    last_update_time TIMESTAMP(3),
    PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.price.agg' = 'max',
    'fields.sales.agg' = 'sum'
    -- last_update_time 未指定彙總函式，預設使用 last_value_ignore_nulls
);

-- 插入資料，相同主鍵的記錄會自動彙總
INSERT INTO product_stats VALUES
    (1, 23.0, 15, TIMESTAMP '2024-01-01 10:00:00'),
    (1, 30.2, 20, TIMESTAMP '2024-01-01 11:00:00');

SELECT * FROM product_stats WHERE product_id = 1;
-- 輸出:
-- +------------+-------+-------+---------------------+
-- | product_id | price | sales | last_update_time    |
-- +------------+-------+-------+---------------------+
-- |          1 |  30.2 |    35 | 2024-01-01 11:00:00 |
-- +------------+-------+-------+---------------------+
-- price: 30.2（23.0 和 30.2 的最大值）
-- sales: 35（15 和 20 的累加和）
-- last_update_time: 2024-01-01 11:00:00（最後一個非空值）

支援的彙總函式

sum

計算多行資料的總和。

支援類型：TINYINT、SMALLINT、INT、BIGINT、FLOAT、DOUBLE、DECIMAL。
空值處理：忽略 null 值。

CREATE TABLE test_sum (
    id BIGINT,
    amount DECIMAL(10, 2),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.amount.agg' = 'sum'
);

INSERT INTO test_sum VALUES (1, 100.50), (1, 200.75);
SELECT * FROM test_sum WHERE id = 1;
-- 輸出:
-- +----+---------+
-- | id | amount  |
-- +----+---------+
-- |  1 | 301.25  |
-- +----+---------+

product

計算多行資料的乘積。

支援類型：TINYINT、SMALLINT、INT、BIGINT、FLOAT、DOUBLE、DECIMAL。
空值處理：忽略 null 值。

CREATE TABLE test_product (
    id BIGINT,
    discount_factor DOUBLE,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.discount_factor.agg' = 'product'
);

INSERT INTO test_product VALUES (1, 0.9), (1, 0.8);
SELECT * FROM test_product WHERE id = 1;
-- 輸出:
-- +----+-----------------------+
-- | id | discount_factor       |
-- +----+-----------------------+
-- |  1 | 0.7200000000000001    |
-- +----+-----------------------+
-- 注意：由於 IEEE 754 雙精確度浮點運算，結果為 0.7200000000000001 而非 0.72
-- 如需精確計算，建議使用 DECIMAL 類型替代 DOUBLE

max

保留最大值。

支援類型：CHAR、STRING、TINYINT、SMALLINT、INT、BIGINT、FLOAT、DOUBLE、DECIMAL、DATE、TIME、TIMESTAMP、TIMESTAMP_LTZ。
空值處理：忽略 null 值。

CREATE TABLE test_max (
    id BIGINT,
    temperature DOUBLE,
    reading_time TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.temperature.agg' = 'max',
    'fields.reading_time.agg' = 'max'
);

INSERT INTO test_max VALUES
    (1, 25.5, TIMESTAMP '2024-01-01 10:00:00'),
    (1, 28.3, TIMESTAMP '2024-01-01 11:00:00');
SELECT * FROM test_max WHERE id = 1;
-- 輸出:
-- +----+-------------+---------------------+
-- | id | temperature | reading_time        |
-- +----+-------------+---------------------+
-- |  1 |        28.3 | 2024-01-01 11:00:00 |
-- +----+-------------+---------------------+

min

保留最小值。

支援類型：CHAR、STRING、TINYINT、SMALLINT、INT、BIGINT、FLOAT、DOUBLE、DECIMAL、DATE、TIME、TIMESTAMP、TIMESTAMP_LTZ。
空值處理：忽略 null 值。

CREATE TABLE test_min (
    id BIGINT,
    lowest_price DECIMAL(10, 2),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.lowest_price.agg' = 'min'
);

INSERT INTO test_min VALUES (1, 99.99), (1, 79.99), (1, 89.99);
SELECT * FROM test_min WHERE id = 1;
-- 輸出:
-- +----+--------------+
-- | id | lowest_price |
-- +----+--------------+
-- |  1 |        79.99 |
-- +----+--------------+

last_value

用最新接收到的值覆蓋舊值。

支援類型：所有資料類型。
空值處理：null 值會直接覆蓋舊值。

CREATE TABLE test_last_value (
    id BIGINT,
    status STRING,
    last_login TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.status.agg' = 'last_value',
    'fields.last_login.agg' = 'last_value'
);

INSERT INTO test_last_value VALUES (1, 'online', TIMESTAMP '2024-01-01 10:00:00');
INSERT INTO test_last_value VALUES (1, 'offline', TIMESTAMP '2024-01-01 11:00:00');
INSERT INTO test_last_value VALUES (1, CAST(NULL AS STRING), TIMESTAMP '2024-01-01 12:00:00');
-- null 會覆蓋之前的 'offline' 值
SELECT * FROM test_last_value WHERE id = 1;
-- 輸出:
-- +----+--------+---------------------+
-- | id | status | last_login          |
-- +----+--------+---------------------+
-- |  1 | null   | 2024-01-01 12:00:00 |
-- +----+--------+---------------------+

last_value_ignore_nulls

用最新的非空值替換之前的值。這是未指定彙總函式時的預設彙總行為。

支援類型：所有資料類型。
空值處理：忽略 null 值，保留之前的值。

CREATE TABLE test_last_value_ignore_nulls (
    id BIGINT,
    email STRING,
    phone STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.email.agg' = 'last_value_ignore_nulls',
    'fields.phone.agg' = 'last_value_ignore_nulls'
);

INSERT INTO test_last_value_ignore_nulls VALUES (1, 'user@example.com', '123-456');
INSERT INTO test_last_value_ignore_nulls VALUES (1, CAST(NULL AS STRING), '789-012');
-- email 為 null，被忽略，保留之前的值
INSERT INTO test_last_value_ignore_nulls VALUES (1, 'new@example.com', CAST(NULL AS STRING));
-- phone 為 null，被忽略，保留之前的值
SELECT * FROM test_last_value_ignore_nulls WHERE id = 1;
-- 輸出:
-- +----+-----------------+---------+
-- | id | email           | phone   |
-- +----+-----------------+---------+
-- |  1 | new@example.com | 789-012 |
-- +----+-----------------+---------+

last_n

與 last_value 僅保留最新單個值不同，last_n 將每次寫入的新資料追加到數組中，並僅保留最近 N 個元素，適合直接在表中維護使用者最近 N 次操作、裝置最近 N 條上報記錄等“最近 N 條”資料，無需在作業側額外開發視窗或過濾邏輯。

支援類型：ARRAY 類型，元素類型不限。
空值處理：null 值不會改變欄位現有內容。
注意：需通過 fields.<列名>.last_n.size 指定保留個數，必填且為正整數，缺失時建表失敗；元素按寫入到達服務的順序保留；每次更新會重寫整個儲存的數組。

CREATE TABLE test_last_n (
    id BIGINT,
    recent_events ARRAY<STRING>,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.recent_events.agg' = 'last_n',
    'fields.recent_events.last_n.size' = '3'
);

INSERT INTO test_last_n VALUES (1, ARRAY['search']);
INSERT INTO test_last_n VALUES (1, ARRAY['view']);
INSERT INTO test_last_n VALUES (1, ARRAY['add_to_cart']);
INSERT INTO test_last_n VALUES (1, ARRAY['pay']);
SELECT * FROM test_last_n WHERE id = 1;
-- 輸出:
-- +----+--------------------------+
-- | id | recent_events            |
-- +----+--------------------------+
-- |  1 | [view, add_to_cart, pay] |
-- +----+--------------------------+

first_value

保留欄位的第一個值，忽略後續所有值。

支援類型：所有資料類型。
空值處理：如果第一個值為 null，則保留 null。

CREATE TABLE test_first_value (
    id BIGINT,
    first_purchase_date DATE,
    first_product STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.first_purchase_date.agg' = 'first_value',
    'fields.first_product.agg' = 'first_value'
);

INSERT INTO test_first_value VALUES (1, '2024-01-01', 'ProductA'), (1, '2024-02-01', 'ProductB');
-- 第二條記錄被忽略，保留第一個值
SELECT * FROM test_first_value WHERE id = 1;
-- 輸出:
-- +----+---------------------+---------------+
-- | id | first_purchase_date | first_product |
-- +----+---------------------+---------------+
-- |  1 | 2024-01-01          | ProductA      |
-- +----+---------------------+---------------+

listagg

將多個字串值用分隔字元拼接為一個字串。

支援類型：STRING、CHAR。
空值處理：跳過 null 值。
分隔字元：預設為逗號 ,，可通過參數自訂。

CREATE TABLE test_listagg (
    id BIGINT,
    tags1 STRING,
    tags2 STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.tags1.agg' = 'listagg',
    'fields.tags2.agg' = 'listagg',
    'fields.tags2.listagg.delimiter' = ';'   -- 指定分隔字元
);

INSERT INTO test_listagg VALUES (1, 'developer', 'developer'), (1, 'java', 'java'), (1, 'flink', 'flink');
SELECT * FROM test_listagg WHERE id = 1;
-- 輸出:
-- +----+----------------------+----------------------+
-- | id | tags1                | tags2                |
-- +----+----------------------+----------------------+
-- |  1 | developer,java,flink | developer;java;flink |
-- +----+----------------------+----------------------+

string_agg

listagg 的別名，功能完全相同。將多個字串值用分隔字元拼接為一個字串。

支援的類型：STRING、CHAR。
空值處理：跳過 null 值。
分隔字元：預設為逗號,，可通過參數自訂。

CREATE TABLE test_string_agg (
    id BIGINT,
    tags STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.tags.agg' = 'string_agg',
    'fields.tags.string_agg.delimiter' = ';'   -- 指定分隔字元
);

INSERT INTO test_string_agg VALUES (1, 'developer'), (1, 'java'), (1, 'flink');
SELECT * FROM test_string_agg WHERE id = 1;
-- 輸出:
-- +----+----------------------+
-- | id | tags                 |
-- +----+----------------------+
-- |  1 | developer;java;flink |
-- +----+----------------------+

rbm32

對序列化的 32 位 RoaringBitmap 值進行並集彙總。

支援類型：BYTES。
空值處理：忽略 null 值。
注意：位元影像值使用 RoaringBitmap 庫的標準 32 位序列化格式。在 Flink SQL 中可直接調用 rb_build、rb_build_agg 等函數產生位元影像，無需在用戶端預先序列化，詳情請參見SQL 函數。

CREATE TABLE user_visits (
    user_id BIGINT,
    visit_bitmap BYTES,
    PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.visit_bitmap.agg' = 'rbm32'
);

-- 方式一：在 Flink SQL 中調用 rb_build 構建位元影像
INSERT INTO user_visits SELECT 1, rb_build(ARRAY[1, 2]);
INSERT INTO user_visits SELECT 1, rb_build(ARRAY[2, 3]);

-- 方式二：插入用戶端預先序列化的位元影像（十六進位字面量）
-- Bitmap {1,2}
INSERT INTO user_visits VALUES (1, x'3A30000001000000000001001000000001000200');
-- Bitmap {2,3}
INSERT INTO user_visits VALUES (1, x'3A30000001000000000001001000000002000300');

SELECT user_id, rb_cardinality(visit_bitmap) AS uv FROM user_visits WHERE user_id = 1;
-- 結果：visit_bitmap 包含並集 {1,2,3}，uv 為 3

rbm64

對序列化的 64 位元 RoaringBitmap 值進行並集彙總。

支援類型：BYTES。
空值處理：忽略 null 值。
注意：Roaring64Bitmap 值需在用戶端預先序列化，使用 RoaringBitmap 庫的 64 位元格式。RoaringBitmap SQL 函數均基於 32 位格式，不適用於 rbm64。

CREATE TABLE session_interactions (
    session_id BIGINT,
    interaction_bitmap BYTES,
    PRIMARY KEY (session_id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.interaction_bitmap.agg' = 'rbm64'
);

-- 插入序列化的 Roaring64Bitmap 值
-- 結果：interaction_bitmap 包含所有插入 bitmap 的並集

bool_and

對布爾值進行邏輯與（AND）彙總，僅當所有值均為 true 時返回 true。

支援類型：BOOLEAN。
空值處理：忽略 null 值。

CREATE TABLE test_bool_and (
    id BIGINT,
    has_all_permissions BOOLEAN,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.has_all_permissions.agg' = 'bool_and'
);

INSERT INTO test_bool_and VALUES (1, true), (1, true), (1, false);
SELECT * FROM test_bool_and WHERE id = 1;
-- 輸出:
-- +----+---------------------+
-- | id | has_all_permissions |
-- +----+---------------------+
-- |  1 | false               |
-- +----+---------------------+

bool_or

對布爾值進行邏輯或（OR）彙總，只要有一個值為 true 即返回 true。

支援類型：BOOLEAN。
空值處理：忽略 null 值。

CREATE TABLE test_bool_or (
    id BIGINT,
    has_any_alert BOOLEAN,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.has_any_alert.agg' = 'bool_or'
);

INSERT INTO test_bool_or VALUES (1, false), (1, false), (1, true);
SELECT * FROM test_bool_or WHERE id = 1;
-- 輸出:
-- +----+---------------+
-- | id | has_any_alert |
-- +----+---------------+
-- |  1 | true          |
-- +----+---------------+