SQL Functions

Apache Fluss registers a set of built-in RoaringBitmap SQL functions in FlussCatalog. These are Flink-side functions that execute within the Flink query engine. They are distinct from the storage-level rbm32 / rbm64 aggregators, which run inside the Fluss TabletServer during write.

How to Use

Create a Fluss catalog first. Its built-in functions are available without any CREATE TEMPORARY FUNCTION statement.

CREATE CATALOG fluss_catalog WITH (
    'type'              = 'fluss',
    'bootstrap.servers' = 'localhost:9123'
);

You can either make the Fluss catalog current or reference its functions by fully qualified name.

Current Fluss Catalog

Switch to the Fluss catalog when the session mainly works with Fluss objects. Function names can then be used directly.

USE CATALOG fluss_catalog;

SELECT rb_cardinality(rb_build(ARRAY[1, 2, 3, 2]));
-- Output: 3

Fully Qualified Names

When working across multiple catalogs, keep the current catalog unchanged and qualify each Fluss function as <catalog>.<database>.<function>.

SELECT fluss_catalog.fluss.rb_cardinality(
    fluss_catalog.fluss.rb_build(ARRAY[1, 2, 3, 2])
);
-- Output: 3

fluss is the default database of FlussCatalog. You can replace it with another Fluss database name, but that database must already exist and the current user must have permission to access it.

All functions operate on BYTES columns containing standard 32-bit RoaringBitmap serialized data, the same wire format used by the rbm32 storage-level aggregator.

Example: Windowed Bitmap Batches

RoaringBitmap SQL functions and the rbm32 merge engine complement each other. A common pattern is to use rb_build_agg to turn each short processing-time window into a bitmap, write those bitmap batches to a Fluss primary-key table, and let the rbm32 merge engine union batches with the same key. Downstream queries can use rb_cardinality to read the cumulative distinct count.

The following example assumes that the Fluss catalog contains an append-only streaming table named click_events with columns page_id BIGINT, user_id INT, and proc_time AS PROCTIME().

The page_uv table uses the Aggregation Merge Engine by setting table.merge-engine=aggregation. The fields.uv_bitmap.agg=rbm32 option enables rbm32 aggregation on uv_bitmap, which unions the serialized 32-bit RoaringBitmap values written for the same primary key.

USE CATALOG fluss_catalog;

-- Store one cumulative bitmap per page in 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'
);

-- Build one bitmap per page every five seconds. Each completed window writes
-- another bitmap batch, and rbm32 unions it into the existing page bitmap.
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;

-- Query the UV value on pages
SELECT page_id,
       rb_cardinality(uv_bitmap) AS uv
FROM page_uv
WHERE page_id = 1;

---

Scalar Functions

Scalar functions operate on a single row and return a single value.

rb_build

Builds a serialized RoaringBitmap from an ARRAY<INT> within a single row.

• Signature: rb_build(values ARRAY<INT>) → BYTES
• Null Handling: Returns NULL if the array argument is NULL. Null elements within the array are ignored. An empty or all-null element array returns an empty bitmap.

SELECT rb_cardinality(rb_build(ARRAY[1, 2, 3, 2]));
-- Output: 3  (duplicate 2 ignored)

SELECT rb_cardinality(rb_build(ARRAY[CAST(NULL AS INT), 1, 2]));
-- Output: 2  (null element ignored)

SELECT rb_build(CAST(NULL AS ARRAY<INT>)) IS NULL;
-- Output: TRUE

---

rb_cardinality

Returns the number of distinct integers in a serialized RoaringBitmap.

• Signature: rb_cardinality(bitmap BYTES) → BIGINT
• Null Handling: Returns NULL for a null input. Returns 0 for an empty bitmap.

SELECT rb_cardinality(rb_build(ARRAY[1, 2, 3, 2]));
-- Output: 3

SELECT rb_cardinality(rb_build(ARRAY[CAST(NULL AS INT)]));
-- Output: 0  (empty bitmap)

---

rb_contains

Returns whether a serialized RoaringBitmap contains a specific integer.

• Signature: rb_contains(bitmap BYTES, value INT) → BOOLEAN
• Null Handling: Returns NULL if either argument is NULL.

SELECT rb_contains(rb_build(ARRAY[1, 2, 3]), 2);
-- Output: TRUE

SELECT rb_contains(rb_build(ARRAY[1, 2, 3]), 5);
-- Output: FALSE

---

rb_to_array

Converts a serialized RoaringBitmap to an ARRAY<INT> in ascending order.

• Signature: rb_to_array(bitmap BYTES) → ARRAY<INT>
• Null Handling: Returns NULL for a null input. Returns an empty array for an empty bitmap.

SELECT rb_to_array(rb_build(ARRAY[3, 1, 2]));
-- Output: [1, 2, 3]  (ascending order)

---

rb_or

Returns the bitwise OR (union) of two serialized RoaringBitmap values.

• Signature: rb_or(left BYTES, right BYTES) → BYTES
• Null Handling: Returns NULL if either argument is NULL. To union bitmaps while ignoring nulls across rows, use rb_or_agg.

SELECT rb_cardinality(rb_or(rb_build(ARRAY[1, 2]), rb_build(ARRAY[2, 3])));
-- Output: 3  ({1, 2, 3})

---

rb_and

Returns the bitwise AND (intersection) of two serialized RoaringBitmap values.

• Signature: rb_and(left BYTES, right BYTES) → BYTES
• Null Handling: Returns NULL if either argument is NULL. Returns an empty serialized bitmap (not NULL) when the intersection is empty.

SELECT rb_cardinality(rb_and(rb_build(ARRAY[1, 2, 3]), rb_build(ARRAY[2, 3, 4])));
-- Output: 2  ({2, 3})

SELECT rb_cardinality(rb_and(rb_build(ARRAY[1, 2]), rb_build(ARRAY[3, 4])));
-- Output: 0  (disjoint sets)

---

rb_xor

Returns the bitwise XOR (symmetric difference) of two serialized RoaringBitmap values — elements present in exactly one of the two inputs.

• Signature: rb_xor(left BYTES, right BYTES) → BYTES
• Null Handling: Returns NULL if either argument is NULL. Returns an empty serialized bitmap (not NULL) when the two inputs are identical.

SELECT rb_cardinality(rb_xor(rb_build(ARRAY[1, 2, 3]), rb_build(ARRAY[2, 3, 4])));
-- Output: 2  ({1, 4})

SELECT rb_cardinality(rb_xor(rb_build(ARRAY[1, 2]), rb_build(ARRAY[1, 2])));
-- Output: 0  (identical inputs cancel)

---

rb_andnot

Returns elements present in the left bitmap but not in the right bitmap.

• Signature: rb_andnot(left BYTES, right BYTES) → BYTES
• Null Handling: Returns NULL if either argument is NULL. Returns an empty serialized bitmap (not NULL) when the right bitmap is a superset of the left.

SELECT rb_cardinality(rb_andnot(rb_build(ARRAY[1, 2, 3, 4]), rb_build(ARRAY[3, 4, 5])));
-- Output: 2  ({1, 2})

-- Users who visited page A but not page B
SELECT rb_cardinality(rb_andnot(a.uv_bitmap, b.uv_bitmap)) AS exclusive_visitors
FROM uv_agg a, uv_agg b
WHERE a.page_id = 1 AND b.page_id = 2 AND a.ymd = b.ymd;

---

Aggregate Functions

Aggregate functions reduce multiple rows into a single bitmap result.

rb_build_agg

Builds a serialized RoaringBitmap from a column of INT values across rows.

• Signature: rb_build_agg(value INT) → BYTES
• Null Handling: Null inputs are ignored. Returns NULL if all inputs are null.

SELECT rb_cardinality(rb_build_agg(user_id)) AS uv
FROM (VALUES (1), (2), (3), (2)) AS t(user_id);
-- Output: 3  (distinct users)

---

rb_or_agg

Unions multiple serialized RoaringBitmap values via bitwise OR across rows.

• Signature: rb_or_agg(bitmap BYTES) → BYTES
• Null Handling: Null and empty inputs are ignored. Returns NULL if all inputs are null.

-- Roll up per-day bitmaps into a weekly unique visitor count
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);
-- Output: 3  (users {1, 2, 3} across both days)

---

rb_and_agg

Intersects multiple serialized RoaringBitmap values via bitwise AND across rows.

• Signature: rb_and_agg(bitmap BYTES) → BYTES
• Null Handling: Null and empty inputs are ignored. Returns NULL if the intersection is empty or all inputs are null.

-- Find users who appeared on every day
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);
-- Output: 1  (only user 2 appeared on both days)

note

rb_and_agg has no server-side counterpart and executes entirely in Flink. Avoid combining with table.merge-engine=aggregation on append-only streams.

---

rb_xor_agg

Aggregates multiple serialized RoaringBitmap values via bitwise XOR across rows. Returns elements that appear in an odd number of input bitmaps.

• Signature: rb_xor_agg(bitmap BYTES) → BYTES
• Null Handling: Null and empty inputs are ignored. Returns NULL only when no non-null input has been accumulated (i.e. net count is zero). Returns an empty serialized bitmap when inputs cancel (e.g. two identical bitmaps XOR to empty) as long as at least one non-null input remains.

-- Find users who appeared on an odd number of days
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);
-- Output: 2  (users {1, 3} each appeared on exactly one day)

note

rb_xor_agg has no server-side counterpart and executes entirely in Flink. Unlike rb_and_agg, it supports retraction on retractable streams (XOR is self-inverse).

---

tip

For a full end-to-end tutorial including Docker setup and multi-dimensional roll-up queries, see the Real-Time UV Deduplication blog post.