diff --git a/README.md b/README.md
index f0673e7..3a3bed1 100644
--- a/README.md
+++ b/README.md
@@ -30,20 +30,23 @@ After install, the `data-transform` console script is on your `PATH`.
 | `omit` | Drop listed top-level keys |
 | `rename` | Rename top-level keys (`OLD NEW` pairs) |
 | `flatten` | Flatten nested objects into joined keys |
+| `unflatten` | Expand joined keys back into nested objects |
 | `filter` | Keep array items matching KEY OP VALUE |
 | `sort` | Sort array of objects by top-level KEY |
 | `unique` | Deduplicate array (by KEY or whole item) |
 | `head` | First N items of a list |
 | `tail` | Last N items of a list |
+| `group-by` | Group array of objects by top-level KEY |
+| `count` | Count items, per KEY value, or per group |
 
 ### 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`, `filter`, `sort`, `unique`, `head`, `tail`)
+- `--minify` — compact JSON output (`to-json`, `select`, `pick`, `omit`, `rename`, `flatten`, `unflatten`, `filter`, `sort`, `unique`, `head`, `tail`, `group-by`, `count`)
 - `--desc` — reverse sort order (`sort` only)
-- `--sep <text>` — separator between nested keys (`flatten` only, default `.`)
+- `--sep <text>` — separator between nested keys (`flatten` and `unflatten`, default `.`)
 
 Format is taken from `--from`, else the file extension (`.ndjson` / `.jsonl` → ndjson), else simple content heuristics.
 
@@ -129,6 +132,23 @@ echo '{"user":{"name":"Ada"}}' \
 # → {"user_name":"Ada"}
 ```
 
+### Unflatten
+
+`unflatten` is the inverse of `flatten`: it splits top-level keys on `--sep` (default `.`) and rebuilds nested objects. Like `flatten`, it maps over a list root and returns non-object roots and items unchanged. It does not rebuild lists, so a segment like `0` stays an object key. Keys are split literally, so an empty segment (`a..b`) becomes an empty key. If an object value and joined keys write to the same branch, they are merged. If a path runs into a non-object value, the later key wins. `unflatten` undoes `flatten` exactly as long as the original keys don't contain the separator.
+
+```bash
+echo '{"user.name":"Ada","user.addr.city":"London","tags":["a","b"]}' \
+  | data-transform unflatten -f json --minify
+# → {"user":{"name":"Ada","addr":{"city":"London"}},"tags":["a","b"]}
+
+echo '[{"user_name":"Ada"},{"user_name":"Bob"}]' \
+  | data-transform unflatten --sep _ -f json --minify
+# → [{"user":{"name":"Ada"}},{"user":{"name":"Bob"}}]
+
+# Round trip
+data-transform flatten nested.json | data-transform unflatten
+```
+
 ### Filter, sort, unique
 
 These commands require a **list** root (JSON/YAML/CSV/NDJSON array). Non-object
@@ -185,6 +205,38 @@ echo '[1,2,3]' | data-transform head 0 -f json --minify
 # → []
 ```
 
+### Group-by and count
+
+`group-by KEY` requires a **list** root. It returns an object that maps each value of top-level `KEY` to the list of items that have it. Groups appear in the order they are first seen, and items keep their input order. Group labels are the value itself for strings; any other value becomes compact JSON text (`1` → `"1"`, `true` → `"true"`, `null` → `"null"`). This means `1` and `"1"` end up in the same group, which suits CSV input where every value is a string. Non-object items and objects without `KEY` are dropped.
+
+`count` counts records:
+
+- `count` on a list prints its length as a number.
+- `count KEY` on a list prints an object of per-value counts, using the same labels and dropping rules as `group-by`.
+- `count` on the output of `group-by` (an object whose values are all lists) prints the size of each group.
+
+Any other root is an error.
+
+```bash
+echo '[{"team":"a","id":1},{"team":"b","id":2},{"team":"a","id":3},{"id":4}]' \
+  | data-transform group-by team -f json --minify
+# → {"a":[{"team":"a","id":1},{"team":"a","id":3}],"b":[{"team":"b","id":2}]}
+
+echo '[{"id":1},{"id":2},{"id":3}]' | data-transform count -f json
+# → 3
+
+echo '[{"team":"a"},{"team":"b"},{"team":"a"}]' \
+  | data-transform count team -f json --minify
+# → {"a":2,"b":1}
+
+# Count per group from group-by output
+data-transform group-by team people.json | data-transform count --minify
+# → {"a":2,"b":1}
+
+# Works with NDJSON / CSV too
+data-transform count status events.ndjson
+```
+
 ### Library use
 
 ```python
@@ -197,11 +249,14 @@ from data_transform import (
     omit_keys,
     rename_keys,
     flatten_object,
+    unflatten_object,
     filter_rows,
     sort_rows,
     unique_rows,
     head_rows,
     tail_rows,
+    group_rows,
+    count_rows,
 )
 
 yaml_text = convert('{"a":1}', "json", "yaml")
@@ -211,11 +266,15 @@ name = get_path({"user": {"name": "Ada"}}, "user.name")
 slim = pick_keys({"id": 1, "name": "a", "note": "x"}, ["id", "name"])
 renamed = rename_keys({"id": 1, "name": "a"}, {"name": "title"})
 flat = flatten_object({"user": {"name": "Ada"}, "tags": ["x"]})
+nested = unflatten_object({"user.name": "Ada"})
 adults = filter_rows([{"age": 36}, {"age": 22}], "age", "ge", "30")
 ordered = sort_rows([{"id": 2}, {"id": 1}], "id")
 deduped = unique_rows([{"id": 1}, {"id": 1}], "id")
 first = head_rows([{"id": 1}, {"id": 2}, {"id": 3}], 2)
 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")
 ```
 
 ## NDJSON notes
diff --git a/pyproject.toml b/pyproject.toml
index 3541611..ff033b5 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "data-transform-cli"
-version = "0.4.0"
+version = "0.5.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 0814627..4404d9e 100644
--- a/src/data_transform/__init__.py
+++ b/src/data_transform/__init__.py
@@ -3,15 +3,18 @@
 from .convert import convert, detect_format, infer_format, parse, serialize
 from .csv_io import escape_csv_field, parse_csv, parse_csv_rows, stringify_csv
 from .shape import (
+    count_rows,
     filter_rows,
     flatten_object,
     get_path,
+    group_rows,
     head_rows,
     omit_keys,
     pick_keys,
     rename_keys,
     sort_rows,
     tail_rows,
+    unflatten_object,
     unique_rows,
 )
 
@@ -30,11 +33,14 @@
     "omit_keys",
     "rename_keys",
     "flatten_object",
+    "unflatten_object",
     "filter_rows",
     "sort_rows",
     "unique_rows",
     "head_rows",
     "tail_rows",
+    "group_rows",
+    "count_rows",
 ]
 
-__version__ = "0.4.0"
+__version__ = "0.5.0"
diff --git a/src/data_transform/cli.py b/src/data_transform/cli.py
index 534c5f2..435e10e 100644
--- a/src/data_transform/cli.py
+++ b/src/data_transform/cli.py
@@ -11,15 +11,18 @@
 from . import __version__
 from .convert import convert, detect_format, infer_format, parse, serialize
 from .shape import (
+    count_rows,
     filter_rows,
     flatten_object,
     get_path,
+    group_rows,
     head_rows,
     omit_keys,
     pick_keys,
     rename_keys,
     sort_rows,
     tail_rows,
+    unflatten_object,
     unique_rows,
 )
 
@@ -601,6 +604,51 @@ def flatten_cmd(
         sys.exit(1)
 
 
+@main.command("unflatten")
+@click.argument("file", required=False)
+@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="input format: json | yaml | csv | ndjson",
+)
+@click.option("--minify", is_flag=True, default=False, help="emit compact JSON")
+@click.option(
+    "--sep",
+    default=".",
+    show_default=True,
+    help="separator between nested keys",
+)
+def unflatten_cmd(
+    file: Optional[str],
+    output: Optional[str],
+    from_format: Optional[str],
+    minify: bool,
+    sep: str,
+) -> None:
+    """Expand separator-joined top-level keys into nested objects.
+
+    The inverse of flatten. Maps over a list root. Lists are not rebuilt
+    from numeric segments. A non-object, non-list root is unchanged.
+
+    Usage: data-transform unflatten [file]
+    """
+    try:
+        text, name = load_input(file)
+        src = resolve_from(text, name, from_format)
+        data = parse(text, src)  # type: ignore[arg-type]
+        result = unflatten_object(data, sep)
+        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)
+
+
 def _run_take(
     command: str,
     tokens: Sequence[str],
@@ -688,5 +736,95 @@ def tail_cmd(
     _run_take("tail", tokens, output, from_format, minify)
 
 
+@main.command("group-by")
+@click.argument("tokens", nargs=-1, required=True)
+@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="input format: json | yaml | csv | ndjson",
+)
+@click.option("--minify", is_flag=True, default=False, help="emit compact JSON")
+def group_by_cmd(
+    tokens: tuple[str, ...],
+    output: Optional[str],
+    from_format: Optional[str],
+    minify: bool,
+) -> None:
+    """Group an array of objects by top-level KEY.
+
+    Emits an object of KEY value → list of items, in first-seen order.
+    Non-string values become JSON text labels. Non-object items and items
+    missing KEY are dropped. Root must be a list.
+
+    Usage: data-transform group-by KEY [file]
+    """
+    try:
+        parts, input_file = _split_tokens_and_file(tokens)
+        if len(parts) != 1:
+            raise ValueError("Usage: group-by KEY [file]")
+        key = parts[0]
+
+        text, name = load_input(input_file)
+        src = resolve_from(text, name, from_format)
+        data = parse(text, src)  # type: ignore[arg-type]
+        result = group_rows(data, key)
+        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)
+
+
+@main.command("count")
+@click.argument("tokens", nargs=-1, required=False)
+@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="input format: json | yaml | csv | ndjson",
+)
+@click.option("--minify", is_flag=True, default=False, help="emit compact JSON")
+def count_cmd(
+    tokens: tuple[str, ...],
+    output: Optional[str],
+    from_format: Optional[str],
+    minify: bool,
+) -> None:
+    """Count records in a list, per KEY value, or per group.
+
+    Without KEY, a list root prints its length and group-by output prints
+    the size of each group. With KEY, prints an object of KEY value → count
+    (same labels and dropping rules as group-by).
+
+    Usage: data-transform count [KEY] [file]
+    """
+    try:
+        parts, input_file = _split_tokens_and_file(tokens)
+        key: Optional[str] = None
+        if len(parts) == 1:
+            key = parts[0]
+        elif len(parts) > 1:
+            raise ValueError("Usage: count [KEY] [file]")
+
+        text, name = load_input(input_file)
+        src = resolve_from(text, name, from_format)
+        data = parse(text, src)  # type: ignore[arg-type]
+        result = count_rows(data, key)
+        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 25285cf..9b13292 100644
--- a/src/data_transform/shape.py
+++ b/src/data_transform/shape.py
@@ -155,6 +155,58 @@ def _flatten_mapping(
             out[path] = value
 
 
+def unflatten_object(data: Any, sep: str = ".") -> Any:
+    """
+    Expand ``sep``-joined top-level keys back into nested mappings.
+
+    The inverse of :func:`flatten_object`. A list root is mapped item-wise;
+    non-mapping items are left unchanged. A non-mapping, non-list root is
+    returned unchanged. Only top-level string keys are split; values are kept
+    as-is, and no list indices are created (segments such as ``"0"`` stay
+    string keys). Keys are split literally, so empty segments become empty
+    keys. Object values merge with branches built from joined keys; when a
+    path collides with a non-object value, the later key wins. An empty
+    separator raises ValueError.
+    """
+    if not isinstance(sep, str) or sep == "":
+        raise ValueError("unflatten: separator must be a non-empty string")
+    if isinstance(data, list):
+        return [_unflatten_one(item, sep) for item in data]
+    return _unflatten_one(data, sep)
+
+
+def _unflatten_one(item: Any, sep: str) -> Any:
+    if not isinstance(item, Mapping):
+        return item
+    nested: dict[Any, Any] = {}
+    for key, value in item.items():
+        parts = key.split(sep) if isinstance(key, str) else [key]
+        current = nested
+        for part in parts[:-1]:
+            child = current.get(part)
+            if not isinstance(child, dict):
+                # Missing or a non-object value: the later key wins.
+                child = {}
+                current[part] = child
+            current = child
+        _merge_value(current, parts[-1], value)
+    return nested
+
+
+def _merge_value(target: dict[Any, Any], key: Any, value: Any) -> None:
+    """Set target[key]; objects merge into an existing object branch."""
+    if isinstance(value, Mapping):
+        branch = target.get(key)
+        if not isinstance(branch, dict):
+            branch = {}
+            target[key] = branch
+        # Copy object values so later keys never mutate the input.
+        for sub_key, sub_value in value.items():
+            _merge_value(branch, sub_key, sub_value)
+    else:
+        target[key] = value
+
+
 def _require_list(data: Any, command: str) -> list[Any]:
     """Raise ValueError unless data is a list; return it typed."""
     if not isinstance(data, list):
@@ -311,6 +363,69 @@ def unique_rows(data: Any, key: str | None = None) -> list[Any]:
     return out
 
 
+def _group_label(value: Any) -> str:
+    """Turn a group value into an object key: strings as-is, else compact JSON."""
+    if isinstance(value, str):
+        return value
+    try:
+        return json.dumps(value, sort_keys=True, separators=(",", ":"), default=str)
+    except (TypeError, ValueError):
+        return str(value)
+
+
+def group_rows(data: Any, key: str) -> dict[str, list[Any]]:
+    """
+    Group an array of objects by top-level KEY.
+
+    Returns an object mapping each group label to the list of items in that
+    group. Groups keep first-seen order and items keep input order. Labels
+    are the value itself for strings and compact JSON otherwise (``1`` →
+    ``"1"``, ``true`` → ``"true"``, ``null`` → ``"null"``), so ``1`` and
+    ``"1"`` share a group. Non-object items and objects without KEY are
+    dropped. Root must be a list.
+    """
+    return _group(_require_list(data, "group-by"), key)
+
+
+def _group(rows: list[Any], key: str) -> dict[str, list[Any]]:
+    groups: dict[str, list[Any]] = {}
+    for item in rows:
+        if not isinstance(item, Mapping) or key not in item:
+            continue
+        groups.setdefault(_group_label(item[key]), []).append(item)
+    return groups
+
+
+def _is_grouped(data: Any) -> bool:
+    """True for a mapping whose values are all lists (``group-by`` output)."""
+    return isinstance(data, Mapping) and all(
+        isinstance(v, list) for v in data.values()
+    )
+
+
+def count_rows(data: Any, key: str | None = None) -> int | dict[str, int]:
+    """
+    Count records.
+
+    Without KEY, a list root returns its length, and an object whose values
+    are all lists (the output of :func:`group_rows`) returns the length of
+    each list under the same group label. With KEY, a list root returns an
+    object of per-value counts using the same labels and dropping rules as
+    :func:`group_rows`. Any other root raises ValueError.
+    """
+    if key is None:
+        if isinstance(data, list):
+            return len(data)
+        if _is_grouped(data):
+            return {str(label): len(items) for label, items in data.items()}
+        raise ValueError(
+            "count: root must be a list (array) or an object of lists "
+            f"(group-by output), got {type(data).__name__}"
+        )
+    rows = _require_list(data, "count")
+    return {label: len(items) for label, items in _group(rows, key).items()}
+
+
 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_cli.py b/tests/test_cli.py
index b73887f..bdc8fd7 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -376,3 +376,107 @@ def test_head_tail_clamp_zero_and_errors(tmp_path):
     assert code == 1
     assert "error:" in stderr
     assert "Usage:" in stderr
+
+
+def test_group_by_from_stdin_and_file():
+    code, stdout, stderr = run_cli(
+        ["group-by", "team", "-f", "json", "--minify"],
+        '[{"team":"a","id":1},{"team":"b","id":2},{"team":"a","id":3},{"id":4}]',
+    )
+    assert code == 0, stderr
+    assert json.loads(stdout) == {
+        "a": [{"team": "a", "id": 1}, {"team": "a", "id": 3}],
+        "b": [{"team": "b", "id": 2}],
+    }
+
+    code, stdout, stderr = run_cli(
+        ["group-by", "id", str(FIXTURE / "sample.json"), "--minify"]
+    )
+    assert code == 0, stderr
+    assert list(json.loads(stdout)) == ["1", "2"]
+
+
+def test_group_by_errors():
+    code, stdout, stderr = run_cli(["group-by", "k", "-f", "json"], '{"k":1}')
+    assert code == 1
+    assert stdout == ""
+    assert "error:" in stderr
+    assert "list" in stderr.lower()
+
+    code, stdout, stderr = run_cli(["group-by", "a", "b", "-f", "json"], "[]")
+    assert code == 1
+    assert "Usage:" in stderr
+
+
+def test_count_total_per_key_and_piped_groups():
+    rows = '[{"t":"a"},{"t":"b"},{"t":"a"}]'
+    code, stdout, stderr = run_cli(["count", "-f", "json"], rows)
+    assert code == 0, stderr
+    assert json.loads(stdout) == 3
+
+    code, stdout, stderr = run_cli(["count", "t", "-f", "json", "--minify"], rows)
+    assert code == 0, stderr
+    assert stdout.strip() == '{"a":2,"b":1}'
+
+    code, grouped, stderr = run_cli(["group-by", "t", "-f", "json"], rows)
+    assert code == 0, stderr
+    code, stdout, stderr = run_cli(["count", "-f", "json", "--minify"], grouped)
+    assert code == 0, stderr
+    assert json.loads(stdout) == {"a": 2, "b": 1}
+
+    code, stdout, stderr = run_cli(["count", str(FIXTURE / "sample.json")])
+    assert code == 0, stderr
+    assert json.loads(stdout) == 2
+
+    code, stdout, stderr = run_cli(
+        ["count", "-f", "ndjson"], '{"id":1}\n{"id":2}\n'
+    )
+    assert code == 0, stderr
+    assert json.loads(stdout) == 2
+
+
+def test_count_errors():
+    code, stdout, stderr = run_cli(["count", "-f", "json"], '{"a":1}')
+    assert code == 1
+    assert "error:" in stderr
+    assert "list" in stderr.lower()
+
+    code, stdout, stderr = run_cli(["count", "a", "b", "-f", "json"], "[]")
+    assert code == 1
+    assert "Usage:" in stderr
+
+
+def test_unflatten_nested_custom_sep_and_round_trip(tmp_path):
+    code, stdout, stderr = run_cli(
+        ["unflatten", "-f", "json", "--minify"],
+        '{"user.name":"Ada","user.addr.city":"London","tags":["a","b"]}',
+    )
+    assert code == 0, stderr
+    assert json.loads(stdout) == {
+        "user": {"name": "Ada", "addr": {"city": "London"}},
+        "tags": ["a", "b"],
+    }
+
+    code, stdout, stderr = run_cli(
+        ["unflatten", "--sep", "_", "-f", "json", "--minify"],
+        '[{"user_name":"Ada"},3]',
+    )
+    assert code == 0, stderr
+    assert json.loads(stdout) == [{"user": {"name": "Ada"}}, 3]
+
+    original = {"user": {"name": "Ada", "addr": {"city": "London"}}, "id": 1}
+    flat_file = tmp_path / "flat.json"
+    code, stdout, stderr = run_cli(
+        ["flatten", "-f", "json", "-o", str(flat_file)], json.dumps(original)
+    )
+    assert code == 0, stderr
+    code, stdout, stderr = run_cli(["unflatten", str(flat_file)])
+    assert code == 0, stderr
+    assert json.loads(stdout) == original
+
+    code, stdout, stderr = run_cli(
+        ["unflatten", "--sep", "", "-f", "json"], '{"a.b":1}'
+    )
+    assert code == 1
+    assert "error:" in stderr
+    assert "separator" in stderr
diff --git a/tests/test_shape.py b/tests/test_shape.py
index 8333c19..3b869ff 100644
--- a/tests/test_shape.py
+++ b/tests/test_shape.py
@@ -2,15 +2,18 @@
 
 import data_transform
 from data_transform.shape import (
+    count_rows,
     filter_rows,
     flatten_object,
     get_path,
+    group_rows,
     head_rows,
     omit_keys,
     pick_keys,
     rename_keys,
     sort_rows,
     tail_rows,
+    unflatten_object,
     unique_rows,
 )
 
@@ -163,8 +166,16 @@ def test_unique_requires_list():
 
 
 def test_public_exports_include_reshape_helpers():
-    assert data_transform.__version__ == "0.4.0"
-    for name in ("rename_keys", "flatten_object", "head_rows", "tail_rows"):
+    assert data_transform.__version__ == "0.5.0"
+    for name in (
+        "rename_keys",
+        "flatten_object",
+        "head_rows",
+        "tail_rows",
+        "group_rows",
+        "count_rows",
+        "unflatten_object",
+    ):
         assert getattr(data_transform, name) is not None
 
 
@@ -281,3 +292,104 @@ def test_head_and_tail_reject_bad_count_and_non_list():
             fn([1, 2], 1.5)  # type: ignore[arg-type]
         with pytest.raises(ValueError, match=label):
             fn("abc", 1)
+
+
+def test_group_rows_first_seen_order_and_drops():
+    data = [
+        {"team": "b", "id": 1},
+        {"team": "a", "id": 2},
+        {"id": 3},
+        "x",
+        {"team": "b", "id": 4},
+    ]
+    result = group_rows(data, "team")
+    assert list(result) == ["b", "a"]
+    assert result == {
+        "b": [{"team": "b", "id": 1}, {"team": "b", "id": 4}],
+        "a": [{"team": "a", "id": 2}],
+    }
+    assert result["b"][0] is data[0]
+    assert group_rows([], "team") == {}
+
+
+def test_group_rows_labels_non_string_values():
+    data = [
+        {"k": 1},
+        {"k": "1"},
+        {"k": True},
+        {"k": None},
+        {"k": [1, 2]},
+        {"k": {"b": 1, "a": 2}},
+    ]
+    result = group_rows(data, "k")
+    assert list(result) == ["1", "true", "null", "[1,2]", '{"a":2,"b":1}']
+    assert result["1"] == [{"k": 1}, {"k": "1"}]
+
+
+def test_group_rows_requires_list():
+    with pytest.raises(ValueError, match="group-by: root must be a list"):
+        group_rows({"k": 1}, "k")
+
+
+def test_count_rows_total_per_key_and_grouped():
+    data = [{"t": "a"}, {"t": "b"}, {"t": "a"}, {"n": 1}, 7]
+    assert count_rows(data) == 5
+    assert count_rows([]) == 0
+    assert count_rows(data, "t") == {"a": 2, "b": 1}
+    assert count_rows(group_rows(data, "t")) == {"a": 2, "b": 1}
+    assert count_rows({}) == {}
+
+
+def test_count_rows_rejects_other_roots():
+    with pytest.raises(ValueError, match="count: root must be a list"):
+        count_rows({"a": 1})
+    with pytest.raises(ValueError, match="count: root must be a list"):
+        count_rows("abc")
+    with pytest.raises(ValueError, match="count: root must be a list"):
+        count_rows({"a": [1]}, "t")
+
+
+def test_unflatten_nested_keys_and_list_root():
+    tags = ["a", "b"]
+    data = {"user.name": "Ada", "user.addr.city": "London", "tags": tags}
+    result = unflatten_object(data)
+    assert result == {
+        "user": {"name": "Ada", "addr": {"city": "London"}},
+        "tags": tags,
+    }
+    assert result["tags"] is tags
+    assert unflatten_object([{"a_b": 1}, 2, "x"], sep="_") == [{"a": {"b": 1}}, 2, "x"]
+    assert unflatten_object({"items.0": "x"}) == {"items": {"0": "x"}}
+    assert unflatten_object(42) == 42
+    assert unflatten_object(None) is None
+    assert unflatten_object({}) == {}
+    assert unflatten_object([]) == []
+
+
+def test_unflatten_round_trips_flatten():
+    data = {
+        "user": {"name": "Ada", "addr": {"city": "London", "zip": None}},
+        "meta": {},
+        "tags": [{"id": 1}],
+        "ok": True,
+    }
+    assert unflatten_object(flatten_object(data)) == data
+    assert unflatten_object(flatten_object(data, sep="__"), sep="__") == data
+
+
+def test_unflatten_collisions_merge_objects_and_later_scalar_wins():
+    assert unflatten_object({"a": 1, "a.b": 2}) == {"a": {"b": 2}}
+    assert unflatten_object({"a.b": 2, "a": 1}) == {"a": 1}
+    assert unflatten_object({"a.b": 1, "a": {"c": 2}}) == {"a": {"b": 1, "c": 2}}
+    inner = {"c": 2}
+    data = {"a": inner, "a.b": 1}
+    assert unflatten_object(data) == {"a": {"c": 2, "b": 1}}
+    assert inner == {"c": 2}
+    assert unflatten_object({"a..b": 1, 1: "x"}) == {"a": {"": {"b": 1}}, 1: "x"}
+
+
+def test_unflatten_rejects_empty_separator():
+    with pytest.raises(ValueError, match="separator"):
+        unflatten_object({"a.b": 1}, sep="")
+    with pytest.raises(ValueError, match="separator"):
+        unflatten_object({"a": 1}, sep=None)  # type: ignore[arg-type]