fix(sql,targets): Delay table preparation until processing the first batch

Author: edgarrmondragon

singer_sdk/sql/sink.py

@@ -56,6 +56,7 @@ def __init__(
         """
         self._connector: _C
         self._connector = connector or self.connector_class(dict(target.config))
+        self._table_prepared = False
         super().__init__(target, stream_name, schema, key_properties)
 
     @property
@@ -250,6 +251,7 @@ def setup(self) -> None:
             primary_keys=self.key_properties,
             as_temp_table=False,
         )
+        self._table_prepared = True
 
     @property
     def key_properties(self) -> t.Sequence[str]:

tests/core/sinks/test_sql_sink.py

@@ -11,6 +11,9 @@
 from singer_sdk.sinks.sql import SQLSink
 from singer_sdk.target_base import SQLTarget
 
+if t.TYPE_CHECKING:
+    from pytest_subtests import SubTests
+
 if sys.version_info >= (3, 12):
     from typing import override  # noqa: ICN003
 else:
@@ -165,3 +168,48 @@ def test_table_name(
 
         assert sink.table_name == expected_table_name
         assert sink.schema_name == expected_schema_name
+
+    def test_table_preparation_deferred_until_first_batch(self, subtests: SubTests):
+        """Test that table preparation is deferred until first batch.
+
+        This test verifies the fix for issue #3237 where table preparation
+        occurred during setup() instead of during the first batch, causing
+        data loss when multiple schema messages arrived for the same stream.
+
+        The test verifies that:
+        1. Table is NOT prepared during sink setup()
+        2. Table IS prepared during the first batch (start_batch)
+        3. This prevents data loss when schemas change mid-stream
+        """
+        target = DummySQLTarget(config={"sqlalchemy_url": "sqlite:///"})
+
+        schema = {
+            "properties": {
+                "id": {"type": ["string", "null"]},
+                "name": {"type": ["string", "null"]},
+            },
+        }
+
+        # Create a sink
+        sink: SQLSink = target.get_sink(
+            "test_stream",
+            schema=schema,
+            key_properties=["id"],
+        )
+
+        with subtests.test("table does not exist yet"):
+            # Verify table is NOT prepared during setup
+            assert sink._table_prepared is False
+            # Verify table does not exist yet
+            assert not sink.connector.table_exists(sink.full_table_name)
+
+        with subtests.test("table is prepared after first batch"):
+            # Process a record (this triggers _get_context which calls start_batch)
+            record = {"id": "1", "name": "Alice"}
+            _ = sink._get_context(record)
+
+            # After getting context (which calls start_batch), table should be prepared
+            assert sink._table_prepared is True
+
+            # Verify table now exists
+            assert sink.connector.table_exists(sink.full_table_name)