fix: Revert "feat(taps): Queue parent contexts and sync child streams only when the queue is full (#3058)" (#3378)

edgarrmondragon · web-flow · commit 8ba143fb203e · 2025-11-18T23:11:11.000-06:00
This reverts commit 16ea7ad. ## Summary by Sourcery Revert the context queueing mechanism introduced earlier for parent-child streams; child streams now sync immediately per context, and related queue configuration and flush logic have been removed along with the associated test case. Bug Fixes: - Restore immediate syncing of child streams per context by removing batch queue processing Enhancements: - Simplify child sync logic by removing QUEUE_MAX_SIZE, context queue, and flush methods Tests: - Remove test for handling exceptions in child stream sync and its snapshots
diff --git a/singer_sdk/streams/core.py b/singer_sdk/streams/core.py
@@ -116,9 +116,6 @@ class Stream(metaclass=abc.ABCMeta):  # noqa: PLR0904
     selected_by_default: bool = True
     """Whether this stream is selected by default in the catalog."""
 
-    QUEUE_MAX_SIZE: int = 1000
-    """Maximum number of contexts to queue before syncing child streams."""
-
     def __init__(
         self,
         tap: Tap,
@@ -170,13 +167,6 @@ def __init__(
         self._is_state_flushed: bool = True
         self._sync_costs: dict[str, int] = {}
         self.child_streams: list[Stream] = []
-
-        # Single queue for all child contexts
-        # Use a simple list for the context queue. Thread safety is not needed because
-        # the queue is only accessed from a single thread during sync. Using a list is
-        # more efficient and simpler than queue.Queue for this use case.
-        self._child_context_queue: list[types.Context] = []
-
         if schema:
             if isinstance(schema, (PathLike, str)):
                 if not Path(schema).is_file():
@@ -1348,8 +1338,6 @@ def _sync_records(  # noqa: C901
                     # Finalize per-partition state only if 1:1 with context
                     state = self.get_context_state(current_context)
                     self._finalize_state(state)
-            # FLUSH any remaining child contexts at the end of parent sync
-            self._flush_child_context_queue()
 
         if not context:
             # Finalize total stream only if we have the full context.
@@ -1444,35 +1432,9 @@ def _sync_children(self, child_context: types.Context | None) -> None:
             )
             return
 
-        self._child_context_queue.append(child_context)
-        if len(self._child_context_queue) >= self.QUEUE_MAX_SIZE:
-            self._flush_child_context_queue()
-
-    def _flush_child_context_queue(self) -> None:
-        """Sync all child streams for each context, then clear the queue."""
-        if not self._child_context_queue or not self.child_streams:
-            return
-
-        self.logger.info(
-            "Flushing %d child contexts for stream '%s'",
-            len(self._child_context_queue),
-            self.name,
-        )
-
-        for context in self._child_context_queue:
-            for child_stream in self.child_streams:
-                if child_stream.selected or child_stream.has_selected_descendents:
-                    try:
-                        child_stream.sync(context=context)
-                    except Exception:  # noqa: BLE001
-                        self.log(
-                            "Error syncing child stream '%s'",
-                            child_stream.name,
-                            level=logging.ERROR,
-                            extra={"context": context},
-                        )
-
-        self._child_context_queue = []
+        for child_stream in self.child_streams:
+            if child_stream.selected or child_stream.has_selected_descendents:
+                child_stream.sync(context=child_context)
 
     # Overridable Methods
 
diff --git a/tests/core/snapshots/test_parent_child/test_child_deselected_parent/singer.jsonl b/tests/core/snapshots/test_parent_child/test_child_deselected_parent/singer.jsonl
@@ -12,4 +12,5 @@
 {"type":"RECORD","stream":"child","record":{"id":1,"pid":3},"time_extracted":"2022-01-01T00:00:00+00:00"}
 {"type":"RECORD","stream":"child","record":{"id":2,"pid":3},"time_extracted":"2022-01-01T00:00:00+00:00"}
 {"type":"RECORD","stream":"child","record":{"id":3,"pid":3},"time_extracted":"2022-01-01T00:00:00+00:00"}
+{"type":"STATE","value":{"bookmarks":{"child":{"partitions":[{"context":{"pid":1}},{"context":{"pid":2}},{"context":{"pid":3}}]}}}}
 {"type":"STATE","value":{"bookmarks":{"child":{"partitions":[{"context":{"pid":1}},{"context":{"pid":2}},{"context":{"pid":3}}]},"parent":{}}}}
diff --git a/tests/core/snapshots/test_parent_child/test_child_deselected_parent/stderr.log b/tests/core/snapshots/test_parent_child/test_child_deselected_parent/stderr.log
@@ -1,6 +1,4 @@
 INFO my-tap.parent Beginning sync of 'parent' in full_table mode
-INFO my-tap.parent Flushing 2 child contexts for stream 'parent'
 INFO my-tap.child Beginning sync of 'child' in full_table mode with context: {'pid': 1}
 INFO my-tap.child Beginning sync of 'child' in full_table mode with context: {'pid': 2}
-INFO my-tap.parent Flushing 1 child contexts for stream 'parent'
 INFO my-tap.child Beginning sync of 'child' in full_table mode with context: {'pid': 3}
diff --git a/tests/core/snapshots/test_parent_child/test_child_sync_exception/singer.jsonl b/tests/core/snapshots/test_parent_child/test_child_sync_exception/singer.jsonl
diff --git a/tests/core/snapshots/test_parent_child/test_child_sync_exception/stderr.log b/tests/core/snapshots/test_parent_child/test_child_sync_exception/stderr.log
diff --git a/tests/core/snapshots/test_parent_child/test_deselected_child/stderr.log b/tests/core/snapshots/test_parent_child/test_deselected_child/stderr.log
@@ -1,4 +1,2 @@
 INFO my-tap Skipping deselected stream 'child'.
 INFO my-tap.parent Beginning sync of 'parent' in full_table mode
-INFO my-tap.parent Flushing 2 child contexts for stream 'parent'
-INFO my-tap.parent Flushing 1 child contexts for stream 'parent'
diff --git a/tests/core/snapshots/test_parent_child/test_one_parent_many_children/singer.jsonl b/tests/core/snapshots/test_parent_child/test_one_parent_many_children/singer.jsonl
@@ -1,11 +1,12 @@
 {"type":"SCHEMA","stream":"parent_many","schema":{"properties":{"id":{"type":"integer"},"children":{"items":{"type":"integer"},"type":"array"}},"type":"object"},"key_properties":[]}
-{"type":"RECORD","stream":"parent_many","record":{"id":"1","children":[1,2,3]},"time_extracted":"2022-01-01T00:00:00+00:00"}
 {"type":"SCHEMA","stream":"child_many","schema":{"properties":{"id":{"type":"integer"},"pid":{"type":"integer"}},"type":"object"},"key_properties":[]}
 {"type":"RECORD","stream":"child_many","record":{"id":1,"pid":"1"},"time_extracted":"2022-01-01T00:00:00+00:00"}
-{"type":"STATE","value":{"bookmarks":{"parent_many":{},"child_many":{"partitions":[{"context":{"child_id":1,"pid":"1"}}]}}}}
+{"type":"STATE","value":{"bookmarks":{"child_many":{"partitions":[{"context":{"child_id":1,"pid":"1"}}]}}}}
 {"type":"SCHEMA","stream":"child_many","schema":{"properties":{"id":{"type":"integer"},"pid":{"type":"integer"}},"type":"object"},"key_properties":[]}
 {"type":"RECORD","stream":"child_many","record":{"id":2,"pid":"1"},"time_extracted":"2022-01-01T00:00:00+00:00"}
-{"type":"STATE","value":{"bookmarks":{"parent_many":{},"child_many":{"partitions":[{"context":{"child_id":1,"pid":"1"}},{"context":{"child_id":2,"pid":"1"}}]}}}}
+{"type":"STATE","value":{"bookmarks":{"child_many":{"partitions":[{"context":{"child_id":1,"pid":"1"}},{"context":{"child_id":2,"pid":"1"}}]}}}}
 {"type":"SCHEMA","stream":"child_many","schema":{"properties":{"id":{"type":"integer"},"pid":{"type":"integer"}},"type":"object"},"key_properties":[]}
 {"type":"RECORD","stream":"child_many","record":{"id":3,"pid":"1"},"time_extracted":"2022-01-01T00:00:00+00:00"}
-{"type":"STATE","value":{"bookmarks":{"parent_many":{},"child_many":{"partitions":[{"context":{"child_id":1,"pid":"1"}},{"context":{"child_id":2,"pid":"1"}},{"context":{"child_id":3,"pid":"1"}}]}}}}
+{"type":"STATE","value":{"bookmarks":{"child_many":{"partitions":[{"context":{"child_id":1,"pid":"1"}},{"context":{"child_id":2,"pid":"1"}},{"context":{"child_id":3,"pid":"1"}}]}}}}
+{"type":"RECORD","stream":"parent_many","record":{"id":"1","children":[1,2,3]},"time_extracted":"2022-01-01T00:00:00+00:00"}
+{"type":"STATE","value":{"bookmarks":{"child_many":{"partitions":[{"context":{"child_id":1,"pid":"1"}},{"context":{"child_id":2,"pid":"1"}},{"context":{"child_id":3,"pid":"1"}}]},"parent_many":{}}}}
diff --git a/tests/core/snapshots/test_parent_child/test_one_parent_many_children/stderr.log b/tests/core/snapshots/test_parent_child/test_one_parent_many_children/stderr.log
@@ -1,7 +1,6 @@
 INFO my-tap-many Skipping parse of env var settings...
 INFO my-tap-many Added 'child_many' as child stream to 'parent_many'
 INFO my-tap-many.parent_many Beginning sync of 'parent_many' in full_table mode
-INFO my-tap-many.parent_many Flushing 3 child contexts for stream 'parent_many'
 INFO my-tap-many.child_many Beginning sync of 'child_many' in full_table mode with context: {'child_id': 1, 'pid': '1'}
 WARNING my-tap-many.child_many Properties ('composite_id', 'child_id') were present in the 'child_many' stream but not found in catalog schema. Ignoring.
 INFO my-tap-many.child_many Beginning sync of 'child_many' in full_table mode with context: {'child_id': 2, 'pid': '1'}
diff --git a/tests/core/snapshots/test_parent_child/test_parent_context_fields_in_child/singer.jsonl b/tests/core/snapshots/test_parent_child/test_parent_context_fields_in_child/singer.jsonl
@@ -1,19 +1,19 @@
 {"type":"SCHEMA","stream":"parent","schema":{"properties":{"id":{"type":"integer"}},"type":"object"},"key_properties":[]}
-{"type":"RECORD","stream":"parent","record":{"id":1},"time_extracted":"2022-01-01T00:00:00+00:00"}
 {"type":"SCHEMA","stream":"child","schema":{"properties":{"id":{"type":"integer"},"pid":{"type":"integer"}},"type":"object"},"key_properties":[]}
 {"type":"RECORD","stream":"child","record":{"id":1,"pid":1},"time_extracted":"2022-01-01T00:00:00+00:00"}
 {"type":"RECORD","stream":"child","record":{"id":2,"pid":1},"time_extracted":"2022-01-01T00:00:00+00:00"}
 {"type":"RECORD","stream":"child","record":{"id":3,"pid":1},"time_extracted":"2022-01-01T00:00:00+00:00"}
-{"type":"STATE","value":{"bookmarks":{"parent":{},"child":{"partitions":[{"context":{"pid":1}}]}}}}
+{"type":"STATE","value":{"bookmarks":{"child":{"partitions":[{"context":{"pid":1}}]}}}}
+{"type":"RECORD","stream":"parent","record":{"id":1},"time_extracted":"2022-01-01T00:00:00+00:00"}
 {"type":"SCHEMA","stream":"child","schema":{"properties":{"id":{"type":"integer"},"pid":{"type":"integer"}},"type":"object"},"key_properties":[]}
 {"type":"RECORD","stream":"child","record":{"id":1,"pid":2},"time_extracted":"2022-01-01T00:00:00+00:00"}
 {"type":"RECORD","stream":"child","record":{"id":2,"pid":2},"time_extracted":"2022-01-01T00:00:00+00:00"}
 {"type":"RECORD","stream":"child","record":{"id":3,"pid":2},"time_extracted":"2022-01-01T00:00:00+00:00"}
-{"type":"STATE","value":{"bookmarks":{"parent":{},"child":{"partitions":[{"context":{"pid":1}},{"context":{"pid":2}}]}}}}
+{"type":"STATE","value":{"bookmarks":{"child":{"partitions":[{"context":{"pid":1}},{"context":{"pid":2}}]},"parent":{}}}}
 {"type":"RECORD","stream":"parent","record":{"id":2},"time_extracted":"2022-01-01T00:00:00+00:00"}
-{"type":"RECORD","stream":"parent","record":{"id":3},"time_extracted":"2022-01-01T00:00:00+00:00"}
 {"type":"SCHEMA","stream":"child","schema":{"properties":{"id":{"type":"integer"},"pid":{"type":"integer"}},"type":"object"},"key_properties":[]}
 {"type":"RECORD","stream":"child","record":{"id":1,"pid":3},"time_extracted":"2022-01-01T00:00:00+00:00"}
 {"type":"RECORD","stream":"child","record":{"id":2,"pid":3},"time_extracted":"2022-01-01T00:00:00+00:00"}
 {"type":"RECORD","stream":"child","record":{"id":3,"pid":3},"time_extracted":"2022-01-01T00:00:00+00:00"}
-{"type":"STATE","value":{"bookmarks":{"parent":{},"child":{"partitions":[{"context":{"pid":1}},{"context":{"pid":2}},{"context":{"pid":3}}]}}}}
+{"type":"STATE","value":{"bookmarks":{"child":{"partitions":[{"context":{"pid":1}},{"context":{"pid":2}},{"context":{"pid":3}}]},"parent":{}}}}
+{"type":"RECORD","stream":"parent","record":{"id":3},"time_extracted":"2022-01-01T00:00:00+00:00"}
diff --git a/tests/core/snapshots/test_parent_child/test_parent_context_fields_in_child/stderr.log b/tests/core/snapshots/test_parent_child/test_parent_context_fields_in_child/stderr.log
@@ -1,6 +1,4 @@
 INFO my-tap.parent Beginning sync of 'parent' in full_table mode
-INFO my-tap.parent Flushing 2 child contexts for stream 'parent'
 INFO my-tap.child Beginning sync of 'child' in full_table mode with context: {'pid': 1}
 INFO my-tap.child Beginning sync of 'child' in full_table mode with context: {'pid': 2}
-INFO my-tap.parent Flushing 1 child contexts for stream 'parent'
 INFO my-tap.child Beginning sync of 'child' in full_table mode with context: {'pid': 3}
diff --git a/tests/core/snapshots/test_parent_child/test_preprocess_context_removes_large_payload/singer.jsonl b/tests/core/snapshots/test_parent_child/test_preprocess_context_removes_large_payload/singer.jsonl
@@ -1,9 +1,9 @@
 {"type":"SCHEMA","stream":"parent_large","schema":{"properties":{"id":{"type":"integer"},"name":{"type":"string"}},"type":"object"},"key_properties":[]}
-{"type":"RECORD","stream":"parent_large","record":{"id":1,"name":"Parent A"},"time_extracted":"2022-01-01T00:00:00+00:00"}
-{"type":"RECORD","stream":"parent_large","record":{"id":2,"name":"Parent B"},"time_extracted":"2022-01-01T00:00:00+00:00"}
 {"type":"SCHEMA","stream":"child_preprocessed","schema":{"properties":{"id":{"type":"integer"},"parent_id":{"type":"integer"},"parent_name":{"type":"string"},"sum":{"type":"integer"}},"type":"object"},"key_properties":[]}
 {"type":"RECORD","stream":"child_preprocessed","record":{"id":1,"parent_id":1,"parent_name":"Parent A","sum":500500},"time_extracted":"2022-01-01T00:00:00+00:00"}
-{"type":"STATE","value":{"bookmarks":{"parent_large":{},"child_preprocessed":{"partitions":[{"context":{"parent_id":1,"parent_name":"Parent A"}}]}}}}
+{"type":"STATE","value":{"bookmarks":{"child_preprocessed":{"partitions":[{"context":{"parent_id":1,"parent_name":"Parent A"}}]}}}}
+{"type":"RECORD","stream":"parent_large","record":{"id":1,"name":"Parent A"},"time_extracted":"2022-01-01T00:00:00+00:00"}
 {"type":"SCHEMA","stream":"child_preprocessed","schema":{"properties":{"id":{"type":"integer"},"parent_id":{"type":"integer"},"parent_name":{"type":"string"},"sum":{"type":"integer"}},"type":"object"},"key_properties":[]}
 {"type":"RECORD","stream":"child_preprocessed","record":{"id":1,"parent_id":2,"parent_name":"Parent B","sum":500500},"time_extracted":"2022-01-01T00:00:00+00:00"}
-{"type":"STATE","value":{"bookmarks":{"parent_large":{},"child_preprocessed":{"partitions":[{"context":{"parent_id":1,"parent_name":"Parent A"}},{"context":{"parent_id":2,"parent_name":"Parent B"}}]}}}}
+{"type":"STATE","value":{"bookmarks":{"child_preprocessed":{"partitions":[{"context":{"parent_id":1,"parent_name":"Parent A"}},{"context":{"parent_id":2,"parent_name":"Parent B"}}]},"parent_large":{}}}}
+{"type":"RECORD","stream":"parent_large","record":{"id":2,"name":"Parent B"},"time_extracted":"2022-01-01T00:00:00+00:00"}
diff --git a/tests/core/snapshots/test_parent_child/test_preprocess_context_removes_large_payload/stderr.log b/tests/core/snapshots/test_parent_child/test_preprocess_context_removes_large_payload/stderr.log
@@ -1,6 +1,5 @@
 INFO tap-preprocess Skipping parse of env var settings...
 INFO tap-preprocess Added 'child_preprocessed' as child stream to 'parent_large'
 INFO tap-preprocess.parent_large Beginning sync of 'parent_large' in full_table mode
-INFO tap-preprocess.parent_large Flushing 2 child contexts for stream 'parent_large'
 INFO tap-preprocess.child_preprocessed Beginning sync of 'child_preprocessed' in full_table mode with context: {'parent_id': 1, 'parent_name': 'Parent A'}
 INFO tap-preprocess.child_preprocessed Beginning sync of 'child_preprocessed' in full_table mode with context: {'parent_id': 2, 'parent_name': 'Parent B'}
diff --git a/tests/core/test_parent_child.py b/tests/core/test_parent_child.py
@@ -20,8 +20,6 @@
 class Parent(Stream):
     """A parent stream."""
 
-    QUEUE_MAX_SIZE = 2
-
     name = "parent"
     schema: t.ClassVar[dict] = {
         "type": "object",
@@ -200,64 +198,6 @@ def test_deselected_child(
     snapshot.assert_match(caplog.text, "stderr.log")
 
 
-@time_machine.travel(DATETIME, tick=False)
-@pytest.mark.snapshot
-def test_child_sync_exception(
-    caplog: pytest.LogCaptureFixture,
-    snapshot: Snapshot,
-):
-    """Test that exceptions in child stream sync are caught and logged."""
-
-    class FailingChild(Stream):
-        """A child stream that raises an exception during sync."""
-
-        name = "failing_child"
-        schema: t.ClassVar[dict] = {
-            "type": "object",
-            "properties": {
-                "id": {"type": "integer"},
-                "pid": {"type": "integer"},
-            },
-        }
-        parent_stream_type = Parent
-
-        def get_records(self, context: dict | None):
-            """Raise an exception."""
-            if context and context.get("pid") == 1:
-                msg = "Intentional failure in child stream"
-                raise ValueError(msg)
-            yield {"id": 1}
-            yield {"id": 2}
-            yield {"id": 3}
-
-    class MyTapWithFailingChild(Tap):
-        """A tap with a failing child stream."""
-
-        name = "my-tap-failing"
-
-        def discover_streams(self):
-            """Discover streams."""
-            return [
-                Parent(self),
-                FailingChild(self),
-            ]
-
-    failing_tap = MyTapWithFailingChild()
-
-    buf = io.StringIO()
-    with (
-        redirect_stdout(buf),
-        caplog.at_level("ERROR"),
-        caplog.filtering(logging.Filter(failing_tap.name)),
-    ):
-        failing_tap.sync_all()
-
-    buf.seek(0)
-
-    snapshot.assert_match(buf.read(), "singer.jsonl")
-    snapshot.assert_match(caplog.text, "stderr.log")
-
-
 @time_machine.travel(DATETIME, tick=False)
 @pytest.mark.snapshot
 def test_one_parent_many_children(
