feat: skip unchanged state-table reads during warehouse sync

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<Vec<u8>, 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<LocationMetadataSync, ControlError> {
+    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<Output = Result<(), ControlError>>),
+) -> Result<Option<(LocationMetadataSync, Option<ClientError>)>, 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
 }

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::<Option<WarehouseHistoryPage>>(

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<DateTime<Utc>>,
+    /// Per scan index: the table's change token after its last complete read.
+    pub(crate) change_tokens: HashMap<usize, ScanToken>,
+    /// The token read before the current scan's first page, recorded once the scan completes.
+    pub(crate) scan_token: Option<String>,
+    /// 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()]
     }