feat: Taps can now opt-in to emitting ACTIVATE_VERSION messages for streams with FULL_TABLE replication

singer_sdk/helpers/_batch.py

@@ -92,6 +92,9 @@ class SDKBatchMessage(Message):
     manifest: list[str] = field(default_factory=list)
     """The manifest of files in the batch."""
 
+    version: int | None = None
+    """If syncing in FULL_TABLE mode, the start time as an epoch timestamp int."""
+
     def __post_init__(self) -> None:
         if isinstance(self.encoding, dict):
             self.encoding = BaseBatchFileEncoding.from_dict(self.encoding)

singer_sdk/sinks/sql.py

@@ -91,7 +91,7 @@ def schema_name(self) -> str | None:
         Returns:
             The target schema name.
         """
-        # Look for a default_target_scheme in the configuraion fle
+        # Look for a default_target_scheme in the configuration fle
         default_target_schema: str = self.config.get("default_target_schema", None)
         parts = self.stream_name.split("-")
 

singer_sdk/streams/core.py

@@ -6,6 +6,7 @@
 import copy
 import datetime
 import json
+import time
 import typing as t
 from os import PathLike
 from pathlib import Path
@@ -136,6 +137,8 @@ def __init__(
         self._config: dict = dict(tap.config)
         self._tap = tap
         self._tap_state = tap.state
+        self._stream_version: int | None = None
+        self._initialized_at = tap.initialized_at
         self._tap_input_catalog: singer.Catalog | None = None
         self._stream_maps: list[StreamMap] | None = None
         self.forced_replication_method: str | None = None
@@ -452,7 +455,14 @@ def get_replication_key_signpost(
         Returns:
             Max allowable bookmark value for this stream's replication key.
         """
-        return utc_now() if self.is_timestamp_replication_key else None
+        return (
+            datetime.datetime.fromtimestamp(
+                self._initialized_at,
+                tz=datetime.timezone.utc,
+            )
+            if self.is_timestamp_replication_key
+            else None
+        )
 
     @property
     def schema_filepath(self) -> Path | Traversable | None:
@@ -811,6 +821,10 @@ def _write_state_message(self) -> None:
             self._last_emitted_state = copy.deepcopy(self.tap_state)
             self._is_state_flushed = True
 
+    def _write_activate_version_message(self, full_table_version: int) -> None:  # noqa: PLR6301
+        """Write out an ACTIVATE_VERSION message."""
+        singer.write_message(singer.ActivateVersionMessage(full_table_version))
+
     def _generate_schema_messages(
         self,
     ) -> t.Generator[singer.SchemaMessage, None, None]:
@@ -876,7 +890,7 @@ def _generate_record_messages(
                 yield singer.RecordMessage(
                     stream=stream_map.stream_alias,
                     record=mapped_record,
-                    version=None,
+                    version=self._stream_version,
                     time_extracted=utc_now(),
                 )
 
@@ -1222,6 +1236,10 @@ def sync(self, context: types.Context | None = None) -> None:
         if self.selected:
             self._write_schema_message()
 
+        if self.replication_method == REPLICATION_FULL_TABLE:
+            self._stream_version = int(time.time())
+            self._write_activate_version_message(self._stream_version)
+
         try:
             batch_config = self.get_batch_config(self.config)
             if batch_config: