🎉 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`

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,18 @@
     ) as _airbyte_nested_data
 {%- endmacro %}
 
+{% macro mysql__cross_join_unnest(stream_name, array_col) -%}
+    cross join json_table(
+        {{ array_col }}
+        '$[*]'
+        COLUMNS(
+            "{{ array_col }}_col"
+            json
+            path '$'
+        )
+    ) as airbyte_nested_data
+{%- endmacro %}
+
 {% macro redshift__cross_join_unnest(stream_name, array_col) -%}
     left join joined on _airbyte_{{ stream_name }}_hashid = joined._airbyte_hashid
 {%- 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/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,10 @@
  {{ "'" ~ json_path_list|join("','") ~ "'" }}
 {%- endmacro %}
 
+{% macro mysql__format_json_path(json_path_list) -%}
+  {{ "\"$." ~ json_path_list|join('.') ~ "\"" }}
+{%- endmacro %}
+
 {% macro redshift__format_json_path(json_path_list) -%}
  {{ "'" ~ json_path_list|join("','") ~ "'" }}
 {%- endmacro %}
@@ -49,6 +54,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 +84,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 +114,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/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/test_ephemeral.py

@@ -42,10 +42,10 @@
 @pytest.fixture(scope="module", autouse=True)
 def before_all_tests(request):
     dbt_test_utils.change_current_test_dir(request)
-    dbt_test_utils.setup_postgres_db()
+    dbt_test_utils.setup_db()
     os.environ["PATH"] = os.path.abspath("../.venv/bin/") + ":" + os.environ["PATH"]
     yield
-    dbt_test_utils.tear_down_postgres_db()
+    dbt_test_utils.tear_down_db()
     for folder in temporary_folders:
         print(f"Deleting temporary test folder {folder}")
         shutil.rmtree(folder, ignore_errors=True)

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

@@ -47,10 +47,10 @@
 @pytest.fixture(scope="module", autouse=True)
 def before_all_tests(request):
     dbt_test_utils.change_current_test_dir(request)
-    dbt_test_utils.setup_postgres_db()
+    dbt_test_utils.setup_db()
     os.environ["PATH"] = os.path.abspath("../.venv/bin/") + ":" + os.environ["PATH"]
     yield
-    dbt_test_utils.tear_down_postgres_db()
+    dbt_test_utils.tear_down_db()
     for folder in temporary_folders:
         print(f"Deleting temporary test folder {folder}")
         shutil.rmtree(folder, ignore_errors=True)
@@ -75,7 +75,6 @@ def setup_test_path(request):
     ),
 )
 @pytest.mark.parametrize("destination_type", list(DestinationType))
-# @pytest.mark.parametrize("destination_type", [DestinationType.POSTGRES])
 def test_normalization(destination_type: DestinationType, test_resource_name: str, setup_test_path):
     print("Testing normalization")
     integration_type = destination_type.value

airbyte-integrations/bases/base-normalization/normalization/destination_type.py

@@ -31,6 +31,7 @@ class DestinationType(Enum):
     POSTGRES = "postgres"
     REDSHIFT = "redshift"
     SNOWFLAKE = "snowflake"
+    MYSQL = "mysql"
 
     @classmethod
     def from_string(cls, string_value: str) -> "DestinationType":

airbyte-integrations/bases/base-normalization/normalization/transform_catalog/destination_name_transformer.py

@@ -31,10 +31,16 @@
 from normalization.transform_catalog.utils import jinja_call
 
 DESTINATION_SIZE_LIMITS = {
+    # https://cloud.google.com/bigquery/quotas#all_tables
     DestinationType.BIGQUERY.value: 1024,
+    # https://docs.snowflake.com/en/sql-reference/identifiers-syntax.html
     DestinationType.SNOWFLAKE.value: 255,
+    # https://docs.aws.amazon.com/redshift/latest/dg/r_names.html
     DestinationType.REDSHIFT.value: 127,
+    # https://www.postgresql.org/docs/12/limits.html
     DestinationType.POSTGRES.value: 63,
+    # https://dev.mysql.com/doc/refman/8.0/en/identifier-length.html
+    DestinationType.MYSQL.value: 64,
 }
 
 # DBT also needs to generate suffix to table names, so we need to make sure it has enough characters to do so...
@@ -176,6 +182,9 @@ def __normalize_identifier_case(self, input_name: str, is_quoted: bool = False)
         elif self.destination_type.value == DestinationType.SNOWFLAKE.value:
             if not is_quoted and not self.needs_quotes(input_name):
                 result = input_name.upper()
+        elif self.destination_type.value == DestinationType.MYSQL.value:
+            if not is_quoted and not self.needs_quotes(input_name):
+                result = input_name.lower()
         else:
             raise KeyError(f"Unknown destination type {self.destination_type}")
         return result