Complete Tier 1 time grains and weekly summary (#3)

Author: CHEUKFUNGWU

README.md

@@ -157,6 +157,8 @@ postgres.public.llm_request_events
 → ods_ai_observability_llm_request_events_di
 → paimon_lake.dwd.dwd_ai_llm_request_di
 → paimon_lake.dws.dws_ai_llm_feature_request_1d
+→ paimon_lake.dws.dws_ai_llm_feature_request_1h
+→ paimon_lake.dws.dws_ai_llm_session_request_1d
 ```
 
 Flink SQL assets are stored in:

docs/data_model.md

@@ -458,6 +458,8 @@ Tier 1 expands the runtime observability model beyond LLM and Agent execution:
 | `dws_ai_retrieval_knowledge_base_request_1d` | One daily row per app, knowledge base, embedding model and strategy | Retrieval volume, zero-result count, hit count, similarity and latency metrics |
 | `dws_ai_feedback_feature_action_1d` | One daily row per app, feature and agent | Feedback volume, positive/negative counts, regeneration/report counts and average rating |
 | `dws_ai_guardrail_rule_check_1d` | One daily row per app, rule category and action | Guardrail check volume, trigger/action counts, latency and distinct users |
+| `dws_ai_llm_feature_request_1h` | One hourly row per app, feature and model | Request, token, cost and latency metrics for near-real-time dashboards |
+| `dws_ai_llm_session_request_1d` | One daily row per app and feature | Session count, turns, tokens, duration and positive-resolution count |
 | `ads_observability_retrieval_daily_quality` | Daily retrieval quality mart | Hit rate, zero-result rate and latency breach flags |
 | `ads_observability_feedback_daily_satisfaction` | Daily satisfaction mart | Satisfaction rate, regeneration rate and breach flags |
 | `ads_observability_guardrail_daily_violation` | Daily guardrail violation mart | Trigger/block rates and policy-latency breach flags |
@@ -480,5 +482,6 @@ Tier 2 starts with cost-governance tables:
 | `dws_ai_llm_feature_env_request_1d` | One daily row per app, feature, model and environment | Prod/staging/dev split of request volume, success/error counts, tokens, cost and latency |
 | `dws_ai_llm_region_request_1d` | One daily row per region, environment, app and model | Regional request volume, success/error counts, tokens, cost and latency |
 | `dws_ai_agent_team_run_1d` | One daily row per team, app, agent and task type | Team-attributed agent run volume, success/error counts, tokens, cost, duration and span metrics |
+| `ads_observability_executive_weekly_summary` | One weekly row per app | Cross-domain LLM, Agent, retrieval, feedback, guardrail, evaluation and cost summary |
 
-Remaining planned extensions include agent, compliance and platform-health dimensions.
+Tier 1 and Tier 2 table inventory is implemented. Remaining planned extensions are Tier 3 compliance, multi-agent orchestration and platform-health facts.

docs/metric_definitions.md

@@ -236,3 +236,30 @@ FROM ai_observability.dws_ai_agent_tool_tool_call_1d
 GROUP BY tool_name
 ORDER BY failure_rate DESC;
 ```
+
+---
+
+## 14. Resolved Session Count
+
+### Definition
+
+Distinct daily sessions with positive feedback. A session is resolved when it has a `thumbs_up` event or a rating value of at least 4.
+
+Session duration is measured from the first request start to the last request completion within the same date, app, feature, and session.
+
+---
+
+## 15. Executive Weekly Weighted Metrics
+
+### Definition
+
+`ads_observability_executive_weekly_summary` is grouped by ISO week start date and application. Average request latency and evaluation score are weighted by their request/evaluation counts rather than averaging daily averages.
+
+```text
+avg_latency_ms = sum(daily_avg_latency_ms * daily_request_count) / sum(daily_request_count)
+avg_evaluation_score = sum(daily_avg_score * daily_evaluation_count) / sum(daily_evaluation_count)
+retrieval_hit_rate_1w = retrieval_hit_cnt_1w / retrieval_returned_cnt_1w
+satisfaction_rate_1w = thumbs_up_cnt_1w / (thumbs_up_cnt_1w + thumbs_down_cnt_1w)
+evaluation_pass_rate_1w = evaluation_pass_cnt_1w / evaluation_cnt_1w
+total_ai_cost_amt_1w = llm_cost_amt_1w + agent_cost_amt_1w
+```

docs/runtime_runbook.md

@@ -49,7 +49,7 @@ Expected long-running SQL jobs:
 | --- | --- | --- | --- |
 | `flink/sql/10_ingest_ods_to_kafka.sql` | CDC ingest | Postgres CDC source table | Kafka ODS topic |
 | `flink/sql/20_build_dwd_from_kafka_ods.sql` | DWD transform and data quality filter | Kafka ODS topic | Paimon DWD |
-| `flink/sql/30_build_dws_from_dwd.sql` | DWS streaming aggregation | Paimon DWD | Paimon DWS |
+| `flink/sql/30_build_dws_from_dwd.sql` | Daily, hourly and session DWS streaming aggregation | Paimon DWD | Paimon DWS |
 
 Submit the streaming jobs:
 
@@ -137,12 +137,24 @@ The health check verifies:
 - Postgres source table `public.llm_request_events` has rows.
 - Flink REST API is reachable.
 - DWD and DWS streaming jobs are running.
+- Hourly feature and daily session DWS jobs are running.
 - Doris FE is queryable when serving checks are enabled.
 
 This is a fast operational check. It does not replace deeper table-level
 verification queries such as `flink/sql/91_verify_dwd_count.sql` and
 `flink/sql/92_verify_dws_metrics.sql`.
 
+The hourly feature job uses a five-second event-time watermark and a one-hour tumble window. The session job resolves sessions from positive feedback (`thumbs_up` or rating >= 4).
+
+Build and load the executive weekly summary after the daily DWS Parquet outputs are available:
+
+```bash
+uv run python -m scripts.spark_build_ads_executive_weekly_summary
+uv run python -m scripts.load_dws_metrics_to_doris \
+  --input data/warehouse/ads/ads_observability_executive_weekly_summary.parquet \
+  --table ads_observability_executive_weekly_summary
+```
+
 ## Flink Recovery
 
 The local Flink runtime is configured with:

docs/three_tier_domain_expansion_plan.md

@@ -532,6 +532,8 @@ Implementation status: Tier 2 granularity expansions are implemented. `dws_ai_co
 | Prompt engineering | Parse prompt registry (Git repo, config DB, or API) | Medium |
 | Model deployment | Parse CI/CD events (GitHub Actions, ArgoCD, internal deploy API) | Medium-High |
 
+Implementation status: `ads_observability_executive_weekly_summary` is implemented at week-and-app grain. It combines LLM, Agent, retrieval, feedback, guardrail, evaluation, and cost metrics for management reporting.
+
 ---
 
 ## 5. Tier 3 — Compliance, Multi-Agent, and Platform Health (Month 5+)
@@ -641,6 +643,8 @@ Source: Kafka JMX, Flink REST API, Paimon catalog metadata, Doris `information_s
 
 Hourly DWS tables use the same schema as daily DWS but with `hour` (int, 0-23) as an additional grouping key. In Flink, hourly aggregation uses a 1-hour tumble window. Daily DWS remains as a rollup from hourly.
 
+Implementation: `dws_ai_llm_feature_request_1h` is implemented with Spark batch aggregation and a Flink 1-hour tumble-window path, with Paimon and Doris serving tables.
+
 ### 6.2 Organization Granularity
 
 ```text
@@ -671,6 +675,8 @@ Implementation: `dim_team` and `dim_user` (Tier 2) enable team-level and user-le
 
 Built from: `dwd_ai_llm_request_di` grouped by `session_id`, then aggregated to daily.
 
+Implementation: `dws_ai_llm_session_request_1d` is implemented. A session is resolved when it has a `thumbs_up` feedback event or a rating of at least 4. Session duration runs from the first request start to the last request completion within the daily app/feature/session grain.
+
 ### 6.4 Region / Environment Granularity
 
 Both `region` and `environment` fields exist in all DWD tables but are not DWS grouping keys.

flink/sql/03_dwd_paimon_tables.sql

@@ -41,6 +41,7 @@ CREATE TABLE IF NOT EXISTS paimon_lake.dwd.dwd_ai_llm_request_di (
     environment STRING,
     created_at TIMESTAMP(3),
     `date` DATE,
+    WATERMARK FOR created_at AS created_at - INTERVAL '5' SECOND,
     PRIMARY KEY (request_id) NOT ENFORCED
 ) WITH (
     'bucket' = '4'

flink/sql/04_dws_paimon_tables.sql

@@ -20,6 +20,41 @@ CREATE TABLE IF NOT EXISTS paimon_lake.dws.dws_ai_llm_feature_request_1d (
     'bucket' = '4'
 );
 
+CREATE TABLE IF NOT EXISTS paimon_lake.dws.dws_ai_llm_feature_request_1h (
+    `date` DATE,
+    `hour` INT,
+    app_name STRING,
+    feature_name STRING,
+    model_name STRING,
+    request_cnt_1h BIGINT,
+    success_cnt_1h BIGINT,
+    error_cnt_1h BIGINT,
+    prompt_token_cnt_1h BIGINT,
+    completion_token_cnt_1h BIGINT,
+    total_token_cnt_1h BIGINT,
+    estimated_cost_amt_1h DOUBLE,
+    avg_latency_ms DOUBLE,
+    max_latency_ms BIGINT,
+    p95_latency_ms BIGINT,
+    PRIMARY KEY (`date`, `hour`, app_name, feature_name, model_name) NOT ENFORCED
+) PARTITIONED BY (`date`) WITH (
+    'bucket' = '4'
+);
+
+CREATE TABLE IF NOT EXISTS paimon_lake.dws.dws_ai_llm_session_request_1d (
+    `date` DATE,
+    app_name STRING,
+    feature_name STRING,
+    session_cnt_1d BIGINT,
+    avg_turns_per_session DOUBLE,
+    avg_tokens_per_session DOUBLE,
+    avg_duration_per_session_ms DOUBLE,
+    resolved_session_cnt_1d BIGINT,
+    PRIMARY KEY (`date`, app_name, feature_name) NOT ENFORCED
+) PARTITIONED BY (`date`) WITH (
+    'bucket' = '4'
+);
+
 CREATE TABLE IF NOT EXISTS paimon_lake.dws.dws_ai_retrieval_knowledge_base_request_1d (
     `date` DATE,
     app_name STRING,

flink/sql/30_build_dws_from_dwd.sql

@@ -1,24 +1,50 @@
--- Build daily feature metrics from DWD.
+-- Build hourly feature metrics from DWD using event-time windows.
 
-INSERT INTO paimon_lake.dws.dws_ai_llm_feature_request_1d
+INSERT INTO paimon_lake.dws.dws_ai_llm_feature_request_1h
 SELECT
-    `date`,
+    CAST(window_start AS DATE) AS `date`,
+    CAST(EXTRACT(HOUR FROM window_start) AS INT) AS `hour`,
     app_name,
     feature_name,
     model_name,
-    COUNT(*) AS request_count,
-    SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) AS success_count,
-    SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) AS error_count,
-    SUM(prompt_tokens) AS prompt_tokens,
-    SUM(completion_tokens) AS completion_tokens,
-    SUM(total_tokens) AS total_tokens,
-    SUM(estimated_cost_usd) AS estimated_cost_usd,
+    COUNT(*) AS request_cnt_1h,
+    SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) AS success_cnt_1h,
+    SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) AS error_cnt_1h,
+    SUM(prompt_tokens) AS prompt_token_cnt_1h,
+    SUM(completion_tokens) AS completion_token_cnt_1h,
+    SUM(total_tokens) AS total_token_cnt_1h,
+    SUM(estimated_cost_usd) AS estimated_cost_amt_1h,
     AVG(latency_ms) AS avg_latency_ms,
-    -- Flink 1.20 SQL does not support PERCENTILE_CONT as a streaming aggregate.
-    -- Store an explicit upper-bound metric instead of naming MAX latency as p95.
     CAST(MAX(latency_ms) AS BIGINT) AS max_latency_ms,
     CAST(0 AS BIGINT) AS p95_latency_ms
-FROM paimon_lake.dwd.dwd_ai_llm_request_di
+FROM TABLE(
+    TUMBLE(
+        TABLE paimon_lake.dwd.dwd_ai_llm_request_di,
+        DESCRIPTOR(created_at),
+        INTERVAL '1' HOUR
+    )
+)
+GROUP BY window_start, app_name, feature_name, model_name;
+
+-- Daily feature metrics roll up the hourly table so both grains share one definition.
+
+INSERT INTO paimon_lake.dws.dws_ai_llm_feature_request_1d
+SELECT
+    `date`,
+    app_name,
+    feature_name,
+    model_name,
+    SUM(request_cnt_1h) AS request_count,
+    SUM(success_cnt_1h) AS success_count,
+    SUM(error_cnt_1h) AS error_count,
+    SUM(prompt_token_cnt_1h) AS prompt_tokens,
+    SUM(completion_token_cnt_1h) AS completion_tokens,
+    SUM(total_token_cnt_1h) AS total_tokens,
+    SUM(estimated_cost_amt_1h) AS estimated_cost_usd,
+    SUM(avg_latency_ms * request_cnt_1h) / SUM(request_cnt_1h) AS avg_latency_ms,
+    MAX(max_latency_ms) AS max_latency_ms,
+    CAST(0 AS BIGINT) AS p95_latency_ms
+FROM paimon_lake.dws.dws_ai_llm_feature_request_1h
 GROUP BY `date`, app_name, feature_name, model_name;
 
 INSERT INTO paimon_lake.dws.dws_ai_retrieval_knowledge_base_request_1d
@@ -129,3 +155,48 @@ SELECT
     CAST(MAX(latency_ms) AS BIGINT) AS p95_latency_ms
 FROM paimon_lake.dwd.dwd_ai_llm_request_di
 GROUP BY `date`, region, environment, app_name, model_name;
+
+INSERT INTO paimon_lake.dws.dws_ai_llm_session_request_1d
+WITH session_metrics AS (
+    SELECT
+        `date`,
+        app_name,
+        feature_name,
+        session_id,
+        COUNT(*) AS turn_cnt,
+        SUM(total_tokens) AS token_cnt,
+        TIMESTAMPDIFF(MILLISECOND, MIN(created_at), MAX(created_at)) + MAX(latency_ms) AS session_duration_ms
+    FROM paimon_lake.dwd.dwd_ai_llm_request_di
+    GROUP BY `date`, app_name, feature_name, session_id
+),
+resolved_sessions AS (
+    SELECT
+        `date`,
+        app_name,
+        feature_name,
+        session_id,
+        MAX(
+            CASE
+                WHEN feedback_type = 'thumbs_up' OR (feedback_type = 'rating' AND rating_value >= 4) THEN 1
+                ELSE 0
+            END
+        ) AS is_resolved
+    FROM paimon_lake.dwd.dwd_ai_feedback_action_di
+    GROUP BY `date`, app_name, feature_name, session_id
+)
+SELECT
+    sessions.`date`,
+    sessions.app_name,
+    sessions.feature_name,
+    COUNT(*) AS session_cnt_1d,
+    AVG(sessions.turn_cnt) AS avg_turns_per_session,
+    AVG(sessions.token_cnt) AS avg_tokens_per_session,
+    AVG(sessions.session_duration_ms) AS avg_duration_per_session_ms,
+    SUM(COALESCE(feedback.is_resolved, 0)) AS resolved_session_cnt_1d
+FROM session_metrics AS sessions
+LEFT JOIN resolved_sessions AS feedback
+    ON sessions.`date` = feedback.`date`
+    AND sessions.app_name = feedback.app_name
+    AND sessions.feature_name = feedback.feature_name
+    AND sessions.session_id = feedback.session_id
+GROUP BY sessions.`date`, sessions.app_name, sessions.feature_name;

flink/sql/92_verify_dws_metrics.sql

@@ -50,3 +50,13 @@ SELECT
     COUNT(*) AS dws_agent_team_metric_rows,
     SUM(run_cnt_1d) AS total_team_run_cnt_1d
 FROM paimon_lake.dws.dws_ai_agent_team_run_1d;
+
+SELECT
+    COUNT(*) AS dws_llm_feature_hourly_metric_rows,
+    SUM(request_cnt_1h) AS total_hourly_request_cnt_1h
+FROM paimon_lake.dws.dws_ai_llm_feature_request_1h;
+
+SELECT
+    COUNT(*) AS dws_llm_session_metric_rows,
+    SUM(session_cnt_1d) AS total_session_cnt_1d
+FROM paimon_lake.dws.dws_ai_llm_session_request_1d;

scripts/check_pipeline_health.sh

@@ -98,6 +98,8 @@ expected = {
     "insert-into_paimon_lake.dws.dws_ai_feedback_feature_action_1d": False,
     "insert-into_paimon_lake.dws.dws_ai_guardrail_rule_check_1d": False,
     "insert-into_paimon_lake.dws.dws_ai_evaluation_feature_judgment_1d": False,
+    "insert-into_paimon_lake.dws.dws_ai_llm_feature_request_1h": False,
+    "insert-into_paimon_lake.dws.dws_ai_llm_session_request_1d": False,
 }
 for name in running_names:
     for expected_name in expected: