diff --git a/crates/sqlbuild-cli/src/native_commands/_helpers/execution.rs b/crates/sqlbuild-cli/src/native_commands/_helpers/execution.rs index a7859224da..81706ccd14 100644 --- a/crates/sqlbuild-cli/src/native_commands/_helpers/execution.rs +++ b/crates/sqlbuild-cli/src/native_commands/_helpers/execution.rs @@ -1,5 +1,6 @@ //! Running a Python-free command: its checks, then its output with the observed services. +use std::collections::HashMap; use std::future::Future; use std::io; use std::path::{Path, PathBuf}; @@ -38,6 +39,7 @@ use sqlbuild_orchestration::warehouse_metadata::constants::MAX_HISTORY_RECORDS; use sqlbuild_orchestration::warehouse_metadata::models::{ MetadataBatch, MetadataScan, StatisticsDueRead, StatisticsRead, StatisticsRetention, WarehouseHistoryKind, WarehouseHistoryPage, WarehouseStateLocation, WarehouseSyncDueRead, + WarehouseWatermarkRead, }; use sqlbuild_orchestration::webhooks::constants::SECURITY_AUDIT_WEBHOOK_VARIABLE; use sqlbuild_orchestration::webhooks::main::{ @@ -69,9 +71,9 @@ use crate::native_commands::_helpers::terminal::{ finish_lost_stdout, terminal_columns, write_output, }; use crate::native_commands::constants::{ - ADMIN_COMMAND, COLUMNS_VARIABLE, FALLBACK_COLUMNS, METADATA_POLL_SECONDS, - METADATA_RECONCILE_SECONDS, METADATA_SYNC_CHECK_SECONDS, SCHEDULER_COMMAND, - UNCAUGHT_EXCEPTION_STATUS, UNKNOWN_COMMAND, + ADMIN_COMMAND, COLUMNS_VARIABLE, FALLBACK_COLUMNS, METADATA_OVERLAP_SECONDS, + METADATA_POLL_SECONDS, METADATA_RECONCILE_SECONDS, METADATA_SYNC_CHECK_SECONDS, + SCHEDULER_COMMAND, UNCAUGHT_EXCEPTION_STATUS, UNKNOWN_COMMAND, }; use crate::native_commands::errors::PythonNeeded; use crate::native_commands::main::_run_admin::run_admin; @@ -80,7 +82,7 @@ use crate::native_commands::main::_run_cost::run_cost; use crate::native_commands::models::{ CommandOutput, InstalledPackage, LocationCompileRequest, LocationCompiler, LocationMetadataPage, LocationMetadataSync, LocationServe, LostStdout, NativeCommand, - NativeInvocation, ProjectLocation, + NativeInvocation, ProjectLocation, ScanToken, }; use crate::native_commands::types::LocationPublication; use crate::workers::main::docker_image_sources::docker_image_sources; @@ -352,6 +354,10 @@ pub(crate) fn metadata_sync( next_at: Instant::now(), idle: true, cycle_started_at: None, + change_tokens: HashMap::new(), + scan_token: None, + scan_delivered: false, + mirror_holds_rows: false, } } @@ -365,10 +371,15 @@ pub(crate) fn history_request(state: &LocationMetadataSync) -> Result, C let kind = state.kind().as_str(); let request = if state.kind() == WarehouseHistoryKind::Statistics { json!({(kind): state.statistics.as_ref().ok_or_else(|| ControlError::Invalid("statistics eligibility page missing".into()))?}) - } else { + } else if state.kind() == WarehouseHistoryKind::Catalog { json!({(kind): { "location":state.location(),"after":state.source_after,"observed_since":state.observed_since,"limit":state.limit }}) + } else { + json!({(kind): { + "location":state.location(),"after":state.source_after,"observed_since":state.observed_since,"limit":state.limit, + "unchanged_token":state.unchanged_token() + }}) }; serde_json::to_vec(&request).map_err(|error| ControlError::Invalid(error.to_string())) } @@ -405,6 +416,11 @@ pub(crate) fn freeze_history_page( page: WarehouseHistoryPage, poll: &LocationWorkPoll, ) -> Result { + if state.source_after.is_none() { + state.scan_token.clone_from(&page.change_token); + state.scan_delivered = false; + } + state.scan_delivered |= !page.records.is_empty(); let batch = MetadataBatch { kind: state.kind(), session_id: poll.session_id, @@ -490,24 +506,91 @@ pub(crate) fn acknowledge_metadata_page(mut state: LocationMetadataSync) -> Loca if state.kind() == WarehouseHistoryKind::Statistics { state.statistics_limit = MAX_HISTORY_RECORDS; } - state.index += 1; - state.scan_id = Uuid::now_v7(); - state.observed_since = None; - state.watermark_loaded = false; - state.limit = MAX_HISTORY_RECORDS; - if state.index == state.locations.len() * WarehouseHistoryKind::ALL.len() { - state.index = 0; - if state.reconcile { - state.reconcile_at = - Instant::now() + Duration::from_secs(u64::from(state.reconcile_seconds)); + match state.scan_token.take() { + Some(token) => { + let recorded = ScanToken { + token, + reconciled: state.reconciling(), + delivered: state.scan_delivered || state.observed_since.is_some(), + }; + state.change_tokens.insert(state.index, recorded); + } + None => { + state.change_tokens.remove(&state.index); } - state.idle = true; - state.next_at = Instant::now(); - drop(write_output(&[( - Stream::Stderr, - "Warehouse history synchronization completed.\n", - )])); } + state = next_scan(state); + } + state +} + +/// The mirror's watermark bounds an incremental scan and shows whether the mirror holds rows. +pub(crate) async fn load_watermark( + mut state: LocationMetadataSync, + (options, client, poll): (&LocationServe, &LocationWorkClient, &LocationWorkPoll), + (started, stopped): (Instant, impl Future>), +) -> Result)>, ControlError> { + tokio::pin!(stopped); + if state.index == 0 { + state.reconcile = Instant::now() >= state.reconcile_at; + } + state.mirror_holds_rows = false; + if (!state.reconcile || state.change_tokens.contains_key(&state.index)) + && !matches!( + state.kind(), + WarehouseHistoryKind::Catalog | WarehouseHistoryKind::Statistics + ) + { + let request = WarehouseWatermarkRead { + session_id: poll.session_id, + generation: state.generation, + location: state.location().clone(), + kind: state.kind(), + }; + let watermark = tokio::select! { + result = &mut stopped => { result?; return Ok(None); }, + result = client.metadata_watermark(&options.target, &request) => result, + }; + let watermark = match watermark { + Ok(watermark) => watermark, + Err(error) => { + tokio::select! { + result = &mut stopped => { result?; return Ok(None); }, + () = metadata_health(&state, (client, poll), (started, Some(WarehouseSyncFailure::Watermark))) => {}, + } + return Ok(Some((state, Some(error)))); + } + }; + state.mirror_holds_rows = watermark.is_some(); + if !state.reconcile { + state.observed_since = watermark.and_then(|time| { + time.checked_sub_signed(chrono::Duration::seconds(METADATA_OVERLAP_SECONDS)) + }); + } + } + state.watermark_loaded = true; + Ok(Some((state, None))) +} + +/// Move to the next scan; after the last one the synchronization completes and waits for a run. +pub(crate) fn next_scan(mut state: LocationMetadataSync) -> LocationMetadataSync { + state.index += 1; + state.scan_id = Uuid::now_v7(); + state.observed_since = None; + state.watermark_loaded = false; + state.limit = MAX_HISTORY_RECORDS; + if state.index == state.locations.len() * WarehouseHistoryKind::ALL.len() { + state.index = 0; + if state.reconcile { + state.reconcile_at = + Instant::now() + Duration::from_secs(u64::from(state.reconcile_seconds)); + } + state.idle = true; + state.next_at = Instant::now(); + drop(write_output(&[( + Stream::Stderr, + "Warehouse history synchronization completed.\n", + )])); } state } diff --git a/crates/sqlbuild-cli/src/native_commands/main/_location_metadata.rs b/crates/sqlbuild-cli/src/native_commands/main/_location_metadata.rs index 90bbbf4164..4e90fadcd5 100644 --- a/crates/sqlbuild-cli/src/native_commands/main/_location_metadata.rs +++ b/crates/sqlbuild-cli/src/native_commands/main/_location_metadata.rs @@ -10,20 +10,18 @@ use sqlbuild_orchestration::platform::models::{ }; use sqlbuild_orchestration::store::errors::ControlError; use sqlbuild_orchestration::warehouse_metadata::models::{ - WarehouseHistoryKind, WarehouseHistoryPage, WarehouseWatermarkRead, + WarehouseHistoryKind, WarehouseHistoryPage, }; use tokio::time::Instant; use crate::argv::models::Stream; use crate::native_commands::_helpers::execution::{ acknowledge_metadata_page, begin_after_run, freeze_history_page, history_request, - location_result, metadata_health, statistics_page, + load_watermark, location_result, metadata_health, next_scan, statistics_page, }; use crate::native_commands::_helpers::paths::location_inputs; use crate::native_commands::_helpers::terminal::write_output; -use crate::native_commands::constants::{ - METADATA_OVERLAP_SECONDS, METADATA_POLL_SECONDS, METADATA_REPLY_BYTES, -}; +use crate::native_commands::constants::{METADATA_POLL_SECONDS, METADATA_REPLY_BYTES}; use crate::native_commands::models::{LocationMetadataSync, LocationServe, LocationWatch}; /// One bounded source or mirror page leaves named work polling a turn between pages. @@ -64,40 +62,15 @@ pub(super) async fn sync_location_metadata( }); } if !state.watermark_loaded { - if state.index == 0 { - state.reconcile = Instant::now() >= state.reconcile_at; - } - if !state.reconcile - && !matches!( - state.kind(), - WarehouseHistoryKind::Catalog | WarehouseHistoryKind::Statistics - ) - { - let request = WarehouseWatermarkRead { - session_id: poll.session_id, - generation: state.generation, - location: state.location().clone(), - kind: state.kind(), - }; - let watermark = tokio::select! { - result = &mut stopped => { result?; return Ok(None); }, - result = client.metadata_watermark(&options.target, &request) => result, - }; - let watermark = match watermark { - Ok(watermark) => watermark, - Err(error) => { - tokio::select! { - result = &mut stopped => { result?; return Ok(None); }, - () = metadata_health(&state, (client, poll), (started, Some(WarehouseSyncFailure::Watermark))) => {}, - } - return retry_metadata(state, error); - } - }; - state.observed_since = watermark.and_then(|time| { - time.checked_sub_signed(chrono::Duration::seconds(METADATA_OVERLAP_SECONDS)) - }); + let Some((loaded, failure)) = + load_watermark(state, (options, client, poll), (started, stopped.as_mut())).await? + else { + return Ok(None); + }; + state = loaded; + if let Some(error) = failure { + return retry_metadata(state, error); } - state.watermark_loaded = true; } let Some(page) = read_metadata_page((options, script), &state, stopped.as_mut()).await? else { @@ -119,6 +92,13 @@ pub(super) async fn sync_location_metadata( { return Ok(Some(state)); } + if page.unchanged { + tokio::select! { + result = &mut stopped => { result?; return Ok(None); }, + () = metadata_health(&state, (client, poll), (started, None)) => {}, + } + return Ok(Some(next_scan(state))); + } state = freeze_history_page(state, page, poll)?; } let Some(pending) = &state.pending else { @@ -173,6 +153,8 @@ async fn read_metadata_page( return Ok(Some(Some(WarehouseHistoryPage { records: Vec::new(), next_after: statistics.next_after.map(|id| vec![Some(id.to_string())]), + change_token: None, + unchanged: false, }))); } let (mut child, reply) = match location_result::>( diff --git a/crates/sqlbuild-cli/src/native_commands/models.rs b/crates/sqlbuild-cli/src/native_commands/models.rs index 99f7eb26df..d1911db3b1 100644 --- a/crates/sqlbuild-cli/src/native_commands/models.rs +++ b/crates/sqlbuild-cli/src/native_commands/models.rs @@ -1,5 +1,6 @@ //! Python-free commands, the invocation facts they run with and what they print. +use std::collections::HashMap; use std::path::PathBuf; use chrono::{DateTime, Utc}; @@ -208,6 +209,24 @@ pub(crate) struct LocationMetadataSync { pub(crate) idle: bool, /// Server time when the current or last synchronization began; `None` before the first. pub(crate) cycle_started_at: Option>, + /// Per scan index: the table's change token after its last complete read. + pub(crate) change_tokens: HashMap, + /// The token read before the current scan's first page, recorded once the scan completes. + pub(crate) scan_token: Option, + /// Whether any page of the current scan delivered records. + pub(crate) scan_delivered: bool, + /// The server already mirrors rows of the current scan's table, per its watermark. + pub(crate) mirror_holds_rows: bool, +} + +/// A table's change token after a complete read, and what that read proves about the mirror. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct ScanToken { + pub(crate) token: String, + /// A complete reconciling read, so the mirror held exactly the table's rows. + pub(crate) reconciled: bool, + /// The read delivered rows or read on top of mirrored rows, so an empty mirror lost rows. + pub(crate) delivered: bool, } impl LocationMetadataSync { @@ -241,6 +260,14 @@ impl LocationMetadataSync { && (self.reconcile || self.kind() == WarehouseHistoryKind::Catalog) } + /// The recorded token a scan may skip on, given whether it reconciles and the mirror holds rows. + pub(crate) fn unchanged_token(&self) -> Option<&str> { + let recorded = self.change_tokens.get(&self.index)?; + ((recorded.reconciled || !self.reconciling()) + && (self.mirror_holds_rows || !recorded.delivered)) + .then_some(recorded.token.as_str()) + } + pub(crate) fn kind(&self) -> WarehouseHistoryKind { WarehouseHistoryKind::ALL[self.index % WarehouseHistoryKind::ALL.len()] } 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 index a888efba54..cb3a03c279 100644 --- a/crates/sqlbuild-cli/src/native_commands/tests/test_location_metadata.rs +++ b/crates/sqlbuild-cli/src/native_commands/tests/test_location_metadata.rs @@ -5,7 +5,7 @@ use sqlbuild_orchestration::warehouse_metadata::models::{ 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; +use crate::native_commands::tests::test_types::{MetadataRetryTestCase, UnchangedScanTestCase}; #[test] fn given_failed_metadata_read_when_back_at_statistics_then_page_shrinks_until_scan_completes() { @@ -53,3 +53,111 @@ fn given_failed_metadata_read_when_back_at_statistics_then_page_shrinks_until_sc ); } } + +#[test] +fn given_recorded_change_token_when_scanning_again_then_only_proven_unchanged_reads_skip() { + let test_cases = [ + UnchangedScanTestCase { + description: "an incremental read lets the next incremental scan skip", + read_reconciled: false, + read_delivered: true, + read_incremental: false, + rereads_without_token: 0, + scan_reconciles: false, + mirror_holds_rows: true, + expected_token: Some("t1"), + }, + UnchangedScanTestCase { + description: "an incremental read never stands in for a reconciling scan", + read_reconciled: false, + read_delivered: true, + read_incremental: false, + rereads_without_token: 0, + scan_reconciles: true, + mirror_holds_rows: true, + expected_token: None, + }, + UnchangedScanTestCase { + description: "a reconciling read lets a reconciling scan skip", + read_reconciled: true, + read_delivered: true, + read_incremental: false, + rereads_without_token: 0, + scan_reconciles: true, + mirror_holds_rows: true, + expected_token: Some("t1"), + }, + UnchangedScanTestCase { + description: "a mirror that lost delivered rows is read again", + read_reconciled: true, + read_delivered: true, + read_incremental: false, + rereads_without_token: 0, + scan_reconciles: false, + mirror_holds_rows: false, + expected_token: None, + }, + UnchangedScanTestCase { + description: "an empty table with an empty mirror stays skipped", + read_reconciled: true, + read_delivered: false, + read_incremental: false, + rereads_without_token: 0, + scan_reconciles: true, + mirror_holds_rows: false, + expected_token: Some("t1"), + }, + UnchangedScanTestCase { + description: "an incremental read that delivered nothing still guards a lost mirror", + read_reconciled: false, + read_delivered: false, + read_incremental: true, + rereads_without_token: 0, + scan_reconciles: false, + mirror_holds_rows: false, + expected_token: None, + }, + UnchangedScanTestCase { + description: "a later read without a token forgets the recorded one", + read_reconciled: false, + read_delivered: true, + read_incremental: false, + rereads_without_token: 1, + scan_reconciles: false, + mirror_holds_rows: true, + expected_token: None, + }, + ]; + 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.reconcile = test_case.read_reconciled; + state.scan_token = Some("t1".into()); + state.scan_delivered = test_case.read_delivered; + state.observed_since = test_case.read_incremental.then(chrono::Utc::now); + let mut state = acknowledge_complete_pages(state, 1); + assert_eq!( + state.kind(), + WarehouseHistoryKind::SourceFreshness, + "{description}", + description = test_case.description + ); + state.index = 0; + let mut state = (0..test_case.rereads_without_token).fold(state, |state, _| { + let mut state = acknowledge_complete_pages(state, 1); + state.index = 0; + state + }); + state.reconcile = test_case.scan_reconciles; + state.mirror_holds_rows = test_case.mirror_holds_rows; + assert_eq!( + state.unchanged_token(), + test_case.expected_token, + "{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 580e0ee5e4..46d3cc8389 100644 --- a/crates/sqlbuild-cli/src/native_commands/tests/test_types.rs +++ b/crates/sqlbuild-cli/src/native_commands/tests/test_types.rs @@ -99,3 +99,15 @@ pub(super) struct MetadataRetryTestCase { pub(super) acknowledged_pages: usize, pub(super) expected_statistics_limit: usize, } + +/// A completed state-table read, the later scan of the same table, and whether that scan may skip. +pub(super) struct UnchangedScanTestCase { + pub(super) description: &'static str, + pub(super) read_reconciled: bool, + pub(super) read_delivered: bool, + pub(super) read_incremental: bool, + pub(super) rereads_without_token: usize, + pub(super) scan_reconciles: bool, + pub(super) mirror_holds_rows: bool, + pub(super) expected_token: Option<&'static str>, +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs index 4c44e2d64e..5281ecee88 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs @@ -415,6 +415,9 @@ pub struct SourceFreshnessRead { /// Inclusive time bound, held fixed across pages; independent of the row-key cursor. pub observed_since: Option>, pub limit: Option, + /// The table's change token after the last complete read; an equal token skips this read. + #[serde(default)] + pub unchanged_token: Option, } /// Fingerprints remain history here; latest-row selection belongs to the existing planner reader. @@ -479,6 +482,9 @@ pub struct FingerprintRead { pub after: Option<[Option; 13]>, pub observed_since: Option>, pub limit: Option, + /// The table's change token after the last complete read; an equal token skips this read. + #[serde(default)] + pub unchanged_token: Option, } /// Node results carry no event identity, so the cursor holds the row's identity columns. @@ -489,6 +495,9 @@ pub struct NodeResultsRead { pub after: Option<[Option; 9]>, pub observed_since: Option>, pub limit: Option, + /// The table's change token after the last complete read; an equal token skips this read. + #[serde(default)] + pub unchanged_token: Option, } /// Consumption rows carry no event identity, so the cursor holds the row's identity columns. @@ -499,6 +508,9 @@ pub struct ConsumptionRead { pub after: Option<[Option; 7]>, pub observed_since: Option>, pub limit: Option, + /// The table's change token after the last complete read; an equal token skips this read. + #[serde(default)] + pub unchanged_token: Option, } /// Stable event identities and timestamps order observations without comparing their payloads. @@ -509,6 +521,9 @@ pub struct StateEventsRead { pub after: Option<[Option; 3]>, pub observed_since: Option>, pub limit: Option, + /// The table's change token after the last complete read; an equal token skips this read. + #[serde(default)] + pub unchanged_token: Option, } /// The private interpreter accepts only known state tables, never arbitrary SQL. @@ -551,6 +566,12 @@ pub enum WarehouseHistoryRecord { pub struct WarehouseHistoryPage { pub records: Vec, pub next_after: Option>>, + /// The table's change token, read before a scan's first page where the warehouse offers one. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub change_token: Option, + /// The token still equals the request's: nothing was read and nothing is delivered. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub unchanged: bool, } /// Typed import columns are selected by code, never from location-supplied field names. 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 1fd2b10c0b..4b57169e6c 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 @@ -104,6 +104,8 @@ pub(crate) fn read_catalog( return Ok(WarehouseHistoryPage { records: Vec::new(), next_after: None, + change_token: None, + unchanged: false, }); } let keys = selected.iter().map(|(_, _, key)| key); @@ -164,5 +166,7 @@ pub(crate) fn read_catalog( Ok(WarehouseHistoryPage { records, next_after, + change_token: None, + unchanged: false, }) } 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 1ec0bd1e50..49e7e71160 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 @@ -262,6 +262,21 @@ fn read_page( return Ok(WarehouseHistoryPage { records: Vec::new(), next_after: None, + change_token: None, + unchanged: false, + }); + } + let change_token: Option = if after.is_none() { + relation_change_token(py, (adapter, connection), (&database, &schema, &name))? + } else { + None + }; + if change_token.is_some() && change_token.as_deref() == unchanged_token(request) { + return Ok(WarehouseHistoryPage { + records: Vec::new(), + next_after: None, + change_token, + unchanged: true, }); } let catalog_columns = call_with( @@ -462,6 +477,52 @@ fn read_page( Ok(WarehouseHistoryPage { records, next_after, + change_token, + unchanged: false, + }) +} + +/// The token a complete earlier read recorded; only state-table scans carry one. +fn unchanged_token(request: &WarehouseHistoryRead) -> Option<&str> { + match request { + WarehouseHistoryRead::SourceFreshness(read) => read.unchanged_token.as_deref(), + WarehouseHistoryRead::Fingerprints(read) => read.unchanged_token.as_deref(), + WarehouseHistoryRead::AuditResults(read) + | WarehouseHistoryRead::JanitorEvents(read) + | WarehouseHistoryRead::Migrations(read) + | WarehouseHistoryRead::OldNameViews(read) + | WarehouseHistoryRead::Microbatches(read) + | WarehouseHistoryRead::ColumnMigrations(read) => read.unchanged_token.as_deref(), + WarehouseHistoryRead::NodeResults(read) => read.unchanged_token.as_deref(), + WarehouseHistoryRead::Consumption(read) => read.unchanged_token.as_deref(), + WarehouseHistoryRead::Catalog(_) | WarehouseHistoryRead::Statistics(_) => None, + } +} + +/// The adapter's change token for one table; without one, or when reading it fails, the scan reads as before. +fn relation_change_token( + py: Python<'_>, + (adapter, connection): (&Bound<'_, PyAny>, &Bound<'_, PyAny>), + (database, schema, name): (&Bound<'_, PyAny>, &Bound<'_, PyAny>, &Bound<'_, PyAny>), +) -> PyResult> { + let reader = py + .import("sqlbuild.adapter.relations.types")? + .getattr("RelationChangeTokenReader")?; + if !adapter.is_instance(&reader)? { + return Ok(None); + } + let token = call_with( + &adapter.getattr("read_relation_change_token")?, + &[ + ("connection", connection.clone()), + ("database", database.clone()), + ("schema", schema.clone()), + ("name", name.clone()), + ], + ); + Ok(match token { + Ok(token) if !token.is_none() => Some(token.extract()?), + _ => None, }) } diff --git a/crates/sqlbuild-python/src/bindings/_helpers/native_host/warehouse_statistics.rs b/crates/sqlbuild-python/src/bindings/_helpers/native_host/warehouse_statistics.rs index 3114d4e852..d991a3102f 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/native_host/warehouse_statistics.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/native_host/warehouse_statistics.rs @@ -57,5 +57,7 @@ pub(crate) fn read_statistics( Ok(WarehouseHistoryPage { records, next_after: request.next_after.map(|id| vec![Some(id.to_string())]), + change_token: None, + unchanged: false, }) } diff --git a/fensu.toml b/fensu.toml index fe1a9384f5..e9afc99dd8 100644 --- a/fensu.toml +++ b/fensu.toml @@ -52,6 +52,11 @@ modules = ["sqlbuild.adapter.contract.classes.base_adapter"] symbols = ["BaseAdapter"] reason = "Native code resolves these through the sqlbuild._native_host name table." +[[targets.python.dead_code.roots]] +modules = ["sqlbuild.adapter.relations.types"] +symbols = ["RelationChangeTokenReader"] +reason = "Native code imports this protocol by path for an isinstance capability check." + [[targets.python.dead_code.roots]] modules = ["sqlbuild.adapter.contract.exceptions"] symbols = ["AdapterUserError"] diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/server.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/server.md index 51d0f0119e..d315dd698d 100644 --- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/server.md +++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/server.md @@ -175,7 +175,8 @@ sqb control sync \ The command prints `{"applied": true}` when the server has queued the request. The location starts the sync within a few seconds, and the top bar shows when it last synced. A sync reads the -warehouse and never writes to it. +warehouse and never writes to it. On Snowflake, it skips state tables that haven't changed since +the last sync, so a sync with nothing new doesn't resume the warehouse. Requesting a sync needs the same access as a connection test: Developer or Admin on the location and target. Viewer and Operator cannot request one. Every request is recorded in the audit log. diff --git a/src/sqlbuild/adapter/relations/types.py b/src/sqlbuild/adapter/relations/types.py index c67c1359a2..4863de5887 100644 --- a/src/sqlbuild/adapter/relations/types.py +++ b/src/sqlbuild/adapter/relations/types.py @@ -69,6 +69,17 @@ def metadata_name_key(self, name: str) -> str: ... +@runtime_checkable +class RelationChangeTokenReader(Protocol): + """Adapter that reads a table's change token without starting warehouse compute.""" + + def read_relation_change_token( + self, *, connection: Any, database: str | None, schema: str, name: str + ) -> str | None: + """A token that differs after every write to the table, or None when unknown.""" + ... + + class ViewDefinitionBatchReader(Protocol): """An adapter's ``read_view_definitions``: stored definitions of named views in one schema.""" diff --git a/src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py b/src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py index 729942f5d8..b8fb21158e 100644 --- a/src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py +++ b/src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py @@ -2199,6 +2199,22 @@ def _relation_info_from_row(*, row: tuple[Any, ...], database: str | None) -> Re retention_days=None if row[6] is None else int(row[6]), ) + def read_relation_change_token( + self, *, connection: _SnowflakeConnection, database: str | None, schema: str, name: str + ) -> str | None: + """The last commit time, which cloud services answer without resuming a warehouse.""" + + qualified: str | None = self.render_qualified_name( + database=database, schema=schema, name=name + ) + if qualified is None: + return None + literal: str = self._quote_sql_string(qualified) + token_sql: str = f"SELECT SYSTEM$LAST_CHANGE_COMMIT_TIME({literal})" + result: QueryResult = self.query(connection=connection, sql=token_sql, limit=1) + token: object = result.rows[0][0] if result.rows else None + return None if token is None else str(token) + def read_schema_relation_listing( self, *, connection: _SnowflakeConnection, database: str | None, schema: str ) -> SchemaRelationListing: 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 30d62cfc26..84ddc86e93 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 @@ -1267,3 +1267,131 @@ def page(after: list[str | None] | None = None, observed_since: str | None = Non (project / "warehouse.db").rename(project / "retained-warehouse.db") assert page() is None assert not (project / "warehouse.db").exists() + + +@pytest.mark.parametrize( + "test_case", + [AutomaticHistoryTestCase("DuckDB change tokens", "duckdb", "database", "legacy-run")], + ids=lambda case: case.description, +) +def test_given_change_tokens_when_a_served_location_resyncs_then_only_changed_or_lost_tables_are_read( + test_case: AutomaticHistoryTestCase, tmp_path: Path +) -> None: + project: Path = tmp_path / "project" + adapter: Any + connection: Any + adapter, connection = open_legacy_warehouse( + project, test_case.adapter_name, test_case.connection_key + ) + adapter.close(connection) + (project / "orchestration/orders.py").write_text("") + fixture: dict[str, Any] = json.loads( + Path( + "benchmarks/duckshop/edge_cases/orchestration/automatic_warehouse_history.json" + ).read_text() + ) + write_files(project_dir=project, files={"models/orders.sql": fixture["model"]}) + hook: Path = tmp_path / "token-hook" + hook.mkdir() + tokens: Path = hook / "tokens.json" + reads: Path = hook / "reads.log" + tokens.write_text( + json.dumps({"_sqlbuild_fingerprints": "f1", "_sqlbuild_source_freshness": "s1"}) + ) + (hook / "sitecustomize.py").write_text( + "import json, re\n" + "from pathlib import Path\n" + "from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter\n" + "def read_relation_change_token(self, *, connection, database, schema, name):\n" + f" return json.loads(Path({str(tokens)!r}).read_text()).get(name)\n" + "DuckDbAdapter.read_relation_change_token = read_relation_change_token\n" + "original_query = DuckDbAdapter.query\n" + "def query(self, *, connection, sql, limit):\n" + " read = re.search(r'FROM\\s+\\S*(_sqlbuild_\\w+)', sql)\n" + " if read:\n" + f" with Path({str(reads)!r}).open('a') as log:\n" + " log.write(read.group(1) + '\\n')\n" + " return original_query(self, connection=connection, sql=sql, limit=limit)\n" + "DuckDbAdapter.query = query\n" + ) + mirrors: tuple[str, str] = ("sqb_warehouse_fingerprints", "sqb_warehouse_freshness") + with snapshot_api(tmp_path / "control") as api: + log: Path = tmp_path / "location.log" + control: Path = tmp_path / "control/.sqlbuild/control.sqlite3" + with location_process( + [ + str(Path(sys.executable).with_name("sqb")), + "--project-dir", + str(project), + "location", + "serve", + "--server", + api.url, + "--location-id", + api.location_id, + "--target", + "development", + ], + log, + tmp_path / "location.stdout", + env=os.environ + | { + "SQLBUILD_LOCATION_TOKEN": api.bearer, + "SQLBUILD_METADATA_RECONCILE_SECONDS": "3600", + "PYTHONPATH": str(hook), + }, + ) as location: + _ = wait_for_generation(api, "development", 1, location, log) + + def mirrored() -> tuple[int, int]: + with closing(sqlite3.connect(control)) as database: + counts: tuple[int, int] = database.execute( + f"SELECT (SELECT COUNT(*) FROM {mirrors[0]})," + f"(SELECT COUNT(*) FROM {mirrors[1]})" + ).fetchone() + return counts + + def resync() -> list[str]: + """Start one synchronization after a finished run and return the tables it read.""" + completed: int = log.read_text().count("synchronization completed") + reads.write_text("") + record_finished_run(control) + deadline: float = time.monotonic() + 45 + while log.read_text().count("synchronization completed") == completed: + assert location.poll() is None, log.read_text() + assert time.monotonic() < deadline, log.read_text() + time.sleep(0.2) + return sorted(set(reads.read_text().split())) + + deadline: float = time.monotonic() + 45 + while log.read_text().count("synchronization completed") == 0 or 0 in mirrored(): + assert location.poll() is None, log.read_text() + assert time.monotonic() < deadline, log.read_text() + time.sleep(0.2) + initial: tuple[int, int] = mirrored() + + assert resync() == [], "unchanged tokens read nothing" + assert mirrored() == initial + + connection = adapter.connect({test_case.connection_key: str(project / "warehouse.db")}) + _ = adapter.execute( + connection=connection, + sql=( + "INSERT INTO _sqlbuild_fingerprints SELECT node_type,'orders-changed'," + "target_database,target_schema,target_name,'changed-run',definition_hash," + "version_hash,schema_fingerprint,definition_b64,metadata_json_b64,ts " + f"FROM _sqlbuild_fingerprints WHERE run_id='{test_case.expected_run_id}'" + ), + ) + adapter.close(connection) + tokens.write_text( + json.dumps({"_sqlbuild_fingerprints": "f2", "_sqlbuild_source_freshness": "s1"}) + ) + assert resync() == ["_sqlbuild_fingerprints"], "only the changed table is read" + assert mirrored() == (initial[0] + 1, initial[1]) + + with closing(sqlite3.connect(control)) as database: + _ = database.execute(f"DELETE FROM {mirrors[1]}") + database.commit() + assert resync() == ["_sqlbuild_source_freshness"], "a lost mirror is read again" + assert mirrored() == (initial[0] + 1, initial[1]) diff --git a/website/src/content/docs/docs/cli/server.mdx b/website/src/content/docs/docs/cli/server.mdx index a580028186..7ac984268b 100644 --- a/website/src/content/docs/docs/cli/server.mdx +++ b/website/src/content/docs/docs/cli/server.mdx @@ -163,7 +163,8 @@ sqb control sync \ The command prints `{"applied": true}` when the server has queued the request. The location starts the sync within a few seconds, and the top bar shows when it last synced. A sync reads the -warehouse and never writes to it. +warehouse and never writes to it. On Snowflake, it skips state tables that haven't changed since +the last sync, so a sync with nothing new doesn't resume the warehouse. Requesting a sync needs the same access as a connection test: Developer or Admin on the location and target. Viewer and Operator cannot request one. Every request is recorded in the audit log.