Greg/guykoh update dbt clickhouse (#14897)

* Update dbt-clickhouse version to 1.1.7 to support AirByte on ClickHouse cloud * Fix quote handling in Clickhouse normalization tests * Update test output for Clickhouse * Bump version and update changelog Co-authored-by: guykohen <guy@clickhouse.com>
airbytehq · Aug 25, 2022 · cb07e8d · cb07e8d
1 parent 7fb587f
commit cb07e8d
Show file tree

Hide file tree

Showing 33 changed files with 191 additions and 479 deletions.
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.2.18
+LABEL io.airbyte.version=0.2.19
 LABEL io.airbyte.name=airbyte/normalization
diff --git a/airbyte-integrations/bases/base-normalization/clickhouse.Dockerfile b/airbyte-integrations/bases/base-normalization/clickhouse.Dockerfile
@@ -20,7 +20,7 @@ WORKDIR /airbyte/normalization_code
 RUN pip install .
 
 WORKDIR /airbyte/normalization_code/dbt-template/
-RUN pip install dbt-clickhouse==1.0.0
+RUN pip install dbt-clickhouse==1.1.7
 # Download external dbt dependencies
 RUN dbt deps
 

diff --git a/...yte-integrations/bases/base-normalization/dbt-project-template-clickhouse/dbt_project.yml b/...yte-integrations/bases/base-normalization/dbt-project-template-clickhouse/dbt_project.yml
@@ -34,7 +34,7 @@ quoting:
   database: true
   # Temporarily disabling the behavior of the ExtendedNameTransformer on table/schema names, see (issue #1785)
   # all schemas should be unquoted
-  schema: false
+  schema: true
   identifier: true
 
 # You can define configurations for models in the `model-paths` directory here.
@@ -46,7 +46,8 @@ models:
     generated:
       airbyte_ctes:
         +tags: airbyte_internal_cte
-        +materialized: ephemeral
+        # ephemeral materialization isn't supported in ClickHouse yet
+        +materialized: view
       airbyte_incremental:
         +tags: incremental_tables
         +materialized: incremental

diff --git a/...ntegration_tests/normalization_test_output/clickhouse/test_simple_streams/dbt_project.yml b/...ntegration_tests/normalization_test_output/clickhouse/test_simple_streams/dbt_project.yml
@@ -22,15 +22,15 @@ clean-targets:
 - dbt_modules
 quoting:
   database: true
-  schema: false
+  schema: true
   identifier: true
 models:
   airbyte_utils:
     +materialized: table
     generated:
       airbyte_ctes:
         +tags: airbyte_internal_cte
-        +materialized: ephemeral
+        +materialized: view
       airbyte_incremental:
         +tags: incremental_tables
         +materialized: incremental

diff --git a/...t_simple_streams/first_output/airbyte_ctes/test_normalization/dedup_exchange_rate_ab1.sql b/...t_simple_streams/first_output/airbyte_ctes/test_normalization/dedup_exchange_rate_ab1.sql
@@ -0,0 +1,25 @@
+
+
+  create view _airbyte_test_normalization.dedup_exchange_rate_ab1__dbt_tmp 
+
+  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(assumeNotNull(_airbyte_data), 'id') as id,
+    JSONExtractRaw(assumeNotNull(_airbyte_data), 'currency') as currency,
+    JSONExtractRaw(assumeNotNull(_airbyte_data), 'date') as date,
+    JSONExtractRaw(assumeNotNull(_airbyte_data), 'timestamp_col') as timestamp_col,
+    JSONExtractRaw(assumeNotNull(_airbyte_data), 'HKD@spéçiäl & characters') as "HKD@spéçiäl & characters",
+    JSONExtractRaw(assumeNotNull(_airbyte_data), 'HKD_special___characters') as HKD_special___characters,
+    JSONExtractRaw(assumeNotNull(_airbyte_data), 'NZD') as NZD,
+    JSONExtractRaw(assumeNotNull(_airbyte_data), 'USD') as USD,
+    _airbyte_ab_id,
+    _airbyte_emitted_at,
+    now() as _airbyte_normalized_at
+from test_normalization._airbyte_raw_dedup_exchange_rate as table_alias
+-- dedup_exchange_rate
+where 1 = 1
+
+  )
diff --git a/...t_simple_streams/first_output/airbyte_ctes/test_normalization/dedup_exchange_rate_ab2.sql b/...t_simple_streams/first_output/airbyte_ctes/test_normalization/dedup_exchange_rate_ab2.sql
@@ -0,0 +1,33 @@
+
+
+  create view _airbyte_test_normalization.dedup_exchange_rate_ab2__dbt_tmp 
+
+  as (
+
+-- SQL model to cast each column to its adequate SQL type converted from the JSON schema type
+-- depends_on: _airbyte_test_normalization.dedup_exchange_rate_ab1
+select
+    accurateCastOrNull(id, '
+    BIGINT
+') as id,
+    nullif(accurateCastOrNull(trim(BOTH '"' from currency), 'String'), 'null') as currency,
+    toDate(parseDateTimeBestEffortOrNull(trim(BOTH '"' from nullif(date, '')))) as date,
+    parseDateTime64BestEffortOrNull(trim(BOTH '"' from nullif(timestamp_col, ''))) as timestamp_col,
+    accurateCastOrNull("HKD@spéçiäl & characters", '
+    Float64
+') as "HKD@spéçiäl & characters",
+    nullif(accurateCastOrNull(trim(BOTH '"' from HKD_special___characters), 'String'), 'null') as HKD_special___characters,
+    accurateCastOrNull(NZD, '
+    Float64
+') as NZD,
+    accurateCastOrNull(USD, '
+    Float64
+') as USD,
+    _airbyte_ab_id,
+    _airbyte_emitted_at,
+    now() as _airbyte_normalized_at
+from _airbyte_test_normalization.dedup_exchange_rate_ab1
+-- dedup_exchange_rate
+where 1 = 1
+
+  )
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
@@ -1,17 +1,9 @@
 
 
-
-
-    create table test_normalization.dedup_cdc_excluded_scd
 
+
+        insert into test_normalization.dedup_cdc_excluded_scd ("_airbyte_unique_key", "_airbyte_unique_key_scd", "id", "name", "_ab_cdc_lsn", "_ab_cdc_updated_at", "_ab_cdc_deleted_at", "_airbyte_start_at", "_airbyte_end_at", "_airbyte_active_row", "_airbyte_ab_id", "_airbyte_emitted_at", "_airbyte_normalized_at", "_airbyte_dedup_cdc_excluded_hashid")
 
-
-    engine = MergeTree()
-
-    order by (tuple())
-
-  as (
-
 -- depends_on: ref('dedup_cdc_excluded_stg')
 with
 
@@ -108,5 +100,5 @@ select
     now() as _airbyte_normalized_at,
     _airbyte_dedup_cdc_excluded_hashid
 from dedup_data where _airbyte_row_num = 1
-  )
+
 
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
@@ -1,17 +1,9 @@
 
 
-
-
-    create table test_normalization.dedup_exchange_rate_scd
 
+
+        insert into test_normalization.dedup_exchange_rate_scd ("_airbyte_unique_key", "_airbyte_unique_key_scd", "id", "currency", "date", "timestamp_col", "HKD@spéçiäl & characters", "HKD_special___characters", "NZD", "USD", "_airbyte_start_at", "_airbyte_end_at", "_airbyte_active_row", "_airbyte_ab_id", "_airbyte_emitted_at", "_airbyte_normalized_at", "_airbyte_dedup_exchange_rate_hashid")
 
-
-    engine = MergeTree()
-
-    order by (tuple())
-
-  as (
-
 -- depends_on: ref('dedup_exchange_rate_stg')
 with
 
@@ -112,5 +104,5 @@ select
     now() as _airbyte_normalized_at,
     _airbyte_dedup_exchange_rate_hashid
 from dedup_data where _airbyte_row_num = 1
-  )
+
 
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
@@ -1,17 +1,9 @@
 
 
-
-
-    create table test_normalization.renamed_dedup_cdc_excluded_scd
 
+
+        insert into test_normalization.renamed_dedup_cdc_excluded_scd ("_airbyte_unique_key", "_airbyte_unique_key_scd", "id", "_ab_cdc_updated_at", "_airbyte_start_at", "_airbyte_end_at", "_airbyte_active_row", "_airbyte_ab_id", "_airbyte_emitted_at", "_airbyte_normalized_at", "_airbyte_renamed_dedup_cdc_excluded_hashid")
 
-
-    engine = MergeTree()
-
-    order by (tuple())
-
-  as (
-
 -- depends_on: ref('renamed_dedup_cdc_excluded_stg')
 with
 
@@ -94,5 +86,5 @@ select
     now() as _airbyte_normalized_at,
     _airbyte_renamed_dedup_cdc_excluded_hashid
 from dedup_data where _airbyte_row_num = 1
-  )
+
 
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
@@ -1,17 +1,9 @@
 
 
-
-
-    create table test_normalization.dedup_exchange_rate
 
+
+        insert into test_normalization.dedup_exchange_rate ("_airbyte_unique_key", "id", "currency", "date", "timestamp_col", "HKD@spéçiäl & characters", "HKD_special___characters", "NZD", "USD", "_airbyte_ab_id", "_airbyte_emitted_at", "_airbyte_normalized_at", "_airbyte_dedup_exchange_rate_hashid")
 
-
-    engine = MergeTree()
-
-    order by (tuple())
-
-  as (
-
 -- Final base SQL model
 -- depends_on: test_normalization.dedup_exchange_rate_scd
 select
@@ -33,5 +25,5 @@ from test_normalization.dedup_exchange_rate_scd
 where 1 = 1
 and _airbyte_active_row = 1
 
-  )
+
 
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
@@ -1,17 +1,9 @@
 
 
-
-
-    create table test_normalization.renamed_dedup_cdc_excluded
 
+
+        insert into test_normalization.renamed_dedup_cdc_excluded ("_airbyte_unique_key", "id", "_ab_cdc_updated_at", "_airbyte_ab_id", "_airbyte_emitted_at", "_airbyte_normalized_at", "_airbyte_renamed_dedup_cdc_excluded_hashid")
 
-
-    engine = MergeTree()
-
-    order by (tuple())
-
-  as (
-
 -- Final base SQL model
 -- depends_on: test_normalization.renamed_dedup_cdc_excluded_scd
 select
@@ -27,5 +19,5 @@ from test_normalization.renamed_dedup_cdc_excluded_scd
 where 1 = 1
 and _airbyte_active_row = 1
 
-  )
+