thermograph/infra/lake-iceberg/README.md

# lake-iceberg — Iceberg conversion for the ERA5 lake

Maintains an Apache Iceberg v2 table `era5_daily` at
`s3://era5-thermograph/iceberg/era5_daily` over the lake's existing hive
parquet (`era5/daily/tile=<ti>_<tj>/year=<Y>/month=<M>/part.parquet`).

Key properties:

- **No data rewrite.** `sync_iceberg.py` uses pyiceberg `add_files`: the
  Iceberg metadata references the existing part files in place. Storage is
  not doubled and nothing under `era5/` is ever written, moved, or deleted —
  the tool only writes under the `iceberg/` prefix.
- **Incremental + idempotent.** Each run diffs the tiles in
  `era5/manifest.parquet` (the source of truth for *complete* tiles — the
  extractor uploads all of a tile's objects before its manifest rows appear)
  against the `thermograph.synced-tiles` table property, and registers only
  the new ones. Files and the property update land in the same commit, so an
  interrupted run never half-syncs a tile; re-running is always safe, even
  while an extraction job is writing new tiles.
- **Self-contained catalog.** A pyiceberg SqlCatalog on local sqlite
  (`THERMOGRAPH_ICEBERG_CATALOG_DB`, default `/state/iceberg-catalog.db`).
  The catalog is only a pointer: if the db is lost, the next run finds the
  newest `*.metadata.json` under the table location and re-registers it —
  the synced-tiles set lives in the table metadata itself, so nothing is
  re-added.
- **Partitioning `(tile, year, month)`** via order-preserving transforms of
  real columns: `truncate[12](lat_idx)` / `truncate[12](lon_idx)` (partition
  values are `ti*12` / `tj*12` — the part files carry no tile column, the 12
  matches the lake's 12x12-point tile grid) plus `year(date)` and
  `month(date)`. Predicates on `lat_idx`/`lon_idx`/`date` prune partitions
  directly — no synthetic tile/year/month columns needed in queries.

## Build

```sh
docker build -t thermograph-lake-iceberg infra/lake-iceberg/
```

## Run (one-shot)

Config is the backend's lake variables (same names and defaults as
`backend/data/era5lake.py`): `THERMOGRAPH_LAKE_S3_ACCESS_KEY`,
`THERMOGRAPH_LAKE_S3_SECRET_KEY`, and optionally
`THERMOGRAPH_LAKE_S3_ENDPOINT` / `_BUCKET` / `_REGION`.

```sh
docker run --rm --env-file /path/to/lake.env \
    -v lake-iceberg-state:/state \
    thermograph-lake-iceberg
```

Flags: `--dry-run` (list pending tiles, write nothing), `--limit N` (cap
tiles per run), `--batch N` (tiles per Iceberg commit, default 5),
`--catalog-db PATH`. `PYICEBERG_MAX_WORKERS` controls the parquet-footer
read parallelism inside `add_files` (32 is a good value for the ~1000 small
footers per tile).

Each run prints the row-count reconciliation (records added vs the
manifest's per-tile row counts) and ends with the table's current metadata
JSON path. It also refreshes `metadata/version-hint.text` (Hadoop-catalog
convention) so readers can scan the bare table root without knowing that
path. Transient S3 errors (Contabo intermittently returns SLOW_DOWN /
connection timeouts under load) are retried per batch with backoff
(`--retries`, default 3); a run that still dies resumes at the next batch
boundary on re-run.

## Scheduling

Meant to run as a one-shot on prod after extraction batches finish — e.g. a
cron or an ops-cron workflow step invoking the `docker run` above, with the
secrets rendered from the SOPS vault like every other consumer of the
`THERMOGRAPH_LAKE_S3_*` pair. Deliberately **not** wired into the Swarm
stack or CI: the sync is an offline maintenance job, not a service, and each
run costs one footer read per new part file (~1000 per tile), so it belongs
after extraction batches, not on a tight loop.

## Querying from DuckDB

No catalog server needed — the version-hint pointer lets DuckDB scan the
table root directly:

```sql
INSTALL iceberg; LOAD iceberg;
INSTALL httpfs;  LOAD httpfs;
CREATE SECRET lake (
    TYPE s3,
    KEY_ID '<THERMOGRAPH_LAKE_S3_ACCESS_KEY>',
    SECRET '<THERMOGRAPH_LAKE_S3_SECRET_KEY>',
    ENDPOINT 'eu2.contabostorage.com',
    REGION 'default',
    URL_STYLE 'path'
);
SELECT count(*)
FROM iceberg_scan('s3://era5-thermograph/iceberg/era5_daily');
```

To pin an exact table version, scan the metadata JSON path the sync run
printed (`iceberg_scan('s3://…/metadata/<NNNNN-uuid>.metadata.json')`).

Partition pruning works on the raw columns, e.g. one tile-month:

```sql
SELECT lat_idx, lon_idx, avg(tmax)
FROM iceberg_scan('s3://era5-thermograph/iceberg/era5_daily')
WHERE lat_idx BETWEEN 120 AND 131 AND lon_idx BETWEEN 120 AND 131
  AND date >= DATE '2020-07-01' AND date < DATE '2020-08-01'
GROUP BY 1, 2;
```

The part files carry no Iceberg field ids; the table sets
`schema.name-mapping.default` so engines resolve columns by name (validated
with pyiceberg 0.11.1 and DuckDB 1.5.5, values byte-identical to the hive
side).

## Tests

```sh
pip install -r infra/lake-iceberg/requirements.txt pytest
pytest infra/lake-iceberg/test_sync.py
```

Local-filesystem fixtures only (tiny parquet files, `file://` warehouse,
sqlite catalog in tmp) — no network, no credentials.