Repository: meltano/sdk
Pull request #1893 (2023-08-03 19:53:51)

feat: Standard configurable load methods
Closes https://github.com/meltano/sdk/issues/1783

I was attempting to add this `overwrite` feature to target-pinecone https://github.com/MeltanoLabs/target-pinecone/issues/4 so I decided it would probably best to draft it in the SDK while I was at it.

- Adds a standard config `load_method` for: append-only, upsert, overwrite
- Implements the overwrite workflow in SQL targets
- By default overwrite isnt supported until the target developer activates it. Is this what we want or should it default to `True`?
- I wasnt sure if the truncate logic should go in `setup` of the SQLSink class or `prepare_table`. For now I chose `prepare_table`
- Ideally a target would throw a nice NotImplementedError if the user choses one of the load methods that arent supported but I'm not totally sure how to do that as of right now. It might require more thought. 

@edgarrmondragon let me know what you think overall and also specifically with the naming of everything.

cc @tayloramurphy in case you have opinions on naming or default behavior.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/meltano/sdk/pull/1893?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) Report
> Merging [#1893](https://app.codecov.io/gh/meltano/sdk/pull/1893?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) (482e06b) into [main](https://app.codecov.io/gh/meltano/sdk/commit/65d9dbacb6a000c7ff2058b7b578f57fec56b725?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) (65d9dba) will **decrease** coverage by `0.11%`.
> The diff coverage is `57.89%`.

```diff
@@            Coverage Diff             @@
##             main    #1893      +/-   ##
==========================================
- Coverage   86.93%   86.82%   -0.11%     
==========================================
  Files          59       59              
  Lines        5073     5092      +19     
  Branches      822      825       +3     
==========================================
+ Hits         4410     4421      +11     
- Misses        466      473       +7     
- Partials      197      198       +1     
```


| [Files Changed](https://app.codecov.io/gh/meltano/sdk/pull/1893?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) | Coverage Δ | |
|---|---|---|
| [singer\_sdk/connectors/sql.py](https://app.codecov.io/gh/meltano/sdk/pull/1893?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano#diff-c2luZ2VyX3Nkay9jb25uZWN0b3JzL3NxbC5weQ==) | `79.59% <38.46%> (-1.87%)` | :arrow_down: |
| [singer\_sdk/helpers/capabilities.py](https://app.codecov.io/gh/meltano/sdk/pull/1893?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano#diff-c2luZ2VyX3Nkay9oZWxwZXJzL2NhcGFiaWxpdGllcy5weQ==) | `92.42% <100.00%> (+0.62%)` | :arrow_up: |
| [singer\_sdk/target\_base.py](https://app.codecov.io/gh/meltano/sdk/pull/1893?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano#diff-c2luZ2VyX3Nkay90YXJnZXRfYmFzZS5weQ==) | `92.33% <100.00%> (+0.02%)` | :arrow_up: |

:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano)

tayloramurphy: @pnadolny13 I like overwrite! It matches Snowflake too https://docs.snowflake.com/en/sql-reference/sql/insert#optional-parameters 
pnadolny13: @edgarrmondragon I added a test for the sqlite sample, and refactored a little. I couldnt get custom query parameterization to work for some reason so I used f strings and had to add noqa flags for ruff to pass.
pnadolny13: @edgarrmondragon I swapped the logic in this to use a drop/recreate flow instead of truncating. I started to implement the rename + rollback workflow but I wasnt able to find anywhere to easily put logic to rollback if an exception is thrown. I tried using the [clean_up](https://github.com/meltano/sdk/blob/36f74363f0eaa9c2e2114af6a0d7de2790a20133/singer_sdk/sinks/core.py#L489) method which worked when the sync was successful to remove the backup table before exiting but I couldnt figure out how to easily catch exceptions and revert to the backup table prior to exiting.
edgarrmondragon: @pnadolny13 I think I was probably overstating the risk of data loss caused by pipeline failures when syncing in _overwrite_ mode. If the user is OK with overriding existing data, then they're probably also OK-ish with losing data if the pipeline fails, as long as they can load data on the next run.
edgarrmondragon: > @edgarrmondragon I swapped the logic in this to use a drop/recreate flow instead of truncating. I started to implement the rename + rollback workflow but I wasnt able to find anywhere to easily put logic to rollback if an exception is thrown. I tried using the [clean_up](https://github.com/meltano/sdk/blob/36f74363f0eaa9c2e2114af6a0d7de2790a20133/singer_sdk/sinks/core.py#L489) method which worked when the sync was successful to remove the backup table before exiting but I couldnt figure out how to easily catch exceptions and revert to the backup table prior to exiting. I create a [draft PR](https://github.com/meltano/sdk/pull/1941) to this branch with the code I was tinkering with.

@pnadolny13 Let'd punt the rollback support and create an issue for it. We'll worry about it if users are annoyed by losing data 🙂 

Patch:
diff --git a/cookiecutter/target-template/{{cookiecutter.target_id}}/{{cookiecutter.library_name}}/sinks.py b/cookiecutter/target-template/{{cookiecutter.target_id}}/{{cookiecutter.library_name}}/sinks.py
index 4e84d128..9edd13a1 100644
--- a/cookiecutter/target-template/{{cookiecutter.target_id}}/{{cookiecutter.library_name}}/sinks.py
+++ b/cookiecutter/target-template/{{cookiecutter.target_id}}/{{cookiecutter.library_name}}/sinks.py
@@ -35,6 +35,7 @@ class {{ cookiecutter.destination_name }}Connector(SQLConnector):
     allow_column_rename: bool = True  # Whether RENAME COLUMN is supported.
     allow_column_alter: bool = False  # Whether altering column types is supported.
     allow_merge_upsert: bool = False  # Whether MERGE UPSERT is supported.
+    allow_overwrite: bool = False  # Whether overwrite load method is supported.
     allow_temp_tables: bool = True  # Whether temp tables are supported.
 
     def get_sqlalchemy_url(self, config: dict) -> str:
diff --git a/samples/sample_target_sqlite/__init__.py b/samples/sample_target_sqlite/__init__.py
index bd759e46..40384fac 100644
--- a/samples/sample_target_sqlite/__init__.py
+++ b/samples/sample_target_sqlite/__init__.py
@@ -19,6 +19,7 @@ class SQLiteConnector(SQLConnector):
     allow_temp_tables = False
     allow_column_alter = False
     allow_merge_upsert = True
+    allow_overwrite: bool = True
 
     def get_sqlalchemy_url(self, config: dict[str, t.Any]) -> str:
         """Generates a SQLAlchemy URL for SQLite."""
diff --git a/singer_sdk/connectors/sql.py b/singer_sdk/connectors/sql.py
index fb49b358..2693bfe2 100644
--- a/singer_sdk/connectors/sql.py
+++ b/singer_sdk/connectors/sql.py
@@ -18,6 +18,7 @@ from sqlalchemy.engine import Engine
 from singer_sdk import typing as th
 from singer_sdk._singerlib import CatalogEntry, MetadataMapping, Schema
 from singer_sdk.exceptions import ConfigValidationError
+from singer_sdk.helpers.capabilities import TargetLoadMethods
 
 if t.TYPE_CHECKING:
     from sqlalchemy.engine.reflection import Inspector
@@ -40,6 +41,7 @@ class SQLConnector:
     allow_column_rename: bool = True  # Whether RENAME COLUMN is supported.
     allow_column_alter: bool = False  # Whether altering column types is supported.
     allow_merge_upsert: bool = False  # Whether MERGE UPSERT is supported.
+    allow_overwrite: bool = False  # Whether overwrite load method is supported.
     allow_temp_tables: bool = True  # Whether temp tables are supported.
     _cached_engine: Engine | None = None
 
@@ -775,6 +777,16 @@ class SQLConnector:
                 as_temp_table=as_temp_table,
             )
             return
+        if self.config["load_method"] == TargetLoadMethods.OVERWRITE:
+            self.get_table(full_table_name=full_table_name).drop(self._engine)
+            self.create_empty_table(
+                full_table_name=full_table_name,
+                schema=schema,
+                primary_keys=primary_keys,
+                partition_keys=partition_keys,
+                as_temp_table=as_temp_table,
+            )
+            return
 
         for property_name, property_def in schema["properties"].items():
             self.prepare_column(
diff --git a/singer_sdk/helpers/capabilities.py b/singer_sdk/helpers/capabilities.py
index f5b5fa30..690f04db 100644
--- a/singer_sdk/helpers/capabilities.py
+++ b/singer_sdk/helpers/capabilities.py
@@ -108,6 +108,40 @@ ADD_RECORD_METADATA_CONFIG = PropertiesList(
 ).to_dict()
 
 
+class TargetLoadMethods(str, Enum):
+    """Target-specific capabilities."""
+
+    # always write all input records whether that records already exists or not
+    APPEND_ONLY = "append-only"
+
+    # update existing records and insert new records
+    UPSERT = "upsert"
+
+    # delete all existing records and insert all input records
+    OVERWRITE = "overwrite"
+
+
+TARGET_LOAD_METHOD_CONFIG = PropertiesList(
+    Property(
+        "load_method",
+        StringType(),
+        description=(
+            "The method to use when loading data into the destination. "
+            "`append-only` will always write all input records whether that records "
+            "already exists or not. `upsert` will update existing records and insert "
+            "new records. `overwrite` will delete all existing records and insert all "
+            "input records."
+        ),
+        allowed_values=[
+            TargetLoadMethods.APPEND_ONLY,
+            TargetLoadMethods.UPSERT,
+            TargetLoadMethods.OVERWRITE,
+        ],
+        default=TargetLoadMethods.APPEND_ONLY,
+    ),
+).to_dict()
+
+
 class DeprecatedEnum(Enum):
     """Base class for capabilities enumeration."""
 
diff --git a/singer_sdk/target_base.py b/singer_sdk/target_base.py
index a5386199..d902f2ac 100644
--- a/singer_sdk/target_base.py
+++ b/singer_sdk/target_base.py
@@ -19,6 +19,7 @@ from singer_sdk.helpers._compat import final
 from singer_sdk.helpers.capabilities import (
     ADD_RECORD_METADATA_CONFIG,
     BATCH_CONFIG,
+    TARGET_LOAD_METHOD_CONFIG,
     TARGET_SCHEMA_CONFIG,
     CapabilitiesEnum,
     PluginCapabilities,
@@ -597,6 +598,7 @@ class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
                     target_jsonschema["properties"][k] = v
 
         _merge_missing(ADD_RECORD_METADATA_CONFIG, config_jsonschema)
+        _merge_missing(TARGET_LOAD_METHOD_CONFIG, config_jsonschema)
 
         capabilities = cls.capabilities
 


Test patch:
diff --git a/tests/samples/test_target_sqlite.py b/tests/samples/test_target_sqlite.py
index abf1bcca..995cb03a 100644
--- a/tests/samples/test_target_sqlite.py
+++ b/tests/samples/test_target_sqlite.py
@@ -508,3 +508,48 @@ def test_hostile_to_sqlite(
         "hname_starts_with_number",
         "name_with_emoji_",
     }
+
+
+def test_overwrite_load_method(
+    sqlite_target_test_config: dict,
+):
+    sqlite_target_test_config["load_method"] = "overwrite"
+    target = SQLiteTarget(config=sqlite_target_test_config)
+    test_tbl = f"zzz_tmp_{str(uuid4()).split('-')[-1]}"
+    schema_msg = {
+        "type": "SCHEMA",
+        "stream": test_tbl,
+        "schema": {
+            "type": "object",
+            "properties": {"col_a": th.StringType().to_dict()},
+        },
+    }
+
+    tap_output_a = "\n".join(
+        json.dumps(msg)
+        for msg in [
+            schema_msg,
+            {"type": "RECORD", "stream": test_tbl, "record": {"col_a": "123"}},
+        ]
+    )
+    # Assert
+    db = sqlite3.connect(sqlite_target_test_config["path_to_db"])
+    cursor = db.cursor()
+
+    target_sync_test(target, input=StringIO(tap_output_a), finalize=True)
+    cursor.execute(f"SELECT col_a FROM {test_tbl} ;")  # noqa: S608
+    records = [res[0] for res in cursor.fetchall()]
+    assert records == ["123"]
+
+    tap_output_b = "\n".join(
+        json.dumps(msg)
+        for msg in [
+            schema_msg,
+            {"type": "RECORD", "stream": test_tbl, "record": {"col_a": "456"}},
+        ]
+    )
+    target = SQLiteTarget(config=sqlite_target_test_config)
+    target_sync_test(target, input=StringIO(tap_output_b), finalize=True)
+    cursor.execute(f"SELECT col_a FROM {test_tbl} ;")  # noqa: S608
+    records = [res[0] for res in cursor.fetchall()]
+    assert records == ["456"]