construct dml merge and copy into

Author: edgarrmondragon

.github/workflows/test.yml

@@ -41,4 +41,4 @@ jobs:
         poetry install
     - name: Test with pytest
       run: |
-        poetry run pytest -n auto
+        poetry run pytest -s -n auto

target_snowflake/connector.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import sys
 from operator import contains, eq
 from typing import TYPE_CHECKING, Any, Iterable, Sequence, cast
 
@@ -9,8 +10,8 @@
 from cryptography.hazmat.primitives import serialization
 from singer_sdk import typing as th
 from singer_sdk.connectors import SQLConnector
-from snowflake.sqlalchemy import URL
 from snowflake.sqlalchemy.base import SnowflakeIdentifierPreparer
+from snowflake.sqlalchemy.custom_commands import CopyInto, MergeInto
 from snowflake.sqlalchemy.snowdialect import SnowflakeDialect
 from sqlalchemy.sql import quoted_name, text
 
@@ -386,6 +387,19 @@ def _get_merge_from_stage_statement(  # noqa: ANN202
         )
         dedup_cols = ", ".join(list(formatted_key_properties))
         dedup = f"QUALIFY ROW_NUMBER() OVER (PARTITION BY {dedup_cols} ORDER BY SEQ8() DESC) = 1"
+
+        merge = MergeInto(
+            target=full_table_name,
+            source=text(f"""
+                (select {json_casting_selects} from '@~/target-snowflake/{sync_id}'
+                (file_format => {file_format}) {dedup}) s
+            """),  # noqa: S608
+            on=join_expr,
+        )
+        merge.when_matched_then_update()
+        merge.when_not_matched_then_insert()
+        print(">>> MERGE INTO:", merge.compile(self._engine), file=sys.stderr)
+
         return (
             text(
                 f"merge into {quoted_name(full_table_name, quote=True)} d using "  # noqa: ISC003
@@ -411,6 +425,15 @@ def _get_copy_statement(self, full_table_name, schema, sync_id, file_format):  #
             column_selections,
             "col_alias",
         )
+        copy_into = CopyInto(
+            from_=text(f"""
+                (select {json_casting_selects} from '@~/target-snowflake/{sync_id}')
+            """),  # noqa: S608
+            into=full_table_name,
+            formatter=formatter,
+        )
+        print(">>> COPY INTO:", copy_into.compile(self._engine), file=sys.stderr)
+
         return (
             text(
                 f"copy into {full_table_name} {col_alias_selects} from "  # noqa: ISC003