feat: Configurable batch size and max wait limit for targets

singer_sdk/helpers/_perftimer.py

@@ -0,0 +1,118 @@
+"""performace timers which deal with dynamic timing events."""
+
+from __future__ import annotations
+
+import time
+
+
+class PerfTimerError(Exception):
+    """A custom exception used to report errors in use of BatchPerfTimer class."""
+
+
+class PerfTimer:
+    """A Basic Performance Timer Class."""
+
+    _start_time: float | None = None
+    _stop_time: float | None = None
+    _lap_time: float | None = None
+
+    @property
+    def start_time(self):
+        return self._start_time
+
+    @property
+    def stop_time(self):
+        return self._stop_time
+
+    @property
+    def lap_time(self):
+        return self._lap_time
+
+    def start(self) -> None:
+        """Start the timer."""
+        if self._start_time is not None:
+            msg = "Timer is running. Use .stop() to stop it"
+            raise PerfTimerError(msg)
+
+        self._start_time = time.perf_counter()
+
+    def stop(self) -> None:
+        """Stop the timer, Stores the elapsed time, and reset."""
+        if self._start_time is None:
+            msg = "Timer is not running. Use .start() to start it"
+            raise PerfTimerError(msg)
+
+        self._stop_time = time.perf_counter()
+        self._lap_time = self._stop_time - self._start_time
+        self._start_time = None
+        self._stop_time = None
+
+
+class BatchPerfTimer(PerfTimer):
+    """The Performance Timer for Target bulk inserts."""
+
+    def __init__(
+        self,
+        max_size: int,
+        max_perf_counter: float,
+    ) -> None:
+        self._sink_max_size: int = max_size
+        self._max_perf_counter: float = max_perf_counter
+
+    SINK_MAX_SIZE_CELING: int = 100000
+    """The max size a bulk insert can be"""
+
+    @property
+    def sink_max_size(self) -> int:
+        """The current MAX_SIZE_DEFAULT."""
+        return self._sink_max_size
+
+    @property
+    def max_perf_counter(self) -> float:
+        """How many seconds can pass before a insert."""
+        return self._max_perf_counter
+
+    @property
+    def perf_diff_allowed_min(self) -> float:
+        """The mininum negative variance allowed, 1/3 worse than wanted."""
+        return -1.0 * (self.max_perf_counter * 0.33)
+
+    @property
+    def perf_diff_allowed_max(self) -> float:
+        """The maximum postive variace allowed, # 3/4 better than wanted."""
+        return self.max_perf_counter * 0.75
+
+    @property
+    def perf_diff(self) -> float:
+        """Difference between wanted elapsed time and actual elpased time."""
+        diff = 0
+        if self._lap_time:
+            diff = self.max_perf_counter - self.lap_time
+        return float(diff)
+
+    def counter_based_max_size(self) -> int:  # noqa: C901
+        """Caclulate performance based batch size."""
+        correction = 0
+        if self.perf_diff < self.perf_diff_allowed_min:
+            if self.sink_max_size >= 15000:  # noqa: PLR2004
+                correction = -5000
+            elif self.sink_max_size >= 10000:  # noqa: PLR2004
+                correction = -1000
+            elif self.sink_max_size >= 1000:  # noqa: PLR2004
+                correction = -100
+            elif self.sink_max_size > 10:  # noqa: PLR2004
+                correction = -10
+        if (
+            self.perf_diff >= self.perf_diff_allowed_max
+            and self.sink_max_size < self.SINK_MAX_SIZE_CELING
+        ):
+            if self.sink_max_size >= 10000:  # noqa: PLR2004
+                correction = 10000
+            elif self.sink_max_size >= 1000:  # noqa: PLR2004
+                correction = 1000
+            elif self.sink_max_size >= 100:  # noqa: PLR2004
+                correction = 100
+            elif self.sink_max_size >= 10:  # noqa: PLR2004
+                correction = 10
+        self._sink_max_size += correction
+        return self.sink_max_size

singer_sdk/sinks/sql.py

@@ -15,6 +15,7 @@
 from singer_sdk.connectors import SQLConnector
 from singer_sdk.exceptions import ConformedNameClashException
 from singer_sdk.helpers._conformers import replace_leading_digit
+from singer_sdk.helpers._perftimer import BatchPerfTimer
 from singer_sdk.sinks.batch import BatchSink
 
 if t.TYPE_CHECKING:
@@ -49,6 +50,23 @@ def __init__(
         """
         self._connector: SQLConnector
         self._connector = connector or self.connector_class(dict(target.config))
+        self._batch_size_rows: int = target.config.get(
+            "batch_size_rows",
+            super().MAX_SIZE_DEFAULT,
+        )
+        self._batch_wait_limit_seconds: int | None = target.config.get(
+            "batch_wait_limit_seconds",
+        )
+
+        self._sink_timer: BatchPerfTimer | None = None
+
+        if self._batch_wait_limit_seconds is not None:
+            self._batch_size_rows = 100
+            self._sink_timer = BatchPerfTimer(
+                self._batch_size_rows,
+                self._batch_wait_limit_seconds,
+            )
+
         super().__init__(target, stream_name, schema, key_properties)
 
     @property
@@ -69,6 +87,51 @@ def connection(self) -> sqlalchemy.engine.Connection:
         """
         return self.connector.connection
 
+    @property
+    def batch_size_rows(self) -> int:
+        """Get batch_size_rows object.
+
+        Returns:
+            A batch_size_rows object.
+        """
+        return self._batch_size_rows
+
+    @batch_size_rows.setter
+    def batch_size_rows(self, new_value: int) -> None:
+        """Set batch_size_rows object to the given value.
+
+        Args:
+            new_value: The value you want to set batch_size_rows too.
+        """
+        self._batch_size_rows = new_value
+
+    @property
+    def batch_wait_limit_seconds(self) -> int | None:
+        """Get batch_wait_limit_seconds object.
+
+        Returns:
+            A batch_wait_limit_seconds object.
+        """
+        return self._batch_wait_limit_seconds
+
+    @property
+    def sink_timer(self) -> BatchPerfTimer | None:
+        """Get sink_timer object.
+
+        Returns:
+            A sink_timer object.
+        """
+        return self._sink_timer
+
+    @property
+    def max_size(self) -> int:
+        """Get max batch size.
+
+        Returns:
+            Max number of records to batch before `is_full=True`
+        """
+        return self.batch_size_rows
+
     @property
     def table_name(self) -> str:
         """Return the table name, with no schema or database part.
@@ -336,6 +399,14 @@ def bulk_insert_records(
         with self.connector._connect() as conn, conn.begin():
             result = conn.execute(insert_sql, new_records)
 
+        # Finish Line for max_size perf counter
+        if self.sink_timer is not None:
+            if self.sink_timer.start_time is not None:
+                self.sink_timer.stop()
+                self.batch_size_rows = self.sink_timer.counter_based_max_size()
+
+            # Starting Line for max_size perf counter
+            self.sink_timer.start()
         return result.rowcount
 
     def merge_upsert_from_table(

tests/core/test_perf_timer.py

@@ -0,0 +1,98 @@
+"""Perf Timer tests."""
+
+from __future__ import annotations
+
+import time
+
+import pytest
+
+from singer_sdk.helpers._perftimer import BatchPerfTimer, PerfTimer, PerfTimerError
+
+
+def test_perftimer_properties():
+    timer: PerfTimer = PerfTimer()
+    timer._start_time = 1.1
+    timer._stop_time = 1.10
+    timer._lap_time = 0.09
+    assert timer.start_time is timer._start_time
+    assert timer.stop_time is timer.stop_time
+    assert timer._lap_time is timer.lap_time
+    assert timer.start_time == 1.1
+    assert timer.stop_time == 1.10
+    assert timer.lap_time == 0.09
+
+
+def test_perftimer_actions():
+    timer: PerfTimer = PerfTimer()
+    timer.start()
+    assert timer.start_time is not None
+    assert timer.stop_time is None
+    assert timer.lap_time is None
+    time.sleep(1.1)
+    timer.stop()
+    assert timer.lap_time >= 1
+    assert timer.lap_time < 1.5
+    assert timer.start_time is None
+    assert timer.stop_time is None
+
+
+def test_perftimer_errors():
+    timer: PerfTimer = PerfTimer()
+    with pytest.raises(
+        PerfTimerError,
+        match=r"Timer is not running. Use .start\(\) to start it",
+    ):
+        timer.stop()
+    # starting a timer to test start() error
+    timer.start()
+    with pytest.raises(
+        PerfTimerError,
+        match=r"Timer is running. Use .stop\(\) to stop it",
+    ):
+        timer.start()
+    # stopping the timer at the end of the test
+    timer.stop()
+
+
+def test_batchperftimer_properties():
+    batchtimer: BatchPerfTimer = BatchPerfTimer(100, 1)
+    batchtimer._lap_time = 0.10
+    assert batchtimer._sink_max_size is batchtimer.sink_max_size
+    assert batchtimer._max_perf_counter is batchtimer.max_perf_counter
+    assert batchtimer.sink_max_size == 100
+    assert batchtimer.max_perf_counter == 1
+    assert batchtimer.perf_diff_allowed_max == 0.75
+    assert batchtimer.perf_diff_allowed_min == -0.33
+    assert batchtimer.perf_diff == 0.90
+
+
+def test_batchperftimer_counter_based_max_size_additive():
+    batchtimer: BatchPerfTimer = BatchPerfTimer(10, 1)
+    batchtimer._lap_time = 0.24
+    assert batchtimer.perf_diff > batchtimer.perf_diff_allowed_max
+    assert batchtimer.counter_based_max_size() == 20
+    batchtimer._sink_max_size = 100
+    assert batchtimer.counter_based_max_size() == 200
+    batchtimer._sink_max_size = 1000
+    assert batchtimer.counter_based_max_size() == 2000
+    batchtimer._sink_max_size = 10000
+    assert batchtimer.counter_based_max_size() == 20000
+    batchtimer._sink_max_size = 100000
+    assert batchtimer.sink_max_size == batchtimer.SINK_MAX_SIZE_CELING
+    assert batchtimer.counter_based_max_size() == 100000
+
+
+def test_batchperftimer_counter_based_max_size_reductive():
+    batchtimer: BatchPerfTimer = BatchPerfTimer(100000, 1)
+    batchtimer._lap_time = 1.5
+    assert batchtimer.perf_diff < batchtimer.perf_diff_allowed_min
+    assert batchtimer.sink_max_size == batchtimer.SINK_MAX_SIZE_CELING
+    assert batchtimer.counter_based_max_size() == 95000
+    batchtimer._sink_max_size = 10000
+    assert batchtimer.counter_based_max_size() == 9000
+    batchtimer._sink_max_size = 1000
+    assert batchtimer.counter_based_max_size() == 900
+    batchtimer._sink_max_size = 100
+    assert batchtimer.counter_based_max_size() == 90
+    batchtimer._sink_max_size = 10
+    assert batchtimer.counter_based_max_size() == 10