Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

🐛 Fix normalization SCD partition by float columns errors with BigQuery #9281

Merged
merged 20 commits into from
Jan 6, 2022
Merged

🐛 Fix normalization SCD partition by float columns errors with BigQuery #9281

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
c6ae621
add normalization-clickhouse docker build step
jzcruiser Dec 22, 2021
b88225e
Merge branch 'patch-4' of github.com:jzcruiser/airbyte into marcos/te…
marcosmarxm Dec 23, 2021
cc499c2
bump normalization version
marcosmarxm Dec 23, 2021
a2517a3
small changes gradle
marcosmarxm Dec 23, 2021
4a51799
Merge branch 'master' into marcos/test-pr-9029
marcosmarxm Dec 23, 2021
a57495c
fix settings gradle
marcosmarxm Dec 27, 2021
c56ef54
fix eof file
marcosmarxm Dec 27, 2021
f8ccfd6
correct clickhouse normalization
marcosmarxm Dec 29, 2021
82b77d3
add test case
ChristopheDuong Jan 3, 2022
df3faf6
Cast cursor to string
ChristopheDuong Dec 20, 2021
a76ba27
Refactor jinja template for scd
ChristopheDuong Jan 4, 2022
34fa54d
Merge branch 'test-pr-9029' into fix-normalization
ChristopheDuong Jan 4, 2022
5e48f8a
Fix normalization scd in oracle
ChristopheDuong Jan 4, 2022
fff86cc
Regen sql
ChristopheDuong Jan 4, 2022
1d9dfb6
Regen clickhouse
ChristopheDuong Jan 4, 2022
138bba3
Merge remote-tracking branch 'origin/master' into fix-normalization
ChristopheDuong Jan 5, 2022
f536e48
Regen file
ChristopheDuong Jan 5, 2022
7b303bb
Re-indent columns
ChristopheDuong Jan 5, 2022
4aed065
Indent clickhouse files
ChristopheDuong Jan 5, 2022
0771ffe
Fix when no cursor
ChristopheDuong Jan 6, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 10 additions & 7 deletions ...d/test_normalization/nested_stream_with_complex_columns_resulting_into_long_names_scd.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,9 +26,9 @@ scd_data as (
), '')) as
string
))) as _airbyte_unique_key,
id,
date,
`partition`,
id,
date,
`partition`,
date as _airbyte_start_at,
lag(date) over (
partition by id
Expand All @@ -54,7 +54,10 @@ dedup_data as (
-- we need to ensure de-duplicated rows for merge/update queries
-- additionally, we generate a unique key for the scd table
row_number() over (
partition by _airbyte_unique_key, _airbyte_start_at, _airbyte_emitted_at
partition by
_airbyte_unique_key,
_airbyte_start_at,
_airbyte_emitted_at
order by _airbyte_active_row desc, _airbyte_ab_id
) as _airbyte_row_num,
to_hex(md5(cast(concat(coalesce(cast(_airbyte_unique_key as
Expand All @@ -72,9 +75,9 @@ dedup_data as (
select
_airbyte_unique_key,
_airbyte_unique_key_scd,
id,
date,
`partition`,
id,
date,
`partition`,
_airbyte_start_at,
_airbyte_end_at,
_airbyte_active_row,
Expand Down
19 changes: 11 additions & 8 deletions ...d/test_normalization/nested_stream_with_complex_columns_resulting_into_long_names_scd.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,11 +57,11 @@ scd_data as (
-- SQL model to build a Type 2 Slowly Changing Dimension (SCD) table for each record identified by their primary key
select
{{ dbt_utils.surrogate_key([
'id',
'id',
]) }} as _airbyte_unique_key,
id,
date,
{{ adapter.quote('partition') }},
id,
date,
{{ adapter.quote('partition') }},
date as _airbyte_start_at,
lag(date) over (
partition by id
Expand All @@ -87,7 +87,10 @@ dedup_data as (
-- we need to ensure de-duplicated rows for merge/update queries
-- additionally, we generate a unique key for the scd table
row_number() over (
partition by _airbyte_unique_key, _airbyte_start_at, _airbyte_emitted_at
partition by
_airbyte_unique_key,
_airbyte_start_at,
_airbyte_emitted_at
order by _airbyte_active_row desc, _airbyte_ab_id
) as _airbyte_row_num,
{{ dbt_utils.surrogate_key([
Expand All @@ -101,9 +104,9 @@ dedup_data as (
select
_airbyte_unique_key,
_airbyte_unique_key_scd,
id,
date,
{{ adapter.quote('partition') }},
id,
date,
{{ adapter.quote('partition') }},
_airbyte_start_at,
_airbyte_end_at,
_airbyte_active_row,
Expand Down
37 changes: 20 additions & 17 deletions ...reams/first_output/airbyte_incremental/scd/test_normalization/dedup_exchange_rate_scd.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,14 +30,14 @@ scd_data as (
), '')) as
string
))) as _airbyte_unique_key,
id,
currency,
date,
timestamp_col,
HKD_special___characters,
HKD_special___characters_1,
NZD,
USD,
id,
currency,
date,
timestamp_col,
HKD_special___characters,
HKD_special___characters_1,
NZD,
USD,
date as _airbyte_start_at,
lag(date) over (
partition by id, currency, cast(NZD as
Expand Down Expand Up @@ -67,7 +67,10 @@ dedup_data as (
-- we need to ensure de-duplicated rows for merge/update queries
-- additionally, we generate a unique key for the scd table
row_number() over (
partition by _airbyte_unique_key, _airbyte_start_at, _airbyte_emitted_at
partition by
_airbyte_unique_key,
_airbyte_start_at,
_airbyte_emitted_at
order by _airbyte_active_row desc, _airbyte_ab_id
) as _airbyte_row_num,
to_hex(md5(cast(concat(coalesce(cast(_airbyte_unique_key as
Expand All @@ -85,14 +88,14 @@ dedup_data as (
select
_airbyte_unique_key,
_airbyte_unique_key_scd,
id,
currency,
date,
timestamp_col,
HKD_special___characters,
HKD_special___characters_1,
NZD,
USD,
id,
currency,
date,
timestamp_col,
HKD_special___characters,
HKD_special___characters_1,
NZD,
USD,
_airbyte_start_at,
_airbyte_end_at,
_airbyte_active_row,
Expand Down
43 changes: 23 additions & 20 deletions ...s/models/generated/airbyte_incremental/scd/test_normalization/dedup_exchange_rate_scd.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,18 +59,18 @@ scd_data as (
-- SQL model to build a Type 2 Slowly Changing Dimension (SCD) table for each record identified by their primary key
select
{{ dbt_utils.surrogate_key([
'id',
'currency',
'NZD',
'id',
'currency',
'NZD',
]) }} as _airbyte_unique_key,
id,
currency,
date,
timestamp_col,
HKD_special___characters,
HKD_special___characters_1,
NZD,
USD,
id,
currency,
date,
timestamp_col,
HKD_special___characters,
HKD_special___characters_1,
NZD,
USD,
date as _airbyte_start_at,
lag(date) over (
partition by id, currency, cast(NZD as {{ dbt_utils.type_string() }})
Expand All @@ -96,7 +96,10 @@ dedup_data as (
-- we need to ensure de-duplicated rows for merge/update queries
-- additionally, we generate a unique key for the scd table
row_number() over (
partition by _airbyte_unique_key, _airbyte_start_at, _airbyte_emitted_at
partition by
_airbyte_unique_key,
_airbyte_start_at,
_airbyte_emitted_at
order by _airbyte_active_row desc, _airbyte_ab_id
) as _airbyte_row_num,
{{ dbt_utils.surrogate_key([
Expand All @@ -110,14 +113,14 @@ dedup_data as (
select
_airbyte_unique_key,
_airbyte_unique_key_scd,
id,
currency,
date,
timestamp_col,
HKD_special___characters,
HKD_special___characters_1,
NZD,
USD,
id,
currency,
date,
timestamp_col,
HKD_special___characters,
HKD_special___characters_1,
NZD,
USD,
_airbyte_start_at,
_airbyte_end_at,
_airbyte_active_row,
Expand Down
43 changes: 23 additions & 20 deletions ...d_models/generated/airbyte_incremental/scd/test_normalization/dedup_exchange_rate_scd.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,18 +59,18 @@ scd_data as (
-- SQL model to build a Type 2 Slowly Changing Dimension (SCD) table for each record identified by their primary key
select
{{ dbt_utils.surrogate_key([
'id',
'currency',
'NZD',
'id',
'currency',
'NZD',
]) }} as _airbyte_unique_key,
id,
currency,
new_column,
date,
timestamp_col,
HKD_special___characters,
NZD,
USD,
id,
currency,
new_column,
date,
timestamp_col,
HKD_special___characters,
NZD,
USD,
date as _airbyte_start_at,
lag(date) over (
partition by cast(id as {{ dbt_utils.type_string() }}), currency, cast(NZD as {{ dbt_utils.type_string() }})
Expand All @@ -96,7 +96,10 @@ dedup_data as (
-- we need to ensure de-duplicated rows for merge/update queries
-- additionally, we generate a unique key for the scd table
row_number() over (
partition by _airbyte_unique_key, _airbyte_start_at, _airbyte_emitted_at
partition by
_airbyte_unique_key,
_airbyte_start_at,
_airbyte_emitted_at
order by _airbyte_active_row desc, _airbyte_ab_id
) as _airbyte_row_num,
{{ dbt_utils.surrogate_key([
Expand All @@ -110,14 +113,14 @@ dedup_data as (
select
_airbyte_unique_key,
_airbyte_unique_key_scd,
id,
currency,
new_column,
date,
timestamp_col,
HKD_special___characters,
NZD,
USD,
id,
currency,
new_column,
date,
timestamp_col,
HKD_special___characters,
NZD,
USD,
_airbyte_start_at,
_airbyte_end_at,
_airbyte_active_row,
Expand Down
46 changes: 25 additions & 21 deletions ...treams/first_output/airbyte_incremental/scd/test_normalization/dedup_cdc_excluded_scd.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,9 +26,10 @@ input_data_with_active_row_num as (
row_number() over (
partition by id
order by
_airbyte_emitted_at is null asc,
_airbyte_emitted_at desc,
_airbyte_emitted_at desc, _ab_cdc_updated_at desc
_ab_cdc_lsn is null asc,
_ab_cdc_lsn desc,
_ab_cdc_updated_at desc,
_airbyte_emitted_at desc
) as _airbyte_active_row_num
from input_data
),
Expand All @@ -40,21 +41,21 @@ scd_data as (
toString(id)

))) as _airbyte_unique_key,
id,
name,
_ab_cdc_lsn,
_ab_cdc_updated_at,
_ab_cdc_deleted_at,
_airbyte_emitted_at as _airbyte_start_at,
id,
name,
_ab_cdc_lsn,
_ab_cdc_updated_at,
_ab_cdc_deleted_at,
_ab_cdc_lsn as _airbyte_start_at,
case when _airbyte_active_row_num = 1 and _ab_cdc_deleted_at is null then 1 else 0 end as _airbyte_active_row,
anyOrNull(_airbyte_emitted_at) over (
anyOrNull(_ab_cdc_lsn) over (
partition by id
order by
_airbyte_emitted_at is null asc,
_airbyte_emitted_at desc,
_airbyte_emitted_at desc, _ab_cdc_updated_at desc
ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING
) as _airbyte_end_at,
_ab_cdc_lsn is null asc,
_ab_cdc_lsn desc,
_ab_cdc_updated_at desc,
_airbyte_emitted_at desc
ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING) as _airbyte_end_at,
_airbyte_ab_id,
_airbyte_emitted_at,
_airbyte_dedup_cdc_excluded_hashid
Expand All @@ -65,7 +66,10 @@ dedup_data as (
-- we need to ensure de-duplicated rows for merge/update queries
-- additionally, we generate a unique key for the scd table
row_number() over (
partition by _airbyte_unique_key, _airbyte_start_at, _airbyte_emitted_at, accurateCastOrNull(_ab_cdc_deleted_at, 'String'), accurateCastOrNull(_ab_cdc_updated_at, 'String')
partition by
_airbyte_unique_key,
_airbyte_start_at,
_airbyte_emitted_at, accurateCastOrNull(_ab_cdc_deleted_at, 'String'), accurateCastOrNull(_ab_cdc_updated_at, 'String')
order by _airbyte_active_row desc, _airbyte_ab_id
) as _airbyte_row_num,
assumeNotNull(hex(MD5(
Expand All @@ -91,11 +95,11 @@ dedup_data as (
select
_airbyte_unique_key,
_airbyte_unique_key_scd,
id,
name,
_ab_cdc_lsn,
_ab_cdc_updated_at,
_ab_cdc_deleted_at,
id,
name,
_ab_cdc_lsn,
_ab_cdc_updated_at,
_ab_cdc_deleted_at,
_airbyte_start_at,
_airbyte_end_at,
_airbyte_active_row,
Expand Down
Loading