diff --git a/README.md b/README.md
index 123de1a..c1eb302 100644
--- a/README.md
+++ b/README.md
@@ -39,15 +39,17 @@ After install, the `data-transform` console script is on your `PATH`.
 | `group-by` | Group array of objects by top-level KEY |
 | `count` | Count items, per KEY value, or per group |
 | `stats` | count / sum / min / max / avg of a numeric FIELD |
+| `join` | Join two arrays of objects on a top-level KEY (`--with` file) |
 
 ### Options (most commands)
 
 - `[file]` — input path; omit or use `-` for stdin
 - `-o, --output <file>` — write to a file instead of stdout
 - `-f, --from <format>` — force input format: `json` | `yaml` | `csv` | `ndjson`
-- `--minify` — compact JSON output (`to-json`, `select`, `pick`, `omit`, `rename`, `flatten`, `unflatten`, `filter`, `sort`, `unique`, `head`, `tail`, `group-by`, `count`)
+- `--minify` — compact JSON output (`to-json`, `select`, `pick`, `omit`, `rename`, `flatten`, `unflatten`, `filter`, `sort`, `unique`, `head`, `tail`, `group-by`, `count`, `stats`, `join`)
 - `--desc` — reverse sort order (`sort` only)
 - `--sep <text>` — separator between nested keys (`flatten` and `unflatten`, default `.`)
+- `-w, --with <file>`, `--how inner|left`, `--right-from <format>` — right-hand file, join type, and right file format (`join` only)
 
 Format is taken from `--from`, else the file extension (`.ndjson` / `.jsonl` → ndjson), else simple content heuristics.
 
@@ -251,6 +253,32 @@ echo '[{"ms":120},{"ms":80},{"ms":"100"},{"ms":"n/a"}]' \
 data-transform group-by service requests.csv | data-transform stats ms
 ```
 
+### Join
+
+`join KEY --with RIGHT_FILE [file]` joins two arrays of objects on a top-level `KEY`. The left side is `[file]` or stdin; the right side is always a file given with `-w` / `--with`. Each side's format is detected on its own, so you can join JSON against CSV, YAML, or NDJSON. Use `-f` to force the left format and `--right-from` to force the right one.
+
+- `--how inner` (default) keeps only rows that match. `--how left` also keeps left rows with no match, unchanged.
+- Output keeps left-row order. A left row that matches several right rows produces one output row per match, in right-file order.
+- Each output row starts with the left row's keys, then adds the right row's keys. If both have the same key (other than `KEY`), the right value wins. `KEY` appears once, with the left value.
+- Values match using the same labels as `group-by`, so `1` and `"1"` are equal — useful when one side is CSV. A missing or `null` `KEY` never matches.
+- Both roots must be lists, and every item must be an object; anything else is an error.
+
+```bash
+echo '[{"id":1,"name":"Ada"},{"id":2,"name":"Bob"},{"id":3,"name":"Cy"}]' > users.json
+printf 'id,team\n1,core\n2,web\n2,ops\n' > teams.csv
+
+data-transform join id --with teams.csv users.json --minify
+# → [{"id":1,"name":"Ada","team":"core"},{"id":2,"name":"Bob","team":"web"},{"id":2,"name":"Bob","team":"ops"}]
+
+data-transform join id -w teams.csv users.json --how left --minify
+# → [... same three rows ..., {"id":3,"name":"Cy"}]
+
+# Left side from stdin, then summarise per team
+cat users.json | data-transform join id -w teams.csv -f json \
+  | data-transform group-by team | data-transform count --minify
+# → {"core":1,"web":1,"ops":1}
+```
+
 ### Library use
 
 ```python
@@ -271,6 +299,8 @@ from data_transform import (
     tail_rows,
     group_rows,
     count_rows,
+    stats_rows,
+    join_rows,
 )
 
 yaml_text = convert('{"a":1}', "json", "yaml")
@@ -289,6 +319,8 @@ last = tail_rows([{"id": 1}, {"id": 2}, {"id": 3}], 1)
 by_team = group_rows([{"team": "a"}, {"team": "b"}], "team")
 total = count_rows([{"id": 1}, {"id": 2}])
 per_team = count_rows([{"team": "a"}, {"team": "a"}], "team")
+summary = stats_rows([{"ms": 120}, {"ms": 80}], "ms")
+joined = join_rows([{"id": 1, "name": "Ada"}], [{"id": "1", "team": "core"}], "id")
 ```
 
 ## NDJSON notes
diff --git a/pyproject.toml b/pyproject.toml
index c9c7363..91dc14a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "data-transform-cli"
-version = "0.6.0"
+version = "0.7.0"
 description = "CLI for YAML ↔ JSON ↔ CSV ↔ NDJSON transforms for pipelines and shell use"
 readme = "README.md"
 license = { text = "MIT" }
diff --git a/src/data_transform/__init__.py b/src/data_transform/__init__.py
index bea440a..646a3e9 100644
--- a/src/data_transform/__init__.py
+++ b/src/data_transform/__init__.py
@@ -9,6 +9,7 @@
     get_path,
     group_rows,
     head_rows,
+    join_rows,
     omit_keys,
     pick_keys,
     rename_keys,
@@ -43,6 +44,7 @@
     "group_rows",
     "count_rows",
     "stats_rows",
+    "join_rows",
 ]
 
-__version__ = "0.6.0"
+__version__ = "0.7.0"
diff --git a/src/data_transform/cli.py b/src/data_transform/cli.py
index 9c1036d..1022817 100644
--- a/src/data_transform/cli.py
+++ b/src/data_transform/cli.py
@@ -18,6 +18,7 @@
     get_path,
     group_rows,
     head_rows,
+    join_rows,
     omit_keys,
     pick_keys,
     rename_keys,
@@ -868,5 +869,86 @@ def stats_cmd(
         sys.exit(1)
 
 
+def _load_right(path: str, right_from: Optional[str]) -> object:
+    """Load and parse the right-hand file of a join (never stdin)."""
+    if path == "-":
+        raise ValueError("join: --with needs a file path; stdin is used for the left side")
+    file = Path(path)
+    if not file.is_file():
+        raise ValueError(f"join: right file not found: {path}")
+    text = file.read_text(encoding="utf-8")
+    src = resolve_from(text, path, right_from)
+    return parse(text, src)  # type: ignore[arg-type]
+
+
+@main.command("join")
+@click.argument("key")
+@click.argument("file", required=False)
+@click.option(
+    "-w",
+    "--with",
+    "right_file",
+    required=True,
+    metavar="RIGHT_FILE",
+    help="right-hand file to join against (not stdin)",
+)
+@click.option("-o", "--output", default=None, help="write to file instead of stdout")
+@click.option(
+    "-f",
+    "--from",
+    "from_format",
+    type=click.Choice(FORMAT_CHOICES, case_sensitive=False),
+    default=None,
+    help="left input format: json | yaml | csv | ndjson",
+)
+@click.option(
+    "--right-from",
+    "right_from",
+    type=click.Choice(FORMAT_CHOICES, case_sensitive=False),
+    default=None,
+    help="right file format: json | yaml | csv | ndjson",
+)
+@click.option(
+    "--how",
+    type=click.Choice(["inner", "left"], case_sensitive=False),
+    default="inner",
+    show_default=True,
+    help="inner keeps matched rows only; left also keeps unmatched left rows",
+)
+@click.option("--minify", is_flag=True, default=False, help="emit compact JSON")
+def join_cmd(
+    key: str,
+    file: Optional[str],
+    right_file: str,
+    output: Optional[str],
+    from_format: Optional[str],
+    right_from: Optional[str],
+    how: str,
+    minify: bool,
+) -> None:
+    """Join two arrays of objects on top-level KEY.
+
+    The left side is FILE (or stdin); the right side is --with. Rows keep
+    left order, one output row per match. Right keys are added after left
+    keys and win on collisions; KEY appears once. Values match like
+    group-by labels, so 1 and "1" are equal. Missing or null KEY never
+    matches.
+
+    Usage: data-transform join KEY --with RIGHT_FILE [file]
+    """
+    try:
+        right = _load_right(right_file, right_from)
+        text, name = load_input(file)
+        src = resolve_from(text, name, from_format)
+        left = parse(text, src)  # type: ignore[arg-type]
+        result = join_rows(left, right, key, how=how)
+        write_output(
+            serialize(result, "json", pretty=not minify, minify=minify), output
+        )
+    except Exception as err:  # noqa: BLE001
+        click.echo(f"error: {err}", err=True)
+        sys.exit(1)
+
+
 if __name__ == "__main__":
     main()
diff --git a/src/data_transform/shape.py b/src/data_transform/shape.py
index 6fba302..f934828 100644
--- a/src/data_transform/shape.py
+++ b/src/data_transform/shape.py
@@ -469,6 +469,80 @@ def stats_rows(data: Any, field: str) -> dict[str, Any]:
     )
 
 
+JOIN_HOWS = frozenset({"inner", "left"})
+
+
+def _join_side(data: Any, side: str) -> list[Mapping[Any, Any]]:
+    """Require a list of objects for one side of a join."""
+    if not isinstance(data, list):
+        raise ValueError(
+            f"join: {side} root must be a list (array) of objects, "
+            f"got {type(data).__name__}"
+        )
+    for index, item in enumerate(data):
+        if not isinstance(item, Mapping):
+            raise ValueError(
+                f"join: {side} item {index} must be an object, "
+                f"got {type(item).__name__}"
+            )
+    return data
+
+
+def _join_label(item: Mapping[Any, Any], key: str) -> str | None:
+    """Match label for KEY, or None when the key is missing or null."""
+    if key not in item or item[key] is None:
+        return None
+    return _group_label(item[key])
+
+
+def join_rows(
+    left: Any, right: Any, key: str, how: str = "inner"
+) -> list[dict[Any, Any]]:
+    """
+    Join two lists of objects on top-level KEY.
+
+    Values match with the same labels as :func:`group_rows`, so ``1`` and
+    ``"1"`` are equal (handy when one side is CSV). Missing or null KEY
+    values never match. Each output row starts with the left object's keys;
+    right keys are added after, and on a collision the right value wins.
+    KEY keeps the left value and appears once. Output follows left-row order,
+    and a left row with several right matches yields one row per match in
+    right-row order.
+
+    ``how="inner"`` keeps only matched pairs. ``how="left"`` also keeps
+    unmatched left rows as they are. Both roots must be lists of objects.
+    """
+    how_norm = how.lower() if isinstance(how, str) else how
+    if how_norm not in JOIN_HOWS:
+        raise ValueError(f"join: unsupported join type {how!r} (use inner or left)")
+    if not isinstance(key, str) or key == "":
+        raise ValueError("join: KEY must be a non-empty string")
+    left_rows = _join_side(left, "left")
+    right_rows = _join_side(right, "right")
+
+    index: dict[str, list[Mapping[Any, Any]]] = {}
+    for item in right_rows:
+        label = _join_label(item, key)
+        if label is not None:
+            index.setdefault(label, []).append(item)
+
+    out: list[dict[Any, Any]] = []
+    for item in left_rows:
+        label = _join_label(item, key)
+        matches = index.get(label, []) if label is not None else []
+        if not matches:
+            if how_norm == "left":
+                out.append(dict(item))
+            continue
+        for match in matches:
+            merged = dict(item)
+            for field, value in match.items():
+                if field != key:
+                    merged[field] = value
+            out.append(merged)
+    return out
+
+
 def _require_count(n: Any, command: str) -> int:
     """Raise ValueError unless n is a non-negative integer (not bool)."""
     if isinstance(n, bool) or not isinstance(n, int) or n < 0:
diff --git a/tests/test_join.py b/tests/test_join.py
new file mode 100644
index 0000000..75d2da3
--- /dev/null
+++ b/tests/test_join.py
@@ -0,0 +1,168 @@
+import json
+
+import pytest
+
+from data_transform import join_rows
+from test_cli import run_cli
+
+USERS = [
+    {"id": 1, "name": "Ada"},
+    {"id": 2, "name": "Bob"},
+    {"id": 3, "name": "Cy"},
+]
+ORDERS = [
+    {"id": 2, "total": 10},
+    {"id": 1, "total": 5},
+    {"id": 2, "total": 7},
+]
+
+
+def test_join_inner_keeps_left_order_and_emits_one_row_per_match():
+    assert join_rows(USERS, ORDERS, "id") == [
+        {"id": 1, "name": "Ada", "total": 5},
+        {"id": 2, "name": "Bob", "total": 10},
+        {"id": 2, "name": "Bob", "total": 7},
+    ]
+
+
+def test_join_left_keeps_unmatched_left_rows_unchanged():
+    out = join_rows(USERS, ORDERS, "id", how="left")
+    assert out[-1] == {"id": 3, "name": "Cy"}
+    assert len(out) == 4
+
+
+def test_join_right_wins_on_collision_and_key_appears_once():
+    left = [{"id": 1, "name": "old", "a": 1}]
+    right = [{"name": "new", "id": 1, "b": 2}]
+    out = join_rows(left, right, "id")
+    assert out == [{"id": 1, "name": "new", "a": 1, "b": 2}]
+    assert list(out[0]) == ["id", "name", "a", "b"]
+
+
+def test_join_missing_and_null_keys_never_match():
+    left = [{"id": None, "v": "l0"}, {"v": "l1"}, {"id": 1, "v": "l2"}]
+    right = [{"id": None, "r": 0}, {"r": 1}, {"id": 1, "r": 2}]
+    assert join_rows(left, right, "id") == [{"id": 1, "v": "l2", "r": 2}]
+    assert join_rows(left, right, "id", how="left") == [
+        {"id": None, "v": "l0"},
+        {"v": "l1"},
+        {"id": 1, "v": "l2", "r": 2},
+    ]
+
+
+def test_join_matches_numbers_and_numeric_strings_keeping_left_value():
+    out = join_rows([{"id": 1}], [{"id": "1", "x": "y"}], "id")
+    assert out == [{"id": 1, "x": "y"}]
+
+
+def test_join_empty_lists():
+    assert join_rows([], ORDERS, "id") == []
+    assert join_rows(USERS, [], "id") == []
+    assert join_rows(USERS, [], "id", how="LEFT") == USERS
+
+
+def test_join_does_not_mutate_inputs():
+    left = [{"id": 1, "a": 1}]
+    right = [{"id": 1, "a": 2}]
+    join_rows(left, right, "id")
+    assert left == [{"id": 1, "a": 1}] and right == [{"id": 1, "a": 2}]
+
+
+@pytest.mark.parametrize(
+    "left,right,message",
+    [
+        ({"id": 1}, [], "left root must be a list"),
+        ([], "x", "right root must be a list"),
+        ([{"id": 1}, 2], [], "left item 1 must be an object"),
+        ([], [None], "right item 0 must be an object"),
+    ],
+)
+def test_join_rejects_non_list_roots_and_non_object_items(left, right, message):
+    with pytest.raises(ValueError, match=message):
+        join_rows(left, right, "id")
+
+
+def test_join_rejects_bad_how_and_empty_key():
+    with pytest.raises(ValueError, match="unsupported join type"):
+        join_rows([], [], "id", how="outer")
+    with pytest.raises(ValueError, match="KEY must be"):
+        join_rows([], [], "")
+
+
+# CLI
+
+
+@pytest.fixture
+def right_csv(tmp_path):
+    path = tmp_path / "teams.csv"
+    path.write_text("id,team\n1,core\n2,web\n2,ops\n", encoding="utf-8")
+    return path
+
+
+def test_cli_join_inner_json_left_from_stdin_csv_right(right_csv):
+    code, stdout, stderr = run_cli(
+        ["join", "id", "--with", str(right_csv), "-f", "json", "--minify"],
+        json.dumps(USERS),
+    )
+    assert code == 0, stderr
+    assert json.loads(stdout) == [
+        {"id": 1, "name": "Ada", "team": "core"},
+        {"id": 2, "name": "Bob", "team": "web"},
+        {"id": 2, "name": "Bob", "team": "ops"},
+    ]
+
+
+def test_cli_join_left_with_file_arg_and_yaml_right(tmp_path):
+    left = tmp_path / "users.json"
+    left.write_text(json.dumps(USERS), encoding="utf-8")
+    right = tmp_path / "roles.yaml"
+    right.write_text("- id: 3\n  role: admin\n", encoding="utf-8")
+    code, stdout, stderr = run_cli(
+        ["join", "id", "-w", str(right), str(left), "--how", "left"]
+    )
+    assert code == 0, stderr
+    assert json.loads(stdout) == [
+        {"id": 1, "name": "Ada"},
+        {"id": 2, "name": "Bob"},
+        {"id": 3, "name": "Cy", "role": "admin"},
+    ]
+
+
+def test_cli_join_right_from_forces_format_and_writes_output(tmp_path):
+    right = tmp_path / "extra.txt"
+    right.write_text('{"id":1,"x":true}\n{"id":9,"x":false}\n', encoding="utf-8")
+    out = tmp_path / "out.json"
+    code, _, stderr = run_cli(
+        [
+            "join", "id", "-w", str(right), "--right-from", "ndjson",
+            "-f", "json", "-o", str(out), "--minify",
+        ],
+        '[{"id":1},{"id":2}]',
+    )
+    assert code == 0, stderr
+    assert json.loads(out.read_text(encoding="utf-8")) == [{"id": 1, "x": True}]
+
+
+def test_cli_join_requires_with_option():
+    code, _, stderr = run_cli(["join", "id", "-f", "json"], "[]")
+    assert code != 0
+    assert "--with" in stderr
+
+
+def test_cli_join_rejects_stdin_and_missing_right_file(tmp_path):
+    code, _, stderr = run_cli(["join", "id", "-w", "-", "-f", "json"], "[]")
+    assert code == 1
+    assert "needs a file path" in stderr
+    code, _, stderr = run_cli(
+        ["join", "id", "-w", str(tmp_path / "nope.json"), "-f", "json"], "[]"
+    )
+    assert code == 1
+    assert "right file not found" in stderr
+
+
+def test_cli_join_reports_non_list_root(right_csv):
+    code, _, stderr = run_cli(
+        ["join", "id", "-w", str(right_csv), "-f", "json"], '{"id":1}'
+    )
+    assert code == 1
+    assert "left root must be a list" in stderr
diff --git a/tests/test_shape.py b/tests/test_shape.py
index 396c395..7e1172e 100644
--- a/tests/test_shape.py
+++ b/tests/test_shape.py
@@ -166,7 +166,7 @@ def test_unique_requires_list():
 
 
 def test_public_exports_include_reshape_helpers():
-    assert data_transform.__version__ == "0.6.0"
+    assert data_transform.__version__ == "0.7.0"
     for name in (
         "rename_keys",
         "flatten_object",
@@ -175,6 +175,7 @@ def test_public_exports_include_reshape_helpers():
         "group_rows",
         "count_rows",
         "stats_rows",
+        "join_rows",
         "unflatten_object",
     ):
         assert getattr(data_transform, name) is not None