Repository: meltano/sdk (python)

Issue:
bug: Incorrect table name resolved if stream name is hyphenated
### Singer SDK Version

0.44.3

### Is this a regression?

- [ ] Yes

### Python Version

NA

### Bug scope

Targets (data type handling, batching, SQL object generation, etc.)

### Operating System

_No response_

### Description

For SQL targets, if a stream name is hyphenated it is assumed to be fully-qualified and the part after the last hyphen is taken as the table name. In our case, we have a tap stream name of `tap_bigquery-weekly_fuel_prices` and expect a table of name in similar shape to be created in Snowflake - instead, `weekly_fuel_prices` is resolved which is not desirable is our case. Is there a reason for this behaviour?

https://github.com/meltano/sdk/blob/881ffd2041c9eca9c50a5e94ffe61a121579ef61/singer_sdk/sinks/sql.py#L76-L85

### Link to Slack/Linen

_No response_

Pull request description:
fix(targets): In SQL targets, use stream name as table name instead of splitting by hyphen if `default_target_schema` is speficied
## Related

- Closes #2909

cc @ReubenFrankel

## Summary by Sourcery

Modify SQL target table naming logic to use the full stream name when a default target schema is specified

New Features:
- Add new test cases to validate table and schema name generation for SQL targets

Bug Fixes:
- Fix table name generation to preserve full stream name when a default target schema is provided

Tests:
- Add comprehensive parametrized test cases for table name and schema name generation in different scenarios

<!-- readthedocs-preview meltano-sdk start -->
----
📚 Documentation preview 📚: https://meltano-sdk--3020.org.readthedocs.build/en/3020/

<!-- readthedocs-preview meltano-sdk end -->

Patch:
diff --git a/singer_sdk/sinks/sql.py b/singer_sdk/sinks/sql.py
index 9a82e59b..43d7a7cc 100644
--- a/singer_sdk/sinks/sql.py
+++ b/singer_sdk/sinks/sql.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import functools
 import re
 import typing as t
 import warnings
@@ -73,6 +74,15 @@ class SQLSink(BatchSink, t.Generic[_C]):
         """
         return self.connector.connection
 
+    @functools.cached_property
+    def default_target_schema(self) -> str | None:
+        """Return the default target schema.
+
+        Returns:
+            The default target schema.
+        """
+        return self.config.get("default_target_schema", None)
+
     @property
     def table_name(self) -> str:
         """Return the table name, with no schema or database part.
@@ -80,6 +90,9 @@ class SQLSink(BatchSink, t.Generic[_C]):
         Returns:
             The target table name.
         """
+        if self.default_target_schema:
+            return self.stream_name
+
         parts = self.stream_name.split("-")
         table = self.stream_name if len(parts) == 1 else parts[-1]
         return self.conform_name(table, "table")
@@ -91,17 +104,14 @@ class SQLSink(BatchSink, t.Generic[_C]):
         Returns:
             The target schema name.
         """
-        # Look for a default_target_scheme in the configuration file
-        default_target_schema: str = self.config.get("default_target_schema", None)
-        parts = self.stream_name.split("-")
-
         # 1) When default_target_scheme is in the configuration use it
         # 2) if the streams are in <schema>-<table> format use the
         #    stream <schema>
         # 3) Return None if you don't find anything
-        if default_target_schema:
-            return default_target_schema
+        if self.default_target_schema:
+            return self.default_target_schema
 
+        parts = self.stream_name.split("-")
         return self.conform_name(parts[-2], "schema") if len(parts) in {2, 3} else None
 
     @property


Test patch:
diff --git a/tests/core/sinks/test_sql_sink.py b/tests/core/sinks/test_sql_sink.py
index d1c921ec..7eaf8f93 100644
--- a/tests/core/sinks/test_sql_sink.py
+++ b/tests/core/sinks/test_sql_sink.py
@@ -64,3 +64,75 @@ class TestDuckDBSink:
         assert str(stmt) == (
             'INSERT INTO foo (id, col_ts, "table") VALUES (:id, :col_ts, :table)'
         )
+
+    @pytest.mark.parametrize(
+        "stream_name, default_target_schema, expected_table_name, expected_schema_name",
+        [
+            pytest.param(
+                "foo",  # stream_name
+                None,  # default_target_schema
+                "foo",  # expected_table_name
+                None,  # expected_schema_name
+                id="no-default-schema",
+            ),
+            pytest.param(
+                "foo-bar",  # stream_name
+                None,  # default_target_schema
+                "bar",  # expected_table_name
+                "foo",  # expected_schema_name
+                id="no-default-schema-2-part",
+            ),
+            pytest.param(
+                "foo-bar-baz",  # stream_name
+                None,  # default_target_schema
+                "baz",  # expected_table_name
+                "bar",  # expected_schema_name
+                id="no-default-schema-3-part",
+            ),
+            pytest.param(
+                "foo",  # stream_name
+                "test",  # default_target_schema
+                "foo",  # expected_table_name
+                "test",  # expected_schema_name
+                id="default-schema",
+            ),
+            pytest.param(
+                "foo-bar",  # stream_name
+                "test",  # default_target_schema
+                "foo-bar",  # expected_table_name
+                "test",  # expected_schema_name
+                id="default-schema-2-part",
+            ),
+            pytest.param(
+                "foo-bar-baz",  # stream_name
+                "test",  # default_target_schema
+                "foo-bar-baz",  # expected_table_name
+                "test",  # expected_schema_name
+                id="default-schema-3-part",
+            ),
+        ],
+    )
+    def test_table_name(
+        self,
+        schema: dict,
+        stream_name: str,
+        default_target_schema: str | None,
+        expected_table_name: str,
+        expected_schema_name: str,
+    ):
+        target = DuckDBTarget(
+            config={
+                "sqlalchemy_url": "duckdb:///",
+                "default_target_schema": default_target_schema,
+            },
+        )
+
+        sink = DuckDBSink(
+            target,
+            stream_name=stream_name,
+            schema=schema,
+            key_properties=["id"],
+        )
+
+        assert sink.table_name == expected_table_name
+        assert sink.schema_name == expected_schema_name