| 136 |
136 |
raise ValueError(f"bad view name {name}") |
| 137 |
137 |
if name not in _referenced_views(sql, {name: spec}): |
| 138 |
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)") |
|
139 |
+ if isinstance(spec, dict) and spec.get("sql"): |
|
140 |
+ con.execute(f"CREATE VIEW {name} AS {spec['sql']}") |
|
141 |
+ elif isinstance(spec, list) and spec: # legacy explicit file list |
|
142 |
+ lst = ", ".join("'" + g.replace("'", "''") + "'" for g in spec) |
|
143 |
+ con.execute(f"CREATE VIEW {name} AS SELECT * FROM read_parquet([{lst}], union_by_name=true, hive_partitioning=false)") |
| 143 |
144 |
else: |
| 144 |
145 |
con.execute(f"CREATE VIEW {name} AS SELECT * FROM (VALUES (NULL)) t(empty) WHERE false") |
| 145 |
146 |
con.execute("SET enable_external_access = false") |
| 174 |
175 |
ds = conn.execute("SELECT id, kind FROM datasets WHERE kind IN ('market_bars','derived','options') AND current_version > 0").fetchall() |
| 175 |
176 |
for d in ds: |
| 176 |
177 |
root = lake.root_for(d["id"]) |
| 177 |
|
− globs = [] |
|
178 |
+ layouts = {} |
| 178 |
179 |
for layout in ("base", "delta"): |
| 179 |
180 |
ldir = root / "curated" / d["id"] / layout |
| 180 |
181 |
if ldir.exists() and next(ldir.rglob("*.parquet"), None) is not None: # skip emptied layouts |
| 181 |
|
− globs.append(str(ldir / "**" / "*.parquet")) |
| 182 |
|
− if globs: |
| 183 |
|
− out[d["id"].replace(".", "_")] = {"globs": globs, "root": str(root), "kind": d["kind"]} |
|
182 |
+ layouts[layout] = str(ldir / "**" / "*.parquet").replace("'", "''") |
|
183 |
+ if not layouts: |
|
184 |
+ continue |
|
185 |
+ parts = [] |
|
186 |
+ if d["kind"] == "options": |
|
187 |
+ parts.append(f"SELECT * FROM read_parquet('{layouts['base']}', hive_partitioning=true)") |
|
188 |
+ else: |
|
189 |
+ if "base" in layouts: # partition columns symbol=, year= come from the path (pruned before any file is opened) |
|
190 |
+ parts.append(f"SELECT * FROM read_parquet('{layouts['base']}', hive_partitioning=true)") |
|
191 |
+ if "delta" in layouts: # session files: derive the same columns so UNION ALL BY NAME lines up |
|
192 |
+ parts.append(f"SELECT *, symbol_as_reported AS symbol, year(session_date)::BIGINT AS year " |
|
193 |
+ f"FROM read_parquet('{layouts['delta']}', hive_partitioning=false)") |
|
194 |
+ out[d["id"].replace(".", "_")] = {"sql": " UNION ALL BY NAME ".join(parts), "layouts": sorted(layouts), "root": str(root), "kind": d["kind"]} |
| 184 |
195 |
_views_cache.update(at=time.monotonic(), value=out) |
| 185 |
196 |
return out |
| 186 |
197 |
|
| 187 |
198 |
|