delta_cdf incremental strategy — Delta CDF POC (draft)

dbt_macros/dune/adapters.sql

@@ -3,6 +3,11 @@
   {%- if config.get('partition_by', None) != None -%}
     {%- do _properties.update({'partitioned_by': "ARRAY['" + (config.get('partition_by') | join("', '") )  + "']"}) -%}
   {%- endif -%}
+  {#-- [dune] forward change_data_feed_enabled (Delta CDF) from model config; never on the temp
+        relation, since the delta_cdf strategy materializes its temp as a table (incremental.sql:14/17). --#}
+  {%- if not temporary and config.get('change_data_feed_enabled', None) != None -%}
+    {%- do _properties.update({'change_data_feed_enabled': config.get('change_data_feed_enabled') | string | lower}) -%}
+  {%- endif -%}
   create or replace table {{ relation }}
     {{ create_table_properties(_properties, relation) }}
   as (
@@ -46,7 +51,12 @@
 {%- endmacro -%}
 
 {% macro create_table_properties(_properties, relation) %}
-  {%- if not (target.name == 'ci' and target.database == 'dune') -%}
+  {#- CDF tables must use a connector-managed location: an explicit `location` bypasses
+      Dune catalog credential vending (VendedCredentialsHandle.empty) so the S3 write
+      fails, and it also triggers a stats-cache read at getStatisticsCollectionMetadataForWrite
+      that errors on the explicit path. Managed location is vended via schema registration. -#}
+  {%- set cdf_enabled = config.get('change_data_feed_enabled', false) -%}
+  {%- if not (target.name == 'ci' and target.database == 'dune') and not cdf_enabled -%}
     {%- set modified_identifier = relation.identifier | replace("__dbt_tmp", "") -%}
     {%- set unique_location = modified_identifier ~ '_' ~ time_salted_md5_prefix() -%}
     {%- set location= 's3a://%s/%s/%s' % (s3_bucket(), relation.schema, unique_location) -%}

dbt_macros/dune/cdf/cdf_advance_watermark.sql

@@ -0,0 +1,33 @@
+{#-
+  Emit an ALTER TABLE ... SET PROPERTIES that stamps dune.cdf.source_version = <version>
+  while preserving any existing dune.* extra_properties keys. SET PROPERTIES
+  extra_properties REPLACES the whole custom map, so every dune.* key must be re-listed.
+  Filtering key LIKE 'dune.%' isolates Dune's custom metadata from native delta.* keys.
+
+  Accepts either a captured literal (incremental: max(_commit_version) of the applied
+  set) or render-time V (bootstrap). In dev the only dune.* key present is the watermark
+  itself (mark_as_spell / expose_spells are prod-gated no-ops), so the preserve-loop is
+  empty; the loop keeps the write forward-compatible for the Phase 2 prod path.
+
+  NOTE: values are interpolated as single-quoted literals; Dune's dune.* values are
+  numbers / booleans / double-quoted JSON, so they contain no single quotes. Revisit if
+  a single-quote-bearing key is ever stored here.
+-#}
+{% macro cdf_advance_watermark(target_relation, version) -%}
+  {%- set entries = [] -%}
+  {%- if execute -%}
+    {%- set probe -%}
+      select key, value
+      from {{ target_relation.database }}.{{ target_relation.schema }}."{{ target_relation.identifier }}$properties"
+      where key like 'dune.%' and key <> 'dune.cdf.source_version'
+    {%- endset -%}
+    {%- set existing = run_query(probe) -%}
+    {%- if existing is not none -%}
+      {%- for row in existing.rows -%}
+        {%- do entries.append("ROW('" ~ row[0] ~ "', '" ~ row[1] ~ "')") -%}
+      {%- endfor -%}
+    {%- endif -%}
+  {%- endif -%}
+  {%- do entries.append("ROW('dune.cdf.source_version', '" ~ version ~ "')") -%}
+  alter table {{ target_relation }} set properties extra_properties = map_from_entries(ARRAY[{{ entries | join(', ') }}])
+{%- endmacro %}

dbt_macros/dune/cdf/cdf_current_source_version.sql

@@ -0,0 +1,22 @@
+{#-
+  Current max committed Delta version of a spell source, via its "$history" table.
+  Cheap for spell sources (one commit per cadence run); NOT safe for high-write raw
+  tables (Phase 2 uses an error-sentinel probe instead). Used to bootstrap the
+  watermark and to pin the bootstrap snapshot (FOR VERSION AS OF V). Returns an int
+  or none. Guarded for parse mode.
+-#}
+{% macro cdf_current_source_version(base_relation) %}
+  {%- if not execute -%}
+    {{ return(none) }}
+  {%- endif -%}
+  {%- set probe -%}
+    select max(version) as v
+    from {{ base_relation.database }}.{{ base_relation.schema }}."{{ base_relation.identifier }}$history"
+  {%- endset -%}
+  {%- set results = run_query(probe) -%}
+  {%- if results is not none and results.rows | length > 0 and results.rows[0][0] is not none -%}
+    {{ return(results.rows[0][0] | int) }}
+  {%- else -%}
+    {{ return(none) }}
+  {%- endif -%}
+{% endmacro %}

dbt_macros/dune/cdf/cdf_get_watermark.sql

@@ -0,0 +1,23 @@
+{#-
+  Read the stored CDF watermark (dune.cdf.source_version) from a target relation's
+  Delta table properties. Returns the integer source version W, or none when absent
+  (the caller should then bootstrap). Trino-Delta specific: reads the "$properties"
+  metadata table, which exposes the Delta metaData.configuration map (where Dune's
+  extra_properties land). Guarded for parse mode (no introspective query).
+-#}
+{% macro cdf_get_watermark(target_relation) %}
+  {%- if not execute -%}
+    {{ return(none) }}
+  {%- endif -%}
+  {%- set probe -%}
+    select value
+    from {{ target_relation.database }}.{{ target_relation.schema }}."{{ target_relation.identifier }}$properties"
+    where key = 'dune.cdf.source_version'
+  {%- endset -%}
+  {%- set results = run_query(probe) -%}
+  {%- if results is not none and results.rows | length > 0 and results.rows[0][0] is not none -%}
+    {{ return(results.rows[0][0] | int) }}
+  {%- else -%}
+    {{ return(none) }}
+  {%- endif -%}
+{% endmacro %}

dbt_macros/dune/cdf/cdf_macros.yml

@@ -0,0 +1,48 @@
+version: 2
+
+macros:
+  - name: source_changes
+    description: >
+      Expand a ref-based Delta spell source into a CDF change-set relation for a
+      delta_cdf incremental model. Bootstraps a full snapshot (FOR VERSION AS OF V)
+      on first build / --full-refresh, otherwise reads table_changes(since_version => W)
+      strictly after the stored watermark. Carries _change_type / _commit_version /
+      _commit_timestamp through for the strategy macro to capture.
+    arguments:
+      - name: base_relation
+        type: relation
+        description: "ref()-resolved physical Delta spell to read changes from (must have change_data_feed_enabled)."
+      - name: change_types
+        type: list
+        description: "_change_type values to keep on the incremental path. Default ['insert','update_postimage']."
+
+  - name: cdf_get_watermark
+    description: >
+      Read dune.cdf.source_version (the exclusive source version W) from a target
+      relation's Delta $properties. Returns an int or none (caller bootstraps).
+    arguments:
+      - name: target_relation
+        type: relation
+        description: "The delta_cdf target whose watermark to read (typically this)."
+
+  - name: cdf_current_source_version
+    description: >
+      Current max committed Delta version of a spell source via its $history table.
+      Spell-only (cheap); used to bootstrap the watermark and pin the bootstrap snapshot.
+    arguments:
+      - name: base_relation
+        type: relation
+        description: "ref()-resolved physical Delta spell source."
+
+  - name: cdf_advance_watermark
+    description: >
+      Emit ALTER TABLE ... SET PROPERTIES that stamps dune.cdf.source_version = version,
+      preserving existing dune.* extra_properties keys. Used by the strategy macro
+      (incremental, captured literal) and the bootstrap post-hook (render-time V).
+    arguments:
+      - name: target_relation
+        type: relation
+        description: "The delta_cdf target to stamp."
+      - name: version
+        type: integer
+        description: "The version literal to store as the new watermark."

dbt_macros/dune/cdf/source_changes.sql

@@ -0,0 +1,88 @@
+{#-
+  Expand a ref-based Delta spell source into a CDF change-set relation for a
+  delta_cdf incremental model. Carries _change_type / _commit_version /
+  _commit_timestamp through so the strategy macro (Task B) can capture the applied
+  max(_commit_version). Three branches:
+
+    - parse mode (not execute): a static, compilable stub. No introspection.
+    - bootstrap (target absent OR --full-refresh => is_incremental() is false): full
+      snapshot of the base pinned at its current version V (FOR VERSION AS OF V), every
+      row tagged 'insert', _commit_version = V.
+    - incremental: <catalog>.system.table_changes(since_version => W), W = the stored
+      watermark. since_version is an EXCLUSIVE lower bound, so the next run reads strictly
+      after the last applied version (no overlap, no skip).
+
+  Column contract: both real branches append, in order, payload cols..., _change_type,
+  _commit_version, _commit_timestamp. The CONSUMING model must keep _commit_version in
+  its FINAL projection only on the incremental path (is_incremental()); on bootstrap it
+  must drop the CDF metadata columns so the target table schema stays clean (the
+  bootstrap CTAS output becomes the table verbatim).
+
+  The catalog is taken from base_relation.database (NOT hardcoded delta_prod) so this
+  works in dev/CI where ref() resolves to the dev catalog. table_changes takes
+  schema_name + table_name only; the catalog lives in the function path.
+
+  Trino-Delta specific; the worker is named trino__ for a future adapter.dispatch swap.
+-#}
+{% macro source_changes(base_relation, change_types=['insert', 'update_postimage']) -%}
+{{ trino__source_changes(base_relation, change_types) }}
+{%- endmacro %}
+
+{% macro trino__source_changes(base_relation, change_types) -%}
+{%- if not execute -%}
+{#-- parse-mode stub: valid, compilable, no introspection --#}
+select *
+  , cast('insert' as varchar) as _change_type
+  , cast(0 as bigint) as _commit_version
+  , cast(null as timestamp(3) with time zone) as _commit_timestamp
+from {{ base_relation }}
+{%- elif not is_incremental() -%}
+{#-- bootstrap: full snapshot pinned at the source's current version, tagged as inserts --#}
+{%- set v = cdf_current_source_version(base_relation) -%}
+{%- if v is none -%}
+{{ exceptions.raise_compiler_error("source_changes: cannot resolve $history version to bootstrap " ~ base_relation ~ " (enable change_data_feed_enabled on the source and ensure it is built first)") }}
+{%- endif -%}
+select *
+  , cast('insert' as varchar) as _change_type
+  , cast({{ v }} as bigint) as _commit_version
+  , cast(null as timestamp(3) with time zone) as _commit_timestamp
+from {{ base_relation }} for version as of {{ v }}
+{%- else -%}
+{#-- incremental: change feed strictly after the stored watermark.
+     table_changes() returns Dune uint256/int256 columns as their raw big-endian
+     varbinary (the logical type annotation is lost through the table function), unlike a
+     normal scan. Re-decode those columns via bytearray_to_uint256 / bytearray_to_int256
+     so the feed schema is identical to the bootstrap / base output; every other column
+     (incl. genuine varbinary like addresses/hashes) passes through untouched. Column
+     types come from information_schema, where the logical uint256/int256 IS preserved. --#}
+{%- set w = cdf_get_watermark(this) -%}
+{%- if w is none -%}
+{{ exceptions.raise_compiler_error("source_changes: target " ~ this ~ " is incremental but has no dune.cdf.source_version watermark; run with --full-refresh to bootstrap") }}
+{%- endif -%}
+{%- set change_types_csv = "'" ~ (change_types | join("', '")) ~ "'" -%}
+{%- set col_rows = run_query(
+      "select column_name, data_type from " ~ base_relation.database
+      ~ ".information_schema.columns where table_schema = '" ~ base_relation.schema
+      ~ "' and table_name = '" ~ base_relation.identifier ~ "' order by ordinal_position") -%}
+{%- set decoded = [] -%}
+{%- for r in col_rows.rows -%}
+  {%- set cname = r[0] -%}
+  {%- set ctype = (r[1] | lower) -%}
+  {%- if 'uint256' in ctype -%}
+    {%- do decoded.append("bytearray_to_uint256(" ~ adapter.quote(cname) ~ ") as " ~ adapter.quote(cname)) -%}
+  {%- elif 'int256' in ctype -%}
+    {%- do decoded.append("bytearray_to_int256(" ~ adapter.quote(cname) ~ ") as " ~ adapter.quote(cname)) -%}
+  {%- else -%}
+    {%- do decoded.append(adapter.quote(cname)) -%}
+  {%- endif -%}
+{%- endfor -%}
+select {{ decoded | join(', ') }}
+  , _change_type, _commit_version, _commit_timestamp
+from table({{ base_relation.database }}.system.table_changes(
+  schema_name => '{{ base_relation.schema }}',
+  table_name => '{{ base_relation.identifier }}',
+  since_version => {{ w | int }}
+))
+where _change_type in ({{ change_types_csv }})
+{%- endif -%}
+{%- endmacro %}

dbt_macros/dune/get_incremental_delta_cdf_sql.sql

@@ -0,0 +1,136 @@
+{#-
+  delta_cdf incremental strategy. Dispatched by the trino incremental materialization
+  because a model sets incremental_strategy='delta_cdf' and dbt finds this macro by name
+  (impl.py:1782 -> get_incremental_<strategy>_sql). 'delta_cdf' is not a builtin, so the
+  adapter allowlist check is skipped (impl.py:1776). No adapter fork.
+
+  The temp_relation is a MATERIALIZED TABLE for this strategy (incremental.sql:14/17:
+  delta_cdf is not in default/append/merge, and unique_key is set), so probing it is a
+  cheap scan, not a feed re-execution. The model body (via transfers_enrich_cdf ->
+  source_changes) carries _change_type and _commit_version into temp on the incremental
+  path.
+
+  Steps:
+    1. run_query max(_commit_version) of the applied change set (and assert no unexpected
+       deletes: spellbook merge bases never emit delete change rows).
+    2. dedup temp to the latest change per unique_key (MERGE requires <=1 source row per
+       target row, and a key can change across multiple commits in one window).
+    3. MERGE upsert against dest_columns; the extra _change_type/_commit_version temp
+       columns are inert (on_schema_change='ignore' keeps dest_columns = target columns).
+       No time-window incremental_predicates on the ON clause (CDF can legitimately update
+       arbitrarily old rows). Instead the ON clause is augmented with a partition-range
+       bound DBT_INTERNAL_DEST.<partcol> BETWEEN min..max read from the change set: a
+       changed row's partition value is immutable, so this prunes only target partitions
+       that cannot contain a match (never dropping a real match) and bounds the target
+       MERGE scan to the touched partitions.
+    4. emit "MERGE ... ; ALTER TABLE ... SET PROPERTIES dune.cdf.source_version=<max_v>"
+       as two statements (dbt-trino splits on ';' via sqlparse, connections.py:547).
+       Skip the ALTER on an empty feed (max_v IS NULL) so the watermark holds.
+-#}
+{% macro get_incremental_delta_cdf_sql(arg_dict) %}
+  {{ return(trino__get_incremental_delta_cdf_sql(arg_dict)) }}
+{% endmacro %}
+
+{% macro trino__get_incremental_delta_cdf_sql(arg_dict) -%}
+  {%- set target = arg_dict["target_relation"] -%}
+  {%- set temp = arg_dict["temp_relation"] -%}
+  {%- set unique_key = arg_dict["unique_key"] -%}
+  {%- set dest_columns = arg_dict["dest_columns"] -%}
+  {%- set cdf_apply_deletes = config.get('cdf_apply_deletes', false) -%}
+
+  {%- if unique_key is string -%}
+    {%- set unique_key_cols = [unique_key] -%}
+  {%- else -%}
+    {%- set unique_key_cols = unique_key -%}
+  {%- endif -%}
+
+  {#-- partition columns are immutable per row -> their change-set range prunes the target scan --#}
+  {%- set partition_by = config.get('partition_by') -%}
+  {%- if partition_by is string -%}
+    {%- set partition_cols = [partition_by] -%}
+  {%- elif partition_by -%}
+    {%- set partition_cols = partition_by -%}
+  {%- else -%}
+    {%- set partition_cols = [] -%}
+  {%- endif -%}
+
+  {#-- 1. capture applied max(_commit_version) + partition-range bounds; assert no unexpected deletes --#}
+  {%- set max_v = none -%}
+  {%- set prune_preds = [] -%}
+  {%- if execute -%}
+    {%- set probe -%}
+      select max(_commit_version) as max_v, count_if(_change_type = 'delete') as n_del
+      {%- for pc in partition_cols %}, min({{ pc }}) as min_{{ loop.index0 }}, max({{ pc }}) as max_{{ loop.index0 }}{% endfor %}
+      from {{ temp }}
+    {%- endset -%}
+    {%- set res = run_query(probe) -%}
+    {%- if res is not none and res.rows | length > 0 -%}
+      {%- set max_v = res.rows[0][0] -%}
+      {%- set n_del = res.rows[0][1] -%}
+      {%- if n_del is not none and n_del | int > 0 and not cdf_apply_deletes -%}
+        {{ exceptions.raise_compiler_error("delta_cdf: " ~ n_del ~ " delete change rows reached " ~ target ~ " but cdf_apply_deletes is false. spellbook merge bases never delete; investigate the source feed.") }}
+      {%- endif -%}
+      {%- for pc in partition_cols -%}
+        {%- set lo = res.rows[0][2 + 2 * loop.index0] -%}
+        {%- set hi = res.rows[0][3 + 2 * loop.index0] -%}
+        {%- if lo is not none and hi is not none -%}
+          {%- do prune_preds.append("DBT_INTERNAL_DEST." ~ adapter.quote(pc) ~ " between " ~ cdf_partition_literal(lo) ~ " and " ~ cdf_partition_literal(hi)) -%}
+        {%- endif -%}
+      {%- endfor -%}
+    {%- endif -%}
+  {%- endif -%}
+
+  {%- set dest_cols_csv = get_quoted_csv(dest_columns | map(attribute="name")) -%}
+  {%- set dest_cols_list = dest_cols_csv.split(', ') -%}
+  {%- set update_columns = get_merge_update_columns(config.get('merge_update_columns'), config.get('merge_exclude_columns'), dest_columns) -%}
+
+  {#-- 2. dedup temp to latest change per unique_key --#}
+  {%- set deduped_source -%}
+    select {{ dest_cols_csv }}
+    from (
+      select {{ dest_cols_csv }}, _change_type, _commit_version,
+             row_number() over (
+               partition by {{ unique_key_cols | join(', ') }}
+               order by _commit_version desc, _change_type desc
+             ) as _cdf_rn
+      from {{ temp }}
+      {%- if not cdf_apply_deletes %}
+      where _change_type <> 'delete'
+      {%- endif %}
+    )
+    where _cdf_rn = 1
+  {%- endset -%}
+
+  {#-- 3. MERGE upsert --#}
+  merge into {{ target }} as DBT_INTERNAL_DEST
+    using ( {{ deduped_source }} ) as DBT_INTERNAL_SOURCE
+    on {% for k in unique_key_cols %}(DBT_INTERNAL_SOURCE.{{ k }} = DBT_INTERNAL_DEST.{{ k }}){% if not loop.last %} and {% endif %}{% endfor %}{% for p in prune_preds %} and {{ p }}{% endfor %}
+  when matched then update set
+    {% for column_name in update_columns -%}
+      {{ column_name }} = DBT_INTERNAL_SOURCE.{{ column_name }}{% if not loop.last %}, {% endif %}
+    {%- endfor %}
+  when not matched then insert
+    ({{ dest_cols_csv }})
+    values
+    ({% for col in dest_cols_list -%}
+      DBT_INTERNAL_SOURCE.{{ col }}{% if not loop.last %}, {% endif %}
+    {%- endfor %})
+  {%- if execute and max_v is not none %}
+  ;
+  {#-- 4. advance the watermark to the exact max applied version --#}
+  {{ cdf_advance_watermark(target, max_v | int) }}
+  {%- endif %}
+{%- endmacro %}
+
+{#-- render a partition-column value as a Trino literal for the prune predicate. Avoids
+    dunder access (dbt's Jinja sandbox blocks __class__): a date/timestamp falls through to
+    the string-repr branch and is wrapped by date '' / timestamp '' based on a time part. --#}
+{% macro cdf_partition_literal(v) -%}
+  {%- if v is none -%}null
+  {%- elif v is string -%}'{{ v }}'
+  {%- elif v is number -%}{{ v }}
+  {%- else -%}
+    {%- set s = v | string -%}
+    {%- if ':' in s -%}timestamp '{{ s }}'{%- else -%}date '{{ s }}'{%- endif -%}
+  {%- endif -%}
+{%- endmacro %}