planner: symbol filter in SQL; views skip emptied layouts
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2 changed files +10 −5
modified
services/ingest/src/findata_ingest/lake.py
+5 −3
@@ -321,7 +321,9 @@ def live_files(conn, dataset_id: str, symbols: list[str] | None = None, start: d | ||
| 321 | 321 | if end is not None: |
| 322 | 322 | sql += " AND date_min <= %s" |
| 323 | 323 | params.append(end) |
| 324 | − rows = conn.execute(sql, params).fetchall() | |
| 325 | 324 | if symbols: |
| 326 | − rows = [r for r in rows if any(r["symbol_min"] <= s <= r["symbol_max"] for s in symbols)] | |
| 327 | − return rows | |
| 325 | + # a file is relevant when its symbol range covers any requested symbol; per-symbol base files hit the | |
| 326 | + # (dataset_id, symbol_min) index, multi-symbol delta files are matched by range | |
| 327 | + sql += " AND EXISTS (SELECT 1 FROM unnest(%s::text[]) s WHERE s BETWEEN symbol_min AND symbol_max)" | |
| 328 | + params.append(list(symbols)) | |
| 329 | + return conn.execute(sql, params).fetchall() | |
modified
services/ingest/src/findata_ingest/reader.py
+5 −2
@@ -174,8 +174,11 @@ def dataset_views() -> dict[str, dict[str, Any]]: | ||
| 174 | 174 | ds = conn.execute("SELECT id, kind FROM datasets WHERE kind IN ('market_bars','derived','options') AND current_version > 0").fetchall() |
| 175 | 175 | for d in ds: |
| 176 | 176 | root = lake.root_for(d["id"]) |
| 177 | − globs = [str(root / "curated" / d["id"] / layout / "**" / "*.parquet") for layout in ("base", "delta") | |
| 178 | − if (root / "curated" / d["id"] / layout).exists()] | |
| 177 | + globs = [] | |
| 178 | + for layout in ("base", "delta"): | |
| 179 | + ldir = root / "curated" / d["id"] / layout | |
| 180 | + if ldir.exists() and next(ldir.rglob("*.parquet"), None) is not None: # skip emptied layouts | |
| 181 | + globs.append(str(ldir / "**" / "*.parquet")) | |
| 179 | 182 | if globs: |
| 180 | 183 | out[d["id"].replace(".", "_")] = {"globs": globs, "root": str(root), "kind": d["kind"]} |
| 181 | 184 | _views_cache.update(at=time.monotonic(), value=out) |
| 182 | 185 | |