-- One-row health summary for short and streaming Delta Funnel captures.
WITH RECURSIVE
selected_operations AS (
  SELECT
    s.track_id AS operation_track_id,
    extract_arg(s.arg_set_id, 'debug.operation_id') AS expected_operation_id
  FROM slice AS s
  WHERE s.category = 'delta_funnel.profile'
    AND s.name IN (
      'Delta Funnel preview',
      'Delta Funnel SQL Server write',
      'Delta Funnel SQL Server write_all'
    )
    AND (
      (
        SELECT capture_scope_id
        FROM delta_funnel_report_selection
      ) IS NULL
      OR extract_arg(s.arg_set_id, 'debug.capture_scope_id') = (
        SELECT capture_scope_id
        FROM delta_funnel_report_selection
      )
    )
),
selected_operation_tracks(track_id, expected_operation_id) AS (
  SELECT operation_track_id, expected_operation_id
  FROM selected_operations
  UNION
  SELECT child.id, parent.expected_operation_id
  FROM track AS child
  JOIN selected_operation_tracks AS parent
    ON child.parent_id = parent.track_id
),
semantic_slices AS (
  SELECT
    s.*,
    extract_arg(s.arg_set_id, 'debug.operation_id') AS operation_id,
    extract_arg(s.arg_set_id, 'debug.query_execution_id') AS query_execution_id,
    extract_arg(s.arg_set_id, 'debug.query_scope') AS query_scope,
    extract_arg(s.arg_set_id, 'debug.query_owner') AS query_owner,
    extract_arg(s.arg_set_id, 'debug.worker_lane_id') AS worker_lane_id,
    extract_arg(s.arg_set_id, 'debug.worker_kind') AS worker_kind,
    extract_arg(s.arg_set_id, 'debug.node_id') AS node_id,
    extract_arg(s.arg_set_id, 'debug.operator_partition') AS operator_partition,
    extract_arg(s.arg_set_id, 'debug.execution_stream_id') AS execution_stream_id,
    extract_arg(s.arg_set_id, 'debug.activity') AS activity,
    extract_arg(s.arg_set_id, 'debug.planning_activity_name') AS planning_activity_name,
    extract_arg(s.arg_set_id, 'debug.execution_activity_name')
      AS execution_activity_name,
    extract_arg(s.arg_set_id, 'debug.time_semantics') AS time_semantics,
    extract_arg(s.arg_set_id, 'debug.result') AS result,
    selected.expected_operation_id
  FROM slice AS s
  LEFT JOIN selected_operation_tracks AS selected
    ON selected.track_id = s.track_id
  WHERE s.category = 'delta_funnel.profile'
    AND (
      (
        SELECT capture_scope_id
        FROM delta_funnel_report_selection
      ) IS NULL
      OR selected.track_id IS NOT NULL
    )
),
classified_slices AS (
  SELECT
    *,
    operation_id IS NOT NULL
      AND operation_id IS expected_operation_id AS operation_identity_valid,
    name NOT IN (
      'Delta Funnel diagnostic group',
      'Operator activity trace truncated'
    ) AS ranked_semantic_candidate,
    CASE
      WHEN execution_activity_name IS NOT NULL THEN 'execution_activity'
      WHEN worker_lane_id IS NOT NULL
        OR worker_kind IS NOT NULL
        OR node_id IS NOT NULL
        OR operator_partition IS NOT NULL
        OR execution_stream_id IS NOT NULL
      THEN 'operator'
      WHEN planning_activity_name IS NOT NULL THEN 'planning_activity'
      WHEN name IN (
        'Delta Funnel preview',
        'Delta Funnel SQL Server write',
        'Delta Funnel SQL Server write_all'
      ) THEN 'operation'
      WHEN name IN ('Planning', 'Execution', 'Finalization') THEN 'phase'
      WHEN name = 'DataFusion query' THEN 'query'
      WHEN name = 'DataFusion query planning' THEN 'query_planning'
    END AS semantic_kind
  FROM semantic_slices
),
missing_fields AS (
  SELECT
    coalesce(sum(
      CASE
        WHEN ranked_semantic_candidate THEN
          (NOT operation_identity_valid) +
          CASE
            WHEN semantic_kind = 'operator' THEN
              (time_semantics IS NULL OR time_semantics != 'active')
            ELSE
              (
                time_semantics IS NULL
                OR time_semantics NOT IN ('wall_clock', 'lifecycle')
              )
          END
        ELSE 0
      END +
      CASE semantic_kind
        WHEN 'operation' THEN
          0
        WHEN 'phase' THEN
          0
        WHEN 'query' THEN
          (query_execution_id IS NULL) + (query_scope IS NULL)
        WHEN 'query_planning' THEN
          (query_execution_id IS NULL) + (query_scope IS NULL)
        WHEN 'planning_activity' THEN
          (query_execution_id IS NULL) +
          (query_scope IS NULL) + (planning_activity_name = '') +
          (activity IS NULL OR activity = '')
        WHEN 'execution_activity' THEN
          (query_execution_id IS NULL) +
          (query_scope IS NULL) + (node_id IS NULL) +
          (operator_partition IS NULL) + (execution_stream_id IS NULL) +
          (execution_activity_name = '') +
          (activity IS NULL OR activity = '')
        WHEN 'operator' THEN
          (query_execution_id IS NULL) +
          (query_scope IS NULL) + (worker_lane_id IS NULL) +
          (worker_kind IS NULL) + (node_id IS NULL) +
          (operator_partition IS NULL) + (execution_stream_id IS NULL) +
          (activity IS NULL)
        ELSE 0
      END
    ), 0) AS missing_identity_field_count,
    coalesce(sum(
      CASE
        WHEN semantic_kind IN (
          'operation',
          'phase',
          'query_planning',
          'planning_activity',
          'execution_activity'
        )
        THEN result IS NULL
        ELSE 0
      END
    ), 0) AS missing_terminal_result_count
  FROM classified_slices
),
ordered_slices AS (
  SELECT
    id,
    semantic_kind,
    ts,
    dur,
    track_id,
    depth,
    max(ts + dur) OVER (
      PARTITION BY track_id, depth
      ORDER BY ts, id
      ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING
    ) AS previous_end
  FROM classified_slices
  WHERE semantic_kind IN (
    'operator',
    'planning_activity',
    'execution_activity'
  ) AND dur > 0
),
crossing_semantic_slice_ids AS (
  SELECT id, semantic_kind
  FROM ordered_slices
  WHERE previous_end > ts
  UNION
  SELECT child.id, child.semantic_kind
  FROM classified_slices AS child
  JOIN classified_slices AS parent ON parent.id = child.parent_id
  WHERE child.semantic_kind IN (
    'operator',
    'planning_activity',
    'execution_activity'
  )
    AND parent.semantic_kind = child.semantic_kind
    AND child.dur > 0
    AND parent.dur > 0
    AND child.ts + child.dur > parent.ts + parent.dur
),
crossing_counts AS (
  SELECT
    coalesce(sum(semantic_kind = 'operator'), 0) AS crossing_worker_slice_count,
    coalesce(sum(semantic_kind = 'planning_activity'), 0)
      AS crossing_planning_activity_slice_count,
    coalesce(sum(semantic_kind = 'execution_activity'), 0)
      AS crossing_execution_activity_slice_count
  FROM crossing_semantic_slice_ids
),
invalid_planning_activity_hierarchy AS (
  SELECT child.id
  FROM classified_slices AS child
  LEFT JOIN classified_slices AS parent ON parent.id = child.parent_id
  WHERE child.semantic_kind = 'planning_activity'
    AND (
      parent.id IS NULL
      OR parent.semantic_kind IS NULL
      OR parent.semantic_kind NOT IN ('query_planning', 'planning_activity')
      OR parent.track_id IS NOT child.track_id
      OR parent.operation_id IS NOT child.operation_id
      OR parent.query_execution_id IS NOT child.query_execution_id
      OR parent.query_scope IS NOT child.query_scope
      OR parent.query_owner IS NOT child.query_owner
      OR (
        child.dur >= 0
        AND parent.dur >= 0
        AND (
          child.ts < parent.ts
          OR child.ts + child.dur > parent.ts + parent.dur
        )
      )
    )
),
invalid_execution_activity_hierarchy AS (
  SELECT child.id
  FROM classified_slices AS child
  LEFT JOIN track AS child_track ON child_track.id = child.track_id
  LEFT JOIN track AS query_track ON query_track.id = child_track.parent_id
  LEFT JOIN track AS operation_track ON operation_track.id = query_track.parent_id
  LEFT JOIN classified_slices AS query
    ON query.track_id = query_track.id
    AND query.semantic_kind = 'query'
    AND query.operation_id IS child.operation_id
    AND query.query_execution_id IS child.query_execution_id
    AND query.query_scope IS child.query_scope
    AND query.query_owner IS child.query_owner
  LEFT JOIN classified_slices AS operation
    ON operation.track_id = operation_track.id
    AND operation.semantic_kind = 'operation'
    AND operation.operation_id IS child.operation_id
  WHERE child.semantic_kind = 'execution_activity'
    AND (
      child.execution_activity_name != 'Await Delta scan output'
      OR child.name != child.execution_activity_name
      OR child.activity != 'await_output'
      OR child.time_semantics != 'wall_clock'
      OR child.result NOT IN ('ok', 'error', 'cancelled')
      OR child_track.id IS NULL
      OR query_track.id IS NULL
      OR operation_track.id IS NULL
      OR query.id IS NULL
      OR operation.id IS NULL
      OR (
        child.dur >= 0
        AND operation.dur >= 0
        AND (
          child.ts < operation.ts
          OR child.ts + child.dur > operation.ts + operation.dur
        )
      )
    )
),
profile_counts AS (
  SELECT
    coalesce(sum(semantic_kind = 'operation'), 0) AS operation_root_count,
    coalesce(sum(semantic_kind = 'operation' AND dur < 0), 0)
      AS incomplete_operation_root_count,
    coalesce(sum(semantic_kind = 'operator'), 0) AS operator_slice_count,
    coalesce(sum(semantic_kind = 'planning_activity'), 0)
      AS planning_activity_slice_count,
    coalesce(sum(semantic_kind = 'execution_activity'), 0)
      AS execution_activity_slice_count,
    coalesce(sum(name = 'Operator activity trace truncated'), 0)
      AS truncation_marker_count
  FROM classified_slices
),
profile_processes AS (
  -- linux.perf can retain rows outside its call-stack scope. A semantic track
  -- provides the trace-local process identity without matching display names.
  SELECT DISTINCT pt.upid
  FROM process_track AS pt
  JOIN semantic_slices AS s ON s.track_id = pt.id
),
sample_counts AS (
  SELECT
    count(*) AS perf_sample_count,
    count(*) - count(ps.callsite_id) AS perf_sample_without_callsite_count
  FROM perf_sample AS ps
  JOIN thread AS t USING (utid)
  JOIN profile_processes AS pp USING (upid)
),
lifecycle_counts AS (
  SELECT coalesce(sum(name = 'tracing_disabled_ns'), 0) AS tracing_disabled_count
  FROM metadata
),
stat_counts AS (
  SELECT
    coalesce(sum(CASE
      WHEN name GLOB 'traced_buf_*' AND severity = 'data_loss' THEN value
      ELSE 0
    END), 0) AS buffer_loss_count,
    -- Every checked-in config reserves buffer 0 for exact semantic events.
    coalesce(sum(CASE
      WHEN name GLOB 'traced_buf_*'
        AND severity = 'data_loss'
        AND idx = 0
      THEN value
      ELSE 0
    END), 0) AS semantic_buffer_loss_count,
    coalesce(sum(CASE
      WHEN severity = 'data_loss'
        AND name NOT GLOB 'traced_buf_*'
        AND name NOT IN (
          'perf_samples_skipped',
          'traced_flushes_failed',
          'traced_final_flush_failed'
        )
      THEN value
      ELSE 0
    END), 0) AS data_source_loss_count,
    coalesce(sum(CASE
      WHEN name = 'traced_flushes_failed' THEN value
      ELSE 0
    END), 0) AS failed_flush_count,
    coalesce(sum(CASE
      WHEN name = 'perf_samples_skipped' THEN value
      ELSE 0
    END), 0) AS perf_samples_skipped,
    coalesce(sum(CASE
      WHEN name = 'traced_final_flush_failed' THEN value
      ELSE 0
    END), 0) AS final_flush_failed
  FROM stats
),
trace_metrics AS (
  SELECT round((end_ts - start_ts) / 1000000000.0, 6) AS trace_duration_seconds
  FROM trace_bounds
),
health_values AS (
  SELECT
    operation_root_count > 0
      AND incomplete_operation_root_count = 0
      AND truncation_marker_count = 0
      AND missing_identity_field_count = 0
      AND missing_terminal_result_count = 0
      AND crossing_worker_slice_count = 0
      AND crossing_planning_activity_slice_count = 0
      AND crossing_execution_activity_slice_count = 0
      AND (SELECT count(*) FROM invalid_planning_activity_hierarchy) = 0
      AND (SELECT count(*) FROM invalid_execution_activity_hierarchy) = 0
      AND semantic_buffer_loss_count = 0
      AS semantic_complete,
    tracing_disabled_count > 0 AS finalization_observed,
    max(failed_flush_count, final_flush_failed) AS flush_failure_count,
    operation_root_count,
    incomplete_operation_root_count,
    operator_slice_count,
    planning_activity_slice_count,
    execution_activity_slice_count,
    truncation_marker_count,
    missing_identity_field_count,
    missing_terminal_result_count,
    crossing_worker_slice_count,
    crossing_planning_activity_slice_count,
    crossing_execution_activity_slice_count,
    (SELECT count(*) FROM invalid_planning_activity_hierarchy)
      AS invalid_planning_activity_hierarchy_count,
    (SELECT count(*) FROM invalid_execution_activity_hierarchy)
      AS invalid_execution_activity_hierarchy_count,
    perf_sample_count,
    perf_sample_without_callsite_count,
    perf_samples_skipped,
    buffer_loss_count,
    data_source_loss_count,
    trace_duration_seconds
  FROM
    profile_counts,
    missing_fields,
    sample_counts,
    lifecycle_counts,
    stat_counts,
    crossing_counts,
    trace_metrics
)
SELECT
  semantic_complete
    AND finalization_observed
    AND flush_failure_count = 0
    AND buffer_loss_count = 0
    AND data_source_loss_count = 0
    AS capture_complete,
  semantic_complete,
  operation_root_count,
  incomplete_operation_root_count,
  operator_slice_count,
  planning_activity_slice_count,
  execution_activity_slice_count,
  truncation_marker_count,
  missing_identity_field_count,
  missing_terminal_result_count,
  crossing_worker_slice_count,
  crossing_planning_activity_slice_count,
  crossing_execution_activity_slice_count,
  invalid_planning_activity_hierarchy_count,
  invalid_execution_activity_hierarchy_count,
  perf_sample_count,
  perf_sample_without_callsite_count,
  perf_samples_skipped,
  buffer_loss_count,
  data_source_loss_count,
  flush_failure_count,
  finalization_observed,
  trace_duration_seconds,
  configured_file_cap_bytes,
  saved_file_bytes
FROM health_values, delta_funnel_capture_health_input;
