import Tabs from ‘@theme/Tabs’; import TabItem from ‘@theme/TabItem’;
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.
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.
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
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.
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 operate on a single row and return a single value.
Builds a serialized RoaringBitmap from an ARRAY<INT> within a single row.
rb_build(values ARRAY<INT>) → BYTESNULL 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
Returns the number of distinct integers in a serialized RoaringBitmap.
rb_cardinality(bitmap BYTES) → BIGINTNULL 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)
Returns whether a serialized RoaringBitmap contains a specific integer.
rb_contains(bitmap BYTES, value INT) → BOOLEANNULL 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
Converts a serialized RoaringBitmap to an ARRAY<INT> in ascending order.
rb_to_array(bitmap BYTES) → ARRAY<INT>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)
Returns the bitwise OR (union) of two serialized RoaringBitmap values.
rb_or(left BYTES, right BYTES) → BYTESNULL 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})
Returns the bitwise AND (intersection) of two serialized RoaringBitmap values.
rb_and(left BYTES, right BYTES) → BYTESNULL 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)
Returns the bitwise XOR (symmetric difference) of two serialized RoaringBitmap values — elements present in exactly one of the two inputs.
rb_xor(left BYTES, right BYTES) → BYTESNULL 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)
Returns elements present in the left bitmap but not in the right bitmap.
rb_andnot(left BYTES, right BYTES) → BYTESNULL 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 reduce multiple rows into a single bitmap result.
Builds a serialized RoaringBitmap from a column of INT values across rows.
rb_build_agg(value INT) → BYTESNULL 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)
Unions multiple serialized RoaringBitmap values via bitwise OR across rows.
rb_or_agg(bitmap BYTES) → BYTESNULL 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)
Intersects multiple serialized RoaringBitmap values via bitwise AND across rows.
rb_and_agg(bitmap BYTES) → BYTESNULL 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. :::
Aggregates multiple serialized RoaringBitmap values via bitwise XOR across rows. Returns elements that appear in an odd number of input bitmaps.
rb_xor_agg(bitmap BYTES) → BYTESNULL 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. :::