Repository: meltano/sdk (python)

Issue:
bug: `table-key-properties` in `metadata` extra does not populate schema `key_properties` for `tap-bigquery`
### Singer SDK Version

0.39.1

### Is this a regression?

- [ ] Yes

### Python Version

3.9

### Bug scope

Taps (catalog, state, etc.)

### Operating System

Ubuntu 22.04 LTS

### Description

I've been trying to use `table-key-properties` to configure key properties for a BigQuery stream:

```yml
plugins:
  extractors:
  - name: tap-bigquery
    variant: meltanolabs
    pip_url: git+https://github.com/Matatika/tap-bigquery.git
    metadata:
      '*-events_*':
        table-key-properties:
        - event_timestamp
        - event_name
        - event_bundle_sequence_id
```

In this case, `key_properties` is empty so  when loading to a database target no primary keys are configured:

<details>
<summary>Generated catalog</summary>

```json
{
  "streams": [
    {
      "tap_stream_id": "analytics_376142012-events_20240825",
      "table_name": "events_20240825",
      "replication_method": "",
      "key_properties": [],
      "schema": ...,
      "is_view": false,
      "stream": "analytics_376142012-events_20240825",
      "metadata": [
        ...,
        {
          "breadcrumb": [],
          "metadata": {
            "inclusion": "available",
            "table-key-properties": [
              "event_timestamp",
              "event_name",
              "event_bundle_sequence_id"
            ],
            "forced-replication-method": "",
            "schema-name": "analytics_376142012",
            "selected": true
          }
        },
        ...
      ],
      "selected": true,
      "table_key_properties": [
        "event_timestamp",
        "event_name",
        "event_bundle_sequence_id"
      ]
    }
  ]
}
```

</details>

Primary keys are configured when using `key-properties`/`key_properties` (not documented [here](https://hub.meltano.com/singer/spec/#metadata)):

```yml
    metadata:
      '*-events_*':
        key-properties:
        - event_timestamp
        - event_name
        - event_bundle_sequence_id
```

<details>
<summary>Generated catalog</summary>

```json
{
  "streams": [
    {
      "tap_stream_id": "analytics_376142012-events_20240825",
      "table_name": "events_20240825",
      "replication_method": "",
      "key_properties": [
        "event_timestamp",
        "event_name",
        "event_bundle_sequence_id"
      ],
      "schema": ...,
      "is_view": false,
      "stream": "analytics_376142012-events_20240825",
      "metadata": [
        ...,
        {
          "breadcrumb": [],
          "metadata": {
            "inclusion": "available",
            "table-key-properties": [],
            "forced-replication-method": "",
            "schema-name": "analytics_376142012",
            "selected": true,
            "key-properties": [
              "event_timestamp",
              "event_name",
              "event_bundle_sequence_id"
            ]
          }
        },
        ...
      ],
      "selected": true
    }
  ]
}
```

</details>

```yml
    metadata:
      '*-events_*':
        key_properties:
        - event_timestamp
        - event_name
        - event_bundle_sequence_id
```

<details>
<summary>Generated catalog</summary>

```json
{
  "streams": [
    {
      "tap_stream_id": "analytics_376142012-events_20240825",
      "table_name": "events_20240825",
      "replication_method": "",
      "key_properties": [
        "event_timestamp",
        "event_name",
        "event_bundle_sequence_id"
      ],
      "schema": ...,
      "is_view": false,
      "stream": "analytics_376142012-events_20240825",
      "metadata": [
        ...,
        {
          "breadcrumb": [],
          "metadata": {
            "inclusion": "available",
            "table-key-properties": [],
            "forced-replication-method": "",
            "schema-name": "analytics_376142012",
            "selected": true,
            "key_properties": [
              "event_timestamp",
              "event_name",
              "event_bundle_sequence_id"
            ]
          }
        },
        ...
      ],
      "selected": true
    }
  ]
}
```

</details>

Why does this work? What is the correct way to do this? I assume `table-key-properties` is the intended approach, so I don't know if I'm going to run into some undefined behaviour down the line.

---

Related:
- https://discuss.meltano.com/t/16277188/hi-all-when-trying-to-specify-custom-pks-in-my-meltano-yml-b
- https://discuss.meltano.com/t/16276147/so-i-have-a-problem-i-need-to-replicate-table-from-postgres-
- https://discuss.meltano.com/t/16272691/need-advice-i-m-trying-to-develop-tap-bigquery-existing-from

### Code

_No response_

### Link to Slack/Linen

https://meltano.slack.com/archives/C069CQNHDNF/p1725976166433339

Pull request description:
fix: Respect standard Singer stream metadata `table-key-properties`, `replication-key` and `forced-replication-method`
- https://hub.meltano.com/singer/spec/#metadata
- https://github.com/singer-io/getting-started/blob/master/docs/DISCOVERY_MODE.md#metadata

**TODO**:

- [x] Tests
~- [ ] Documentation~

Related:

- Closes #2660 

<!-- readthedocs-preview meltano-sdk start -->
----
📚 Documentation preview 📚: https://meltano-sdk--2770.org.readthedocs.build/en/2770/

<!-- readthedocs-preview meltano-sdk end -->

Patch:
diff --git a/docs/implementation/catalog_metadata.md b/docs/implementation/catalog_metadata.md
index b255a553..090a3896 100644
--- a/docs/implementation/catalog_metadata.md
+++ b/docs/implementation/catalog_metadata.md
@@ -1,6 +1,6 @@
 # Catalog Metadata
 
-The SDK automatically generates catalog metadata during catalog discovery. Selection rules overrided by a user will be respected.
+The SDK automatically generates catalog metadata during catalog discovery. Selection rules overridden by a user will be respected.
 
 Primary key properties may not be deselected, as these are required for `key_properties` to be declared in stream messages.
 
diff --git a/singer_sdk/_singerlib/catalog.py b/singer_sdk/_singerlib/catalog.py
index 986272e3..c1281e09 100644
--- a/singer_sdk/_singerlib/catalog.py
+++ b/singer_sdk/_singerlib/catalog.py
@@ -87,6 +87,7 @@ class StreamMetadata(Metadata):
     """Stream metadata."""
 
     table_key_properties: t.Sequence[str] | None = None
+    replication_key: str | None = None
     forced_replication_method: str | None = None
     valid_replication_keys: list[str] | None = None
     schema_name: str | None = None
diff --git a/singer_sdk/streams/core.py b/singer_sdk/streams/core.py
index 009726b7..fc99b82b 100644
--- a/singer_sdk/streams/core.py
+++ b/singer_sdk/streams/core.py
@@ -56,6 +56,7 @@ from singer_sdk.mapper import RemoveRecordTransform, SameRecordTransform, Stream
 if t.TYPE_CHECKING:
     import logging
 
+    from singer_sdk._singerlib.catalog import StreamMetadata
     from singer_sdk.helpers import types
     from singer_sdk.helpers._compat import Traversable
     from singer_sdk.tap_base import Tap
@@ -1277,10 +1278,26 @@ class Stream(metaclass=abc.ABCMeta):  # noqa: PLR0904
 
         catalog_entry = catalog.get_stream(self.name)
         if catalog_entry:
-            self.primary_keys = catalog_entry.key_properties
-            self.replication_key = catalog_entry.replication_key
-            if catalog_entry.replication_method:
-                self.forced_replication_method = catalog_entry.replication_method
+            stream_metadata: StreamMetadata | None
+            if stream_metadata := catalog_entry.metadata.get(()):  # type: ignore[assignment]
+                table_key_properties = stream_metadata.table_key_properties
+                table_replication_key = stream_metadata.replication_key
+                table_replication_method = stream_metadata.forced_replication_method
+            else:
+                table_key_properties = None
+                table_replication_key = None
+                table_replication_method = None
+
+            self.primary_keys = catalog_entry.key_properties or table_key_properties
+            self.replication_key = (
+                catalog_entry.replication_key or table_replication_key
+            )
+
+            replication_method = (
+                catalog_entry.replication_method or table_replication_method
+            )
+            if replication_method:
+                self.forced_replication_method = replication_method
 
     def _get_state_partition_context(
         self,


Test patch:
diff --git a/tests/core/test_streams.py b/tests/core/test_streams.py
index 038f8765..6dabf1f2 100644
--- a/tests/core/test_streams.py
+++ b/tests/core/test_streams.py
@@ -106,6 +106,43 @@ def test_stream_apply_catalog(stream: Stream):
     assert stream.forced_replication_method == REPLICATION_FULL_TABLE
 
 
+def test_stream_apply_catalog__singer_standard(stream: Stream):
+    """Applying a catalog to a stream should overwrite fields."""
+    assert stream.primary_keys == []
+    assert stream.replication_key == "updatedAt"
+    assert stream.replication_method == REPLICATION_INCREMENTAL
+    assert stream.forced_replication_method is None
+
+    stream.apply_catalog(
+        catalog=Catalog.from_dict(
+            {
+                "streams": [
+                    {
+                        "tap_stream_id": stream.name,
+                        "stream": stream.name,
+                        "schema": stream.schema,
+                        "metadata": [
+                            {
+                                "breadcrumb": [],
+                                "metadata": {
+                                    "table-key-properties": ["id"],
+                                    "replication-key": "newReplicationKey",
+                                    "forced-replication-method": REPLICATION_FULL_TABLE,
+                                },
+                            },
+                        ],
+                    },
+                ],
+            },
+        ),
+    )
+
+    assert stream.primary_keys == ["id"]
+    assert stream.replication_key == "newReplicationKey"
+    assert stream.replication_method == REPLICATION_FULL_TABLE
+    assert stream.forced_replication_method == REPLICATION_FULL_TABLE
+
+
 @pytest.mark.parametrize(
     "stream_name,forced_replication_method,bookmark_value,expected_starting_value",
     [