Merge pull request #3 from bluedynamics/feat/incremental-parallel-import

Pass start_tid through parallel delegation

Author: jensens

CHANGES.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## 1.0.0b9
+
+- Support `--incremental -w N` together. Previously, `--incremental`
+  forced a fallback to sequential copy even when `-w` was specified.
+  Now passes `start_tid` through to the destination's parallel
+  `copyTransactionsFrom()` for resumable parallel imports.
+
 ## 1.0.0b8
 
 - Remove source transaction counting (saves a full iteration over 200k+

src/zodb_convert/copier.py

@@ -81,7 +81,7 @@ def get_incremental_start_tid(source, destination):
     return p64(last_valid_int + 1)
 
 
-def _try_parallel_delegation(source, destination, workers):
+def _try_parallel_delegation(source, destination, workers, start_tid=None):
     """Try delegating to the destination's parallel copyTransactionsFrom.
 
     Returns (txn_count, obj_count, blob_count) on success, or None if the
@@ -93,19 +93,25 @@ def _try_parallel_delegation(source, destination, workers):
         log.info("Destination has no copyTransactionsFrom, using generic copier.")
         return None
 
+    kwargs = {"workers": workers}
+    if start_tid is not None:
+        kwargs["start_tid"] = start_tid
+
     try:
-        log.info("Delegating to destination.copyTransactionsFrom(workers=%d).", workers)
-        copy_method(source, workers=workers)
+        log.info(
+            "Delegating to destination.copyTransactionsFrom(%s).",
+            ", ".join(f"{k}={v!r}" for k, v in kwargs.items()),
+        )
+        copy_method(source, **kwargs)
     except TypeError:
-        # Destination's copyTransactionsFrom doesn't accept 'workers'.
+        # Destination's copyTransactionsFrom doesn't accept these kwargs.
         log.info(
-            "Destination doesn't support parallel copy (no workers parameter), "
+            "Destination doesn't support parallel copy, "
             "falling back to generic sequential copier."
         )
         return None
 
     # Destination handled everything including its own progress logging.
-    # Exact per-object counts are unavailable; return None as sentinel.
     return None, None, None
 
 
@@ -125,8 +131,8 @@ def copy_transactions(
 
     Returns (txn_count, obj_count, blob_count).
     """
-    if workers > 1 and not dry_run and start_tid is None:
-        result = _try_parallel_delegation(source, destination, workers)
+    if workers > 1 and not dry_run:
+        result = _try_parallel_delegation(source, destination, workers, start_tid)
         if result is not None:
             return result
         # Fall through to generic sequential copier.

tests/test_copier.py

@@ -326,6 +326,21 @@ def test_delegation_success(self):
         assert result == (None, None, None)
         dest.copyTransactionsFrom.assert_called_once_with(source, workers=4)
 
+    def test_delegation_with_start_tid(self):
+        """Delegation passes start_tid when set."""
+        from ZODB.utils import p64
+
+        source = MagicMock()
+        dest = MagicMock()
+        dest.copyTransactionsFrom = MagicMock()
+
+        start = p64(100)
+        result = _try_parallel_delegation(source, dest, workers=4, start_tid=start)
+        assert result == (None, None, None)
+        dest.copyTransactionsFrom.assert_called_once_with(
+            source, workers=4, start_tid=start
+        )
+
     def test_delegation_no_method(self):
         """Destination has no copyTransactionsFrom at all."""
         source = MagicMock()
@@ -343,38 +358,46 @@ def test_delegation_no_workers_support(self):
         result = _try_parallel_delegation(source, dest, workers=2)
         assert result is None
 
+    def test_delegation_start_tid_not_supported(self):
+        """Destination doesn't support start_tid -- falls back."""
+        from ZODB.utils import p64
+
+        source = MagicMock()
+        dest = MagicMock()
+        dest.copyTransactionsFrom = MagicMock(side_effect=TypeError("unexpected kwarg"))
+
+        result = _try_parallel_delegation(source, dest, workers=2, start_tid=p64(100))
+        assert result is None
+
     def test_copy_transactions_delegates_when_workers_gt_1(
         self, populated_source, dest_filestorage
     ):
         """copy_transactions with workers>1 tries delegation, falls back on TypeError."""
-        # FileStorage.copyTransactionsFrom doesn't accept workers → TypeError fallback
         txn_count, obj_count, _blob_count = copy_transactions(
             populated_source, dest_filestorage, workers=2
         )
-        # Should have fallen back to sequential copy and succeeded
         assert txn_count == 4  # initial root + 3 explicit
         assert obj_count > 0
 
-    def test_copy_transactions_skips_delegation_on_dry_run(
+    def test_copy_transactions_delegates_with_start_tid(
         self, populated_source, dest_filestorage
     ):
-        """workers>1 is ignored when dry_run=True."""
+        """copy_transactions with workers>1 and start_tid delegates both."""
+        from ZODB.utils import p64
+
         txn_count, obj_count, _blob_count = copy_transactions(
-            populated_source, dest_filestorage, dry_run=True, workers=4
+            populated_source, dest_filestorage, start_tid=p64(0), workers=4
         )
         assert txn_count == 4
         assert obj_count > 0
-        # Destination should still be empty (dry run)
-        assert storage_has_data(dest_filestorage) is False
 
-    def test_copy_transactions_skips_delegation_on_incremental(
+    def test_copy_transactions_skips_delegation_on_dry_run(
         self, populated_source, dest_filestorage
     ):
-        """workers>1 is ignored when start_tid is set."""
-        from ZODB.utils import p64
-
+        """workers>1 is ignored when dry_run=True."""
         txn_count, obj_count, _blob_count = copy_transactions(
-            populated_source, dest_filestorage, start_tid=p64(0), workers=4
+            populated_source, dest_filestorage, dry_run=True, workers=4
         )
         assert txn_count == 4
         assert obj_count > 0
+        assert storage_has_data(dest_filestorage) is False