diff --git a/ddl/20_ods.sql b/ddl/20_ods.sql
index f19318b..0b2c0c6 100644
--- a/ddl/20_ods.sql
+++ b/ddl/20_ods.sql
@@ -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
+    );
diff --git a/jobs/30_dds_refresh.sql b/jobs/30_dds_refresh.sql
index aa4dcb3..1597035 100644
--- a/jobs/30_dds_refresh.sql
+++ b/jobs/30_dds_refresh.sql
@@ -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
diff --git a/jobs/40_dm_refresh.sql b/jobs/40_dm_refresh.sql
index 9637286..d2f0c42 100644
--- a/jobs/40_dm_refresh.sql
+++ b/jobs/40_dm_refresh.sql
@@ -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,