Repository: meltano/sdk (python)

Issue:
__filter__ of stream map breaks tap-postgres
### Singer SDK Version

2.19.1

### Is this a regression?

- [ ] Yes

### Python Version

3.10

### Bug scope

Mapping (stream maps, flattening, etc.)

### Operating System

MacOS

### Description

my setup looks like this:
```yaml
  extractors:
  - name: tap-postgres
    pip_url: git+https://github.com/MeltanoLabs/tap-postgres.git@5320d1fa47442466785f6821af3ebfbe3ebff327
    config:
      stream_maps:
        public-time_series_meta:
          __filter__: ts_id in config['allowed_id']
      stream_map_config:
        allowed_id:
          - 2364
          - 2363
    select:
      - public-time_series_meta.*
```

The catalog from the tap contains the following:
```json
{
      "tap_stream_id": "public-time_series_meta",
      "table_name": "time_series_meta",
      "replication_method": "",
      "key_properties": [
        "ts_id"
      ],
      "schema": {
        "properties": {
          "ts_id": {
            "type": [
              "integer"
            ]
          },
          "name": {
            "type": [
              "string",
              "null"
            ]
          },
          "region": {
            "type": [
              "string",
              "null"
            ]
          },
          "source": {
            "type": [
              "string",
              "null"
            ]
          },
          "refers_to": {
            "type": [
              "integer",
              "null"
            ]
          },
          "default_source": {
            "type": [
              "boolean",
              "null"
            ]
          },
          "frequency": {
            "type": [
              "string",
              "null"
            ]
          },
          "kwargs": {
            "type": [
              "string",
              "null"
            ]
          },
          "description": {
            "type": [
              "string",
              "null"
            ]
          },
          "unit": {
            "type": [
              "string",
              "null"
            ]
          },
          "method": {
            "type": [
              "string",
              "null"
            ]
          },
          "load_interval": {
            "type": [
              "string",
              "null"
            ]
          },
          "access": {
            "type": [
              "string",
              "null"
            ]
          },
          "orig_timezone": {
            "type": [
              "string",
              "null"
            ]
          }
        },
        "type": "object",
        "required": [
          "ts_id"
        ]
      }
```

but when I run `meltano invoke tap postgres` I get:
`singer_sdk.exceptions.StreamMapConfigError: Invalid key properties for 'public-time_series_meta': [ts_id]. Property 'ts_id' was not detected in schema.`

Running the same config in meltano-map-transformer works fine. 

### Code

```python
-
```


Pull request:
fix(taps): Avoid setting up mapper in discovery mode
Closes #1821


<!-- readthedocs-preview meltano-sdk start -->
----
:books: Documentation preview :books:: https://meltano-sdk--1835.org.readthedocs.build/en/1835/

<!-- readthedocs-preview meltano-sdk end -->

Patch:
diff --git a/singer_sdk/plugin_base.py b/singer_sdk/plugin_base.py
index 1edc34c7..d81e8f7c 100644
--- a/singer_sdk/plugin_base.py
+++ b/singer_sdk/plugin_base.py
@@ -31,11 +31,9 @@ from singer_sdk.helpers.capabilities import (
     CapabilitiesEnum,
     PluginCapabilities,
 )
+from singer_sdk.mapper import PluginMapper
 from singer_sdk.typing import extend_validator_with_defaults
 
-if t.TYPE_CHECKING:
-    from singer_sdk.mapper import PluginMapper
-
 SDK_PACKAGE_NAME = "singer_sdk"
 CHECK_SUPPORTED_PYTHON_VERSIONS = (
     # unsupported versions
@@ -65,6 +63,14 @@ CHECK_SUPPORTED_PYTHON_VERSIONS = (
 JSONSchemaValidator = extend_validator_with_defaults(Draft7Validator)
 
 
+class MapperNotInitialized(Exception):
+    """Raised when the mapper is not initialized."""
+
+    def __init__(self) -> None:
+        """Initialize the exception."""
+        super().__init__("Mapper not initialized. Please call setup_mapper() first.")
+
+
 class PluginBase(metaclass=abc.ABCMeta):
     """Abstract base class for taps."""
 
@@ -144,11 +150,41 @@ class PluginBase(metaclass=abc.ABCMeta):
                 config_dict[k] = SecretString(v)
         self._config = config_dict
         self._validate_config(raise_errors=validate_config)
-        self.mapper: PluginMapper
+        self._mapper: PluginMapper | None = None
 
         metrics._setup_logging(self.config)
         self.metrics_logger = metrics.get_metrics_logger()
 
+    def setup_mapper(self) -> None:
+        """Initialize the plugin mapper for this tap."""
+        self._mapper = PluginMapper(
+            plugin_config=dict(self.config),
+            logger=self.logger,
+        )
+
+    @property
+    def mapper(self) -> PluginMapper:
+        """Plugin mapper for this tap.
+
+        Returns:
+            A PluginMapper object.
+
+        Raises:
+            MapperNotInitialized: If the mapper has not been initialized.
+        """
+        if self._mapper is None:
+            raise MapperNotInitialized
+        return self._mapper
+
+    @mapper.setter
+    def mapper(self, mapper: PluginMapper) -> None:
+        """Set the plugin mapper for this plugin.
+
+        Args:
+            mapper: A PluginMapper object.
+        """
+        self._mapper = mapper
+
     @classproperty
     def capabilities(self) -> list[CapabilitiesEnum]:
         """Get capabilities.
diff --git a/singer_sdk/tap_base.py b/singer_sdk/tap_base.py
index 150f0ee8..28756811 100644
--- a/singer_sdk/tap_base.py
+++ b/singer_sdk/tap_base.py
@@ -25,12 +25,12 @@ from singer_sdk.helpers.capabilities import (
     PluginCapabilities,
     TapCapabilities,
 )
-from singer_sdk.mapper import PluginMapper
 from singer_sdk.plugin_base import PluginBase
 
 if t.TYPE_CHECKING:
     from pathlib import PurePath
 
+    from singer_sdk.mapper import PluginMapper
     from singer_sdk.streams import SQLStream, Stream
 
 STREAM_MAPS_CONFIG = "stream_maps"
@@ -61,6 +61,7 @@ class Tap(PluginBase, metaclass=abc.ABCMeta):
         state: PurePath | str | dict | None = None,
         parse_env_config: bool = False,
         validate_config: bool = True,
+        setup_mapper: bool = True,
     ) -> None:
         """Initialize the tap.
 
@@ -73,6 +74,7 @@ class Tap(PluginBase, metaclass=abc.ABCMeta):
             parse_env_config: Whether to look for configuration values in environment
                 variables.
             validate_config: True to require validation of config settings.
+            setup_mapper: True to initialize the plugin mapper.
         """
         super().__init__(
             config=config,
@@ -94,14 +96,10 @@ class Tap(PluginBase, metaclass=abc.ABCMeta):
         elif catalog is not None:
             self._input_catalog = Catalog.from_dict(read_json_file(catalog))
 
-        # Initialize mapper
-        self.mapper: PluginMapper
-        self.mapper = PluginMapper(
-            plugin_config=dict(self.config),
-            logger=self.logger,
-        )
+        self._mapper: PluginMapper | None = None
 
-        self.mapper.register_raw_streams_from_catalog(self.catalog)
+        if setup_mapper:
+            self.setup_mapper()
 
         # Process state
         state_dict: dict = {}
@@ -168,6 +166,11 @@ class Tap(PluginBase, metaclass=abc.ABCMeta):
 
         return self._catalog
 
+    def setup_mapper(self) -> None:
+        """Initialize the plugin mapper for this tap."""
+        super().setup_mapper()
+        self.mapper.register_raw_streams_from_catalog(self.catalog)
+
     @classproperty
     def capabilities(self) -> list[CapabilitiesEnum]:
         """Get tap capabilities.
@@ -520,6 +523,7 @@ class Tap(PluginBase, metaclass=abc.ABCMeta):
             config=config_files,  # type: ignore[arg-type]
             parse_env_config=parse_env_config,
             validate_config=False,
+            setup_mapper=False,
         )
         tap.run_discovery()
         ctx.exit()
diff --git a/singer_sdk/target_base.py b/singer_sdk/target_base.py
index 4ffc7446..d62bbbfd 100644
--- a/singer_sdk/target_base.py
+++ b/singer_sdk/target_base.py
@@ -23,12 +23,12 @@ from singer_sdk.helpers.capabilities import (
     TargetCapabilities,
 )
 from singer_sdk.io_base import SingerMessageType, SingerReader
-from singer_sdk.mapper import PluginMapper
 from singer_sdk.plugin_base import PluginBase
 
 if t.TYPE_CHECKING:
     from pathlib import PurePath
 
+    from singer_sdk.mapper import PluginMapper
     from singer_sdk.sinks import Sink
 
 _MAX_PARALLELISM = 8
@@ -56,6 +56,7 @@ class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
         config: dict | PurePath | str | list[PurePath | str] | None = None,
         parse_env_config: bool = False,
         validate_config: bool = True,
+        setup_mapper: bool = True,
     ) -> None:
         """Initialize the target.
 
@@ -66,6 +67,7 @@ class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
             parse_env_config: Whether to look for configuration values in environment
                 variables.
             validate_config: True to require validation of config settings.
+            setup_mapper: True to setup the mapper. Set to False if you want to
         """
         super().__init__(
             config=config,
@@ -82,12 +84,10 @@ class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
         # Approximated for max record age enforcement
         self._last_full_drain_at: float = time.time()
 
-        # Initialize mapper
-        self.mapper: PluginMapper
-        self.mapper = PluginMapper(
-            plugin_config=dict(self.config),
-            logger=self.logger,
-        )
+        self._mapper: PluginMapper | None = None
+
+        if setup_mapper:
+            self.setup_mapper()
 
     @classproperty
     def capabilities(self) -> list[CapabilitiesEnum]:


Test patch:
diff --git a/tests/core/test_plugin_base.py b/tests/core/test_plugin_base.py
index b8dfbb69..7d1021c1 100644
--- a/tests/core/test_plugin_base.py
+++ b/tests/core/test_plugin_base.py
@@ -3,7 +3,9 @@ from __future__ import annotations
 import os
 from unittest import mock
 
-from singer_sdk.plugin_base import PluginBase
+import pytest
+
+from singer_sdk.plugin_base import MapperNotInitialized, PluginBase
 from singer_sdk.typing import IntegerType, PropertiesList, Property, StringType
 
 
@@ -41,3 +43,13 @@ def test_get_env_var_config():
     assert "PROP2" not in env_config
     assert "prop3" not in no_env_config
     assert "PROP3" not in env_config
+
+
+def test_mapper_not_initialized():
+    """Test that the mapper is not initialized before the plugin is started."""
+    plugin = PluginTest(
+        parse_env_config=False,
+        validate_config=False,
+    )
+    with pytest.raises(MapperNotInitialized):
+        _ = plugin.mapper