diff --git a/fluree-db-core/src/cancellation.rs b/fluree-db-core/src/cancellation.rs index be4a68ee92..7d68be95c3 100644 --- a/fluree-db-core/src/cancellation.rs +++ b/fluree-db-core/src/cancellation.rs @@ -65,9 +65,12 @@ impl fmt::Display for QueryCancellationReason { struct QueryCancellationInner { reason: AtomicU8, /// Bytes of retained query memory recorded via [`QueryCancellation::record_alloc`]. - /// A monotonic, deliberately-conservative high-water accumulator: callers record - /// where a retained buffer grows, so the total tracks a query's live post-scan - /// memory across all its operators. + /// A deliberately-conservative accumulator: callers record where a retained buffer + /// grows, so the total tracks a query's live post-scan memory across all its + /// operators. Mostly grows (persistent join/aggregate builds, retained lookups); + /// an allocation with a PROVABLE drop point (a scan window handed off and dropped) + /// may be paired with [`QueryCancellation::release`], so this is a high-water of + /// *live* accounted memory, not a monotonic all-time sum. allocated: AtomicUsize, /// Optional per-query memory ceiling in bytes (`NO_MEMORY_LIMIT` = unset). Stored /// as an opaque number — this crate never compares or enforces it; the query @@ -152,8 +155,10 @@ impl QueryCancellation { /// Record `bytes` of retained query memory into the shared counter. Callers record /// at the points where a retained buffer (a hash-join build table, a GROUP BY map, /// a fused dim-map) grows, so the counter tracks the query's live post-scan memory. - /// Monotonic and intentionally conservative — over-counting can only trip the guard - /// on a query already near its ceiling. No-op on a disabled handle. + /// Intentionally conservative — over-counting can only trip the guard on a query + /// already near its ceiling. Persistent allocations are never released; an + /// allocation with a provable drop point is paired with [`release`](Self::release). + /// No-op on a disabled handle. #[inline] pub fn record_alloc(&self, bytes: usize) { if let Some(inner) = &self.inner { @@ -161,6 +166,26 @@ impl QueryCancellation { } } + /// Release `bytes` previously recorded via [`record_alloc`](Self::record_alloc), + /// for an allocation whose drop point the caller can prove — e.g. a scan window + /// that has been handed off downstream and is about to drop, so a streaming scan + /// of N sequentially-freed windows accounts only the resident one rather than + /// their all-time sum. Saturating, so a double- or over-release can never + /// underflow the counter. It is a CALLER INVARIANT that `release` is paired only + /// with a matching non-persistent `record_alloc`: persistent buffers (join / + /// aggregate builds, retained lookups) must never be released, or the guard would + /// under-count live memory. No-op on a disabled handle. + #[inline] + pub fn release(&self, bytes: usize) { + if let Some(inner) = &self.inner { + let _ = inner + .allocated + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |cur| { + Some(cur.saturating_sub(bytes)) + }); + } + } + /// Bytes recorded via [`record_alloc`](Self::record_alloc) so far (0 on a disabled /// handle). #[inline] @@ -261,6 +286,34 @@ mod tests { assert_eq!(derived.allocated_bytes(), 150); } + #[test] + fn release_subtracts_and_saturates_at_zero() { + let cancellation = QueryCancellation::new(); + let derived = cancellation.clone(); + + cancellation.record_alloc(1000); + // A provable-drop allocation is released — the counter tracks LIVE memory, so + // charging then releasing nets zero (a streaming scan's per-window pattern). + cancellation.release(400); + assert_eq!(cancellation.allocated_bytes(), 600); + assert_eq!( + derived.allocated_bytes(), + 600, + "release is shared across clones" + ); + + // Over-release saturates at zero rather than underflowing. + derived.release(10_000); + assert_eq!(cancellation.allocated_bytes(), 0); + } + + #[test] + fn disabled_handle_release_is_a_noop() { + let cancellation = QueryCancellation::disabled(); + cancellation.release(1 << 20); + assert_eq!(cancellation.allocated_bytes(), 0); + } + #[test] fn memory_limit_defaults_unset_and_round_trips() { let cancellation = QueryCancellation::new(); diff --git a/fluree-db-query/src/context.rs b/fluree-db-query/src/context.rs index c4fbf71a6b..98fd55b2de 100644 --- a/fluree-db-query/src/context.rs +++ b/fluree-db-query/src/context.rs @@ -49,13 +49,60 @@ pub fn query_memory_budget_bytes() -> usize { }) } -/// R3-B: conservative per-binding byte estimate for the approximate memory-budget -/// accounting (a `Binding` is a small enum, occasionally with a heap `String`; -/// over-counting is safe — a too-tight budget only aborts a query already near OOM). -pub const BINDING_EST_BYTES: usize = 64; +/// R3-B: per-binding byte estimate for the approximate memory-budget accounting. +/// Derived from the true stack size of a [`Binding`](crate::binding::Binding) +/// (`size_of::()` = 88 bytes, documented at `binding.rs:14-17`) rather +/// than a hand-picked number, so it can never silently under-count the stack +/// footprint the way the previous `64` did (a 27% under-count of the 88-byte enum). +/// +/// This still under-counts the HEAP a binding owns: an IRI-bearing row +/// (`Binding::Iri(Arc)`, `IriMatch { iri: Arc, ledger_alias: Arc }`, +/// or a `Lit` with a `String` value) carries ~50-70 bytes of `Arc` payload +/// the stack size does not see, so a wide R2RML/Iceberg crawl of IRI rows is still +/// counted at roughly 1/2.2 of its true resident bytes. Over-counting is safe (a +/// too-tight budget only aborts a query already near OOM); this constant deliberately +/// stays a floor. A heap-aware per-binding estimate is a documented follow-up. +pub const BINDING_EST_BYTES: usize = std::mem::size_of::(); +/// F-AUD-3 site D: compile-time guard. The estimate must never drop below the +/// true `Binding` stack size — refusing any future edit that pins a smaller magic +/// number (the very regression the `64` was). Trivially holds while it is DEFINED +/// as `size_of`, and bites the moment someone changes that. +const _: () = assert!(BINDING_EST_BYTES >= std::mem::size_of::()); /// R3-B: conservative per-group overhead estimate (key bindings + aggregate state). +/// A flat estimate — like [`BINDING_EST_BYTES`] it ignores per-group heap (e.g. a +/// `GROUP_CONCAT`/`Collect` accumulator), the same documented conservatism. pub const GROUP_EST_BYTES: usize = 128; +/// F-AUD-3 site C: divisor applied to the process memory budget to derive a +/// per-query ceiling, read from `FLUREE_QUERY_BUDGET_SHARE_DIV`. Default `1` (and +/// any unparseable/zero value floors to `1`), which makes the per-query ceiling +/// equal to the full budget — byte-for-byte today's behavior. Set it to the +/// deployment's expected max query concurrency (e.g. a Lambda's reserved +/// concurrency) so N concurrent queries share the budget instead of each +/// comparing its own counter against the FULL budget (the over-admission V2 §3 +/// describes: two queries each accounting 5 GB both read "under 8 GB" while the +/// node sits at 10 GB). See [`per_query_memory_ceiling`]. +/// +/// This is the sound, minimal static form. A dynamic divisor equal to the ACTUAL +/// live concurrency (so a lone query keeps the full budget) needs a process-wide +/// active-*top-level*-query count — which only the server request boundary +/// (`fluree-db-server/src/query_control.rs`) can measure without miscounting +/// nested policy/reasoning/sub-queries that re-enter the engine. Deferred there. +pub fn query_budget_share_div() -> usize { + std::env::var("FLUREE_QUERY_BUDGET_SHARE_DIV") + .ok() + .and_then(|v| v.trim().parse::().ok()) + .unwrap_or(1) + .max(1) +} + +/// F-AUD-3 site C: the per-query memory ceiling = `full_budget / share_div`, with +/// `share_div` floored at 1. `share_div == 1` returns the full budget unchanged. +/// Pure so the division is unit-testable without touching the environment. +pub fn per_query_memory_ceiling(full_budget: usize, share_div: usize) -> usize { + full_budget / share_div.max(1) +} + /// Best-effort container/system memory limit (cgroup v2 → cgroup v1 → /// `/proc/meminfo`). `None` where none is readable (e.g. macOS dev), where the /// caller uses the absolute fallback. A cgroup "unlimited" sentinel (non-numeric @@ -759,6 +806,16 @@ impl<'a> ExecutionContext<'a> { self.cancellation.record_alloc(bytes); } + /// Release `bytes` previously recorded via [`record_alloc`](Self::record_alloc) + /// for an allocation with a provable drop point (e.g. a materialized scan window + /// that has been emitted and is about to drop). Saturating. Only valid for + /// non-persistent allocations — persistent join/aggregate/lookup buffers must + /// never be released. See [`QueryCancellation::release`]. + #[inline] + pub fn release(&self, bytes: usize) { + self.cancellation.release(bytes); + } + /// Retained query memory recorded so far via [`record_alloc`](Self::record_alloc). #[inline] pub fn mem_used(&self) -> usize { @@ -1581,3 +1638,88 @@ impl WellKnownDatatypes { false } } + +#[cfg(test)] +mod budget_tests { + use super::*; + use crate::binding::Binding; + use crate::error::QueryError; + use crate::var_registry::VarRegistry; + use fluree_db_core::{LedgerSnapshot, QueryCancellation}; + + /// F-AUD-3 site D: the per-binding estimate must never be below the true stack + /// size of a `Binding` (the 64→88 fix). Deriving it from `size_of` makes the + /// `>=` hold by construction; the equality canary documents the 88-byte size + /// (binding.rs:14-17) and fails loudly if the enum grows so the estimate is + /// re-examined for the heap it still omits. + #[test] + fn binding_est_bytes_is_at_least_binding_stack_size() { + // Derived from the type, so it equals the stack size and can never silently + // under-count it the way the previous hardcoded 64 did. The 88 canary + // documents binding.rs:14-17 and fails loudly if the enum grows (prompting a + // re-look at the heap the estimate still omits). The `>= size_of` invariant + // itself is a compile-time `const _` assertion next to the constant. + assert_eq!(BINDING_EST_BYTES, std::mem::size_of::()); + assert_eq!( + std::mem::size_of::(), + 88, + "binding.rs:14-17 documents size_of::() == 88" + ); + } + + /// F-AUD-3 site C: the ceiling divides the full budget and floors the divisor + /// at 1, so `div == 1` (the default) is byte-for-byte the full budget. + #[test] + fn per_query_ceiling_divides_and_floors() { + let full = 8usize << 30; // 8 GiB + assert_eq!( + per_query_memory_ceiling(full, 1), + full, + "div=1 → full budget" + ); + assert_eq!( + per_query_memory_ceiling(full, 4), + 2usize << 30, + "div=4 → quarter" + ); + assert_eq!(per_query_memory_ceiling(full, 0), full, "div=0 floors to 1"); + } + + /// F-AUD-3 site C: two concurrent queries pinned to a shared (divided) ceiling + /// each trip at their divided budget — neither can consume the full budget the + /// way today's undivided per-query counter allows (V2 §3 over-admission). + #[test] + fn shared_ceiling_trips_each_query_at_its_divided_budget() { + let full = 8usize << 30; + let ceiling = per_query_memory_ceiling(full, 2); // 4 GiB each + assert_eq!(ceiling, 4usize << 30); + + let snapshot = LedgerSnapshot::genesis("test/main"); + let vars = VarRegistry::new(); + // Two independent queries, each pinned to the divided ceiling as the runner + // attach point does under FLUREE_QUERY_BUDGET_SHARE_DIV=2. + for _ in 0..2 { + let cancel = QueryCancellation::new(); + cancel.set_memory_limit(ceiling); + let ctx = ExecutionContext::new(&snapshot, &vars).with_cancellation(cancel); + // Recording just over the DIVIDED ceiling trips, though it is well under + // the full 8 GiB budget an undivided query would compare against. + ctx.record_alloc(ceiling + 1); + match ctx.checkpoint() { + Err(QueryError::MemoryBudgetExceeded { + used_bytes, + budget_bytes, + }) => { + assert_eq!( + budget_bytes, ceiling, + "enforced ceiling is the divided budget" + ); + assert_eq!(used_bytes, ceiling + 1); + } + other => { + panic!("expected MemoryBudgetExceeded at the divided ceiling, got {other:?}") + } + } + } + } +} diff --git a/fluree-db-query/src/execute/runner.rs b/fluree-db-query/src/execute/runner.rs index a0e618dd75..4fa89f7691 100644 --- a/fluree-db-query/src/execute/runner.rs +++ b/fluree-db-query/src/execute/runner.rs @@ -826,6 +826,19 @@ async fn execute_prepared_into<'a, S: BatchSink>( ctx = ctx.with_tracker(tracker.clone()); } if let Some(cancellation) = config.cancellation { + // F-AUD-3 site C: give this query its share of the process memory budget + // instead of letting it (and every concurrent query) compare its own counter + // against the FULL budget. `FLUREE_QUERY_BUDGET_SHARE_DIV` (default 1) is the + // divisor; div==1 pins nothing, so the checkpoint falls back to the full + // process budget exactly as before. An explicit ceiling already pinned by the + // embedder wins (never clobbered). See `context::per_query_memory_ceiling`. + let div = crate::context::query_budget_share_div(); + if div > 1 && cancellation.memory_limit().is_none() { + let full = crate::context::query_memory_budget_bytes(); + if full != 0 { + cancellation.set_memory_limit(crate::context::per_query_memory_ceiling(full, div)); + } + } ctx = ctx.with_cancellation(cancellation); } if let Some(enforcer) = config.policy_enforcer { diff --git a/fluree-db-query/src/r2rml/operator.rs b/fluree-db-query/src/r2rml/operator.rs index 6c0cc47583..377298b6ee 100644 --- a/fluree-db-query/src/r2rml/operator.rs +++ b/fluree-db-query/src/r2rml/operator.rs @@ -236,6 +236,35 @@ fn topk_pushdown_enabled() -> bool { *ENABLED.get_or_init(|| super::env_switch_enabled("FLUREE_R2RML_TOPK_PUSHDOWN")) } +/// F-AUD-3 kill switch: whether the non-aggregate scan/crawl path records its +/// materialized windows and fact-parent lookup builds against the query memory +/// budget. Default ON (`FLUREE_SCAN_MEM_ACCOUNTING`, family falsy spellings via +/// [`super::env_switch_enabled`]). ON makes a wide crawl trip a typed +/// `MemoryBudgetExceeded` (507) instead of OOMing — closing the blind spot the +/// audit's specimen 071cd59f (a point-lookup crawl that hard-OOM'd at 10 GB) fell +/// into, which the hash-join / group-aggregate `record_alloc`s never covered. +/// OFF is a clean revert to the prior behavior (scan path invisible to the budget; +/// the `checkpoint()` polls degrade to pure cancellation checks because the scan +/// records nothing). Each materialized window is charged then RELEASED once emitted +/// (it has a provable drop point), so a long streaming scan accounts only its +/// resident window rather than the all-time sum of every freed window — that sum +/// otherwise false-aborts a bounded-memory long scan (q038). The retained parent-map +/// build (A2) and the per-file buffers (V2 site B, still excluded) are the remaining +/// non-released allocations; A2 genuinely persists so its cumulative charge is +/// correct, and site B needs its own release pairing. +fn scan_mem_accounting_enabled() -> bool { + static ENABLED: std::sync::OnceLock = std::sync::OnceLock::new(); + *ENABLED.get_or_init(|| super::env_switch_enabled("FLUREE_SCAN_MEM_ACCOUNTING")) +} + +/// F-AUD-3: conservative per-entry byte estimate for a `build_parent_lookup` +/// entry (a join key vec + a materialized subject `RdfTerm`, typically an IRI +/// string) used to account the fact-as-parent build against the budget. ~200 B is +/// the V2 estimate; like [`crate::context::BINDING_EST_BYTES`] it is a floor +/// (ignores per-entry heap beyond the term), safe because over-counting only trips +/// a build already near OOM. +const PARENT_ENTRY_EST_BYTES: usize = 200; + /// How a window of produced rows is combined with the buffered child rows. /// /// The join is *flipped* relative to a naive per-child probe: the (small, @@ -563,8 +592,7 @@ impl R2rmlScanOperator { /// — the production build calls THIS fn, so the test exercises the real /// admission, not a copy. fn ref_template_shortcut_enabled(trust_fk_refs: bool, pattern: &R2rmlPattern) -> bool { - let star_free = - pattern.star_bindings.is_empty() && pattern.star_constraints.is_empty(); + let star_free = pattern.star_bindings.is_empty() && pattern.star_constraints.is_empty(); trust_fk_refs && (star_free || Self::folded_wildcard_all_scalar(pattern)) && pattern.predicate_filter.is_none() @@ -833,17 +861,15 @@ impl R2rmlScanOperator { // WILDCARD emits per-(p,o) across co-subject maps, so a map lacking the // folded constraint predicate can still contribute rows — pruning it would // drop those rows for vertically-partitioned subjects (F10-class mappings). - let star_required_preds: Vec = if star_prune_on - && self.pattern.predicate_var.is_none() - && self.has_star_members() - { - self.pattern_predicates() - .iter() - .map(|s| (*s).to_string()) - .collect() - } else { - Vec::new() - }; + let star_required_preds: Vec = + if star_prune_on && self.pattern.predicate_var.is_none() && self.has_star_members() { + self.pattern_predicates() + .iter() + .map(|s| (*s).to_string()) + .collect() + } else { + Vec::new() + }; // PR-3 fix (b'): resolution-only class prune (template-disjoint; see // `R2rmlPattern::class_prune_hint`). Gated by the same switch as fix (a). let prune_class: Option = if star_prune_on { @@ -1286,6 +1312,7 @@ impl R2rmlScanOperator { let parent_batches = collect_stream(parent_stream, ctx).await?; let lookup = Arc::new(build_parent_lookup( + ctx, parent_tm, &parent_join_cols, parent_batches, @@ -1431,6 +1458,29 @@ impl R2rmlScanOperator { ctx, )?; + // F-AUD-3 site A1: account the materialized window against the query + // memory budget so a wide non-aggregate crawl (the previously-blind scan + // path) trips a typed `MemoryBudgetExceeded` instead of OOMing. Charge the + // resident window and `checkpoint()` BEFORE doing more work, so an oversized + // window (or this window on top of a retained A2 fact-parent build / an + // upstream fold) aborts typed. The window is then RELEASED once emitted + // (below) — it has a provable drop point, so a streaming scan of N + // sequentially-freed windows accounts only the resident one instead of + // their all-time sum. Without the release, ~70 freed 512K-row windows of a + // long un-fused-COUNT scan (q038) SUM past the budget and false-abort a + // query whose per-window resident memory is bounded and fine. + let window_est = if scan_mem_accounting_enabled() { + let est = produced + .len() + .saturating_mul(num_cols) + .saturating_mul(crate::context::BINDING_EST_BYTES); + ctx.record_alloc(est); + ctx.checkpoint()?; + est + } else { + 0 + }; + if !produced.is_empty() { emit_produced_window( &self.out_pos, @@ -1445,6 +1495,16 @@ impl R2rmlScanOperator { ctx, )?; } + + // The window has been handed off (its rows copied into `columns` / the + // bounded `self.pending` overflow) and `produced` drops at the end of this + // iteration — release its charge so only the in-flight window is counted. + // The retained overflow is bounded (≤ one window, drained before the next + // pull) and intentionally left untracked (minimal per V2 site B). The A2 + // fact-parent build charge is NOT released — that map genuinely persists. + if window_est != 0 { + ctx.release(window_est); + } // Geometric window growth. A budgeted (LIMIT) scan starts with a small // window (~the remaining budget) so a selective query does not explode // a full window into bindings before the first output row. But when the @@ -2496,6 +2556,7 @@ fn materialize_batch( /// /// HashMap mapping join key (as `Vec`) to parent subject `RdfTerm`. fn build_parent_lookup( + ctx: &ExecutionContext<'_>, parent_tm: &TriplesMap, parent_columns: &[String], batches: Vec, @@ -2503,6 +2564,21 @@ fn build_parent_lookup( let mut lookup = ParentLookup::new(); for batch in batches { + // F-AUD-3 site A2: the fact-as-parent hazard (V2) — a RefObjectMap whose + // parent is a FACT table transiently builds a full parent-sized map (tens of + // millions of entries) here, unbounded by the memo cap, which only refuses to + // RETAIN an over-window lookup AFTER it is fully built. Account each batch's + // worst-case contribution and checkpoint inside the build loop so a + // budget-exceeding build aborts typed (`MemoryBudgetExceeded`) BEFORE the + // whole map is resident, instead of OOMing. `num_rows` over-counts skipped + // (null-subject / null-key) rows — deliberately conservative. Unlike the A1 + // scan window (released on hand-off), this charge is NOT released: the lookup + // is genuinely retained for the whole join, so the cumulative charge correctly + // equals its resident footprint. + if scan_mem_accounting_enabled() { + ctx.record_alloc(batch.num_rows.saturating_mul(PARENT_ENTRY_EST_BYTES)); + ctx.checkpoint()?; + } for row_idx in 0..batch.num_rows { // Materialize parent subject let subject_term = @@ -2607,13 +2683,16 @@ impl Operator for R2rmlScanOperator { .collect(); loop { - // Cancellation checkpoint at the top of the internal loop: this loop - // can pull many windows / files / child batches before returning a + // Cancellation + memory checkpoint at the top of the internal loop: this + // loop can pull many windows / files / child batches before returning a // full output batch, so the runner's between-`next_batch` check would // otherwise never run for a whole-table scan. Covers the loop's - // non-advancing branches (overflow drain, child pull) that site 2 - // does not. - ctx.check_cancelled()?; + // non-advancing branches (overflow drain, child pull) that site 2 does + // not. Upgraded from `check_cancelled()` to `checkpoint()` (F-AUD-3): it + // also enforces the query memory budget against the window bytes recorded + // by site A1 (a no-op for the budget when scan accounting is off, since + // nothing on this path records then). + ctx.checkpoint()?; // 1. Drain overflow from a prior window before doing more work. while !self.pending.is_empty() && columns[0].len() < ctx.batch_size { let row = self.pending.pop_front().unwrap(); @@ -2763,6 +2842,239 @@ mod tests { )); } + /// F-AUD-3 site A1 — specimen 071cd59f regression. A wide non-aggregate crawl + /// (`?s ?p ?o`) materializes a window of bindings that the pre-fix scan path + /// never recorded against the memory budget, so a runaway crawl OOM'd instead of + /// aborting typed. With scan accounting on (the default), the materialized + /// window is recorded and a tiny 1-byte ceiling makes the window checkpoint + /// abort TYPED (`MemoryBudgetExceeded`, a 507) rather than completing/OOMing. + #[tokio::test] + async fn r3b_scan_window_budget_aborts_typed() { + use crate::r2rml::R2rmlTableProvider; + use crate::seed::EmptyOperator; + use crate::var_registry::VarRegistry; + use fluree_db_core::QueryCancellation; + use fluree_db_r2rml::mapping::{ + CompiledR2rmlMapping, ObjectMap, PredicateMap, PredicateObjectMap, TriplesMap, + }; + use fluree_db_tabular::{BatchSchema, FieldInfo, FieldType}; + + #[derive(Debug)] + struct StoreProvider; + #[async_trait::async_trait] + impl R2rmlTableProvider for StoreProvider { + async fn scan_table( + &self, + _graph_source_id: &str, + _table_name: &str, + _projection: &[String], + _filters: &[crate::r2rml::ScanFilter], + _topk: Option<&crate::r2rml::ScanTopK>, + _as_of_t: Option, + ) -> Result { + let schema = Arc::new(BatchSchema::new(vec![FieldInfo { + name: "STORE_KEY".to_string(), + field_type: FieldType::Int64, + nullable: true, + field_id: 1, + }])); + let batch = + ColumnBatch::new(schema, vec![Column::Int64(vec![Some(1), Some(2), Some(3)])]) + .unwrap(); + Ok(Box::pin(futures::stream::once(async move { Ok(batch) }))) + } + } + + let tm = TriplesMap::new("#Store", "DIM_STORE") + .with_subject_template("http://ex/store/{STORE_KEY}") + .with_class("http://ex/Store") + .with_predicate_object(PredicateObjectMap { + predicate_map: PredicateMap::constant("http://ex/storeKey"), + object_map: ObjectMap::column("STORE_KEY"), + }); + let mapping = Arc::new(CompiledR2rmlMapping::new(vec![tm])); + let snapshot = fluree_db_core::LedgerSnapshot::genesis("test/main"); + let vars = VarRegistry::new(); + let provider = StoreProvider; + + let cancel = QueryCancellation::new(); + cancel.set_memory_limit(1); // 1-byte ceiling → the first window's record crosses it. + let mut ctx = ExecutionContext::new(&snapshot, &vars).with_cancellation(cancel); + ctx.r2rml_table_provider = Some(&provider); + + // `?s ?p ?o` — the true-wildcard crawl the blind spot lived on. + let pattern = + R2rmlPattern::new("gs:main", VarId(0), Some(VarId(2))).with_predicate_var(VarId(1)); + let mut op = R2rmlScanOperator::new(Box::new(EmptyOperator::new()), pattern); + op.mapping = Some(Arc::clone(&mapping)); + + let mut progress = op + .build_progress(&ctx, Batch::single_empty()) + .await + .expect("build_progress") + .expect("wildcard crawl resolves the one TriplesMap"); + let num_cols = op.schema().len(); + let mut columns: Vec> = (0..num_cols).map(|_| Vec::new()).collect(); + let err = op + .advance_one_window(&ctx, &mut progress, num_cols, &mut columns) + .await + .expect_err("the materialized window must trip the 1-byte budget"); + assert!( + matches!(err, QueryError::MemoryBudgetExceeded { .. }), + "wide crawl window must abort typed, got {err:?}" + ); + } + + /// F-AUD-3 site A1 — q038 regression (the false-abort the live re-bless caught). + /// A long non-aggregate scan streams many windows that are each materialized then + /// FREED; the per-window budget charge is released on hand-off, so the windows do + /// not SUM to a false over-budget. Here 64 one-row windows (charge ~528 B each) + /// run under an 8000 B ceiling: each resident window fits and the scan COMPLETES — + /// whereas the pre-fix cumulative counter crossed the ceiling within ~16 windows + /// and false-aborted a bounded-memory scan. The single-window abort test above + /// (an oversized window on a 1-byte ceiling) still passes: the checkpoint fires + /// while the window is charged, before it is released. + #[tokio::test] + async fn r3b_scan_windows_release_no_false_abort() { + use crate::r2rml::R2rmlTableProvider; + use crate::seed::EmptyOperator; + use crate::var_registry::VarRegistry; + use fluree_db_core::QueryCancellation; + use fluree_db_r2rml::mapping::{ + CompiledR2rmlMapping, ObjectMap, PredicateMap, PredicateObjectMap, TriplesMap, + }; + use fluree_db_tabular::{BatchSchema, FieldInfo, FieldType}; + + const N_WINDOWS: usize = 64; + + #[derive(Debug)] + struct ManyRowsProvider; + #[async_trait::async_trait] + impl R2rmlTableProvider for ManyRowsProvider { + async fn scan_table( + &self, + _graph_source_id: &str, + _table_name: &str, + _projection: &[String], + _filters: &[crate::r2rml::ScanFilter], + _topk: Option<&crate::r2rml::ScanTopK>, + _as_of_t: Option, + ) -> Result { + let schema = Arc::new(BatchSchema::new(vec![FieldInfo { + name: "STORE_KEY".to_string(), + field_type: FieldType::Int64, + nullable: true, + field_id: 1, + }])); + // N single-row batches → with window_rows pinned to 1, one window each. + let batches: Vec> = (0..N_WINDOWS as i64) + .map(|k| { + Ok(ColumnBatch::new( + Arc::clone(&schema), + vec![Column::Int64(vec![Some(k + 1)])], + ) + .unwrap()) + }) + .collect(); + Ok(Box::pin(futures::stream::iter(batches))) + } + } + + let tm = TriplesMap::new("#Store", "DIM_STORE") + .with_subject_template("http://ex/store/{STORE_KEY}") + .with_class("http://ex/Store") + .with_predicate_object(PredicateObjectMap { + predicate_map: PredicateMap::constant("http://ex/storeKey"), + object_map: ObjectMap::column("STORE_KEY"), + }); + let mapping = Arc::new(CompiledR2rmlMapping::new(vec![tm])); + let snapshot = fluree_db_core::LedgerSnapshot::genesis("test/main"); + let vars = VarRegistry::new(); + let provider = ManyRowsProvider; + + let cancel = QueryCancellation::new(); + // Between one window (~528 B) and the naive cumulative (64×528 ≈ 34 KB): the + // pre-fix counter crosses this within ~16 windows; released windows never do. + cancel.set_memory_limit(8000); + let mut ctx = ExecutionContext::new(&snapshot, &vars).with_cancellation(cancel); + ctx.r2rml_table_provider = Some(&provider); + + let pattern = + R2rmlPattern::new("gs:main", VarId(0), Some(VarId(2))).with_predicate_var(VarId(1)); + let mut op = R2rmlScanOperator::new(Box::new(EmptyOperator::new()), pattern); + op.mapping = Some(Arc::clone(&mapping)); + + let mut progress = op + .build_progress(&ctx, Batch::single_empty()) + .await + .expect("build_progress") + .expect("wildcard crawl resolves the one TriplesMap"); + let num_cols = op.schema().len(); + let mut columns: Vec> = (0..num_cols).map(|_| Vec::new()).collect(); + + let mut windows = 0usize; + loop { + progress.window_rows = 1; // pin one row per window (bypass geometric growth) — env-free. + let more = op + .advance_one_window(&ctx, &mut progress, num_cols, &mut columns) + .await + .expect("a bounded-resident streaming scan must NOT false-abort on the budget"); + // Simulate the consumer draining the emitted batch so nothing accumulates + // outside the released window charge. + for c in &mut columns { + c.clear(); + } + op.pending.clear(); + // The window charge is released on hand-off, so live accounted memory + // never approaches the naive cumulative — it stays within one window. + assert!( + ctx.mem_used() < 8000, + "released window charge must keep live memory under budget, got {}", + ctx.mem_used() + ); + if !more { + break; + } + windows += 1; + assert!(windows < 1000, "safety bound"); + } + assert!( + windows >= N_WINDOWS, + "the whole streaming scan must complete; got {windows} windows" + ); + } + + /// F-AUD-3 site A2 — the fact-as-parent build. `build_parent_lookup` transiently + /// materializes a full parent-sized map; with a RefObjectMap whose parent is a + /// FACT table this is tens of millions of entries, unbounded by the memo cap. + /// The per-batch accounting + checkpoint makes a budget-exceeding build abort + /// TYPED before the whole map is resident (a 1-byte ceiling trips on the first + /// batch here) instead of OOMing. + #[test] + fn r3b_parent_build_budget_aborts_typed() { + use crate::var_registry::VarRegistry; + use fluree_db_core::{LedgerSnapshot, QueryCancellation}; + use fluree_db_r2rml::mapping::TriplesMap; + + let snapshot = LedgerSnapshot::genesis("test/main"); + let vars = VarRegistry::new(); + let cancel = QueryCancellation::new(); + cancel.set_memory_limit(1); // 1-byte ceiling. + let ctx = ExecutionContext::new(&snapshot, &vars).with_cancellation(cancel); + + let parent_tm = TriplesMap::new("#Customer", "customers") + .with_subject_template("http://ex/customer/{ID}"); + // One batch of rows is enough — the per-batch record_alloc crosses the ceiling + // before any row is inserted, so the build aborts on the first batch. + let batch = single_col_batch("ID", vec![Some(1), Some(2), Some(3)]); + let err = build_parent_lookup(&ctx, &parent_tm, &["ID".to_string()], vec![batch]) + .expect_err("the fact-parent build must trip the 1-byte budget"); + assert!( + matches!(err, QueryError::MemoryBudgetExceeded { .. }), + "fact-parent build must abort typed, got {err:?}" + ); + } + /// PR-5 soundness: the top-k pushdown must be declined whenever the scan /// carries a residual filter the operator enforces after emitting rows — /// otherwise the heap is fed pre-filter values and could prune files whose @@ -3607,12 +3919,7 @@ mod tests { #[test] fn f16_folded_wildcard_all_scalar_gate() { use crate::r2rml::{ObjectConstant, ScanValue}; - let scalar = |p: &str| { - ( - p.to_string(), - ObjectConstant::Scalar(ScanValue::Int(7)), - ) - }; + let scalar = |p: &str| (p.to_string(), ObjectConstant::Scalar(ScanValue::Int(7))); let iri = |p: &str| { ( p.to_string(), @@ -3633,8 +3940,7 @@ mod tests { // Mixed scalar + IRI → disqualified (every constraint must be scalar). let mut mixed = pass.clone(); - mixed.star_constraints = - vec![scalar("http://ex/lineNumber"), iri("http://ex/order")]; + mixed.star_constraints = vec![scalar("http://ex/lineNumber"), iri("http://ex/order")]; assert!(!R2rmlScanOperator::folded_wildcard_all_scalar(&mixed)); // Fixed-predicate star member present → not the crawl fold. @@ -3667,8 +3973,8 @@ mod tests { use crate::r2rml::{ObjectConstant, ScanValue}; // The ref side: single-column templated FK — shortcut-eligible (the // deployed `edw:order` → FactOrder shape). - let parent = - TriplesMap::new("#Order", "FACT_ORDER").with_subject_template("http://ex/order/{ORDER_KEY}"); + let parent = TriplesMap::new("#Order", "FACT_ORDER") + .with_subject_template("http://ex/order/{ORDER_KEY}"); let rom = RefObjectMap::new("#Order", "ORDER_KEY", "ORDER_KEY"); assert!( build_ref_shortcut(&parent, &rom).is_some(), @@ -3683,7 +3989,9 @@ mod tests { "http://ex/lineNumber".to_string(), ObjectConstant::Scalar(ScanValue::Int(1)), )]; - assert!(R2rmlScanOperator::ref_template_shortcut_enabled(true, &folded)); + assert!(R2rmlScanOperator::ref_template_shortcut_enabled( + true, &folded + )); // Iri-folded → admission false (sound parent-scan path; catch #11). let mut iri_folded = folded.clone(); @@ -3705,7 +4013,9 @@ mod tests { // unchanged by the F-16 amendment). let plain = R2rmlPattern::new("gs:main", VarId(0), Some(VarId(2))).with_predicate_var(VarId(1)); - assert!(R2rmlScanOperator::ref_template_shortcut_enabled(true, &plain)); + assert!(R2rmlScanOperator::ref_template_shortcut_enabled( + true, &plain + )); // Fixed-predicate star + trust ON → admission false (star shapes keep // parent-scan + dangling-FK semantics; unchanged). @@ -3714,7 +4024,9 @@ mod tests { "http://ex/lineNumber".to_string(), ObjectConstant::Scalar(ScanValue::Int(1)), )]; - assert!(!R2rmlScanOperator::ref_template_shortcut_enabled(true, &star)); + assert!(!R2rmlScanOperator::ref_template_shortcut_enabled( + true, &star + )); } /// A templated (non-constant) predicate binds `?p` from the row when the