Operate ingestion through control-table evidence.

Control schema model

Control tables are created by ensure_ctrl_tables() in the configured control catalog and schema. They are additive: framework upgrades add known columns when missing and do not remove existing columns automatically.

Table map

Field reference

Use this section as the operational data dictionary for dashboards, alert rules and support queries. JSON fields are stored as strings; parse them with from_json, get_json_object or SQL JSON functions when building curated operational views.

Useful SQL queries

Replace main.ops with the catalog and schema used for control tables.

Recent run health

SELECT
  status,
  target_table,
  mode,
  source_connector,
  runtime_type,
  rows_read,
  rows_written,
  rows_quarantined,
  duration_seconds,
  error_message,
  run_id
FROM main.ops.ctrl_ingestion_runs
WHERE run_date >= current_date() - INTERVAL 7 DAYS
ORDER BY started_at_utc DESC
LIMIT 100;

Failure triage with full stack trace

SELECT
  r.started_at_utc,
  r.target_table,
  r.source_connector,
  r.mode,
  r.runtime_type,
  r.error_message AS short_error,
  e.error_type,
  e.stack_trace
FROM main.ops.ctrl_ingestion_runs r
LEFT JOIN main.ops.ctrl_ingestion_errors e
  ON r.run_id = e.run_id
WHERE r.status = 'FAILED'
  AND r.run_date >= current_date() - INTERVAL 7 DAYS
ORDER BY r.started_at_utc DESC;

Freshness and watermark drift

SELECT
  target_table,
  watermark_column,
  watermark_value,
  last_success_at_utc,
  last_status,
  last_rows_written,
  last_delta_version,
  current_timestamp() - last_success_at_utc AS time_since_success
FROM main.ops.ctrl_ingestion_state
ORDER BY last_success_at_utc ASC NULLS FIRST;

Quality rules requiring attention

SELECT
  q.target_table,
  q.rule_name,
  q.severity,
  q.status,
  q.failed_count,
  q.message,
  r.run_id,
  r.started_at_utc
FROM main.ops.ctrl_ingestion_quality q
JOIN main.ops.ctrl_ingestion_runs r
  ON q.run_id = r.run_id
WHERE r.run_date >= current_date() - INTERVAL 14 DAYS
  AND q.failed_count > 0
ORDER BY r.started_at_utc DESC, q.failed_count DESC;

Quarantine volume by target and rule

SELECT
  target_table,
  rule_name,
  count(*) AS quarantined_records,
  max(quarantined_at_utc) AS latest_quarantine
FROM main.ops.ctrl_ingestion_quarantine
GROUP BY target_table, rule_name
ORDER BY quarantined_records DESC;

Schema changes by table

SELECT
  target_table,
  change_type,
  column_name,
  source_type,
  target_type,
  applied,
  change_ts_utc,
  details_json
FROM main.ops.ctrl_ingestion_schema_changes
ORDER BY change_ts_utc DESC;

Streaming aggregate versus child runs

WITH child AS (
  SELECT
    parent_run_id AS stream_run_id,
    count(*) AS child_batches,
    sum(coalesce(rows_read, 0)) AS child_rows_read,
    sum(coalesce(rows_written, 0)) AS child_rows_written,
    sum(coalesce(rows_quarantined, 0)) AS child_rows_quarantined
  FROM main.ops.ctrl_ingestion_runs
  WHERE parent_run_id IS NOT NULL
  GROUP BY parent_run_id
)
SELECT
  s.stream_run_id,
  s.status,
  s.batches_processed,
  c.child_batches,
  s.total_rows_read,
  c.child_rows_read,
  s.total_rows_written,
  c.child_rows_written,
  s.total_rows_quarantined,
  c.child_rows_quarantined
FROM main.ops.ctrl_ingestion_streams s
LEFT JOIN child c
  ON s.stream_run_id = c.stream_run_id
ORDER BY s.started_at_utc DESC;

Slow stages from stage durations

SELECT
  run_id,
  target_table,
  mode,
  duration_seconds,
  get_json_object(stage_durations_json, '$.read') AS read_seconds,
  get_json_object(stage_durations_json, '$.quality') AS quality_seconds,
  get_json_object(stage_durations_json, '$.write') AS write_seconds,
  get_json_object(stage_durations_json, '$.state') AS state_seconds,
  get_json_object(stage_durations_json, '$.lineage') AS lineage_seconds
FROM main.ops.ctrl_ingestion_runs
WHERE run_date >= current_date() - INTERVAL 7 DAYS
ORDER BY duration_seconds DESC
LIMIT 50;

Throughput and operational cost estimate

WITH rates AS (
  SELECT 2.0 AS dbu_per_hour, 0.40 AS currency_per_dbu
)
SELECT
  r.target_table,
  r.mode,
  r.source_connector,
  r.rows_read,
  r.rows_written,
  r.duration_seconds,
  CASE WHEN r.duration_seconds > 0 THEN r.rows_written / r.duration_seconds END AS rows_written_per_second,
  (r.duration_seconds / 3600.0) * rates.dbu_per_hour AS estimated_dbu,
  (r.duration_seconds / 3600.0) * rates.dbu_per_hour * rates.currency_per_dbu AS estimated_cost
FROM main.ops.ctrl_ingestion_runs r
CROSS JOIN rates
WHERE r.status = 'SUCCESS'
  AND r.run_date >= current_date() - INTERVAL 30 DAYS
ORDER BY estimated_cost DESC;

Governance application status

SELECT
  target_table,
  annotation_scope,
  annotation_type,
  column_name,
  key,
  status,
  error_message,
  annotation_ts_utc
FROM main.ops.ctrl_ingestion_annotations
ORDER BY annotation_ts_utc DESC;

Access drift and failed security operations

SELECT
  target_table,
  access_type,
  principal,
  privilege,
  column_name,
  function_name,
  mode,
  drift_policy,
  status,
  error_message,
  access_ts_utc
FROM main.ops.ctrl_ingestion_access
WHERE status <> 'APPLIED'
ORDER BY access_ts_utc DESC;

Active or expired locks

SELECT
  target_table,
  run_id,
  owner,
  status,
  acquired_at_utc,
  expires_at_utc,
  released_at_utc,
  CASE
    WHEN status = 'ACTIVE' AND expires_at_utc < current_timestamp() THEN 'EXPIRED_BUT_NOT_RELEASED'
    ELSE status
  END AS effective_status
FROM main.ops.ctrl_ingestion_locks
ORDER BY acquired_at_utc DESC;

Framework version inventory

SELECT
  framework_version,
  ctrl_schema_version,
  runtime_type,
  spark_version,
  python_version,
  count(*) AS runs
FROM main.ops.ctrl_ingestion_runs
WHERE run_date >= current_date() - INTERVAL 30 DAYS
GROUP BY framework_version, ctrl_schema_version, runtime_type, spark_version, python_version
ORDER BY runs DESC;

Operational workflows

Maintenance commands

Retention and cost reports are operational utilities. Cost reports estimate runtime cost from user-provided rates; they are not a substitute for cloud billing exports.

contractforge maintenance ctrl-retention --catalog main --ctrl-schema ops --retention-days 30
contractforge maintenance cost-report --catalog main --ctrl-schema ops --dbu-per-hour 2 --currency-per-dbu 0.40