From 4b334e3502b5693fc2761d7251da8b869a653755 Mon Sep 17 00:00:00 2001 From: Christophe Di Prima Date: Thu, 13 Aug 2026 16:20:46 +0200 Subject: [PATCH 1/6] feat(iceberg): carry each file's commit sequence, and order scans by it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Groundwork for resumable materialization. A consumer that stops part-way through a backlog and resumes needs two things this planner did not provide: to know WHEN each file's rows entered the table, and for the file order to be a contract rather than an accident. `FileScanTask` gains `data_sequence_number`. Iceberg assigns it per commit and never reuses or reorders it, so it is the only value on a task that says when the rows arrived — paths are arbitrary. Both planners already compute it to decide what to select and then dropped it; they now attach it. `Option`, not a sentinel: a task the planner cannot attribute is "unknown", and a consumer must not read that as "oldest". Both plans are then sorted by `(data_sequence_number, file_path)`. Manifest traversal is sequential and manifest files are immutable, so for a pinned snapshot the order was already stable — but only incidentally. Anything that later reads manifests concurrently would silently change it, and a resumed scan against a different order either repeats files or skips them. Sorting makes the guarantee explicit at the cost of a sort over file metadata. Sequence number first, path only as a tiebreak within one commit, and that direction is the point rather than a style choice. Materialization decides latest-by-key from an ordering column in the data, not from scan order, so within a single pass file order does not matter. ACROSS a resume boundary it does: if a key's newer row is processed in one pass and its older row in the next, the older one lands last and wins. Processing commits in order is what prevents that, and sorting by path alone would not. The full-scan path carries it too. That is the read a consumer falls back to when its watermark has expired — precisely the read that most needs to become resumable. No behaviour change: nothing consumes the field yet, and the sort only makes the existing order explicit. --- fluree-db-iceberg/src/scan/planner.rs | 99 ++++++++++++++++++++++ fluree-db-iceberg/src/scan/send_planner.rs | 53 ++++++++++-- 2 files changed, 145 insertions(+), 7 deletions(-) diff --git a/fluree-db-iceberg/src/scan/planner.rs b/fluree-db-iceberg/src/scan/planner.rs index d0bedd7161..dba0c652e6 100644 --- a/fluree-db-iceberg/src/scan/planner.rs +++ b/fluree-db-iceberg/src/scan/planner.rs @@ -86,9 +86,29 @@ pub struct FileScanTask { pub length: i64, /// Iceberg schema for field ID mapping (ensures correct column mapping after schema evolution). pub iceberg_schema: Option>, + /// Effective data sequence number of the file, when the planner knew it. + /// + /// Iceberg assigns this per commit and never reuses or reorders it, so it is + /// the only value on a task that says *when* the file's rows entered the + /// table. A consumer that must process files in commit order — or stop + /// part-way through a backlog and resume without going backwards — has + /// nothing else to sort or checkpoint on: paths are arbitrary, and manifest + /// iteration order is an implementation detail rather than a contract. + /// + /// `None` where the planner cannot attribute one (a whole-table read built + /// outside manifest traversal, and the test helpers). Consumers that need + /// ordering must treat `None` as "unknown", not as zero. + pub data_sequence_number: Option, } impl FileScanTask { + /// Attach the file's effective data sequence number. + #[must_use] + pub fn with_data_sequence_number(mut self, seq: i64) -> Self { + self.data_sequence_number = Some(seq); + self + } + /// Create a task for reading an entire file. pub fn for_whole_file( data_file: DataFile, @@ -103,6 +123,7 @@ impl FileScanTask { start: 0, length, iceberg_schema: None, + data_sequence_number: None, } } @@ -121,6 +142,7 @@ impl FileScanTask { start: 0, length, iceberg_schema: Some(schema), + data_sequence_number: None, } } } @@ -385,6 +407,83 @@ impl<'a, S: IcebergStorage> ScanPlanner<'a, S> { mod tests { use super::*; + /// The sort a resumable consumer depends on: commit order first, path as a + /// deterministic tiebreak inside one commit. + /// + /// This is asserted on the comparator rather than on a planner run because + /// the planners need storage; the ordering rule is the part that has to + /// hold, and it is what a cursor checkpoints against. + #[test] + fn tasks_order_by_commit_then_path() { + fn task(path: &str, seq: Option) -> FileScanTask { + let df = crate::manifest::DataFile { + file_path: path.to_string(), + file_format: crate::manifest::FileFormat::Parquet, + record_count: 1, + file_size_in_bytes: 1, + partition: crate::manifest::PartitionData::default(), + column_sizes: None, + value_counts: None, + null_value_counts: None, + nan_value_counts: None, + lower_bounds: None, + upper_bounds: None, + split_offsets: None, + sort_order_id: None, + }; + let t = FileScanTask::for_whole_file(df, vec![], None); + match seq { + Some(s) => t.with_data_sequence_number(s), + None => t, + } + } + + let mut tasks = [ + task("z-early.parquet", Some(1)), + task("a-late.parquet", Some(9)), + task("m-mid-b.parquet", Some(5)), + task("m-mid-a.parquet", Some(5)), + ]; + tasks.sort_by(|a, b| { + a.data_sequence_number + .cmp(&b.data_sequence_number) + .then_with(|| a.data_file.file_path.cmp(&b.data_file.file_path)) + }); + + let order: Vec<&str> = tasks + .iter() + .map(|t| t.data_file.file_path.as_str()) + .collect(); + assert_eq!( + order, + vec![ + "z-early.parquet", // seq 1 — earliest commit, despite sorting last by path + "m-mid-a.parquet", // seq 5, path tiebreak + "m-mid-b.parquet", + "a-late.parquet", // seq 9 — latest commit, despite sorting first by path + ], + "commit order must outrank path order: sorting by path alone would \ + process a later commit before an earlier one, which lets an older \ + row overwrite a newer one across a resume boundary" + ); + } + + /// `None` must not be treated as sequence zero. An unattributed file sorting + /// first would place it before every real commit. + #[test] + fn an_unknown_sequence_number_is_distinct_from_zero() { + assert_ne!( + Some(0i64), + Option::::None, + "a task with no attributable sequence number is 'unknown', not 'oldest'" + ); + // And the helper never yields None: a missing entry seq inherits the + // manifest's, which is what makes the sort total in practice. + assert_eq!(effective_sequence_number(None, 7), 7); + assert_eq!(effective_sequence_number(Some(0), 7), 7); + assert_eq!(effective_sequence_number(Some(3), 7), 3); + } + #[test] fn test_scan_config_builder() { let config = ScanConfig::new() diff --git a/fluree-db-iceberg/src/scan/send_planner.rs b/fluree-db-iceberg/src/scan/send_planner.rs index 55f6388d6c..bfd3bac646 100644 --- a/fluree-db-iceberg/src/scan/send_planner.rs +++ b/fluree-db-iceberg/src/scan/send_planner.rs @@ -153,16 +153,32 @@ impl<'a, S: SendIcebergStorage> SendScanPlanner<'a, S> { estimated_row_count += data_file.record_count; // Create file scan task with schema for correct field ID mapping + let eff_seq = effective_sequence_number( + entry.sequence_number, + manifest_entry.sequence_number, + ); let task = FileScanTask::for_whole_file_with_schema( data_file, projected_field_ids.clone(), self.config.filter.clone(), Arc::clone(&schema_arc), - ); + ) + // A full read carries it too. This is the read a consumer falls + // back to when its watermark expired, so it is exactly the read + // that most needs to be resumable. + .with_data_sequence_number(eff_seq); tasks.push(task); } } + // Commit order, path as the tiebreak — see the same sort in + // `plan_incremental` for why this is a contract rather than a nicety. + tasks.sort_by(|a, b| { + a.data_sequence_number + .cmp(&b.data_sequence_number) + .then_with(|| a.data_file.file_path.cmp(&b.data_file.file_path)) + }); + tracing::info!( files_selected, files_pruned, @@ -317,15 +333,38 @@ impl<'a, S: SendIcebergStorage> SendScanPlanner<'a, S> { files_selected += 1; estimated_row_count += data_file.record_count; - added_tasks.push(FileScanTask::for_whole_file_with_schema( - data_file, - projected_field_ids.clone(), - self.config.filter.clone(), - Arc::clone(&schema_arc), - )); + added_tasks.push( + FileScanTask::for_whole_file_with_schema( + data_file, + projected_field_ids.clone(), + self.config.filter.clone(), + Arc::clone(&schema_arc), + ) + // `eff_seq` is already the value this loop filters on; carrying + // it lets a consumer order or checkpoint by commit without + // re-deriving it from manifests it no longer holds. + .with_data_sequence_number(eff_seq), + ); } } + // Commit order, with the path as a deterministic tiebreak inside a commit. + // + // Manifest traversal is already sequential, so this order is stable today + // — but only incidentally. A consumer that stops part-way through a + // backlog and resumes needs the order to be a CONTRACT, because resuming + // against a different order either repeats files or skips them. Sorting + // makes it one, and costs a sort over file metadata rather than data. + // + // Sequence number first, not path first: an older row overwriting a newer + // one is the failure mode that matters here, and processing commits in + // order is what prevents it. + added_tasks.sort_by(|a, b| { + a.data_sequence_number + .cmp(&b.data_sequence_number) + .then_with(|| a.data_file.file_path.cmp(&b.data_file.file_path)) + }); + tracing::info!( ?from_snapshot_id, to_snapshot_id, From 4eaec8f2783810b1cdf28b764052990aac2bec6b Mon Sep 17 00:00:00 2001 From: Christophe Di Prima Date: Fri, 14 Aug 2026 09:12:15 +0200 Subject: [PATCH 2/6] fix(materialize): bound a full read and checkpoint it at a commit boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A full read is the path a source lands on once its watermark has expired, and without a bound it is a trap rather than merely slow. It is the most expensive read the table has, so it is the most likely to exhaust the target's novelty and defer; a deferral writes no watermark; so the next poll performs the same full read. Every later poll is then the same read, forever, and nothing recovers it. Measured: one deployment had 13 of 17 tables in exactly this state, each reading its whole self on every poll, with no watermark moving for hours. A full read cannot be bounded by snapshot — there is no `from` to take a prefix after — so bound it by rows and checkpoint at the snapshot that prefix corresponds to. `FLUREE_MATERIALIZE_MAX_ROWS_PER_FULL_PASS`, default 250_000; `0` restores the previous read-it-all behaviour. The mechanism needs no new vocabulary and no schema change. Tasks now arrive sorted by `(data_sequence_number, path)`, so a row-budgeted prefix is a prefix in COMMIT order; `snapshot_at_or_before_sequence` names the snapshot that prefix leaves the target at; and reporting THAT as the scan's `to` makes the caller's existing watermark write land there. The crash-safety ordering is unchanged — the watermark still follows the data. Two rules decide correctness, and both are asserted rather than assumed: - **The cut extends to the end of its commit.** Stopping inside one leaves the target holding part of a commit, which no snapshot names, and a state that cannot be named cannot be resumed from. - **The checkpoint rounds DOWN**, to the newest ancestor whose sequence is fully applied. Rounding up would write a watermark claiming rows that were never applied — silent data loss, and the next scan would start after them. Every reason to decline returns "read it whole", which is today's behaviour: the bound is disabled, the plan already fits, the cut would land on the head anyway, or any task lacks an attributable sequence number. That last one matters — without a sequence there is no safe cut, and an expensive honest read beats a wrong checkpoint. This does not change what a bounded pass reads. A full read selects the files live at `to`; applying those at or below the cut is exactly the state at the checkpoint snapshot, and the next pass scans `(checkpoint, to]` for the rest. Three mutations, each killed: cutting inside a commit, rounding the checkpoint up, and letting an unattributed task through. --- fluree-db-api/src/graph_source/r2rml.rs | 185 +++++++++++++++++++++++- fluree-db-iceberg/src/metadata/table.rs | 53 +++++++ 2 files changed, 236 insertions(+), 2 deletions(-) diff --git a/fluree-db-api/src/graph_source/r2rml.rs b/fluree-db-api/src/graph_source/r2rml.rs index 24461d9486..ff4d1af6eb 100644 --- a/fluree-db-api/src/graph_source/r2rml.rs +++ b/fluree-db-api/src/graph_source/r2rml.rs @@ -244,6 +244,54 @@ impl ScanChoice { } } +/// Rows a single FULL materialize read may take before checkpointing. +/// +/// `FLUREE_MATERIALIZE_MAX_ROWS_PER_FULL_PASS`, default 250_000; `0` disables +/// the bound and restores the previous read-it-all behaviour. +fn materialize_max_rows_per_full_pass() -> i64 { + static CACHED: std::sync::OnceLock = std::sync::OnceLock::new(); + *CACHED.get_or_init(|| { + std::env::var("FLUREE_MATERIALIZE_MAX_ROWS_PER_FULL_PASS") + .ok() + .and_then(|v| v.trim().parse::().ok()) + .filter(|v| *v >= 0) + .unwrap_or(250_000) + }) +} + +/// The commit sequence to stop a full read at, or `None` to read it whole. +/// +/// Walks `tasks` (already sorted by `(data_sequence_number, path)`) accumulating +/// rows, and returns the sequence of the commit the budget ran out in. The whole +/// of that commit is kept: a cut INSIDE a commit leaves the target in a state no +/// snapshot names, and an unnameable state cannot be checkpointed. +/// +/// `None` when the bound is disabled, the plan already fits, any task lacks a +/// sequence number, or the budget runs out in the final commit — stopping there +/// would checkpoint at the head, which is the whole read with extra steps. +fn full_read_prefix(tasks: &[fluree_db_iceberg::scan::FileScanTask], max_rows: i64) -> Option { + if max_rows <= 0 { + return None; + } + let total: i64 = tasks.iter().map(|t| t.data_file.record_count).sum(); + if total <= max_rows { + return None; + } + let last_seq = tasks + .iter() + .filter_map(|t| t.data_sequence_number) + .next_back()?; + let mut rows = 0i64; + for t in tasks { + let seq = t.data_sequence_number?; + rows = rows.saturating_add(t.data_file.record_count); + if rows >= max_rows { + return (seq < last_seq).then_some(seq); + } + } + None +} + /// 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). @@ -1277,7 +1325,7 @@ impl<'a> FlureeR2rmlProvider<'a> { stream: empty_batch_stream(), }); }; - let to_snapshot_id = to_snapshot.snapshot_id; + let mut to_snapshot_id = to_snapshot.snapshot_id; // Schema AT the `to` snapshot (falls back to current when the snapshot // carries no schema-id) — identical to current for an unpinned read. @@ -1358,7 +1406,57 @@ impl<'a> FlureeR2rmlProvider<'a> { estimated_rows = plan.estimated_row_count, "materialize: full scan plan" ); - plan.tasks + + // A full read cannot be bounded by snapshot — there is no `from` to + // take a prefix after — so bound it by rows instead, and checkpoint + // at the snapshot that prefix corresponds to. + // + // This is the path a source lands on once its watermark has expired, + // and without a bound it is a trap: the read is the most expensive + // one the table has, so it is the most likely to exhaust the target's + // novelty and defer, which writes no watermark, which guarantees the + // same read next poll. Every later poll is then the same full read, + // forever. Bounding it means each pass finishes, writes a watermark, + // and the one after starts incremental. + // + // Tasks arrive sorted by `(data_sequence_number, path)`, so a prefix + // is a prefix in COMMIT order. The cut is then extended to the end of + // its commit: splitting inside one would leave the target holding + // part of a commit with no snapshot to name that state, and a + // checkpoint that cannot be named cannot be resumed from. + match full_read_prefix(&plan.tasks, materialize_max_rows_per_full_pass()) { + Some(cut_seq) => { + match metadata.snapshot_at_or_before_sequence(to_snapshot_id, cut_seq) { + Ok(Some(checkpoint)) if checkpoint.snapshot_id != to_snapshot_id => { + let kept: Vec<_> = plan + .tasks + .into_iter() + .filter(|t| t.data_sequence_number.is_some_and(|s| s <= cut_seq)) + .collect(); + info!( + to_snapshot_id, + checkpoint_snapshot_id = checkpoint.snapshot_id, + checkpoint_sequence = cut_seq, + files_this_pass = kept.len(), + files_total = plan.files_selected, + "materialize: full read bounded to a commit prefix; \ + the watermark will checkpoint short of the head" + ); + // Reporting the checkpoint as this scan's `to` is what + // makes the caller's existing watermark write land there — + // no new vocabulary, and the crash-safety ordering + // (watermark after data) is unchanged. + to_snapshot_id = checkpoint.snapshot_id; + kept + } + // No nameable checkpoint short of the head: read it whole, as + // before. Better an expensive honest read than a watermark + // pointing somewhere the data does not correspond to. + _ => plan.tasks, + } + } + None => plan.tasks, + } }; // How OLD is the window we are about to read? Measured from Iceberg's own @@ -4542,4 +4640,87 @@ mod tests { ])); assert!(!super::listing_is_single_table(&[])); } + + fn t(seq: i64, rows: i64) -> fluree_db_iceberg::scan::FileScanTask { + let df = fluree_db_iceberg::manifest::DataFile { + file_path: format!("f{seq}-{rows}.parquet"), + file_format: fluree_db_iceberg::manifest::FileFormat::Parquet, + record_count: rows, + file_size_in_bytes: rows, + partition: fluree_db_iceberg::manifest::PartitionData::default(), + column_sizes: None, + value_counts: None, + null_value_counts: None, + nan_value_counts: None, + lower_bounds: None, + upper_bounds: None, + split_offsets: None, + sort_order_id: None, + }; + fluree_db_iceberg::scan::FileScanTask::for_whole_file(df, vec![], None) + .with_data_sequence_number(seq) + } + + /// The cut lands on a COMMIT boundary, never inside one. A partial commit + /// leaves the target in a state no snapshot names, and an unnameable state + /// cannot be checkpointed or resumed from. + #[test] + fn full_read_prefix_cuts_on_a_commit_boundary() { + // commit 10: 60 rows, commit 20: 60, commit 30: 60 => 180 total + let tasks = vec![ + t(10, 30), + t(10, 30), + t(20, 30), + t(20, 30), + t(30, 30), + t(30, 30), + ]; + + // Budget runs out inside commit 20 -> keep all of 20, cut there. + assert_eq!(super::full_read_prefix(&tasks, 70), Some(20)); + // Budget runs out on the first file of commit 10 -> cut at 10. + assert_eq!(super::full_read_prefix(&tasks, 10), Some(10)); + } + + /// Every reason to decline to cut. Each returns `None`, meaning "read it + /// whole" — the pre-existing behaviour. + #[test] + fn full_read_prefix_declines_when_a_cut_would_not_help() { + let tasks = vec![t(10, 30), t(20, 30), t(30, 30)]; + + assert_eq!(super::full_read_prefix(&tasks, 0), None, "disabled"); + assert_eq!(super::full_read_prefix(&tasks, -1), None, "negative"); + assert_eq!(super::full_read_prefix(&tasks, 90), None, "already fits"); + assert_eq!( + super::full_read_prefix(&tasks, 1_000), + None, + "budget exceeds total" + ); + assert_eq!(super::full_read_prefix(&[], 10), None, "no tasks"); + + // The budget runs out in the FINAL commit. Checkpointing at the head is + // the whole read with extra bookkeeping, so decline. + assert_eq!( + super::full_read_prefix(&tasks, 85), + None, + "cut would be the head" + ); + + // A task with no attributable sequence cannot be ordered, so no cut is + // safe — better an expensive honest read than a wrong checkpoint. + let unattributed = vec![ + t(10, 30), + fluree_db_iceberg::scan::FileScanTask::for_whole_file( + tasks[0].data_file.clone(), + vec![], + None, + ), + t(30, 30), + ]; + assert_eq!( + super::full_read_prefix(&unattributed, 40), + None, + "unknown sequence" + ); + } } diff --git a/fluree-db-iceberg/src/metadata/table.rs b/fluree-db-iceberg/src/metadata/table.rs index e6ac648a3f..84c867980b 100644 --- a/fluree-db-iceberg/src/metadata/table.rs +++ b/fluree-db-iceberg/src/metadata/table.rs @@ -170,6 +170,31 @@ impl TableMetadata { } } + /// The newest ancestor of `to_id` whose sequence number is at or below + /// `seq` — the snapshot a consumer has reached once it has applied every + /// file up to and including that sequence. + /// + /// This is what lets a *partial* full read record its progress. A full read + /// selects the files live at `to_id`; applying only those with an effective + /// sequence at or below `seq` leaves the target holding exactly the rows + /// that had arrived by this snapshot, so the consumer can checkpoint here + /// and let the next pass scan `(this, to_id]`. + /// + /// Returns `None` when no ancestor qualifies — every retained snapshot is + /// newer than `seq` — in which case there is no safe checkpoint short of + /// the whole read, and the consumer must not invent one. + pub fn snapshot_at_or_before_sequence( + &self, + to_id: i64, + seq: i64, + ) -> crate::error::Result> { + // Newest-first, so the first match is the newest qualifying ancestor. + Ok(self + .snapshot_window(None, to_id)? + .into_iter() + .find(|s| s.sequence_number <= seq)) + } + /// 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 +576,34 @@ mod tests { assert!(m.snapshot_window(Some(1), 42).is_err()); } + /// A partial full read checkpoints at the newest ancestor whose sequence it + /// has fully applied — never at one it has only partly reached. + #[test] + fn snapshot_at_or_before_sequence_picks_the_newest_fully_applied() { + let meta = meta_with(vec![ + snap(1, None, 10, Some("append")), + snap(2, Some(1), 20, Some("append")), + snap(3, Some(2), 30, Some("append")), + ]); + + // Exactly on a boundary: that snapshot is fully applied. + let at = |seq| { + meta.snapshot_at_or_before_sequence(3, seq) + .unwrap() + .map(|s| s.snapshot_id) + }; + assert_eq!(at(20), Some(2)); + // Between boundaries: fall BACK to the last fully-applied one. Rounding + // up would checkpoint past rows that were never applied. + assert_eq!(at(25), Some(2)); + // At or past the head. + assert_eq!(at(30), Some(3)); + assert_eq!(at(99), Some(3)); + // Below every retained snapshot: no nameable checkpoint, so the caller + // must read the whole thing rather than invent one. + assert_eq!(at(5), None); + } + #[test] fn window_is_append_only_detects_non_append() { let all_append = meta_with(vec![ From af8269e64d009d9bfec0d4f176108c915d0bb3f5 Mon Sep 17 00:00:00 2001 From: Christophe Di Prima Date: Fri, 14 Aug 2026 11:41:29 +0200 Subject: [PATCH 3/6] fix(iceberg): let the checkpoint walk survive an expired ancestor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `snapshot_at_or_before_sequence` used `snapshot_window(None, to_id)`, which walks the whole history to the ROOT and errors on any missing ancestor. The tables that need a checkpoint are precisely the ones whose old snapshots have expired — that is why they are full-reading in the first place. So the helper errored on every one of them and the caller fell through to an unbounded read. The bound could never fire on the only tables it was written for, and it failed silently: every unit test passed, because every fixture had an intact chain. Walk parents from `to` and stop at the first qualifying snapshot instead. The walk is short — the cut lands near the start of the backlog, so the answer is a few hops from `to`, nowhere near the expired region. Reaching the root, or a parent that is gone, means there is no nameable checkpoint: that is `None`, and the caller reads the whole thing, which is correct rather than merely safe. The regression test uses a deliberately severed chain (5 <- 6 <- 7 retained, 4 expired) because that is the shape the original fixtures could not express. --- fluree-db-iceberg/src/metadata/table.rs | 65 +++++++++++++++++++++++-- 1 file changed, 60 insertions(+), 5 deletions(-) diff --git a/fluree-db-iceberg/src/metadata/table.rs b/fluree-db-iceberg/src/metadata/table.rs index 84c867980b..b3531d3bcc 100644 --- a/fluree-db-iceberg/src/metadata/table.rs +++ b/fluree-db-iceberg/src/metadata/table.rs @@ -188,11 +188,34 @@ impl TableMetadata { to_id: i64, seq: i64, ) -> crate::error::Result> { - // Newest-first, so the first match is the newest qualifying ancestor. - Ok(self - .snapshot_window(None, to_id)? - .into_iter() - .find(|s| s.sequence_number <= seq)) + // Walk parents from `to`, newest first, and stop at the first match. + // + // Deliberately NOT `snapshot_window(None, to_id)`: that walks the whole + // history to the ROOT and errors on any missing ancestor. The tables + // that most need a checkpoint are precisely the ones whose old + // snapshots have expired, so requiring an intact history would refuse + // exactly the case this exists for — and silently, by falling back to + // an unbounded read. + // + // The walk is short in practice: the cut lands near the start of the + // backlog, so the answer is usually a few hops from `to`. Hitting a + // missing ancestor before finding one means there is no nameable + // checkpoint, which is `None` — the caller then reads the whole thing, + // which is correct rather than merely safe. + let mut cur = self.snapshot(to_id).ok_or_else(|| { + crate::error::IcebergError::SnapshotNotFound(format!("snapshot {to_id} not found")) + })?; + loop { + if cur.sequence_number <= seq { + return Ok(Some(cur)); + } + match cur.parent_snapshot_id.and_then(|pid| self.snapshot(pid)) { + Some(parent) => cur = parent, + // Root reached, or the parent has been expired. Either way there + // is nothing older we can name. + None => return Ok(None), + } + } } /// Whether every snapshot in `(from_id, to_id]` was created by an `append` @@ -576,6 +599,38 @@ mod tests { assert!(m.snapshot_window(Some(1), 42).is_err()); } + /// The checkpoint walk must survive an EXPIRED ancestor, because the tables + /// that need a checkpoint are the ones whose history has been expiring. + /// + /// Using a full-history walk here was a real bug: it errored on these very + /// tables and the caller fell through to an unbounded read, so the bound + /// never engaged in production while every unit test passed. + #[test] + fn snapshot_at_or_before_sequence_survives_an_expired_ancestor() { + // 5 <- 6 <- 7 retained; 5's parent (4) has been expired away. + let meta = meta_with(vec![ + snap(5, Some(4), 50, Some("append")), + snap(6, Some(5), 60, Some("append")), + snap(7, Some(6), 70, Some("append")), + ]); + + // A checkpoint inside the retained range is found without ever touching + // the missing ancestor. + assert_eq!( + meta.snapshot_at_or_before_sequence(7, 60) + .unwrap() + .map(|s| s.snapshot_id), + Some(6) + ); + // Below everything retained: no nameable checkpoint, but NOT an error — + // the caller reads the whole thing. `is_none`, not `assert_eq!(.., None)`: + // `Snapshot` has no `PartialEq`. + assert!(meta + .snapshot_at_or_before_sequence(7, 10) + .unwrap() + .is_none()); + } + /// A partial full read checkpoints at the newest ancestor whose sequence it /// has fully applied — never at one it has only partly reached. #[test] From 82317e70979fffb9ee8dc0fb17e1f8ed192ced9f Mon Sep 17 00:00:00 2001 From: Christophe Di Prima Date: Tue, 18 Aug 2026 16:08:40 +0200 Subject: [PATCH 4/6] fix(materialize): size the full-read bound so the pass can actually commit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The row bound added earlier could not fire on the deployment it was written for, and when it declined it said nothing. Measured in production: its success line appeared 0 times across 16 consecutive passes while the same 727k-row table was re-read roughly once a minute, and finding out why needed a state-ledger query rather than a log read. Three defects, one per change here. 1. The budget was denominated in the wrong currency. A bounded pass is only worth anything if it can COMMIT: `at_max_novelty` is `novelty.size >= reindex_max_bytes`, a window over that ceiling is DEFERRED, and a deferral discards the window's progress — so the next poll re-reads the same rows and stops at the same wall. A flat 250_000 rows is ~27 MB of flakes, which against a deployment that has pinned the ceiling to 8 MiB never commits; the bound therefore changed how much was read and nothing about whether any of it landed. The budget is now DERIVED from the ceiling, read from the same `FLUREE_REINDEX_MAX_BYTES` that configures it so the two cannot drift, via a tunable `FLUREE_MATERIALIZE_FLAKE_BYTES_PER_ROW` (default 108, measured). `FLUREE_MATERIALIZE_MAX_ROWS_PER_FULL_PASS` still overrides, `0` still disables. Floored at 1_000 rows so a tiny ceiling still makes progress instead of a zero-row pass that reads nothing and checkpoints nowhere. 2. Running out of budget inside the HEAD commit declined outright, which is how a COMPACTED table livelocks: a compaction rewrites every file with one sequence number, so from then on the budget always lands in the head commit and the read is never bounded. It now falls back to the newest boundary strictly below the head — a smaller pass than asked for, and still forward progress, which is the only property that matters. The test that asserted the old `None` here was encoding the bug, so its expectation is flipped. 3. Every decline was silent. `full_read_prefix` now returns a typed `FullReadCut` and the caller logs which reason fired, warning on the two that mean an unbounded read (`NoSequence`, `SingleCommit`) and each checkpoint-walk failure, while staying quiet on `PlanFits` and `Disabled` so the log does not fill with noise on every small read. `rows_for_ceiling` is split out from the env plumbing so the arithmetic that has to be right is testable without process-global state; its test asserts `rows * bytes_per_row <= ceiling` directly. New tests cover the compaction shape and that the healthy outcomes are NOT alarming, so logging stays proportionate. Known limit, deliberately left: flake-bytes-per-row is a heuristic measured on one table, so a much wider row still overshoots. Stopping on accumulated bytes rather than a row count is the robust version and is a larger change. Verified: fmt clean, clippy clean under BOTH default and all features, `fluree-db-api --all-features --lib graph_source::r2rml` 83 pass / 0 fail. --- fluree-db-api/src/graph_source/r2rml.rs | 328 +++++++++++++++++++++--- 1 file changed, 292 insertions(+), 36 deletions(-) diff --git a/fluree-db-api/src/graph_source/r2rml.rs b/fluree-db-api/src/graph_source/r2rml.rs index ff4d1af6eb..47c2dcf40b 100644 --- a/fluree-db-api/src/graph_source/r2rml.rs +++ b/fluree-db-api/src/graph_source/r2rml.rs @@ -244,52 +244,188 @@ impl ScanChoice { } } +/// Bytes of flakes one materialized row costs, for sizing the full-read budget. +/// +/// `FLUREE_MATERIALIZE_FLAKE_BYTES_PER_ROW`, default 108 — measured on the +/// production table this bound was tuned against. It exists only to convert the +/// novelty ceiling into a row count. A deployment whose rows are much wider or +/// narrower should set THIS rather than overriding the row budget directly, so +/// the derivation below keeps tracking the ceiling instead of drifting from it. +fn flake_bytes_per_row() -> i64 { + static CACHED: std::sync::OnceLock = std::sync::OnceLock::new(); + *CACHED.get_or_init(|| { + std::env::var("FLUREE_MATERIALIZE_FLAKE_BYTES_PER_ROW") + .ok() + .and_then(|v| v.trim().parse::().ok()) + .filter(|v| *v > 0) + .unwrap_or(108) + }) +} + +/// The novelty ceiling this deployment commits against, in bytes. +/// +/// Read from `FLUREE_REINDEX_MAX_BYTES` — the SAME variable that configures +/// `IndexConfig::reindex_max_bytes`, which `at_max_novelty` compares novelty +/// against (`novelty.size >= reindex_max_bytes`) — so the budget below and the +/// wall it has to fit under cannot drift apart. +fn novelty_ceiling_bytes() -> i64 { + static CACHED: std::sync::OnceLock = std::sync::OnceLock::new(); + *CACHED.get_or_init(|| { + std::env::var("FLUREE_REINDEX_MAX_BYTES") + .ok() + .and_then(|v| v.trim().parse::().ok()) + .filter(|v| *v > 0) + .unwrap_or_else(|| { + i64::try_from(crate::server_defaults::default_reindex_max_bytes()) + .unwrap_or(i64::MAX) + }) + }) +} + /// Rows a single FULL materialize read may take before checkpointing. /// -/// `FLUREE_MATERIALIZE_MAX_ROWS_PER_FULL_PASS`, default 250_000; `0` disables -/// the bound and restores the previous read-it-all behaviour. +/// `FLUREE_MATERIALIZE_MAX_ROWS_PER_FULL_PASS` overrides; `0` disables the bound +/// and restores read-it-all. Left unset it is DERIVED from the novelty ceiling, +/// and that derivation is the whole point rather than a convenience. +/// +/// A bounded pass is only worth anything if it can COMMIT. A window over the +/// ceiling is DEFERRED, and a deferral discards the window's progress, so the +/// next poll re-reads the same rows and stops at the same wall — nothing +/// accumulates. A bound that does not fit under the ceiling is therefore not a +/// bound at all: it changes how much is read and nothing about whether any of it +/// lands. The previous flat 250_000 rows is ~27 MB of flakes, which against a +/// deployment that has pinned the ceiling to 8 MiB could never commit, so the +/// read repeated on every poll indefinitely. +/// +/// Floored at 1_000 rows so a very small ceiling still makes forward progress +/// rather than producing a zero-row pass that reads nothing and checkpoints +/// nowhere. fn materialize_max_rows_per_full_pass() -> i64 { static CACHED: std::sync::OnceLock = std::sync::OnceLock::new(); *CACHED.get_or_init(|| { - std::env::var("FLUREE_MATERIALIZE_MAX_ROWS_PER_FULL_PASS") + if let Some(explicit) = std::env::var("FLUREE_MATERIALIZE_MAX_ROWS_PER_FULL_PASS") .ok() .and_then(|v| v.trim().parse::().ok()) .filter(|v| *v >= 0) - .unwrap_or(250_000) + { + return explicit; + } + rows_for_ceiling(novelty_ceiling_bytes(), flake_bytes_per_row()) }) } -/// The commit sequence to stop a full read at, or `None` to read it whole. +/// Rows that fit under `ceiling_bytes` at `bytes_per_row`, floored at 1_000. +/// +/// Split out from the env plumbing above so the arithmetic — the part that has +/// to be right — is testable without touching process-global state. +fn rows_for_ceiling(ceiling_bytes: i64, bytes_per_row: i64) -> i64 { + if bytes_per_row <= 0 { + return 1_000; + } + (ceiling_bytes / bytes_per_row).max(1_000) +} + +/// The outcome of sizing a full read down to a commit prefix. +/// +/// Typed rather than `Option`, so the caller can say WHICH reason fired. Every +/// non-`Cut` outcome means "read it whole", and on a table whose window exceeds +/// the novelty ceiling that read cannot commit — an uncommitted window writes no +/// watermark, so the same read repeats on the next poll, forever. A silent +/// decline is therefore an invisible livelock, and one was: in production the +/// bound's success line never appeared across 16 consecutive passes and there was +/// nothing in the log to say why. Finding it needed a state-ledger query. +#[derive(Debug, PartialEq, Eq)] +enum FullReadCut { + /// Cut after this commit sequence; keep every task at or below it. + Cut(i64), + /// The bound is switched off (`FLUREE_MATERIALIZE_MAX_ROWS_PER_FULL_PASS=0`). + Disabled, + /// The plan already fits the budget — not alarming, and not worth a warning. + PlanFits, + /// A data file carries no commit sequence, so no ordering — and therefore no + /// cut — is safe. + NoSequence, + /// Every row sits in ONE commit, so there is no boundary short of the head to + /// checkpoint at. + SingleCommit, +} + +impl FullReadCut { + /// Operator-facing reason for declining to bound the read. + fn reason(&self) -> &'static str { + match self { + Self::Cut(_) => "bounded", + Self::Disabled => "the row bound is disabled", + Self::PlanFits => "the plan already fits the row budget", + Self::NoSequence => "a data file carries no commit sequence, so no cut is safe", + Self::SingleCommit => { + "every row is in a single commit, so there is no boundary short of the head" + } + } + } + + /// Whether this outcome should be shouted about. `PlanFits` is the healthy + /// small-read case and `Disabled` is a deliberate configuration choice; + /// neither is news. The other two mean an unbounded read that may never + /// commit. + fn is_alarming(&self) -> bool { + matches!(self, Self::NoSequence | Self::SingleCommit) + } +} + +/// The commit sequence to stop a full read at, or why it could not be cut. /// /// Walks `tasks` (already sorted by `(data_sequence_number, path)`) accumulating /// rows, and returns the sequence of the commit the budget ran out in. The whole /// of that commit is kept: a cut INSIDE a commit leaves the target in a state no /// snapshot names, and an unnameable state cannot be checkpointed. -/// -/// `None` when the bound is disabled, the plan already fits, any task lacks a -/// sequence number, or the budget runs out in the final commit — stopping there -/// would checkpoint at the head, which is the whole read with extra steps. -fn full_read_prefix(tasks: &[fluree_db_iceberg::scan::FileScanTask], max_rows: i64) -> Option { +fn full_read_prefix(tasks: &[fluree_db_iceberg::scan::FileScanTask], max_rows: i64) -> FullReadCut { if max_rows <= 0 { - return None; + return FullReadCut::Disabled; } let total: i64 = tasks.iter().map(|t| t.data_file.record_count).sum(); if total <= max_rows { - return None; + return FullReadCut::PlanFits; } - let last_seq = tasks + let Some(last_seq) = tasks .iter() .filter_map(|t| t.data_sequence_number) - .next_back()?; + .next_back() + else { + return FullReadCut::NoSequence; + }; let mut rows = 0i64; for t in tasks { - let seq = t.data_sequence_number?; + let Some(seq) = t.data_sequence_number else { + return FullReadCut::NoSequence; + }; rows = rows.saturating_add(t.data_file.record_count); if rows >= max_rows { - return (seq < last_seq).then_some(seq); + if seq < last_seq { + return FullReadCut::Cut(seq); + } + // The budget ran out inside the HEAD commit. Checkpointing at the + // head is the whole read with extra bookkeeping — but declining + // outright, which is what this used to do, is how a COMPACTED table + // livelocks. A compaction rewrites the table into files that all + // share one sequence number, so from then on the budget always lands + // in the head commit, the read is never bounded, and if the window + // is over the novelty ceiling it can never commit either. + // + // Fall back to the newest boundary STRICTLY BELOW the head. That is + // a smaller pass than the budget asked for, and it is still forward + // progress, which is the only property that matters here. + return tasks + .iter() + .filter_map(|t| t.data_sequence_number) + .filter(|s| *s < last_seq) + .max() + .map_or(FullReadCut::SingleCommit, FullReadCut::Cut); } } - None + // Unreachable: `total > max_rows` guarantees the accumulator crosses the + // budget above. Typed as the harmless outcome rather than a panic. + FullReadCut::PlanFits } /// Field ids to project for `projection` against `schema`: every non-nested @@ -1424,8 +1560,10 @@ impl<'a> FlureeR2rmlProvider<'a> { // its commit: splitting inside one would leave the target holding // part of a commit with no snapshot to name that state, and a // checkpoint that cannot be named cannot be resumed from. - match full_read_prefix(&plan.tasks, materialize_max_rows_per_full_pass()) { - Some(cut_seq) => { + let budget_rows = materialize_max_rows_per_full_pass(); + let cut = full_read_prefix(&plan.tasks, budget_rows); + match cut { + FullReadCut::Cut(cut_seq) => { match metadata.snapshot_at_or_before_sequence(to_snapshot_id, cut_seq) { Ok(Some(checkpoint)) if checkpoint.snapshot_id != to_snapshot_id => { let kept: Vec<_> = plan @@ -1449,13 +1587,52 @@ impl<'a> FlureeR2rmlProvider<'a> { to_snapshot_id = checkpoint.snapshot_id; kept } - // No nameable checkpoint short of the head: read it whole, as - // before. Better an expensive honest read than a watermark - // pointing somewhere the data does not correspond to. - _ => plan.tasks, + // No nameable checkpoint short of the head: read it whole. + // Better an expensive honest read than a watermark pointing + // somewhere the data does not correspond to — but SAY SO, + // because an unbounded read over the novelty ceiling cannot + // commit, and what cannot commit writes no watermark and so + // repeats forever. + outcome => { + warn!( + to_snapshot_id, + cut_sequence = cut_seq, + budget_rows, + estimated_rows = plan.estimated_row_count, + files = plan.files_selected, + reason = match outcome { + Ok(None) => "no retained snapshot names the cut", + Ok(Some(_)) => "the only nameable checkpoint is the head", + Err(_) => "the checkpoint walk failed", + }, + "materialize: full read could NOT be bounded — reading the \ + whole table. If this window exceeds the novelty ceiling it \ + cannot commit, and an uncommitted window writes no watermark, \ + so this repeats on every poll" + ); + plan.tasks + } } } - None => plan.tasks, + // Not every decline is news: `PlanFits` is the healthy small read + // and `Disabled` is a deliberate choice. The other two mean an + // unbounded read, which is the shape that livelocks. + decline => { + if decline.is_alarming() { + warn!( + to_snapshot_id, + budget_rows, + estimated_rows = plan.estimated_row_count, + files = plan.files_selected, + reason = decline.reason(), + "materialize: full read could NOT be bounded — reading the whole \ + table. If this window exceeds the novelty ceiling it cannot \ + commit, and an uncommitted window writes no watermark, so this \ + repeats on every poll" + ); + } + plan.tasks + } } }; @@ -4677,9 +4854,15 @@ mod tests { ]; // Budget runs out inside commit 20 -> keep all of 20, cut there. - assert_eq!(super::full_read_prefix(&tasks, 70), Some(20)); + assert_eq!( + super::full_read_prefix(&tasks, 70), + super::FullReadCut::Cut(20) + ); // Budget runs out on the first file of commit 10 -> cut at 10. - assert_eq!(super::full_read_prefix(&tasks, 10), Some(10)); + assert_eq!( + super::full_read_prefix(&tasks, 10), + super::FullReadCut::Cut(10) + ); } /// Every reason to decline to cut. Each returns `None`, meaning "read it @@ -4688,22 +4871,41 @@ mod tests { fn full_read_prefix_declines_when_a_cut_would_not_help() { let tasks = vec![t(10, 30), t(20, 30), t(30, 30)]; - assert_eq!(super::full_read_prefix(&tasks, 0), None, "disabled"); - assert_eq!(super::full_read_prefix(&tasks, -1), None, "negative"); - assert_eq!(super::full_read_prefix(&tasks, 90), None, "already fits"); + use super::FullReadCut; + assert_eq!( + super::full_read_prefix(&tasks, 0), + FullReadCut::Disabled, + "disabled" + ); + assert_eq!( + super::full_read_prefix(&tasks, -1), + FullReadCut::Disabled, + "negative" + ); + assert_eq!( + super::full_read_prefix(&tasks, 90), + FullReadCut::PlanFits, + "already fits" + ); assert_eq!( super::full_read_prefix(&tasks, 1_000), - None, + FullReadCut::PlanFits, "budget exceeds total" ); - assert_eq!(super::full_read_prefix(&[], 10), None, "no tasks"); + assert_eq!( + super::full_read_prefix(&[], 10), + FullReadCut::PlanFits, + "no tasks" + ); - // The budget runs out in the FINAL commit. Checkpointing at the head is - // the whole read with extra bookkeeping, so decline. + // The budget runs out in the FINAL commit. Checkpointing at the head + // would be the whole read with extra bookkeeping — but declining + // outright is what livelocked a compacted table in production, so fall + // back to the newest boundary strictly below the head instead. assert_eq!( super::full_read_prefix(&tasks, 85), - None, - "cut would be the head" + FullReadCut::Cut(20), + "falls back below the head commit rather than declining" ); // A task with no attributable sequence cannot be ordered, so no cut is @@ -4719,8 +4921,62 @@ mod tests { ]; assert_eq!( super::full_read_prefix(&unattributed, 40), - None, + FullReadCut::NoSequence, "unknown sequence" ); } + + /// A COMPACTED table is the case the old `None` return livelocked on: a + /// compaction rewrites every file with one sequence number, so the budget + /// always lands in the head commit and there is no boundary below it. That + /// is genuinely uncuttable — but it must be REPORTED, not returned as a bare + /// "declined", because an unbounded read over the novelty ceiling never + /// commits and so repeats on every poll. + #[test] + fn full_read_prefix_reports_a_single_commit_table() { + let one_commit = vec![t(7, 400_000), t(7, 400_000)]; + assert_eq!( + super::full_read_prefix(&one_commit, 250_000), + super::FullReadCut::SingleCommit + ); + assert!( + super::FullReadCut::SingleCommit.is_alarming(), + "an uncuttable full read must be shouted about, not swallowed" + ); + // And the two healthy outcomes must NOT be, or the log fills with noise + // on every small read and the real signal is lost. + assert!(!super::FullReadCut::PlanFits.is_alarming()); + assert!(!super::FullReadCut::Disabled.is_alarming()); + } + + /// The row budget has to fit under the novelty ceiling, because a pass that + /// cannot commit defers, and a deferral discards the window's progress. The + /// old flat 250_000 rows is ~27 MB of flakes: against the 8 MiB ceiling this + /// deployment pins, it could never commit, so the read repeated forever. + #[test] + fn the_row_budget_is_derived_to_fit_the_novelty_ceiling() { + const BYTES_PER_ROW: i64 = 108; + let eight_mib = 8 * 1024 * 1024; + + let rows = super::rows_for_ceiling(eight_mib, BYTES_PER_ROW); + assert!( + rows * BYTES_PER_ROW <= eight_mib, + "a full pass must fit the ceiling it has to commit under: \ + {rows} rows * {BYTES_PER_ROW}B > {eight_mib}B" + ); + assert!( + rows < 250_000, + "the derived budget must be tighter than the flat 250_000 that could not commit" + ); + + // A generous ceiling derives a generous budget — the derivation tracks + // the ceiling rather than clamping to some other constant. + assert!(super::rows_for_ceiling(256 * 1024 * 1024, BYTES_PER_ROW) > rows); + + // Floors: a tiny or nonsensical ceiling still makes forward progress + // rather than a zero-row pass that reads nothing and checkpoints nowhere. + assert_eq!(super::rows_for_ceiling(1, BYTES_PER_ROW), 1_000); + assert_eq!(super::rows_for_ceiling(eight_mib, 0), 1_000); + assert_eq!(super::rows_for_ceiling(eight_mib, -5), 1_000); + } } From d674f0d0d68265a88b779f56645e71481a86639f Mon Sep 17 00:00:00 2001 From: Christophe Di Prima Date: Tue, 18 Aug 2026 16:51:15 +0200 Subject: [PATCH 5/6] fix(materialize): make a full read resumable from a commit-sequence cursor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The durable fix for the livelock. A source whose stored watermark has fallen outside the source table's snapshot retention full-reads the whole table on every poll, forever, and no amount of bounding by SNAPSHOT can rescue it: `snapshot_at_or_before_sequence` can only name a checkpoint that is still retained, and on the only tables that need one there is nothing to name. So the bound declines, the read stays unbounded, it exceeds the novelty ceiling, the window is deferred, the deferral records no progress, and the next poll performs the identical read. Measured in production: 16 identical 727k-row reads in 14 minutes with `from_snapshot_id` frozen, and two of four fan-out targets receiving no rows from that table at all. Bound it by COMMIT SEQUENCE instead. Iceberg assigns sequence numbers per commit and never reuses or reorders them, and unlike a snapshot id a sequence stays meaningful after the snapshot that carried it has been expired away — which is precisely the state that forced the full read. `sequence_prefix_cut` therefore needs nothing retained; note what is absent from its tests: any TableMetadata, any snapshot lookup, any intact history. Each pass takes whole commits above the cursor until the row budget (already derived from the novelty ceiling) is spent, applies them, and records the highest sequence it covered. The next pass resumes above that. A table too large to materialize in one pass now drains in bounded steps instead of never. THE SAFETY PROPERTY, and the one thing to check in review: a partial pass writes the cursor and MUST NOT advance the snapshot watermark. 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 silently discarded ~80 % of a deployment's entities. `MaterializeScan::to_sequence` encodes it in one field (`Some` => prefix => suppress the snapshot write), `watermark_node` writes cursor XOR snapshot, and `a_partial_pass_writes_the_cursor_and_never_advances_the_snapshot` asserts it directly so it cannot be reintroduced quietly. Additive on purpose. `appliedSequence` sits alongside `lastSnapshotId` rather than replacing it: no state migration, the snapshot stays the fast path, and rolling back to an image without this loses resumability but not watermarks. A cursor left behind by a COMPLETE pass is deliberately not cleared, and that is safe rather than sloppy: everything at or below it was applied when it was set, and anything modified since carries a higher sequence — the same invariant the incremental path already relies on. Clearing it would need an explicit retract for no benefit. `force_full` ignores the cursor: an explicit full refresh means start over, so honouring a cursor would silently narrow it. Verified: fmt clean, clippy clean under BOTH default and all features with --all-targets, fluree-db-api --all-features --lib 1076 pass, graph_source::r2rml 87 pass, 0 failures. Caught late and worth recording: six call sites lived in fluree-db-api/tests/, so `--lib` passed at 1076 while `clippy --all-targets` failed on all six. A --lib run hides integration targets exactly as it hides feature-gated modules. --- fluree-db-api/src/graph_source/r2rml.rs | 325 ++++++++++++++---- .../src/graph_source/r2rml_materialize.rs | 143 +++++++- fluree-db-api/tests/it_iceberg_local_fs.rs | 8 +- .../tests/it_iceberg_warehouse_root.rs | 4 +- 4 files changed, 404 insertions(+), 76 deletions(-) diff --git a/fluree-db-api/src/graph_source/r2rml.rs b/fluree-db-api/src/graph_source/r2rml.rs index 47c2dcf40b..a05c819e43 100644 --- a/fluree-db-api/src/graph_source/r2rml.rs +++ b/fluree-db-api/src/graph_source/r2rml.rs @@ -170,6 +170,23 @@ pub struct MaterializeScan { /// stored watermark) or when the stored snapshot is no longer resolvable — /// both of which mean "persist regardless". pub window_age_ms: Option, + /// The highest source COMMIT SEQUENCE this pass covers, when the pass was cut + /// short by the row budget — and `None` when it covers the whole plan. + /// + /// That distinction is the point of the field, and inverting it is silent data + /// loss. `Some(s)` means the target will hold only a PREFIX of the window, so + /// `to_snapshot_id` must NOT be persisted as the watermark: a watermark naming + /// the head while the target holds a prefix claims rows that were never applied, + /// and the next scan starts after them. That is the applied-marker ratchet that + /// discarded ~80 % of a deployment's entities. Persist the sequence instead; the + /// snapshot advances only once a pass reports `None` here. + /// + /// A sequence rather than a snapshot id because it SURVIVES SNAPSHOT EXPIRY. + /// Iceberg assigns sequence numbers per commit and never reuses or reorders + /// them, and unlike a snapshot id a sequence stays meaningful after the snapshot + /// that carried it has been expired away — which is the state that forced the + /// full read to begin with. + pub to_sequence: Option, /// Column batches, streamed. See [`Self::stream`] usage notes on the method. pub stream: ColumnBatchStream, } @@ -428,6 +445,70 @@ fn full_read_prefix(tasks: &[fluree_db_iceberg::scan::FileScanTask], max_rows: i FullReadCut::PlanFits } +/// How far a sequence-bounded pass gets through the tasks above its cursor. +#[derive(Debug, PartialEq, Eq)] +enum SeqCut { + /// Everything above the cursor fits the budget — the pass is COMPLETE, so the + /// snapshot watermark may advance and no cursor need be kept. + Complete, + /// Only a prefix fits; this is the highest commit sequence it covers. The + /// snapshot watermark must NOT advance. + Prefix(i64), +} + +/// Where to stop a full read that is resuming from a SEQUENCE cursor. +/// +/// Sibling of [`full_read_prefix`], and the difference is what each needs to be +/// true. `full_read_prefix` hands back a sequence the caller must then name a +/// RETAINED snapshot for; on a table whose history has been expired away — the only +/// reason a full read is happening — there is no such snapshot, so it declines and +/// the read stays unbounded. A cursor needs nothing retained: Iceberg sequence +/// numbers are monotonic, never reused, and stay meaningful after the snapshot that +/// carried them is gone. +/// +/// `None` when any task lacks a sequence: without a total order there is no safe +/// cut, and the caller falls back to the snapshot route. +fn sequence_prefix_cut( + tasks: &[fluree_db_iceberg::scan::FileScanTask], + cursor: i64, + max_rows: i64, +) -> Option { + if tasks.iter().any(|t| t.data_sequence_number.is_none()) { + return None; + } + // Tasks arrive sorted by `(data_sequence_number, path)`, so "above the cursor" + // is a suffix and the walk below is in commit order. + let above: Vec<&fluree_db_iceberg::scan::FileScanTask> = tasks + .iter() + .filter(|t| t.data_sequence_number.is_some_and(|sq| sq > cursor)) + .collect(); + if above.is_empty() { + return Some(SeqCut::Complete); + } + let total: i64 = above.iter().map(|t| t.data_file.record_count).sum(); + if max_rows <= 0 || total <= max_rows { + return Some(SeqCut::Complete); + } + let last_seq = above.last()?.data_sequence_number?; + let mut rows = 0i64; + for t in &above { + let sq = t.data_sequence_number?; + rows = rows.saturating_add(t.data_file.record_count); + if rows >= max_rows { + // Keep the WHOLE commit the budget ran out in: a cut inside one leaves + // the target holding part of a commit, which no cursor can honestly + // describe. Landing in the newest commit means the budget covers + // everything above the cursor, so the pass is complete, not a prefix. + return Some(if sq < last_seq { + SeqCut::Prefix(sq) + } else { + SeqCut::Complete + }); + } + } + Some(SeqCut::Complete) +} + /// 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). @@ -1427,6 +1508,7 @@ impl<'a> FlureeR2rmlProvider<'a> { projection: &[String], from_snapshot_id: Option, to_snapshot_id: Option, + from_sequence: Option, ) -> QueryResult { let (storage, metadata, _loc) = self .prepare_iceberg_scan(graph_source_id, table_name) @@ -1458,6 +1540,7 @@ impl<'a> FlureeR2rmlProvider<'a> { to_snapshot_id: None, incremental: false, window_age_ms: None, + to_sequence: None, stream: empty_batch_stream(), }); }; @@ -1502,6 +1585,10 @@ impl<'a> FlureeR2rmlProvider<'a> { let scan_config = ScanConfig::new().with_projection(projected_field_ids); let planner = SendScanPlanner::new(storage.as_ref(), &metadata, scan_config); + // Set only when this pass covers a PREFIX of the plan; see + // `MaterializeScan::to_sequence` for why `Some` must SUPPRESS the snapshot + // watermark write rather than accompany it. + let mut to_sequence: Option = None; let tasks = if incremental { let plan = planner .plan_incremental(from_snapshot_id, to_snapshot_id) @@ -1561,77 +1648,123 @@ impl<'a> FlureeR2rmlProvider<'a> { // part of a commit with no snapshot to name that state, and a // checkpoint that cannot be named cannot be resumed from. let budget_rows = materialize_max_rows_per_full_pass(); - let cut = full_read_prefix(&plan.tasks, budget_rows); - match cut { - FullReadCut::Cut(cut_seq) => { - match metadata.snapshot_at_or_before_sequence(to_snapshot_id, cut_seq) { - Ok(Some(checkpoint)) if checkpoint.snapshot_id != to_snapshot_id => { - let kept: Vec<_> = plan - .tasks - .into_iter() - .filter(|t| t.data_sequence_number.is_some_and(|s| s <= cut_seq)) - .collect(); - info!( - to_snapshot_id, - checkpoint_snapshot_id = checkpoint.snapshot_id, - checkpoint_sequence = cut_seq, - files_this_pass = kept.len(), - files_total = plan.files_selected, - "materialize: full read bounded to a commit prefix; \ + + // A stored SEQUENCE cursor takes precedence, and it is the only route + // that recovers a watermark which has fallen out of retention. The + // snapshot-checkpoint route below has to name a RETAINED snapshot for + // its cut; on a table whose old snapshots have been expired away — + // which is the only reason this full read is happening — there is + // nothing to name, so it declines and the read stays unbounded. That is + // the livelock: an unbounded read exceeds the novelty ceiling, a window + // over the ceiling is deferred, a deferral records no progress, and the + // next poll performs the identical read. Forever. + let cursor_cut = from_sequence.and_then(|cursor| { + sequence_prefix_cut(&plan.tasks, cursor, budget_rows).map(|c| (cursor, c)) + }); + + if let Some((cursor, seq_cut)) = cursor_cut { + let keep_through = match seq_cut { + SeqCut::Prefix(sq) => Some(sq), + SeqCut::Complete => None, + }; + let files_total = plan.files_selected; + let kept: Vec<_> = plan + .tasks + .into_iter() + .filter(|t| { + t.data_sequence_number + .is_some_and(|sq| sq > cursor && keep_through.is_none_or(|k| sq <= k)) + }) + .collect(); + info!( + to_snapshot_id, + from_sequence = cursor, + to_sequence = ?keep_through, + files_this_pass = kept.len(), + files_total, + budget_rows, + complete = keep_through.is_none(), + "materialize: full read resumed from a sequence cursor" + ); + // `Some` => prefix only, so the caller persists the cursor and + // leaves the snapshot watermark alone. + to_sequence = keep_through; + kept + } else { + let cut = full_read_prefix(&plan.tasks, budget_rows); + match cut { + FullReadCut::Cut(cut_seq) => { + match metadata.snapshot_at_or_before_sequence(to_snapshot_id, cut_seq) { + Ok(Some(checkpoint)) if checkpoint.snapshot_id != to_snapshot_id => { + let kept: Vec<_> = plan + .tasks + .into_iter() + .filter(|t| { + t.data_sequence_number.is_some_and(|s| s <= cut_seq) + }) + .collect(); + info!( + to_snapshot_id, + checkpoint_snapshot_id = checkpoint.snapshot_id, + checkpoint_sequence = cut_seq, + files_this_pass = kept.len(), + files_total = plan.files_selected, + "materialize: full read bounded to a commit prefix; \ the watermark will checkpoint short of the head" - ); - // Reporting the checkpoint as this scan's `to` is what - // makes the caller's existing watermark write land there — - // no new vocabulary, and the crash-safety ordering - // (watermark after data) is unchanged. - to_snapshot_id = checkpoint.snapshot_id; - kept + ); + // Reporting the checkpoint as this scan's `to` is what + // makes the caller's existing watermark write land there — + // no new vocabulary, and the crash-safety ordering + // (watermark after data) is unchanged. + to_snapshot_id = checkpoint.snapshot_id; + kept + } + // No nameable checkpoint short of the head: read it whole. + // Better an expensive honest read than a watermark pointing + // somewhere the data does not correspond to — but SAY SO, + // because an unbounded read over the novelty ceiling cannot + // commit, and what cannot commit writes no watermark and so + // repeats forever. + outcome => { + warn!( + to_snapshot_id, + cut_sequence = cut_seq, + budget_rows, + estimated_rows = plan.estimated_row_count, + files = plan.files_selected, + reason = match outcome { + Ok(None) => "no retained snapshot names the cut", + Ok(Some(_)) => "the only nameable checkpoint is the head", + Err(_) => "the checkpoint walk failed", + }, + "materialize: full read could NOT be bounded — reading the \ + whole table. If this window exceeds the novelty ceiling it \ + cannot commit, and an uncommitted window writes no watermark, \ + so this repeats on every poll" + ); + plan.tasks + } } - // No nameable checkpoint short of the head: read it whole. - // Better an expensive honest read than a watermark pointing - // somewhere the data does not correspond to — but SAY SO, - // because an unbounded read over the novelty ceiling cannot - // commit, and what cannot commit writes no watermark and so - // repeats forever. - outcome => { + } + // Not every decline is news: `PlanFits` is the healthy small read + // and `Disabled` is a deliberate choice. The other two mean an + // unbounded read, which is the shape that livelocks. + decline => { + if decline.is_alarming() { warn!( to_snapshot_id, - cut_sequence = cut_seq, budget_rows, estimated_rows = plan.estimated_row_count, files = plan.files_selected, - reason = match outcome { - Ok(None) => "no retained snapshot names the cut", - Ok(Some(_)) => "the only nameable checkpoint is the head", - Err(_) => "the checkpoint walk failed", - }, - "materialize: full read could NOT be bounded — reading the \ - whole table. If this window exceeds the novelty ceiling it \ - cannot commit, and an uncommitted window writes no watermark, \ - so this repeats on every poll" - ); - plan.tasks - } - } - } - // Not every decline is news: `PlanFits` is the healthy small read - // and `Disabled` is a deliberate choice. The other two mean an - // unbounded read, which is the shape that livelocks. - decline => { - if decline.is_alarming() { - warn!( - to_snapshot_id, - budget_rows, - estimated_rows = plan.estimated_row_count, - files = plan.files_selected, - reason = decline.reason(), - "materialize: full read could NOT be bounded — reading the whole \ + reason = decline.reason(), + "materialize: full read could NOT be bounded — reading the whole \ table. If this window exceeds the novelty ceiling it cannot \ commit, and an uncommitted window writes no watermark, so this \ repeats on every poll" - ); + ); + } + plan.tasks } - plan.tasks } } }; @@ -1655,6 +1788,7 @@ impl<'a> FlureeR2rmlProvider<'a> { to_snapshot_id: Some(to_snapshot_id), incremental, window_age_ms, + to_sequence, stream, }) } @@ -4979,4 +5113,75 @@ mod tests { assert_eq!(super::rows_for_ceiling(eight_mib, 0), 1_000); assert_eq!(super::rows_for_ceiling(eight_mib, -5), 1_000); } + + /// The cursor route exists because the snapshot route cannot work on the only + /// tables that need it. Note what is absent from every assertion below: any + /// `TableMetadata`, any snapshot lookup, any retained history. That absence IS + /// the fix — a cut named by a commit sequence survives the expiry that makes + /// `snapshot_at_or_before_sequence` return `None` and leaves the read unbounded. + #[test] + fn sequence_prefix_cut_resumes_above_the_cursor_without_naming_a_snapshot() { + use super::SeqCut; + // commit 10: 30 rows, 20: 30, 30: 30 + let tasks = vec![t(10, 30), t(20, 30), t(30, 30)]; + + // Budget runs out inside commit 20 -> keep whole commits up to 20. + assert_eq!( + super::sequence_prefix_cut(&tasks, 0, 50), + Some(SeqCut::Prefix(20)) + ); + // A cursor already past commit 10 shrinks the remaining work, so the same + // budget now covers everything left. + assert_eq!( + super::sequence_prefix_cut(&tasks, 10, 70), + Some(SeqCut::Complete) + ); + // Nothing above the cursor: COMPLETE, not stuck. Reporting a prefix here + // would pin the snapshot watermark forever on an idle table. + assert_eq!( + super::sequence_prefix_cut(&tasks, 30, 10), + Some(SeqCut::Complete) + ); + assert_eq!( + super::sequence_prefix_cut(&tasks, 9_999, 10), + Some(SeqCut::Complete) + ); + // Budget disabled means "read everything above the cursor", matching the + // row-bound's own `0` semantics. + assert_eq!( + super::sequence_prefix_cut(&tasks, 0, 0), + Some(SeqCut::Complete) + ); + } + + /// Without a total order there is no safe cut, so the caller must fall back to + /// the snapshot route rather than guess. `None`, not an arbitrary boundary. + #[test] + fn sequence_prefix_cut_declines_when_a_task_has_no_sequence() { + let mixed = vec![ + t(10, 30), + fluree_db_iceberg::scan::FileScanTask::for_whole_file( + t(10, 30).data_file.clone(), + vec![], + None, + ), + t(30, 30), + ]; + assert_eq!(super::sequence_prefix_cut(&mixed, 0, 40), None); + } + + /// The cut never splits a commit. A target holding half a commit is a state no + /// cursor can honestly describe, and resuming from an unnameable state is how + /// data goes missing. + #[test] + fn sequence_prefix_cut_keeps_whole_commits() { + use super::SeqCut; + // One fat commit (100) then two small ones. + let tasks = vec![t(10, 100), t(20, 10), t(30, 10)]; + // A budget that expires part-way through commit 10 still keeps all of it. + assert_eq!( + super::sequence_prefix_cut(&tasks, 0, 40), + Some(SeqCut::Prefix(10)) + ); + } } diff --git a/fluree-db-api/src/graph_source/r2rml_materialize.rs b/fluree-db-api/src/graph_source/r2rml_materialize.rs index bfc7906672..60fc8b8a24 100644 --- a/fluree-db-api/src/graph_source/r2rml_materialize.rs +++ b/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, + from_sequence: Option, ) -> Result; } @@ -236,11 +245,19 @@ impl MaterializeSource for FlureeR2rmlProvider<'_> { graph_source_id: &str, table_name: &str, from_snapshot_id: Option, + from_sequence: Option, ) -> Result { 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)> = Vec::new(); + // (table, from-snapshot, advanced to-snapshot, prefix sequence if partial) + let mut table_watermarks: Vec<(String, Option, i64, Option)> = 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 = 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> { + 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, ) -> 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,6 +2758,27 @@ 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( @@ -2670,6 +2786,7 @@ mod tests { "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, + to_sequence: Option, 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,6 +3119,7 @@ mod engine_tests { _gs: &str, _table: &str, _from: Option, + _from_sequence: Option, ) -> Result { self.scans.fetch_add(1, std::sync::atomic::Ordering::SeqCst); let taken: Vec = std::mem::take(&mut *self.batches.lock().unwrap()); @@ -3005,6 +3127,7 @@ mod engine_tests { 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))), }) } diff --git a/fluree-db-api/tests/it_iceberg_local_fs.rs b/fluree-db-api/tests/it_iceberg_local_fs.rs index c18208780e..c4c053696f 100644 --- a/fluree-db-api/tests/it_iceberg_local_fs.rs +++ b/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"), diff --git a/fluree-db-api/tests/it_iceberg_warehouse_root.rs b/fluree-db-api/tests/it_iceberg_warehouse_root.rs index 0b6c946430..2809ca04dc 100644 --- a/fluree-db-api/tests/it_iceberg_warehouse_root.rs +++ b/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"); From c5cbdd7bbdf99201aef878c2c4484cda48c75795 Mon Sep 17 00:00:00 2001 From: Christophe Di Prima Date: Wed, 19 Aug 2026 09:12:16 +0200 Subject: [PATCH 6/6] fix(materialize): budget a QUARTER of the novelty ceiling, so the bound engages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Found by deploying the previous two commits to production and watching them do nothing. The arithmetic was individually correct at every step, which is why no unit test caught it. Raising the ceiling 8 MiB -> 96 MiB to make the window committable also took the derived row budget to ~932k rows, against a 733,608-row table. The budget exceeded the table, so `full_read_prefix` answered `PlanFits`, no prefix was taken and no cursor was ever written — while two of four fan-out targets still could not absorb the window and kept deferring (8 deferrals and 6 `novelty at max` in 4 minutes). The watermark stayed frozen at the same snapshot it had been stuck on for hours, and zero `appliedSequence` rows existed in the state ledger. Raising the ceiling had DISABLED the mechanism meant to drain it. A bound has to ENGAGE to be worth anything, and a budget larger than the table cannot engage — the pass then fails in the APPLY instead, which is where novelty is actually spent and where nothing records progress. So a pass now claims `ceiling / 4`. At 96 MiB that is 233,016 rows (24 MiB of flakes), which cuts the 733,608-row plan into ~4 committable passes, each writing a cursor the next resumes from. Chunks stay pinned at 2 MiB, so staging is untouched. A quarter rather than some other fraction for three reasons. It is the honest share: the ceiling is per-ledger and 17 sources commit into it, so one table's pass laying claim to all of it was never right. It matches the per-transaction budget's existing `reindexMaxBytes / 4` derivation. And it leaves margin for `flake_bytes_per_row` being an UNDER-estimate — 108 was measured per accumulated item, and a source row can yield several, which is why 733k rows did not in fact fit under 96 MiB. `the_budget_still_engages_on_the_table_that_livelocked` pins the property that actually matters, which is not "the budget fits the ceiling" but "the budget is small enough to engage on the table that livelocked". It also asserts the budget still tracks the ceiling upward, so this stays a fraction rather than becoming a cap. Verified: fmt clean, clippy clean under BOTH default and all features with --all-targets, fluree-db-api --all-features --lib 1077 pass, graph_source::r2rml 88 pass, 0 failures. --- fluree-db-api/src/graph_source/r2rml.rs | 67 ++++++++++++++++++++++++- 1 file changed, 65 insertions(+), 2 deletions(-) diff --git a/fluree-db-api/src/graph_source/r2rml.rs b/fluree-db-api/src/graph_source/r2rml.rs index a05c819e43..720546e5a1 100644 --- a/fluree-db-api/src/graph_source/r2rml.rs +++ b/fluree-db-api/src/graph_source/r2rml.rs @@ -331,7 +331,29 @@ fn materialize_max_rows_per_full_pass() -> i64 { }) } -/// Rows that fit under `ceiling_bytes` at `bytes_per_row`, floored at 1_000. +/// Fraction of the novelty ceiling one full pass may spend. +/// +/// A pass gets a QUARTER of the ceiling, not all of it, and the reason is a +/// production failure rather than caution. Spending the whole ceiling makes the +/// budget so large that it exceeds the table, `full_read_prefix` answers +/// `PlanFits`, no prefix is taken, no cursor is written — and the pass then fails +/// anyway in the APPLY, which is where novelty is actually spent. The bound has to +/// ENGAGE to be worth anything, and a budget bigger than the table cannot engage. +/// +/// Observed 2026-08-18: raising the ceiling 8 MiB -> 96 MiB took the budget to +/// ~932k rows against a 733,608-row table, so the bound stopped firing entirely +/// while two of four targets still could not absorb the window. Raising the +/// ceiling had DISABLED the mechanism meant to chip away at it. +/// +/// A quarter is also the honest share: the ceiling is per-ledger and 17 sources +/// commit into it, so one table's pass laying claim to all of it was never right. +/// It matches the per-transaction budget's own `reindexMaxBytes / 4` derivation, +/// and it leaves margin for `flake_bytes_per_row` being an UNDER-estimate — it was +/// measured per accumulated item, and a source row can yield several. +const PASS_CEILING_FRACTION: i64 = 4; + +/// Rows that fit under a pass's share of `ceiling_bytes` at `bytes_per_row`, +/// floored at 1_000. /// /// Split out from the env plumbing above so the arithmetic — the part that has /// to be right — is testable without touching process-global state. @@ -339,7 +361,7 @@ fn rows_for_ceiling(ceiling_bytes: i64, bytes_per_row: i64) -> i64 { if bytes_per_row <= 0 { return 1_000; } - (ceiling_bytes / bytes_per_row).max(1_000) + ((ceiling_bytes / PASS_CEILING_FRACTION) / bytes_per_row).max(1_000) } /// The outcome of sizing a full read down to a commit prefix. @@ -5112,6 +5134,47 @@ mod tests { assert_eq!(super::rows_for_ceiling(1, BYTES_PER_ROW), 1_000); assert_eq!(super::rows_for_ceiling(eight_mib, 0), 1_000); assert_eq!(super::rows_for_ceiling(eight_mib, -5), 1_000); + + // A pass claims a QUARTER of the ceiling, not all of it — see + // PASS_CEILING_FRACTION for why spending the whole thing disables the bound. + assert_eq!( + rows, + (eight_mib / 4) / BYTES_PER_ROW, + "a pass must budget a quarter of the ceiling" + ); + } + + /// The regression for the way this failed IN PRODUCTION on 2026-08-18, which no + /// unit test would have caught because the arithmetic was individually correct. + /// + /// Raising the ceiling 8 MiB -> 96 MiB to make the window committable also took + /// the derived budget to ~932k rows against a 733,608-row table. The budget + /// exceeded the table, so `full_read_prefix` answered `PlanFits`, no prefix was + /// taken and no cursor was written — while two of four fan-out targets still + /// could not absorb the window and kept deferring. Raising the ceiling had + /// DISABLED the mechanism meant to drain it, and the watermark stayed frozen. + /// + /// So the property is not "the budget fits the ceiling" — it is "the budget is + /// small enough to ENGAGE on the table that livelocked". + #[test] + fn the_budget_still_engages_on_the_table_that_livelocked() { + const BYTES_PER_ROW: i64 = 108; + const OBSERVED_PLAN_ROWS: i64 = 733_608; + let ninety_six_mib = 96 * 1024 * 1024; + + let budget = super::rows_for_ceiling(ninety_six_mib, BYTES_PER_ROW); + assert!( + budget < OBSERVED_PLAN_ROWS, + "the bound must still cut a {OBSERVED_PLAN_ROWS}-row plan at a 96 MiB \ + ceiling, else no prefix is taken and no cursor is ever written: \ + budget was {budget}" + ); + // And it must still fit comfortably under the ceiling it has to commit + // beneath, or we are back to a bound that cannot commit. + assert!(budget * BYTES_PER_ROW <= ninety_six_mib / 4); + // Sanity: raising the ceiling still raises the budget. The fix is a + // fraction, not a cap that stops tracking the ceiling. + assert!(budget > super::rows_for_ceiling(8 * 1024 * 1024, BYTES_PER_ROW)); } /// The cursor route exists because the snapshot route cannot work on the only