diff --git a/fluree-db-api/src/graph_source/r2rml.rs b/fluree-db-api/src/graph_source/r2rml.rs index 24461d9486..720546e5a1 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, } @@ -244,6 +261,276 @@ 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` 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(|| { + 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) + { + return explicit; + } + rows_for_ceiling(novelty_ceiling_bytes(), flake_bytes_per_row()) + }) +} + +/// 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. +fn rows_for_ceiling(ceiling_bytes: i64, bytes_per_row: i64) -> i64 { + if bytes_per_row <= 0 { + return 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. +/// +/// 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. +fn full_read_prefix(tasks: &[fluree_db_iceberg::scan::FileScanTask], max_rows: i64) -> FullReadCut { + if max_rows <= 0 { + return FullReadCut::Disabled; + } + let total: i64 = tasks.iter().map(|t| t.data_file.record_count).sum(); + if total <= max_rows { + return FullReadCut::PlanFits; + } + let Some(last_seq) = tasks + .iter() + .filter_map(|t| t.data_sequence_number) + .next_back() + else { + return FullReadCut::NoSequence; + }; + let mut rows = 0i64; + for t in tasks { + let Some(seq) = t.data_sequence_number else { + return FullReadCut::NoSequence; + }; + rows = rows.saturating_add(t.data_file.record_count); + if rows >= max_rows { + 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); + } + } + // Unreachable: `total > max_rows` guarantees the accumulator crosses the + // budget above. Typed as the harmless outcome rather than a panic. + 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). @@ -1243,6 +1530,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) @@ -1274,10 +1562,11 @@ impl<'a> FlureeR2rmlProvider<'a> { to_snapshot_id: None, incremental: false, window_age_ms: None, + to_sequence: None, 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. @@ -1318,6 +1607,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) @@ -1358,7 +1651,144 @@ 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. + let budget_rows = materialize_max_rows_per_full_pass(); + + // 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 + } + // 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 + } + } + } + // 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 + } + } + } }; // How OLD is the window we are about to read? Measured from Iceberg's own @@ -1380,6 +1810,7 @@ impl<'a> FlureeR2rmlProvider<'a> { to_snapshot_id: Some(to_snapshot_id), incremental, window_age_ms, + to_sequence, stream, }) } @@ -4542,4 +4973,278 @@ 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), + 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), + super::FullReadCut::Cut(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)]; + + 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), + FullReadCut::PlanFits, + "budget exceeds total" + ); + assert_eq!( + super::full_read_prefix(&[], 10), + FullReadCut::PlanFits, + "no tasks" + ); + + // 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), + 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 + // 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), + 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); + + // 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 + /// 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"); diff --git a/fluree-db-iceberg/src/metadata/table.rs b/fluree-db-iceberg/src/metadata/table.rs index e6ac648a3f..b3531d3bcc 100644 --- a/fluree-db-iceberg/src/metadata/table.rs +++ b/fluree-db-iceberg/src/metadata/table.rs @@ -170,6 +170,54 @@ 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> { + // 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` /// operation. Only then does an added-files incremental scan capture all /// changes (no `overwrite`/`delete`/`replace` => no updates or deletions to @@ -551,6 +599,66 @@ 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] + 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![ 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,