diff --git a/crates/sqlbuild-cli/src/native_commands/_helpers/execution.rs b/crates/sqlbuild-cli/src/native_commands/_helpers/execution.rs index 035882d3e..77ff315cd 100644 --- a/crates/sqlbuild-cli/src/native_commands/_helpers/execution.rs +++ b/crates/sqlbuild-cli/src/native_commands/_helpers/execution.rs @@ -1232,7 +1232,13 @@ pub(crate) async fn read_metadata_page( Some(reader) => reader, None => match open_location_runtime(options, script, Vec::new()) { Ok(reader) => reader, - Err(_) => return Ok(Some((state, None))), + Err(error) => { + drop(write_output(&[( + Stream::Stderr, + &format!("Warehouse metadata reader did not start: {error}\n"), + )])); + return Ok(Some((state, None))); + } }, }; tokio::pin!(stopped); @@ -1254,11 +1260,23 @@ pub(crate) async fn read_metadata_page( state.reader = Some(reader); Ok(Some((state, None))) } - Ok((reader, Err(_))) => { + Ok((reader, Err(error))) => { + drop(write_output(&[( + Stream::Stderr, + &format!("Warehouse metadata reader failed: {error}\n"), + )])); stop_location_runtime(Some(reader)).await?; Ok(Some((state, None))) } - Err(_) => Ok(Some((state, None))), + Err(_) => { + drop(write_output(&[( + Stream::Stderr, + &format!( + "Warehouse metadata read did not answer within {METADATA_POLL_SECONDS}s.\n" + ), + )])); + Ok(Some((state, None))) + } }, } } diff --git a/crates/sqlbuild-python/src/bindings/_helpers/location_command/runtime.rs b/crates/sqlbuild-python/src/bindings/_helpers/location_command/runtime.rs index 6c4af1c5e..2a9548897 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/location_command/runtime.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/location_command/runtime.rs @@ -2,9 +2,11 @@ use std::io::{self, Write}; +use pyo3::PyErr; use pyo3::exceptions::PyValueError; use pyo3::prelude::{Bound, PyAnyMethods, PyResult, Python}; use pyo3::types::{PyDict, PyDictMethods}; +use sqlbuild_core::redaction::main::redact_error_text::redact_error_text; use sqlbuild_orchestration::sensors::models::{CustomSensorOutcome, SensorEvaluationRequest}; use sqlbuild_orchestration::warehouse_metadata::models::{ WarehouseHistoryKind, WarehouseHistoryRead, @@ -55,9 +57,9 @@ pub(super) fn run_runtime(py: Python<'_>, options: Bound<'_, PyDict>) -> PyResul } }; serde_json::to_string(&record) - } else if WarehouseHistoryKind::ALL + } else if let Some(kind) = WarehouseHistoryKind::ALL .iter() - .any(|kind| request.get(kind.as_str()).is_some()) + .find(|kind| request.get(kind.as_str()).is_some()) { let metadata: WarehouseHistoryRead = serde_json::from_value(request).map_err(|_| { @@ -73,7 +75,10 @@ pub(super) fn run_runtime(py: Python<'_>, options: Bound<'_, PyDict>) -> PyResul if !busy { write_stderr( py, - "Warehouse connection for metadata reads failed.\n", + &format!( + "Warehouse connection for metadata reads failed: {}\n", + failure_reason(&error) + ), )?; } else if !reported_busy { write_stderr( @@ -86,16 +91,13 @@ pub(super) fn run_runtime(py: Python<'_>, options: Bound<'_, PyDict>) -> PyResul }; } let page = match &session { - Some(opened) => { - match read_history(py, (opened, stream.take()), &metadata) { - Ok((page, open)) => { - stream = open; - Some(page) - } - Err(_) => None, - } - } - None => None, + Some(opened) => read_history(py, (opened, stream.take()), &metadata) + .map(|(page, open)| { + stream = open; + page + }) + .map_err(|error| format!(": {}", failure_reason(&error))), + None => Err(String::new()), }; if session.as_ref().is_some_and(|opened| opened.local) && let Some(opened) = session.take() @@ -106,16 +108,23 @@ pub(super) fn run_runtime(py: Python<'_>, options: Bound<'_, PyDict>) -> PyResul if busy { serde_json::to_string(&serde_json::json!({ "busy": true })) } else { - if page.is_none() { - if let Some(opened) = session.take() { - drop(close_history_session(opened, None)); + let delivered = match &page { + Ok(page) => Some(page), + Err(reason) => { + if let Some(opened) = session.take() { + drop(close_history_session(opened, None)); + } + write_stderr( + py, + &format!( + "Warehouse metadata read ({}) failed{reason}. No page delivered.\n", + kind.as_str() + ), + )?; + None } - write_stderr( - py, - "Warehouse metadata read failed. No page delivered.\n", - )?; - } - serde_json::to_string(&page) + }; + serde_json::to_string(&delivered) } } else { let evaluation: SensorEvaluationRequest = serde_json::from_value(request) @@ -162,3 +171,11 @@ pub(super) fn run_runtime(py: Python<'_>, options: Bound<'_, PyDict>) -> PyResul result }) } + +/// The driver's class and message on one line, redacted as warehouse error text. +fn failure_reason(error: &PyErr) -> String { + redact_error_text(&error.to_string()) + .split_whitespace() + .collect::>() + .join(" ") +} 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 9f9184979..4345a0055 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 @@ -29,6 +29,7 @@ use sqlbuild_state::freshness::constants::{ }; use sqlbuild_state::observations::constants::OBSERVATION_COLUMN_TYPES; use sqlbuild_state::resource_costs::constants::RESOURCE_COST_COLUMN_TYPES; +use sqlbuild_state::state_tables::main::event_id_index::event_id_index; use sqlbuild_state::state_tables::main::read_history_sql::read_history_sql; use sqlbuild_state::state_tables::models::HistoryRead; @@ -445,7 +446,7 @@ fn read_page<'py>( .then_some(index) }) .collect(); - cursor_indexes.push(columns.len()); + cursor_indexes.push(event_id_index(columns)); let mut literals = vec![None; key_columns.unwrap_or(columns).len() + 1]; if let Some(after) = after { let classes = SqlValueClasses::load(py)?; @@ -771,7 +772,7 @@ fn decode_record( let mut value = json!({ "state_database": location.database, "state_schema": location.schema, - "event_id": row.get_item(columns.len())?.extract::>()?, + "event_id": row.get_item(event_id_index(columns))?.extract::>()?, }); for name in columns { let name = match *name { diff --git a/crates/sqlbuild-state/src/state_tables/constants.rs b/crates/sqlbuild-state/src/state_tables/constants.rs index 1dd50fb1e..7ebe8b34c 100644 --- a/crates/sqlbuild-state/src/state_tables/constants.rs +++ b/crates/sqlbuild-state/src/state_tables/constants.rs @@ -3,3 +3,5 @@ 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"; +/// The column every event table identifies its rows by. +pub(crate) const EVENT_ID_COLUMN: &str = "event_id"; diff --git a/crates/sqlbuild-state/src/state_tables/main/event_id_index.rs b/crates/sqlbuild-state/src/state_tables/main/event_id_index.rs new file mode 100644 index 000000000..00b1e12e8 --- /dev/null +++ b/crates/sqlbuild-state/src/state_tables/main/event_id_index.rs @@ -0,0 +1,12 @@ +//! Where a history read's row carries its event ID. + +use crate::state_tables::constants::EVENT_ID_COLUMN; + +/// A selected `event_id` column, else the one the history read appends after the columns. +#[must_use] +pub fn event_id_index(columns: &[&str]) -> usize { + columns + .iter() + .position(|column| *column == EVENT_ID_COLUMN) + .unwrap_or(columns.len()) +} diff --git a/crates/sqlbuild-state/src/state_tables/main/mod.rs b/crates/sqlbuild-state/src/state_tables/main/mod.rs index 6b46ba4e9..b83c29f51 100644 --- a/crates/sqlbuild-state/src/state_tables/main/mod.rs +++ b/crates/sqlbuild-state/src/state_tables/main/mod.rs @@ -1,3 +1,4 @@ pub mod create_table_sql; +pub mod event_id_index; pub mod read_history_sql; pub mod values_insert_sql; 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 8739ededc..ab2b046bf 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,9 +1,9 @@ //! Keyset pages preserve distinct observations without rereading or ranking all preceding rows. -use crate::state_tables::constants::{MISSING_EVENT_ID_KEY, SQLSERVER_ADAPTER}; +use crate::state_tables::constants::{EVENT_ID_COLUMN, 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. +/// Rendered cursor values, `None` first; a trailing event ID repeating an `event_id` key is ignored. pub fn read_history_sql( qualified_name: &str, after: Option<&[Option]>, @@ -19,17 +19,21 @@ pub fn read_history_sql( limit, } = read; let mut projection = columns.join(", "); - projection.push_str(if has_event_id { - ", event_id" - } else { - ", NULL AS event_id" - }); - let mut columns = key_columns.unwrap_or(columns).to_vec(); - columns.push(if has_event_id { - "event_id" - } else { - MISSING_EVENT_ID_KEY - }); + if !columns.contains(&EVENT_ID_COLUMN) { + projection.push_str(if has_event_id { + ", event_id" + } else { + ", NULL AS event_id" + }); + } + let mut columns: Vec<&str> = key_columns.unwrap_or(columns).to_vec(); + if !columns.contains(&EVENT_ID_COLUMN) { + columns.push(if has_event_id { + EVENT_ID_COLUMN + } else { + MISSING_EVENT_ID_KEY + }); + } let ordered = columns .iter() .map(|column| { 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 adcb4e0b1..eab9c3779 100644 --- a/crates/sqlbuild-state/src/state_tables/tests/test_sql.rs +++ b/crates/sqlbuild-state/src/state_tables/tests/test_sql.rs @@ -1,10 +1,13 @@ +use crate::events::constants::{ + COLUMN_MIGRATION_COLUMNS, MIGRATION_COLUMNS, NODE_RESULT_COLUMNS, OLD_NAME_VIEW_COLUMNS, +}; 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, HistoryReadTestCase, ValuesInsertTestCase, + CreateTableTestCase, EventKeyTestCase, HistoryReadTestCase, ValuesInsertTestCase, }; #[test] @@ -110,3 +113,126 @@ fn given_a_table_without_event_ids_when_paging_history_then_the_missing_key_neve ); } } + +#[test] +fn given_event_tables_keyed_by_event_id_when_paging_history_then_each_column_and_key_appears_once() +{ + let test_cases = [ + EventKeyTestCase { + description: "migrations", + columns: &MIGRATION_COLUMNS, + key_columns: &["event_id", "created_at"], + timestamp_column: "created_at", + expected_ordering_keys: &["event_id", "created_at"], + }, + EventKeyTestCase { + description: "column migrations", + columns: &COLUMN_MIGRATION_COLUMNS, + key_columns: &["event_id", "created_at"], + timestamp_column: "created_at", + expected_ordering_keys: &["event_id", "created_at"], + }, + EventKeyTestCase { + description: "old-name views", + columns: &OLD_NAME_VIEW_COLUMNS, + key_columns: &["event_id", "created_at"], + timestamp_column: "created_at", + expected_ordering_keys: &["event_id", "created_at"], + }, + EventKeyTestCase { + description: "microbatches", + columns: &["event_id", "record_type", "scope_kind", "created_at"], + key_columns: &["event_id", "created_at"], + timestamp_column: "created_at", + expected_ordering_keys: &["event_id", "created_at"], + }, + EventKeyTestCase { + description: "janitor events", + columns: &["event_id", "schema_version", "event_type", "occurred_at"], + key_columns: &["event_id", "occurred_at"], + timestamp_column: "occurred_at", + expected_ordering_keys: &["event_id", "occurred_at"], + }, + EventKeyTestCase { + description: "node results, which carry no event ID column", + columns: &NODE_RESULT_COLUMNS, + key_columns: &[ + "node_type", + "node_name", + "target_database", + "target_schema", + "target_name", + "run_id", + "status", + "ts", + ], + timestamp_column: "ts", + expected_ordering_keys: &[ + "node_type", + "node_name", + "target_database", + "target_schema", + "target_name", + "run_id", + "status", + "ts", + "event_id", + ], + }, + ]; + let adapters = [ + "snowflake", + "duckdb", + "motherduck", + "sqlite", + "postgres", + "bigquery", + "databricks", + "sqlserver", + "clickhouse", + ]; + for test_case in test_cases { + for adapter in adapters { + let protocol_cursor = vec![Some("'x'".to_owned()); test_case.key_columns.len() + 1]; + let sql = read_history_sql( + "s.t", + Some(&protocol_cursor), + HistoryRead { + columns: test_case.columns, + key_columns: Some(test_case.key_columns), + timestamp_column: test_case.timestamp_column, + observed_since: None, + adapter, + has_event_id: true, + limit: 5, + }, + ) + .expect("supported adapter"); + let head = sql.strip_prefix("SELECT ").expect("select"); + let head = head.strip_prefix("TOP 5 ").unwrap_or(head); + let (projection, rest) = head.split_once(" FROM ").expect("from"); + let mut selected: Vec<&str> = projection.split(", ").collect(); + selected.sort_unstable(); + selected.dedup(); + assert_eq!( + selected.len(), + test_case.columns.len() + usize::from(!test_case.columns.contains(&"event_id")), + "{} on {adapter}: each column selected once in {sql}", + test_case.description + ); + let ordering_keys: Vec<&str> = rest + .split_once(" ORDER BY ") + .expect("order") + .1 + .split("CASE WHEN ") + .skip(1) + .filter_map(|item| item.split_once(" IS NULL").map(|(key, _)| key)) + .collect(); + assert_eq!( + ordering_keys, test_case.expected_ordering_keys, + "{} on {adapter}: {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 6914593ee..d33808bbb 100644 --- a/crates/sqlbuild-state/src/state_tables/tests/test_types.rs +++ b/crates/sqlbuild-state/src/state_tables/tests/test_types.rs @@ -19,3 +19,11 @@ pub(crate) struct HistoryReadTestCase { pub(crate) after: Option>>, pub(crate) expected_sql: &'static str, } + +pub(crate) struct EventKeyTestCase { + pub(crate) description: &'static str, + pub(crate) columns: &'static [&'static str], + pub(crate) key_columns: &'static [&'static str], + pub(crate) timestamp_column: &'static str, + pub(crate) expected_ordering_keys: &'static [&'static str], +} diff --git a/fensu.toml b/fensu.toml index 48fd8be4c..742fc514f 100644 --- a/fensu.toml +++ b/fensu.toml @@ -1647,8 +1647,9 @@ select = ["FPRS", "XSB072", "XSB073", "XSB075", "XSB076", "XSB077", "XSB078", "X [[targets.rust.threshold_overrides]] paths = ["crates/sqlbuild-python/Cargo.toml"] -# Raised from 217,769 for the identity-pass hooks, the sink operation filter and the checks-only debug loader. -thresholds = { max_crate_code_lines = 217820 } +# Raised from 217,769 for the identity-pass hooks, the sink operation filter and the checks-only debug loader, +# then from 217,820 for reporting the driver error of failed warehouse metadata reads. +thresholds = { max_crate_code_lines = 217836 } reason = "The Python bindings crate is above the code-line limit; this caps it at its current size until it is split into crates by area. Lower the cap as code moves out." [[targets.rust.threshold_overrides]] diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py index 924ff683f..a5598075c 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py @@ -317,6 +317,13 @@ class CatalogBoundaryTestCase: expected_page_delivered: bool +@dataclass(frozen=True) +class ReadFailureTestCase: + description: str + table: str + expected_stderr: str + + @dataclass(frozen=True) class StatisticsRecoveryTestCase: description: str 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 4007281a0..22e684b4a 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py @@ -444,7 +444,21 @@ def warehouse_history_page( observed_since: str | None = None, ) -> Any: """Exercise the real isolated reader without a server, state database or declaration import.""" - result: subprocess.CompletedProcess[str] = subprocess.run( + result: subprocess.CompletedProcess[str] = warehouse_history_read( + project, table, after, observed_since + ) + assert result.returncode == 0, result.stderr + return json.loads(result.stdout) + + +def warehouse_history_read( + project: Path, + table: str, + after: list[str | None] | None = None, + observed_since: str | None = None, +) -> subprocess.CompletedProcess[str]: + """One isolated reader exchange, with the reader's own report of a failed read.""" + return subprocess.run( [ str(Path(sys.executable).with_name("sqb")), "--project-dir", @@ -472,9 +486,8 @@ def warehouse_history_page( capture_output=True, timeout=20, env=os.environ | {"SQLBUILD_CONTROL_DATABASE_URL": "unsupported://must-not-open"}, + check=False, ) - assert result.returncode == 0, result.stderr - return json.loads(result.stdout) def warehouse_history_pages( diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_warehouse_catalog.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_warehouse_catalog.py index 2885a202c..2567dd67a 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_warehouse_catalog.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_warehouse_catalog.py @@ -168,7 +168,10 @@ def test_given_non_authoritative_adapter_scope_when_reading_catalog_then_no_page ) assert result.returncode == 0, result.stderr assert (json.loads(result.stdout) is not None) is test_case.expected_page_delivered - assert "No page delivered" in result.stderr + assert ( + "Warehouse metadata read (catalog) failed: ValueError: catalog listing is not " + "authoritative for this namespace; scan incomplete. No page delivered." + ) in result.stderr @pytest.mark.parametrize( 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 a20a7e800..81872d480 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 @@ -7,6 +7,7 @@ import os import shutil import sqlite3 +import subprocess import sys import time from contextlib import closing @@ -42,6 +43,7 @@ AutomaticHistoryTestCase, CatalogConnectionTestCase, FingerprintHistoryTestCase, + ReadFailureTestCase, StateEventsTestCase, SyncPacingTestCase, WarehouseHistoryTestCase, @@ -74,10 +76,43 @@ wait_for_warehouse_sync_health, warehouse_history_page, warehouse_history_pages, + warehouse_history_read, ) from tests.integration.src.sqlbuild.compiler.graph.helpers import write_files +@pytest.mark.parametrize( + "test_case", + [ + ReadFailureTestCase( + "Unopenable warehouse file", + "migrations", + "Warehouse connection for metadata reads failed: OperationalError: " + "unable to open database file\n" + "Warehouse metadata read (migrations) failed. No page delivered.\n", + ) + ], + ids=lambda case: case.description, +) +def test_given_an_unreachable_warehouse_when_reading_history_then_the_driver_error_is_reported( + test_case: ReadFailureTestCase, tmp_path: Path +) -> None: + project: Path = tmp_path / "project" + write_files( + project_dir=project, + files={ + "sqlbuild_project.toml": ( + 'name = "orders"\nadapter = "sqlite"\n' + '[connections.local]\npath = "missing/warehouse.db"\n' + '[targets.development]\nschema = "main"\n' + ) + }, + ) + result: subprocess.CompletedProcess[str] = warehouse_history_read(project, test_case.table) + assert (result.returncode, result.stdout) == (0, "null\n") + assert test_case.expected_stderr in result.stderr + + @pytest.mark.parametrize( "test_case", [