"""DuckDB-only fallback: same surface as the cloud adapters, but the
"warehouse" is the in-process DuckDB engine reading the local Iceberg
table from L10."""

from __future__ import annotations

from pathlib import Path

# The warehouse/ dir sits alongside iceberg/ inside sample-pipeline/.
# We import relative to the package root so this module is importable
# without any env vars set (no cloud SDK dependency).
from iceberg.query_duckdb import open_with_iceberg  # noqa: E402

LOCAL_DUCKDB_PATH = (Path(__file__).resolve().parent.parent
                     / "warehouse" / "duckdb_local.duckdb").as_posix()


def materialize_daily_counts() -> None:
    """Materialise daily_counts into a persistent .duckdb file so analyst
    queries don't re-scan the Iceberg metadata each time.

    Same SQL shape as the Snowflake / BigQuery adapters' view, with the
    materialisation step expressed as CREATE TABLE AS SELECT (CTAS).
    """
    import duckdb

    in_mem = open_with_iceberg()
    persistent = duckdb.connect(LOCAL_DUCKDB_PATH)
    # Move the events view into the persistent DB by copying rows.
    rows = in_mem.execute("SELECT * FROM events").arrow()
    persistent.register("events_arrow", rows)
    persistent.execute(
        "CREATE OR REPLACE TABLE events AS SELECT * FROM events_arrow"
    )
    persistent.execute("""
        CREATE OR REPLACE TABLE daily_counts AS
        SELECT event_date, type, count(*) AS n
        FROM events
        GROUP BY event_date, type
    """)
    n = persistent.execute("SELECT count(*) FROM daily_counts").fetchone()[0]
    print(f"duckdb_local: daily_counts materialised, {n} rows.")


if __name__ == "__main__":
    materialize_daily_counts()