🎉 MySQL destination: normalization

airbyte-integrations/bases/base-normalization/README.md

@@ -54,6 +54,7 @@ allowed characters, if quotes are needed or not, and the length limitations:
 - [postgres](../../../docs/integrations/destinations/postgres.md)
 - [redshift](../../../docs/integrations/destinations/redshift.md)
 - [snowflake](../../../docs/integrations/destinations/snowflake.md)
+- [mysql](../../../docs/integrations/destinations/mysql.md)
 
 Rules about truncations, for example for both of these strings which are too long for the postgres 64 limit:
 - `Aaaa_Bbbb_Cccc_Dddd_Eeee_Ffff_Gggg_Hhhh_Iiii`
@@ -216,6 +217,14 @@ A nice improvement would be to add csv/json seed files as expected output data f
 The integration tests would verify that the content of such tables in the destination would match
 these seed files or fail.
 
+### Debug dbt operations with local database
+This only works for testing databases launched in local containers (e.g. postgres and mysql).
+
+- In `dbt_integration_test.py`, comment out the `tear_down_db` method so that the relevant database container is not deleted.
+- Find the name of the database container in the logs (e.g. by searching `Executing`).
+- Connect to the container by running `docker exec -it <container-name> bash` in the commandline.
+- Connect to the database inside the container (e.g. `mysql -u root` for mysql).
+
 ## Standard Destination Tests
 
 Generally, to invoke standard destination tests, you run with gradle using:

airbyte-integrations/bases/base-normalization/build.gradle

@@ -25,6 +25,7 @@ task("customIntegrationTestPython", type: PythonTask, dependsOn: installTestReqs
     dependsOn ':airbyte-integrations:connectors:destination-postgres:airbyteDocker'
     dependsOn ':airbyte-integrations:connectors:destination-redshift:airbyteDocker'
     dependsOn ':airbyte-integrations:connectors:destination-snowflake:airbyteDocker'
+    dependsOn ':airbyte-integrations:connectors:destination-mysql:airbyteDocker'
 }
 
 integrationTest.dependsOn("customIntegrationTestPython")

airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/array.sql

@@ -28,6 +28,10 @@
     ) as _airbyte_nested_data
 {%- endmacro %}
 
+{% macro mysql__cross_join_unnest(stream_name, array_col) -%}
+    left join joined on _airbyte_{{ stream_name }}_hashid = joined._airbyte_hashid
+{%- endmacro %}
+
 {% macro redshift__cross_join_unnest(stream_name, array_col) -%}
     left join joined on _airbyte_{{ stream_name }}_hashid = joined._airbyte_hashid
 {%- endmacro %}
@@ -36,7 +40,7 @@
     cross join table(flatten({{ array_col }})) as {{ array_col }}
 {%- endmacro %}
 
-{# unnested_column_value -------------------------------------------------     #}
+{# unnested_column_value -- this macro is related to unnest_cte #}
 
 {% macro unnested_column_value(column_col) -%}
   {{ adapter.dispatch('unnested_column_value')(column_col) }}
@@ -58,6 +62,10 @@
     _airbyte_nested_data
 {%- endmacro %}
 
+{% macro mysql__unnested_column_value(column_col) -%}
+    _airbyte_nested_data
+{%- endmacro %}
+
 {# unnest_cte -------------------------------------------------     #}
 
 {% macro unnest_cte(table_name, stream_name, column_col) -%}
@@ -97,3 +105,37 @@ joined as (
     where numbers.generated_number <= json_array_length({{ column_col }}, true)
 )
 {%- endmacro %}
+
+{% macro mysql__unnest_cte(table_name, stream_name, column_col) -%}
+    {%- if not execute -%}
+        {{ return('') }}
+    {% endif %}
+
+    {%- call statement('max_json_array_length', fetch_result=True) -%}
+        with max_value as (
+            select max(json_length({{ column_col }})) as max_number_of_items
+            from {{ ref(table_name) }}
+        )
+        select
+            case when max_number_of_items is not null and max_number_of_items > 1
+            then max_number_of_items
+            else 1 end as max_number_of_items
+        from max_value
+    {%- endcall -%}
+
+    {%- set max_length = load_result('max_json_array_length') -%}
+    with numbers as (
+        {{ dbt_utils.generate_series(max_length["data"][0][0]) }}
+    ),
+    joined as (
+        select
+            _airbyte_{{ stream_name }}_hashid as _airbyte_hashid,
+            {# -- json_extract(column_col, '$[i][0]') as _airbyte_nested_data #}
+            json_extract({{ column_col }}, concat("$[", numbers.generated_number - 1, "][0]")) as _airbyte_nested_data
+        from {{ ref(table_name) }}
+        cross join numbers
+        -- only generate the number of records in the cross join that corresponds
+        -- to the number of items in {{ table_name }}.{{ column_col }}
+        where numbers.generated_number <= json_length({{ column_col }})
+    )
+{%- endmacro %}

airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/datatypes.sql

@@ -19,3 +19,11 @@
 {% macro snowflake__type_json() %}
     variant
 {% endmacro %}
+
+{%- macro mysql__type_json() -%}
+    json
+{%- endmacro -%}
+
+{%- macro mysql__type_string() -%}
+    char
+{%- endmacro -%}

airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/except.sql

@@ -0,0 +1,3 @@
+{% macro mysql__except() %}
+    {% do exceptions.warn("MySQL does not support EXCEPT operator") %}
+{% endmacro %}

airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/json_operations.sql

@@ -2,8 +2,9 @@
     Adapter Macros for the following functions:
     - Bigquery: JSON_EXTRACT(json_string_expr, json_path_format) -> https://cloud.google.com/bigquery/docs/reference/standard-sql/json_functions
     - Snowflake: JSON_EXTRACT_PATH_TEXT( <column_identifier> , '<path_name>' ) -> https://docs.snowflake.com/en/sql-reference/functions/json_extract_path_text.html
-    - Redshift: json_extract_path_text('json_string', 'path_elem' [,'path_elem'[, …] ] [, null_if_invalid ] ) -> https://docs.aws.amazon.com/redshift/latest/dg/JSON_EXTRACT_PATH_TEXT.html
+    - Redshift: json_extract_path_text('json_string', 'path_elem' [,'path_elem'[, ...] ] [, null_if_invalid ] ) -> https://docs.aws.amazon.com/redshift/latest/dg/JSON_EXTRACT_PATH_TEXT.html
     - 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
 #}
 
 {# format_json_path --------------------------------------------------     #}
@@ -23,6 +24,11 @@
  {{ "'" ~ json_path_list|join("','") ~ "'" }}
 {%- endmacro %}
 
+{% macro mysql__format_json_path(json_path_list) -%}
+  {# -- '$."x"."y"."z"' #}
+  {{ "'$.\"" ~ json_path_list|join(".") ~ "\"'" }}
+{%- endmacro %}
+
 {% macro redshift__format_json_path(json_path_list) -%}
  {{ "'" ~ json_path_list|join("','") ~ "'" }}
 {%- endmacro %}
@@ -49,6 +55,10 @@
     jsonb_extract_path({{ json_column }}, {{ format_json_path(json_path_list) }})
 {%- endmacro %}
 
+{% macro mysql__json_extract(json_column, json_path_list) -%}
+    json_extract({{ json_column }}, {{ format_json_path(json_path_list) }})
+{%- endmacro %}
+
 {% macro redshift__json_extract(json_column, json_path_list) -%}
     case when json_extract_path_text({{ json_column }}, {{ format_json_path(json_path_list) }}, true) != '' then json_extract_path_text({{ json_column }}, {{ format_json_path(json_path_list) }}, true) end
 {%- endmacro %}
@@ -75,6 +85,10 @@
     jsonb_extract_path_text({{ json_column }}, {{ format_json_path(json_path_list) }})
 {%- endmacro %}
 
+{% macro mysql__json_extract_scalar(json_column, json_path_list) -%}
+    json_extract({{ json_column }}, {{ format_json_path(json_path_list) }})
+{%- endmacro %}
+
 {% macro redshift__json_extract_scalar(json_column, json_path_list) -%}
     case when json_extract_path_text({{ json_column }}, {{ format_json_path(json_path_list) }}, true) != '' then json_extract_path_text({{ json_column }}, {{ format_json_path(json_path_list) }}, true) end
 {%- endmacro %}
@@ -101,6 +115,10 @@
     jsonb_extract_path({{ json_column }}, {{ format_json_path(json_path_list) }})
 {%- endmacro %}
 
+{% macro mysql__json_extract_array(json_column, json_path_list) -%}
+    json_extract({{ json_column }}, {{ format_json_path(json_path_list) }})
+{%- endmacro %}
+
 {% macro redshift__json_extract_array(json_column, json_path_list) -%}
     json_extract_path_text({{ json_column }}, {{ format_json_path(json_path_list) }}, true)
 {%- endmacro %}

airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/type_conversions.sql

@@ -38,3 +38,8 @@
 {% macro redshift__cast_to_boolean(field) -%}
     cast(decode({{ field }}, 'true', '1', 'false', '0')::integer as boolean)
 {%- endmacro %}
+
+{# cast_to_bigint -------------------------------------------------     #}
+{% macro mysql__type_bigint() %}
+    signed
+{% endmacro %}

airbyte-integrations/bases/base-normalization/dbt-project-template/macros/schema_tests/equality.sql

@@ -0,0 +1,69 @@
+{#
+-- Adapted from https://github.com/fishtown-analytics/dbt-utils/blob/bbba960726667abc66b42624f0d36bbb62c37593/macros/schema_tests/equality.sql
+-- dbt-utils version: 0.6.4
+-- This macro needs to be updated accordingly when dbt-utils is upgraded.
+-- This is needed because MySQL does not support the EXCEPT operator!
+#}
+
+{% macro mysql__test_equality(model, compare_model, compare_columns=None) %}
+
+    {% set set_diff %}
+        count(*) + coalesce(abs(
+            sum(case when which_diff = 'a_minus_b' then 1 else 0 end) -
+            sum(case when which_diff = 'b_minus_a' then 1 else 0 end)
+        ), 0)
+    {% endset %}
+
+    {{ config(fail_calc = set_diff) }}
+
+    {%- if not execute -%}
+        {{ return('') }}
+    {% endif %}
+
+    {%- do dbt_utils._is_relation(model, 'test_equality') -%}
+
+    {%- if not compare_columns -%}
+        {%- do dbt_utils._is_ephemeral(model, 'test_equality') -%}
+        {%- set compare_columns = adapter.get_columns_in_relation(model) | map(attribute='quoted') -%}
+    {%- endif -%}
+
+    {% set compare_cols_csv = compare_columns | join(', ') %}
+
+    with a as (
+        select * from {{ model }}
+    ),
+
+    b as (
+        select * from {{ compare_model }}
+    ),
+
+    a_minus_b as (
+        select {{ compare_cols_csv }} from a
+        where ({{ compare_cols_csv }}) not in
+            (select {{ compare_cols_csv }} from b)
+    ),
+
+    b_minus_a as (
+        select {{ compare_cols_csv }} from b
+        where ({{ compare_cols_csv }}) not in
+            (select {{ compare_cols_csv }} from a)
+    ),
+
+    unioned as (
+        select * from a_minus_b
+        union all
+        select * from b_minus_a
+    ),
+
+    final as (
+        select (select count(*) from unioned) +
+        (select abs(
+            (select count(*) from a_minus_b) -
+            (select count(*) from b_minus_a)
+            ))
+        as count
+    )
+
+    select count from final
+
+{% endmacro %}

airbyte-integrations/bases/base-normalization/integration_tests/dbt_integration_test.py

@@ -42,12 +42,17 @@
 class DbtIntegrationTest(object):
     def __init__(self):
         self.target_schema = "test_normalization"
-        self.container_name = "test_normalization_db_" + self.random_string(3)
+        self.container_prefix = f"test_normalization_db_{self.random_string(3)}"
+        self.db_names = ["postgres", "mysql"]
 
     @staticmethod
     def random_string(length: int) -> str:
         return "".join(random.choice(string.ascii_lowercase) for i in range(length))
 
+    def setup_db(self):
+        self.setup_postgres_db()
+        self.setup_mysql_db()
+
     def setup_postgres_db(self):
         print("Starting localhost postgres container for tests")
         port = self.find_free_port()
@@ -64,7 +69,7 @@ def setup_postgres_db(self):
             "run",
             "--rm",
             "--name",
-            f"{self.container_name}",
+            f"{self.container_prefix}_postgres",
             "-e",
             f"POSTGRES_USER={config['username']}",
             "-e",
@@ -81,6 +86,42 @@ def setup_postgres_db(self):
         with open("../secrets/postgres.json", "w") as fh:
             fh.write(json.dumps(config))
 
+    def setup_mysql_db(self):
+        print("Starting localhost mysql container for tests")
+        port = self.find_free_port()
+        config = {
+            "type": "mysql",
+            "host": "localhost",
+            "port": port,
+            "database": self.target_schema,
+            "username": "root",
+            "password": "",
+        }
+        commands = [
+            "docker",
+            "run",
+            "--rm",
+            "--name",
+            f"{self.container_prefix}_mysql",
+            "-e",
+            "MYSQL_ALLOW_EMPTY_PASSWORD=yes",
+            "-e",
+            "MYSQL_INITDB_SKIP_TZINFO=yes",
+            "-e",
+            f"MYSQL_DATABASE={config['database']}",
+            "-p",
+            f"{config['port']}:3306",
+            "-d",
+            "mysql",
+        ]
+        print("Executing: ", " ".join(commands))
+        subprocess.call(commands)
+
+        if not os.path.exists("../secrets"):
+            os.makedirs("../secrets")
+        with open("../secrets/mysql.json", "w") as fh:
+            fh.write(json.dumps(config))
+
     @staticmethod
     def find_free_port():
         """
@@ -92,12 +133,13 @@ def find_free_port():
         s.close()
         return addr[1]
 
-    def tear_down_postgres_db(self):
-        print("Stopping localhost postgres container for tests")
-        try:
-            subprocess.call(["docker", "kill", f"{self.container_name}"])
-        except Exception as e:
-            print(f"WARN: Exception while shutting down postgres db: {e}")
+    def tear_down_db(self):
+        for db_name in self.db_names:
+            print(f"Stopping localhost {db_name} container for tests")
+            try:
+                subprocess.call(["docker", "kill", f"{self.container_prefix}_{db_name}"])
+            except Exception as e:
+                print(f"WARN: Exception while shutting down {db_name}: {e}")
 
     @staticmethod
     def change_current_test_dir(request):

airbyte-integrations/bases/base-normalization/integration_tests/resources/test_primary_key_streams/data_input/replace_identifiers.json

@@ -22,5 +22,26 @@
     { "#quote: true in postgres": "quote: true" }
   ],
   "snowflake": [{ "HKD@SPÉÇIÄL & CHARACTERS": "HKD@spéçiäl & characters" }],
-  "redshift": []
+  "redshift": [],
+  "mysql": [
+    {
+      "_airbyte_raw_nested_stream_with_complex_columns_resulting_into_long_names": "_airbyte_raw_nested_s__lting_into_long_names"
+    },
+    {
+      "nested_stream_with_complex_columns_resulting_into_long_names_partition_double_array_data": "nested_stream_with_co__ion_double_array_data"
+    },
+    {
+      "nested_stream_with_complex_columns_resulting_into_long_names_partition_data": "nested_stream_with_co___names_partition_data"
+    },
+    {
+      "nested_stream_with_complex_columns_resulting_into_long_names_partition": "nested_stream_with_co___long_names_partition"
+    },
+    {
+      "'nested_stream_with_complex_columns_resulting_into_long_names'": "'nested_stream_with_co__lting_into_long_names'"
+    },
+    {
+      "non_nested_stream_without_namespace_resulting_into_long_names": "non_nested_stream_wit__lting_into_long_names"
+    },
+    { "#quote: true in mysql": "quote: true" }
+  ]
 }

airbyte-integrations/bases/base-normalization/integration_tests/resources/test_primary_key_streams/dbt_data_tests/test_check_row_counts.sql

@@ -34,7 +34,7 @@ union all
 union all
     select count(distinct currency) as row_count, 1 as expected_count
     from {{ ref('nested_stream_with_complex_columns_resulting_into_long_names_partition_DATA') }}
---union all
+-- union all
 --    select count(distinct id) as row_count, 3 as expected_count
 --    from {{ ref('nested_stream_with_complex_columns_resulting_into_long_names_partition_double_array_data') }}
 )