"""Read the L02 Parquet partitions and append them into the Iceberg
``default.events`` table.  Each invocation creates a new snapshot.

Source layout (L02 output):
  data/parquet/date=YYYY-MM-DD/*.parquet
  Columns: type, actor, repo, created_at  +  partition column ``date``

Iceberg-internal mapping:
  - ``date``  (source partition column) -> renamed to ``event_date``
  - ``actor`` is kept as-is (canonical column name in the Iceberg schema)
  - A synthetic ``event_id`` column is added (row index within the batch)
  - ``repo``  is kept as-is

The warehouse data directories will use ``event_date=YYYY-MM-DD/`` Hive-
style names because that is the Iceberg-internal partition column name.
This is completely separate from the source ``data/parquet/date=*/``
directory layout.
"""

from __future__ import annotations

import pyarrow as pa
import pyarrow.dataset as ds

from pathlib import Path

from .tables import create_or_load_events

# Resolved relative to this file: ../../data/parquet
# Source Parquet uses Hive partition column ``date`` (NOT ``event_date``).
PARQUET_ROOT = (Path(__file__).resolve().parent.parent / "data" / "parquet").as_posix()


def append_partitions() -> None:
    table = create_or_load_events()

    # Read source Parquet with Hive partitioning; partition column is ``date``.
    dataset = ds.dataset(
        PARQUET_ROOT,
        format="parquet",
        partitioning="hive",
    )
    raw = dataset.to_table()

    # Rename source ``date`` column -> Iceberg logical column ``event_date``.
    # Cast to pa.date32() to match the Iceberg DateType field.
    col_names = raw.schema.names
    if "date" in col_names and "event_date" not in col_names:
        idx = col_names.index("date")
        raw = raw.rename_columns(
            [("event_date" if n == "date" else n) for n in col_names]
        )
    # Cast event_date to date32 if it came in as string from Hive partitioning.
    if raw.schema.field("event_date").type != pa.date32():
        raw = raw.set_column(
            raw.schema.get_field_index("event_date"),
            "event_date",
            raw.column("event_date").cast(pa.date32()),
        )

    # Add a synthetic event_id (row index) if missing.
    if "event_id" not in raw.schema.names:
        raw = raw.append_column(
            "event_id",
            pa.array(range(raw.num_rows), type=pa.int64()),
        )

    # Select and reorder columns to match the Iceberg SCHEMA field order.
    iceberg_cols = ["event_id", "type", "actor", "repo", "created_at", "event_date"]
    available = [c for c in iceberg_cols if c in raw.schema.names]
    arrow_table = raw.select(available)

    table.append(arrow_table)
    snap_id = table.current_snapshot().snapshot_id
    print(
        f"appended {arrow_table.num_rows} rows; "
        f"current_snapshot_id = {snap_id}"
    )


if __name__ == "__main__":
    append_partitions()