fix(targets): reuse SQLAlchemy inspector reflection cache

AGENTS.md

@@ -0,0 +1,124 @@
+# AGENTS.md
+
+This is the Meltano Singer SDK, a Python framework for building Singer taps (data extractors) and targets (data loaders).
+
+## Development Setup
+
+```bash
+uv sync --all-groups --all-extras --all-packages  # Full development environment
+```
+
+## Common Commands
+
+### Testing
+
+```bash
+nox -s tests              # Run core tests
+nox -s test-contrib       # Run contrib (experimental) tests
+nox -s test-packages      # Run package integration tests
+nox -t typing             # Type checking with mypy
+```
+
+### Linting & Formatting
+
+```bash
+pre-commit run --all      # Run all pre-commit hooks
+ruff check --fix          # Lint and auto-fix
+ruff format               # Format code
+```
+
+### Documentation
+
+```bash
+nox -s docs               # Build Sphinx documentation
+```
+
+## Code Style
+
+- Required: `from __future__ import annotations` at top of every file
+- Use `typing as t` for type imports (abbreviated import per ruff config)
+- Google-style docstrings with full parameter documentation
+- Line length: 88 characters
+- Ruff handles all formatting and linting
+
+## Exceptions
+
+All SDK exceptions live in `singer_sdk/exceptions.py` and inherit from `SingerSDKError`.
+Choose the right base class by recovery strategy:
+
+| Situation | Raise |
+|---|---|
+| HTTP/API error, must abort sync | `FatalAPIError` |
+| HTTP/API error, safe to retry | `RetriableAPIError` |
+| HTTP/API error, expected / skip silently | `IgnorableAPIError` |
+| Config value is wrong | `ConfigValidationError` |
+| Discovery / catalog problem | `DiscoveryError` (or a subclass) |
+| Stream map config or expression fails | `MappingError` (or a subclass) |
+
+When adding a new exception:
+
+1. Place it in `singer_sdk/exceptions.py` — never define public exceptions in other files.
+1. Inherit from the appropriate intermediate base (`FatalSyncError`, `RetriableSyncError`,
+   `IgnorableSyncError`, `DataError`, `ConfigurationError`, `MappingError`, etc.) rather
+   than from `Exception` or `SingerSDKError` directly.
+1. Add it to `__all__` in that file.
+1. Add `issubclass` assertions to `tests/core/test_exceptions.py`.
+
+See `docs/implementation/errors/hierarchy.md` for the full hierarchy and
+`docs/implementation/errors/design.md` for the design rationale.
+
+## Deprecation Warnings
+
+Three warning classes in `singer_sdk/helpers/_compat.py`:
+
+| Class | Base | When to use |
+|---|---|---|
+| `SingerSDKDeprecationWarning` | `DeprecationWarning` | Removal version is known |
+| `SingerSDKPendingDeprecationWarning` | `PendingDeprecationWarning` | No committed removal timeline; silenced by default |
+| `SingerSDKPythonEOLWarning` | `FutureWarning` | Python version nearing/past EOL |
+
+Use `deprecated(msg, *, category=SingerSDKDeprecationWarning, stacklevel=1)` as a decorator factory for deprecating classes and functions. `category` is **required** — it is appended to the message automatically. Never embed the version in the message string manually.
+
+```python
+@deprecated(
+    "Use Bar instead. Foo will be removed in <removal_version>.",
+    category=SingerSDKDeprecationWarning,
+)
+class Foo: ...
+```
+
+For `warnings.warn` call sites with no committed removal timeline, use `SingerSDKPendingDeprecationWarning` directly:
+
+```python
+warnings.warn("...", SingerSDKPendingDeprecationWarning, stacklevel=2)
+```
+
+**Deprecation policy**: at least 3 months / 3 feature releases notice, named removal version required. See `docs/release_process.md` and `docs/deprecation.md`.
+
+## Architecture
+
+The SDK uses abstract base classes for its plugin system:
+
+- `Tap` / `Target` - Main plugin entry points
+- `Stream` / `RESTStream` / `GraphQLStream` / `SQLStream` - Data extraction
+- `Sink` / `BatchSink` / `SQLSink` - Data loading
+- `SQLConnector` - Database connections
+
+## Testing
+
+- pytest-based with custom pytest plugin (`singer_testing`)
+- Test markers: `contrib`, `external`, `packages`, `snapshot`
+- Platform markers: `@pytest.mark.darwin`, `@pytest.mark.linux`, `@pytest.mark.windows`
+- Standard test suites: `singer_sdk.testing.get_standard_tap_tests()` / `get_standard_target_tests()`
+
+## Project Structure
+
+- `singer_sdk/` - Main SDK package
+- `tests/` - Test suite (core, contrib, external, packages)
+- `packages/` - Reference implementations for E2E testing
+- `cookiecutter/` - Templates for scaffolding new taps/targets
+- `docs/` - Sphinx documentation
+
+## Python Version
+
+Supports Python 3.10-3.14. Primary development version is 3.14.

singer_sdk/sql/connector.py

@@ -13,6 +13,7 @@
 
 import sqlalchemy as sa
 import sqlalchemy.types
+from sqlalchemy import event
 from sqlalchemy.engine import reflection
 from sqlalchemy.sql import ddl
 
@@ -617,6 +618,7 @@ def __init__(
         self._config = config or {}
         self._sqlalchemy_url: str | None = sqlalchemy_url or None
         self._tables_prepared: dict[str, bool] = {}
+        self._cached_inspector: reflection.Inspector | None = None
 
     @property
     def config(self) -> Mapping[str, t.Any]:
@@ -873,8 +875,46 @@ def _engine(self) -> sa.Engine:
         """
         if not self._cached_engine:
             self._cached_engine = self.create_engine()
+            event.listen(
+                self._cached_engine,
+                "after_cursor_execute",
+                self._clear_reflection_cache_after_ddl,
+            )
         return self._cached_engine
 
+    @property
+    def _inspector(self) -> reflection.Inspector:
+        """Cached SQLAlchemy inspector for this connector."""
+        if self._cached_inspector is None:
+            self._cached_inspector = sa.inspect(self._engine)
+        return self._cached_inspector
+
+    def clear_reflection_cache(self) -> None:
+        """Clear cached SQLAlchemy reflection state, if an inspector exists."""
+        if self._cached_inspector is None:
+            return
+
+        clear_cache = getattr(self._cached_inspector, "clear_cache", None)
+        if clear_cache:
+            clear_cache()
+            return
+
+        self._cached_inspector.info_cache.clear()
+
+    def _clear_reflection_cache_after_ddl(
+        self,
+        _conn: sa.Connection,
+        _cursor: object,
+        statement: str,
+        _parameters: object,
+        _context: object,
+        _executemany: bool,
+    ) -> None:
+        """Clear cached reflection state after SQLAlchemy DDL executes."""
+        statement_prefix = statement.lstrip().partition(" ")[0].upper()
+        if statement_prefix in {"ALTER", "CREATE", "DROP"}:
+            self.clear_reflection_cache()
+
     def create_engine(self) -> sa.Engine:
         """Creates and returns a new engine. Do not call outside of _engine.
 
@@ -1187,7 +1227,7 @@ def table_exists(self, full_table_name: str | FullyQualifiedName) -> bool:
         """
         _, schema_name, table_name = self.parse_full_table_name(full_table_name)
 
-        return sa.inspect(self._engine).has_table(table_name, schema_name)
+        return self._inspector.has_table(table_name, schema_name)
 
     def schema_exists(self, schema_name: str) -> bool:
         """Determine if the target database schema already exists.
@@ -1198,7 +1238,7 @@ def schema_exists(self, schema_name: str) -> bool:
         Returns:
             True if the database schema exists, False if not.
         """
-        schemas = set(sa.inspect(self._engine).get_schema_names())
+        schemas = set(self._inspector.get_schema_names())
         return schema_name in schemas
 
     def get_table_columns(
@@ -1216,8 +1256,7 @@ def get_table_columns(
             An ordered list of column objects.
         """
         _, schema_name, table_name = self.parse_full_table_name(full_table_name)
-        inspector = sa.inspect(self._engine)
-        columns = inspector.get_columns(table_name, schema_name)
+        columns = self._inspector.get_columns(table_name, schema_name)
 
         columns_dict: dict[str, sa.Column] = {
             col_meta["name"]: sa.Column(