# v2.17 Rows Stream TODO

Private planning file. Do not commit.

## Goal

Add an architecturally clean, high-performance, expression-level stream mode:

```jetro
$.rows().reverse().distinct_by($.id).take(100).map({id: $.id, v: $.v})
```

Rules:

- Keep `--ndjson` as the source-format switch.
- Without `$.rows()`, NDJSON keeps current row-local behavior.
- With root-level `$.rows()`, the expression is planned as one whole-stream query.
- Normal JSON can also use `$.rows()`:
  - array input streams array elements
  - object/scalar input is one row unless a better error policy is chosen
- No query-shape-specific fusions.
- Reuse existing builtin metadata, demand model, direct byte/tape writers, adaptive distinct state, and row drivers.
- Keep the architecture open to future stream operators: reducers, group_by, windows, joins, multi-source, schema hints.

## Target UX

Row-local NDJSON mode stays unchanged:

```bash
jetrocli --ndjson -i events.ndjson '$.id'
```

Whole-stream NDJSON mode:

```bash
jetrocli --ndjson -i events.ndjson \
  '$.rows().reverse().distinct_by($.id).take(100).map({id: $.id, v: $.v})'
```

Normal JSON array mode should use the same expression shape:

```jetro
$.rows().distinct_by($.id).map($.name)
```

## Design Principles

- SOLID:
  - sources own IO/access strategy
  - stages own stream semantics
  - planners own lowering and legality
  - executors execute plans, not ad hoc query patterns
- No duplicate distinct, projection, filter, or writer logic.
- Planner facts come from builtin registry metadata where possible.
- Fast paths must be generic and algorithmic.
- Fallback must preserve correctness and lose only performance.
- Late projection should remain legal only for pure one-to-one projections after row selection.
- Materialization only at barriers, fallback boundaries, mutation/update semantics, or final conversion.

## Phase 1: Stream IR

- [x] Add a small stream IR independent from NDJSON helper functions.
- Proposed types:

```rust
enum RowStreamSourceKind {
    DocumentRows,
    NdjsonRows,
}

enum RowStreamDirection {
    Forward,
    Reverse,
}

struct RowStreamPlan {
    source: RowStreamSourceKind,
    direction: RowStreamDirection,
    stages: Vec<RowStreamStage>,
    output: RowStreamOutput,
}

enum RowStreamStage {
    Filter(RowPredicatePlan),
    DistinctBy(RowKeyPlan),
    Take(usize),
    Map(RowProjectorPlan),
}
```

- [x] Keep this small for first version.
- [x] Do not tie the IR to NDJSON-only implementation details.

## Phase 2: `rows()` Builtin Metadata

- [x] Add `rows()` to builtin definitions.
- [x] Mark it as a source-lifting builtin rather than a regular array method.
- [x] Metadata should describe:
  - arity: zero
  - stream-source behavior
  - barrier behavior: source boundary
  - demand behavior: forwards stream demand to source
  - lowering shape: `RowsSource`
- [x] Keep parser/planner decisions based on registry facts where possible.

## Phase 3: Parser and Planner Detection

- [x] Detect root-level:

```jetro
$.rows()...
```

- [x] First version supports only root/source boundary `rows()`.
- [ ] Decide later whether nested `$.books.rows()` is allowed.
- [x] In normal JSON mode:
  - lower to `DocumentRows`
- [x] In NDJSON mode:
  - lower to `NdjsonRows`
- [x] If `$.rows()` is used in a context with no stream-capable source, return a clear planning error.

## Phase 4: Generic Stream Lowerer

- [x] Convert the chain after `rows()` into `RowStreamPlan`.
- [x] First supported operators:
  - `reverse()`
  - `filter(pred)`
  - `distinct_by(key)`
  - `take(n)` / `first()`
  - `map(expr)`
- [x] Enforce ordering semantics exactly:
  - `reverse().distinct_by($.id).take(n)` means latest-per-key early stop on reverse-capable sources
  - `filter(...).distinct_by(...).take(n)` filters before distinct
  - `distinct_by(...).map(...)` projects only retained rows
  - `map(...).distinct_by(...)` key observes mapped row unless planner proves legal transform reordering
- [x] Unsupported stream chains must fail clearly or fall back conservatively.

## Phase 5: Stream Demand

- [x] Add stream-level demand annotations:
  - direction
  - retained-row limit
  - key expression demand
  - predicate demand
  - projection demand
- `take(n)` after `distinct_by` means stop after `n` retained rows.
- `take(n)` before `distinct_by` means only first `n` stream rows are considered.
- `reverse()` changes source order, not order inside row values.
- Preserve late projection when legal.

## Phase 6: Executor Traits

- Introduce minimal traits:

```rust
trait RowStreamSource {
    fn next_row(&mut self) -> Result<Option<RowHandle>, JetroEngineError>;
}

trait RowPredicate {
    fn test(&mut self, row: &RowHandle) -> Result<bool, JetroEngineError>;
}

trait RowKey {
    fn key(&mut self, row: &RowHandle) -> Result<RowKeyBytes, JetroEngineError>;
}

trait RowProjector {
    fn write(&mut self, row: &RowHandle, out: &mut dyn Write) -> Result<(), JetroEngineError>;
}
```

- Keep traits lean to avoid abstraction overhead on hot paths.
- Source implementations:
  - NDJSON forward file/reader source
  - NDJSON reverse file source
  - [x] document array/value source
- Stage implementations are shared across sources.

## Phase 7: Reuse Existing Fast Paths

- [ ] Move reusable components behind generic stream modules:
- [x] adaptive exact/Bloom/Cuckoo distinct state
- [x] direct row key extraction for root-path keys in `$.rows()` executor
- [x] direct row projection writer for final raw-row maps in `$.rows()` executor
  - stats counters
- [x] Existing `ndjson_rev` should call shared components rather than duplicate distinct key canonicalization.
- [x] Avoid adding hardcoded chain kernels.
- [x] Use direct byte/tape path for stream filters and raw retained-row output when source row is raw NDJSON bytes.
- [x] Use existing `Val`/view path for document rows.

## Phase 8: NDJSON Execution

- [x] Dispatch rules:

```text
--ndjson + no rows()  => current row-local NDJSON execution
--ndjson + rows()     => stream executor
```

- [x] `rows().reverse()` on regular file uses reverse NDJSON driver.
- [x] `rows().reverse()` on non-seekable reader:
  - either clear unsupported error
  - or materializing fallback only if explicitly allowed later
- [x] `rows().take(n)` should stop after `n` emitted/retained rows based on chain position.
- [x] `rows().distinct_by(k).take(n)` should stop after `n` retained unique rows.

## Phase 9: Document Execution

- [x] Normal JSON document:
  - array: stream array elements
  - object/scalar: one row
- [x] Use same `RowStreamPlan` stage execution.
- [x] This proves `rows()` is not NDJSON-specific.
- [x] Add compatibility path to collect full output as JSON array when normal document mode uses stream plan.

## Phase 10: Stats and Observability

- [x] Generalize current reverse distinct stats into stream stats:

```rust
struct RowStreamStats {
    rows_scanned: usize,
    rows_emitted: usize,
    rows_filtered: usize,
    duplicate_rows: usize,
    direct_key_rows: usize,
    fallback_key_rows: usize,
    direct_project_rows: usize,
    fallback_project_rows: usize,
    source_kind: RowStreamSourceKind,
    direction: RowStreamDirection,
    front_filter: DistinctFrontFilterKind,
}
```

- Keep stats optional.
- CLI does not need to print stats initially.
- [x] Tests should use stats to prove path selection.

## Phase 11: Tests

- [x] Parser/builtin tests:
  - parses `$.rows()`
  - builtin metadata exists
  - row-local `$` semantics unchanged
- [x] Normal JSON tests:
  - `$.rows().take(2)`
  - `$.rows().map($.id)`
  - `$.rows().distinct_by($.id)`
  - object/scalar one-row policy
- [x] NDJSON row-local compatibility:
  - `--ndjson '$.id'` remains per-row
- [x] NDJSON stream tests:
- [x] `$.rows().take(2)`
- [x] `$.rows().first()`
- [x] `$.rows().reverse().take(2)`
- [x] `$.rows().reverse().distinct_by($.id).take(2).map($.v)`
- [x] `$.rows().filter($.active).distinct_by($.id).take(3)`
- [x] early stop before invalid old rows
- [x] escaped string keys
  - compound keys
- [x] direct-path stats
- Unsupported cases:
  - reverse on non-seekable stream
  - unknown stream builtin
  - illegal nested `rows()` if not implemented

## Phase 12: Benchmarks

- [x] Add stream-expression benchmark cases:
  - [x] `rows().take(n)`
  - [x] `rows().reverse().take(n)`
  - [x] `rows().reverse().distinct_by($.id).take(n)`
  - [x] `rows().filter($.active).distinct_by($.id).take(n)`
  - [x] direct projection after distinct
  - [x] fallback key expression
- [x] Compare to current direct API to prove no architecture overhead.
- [x] Keep benchmark methodology cold-path friendly.

## Phase 13: CLI Integration

- No new CLI flags for distinct/reverse/limit.
- Keep `--ndjson` as source-format switch.
- CLI dispatch:

```text
--ndjson + rows() expression     => stream executor
--ndjson + no rows() expression  => current row-local executor
normal JSON + rows()             => document stream executor
```

- Keep existing `--reverse` and `--limit` for backward compatibility only.
- Later deprecate stream behavior flags if expression mode supersedes them.
- [x] Core file-source dispatch now covers the API shape used by jetrocli:
  row-local expressions remain per-row, root `$.rows()` expressions become one
  stream over the file source.

## Phase 14: Architecture Cleanup

- Candidate modules:
  - `io/stream_plan.rs`
  - `io/stream_exec.rs`
  - [x] `io/stream_source.rs`
  - `io/stream_stage.rs`
  - `io/stream_distinct.rs`
- NDJSON-specific IO stays separate:
  - forward reader
  - reverse file reader
  - byte/tape row access
- Generic stages should not know whether rows come from NDJSON or document arrays.
- Keep existing APIs stable while migrating call sites.

## Current Remaining Work

- Add benchmark coverage for expression-level `$.rows()` stream mode.
- Validate real `../jetrocli` binary dispatch against this branch when the CLI
  crate points at the local/core 0.5.10 dependency.
- Decide whether row-stream stats become public, explain-only, or test-only.
- Split generic stage/key/projector helpers further if `stream_exec.rs` grows
  with reducers/grouping/window operators.
- Decide nested `$.books.rows()` semantics later; root `$.rows()` is the only
  supported source boundary for now.

## Acceptance Criteria

- `$.rows()` works in normal JSON and NDJSON mode.
- NDJSON row-local mode is unchanged.
- `$.rows().reverse().distinct_by($.id).take(n).map(...)` uses reverse NDJSON source, adaptive distinct, retained-row early stop, and late projection.
- No CLI `--distinct-by` flag.
- No handwritten query-chain fusion.
- No duplicated distinct/key/projection implementations.
- Tests prove direct and fallback paths.
- Benchmarks show stream expression path is near current direct API path.

## Estimate

14-18 focused sessions for a clean first production version.
