SPB Git
spboucher.ai

spb/findata

Public

SQL views: hive-partitioned directory globs, lazy + cached; coverage limited to primary datasets; 4 bulk workers

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Simon-Pierre Boucher committed (Nov 2, 2026) parent e8e99da Browse files

4 changed files +52 −111

4 changed files
Unified Split
modified infra/mld/findata.json +0 −96
@@ -192,102 +192,6 @@
192 192 "autorestart": true,
193 193 "max_memory_restart": "12G",
194 194 "kill_timeout": 3600000
195 − },
196 − {
197 − "name": "findata-bulk-5",
198 − "manager": "pm2",
199 − "script": "{{HOME}}/apps/findata/infra/scripts/run-under-sshd.sh",
200 − "args": [
201 − "{{HOME}}/apps/findata",
202 − "{{HOME}}/apps/findata/.venv/bin/python",
203 − "-m",
204 − "findata_ingest.cli",
205 − "worker",
206 − "--role",
207 − "bulk"
208 − ],
209 − "interpreter": null,
210 − "cwd": "{{HOME}}/apps/findata",
211 − "env": {
212 − "FINDATA_ROOT": "{{HOME}}/apps/findata",
213 − "PATH": "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin"
214 − },
215 − "cron_restart": null,
216 − "autorestart": true,
217 − "max_memory_restart": "12G",
218 − "kill_timeout": 3600000
219 − },
220 − {
221 − "name": "findata-bulk-6",
222 − "manager": "pm2",
223 − "script": "{{HOME}}/apps/findata/infra/scripts/run-under-sshd.sh",
224 − "args": [
225 − "{{HOME}}/apps/findata",
226 − "{{HOME}}/apps/findata/.venv/bin/python",
227 − "-m",
228 − "findata_ingest.cli",
229 − "worker",
230 − "--role",
231 − "bulk"
232 − ],
233 − "interpreter": null,
234 − "cwd": "{{HOME}}/apps/findata",
235 − "env": {
236 − "FINDATA_ROOT": "{{HOME}}/apps/findata",
237 − "PATH": "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin"
238 − },
239 − "cron_restart": null,
240 − "autorestart": true,
241 − "max_memory_restart": "12G",
242 − "kill_timeout": 3600000
243 − },
244 − {
245 − "name": "findata-bulk-7",
246 − "manager": "pm2",
247 − "script": "{{HOME}}/apps/findata/infra/scripts/run-under-sshd.sh",
248 − "args": [
249 − "{{HOME}}/apps/findata",
250 − "{{HOME}}/apps/findata/.venv/bin/python",
251 − "-m",
252 − "findata_ingest.cli",
253 − "worker",
254 − "--role",
255 − "bulk"
256 − ],
257 − "interpreter": null,
258 − "cwd": "{{HOME}}/apps/findata",
259 − "env": {
260 − "FINDATA_ROOT": "{{HOME}}/apps/findata",
261 − "PATH": "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin"
262 − },
263 − "cron_restart": null,
264 − "autorestart": true,
265 − "max_memory_restart": "12G",
266 − "kill_timeout": 3600000
267 − },
268 − {
269 − "name": "findata-bulk-8",
270 − "manager": "pm2",
271 − "script": "{{HOME}}/apps/findata/infra/scripts/run-under-sshd.sh",
272 − "args": [
273 − "{{HOME}}/apps/findata",
274 − "{{HOME}}/apps/findata/.venv/bin/python",
275 − "-m",
276 − "findata_ingest.cli",
277 − "worker",
278 − "--role",
279 − "bulk"
280 − ],
281 − "interpreter": null,
282 − "cwd": "{{HOME}}/apps/findata",
283 − "env": {
284 − "FINDATA_ROOT": "{{HOME}}/apps/findata",
285 − "PATH": "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin"
286 − },
287 − "cron_restart": null,
288 − "autorestart": true,
289 − "max_memory_restart": "12G",
290 − "kill_timeout": 3600000
291 195 }
292 196 ],
293 197 "ngrok": null,
modified services/api/src/findata_api/routers/query.py +12 −4
@@ -15,6 +15,11 @@ from findata_api.ratelimit import record_usage
15 15 router = APIRouter(prefix="/query", tags=["query"])
16 16
17 17
18 +def _used(sql: str, views: dict) -> list[str]:
19 + from findata_ingest.reader import _referenced_views
20 + return _referenced_views(sql, views)
21 +
22 +
18 23 class QueryIn(BaseModel):
19 24 sql: str = Field(..., max_length=20000)
20 25 max_rows: int = Field(1000, ge=1, le=100_000)
@@ -43,12 +48,15 @@ def query(body: QueryIn, request: Request, p: A.Principal = Depends(A.require("q
43 48 A.audit(request, "query.sql", None, rows=len(rows), ms=int((time.monotonic() - t0) * 1000), sql=body.sql[:500])
44 49 return {"columns": [{"name": f.name, "type": str(f.type)} for f in tbl.schema], "rows": rows,
45 50 "meta": {**meta, "rows": len(rows), "elapsed_ms": int((time.monotonic() - t0) * 1000),
46 − "views": sorted(views), "engine": "duckdb (read-only, sandboxed, 1 statement)"}}
51 + "views_used": _used(body.sql, views), "engine": "duckdb (read-only, sandboxed, 1 statement, hive-partition pruning)"}}
47 52
48 53
49 54 @router.get("/views")
50 55 def views(p: A.Principal = Depends(A.require("query"))):
51 56 v = dataset_views()
52 − return {"views": [{"name": k, "files": len(paths)} for k, paths in sorted(v.items())],
53 − "examples": ["SELECT symbol_as_reported, count(*) bars FROM market_bars_stock_1m_raw GROUP BY 1 ORDER BY 2 DESC LIMIT 20",
54 − "SELECT session_date, min(low) lo, max(high) hi, sum(volume) vol FROM market_bars_stock_1m_raw WHERE symbol_as_reported='AAPL' GROUP BY 1 ORDER BY 1"]}
57 + return {"views": [{"name": k, "kind": spec.get("kind"), "root": spec.get("root")} for k, spec in sorted(v.items())],
58 + "note": "Views are Hive-partitioned directory scans: filter on the partition columns (symbol, year, session_date; underlying, period for options) "
59 + "to prune files — e.g. WHERE symbol='AAPL' AND year=2024. Whole-dataset scans over millions of files time out (30 s).",
60 + "examples": ["SELECT session_date, min(low) lo, max(high) hi, sum(volume) vol FROM market_bars_stock_1m_raw WHERE symbol='AAPL' AND year=2024 GROUP BY 1 ORDER BY 1",
61 + "SELECT trade_date, strike, call_put, bid, ask, open_interest, delta FROM options_eod WHERE underlying='SPY' AND period='2024q1' AND expiry=DATE '2024-03-28' ORDER BY trade_date, strike LIMIT 100",
62 + "SELECT symbol, count(*) bars FROM market_bars_futures_1m_continuous_ratio WHERE symbol IN ('ES','NQ') GROUP BY 1"]}
modified services/ingest/src/findata_ingest/reader.py +34 −11
@@ -114,7 +114,7 @@ def apply_adjustment(table: pa.Table, adjustment: str, lake: Lake | None = None)
114 114 return res.to_arrow().cast(BARS_SCHEMA), notes
115 115
116 116
117 −def run_sql(sql: str, max_rows: int, timeout_s: int, views: dict[str, list[str]] | None = None) -> tuple[pa.Table, dict[str, Any]]:
117 +def run_sql(sql: str, max_rows: int, timeout_s: int, views: dict[str, Any] | None = None) -> tuple[pa.Table, dict[str, Any]]:
118 118 """Sandboxed read-only SQL: DuckDB in-memory, external access limited to the registered dataset views."""
119 119 import re
120 120 import threading
@@ -131,12 +131,15 @@ def run_sql(sql: str, max_rows: int, timeout_s: int, views: dict[str, list[str]]
131 131 rr = str(r.resolve()).replace("'", "''")
132 132 dirs += [f"'{rr}/curated'", f"'{rr}/_retired'"]
133 133 con.execute(f"SET allowed_directories = [{', '.join(dirs)}]")
134 − for name, paths in (views or {}).items():
134 + for name, spec in (views or {}).items():
135 135 if not re.fullmatch(r"[a-z][a-z0-9_]*", name):
136 136 raise ValueError(f"bad view name {name}")
137 − if paths:
138 − lst = ", ".join("'" + p.replace("'", "''") + "'" for p in paths)
139 − con.execute(f"CREATE VIEW {name} AS SELECT * FROM read_parquet([{lst}], union_by_name=true, hive_partitioning=false)")
137 + if name not in _referenced_views(sql, {name: spec}):
138 + continue # only materialize views the statement uses (directory globs are not free)
139 + globs = (spec.get("globs") or []) if isinstance(spec, dict) else list(spec or [])
140 + if globs:
141 + lst = ", ".join("'" + g.replace("'", "''") + "'" for g in globs)
142 + con.execute(f"CREATE VIEW {name} AS SELECT * FROM read_parquet([{lst}], union_by_name=true, hive_partitioning=true)")
140 143 else:
141 144 con.execute(f"CREATE VIEW {name} AS SELECT * FROM (VALUES (NULL)) t(empty) WHERE false")
142 145 con.execute("SET enable_external_access = false")
@@ -153,13 +156,33 @@ def run_sql(sql: str, max_rows: int, timeout_s: int, views: dict[str, list[str]]
153 156 return tbl.slice(0, max_rows), {"truncated": truncated}
154 157
155 158
156 −def dataset_views() -> dict[str, list[str]]:
157 − """Map SQL view names (dataset ids with dots → underscores) to live file lists."""
159 +_views_cache: dict[str, Any] = {"at": 0.0, "value": None}
160 +
161 +
162 +def dataset_views() -> dict[str, dict[str, Any]]:
163 + """SQL views for the playground/assistant: one per dataset with data, built from Hive-style directory globs
164 + (`curated/<dataset>/{base,delta}/**`) so that DuckDB prunes partitions (`symbol=`, `year=`, `session_date=`,
165 + `underlying=`, `period=`) from the path instead of receiving millions of file names. The directory tree is the
166 + live set (retired files are moved out); files of an in-progress publication may be visible for a few minutes.
167 + Cached 5 minutes."""
168 + import time
169 + if _views_cache["value"] is not None and time.monotonic() - _views_cache["at"] < 300:
170 + return _views_cache["value"]
158 171 lake = Lake()
159 − out: dict[str, list[str]] = {}
172 + out: dict[str, dict[str, Any]] = {}
160 173 with connection() as conn:
161 − ds = conn.execute("SELECT id FROM datasets WHERE kind IN ('market_bars','derived','options') AND current_version > 0").fetchall()
174 + ds = conn.execute("SELECT id, kind FROM datasets WHERE kind IN ('market_bars','derived','options') AND current_version > 0").fetchall()
162 175 for d in ds:
163 − files = live_files(conn, d["id"])
164 − out[d["id"].replace(".", "_")] = [str(lake.resolve(f["path"])) for f in files]
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()]
179 + if globs:
180 + out[d["id"].replace(".", "_")] = {"globs": globs, "root": str(root), "kind": d["kind"]}
181 + _views_cache.update(at=time.monotonic(), value=out)
165 182 return out
183 +
184 +
185 +def _referenced_views(sql: str, views: dict[str, Any]) -> list[str]:
186 + import re as _re
187 + low = sql.lower()
188 + return [name for name in views if _re.search(r"\b" + _re.escape(name) + r"\b", low)]
modified services/ingest/src/findata_ingest/tasks/common.py +6 −0
@@ -31,7 +31,13 @@ def record_raw_archive(conn, source_id: str, dataset_id: str, request: dict[str,
31 31 return str(r["id"]), True
32 32
33 33
34 +COVERAGE_DATASETS = {"market_bars.stock.1m.raw", "market_bars.etf.1m.raw", "market_bars.index.1m.raw",
35 + "market_bars.stock.5m.raw", "market_bars.etf.5m.raw", "market_bars.stock.1d.raw", "market_bars.etf.1d.raw"}
36 +
37 +
34 38 def upsert_coverage(conn, dataset_id: str, table: pa.Table, interval: Interval) -> int:
39 + if dataset_id not in COVERAGE_DATASETS:
40 + return 0 # adjusted/coarse variants share the raw datasets' sessions; keeping per-session rows for all 64 datasets cost 130 GB
35 41 if table.num_rows and table.column("asset_class")[0].as_py() in ("fx", "crypto", "futures"):
36 42 return 0 # no exchange-session expectation for 24h markets
37 43 rows = session_coverage(table, interval)
38 44
Menu

Theme
↑↓ navigate ↵ open esc close / or ⌘K anywhere