logic

def aggregate_unsafe(
    data: Iterator[StatementDict], dataset: str | None = None
) -> Iterator[EntityPayload]:
    """
    Aggregate statement dicts (e.g. from DuckDB rows) to entity payloads.

    Completely circumvents the dict -> Statement -> StatementEntity -> dict
    Python path, but therefore has no validation checks. Input must be sorted
    by entity_id.
    """
    current: EntityPayload | None = None
    for statement in data:
        if current is None or statement["entity_id"] != current.id:
            if current is not None:
                yield current
            current = EntityPayload(id=statement["entity_id"], dataset=dataset)
        current.add(statement)
    if current is not None:
        yield current

def duckdb_config() -> dict[str, str]:
    """LakeStore DuckDB config derived from lakehouse settings.

    Per-query memory is bounded by `Settings.duckdb_memory_limit`
    (env: ``LAKEHOUSE_DUCKDB_MEMORY_LIMIT``, default ``8GB``); queries
    exceeding the limit spill to `Settings.duckdb_temp_directory`
    (env: ``LAKEHOUSE_DUCKDB_TEMP_DIRECTORY``) when set, otherwise to
    the OS temp directory DuckDB picks by default. Extensions (notably
    ``delta``) are loaded from `Settings.duckdb_extension_directory`
    (env: ``LAKEHOUSE_DUCKDB_EXTENSION_DIRECTORY``) when set, otherwise
    from ``$HOME/.duckdb/extensions``. Passed to
    `LakeStore` via the ``duckdb_config`` kwarg.
    """
    settings = Settings()
    config: dict[str, str] = {"memory_limit": settings.duckdb_memory_limit}
    if settings.duckdb_temp_directory:
        config["temp_directory"] = settings.duckdb_temp_directory
    if settings.duckdb_extension_directory:
        config["extension_directory"] = settings.duckdb_extension_directory
    return config

def raw_view_sql(dt: DeltaTable) -> str:
    """SELECT body for the ``statement_raw`` view.

    Surfaces every physical row in the Delta table, including
    tombstones and pre-merge duplicates. Used by [`build_merge_sql`][build_merge_sql]
    and raw-source queries (diff exports) – any path that needs the
    physical layout visible.
    """
    return f"SELECT * FROM {_delta_scan_clause(dt)}"

def live_view_sql(dt: DeltaTable) -> str:
    """SELECT body for the live ``statement`` view.

    On a store kept canonical by [`build_merge_sql`][build_merge_sql] (one row per
    statement id, fragment supersession applied, ``first_seen`` /
    ``last_seen`` folded) the live rows are simply the non-tombstoned
    physical rows – so the view is a plain filtered scan, no window
    function. Predicate pushdown works natively: ``schema`` / ``prop`` /
    ``entity_id`` filters reach ``delta_scan``'s per-file statistics (a
    window would be a pushdown barrier for any non-partition column).

    ``canonical_id`` is not stored – this is a single-dataset store with no
    entity resolution, so it always equals ``entity_id`` – and is synthesised
    here as ``entity_id AS canonical_id`` so ftmq's query layer (which keys
    entity identity on ``canonical_id``) resolves against the view unchanged.
    [`raw_view_sql`][raw_view_sql] deliberately omits it so ``merge`` never materialises
    the duplicate column.

    Correctness holds only on an **optimized** store: between a write and
    the next `merge` this view can surface duplicate ids and rows
    whose delete has not been applied yet. Run ``optimize`` before
    querying – the dedupe / supersession / grace logic lives solely in
    [`build_merge_sql`][build_merge_sql].
    """
    return (
        f"SELECT *, entity_id AS canonical_id "
        f"FROM {_delta_scan_clause(dt)} WHERE deleted_at IS NULL"
    )

def build_merge_sql(
    shard: str,
    bucket: str,
    origin: str,
    grace_cutoff: datetime,
    entity_id_range: tuple[str | None, str | None] = (None, None),
) -> str:
    """DuckDB SQL that collapses one partition for physical merge.

    `_dedupe_sql` over the **raw** ``statement_raw`` view (not the
    deduped ``statement``) because ``merge`` needs every row visible –
    including tombstones within the grace window, which must persist
    physically to keep shadowing their live rows – scoped to one
    ``(shard, bucket, origin)`` partition. Output is ordered by
    ``(entity_id, fragment, role, prop, id, last_seen DESC)`` – the file sort
    key – so the rewritten parquet file is ready for future merges
    without re-sort.

    Args:
        shard: Target shard value (hex-padded).
        bucket: Target bucket (``thing`` / ``interval`` / ``document`` /
            ``page`` / ``pages`` / ``mention``).
        origin: Target origin tag – re-validated here, so it is safe to
            interpolate: `validate_origin` rejects quote characters.
        grace_cutoff: Tombstones with ``deleted_at <= grace_cutoff`` are
            dropped. Typically ``now - LAKEHOUSE_GRACE_PERIOD_DAYS``.
        entity_id_range: Optional half-open ``[lo, hi)`` bound on
            ``entity_id`` (``None`` = unbounded on that side) scoping the
            merge to one range slice (`slice_ranges`). Every dedupe
            group is a function of a single entity – the non-fragment key
            ends in the statement ``id`` (owned by exactly one entity),
            the fragment key contains ``entity_id`` itself – so an
            ``entity_id`` predicate can never split a group.

    Returns:
        Executable DuckDB SQL.
    """
    origin = validate_origin(origin)
    lo, hi = entity_id_range
    where = f"WHERE shard = '{shard}' AND bucket = '{bucket}' AND origin = '{origin}'"
    if lo is not None:
        where += f" AND entity_id >= '{_string_literal(lo)}'"
    if hi is not None:
        where += f" AND entity_id < '{_string_literal(hi)}'"
    return _dedupe_sql(
        source=TABLE_RAW.name,
        where=where,
        tombstone=(
            "(deleted_at IS NULL OR deleted_at > "
            f"TIMESTAMPTZ '{grace_cutoff.isoformat()}')"
        ),
        order_by="ORDER BY entity_id, fragment, role, prop, id, last_seen DESC",
    )

def build_shard_sql(shard: str, bucket: str, origin: str, shards: int) -> str:
    """DuckDB SQL re-keying one partition's rows onto ``shards`` shards.

    ``SELECT *`` over the **raw** ``statement_raw`` view (tombstones and
    pre-merge duplicates included – a re-shard moves rows, it does not
    decide what survives) with the stored ``shard`` swapped for the one
    [`shard_expr_sql`][shard_expr_sql] computes from ``entity_id``. ``REPLACE`` keeps
    the projection positional, so the result still *is*
    `SHARDED_SCHEMA` and streams
    straight into ``write_deltalake``.

    Deliberately unordered and un-deduped: the caller
    ([`shard`][ftm_lakehouse.storage.parquet.ParquetStore.shard]) re-stamps
    every rewritten partition as dirty, so the next ``merge`` restores the
    file sort order – paying for a sort here would only make the rewrite
    slower.

    Args:
        shard: Source shard value (hex-padded) to read.
        bucket: Source bucket – invariant under re-sharding.
        origin: Source origin tag – invariant under re-sharding.
            Re-validated here, so it is safe to interpolate.
        shards: Target shard count.

    Returns:
        Executable DuckDB SQL.
    """
    origin = validate_origin(origin)
    return (
        f"SELECT * REPLACE ({shard_expr_sql(shards)} AS shard) "
        f"FROM {TABLE_RAW.name} "
        f"WHERE shard = '{shard}' AND bucket = '{bucket}' AND origin = '{origin}'"
    )

def shard_expr_sql(shards: int, column: str = "entity_id") -> str:
    """DuckDB expression computing the shard key of ``column``.

    The SQL twin of `helpers.shards.entity_shard`,
    used by [`build_shard_sql`][build_shard_sql] so a re-shard recomputes every row's
    partition inside DuckDB's vectorised pipeline instead of marshaling
    ids into Python. ``banal.hash_data`` of a ``str`` is a plain SHA-1 of
    its UTF-8 bytes, which is exactly what DuckDB's ``sha1()`` returns –
    the two are pinned to agree by
    ``tests/test_logic_parquet.py::test_shard_expr_sql_parity``.

    Args:
        shards: Target shard count; ``<= 1`` collapses to the constant
            single-shard key, matching ``entity_shard``.
        column: Column holding the entity id.

    Returns:
        A DuckDB scalar expression yielding the hex-padded shard key.
    """
    if shards <= 1:
        return "'0'"
    width = shard_hex_width(shards)
    return (
        f"printf('%0{width}x', "
        f"(('0x' || substr(sha1({column}), 1, 8))::BIGINT) % {int(shards)})"
    )