Snapshots Error: Correlated subqueries are not supported in JOINs yet

[enqueue]

Describe the bug

After the initial dbt snapshot run, an error occurs:

DB::Exception: Correlated subqueries are not supported in JOINs yet,
but found in expression: snapshotted_data

Steps to reproduce

1. Create model foo
2. Create snapshot foo_snapshot (see YAML below)
3. Run dbt snapshot
4. Run dbt snapshot

Expected behaviour

Snapshot table being updated.

Code examples, such as models or profile settings

I have tried to obfuscate our table names manually. I hope I did not break anything else while doing so.

Source SQL

{{
  config(
    materialized = 'table'
  )
}}


with foos as (
    select *
    from {{ ref('stg_foos') }}
)

select
    foo_id,
    foo_data,
    toDateTime64(coalesce(last_modified_time, '1970-01-01 00:00:00'), 3)
        as last_update,
    toStartOfDay(last_update) as last_update_day
from foos

Snapshot YAML

snapshots:
  - name: foo_snapshot
    relation: ref('sourceschema', 'foo')
    config:
      database: targetschema
      unique_key: foo_id
      strategy: timestamp
      updated_at: last_update
      dbt_valid_to_current: "toDate('2149-06-06')"
      snapshot_meta_column_names:
        dbt_valid_from: valid_from
        dbt_valid_to: valid_to
        dbt_is_deleted: is_locked

dbt and/or ClickHouse server logs

Error message

  Code: 48.
  DB::Exception: Correlated subqueries are not supported in JOINs yet, but found in expression:  snapshotted_data. Stack trace:
  
  0. ? @ 0x0000000013c42d9f
  1. ? @ 0x000000000c89690e
  2. ? @ 0x000000000c8963c0
  3. ? @ 0x000000000c895f2b
  4. ? @ 0x0000000017ddaee8
  5. ? @ 0x0000000017dd6378
  6. ? @ 0x0000000017da69ef
  7. ? @ 0x0000000017db1161
  8. ? @ 0x0000000017dd6977
  9. ? @ 0x0000000017da69ef
  10. ? @ 0x0000000017dac084
  11. ? @ 0x0000000017da41cd
  12. ? @ 0x0000000017da381c
  13. ? @ 0x0000000017df9736
  14. ? @ 0x0000000018949c8d
  15. ? @ 0x0000000018947baa
  16. ? @ 0x000000001894a0ea
  17. ? @ 0x000000001886b9f3
  18. ? @ 0x00000000188747e5
  19. ? @ 0x000000001888a778
  20. ? @ 0x0000000018d1fe44
  21. ? @ 0x0000000018d19c73
  22. ? @ 0x000000001a4837b7
  23. ? @ 0x000000001a4a5d59
  24. ? @ 0x000000001f561c07
  25. ? @ 0x000000001f562099
  26. ? @ 0x000000001f528847
  27. ? @ 0x000000001f526c41
  28. ? @ 0x000000000009698b
  29. ? @ 0x000000000011a9cc
16:00:59  
16:00:59  Done. PASS=0 WARN=0 ERROR=1 SKIP=0 NO-OP=0 TOTAL=1

Generated SQL statement

create table `targetschema`.`foo_snapshot__dbt_tmp`



  engine = MergeTree()

      order by (tuple())




                    -- end_of_sql


            empty
          as (


        with snapshot_query as (

        select * from `sourceschema`.`foo`

    ),

    snapshotted_data as (

        select *,
            foo_id as dbt_unique_key

        from `targetschema`.`foo_snapshot`
        where dbt_valid_to is null

    ),

    insertions_source_data as (

        select
            *,
            foo_id as dbt_unique_key,
            last_update_day as dbt_updated_at,
            last_update_day as dbt_valid_from,
            nullif(last_update_day, last_update_day) as dbt_valid_to,
            halfMD5(coalesce(cast(foo_id as varchar ), '')
     || '|' || coalesce(cast(last_update_day as varchar ), '')
    ) as dbt_scd_id

        from snapshot_query
    ),

    updates_source_data as (

        select
            *,
            foo_id as dbt_unique_key,
            last_update_day as dbt_updated_at,
            last_update_day as dbt_valid_from,
            last_update_day as dbt_valid_to

        from snapshot_query
    ),

    insertions as (

        select
            'insert' as dbt_change_type,
            source_data.*

        from insertions_source_data as source_data
        left outer join snapshotted_data on snapshotted_data.dbt_unique_key = source_data.dbt_unique_key
        where snapshotted_data.dbt_unique_key is null
           or (
                snapshotted_data.dbt_unique_key is not null
            and (
                (snapshotted_data.valid_from < source_data.last_update_day)
            )
        )

    ),

    updates as (

        select
            'update' as dbt_change_type,
            source_data.*,
            snapshotted_data.dbt_scd_id

        from updates_source_data as source_data
        join snapshotted_data on snapshotted_data.dbt_unique_key = source_data.dbt_unique_key
        where (
            (snapshotted_data.valid_from < source_data.last_update_day)
        )
    )

    select * from insertions
    union all
    select * from updates


          )

Configuration

Environment

• dbt version: 1.10.15
• dbt-clickhouse version: 1.9.6
• Python version: 3.13.7
• Operating system: Linux

ClickHouse server

• ClickHouse Server version: 25.10.2
• ClickHouse Server non-default settings, if any: none
• CREATE TABLE statements for tables involved: I hope you can figure out without me constructing full test case, but let me know if you need complete reproducer from me

[koletzilla]

Hi @enqueue ! I faced a similar situation while working on supporting ClickHouse on the dbt's Jaffle-shop project. Looks like this is somehow a limitation on the ClickHouse analyser that interprets some column names wrongly, making it think it's a correlated subquery.

In my case, I fixed it with these lines: https://github.com/koletzilla/jaffle-shop-clickhouse/blob/main/models/marts/order_items.sql#L44-L46

By default this sql adds the columns from the right side of the join as order_items.* and this raised the correlated errors:

joined as (

    select
        order_items.*, -- This was the part that ClickHouse doesn't manage well.

        orders.ordered_at,
        ....

    from order_items

    left join orders on order_items.order_id = orders.order_id
    ....

The way I fixed it was by doing changing the order_items.* and renaming all columns to their original name:

joined as (

    select
        order_items.order_item_id as order_item_id, -- new line
        order_items.order_id as order_id, -- new line
        order_items.product_id as product_id, -- new line

        orders.ordered_at,
        ....

    from order_items

    left join orders on order_items.order_id = orders.order_id

The solution is not pretty, but it overcomes this ClickHouse limitation. Would you try it? Just try changing all the table.* in your models to see if that fix the issue. I see some of them are created inside the internal SQL generated, but let first try to fix it from your's

[enqueue]

So in the above example I would replace

select
    foo_id,
    foo_data,
    toDateTime64(coalesce(last_modified_time, '1970-01-01 00:00:00'), 3)
        as last_update,
    toStartOfDay(last_update) as last_update_day
from foos

with

select
    foos.foo_id as foo_id,
    foos.foo_data as foo_data,
    toDateTime64(coalesce(foos.last_modified_time, '1970-01-01 00:00:00'), 3)
        as last_modified_time,
    toStartOfDay(foos.last_update) as last_update
from foos

?

On a side note I made snapshot work using a different configuration:

1. Make valid_to column nullable
2. Reduce configuration to read like this:

snapshots:
  - name: foo_snapshot
    relation: ref('sourceschema', 'foo')
    config:
      database: targetschema
      unique_key: foo_id
      strategy: timestamp
      updated_at: last_update

I cannot find resulting SQL compiled in dbt, here is a query from ClickHouse query log:

select * from (
    select * from (

        with snapshot_query as (

            select * from `sourceschema`.`foo`

        ),

        snapshotted_data as (

            select
                *,
                foo_id as dbt_unique_key

            from `targetschema`.`foo_snapshots`
            where dbt_valid_to is null

        ),

        insertions_source_data as (

            select
                *,
                foo_id as dbt_unique_key,
                last_update_day as dbt_updated_at,
                last_update_day as dbt_valid_from,
                nullif(last_update_day, last_update_day) as dbt_valid_to,
                halfMD5(
                    coalesce(cast(foo_id as Varchar), '')
                    || '|' || coalesce(cast(last_update_day as Varchar), '')
                ) as dbt_scd_id

            from snapshot_query
        ),

        updates_source_data as (

            select
                *,
                foo_id as dbt_unique_key,
                last_update_day as dbt_updated_at,
                last_update_day as dbt_valid_from,
                last_update_day as dbt_valid_to

            from snapshot_query
        ),

        insertions as (

            select
                'insert' as dbt_change_type,
                source_data.*

            from insertions_source_data as source_data
            left outer join
                snapshotted_data
                on source_data.dbt_unique_key = snapshotted_data.dbt_unique_key
            where
                snapshotted_data.dbt_unique_key is null
                or (
                    snapshotted_data.dbt_unique_key is not null
                    and (
                        (
                            snapshotted_data.dbt_valid_from
                            < source_data.last_update_day
                        )
                    )
                )

        ),

        updates as (

            select
                'update' as dbt_change_type,
                source_data.*,
                snapshotted_data.dbt_scd_id

            from updates_source_data as source_data
            inner join
                snapshotted_data
                on source_data.dbt_unique_key = snapshotted_data.dbt_unique_key
            where (
                (snapshotted_data.dbt_valid_from < source_data.last_update_day)
            )
        )

        select * from insertions
        union all
        select * from updates

    ) as __dbt_sbq
    where false
    limit 0

) limit 0