From c24180c848d86ced9f11b5380b56bbb5a071921f Mon Sep 17 00:00:00 2001 From: Christophe Di Prima Date: Thu, 13 Aug 2026 15:28:10 +0200 Subject: [PATCH 1/2] feat(materialize): advance a backlogged window by a bounded prefix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The materialize watermark only advances when a whole pass succeeds. A pass that cannot finish its window writes no watermark, so the next pass re-reads a window one poll wider, and so on without bound. That is not merely slow, it is a one-way door. Once the window outgrows the source table's snapshot retention, the stored watermark can no longer be resolved, so every later poll falls back to a full table read — which is far more expensive and therefore even less likely to finish. The entry condition for the incremental path is exactly what is lost, so nothing recovers it. Measured on a 17-table deployment: one affected table became 13 of 17 in about four hours. 126 consecutive polls, every one "partial window, remainder deferred", zero completions, and every affected table reading its whole self on every poll. Resident memory rose 19.5 -> 24.7 GiB from the full reads alone. So bound the pass. `TableMetadata::window_end_capped` takes the OLDEST prefix of `(from, head]` — at most `FLUREE_MATERIALIZE_MAX_SNAPSHOTS_PER_PASS` snapshots, default 64 — and an unpinned materialize scan reads to there instead of to head. Three properties make this safe rather than merely smaller: - capping only ever moves `to` EARLIER, so no snapshot is skipped; the next poll resumes from the watermark this one wrote. - an initial full read (`from = None`) is never capped. A partial "full" read would be worse than an unbounded one. - any window error — expired ancestor, non-ancestor, rollback — falls through to head unchanged, leaving the existing full-read fallback to own that case. 64 is deliberately generous. A source committing 37-72 snapshots an hour stays under it at any sane poll interval, so a healthy job never reaches the cap; only a backlogged one does, and it then drains in bounded steps instead of never. `0` disables it and restores the previous behaviour exactly. The test asserts the DIRECTION explicitly (oldest prefix, not newest): taking the newest would skip everything between the watermark and the chosen end, which is silent data loss rather than a performance bug. --- fluree-db-api/src/graph_source/r2rml.rs | 73 ++++++++++++++++++- fluree-db-iceberg/src/metadata/table.rs | 93 +++++++++++++++++++++++++ 2 files changed, 165 insertions(+), 1 deletion(-) diff --git a/fluree-db-api/src/graph_source/r2rml.rs b/fluree-db-api/src/graph_source/r2rml.rs index 24461d9486..4f10d138ab 100644 --- a/fluree-db-api/src/graph_source/r2rml.rs +++ b/fluree-db-api/src/graph_source/r2rml.rs @@ -244,6 +244,38 @@ impl ScanChoice { } } +/// How many source snapshots one unpinned materialize pass may advance through. +/// +/// `FLUREE_MATERIALIZE_MAX_SNAPSHOTS_PER_PASS`, default 64; `0` disables the cap +/// and restores the previous "always read to head" behaviour. +/// +/// **Why a cap exists at all.** The materialize watermark only advances when a +/// whole pass succeeds. A pass that cannot finish its window writes no +/// watermark, so the next pass re-reads a window one poll wider, and so on. Once +/// that window outgrows the source's snapshot retention the stored watermark can +/// no longer be resolved, and every subsequent poll falls back to a full table +/// read — which is far more expensive, so it is even less likely to finish. The +/// failure is self-reinforcing and has no exit: the entry condition for the +/// incremental path is exactly what was lost. +/// +/// Observed on a 17-table deployment: one affected table became 13 of 17 in +/// about four hours, with every table then reading its whole self on every poll. +/// +/// A cap bounds the work per pass, so the watermark advances every time and +/// therefore stays inside retention. 64 is deliberately generous — a source +/// committing 37-72 snapshots an hour stays under it at any sane poll interval, +/// so a healthy job never notices the cap, while a backlogged one drains in +/// bounded steps instead of never. +fn materialize_max_snapshots_per_pass() -> usize { + static CACHED: std::sync::OnceLock = std::sync::OnceLock::new(); + *CACHED.get_or_init(|| { + std::env::var("FLUREE_MATERIALIZE_MAX_SNAPSHOTS_PER_PASS") + .ok() + .and_then(|v| v.trim().parse::().ok()) + .unwrap_or(64) + }) +} + /// Field ids to project for `projection` against `schema`: every non-nested /// field when `projection` is empty, else the named columns that exist in the /// schema (unknown names are skipped — the consumer treats them as absent). @@ -1266,7 +1298,46 @@ impl<'a> FlureeR2rmlProvider<'a> { })?, ) } - None => metadata.current_snapshot(), + // Unpinned: read to the source's head, but advance through a long + // backlog in bounded steps rather than in one unbounded pass. + // + // A materialization whose watermark cannot advance re-reads a window + // that grows without bound, and once that window outgrows the source's + // snapshot retention the watermark can never be resolved again — every + // later poll degrades to a full table read, with no path back. Capping + // the pass keeps the watermark moving, which keeps it inside retention. + // + // Capping only ever moves `to` EARLIER, so it cannot skip a snapshot: + // the next poll resumes from the watermark this pass wrote. On any + // error (expired ancestor, non-ancestor, rollback) fall through to the + // head unchanged — the existing full-read fallback owns that case and + // must keep owning it. + None => { + let head = metadata.current_snapshot(); + let cap = materialize_max_snapshots_per_pass(); + match head { + Some(head) if cap > 0 => { + let bounded = metadata + .window_end_capped(from_snapshot_id, head.snapshot_id, cap) + .ok() + .filter(|id| *id != head.snapshot_id) + .and_then(|id| metadata.snapshot(id)); + if let Some(b) = bounded { + info!( + graph_source_id = %graph_source_id, + table = %table_name, + from_snapshot_id = ?from_snapshot_id, + head_snapshot_id = head.snapshot_id, + bounded_to_snapshot_id = b.snapshot_id, + max_snapshots_per_pass = cap, + "materialize: backlog exceeds the per-pass cap, advancing by a prefix" + ); + } + bounded.or(Some(head)) + } + other => other, + } + } }; let Some(to_snapshot) = to_snapshot else { // Table has no snapshots: nothing to materialize. diff --git a/fluree-db-iceberg/src/metadata/table.rs b/fluree-db-iceberg/src/metadata/table.rs index e6ac648a3f..e3a277b487 100644 --- a/fluree-db-iceberg/src/metadata/table.rs +++ b/fluree-db-iceberg/src/metadata/table.rs @@ -170,6 +170,44 @@ impl TableMetadata { } } + /// The end of a window `(from_id, to_id]` capped to at most `max_snapshots` + /// snapshots, so a consumer can advance through a long backlog in bounded + /// steps instead of one unbounded pass. + /// + /// Returns `to_id` unchanged when the window already fits, when `from_id` is + /// `None` (an initial full read has no prefix to take), or when + /// `max_snapshots` is `0` (disabled). + /// + /// **Why a consumer wants this.** A materialization whose watermark cannot + /// advance re-reads a window that grows without bound, and once that window + /// outgrows the source's snapshot retention the watermark can never be + /// resolved again — every later poll degrades to a full table read, forever. + /// Advancing by a bounded prefix keeps the watermark moving, which keeps it + /// inside retention, which is what makes the incremental path recoverable + /// rather than a one-way door. + /// + /// Errors propagate from [`Self::snapshot_window`] (unknown/expired/ + /// non-ancestor), where the caller must already fall back to a full re-read. + pub fn window_end_capped( + &self, + from_id: Option, + to_id: i64, + max_snapshots: usize, + ) -> crate::error::Result { + if max_snapshots == 0 || from_id.is_none() { + return Ok(to_id); + } + let window = self.snapshot_window(from_id, to_id)?; + if window.len() <= max_snapshots { + return Ok(to_id); + } + // `snapshot_window` is NEWEST-first, and we want the OLDEST + // `max_snapshots` of them — the prefix adjacent to `from_id`. Counting + // that many back from the old end lands on the last snapshot of the + // prefix, which becomes this pass's `to`. + Ok(window[window.len() - max_snapshots].snapshot_id) + } + /// Whether every snapshot in `(from_id, to_id]` was created by an `append` /// operation. Only then does an added-files incremental scan capture all /// changes (no `overwrite`/`delete`/`replace` => no updates or deletions to @@ -551,6 +589,61 @@ mod tests { assert!(m.snapshot_window(Some(1), 42).is_err()); } + /// A backlog longer than the cap is advanced in bounded steps, taking the + /// OLDEST snapshots first. Taking the newest instead would skip everything + /// between `from` and the chosen end — silent data loss, and the reason the + /// direction is asserted rather than assumed. + #[test] + fn window_end_capped_takes_the_oldest_prefix() { + // 1 <- 2 <- 3 <- 4 <- 5, watermark at 1, so the window is (1, 5] = 4 wide. + let meta = meta_with(vec![ + snap(1, None, 1, Some("append")), + snap(2, Some(1), 2, Some("append")), + snap(3, Some(2), 3, Some("append")), + snap(4, Some(3), 4, Some("append")), + snap(5, Some(4), 5, Some("append")), + ]); + + // Cap 2 -> advance to snapshot 3, NOT 5: the two oldest after the + // watermark. Snapshot 2 must not be skipped. + assert_eq!(meta.window_end_capped(Some(1), 5, 2).unwrap(), 3); + // Cap 1 -> one step at a time. + assert_eq!(meta.window_end_capped(Some(1), 5, 1).unwrap(), 2); + // Successive passes converge on the head rather than stalling short of it. + assert_eq!(meta.window_end_capped(Some(3), 5, 2).unwrap(), 5); + } + + #[test] + fn window_end_capped_is_a_no_op_when_it_cannot_help() { + let meta = meta_with(vec![ + snap(1, None, 1, Some("append")), + snap(2, Some(1), 2, Some("append")), + snap(3, Some(2), 3, Some("append")), + ]); + + // Window already fits. + assert_eq!(meta.window_end_capped(Some(1), 3, 5).unwrap(), 3); + // Disabled. + assert_eq!(meta.window_end_capped(Some(1), 3, 0).unwrap(), 3); + // An initial full read has no prefix to take — capping it would produce a + // partial "full" read, which is worse than the unbounded one. + assert_eq!(meta.window_end_capped(None, 3, 1).unwrap(), 3); + // from == to: empty window, nothing to cap. + assert_eq!(meta.window_end_capped(Some(3), 3, 1).unwrap(), 3); + } + + /// An expired ancestor must still ERROR rather than silently capping to + /// something arbitrary — the caller falls back to a full re-read, and that + /// decision has to stay with the caller. + #[test] + fn window_end_capped_propagates_an_expired_ancestor() { + let meta = meta_with(vec![ + snap(3, Some(2), 3, Some("append")), + snap(4, Some(3), 4, Some("append")), + ]); + assert!(meta.window_end_capped(Some(1), 4, 2).is_err()); + } + #[test] fn window_is_append_only_detects_non_append() { let all_append = meta_with(vec![ From 06a96d6e2389de893554ef971c744b0a6be2b6b0 Mon Sep 17 00:00:00 2001 From: Christophe Di Prima Date: Thu, 13 Aug 2026 16:00:37 +0200 Subject: [PATCH 2/2] test(materialize): make the per-pass cap decision testable without a backend MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The cap had three unit tests on `window_end_capped`, but the DECISION — head or prefix, and every way of declining — lived inline in `scan_for_materialize_stream` and had none. Testing it there needs a storage backend and an Iceberg fixture, which in this repo means `aws-testcontainers`; that is a high price for asserting a branch, and the usual result is that the branch stays unasserted. So move the decision into `TableMetadata::capped_scan_end`, where the metadata fixtures already exist. The scan path now calls it and uses the answer, which leaves the caller with no logic to get wrong and puts every decline path under test: - no snapshots at all, no panic - `max_snapshots == 0`, the disable switch - `from_id = None`, an initial full read - backlog already fits - the window cannot be walked (expired ancestor, non-ancestor, rollback) Two properties are asserted rather than assumed, because both are silent when wrong: the prefix is the OLDEST snapshots (taking the newest would skip everything between the watermark and the chosen end — data loss, not slowness), and the result never exceeds the head. Mutation-tested, each killed: taking the newest prefix, capping an initial full read, and dropping the fall-back-to-head on a window error. --- fluree-db-api/src/graph_source/r2rml.rs | 35 +++++------ fluree-db-iceberg/src/metadata/table.rs | 82 +++++++++++++++++++++++++ 2 files changed, 96 insertions(+), 21 deletions(-) diff --git a/fluree-db-api/src/graph_source/r2rml.rs b/fluree-db-api/src/graph_source/r2rml.rs index 4f10d138ab..cb46a4dd9a 100644 --- a/fluree-db-api/src/graph_source/r2rml.rs +++ b/fluree-db-api/src/graph_source/r2rml.rs @@ -1313,30 +1313,23 @@ impl<'a> FlureeR2rmlProvider<'a> { // head unchanged — the existing full-read fallback owns that case and // must keep owning it. None => { - let head = metadata.current_snapshot(); let cap = materialize_max_snapshots_per_pass(); - match head { - Some(head) if cap > 0 => { - let bounded = metadata - .window_end_capped(from_snapshot_id, head.snapshot_id, cap) - .ok() - .filter(|id| *id != head.snapshot_id) - .and_then(|id| metadata.snapshot(id)); - if let Some(b) = bounded { - info!( - graph_source_id = %graph_source_id, - table = %table_name, - from_snapshot_id = ?from_snapshot_id, - head_snapshot_id = head.snapshot_id, - bounded_to_snapshot_id = b.snapshot_id, - max_snapshots_per_pass = cap, - "materialize: backlog exceeds the per-pass cap, advancing by a prefix" - ); - } - bounded.or(Some(head)) + let chosen = metadata.capped_scan_end(from_snapshot_id, cap); + // Log only when the cap actually bit, so a healthy job stays quiet. + if let (Some(chosen), Some(head)) = (chosen, metadata.current_snapshot()) { + if chosen.snapshot_id != head.snapshot_id { + info!( + graph_source_id = %graph_source_id, + table = %table_name, + from_snapshot_id = ?from_snapshot_id, + head_snapshot_id = head.snapshot_id, + bounded_to_snapshot_id = chosen.snapshot_id, + max_snapshots_per_pass = cap, + "materialize: backlog exceeds the per-pass cap, advancing by a prefix" + ); } - other => other, } + chosen } }; let Some(to_snapshot) = to_snapshot else { diff --git a/fluree-db-iceberg/src/metadata/table.rs b/fluree-db-iceberg/src/metadata/table.rs index e3a277b487..44694873d7 100644 --- a/fluree-db-iceberg/src/metadata/table.rs +++ b/fluree-db-iceberg/src/metadata/table.rs @@ -208,6 +208,37 @@ impl TableMetadata { Ok(window[window.len() - max_snapshots].snapshot_id) } + /// Where an unpinned incremental consumer should end this pass: the head, + /// or an earlier snapshot when the backlog from `from_id` exceeds + /// `max_snapshots`. + /// + /// This is the whole decision in one place, so it is testable without a + /// storage backend — the caller does nothing but use the answer. Every way + /// of declining to cap returns the head unchanged: + /// + /// - no snapshots at all (`None`; there is nothing to read); + /// - `max_snapshots == 0`, the disable switch; + /// - `from_id` is `None` — an initial full read has no prefix to take, and a + /// partial "full" read would be worse than an unbounded one; + /// - the backlog already fits; + /// - the window cannot be walked (expired ancestor, non-ancestor, rollback). + /// That is the existing full-read fallback's case and it must keep it. + /// + /// Capping only ever returns an EARLIER snapshot than the head, never a + /// later one, so a consumer that records where it stopped cannot skip data. + pub fn capped_scan_end( + &self, + from_id: Option, + max_snapshots: usize, + ) -> Option<&super::Snapshot> { + let head = self.current_snapshot()?; + self.window_end_capped(from_id, head.snapshot_id, max_snapshots) + .ok() + .filter(|id| *id != head.snapshot_id) + .and_then(|id| self.snapshot(id)) + .or(Some(head)) + } + /// Whether every snapshot in `(from_id, to_id]` was created by an `append` /// operation. Only then does an added-files incremental scan capture all /// changes (no `overwrite`/`delete`/`replace` => no updates or deletions to @@ -632,6 +663,57 @@ mod tests { assert_eq!(meta.window_end_capped(Some(3), 3, 1).unwrap(), 3); } + /// `capped_scan_end` is the decision a scan actually makes, so every way of + /// declining to cap is pinned here — a consumer calls this and nothing else. + #[test] + fn capped_scan_end_covers_every_decline_path() { + let meta = meta_with(vec![ + snap(1, None, 1, Some("append")), + snap(2, Some(1), 2, Some("append")), + snap(3, Some(2), 3, Some("append")), + snap(4, Some(3), 4, Some("append")), + ]); + let head = 4; + + // Caps when the backlog exceeds the limit. + assert_eq!(meta.capped_scan_end(Some(1), 2).unwrap().snapshot_id, 3); + + // Declines: disabled, initial full read, backlog already fits. + assert_eq!(meta.capped_scan_end(Some(1), 0).unwrap().snapshot_id, head); + assert_eq!(meta.capped_scan_end(None, 1).unwrap().snapshot_id, head); + assert_eq!(meta.capped_scan_end(Some(1), 99).unwrap().snapshot_id, head); + + // Declines: the window cannot be walked. `from` is not an ancestor here, + // which is the rollback/branch case — the full-read fallback owns it, so + // this must return the head rather than inventing a bound. + let orphan = meta_with(vec![ + snap(7, None, 7, Some("append")), + snap(8, Some(7), 8, Some("append")), + ]); + assert_eq!(orphan.capped_scan_end(Some(1), 1).unwrap().snapshot_id, 8); + + // No snapshots: nothing to read, and no panic. + assert!(meta_with(vec![]).capped_scan_end(Some(1), 1).is_none()); + } + + /// The cap must never hand back a snapshot NEWER than the head — that would + /// read past what the caller asked for. + #[test] + fn capped_scan_end_never_exceeds_the_head() { + let meta = meta_with(vec![ + snap(1, None, 1, Some("append")), + snap(2, Some(1), 2, Some("append")), + snap(3, Some(2), 3, Some("append")), + ]); + for cap in 0..6 { + let chosen = meta.capped_scan_end(Some(1), cap).unwrap().sequence_number; + assert!( + chosen <= 3, + "cap {cap} chose sequence {chosen}, past the head" + ); + } + } + /// An expired ancestor must still ERROR rather than silently capping to /// something arbitrary — the caller falls back to a full re-read, and that /// decision has to stay with the caller.