Realtime Compute for Apache Flink:Use BITMAP type for exact deduplication

-- Source: real-time event stream
CREATE TEMPORARY TABLE user_events
(
      user_id     INT
      ,tag        STRING          -- Event tag
      ,event_time TIMESTAMP(3)
      ,WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
)
WITH ('connector' = 'kafka' ...)
;

-- Sink: minute-level UV statistics
CREATE TEMPORARY TABLE minute_uv
(
      ts_date    STRING
      ,ts_hour   STRING
      ,ts_minute STRING
      ,tag       STRING           -- Event tag
      ,minute_uv BIGINT           -- Deduplicated UV count
)
WITH ('connector' = 'jdbc' ...)
;

INSERT INTO minute_uv
SELECT
    DATE_FORMAT(window_start, 'yyyy-MM-dd') AS ts_date,
    DATE_FORMAT(window_start, 'HH')         AS ts_hour,
    DATE_FORMAT(window_start, 'mm')         AS ts_minute,
    tag,
    -- Builds a bitmap from user_ids in the window, then immediately returns
    -- its cardinality (the exact deduplicated count).
    -- Equivalent to COUNT(DISTINCT user_id) but with lower memory overhead.
    BITMAP_BUILD_CARDINALITY_AGG(user_id)   AS uv
FROM TABLE(
    TUMBLE(
        TABLE user_events,
        DESCRIPTOR(event_time),
        INTERVAL '1' MINUTE
    )
)
GROUP BY
    window_start,
    window_end,
    tag;

-- Source: real-time event stream (same schema as Example 1)
CREATE TEMPORARY TABLE user_events
(
      user_id     INT
      ,tag        STRING          -- Event tag
      ,event_time TIMESTAMP(3)
      ,WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
)
WITH ('connector' = 'kafka' ...)
;

-- Sink: minute-level serialized bitmaps
CREATE TEMPORARY TABLE minute_bitmaps
(
      ts_date    STRING
      ,ts_hour   STRING
      ,ts_minute STRING
      ,tag       STRING           -- Event tag
      ,minute_bm BYTES            -- Serialized bitmap (RoaringBitmap binary format)
)
WITH ('connector' = 'jdbc' ...)
;

INSERT INTO minute_bitmaps
SELECT
    DATE_FORMAT(window_start, 'yyyy-MM-dd') AS ts_date,
    DATE_FORMAT(window_start, 'HH')         AS ts_hour,
    DATE_FORMAT(window_start, 'mm')         AS ts_minute,
    tag,
    -- Two-step process:
    -- 1. BITMAP_BUILD_AGG: aggregates user_ids within the tumbling window
    --    into an in-memory bitmap object.
    -- 2. BITMAP_TO_BYTES: encodes the bitmap into RoaringBitmap binary format
    --    for persistent storage.
    BITMAP_TO_BYTES(
        BITMAP_BUILD_AGG(user_id)
    ) AS user_id_bitmap
FROM TABLE(
    TUMBLE(
        TABLE user_events,
        DESCRIPTOR(event_time),
        INTERVAL '1' MINUTE
    )
)
GROUP BY
    window_start,
    window_end,
    tag;

-- Source: pre-computed minute-level bitmaps (produced by Example 2)
CREATE TEMPORARY TABLE minute_bitmaps
(
      ts_date    STRING
      ,ts_hour   STRING
      ,ts_minute STRING
      ,tag       STRING           -- Event tag
      ,minute_bm BYTES            -- Pre-computed serialized minute-level bitmap
)
WITH ('connector' = 'jdbc' ...)
;

-- Sink: hourly filtered UV counts
CREATE TEMPORARY TABLE hour_combined_uv
(
      ts_date  STRING
      ,ts_hour STRING
      ,tag     STRING             -- Metric label
      ,uv      BIGINT
)
WITH ('connector' = 'jdbc' ...)
;

INSERT INTO hour_combined_uv
SELECT
    ts_date,
    ts_hour,
    'A and B andnot C' AS metric_name,
    -- Step 3: Count the elements in the final bitmap.
    BITMAP_CARDINALITY(
        -- Step 2: Set operations -- (Tag A ∩ Tag B) − Tag C.
        BITMAP_ANDNOT(
            BITMAP_AND(hour_bm_a, hour_bm_b),
            hour_bm_c
        )
    ) AS uv
FROM (
    SELECT
        ts_date,
        ts_hour,
        -- Step 1: Deserialize stored BYTES and union all minute-level bitmaps
        -- within the hour into a single hourly bitmap per tag.
        BITMAP_OR_AGG(BITMAP_FROM_BYTES(minute_bm))
            FILTER (WHERE tag = 'A') AS hour_bm_a,

        BITMAP_OR_AGG(BITMAP_FROM_BYTES(minute_bm))
            FILTER (WHERE tag = 'B') AS hour_bm_b,

        BITMAP_OR_AGG(BITMAP_FROM_BYTES(minute_bm))
            FILTER (WHERE tag = 'C') AS hour_bm_c
    FROM minute_bitmaps
    WHERE tag IN ('A', 'B', 'C')
    GROUP BY
        ts_date,
        ts_hour
);

-- Source 1: real-time minute-level bitmaps for today (produced by Example 2)
CREATE TEMPORARY TABLE minute_bitmaps
(
      ts_date    STRING
      ,ts_hour   STRING
      ,ts_minute STRING
      ,tag       STRING           -- Page name
      ,minute_bm BYTES            -- Pre-computed serialized minute-level bitmap
)
WITH ('connector' = 'jdbc' ...)
;

-- Source 2: pre-computed day-level bitmaps from storage (historical data)
CREATE TEMPORARY TABLE daily_bitmaps
(
      ts_date   STRING
      ,tag      STRING            -- Page name
      ,daily_bm BYTES             -- Pre-computed serialized day-level bitmap
)
WITH ('connector' = 'jdbc' ...)
;

-- Sink: real-time retention metrics
CREATE TEMPORARY TABLE realtime_retention_metrics
(
      ts_date         STRING
      ,tag            STRING      -- Page name
      ,retained_users BIGINT      -- Users present both today and yesterday
      ,total_users    BIGINT      -- Total visitors yesterday (retention denominator)
      ,update_time    TIMESTAMP(3)
      ,PRIMARY KEY (ts_date, tag) NOT ENFORCED
)
WITH ('connector' = 'jdbc' ...)
;

INSERT INTO realtime_retention_metrics
SELECT
    T.ts_date,
    T.tag,
    -- Metric 1: retained users -- appeared today AND yesterday.
    BITMAP_CARDINALITY(
        BITMAP_AND(
            T.today_bm,                           -- Real-time bitmap for today
            BITMAP_FROM_BYTES(Y.daily_bm)         -- Yesterday's historical bitmap
        )
    ) AS retained_users,

    -- Metric 2: yesterday's total visitors (retention denominator).
    BITMAP_CARDINALITY(
        BITMAP_FROM_BYTES(Y.daily_bm)
    ) AS total_users,

    CURRENT_TIMESTAMP AS update_time
FROM (
    -- Subquery T: aggregate today's minute-level slices in real time
    -- into a single cumulative bitmap for the current day.
    SELECT
        ts_date,
        tag,
        BITMAP_OR_AGG(BITMAP_FROM_BYTES(minute_bm)) AS today_bm
    FROM minute_bitmaps
    GROUP BY ts_date, tag
) T
-- Stream-batch join: match each real-time record with the previous day's bitmap.
INNER JOIN daily_bitmaps Y
    ON T.tag = Y.tag
    AND CAST(T.ts_date AS DATE) = CAST(Y.ts_date AS DATE) + INTERVAL '1' DAY;