Publish PR 9029: clickhouse normalization (#9072)

* add normalization-clickhouse docker build step * bump normalization version * small changes gradle * fix settings gradle * fix eof file * correct clickhouse normalization * Refactor jinja template for scd (#9278) * merge chris code and regenerate sql files Co-authored-by: James Zhao <james.zhao@sinoreps.com> Co-authored-by: Edward Gao <edward.gao@airbyte.io> Co-authored-by: Christophe Duong <christophe.duong@gmail.com>
airbytehq · Jan 5, 2022 · de56d47 · de56d47
1 parent bd71999
commit de56d47
Show file tree

Hide file tree

Showing 46 changed files with 254 additions and 601 deletions.
diff --git a/airbyte-integrations/bases/base-normalization/.dockerignore b/airbyte-integrations/bases/base-normalization/.dockerignore
@@ -8,3 +8,4 @@
 !dbt-project-template-mssql
 !dbt-project-template-mysql
 !dbt-project-template-oracle
+!dbt-project-template-clickhouse
diff --git a/airbyte-integrations/bases/base-normalization/Dockerfile b/airbyte-integrations/bases/base-normalization/Dockerfile
@@ -28,5 +28,5 @@ WORKDIR /airbyte
 ENV AIRBYTE_ENTRYPOINT "/airbyte/entrypoint.sh"
 ENTRYPOINT ["/airbyte/entrypoint.sh"]
 
-LABEL io.airbyte.version=0.1.61
+LABEL io.airbyte.version=0.1.62
 LABEL io.airbyte.name=airbyte/normalization
diff --git a/airbyte-integrations/bases/base-normalization/build.gradle b/airbyte-integrations/bases/base-normalization/build.gradle
@@ -69,10 +69,15 @@ task airbyteDockerOracle(type: Exec, dependsOn: checkSshScriptCopy) {
     configure buildAirbyteDocker('oracle')
     dependsOn assemble
 }
+task airbyteDockerClickhouse(type: Exec, dependsOn: checkSshScriptCopy) {
+    configure buildAirbyteDocker('clickhouse')
+    dependsOn assemble
+}
 
 airbyteDocker.dependsOn(airbyteDockerMSSql)
 airbyteDocker.dependsOn(airbyteDockerMySql)
 airbyteDocker.dependsOn(airbyteDockerOracle)
+airbyteDocker.dependsOn(airbyteDockerClickhouse)
 
 task("customIntegrationTestPython", type: PythonTask, dependsOn: installTestReqs) {
     module = "pytest"
@@ -86,6 +91,7 @@ task("customIntegrationTestPython", type: PythonTask, dependsOn: installTestReqs
     dependsOn ':airbyte-integrations:connectors:destination-snowflake:airbyteDocker'
     dependsOn ':airbyte-integrations:connectors:destination-oracle:airbyteDocker'
     dependsOn ':airbyte-integrations:connectors:destination-mssql:airbyteDocker'
+    dependsOn ':airbyte-integrations:connectors:destination-clickhouse:airbyteDocker'
 }
 
 integrationTest.dependsOn("customIntegrationTestPython")

diff --git a/airbyte-integrations/bases/base-normalization/docker-compose.yaml b/airbyte-integrations/bases/base-normalization/docker-compose.yaml
@@ -10,3 +10,5 @@ services:
     image: airbyte/normalization-mysql:${VERSION}
   normalization-oracle:
     image: airbyte/normalization-oracle:${VERSION}
+  normalization-clickhouse:
+    image: airbyte/normalization-clickhouse:${VERSION}
diff --git a/...treams/first_output/airbyte_incremental/scd/test_normalization/dedup_cdc_excluded_scd.sql b/...treams/first_output/airbyte_incremental/scd/test_normalization/dedup_cdc_excluded_scd.sql
@@ -12,11 +12,12 @@
 
   as (
 
+-- depends_on: ref('dedup_cdc_excluded_stg')
 with
 
 input_data as (
     select *
-    from _airbyte_test_normalization.dedup_cdc_excluded_ab3
+    from _airbyte_test_normalization.dedup_cdc_excluded_stg
     -- dedup_cdc_excluded from test_normalization._airbyte_raw_dedup_cdc_excluded
 ),
 
@@ -45,15 +46,15 @@ scd_data as (
         _ab_cdc_updated_at,
         _ab_cdc_deleted_at,
       _airbyte_emitted_at 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 (
         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
+            ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING
       ) as _airbyte_end_at,
-      case when _airbyte_active_row_num = 1 and _ab_cdc_deleted_at is null then 1 else 0 end as _airbyte_active_row,
       _airbyte_ab_id,
       _airbyte_emitted_at,
       _airbyte_dedup_cdc_excluded_hashid
@@ -65,7 +66,7 @@ dedup_data as (
         -- 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')
-            order by _airbyte_ab_id
+            order by _airbyte_active_row desc, _airbyte_ab_id
         ) as _airbyte_row_num,
         assumeNotNull(hex(MD5(
 

diff --git a/...reams/first_output/airbyte_incremental/scd/test_normalization/dedup_exchange_rate_scd.sql b/...reams/first_output/airbyte_incremental/scd/test_normalization/dedup_exchange_rate_scd.sql
@@ -12,11 +12,12 @@
 
   as (
 
+-- depends_on: ref('dedup_exchange_rate_stg')
 with
 
 input_data as (
     select *
-    from _airbyte_test_normalization.dedup_exchange_rate_ab3
+    from _airbyte_test_normalization.dedup_exchange_rate_stg
     -- dedup_exchange_rate from test_normalization._airbyte_raw_dedup_exchange_rate
 ),
 
@@ -54,15 +55,15 @@ scd_data as (
         NZD,
         USD,
       date as _airbyte_start_at,
+      case when _airbyte_active_row_num = 1 then 1 else 0 end as _airbyte_active_row,
       anyOrNull(date) over (
         partition by id, currency, cast(NZD as String)
         order by
             date is null asc,
             date desc,
             _airbyte_emitted_at desc
-        ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING
+            ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING
       ) as _airbyte_end_at,
-      case when _airbyte_active_row_num = 1 then 1 else 0 end as _airbyte_active_row,
       _airbyte_ab_id,
       _airbyte_emitted_at,
       _airbyte_dedup_exchange_rate_hashid
@@ -74,7 +75,7 @@ dedup_data as (
         -- additionally, we generate a unique key for the scd table
         row_number() over (
             partition by _airbyte_unique_key, _airbyte_start_at, _airbyte_emitted_at
-            order by _airbyte_ab_id
+            order by _airbyte_active_row desc, _airbyte_ab_id
         ) as _airbyte_row_num,
         assumeNotNull(hex(MD5(
 

diff --git a/...irst_output/airbyte_incremental/scd/test_normalization/renamed_dedup_cdc_excluded_scd.sql b/...irst_output/airbyte_incremental/scd/test_normalization/renamed_dedup_cdc_excluded_scd.sql
@@ -12,11 +12,12 @@
 
   as (
 
+-- depends_on: ref('renamed_dedup_cdc_excluded_stg')
 with
 
 input_data as (
     select *
-    from _airbyte_test_normalization.renamed_dedup_cdc_excluded_ab3
+    from _airbyte_test_normalization.renamed_dedup_cdc_excluded_stg
     -- renamed_dedup_cdc_excluded from test_normalization._airbyte_raw_renamed_dedup_cdc_excluded
 ),
 
@@ -41,15 +42,15 @@ scd_data as (
     ))) as _airbyte_unique_key,
         id,
       _airbyte_emitted_at as _airbyte_start_at,
+      case when _airbyte_active_row_num = 1 then 1 else 0 end as _airbyte_active_row,
       anyOrNull(_airbyte_emitted_at) over (
         partition by id
         order by
             _airbyte_emitted_at is null asc,
             _airbyte_emitted_at desc,
             _airbyte_emitted_at desc
-        ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING
+            ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING
       ) as _airbyte_end_at,
-      case when _airbyte_active_row_num = 1 then 1 else 0 end as _airbyte_active_row,
       _airbyte_ab_id,
       _airbyte_emitted_at,
       _airbyte_renamed_dedup_cdc_excluded_hashid
@@ -61,7 +62,7 @@ dedup_data as (
         -- additionally, we generate a unique key for the scd table
         row_number() over (
             partition by _airbyte_unique_key, _airbyte_start_at, _airbyte_emitted_at
-            order by _airbyte_ab_id
+            order by _airbyte_active_row desc, _airbyte_ab_id
         ) as _airbyte_row_num,
         assumeNotNull(hex(MD5(
 

diff --git a/...simple_streams/first_output/airbyte_incremental/test_normalization/dedup_cdc_excluded.sql b/...simple_streams/first_output/airbyte_incremental/test_normalization/dedup_cdc_excluded.sql
@@ -13,6 +13,7 @@
   as (
 
 -- Final base SQL model
+-- depends_on: test_normalization.dedup_cdc_excluded_scd
 select
     _airbyte_unique_key,
     id,

diff --git a/...imple_streams/first_output/airbyte_incremental/test_normalization/dedup_exchange_rate.sql b/...imple_streams/first_output/airbyte_incremental/test_normalization/dedup_exchange_rate.sql
@@ -13,6 +13,7 @@
   as (
 
 -- Final base SQL model
+-- depends_on: test_normalization.dedup_exchange_rate_scd
 select
     _airbyte_unique_key,
     id,

diff --git a/...treams/first_output/airbyte_incremental/test_normalization/renamed_dedup_cdc_excluded.sql b/...treams/first_output/airbyte_incremental/test_normalization/renamed_dedup_cdc_excluded.sql
@@ -13,6 +13,7 @@
   as (
 
 -- Final base SQL model
+-- depends_on: test_normalization.renamed_dedup_cdc_excluded_scd
 select
     _airbyte_unique_key,
     id,

diff --git a/...ouse/test_simple_streams/first_output/airbyte_tables/test_normalization/exchange_rate.sql b/...ouse/test_simple_streams/first_output/airbyte_tables/test_normalization/exchange_rate.sql
@@ -14,6 +14,7 @@
 with __dbt__cte__exchange_rate_ab1 as (
 
 -- SQL model to parse JSON blob stored in a single column and extract into separated field columns as described by the JSON Schema
+-- depends_on: test_normalization._airbyte_raw_exchange_rate
 select
     JSONExtractRaw(_airbyte_data, 'id') as id,
     JSONExtractRaw(_airbyte_data, 'currency') as currency,
@@ -33,6 +34,7 @@ where 1 = 1
 ),  __dbt__cte__exchange_rate_ab2 as (
 
 -- SQL model to cast each column to its adequate SQL type converted from the JSON schema type
+-- depends_on: __dbt__cte__exchange_rate_ab1
 select
     accurateCastOrNull(id, '
     BIGINT
@@ -60,6 +62,7 @@ where 1 = 1
 ),  __dbt__cte__exchange_rate_ab3 as (
 
 -- SQL model to build a hash column based on the values of this record
+-- depends_on: __dbt__cte__exchange_rate_ab2
 select
     assumeNotNull(hex(MD5(
 
@@ -95,6 +98,7 @@ from __dbt__cte__exchange_rate_ab2 tmp
 -- exchange_rate
 where 1 = 1
 )-- Final base SQL model
+-- depends_on: __dbt__cte__exchange_rate_ab3
 select
     id,
     currency,

diff --git a/...t_simple_streams/first_output/airbyte_views/test_normalization/dedup_cdc_excluded_ab3.sql b/...t_simple_streams/first_output/airbyte_views/test_normalization/dedup_cdc_excluded_ab3.sql
diff --git a/...normalization/dedup_exchange_rate_ab3.sql → ...normalization/dedup_exchange_rate_stg.sql b/...normalization/dedup_exchange_rate_ab3.sql → ...normalization/dedup_exchange_rate_stg.sql
@@ -1,12 +1,13 @@
 
 
-  create view _airbyte_test_normalization.dedup_exchange_rate_ab3__dbt_tmp 
+  create view _airbyte_test_normalization.dedup_exchange_rate_stg__dbt_tmp 
 
   as (
 
 with __dbt__cte__dedup_exchange_rate_ab1 as (
 
 -- SQL model to parse JSON blob stored in a single column and extract into separated field columns as described by the JSON Schema
+-- depends_on: test_normalization._airbyte_raw_dedup_exchange_rate
 select
     JSONExtractRaw(_airbyte_data, 'id') as id,
     JSONExtractRaw(_airbyte_data, 'currency') as currency,
@@ -26,6 +27,7 @@ where 1 = 1
 ),  __dbt__cte__dedup_exchange_rate_ab2 as (
 
 -- SQL model to cast each column to its adequate SQL type converted from the JSON schema type
+-- depends_on: __dbt__cte__dedup_exchange_rate_ab1
 select
     accurateCastOrNull(id, '
     BIGINT
@@ -51,6 +53,7 @@ from __dbt__cte__dedup_exchange_rate_ab1
 where 1 = 1
 
 )-- SQL model to build a hash column based on the values of this record
+-- depends_on: __dbt__cte__dedup_exchange_rate_ab2
 select
     assumeNotNull(hex(MD5(
-Original file line number
+Diff line change
@@ Expand Up / @@ -13,6 +13,7 @@ @@
       as (
     -- Final base SQL model
+    -- depends_on: test_normalization.dedup_cdc_excluded_scd
     select
         _airbyte_unique_key,
         id,
@@ Expand Down @@