fix: read event-keyed warehouse history on Snowflake and report metadata read errors

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)))
+            }
         },
     }
 }

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::<Vec<_>>()
+        .join(" ")
+}

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::<Option<String>>()?,
+        "event_id": row.get_item(event_id_index(columns))?.extract::<Option<String>>()?,
     });
     for name in columns {
         let name = match *name {

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";

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())
+}

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;

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<String>]>,
@@ -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| {

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
+            );
+        }
+    }
+}

crates/sqlbuild-state/src/state_tables/tests/test_types.rs

@@ -19,3 +19,11 @@ pub(crate) struct HistoryReadTestCase {
     pub(crate) after: Option<Vec<Option<String>>>,
     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],
+}