SQL 函數 - Realtime Compute for Apache Flink

本文介紹 Fluss Catalog 內建的 RoaringBitmap SQL 函數，包括函數列表、簽名、空值語義與典型用法。

功能說明

RoaringBitmap SQL 函數由 Fluss Catalog 註冊，在 Flink 查詢引擎內執行，用於位元影像的構建、集合運算與基數統計，適用於即時 UV、使用者留存與人群圈選等精確去重情境。

這類函數與儲存層的 rbm32、rbm64 彙總器不同：

對比項	RoaringBitmap SQL 函數	rbm32 / rbm64 彙總器
執行位置	Flink 查詢引擎	Fluss TabletServer
觸發方式	SQL 中顯式調用	主鍵表寫入合并時自動執行
作用範圍	單行位元影像運算與跨行位元影像彙總	相同主鍵的多批次位元影像並集

兩者配合使用：SQL 函數在 Flink 側產生位元影像，rbm32 彙總器在服務端按主鍵累積合并。合并引擎詳情請參見Aggregation Merge Engine。

前提條件

已建立 Fluss 叢集，具體操作請參見開通流儲存 Apache Fluss 版。
已在 Flink 工作空間註冊 Fluss Catalog，具體操作請參見管理 Fluss Catalog。

使用限制

函數隨 Fluss Catalog 註冊，不是 Flink 引擎內建函數。在 Fluss Catalog 之外的作業中調用時，需使用全限定名。
全部函數基於 32 位 RoaringBitmap，位元影像值使用 BYTES 類型表示，序列化格式與 rbm32 彙總器一致，不支援 rbm64 彙總器使用的 64 位元 RoaringBitmap 資料。各函數的其他輸入和傳回型別有所不同，具體以函數簽名為準BYTES 類型，序列化格式與 rbm32 彙總器一致，不適用於 rbm64 彙總器寫入的 64 位元 RoaringBitmap 資料。
rb_and_agg、rb_xor_agg 無服務端對應實現，全部在 Flink 側計算。
本文函數由 Fluss Catalog 提供，使用 BYTES 表示位元影像；Realtime Compute Flink 版的 BITMAP 位元影像函數由引擎提供，使用 BITMAP 類型。兩類函數的位元影像參數不能直接互傳。在支援相應轉換函式的引擎版本中，可通過 BITMAP_FROM_BYTES 和 BITMAP_TO_BYTES，在標準 32 位 RoaringBitmap 位元組資料與 BITMAP 類型之間顯式轉換。當前 Catalog 不是 Fluss Catalog 時，可通過全限定名調用本文函數。

說明

RoaringBitmap SQL 函數由 Flink 作業載入的 Fluss Connector 中的 Catalog 實現提供。若提示函數不存在，請先確認當前 Connector 版本是否包含這些函數，並檢查當前 Catalog、資料庫及函數全限定名是否正確。僅升級 Fluss 叢集服務端不能保證這些函數可用，請參見叢集升級說明。

函數列表

純量涵式

函數	簽名	說明
`rb_build`	`rb_build(values ARRAY<INT>) → BYTES`	由單行的 `ARRAY<INT>` 構建序列化位元影像。
`rb_cardinality`	`rb_cardinality(bitmap BYTES) → BIGINT`	返回位元影像中不同整數的個數。
`rb_contains`	`rb_contains(bitmap BYTES, value INT) → BOOLEAN`	判斷位元影像是否包含指定整數。
`rb_to_array`	`rb_to_array(bitmap BYTES) → ARRAY<INT>`	將位元影像轉換為升序 `ARRAY<INT>`。
`rb_or`	`rb_or(left BYTES, right BYTES) → BYTES`	求兩個位元影像的並集。
`rb_and`	`rb_and(left BYTES, right BYTES) → BYTES`	求兩個位元影像的交集。
`rb_xor`	`rb_xor(left BYTES, right BYTES) → BYTES`	求兩個位元影像的對稱差集。
`rb_andnot`	`rb_andnot(left BYTES, right BYTES) → BYTES`	求左位元影像減去右位元影像的差集。

彙總函式

函數	簽名	說明
`rb_build_agg`	`rb_build_agg(value INT) → BYTES`	跨行將 `INT` 列彙總為位元影像。
`rb_or_agg`	`rb_or_agg(bitmap BYTES) → BYTES`	跨行按位或合并位元影像。
`rb_and_agg`	`rb_and_agg(bitmap BYTES) → BYTES`	跨行按位與合并位元影像。
`rb_xor_agg`	`rb_xor_agg(bitmap BYTES) → BYTES`	跨行按位異或合并位元影像。

調用方式

註冊 Fluss Catalog 後，內建函數即可直接調用，無需執行 CREATE TEMPORARY FUNCTION。

切換至 Fluss Catalog

會話主要操作 Fluss 對象時，切換當前 Catalog 後可直接使用函數名。

USE CATALOG `fluss_catalog`;

SELECT rb_cardinality(rb_build(ARRAY[1, 2, 3, 2]));
-- 輸出：3

使用全限定名

跨 Catalog 作業時保持當前 Catalog 不變，按 <catalog>.<database>.<function> 格式調用函數。

SELECT `fluss_catalog`.`fluss`.rb_cardinality(
    `fluss_catalog`.`fluss`.rb_build(ARRAY[1, 2, 3, 2])
);
-- 輸出：3

fluss 為 Fluss Catalog 的預設資料庫，可替換為其他 Fluss 資料庫名。目標資料庫需已存在，且目前使用者具備存取權限。

情境樣本：視窗位元影像累積去重

統計各頁面的累計 UV。rb_build_agg 將每個處理時間視窗內的使用者識別碼彙總為位元影像並寫入 Fluss 主鍵表，rbm32 彙總器在服務端將相同主鍵的位元影像批次併入累計位元影像，查詢時用 rb_cardinality 讀取累計去重值。

以下樣本假設 Fluss Catalog 中已存在日誌表 click_events，包含 page_id BIGINT、user_id INT 與 proc_time AS PROCTIME() 三列。

USE CATALOG `fluss_catalog`;

-- 每個頁面在 Fluss 中儲存一份累計位元影像
CREATE TABLE page_uv (
    page_id   BIGINT,
    uv_bitmap BYTES,
    PRIMARY KEY (page_id) NOT ENFORCED
) WITH (
    'table.merge-engine'   = 'aggregation',
    'fields.uv_bitmap.agg' = 'rbm32'
);

-- 每 5 秒為每個頁面產生一份位元影像。
-- 每個視窗寫入一個位元影像批次，rbm32 將其併入該頁面已有的位元影像。
INSERT INTO page_uv
SELECT page_id,
       rb_build_agg(user_id) AS uv_bitmap
FROM TABLE(
    TUMBLE(
        TABLE click_events,
        DESCRIPTOR(proc_time),
        INTERVAL '5' SECOND
    )
)
GROUP BY page_id, window_start, window_end;

-- 查詢頁面 UV
SELECT page_id,
       rb_cardinality(uv_bitmap) AS uv
FROM page_uv
WHERE page_id = 1;

'table.merge-engine' = 'aggregation' 啟用合并引擎，'fields.uv_bitmap.agg' = 'rbm32' 指定 uv_bitmap 欄位按 32 位 RoaringBitmap 並集彙總。

純量涵式

純量涵式作用於單行，返回單個值。

rb_build

由單行的 ARRAY<INT> 構建序列化 RoaringBitmap。

簽名：rb_build(values ARRAY<INT>) → BYTES
空值處理：入參數組為 NULL 時返回 NULL。數組內的 NULL 元素被忽略。空數組或元素全為 NULL 的數組返回空位元影像。

SELECT rb_cardinality(rb_build(ARRAY[1, 2, 3, 2]));
-- 輸出：3，重複值 2 被忽略

SELECT rb_cardinality(rb_build(ARRAY[CAST(NULL AS INT), 1, 2]));
-- 輸出：2，NULL 元素被忽略

SELECT rb_build(CAST(NULL AS ARRAY<INT>)) IS NULL;
-- 輸出：TRUE

rb_cardinality

返回序列化 RoaringBitmap 中不同整數的個數。

簽名：rb_cardinality(bitmap BYTES) → BIGINT
空值處理：入參為 NULL 時返回 NULL。空位元影像返回 0。

SELECT rb_cardinality(rb_build(ARRAY[1, 2, 3, 2]));
-- 輸出：3

SELECT rb_cardinality(rb_build(ARRAY[CAST(NULL AS INT)]));
-- 輸出：0，空位元影像

rb_contains

判斷序列化 RoaringBitmap 是否包含指定整數。

簽名：rb_contains(bitmap BYTES, value INT) → BOOLEAN
空值處理：任一入參為 NULL 時返回 NULL。

SELECT rb_contains(rb_build(ARRAY[1, 2, 3]), 2);
-- 輸出：TRUE

SELECT rb_contains(rb_build(ARRAY[1, 2, 3]), 5);
-- 輸出：FALSE

rb_to_array

將序列化 RoaringBitmap 轉換為升序 ARRAY<INT>。

簽名：rb_to_array(bitmap BYTES) → ARRAY<INT>
空值處理：入參為 NULL 時返回 NULL。空位元影像返回空數組。

SELECT rb_to_array(rb_build(ARRAY[3, 1, 2]));
-- 輸出：[1, 2, 3]，升序排列

rb_or

求兩個序列化 RoaringBitmap 的並集（按位或）。

簽名：rb_or(left BYTES, right BYTES) → BYTES
空值處理：任一入參為 NULL 時返回 NULL。需跨行忽略空值求並集時，使用 rb_or_agg。

SELECT rb_cardinality(rb_or(rb_build(ARRAY[1, 2]), rb_build(ARRAY[2, 3])));
-- 輸出：3，即 {1, 2, 3}

rb_and

求兩個序列化 RoaringBitmap 的交集（按位與）。

簽名：rb_and(left BYTES, right BYTES) → BYTES
空值處理：任一入參為 NULL 時返回 NULL。交集為空白時返回空位元影像，不返回 NULL。

SELECT rb_cardinality(rb_and(rb_build(ARRAY[1, 2, 3]), rb_build(ARRAY[2, 3, 4])));
-- 輸出：2，即 {2, 3}

SELECT rb_cardinality(rb_and(rb_build(ARRAY[1, 2]), rb_build(ARRAY[3, 4])));
-- 輸出：0，兩個集合無交集

rb_xor

求兩個序列化 RoaringBitmap 的對稱差集（按位異或），即僅出現在其中一個入參中的元素。

簽名：rb_xor(left BYTES, right BYTES) → BYTES
空值處理：任一入參為 NULL 時返回 NULL。兩個入參完全相同時返回空位元影像，不返回 NULL。

SELECT rb_cardinality(rb_xor(rb_build(ARRAY[1, 2, 3]), rb_build(ARRAY[2, 3, 4])));
-- 輸出：2，即 {1, 4}

SELECT rb_cardinality(rb_xor(rb_build(ARRAY[1, 2]), rb_build(ARRAY[1, 2])));
-- 輸出：0，相同入參相互抵消

rb_andnot

求出現在左位元影像但不出現在右位元影像中的元素。

簽名：rb_andnot(left BYTES, right BYTES) → BYTES
空值處理：任一入參為 NULL 時返回 NULL。右位元影像是左位元影像的超集時返回空位元影像，不返回 NULL。

SELECT rb_cardinality(rb_andnot(rb_build(ARRAY[1, 2, 3, 4]), rb_build(ARRAY[3, 4, 5])));
-- 輸出：2，即 {1, 2}

-- 訪問過頁面 1 但未訪問過頁面 2 的使用者數
SELECT rb_cardinality(rb_andnot(a.uv_bitmap, b.uv_bitmap)) AS exclusive_visitors
FROM page_uv a, page_uv b
WHERE a.page_id = 1 AND b.page_id = 2;

彙總函式

彙總函式將多行歸併為單個位元影像結果。

rb_build_agg

跨行將 INT 列彙總為序列化 RoaringBitmap。

簽名：rb_build_agg(value INT) → BYTES
空值處理：忽略 NULL 入參。入參全為 NULL 時返回 NULL。

SELECT rb_cardinality(rb_build_agg(user_id)) AS uv
FROM (VALUES (1), (2), (3), (2)) AS t(user_id);
-- 輸出：3，去重後的使用者數

rb_or_agg

跨行按位或合并多個序列化 RoaringBitmap。

簽名：rb_or_agg(bitmap BYTES) → BYTES
空值處理：忽略 NULL 與空位元影像入參。入參全為 NULL 時返回 NULL。

-- 將按天統計的位元影像匯總為周 UV
SELECT rb_cardinality(rb_or_agg(daily_bitmap)) AS weekly_uv
FROM (
    VALUES
        (1, rb_build(ARRAY[1, 2])),
        (2, rb_build(ARRAY[2, 3]))
) AS t(day_id, daily_bitmap);
-- 輸出：3，兩天合計使用者 {1, 2, 3}

rb_and_agg

跨行按位與合并多個序列化 RoaringBitmap。

簽名：rb_and_agg(bitmap BYTES) → BYTES
空值處理：忽略 NULL 與空位元影像入參。交集為空白或入參全為 NULL 時返回 NULL。

-- 找出每天都出現的使用者
SELECT rb_cardinality(rb_and_agg(daily_bitmap)) AS retained_users
FROM (
    VALUES
        (1, rb_build(ARRAY[1, 2])),
        (2, rb_build(ARRAY[2, 3]))
) AS t(day_id, daily_bitmap);
-- 輸出：1，僅使用者 2 兩天都出現

說明

rb_and_agg 無服務端對應實現，全部在 Flink 側執行。在僅追加（Append-Only）流上，避免與 'table.merge-engine' = 'aggregation' 組合使用。

rb_xor_agg

跨行按位異或合并多個序列化 RoaringBitmap，返回出現次數為奇數的元素。

簽名：rb_xor_agg(bitmap BYTES) → BYTES
空值處理：忽略 NULL 與空位元影像入參。僅在未累積任何非 NULL 入參時返回 NULL。只要存在至少一個非 NULL 入參，入參相互抵消時返回空位元影像。

-- 找出出現天數為奇數的使用者
SELECT rb_cardinality(rb_xor_agg(daily_bitmap)) AS changed_users
FROM (
    VALUES
        (1, rb_build(ARRAY[1, 2])),
        (2, rb_build(ARRAY[2, 3]))
) AS t(day_id, daily_bitmap);
-- 輸出：2，使用者 {1, 3} 各只出現一天

說明

rb_xor_agg 無服務端對應實現，全部在 Flink 側執行。與 rb_and_agg 不同，異或運算自逆，因此支援可回撤流上的回撤語義。