diff --git a/benchmarks/duckshop/edge_cases/orchestration/column_migration_history.json b/benchmarks/duckshop/edge_cases/orchestration/column_migration_history.json new file mode 100644 index 0000000000..7b7cb83187 --- /dev/null +++ b/benchmarks/duckshop/edge_cases/orchestration/column_migration_history.json @@ -0,0 +1,46 @@ +{ + "record": { + "state_database": null, + "state_schema": "main", + "event_id": "rename-000", + "target_name": null, + "model_name": "orders", + "relation_database": null, + "relation_schema": "main", + "relation_name": "orders_🧾", + "origin_column": "customer_🧾", + "destination_column": "customer_id", + "discovery": "manual", + "decision": "rename", + "run_id": "external-rename", + "created_at": "2026-02-03T04:05:06Z" + }, + "source_rows": 104, + "duplicate_rows": 3, + "later_time": "2026-02-03T04:05:06.000001Z", + "page_lengths": [ + 100, + 4 + ], + "later_index": 103, + "source_count": 104, + "model": "MODEL (description 'Orders.', materialized table); SELECT 1 AS order_id", + "retained_result": "rename-103", + "kind": "column_migrations", + "source_key": "event_id", + "timestamp_key": "created_at", + "id_prefix": "rename", + "later_record": { + "event_id": "rename-103", + "created_at": "2026-02-03T04:05:06.000001Z" + }, + "last_record": { + "event_id": "rename-103", + "created_at": "2026-02-03T04:05:06.000001Z" + }, + "record_overrides": { + "rename-103": { + "created_at": "2026-02-03T04:05:06.000001Z" + } + } +} diff --git a/benchmarks/duckshop/edge_cases/orchestration/consumption_history.json b/benchmarks/duckshop/edge_cases/orchestration/consumption_history.json new file mode 100644 index 0000000000..90f8ba0007 --- /dev/null +++ b/benchmarks/duckshop/edge_cases/orchestration/consumption_history.json @@ -0,0 +1,63 @@ +{ + "record": { + "state_database": null, + "state_schema": "main", + "execution_id": "execution-000", + "phase": "succeeded", + "run_id": "external-run", + "node_type": "model", + "node_name": "orders", + "target_database": null, + "target_schema": "main", + "target_name": "orders_🧾", + "version_hash": "orders-version", + "input_count": 1, + "input_kind": "source", + "input_name": "shop.orders_🧾", + "input_version": "source-version", + "recorded_at": "2026-02-03T04:05:06Z" + }, + "source_rows": 104, + "duplicate_rows": 3, + "later_time": "2026-02-03T04:05:06.000001Z", + "page_lengths": [ + 100, + 7 + ], + "later_index": 106, + "source_count": 107, + "model": "MODEL (description 'Orders.', materialized table); SELECT 1 AS order_id", + "retained_result": "execution-103", + "kind": "consumption", + "source_key": "execution_id", + "timestamp_key": "recorded_at", + "id_prefix": "execution", + "later_record": { + "execution_id": "execution-103", + "phase": "started", + "input_count": 0, + "input_kind": null, + "input_name": null, + "input_version": null, + "recorded_at": "2026-02-03T04:05:06.000001Z" + }, + "last_record": { + "execution_id": "execution-103", + "phase": "started", + "input_count": 0, + "input_kind": null, + "input_name": null, + "input_version": null, + "recorded_at": "2026-02-03T04:05:06.000001Z" + }, + "record_overrides": { + "execution-103": { + "phase": "started", + "input_count": 0, + "input_kind": null, + "input_name": null, + "input_version": null, + "recorded_at": "2026-02-03T04:05:06.000001Z" + } + } +} diff --git a/benchmarks/duckshop/edge_cases/orchestration/node_result_history.json b/benchmarks/duckshop/edge_cases/orchestration/node_result_history.json new file mode 100644 index 0000000000..db3a8f2dae --- /dev/null +++ b/benchmarks/duckshop/edge_cases/orchestration/node_result_history.json @@ -0,0 +1,52 @@ +{ + "record": { + "state_database": null, + "state_schema": "main", + "node_type": "task", + "node_name": "result-000", + "target_database": null, + "target_schema": "main", + "target_name": null, + "run_id": "external-run", + "status": "success", + "payload_json_b64": "eyJub3RlIjoiUmV0YWluZWQgJ29yZGVycycgXHVkODNlXHVkZGZlIiwicm93cyI6N30=", + "metadata_json_b64": "e30=", + "error_message": null, + "materialized": "true", + "ts": "2026-02-03T04:05:06Z" + }, + "source_rows": 104, + "duplicate_rows": 3, + "later_time": "2026-02-03T04:05:06.000001Z", + "page_lengths": [ + 100, + 7 + ], + "later_index": 106, + "source_count": 107, + "model": "MODEL (description 'Orders.', materialized table); SELECT 1 AS order_id", + "retained_result": "result-103", + "kind": "node_results", + "source_key": "node_name", + "timestamp_key": "ts", + "id_prefix": "result", + "later_record": { + "node_name": "result-103", + "status": "failed", + "error_message": "Rejected 'orders' 🧾", + "ts": "2026-02-03T04:05:06.000001Z" + }, + "last_record": { + "node_name": "result-103", + "status": "failed", + "error_message": "Rejected 'orders' 🧾", + "ts": "2026-02-03T04:05:06.000001Z" + }, + "record_overrides": { + "result-103": { + "status": "failed", + "error_message": "Rejected 'orders' 🧾", + "ts": "2026-02-03T04:05:06.000001Z" + } + } +} diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index 0a2c10dcbb..8a7d06b209 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -3039,7 +3039,7 @@ export interface components { * @description Catalog observations require full scans; only immutable history supports a watermark. * @enum {string} */ - WarehouseHistoryKind: "fingerprints" | "source_freshness" | "audit_results" | "janitor_events" | "migrations" | "old_name_views" | "microbatches" | "catalog" | "statistics"; + WarehouseHistoryKind: "fingerprints" | "source_freshness" | "audit_results" | "janitor_events" | "migrations" | "old_name_views" | "microbatches" | "column_migrations" | "node_results" | "consumption" | "catalog" | "statistics"; /** @description Warehouse cleanup facts do not authorize control-plane deletion or worker recovery. */ WarehouseJanitorRecord: { archive_name: string; diff --git a/crates/sqlbuild-analysis/src/janitor_planning/constants.rs b/crates/sqlbuild-analysis/src/janitor_planning/constants.rs index 29ab3e1097..c68aa2bfc7 100644 --- a/crates/sqlbuild-analysis/src/janitor_planning/constants.rs +++ b/crates/sqlbuild-analysis/src/janitor_planning/constants.rs @@ -1,25 +1,9 @@ //! Janitor planning names, prefixes and skip reasons. -use sqlbuild_core::constants::COLUMN_MIGRATION_TABLE_NAME; -use sqlbuild_core::constants::CONSUMPTION_TABLE_NAME; -use sqlbuild_core::constants::FINGERPRINT_TABLE_NAME; -use sqlbuild_core::constants::MIGRATION_TABLE_NAME; -use sqlbuild_core::constants::OLD_NAME_VIEW_TABLE_NAME; -use sqlbuild_core::constants::SOURCE_FRESHNESS_TABLE_NAME; +use sqlbuild_core::constants::STATE_TABLE_NAMES; /// SQLBuild state tables the janitor never acts on. -pub const BUILT_IN_EXCLUDE_PATTERNS: [&str; 10] = [ - FINGERPRINT_TABLE_NAME, - "_sqlbuild_microbatches", - SOURCE_FRESHNESS_TABLE_NAME, - CONSUMPTION_TABLE_NAME, - "_sqlbuild_node_results", - "_sqlbuild_audit_results", - "_sqlbuild_janitor_events", - MIGRATION_TABLE_NAME, - COLUMN_MIGRATION_TABLE_NAME, - OLD_NAME_VIEW_TABLE_NAME, -]; +pub const BUILT_IN_EXCLUDE_PATTERNS: [&str; 10] = STATE_TABLE_NAMES; /// Prefix of every strict janitor archive name. pub const ARCHIVE_NAME_PREFIX: &str = "_SQB_ARCHIVE__"; /// Separator between archive name segments. diff --git a/crates/sqlbuild-cli/src/native_commands/_helpers/execution.rs b/crates/sqlbuild-cli/src/native_commands/_helpers/execution.rs index a7a8b314c4..9d5ce63b7e 100644 --- a/crates/sqlbuild-cli/src/native_commands/_helpers/execution.rs +++ b/crates/sqlbuild-cli/src/native_commands/_helpers/execution.rs @@ -346,6 +346,7 @@ pub(crate) fn metadata_sync( reconcile_at: Instant::now(), reconcile_seconds, limit: MAX_HISTORY_RECORDS, + statistics_limit: MAX_HISTORY_RECORDS, pending: None, next_at: Instant::now(), } @@ -389,7 +390,7 @@ pub(crate) async fn statistics_page( generation: state.generation, location: state.location().clone(), after, - limit: state.limit, + limit: state.statistics_limit, }, ) .await @@ -423,8 +424,13 @@ pub(crate) fn freeze_history_page( > BODY_BYTES { state.statistics = None; - if state.limit > 1 { - state.limit = (state.limit / 2).max(1); + let limit = if state.kind() == WarehouseHistoryKind::Statistics { + &mut state.statistics_limit + } else { + &mut state.limit + }; + if *limit > 1 { + *limit /= 2; } else { state.next_at = Instant::now() + Duration::from_secs(METADATA_POLL_SECONDS); drop(write_output(&[( @@ -450,6 +456,9 @@ pub(crate) fn acknowledge_metadata_page(mut state: LocationMetadataSync) -> Loca state.statistics = None; } if state.source_after.is_none() { + if state.kind() == WarehouseHistoryKind::Statistics { + state.statistics_limit = MAX_HISTORY_RECORDS; + } state.index += 1; state.scan_id = Uuid::now_v7(); state.observed_since = None; diff --git a/crates/sqlbuild-cli/src/native_commands/models.rs b/crates/sqlbuild-cli/src/native_commands/models.rs index 61f4d42fd2..a1bbf08388 100644 --- a/crates/sqlbuild-cli/src/native_commands/models.rs +++ b/crates/sqlbuild-cli/src/native_commands/models.rs @@ -200,6 +200,8 @@ pub(crate) struct LocationMetadataSync { pub(crate) reconcile_at: tokio::time::Instant, pub(crate) reconcile_seconds: u32, pub(crate) limit: usize, + /// Survives a failed statistics read's rewind through catalog, which resets `limit`. + pub(crate) statistics_limit: usize, pub(crate) pending: Option, pub(crate) next_at: tokio::time::Instant, } @@ -207,14 +209,16 @@ pub(crate) struct LocationMetadataSync { impl LocationMetadataSync { /// Failed samples revisit catalog before retrying; no frozen delivery is discarded. pub(crate) fn retry_read(&mut self) { - self.limit = (self.limit / 2).max(1); self.statistics = None; if self.kind() == WarehouseHistoryKind::Statistics { + self.statistics_limit = (self.statistics_limit / 2).max(1); self.index -= 1; self.source_after = None; self.scan_id = Uuid::now_v7(); self.observed_since = None; self.watermark_loaded = false; + } else { + self.limit = (self.limit / 2).max(1); } self.next_at = tokio::time::Instant::now() + std::time::Duration::from_secs(super::constants::METADATA_POLL_SECONDS); diff --git a/crates/sqlbuild-cli/src/native_commands/tests.rs b/crates/sqlbuild-cli/src/native_commands/tests.rs index 36ed1676b5..6aecb77547 100644 --- a/crates/sqlbuild-cli/src/native_commands/tests.rs +++ b/crates/sqlbuild-cli/src/native_commands/tests.rs @@ -6,6 +6,8 @@ mod decision; mod helpers; #[path = "tests/test_init.rs"] mod init; +#[path = "tests/test_location_metadata.rs"] +mod location_metadata; #[path = "tests/test_location_plan.rs"] mod location_plan; #[path = "tests/test_paths.rs"] diff --git a/crates/sqlbuild-cli/src/native_commands/tests/helpers.rs b/crates/sqlbuild-cli/src/native_commands/tests/helpers.rs index ec817e0ae9..ed08157d2e 100644 --- a/crates/sqlbuild-cli/src/native_commands/tests/helpers.rs +++ b/crates/sqlbuild-cli/src/native_commands/tests/helpers.rs @@ -1,8 +1,13 @@ use std::path::Path; +use sqlbuild_orchestration::warehouse_metadata::models::MetadataBatch; +use uuid::Uuid; + use crate::argv::main::parse_cli_arguments::parse_cli_arguments; use crate::argv::models::{ArgvOutcome, ArgvSettings}; use crate::native_commands::_helpers::argv_host::{NativeArgvHost, namespace_defaults}; +use crate::native_commands::_helpers::execution::acknowledge_metadata_page; +use crate::native_commands::models::{LocationMetadataPage, LocationMetadataSync}; /// Parse `argv` onto the `CliNamespace` defaults with the native host. pub(super) fn parse(argv: &[&str]) -> (ArgvOutcome, NativeArgvHost) { @@ -35,3 +40,23 @@ pub(super) fn plan_output(diff_line: &str, count: usize) -> Vec { serde_json::to_vec(&serde_json::json!({"selected_count": count, "models": models})) .expect("plan output") } + +/// Acknowledges `pages` complete, empty pages, each finishing the kind it was read for. +pub(super) fn acknowledge_complete_pages( + state: LocationMetadataSync, + pages: usize, +) -> LocationMetadataSync { + (0..pages).fold(state, |mut state, _| { + state.pending = Some(LocationMetadataPage { + batch: MetadataBatch { + kind: state.kind(), + session_id: Uuid::nil(), + generation: 1, + records: Vec::new(), + scan: None, + }, + next_source_after: None, + }); + acknowledge_metadata_page(state) + }) +} diff --git a/crates/sqlbuild-cli/src/native_commands/tests/test_location_metadata.rs b/crates/sqlbuild-cli/src/native_commands/tests/test_location_metadata.rs new file mode 100644 index 0000000000..a888efba54 --- /dev/null +++ b/crates/sqlbuild-cli/src/native_commands/tests/test_location_metadata.rs @@ -0,0 +1,55 @@ +use sqlbuild_orchestration::warehouse_metadata::constants::MAX_HISTORY_RECORDS; +use sqlbuild_orchestration::warehouse_metadata::models::{ + WarehouseHistoryKind, WarehouseStateLocation, +}; + +use crate::native_commands::_helpers::execution::metadata_sync; +use crate::native_commands::tests::helpers::acknowledge_complete_pages; +use crate::native_commands::tests::test_types::MetadataRetryTestCase; + +#[test] +fn given_failed_metadata_read_when_back_at_statistics_then_page_shrinks_until_scan_completes() { + let test_cases = [ + MetadataRetryTestCase { + description: "a timed-out statistics page is retried smaller after catalog is revisited", + failed_kind: WarehouseHistoryKind::Statistics, + acknowledged_pages: 1, + expected_statistics_limit: MAX_HISTORY_RECORDS / 2, + }, + MetadataRetryTestCase { + description: "a failed catalog page leaves the statistics page size alone", + failed_kind: WarehouseHistoryKind::Catalog, + acknowledged_pages: 1, + expected_statistics_limit: MAX_HISTORY_RECORDS, + }, + MetadataRetryTestCase { + description: "a completed statistics scan restores the full statistics page", + failed_kind: WarehouseHistoryKind::Statistics, + acknowledged_pages: 1 + WarehouseHistoryKind::ALL.len(), + expected_statistics_limit: MAX_HISTORY_RECORDS, + }, + ]; + for test_case in test_cases { + let location = WarehouseStateLocation { + database: Some("analytics".into()), + schema: "state".into(), + }; + let mut state = metadata_sync((vec![location], 1), 3600); + state.index = WarehouseHistoryKind::ALL + .iter() + .position(|kind| *kind == test_case.failed_kind) + .expect("kind is listed"); + state.retry_read(); + let state = acknowledge_complete_pages(state, test_case.acknowledged_pages); + assert_eq!( + (state.kind(), state.statistics_limit, state.limit), + ( + WarehouseHistoryKind::Statistics, + test_case.expected_statistics_limit, + MAX_HISTORY_RECORDS, + ), + "{description}", + description = test_case.description + ); + } +} diff --git a/crates/sqlbuild-cli/src/native_commands/tests/test_types.rs b/crates/sqlbuild-cli/src/native_commands/tests/test_types.rs index 93270cf91a..580e0ee5e4 100644 --- a/crates/sqlbuild-cli/src/native_commands/tests/test_types.rs +++ b/crates/sqlbuild-cli/src/native_commands/tests/test_types.rs @@ -1,3 +1,5 @@ +use sqlbuild_orchestration::warehouse_metadata::models::WarehouseHistoryKind; + use crate::argv::models::{TypeFailure, Value}; use crate::native_commands::errors::PythonNeeded; use crate::native_commands::models::{CommandOutput, NativeCommand, RulesAction}; @@ -89,3 +91,11 @@ pub(super) struct PlanDocumentTestCase { pub(super) planned: bool, pub(super) expected_first_model: serde_json::Value, } + +/// One failed metadata read and the statistics page size the sync uses afterwards. +pub(super) struct MetadataRetryTestCase { + pub(super) description: &'static str, + pub(super) failed_kind: WarehouseHistoryKind, + pub(super) acknowledged_pages: usize, + pub(super) expected_statistics_limit: usize, +} diff --git a/crates/sqlbuild-core/src/constants.rs b/crates/sqlbuild-core/src/constants.rs index 7f5096db14..7eeb30d732 100644 --- a/crates/sqlbuild-core/src/constants.rs +++ b/crates/sqlbuild-core/src/constants.rs @@ -30,6 +30,27 @@ pub const MIGRATION_TABLE_NAME: &str = "_sqlbuild_migrations"; pub const COLUMN_MIGRATION_TABLE_NAME: &str = "_sqlbuild_column_migrations"; /// State table of old-name view events. pub const OLD_NAME_VIEW_TABLE_NAME: &str = "_sqlbuild_old_name_views"; +/// State table of each node's outcome in each run. +pub const NODE_RESULTS_TABLE_NAME: &str = "_sqlbuild_node_results"; +/// State table of microbatch requirements and completions. +pub const MICROBATCH_TABLE_NAME: &str = "_sqlbuild_microbatches"; +/// State table of audit outcomes. +pub const AUDIT_RESULTS_TABLE_NAME: &str = "_sqlbuild_audit_results"; +/// State table of janitor archive and drop events. +pub const JANITOR_EVENTS_TABLE_NAME: &str = "_sqlbuild_janitor_events"; +/// Every state table SQLBuild writes in a target's state schema. +pub const STATE_TABLE_NAMES: [&str; 10] = [ + FINGERPRINT_TABLE_NAME, + MICROBATCH_TABLE_NAME, + SOURCE_FRESHNESS_TABLE_NAME, + CONSUMPTION_TABLE_NAME, + NODE_RESULTS_TABLE_NAME, + AUDIT_RESULTS_TABLE_NAME, + JANITOR_EVENTS_TABLE_NAME, + MIGRATION_TABLE_NAME, + COLUMN_MIGRATION_TABLE_NAME, + OLD_NAME_VIEW_TABLE_NAME, +]; /// A zero-length cursor duration. pub const ZERO_DAY_DURATION: &str = "0d"; /// Node key of a common table expression in the polyglot SQL AST. diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_operations.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_operations.rs index 602c4e4621..fd51e5d8cc 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_operations.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_operations.rs @@ -26,11 +26,11 @@ use crate::runs::models::{ use crate::runs::types::{CommandPlanner, RunsListing}; use crate::store::errors::ControlError; use crate::store::models::ControlStore; -use crate::warehouse_metadata::main::read_person_catalog::read_person_catalog; -use crate::warehouse_metadata::main::read_person_column_lineage::read_person_column_lineage; -use crate::warehouse_metadata::main::read_person_fingerprint::read_person_fingerprint; -use crate::warehouse_metadata::main::read_person_node_status::read_person_node_status; -use crate::warehouse_metadata::main::read_person_statistics::read_person_statistics; +use crate::warehouse_metadata::main::person_reads::read_person_catalog::read_person_catalog; +use crate::warehouse_metadata::main::person_reads::read_person_column_lineage::read_person_column_lineage; +use crate::warehouse_metadata::main::person_reads::read_person_fingerprint::read_person_fingerprint; +use crate::warehouse_metadata::main::person_reads::read_person_node_status::read_person_node_status; +use crate::warehouse_metadata::main::person_reads::read_person_statistics::read_person_statistics; use crate::warehouse_metadata::models::{ CatalogPage, CatalogQuery, ColumnLineageRead, FingerprintObservation, FingerprintQuery, NodeStatus, NodeStatusQuery, StatisticsPage, StatisticsQuery, StatisticsRetention, diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs index 66d81a041a..1e2b6c3f05 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs @@ -41,7 +41,7 @@ use crate::warehouse_history::main::read_person_run_node_error::read_person_run_ use crate::warehouse_history::main::read_person_run_output::read_person_run_output; use crate::warehouse_history::main::read_person_run_statement_sql::read_person_run_statement_sql; use crate::warehouse_history::main::read_person_run_statements::read_person_run_statements; -use crate::warehouse_metadata::main::list_person_catalog_nodes::list_person_catalog_nodes; +use crate::warehouse_metadata::main::person_reads::list_person_catalog_nodes::list_person_catalog_nodes; use crate::warehouse_metadata::models::CatalogNodeRecord; #[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/runs/{run_id}/nodes", operation_id = "listPersonRunNodes", diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/warehouse_metadata.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/warehouse_metadata.rs index 9354aada8d..4632b8b76a 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/warehouse_metadata.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/warehouse_metadata.rs @@ -8,17 +8,20 @@ use crate::api::_helpers::authentication::bearer::require_location; use crate::api::errors::ApiError; use crate::credentials::models::TokenIdentity; use crate::store::models::ControlStore; -use crate::warehouse_metadata::main::ingest_audit_results::ingest_audit_results; -use crate::warehouse_metadata::main::ingest_catalog::ingest_catalog; -use crate::warehouse_metadata::main::ingest_fingerprints::ingest_fingerprints; -use crate::warehouse_metadata::main::ingest_janitor_events::ingest_janitor_events; -use crate::warehouse_metadata::main::ingest_microbatches::ingest_microbatches; -use crate::warehouse_metadata::main::ingest_migrations::ingest_migrations; -use crate::warehouse_metadata::main::ingest_old_name_views::ingest_old_name_views; -use crate::warehouse_metadata::main::ingest_source_freshness::ingest_source_freshness; -use crate::warehouse_metadata::main::ingest_statistics::ingest_statistics; -use crate::warehouse_metadata::main::read_statistics_due::read_statistics_due; -use crate::warehouse_metadata::main::read_watermark::read_watermark; +use crate::warehouse_metadata::main::ingest::ingest_audit_results::ingest_audit_results; +use crate::warehouse_metadata::main::ingest::ingest_catalog::ingest_catalog; +use crate::warehouse_metadata::main::ingest::ingest_column_migrations::ingest_column_migrations; +use crate::warehouse_metadata::main::ingest::ingest_consumption::ingest_consumption; +use crate::warehouse_metadata::main::ingest::ingest_fingerprints::ingest_fingerprints; +use crate::warehouse_metadata::main::ingest::ingest_janitor_events::ingest_janitor_events; +use crate::warehouse_metadata::main::ingest::ingest_microbatches::ingest_microbatches; +use crate::warehouse_metadata::main::ingest::ingest_migrations::ingest_migrations; +use crate::warehouse_metadata::main::ingest::ingest_node_results::ingest_node_results; +use crate::warehouse_metadata::main::ingest::ingest_old_name_views::ingest_old_name_views; +use crate::warehouse_metadata::main::ingest::ingest_source_freshness::ingest_source_freshness; +use crate::warehouse_metadata::main::ingest::ingest_statistics::ingest_statistics; +use crate::warehouse_metadata::main::location_reads::read_statistics_due::read_statistics_due; +use crate::warehouse_metadata::main::location_reads::read_watermark::read_watermark; use crate::warehouse_metadata::models::{ MetadataBatch, StatisticsDueRead, StatisticsRead, WarehouseHistoryKind, WarehouseSyncResponse, WarehouseWatermarkRead, @@ -47,6 +50,11 @@ pub(crate) async fn report( WarehouseHistoryKind::Migrations => ingest_migrations(&store, scope, &batch).await, WarehouseHistoryKind::OldNameViews => ingest_old_name_views(&store, scope, &batch).await, WarehouseHistoryKind::Microbatches => ingest_microbatches(&store, scope, &batch).await, + WarehouseHistoryKind::ColumnMigrations => { + ingest_column_migrations(&store, scope, &batch).await + } + WarehouseHistoryKind::NodeResults => ingest_node_results(&store, scope, &batch).await, + WarehouseHistoryKind::Consumption => ingest_consumption(&store, scope, &batch).await, WarehouseHistoryKind::Catalog => ingest_catalog(&store, scope, &batch).await, WarehouseHistoryKind::Statistics => ingest_statistics(&store, scope, &batch).await, }?; diff --git a/crates/sqlbuild-orchestration/src/platform/_helpers/retention.rs b/crates/sqlbuild-orchestration/src/platform/_helpers/retention.rs index bacfb54d03..05bd7f95eb 100644 --- a/crates/sqlbuild-orchestration/src/platform/_helpers/retention.rs +++ b/crates/sqlbuild-orchestration/src/platform/_helpers/retention.rs @@ -14,7 +14,7 @@ use crate::store::errors::ControlError; use crate::store::main::{query::query, write::write}; use crate::store::models::{ControlStore, Parameter}; use crate::warehouse_metadata::constants::MAX_HISTORY_RECORDS; -use crate::warehouse_metadata::main::prune_statistics::prune_statistics; +use crate::warehouse_metadata::main::ingest::prune_statistics::prune_statistics; use crate::warehouse_metadata::models::StatisticsRetention; /// Old evaluation phases go once a later attempt of the same sensor exists. diff --git a/crates/sqlbuild-orchestration/src/store/schema.sql b/crates/sqlbuild-orchestration/src/store/schema.sql index a7e4973c6d..f12fd87027 100644 --- a/crates/sqlbuild-orchestration/src/store/schema.sql +++ b/crates/sqlbuild-orchestration/src/store/schema.sql @@ -465,6 +465,42 @@ CREATE TABLE sqb_warehouse_microbatches ( ); CREATE INDEX sqb_microbatches_namespace ON sqb_warehouse_microbatches (location_id,target,state_database,state_schema,observed_at,event_id); CREATE INDEX sqb_microbatches_model ON sqb_warehouse_microbatches (location_id,target,model_name); +CREATE TABLE sqb_warehouse_column_migrations ( + event_id {uuid} PRIMARY KEY, + location_id {uuid} NOT NULL REFERENCES sqb_locations(location_id), + target TEXT NOT NULL, + observed_at {time} NOT NULL, + state_database TEXT, + state_schema TEXT NOT NULL, + record_json {json} NOT NULL, + received_at {time} NOT NULL, + scan_id {uuid} +); +CREATE INDEX sqb_column_migrations_namespace ON sqb_warehouse_column_migrations (location_id,target,state_database,state_schema,observed_at,event_id); +CREATE TABLE sqb_warehouse_node_results ( + event_id {uuid} PRIMARY KEY, + location_id {uuid} NOT NULL REFERENCES sqb_locations(location_id), + target TEXT NOT NULL, + observed_at {time} NOT NULL, + state_database TEXT, + state_schema TEXT NOT NULL, + record_json {json} NOT NULL, + received_at {time} NOT NULL, + scan_id {uuid} +); +CREATE INDEX sqb_node_results_namespace ON sqb_warehouse_node_results (location_id,target,state_database,state_schema,observed_at,event_id); +CREATE TABLE sqb_warehouse_consumption ( + event_id {uuid} PRIMARY KEY, + location_id {uuid} NOT NULL REFERENCES sqb_locations(location_id), + target TEXT NOT NULL, + observed_at {time} NOT NULL, + state_database TEXT, + state_schema TEXT NOT NULL, + record_json {json} NOT NULL, + received_at {time} NOT NULL, + scan_id {uuid} +); +CREATE INDEX sqb_consumption_namespace ON sqb_warehouse_consumption (location_id,target,state_database,state_schema,observed_at,event_id); CREATE INDEX sqb_freshness_source_latest ON sqb_warehouse_freshness ( location_id, target, (record_json ->> 'source_name'), (record_json ->> 'target_database'), (record_json ->> 'target_schema'), diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/events.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/events.rs index 8b1e5693c8..fd614ea6eb 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/events.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/events.rs @@ -27,7 +27,12 @@ pub(crate) async fn ingest_events( ), )?; let table = batch.kind.mirror_table(); - let source_table = format!("_sqlbuild_{}", batch.kind.as_str()); + let source_table = batch.kind.source_table().ok_or_else(|| { + ControlError::Invalid(format!( + "{} is not a state-table history", + batch.kind.as_str() + )) + })?; let mut transaction = begin_ingestion( store, (location_id, target), @@ -42,7 +47,7 @@ pub(crate) async fn ingest_events( ( record.location.database.as_deref(), &record.location.schema, - &source_table, + source_table, ), &record.source_id, )?; diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_audit_results.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_audit_results.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_audit_results.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_audit_results.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_catalog.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_catalog.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_catalog.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_catalog.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_column_migrations.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_column_migrations.rs new file mode 100644 index 0000000000..0c2b149a9c --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_column_migrations.rs @@ -0,0 +1,46 @@ +//! Mirror column rename decisions; the warehouse state table stays the owner of the history. + +use serde_json::json; +use uuid::Uuid; + +use crate::store::errors::ControlError; +use crate::store::models::ControlStore; +use crate::warehouse_metadata::_helpers::events::ingest_events; +use crate::warehouse_metadata::_helpers::ingestion::batch_records; +use crate::warehouse_metadata::models::{ + MetadataBatch, StateEventImport, WarehouseColumnMigrationRecord, WarehouseStateLocation, + WarehouseSyncResponse, +}; + +pub async fn ingest_column_migrations( + store: &ControlStore, + scope: (Uuid, &str), + batch: &MetadataBatch, +) -> Result { + let records: Vec = batch_records(batch)?; + ingest_events( + store, + scope, + batch, + ( + records + .iter() + .map(|record| StateEventImport { + source_id: record.event_id.clone(), + location: WarehouseStateLocation { + database: record.state_database.clone(), + schema: record.state_schema.clone(), + }, + observed_at: record.created_at, + columns: Vec::new(), + record_json: json!(record), + }) + .collect(), + records + .iter() + .flat_map(|record| [&record.event_id, &record.state_schema]) + .any(|value| value.trim().is_empty()), + ), + ) + .await +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_consumption.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_consumption.rs new file mode 100644 index 0000000000..7f8520483d --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_consumption.rs @@ -0,0 +1,57 @@ +//! Mirror consumed input versions; rows carry no event identity, so their identity columns derive one. + +use serde_json::json; +use sqlbuild_recall::records::main::sha256_hex::sha256_hex; +use uuid::Uuid; + +use crate::store::errors::ControlError; +use crate::store::models::ControlStore; +use crate::warehouse_metadata::_helpers::events::ingest_events; +use crate::warehouse_metadata::_helpers::ingestion::batch_records; +use crate::warehouse_metadata::models::{ + MetadataBatch, StateEventImport, WarehouseConsumptionRecord, WarehouseStateLocation, + WarehouseSyncResponse, +}; + +pub async fn ingest_consumption( + store: &ControlStore, + scope: (Uuid, &str), + batch: &MetadataBatch, +) -> Result { + let records: Vec = batch_records(batch)?; + ingest_events( + store, + scope, + batch, + ( + records + .iter() + .map(|record| StateEventImport { + source_id: sha256_hex( + json!([ + record.execution_id, + record.phase, + record.input_kind, + record.input_name, + record.input_version, + record.recorded_at, + ]) + .to_string() + .as_bytes(), + ), + location: WarehouseStateLocation { + database: record.state_database.clone(), + schema: record.state_schema.clone(), + }, + observed_at: record.recorded_at, + columns: Vec::new(), + record_json: json!(record), + }) + .collect(), + records + .iter() + .any(|record| record.state_schema.trim().is_empty()), + ), + ) + .await +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_fingerprints.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_fingerprints.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_fingerprints.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_fingerprints.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_janitor_events.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_janitor_events.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_janitor_events.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_janitor_events.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_microbatches.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_microbatches.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_microbatches.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_microbatches.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_migrations.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_migrations.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_migrations.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_migrations.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_node_results.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_node_results.rs new file mode 100644 index 0000000000..b1f69c33d2 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_node_results.rs @@ -0,0 +1,59 @@ +//! Mirror node outcomes; rows carry no event identity, so their identity columns derive one. + +use serde_json::json; +use sqlbuild_recall::records::main::sha256_hex::sha256_hex; +use uuid::Uuid; + +use crate::store::errors::ControlError; +use crate::store::models::ControlStore; +use crate::warehouse_metadata::_helpers::events::ingest_events; +use crate::warehouse_metadata::_helpers::ingestion::batch_records; +use crate::warehouse_metadata::models::{ + MetadataBatch, StateEventImport, WarehouseNodeResultRecord, WarehouseStateLocation, + WarehouseSyncResponse, +}; + +pub async fn ingest_node_results( + store: &ControlStore, + scope: (Uuid, &str), + batch: &MetadataBatch, +) -> Result { + let records: Vec = batch_records(batch)?; + ingest_events( + store, + scope, + batch, + ( + records + .iter() + .map(|record| StateEventImport { + source_id: sha256_hex( + json!([ + record.node_type, + record.node_name, + record.target_database, + record.target_schema, + record.target_name, + record.run_id, + record.status, + record.ts, + ]) + .to_string() + .as_bytes(), + ), + location: WarehouseStateLocation { + database: record.state_database.clone(), + schema: record.state_schema.clone(), + }, + observed_at: record.ts, + columns: Vec::new(), + record_json: json!(record), + }) + .collect(), + records + .iter() + .any(|record| record.state_schema.trim().is_empty()), + ), + ) + .await +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_old_name_views.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_old_name_views.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_old_name_views.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_old_name_views.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_source_freshness.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_source_freshness.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_source_freshness.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_source_freshness.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_statistics.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_statistics.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_statistics.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_statistics.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/mod.rs new file mode 100644 index 0000000000..aa95f66a37 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/mod.rs @@ -0,0 +1,13 @@ +pub mod ingest_audit_results; +pub mod ingest_catalog; +pub mod ingest_column_migrations; +pub mod ingest_consumption; +pub mod ingest_fingerprints; +pub mod ingest_janitor_events; +pub mod ingest_microbatches; +pub mod ingest_migrations; +pub mod ingest_node_results; +pub mod ingest_old_name_views; +pub mod ingest_source_freshness; +pub mod ingest_statistics; +pub(crate) mod prune_statistics; diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/prune_statistics.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/prune_statistics.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/prune_statistics.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/prune_statistics.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/location_reads/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/location_reads/mod.rs new file mode 100644 index 0000000000..af7a2a5d7a --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/location_reads/mod.rs @@ -0,0 +1,2 @@ +pub mod read_statistics_due; +pub mod read_watermark; diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_statistics_due.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/location_reads/read_statistics_due.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_statistics_due.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/location_reads/read_statistics_due.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_watermark.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/location_reads/read_watermark.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_watermark.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/location_reads/read_watermark.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs index f3f9f53dc5..d3bb217ec4 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs @@ -1,18 +1,3 @@ -pub mod ingest_audit_results; -pub mod ingest_catalog; -pub mod ingest_fingerprints; -pub mod ingest_janitor_events; -pub mod ingest_microbatches; -pub mod ingest_migrations; -pub mod ingest_old_name_views; -pub mod ingest_source_freshness; -pub mod ingest_statistics; -pub mod list_person_catalog_nodes; -pub(crate) mod prune_statistics; -pub mod read_person_catalog; -pub mod read_person_column_lineage; -pub mod read_person_fingerprint; -pub mod read_person_node_status; -pub mod read_person_statistics; -pub mod read_statistics_due; -pub mod read_watermark; +pub mod ingest; +pub mod location_reads; +pub mod person_reads; diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/list_person_catalog_nodes.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/list_person_catalog_nodes.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/list_person_catalog_nodes.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/list_person_catalog_nodes.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/mod.rs new file mode 100644 index 0000000000..eb7fc45d0f --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/mod.rs @@ -0,0 +1,6 @@ +pub mod list_person_catalog_nodes; +pub mod read_person_catalog; +pub mod read_person_column_lineage; +pub mod read_person_fingerprint; +pub mod read_person_node_status; +pub mod read_person_statistics; diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/read_person_catalog.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/read_person_catalog.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_column_lineage.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/read_person_column_lineage.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_column_lineage.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/read_person_column_lineage.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_fingerprint.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/read_person_fingerprint.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_fingerprint.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/read_person_fingerprint.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_node_status.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/read_person_node_status.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_node_status.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/read_person_node_status.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/read_person_statistics.rs similarity index 100% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs rename to crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/read_person_statistics.rs diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs index f0c0c2297e..d678e6d29a 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs @@ -4,6 +4,12 @@ use chrono::{DateTime, Duration, Utc}; use serde::{Deserialize, Serialize}; use uuid::Uuid; +use sqlbuild_core::constants::{ + AUDIT_RESULTS_TABLE_NAME, COLUMN_MIGRATION_TABLE_NAME, CONSUMPTION_TABLE_NAME, + FINGERPRINT_TABLE_NAME, JANITOR_EVENTS_TABLE_NAME, MICROBATCH_TABLE_NAME, MIGRATION_TABLE_NAME, + NODE_RESULTS_TABLE_NAME, OLD_NAME_VIEW_TABLE_NAME, SOURCE_FRESHNESS_TABLE_NAME, +}; + use crate::store::errors::ControlError; use crate::warehouse_metadata::constants::DEFAULT_STATISTICS_RETENTION_DAYS; @@ -202,6 +208,68 @@ pub struct WarehouseStateLocation { pub schema: String, } +/// Column renames a model migration decided; only the event identity and time are required. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +#[serde(deny_unknown_fields)] +pub struct WarehouseColumnMigrationRecord { + pub state_database: Option, + pub state_schema: String, + pub event_id: String, + pub target_name: Option, + pub model_name: Option, + pub relation_database: Option, + pub relation_schema: Option, + pub relation_name: Option, + pub origin_column: Option, + pub destination_column: Option, + pub discovery: Option, + pub decision: Option, + pub run_id: Option, + pub created_at: DateTime, +} + +/// One node's recorded outcome in one run, with its encoded payloads kept as written. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +#[serde(deny_unknown_fields)] +pub struct WarehouseNodeResultRecord { + pub state_database: Option, + pub state_schema: String, + pub node_type: Option, + pub node_name: Option, + pub target_database: Option, + pub target_schema: Option, + pub target_name: Option, + pub run_id: Option, + pub status: Option, + pub payload_json_b64: Option, + pub metadata_json_b64: Option, + pub error_message: Option, + pub materialized: Option, + pub ts: DateTime, +} + +/// One input version an execution consumed, or the row marking that the execution started. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +#[serde(deny_unknown_fields)] +pub struct WarehouseConsumptionRecord { + pub state_database: Option, + pub state_schema: String, + pub execution_id: Option, + pub phase: Option, + pub run_id: Option, + pub node_type: Option, + pub node_name: Option, + pub target_database: Option, + pub target_schema: Option, + pub target_name: Option, + pub version_hash: Option, + pub input_count: Option, + pub input_kind: Option, + pub input_name: Option, + pub input_version: Option, + pub recorded_at: DateTime, +} + /// Catalog observations require full scans; only immutable history supports a watermark. #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] #[serde(rename_all = "snake_case")] @@ -213,12 +281,16 @@ pub enum WarehouseHistoryKind { Migrations, OldNameViews, Microbatches, + ColumnMigrations, + NodeResults, + Consumption, Catalog, Statistics, } impl WarehouseHistoryKind { - pub const ALL: [Self; 9] = [ + /// Catalog and statistics stay last: a failed statistics read rewinds one kind, to catalog. + pub const ALL: [Self; 12] = [ Self::Fingerprints, Self::SourceFreshness, Self::AuditResults, @@ -226,38 +298,85 @@ impl WarehouseHistoryKind { Self::Migrations, Self::OldNameViews, Self::Microbatches, + Self::ColumnMigrations, + Self::NodeResults, + Self::Consumption, Self::Catalog, Self::Statistics, ]; - #[must_use] - pub const fn as_str(self) -> &'static str { + /// Wire name, control-store mirror table and warehouse state table of each kind. + const fn names(self) -> (&'static str, &'static str, Option<&'static str>) { match self { - Self::Fingerprints => "fingerprints", - Self::SourceFreshness => "source_freshness", - Self::AuditResults => "audit_results", - Self::JanitorEvents => "janitor_events", - Self::Migrations => "migrations", - Self::OldNameViews => "old_name_views", - Self::Microbatches => "microbatches", - Self::Catalog => "catalog", - Self::Statistics => "statistics", + Self::Fingerprints => ( + "fingerprints", + "sqb_warehouse_fingerprints", + Some(FINGERPRINT_TABLE_NAME), + ), + Self::SourceFreshness => ( + "source_freshness", + "sqb_warehouse_freshness", + Some(SOURCE_FRESHNESS_TABLE_NAME), + ), + Self::AuditResults => ( + "audit_results", + "sqb_warehouse_audit_results", + Some(AUDIT_RESULTS_TABLE_NAME), + ), + Self::JanitorEvents => ( + "janitor_events", + "sqb_warehouse_janitor_events", + Some(JANITOR_EVENTS_TABLE_NAME), + ), + Self::Migrations => ( + "migrations", + "sqb_warehouse_migrations", + Some(MIGRATION_TABLE_NAME), + ), + Self::OldNameViews => ( + "old_name_views", + "sqb_warehouse_old_name_views", + Some(OLD_NAME_VIEW_TABLE_NAME), + ), + Self::Microbatches => ( + "microbatches", + "sqb_warehouse_microbatches", + Some(MICROBATCH_TABLE_NAME), + ), + Self::ColumnMigrations => ( + "column_migrations", + "sqb_warehouse_column_migrations", + Some(COLUMN_MIGRATION_TABLE_NAME), + ), + Self::NodeResults => ( + "node_results", + "sqb_warehouse_node_results", + Some(NODE_RESULTS_TABLE_NAME), + ), + Self::Consumption => ( + "consumption", + "sqb_warehouse_consumption", + Some(CONSUMPTION_TABLE_NAME), + ), + Self::Catalog => ("catalog", "sqb_warehouse_catalog", None), + Self::Statistics => ("statistics", "sqb_warehouse_statistics", None), } } + #[must_use] + pub const fn as_str(self) -> &'static str { + self.names().0 + } + + /// The state table a kind mirrors; catalog and statistics read the warehouse catalog instead. + #[must_use] + pub const fn source_table(self) -> Option<&'static str> { + self.names().2 + } + #[must_use] pub const fn mirror_table(self) -> &'static str { - match self { - Self::Fingerprints => "sqb_warehouse_fingerprints", - Self::SourceFreshness => "sqb_warehouse_freshness", - Self::AuditResults => "sqb_warehouse_audit_results", - Self::JanitorEvents => "sqb_warehouse_janitor_events", - Self::Migrations => "sqb_warehouse_migrations", - Self::OldNameViews => "sqb_warehouse_old_name_views", - Self::Microbatches => "sqb_warehouse_microbatches", - Self::Catalog => "sqb_warehouse_catalog", - Self::Statistics => "sqb_warehouse_statistics", - } + self.names().1 } } @@ -346,6 +465,26 @@ pub struct FingerprintRead { pub limit: Option, } +/// Node results carry no event identity, so the cursor holds the row's identity columns. +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct NodeResultsRead { + pub location: WarehouseStateLocation, + pub after: Option<[Option; 9]>, + pub observed_since: Option>, + pub limit: Option, +} + +/// Consumption rows carry no event identity, so the cursor holds the row's identity columns. +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ConsumptionRead { + pub location: WarehouseStateLocation, + pub after: Option<[Option; 7]>, + pub observed_since: Option>, + pub limit: Option, +} + /// Stable event identities and timestamps order observations without comparing their payloads. #[derive(Clone, Debug, Serialize, Deserialize)] #[serde(deny_unknown_fields)] @@ -367,6 +506,9 @@ pub enum WarehouseHistoryRead { Migrations(StateEventsRead), OldNameViews(StateEventsRead), Microbatches(StateEventsRead), + ColumnMigrations(StateEventsRead), + NodeResults(NodeResultsRead), + Consumption(ConsumptionRead), Catalog(CatalogRead), Statistics(StatisticsRead), } @@ -381,6 +523,9 @@ pub enum WarehouseHistoryRecord { Migration(Box), OldNameView(Box), Microbatch(Box), + ColumnMigration(Box), + NodeResult(Box), + Consumption(Box), Catalog(WarehouseCatalogRecord), Statistics(WarehouseStatisticsRecord), } diff --git a/crates/sqlbuild-orchestration/tests/api/fingerprint_reads/helpers.rs b/crates/sqlbuild-orchestration/tests/api/fingerprint_reads/helpers.rs index 3018d28bb1..45eef6062a 100644 --- a/crates/sqlbuild-orchestration/tests/api/fingerprint_reads/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/api/fingerprint_reads/helpers.rs @@ -9,7 +9,8 @@ use crate::platform::models::{LocationHealth, LocationHealthReport}; use crate::store::errors::ControlError; use crate::store::models::ControlStore; use crate::warehouse_metadata::main::{ - ingest_fingerprints::ingest_fingerprints, read_person_fingerprint::read_person_fingerprint, + ingest::ingest_fingerprints::ingest_fingerprints, + person_reads::read_person_fingerprint::read_person_fingerprint, }; use crate::warehouse_metadata::models::{FingerprintQuery, MetadataBatch}; diff --git a/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs b/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs index 00d6459944..8287aa9c47 100644 --- a/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs @@ -14,10 +14,10 @@ use crate::store::errors::ControlError; use crate::store::main::{begin::begin, commit::commit, query::query}; use crate::store::models::{ControlStore, Parameter, TransactionMode}; use crate::warehouse_metadata::main::{ - ingest_catalog::ingest_catalog, ingest_fingerprints::ingest_fingerprints, - ingest_source_freshness::ingest_source_freshness, - list_person_catalog_nodes::list_person_catalog_nodes, - read_person_node_status::read_person_node_status, + ingest::ingest_catalog::ingest_catalog, ingest::ingest_fingerprints::ingest_fingerprints, + ingest::ingest_source_freshness::ingest_source_freshness, + person_reads::list_person_catalog_nodes::list_person_catalog_nodes, + person_reads::read_person_node_status::read_person_node_status, }; use crate::warehouse_metadata::models::{MetadataBatch, NodeStatusQuery}; diff --git a/crates/sqlbuild-orchestration/tests/api/statistics/eligibility/helpers.rs b/crates/sqlbuild-orchestration/tests/api/statistics/eligibility/helpers.rs index 84a61e7b88..e9f5fb4264 100644 --- a/crates/sqlbuild-orchestration/tests/api/statistics/eligibility/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/api/statistics/eligibility/helpers.rs @@ -7,8 +7,9 @@ use uuid::Uuid; use crate::credentials::models::{AuditContext, DefaultRole}; use crate::store::models::ControlStore; use crate::warehouse_metadata::main::{ - ingest_statistics::ingest_statistics, read_person_statistics::read_person_statistics, - read_statistics_due::read_statistics_due, + ingest::ingest_statistics::ingest_statistics, + location_reads::read_statistics_due::read_statistics_due, + person_reads::read_person_statistics::read_person_statistics, }; use crate::warehouse_metadata::models::{ MetadataBatch, StatisticsDueRead, StatisticsQuery, WarehouseStateLocation, diff --git a/crates/sqlbuild-orchestration/tests/api/statistics/helpers.rs b/crates/sqlbuild-orchestration/tests/api/statistics/helpers.rs index eb74eba6b0..31b328458b 100644 --- a/crates/sqlbuild-orchestration/tests/api/statistics/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/api/statistics/helpers.rs @@ -9,7 +9,8 @@ use crate::store::errors::ControlError; use crate::store::main::{begin::begin, commit::commit, query::query}; use crate::store::models::{ControlStore, TransactionMode}; use crate::warehouse_metadata::main::{ - ingest_statistics::ingest_statistics, read_person_statistics::read_person_statistics, + ingest::ingest_statistics::ingest_statistics, + person_reads::read_person_statistics::read_person_statistics, }; use crate::warehouse_metadata::models::{MetadataBatch, StatisticsQuery}; diff --git a/crates/sqlbuild-orchestration/tests/api/statistics/retention/helpers.rs b/crates/sqlbuild-orchestration/tests/api/statistics/retention/helpers.rs index 3f63670636..179fd207a3 100644 --- a/crates/sqlbuild-orchestration/tests/api/statistics/retention/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/api/statistics/retention/helpers.rs @@ -6,9 +6,10 @@ use uuid::Uuid; use crate::credentials::models::{AuditContext, DefaultRole}; use crate::store::models::ControlStore; -use crate::warehouse_metadata::main::prune_statistics::prune_statistics; +use crate::warehouse_metadata::main::ingest::prune_statistics::prune_statistics; use crate::warehouse_metadata::main::{ - ingest_statistics::ingest_statistics, read_person_statistics::read_person_statistics, + ingest::ingest_statistics::ingest_statistics, + person_reads::read_person_statistics::read_person_statistics, }; use crate::warehouse_metadata::models::{MetadataBatch, StatisticsQuery, StatisticsRetention}; diff --git a/crates/sqlbuild-orchestration/tests/api/test_catalog_reads.rs b/crates/sqlbuild-orchestration/tests/api/test_catalog_reads.rs index c46382999f..96bfd0d3d9 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_catalog_reads.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_catalog_reads.rs @@ -24,7 +24,7 @@ use crate::store::main::migrate_store::migrate_store; use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; use crate::store::tests::store_test_types::Backend; use crate::warehouse_metadata::main::{ - ingest_catalog::ingest_catalog, read_person_catalog::read_person_catalog, + ingest::ingest_catalog::ingest_catalog, person_reads::read_person_catalog::read_person_catalog, }; use crate::warehouse_metadata::models::{CatalogQuery, MetadataBatch}; @@ -307,7 +307,7 @@ fn given_catalog_observations_when_people_page_then_live_grants_and_exact_namesp .await .expect("revoke"); assert!(matches!( - crate::warehouse_metadata::main::read_person_statistics::read_person_statistics( + crate::warehouse_metadata::main::person_reads::read_person_statistics::read_person_statistics( store, credentials, ( @@ -328,7 +328,7 @@ fn given_catalog_observations_when_people_page_then_live_grants_and_exact_namesp Err(ControlError::Denied { .. }) )); assert!(matches!( - crate::warehouse_metadata::main::read_person_node_status::read_person_node_status( + crate::warehouse_metadata::main::person_reads::read_person_node_status::read_person_node_status( store, credentials, ( @@ -351,7 +351,7 @@ fn given_catalog_observations_when_people_page_then_live_grants_and_exact_namesp name: Some("orders".into()), }; assert!(matches!( - crate::warehouse_metadata::main::read_person_fingerprint::read_person_fingerprint( + crate::warehouse_metadata::main::person_reads::read_person_fingerprint::read_person_fingerprint( store, credentials, (location, "development", &evidence), diff --git a/crates/sqlbuild-orchestration/tests/api/test_checks.rs b/crates/sqlbuild-orchestration/tests/api/test_checks.rs index 46e3c3bf7a..2c1c86bbc8 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_checks.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_checks.rs @@ -13,7 +13,7 @@ use crate::store::tests::fault_helpers::{install_fault, remove_fault}; use crate::store::tests::fault_types::TransactionFault; use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; use crate::store::tests::store_test_types::Backend; -use crate::warehouse_metadata::main::ingest_audit_results::ingest_audit_results; +use crate::warehouse_metadata::main::ingest::ingest_audit_results::ingest_audit_results; #[test] fn given_checks_sharing_an_audit_name_when_people_read_them_then_binding_keys_and_sql_permissions_hold() diff --git a/crates/sqlbuild-orchestration/tests/api/test_column_lineage.rs b/crates/sqlbuild-orchestration/tests/api/test_column_lineage.rs index 68593a8c42..eb255b6e40 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_column_lineage.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_column_lineage.rs @@ -15,7 +15,7 @@ use crate::store::errors::ControlError; use crate::store::main::migrate_store::migrate_store; use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; use crate::store::tests::store_test_types::Backend; -use crate::warehouse_metadata::main::read_person_column_lineage::read_person_column_lineage; +use crate::warehouse_metadata::main::person_reads::read_person_column_lineage::read_person_column_lineage; #[test] fn given_published_column_lineage_when_people_read_a_node_then_only_its_reachable_columns_return() { diff --git a/crates/sqlbuild-orchestration/tests/api/test_maintenance.rs b/crates/sqlbuild-orchestration/tests/api/test_maintenance.rs index f395d47821..9fe6f671fa 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_maintenance.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_maintenance.rs @@ -15,7 +15,7 @@ use crate::runs::tests::helpers::snapshot; use crate::snapshots::main::publish_snapshot::publish_snapshot; use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; use crate::store::tests::store_test_types::Backend; -use crate::warehouse_metadata::main::{ +use crate::warehouse_metadata::main::ingest::{ ingest_janitor_events::ingest_janitor_events, ingest_migrations::ingest_migrations, ingest_old_name_views::ingest_old_name_views, }; diff --git a/crates/sqlbuild-orchestration/tests/api/test_microbatch_coverage.rs b/crates/sqlbuild-orchestration/tests/api/test_microbatch_coverage.rs index 79bffe3bbc..734f5b50e3 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_microbatch_coverage.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_microbatch_coverage.rs @@ -10,7 +10,7 @@ use crate::api::tests::warehouse_history_helpers::{ }; use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; use crate::store::tests::store_test_types::Backend; -use crate::warehouse_metadata::main::ingest_microbatches::ingest_microbatches; +use crate::warehouse_metadata::main::ingest::ingest_microbatches::ingest_microbatches; const HOUR_0: &str = "2026-01-01 00:00:00+00"; const HOUR_1: &str = "2026-01-01T09:00:00+08:00"; diff --git a/crates/sqlbuild-orchestration/tests/api/test_types.rs b/crates/sqlbuild-orchestration/tests/api/test_types.rs index e7c650905a..94f15bc1ef 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_types.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_types.rs @@ -1,6 +1,7 @@ use crate::runs::models::RunStatus; use crate::store::errors::ControlError; use crate::store::tests::store_test_types::Backend; +use crate::warehouse_metadata::models::WarehouseHistoryKind; pub(crate) struct ApiContractTestCase { pub(crate) description: &'static str, @@ -103,6 +104,12 @@ pub(crate) struct ColumnLineageTestCase { pub(crate) expected_leaf_edges: usize, } +pub(crate) struct StateTableClassificationTestCase { + pub(crate) description: &'static str, + pub(crate) expected_tables: Vec<&'static str>, + pub(crate) expected_last_kinds: [WarehouseHistoryKind; 2], +} + pub(crate) struct WarehouseMetadataTestCase { pub(crate) description: &'static str, pub(crate) backend: Backend, diff --git a/crates/sqlbuild-orchestration/tests/api/test_warehouse_metadata.rs b/crates/sqlbuild-orchestration/tests/api/test_warehouse_metadata.rs index ef1a0ad714..9aae0418f6 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_warehouse_metadata.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_warehouse_metadata.rs @@ -3,11 +3,12 @@ use chrono::{Duration, Utc}; use reqwest::StatusCode; use serde_json::json; +use sqlbuild_core::constants::STATE_TABLE_NAMES; use uuid::Uuid; use crate::api::tests::changes_helpers::{connect, expect_topics}; use crate::api::tests::helpers::{start_api, stop_api}; -use crate::api::tests::test_types::WarehouseMetadataTestCase; +use crate::api::tests::test_types::{StateTableClassificationTestCase, WarehouseMetadataTestCase}; use crate::credentials::main::tokens::issue_token::issue_token; use crate::credentials::models::{AuditContext, CredentialScope}; use crate::credentials::tests::location_helpers::create_location; @@ -26,6 +27,38 @@ use crate::store::tests::store_helpers::{close_test_store, open_test_store, run} use crate::store::tests::store_test_types::Backend; use crate::warehouse_metadata::models::{MetadataBatch, WarehouseHistoryKind}; +#[test] +fn given_state_tables_when_classified_then_each_is_mirrored_by_exactly_one_kind() { + let mut written: Vec<&str> = STATE_TABLE_NAMES.to_vec(); + written.sort_unstable(); + let test_cases = [StateTableClassificationTestCase { + description: "every state table SQLBuild writes, with statistics rewinding to catalog", + expected_tables: written, + expected_last_kinds: [ + WarehouseHistoryKind::Catalog, + WarehouseHistoryKind::Statistics, + ], + }]; + for test_case in test_cases { + let mut mirrored: Vec<&str> = WarehouseHistoryKind::ALL + .iter() + .filter_map(|kind| kind.source_table()) + .collect(); + mirrored.sort_unstable(); + assert_eq!( + mirrored, test_case.expected_tables, + "{}", + test_case.description + ); + assert_eq!( + WarehouseHistoryKind::ALL[WarehouseHistoryKind::ALL.len() - 2..], + test_case.expected_last_kinds, + "{}", + test_case.description + ); + } +} + #[test] fn given_warehouse_observations_when_pages_retry_then_scope_history_and_atomicity_are_preserved() { let test_cases = [ @@ -348,6 +381,21 @@ fn given_warehouse_observations_when_pages_retry_then_scope_history_and_atomicit "microbatch_history", "event_id", ), + ( + WarehouseHistoryKind::ColumnMigrations, + "column_migration_history", + "event_id", + ), + ( + WarehouseHistoryKind::NodeResults, + "node_result_history", + "node_name", + ), + ( + WarehouseHistoryKind::Consumption, + "consumption_history", + "execution_id", + ), ] { crate::api::tests::warehouse_events_helpers::event_history( store, diff --git a/crates/sqlbuild-orchestration/tests/api/warehouse_metadata/events/helpers.rs b/crates/sqlbuild-orchestration/tests/api/warehouse_metadata/events/helpers.rs index 3a437d2cb4..c8f58887a4 100644 --- a/crates/sqlbuild-orchestration/tests/api/warehouse_metadata/events/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/api/warehouse_metadata/events/helpers.rs @@ -136,7 +136,8 @@ pub(crate) async fn event_history( description: "event page rolls back atomically", table: kind.mirror_table(), operation: "INSERT", - condition: "WHEN 'rejected-result' IN (NEW.record_json ->> 'event_id', NEW.record_json ->> 'result_id')", + condition: "WHEN 'rejected-result' IN (NEW.record_json ->> 'event_id', NEW.record_json ->> 'result_id', \ + NEW.record_json ->> 'node_name', NEW.record_json ->> 'execution_id')", sqlite_body: "SELECT RAISE(ABORT, 'injected audit failure')", postgres_body: "RAISE EXCEPTION 'injected audit failure'", }; diff --git a/crates/sqlbuild-orchestration/tests/credentials/roles/helpers.rs b/crates/sqlbuild-orchestration/tests/credentials/roles/helpers.rs index fa5ba26b5e..aeb4358e1c 100644 --- a/crates/sqlbuild-orchestration/tests/credentials/roles/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/credentials/roles/helpers.rs @@ -29,8 +29,8 @@ use crate::snapshots::main::publish_snapshot::publish_snapshot; use crate::store::errors::ControlError; use crate::store::main::{begin::begin, commit::commit, query::query}; use crate::store::models::{ControlStore, TransactionMode}; -use crate::warehouse_metadata::main::read_person_catalog::read_person_catalog; -use crate::warehouse_metadata::main::read_person_statistics::read_person_statistics; +use crate::warehouse_metadata::main::person_reads::read_person_catalog::read_person_catalog; +use crate::warehouse_metadata::main::person_reads::read_person_statistics::read_person_statistics; use crate::warehouse_metadata::models::{CatalogQuery, StatisticsQuery, StatisticsRetention}; pub(crate) const SUPPORT_ROLE: &str = "orders_support"; diff --git a/crates/sqlbuild-orchestration/tests/sensors/test_sources.rs b/crates/sqlbuild-orchestration/tests/sensors/test_sources.rs index 2f4831ba58..d16bcc357d 100644 --- a/crates/sqlbuild-orchestration/tests/sensors/test_sources.rs +++ b/crates/sqlbuild-orchestration/tests/sensors/test_sources.rs @@ -28,7 +28,7 @@ use crate::store::tests::fault_helpers::{install_fault, remove_fault}; use crate::store::tests::fault_types::TransactionFault; use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; use crate::store::tests::store_test_types::Backend; -use crate::warehouse_metadata::main::ingest_source_freshness::ingest_source_freshness; +use crate::warehouse_metadata::main::ingest::ingest_source_freshness::ingest_source_freshness; use crate::warehouse_metadata::models::{ MetadataBatch, MetadataScan, WarehouseFreshnessRecord, WarehouseHistoryKind, WarehouseStateLocation, diff --git a/crates/sqlbuild-python/src/bindings/_helpers/native_host/warehouse_catalog.rs b/crates/sqlbuild-python/src/bindings/_helpers/native_host/warehouse_catalog.rs index 112caea0a2..1fd2b10c0b 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/native_host/warehouse_catalog.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/native_host/warehouse_catalog.rs @@ -3,7 +3,7 @@ use pyo3::IntoPyObjectExt; use pyo3::exceptions::PyValueError; use pyo3::prelude::{Bound, PyAny, PyAnyMethods, PyResult, Python}; -use pyo3::types::{PyDict, PyDictMethods, PyTuple}; +use pyo3::types::{PyDict, PyDictMethods, PyFrozenSet, PyTuple}; use sqlbuild_orchestration::warehouse_metadata::constants::MAX_HISTORY_RECORDS; use sqlbuild_orchestration::warehouse_metadata::models::{ CatalogRead, WarehouseCatalogColumn, WarehouseCatalogRecord, WarehouseHistoryPage, @@ -49,23 +49,43 @@ pub(crate) fn read_catalog( "catalog listing is not authoritative for this namespace; scan incomplete", )); } - let relations = call_with( - &adapter.getattr("list_relations")?, - &[ - ("connection", connection.clone()), - ( - "database", - request.location.database.clone().into_bound_py_any(py)?, - ), - ( - "schemas", - (request.location.schema.clone(),).into_bound_py_any(py)?, - ), - ], + let schema_scoped = adapter.is_instance( + &py.import("sqlbuild.adapter.relations.types")? + .getattr("SchemaScopedMetadataReader")?, )?; - let mut selected: Vec<(String, Bound<'_, PyAny>)> = Vec::new(); - for relation in relations.try_iter()? { - let relation = relation?; + let mut selected: Vec<(String, String, Bound<'_, PyAny>)> = Vec::new(); + let listed = if schema_scoped { + field( + &call_with( + &adapter.getattr("read_schema_relation_listing")?, + &[ + ("connection", connection.clone()), + namespace[0].clone(), + namespace[1].clone(), + ], + )?, + "entries", + )? + } else { + call_with( + &adapter.getattr("list_relations")?, + &[ + ("connection", connection.clone()), + namespace[0].clone(), + ( + "schemas", + (request.location.schema.clone(),).into_bound_py_any(py)?, + ), + ], + )? + }; + for item in listed.try_iter()? { + let item = item?; + let (relation, key) = if schema_scoped { + (field(&item, "relation")?, field(&item, "stored_name")?) + } else { + (item.clone(), field(&item, "name")?) + }; let name: String = field(&relation, "name")?.extract()?; if request .after @@ -73,7 +93,8 @@ pub(crate) fn read_catalog( .and_then(|cursor| cursor[0].as_deref()) .is_none_or(|after| name.as_str() > after) { - selected.push((name, relation)); + let relation_type: String = field(&relation, "relation_type")?.extract()?; + selected.push((name, relation_type, key)); } } selected.sort_by(|left, right| left.0.cmp(&right.0)); @@ -85,28 +106,43 @@ pub(crate) fn read_catalog( next_after: None, }); } - let names = PyTuple::new(py, selected.iter().map(|(name, _)| name))?; - let columns = call_with( - &adapter.getattr("get_all_columns")?, - &[ - ("connection", connection.clone()), - ("database", namespace[0].1.clone()), - ( - "schemas", - (request.location.schema.clone(),).into_bound_py_any(py)?, - ), - ("names", names.into_any()), - ], - )?; + let keys = selected.iter().map(|(_, _, key)| key); + let columns = if schema_scoped { + field( + &call_with( + &adapter.getattr("read_schema_column_listing")?, + &[ + ("connection", connection.clone()), + namespace[0].clone(), + namespace[1].clone(), + ("stored_names", PyFrozenSet::new(py, keys)?.into_any()), + ], + )?, + "columns_by_stored_name", + )? + } else { + call_with( + &adapter.getattr("get_all_columns")?, + &[ + ("connection", connection.clone()), + namespace[0].clone(), + ( + "schemas", + (request.location.schema.clone(),).into_bound_py_any(py)?, + ), + ("names", PyTuple::new(py, keys)?.into_any()), + ], + )? + }; let columns = columns.cast::()?; let next_after = if more { - selected.last().map(|(name, _)| vec![Some(name.clone())]) + selected.last().map(|(name, _, _)| vec![Some(name.clone())]) } else { None }; let mut records: Vec = Vec::new(); - for (name, relation) in selected { - let relation_columns = columns.get_item(name.as_str())?.ok_or_else(|| { + for (name, relation_type, key) in selected { + let relation_columns = columns.get_item(key)?.ok_or_else(|| { PyValueError::new_err("catalog relation columns are unavailable; scan incomplete") })?; let mut decoded: Vec = Vec::new(); @@ -121,7 +157,7 @@ pub(crate) fn read_catalog( state_database: request.location.database.clone(), state_schema: request.location.schema.clone(), name, - relation_type: field(&relation, "relation_type")?.extract()?, + relation_type, columns: decoded, })); } diff --git a/crates/sqlbuild-python/src/bindings/_helpers/native_host/warehouse_metadata.rs b/crates/sqlbuild-python/src/bindings/_helpers/native_host/warehouse_metadata.rs index a5c265e81e..b00d15d6b8 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/native_host/warehouse_metadata.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/native_host/warehouse_metadata.rs @@ -8,13 +8,20 @@ use sqlbuild_analysis::fingerprint_storage::constants::{COLUMN_TIMESTAMP, FINGER use sqlbuild_config::command_connection::constants::{ DUCKDB_ADAPTER, DUCKDB_MEMORY_DATABASE, LOCAL_DATABASE_PATH_KEYS, }; -use sqlbuild_core::constants::{FINGERPRINT_TABLE_NAME, SOURCE_FRESHNESS_TABLE_NAME}; +use sqlbuild_core::constants::{ + COLUMN_MIGRATION_TABLE_NAME, CONSUMPTION_TABLE_NAME, FINGERPRINT_TABLE_NAME, + NODE_RESULTS_TABLE_NAME, SOURCE_FRESHNESS_TABLE_NAME, +}; use sqlbuild_core::models::SqlValue; use sqlbuild_orchestration::warehouse_metadata::constants::MAX_HISTORY_RECORDS; use sqlbuild_orchestration::warehouse_metadata::models::{ - WarehouseHistoryPage, WarehouseHistoryRead, WarehouseHistoryRecord, WarehouseStateLocation, + ConsumptionRead, NodeResultsRead, WarehouseHistoryPage, WarehouseHistoryRead, + WarehouseHistoryRecord, WarehouseStateLocation, +}; +use sqlbuild_state::consumption::constants::{CONSUMPTION_COLUMN_TYPES, RECORDED_AT_COLUMN}; +use sqlbuild_state::events::constants::{ + COLUMN_MIGRATION_COLUMNS, MIGRATION_COLUMNS, NODE_RESULT_COLUMNS, OLD_NAME_VIEW_COLUMNS, }; -use sqlbuild_state::events::constants::{MIGRATION_COLUMNS, OLD_NAME_VIEW_COLUMNS}; use sqlbuild_state::freshness::constants::{ OBSERVED_AT_COLUMN, SOURCE_FRESHNESS_COLUMN_TYPES, SQLITE_ADAPTER, }; @@ -50,7 +57,10 @@ pub(crate) fn read_history( | WarehouseHistoryRead::JanitorEvents(read) | WarehouseHistoryRead::Migrations(read) | WarehouseHistoryRead::OldNameViews(read) - | WarehouseHistoryRead::Microbatches(read) => &read.location, + | WarehouseHistoryRead::Microbatches(read) + | WarehouseHistoryRead::ColumnMigrations(read) => &read.location, + WarehouseHistoryRead::NodeResults(read) => &read.location, + WarehouseHistoryRead::Consumption(read) => &read.location, }; if location.schema.trim().is_empty() { return Err(PyValueError::new_err( @@ -110,6 +120,7 @@ fn read_page( request: &WarehouseHistoryRead, ) -> PyResult { let freshness_columns = SOURCE_FRESHNESS_COLUMN_TYPES.map(|(name, _)| name); + let consumption_columns = CONSUMPTION_COLUMN_TYPES.map(|(name, _)| name); let event_columns: Vec = match request { WarehouseHistoryRead::AuditResults(_) => py .import("sqlbuild.executor.audit_results.constants")? @@ -196,6 +207,33 @@ fn read_page( event_columns.as_slice(), "created_at", ), + WarehouseHistoryRead::ColumnMigrations(read) => ( + &read.location, + read.after.as_ref().map(|cursor| cursor.as_slice()), + read.observed_since, + read.limit, + COLUMN_MIGRATION_TABLE_NAME, + COLUMN_MIGRATION_COLUMNS.as_slice(), + "created_at", + ), + WarehouseHistoryRead::NodeResults(read) => ( + &read.location, + read.after.as_ref().map(|cursor| cursor.as_slice()), + read.observed_since, + read.limit, + NODE_RESULTS_TABLE_NAME, + NODE_RESULT_COLUMNS.as_slice(), + "ts", + ), + WarehouseHistoryRead::Consumption(read) => ( + &read.location, + read.after.as_ref().map(|cursor| cursor.as_slice()), + read.observed_since, + read.limit, + CONSUMPTION_TABLE_NAME, + consumption_columns.as_slice(), + RECORDED_AT_COLUMN, + ), }; let limit = limit.unwrap_or(MAX_HISTORY_RECORDS); if limit == 0 || limit > MAX_HISTORY_RECORDS { @@ -268,7 +306,26 @@ fn read_page( WarehouseHistoryRead::JanitorEvents(_) => Some(&["event_id", "occurred_at"]), WarehouseHistoryRead::Migrations(_) | WarehouseHistoryRead::OldNameViews(_) - | WarehouseHistoryRead::Microbatches(_) => Some(&["event_id", "created_at"]), + | WarehouseHistoryRead::Microbatches(_) + | WarehouseHistoryRead::ColumnMigrations(_) => Some(&["event_id", "created_at"]), + WarehouseHistoryRead::NodeResults(_) => Some(&[ + "node_type", + "node_name", + "target_database", + "target_schema", + "target_name", + "run_id", + "status", + "ts", + ]), + WarehouseHistoryRead::Consumption(_) => Some(&[ + "execution_id", + "phase", + "input_kind", + "input_name", + "input_version", + RECORDED_AT_COLUMN, + ]), _ => None, }; let mut cursor_indexes: Vec = columns @@ -389,6 +446,9 @@ fn read_page( WarehouseHistoryRecord::Migration(record) => record.created_at, WarehouseHistoryRecord::OldNameView(record) => record.created_at, WarehouseHistoryRecord::Microbatch(record) => record.created_at, + WarehouseHistoryRecord::ColumnMigration(record) => record.created_at, + WarehouseHistoryRecord::NodeResult(record) => record.ts, + WarehouseHistoryRecord::Consumption(record) => record.recorded_at, }; if since.is_none_or(|since| observed >= since) { records.push(record); @@ -427,9 +487,15 @@ fn decode_record( qualified, )?, ), - WarehouseHistoryRead::AuditResults(read) | WarehouseHistoryRead::Migrations(read) => { + WarehouseHistoryRead::AuditResults(read) + | WarehouseHistoryRead::Migrations(read) + | WarehouseHistoryRead::ColumnMigrations(read) => { return decode_event_record(&read.location, row, (columns, &[timestamp])); } + WarehouseHistoryRead::NodeResults(NodeResultsRead { location, .. }) + | WarehouseHistoryRead::Consumption(ConsumptionRead { location, .. }) => { + return decode_event_record(location, row, (columns, &[timestamp])); + } WarehouseHistoryRead::JanitorEvents(read) => { return decode_event_record( &read.location, diff --git a/crates/sqlbuild-state/src/events/constants.rs b/crates/sqlbuild-state/src/events/constants.rs index c89814977d..51b8b09a1d 100644 --- a/crates/sqlbuild-state/src/events/constants.rs +++ b/crates/sqlbuild-state/src/events/constants.rs @@ -34,6 +34,21 @@ pub const MIGRATION_COLUMNS: [&str; 15] = [ "run_id", "created_at", ]; +/// Columns of `_sqlbuild_node_results`, in the order the executor writes them. +pub const NODE_RESULT_COLUMNS: [&str; 12] = [ + "node_type", + "node_name", + "target_database", + "target_schema", + "target_name", + "run_id", + "status", + "payload_json_b64", + "metadata_json_b64", + "error_message", + "materialized", + "ts", +]; pub const COLUMN_MIGRATION_COLUMNS: [&str; 12] = [ "event_id", "target_name", diff --git a/crates/sqlbuild-state/src/state_tables/constants.rs b/crates/sqlbuild-state/src/state_tables/constants.rs index 8bf6191df2..1dd50fb1ef 100644 --- a/crates/sqlbuild-state/src/state_tables/constants.rs +++ b/crates/sqlbuild-state/src/state_tables/constants.rs @@ -1,3 +1,5 @@ //! Adapter decisions shared by append-only state history reads. pub(crate) const SQLSERVER_ADAPTER: &str = "sqlserver"; +/// Cursor key of a table without event IDs: always NULL, so it never orders or binds a value. +pub(crate) const MISSING_EVENT_ID_KEY: &str = "NULL"; diff --git a/crates/sqlbuild-state/src/state_tables/main/read_history_sql.rs b/crates/sqlbuild-state/src/state_tables/main/read_history_sql.rs index 8a415a7352..8739ededca 100644 --- a/crates/sqlbuild-state/src/state_tables/main/read_history_sql.rs +++ b/crates/sqlbuild-state/src/state_tables/main/read_history_sql.rs @@ -1,6 +1,6 @@ //! Keyset pages preserve distinct observations without rereading or ranking all preceding rows. -use crate::state_tables::constants::SQLSERVER_ADAPTER; +use crate::state_tables::constants::{MISSING_EVENT_ID_KEY, SQLSERVER_ADAPTER}; use crate::state_tables::models::HistoryRead; /// Cursor values are already rendered by the adapter; `None` sorts before every SQL value. @@ -25,11 +25,15 @@ pub fn read_history_sql( ", NULL AS event_id" }); let mut columns = key_columns.unwrap_or(columns).to_vec(); - columns.push(if has_event_id { "event_id" } else { "NULL" }); + columns.push(if has_event_id { + "event_id" + } else { + MISSING_EVENT_ID_KEY + }); let ordered = columns .iter() .map(|column| { - if *column == timestamp_column { + if *column == timestamp_column || *column == MISSING_EVENT_ID_KEY { Ok((*column).to_owned()) } else { binary(column, adapter) @@ -81,6 +85,7 @@ pub fn read_history_sql( let order: Vec = columns .iter() .zip(&ordered) + .filter(|(column, _)| **column != MISSING_EVENT_ID_KEY) .map(|(column, ordered)| format!("CASE WHEN {column} IS NULL THEN 0 ELSE 1 END, {ordered}")) .collect(); let top = adapter == SQLSERVER_ADAPTER; diff --git a/crates/sqlbuild-state/src/state_tables/tests/test_sql.rs b/crates/sqlbuild-state/src/state_tables/tests/test_sql.rs index 5c9ef41669..adcb4e0b11 100644 --- a/crates/sqlbuild-state/src/state_tables/tests/test_sql.rs +++ b/crates/sqlbuild-state/src/state_tables/tests/test_sql.rs @@ -1,7 +1,11 @@ use crate::state_tables::main::create_table_sql::create_table_sql; +use crate::state_tables::main::read_history_sql::read_history_sql; use crate::state_tables::main::values_insert_sql::values_insert_sql; +use crate::state_tables::models::HistoryRead; use crate::state_tables::tests::helpers::column; -use crate::state_tables::tests::test_types::{CreateTableTestCase, ValuesInsertTestCase}; +use crate::state_tables::tests::test_types::{ + CreateTableTestCase, HistoryReadTestCase, ValuesInsertTestCase, +}; #[test] fn given_columns_when_rendering_ddl_then_required_columns_are_not_null() { @@ -48,3 +52,61 @@ fn given_rows_when_rendering_insert_then_each_row_is_one_values_tuple() { ); } } + +#[test] +fn given_a_table_without_event_ids_when_paging_history_then_the_missing_key_never_orders_or_converts() + { + let test_cases = [ + HistoryReadTestCase { + description: "Snowflake rejects TO_BINARY of a bare NULL", + adapter: "snowflake", + has_event_id: false, + after: Some(vec![Some("'e'".to_owned()), Some("ts1".to_owned()), None]), + expected_sql: "SELECT k, ts, NULL AS event_id FROM s.t WHERE ((TO_BINARY(k, 'UTF-8') > \ + TO_BINARY('e', 'UTF-8')) OR (TO_BINARY(k, 'UTF-8') = TO_BINARY('e', 'UTF-8') AND ts > ts1) \ + OR (TO_BINARY(k, 'UTF-8') = TO_BINARY('e', 'UTF-8') AND ts = ts1 AND NULL IS NOT NULL)) \ + ORDER BY CASE WHEN k IS NULL THEN 0 ELSE 1 END, TO_BINARY(k, 'UTF-8'), \ + CASE WHEN ts IS NULL THEN 0 ELSE 1 END, ts LIMIT 5", + }, + HistoryReadTestCase { + description: "SQL Server rejects a constant in ORDER BY", + adapter: "sqlserver", + has_event_id: false, + after: None, + expected_sql: "SELECT TOP 5 k, ts, NULL AS event_id FROM s.t ORDER BY \ + CASE WHEN k IS NULL THEN 0 ELSE 1 END, CONVERT(VARBINARY(MAX), CONVERT(NVARCHAR(MAX), k)), \ + CASE WHEN ts IS NULL THEN 0 ELSE 1 END, ts", + }, + HistoryReadTestCase { + description: "a real event ID stays the last ordering key", + adapter: "snowflake", + has_event_id: true, + after: None, + expected_sql: "SELECT k, ts, event_id FROM s.t ORDER BY CASE WHEN k IS NULL THEN 0 ELSE 1 END, \ + TO_BINARY(k, 'UTF-8'), CASE WHEN ts IS NULL THEN 0 ELSE 1 END, ts, \ + CASE WHEN event_id IS NULL THEN 0 ELSE 1 END, TO_BINARY(event_id, 'UTF-8') LIMIT 5", + }, + ]; + + for test_case in test_cases { + assert_eq!( + read_history_sql( + "s.t", + test_case.after.as_deref(), + HistoryRead { + columns: &["k", "ts"], + key_columns: None, + timestamp_column: "ts", + observed_since: None, + adapter: test_case.adapter, + has_event_id: test_case.has_event_id, + limit: 5, + }, + ) + .expect("supported adapter"), + test_case.expected_sql, + "{}", + test_case.description + ); + } +} diff --git a/crates/sqlbuild-state/src/state_tables/tests/test_types.rs b/crates/sqlbuild-state/src/state_tables/tests/test_types.rs index 836ccf548d..6914593ee1 100644 --- a/crates/sqlbuild-state/src/state_tables/tests/test_types.rs +++ b/crates/sqlbuild-state/src/state_tables/tests/test_types.rs @@ -11,3 +11,11 @@ pub(crate) struct ValuesInsertTestCase { pub(crate) rows: Vec>, pub(crate) expected_sql: &'static str, } + +pub(crate) struct HistoryReadTestCase { + pub(crate) description: &'static str, + pub(crate) adapter: &'static str, + pub(crate) has_event_id: bool, + pub(crate) after: Option>>, + pub(crate) expected_sql: &'static str, +} diff --git a/fensu.toml b/fensu.toml index b2ee2457c4..7a8eac2005 100644 --- a/fensu.toml +++ b/fensu.toml @@ -1381,7 +1381,7 @@ paths = ["crates/sqlbuild-orchestration/src/api/_helpers/resources/administrator reason = "Each function is one utoipa-annotated axum route: OpenAPI needs one annotated handler per operation, and each body only checks the caller's scope and calls one store operation. Scope checks and store work live in their shared owners." [[dupes.allowlist]] -paths = ["crates/sqlbuild-orchestration/src/authorization/main/list_role_definitions.rs", "crates/sqlbuild-orchestration/src/authorization/main/read_security_audit.rs", "crates/sqlbuild-orchestration/src/credentials/_helpers/active_session.rs", "crates/sqlbuild-orchestration/src/declarations/_helpers/reads.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_person_jobs.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_person_trigger_changes.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_person_triggers.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_triggers.rs", "crates/sqlbuild-orchestration/src/platform/main/read_location_work.rs", "crates/sqlbuild-orchestration/src/platform/main/read_person_location_health.rs", "crates/sqlbuild-orchestration/src/platform/main/location_health.rs", "crates/sqlbuild-orchestration/src/snapshots/main/current_snapshot_record.rs", "crates/sqlbuild-orchestration/src/runs/main/list_person_runs.rs", "crates/sqlbuild-orchestration/src/runs/main/list_person_run_nodes.rs", "crates/sqlbuild-orchestration/src/runs/main/list_run_nodes.rs", "crates/sqlbuild-orchestration/src/runs/main/list_runs.rs", "crates/sqlbuild-orchestration/src/runs/_helpers/lifecycle/reads.rs", "crates/sqlbuild-orchestration/src/warehouse_history/_helpers/checks.rs", "crates/sqlbuild-orchestration/src/scheduler/main/list_person_ticks.rs", "crates/sqlbuild-orchestration/src/store/main/read_records.rs", "crates/sqlbuild-orchestration/src/platform/main/location_work.rs", "crates/sqlbuild-orchestration/src/scheduler/main/list_ticks.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/attempts.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/facts.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_current_record.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_documents.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_nodes.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_checks.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_microbatch_coverage.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_node_runs.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_run_events.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_run_statements.rs", "crates/sqlbuild-orchestration/src/warehouse_history/_helpers/node_runs.rs"] +paths = ["crates/sqlbuild-orchestration/src/authorization/main/list_role_definitions.rs", "crates/sqlbuild-orchestration/src/authorization/main/read_security_audit.rs", "crates/sqlbuild-orchestration/src/credentials/_helpers/active_session.rs", "crates/sqlbuild-orchestration/src/declarations/_helpers/reads.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_person_jobs.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_person_trigger_changes.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_person_triggers.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_triggers.rs", "crates/sqlbuild-orchestration/src/platform/main/read_location_work.rs", "crates/sqlbuild-orchestration/src/platform/main/read_person_location_health.rs", "crates/sqlbuild-orchestration/src/platform/main/location_health.rs", "crates/sqlbuild-orchestration/src/snapshots/main/current_snapshot_record.rs", "crates/sqlbuild-orchestration/src/runs/main/list_person_runs.rs", "crates/sqlbuild-orchestration/src/runs/main/list_person_run_nodes.rs", "crates/sqlbuild-orchestration/src/runs/main/list_run_nodes.rs", "crates/sqlbuild-orchestration/src/runs/main/list_runs.rs", "crates/sqlbuild-orchestration/src/runs/_helpers/lifecycle/reads.rs", "crates/sqlbuild-orchestration/src/warehouse_history/_helpers/checks.rs", "crates/sqlbuild-orchestration/src/scheduler/main/list_person_ticks.rs", "crates/sqlbuild-orchestration/src/store/main/read_records.rs", "crates/sqlbuild-orchestration/src/platform/main/location_work.rs", "crates/sqlbuild-orchestration/src/scheduler/main/list_ticks.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/attempts.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/facts.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_current_record.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_documents.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_nodes.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/read_person_catalog.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_checks.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_microbatch_coverage.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/read_person_statistics.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_node_runs.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_run_events.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_run_statements.rs", "crates/sqlbuild-orchestration/src/warehouse_history/_helpers/node_runs.rs"] reason = "Each is one control-store read with its own SQL and record type. Transactions, person authorization and decoding go through the shared begin_authorized, row_in_target, query_records, read_records and query_bodies helpers; what remains alike is the call shape." [[dupes.allowlist]] @@ -1823,7 +1823,7 @@ paths = ["crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/fingerpr reason = "Typed observation field projections intentionally mirror provenance checks while preserving different required identity fields and timestamp names. Bounds, rejection rules and reconciliation completion share the ingestion owner." [[dupes.allowlist]] -paths = ["crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_audit_results.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_janitor_events.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_microbatches.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_migrations.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest_old_name_views.rs"] +paths = ["crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_audit_results.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_column_migrations.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_consumption.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_janitor_events.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_microbatches.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_migrations.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_node_results.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_old_name_views.rs"] reason = "Typed event projections retain distinct required identities and indexed payload columns. Event persistence, fencing, atomic retries and reconciliation share the events ingestion owner; no per-table lifecycle is copied." [[targets.python.rule_exceptions]] diff --git a/src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py b/src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py index afe6a25c9d..729942f5d8 100644 --- a/src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py +++ b/src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py @@ -4201,6 +4201,25 @@ def schema_exists( database: str | None, schema: str, ) -> bool: + """Check with SHOW SCHEMAS, which needs no running warehouse, where the name allows.""" + + show_database: str | None = ( + database if database is not None else self._current_database(connection=connection) + ) + if show_database is not None and SHOW_LIKE_ESCAPE_CHARACTER not in schema: + try: + rows: list[dict[str, object]] = self._fetch_show_rows( + connection=connection, + query=( + f"SHOW SCHEMAS LIKE '{show_like_pattern(schema.upper())}' " + f"IN DATABASE {self.render_identifier(show_database)}" + ), + ) + except Exception as error: + if not is_missing_object_error(error): + raise + return False + return any(str(row.get("name")).upper() == schema.upper() for row in rows) query: str = ( "SELECT 1 FROM information_schema.schemata WHERE UPPER(schema_name) = UPPER(%s)" ) diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py index dff015df40..aa7f70188d 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py @@ -2,6 +2,7 @@ from __future__ import annotations +import base64 import hashlib import json import os @@ -28,6 +29,8 @@ import pytest from sqlbuild._native import ( + ColumnMigrationDecision, + ColumnMigrationEvent, Fingerprint, MigrationDecision, MigrationDiscovery, @@ -35,6 +38,9 @@ MigrationRelation, OldNameViewEvent, OldNameViewEventType, + StateSqlValueType, + build_consumption_create_table_sql, + render_state_sql_literal, write_fingerprint, write_migration_state_event, ) @@ -47,6 +53,8 @@ ) from sqlbuild.executor.janitor_events.models import JanitorEventRecord from sqlbuild.executor.janitor_events.types import JanitorEventType +from sqlbuild.executor.node_results.classes.direct_store import DirectNodeResultStore +from sqlbuild.executor.node_results.models import NodeResultRecord from sqlbuild.microbatches.classes.direct_store import DirectMicrobatchEventStore from sqlbuild.microbatches.classes.event_codec import MicrobatchEventCodec from sqlbuild.microbatches.constants import MICROBATCH_COLUMNS @@ -233,6 +241,92 @@ def append_migration_history(adapter: Any, connection: Any, fixture: dict[str, A ) +def append_column_migration_history(adapter: Any, connection: Any, fixture: dict[str, Any]) -> None: + """Real column-rename facts, including idempotent duplicate attempts.""" + data: dict[str, Any] = migration_history_data(fixture, ("relation",)) + data["discovery"] = MigrationDiscovery(data["discovery"]) + data["decision"] = ColumnMigrationDecision(data["decision"]) + record: ColumnMigrationEvent = ColumnMigrationEvent(**data) + write_migration_history( + adapter, + connection, + fixture, + record, + "column", + adapter.render_create_column_migration_state_table_sql(database=None, schema="main"), + {}, + ) + + +def append_node_result_history(adapter: Any, connection: Any, fixture: dict[str, Any]) -> None: + """Real node-result writer; retried writes repeat rows, which carry no event identity.""" + store: DirectNodeResultStore = DirectNodeResultStore( + adapter=adapter, connection=connection, database=None, schema="main" + ) + rows: list[dict[str, Any]] = identified_rows(fixture) + for row in rows + [rows[0]] * fixture["duplicate_rows"]: + store.write( + NodeResultRecord( + node_type=row["node_type"], + node_name=row["node_name"], + target_database=row["target_database"], + target_schema=row["target_schema"], + target_name=row["target_name"], + run_id=row["run_id"], + status=row["status"], + payload=json.loads(base64.b64decode(row["payload_json_b64"])), + metadata=json.loads(base64.b64decode(row["metadata_json_b64"])), + error_message=row["error_message"], + materialized=row["materialized"] == "true", + ts=datetime.fromisoformat(row["ts"]), + ) + ) + + +def append_consumption_history(adapter: Any, connection: Any, fixture: dict[str, Any]) -> None: + """Consumption rows in the real table layout; retried appends repeat identical rows.""" + _ = adapter.execute( + connection=connection, + sql=build_consumption_create_table_sql( + database=None, + schema="main", + render_qualified_name=adapter.render_qualified_name, + render_framework_type=adapter.render_framework_type, + ), + ) + columns: list[str] = list(fixture["record"])[2:] + types: dict[str, StateSqlValueType] = dict.fromkeys(columns, StateSqlValueType.STRING) | { + "input_count": StateSqlValueType.INTEGER, + "recorded_at": StateSqlValueType.TIMESTAMP, + } + rows: list[dict[str, Any]] = [ + row | {"recorded_at": datetime.fromisoformat(row["recorded_at"])} + for row in identified_rows(fixture) + ] + for row in rows + [rows[0]] * fixture["duplicate_rows"]: + values: list[str] = [ + render_state_sql_literal( + value=row[column], declared_type=types[column], timestamp_type="TIMESTAMP" + ) + for column in columns + ] + _ = adapter.execute( + connection=connection, + sql=f"INSERT INTO main._sqlbuild_consumption ({', '.join(columns)}) " + f"VALUES ({', '.join(values)})", + ) + + +def identified_rows(fixture: dict[str, Any]) -> list[dict[str, Any]]: + """The fixture record once per source identity, with its per-identity overrides.""" + return [ + fixture["record"] + | {fixture["source_key"]: f"{fixture['id_prefix']}-{index:03}"} + | fixture["record_overrides"].get(f"{fixture['id_prefix']}-{index:03}", {}) + for index in range(fixture["source_rows"]) + ] + + def append_old_name_history(adapter: Any, connection: Any, fixture: dict[str, Any]) -> None: """Real compatibility-view facts, including nullable and populated expiry.""" data: dict[str, Any] = migration_history_data(fixture, ("old", "new")) @@ -254,7 +348,7 @@ def append_old_name_history(adapter: Any, connection: Any, fixture: dict[str, An ) -def migration_history_data(fixture: dict[str, Any], relations: tuple[str, str]) -> dict[str, Any]: +def migration_history_data(fixture: dict[str, Any], relations: tuple[str, ...]) -> dict[str, Any]: """Reconstruct native relation values from the warehouse-shaped fixture.""" data: dict[str, Any] = fixture["record"].copy() del data["state_database"] @@ -273,8 +367,8 @@ def write_migration_history( adapter: Any, connection: Any, fixture: dict[str, Any], - record: MigrationEvent | OldNameViewEvent, - kind: Literal["model", "old_name"], + record: MigrationEvent | ColumnMigrationEvent | OldNameViewEvent, + kind: Literal["model", "column", "old_name"], create_sql: str, last_values: dict[str, Any], ) -> None: diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_warehouse_metadata.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_warehouse_metadata.py index 4d466244e5..83f43139f4 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_warehouse_metadata.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_warehouse_metadata.py @@ -45,11 +45,14 @@ ) from tests.integration.src.sqlbuild.compiler.compiled_project.publication.helpers import ( append_audit_history, + append_column_migration_history, + append_consumption_history, append_contended_fingerprints, append_janitor_history, append_legacy_history, append_microbatch_history, append_migration_history, + append_node_result_history, append_old_name_history, append_retried_fingerprints, location_process, @@ -122,6 +125,38 @@ ), ("sqlite", "path", "migration_history", append_migration_history, "external-move", ()), ("sqlite", "path", "old_name_history", append_old_name_history, "external-move", ()), + ( + "duckdb", + "database", + "column_migration_history", + append_column_migration_history, + "external-rename", + (), + ), + ( + "sqlite", + "path", + "column_migration_history", + append_column_migration_history, + "external-rename", + (), + ), + ( + "duckdb", + "database", + "node_result_history", + append_node_result_history, + "external-run", + (), + ), + ( + "duckdb", + "database", + "consumption_history", + append_consumption_history, + "external-run", + (), + ), ( "duckdb", "database", @@ -259,6 +294,38 @@ def test_given_external_state_events_when_reading_pages_then_identities_text_and ), ("sqlite", "path", "migration_history", append_migration_history, "external-move", ()), ("sqlite", "path", "old_name_history", append_old_name_history, "external-move", ()), + ( + "duckdb", + "database", + "column_migration_history", + append_column_migration_history, + "external-rename", + (), + ), + ( + "sqlite", + "path", + "column_migration_history", + append_column_migration_history, + "external-rename", + (), + ), + ( + "duckdb", + "database", + "node_result_history", + append_node_result_history, + "external-run", + (), + ), + ( + "duckdb", + "database", + "consumption_history", + append_consumption_history, + "external-run", + (), + ), ] ], ids=lambda case: case.description, diff --git a/tests/unit/src/sqlbuild/adapters/snowflake/_test_types.py b/tests/unit/src/sqlbuild/adapters/snowflake/_test_types.py index 30ed59be03..e06ff737f7 100644 --- a/tests/unit/src/sqlbuild/adapters/snowflake/_test_types.py +++ b/tests/unit/src/sqlbuild/adapters/snowflake/_test_types.py @@ -417,3 +417,12 @@ class SnowflakePlatformDetectionTestCase: description: str environment: dict[str, str] expected_value: str + + +@dataclass(frozen=True) +class SnowflakeSchemaExistsTestCase: + description: str + schema: str + cursor: object + expected_exists: bool + expected_sql_fragment: str diff --git a/tests/unit/src/sqlbuild/adapters/snowflake/helpers.py b/tests/unit/src/sqlbuild/adapters/snowflake/helpers.py index 4882cc8bc5..0c6d24b4f4 100644 --- a/tests/unit/src/sqlbuild/adapters/snowflake/helpers.py +++ b/tests/unit/src/sqlbuild/adapters/snowflake/helpers.py @@ -78,6 +78,14 @@ def fetchmany(self, size: int) -> list[tuple[object, ...]]: return self._rows[:size] +class FakeSnowflakeMissingObjectCursor(FakeSnowflakeQueryCursor): + """Cursor double whose statement fails as Snowflake reports a missing object.""" + + def execute(self, sql: str) -> None: + super().execute(sql) + raise RuntimeError("Database 'GONE' does not exist or not authorized.") + + class FakeSnowflakeDescribeConnection: """Connection double returning a preconfigured describe cursor.""" diff --git a/tests/unit/src/sqlbuild/adapters/snowflake/test_client.py b/tests/unit/src/sqlbuild/adapters/snowflake/test_client.py index 6a9dcf2cdd..5c64c55bcf 100644 --- a/tests/unit/src/sqlbuild/adapters/snowflake/test_client.py +++ b/tests/unit/src/sqlbuild/adapters/snowflake/test_client.py @@ -47,6 +47,7 @@ SnowflakeRenderPythonFunctionTestCase, SnowflakeRenderTableFunctionTestCase, SnowflakeSchemaDiffTestCase, + SnowflakeSchemaExistsTestCase, SnowflakeTableFreshnessBatchTestCase, SnowflakeTableFreshnessMetadataErrorTestCase, SnowflakeTableFreshnessMetadataTestCase, @@ -59,6 +60,7 @@ FakeSnowflakeMetadataConnection, FakeSnowflakeMetadataCursor, FakeSnowflakeMetadataSequenceConnection, + FakeSnowflakeMissingObjectCursor, FakeSnowflakeQueryCursor, FakeSnowflakeRawConnection, build_bulk_columns_rows, @@ -1357,3 +1359,52 @@ def test_given_snowflake_relation_timestamps_when_listing_then_exposes_utc_age_m str(test_case.expected_created_at), str(test_case.expected_last_altered_at), ) + + +@pytest.mark.parametrize( + "test_case", + [ + SnowflakeSchemaExistsTestCase( + description="SHOW SCHEMAS finds a schema stored in another case", + schema="staging", + cursor=FakeSnowflakeQueryCursor((("name",),), [("Staging",)]), + expected_exists=True, + expected_sql_fragment="SHOW SCHEMAS LIKE 'STAGING' IN DATABASE", + ), + SnowflakeSchemaExistsTestCase( + description="a neighbour matched by the LIKE wildcard is not the schema", + schema="stg_a", + cursor=FakeSnowflakeQueryCursor((("name",),), [("STGXA",)]), + expected_exists=False, + expected_sql_fragment="SHOW SCHEMAS LIKE 'STG_A' IN DATABASE", + ), + SnowflakeSchemaExistsTestCase( + description="a missing database holds no schema", + schema="staging", + cursor=FakeSnowflakeMissingObjectCursor((("name",),), []), + expected_exists=False, + expected_sql_fragment="SHOW SCHEMAS LIKE 'STAGING' IN DATABASE", + ), + SnowflakeSchemaExistsTestCase( + description="a name with a backslash reads the information schema", + schema="stag\\ing", + cursor=FakeSnowflakeMetadataCursor(row=(1,)), + expected_exists=True, + expected_sql_fragment="information_schema.schemata", + ), + ], + ids=lambda case: case.description, +) +def test_given_schema_when_checking_existence_then_show_needs_no_running_warehouse( + test_case: SnowflakeSchemaExistsTestCase, +) -> None: + connection: FakeSnowflakeDescribeConnection = FakeSnowflakeDescribeConnection( + cast(Any, test_case.cursor) + ) + + exists: bool = SnowflakeAdapter().schema_exists( + connection=cast(Any, connection), database="ANALYTICS", schema=test_case.schema + ) + + assert exists is test_case.expected_exists + assert test_case.expected_sql_fragment in str(cast(Any, test_case.cursor).executed_sql)