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

docs/api/endpoints.md

@@ -2542,11 +2542,16 @@ POST http://localhost:8090/v1/fluree/iceberg/map
 | `r2rml` | string | Inline R2RML mapping (Turtle/JSON-LD). Omit to auto-generate a direct mapping. |
 | `r2rml_type` | string | Media type of `r2rml` (`text/turtle`, `application/ld+json`) |
 | `branch` | string | Branch name (default: `main`) |
-| `auth_bearer` | string | Bearer token for catalog auth |
-| `oauth2_*` | string | OAuth2 client-credentials flow for the catalog |
+| `auth_bearer` | string | Static bearer token for catalog auth (does not refresh — a Google OAuth token will expire after ~1h) |
+| `oauth2_*` | string | OAuth2 client-credentials flow for the catalog (refreshes) |
+| `auth_google_metadata` | bool | Use the GCE/GKE metadata server (Workload Identity) for catalog auth, minting + auto-refreshing tokens — for Google Iceberg REST catalogs (BigLake). Overrides `auth_bearer`. Only works when running on GCP. |
+| `auth_google_scopes` | string | Optional OAuth scopes for `auth_google_metadata` (default `cloud-platform`) |
 | `warehouse` | string | Warehouse identifier |
 | `no_vended_credentials` | bool | Disable vended credentials |
 | `s3_region`, `s3_endpoint`, `s3_path_style` | | S3 overrides for `direct` mode |
+| `order_by` | string | Latest-by-key ordering column for materialization (int/date/timestamp) |
+| `delete_column` | string | Column that marks a row as a delete (tombstone) during materialization |
+| `delete_values` | (string\|null)[] | Values of `delete_column` that mean "deleted"; a `null` entry matches a NULL column (null-payload delete). Required when `delete_column` is set. |
 
 **Response:**
 
@@ -2570,6 +2575,78 @@ POST http://localhost:8090/v1/fluree/iceberg/map
 
 See also the CLI wrapper: [fluree iceberg map](../cli/iceberg.md).
 
+### POST {api_base_url}/iceberg/materialize
+
+Materialize a graph source into a native ledger (so BM25 / vector / reasoning can run over it). Reads incrementally from a per-table watermark persisted in the target, or fully with `force_full`. Admin-protected; `iceberg` feature only. See [Materialization](../graph-sources/iceberg.md#materialization-into-a-native-ledger).
+
+**Request Body:**
+
+```json
+{ "source": "orders:main", "target": "orders-native:main", "force_full": false }
+```
+
+| Field | Type | Description |
+|-------|------|-------------|
+| `source` | string | Source graph source id to read (required) |
+| `target` | string | Target native ledger to write (required; created if absent) |
+| `force_full` | bool | Ignore the watermark and re-read the whole table (default `false`) |
+
+**Response:**
+
+```json
+{
+  "source": "orders:main",
+  "target": "orders-native:main",
+  "from_snapshot_id": null,
+  "to_snapshot_id": 5648190075564901028,
+  "incremental": false,
+  "committed": true,
+  "rows_read": 1200,
+  "subjects_upserted": 1200,
+  "subjects_retracted": 0
+}
+```
+
+`committed: false` means a no-delta poll (nothing changed since the watermark). `from_snapshot_id`/`to_snapshot_id` are populated for single-table mappings (multi-table watermarks live per-table in the target ledger).
+
+**Status Codes:** `200 OK`; `400` invalid request / config; `401/403` admin auth; `500` scan or commit failure.
+
+### POST {api_base_url}/iceberg/track
+
+Register a `source → target` materialization job with the in-process tracking worker, run an immediate first sync, and keep the target fresh on a timer. Admin-protected; `iceberg` feature only.
+
+**Request Body:**
+
+```json
+{ "source": "orders:main", "target": "orders-native:main", "poll_interval_secs": 300 }
+```
+
+| Field | Type | Description |
+|-------|------|-------------|
+| `source` | string | Graph-source id to read. Required. |
+| `target` | string | Native ledger to keep materialized. Required. |
+| `poll_interval_secs` | number | This job's own re-sync cadence, in seconds. Optional — defaults to the worker's configured interval (30s). Must be > 0. Each tracked job runs on its own timer. |
+
+**Response:** `{ "source", "target", "tracked": true, "poll_interval_secs": 300, "tracked_jobs": 1, "initial": { …materialize response… } }`
+
+Tracked jobs are held in memory (re-issue `track` after a restart; the watermark in the target ledger means it resumes incrementally). The worker runs on write nodes only.
+
+### POST {api_base_url}/iceberg/untrack
+
+Stop tracking a `source → target` pair (already-materialized data is left in place). Body: `{ "source", "target" }`. Response: `{ "source", "target", "removed": true, "tracked_jobs": 0 }`.
+
+### GET {api_base_url}/iceberg/tracking
+
+List the tracking worker's jobs and cumulative stats.
+
+```json
+{
+  "running": true,
+  "jobs": [ { "source": "orders:main", "target": "orders-native:main", "poll_interval_secs": 300 } ],
+  "stats": { "polls": 12, "syncs_committed": 1, "syncs_noop": 11, "syncs_failed": 0, "tracked_jobs": 1 }
+}
+```
+
 ## Admin Endpoints
 
 ### POST /reindex

docs/graph-sources/iceberg.md

@@ -237,7 +237,9 @@ export AWS_SECRET_ACCESS_KEY=<gcs-hmac-secret>
 
 A signing region is required and must match the bucket location — SigV4 scopes the signature to a region, and GCS interop rejects a mismatched or unsigned region. Set it via `s3_region` in the config (recommended, and what the examples use) or via the ambient `AWS_REGION` in the server environment. `s3_endpoint` must be the interop host, and `s3_path_style` must be `true`. HMAC keys do not expire; in `rest` mode, credentials vended by the catalog for a GCS-backed table are used instead (and refreshed by the SDK).
 
-GCS-backed Iceberg tables are typically read via `direct` mode — point `table_location` at the table root. GCS-native conventions are handled automatically: `gs://` paths in metadata/manifests, a Hadoop-style integer `version-hint.text` (resolved to `vN.metadata.json`), and Snappy-compressed Parquet. As with any direct-mode table, the Iceberg layout (the `metadata/` directory and a current `version-hint.text`) must already exist in the bucket.
+GCS-backed Iceberg tables are typically read via `direct` mode — point `table_location` at the table root. GCS-native conventions are handled automatically: `gs://` paths in metadata/manifests, a Hadoop-style integer `version-hint.text` (resolved to `vN.metadata.json`), and Snappy/GZIP-compressed Parquet. As with any direct-mode table, the Iceberg layout (the `metadata/` directory and a current `version-hint.text`) must already exist in the bucket.
+
+**BigLake REST catalog (catalog auth, distinct from the storage HMAC above).** For tables discovered through Google's BigLake Iceberg REST catalog, the catalog `loadTable` call authenticates with a **Google OAuth token** — separate from the HMAC keys that read the `gs://` data files. A **static `auth_bearer`** (e.g. `gcloud auth print-access-token`) works for a one-shot map/query but **expires after ~1h and cannot renew**, so a long-running tracking worker starts returning 401s. For a workload running as a GCP service account (GKE **Workload Identity**), set **`auth_google_metadata: true`** instead: it mints and auto-refreshes tokens from the instance metadata server, so tracked jobs keep authenticating. (The metadata server is only reachable on GCE/GKE; locally, use a static `auth_bearer`.) The storage HMAC keys are unaffected — they don't expire.
 
 ## RDF Mapping (R2RML)
 
@@ -315,6 +317,147 @@ ORDER BY DESC(?date)
 LIMIT 100
 ```
 
+## Materialization (into a native ledger)
+
+Querying a graph source reads the Iceberg table on the fly. Native Fluree
+features — **BM25 full-text search, vector / RAG, and reasoning** — operate only
+on facts committed to a *native* ledger. To use those over an Iceberg table,
+**materialize** it: Fluree expands the R2RML mapping over the source rows and
+`upsert`s the resulting triples into a target native ledger, which you can then
+index and reason over like any other ledger.
+
+> Requires the `iceberg` feature. The endpoints are admin-protected (send the
+> admin Bearer token when one is configured).
+
+### One-shot materialize
+
+`POST {api_base_url}/iceberg/materialize` reads the source and writes it into the
+target ledger (created if it does not exist):
+
+```bash
+curl -X POST http://localhost:8090/v1/fluree/iceberg/materialize \
+  -H 'Content-Type: application/json' \
+  -H "Authorization: Bearer $ADMIN_TOKEN" \
+  -d '{ "source": "orders:main", "target": "orders-native:main" }'
+```
+
+```json
+{
+  "source": "orders:main",
+  "target": "orders-native:main",
+  "from_snapshot_id": null,
+  "to_snapshot_id": 5648190075564901028,
+  "incremental": false,
+  "committed": true,
+  "rows_read": 1200,
+  "subjects_upserted": 1200,
+  "subjects_retracted": 0
+}
+```
+
+The materialized snapshot id is persisted as a **watermark** inside the target
+ledger (one per source table), so re-running resumes **incrementally** — only
+the rows added since the last run are read. You track nothing; just call it
+again. A run with no new data commits nothing and returns `committed: false`.
+Pass `"force_full": true` to ignore the watermark and re-read the whole table.
+
+Incremental reads apply when the source's snapshot window is append- or
+compaction-only; an `overwrite`/`delete` snapshot, or expired history, falls back
+to a full re-read automatically.
+
+### Tracking (keep the target fresh automatically)
+
+`POST {api_base_url}/iceberg/track` registers a `source → target` job with the
+in-process tracking worker, runs an immediate first sync, and then refreshes the
+target on a timer (default every 30s):
+
+```bash
+curl -X POST http://localhost:8090/v1/fluree/iceberg/track \
+  -H 'Content-Type: application/json' \
+  -H "Authorization: Bearer $ADMIN_TOKEN" \
+  -d '{ "source": "orders:main", "target": "orders-native:main" }'
+```
+
+- `POST {api_base_url}/iceberg/untrack` — stop tracking (leaves materialized data in place).
+- `GET {api_base_url}/iceberg/tracking` — list tracked jobs and worker stats.
+
+The worker runs on write nodes (not peers). Tracked jobs are held in memory, so
+after a restart re-issue `track` — the watermark in the target ledger means it
+resumes incrementally rather than re-reading everything.
+
+### Change data capture: updates and deletes (latest-by-key)
+
+By default materialization is **additive** (it inserts/updates triples; it never
+removes them). For a change-data-capture source — an append-only log where each
+change is a new row and a delete is a **tombstone row** — configure two options
+so Fluree applies *latest-by-key* semantics that match a
+`ROW_NUMBER() OVER (PARTITION BY id ORDER BY <ts> DESC) = 1` view:
+
+- **`order_by`** — a column that orders a key's revisions (e.g. an event
+  timestamp or offset). Must be an **integer / date / timestamp** column. The
+  latest row per subject wins, and a whole-subject *replace* clears fields that
+  were dropped in the newer revision.
+- **`delete_column`** + **`delete_values`** — how a row is recognized as a
+  delete. `delete_values` lists the `delete_column` values that mean "deleted";
+  a `null` entry matches a NULL column (the Debezium null-payload convention).
+  When the latest row for a key is a tombstone, the **entire subject** (all its
+  triples) is retracted.
+
+Two ways to encode a delete:
+
+```bash
+# (a) value-match: an op column carries "d" on a delete (Debezium-style).
+curl -X POST http://localhost:8090/v1/fluree/iceberg/map \
+  -H 'Content-Type: application/json' \
+  -d '{ "name": "orders", "mode": "direct",
+        "table_location": "s3://my-bucket/warehouse/sales/orders",
+        "r2rml": "...", "r2rml_type": "text/turtle",
+        "order_by": "event_timestamp",
+        "delete_column": "_op", "delete_values": ["d", "delete"] }'
+
+# (b) null-payload: a delete row has the key set but content columns null;
+#     pick a column always set on a live row but null on a delete, and list null.
+curl -X POST http://localhost:8090/v1/fluree/iceberg/map \
+  -H 'Content-Type: application/json' \
+  -d '{ "name": "orders", "mode": "direct",
+        "table_location": "s3://my-bucket/warehouse/sales/orders",
+        "r2rml": "...", "r2rml_type": "text/turtle",
+        "order_by": "event_timestamp",
+        "delete_column": "status", "delete_values": [null] }'
+
+# (combine both: a "d" op value OR a null column both mean delete)
+#   "delete_column": "_op", "delete_values": ["d", null]
+```
+
+These live in the graph source's stored config (`IcebergGsConfig.delete` and
+`order_by`); set them once at `iceberg map` time. A delete removes the **whole
+entity**, not individual columns — a `null` in an ordinary column of a *live* row
+just clears that one predicate.
+
+### Assumptions and limitations
+
+Latest-by-key mode (i.e. when `order_by` and/or a delete convention is set)
+assumes the source matches the append-only, full-image CDC shape these features
+target. The materializer enforces what it can and documents the rest:
+
+- **One complete row per subject revision.** Each row is a full snapshot of its
+  subject. A source that assembles one subject across multiple rows (e.g. an
+  unpivoted join table) is not supported in latest-by-key mode — use additive
+  mode (omit `order_by`/`delete`) or a one-row-per-subject view.
+- **One triples map per logical table** (enforced — multiple would clobber under
+  whole-subject replace).
+- **`order_by` must be populated and value-orderable** (int/date/timestamp,
+  enforced). A row with a null ordering value sorts as oldest.
+- **The target ledger is dedicated to one source.** Whole-subject retraction owns
+  the subject; don't mix other sources or hand-written data about the same IRIs
+  into the same target.
+- **Deletes must be expressed as tombstone rows.** A key that simply stops
+  appearing — with no tombstone — is not reconciled (a set-difference pass is a
+  possible future addition).
+- Retract and re-assert are two ordered commits; the watermark advances only in
+  the second, so an interrupted run re-materializes the same window on the next
+  pass (self-healing).
+
 ## Partition Pruning
 
 Iceberg's partition pruning optimizes queries: