Merge pull request #31 from GregoryKogan/feature/map-reduce-dev

feat: update dev vs prod documentation and enhance command-mode descr…

Author: GregoryKogan

docs/dev-vs-prod.md

@@ -88,6 +88,34 @@ Typical flow:
 
 See [Map operations — Append output](operations/map.md) (`append: true` section).
 
+Command-mode mappers only (string commands). TypedJob map legs run on the cluster in prod.
+
+### MapReduce (dev)
+
+Typical flow:
+
+1. Sandbox: `.dev/sandbox_mr_<input>-><output>/` with `input.jsonl`, `intermediate.jsonl`, and `output.jsonl`.
+2. Copy input JSONL into the sandbox and upload file dependencies (same as map).
+3. Run the mapper command as a subprocess; stdout becomes `intermediate.jsonl`.
+4. Sort intermediate rows by `sort_by` when set, otherwise by `reduce_by` (loads the JSONL into memory; for small local fixtures only).
+5. Run the reducer command; stdout becomes the output table at `.dev/<output>.jsonl`.
+6. Stderr for each leg: `.dev/<output_basename>_mapper.log` and `_reducer.log`.
+
+String commands only; TypedJob map-reduce legs are prod-only (same rule as map).
+
+Dev runs one mapper and one reducer process (no shuffle partitions). For command-mode reducers that expect sorted keys, dev sorting matches what the cluster provides after shuffle.
+
+### Reduce (dev)
+
+Typical flow:
+
+1. Sandbox: `.dev/sandbox_reduce_<input>-><output>/`.
+2. Copy input JSONL, upload dependencies, auto-sort rows by `reduce_by` (in-memory; small fixtures only).
+3. Run the reducer subprocess; stdout becomes `.dev/<output>.jsonl`.
+4. Stderr: `.dev/<output_basename>_reducer.log`.
+
+String commands only. Dev auto-sorts before the reducer so you do not need a separate `run_sort` stage locally. In prod, the input table must already be sorted by `reduce_by` (or run sort first).
+
 ### Vanilla (dev)
 
 1. Sandbox under `.dev/<stage>_sandbox/` (name depends on stage).

docs/operations/command-mode-map-reduce.md

@@ -72,4 +72,6 @@ If omitted, the uploader picks defaults (e.g. first existing among `reducer.py`,
 
 ## Dev client
 
-`client_dev` does not execute map-reduce legs; it logs whether each leg would be TypedJob or command (`JsonFormat` in prod) for sanity checks.
+In dev mode, map-reduce and reduce run locally as subprocesses (JSONL stdin/stdout) with the same contract as command-mode map. Map-reduce runs mapper → sort by `sort_by` or `reduce_by` → reducer. Reduce-only auto-sorts by `reduce_by` before the reducer.
+
+String commands only; TypedJob legs stay prod-only. See [Dev vs prod — MapReduce and Reduce](../dev-vs-prod.md).

tests/test_client_dev.py

@@ -570,63 +570,163 @@ def test_dev_client_run_yql_warns_when_input_table_jsonl_missing(
     )
 
 
-def test_dev_client_run_map_reduce_copies_input_jsonl_to_output(
-    tmp_path: Path,
-) -> None:
+_DEV_MR_MAPPER = """#!/usr/bin/env python3
+import json
+import sys
+
+for line in sys.stdin:
+    row = json.loads(line)
+    print(json.dumps({"k": row["k"], "v": int(row["v"])}), flush=True)
+"""
+
+_DEV_MR_REDUCER = """#!/usr/bin/env python3
+import json
+import sys
+
+cur_k = None
+s = 0
+
+
+def flush() -> None:
+    global cur_k, s
+    if cur_k is not None:
+        print(json.dumps({"k": cur_k, "sum_v": s}), flush=True)
+
+
+for line in sys.stdin:
+    row = json.loads(line)
+    k, v = row["k"], int(row["v"])
+    if cur_k is None:
+        cur_k, s = k, v
+    elif k == cur_k:
+        s += v
+    else:
+        flush()
+        cur_k, s = k, v
+flush()
+"""
+
+
+def _write_mr_job_scripts(pipeline_dir: Path) -> list[tuple[str, str]]:
+    (pipeline_dir / "mapper.py").write_text(_DEV_MR_MAPPER, encoding="utf-8")
+    (pipeline_dir / "reducer.py").write_text(_DEV_MR_REDUCER, encoding="utf-8")
+    return [
+        ("//yt/mapper.py", "mapper.py"),
+        ("//yt/reducer.py", "reducer.py"),
+    ]
+
+
+def test_dev_client_run_map_reduce_sums_by_key(tmp_path: Path) -> None:
     client = YTDevClient(_null_logger("tests.client_dev.mr"), pipeline_dir=tmp_path)
-    client.write_table("//tmp/mr_in", [{"row": 1}], append=False)
+    client.write_table(
+        "//tmp/mr_in",
+        [
+            {"k": "a", "v": 1},
+            {"k": "a", "v": 2},
+            {"k": "b", "v": 3},
+        ],
+        append=False,
+    )
+    deps = _write_mr_job_scripts(tmp_path)
     op = client.run_map_reduce(
         "python3 mapper.py",
         "python3 reducer.py",
         "//tmp/mr_in",
         "//tmp/mr_out",
-        ["id"],
-        [],
+        ["k"],
+        deps,
         OperationResources(),
         {},
     )
-    assert op.get_state() == "completed" and client.read_table("//tmp/mr_out") == [
-        {"row": 1}
-    ], "dev run_map_reduce copies input table jsonl to output"
+    assert op.get_state() == "completed", "map-reduce must finish successfully"
+    got = {r["k"]: r["sum_v"] for r in client.read_table("//tmp/mr_out")}
+    assert got == {"a": 3, "b": 3}, "dev map-reduce runs mapper then reducer"
 
 
-def test_dev_client_run_map_reduce_writes_empty_output_when_input_missing(
+def test_dev_client_run_map_reduce_raises_when_input_missing(tmp_path: Path) -> None:
+    client = YTDevClient(
+        _null_logger("tests.client_dev.mr_miss"), pipeline_dir=tmp_path
+    )
+    with pytest.raises(FileNotFoundError, match="input table file not found"):
+        client.run_map_reduce(
+            "python3 mapper.py",
+            "python3 reducer.py",
+            "//tmp/missing_in",
+            "//tmp/mr_empty_out",
+            ["k"],
+            [],
+            OperationResources(),
+            {},
+        )
+
+
+def test_dev_client_run_map_reduce_fails_when_mapper_exits_nonzero(
     tmp_path: Path,
 ) -> None:
     client = YTDevClient(
-        _null_logger("tests.client_dev.mr_miss"), pipeline_dir=tmp_path
+        _null_logger("tests.client_dev.mr_fail"), pipeline_dir=tmp_path
     )
+    client.write_table("//tmp/mr_in", [{"k": "a", "v": 1}], append=False)
     op = client.run_map_reduce(
-        "m",
-        "r",
-        "//tmp/missing_in",
-        "//tmp/mr_empty_out",
-        [],
-        [],
+        "exit 1",
+        "python3 reducer.py",
+        "//tmp/mr_in",
+        "//tmp/mr_out",
+        ["k"],
+        _write_mr_job_scripts(tmp_path),
         OperationResources(),
         {},
     )
-    assert op.get_state() == "completed"
-    assert client.read_table("//tmp/mr_empty_out") == []
+    assert op.get_state() == "failed", "mapper failure should fail the operation"
+    assert not client._table_local_path("//tmp/mr_out").exists(), (
+        "reducer must not run when mapper fails"
+    )
 
 
-def test_dev_client_run_reduce_copies_input_jsonl_to_output(
+def test_dev_client_run_reduce_sums_unsorted_input_after_auto_sort(
     tmp_path: Path,
 ) -> None:
     client = YTDevClient(_null_logger("tests.client_dev.red"), pipeline_dir=tmp_path)
-    client.write_table("//tmp/rd_in", [{"k": "v"}], append=False)
+    client.write_table(
+        "//tmp/rd_in",
+        [
+            {"k": "b", "v": 1},
+            {"k": "a", "v": 2},
+            {"k": "a", "v": 1},
+        ],
+        append=False,
+    )
+    (tmp_path / "reducer.py").write_text(_DEV_MR_REDUCER, encoding="utf-8")
     op = client.run_reduce(
         "python3 reducer.py",
         "//tmp/rd_in",
         "//tmp/rd_out",
         ["k"],
-        [],
+        [("//yt/reducer.py", "reducer.py")],
         OperationResources(),
         {},
     )
-    assert op.get_state() == "completed" and client.read_table("//tmp/rd_out") == [
-        {"k": "v"}
-    ], "dev run_reduce copies input table jsonl to output"
+    assert op.get_state() == "completed", "reduce must finish successfully"
+    got = {r["k"]: r["sum_v"] for r in client.read_table("//tmp/rd_out")}
+    assert got == {"a": 3, "b": 1}, "dev reduce auto-sorts then runs reducer"
+
+
+def test_dev_client_run_map_reduce_raises_for_typed_job_legs(tmp_path: Path) -> None:
+    client = YTDevClient(
+        _null_logger("tests.client_dev.mr_typed"), pipeline_dir=tmp_path
+    )
+    client.write_table("//tmp/mr_in", [{"k": "a", "v": 1}], append=False)
+    with pytest.raises(NotImplementedError, match="string commands"):
+        client.run_map_reduce(
+            TypedJob(),
+            "python3 reducer.py",
+            "//tmp/mr_in",
+            "//tmp/mr_out",
+            ["k"],
+            [],
+            OperationResources(),
+            {},
+        )
 
 
 def test_dev_client_join_tables_materializes_joined_rows_in_output_jsonl(

tests/test_client_dev_runtime.py

@@ -0,0 +1,108 @@
+"""Unit tests for yt_framework.yt.support._client_dev_runtime helpers."""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+
+from yt_framework.yt.support._client_dev_runtime import (
+    dev_copy_output_to_table,
+    dev_resolve_sort_keys,
+    dev_sort_jsonl_file,
+)
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def test_dev_resolve_sort_keys_prefers_sort_by_when_set() -> None:
+    keys = dev_resolve_sort_keys(reduce_by=["a"], sort_by=["b", "a"])
+    assert keys == ["b", "a"], "sort_by wins when non-empty"
+
+
+def test_dev_resolve_sort_keys_falls_back_to_reduce_by() -> None:
+    keys = dev_resolve_sort_keys(reduce_by=["k"], sort_by=None)
+    assert keys == ["k"], "reduce_by used when sort_by absent"
+
+
+def test_dev_sort_jsonl_file_missing_sort_key_sorts_before_present(
+    tmp_path: Path,
+) -> None:
+    path = tmp_path / "rows.jsonl"
+    path.write_text(
+        '{"k":"b"}\n{}\n{"k":"a"}\n',
+        encoding="utf-8",
+    )
+    dev_sort_jsonl_file(path, ["k"])
+    rows = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
+    assert rows == [{}, {"k": "a"}, {"k": "b"}], (
+        "rows missing sort key sort before rows that define the key"
+    )
+
+
+def test_dev_sort_jsonl_file_mixed_types_do_not_raise(tmp_path: Path) -> None:
+    path = tmp_path / "rows.jsonl"
+    path.write_text(
+        '{"k": 2}\n{"k": "1"}\n{"k": 10}\n',
+        encoding="utf-8",
+    )
+    dev_sort_jsonl_file(path, ["k"])
+    rows = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
+    assert rows == [{"k": "1"}, {"k": 10}, {"k": 2}], (
+        "mixed types sort by JSON canonical string without TypeError"
+    )
+
+
+def test_dev_sort_jsonl_file_multi_key_missing_secondary_key(tmp_path: Path) -> None:
+    path = tmp_path / "rows.jsonl"
+    path.write_text(
+        '{"k":"a","v":2}\n{"k":"a"}\n{"k":"a","v":1}\n',
+        encoding="utf-8",
+    )
+    dev_sort_jsonl_file(path, ["k", "v"])
+    rows = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
+    assert rows == [{"k": "a"}, {"k": "a", "v": 1}, {"k": "a", "v": 2}], (
+        "missing secondary sort key sorts before rows that define it"
+    )
+
+
+def test_dev_sort_jsonl_file_orders_rows_by_keys(tmp_path: Path) -> None:
+    path = tmp_path / "rows.jsonl"
+    path.write_text(
+        '{"k":"b","v":1}\n{"k":"a","v":2}\n{"k":"a","v":1}\n',
+        encoding="utf-8",
+    )
+    dev_sort_jsonl_file(path, ["k"])
+    rows = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
+    assert [r["k"] for r in rows] == ["a", "a", "b"], "stable sort by reduce key"
+
+
+def test_dev_sort_jsonl_file_noop_when_sort_keys_empty(tmp_path: Path) -> None:
+    path = tmp_path / "rows.jsonl"
+    original = '{"k":"b"}\n{"k":"a"}\n'
+    path.write_text(original, encoding="utf-8")
+    dev_sort_jsonl_file(path, [])
+    assert path.read_text(encoding="utf-8") == original
+
+
+def test_dev_copy_output_to_table_skips_on_nonzero_returncode(tmp_path: Path) -> None:
+    sandbox_output = tmp_path / "sand.jsonl"
+    sandbox_output.write_text('{"x": 1}\n', encoding="utf-8")
+    out_table = tmp_path / "out.jsonl"
+    dev_copy_output_to_table(
+        proc_returncode=1,
+        sandbox_output=sandbox_output,
+        output_table_local_path=out_table,
+    )
+    assert not out_table.exists(), "failed leg must not publish output"
+
+
+def test_dev_sort_jsonl_file_uses_sort_by_columns(tmp_path: Path) -> None:
+    path = tmp_path / "rows.jsonl"
+    path.write_text(
+        '{"k":"x","ord":2}\n{"k":"x","ord":1}\n',
+        encoding="utf-8",
+    )
+    dev_sort_jsonl_file(path, ["ord"])
+    rows = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
+    assert [r["ord"] for r in rows] == [1, 2]