Lake /query reads era5_daily through Iceberg metadata #44

Merged
admin_emi merged 1 commit from fix/lake-query-iceberg into dev 2026-07-24 19:09:23 +00:00
2 changed files with 44 additions and 25 deletions

View file

@ -53,10 +53,12 @@ RUN useradd --system --create-home --uid 10001 thermograph \
&& chown -R thermograph:thermograph /app /state
USER thermograph
# Bake DuckDB's httpfs extension AS THE RUNTIME USER — extensions install to
# the invoking user's ~/.duckdb, and a root-time bake leaves uid 10001 unable
# to LOAD it (seen live: prod lake /query 500'd on exactly this).
RUN python -c "import duckdb; duckdb.connect().execute('INSTALL httpfs')"
# Bake DuckDB's httpfs + iceberg extensions AS THE RUNTIME USER — extensions
# install to the invoking user's ~/.duckdb, and a root-time bake leaves uid
# 10001 unable to LOAD them (seen live: prod lake /query 500'd on exactly
# this). iceberg powers the era5_daily view via table metadata instead of a
# bucket-wide glob LIST.
RUN python -c "import duckdb; c = duckdb.connect(); c.execute('INSTALL httpfs'); c.execute('INSTALL iceberg')"
WORKDIR /app
ENV PORT=8137 \

View file

@ -135,34 +135,51 @@ _FORBIDDEN = re.compile(
def _connect():
"""A fresh in-memory DuckDB with the lake views registered. Per-request:
connections are cheap, and a throwaway one means a query can't leave state
behind for the next. Returns None when no lake is configured."""
"""A fresh in-memory DuckDB with the lake views registered. Returns None
when no lake is configured.
Bucket mode reads era5_daily through the ICEBERG table, not a hive glob:
a glob over era5/daily/ must LIST every object under it (~1k per tile
minutes at a few hundred tiles, seen live timing out /query at 390),
while Iceberg gets the file list from table metadata in a handful of
reads. Local mode (tests, small trees) keeps the plain hive glob no
Iceberg table exists in the fixtures and listing a local dir is free."""
import duckdb # noqa: PLC0415 - only the lake role pays the import
mode, cfg = _source()
if mode == "none":
return None
con = duckdb.connect()
if mode == "local":
daily_glob = os.path.join(era5lake.local_dir(), era5lake.PREFIX,
"daily/*/*/*/*.parquet")
con.execute(f"""
CREATE VIEW era5_daily AS
SELECT * FROM read_parquet('{daily_glob}', hive_partitioning=1)""")
manifest_src = os.path.join(era5lake.local_dir(), era5lake.MANIFEST_KEY)
elif mode == "bucket":
daily_glob = f"s3://{cfg['bucket']}/{era5lake.PREFIX}/daily/*/*/*/*.parquet"
manifest_src = f"s3://{cfg['bucket']}/{era5lake.MANIFEST_KEY}"
else:
return None
con = duckdb.connect()
if mode == "bucket":
con.execute("LOAD httpfs") # baked into the image at build time
host = cfg["endpoint"].split("://", 1)[-1]
con.execute(f"SET s3_endpoint='{host}'")
con.execute("SET s3_url_style='path'")
con.execute(f"SET s3_region='{cfg['region']}'")
con.execute("SET s3_access_key_id=?", [cfg["access_key"]])
con.execute("SET s3_secret_access_key=?", [cfg["secret_key"]])
con.execute(f"""
CREATE VIEW era5_daily AS
SELECT * FROM read_parquet('{daily_glob}', hive_partitioning=1)""")
con.execute(f"CREATE VIEW manifest AS SELECT * FROM read_parquet('{manifest_src}')")
con.execute(
f"CREATE VIEW manifest AS SELECT * FROM read_parquet('{manifest_src}')")
return con
con.execute("LOAD httpfs") # both baked into the image at build time,
con.execute("LOAD iceberg") # as the runtime user (see Dockerfile)
host = cfg["endpoint"].split("://", 1)[-1]
con.execute(f"SET s3_endpoint='{host}'")
con.execute("SET s3_url_style='path'")
con.execute(f"SET s3_region='{cfg['region']}'")
con.execute("SET s3_access_key_id=?", [cfg["access_key"]])
con.execute("SET s3_secret_access_key=?", [cfg["secret_key"]])
# Newest metadata JSON wins (one single-page LIST of metadata/ — cheap);
# pyiceberg names them NNNNN-uuid.metadata.json, so lexicographic max is
# the numeric max.
meta = con.execute(
f"SELECT max(file) FROM glob('s3://{cfg['bucket']}/iceberg/era5_daily/"
f"metadata/*.metadata.json')").fetchone()[0]
if meta is None:
return None # lake bucket exists but the Iceberg table doesn't yet
con.execute(f"CREATE VIEW era5_daily AS SELECT * FROM iceberg_scan('{meta}')")
con.execute(f"""CREATE VIEW manifest AS SELECT * FROM
read_parquet('s3://{cfg['bucket']}/{era5lake.MANIFEST_KEY}')""")
return con