Repository: meltano/sdk
Pull request #3165 (2025-07-15 20:19:19)

fix: Honor order of primary key columns when creating an empty table
## Summary by Sourcery

Fix the SQLConnector create_empty_table method to preserve the specified order of primary key columns, ignore non‐existent keys, and skip constraints when no keys exist, and add comprehensive tests to validate these behaviors

Bug Fixes:
- Honor the order of primary key columns when defining an empty table
- Ignore missing primary key columns rather than raising errors
- Omit primary key constraint when no specified keys exist in the schema

Enhancements:
- Refactor create_empty_table to use a PrimaryKeyConstraint for ordered key enforcement instead of inline column flags

Tests:
- Add parameterized tests for key ordering, partial keys, absent keys, and schema order independence in create_empty_table
- Add assertion for cursor presence in DuckDB connector tests


Patch:
diff --git a/singer_sdk/connectors/sql.py b/singer_sdk/connectors/sql.py
index f1130593..c9a86d9c 100644
--- a/singer_sdk/connectors/sql.py
+++ b/singer_sdk/connectors/sql.py
@@ -1314,16 +1314,21 @@ class SQLConnector:  # noqa: PLR0904
             msg = f"Schema for '{full_table_name}' does not define properties: {schema}"
             raise RuntimeError(msg) from e
         for property_name, property_jsonschema in properties.items():
-            is_primary_key = property_name in primary_keys
             columns.append(
                 sa.Column(
                     property_name,
                     self.to_sql_type(property_jsonschema),
-                    primary_key=is_primary_key,
                 ),
             )
 
-        _ = sa.Table(table_name, meta, *columns)
+        table_args = []
+        if primary_keys:
+            # Only include primary keys that exist in the schema properties
+            existing_pk_columns = [col for col in primary_keys if col in properties]
+            if existing_pk_columns:
+                table_args.append(sa.PrimaryKeyConstraint(*existing_pk_columns))
+
+        _ = sa.Table(table_name, meta, *columns, *table_args)
         meta.create_all(self._engine)
 
     def _create_empty_column(


Test patch:
diff --git a/tests/core/test_connector_sql.py b/tests/core/test_connector_sql.py
index 478c7f72..59d426b6 100644
--- a/tests/core/test_connector_sql.py
+++ b/tests/core/test_connector_sql.py
@@ -303,6 +303,165 @@ class TestConnectorSQL:  # noqa: PLR0904
                 (2, {"x": Decimal("2.0"), "y": [1, 2, 3]}),
             ]
 
+    @pytest.mark.parametrize(
+        "primary_keys",
+        [
+            pytest.param(["col_b", "col_d", "col_a"], id="mixed-order"),
+            pytest.param(["col_d", "col_c", "col_b", "col_a"], id="reverse-order"),
+            pytest.param(["col_a", "col_b", "col_c", "col_d"], id="alphabetical-order"),
+        ],
+    )
+    def test_create_empty_table_primary_key_order(
+        self,
+        connector: SQLConnector,
+        primary_keys: list[str],
+    ):
+        """Test that primary key columns maintain their specified order."""
+        schema = {
+            "type": "object",
+            "properties": {
+                "col_a": {"type": "string"},
+                "col_b": {"type": "integer"},
+                "col_c": {"type": "string"},
+                "col_d": {"type": "integer"},
+            },
+        }
+
+        table_name = "test_pk_order"
+
+        connector.create_empty_table(
+            full_table_name=table_name,
+            schema=schema,
+            primary_keys=primary_keys,
+        )
+
+        inspector = sa.inspect(connector._engine)
+        pk_constraint = inspector.get_pk_constraint(table_name)
+
+        assert pk_constraint["constrained_columns"] == primary_keys
+
+        with connector._engine.connect() as conn, conn.begin():
+            conn.execute(sa.text(f"DROP TABLE {table_name}"))
+
+    def test_create_empty_table_partial_primary_keys(self, connector: SQLConnector):
+        """Test that only existing columns are included in primary key constraint."""
+        schema = {
+            "type": "object",
+            "properties": {
+                "existing_col_1": {"type": "string"},
+                "existing_col_2": {"type": "integer"},
+                "existing_col_3": {"type": "string"},
+            },
+        }
+
+        # Primary keys include some columns not in the schema
+        primary_keys = [
+            "missing_col",
+            "existing_col_2",
+            "existing_col_1",
+            "another_missing",
+        ]
+        table_name = "test_partial_pk"
+
+        # Should not raise an error
+        connector.create_empty_table(
+            full_table_name=table_name,
+            schema=schema,
+            primary_keys=primary_keys,
+        )
+
+        # Inspect the created table
+        inspector = sa.inspect(connector._engine)
+        pk_constraint = inspector.get_pk_constraint(table_name)
+
+        # Should only include existing columns in the order they were specified
+        expected_pk_columns = ["existing_col_2", "existing_col_1"]
+        assert pk_constraint["constrained_columns"] == expected_pk_columns
+
+        # Clean up
+        with connector._engine.connect() as conn, conn.begin():
+            conn.execute(sa.text(f"DROP TABLE {table_name}"))
+
+    def test_create_empty_table_no_primary_keys_in_schema(
+        self,
+        connector: SQLConnector,
+    ):
+        """Test behavior when no primary key columns exist in schema."""
+        schema = {
+            "type": "object",
+            "properties": {
+                "col_a": {"type": "string"},
+                "col_b": {"type": "integer"},
+            },
+        }
+
+        # Primary keys reference columns not in schema
+        primary_keys = ["missing_col_1", "missing_col_2"]
+        table_name = "test_no_pk_in_schema"
+
+        # Should not raise an error
+        connector.create_empty_table(
+            full_table_name=table_name,
+            schema=schema,
+            primary_keys=primary_keys,
+        )
+
+        # Inspect the created table
+        inspector = sa.inspect(connector._engine)
+        pk_constraint = inspector.get_pk_constraint(table_name)
+
+        # Should have no primary key constraint
+        assert pk_constraint["constrained_columns"] == []
+
+        # Clean up
+        with connector._engine.connect() as conn, conn.begin():
+            conn.execute(sa.text(f"DROP TABLE {table_name}"))
+
+    def test_create_empty_table_schema_property_order_independence(
+        self,
+        connector: SQLConnector,
+    ):
+        """Test that primary key order is independent of schema property order."""
+        # Same properties in different orders
+        schema1 = {
+            "type": "object",
+            "properties": {
+                "col_a": {"type": "string"},
+                "col_b": {"type": "integer"},
+                "col_c": {"type": "string"},
+            },
+        }
+
+        schema2 = {
+            "type": "object",
+            "properties": {
+                "col_c": {"type": "string"},
+                "col_a": {"type": "string"},
+                "col_b": {"type": "integer"},
+            },
+        }
+
+        primary_keys = ["col_b", "col_c", "col_a"]
+
+        # Create tables with different schema property orders
+        table1 = "test_schema_order_1"
+        table2 = "test_schema_order_2"
+
+        connector.create_empty_table(table1, schema1, primary_keys)
+        connector.create_empty_table(table2, schema2, primary_keys)
+
+        # Both should have identical primary key order
+        inspector = sa.inspect(connector._engine)
+        pk1 = inspector.get_pk_constraint(table1)["constrained_columns"]
+        pk2 = inspector.get_pk_constraint(table2)["constrained_columns"]
+
+        assert pk1 == pk2 == primary_keys
+
+        # Clean up
+        with connector._engine.connect() as conn, conn.begin():
+            conn.execute(sa.text(f"DROP TABLE {table1}"))
+            conn.execute(sa.text(f"DROP TABLE {table2}"))
+
 
 class TestDuckDBConnector:
     @pytest.fixture
@@ -349,6 +508,7 @@ class TestDuckDBConnector:
         with engine.connect() as conn:
             result = conn.execute(sa.text("SELECT * FROM test_table"))
             assert result.keys() == ["id", "name"]
+            assert result.cursor is not None
             assert result.cursor.description[1][1] == "STRING"
 
     @pytest.mark.parametrize(