feat(sql): Prioritize source_defined_primary_key and return None when unset

- Reverse priority order to favor source_defined_primary_key over primary_key
- Return None when neither primary key field is set
- Update all callers to handle None gracefully with 'pks or []' coalescing
- Update unit tests to reflect new behavior and priority order
- Addresses @aaronsteers feedback on primary key fallback logic

Co-Authored-By: AJ Steers <[email protected]>

Author: devin-ai-integration[bot]

airbyte_cdk/sql/shared/catalog_providers.py

@@ -119,18 +119,16 @@ def get_stream_properties(
     def get_primary_keys(
         self,
         stream_name: str,
-    ) -> list[str]:
+    ) -> list[str] | None:
         """Return the primary keys for the given stream.
 
-        We will use `primary_key` if it is set explicitly in the configured catalog,
-        otherwise we will fall back to `source_defined_primary_key`, if set.
+        We return `source_defined_primary_key` if set, or `primary_key` otherwise. If both are set, we assume they should not should differ, since Airbyte data integrity constraints do not permit overruling a source's pre-defined primary keys. If neither is set, we return `None`.
         """
         configured_stream = self.get_configured_stream_info(stream_name)
-        pks = (
-            configured_stream.primary_key
-            or configured_stream.stream.source_defined_primary_key
-            or []
-        )
+        pks = configured_stream.stream.source_defined_primary_key or configured_stream.primary_key
+
+        if not pks:
+            return None
 
         normalized_pks: list[list[str]] = [
             [LowerCaseNormalizer.normalize(c) for c in pk] for pk in pks

airbyte_cdk/sql/shared/sql_processor.py

@@ -667,7 +667,8 @@ def _merge_temp_table_to_final_table(
         nl = "\n"
         columns = {self._quote_identifier(c) for c in self._get_sql_column_definitions(stream_name)}
         pk_columns = {
-            self._quote_identifier(c) for c in self.catalog_provider.get_primary_keys(stream_name)
+            self._quote_identifier(c)
+            for c in (self.catalog_provider.get_primary_keys(stream_name) or [])
         }
         non_pk_columns = columns - pk_columns
         join_clause = f"{nl} AND ".join(f"tmp.{pk_col} = final.{pk_col}" for pk_col in pk_columns)
@@ -724,7 +725,7 @@ def _emulated_merge_temp_table_to_final_table(
         """
         final_table = self._get_table_by_name(final_table_name)
         temp_table = self._get_table_by_name(temp_table_name)
-        pk_columns = self.catalog_provider.get_primary_keys(stream_name)
+        pk_columns = self.catalog_provider.get_primary_keys(stream_name) or []
 
         columns_to_update: set[str] = self._get_sql_column_definitions(
             stream_name=stream_name

unit_tests/sql/shared/test_catalog_providers.py

@@ -12,11 +12,23 @@ class TestCatalogProvider:
     @pytest.mark.parametrize(
         "configured_primary_key,source_defined_primary_key,expected_result,test_description",
         [
-            (["configured_id"], ["source_id"], ["configured_id"], "uses configured when both set"),
-            ([], ["source_id"], ["source_id"], "falls back to source when configured empty"),
-            (None, ["source_id"], ["source_id"], "falls back to source when configured None"),
-            ([], [], [], "returns empty when both empty"),
-            (None, None, [], "returns empty when both None"),
+            (["configured_id"], ["source_id"], ["source_id"], "prioritizes source when both set"),
+            ([], ["source_id"], ["source_id"], "uses source when configured empty"),
+            (None, ["source_id"], ["source_id"], "uses source when configured None"),
+            (
+                ["configured_id"],
+                [],
+                ["configured_id"],
+                "falls back to configured when source empty",
+            ),
+            (
+                ["configured_id"],
+                None,
+                ["configured_id"],
+                "falls back to configured when source None",
+            ),
+            ([], [], None, "returns None when both empty"),
+            (None, None, None, "returns None when both None"),
             ([], ["id1", "id2"], ["id1", "id2"], "handles composite keys from source"),
         ],
     )