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

Jump to bottom

Minor fixes to incremental normalization and nesting #7669

Merged
merged 10 commits into from
Nov 8, 2021
Merged

Minor fixes to incremental normalization and nesting #7669

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
5353994
Fix incremental and add safety cast
ChristopheDuong Nov 4, 2021
b2b3f57
add sql examples
ChristopheDuong Nov 5, 2021
90fbf86
Fix mypy check
ChristopheDuong Nov 5, 2021
05f28a9
Postgres materialization
ChristopheDuong Nov 5, 2021
f332c28
make postgres index unique for dedup table
ChristopheDuong Nov 5, 2021
b4edde0
Fix destination acceptance tests
ChristopheDuong Nov 8, 2021
bb84c84
Add tests for dedup on null cursor
ChristopheDuong Nov 8, 2021
40a0f0f
Bumpversion of normalization
ChristopheDuong Nov 5, 2021
5bfbdb2
Fix acceptance test
ChristopheDuong Nov 8, 2021
62c2992
Fix acceptance tests
ChristopheDuong Nov 8, 2021
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
40 changes: 31 additions & 9 deletions airbyte-integrations/bases/base-normalization/.gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,12 +6,34 @@ dbt_modules/
secrets/
dist/

integration_tests/normalization_test_output/*/*/*.log
integration_tests/normalization_test_output/*/*/*.yml
integration_tests/normalization_test_output/*/*/*.json
integration_tests/normalization_test_output/*/*/*.md
integration_tests/normalization_test_output/*/*/macros/
integration_tests/normalization_test_output/*/*/tests/
integration_tests/normalization_test_output/*/*/models/dbt_data_tests/
integration_tests/normalization_test_output/*/*/models/dbt_schema_tests/
integration_tests/normalization_test_output/*/*/modified_models/
integration_tests/normalization_test_output/*/*/macros
integration_tests/normalization_test_output/*/*/tests
integration_tests/normalization_test_output/**/*.json
integration_tests/normalization_test_output/**/*.log
integration_tests/normalization_test_output/**/*.md
integration_tests/normalization_test_output/**/*.sql
integration_tests/normalization_test_output/**/*.yml
!integration_tests/normalization_test_output/**/*dbt_project.yml
!integration_tests/normalization_test_output/**/generated/sources.yml

# We keep a minimal/restricted subset of sql files for all destinations to avoid noise in diff
# Simple Streams
!integration_tests/normalization_test_output/**/dedup_exchange_rate*.sql
!integration_tests/normalization_test_output/**/exchange_rate.sql
# Nested Streams
# Parent table
!integration_tests/normalization_test_output/**/nested_stream_with*_names_ab*.sql
!integration_tests/normalization_test_output/**/nested_stream_with*_names_scd.sql
!integration_tests/normalization_test_output/**/nested_stream_with*_names.sql
# Nested table
!integration_tests/normalization_test_output/**/nested_stream_with_*_partition_ab1.sql
!integration_tests/normalization_test_output/**/nested_stream_with_*_data_ab1.sql
!integration_tests/normalization_test_output/**/nested_stream_with*_partition_scd.sql
!integration_tests/normalization_test_output/**/nested_stream_with*_data_scd.sql
!integration_tests/normalization_test_output/**/nested_stream_with*_partition.sql
!integration_tests/normalization_test_output/**/nested_stream_with*_data.sql

# but we keep all sql files for Postgres
!integration_tests/normalization_test_output/postgres/**/*.sql
integration_tests/normalization_test_output/postgres/**/dbt_data_tests
integration_tests/normalization_test_output/postgres/**/dbt_schema_tests
4 changes: 2 additions & 2 deletions airbyte-integrations/bases/base-normalization/dbt-project-template/macros/incremental.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,14 +10,14 @@

{%- macro default__incremental_clause(col_emitted_at) -%}
{% if is_incremental() %}
and {{ col_emitted_at }} >= (select max({{ col_emitted_at }}) from {{ this }})
and cast({{ col_emitted_at }} as {{ type_timestamp_with_timezone() }}) >= (select max(cast({{ col_emitted_at }} as {{ type_timestamp_with_timezone() }})) from {{ this }})
{% endif %}
{%- endmacro -%}

{# -- see https://on-systems.tech/113-beware-dbt-incremental-updates-against-snowflake-external-tables/ #}
{%- macro snowflake__incremental_clause(col_emitted_at) -%}
{% if is_incremental() %}
and {{ col_emitted_at }} >= cast('{{ get_max_normalized_cursor(col_emitted_at) }}' as {{ type_timestamp_with_timezone() }})
and cast({{ col_emitted_at }} as {{ type_timestamp_with_timezone() }}) >= cast('{{ get_max_normalized_cursor(col_emitted_at) }}' as {{ type_timestamp_with_timezone() }})
{% endif %}
{%- endmacro -%}

Expand Down
20 changes: 20 additions & 0 deletions ...ed/airbyte_incremental/test_normalization/nested_stream_with_c___long_names_partition.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
{{ config(
indexes = [{'columns':['_airbyte_emitted_at'],'type':'hash'}],
unique_key = '_airbyte_ab_id',
schema = "test_normalization",
tags = [ "nested" ]
) }}
Comment on lines +1 to +6
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Un-nesting an object type column from the parent stream can re-use the same _airbyte_ab_id column as unique_key

-- Final base SQL model
select
_airbyte_nested_stre__nto_long_names_hashid,
double_array_data,
{{ adapter.quote('DATA') }},
_airbyte_ab_id,
_airbyte_emitted_at,
{{ current_timestamp() }} as _airbyte_normalized_at,
_airbyte_partition_hashid
from {{ ref('nested_stream_with_c___long_names_partition_ab3') }}
-- partition at nested_stream_with_complex_columns_resulting_into_long_names/partition from {{ ref('nested_stream_with_c__lting_into_long_names_scd') }}
where 1 = 1
{{ incremental_clause('_airbyte_emitted_at') }}

18 changes: 18 additions & 0 deletions ...ed/airbyte_incremental/test_normalization/nested_stream_with_c___names_partition_data.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
{{ config(
indexes = [{'columns':['_airbyte_emitted_at'],'type':'hash'}],
schema = "test_normalization",
tags = [ "nested" ]
) }}
Comment on lines +1 to +5
Copy link
Contributor Author

@ChristopheDuong ChristopheDuong Nov 5, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Un-nesting an array type column from the parent stream can NOT re-use the same _airbyte_ab_id column as unique_key... each element of the nested array would have the same unique_key and it wouldn't be unique anymore...

(this would fail with exceptions on certain destinations)

-- Final base SQL model
select
_airbyte_partition_hashid,
currency,
_airbyte_ab_id,
_airbyte_emitted_at,
{{ current_timestamp() }} as _airbyte_normalized_at,
_airbyte_data_hashid
from {{ ref('nested_stream_with_c___names_partition_data_ab3') }}
-- DATA at nested_stream_with_complex_columns_resulting_into_long_names/partition/DATA from {{ ref('nested_stream_with_c___long_names_partition') }}
where 1 = 1
{{ incremental_clause('_airbyte_emitted_at') }}

6 changes: 3 additions & 3 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
@@ -1,7 +1,7 @@



create table "postgres".test_normalization."dedup_exchange_rate_scd"
create table "postgres"."test_normalization"."dedup_exchange_rate_scd__dbt_tmp"
as (

with
Expand Down Expand Up @@ -42,15 +42,15 @@ scd_data as (
"date" desc,
_airbyte_emitted_at desc
) as _airbyte_end_at,
case when lag("date") over (
case when row_number() over (
partition by "id", currency, cast(nzd as
varchar
)
order by
"date" is null asc,
"date" desc,
_airbyte_emitted_at desc
) is null then 1 else 0 end as _airbyte_active_row,
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

when all (multiple) rows for the same primary key have a cursor (date here) equal to NULL, then this will flag multiple rows as active...

) = 1 then 1 else 0 end as _airbyte_active_row,
_airbyte_ab_id,
_airbyte_emitted_at,
_airbyte_dedup_exchange_rate_hashid
Expand Down
20 changes: 10 additions & 10 deletions ...ase-normalization/integration_tests/resources/test_nested_streams/data_input/catalog.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,16 +37,6 @@
}
}
}
},
"column`_'with\"_quotes": {
"type": ["null", "array"],
"items": {
"properties": {
"currency": {
"type": ["null", "string"]
}
}
}
}
}
}
Expand Down Expand Up @@ -234,6 +224,16 @@
"properties": {
"owner_id": {
"type": ["null", "integer"]
},
"column`_'with\"_quotes": {
"type": ["null", "array"],
"items": {
"properties": {
"currency": {
"type": ["null", "string"]
}
}
}
}
}
}
Expand Down
4 changes: 2 additions & 2 deletions ...ase-normalization/integration_tests/resources/test_nested_streams/data_input/messages.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,6 @@
{"type":"RECORD","record":{"stream":"conflict_stream_scalar","data":{"id":1,"conflict_stream_scalar": 2},"emitted_at":1623861660}}
{"type":"RECORD","record":{"stream":"conflict_stream_scalar","data":{"id":2,"conflict_stream_scalar": 2},"emitted_at":1623861660}}

{"type":"RECORD","record":{"stream":"unnest_alias","data":{"id":1, "children": [{"ab_id": 1, "owner": {"owner_id": 1}},{"ab_id": 2, "owner": {"owner_id": 2}}]},"emitted_at":1623861660}}
{"type":"RECORD","record":{"stream":"unnest_alias","data":{"id":2, "children": [{"ab_id": 3, "owner": {"owner_id": 3}},{"ab_id": 4, "owner": {"owner_id": 4}}]},"emitted_at":1623861660}}
{"type":"RECORD","record":{"stream":"unnest_alias","data":{"id":1, "children": [{"ab_id": 1, "owner": {"owner_id": 1, "column`_'with\"_quotes": [ {"currency": "EUR" } ]}},{"ab_id": 2, "owner": {"owner_id": 2, "column`_'with\"_quotes": [ {"currency": "EUR" } ]}}]},"emitted_at":1623861660}}
{"type":"RECORD","record":{"stream":"unnest_alias","data":{"id":2, "children": [{"ab_id": 3, "owner": {"owner_id": 3, "column`_'with\"_quotes": [ {"currency": "EUR" } ]}},{"ab_id": 4, "owner": {"owner_id": 4, "column`_'with\"_quotes": [ {"currency": "EUR" } ]}}]},"emitted_at":1623861660}}

5 changes: 2 additions & 3 deletions ...ation/integration_tests/resources/test_nested_streams/data_input/messages_incremental.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,5 @@
{"type":"RECORD","record":{"stream":"conflict_stream_scalar","data":{"id":1,"conflict_stream_scalar": 2},"emitted_at":1623861660}}
{"type":"RECORD","record":{"stream":"conflict_stream_scalar","data":{"id":2,"conflict_stream_scalar": 2},"emitted_at":1623861660}}

{"type":"RECORD","record":{"stream":"unnest_alias","data":{"id":1, "children": [{"ab_id": 1, "owner": {"owner_id": 1}},{"ab_id": 2, "owner": {"owner_id": 2}}]},"emitted_at":1623861660}}
{"type":"RECORD","record":{"stream":"unnest_alias","data":{"id":2, "children": [{"ab_id": 3, "owner": {"owner_id": 3}},{"ab_id": 4, "owner": {"owner_id": 4}}]},"emitted_at":1623861660}}

{"type":"RECORD","record":{"stream":"unnest_alias","data":{"id":1, "children": [{"ab_id": 1, "owner": {"owner_id": 1, "column`_'with\"_quotes": [ {"currency": "EUR" } ]}},{"ab_id": 2, "owner": {"owner_id": 2, "column`_'with\"_quotes": [ {"currency": "EUR" } ]}}]},"emitted_at":1623861660}}
{"type":"RECORD","record":{"stream":"unnest_alias","data":{"id":2, "children": [{"ab_id": 3, "owner": {"owner_id": 3, "column`_'with\"_quotes": [ {"currency": "EUR" } ]}},{"ab_id": 4, "owner": {"owner_id": 4, "column`_'with\"_quotes": [ {"currency": "EUR" } ]}}]},"emitted_at":1623861660}}
3 changes: 0 additions & 3 deletions ...s/dbt_test_config/dbt_data_tests_tmp_incremental/nested_streams_second_run_row_counts.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,6 @@ with table_row_counts as (
union all
select distinct 'nested_stream_with_complex_columns_resulting_into_long_names' as label, count(*) as row_count, 3 as expected_count
from {{ ref('nested_stream_with_complex_columns_resulting_into_long_names') }}
union all
select distinct 'nested_stream_with_complex_columns_resulting_into_long_names_partition' as label, count(*) as row_count, 3 as expected_count
from {{ ref('nested_stream_with_complex_columns_resulting_into_long_names_partition') }}
union all
select 'nested_stream_with_complex_columns_resulting_into_long_names_partition_DATA' as label, count(distinct currency) as row_count, 1 as expected_count
from {{ ref('nested_stream_with_complex_columns_resulting_into_long_names_partition_DATA') }}
Expand Down
14 changes: 8 additions & 6 deletions ...tion_tests/resources/test_nested_streams/dbt_test_config/dbt_schema_tests/schema_test.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,15 +7,17 @@ models:
expression: "double_array_data is not null"
- dbt_utils.expression_is_true:
expression: "DATA is not null"
- dbt_utils.expression_is_true:
expression: "\"column`_'with\"\"_quotes\" is not null"
- name: nested_stream_with_complex_columns_resulting_into_long_names_partition_DATA
columns:
- name: currency
tests:
- not_null
- name: nested_stream_with_complex_columns_resulting_into_long_names_partition_double_array_data
columns:
- name: id
tests:
# - name: nested_stream_with_complex_columns_resulting_into_long_names_partition_double_array_data
# columns:
# - name: id
# tests:
# - not_null # TODO Fix bug here
- name: unnest_alias_children_owner
tests:
- dbt_utils.expression_is_true:
expression: "\"column`_'with\"\"_quotes\" is not null"
14 changes: 8 additions & 6 deletions ...esources/test_nested_streams/dbt_test_config/dbt_schema_tests_incremental/schema_test.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,15 +7,17 @@ models:
expression: "double_array_data is not null"
- dbt_utils.expression_is_true:
expression: "DATA is not null"
- dbt_utils.expression_is_true:
expression: "\"column`_'with\"\"_quotes\" is not null"
- name: nested_stream_with_complex_columns_resulting_into_long_names_partition_DATA
columns:
- name: currency
tests:
- not_null
- name: nested_stream_with_complex_columns_resulting_into_long_names_partition_double_array_data
columns:
- name: id
tests:
# - name: nested_stream_with_complex_columns_resulting_into_long_names_partition_double_array_data
# columns:
# - name: id
# tests:
# - not_null # TODO Fix bug here
- name: unnest_alias_children_owner
tests:
- dbt_utils.expression_is_true:
expression: "\"column`_'with\"\"_quotes\" is not null"
3 changes: 3 additions & 0 deletions ...ase-normalization/integration_tests/resources/test_simple_streams/data_input/catalog.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,9 @@
},
"USD": {
"type": "number"
},
"column`_'with\"_quotes": {
"type": "string"
}
}
},
Expand Down
3 changes: 3 additions & 0 deletions ...ion/integration_tests/resources/test_simple_streams/data_input/catalog_schema_change.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,9 @@
},
"USD": {
"type": "number"
},
"column`_'with\"_quotes": {
"type": "string"
}
}
},
Expand Down
Loading