fix(materialize): bound the full read a source falls back to when its watermark expires

fluree-db-api/src/graph_source/r2rml_materialize.rs

@@ -92,6 +92,14 @@ const WATERMARK_SUBJECT_PREFIX: &str = "urn:fluree:materialize-state:";
 /// Predicate holding the last materialized source snapshot id (stored as a
 /// string to preserve full i64 precision for 19-digit snapshot ids).
 const WATERMARK_SNAPSHOT_PRED: &str = "urn:fluree:materialize#lastSnapshotId";
+/// The resumable SEQUENCE cursor, written when a pass covered only a prefix.
+///
+/// Additive on purpose: `lastSnapshotId` keeps its meaning and its readers, so a
+/// deployment can roll back to an image without this and lose only the resumability,
+/// not its watermarks. A sequence is stored alongside rather than instead of the
+/// snapshot because the snapshot is still the fast path — the cursor only matters
+/// once the snapshot stops resolving.
+const WATERMARK_SEQUENCE_PRED: &str = "urn:fluree:materialize#appliedSequence";
 /// Predicate recording which source the watermark belongs to (informational).
 const WATERMARK_SOURCE_PRED: &str = "urn:fluree:materialize#source";
 /// Predicate recording which target-spec (ledger id or template) the watermark
@@ -213,6 +221,7 @@ pub trait MaterializeSource: Send + Sync {
         graph_source_id: &str,
         table_name: &str,
         from_snapshot_id: Option<i64>,
+        from_sequence: Option<i64>,
     ) -> Result<MaterializeScan>;
 }
 
@@ -236,11 +245,19 @@ impl MaterializeSource for FlureeR2rmlProvider<'_> {
         graph_source_id: &str,
         table_name: &str,
         from_snapshot_id: Option<i64>,
+        from_sequence: Option<i64>,
     ) -> Result<MaterializeScan> {
         Ok(self
             // `to = None`: the sync-to-head worker always reads to the source's
             // current snapshot; explicit pins are for point-in-time consumers.
-            .scan_for_materialize_stream(graph_source_id, table_name, &[], from_snapshot_id, None)
+            .scan_for_materialize_stream(
+                graph_source_id,
+                table_name,
+                &[],
+                from_snapshot_id,
+                None,
+                from_sequence,
+            )
             .await?)
     }
 }
@@ -383,7 +400,8 @@ impl Fluree {
         // no-data poll must still persist its watermark to keep it resolvable.
         let mut watermark_refresh_due = false;
         // (table, from-snapshot, advanced to-snapshot) per source table.
-        let mut table_watermarks: Vec<(String, Option<i64>, i64)> = Vec::new();
+        // (table, from-snapshot, advanced to-snapshot, prefix sequence if partial)
+        let mut table_watermarks: Vec<(String, Option<i64>, i64, Option<i64>)> = Vec::new();
 
         for (table_name, tms) in &tables {
             let from_t = if force_full {
@@ -397,6 +415,22 @@ impl Fluree {
                 )
                 .await?
             };
+            // The SEQUENCE cursor, when a previous pass stopped short. It is what
+            // makes a full read resumable on a table whose snapshot history has been
+            // expired away: a snapshot id stops resolving, a commit sequence does
+            // not. `force_full` deliberately ignores it — an explicit full refresh
+            // means "start over", so honouring a cursor would silently narrow it.
+            let from_seq = if force_full {
+                None
+            } else {
+                self.materialize_sequence_cursor(
+                    MATERIALIZE_STATE_LEDGER,
+                    source_graph_source_id,
+                    target_ledger_id,
+                    table_name,
+                )
+                .await?
+            };
 
             // STREAM the scan; do not collect it. A full read is mandatory whenever
             // the snapshot window contains overwrite/delete, so on some sources this
@@ -413,9 +447,12 @@ impl Fluree {
             // and more predictable term than the raw columnar data, but it is not
             // O(1) — a window with millions of distinct subjects is still large.
             let scan = provider
-                .scan_window(source_graph_source_id, table_name, from_t)
+                .scan_window(source_graph_source_id, table_name, from_t, from_seq)
                 .await?;
             let (to_id, incremental) = (scan.to_snapshot_id, scan.incremental);
+            // `Some` => this pass covers only a PREFIX, so the snapshot watermark
+            // must not advance; see `MaterializeScan::to_sequence`.
+            let to_seq = scan.to_sequence;
             // A window older than the refresh bound must persist its watermark even
             // with zero rows — see `watermark_refresh_bound_ms`.
             if scan
@@ -436,7 +473,7 @@ impl Fluree {
             if let Some(to) = to_id {
                 any_table = true;
                 incremental_all = incremental_all && incremental;
-                table_watermarks.push(((*table_name).to_string(), from_t, to));
+                table_watermarks.push(((*table_name).to_string(), from_t, to, to_seq));
             }
 
             while let Some(batch) = batch_stream.next().await {
@@ -728,8 +765,14 @@ impl Fluree {
             let state = self.materialize_state_ledger().await?;
             let watermark_nodes: Vec<JsonValue> = table_watermarks
                 .iter()
-                .map(|(table, _from, to)| {
-                    watermark_node(source_graph_source_id, target_ledger_id, table, *to)
+                .map(|(table, _from, to, to_seq)| {
+                    watermark_node(
+                        source_graph_source_id,
+                        target_ledger_id,
+                        table,
+                        *to,
+                        *to_seq,
+                    )
                 })
                 .collect();
             // Through the backpressure helper, not a bare upsert. This write is
@@ -794,6 +837,42 @@ impl Fluree {
         Ok(extract_first_i64(&json))
     }
 
+    /// The stored SEQUENCE cursor for one (source, target, table), if any.
+    ///
+    /// Mirrors [`Self::materialize_watermark`] but reads
+    /// [`WATERMARK_SEQUENCE_PRED`]. Absent means "no pass has stopped short", which
+    /// is both the first-run state and the steady state — a healthy job never writes
+    /// one, so its absence must read as "no constraint", never as zero. Zero would be
+    /// a cursor at the very beginning of history and would silently narrow every
+    /// subsequent read to "everything after the start", which is a different query.
+    pub async fn materialize_sequence_cursor(
+        &self,
+        state_ledger_id: &str,
+        source_graph_source_id: &str,
+        target_spec: &str,
+        table_name: &str,
+    ) -> Result<Option<i64>> {
+        if !self.ledger_exists(state_ledger_id).await? {
+            return Ok(None);
+        }
+        let db = self.db(state_ledger_id).await?;
+
+        let subject = watermark_subject(source_graph_source_id, target_spec, table_name);
+        let mut where_obj = Map::new();
+        where_obj.insert("@id".to_string(), JsonValue::String(subject));
+        where_obj.insert(
+            WATERMARK_SEQUENCE_PRED.to_string(),
+            JsonValue::String("?v".to_string()),
+        );
+        let query = json!({ "select": ["?v"], "where": JsonValue::Object(where_obj) });
+
+        let result = self.query(&db, &query).await?;
+        let json = result.to_jsonld(&db.snapshot).map_err(|e| {
+            ApiError::Internal(format!("Failed to format sequence-cursor query: {e}"))
+        })?;
+        Ok(extract_first_i64(&json))
+    }
+
     /// Open the shared materialization-state ledger, creating it if absent.
     ///
     /// Tolerates losing the create race: two concurrent pollers (or a poller and
@@ -908,6 +987,7 @@ fn watermark_node(
     target_spec: &str,
     table_name: &str,
     to_snapshot_id: i64,
+    to_sequence: Option<i64>,
 ) -> JsonValue {
     let mut node = Map::new();
     node.insert(
@@ -918,10 +998,25 @@ fn watermark_node(
             table_name,
         )),
     );
-    node.insert(
-        WATERMARK_SNAPSHOT_PRED.to_string(),
-        JsonValue::String(to_snapshot_id.to_string()),
-    );
+    // A PARTIAL pass must not advance the snapshot. Writing `to` while the target
+    // holds only a prefix would claim rows that were never applied and the next
+    // scan would start after them — the applied-marker ratchet, which discarded
+    // ~80 % of a deployment's entities. So a prefix writes the resumable SEQUENCE
+    // instead, and the snapshot stays exactly where it was.
+    match to_sequence {
+        Some(seq) => {
+            node.insert(
+                WATERMARK_SEQUENCE_PRED.to_string(),
+                JsonValue::String(seq.to_string()),
+            );
+        }
+        None => {
+            node.insert(
+                WATERMARK_SNAPSHOT_PRED.to_string(),
+                JsonValue::String(to_snapshot_id.to_string()),
+            );
+        }
+    }
     node.insert(
         WATERMARK_SOURCE_PRED.to_string(),
         JsonValue::String(source_graph_source_id.to_string()),
@@ -2663,13 +2758,35 @@ mod tests {
         );
     }
 
+    /// A PARTIAL pass must write the sequence cursor and must NOT advance the
+    /// snapshot. Advancing it while the target holds only a prefix claims rows that
+    /// were never applied and makes the next scan start after them — the
+    /// applied-marker ratchet that discarded ~80 % of a deployment's entities. This
+    /// is the assertion that stops that from being reintroduced.
+    #[test]
+    fn a_partial_pass_writes_the_cursor_and_never_advances_the_snapshot() {
+        let node = watermark_node(
+            "people:main",
+            "silver:main",
+            "demo.actors",
+            5_648_190_075_564_901_028,
+            Some(4_242),
+        );
+        assert_eq!(node[WATERMARK_SEQUENCE_PRED], json!("4242"));
+        assert!(
+            node.get(WATERMARK_SNAPSHOT_PRED).is_none(),
+            "a prefix pass must not advance the snapshot watermark"
+        );
+    }
+
     #[test]
     fn watermark_node_is_per_job_and_string_encoded() {
         let node = watermark_node(
             "people:main",
             "silver:main",
             "demo.actors",
             5_648_190_075_564_901_028,
+            None,
         );
         // Every segment's ':' is escaped (%3A) so the (source, target, table)
         // encoding is injective.
@@ -2680,6 +2797,8 @@ mod tests {
         // String-encoded to preserve full i64 precision.
         assert_eq!(node[WATERMARK_SNAPSHOT_PRED], json!("5648190075564901028"));
         assert_eq!(node[WATERMARK_SOURCE_PRED], json!("people:main"));
+        // A COMPLETE pass writes no sequence cursor.
+        assert!(node.get(WATERMARK_SEQUENCE_PRED).is_none());
         assert_eq!(node[WATERMARK_TARGET_PRED], json!("silver:main"));
         assert_eq!(node[WATERMARK_TABLE_PRED], json!("demo.actors"));
     }
@@ -2962,6 +3081,7 @@ mod engine_tests {
         /// `None` means "first run / watermark unresolvable", which C3 treats as
         /// persist-regardless. Default to a FRESH window so tests opt in to staleness.
         window_age_ms: Option<i64>,
+        to_sequence: Option<i64>,
         scans: std::sync::atomic::AtomicUsize,
     }
 
@@ -2974,6 +3094,7 @@ mod engine_tests {
                 order_by: None,
                 to_snapshot_id: Some(7),
                 window_age_ms: Some(0),
+                to_sequence: None,
                 scans: std::sync::atomic::AtomicUsize::new(0),
             }
         }
@@ -2998,13 +3119,15 @@ mod engine_tests {
             _gs: &str,
             _table: &str,
             _from: Option<i64>,
+            _from_sequence: Option<i64>,
         ) -> Result<MaterializeScan> {
             self.scans.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
             let taken: Vec<ColumnBatch> = std::mem::take(&mut *self.batches.lock().unwrap());
             Ok(MaterializeScan {
                 to_snapshot_id: self.to_snapshot_id,
                 incremental: false,
                 window_age_ms: self.window_age_ms,
+                to_sequence: self.to_sequence,
                 stream: Box::pin(futures::stream::iter(taken.into_iter().map(Ok))),
             })
         }

fluree-db-api/tests/it_iceberg_local_fs.rs

@@ -134,7 +134,7 @@ async fn local_table_end_to_end() {
 
     // Full unpinned streaming read: all 5 rows.
     let scan = provider
-        .scan_for_materialize_stream(gs, "silver.people", &[], None, None)
+        .scan_for_materialize_stream(gs, "silver.people", &[], None, None, None)
         .await
         .expect("full scan");
     assert_eq!(scan.to_snapshot_id, Some(current));
@@ -170,7 +170,7 @@ async fn local_table_end_to_end() {
     };
 
     let scan = provider
-        .scan_for_materialize_stream(gs, "silver.people", &[], Some(first), None)
+        .scan_for_materialize_stream(gs, "silver.people", &[], Some(first), None, None)
         .await
         .expect("incremental scan");
     assert!(scan.incremental, "append-only window scans incrementally");
@@ -184,7 +184,7 @@ async fn local_table_end_to_end() {
     // PINNED read: to = the FIRST snapshot → only the first append's rows,
     // and the resolved watermark is the pin, not current.
     let scan = provider
-        .scan_for_materialize_stream(gs, "silver.people", &[], None, Some(first))
+        .scan_for_materialize_stream(gs, "silver.people", &[], None, Some(first), None)
         .await
         .expect("pinned scan");
     assert_eq!(scan.to_snapshot_id, Some(first), "pin is honored");
@@ -198,7 +198,7 @@ async fn local_table_end_to_end() {
     // An expired/unknown pin is the typed error, never a fall-forward.
     // (`MaterializeScan` has no Debug — a stream field — so match manually.)
     match provider
-        .scan_for_materialize_stream(gs, "silver.people", &[], None, Some(999))
+        .scan_for_materialize_stream(gs, "silver.people", &[], None, Some(999), None)
         .await
     {
         Ok(_) => panic!("unknown pin must fail, not fall forward"),

fluree-db-api/tests/it_iceberg_warehouse_root.rs

@@ -107,7 +107,7 @@ async fn warehouse_root_resolves_at_every_direct_entry_point() {
         .expect("fixture table has snapshots");
 
     let scan = provider
-        .scan_for_materialize_stream(gs, "silver.people", &[], None, None)
+        .scan_for_materialize_stream(gs, "silver.people", &[], None, None, None)
         .await
         .expect("scan_for_materialize_stream must resolve the table under the root");
     assert_eq!(
@@ -121,7 +121,7 @@ async fn warehouse_root_resolves_at_every_direct_entry_point() {
 
     // A pinned scan takes the same resolution path with a caller-supplied `to`.
     let scan = provider
-        .scan_for_materialize_stream(gs, "silver.people", &[], None, Some(current))
+        .scan_for_materialize_stream(gs, "silver.people", &[], None, Some(current), None)
         .await
         .expect("pinned scan under a warehouse root");
     assert_eq!(scan.to_snapshot_id, Some(current), "pin is honored");