{% snapshot daily_data_snapshot %}

{{
  config(
    target_database=[BigQueryで言うところのプロジェクト名]
    target_schema=[データセット],
    unique_key=[主キー],
    strategy='timestamp',
    updated_at=[date型のカラム。ここに指定した列を最終更新日時として、既存テーブルと比較して更新された行をINSERTする],
  )
}}

select
    *
from
    {{ source([データセット名], [テーブル名]) }}

{% endsnapshot %}

create or replace table `スナップショットテーブル`
partition by date_trunc(updated_at, day)
cluster by ~, ~,~,~ as (

with
    base_data as (select * from `増分更新テーブル`),

    data_with_row_number as (
        select *, row_number() over (partition by id order by updated_at) as row_number
        from base_data
    ),

    previous_records as (
        select id, updated_at as dbt_valid_to, row_number - 1 as row_number
        from data_with_row_number
        where row_number >= 2
    ),

    final as (
        select
            data_with_row_number.* except (row_number),
            to_hex(
                md5(
                    concat(
                        coalesce(cast(data_with_row_number.id as string), ''),
                        '|',
                        coalesce(cast(data_with_row_number.updated_at as string), '')
                    )
                )
            ) as dbt_scd_id,
            data_with_row_number.updated_at as dbt_updated_at,
            data_with_row_number.updated_at as dbt_valid_from,
            previous_records.dbt_valid_to
        from data_with_row_number
        left join
            previous_records
            on data_with_row_number.id = previous_records.id
            and data_with_row_number.row_number = previous_records.row_number
    ),

    test as ( -- 確認用
        select id, updated_at, dbt_scd_id, dbt_updated_at, dbt_valid_from, dbt_valid_to
        from final
        where id in (1945, 1813)
        order by id asc, updated_at asc
    ),

    test2 as ( -- 確認用
        select id, count(*) as cnt
        from base_data
        group by 1
        having cnt > 1
    )

select *
from final
)

SELECT *
FROM INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS

SELECT 
	REFERENCED_TABLE_NAME,TABLE_NAME,DELETE_RULE,CONSTRAINT_NAME
FROM INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS
WHERE REFERENCED_TABLE_NAME in (
	"〜",
	"〜",
	"〜"
	)

SELECT `TABLE_SCHEMA`,
       `KEY_COLUMN_USAGE`.`TABLE_NAME`,
       `COLUMN_NAME`,
       `KEY_COLUMN_USAGE`.`REFERENCED_TABLE_NAME`,
       `REFERENCED_COLUMN_NAME`,
       `UPDATE_RULE`,
       `DELETE_RULE`
FROM `INFORMATION_SCHEMA`.`KEY_COLUMN_USAGE`
         JOIN `INFORMATION_SCHEMA`.`REFERENTIAL_CONSTRAINTS`
              ON `INFORMATION_SCHEMA`.`KEY_COLUMN_USAGE`.`TABLE_SCHEMA` =
                 `INFORMATION_SCHEMA`.`REFERENTIAL_CONSTRAINTS`.`CONSTRAINT_SCHEMA`
                  AND
                 `INFORMATION_SCHEMA`.`KEY_COLUMN_USAGE`.`TABLE_NAME` =
                 `INFORMATION_SCHEMA`.`REFERENTIAL_CONSTRAINTS`.`TABLE_NAME`
                  AND
                 `INFORMATION_SCHEMA`.`KEY_COLUMN_USAGE`.`CONSTRAINT_NAME` =
                 `INFORMATION_SCHEMA`.`REFERENTIAL_CONSTRAINTS`.`CONSTRAINT_NAME`
WHERE `DELETE_RULE` = 'CASCADE'
   OR `UPDATE_RULE` = 'CASCADE';

{% test validity_period_validation(model, column_name) %}

{# スナップショットの有効期間に隙間や重複がないか 
https://qiita.com/hanon52_/items/6b1ed1a5247d0cac759f
#}

with ordered_periods as (
  select
    {{ column_name }},
    dbt_valid_from,
    dbt_valid_to,
    lead(dbt_valid_from) over (partition by {{ column_name }} order by dbt_valid_from) as next_valid_from
  from
    {{ model }}
)
select *
from ordered_periods
where
  dbt_valid_to != next_valid_from

{% endtest %}

-- 異常発生後に挿入された行を削除
delete from `ucchi_test_snapshot`
where dbt_valid_from >= timestamp("2024-12-08 16:25:52.849929 UTC");

-- 異常発生後に無効となった行を有効化
update `ucchi_test_snapshot`
set
  dbt_valid_to = null
where dbt_valid_to >= timestamp("2024-12-08 16:25:52.849929 UTC");

if(dbt_valid_to is null, true, false) as is_current

WITH your_table AS (
  SELECT 1 AS id, DATE '2024-01-01' AS dbt_updated_at UNION ALL
  SELECT 2, DATE '2024-01-02' UNION ALL
  SELECT 3, DATE '2024-01-03' UNION ALL
  SELECT 3, DATE '2024-01-04' UNION ALL
  SELECT 3, DATE '2024-01-05' UNION ALL
  SELECT 4, DATE '2024-01-06'
)
SELECT
  *,
  ROW_NUMBER() OVER (PARTITION BY id ORDER BY dbt_updated_at DESC) = 1 AS is_current
FROM
  your_table
ORDER BY
  id, dbt_updated_at

PARTITION BY dbt_valid_to
CLUSTER BY dbt_unique_key, dbt_scd_id, dbt_valid_from

PARTITION BY dbt_valid_from
CLUSTER BY dbt_unique_key, dbt_valid_to

PARTITION BY dbt_valid_to
CLUSTER BY dbt_unique_key, dbt_scd_id, dbt_valid_from