feat(data): add error handling for ODS layer and improve partial data support

Add materialized views to capture parsing errors from browser, location, device, and geo raw staging tables and route them to dedicated error tables in the ODS layer. Refactor DDS refresh logic to handle partial data arrivals where device and geo events may arrive independently by using a unified click_id source with LEFT JOINs. Add TRUNCATE command to prevent duplicate data accumulation in DQ summary table.
2026-02-06 22:10:51 +03:00
parent 6bbb26b9b3
commit e44b988d76
3 changed files with 128 additions and 6 deletions
@@ -65,6 +65,34 @@ SELECT
 FROM stg.browser_raw
 WHERE event_id IS NOT NULL;  -- Filter NULL keys to error table

+CREATE MATERIALIZED VIEW IF NOT EXISTS stg.mv_browser_raw_to_ods_errors
+TO ods.browser_event_errors
+AS
+WITH
+    toUUIDOrNull(JSONExtractString(raw, 'event_id')) AS event_id,
+    parseDateTime64BestEffortOrNull(JSONExtractString(raw, 'event_timestamp'), 6) AS event_ts,
+    toUUIDOrNull(JSONExtractString(raw, 'click_id')) AS click_id,
+    arrayFilter(x -> x != '', [
+        if(event_id IS NULL, 'bad_event_id', ''),
+        if(event_ts IS NULL, 'bad_event_timestamp', ''),
+        if(click_id IS NULL, 'bad_click_id', '')
+    ]) AS parse_errors
+SELECT
+    ingest_ts,
+    kafka_topic,
+    kafka_partition,
+    kafka_offset,
+    kafka_ts,
+    raw,
+    arrayStringConcat(parse_errors, ',') AS error_reason
+FROM stg.browser_raw
+WHERE length(parse_errors) > 0
+    AND (
+        event_id IS NULL
+        OR event_ts IS NULL
+        OR click_id IS NULL
+    );
+
 -- ODS: location_events
 CREATE TABLE IF NOT EXISTS ods.location_event
 (
@@ -123,6 +151,26 @@ SELECT
 FROM stg.location_raw
 WHERE event_id IS NOT NULL;

+CREATE MATERIALIZED VIEW IF NOT EXISTS stg.mv_location_raw_to_ods_errors
+TO ods.location_event_errors
+AS
+WITH
+    toUUIDOrNull(JSONExtractString(raw, 'event_id')) AS event_id,
+    arrayFilter(x -> x != '', [
+        if(event_id IS NULL, 'bad_event_id', '')
+    ]) AS parse_errors
+SELECT
+    ingest_ts,
+    kafka_topic,
+    kafka_partition,
+    kafka_offset,
+    kafka_ts,
+    raw,
+    arrayStringConcat(parse_errors, ',') AS error_reason
+FROM stg.location_raw
+WHERE length(parse_errors) > 0
+    AND event_id IS NULL;
+
 -- ODS: device_events
 CREATE TABLE IF NOT EXISTS ods.device_by_click
 (
@@ -182,6 +230,31 @@ SELECT
 FROM stg.device_raw
 WHERE click_id IS NOT NULL;

+CREATE MATERIALIZED VIEW IF NOT EXISTS stg.mv_device_raw_to_ods_errors
+TO ods.device_by_click_errors
+AS
+WITH
+    toUUIDOrNull(JSONExtractString(raw, 'click_id')) AS click_id,
+    toUUIDOrNull(JSONExtractString(raw, 'user_domain_id')) AS user_domain_id,
+    arrayFilter(x -> x != '', [
+        if(click_id IS NULL, 'bad_click_id', ''),
+        if(user_domain_id IS NULL, 'bad_user_domain_id', '')
+    ]) AS parse_errors
+SELECT
+    ingest_ts,
+    kafka_topic,
+    kafka_partition,
+    kafka_offset,
+    kafka_ts,
+    raw,
+    arrayStringConcat(parse_errors, ',') AS error_reason
+FROM stg.device_raw
+WHERE length(parse_errors) > 0
+    AND (
+        click_id IS NULL
+        OR user_domain_id IS NULL
+    );
+
 -- ODS: geo_events
 CREATE TABLE IF NOT EXISTS ods.geo_by_click
 (
@@ -239,3 +312,31 @@ SELECT
    ]) AS parse_errors
 FROM stg.geo_raw
 WHERE click_id IS NOT NULL;
+
+CREATE MATERIALIZED VIEW IF NOT EXISTS stg.mv_geo_raw_to_ods_errors
+TO ods.geo_by_click_errors
+AS
+WITH
+    toUUIDOrNull(JSONExtractString(raw, 'click_id')) AS click_id,
+    toFloat64OrNull(JSONExtractString(raw, 'geo_latitude')) AS geo_latitude,
+    toFloat64OrNull(JSONExtractString(raw, 'geo_longitude')) AS geo_longitude,
+    arrayFilter(x -> x != '', [
+        if(click_id IS NULL, 'bad_click_id', ''),
+        if(geo_latitude IS NULL, 'bad_geo_latitude', ''),
+        if(geo_longitude IS NULL, 'bad_geo_longitude', '')
+    ]) AS parse_errors
+SELECT
+    ingest_ts,
+    kafka_topic,
+    kafka_partition,
+    kafka_offset,
+    kafka_ts,
+    raw,
+    arrayStringConcat(parse_errors, ',') AS error_reason
+FROM stg.geo_raw
+WHERE length(parse_errors) > 0
+    AND (
+        click_id IS NULL
+        OR geo_latitude IS NULL
+        OR geo_longitude IS NULL
+    );
@@ -6,7 +6,7 @@
 -- Strategy: full rebuild for demo; incremental for production
 INSERT INTO dds.click
 SELECT
-    d.click_id,
+    c.click_id,
    d.user_domain_id,
    d.user_custom_id,
    d.device_type,
@@ -22,14 +22,33 @@ SELECT
    g.ip_address,
    now64(3) AS dds_update_ts,
    arrayFilter(x -> x != '', arrayConcat(
-        d.parse_errors,
+        ifNull(d.parse_errors, []),
+        if(d.click_id IS NULL, ['device_not_found'], []),
        if(g.click_id IS NULL, ['geo_not_found'], []),
        if(g.geo_country IS NULL, ['geo_country_missing'], [])
    )) AS ods_parse_errors
 FROM (
+    -- Union of all click ids to keep geo-only/device-only arrivals
+    SELECT click_id
+    FROM (
+        SELECT assumeNotNull(click_id) AS click_id
+        FROM ods.device_by_click
+        WHERE click_id IS NOT NULL
+        GROUP BY click_id
+    )
+    UNION DISTINCT
+    SELECT click_id
+    FROM (
+        SELECT assumeNotNull(click_id) AS click_id
+        FROM ods.geo_by_click
+        WHERE click_id IS NOT NULL
+        GROUP BY click_id
+    )
+) AS c
+LEFT JOIN (
    -- Latest device snapshot from ODS
    SELECT
-        click_id,
+        assumeNotNull(click_id) AS click_id,
        argMax(user_domain_id, src_ingest_ts) AS user_domain_id,
        argMax(user_custom_id, src_ingest_ts) AS user_custom_id,
        argMax(device_type, src_ingest_ts) AS device_type,
@@ -41,11 +60,11 @@ FROM (
    FROM ods.device_by_click
    WHERE click_id IS NOT NULL
    GROUP BY click_id
-) AS d
+) AS d ON d.click_id = c.click_id
 LEFT JOIN (
    -- Latest geo snapshot from ODS
    SELECT
-        click_id,
+        assumeNotNull(click_id) AS click_id,
        argMax(geo_country, src_ingest_ts) AS geo_country,
        argMax(geo_region_name, src_ingest_ts) AS geo_region_name,
        argMax(geo_timezone, src_ingest_ts) AS geo_timezone,
@@ -55,7 +74,7 @@ LEFT JOIN (
    FROM ods.geo_by_click
    WHERE click_id IS NOT NULL
    GROUP BY click_id
-) AS g ON g.click_id = d.click_id;
+) AS g ON g.click_id = c.click_id;

 -- Refresh DDS.event (from browser + location)
 INSERT INTO dds.event
@@ -40,6 +40,8 @@ PARTITION BY toYYYYMM(check_date)
 ORDER BY (check_date, layer, table_name, check_name);

 -- Truncate and refill DQ summary
+TRUNCATE TABLE dm.dq_summary;
+
 INSERT INTO dm.dq_summary
 SELECT
    today() AS check_date,