# KinetiShift change log -> Lakeflow Declarative Pipelines AUTO CDC (Python source, list of tables).
#
# KinetiShift (--sink databricks --databricks-write changelog) appends every source change to
# <catalog>.<schema>.<table>__changes: _op ('I'/'U'/'D'/'T'), _commit_scn (STRING), _txid, _seq
# (BIGINT, position in the source transaction), _rowid (informational), then the source columns.
# This file builds, for each configured table, a current-state table (SCD type 1, default) or a
# history table (SCD type 2) with AUTO CDC.
#
# Pipeline configuration (Settings > Advanced > Configuration):
#   kinetishift.source_catalog  catalog of the change-log tables (optional if fully-qualified below)
#   kinetishift.source_schema   schema of the change-log tables   (optional if fully-qualified below)
#   kinetishift.tables          JSON list, one entry per table, e.g.
#     [{"table": "orders", "keys": ["ORDER_ID"]},
#      {"table": "order_lines", "keys": ["ORDER_ID", "LINE_NO"]},
#      {"table": "orders", "keys": ["ORDER_ID"], "scd": 2, "target": "orders_history"}]
#   Per entry: "table" (source table name; the change log is <table>__changes), "keys" (primary-key
#   columns), optional "scd" (1 or 2, default 1), optional "target" (default: <table> for SCD 1,
#   <table>_history for SCD 2), optional "source" (full change-log table name, overrides the default).
#
# Order: by commit SCN, then _seq. _commit_scn is a STRING and must be cast to a number (a text
# comparison would order '10' before '9'). The pair is folded into ONE DECIMAL(38,0) value,
#   CAST(_commit_scn AS DECIMAL(38,0)) * 10^12 + _seq
# (exact for any SCN up to 2^64 and fewer than 10^12 changes per transaction), because AUTO CDC's
# apply_as_truncates does not accept a struct sequence (UNSUPPORTED_FEATURE.LITERAL_TYPE).

import json

from pyspark.sql import functions as F

try:  # Lakeflow Declarative Pipelines (current module name)
    from pyspark import pipelines as dp

    _view = dp.temporary_view
except ImportError:  # earlier runtimes
    import dlt as dp

    _view = dp.view

KINETISHIFT_META_COLUMNS = ["_op", "_commit_scn", "_txid", "_seq", "_rowid"]

_tables = json.loads(spark.conf.get("kinetishift.tables"))  # noqa: F821 (spark is provided)
_catalog = spark.conf.get("kinetishift.source_catalog", "")  # noqa: F821
_schema = spark.conf.get("kinetishift.source_schema", "")  # noqa: F821


def _sequence():
    return F.expr("CAST(CAST(_commit_scn AS DECIMAL(38,0)) * 1000000000000 + _seq AS DECIMAL(38,0))")


def _define(entry):
    table = entry["table"]
    keys = entry["keys"]
    scd = int(entry.get("scd", 1))
    if scd not in (1, 2):
        raise ValueError(f"kinetishift.tables: scd must be 1 or 2 for {table}")
    target = entry.get("target") or (table if scd == 1 else f"{table}_history")
    source = entry.get("source") or ".".join(
        p for p in (_catalog, _schema, f"{table}__changes") if p
    )
    view_name = f"{target}_kinetishift_changes"

    @_view(name=view_name)
    def _changes():
        df = spark.readStream.table(source)  # noqa: F821
        # AUTO CDC supports TRUNCATE only for SCD type 1; drop 'T' rows for a history table.
        return df if scd == 1 else df.where(F.col("_op") != "T")

    dp.create_streaming_table(
        name=target,
        comment=f"Built by AUTO CDC (SCD type {scd}) from the KinetiShift change log {source}",
    )
    dp.create_auto_cdc_flow(
        target=target,
        source=view_name,
        keys=keys,
        sequence_by=_sequence(),
        apply_as_deletes=F.expr("_op = 'D'"),
        apply_as_truncates=F.expr("_op = 'T'") if scd == 1 else None,
        except_column_list=KINETISHIFT_META_COLUMNS,
        stored_as_scd_type=scd,
        name=f"{target}_kinetishift_auto_cdc",
    )


for _entry in _tables:
    _define(_entry)