# Iceberg-to-FalkorDB Loader

A Rust CLI tool that loads and incrementally syncs Apache Iceberg tables into a FalkorDB graph using a declarative JSON/YAML configuration.

Design mirrors other connectors in this repo (ClickHouse, Databricks, …): you describe how Iceberg tables map to graph nodes and edges, and the tool handles Arrow batch conversion, UNWIND+MERGE upserts, optional soft deletes, and incremental updates based on a watermark column.

## Features

- **Iceberg source**
  - Catalogs: REST (`iceberg-catalog-rest`), AWS Glue (`iceberg-catalog-glue`), SQL/sqlite (`iceberg-catalog-sql`) for local/lightweight setups.
  - Storage via `iceberg-storage-opendal`'s `OpenDalResolvingStorageFactory` (auto-detects `file://`, `s3://`, `gs://`, `azblob://`).
  - Full `table.scan().to_arrow()` path (stable; correctly applies merge-on-read deletes — see `ADR-0001-source-access.md`).
- **Schema scaffolding**
  - `--introspect-schema` and `--generate-template` derive starter node mappings from Iceberg table metadata.
  - No automatic cross-table FK/edge inference in v1 — edges are defined manually.
- **FalkorDB sink**
  - Writes nodes and edges using Cypher `UNWIND` + `MERGE`.
  - Applies explicit `falkordb.indexes` plus implicit indexes on node keys and edge endpoint `match_on` properties.
- **Incremental sync with column watermark**
  - Per-mapping `mode: full` or `mode: incremental`.
  - Watermark column (`delta.updated_at_column`) filters rows client-side after a full scan.
  - Optional soft delete via `delta.deleted_flag_column` / `delta.deleted_flag_value`.
  - Native Iceberg snapshot-incremental scanning is intentionally **not** used in v1 (see ADR-0001).
- **Persistent state**
  - File-backed state (`state.backend: file`) stores per-mapping watermarks between runs.
- **Daemon / purge / metrics**
  - `--daemon`, `--purge-graph`, `--purge-mapping`, Prometheus metrics on port 9995 (prefix `iceberg_to_falkordb_`).

## Quick Start

### 1. Generate the local demo data

`iceberg_sample_to_falkordb.yaml` is a working end-to-end demo out of the
box: it reads a local Iceberg table (SQL/sqlite catalog + `file://`
warehouse) generated by a small helper script, and writes `Customer` /
`Order` nodes plus `PLACED` edges into FalkorDB.

```bash
cd Iceberg-to-FalkorDB
pip install -r sample_data/requirements.txt
eval "$(python3 sample_data/generate_sample_table.py)"
```

The script (re)creates `sample_data/catalog.db` + `sample_data/warehouse/`
with `sales.customers` and `sales.orders` tables (the latter including a
merge-on-read delete, to exercise that path), and prints two `export`
lines consumed by `eval` above: `ICEBERG_SAMPLE_CATALOG_URI` and
`ICEBERG_SAMPLE_WAREHOUSE`, referenced by the sample config via the usual
`$VARIABLE` convention. Re-run it any time to reset the demo data.

### 2. Build and run

```bash
cd iceberg-to-falkordb
cargo build --release
cargo run --release -- --config ../iceberg_sample_to_falkordb.yaml
```

Daemon mode:

```bash
cargo run --release -- --config ../iceberg_sample_to_falkordb.yaml --daemon --interval-secs 300
```

### 3. Connect to a real catalog

For a managed/production REST catalog, replace the `iceberg.catalog` block
with (see the commented example at the top of `iceberg_sample_to_falkordb.yaml`):

```yaml
iceberg:
  catalog:
    type: rest
    uri: "$ICEBERG_CATALOG_URI"
    warehouse: "s3://my-bucket/warehouse"
    properties:
      token: "$ICEBERG_TOKEN"
  storage_properties:
    s3.region: "us-east-1"
    s3.access-key-id: "$AWS_ACCESS_KEY_ID"
    s3.secret-access-key: "$AWS_SECRET_ACCESS_KEY"
```

Secrets support the `$VARIABLE` environment-reference convention used by every other connector.

### Scaffold from Iceberg metadata

```bash
cargo run --release -- --config path/to/config.yaml --introspect-schema
cargo run --release -- --config path/to/config.yaml --generate-template --output iceberg.scaffold.yaml
```

### Metrics

Default endpoint: `http://127.0.0.1:9995/`

- CLI: `--metrics-port <port>`
- Env: `ICEBERG_TO_FALKORDB_METRICS_PORT`

Metric names use the `iceberg_to_falkordb_` prefix (runs, failed_runs, rows_fetched/written/deleted, plus per-mapping variants).

## Architecture notes

See `ADR-0001-source-access.md` for the feasibility spike findings and the binding v1 decisions:

1. Full `table.scan()` only (no snapshot-incremental).
2. Column-watermark incremental model shared with other connectors.
3. Catalogs: REST + Glue + SQL; HMS deferred.
4. Storage: `OpenDalResolvingStorageFactory` directly.

## Tuning guidance and known limitations

Validated locally against a SQL/sqlite catalog with a `file://` warehouse,
including append, merge-on-read delete, and watermark-update scenarios:

- **Memory characteristics**: per ADR-0001, `fetch_rows_for_mapping` collects
  the entire `table.scan().to_arrow()` result (all Arrow batches for the
  mapped table) into memory before converting and writing to FalkorDB; the
  column-watermark filter is then applied client-side. Memory usage scales
  with the full size of each mapped table, not just the incremental delta.
  For very large tables, consider `source.columns` projection to reduce
  per-row width, or splitting a table into multiple narrower mappings.
- **Arrow version bridging overhead**: because `iceberg` 0.10.1 emits Arrow 58
  `RecordBatch`es and the shared bridge crate expects Arrow 60, batches are
  round-tripped through Arrow IPC (`iceberg_batch_to_logical_rows`). This adds
  CPU/serialization overhead proportional to scanned data volume; revisit once
  upstream `iceberg` aligns on a newer Arrow version.
- **SQL catalog name must match the connector's load name**: when testing
  locally against a SQL/sqlite catalog created via `pyiceberg` or another
  client, the catalog name used (e.g. `SqlCatalog("sql", ...)`) must match the
  name this connector passes internally (`"sql"`), since `iceberg-catalog-sql`
  scopes table rows by catalog name. A mismatch surfaces as `TableNotFound`
  even though the table exists in the catalog database.
- **Credentials**: catalog `properties` (e.g. REST bearer tokens) and
  `storage_properties` (e.g. S3 keys) are plain strings resolved from `$VAR`
  env references at startup. No code path currently logs the full config via
  `{:?}`, but the config structs derive `Debug` without redaction (consistent
  with other connectors in this repo), so avoid adding debug-logging of raw
  config values in future changes.

## License

Apache-2.0 (same as this repository).