perf: Iceberg/R2RML graph-source performance — planning, catalog caching, LIMIT + row-group pruning - #1406
Merged
Merged
Conversation
A subject-only / rdf:type triple pattern lowered to predicate_filter=None, object_var=None, which (a) projected every POM column, (b) scanned every RefObjectMap parent table to build lookups that subject-only materialization never reads, and (c) ran as a separate correlated operator that re-scanned the base table once per child batch. The same-subject star also resolved TriplesMaps by predicate alone, so shared predicates fanned out to every TriplesMap sharing them. - TriplesMap::subject_columns(): project only subject template/column for subject-only patterns instead of columns_for_predicate(None). - R2rmlScanOperator: project subject-only and select no POMs/parents when the pattern has no object var; keep the all-POMs path for a true wildcard (object_var=Some). - rewrite: fuse a same-subject rdf:type into the star base via class_filter, which constrains TriplesMap resolution to the class and drops the redundant correlated scan; emit unfused classes as subject-only scans. Against live Snowflake Horizon (enterprise R2RML graph source), results byte-identical: simple Store query 6 scans -> 1; Store join Geography 36 -> 5 and 33 -> 2; the unrelated-dimension fan-out is eliminated and peak RSS drops ~8x. Adds scan-count + projection guardrail tests covering the type/star, subject-only, referenced-RefObjectMap (parent kept), and wildcard cases.
Every REST-catalog table scan rebuilt the auth provider + RestCatalogClient and called loadTable, so each scan paid a fresh OAuth token exchange plus a GET /tables/<t> round-trip. The downstream metadata/scan-files caches sit below loadTable (keyed by its output, metadata_location), so they never prevented the REST call. There was no loadTable-response cache in REST mode at all, and N scans of one table did N OAuth exchanges and N loadTable GETs returning the same location. FlureeR2rmlProvider is constructed once per query, so it now owns an IcebergCatalogSession: one RestCatalogClient per source (its OAuth CachedToken services every scan from a single token exchange) and a loadTable response cache keyed (graph_source_id, namespace.table). The first scan of a table loads it; later scans reuse the pinned snapshot + vended credentials with no round-trip. This also pins one Iceberg snapshot across the whole query instead of letting independent per-scan loads observe different snapshots mid-query. Cached vended credentials are never served at/after their 30s-buffered expiry (a late scan reloads). FLUREE_ICEBERG_LOADTABLE_CACHE=0 restores per-scan loads. Against live Snowflake Horizon (results byte-identical): the Store-join-Geography query drops OAuth exchanges 5 -> 1 and loadTable round-trips 5 -> 2, wall 18.9s -> 9.1s. Unit tests cover cache hit, expired-creds miss, no-creds entries, and key isolation.
The query-scoped loadTable cache dropped an entry once its vended credentials neared expiry, and the reload stored the fresh response wholesale. If the table committed between the first load and the refresh, load_table returns a newer metadata_location, so later scans in the same query could observe a different Iceberg snapshot — breaking the per-query snapshot pin the cache is meant to provide. Pin metadata_location on the first load and keep it across refreshes: a reload now updates only the credentials, and scan_table overrides the reloaded metadata_location with the pinned one. Vended credentials are bucket/prefix scoped, so the fresh creds still read the pinned snapshot's immutable data files. Adds a unit test asserting the pin survives a store that lands on a newer snapshot.
R2rmlScanOperator is a correlated streaming nested-loop: next_batch pulls one child batch and calls build_progress, which re-invokes scan_table for the inner table. A cross-table join (the inner pattern's child is the outer scan) therefore re-scanned the inner dimension table once per child batch — the Store-join- Geography query decoded DIM_GEOGRAPHY once per outer batch on top of the RefObjectMap parent lookup. Cache the inner scan on the operator keyed by (table_name, projection): the first child batch collects it (up to one materialize window) and later batches replay the cached batches instead of re-scanning. An inner larger than one window is not cached and streams fresh per batch as before, so resident memory is bounded by the same window the scan already used. FLUREE_R2RML_SCAN_CACHE=0 disables it. Against live Snowflake Horizon (results byte-identical): Store-join-Geography LIMIT 10 drops DIM_GEOGRAPHY scans 4 -> 2 and total scans 5 -> 3 (the residual GEOGRAPHY scan is the separate parent lookup). Adds a guardrail that a 2500-row (3-batch) outer scans the inner main table exactly once; it fails with the cache disabled.
The inner-scan cache is keyed by (table_name, projection) but a scan also depends on its pushdown scan_filters: a filtered scan can prune files and return a row SUBSET. Replaying that subset for a differently-filtered (or unfiltered) scan of the same table/projection would silently drop rows. Today a pattern's filter is constant per operator so this cannot misfire, but it is a latent footgun as filter pushdown expands. Gate the cache (both read and write) on scan_filters.is_empty(): filtered scans bypass the cache and scan fresh. Also clear scan_cache in close() so a cached window of ColumnBatch data is not retained past the operator's useful life.
…ache The query-scoped catalog session eliminated the per-scan REST storm within one query, but every new query still paid a fresh OAuth token exchange (~0.5s) and a loadTable GET (~1.3-3s per distinct table) — the whole per-query floor against Snowflake Horizon. For a long-lived server answering repeated queries against the same tables, that is pure waste: the OAuth token is valid ~1h and a table's metadata rarely changes between queries. Move two caches from the per-query session into the process-wide R2rmlCache: - rest_clients: the RestCatalogClient (and its OAuth CachedToken + connection pool) keyed by a source config fingerprint, so a warm server does one token exchange per ~hour instead of one per query. A rotated PAT changes the fingerprint and rebuilds the client. - rest_load_tables: loadTable responses keyed by (source, table) with a 60s TTL (FLUREE_ICEBERG_LOADTABLE_TTL_SECS, 0 disables) and a vended-credential-expiry gate, so a burst of queries skips the catalog GET. Staleness is bounded by the TTL; a new query still pins its own snapshot, and the cross-query cache always records the actual current catalog state (never a query's pin), so pin preservation never leaks a stale metadata_location to other queries. The per-query session keeps only the snapshot pin. FLUREE_ICEBERG_LOADTABLE_CACHE=0 still disables everything. Live (server, two identical queries, results byte-identical): the second query drops OAuth 1->0, loadTable GET 1->0 (cross-query hit), wall 3.02s -> 0.20s. Unit tests cover the cross-query cache put/get, credential-expiry gate, and clear().
Add a configuration section covering the catalog caching knobs (FLUREE_ICEBERG_LOADTABLE_CACHE, FLUREE_ICEBERG_LOADTABLE_TTL_SECS, FLUREE_R2RML_SCAN_CACHE, FLUREE_R2RML_MATERIALIZE_WINDOW_ROWS, FLUREE_ICEBERG_SCAN_CONCURRENCY), the two-scope caching model (per-query snapshot pin + process-wide client/loadTable reuse), and the freshness/latency tradeoff of the loadTable TTL.
A LIMIT never reached an Iceberg/R2RML scan: R2rmlScanOperator ignored the row-budget (no-op default), and the GraphOperator that wraps every graph-source query neither forwarded the budget nor stopped draining its inner subplan. So a LIMIT 5 still materialized a full 512K-row window (and a fact-table LIMIT drained the whole table — effectively infeasible). - R2rmlScanOperator now honors set_row_budget: it caps the materialize window at the remaining budget and stops emitting once the budget is met (counting output rows, not scanned rows). It does NOT forward the budget to its child (an inner scan feeding this operator's join is not row-preserving). A budgeted scan also bypasses the inner-scan cache, whose full-window collect would otherwise defeat early termination. FLUREE_R2RML_LIMIT_PUSHDOWN=0 disables it. - GraphOperator forwards its LIMIT budget into the per-parent-batch inner subplan so the wrapped scan can terminate early instead of buffering the whole table into result_buffer. Budget is stored (not forwarded to the parent child) and applied to both the R2RML batched fast path and the per-row path. Correctness is bounded by the outer LIMIT; a per-parent-batch budget may over-read across parent batches but never under-reads. The budget only reaches a scan through row-preserving operators (Project/Offset), so a scan under a FILTER/join/aggregate/ORDER BY is never early-terminated. Live (Snowflake Horizon): FACT_ORDER_LINE LIMIT 5 goes from >90s (was timing out at 5min) to 4.4s. Adds a guardrail asserting a LIMIT 5 pulls ~1 of 40 streamed chunks while the un-limited query drains all 40.
The Parquet reader decoded every row group of every selected file; the residual filter was carried on FileScanTask but never consulted. For a multi-row-group fact file, a date/int/bool predicate could skip most groups but didn't. Add row-group pruning: before decoding a row group, evaluate the residual filter against its column statistics (min/max) and skip groups that cannot match. The min/max reasoning is factored into a shared bounds_can_contain used by both the existing DataFile (file-level) path and the new Parquet-stats (row-group) path, so the logic is identical and already covered by the file-pruning tests. Row-group bounds are read only for the pushdown-supported physical types (bool / int32 / int64, including int32-backed dates); anything else keeps the group (conservative). The field id -> Parquet column mapping reused is the same one the decoder uses, so a filter resolves to exactly the column it reads. Wired into both reader paths (small sparse-buffer and large range-backed); for the range-backed reader a skipped group's column chunks are never requested, so its on-demand reads are avoided too. FLUREE_ICEBERG_PREDICATE_PUSHDOWN=0 disables it; row_groups_pruned is logged. Adds a reader-level test that writes a two-row-group Parquet file and asserts pruning against its real statistics. Note: pruning only reduces decode within files a query already reads. A FILTER + LIMIT fact query still cannot early-terminate (FILTER is not row-preserving, so it blocks the LIMIT budget); making that path fast requires consuming the pushed filter into the scan and dropping the redundant FilterOperator, handled next.
aaj3f
approved these changes
Jul 1, 2026
aaj3f
left a comment
aaj3f
left a comment
Contributor
There was a problem hiding this comment.
Thanks, @bplatz! Excited for these wins!
I have one in-line comment below, and then I'm going to provide some feedback direct-from-Claude here, but otherwise things look good!
### 1. `fluree-db-query/src/r2rml/rewrite.rs:159-166` (and `:181-184`) — CORRECTNESS: class fusion silently empties a legal cross-TriplesMap query
Fusing a `?s a ex:C` pattern into a same-subject star sets `base.class_filter =
Some(class)` on a base that already carries `predicate_filter`. In
`operator.rs::build_progress` (~lines 408-428) TriplesMap resolution then requires
**one** TriplesMap to satisfy *both* the class **and** the predicate:
```rust
if let Some(ref class_filter) = self.pattern.class_filter {
if !tm.classes().contains(class_filter) { return false; }
}
if let Some(ref pred_filter) = self.pattern.predicate_filter {
if !tm.predicate_object_maps.iter().any(|pom|
pom.predicate_map.as_constant() == Some(pred_filter.as_str())) { return false; }
}
```
Same subject **variable** does not imply same TriplesMap. A legal R2RML mapping can
vertically partition a subject: `TM_A { subject person/{id}, rr:class ex:Person }`
and `TM_B { subject person/{id}, predicate ex:name }`. For
`?s a ex:Person ; ex:name ?name`:
- **Before this PR:** the type pattern was a standalone subject-only R2RML scan
(resolves `TM_A`) and the name pattern a standalone scan (resolves `TM_B`), joined
on `?s` — correct.
- **After this PR:** the name star fuses `class_filter=Person`, so `build_progress`
needs one TM with class Person **and** predicate `ex:name`. `TM_A` fails (no name),
`TM_B` fails (no class) → `triples_maps.is_empty()` → **the query returns zero
rows.**
This is a silent wrong-result regression, not a crash. It's gated on the
class and the queried predicate living in different TriplesMaps that share a
subject template — uncommon in the tested star-schema dataset (dimensions joined via
RefObjectMap, not shared-subject partitioning), but valid R2RML.
**Suggested fix:** only fuse the class when exactly one TriplesMap resolves the
star predicate **and** declares the class; otherwise leave the class as a joined
subject-only scan (the pre-fusion path, already correct). A guard test covering the
split-TriplesMap shape would lock this down.
---
### 2. `fluree-db-query/src/r2rml/operator.rs:673-674` — PERF: a small LIMIT over a selective *internal* R2RML join scans the whole table in tiny windows
`window_rows = materialize_window_rows().min(b.saturating_sub(self.emitted).max(1))`
is computed **once** per `ScanProgress` and never grows. For the common R2RML case
(single seed child → one `ScanProgress`), a `LIMIT 5` fixes the materialize window at
5 rows for the entire scan.
When the scan's produced rows feed an internal join (RefObjectMap / correlated child)
that filters most of them out, the output budget is never met, so the operator keeps
pulling **5-row windows** across the whole table — many small rayon materialize passes
instead of the full 512K-row windows the un-budgeted path uses. On `main` this same
query drains the table in big windows; post-PR it drains it in tiny ones, so `LIMIT`
can make this shape **slower**. Correctness is unaffected.
**Suggested fix:** floor the window at a reasonable minimum (e.g.
`materialize_window_rows().min(max(budget_remaining, MIN_WINDOW))`) or grow the window
geometrically across iterations when the budget isn't being met. Alternatively, note
the pathology next to the `FLUREE_R2RML_LIMIT_PUSHDOWN` kill-switch.
---
### 3. `fluree-db-iceberg/src/scan/pruning.rs:279-282` — LATENT CORRECTNESS: row-group pruning has no Decimal-logical-type guard
`stat_bounds` keys purely on the Parquet **physical** variant
(`Statistics::Int64` etc.), with no check of the column's logical/converted type. If
a column is stored as INT64 with a *decimal* logical type (unscaled integers) and the
R2RML `xsd:integer → Decimal` coercion emits an `Int64` filter literal, pruning
compares the scaled query value against unscaled bounds → can drop a matching row
group (e.g. Decimal(10,2) storing `5.00` as `500`, filter `= 5`, bounds `[500,504]` →
`Eq` prunes, matching row vanishes). Currently safe only because the Iceberg spec
mandates `fixed_len_byte_array` for decimals (→ `(None,None)` → conservative keep) —
the PR's own "Not included" note flags this xsd:integer↔Decimal mismatch as live.
**Suggested fix:** skip pruning (return `(None,None)`) when the Parquet column's
`LogicalType`/`ConvertedType` is Decimal, so the safety no longer rests on an
implicit encoding assumption.
---
### 4. `fluree-db-api/src/graph_source/r2rml.rs:725-781` — MINOR/UNLIKELY: snapshot pin can slip under concurrent first-load of the same table
The pin is deduped in `store_load_table` (first insert wins), but the caller scans
with its own local `resp.metadata_location`, not the pinned value. If two scans of the
same table both miss the per-query pin and the cross-query cache and both issue a real
`load_table` concurrently, scan A pins `A_loc` while scan B proceeds with `B_loc`; a
mid-query commit then yields two snapshots in one query. Almost certainly unreachable
given the pull-based, sequential operator execution, but worth a defensive re-read of
the pin after `store_load_table` if you want the invariant to hold unconditionally.
---
### 5. `fluree-db-query/src/r2rml/rewrite.rs:162-164, 182-184` — CLEANUP: duplicated fusion block
```rust
if let Some(class) = take_single_class(&mut class_groups, subject) {
base.class_filter = Some(class);
}
```
appears verbatim in both the single-member and multi-member star arms. Fold into a
small helper (`fuse_class(&mut base, &mut class_groups, subject)`) once `base` is
chosen. Minor.
Comment on lines
+1019
to
+1020
| cannot leak a stale location to other queries. The client cache is keyed by a | ||
| config fingerprint, so rotating the source's access token rebuilds the client. |
Contributor
There was a problem hiding this comment.
This is less a doc comment and more of a claim in the docs that I think sounds good to me, but I'm not sure it's right vis-a-vis the code in fluree-db-api/src/graph_source/r2rml.rs + cache.rs.
It seems config_fingerprint hashes only the raw config JSON, while an OAuth2/Bearer secret is a ConfigValue, which in practice is Dynamic { env_var: "…" }. The JSON stores the env var name, not the secret. An operator might rotate the secret, but record.config is unchanged --> fingerprint unchanged --> cache.rest_client(&fp) returns the stale Arc<RestCatalogClient> holding the old secret and its cached OAuth token. rest_clients is moka::sync::Cache::new(64) with no TTL (cache.rs:145), so the stale client survives until LRU eviction or process restart. When the cached OAuth token expires (~1h) the client refreshes with the stale secret --> 401s process-wide. Only literal-in-config secrets get the behavior the comment advertises.
We could fold the resolved secret into the fingerprint, or put a TTL on rest_clients so it self-heals
Contributor
Author
There was a problem hiding this comment.
Fixed in e738077: the client cache now has a TTL (FLUREE_ICEBERG_REST_CLIENT_TTL_SECS, default 15m) so an env-var/secret-store secret rotation self-heals rather than serving 401s until eviction. Also corrected this doc line and the config_fingerprint / rest_clients docstrings to stop claiming rotation rebuilds the client — the fingerprint only sees inline config, so the TTL is what bounds staleness.
…w, cache TTL Follow-ups from the #1406 review: - Class fusion could silently drop rows. Fusing `?s a ex:C` into a same-subject star sets class_filter, and TriplesMap resolution then needs one map with both the class and the star predicate. A vertically partitioned mapping (class and predicate in separate maps sharing a subject template) has none, so a fused scan resolved zero maps and returned no rows. Fuse only when every map resolving the base predicate also declares the class (class_fusion_is_safe); otherwise leave the class as its own subject-only scan joined on the subject. The rewrite now takes the compiled mapping (loaded best-effort at the call sites; None disables fusion but stays correct). - Budgeted (LIMIT) materialize window was fixed at the remaining budget, so a LIMIT over an internal join that filters most rows re-scanned the whole table in tiny windows. Grow the window geometrically each pass so it ramps to the full size after a few low-yield passes; a selective LIMIT still stops after its cheap first window. - REST catalog client cache had no TTL and was keyed by a fingerprint of the raw config JSON, so rotating a secret referenced by env var / secret store never invalidated the client — stale token → process-wide 401s until eviction. Add a TTL (FLUREE_ICEBERG_REST_CLIENT_TTL_SECS, default 15m) so a rotation self-heals, and correct the docstrings / configuration docs that claimed rotation rebuilt the client. - Row-group pruning keyed purely on the Parquet physical type; guard against a Decimal logical/converted type (skip pruning) so correctness no longer rests on the Iceberg spec's fixed_len_byte_array-for-decimals encoding assumption. - Adopt the pinned metadata_location after store_load_table so a concurrent first-load of the same table cannot leave two scans on different snapshots. Also fold the duplicated class-fusion block into one helper. Adds unit tests for class_fusion_is_safe and an end-to-end split-TriplesMap regression test.
Contributor
Author
|
All review points addressed in e738077:
(The inline config-fingerprint note is answered in its thread — REST client cache now has a TTL.) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Performance work on the Iceberg/R2RML graph-source read path, from a live diagnosis against Snowflake Horizon (
ENTERPRISE_DEMO.DW). Correct-but-slow queries — a 10-row join took ~4 minutes with 36 Snowflake round-trips — traced to R2RML query planning amplifying REST and Parquet costs, not to IO (every data read in every trace was a disk-cache hit). Each change is independently shippable and kill-switchable.What's here
Query planning
rdf:typeinto the same-subject star + prune subject-only scans: a subject-only/rdf:typepattern no longer projects every POM column, scans RefObjectMap parent tables it never reads, or fans out to predicate-sharing TriplesMaps. Simple Store query 6 scans → 1; Store ⋈ Geography 36 → 5.Catalog access
loadTableresponse reused across a query's scans, with the Iceberg snapshot pinned. OAuth 36 → 1,loadTable36 → 2 within a query.metadata_locationacross a mid-query credential refresh so a creds refresh cannot shift the query onto a newer snapshot.loadTablecache (60s TTL, vended-credential-expiry gated): on a warm server the 2nd query against a table skips OAuth +loadTableentirely.Scan execution
LIMIT nnow reaches the scan (GraphOperatorforwards the row budget through the R2RML subplan), so it stops after ~one materialize window instead of draining the table.Live results (Snowflake Horizon; results byte-identical to baseline)
Store; storeId; nameLIMIT 5Store ⋈ GeographyLIMIT 10FACT_ORDER_LINELIMIT 5Config (all default-on, kill-switchable)
Documented in
docs/operations/configuration.md:FLUREE_ICEBERG_LOADTABLE_CACHE,FLUREE_ICEBERG_LOADTABLE_TTL_SECS(default 60),FLUREE_R2RML_SCAN_CACHE,FLUREE_R2RML_LIMIT_PUSHDOWN,FLUREE_ICEBERG_PREDICATE_PUSHDOWN, plus the existingFLUREE_R2RML_MATERIALIZE_WINDOW_ROWSandFLUREE_ICEBERG_SCAN_CONCURRENCY.Testing
cargo fmt --all -- --checkclean;cargo clippy --all --all-features --all-targetsclean.LIMIT 5pulls ~1 of 40 streamed chunks); row-group pruning against a real two-row-group Parquet fixture.enterprisegraph source for every change, with byte-identical results pushdown on vs off.Not included
Row-level residual filtering + planner FILTER-consumption (so a
FILTER + LIMITfact query early-terminates) is deferred. Live testing surfaced thatxsd:integerR2RML columns are physically IcebergDecimal, and a residual filter needs conservative "keep on unevaluable" semantics distinct from the existing row matcher — a follow-up, scoped in the branch notes.