feat: materialize Iceberg/R2RML graph sources into native ledgers + tracking worker

[christophediprima]

Summary

Adds the ability to materialize an Iceberg / R2RML graph source into a native Fluree ledger and
keep it fresh with a background tracking worker. Querying a graph source directly re-reads and
re-joins the raw Iceberg rows on every request; materializing into a native ledger gives deduped,
indexed, time-travelable state that queries at native speed and refreshes incrementally.

Five commits:

1. Append-only incremental scan + snapshot-window helpers (fluree-db-iceberg).
2. R2RML incremental scan + snapshot helpers (fluree-db-api).
3. Materialize into native ledgers + tracking worker + endpoints (fluree-db-api, fluree-db-server).
4. Tombstone deletion + latest-by-key dedup for materialization.
5. Refreshable Google metadata-server catalog auth — so long-running tracking against a Google
   Iceberg REST catalog keeps authenticating past the ~1h a static bearer lasts.

Builds on the existing Iceberg graph-source support on main.

Motivation

A graph source exposes an Iceberg table as read-on-query triples. That's ideal for ad-hoc access,
but not for a dataset you query repeatedly:

• No dedup. Iceberg tables written by streaming/CDC pipelines are append-only — many row
  versions per entity. A graph source surfaces all of them; there's no "current state" of a subject.
• Re-read + re-join every query. Multi-pattern queries (?s a X ; p1 ?a ; p2 ?b …) join the raw
  append-only rows on each request, which blows up combinatorially and re-does the scan every time.
• No native indexes / time travel over the sourced data.

Materializing solves all three: collapse the append-only rows to one node per subject (upsert),
write them into a native ledger (native indexes, SPARQL, history), and refresh incrementally off
the source's snapshot log. The tracking worker automates the refresh.

Changes

• fluree-db-iceberg — incremental scan + snapshot windows (scan/planner.rs,
  scan/send_planner.rs, metadata/table.rs, config.rs): plan a scan over only the data files
  added between two snapshots (append-only incremental), plus helpers to resolve the current
  snapshot and the delta from a watermark. Falls back to a full scan when an incremental window
  isn't safe (e.g. a non-append snapshot). + unit tests.

• fluree-db-api — R2RML materialization (graph_source/r2rml_materialize.rs): read
  graph-source rows (full or incremental) and materialize them into a native ledger.
  Latest-by-key dedup collapses append-only rows to one node per subject (upsert); whole-subject
  tombstone retraction removes a subject when its key disappears from the source. A per-(source, table) watermark node (urn:fluree:materialize-state:{source}:{table}, source : escaped so
  the encoding is injective) persists the last-materialized snapshot id in the target ledger, so a
  refresh resumes incrementally; force_full ignores it. + unit tests (watermark encoding/injectivity,
  latest-by-key, retract shape).

• fluree-db-api — tracking worker (materialize_worker.rs): a per-job worker
  (TrackedJob { source, target, interval, next_due }) driven by a base-granularity ticker; each job
  re-syncs on its own poll interval. Public API: MaterializeTrackingWorker,
  MaterializeWorkerConfig / Handle / Stats. + unit tests (due-job selection + rescheduling).

• fluree-db-server — endpoints (routes/iceberg.rs, routes/mod.rs, state.rs): the four
  endpoints below. /track runs an immediate first sync on registration so the target is
  populated without waiting a cycle; /tracking is an admin-protected read of this node's worker
  state. Peer nodes forward writes to the write node, consistent with the existing iceberg routes.

• fluree-db-iceberg — refreshable catalog auth (auth/google_metadata.rs, auth/token.rs,
  auth/mod.rs): a google_metadata AuthConfig variant that mints and auto-refreshes catalog
  OAuth tokens from the GCE/GKE metadata server (Workload Identity). A CachedToken helper is
  extracted from the existing OAuth2 provider and shared, so the new provider is the same small shape
  as oauth2.rs (lean reqwest client + cache, no new dependencies). Exposed via
  auth_google_metadata on the map endpoint. + unit tests (wiremock: Metadata-Flavor header,
  scopes, caching, failure surfacing).

• fluree-db-api (graph_source/r2rml.rs): incremental-scan wiring on FlureeR2rmlProvider —
  prepare_iceberg_scan (shared REST/Direct storage + table-metadata setup over S3IcebergStorage),
  read_scan_tasks, current_snapshot_id, and scan_table_incremental.

• docs: materialize / track / untrack / tracking in docs/api/endpoints.md; a materialization +
  tracking section (incl. append-only dedup and the watermark) in docs/graph-sources/iceberg.md; the
  auth_google_metadata option.

New endpoints

Endpoint	Method	Body	Purpose
/iceberg/materialize	POST	{source, target, force_full?}	One-shot: read the source and upsert it into target. Returns {from_snapshot_id, to_snapshot_id, incremental, committed, rows_read, subjects_upserted, subjects_retracted}.
/iceberg/track	POST	{source, target, poll_interval_secs?}	Register a source → target job, run an immediate first sync, then keep target fresh on the interval (default 30s). Returns the effective interval + the first-sync result.
/iceberg/untrack	POST	{source, target}	Stop tracking (leaves already-materialized data in place).
/iceberg/tracking	GET	—	Worker status for this node (admin-protected read).

Auth

The tracking worker holds a job open for hours, so catalog auth has to survive token expiry. For a
workload running as a GCP service account (GKE Workload Identity / GCE), set
auth_google_metadata: true: tokens are minted and auto-refreshed from the instance metadata server,
so tracked jobs against a Google Iceberg REST catalog (e.g. BigLake) keep authenticating. A static
auth_bearer still works for one-shot map/query and for local use, but expires after ~1h.

Design note: the provider talks to the metadata endpoint directly rather than pulling a full
Application-Default-Credentials crate. That keeps fluree-db-iceberg lean and runtime-agnostic
(AuthConfig stays an explicit, declared config rather than ambient credential discovery), and it
mirrors the existing oauth2_client_credentials provider exactly — same struct shape, same
CachedToken, no new dependencies. It's precisely how a GKE workload authenticates. Storage-side
HMAC interop keys are unaffected — they don't expire.

Testing

• Unit + integration (Docker fluree-rust-dev, workspace pinned toolchain):
  • cargo test -p fluree-db-iceberg --lib → 148 passed (incremental scan / snapshot windows,
    google_metadata provider, CachedToken).
  • cargo test -p fluree-db-api --features iceberg --lib → 669 passed (watermark
    encoding/injectivity, latest-by-key dedup, worker due-job scheduling/rescheduling).
  • cargo test -p fluree-db-api --features iceberg --test grp_graphsource → 103 passed, incl.
    it_materialize_retract — a regression test that whole-subject retraction binds ?s as a typed
    @id (a bare-string binding parses to a literal that joins nothing and silently retracts zero
    rows) and asserts the triples are actually gone end-to-end.
• cargo fmt --all --check clean; cargo clippy --all-targets clean on the touched crates,
  with the iceberg feature both off and on (the materialize code is #[cfg(feature = "iceberg")],
  so it must be linted with the feature enabled).
• Manual end-to-end against a real Iceberg dataset behind a Google Iceberg REST catalog: registered
  R2RML mappings, tracked several tables into a single native ledger via /iceberg/track, confirmed the
  immediate first sync + an incremental re-sync on the worker (only added files re-read), and ran SPARQL
  over the deduped materialized ledger — one node per subject, expected counts.

Known limitations / follow-ups

• Worker state is in-memory per node. Tracked jobs are not persisted across restarts (they must be
  re-registered); persisting the ApplyRecord is noted as a follow-up in the routes. The materialized
  data and its watermark are durable (they live in the target ledger), so a re-registered job
  resumes incrementally rather than re-reading everything.
• Incremental scan is append-only. Snapshots that delete/rewrite files fall back to a full-window
  read; combined with latest-by-key upserts this is still correct (idempotent), just less cheap.
• google_metadata requires GCE/GKE (the metadata server isn't reachable elsewhere); locally, use
  a static auth_bearer.

Breaking changes

None. Everything is additive: four new endpoints, a new AuthConfig variant, and additive request
fields (poll_interval_secs, auth_google_metadata). All new server/API code is behind the existing
iceberg feature; nothing changes for builds without it.

[christophediprima]

Hi there! What do you think about this feature? Any chance it gets merged at some point? Do you think it is a good idea?

[christophediprima]

Update — rebased onto latest main, plus additive @type union

Rebased the branch onto current main (v4.1.3) — conflict-free. This adapts to the Iceberg→R2RML
catalog engine that merged in the meantime: delete / order_by now sit on IcebergCreateConfig (materialization options, not connection concerns) under the new IcebergConnectionConfig split, and the refreshing providers share a single CachedToken in auth::token (keeping the redacting Debug).

Added one materialization-correctness commit: rdf:type now UNIONs across sources in additive mode. Previously each subject was applied with a single upsert (which retracts-then-inserts per predicate), so when more than one source contributed to the same subject IRI — a shared target ledger fed by several graph sources, a join table that only adds an edge to a parent entity, or a
secondary-type table — whichever source wrote last clobbered the others' @type. Now additive-mode materialize asserts @type via an idempotent insert and upserts only the non-type predicates, so classes accumulate instead of overwriting. A rr:predicate rdf:type object map is likewise routed to
the subject's @type (the only way to express per-row, data-driven typing, since rr:class is constant-only) rather than being treated as an ordinary predicate. Covered by new tests: a type-union case, a control proving the single-upsert path clobbers, and a build_live_node unit test for the
rdf:type-as-class routing.

Relationship to #1529 (feat/materialize-builder, DEC-003 D1)

#1529 lands the bulk "native twin" builder: a whole-graph R2RML enumerator feeding the native bulk-import pipeline, a post-build parity gate, and a fluree materialize CLI. It stamps each twin's final commit with a WatermarkStamp — per-table pinned Iceberg snapshot (metadata_location + snapshot_id/sequence_number) + mapping hash + builder version, readable via read_stamp. That is a one-shot build; DEC-003 calls out a separate Deliverable 3 — delta-sync for keeping a twin fresh, which isn't implemented there yet (the per-table snapshot_id/sequence_number are captured in
the stamp precisely so delta-sync can diff).

This PR is the complementary half: incremental refresh + a tracking worker + HTTP endpoints (/materialize, /track, /untrack, /tracking). Its append-only snapshot-window scan (plan_incremental over (from, to]) is exactly the delta-sync operation D3 describes — read the twin's WatermarkStamp for each table's from snapshot, scan only the files added since, and upsert the delta into the twin (with the @type union above so a re-typed subject isn't clobbered).

Happy to reframe this PR as the delta-sync layer on top of #1529's twin rather than a parallel bulk path — reusing #1529's WatermarkStamp / build_watermark as the shared contract and dropping any overlap in graph_source/{catalog_session,r2rml}.rs. Flagging the overlap now so we can decide how the two should compose.

[christophediprima]

Update — named-graph routing in the materializer (rr:graphMap / per-graph statements)

The materializer now honors R2RML rr:graphMap / rr:graph at the subject-map level: each source row's triples are placed into a named graph derived from the row, instead of always the default graph. No graph map ⇒ default graph (existing behavior, unchanged).

Mechanics — no transaction-core changes (named graphs are already first-class: flakes carry a graph id, staging keys on (GraphId, Sid), and generate_upsert_deletions already scopes retract-existing by graph_id):

• fluree-db-r2rml — a GraphMap term map (rr:graph constant shortcut, or rr:graphMap [ rr:template | rr:column | rr:constant ]) parsed onto SubjectMap; materialize_graph_from_batch resolves the per-row graph IRI, reusing the exact template/column expansion the subject materializer uses.
• fluree-db-api (r2rml_materialize.rs) — the accumulator now keys on (graph, subject), so the same IRI in two graphs is two independent keys. Live nodes are emitted with a per-node @graph selector ({"@id": <s>, "@graph": "<g>", … }), which the insert/upsert transaction parser resolves per node — scoping every triple (including @type) to that graph; default-graph nodes (and the materialization watermark — bookkeeping, kept in the default graph) stay plain top-level. Whole-subject retraction (latest-by-key mode) is scoped to each graph via the UPDATE graph key; the upsert's own retract-existing is already graph-scoped, so per-graph subjects never clobber across graphs.

Additive mode composes unchanged: @type still UNIONs and predicates still upsert — but now within each subject's graph. Tests cover rr:graphMap/rr:graph parsing, (graph, subject) isolation of the same IRI across graphs (record + merge_live), the graph-scoped retract doc, per-graph doc flattening, and an end-to-end round-trip through the transaction parser (it_materialize_named_graph) that inserts/upserts the emitted per-node @graph shape into a memory ledger and reads it back per graph — asserting each subject's triples land in its named graph (not the default) and stay isolated across graphs.

[christophediprima]

Update — templated materialize target (fan out into a ledger per partition) + a shared watermark state ledger

I added a templated materialize target so one tracking job can fan out into many target ledgers, resolved per row from the source's own columns. Motivation: I need per-user isolation (each user sees only their own data), not just per-tenant. Fluree's read-policy engine is deliberately graph-blind (it targets subjects/properties/classes, never named graphs), and the docs recommend separate ledgers for separate access regimes — so I make the isolation boundary the ledger, one per (tenant, user), which rides the existing bearer-scope (can_read) at the maintainer-endorsed granularity rather than inventing graph-as-authorization. (The rr:graphMap named-graph routing from the previous section stays as an orthogonal capability; it is just not what I use for per-user access isolation.)

How it works. The target on /iceberg/materialize and /iceberg/track may now contain {column}
placeholders (e.g. silver_{tenant_id}_{user_id}:main). I resolve the concrete target ledger per row by expanding that template against the row's columns (reusing the same expand_template_from_batch the subject/graph materializers use), so a single scan of the source routes each row into its own ledger. A placeholder-free target is the existing single-ledger behavior, unchanged — this is fully backward compatible. Concretely: the latest-by-key accumulator now keys on (target, graph, subject) instead of (graph, subject); after finalizing I group the state by target ledger and run each target's retract/@type-insert/predicate-upsert as its own independent commit chain (creating the ledger on
first sight — correct for append-only CDC, where a user's rows only arrive once the user exists). A row whose template columns are null can't be routed and is skipped.

Watermark moved to a shared state ledger. Previously the per-(source, table) watermark was written as a triple inside the target ledger's default graph, mixed with materialized entity data. With fan-out there is no single target — and the incremental scan needs the watermark before it discovers which per-row targets exist. So I moved the watermark into one shared fluree_materialize_state:main ledger, keyed by (source, target-spec, table) (the target-spec is the template for a fan-out job, so one templated job keeps one watermark per source table — a single scan feeds all its targets). This also removes bookkeeping triples from user data. I advance the watermark after every target's data commit, never before, so a crash in the gap re-reads the window (safe because the writes are idempotent:
whole-subject replace / idempotent insert+upsert). This unifies plain and templated jobs on one generic, job-linked state location instead of storing the watermark in two different places.

Tests: a new record_isolates_same_subject_across_targets (the same IRI in two target ledgers is two independent keys), the watermark helpers updated for the (source, target-spec, table) key with an injectivity test across all three segments, and the existing single-target + named-graph suites still green (19 unit + 8 materialize-integration). --features iceberg; fmt + clippy (-D warnings) clean.

[christophediprima]

Update — don't upsert an empty doc for type-only sources

A type-only source (subject + rdf:type, no other predicates) produces an empty additive-mode predicate doc, and I was upserting it unconditionally. The transactor rejects an empty upsert, which aborted the sync before its watermark advanced — so that source re-scanned and re-committed the same @type triples on every poll (churn). Fix: guard the upsert with if !pred_doc.is_empty(), mirroring the existing @type-insert guard (same guard added to the latest-by-key delete-only path). +1 test; clippy/fmt clean.

[christophediprima]

Update — tracking jobs survive a restart

Closes the "worker state is in-memory" limitation above, which turned out to be worse in practice than it reads. A restart silently stopped every materialization until a client noticed and re-issued POST /iceberg/track — and nothing surfaced the gap: GET /iceberg/tracking reported the worker running with zero jobs, which is indistinguishable from "nobody has tracked anything yet". For a fan-out job feeding a ledger per partition, that is an outage that looks like an idle system.

Jobs now persist as urn:fluree:materialize#Job rows in the shared fluree_materialize_state:main ledger — the store this worker already owns, already writes on every committed poll, and which already carries each job's source and target-spec on its watermark rows. MaterializeTrackingWorker::run restores the set before its first tick.

• The job is stored as one serialized JSON blob, plus duplicate source/target triples so operators can query jobs with the same SPARQL they already use for watermarks. The blob keeps the restore decoder trivial (the only existing state-ledger reader is extract_first_i64, which cannot decode N rows × M columns) and lets a job gain fields without a schema migration. It is also self-validating on read: only a string that deserializes into a whole PersistedMaterializeJob is taken, so the sibling triples on the row can never be mistaken for one.
• Restore lives in run(), not in the server's construction path, which inherits peer-exclusion for free — peers never spawn a worker, and /iceberg/track already 400s on them.
• Recovery is incremental. The per-(source, target-spec, table) watermarks live in the same ledger and are read by the materialize call itself, so a restored job re-reads nothing it had already materialized. A restored job's first poll fires one interval from now, matching track.
• untrack is durable too — it sets tracked = false rather than retracting the row, keeping an audit trail and avoiding a delete path. Restore filters on that flag.

Why not the graph-source record, which would have been the cheaper mirror of how the BM25 maintenance worker persists its own tracked flag: IcebergGsConfig has no catch-all field and both create paths serialize a fresh struct (r2rml.rs:195-198, :290-293), so any key written onto the record is dropped on the next config write; and rest_client_cache_key hashes the raw config JSON (r2rml.rs:69-71), so writing to it would discard the cached RestCatalogClient and its OAuth token on every change. A job is also a relationship (source → target-spec), not a property of the source — one source can have several targets, which the record shape handles badly.

Also fixes a latent race the new writes would have made hotter: opening the state ledger was a bare check-then-create, and the loser of a race between two pollers got ApiError::LedgerExists instead of a ledger, failing the whole materialization pass. Both call sites now share one helper that tolerates it.

Tests (fluree-db-server/tests/iceberg_track_durability.rs, +4): a job tracked before a restart is running after it with its own interval intact and nothing re-issuing track; an untracked job stays gone across a restart; re-tracking the same pair updates in place rather than accumulating rows; and a server that has never tracked anything treats the absent state ledger as "no jobs" rather than an error that kills the worker. The tests drive the durable record directly rather than through POST /iceberg/track, since that route also runs an immediate materialize and would need a reachable catalog — the restore path under test is the real one either way. clippy/fmt clean.

[aaj3f]

🔄 Request changes — with genuine appreciation for the work, and a concrete proposal below for how we get all of it landed.

@christophediprima this is a substantial contribution, and the sustained engagement shows — the rejected-alternatives reasoning in the job-persistence commit and the churn analysis in the empty-upsert commit are the kind of commit bodies I wish more PRs had. Your core premise also survives scrutiny: I verified fluree-db-policy has no graph-scoped enforcement, so you are right that per-ledger scope is the real isolation boundary today. And the CDC semantics you've worked out — tombstone deletion, latest-by-key, the additive rdf:type union, watermarks, restart-surviving jobs — are exactly the hard-won details this capability needs.

Five things block the PR in its current shape; the inline comments carry the detail and the suggested fixes. (1) plan_incremental reimplements the Iceberg manifest walk and so misses the merge-on-read fail-closed guard that merged in #1520 roughly nine hours after your rebase — and materialization raises the stakes, because a caveated query result becomes permanent, uncaveated state in a ledger the watermark never revisits. (2) The materialize pass buffers the whole table, the whole subject accumulator, and the full JSON-LD expansion simultaneously inside the server process, so the first sync of a large CDC table — the stated use case — can take query serving down with it. (3) Target-ledger creation is a check-then-create race; your newest commit fixed exactly this for the state ledger, and the target path needs the same tolerant fallback. (4) Mutation testing shows no test calls the materialization engine: reintroducing the exact bugs that commits cf472f9ab and a985bc6b5 fix leaves all 28 tests green. (5) rr:graphMap silently drops POM-level graph maps, repeated graph maps, and rr:defaultGraph, which collapses two tenants' rows into one accumulator key — the exact clobbering named-graph routing exists to prevent.

Now the structural part, which I'd rather be transparent about than let surface awkwardly later: we have an internal draft (#1529) exploring the same native-twin capability through the bulk-import finalizer rather than transaction replay. That architecture resolves the memory blocker structurally — chunked, bounded, parallel ingest with serial finalization — rather than by adding knobs to the replay path. Neither effort should win by default, but having now reviewed both closely, I think the right landing is: the bulk-builder as the engine foundation, with your CDC semantics on top of it — #1529 has none of the tombstone/latest-by-key/watermark/job-persistence thinking this PR gets right, and that half is the harder half to get right.

So here is my concrete proposal for landing this work rather than orbiting it. Split the PR: (1) the Google metadata-server auth commit (10e27961a) is self-contained, fixes a real expiring-token bug, and I would approve it in its own PR today; (2) rr:graphMap named-graph routing stands alone as a genuine R2RML vocabulary gap — with the fail-loud fixes from the inline comment; (3) the append-only incremental scan stands alone once it routes through the shared MoR-guarded planner; (4) for the materialization engine itself, let's converge on the bulk-import foundation and port your CDC semantics onto it — I'd genuinely value doing that part together, since you've thought harder about CDC ordering and deletion semantics than anyone else on this codebase. The ledger-per-(tenant,user) fan-out deserves its own design discussion before it ships under either engine, since it converts a policy gap into unbounded ledger multiplication and graph-scoped policy may be the better long-term investment.

One housekeeping items (somewhat implied by the above). The branch currently has 5 conflicts against main, two of them exactly where #1520's guard exports live, so the MoR interaction should be resolved deliberately rather than by whatever conflict resolution happens to produce.

Adherence to repo commitments:

• Patterns/abstractions: ✖ Reinvents bulk ingest instead of extending the memory-bounded chunked path in fluree-db-api/src/import.rs, adds a second Iceberg manifest walk instead of routing through the guarded planner, and implements a partial rr:graphMap that silently drops the R2RML constructs it does not support.
• Performance (speed first, memory second): ✖ CRITICAL — unbounded buffering on a task spawned inside the server process. No query hot path is touched.
• Testing: ⚠️ The new tests run (correctly wired into grp_graphsource.rs) and the durability suite is excellent, but the engine itself has zero coverage — two mutations reintroducing shipped bugs leave all 28 tests green.

[aaj3f]

Blocking. CRITICAL: this is a third Iceberg read entry point that will not inherit the merge-on-read fail-closed guard, and the merge that introduces the gap is textually clean.

plan_incremental reimplements the manifest-list walk instead of routing through plan_scan_for_snapshot_inner. Compare :219 (parse_manifest_list) and :226 (if manifest_entry.is_deletes() { continue; }) against the identical two lines at :94/:109 in the full-scan path. PR #1520 ("fail closed on merge-on-read delete files", still open on fix/iceberg-mor-delete-guard) installs mor_guard::ensure_no_summary_deletes + ensure_no_delete_manifests at exactly those two points in plan_scan_for_snapshot_inner, in both planners — and nowhere else. I verified git merge-tree --write-tree HEAD 771f6e88b returns exit 0 with no conflict, so whichever of these two merges second, git will report nothing and the incremental path will silently ship ungated.

To be fair about the exposure: the common MoR timeline is caught, because MoR delete files arrive under overwrite/delete snapshot operations, window_is_incremental_safe rejects those, and the fallback goes through plan_scan which #1520 does gate. The residual holes are (a) replace snapshots, which the Iceberg spec defines as "data and delete files were added and removed" and which table.rs:187 explicitly safelists, and (b) any writer whose summary operation string understates what the snapshot did. Both produce a materialized native twin that durably contains rows the source considers deleted — and unlike a virtual query, a twin is not re-read, so the wrong answer persists and is served at native speed as ground truth.

Suggested fix: hoist the manifest-list load + guard into one shared helper used by all three planners, or at minimum add the same two calls at the top of plan_incremental:

let allow_mor = crate::mor_guard::mor_deletes_allowed();
crate::mor_guard::ensure_no_summary_deletes(to_snapshot, &self.metadata.location, allow_mor)?;
// ...then parse_manifest_list_with_deletes + ensure_no_delete_manifests, as in plan_scan_for_snapshot_inner

Whichever PR lands second should carry this. Worth coordinating explicitly with #1520 rather than leaving it to the merge.

[aaj3f]

Blocking. A 1365-line engine plus a 381-line worker ship with zero end-to-end test coverage, and the tests that claim to cover them do not.

grep -rn "materialize_r2rml_graph_source" --include=*.rs . finds no test caller anywhere in the repo — only routes/iceberg.rs:288,389 and materialize_worker.rs:277. The three new it_materialize_*.rs files are wired correctly into grp_graphsource.rs:18-23 and their 7 tests do run (confirmed by name under --all-features), but they hand-copy the JSON shapes rather than calling the code that produces them — it_materialize_retract.rs:38 even says so: "mirrors build_retract_doc".

I proved this is not theoretical. Reverting build_retract_doc:699 to the bare-string binding — the exact bug the PR body says it_materialize_retract is a regression test for — turns the unit test build_retract_doc_shape red (so there is a real guard, credit where due) but leaves all 7 PR-new integration tests green, materialize_retract_shape_actually_retracts included. The PR body's claim that it "asserts the triples are actually gone end-to-end" is true of the shape and false of the producer.

What's missing is a test that drives materialize_r2rml_graph_source itself. That needs an Iceberg source, which is the hard part — but the accumulator → doc → transaction → query round trip could be exercised against a memory ledger with a stubbed scan_for_materialize, and that is where every one of the bugs in this PR's own commit history lived (the bare-string retract, the @graph envelope, the empty-upsert abort). Three shipped-then-fixed bugs in the same seam is the argument for the test.

[aaj3f]

Blocking. CRITICAL (performance/memory): the materialize path buffers the entire table, twice, with no streaming or byte budget.

read_scan_tasks does try_collect::<Vec<Vec<ColumnBatch>>>(), so every Parquet file in the plan is decoded and held before a single row is processed. materialize_r2rml_graph_source:234-288 then accumulates one SubjectNode per distinct subject across all tables into MaterializeAccum, and :398 serializes the whole set into a single JSON-LD array per target. build_retract_doc:696 has the same shape — every subject IRI in the window goes into one values array in one update call.

An initial materialization (from = None → full scan, which is also what force_full and every expired/branched-history fallback take) therefore holds the decoded table plus its full JSON-LD expansion simultaneously. This does not degrade gracefully on a large table; it OOMs. Given the PR's stated motivation is large CDC tables, that ceiling deserves at least a documented row limit, and ideally chunked commits keyed off a byte budget.

Note the query engine itself is unaffected — I verified plan_scan_for_snapshot_inner and r2rml.rs::scan_table are byte-unchanged and the new helpers have no query-path callers, so there is no virtual-query regression.

[aaj3f]

Blocking. rr:graphMap is honored only by the materialize path, so the virtual and materialized views of the same source disagree about where the triples live.

grep -rn "graph_map" across fluree-db-query/ and fluree-db-api/ returns exactly one consumer: this line. The query path never reads SubjectMap::graph_map. So for a mapping with a subject-level rr:graphMap, querying the graph source virtually returns every triple in the default graph, while the materialized twin puts them in per-row named graphs — and a query written against one returns nothing against the other. That breaks the PR's own premise that materializing gives you the same data at native speed, and it is silent.

Either teach the query path to honor graph maps too (parity), or gate materialization on the mapping having no graph map until it does.

Separately, R2RML permits rr:graphMap on predicate-object maps as well as subject maps; extractor.rs:extract_graph_map only reads the subject-map position, so a POM-level graph map is silently ignored rather than rejected — that one should at minimum fail loud.

[aaj3f]

Optional, but please confirm the deployment story. (Covers :285-287.)

The code comment concedes it: "In Raft mode the worker currently runs on every node and writes directly via Fluree::upsert; gating it to the leader is a follow-up." Every non-Peer node spawns its own worker with its own in-memory job set.

If a multi-node non-Peer deployment is reachable today, two workers polling the same (source, target) will interleave their retract and upsert commits — and since the retract at r2rml_materialize.rs:376 and the upsert at :400 are two separate transactions, node A's retract landing after node B's upsert leaves subjects missing until the next poll.

The same race exists between the worker and a concurrent POST /iceberg/materialize for the same pair even on a single node, since nothing serializes them. A per-(source, target) mutex around the pass would close the single-node half cheaply.

[aaj3f]

Design discussion — worth resolving before the fan-out ships; non-blocking for the rest.

First, your stated premise is correct, and I verified it rather than assuming: fluree-db-policy/src/ has no graph-scoped enforcement (12 "graph" references, all about the policy model graph; graphs: None at index.rs:327,545). Named graphs genuinely are not an authorization axis here, so per-ledger bearer scope genuinely is the isolation boundary. The reasoning in eb62eaa4e is sound.

The open question is whether ledger-per-(tenant,user) is the right answer to that constraint: it converts a policy gap into unbounded infrastructure multiplication — every partition gets a nameservice record, a commit chain, and its own index set, created implicitly from row data with no bound, no quota, and no operator visibility beyond the ledger list. It also puts the PR in the odd position of shipping two isolation mechanisms (named graphs and separate ledgers) where only one can be the boundary. I'd like us to weigh graph-scoped policy as the longer-term investment before the fan-out lands, rather than settling it implicitly inside this PR.

[aaj3f]

Optional.

A permanently failing sync is a warn! plus a syncs_failed counter, so a target ledger can diverge from its source indefinitely while GET /iceberg/tracking still reports the worker healthy. Given the MoR refusal recommended in the scan-path comment will make exactly this happen post-merge for MoR tables, consider surfacing per-job last_error and last_success timestamps on the tracking endpoint.

[aaj3f]

Optional.

The "Assumptions and limitations" section is genuinely good on CDC shape (one row per subject revision, one triples map per table, orderable order_by, dedicated target, tombstones required). It omits four things the code does: merge-on-read is unhandled on the incremental path, the full sync is unbounded in memory, the worker runs on every non-peer node so multi-node deployments double-materialize, and a templated target creates ledgers without bound. #1520 added an MoR section to this same file on main — the materialization section should cross-reference it after the rebase.

[aaj3f]

Praise.

This is the strongest-tested part of the PR: four tests that drive the real persist_materialize_job / forget_materialize_job / tracked_materialize_jobs API and the worker's actual restore path, including the negative cases (untracked jobs stay out, re-tracking replaces rather than duplicates, restore is a no-op with no state ledger). This is exactly the pattern the materialization engine tests should follow.

[aaj3f]

Praise.

The four new routes are correctly mounted on the existing v1_admin_protected_writes / v1_admin_protected_reads layers rather than inventing a new auth path, so they inherit admin-token enforcement and leader-forward ordering for free. Correct altitude.

[christophediprima]

Taking the split — it's the right call, and thank you for laying it out that concretely rather than just blocking. Answering per part.

Part 1 — up as #1567

10e27961a cherry-picked onto current main with one conflict (upstream's with_auth_bearer_token_ref landed at the same insertion point as with_auth_google_metadata — resolved as a union, neither modified), plus one adaptation: hydrate at auth/mod.rs:189 needed a GoogleMetadata arm, written out explicitly rather than as a wildcard so the match keeps failing compilation when a variant appears.

Your wildcard-free canary in ledger_info.rs then caught what I'd missed — a new AuthConfig variant has to earn an explicit redaction decision and I hadn't made one. For the record it needs no allowlist entry: both fields are plain String rather than ConfigValue, so there's no secret-capable field to seed, and the credential is minted per call and only ever lives in the in-memory CachedToken, which isn't serialized. I added the variant to the fixture anyway so it runs the assertions with the rest. Good guard — it failed me in precisely the way it was written to.

Part 2 — rr:graphMap routing — up as #1581

Agreed it stands alone, so I've sent the vocabulary layer: #1581, fluree-db-r2rml only, 7 files. It carries every fail-loud fix from your inline comment — POM-level graph maps, more than one graph per term map, and rr:graph rr:defaultGraph all now fail at mapping-load time with a ConfigError naming the construct, instead of being dropped into the default graph and collapsing two partitions into one accumulator key.

One I added beyond your list, because it reaches the identical failure by another route: a rr:graphMap with no rr:template, rr:column or rr:constant was treated as absent rather than malformed.

I deliberately left out the 313 lines of r2rml_materialize.rs from 7979c338f — that's engine consumption and belongs with part 4. Splitting at the crate boundary is what makes this the standalone vocabulary gap you described.

Two things #1581 does not resolve, both called out in its body rather than left for you to find:

Your virtual/materialized divergence finding. The query path still never reads SubjectMap::graph_map. Worth flagging a trap in your second option though — "gate materialization on the mapping having no graph map" would disable exactly the feature this enables, so parity looks like the real requirement rather than a choice between two. That's query-engine work and I haven't sized it; happy to, if you agree that's the direction.

graphMap routing and BM25 are mutually exclusive today. The indexing query runs at ledger.as_graph_db_ref(0) (bm25.rs:241, :259) and DocKey is {ledger_alias, subject_iri} with no graph field (bm25/index.rs:29-34), so anything routed into a named graph silently disappears from full-text search. Pre-existing rather than introduced by part 2 — but part 2 is what makes it reachable. Would you rather I document the limitation, or is a graph-aware DocKey something you'd take? The latter is a snapshot format change plus a forced rebuild of every existing index, so I'd want your call before starting rather than presenting you with one.

Also up: #1580, a policy bug found while auditing how far the graph identifier travels through the read path. PolicyContext::class_cache is keyed on Sid with no graph, so the same subject IRI in two named graphs with different types resolves to whichever populated the cache first and f:onClass then decides on the wrong classes — order-dependent behaviour in an authorization path. It's reachable by any from-named user, not just through R2RML; docs/security/policy-in-queries.md:203-205 actually recommends f:onClass for exactly the graph-differentiated case it corrupts. Sent separately since it's pre-existing and independent of this PR. Two smaller siblings in the same family — two commit/block paths that hardcode graph 0, and the read-path condition executor pinned to g_id 0 — I'll send the same way rather than clutter this thread.

Part 3 — append-only incremental scan

Agreed, and it's blocked exactly where you said. You were more right than you knew about the housekeeping note. I attempted the rebase to size it: four of the five conflicts are import-list unions, but the fifth isn't a conflict at all — it's a compile error, send_planner.rs:235: cannot find function parse_manifest_list. plan_incremental calls a function #1520 restructured away when it introduced the MoR-guarded planner. So finishing the rebase is the fix, there's no mechanical resolution, and I'd have papered straight over it if I hadn't compiled the result. I've reset the branch to what you reviewed rather than push a rebase that hides that.

Part 4 — the engine

Yes, and thank you — I'd much rather converge than have two implementations orbit each other, and I'll take you up on doing it together. Two things I'd want to carry across explicitly, because they're the easiest to lose in a port and the most expensive to rediscover:

• The retract-then-upsert pair is ordered, and its crash window is documented as self-healing rather than merely tolerated. If the finalizer reorders or batches those, the guarantee changes shape.
• The watermark deliberately never rides in the same commit as the data. It advances after every target's data commit, never before, so a crash in the gap re-reads the window — safe only because the writes are idempotent.

Point me at #1529 whenever it's shareable and I'll start by mapping our CDC semantics onto its seams rather than proposing a shape.

On the fan-out, and on graph-scoped policy

Agreed it deserves its own discussion, and I'll bring numbers. We've run the ledger-per-(tenant,user) shape in anger, and it bends in three specific places:

1. Boot is O(tenants). preload_all_ledgers structurally loads every non-retracted ledger at startup — the warm budget caps page warming, not the structural load.
2. reindex_max_bytes has no aggregate ceiling. It's the hard threshold that blocks commits, and it's set per LedgerState, so the safe per-ledger value has to shrink as tenants grow and nothing surfaces that. At 64 MiB and 1000 ledgers the worst case is 62 GiB of novelty.
3. BM25 maintenance saturates first — at our data density, somewhere around 25-50 concurrently-active users — because the "incremental" sync re-runs the whole indexing query. It's also the only one of the three that gets worse as data grows at a fixed tenant count.

On graph-scoped policy specifically — I went and sized it, because it's what I originally wanted and I'd have argued for it if I thought it carried its weight. I don't think it does, at least not as a fix for those three, and I'd rather say so than sell it.

The mechanism is fine: graph is a physical partition rather than a key column (run_record.rs:231), so restricting a scan to one graph is cheaper than a prefix seek, and g_id already reaches every read enforcement site — it's dropped at exactly one boundary, policy/enforcer.rs:102-109. The materializer's @graph IRI would resolve and match an f:onGraph target exactly.

What it doesn't do is fix (1), (2) or (3) — collapsing 1000 ledgers into 1000 graphs in one ledger trades each of them for a different problem rather than removing it:

1. Boot — the N ledger loads go away, but store construction eagerly fetches every named graph's four branch manifests in a serial await loop (binary_index_store.rs:316-330), so it becomes O(4xtenants) sequential CAS round-trips on every cold open. Net win only once that loop is lazy or concurrent. There's also a hard u16 ceiling at ~65,533 graphs with no cap check.
2. Novelty budget — the 62 GiB aggregate worst case does collapse, which is a real gain. But 1000 independent budgets become one shared 64 MiB budget, so a single busy tenant applies commit backpressure to every other tenant. Lower ceiling, much wider blast radius.
3. BM25 — this one gets actively worse, for the DocKey reason under part 2: routed data isn't indexed at all today, and a naive shared DocKey would merge two tenants' text into one document. And the saturation itself is per-index maintenance work, which consolidating ledgers doesn't reduce.

So f:onGraph is worth doing on its own merits, but not as multi-tenancy plumbing, and I don't want to argue for it on a benefit it doesn't deliver.

If you do want it, three decisions look like yours rather than mine:

• for_graphs as an orthogonal narrowing field mirroring for_classes, not a TargetMode::OnGraph variant. Under the narrowing form the covers_predicate fast path (types.rs:504-521) stays sound by construction. Under a variant with its own bucket it over-approximates the wrong way, and the two non-exhaustive matches!(target_mode, OnProperty|OnSubject|OnClass) sites take the fail-open branch.
• Fail-closed semantics for unresolved graph targets. Target-mode precedence falls through to TargetMode::Default when all target sets are empty (policy_builder.rs:1058-1068), and Default applies to every flake. A graph has no GraphId until something is written to it — so f:allow + f:onGraph <not-yet-created> wouldn't be inert, it'd be blanket allow on the ledger. That contradicts the for_classes convention, which is why it needs your call rather than my guess.
• A written position on the write side. WriteFlakeInfo has no graph field (fluree-db-policy/src/types.rs:172-196) and a transaction naming a new graph IRI implicitly registers it, so graph would be a confidentiality boundary with no integrity half. Fine for my case — only the materializer writes — but it should be stated rather than discovered.

Worth noting the docs currently point the other way: docs/security/policy-in-queries.md:201-205 says to use separate ledgers for wholly separate access regimes. I read your comment as revisiting that, but I'd rather name the contradiction than let it surface later.

Blockers

2–5 I'll pick up in their respective parts. Blocker 4 is fair and slightly embarrassing — mutation testing showing all 28 tests green with the engine's bugs reintroduced is exactly the check I should have run myself.