Brigad
diff --git a/‎airbyte-db/db-lib/src/main/java/io/airbyte/db/factory/DataSourceFactory.java
Lines changed: 5 additions & 0 deletions b/‎airbyte-db/db-lib/src/main/java/io/airbyte/db/factory/DataSourceFactory.java
Lines changed: 5 additions & 0 deletions
diff --git a/‎airbyte-integrations/bases/base-normalization/.dockerignore
Lines changed: 1 addition & 0 deletions b/‎airbyte-integrations/bases/base-normalization/.dockerignore
Lines changed: 1 addition & 0 deletions
diff --git a/‎airbyte-integrations/bases/base-normalization/build.gradle
Lines changed: 6 additions & 0 deletions b/‎airbyte-integrations/bases/base-normalization/build.gradle
Lines changed: 6 additions & 0 deletions
diff --git a/‎airbyte-integrations/bases/base-normalization/databricks.Dockerfile
Lines changed: 34 additions & 0 deletions b/‎airbyte-integrations/bases/base-normalization/databricks.Dockerfile
Lines changed: 34 additions & 0 deletions
diff --git a/‎airbyte-integrations/bases/base-normalization/dbt-project-template-databricks/dbt_project.yml
Lines changed: 72 additions & 0 deletions b/‎airbyte-integrations/bases/base-normalization/dbt-project-template-databricks/dbt_project.yml
Lines changed: 72 additions & 0 deletions
diff --git a/‎airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/array.sql
Lines changed: 9 additions & 0 deletions b/‎airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/array.sql
Lines changed: 9 additions & 0 deletions
diff --git a/‎airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/columns.sql
Lines changed: 28 additions & 0 deletions b/‎airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/columns.sql
Lines changed: 28 additions & 0 deletions
diff --git a/‎airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/current_timestamp.sql
Lines changed: 4 additions & 0 deletions b/‎airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/current_timestamp.sql
Lines changed: 4 additions & 0 deletions
diff --git a/‎airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/datatypes.sql
Lines changed: 22 additions & 0 deletions b/‎airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/datatypes.sql
Lines changed: 22 additions & 0 deletions
diff --git a/‎airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/json_operations.sql
Lines changed: 26 additions & 0 deletions b/‎airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/json_operations.sql
Lines changed: 26 additions & 0 deletions
diff --git a/‎airbyte-integrations/bases/base-normalization/dbt-project-template/macros/should_full_refresh.sql
Lines changed: 33 additions & 2 deletions b/‎airbyte-integrations/bases/base-normalization/dbt-project-template/macros/should_full_refresh.sql
Lines changed: 33 additions & 2 deletions
@@ -267,6 +267,11 @@ public DataSource build() {
        * will preserve existing behavior that tests for the connection on first use, not on creation.
        */
       config.setInitializationFailTimeout(Integer.MIN_VALUE);
+      /*
+       * Default timeout is 30 sec, which is too short when you work with cloud data warehouses clusters
+       * that can take 4-5 min to start up. Set it to 30 min to be sure
+       */
+      config.setConnectionTimeout(30 * 60 * 1000);
 
       connectionProperties.forEach(config::addDataSourceProperty);
 
 
@@ -10,4 +10,5 @@
 !dbt-project-template-oracle
 !dbt-project-template-clickhouse
 !dbt-project-template-snowflake
+!dbt-project-template-databricks
 !dbt-project-template-redshift
@@ -75,6 +75,10 @@ task airbyteDockerSnowflake(type: Exec, dependsOn: checkSshScriptCopy) {
     configure buildAirbyteDocker('snowflake')
     dependsOn assemble
 }
+task airbyteDockerDatabricks(type: Exec, dependsOn: checkSshScriptCopy) {
+    configure buildAirbyteDocker('databricks')
+    dependsOn assemble
+}
 task airbyteDockerRedshift(type: Exec, dependsOn: checkSshScriptCopy) {
     configure buildAirbyteDocker('redshift')
     dependsOn assemble
@@ -85,6 +89,7 @@ airbyteDocker.dependsOn(airbyteDockerMySql)
 airbyteDocker.dependsOn(airbyteDockerOracle)
 airbyteDocker.dependsOn(airbyteDockerClickhouse)
 airbyteDocker.dependsOn(airbyteDockerSnowflake)
+airbyteDocker.dependsOn(airbyteDockerDatabricks)
 airbyteDocker.dependsOn(airbyteDockerRedshift)
 
 task("customIntegrationTestPython", type: PythonTask, dependsOn: installTestReqs) {
@@ -100,6 +105,7 @@ task("customIntegrationTestPython", type: PythonTask, dependsOn: installTestReqs
     dependsOn ':airbyte-integrations:connectors:destination-oracle:airbyteDocker'
     dependsOn ':airbyte-integrations:connectors:destination-mssql:airbyteDocker'
     dependsOn ':airbyte-integrations:connectors:destination-clickhouse:airbyteDocker'
+    dependsOn ':airbyte-integrations:connectors:destination-databricks:airbyteDocker'
 }
 
 // DATs have some additional tests that exercise normalization code paths,
 
@@ -0,0 +1,34 @@
+FROM fishtownanalytics/dbt:1.0.0
+COPY --from=airbyte/base-airbyte-protocol-python:0.1.1 /airbyte /airbyte
+
+# Install SSH Tunneling dependencies
+RUN apt-get update && apt-get install -y jq sshpass
+
+WORKDIR /airbyte
+COPY entrypoint.sh .
+COPY build/sshtunneling.sh .
+
+WORKDIR /airbyte/normalization_code
+COPY normalization ./normalization
+COPY setup.py .
+COPY dbt-project-template/ ./dbt-template/
+COPY dbt-project-template-databricks/* ./dbt-template/
+
+# Install python dependencies
+WORKDIR /airbyte/base_python_structs
+RUN pip install .
+
+WORKDIR /airbyte/normalization_code
+RUN pip install .
+
+WORKDIR /airbyte/normalization_code/dbt-template/
+# Download external dbt dependencies
+RUN pip install dbt-databricks==1.0.0
+RUN dbt deps
+
+WORKDIR /airbyte
+ENV AIRBYTE_ENTRYPOINT "/airbyte/entrypoint.sh"
+ENTRYPOINT ["/airbyte/entrypoint.sh"]
+
+LABEL io.airbyte.version=0.1.73
+LABEL io.airbyte.name=airbyte/normalization-databricks
@@ -0,0 +1,72 @@
+# This file is necessary to install dbt-utils with dbt deps
+# the content will be overwritten by the transform function
+
+# Name your package! Package names should contain only lowercase characters
+# and underscores. A good package name should reflect your organization's
+# name or the intended use of these models
+name: "airbyte_utils"
+version: "1.0"
+config-version: 2
+
+# This setting configures which "profile" dbt uses for this project. Profiles contain
+# database connection information, and should be configured in the  ~/.dbt/profiles.yml file
+profile: "normalize"
+
+# These configurations specify where dbt should look for different types of files.
+# The `model-paths` config, for example, states that source models can be found
+# in the "models/" directory. You probably won't need to change these!
+model-paths: ["models"]
+docs-paths: ["docs"]
+analysis-paths: ["analysis"]
+test-paths: ["tests"]
+seed-paths: ["data"]
+macro-paths: ["macros"]
+
+target-path: "../build" # directory which will store compiled SQL files
+log-path: "../logs" # directory which will store DBT logs
+packages-install-path: "/tmp/dbt_modules" # directory which will store external DBT dependencies
+
+clean-targets: # directories to be removed by `dbt clean`
+  - "build"
+  - "dbt_modules"
+
+quoting:
+  database: true
+  # Temporarily disabling the behavior of the ExtendedNameTransformer on table/schema names, see (issue #1785)
+  # all schemas should be unquoted
+  schema: false
+  identifier: false
+
+# You can define configurations for models in the `model-paths` directory here.
+# Using these configurations, you can enable or disable models, change how they
+# are materialized, and more!
+models:
+  +transient: false
+  airbyte_utils:
+    +materialized: table
+    generated:
+      airbyte_ctes:
+        +tags: airbyte_internal_cte
+        +materialized: ephemeral
+      airbyte_incremental:
+        +tags: incremental_tables
+        +materialized: incremental
+        +incremental_strategy: merge
+        # schema change test is supported automatically by the merge operation
+        # need to be run against a cluster with spark.databricks.delta.schema.autoMerge.enabled = True
+        # schema merge being handled at the final step, if a schema changes in one of the primary keys
+        # that coalesce differently to string, unicity will be broken
+        +on_schema_change: "ignore"
+        +file_format: delta
+        +pre-hook: 'SET spark.databricks.delta.schema.autoMerge.enabled = True'
+      airbyte_tables:
+        +tags: normalized_tables
+        +materialized: table
+        +file_format: delta
+      airbyte_views:
+        +tags: airbyte_internal_views
+        +materialized: view
+
+dispatch:
+  - macro_namespace: dbt_utils
+    search_order: ["airbyte_utils", "dbt_utils"]
@@ -6,6 +6,7 @@
     - postgres: unnest() -> https://www.postgresqltutorial.com/postgresql-array/
     - MSSQL: openjson() –> https://docs.microsoft.com/en-us/sql/relational-databases/json/validate-query-and-change-json-data-with-built-in-functions-sql-server?view=sql-server-ver15
     - ClickHouse: ARRAY JOIN –> https://clickhouse.com/docs/zh/sql-reference/statements/select/array-join/
+    - Databricks: LATERAL VIEW -> https://docs.databricks.com/spark/latest/spark-sql/language-manual/sql-ref-syntax-qry-select-lateral-view.html
 #}
 
 {# cross_join_unnest -------------------------------------------------     #}
@@ -50,6 +51,10 @@
     cross join table(flatten({{ array_col }})) as {{ array_col }}
 {%- endmacro %}
 
+{% macro databricks__cross_join_unnest(stream_name, array_col) -%}
+    lateral view outer explode(from_json({{ array_col }}, 'array<string>')) as _airbyte_nested_data
+{%- endmacro %}
+
 {% macro sqlserver__cross_join_unnest(stream_name, array_col) -%}
 {# https://docs.microsoft.com/en-us/sql/relational-databases/json/convert-json-data-to-rows-and-columns-with-openjson-sql-server?view=sql-server-ver15#option-1---openjson-with-the-default-output #}
     CROSS APPLY (
@@ -87,6 +92,10 @@
     _airbyte_nested_data
 {%- endmacro %}
 
+{% macro databricks__unnested_column_value(column_col) -%}
+    _airbyte_nested_data
+{%- endmacro %}
+
 {% macro oracle__unnested_column_value(column_col) -%}
     {{ column_col }}
 {%- endmacro %}
 
@@ -14,3 +14,31 @@
   {% endcall %}
 
 {% endmacro %}
+
+{#
+  This changes the behaviour of the default adapter macro, since DBT defaults to 256 when there are no explicit varchar limits
+  (cf : https://github.com/dbt-labs/dbt-core/blob/3996a69861d5ba9a460092c93b7e08d8e2a63f88/core/dbt/adapters/base/column.py#L91)
+  Since normalization code uses varchar for string type (and not text) on postgres, we need to set the max length possible when using unlimited varchars
+  (cf : https://dba.stackexchange.com/questions/189876/size-limit-of-character-varying-postgresql)
+#}
+
+{% macro postgres__get_columns_in_relation(relation) -%}
+  {% call statement('get_columns_in_relation', fetch_result=True) %}
+      select
+          column_name,
+          data_type,
+          COALESCE(character_maximum_length, 10485760),
+          numeric_precision,
+          numeric_scale
+
+      from {{ relation.information_schema('columns') }}
+      where table_name = '{{ relation.identifier }}'
+        {% if relation.schema %}
+        and table_schema = '{{ relation.schema }}'
+        {% endif %}
+      order by ordinal_position
+
+  {% endcall %}
+  {% set table = load_result('get_columns_in_relation').table %}
+  {{ return(sql_convert_columns_in_relation(table)) }}
+{% endmacro %}
@@ -5,3 +5,7 @@
 {% macro oracle__current_timestamp() %}
     CURRENT_TIMESTAMP
 {% endmacro %}
+
+{% macro databricks__current_timestamp() %}
+    CURRENT_TIMESTAMP
+{% endmacro %}
@@ -8,6 +8,10 @@
     string
 {% endmacro %}
 
+{%- macro databricks__type_json() -%}
+    string
+{%- endmacro -%}
+
 {%- macro redshift__type_json() -%}
   {%- if redshift_super_type() -%}
     super
@@ -91,6 +95,10 @@
     INT
 {% endmacro %}
 
+{% macro databricks__type_int() %}
+    INT
+{% endmacro %}
+
 
 {# bigint ------------------------------------------------- #}
 {% macro mysql__type_bigint() %}
@@ -105,6 +113,10 @@
     BIGINT
 {% endmacro %}
 
+{% macro databricks__type_bigint() %}
+    BIGINT
+{% endmacro %}
+
 
 {# numeric ------------------------------------------------- --#}
 {% macro mysql__type_numeric() %}
@@ -115,6 +127,10 @@
     Float64
 {% endmacro %}
 
+{% macro databricks__type_numeric() %}
+    FLOAT
+{% endmacro %}
+
 
 {# timestamp ------------------------------------------------- --#}
 {% macro mysql__type_timestamp() %}
@@ -146,6 +162,12 @@
     timestamp
 {% endmacro %}
 
+{#-- Spark timestamps are already 'point in time', even if converted / stored without the original tz info, relative to session tz  --#}
+{#-- cf: https://docs.databricks.com/spark/latest/dataframes-datasets/dates-timestamps.html --#}
+{% macro databricks__type_timestamp_with_timezone() %}
+    timestamp
+{% endmacro %}
+
 {#-- MySQL doesnt allow cast operation to work with TIMESTAMP so we have to use char --#}
 {%- macro mysql__type_timestamp_with_timezone() -%}
     char
 
@@ -6,6 +6,7 @@
     - Postgres: json_extract_path_text(<from_json>, 'path' [, 'path' [, ...}}) -> https://www.postgresql.org/docs/12/functions-json.html
     - MySQL: JSON_EXTRACT(json_doc, 'path' [, 'path'] ...) -> https://dev.mysql.com/doc/refman/8.0/en/json-search-functions.html
     - ClickHouse: JSONExtractString(json_doc, 'path' [, 'path'] ...) -> https://clickhouse.com/docs/en/sql-reference/functions/json-functions/
+    - Databricks: get_json_object(json_txt, 'path') -> https://spark.apache.org/docs/latest/api/sql/#get_json_object
 #}
 
 {# format_json_path --------------------------------------------------     #}
@@ -42,6 +43,15 @@
     {{ "'$.\"" ~ json_path_list|join(".") ~ "\"'" }}
 {%- endmacro %}
 
+{% macro databricks__format_json_path(json_path_list) -%}
+    {# -- '$.x.y.z' #}
+    {%- set str_list = [] -%}
+    {%- for json_path in json_path_list -%}
+        {%- if str_list.append(json_path.replace("'", "\\'")) -%} {%- endif -%}
+    {%- endfor -%}
+    {{ "'$." ~ str_list|join(".") ~ "'" }}
+{%- endmacro %}
+
 {% macro redshift__format_json_path(json_path_list) -%}
     {%- set quote = '"' if redshift_super_type() else "'" -%}
     {%- set str_list = [] -%}
@@ -86,6 +96,14 @@
     json_extract({{ from_table}}.{{ json_column }}, {{ format_json_path(json_path_list) }})
 {%- endmacro %}
 
+{% macro databricks__json_extract(from_table, json_column, json_path_list, normalized_json_path) -%}
+    {%- if from_table|string() == '' %}
+        get_json_object({{ json_column }}, {{ format_json_path(json_path_list) }})
+    {% else %}
+        get_json_object({{ from_table }}.{{ json_column }}, {{ format_json_path(json_path_list) }})
+    {% endif -%}
+{%- endmacro %}
+
 {% macro oracle__json_extract(from_table, json_column, json_path_list, normalized_json_path) -%}
     json_value({{ json_column }}, {{ format_json_path(normalized_json_path) }})
 {%- endmacro %}
@@ -191,6 +209,10 @@
     JSONExtractRaw(assumeNotNull({{ json_column }}), {{ format_json_path(json_path_list) }})
 {%- endmacro %}
 
+{% macro databricks__json_extract_scalar(json_column, json_path_list, normalized_json_path) -%}
+    get_json_object({{ json_column }}, {{ format_json_path(json_path_list) }})
+{%- endmacro %}
+
 {# json_extract_array -------------------------------------------------     #}
 
 {% macro json_extract_array(json_column, json_path_list, normalized_json_path) -%}
@@ -237,6 +259,10 @@
     JSONExtractArrayRaw(assumeNotNull({{ json_column }}), {{ format_json_path(json_path_list) }})
 {%- endmacro %}
 
+{% macro databricks__json_extract_array(json_column, json_path_list, normalized_json_path) -%}
+    get_json_object({{ json_column }}, {{ format_json_path(json_path_list) }})
+{%- endmacro %}
+
 {# json_extract_string_array -------------------------------------------------     #}
 
 {% macro json_extract_string_array(json_column, json_path_list, normalized_json_path) -%}
 
@@ -4,12 +4,43 @@
     - the column _airbyte_ab_id does not exists in the normalized tables and make sure it is well populated.
 #}
 
+{%- macro get_columns_in_relation_if_exist(target_table) -%}
+    {{ return(adapter.dispatch('get_columns_in_relation_if_exist')(target_table)) }}
+{%- endmacro -%}
+
+{%- macro default__get_columns_in_relation_if_exist(target_table) -%}
+    {{ return(adapter.get_columns_in_relation(target_table)) }}
+{%- endmacro -%}
+
+{%- macro databricks__get_columns_in_relation_if_exist(target_table) -%}
+    {%- if target_table.schema is none -%}
+        {%- set found_table = True %}
+    {%- else -%}
+    {% call statement('list_table_infos', fetch_result=True) -%}
+        show tables in {{ target_table.schema }} like '*'
+    {% endcall %}
+    {%- set existing_tables = load_result('list_table_infos').table -%}
+    {%- set found_table = [] %}
+    {%- for table in existing_tables -%}
+        {%- if table.tableName == target_table.identifier -%}
+            {% do found_table.append(table.tableName) %}
+        {%- endif -%}
+    {%- endfor -%}
+    {%- endif -%}
+    {%- if found_table -%}
+        {%- set cols = adapter.get_columns_in_relation(target_table) -%}
+        {{ return(cols) }}
+    {%- else -%}
+        {{ return ([]) }}
+    {%- endif -%}
+{%- endmacro -%}
+
 {%- macro need_full_refresh(col_ab_id, target_table=this) -%}
     {%- if not execute -%}
         {{ return(false) }}
     {%- endif -%}
     {%- set found_column = [] %}
-    {%- set cols = adapter.get_columns_in_relation(target_table) -%}
+    {%- set cols = get_columns_in_relation_if_exist(target_table) -%}
     {%- for col in cols -%}
         {%- if col.column == col_ab_id -%}
             {% do found_column.append(col.column) %}
@@ -18,7 +49,7 @@
     {%- if found_column -%}
         {{ return(false) }}
     {%- else -%}
-        {{ dbt_utils.log_info(target_table ~ "." ~ col_ab_id ~ " does not exist yet. The table will be created or rebuilt with dbt.full_refresh") }}
+        {{ dbt_utils.log_info(target_table ~ "." ~ col_ab_id ~ " does not exist. The table needs to be rebuilt in full_refresh") }}
         {{ return(true) }}
     {%- endif -%}
 {%- endmacro -%}