From aabf00fb93935a35384dff8d4dbb429f2d2ea77c Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Wed, 11 Nov 2026 11:58:22 +0000 Subject: [PATCH 1/7] fix(location): keep a smaller statistics page after a failed statistics read A statistics read that timed out halved the shared page limit and rewound to catalog. Finishing catalog reset that limit to the maximum, so statistics retried at full size, timed out again and looped, keeping the warehouse busy. Statistics now has its own page limit: it survives the rewind and returns to the maximum once a statistics scan completes. --- .../src/native_commands/_helpers/execution.rs | 15 ++++- .../src/native_commands/models.rs | 6 +- .../sqlbuild-cli/src/native_commands/tests.rs | 2 + .../src/native_commands/tests/helpers.rs | 25 +++++++++ .../tests/test_location_metadata.rs | 55 +++++++++++++++++++ .../src/native_commands/tests/test_types.rs | 10 ++++ 6 files changed, 109 insertions(+), 4 deletions(-) create mode 100644 crates/sqlbuild-cli/src/native_commands/tests/test_location_metadata.rs 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, +} From 84691240e549ed83781404148820cf8ec522542f Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Wed, 11 Nov 2026 14:00:19 +0000 Subject: [PATCH 2/7] fix(snowflake): sync the catalog and check schemas without a running warehouse A code location's catalog sync listed relations and columns through information_schema, and every Snowflake connection checked its session schema through information_schema.schemata. Both need a running warehouse, so each sync cycle resumed it and kept it from suspending. The catalog page now uses the schema-scoped SHOW listings the build already uses when the adapter provides them, and schema_exists asks SHOW SCHEMAS, falling back to the information schema for names SHOW LIKE cannot express. --- .../_helpers/native_host/warehouse_catalog.rs | 106 ++++++++++++------ .../snowflake/classes/snowflake_adapter.py | 19 ++++ .../adapters/snowflake/_test_types.py | 9 ++ .../sqlbuild/adapters/snowflake/helpers.py | 8 ++ .../adapters/snowflake/test_client.py | 51 +++++++++ 5 files changed, 158 insertions(+), 35 deletions(-) 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/src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py b/src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py index ddf90d3385..1c19faa5bf 100644 --- a/src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py +++ b/src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py @@ -4165,6 +4165,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/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) From ccba10450162af6e4b6265f5f92662b30d6c6bff Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Wed, 11 Nov 2026 14:43:11 +0000 Subject: [PATCH 3/7] feat(location): mirror column migrations, node results and consumption state The location sync now reads every state table SQLBuild writes. Column migrations, node results and consumption join the other kinds, ahead of catalog and statistics. Node result and consumption rows have no event ID, so the server derives one from their identity columns, and retried identical rows collapse to one. Each kind names its source state table, and a test fails when a state table has no kind. The janitor's built-in exclusions now use the same list. The warehouse_metadata main modules are grouped into ingest, location-read and person-read buckets. --- .../column_migration_history.json | 46 +++++++ .../orchestration/consumption_history.json | 63 +++++++++ .../orchestration/node_result_history.json | 52 ++++++++ clients/typescript/src/schema.d.ts | 2 +- .../src/janitor_planning/constants.rs | 20 +-- crates/sqlbuild-core/src/constants.rs | 21 +++ .../_helpers/resources/person_operations.rs | 10 +- .../api/_helpers/resources/person_reads.rs | 2 +- .../_helpers/resources/warehouse_metadata.rs | 30 +++-- .../src/platform/_helpers/retention.rs | 2 +- .../src/store/schema.sql | 36 +++++ .../src/warehouse_metadata/_helpers/events.rs | 9 +- .../main/{ => ingest}/ingest_audit_results.rs | 0 .../main/{ => ingest}/ingest_catalog.rs | 0 .../main/ingest/ingest_column_migrations.rs | 46 +++++++ .../main/ingest/ingest_consumption.rs | 57 ++++++++ .../main/{ => ingest}/ingest_fingerprints.rs | 0 .../{ => ingest}/ingest_janitor_events.rs | 0 .../main/{ => ingest}/ingest_microbatches.rs | 0 .../main/{ => ingest}/ingest_migrations.rs | 0 .../main/ingest/ingest_node_results.rs | 59 ++++++++ .../{ => ingest}/ingest_old_name_views.rs | 0 .../{ => ingest}/ingest_source_freshness.rs | 0 .../main/{ => ingest}/ingest_statistics.rs | 0 .../src/warehouse_metadata/main/ingest/mod.rs | 13 ++ .../main/{ => ingest}/prune_statistics.rs | 0 .../main/location_reads/mod.rs | 2 + .../read_statistics_due.rs | 0 .../{ => location_reads}/read_watermark.rs | 0 .../src/warehouse_metadata/main/mod.rs | 21 +-- .../list_person_catalog_nodes.rs | 0 .../main/person_reads/mod.rs | 6 + .../{ => person_reads}/read_person_catalog.rs | 0 .../read_person_column_lineage.rs | 0 .../read_person_fingerprint.rs | 0 .../read_person_node_status.rs | 0 .../read_person_statistics.rs | 0 .../src/warehouse_metadata/models.rs | 126 +++++++++++++++++- .../tests/api/fingerprint_reads/helpers.rs | 3 +- .../tests/api/node_status/helpers.rs | 8 +- .../api/statistics/eligibility/helpers.rs | 5 +- .../tests/api/statistics/helpers.rs | 3 +- .../tests/api/statistics/retention/helpers.rs | 5 +- .../tests/api/test_catalog_reads.rs | 8 +- .../tests/api/test_checks.rs | 2 +- .../tests/api/test_column_lineage.rs | 2 +- .../tests/api/test_maintenance.rs | 2 +- .../tests/api/test_microbatch_coverage.rs | 2 +- .../tests/api/test_types.rs | 7 + .../tests/api/test_warehouse_metadata.rs | 50 ++++++- .../api/warehouse_metadata/events/helpers.rs | 3 +- .../tests/credentials/roles/helpers.rs | 4 +- .../tests/sensors/test_sources.rs | 2 +- .../native_host/warehouse_metadata.rs | 77 ++++++++++- crates/sqlbuild-state/src/events/constants.rs | 15 +++ .../compiled_project/publication/helpers.py | 98 +++++++++++++- .../publication/test_warehouse_metadata.py | 67 ++++++++++ 57 files changed, 898 insertions(+), 88 deletions(-) create mode 100644 benchmarks/duckshop/edge_cases/orchestration/column_migration_history.json create mode 100644 benchmarks/duckshop/edge_cases/orchestration/consumption_history.json create mode 100644 benchmarks/duckshop/edge_cases/orchestration/node_result_history.json rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => ingest}/ingest_audit_results.rs (100%) rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => ingest}/ingest_catalog.rs (100%) create mode 100644 crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_column_migrations.rs create mode 100644 crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_consumption.rs rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => ingest}/ingest_fingerprints.rs (100%) rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => ingest}/ingest_janitor_events.rs (100%) rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => ingest}/ingest_microbatches.rs (100%) rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => ingest}/ingest_migrations.rs (100%) create mode 100644 crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/ingest_node_results.rs rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => ingest}/ingest_old_name_views.rs (100%) rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => ingest}/ingest_source_freshness.rs (100%) rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => ingest}/ingest_statistics.rs (100%) create mode 100644 crates/sqlbuild-orchestration/src/warehouse_metadata/main/ingest/mod.rs rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => ingest}/prune_statistics.rs (100%) create mode 100644 crates/sqlbuild-orchestration/src/warehouse_metadata/main/location_reads/mod.rs rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => location_reads}/read_statistics_due.rs (100%) rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => location_reads}/read_watermark.rs (100%) rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => person_reads}/list_person_catalog_nodes.rs (100%) create mode 100644 crates/sqlbuild-orchestration/src/warehouse_metadata/main/person_reads/mod.rs rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => person_reads}/read_person_catalog.rs (100%) rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => person_reads}/read_person_column_lineage.rs (100%) rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => person_reads}/read_person_fingerprint.rs (100%) rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => person_reads}/read_person_node_status.rs (100%) rename crates/sqlbuild-orchestration/src/warehouse_metadata/main/{ => person_reads}/read_person_statistics.rs (100%) 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-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..74988e49c9 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,6 +298,9 @@ impl WarehouseHistoryKind { Self::Migrations, Self::OldNameViews, Self::Microbatches, + Self::ColumnMigrations, + Self::NodeResults, + Self::Consumption, Self::Catalog, Self::Statistics, ]; @@ -240,12 +315,32 @@ impl WarehouseHistoryKind { Self::Migrations => "migrations", Self::OldNameViews => "old_name_views", Self::Microbatches => "microbatches", + Self::ColumnMigrations => "column_migrations", + Self::NodeResults => "node_results", + Self::Consumption => "consumption", Self::Catalog => "catalog", Self::Statistics => "statistics", } } #[must_use] + /// The state table a kind mirrors; catalog and statistics read the warehouse catalog instead. + pub const fn source_table(self) -> Option<&'static str> { + match self { + Self::Fingerprints => Some(FINGERPRINT_TABLE_NAME), + Self::SourceFreshness => Some(SOURCE_FRESHNESS_TABLE_NAME), + Self::AuditResults => Some(AUDIT_RESULTS_TABLE_NAME), + Self::JanitorEvents => Some(JANITOR_EVENTS_TABLE_NAME), + Self::Migrations => Some(MIGRATION_TABLE_NAME), + Self::OldNameViews => Some(OLD_NAME_VIEW_TABLE_NAME), + Self::Microbatches => Some(MICROBATCH_TABLE_NAME), + Self::ColumnMigrations => Some(COLUMN_MIGRATION_TABLE_NAME), + Self::NodeResults => Some(NODE_RESULTS_TABLE_NAME), + Self::Consumption => Some(CONSUMPTION_TABLE_NAME), + Self::Catalog | Self::Statistics => None, + } + } + pub const fn mirror_table(self) -> &'static str { match self { Self::Fingerprints => "sqb_warehouse_fingerprints", @@ -255,6 +350,9 @@ impl WarehouseHistoryKind { Self::Migrations => "sqb_warehouse_migrations", Self::OldNameViews => "sqb_warehouse_old_name_views", Self::Microbatches => "sqb_warehouse_microbatches", + Self::ColumnMigrations => "sqb_warehouse_column_migrations", + Self::NodeResults => "sqb_warehouse_node_results", + Self::Consumption => "sqb_warehouse_consumption", Self::Catalog => "sqb_warehouse_catalog", Self::Statistics => "sqb_warehouse_statistics", } @@ -346,6 +444,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 +485,9 @@ pub enum WarehouseHistoryRead { Migrations(StateEventsRead), OldNameViews(StateEventsRead), Microbatches(StateEventsRead), + ColumnMigrations(StateEventsRead), + NodeResults(NodeResultsRead), + Consumption(ConsumptionRead), Catalog(CatalogRead), Statistics(StatisticsRead), } @@ -381,6 +502,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_metadata.rs b/crates/sqlbuild-python/src/bindings/_helpers/native_host/warehouse_metadata.rs index a5c265e81e..005fff9c83 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,19 @@ 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, }; -use sqlbuild_state::events::constants::{MIGRATION_COLUMNS, OLD_NAME_VIEW_COLUMNS}; +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::freshness::constants::{ OBSERVED_AT_COLUMN, SOURCE_FRESHNESS_COLUMN_TYPES, SQLITE_ADAPTER, }; @@ -50,7 +56,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 +119,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 +206,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 +305,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 +445,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,7 +486,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(read) => { + return decode_event_record(&read.location, row, (columns, &[timestamp])); + } + WarehouseHistoryRead::Consumption(read) => { return decode_event_record(&read.location, row, (columns, &[timestamp])); } WarehouseHistoryRead::JanitorEvents(read) => { 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/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py index dff015df40..7d9fb1c271 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")) @@ -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, From af4a785b51df58f400ca50b3c1db6edd76371cfe Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Wed, 11 Nov 2026 14:51:50 +0000 Subject: [PATCH 4/7] fix(state): page state tables without event IDs on Snowflake and SQL Server A table without an event_id column pages with a NULL placeholder key. The history SQL wrapped that key in the adapter's binary conversion, and Snowflake rejects TO_BINARY(NULL, 'UTF-8'). The placeholder now stays a bare NULL and is left out of ORDER BY, where SQL Server rejects constants. --- .../src/state_tables/constants.rs | 2 + .../src/state_tables/main/read_history_sql.rs | 11 +++- .../src/state_tables/tests/test_sql.rs | 64 ++++++++++++++++++- .../src/state_tables/tests/test_types.rs | 8 +++ 4 files changed, 81 insertions(+), 4 deletions(-) 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, +} From 5ca9d1f6a881f460b8f2df43c0f1b752e0e1b43e Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Wed, 11 Nov 2026 15:04:05 +0000 Subject: [PATCH 5/7] chore(location): point duplicate-code exemptions at the moved ingest and person-read modules --- .../src/warehouse_metadata/models.rs | 3 ++- .../_helpers/native_host/warehouse_metadata.rs | 11 +++++------ fensu.toml | 4 ++-- 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs index 74988e49c9..c79e16f910 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs @@ -323,8 +323,8 @@ impl WarehouseHistoryKind { } } - #[must_use] /// 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> { match self { Self::Fingerprints => Some(FINGERPRINT_TABLE_NAME), @@ -341,6 +341,7 @@ impl WarehouseHistoryKind { } } + #[must_use] pub const fn mirror_table(self) -> &'static str { match self { Self::Fingerprints => "sqb_warehouse_fingerprints", 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 005fff9c83..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 @@ -15,7 +15,8 @@ use sqlbuild_core::constants::{ 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::{ @@ -491,11 +492,9 @@ fn decode_record( | WarehouseHistoryRead::ColumnMigrations(read) => { return decode_event_record(&read.location, row, (columns, &[timestamp])); } - WarehouseHistoryRead::NodeResults(read) => { - return decode_event_record(&read.location, row, (columns, &[timestamp])); - } - WarehouseHistoryRead::Consumption(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( 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]] From c9efef0527dcd12f73dd679275167f1ab7731fb8 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Wed, 11 Nov 2026 15:16:01 +0000 Subject: [PATCH 6/7] refactor(location): keep each sync kind's names in one table --- .../src/warehouse_metadata/models.rs | 102 +++++++++++------- 1 file changed, 61 insertions(+), 41 deletions(-) diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs index c79e16f910..d678e6d29a 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs @@ -305,58 +305,78 @@ impl WarehouseHistoryKind { 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::ColumnMigrations => "column_migrations", - Self::NodeResults => "node_results", - Self::Consumption => "consumption", - 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> { - match self { - Self::Fingerprints => Some(FINGERPRINT_TABLE_NAME), - Self::SourceFreshness => Some(SOURCE_FRESHNESS_TABLE_NAME), - Self::AuditResults => Some(AUDIT_RESULTS_TABLE_NAME), - Self::JanitorEvents => Some(JANITOR_EVENTS_TABLE_NAME), - Self::Migrations => Some(MIGRATION_TABLE_NAME), - Self::OldNameViews => Some(OLD_NAME_VIEW_TABLE_NAME), - Self::Microbatches => Some(MICROBATCH_TABLE_NAME), - Self::ColumnMigrations => Some(COLUMN_MIGRATION_TABLE_NAME), - Self::NodeResults => Some(NODE_RESULTS_TABLE_NAME), - Self::Consumption => Some(CONSUMPTION_TABLE_NAME), - Self::Catalog | Self::Statistics => None, - } + 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::ColumnMigrations => "sqb_warehouse_column_migrations", - Self::NodeResults => "sqb_warehouse_node_results", - Self::Consumption => "sqb_warehouse_consumption", - Self::Catalog => "sqb_warehouse_catalog", - Self::Statistics => "sqb_warehouse_statistics", - } + self.names().1 } } From 2de4fe3e1acac5e155422c43abfd0e3dc0ac77fb Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Wed, 11 Nov 2026 15:19:20 +0000 Subject: [PATCH 7/7] test(location): accept one relation prefix in migration history fixtures --- .../sqlbuild/compiler/compiled_project/publication/helpers.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 7d9fb1c271..aa7f70188d 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py @@ -348,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"]