fix(sql): Add fallback to source_defined_primary_key in CatalogProvider

[devin-ai-integration]

Fix primary key fallback in CatalogProvider for SQL destinations

Summary

This PR fixes a critical bug in the Airbyte Python CDK where the CatalogProvider.get_primary_keys() method ignores source-defined primary keys when configured primary keys are empty or None. This affects all destinations using the CDK's SQL processor base classes, not just MotherDuck.

Root Cause: The CDK's CatalogProvider.get_primary_keys() method only uses ConfiguredAirbyteStream.primary_key and returns an empty list when it's falsy, without falling back to stream.source_defined_primary_key as specified in the Airbyte protocol.

Fix: Added fallback logic to use stream.source_defined_primary_key when primary_key is empty/None, with comprehensive unit test coverage.

Review & Testing Checklist for Human

⚠️ HIGH RISK CHANGE - This modifies core CDK logic affecting all SQL destinations

• Verify protocol compliance: Confirm that the Airbyte protocol specification actually requires fallback to source_defined_primary_key when primary_key is empty/None
• End-to-end testing: Test with actual destination connectors (especially MotherDuck, DuckDB, PostgreSQL) to verify the fix works correctly
• Check for existing workarounds: Review other SQL destinations to see if any have implemented workarounds for this bug that might break with this change
• Regression testing: Verify that destinations with explicitly set primary_key still work correctly (configured primary key should take precedence)
• Performance impact: Confirm the additional configured_stream.stream.source_defined_primary_key lookup doesn't cause performance issues

Recommended Test Plan

1. Create test cases with catalogs having empty primary_key but non-empty source_defined_primary_key
2. Test sync operations with deduplication modes that rely on primary keys
3. Verify SQL generation includes correct primary key columns from source-defined keys
4. Test with multiple destinations to ensure no regressions

---

Diagram

graph TD
    A[airbyte_cdk/sql/shared/catalog_providers.py]:::major-edit
    B[unit_tests/sql/shared/test_catalog_providers.py]:::major-edit
    C[ConfiguredAirbyteStream.primary_key]:::context
    D[AirbyteStream.source_defined_primary_key]:::context
    E[SQL Destinations]:::context
    F[MotherDuck Destination]:::context
    G[DuckDB Destination]:::context
    H[Other SQL Destinations]:::context

    A --> C
    A --> D
    A --> E
    E --> F
    E --> G  
    E --> H
    B --> A

    subgraph Legend
        L1[Major Edit]:::major-edit
        L2[Minor Edit]:::minor-edit
        L3[Context/No Edit]:::context
    end

    classDef major-edit fill:#90EE90
    classDef minor-edit fill:#87CEEB
    classDef context fill:#FFFFFF

Loading

Notes

• Documentation References: According to the Airbyte protocol, primary_key in ConfiguredAirbyteStream overrides source-defined primary keys when set, and source_defined_primary_key provides the default from the source
• CI Test Failure: The MotherDuck destination test failure in CI is unrelated to this change (destinations don't support 'discover' operation)
• Backward Compatibility: This change should be backward compatible as it only adds fallback behavior when no configured primary key exists
• Test Coverage: Added 7 comprehensive unit tests covering all scenarios including configured precedence, fallback behavior, composite keys, and case normalization

[devin-ai-integration]

Original prompt from AJ Steers:

Received message in Slack channel #ask-devin-ai:

@Devin - we've identified a bug in the MotherDuck destination. Specifically, it looks like that implementation does not consider overridden primary key for streams described in the configured catalog. Instead, it seems to always use the source defined primary key. Can you look into this and (1) confirm, and (2) if confirmed propose a fix?

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment and CI monitoring

[github-actions]

PyTest Results (Fast)

3 683 tests  +7   3 672 ✅ +7   6m 12s ⏱️ -4s
    1 suites ±0      11 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit 821848b. ± Comparison against base commit 5824a5e.

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

3 686 tests  +7   3 675 ✅ +7   17m 57s ⏱️ -3s
    1 suites ±0      11 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit 821848b. ± Comparison against base commit 5824a5e.

♻️ This comment has been updated with latest results.

[aaronsteers]

Looks like our new test suite assumes all Python connectors are sources. Since this is a destination and doesn't support 'discover', the discover test fails.

Nothing we need to solve here in this PR and not related to our change.

E           AssertionError: Expected no errors but got 1: 
E           
E           AirbyteErrorTraceMessage(message='Something went wrong in the connector. See the logs for more details.', internal_message="'DestinationMotherDuck' object has no attribute 'discover'", stack_trace='Traceback (most recent call last):
E             File "/home/runner/.cache/pypoetry/virtualenvs/airbyte-destination-motherduck-BV7hPlm3-py3.11/lib/python3.11/site-packages/airbyte_cdk/test/entrypoint_wrapper.py", line 325, in _run_command
E               for message in source_entrypoint.run(parsed_args):
E             File "/home/runner/.cache/pypoetry/virtualenvs/airbyte-destination-motherduck-BV7hPlm3-py3.11/lib/python3.11/site-packages/airbyte_cdk/entrypoint.py", line 199, in run
E               yield from map(
E             File "/home/runner/.cache/pypoetry/virtualenvs/airbyte-destination-motherduck-BV7hPlm3-py3.11/lib/python3.11/site-packages/airbyte_cdk/entrypoint.py", line 266, in discover
E               catalog = self.source.discover(self.logger, config)
E                         ^^^^^^^^^^^^^^^^^^^^
E           AttributeError: \'DestinationMotherDuck\' object has no attribute \'discover\'
E           ', failure_type=<FailureType.system_error: 'system_error'>, stream_descriptor=None)

/home/runner/.cache/pypoetry/virtualenvs/airbyte-destination-motherduck-BV7hPlm3-py3.11/lib/python3.11/site-packages/airbyte_cdk/test/standard_tests/_job_runner.py:128: AssertionError
______________ TestSuite.test_basic_read['config' Test Scenario] _______________

[aaronsteers]

/bump-version

[aaronsteers]

This would be simpler:

pks = configured_stream.primary_key or configured_stream.stream.source_defined_primary_key or []

You'll have to wrap to make not overflow line length, but this version seems more concise and also more readable.