From 14869e79f7c3e867d010155e1a743be830ab2d84 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Thu, 12 Nov 2026 15:09:16 +0000 Subject: [PATCH] feat: skip unchanged state-table reads during warehouse sync A code location's sync now asks the adapter for a change token before reading each state table. When the token matches the last complete read of that table, the sync skips it without reading any rows. On Snowflake the token is SYSTEM$LAST_CHANGE_COMMIT_TIME, which cloud services answer without resuming the warehouse, so a sync with nothing new leaves the warehouse suspended. A reconciling scan skips only on a token recorded by a reconciling read, and a scan never skips when the mirror has lost rows that read had. Adapters without the capability, missing tables and token errors fall back to reading as before. --- .../src/native_commands/_helpers/execution.rs | 125 ++++++++++++++--- .../main/_location_metadata.rs | 58 +++----- .../src/native_commands/models.rs | 27 ++++ .../tests/test_location_metadata.rs | 110 ++++++++++++++- .../src/native_commands/tests/test_types.rs | 12 ++ .../src/warehouse_metadata/models.rs | 21 +++ .../_helpers/native_host/warehouse_catalog.rs | 4 + .../native_host/warehouse_metadata.rs | 61 +++++++++ .../native_host/warehouse_statistics.rs | 2 + fensu.toml | 5 + .../sqlbuild/references/docs/cli/server.md | 3 +- src/sqlbuild/adapter/relations/types.py | 11 ++ .../snowflake/classes/snowflake_adapter.py | 16 +++ .../publication/test_warehouse_metadata.py | 128 ++++++++++++++++++ website/src/content/docs/docs/cli/server.mdx | 3 +- 15 files changed, 524 insertions(+), 62 deletions(-) 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.