Dev

pyproject.toml

@@ -94,6 +94,7 @@ ignore = [
 "yt_framework/yt/clients/client_dev.py" = ["ARG002", "PLR0913"]
 # Complexity: remaining high-arity API surfaces are isolated to explicit modules.
 "yt_framework/operations/_internal/dependency_strategy.py" = ["PLR0913"]
+"yt_framework/operations/command_ops/map.py" = ["PLR0913"]
 "yt_framework/operations/command_ops/map_reduce.py" = ["PLR0913"]
 "yt_framework/operations/command_ops/map_reduce_support.py" = ["D103"]
 "yt_framework/operations/s3.py" = ["PLR0913"]

tests/test_map.py

@@ -156,3 +156,12 @@ def test_run_map_raises_when_mapper_missing_and_builder_command_missing(
     ctx = _map_stage_context(tmp_path)
     with pytest.raises(ValueError, match="Command not provided"):
         run_map(ctx, _minimal_map_config())
+
+
+def test_run_map_sync_false_returns_operation_without_waiting(
+    tmp_path: Path,
+) -> None:
+    ctx = _map_stage_context(tmp_path)
+    result = run_map(ctx, _minimal_map_config(), sync=False)
+    assert result is ctx.deps.yt_client.run_map_submit.return_value
+    ctx.deps.yt_client.wait_for_operation.assert_not_called()

yt_framework/operations/command_ops/map.py

@@ -32,6 +32,7 @@
     import logging
     from pathlib import Path
 
+    from yt.wrapper import Operation
     from yt.wrapper.schema import TableSchema
 
     from yt_framework.contracts import StageContext
@@ -173,8 +174,10 @@ def run_map(
     output_schema: TableSchema | None = None,
     mapper: object | None = None,
     job: object | None = None,
-) -> bool:
-    """Run YT map operation and wait for completion.
+    *,
+    sync: bool = True,
+) -> bool | Operation:
+    """Run YT map operation and optionally wait for completion.
 
     All job parameters (pool, memory, CPU, Docker image, etc.) are automatically
     extracted from operation_config. Operation config should be passed from
@@ -188,9 +191,12 @@ def run_map(
         mapper: Optional mapper leg (legacy name). When omitted, framework uses command wrapper.
             Can be a TypedJob instance or command string.
         job: Preferred mapper leg alias. Can be a TypedJob instance or command string.
+        sync: If True (default), wait for operation completion and return bool.
+            If False, submit the operation and return the Operation object immediately.
 
     Returns:
-        True if successful, False otherwise
+        If ``sync=True``: True if successful, False otherwise.
+        If ``sync=False``: the submitted ``yt.wrapper.Operation`` object.
 
     """
     logger = context.logger
@@ -274,6 +280,9 @@ def run_map(
         ),
     )
 
+    if not sync:
+        return operation
+
     # Wait for completion
     success = context.deps.yt_client.wait_for_operation(operation)