-- KinetiShift change log -> Lakeflow Declarative Pipelines AUTO CDC (SQL source, one table).
--
-- KinetiShift (--sink databricks --databricks-write changelog) appends every source change to
-- <catalog>.<schema>.<table>__changes with the columns
--   _op          STRING  'I' insert, 'U' update, 'D' delete, 'T' truncate
--   _commit_scn  STRING  commit SCN of the source transaction (a number, stored as text)
--   _txid        STRING  source transaction id
--   _seq         BIGINT  position of the change within its source transaction
--   _rowid       STRING  Oracle ROWID when available (informational, not a key)
-- followed by the source columns (STRING, or BOOLEAN). This flow builds the current-state table
-- (SCD type 1) from that log.
--
-- Order: by commit SCN, then _seq. _commit_scn is a STRING, so it 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 TRUNCATE does not accept a STRUCT sequence (UNSUPPORTED_FEATURE.LITERAL_TYPE). Without
-- TRUNCATE, STRUCT(CAST(_commit_scn AS DECIMAL(38,0)), _seq) orders the same way.
--
-- Pipeline configuration (Settings > Advanced > Configuration):
--   kinetishift.source  full name of the change-log table, e.g. main.sales.orders__changes
--   kinetishift.target  name of the table to build in the pipeline's target schema, e.g. orders
--   kinetishift.keys    primary-key column(s), comma-separated, e.g. ORDER_ID or ORDER_ID, LINE_NO
--
-- SQL cannot loop over a list of tables. For several tables, copy this file once per table with its
-- own configuration keys, or use auto_cdc_pipeline.py, which reads a list.

CREATE OR REFRESH STREAMING TABLE ${kinetishift.target}
COMMENT 'Current state built by AUTO CDC from the KinetiShift change log ${kinetishift.source}';

CREATE FLOW ${kinetishift.target}_kinetishift_auto_cdc AS AUTO CDC INTO ${kinetishift.target}
FROM STREAM(${kinetishift.source})
KEYS (${kinetishift.keys})
APPLY AS DELETE WHEN _op = 'D'
APPLY AS TRUNCATE WHEN _op = 'T'
SEQUENCE BY CAST(CAST(_commit_scn AS DECIMAL(38,0)) * 1000000000000 + _seq AS DECIMAL(38,0))
COLUMNS * EXCEPT (_op, _commit_scn, _txid, _seq, _rowid)
STORED AS SCD TYPE 1;

-- SCD type 2 variant (row history with __START_AT / __END_AT). AUTO CDC does not support TRUNCATE
-- for SCD type 2, so 'T' rows are filtered out in a view first. To use it, replace the two
-- statements above with:
--
-- CREATE TEMPORARY VIEW ${kinetishift.target}_kinetishift_changes AS
--   SELECT * FROM STREAM(${kinetishift.source}) WHERE _op <> 'T';
--
-- CREATE OR REFRESH STREAMING TABLE ${kinetishift.target};
--
-- CREATE FLOW ${kinetishift.target}_kinetishift_auto_cdc AS AUTO CDC INTO ${kinetishift.target}
-- FROM STREAM(${kinetishift.target}_kinetishift_changes)
-- KEYS (${kinetishift.keys})
-- APPLY AS DELETE WHEN _op = 'D'
-- SEQUENCE BY CAST(CAST(_commit_scn AS DECIMAL(38,0)) * 1000000000000 + _seq AS DECIMAL(38,0))
-- COLUMNS * EXCEPT (_op, _commit_scn, _txid, _seq, _rowid)
-- STORED AS SCD TYPE 2;