diff --git a/CHANGELOG.md b/CHANGELOG.md index c267ac1..96f18ad 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - Added `elasticsearch:/index` and `es:/index` Elastic CLI context targets using `ELASTIC_ES_URL` and `ELASTIC_ES_API_KEY`. +- Added `--split ` for bounded, parallel ingestion of JSON arrays and objects, including nested collection selection and object-key `id` fields (#14). ## [0.5.0] - 2026-08-12 diff --git a/Cargo.toml b/Cargo.toml index 4329623..8924699 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,6 +23,7 @@ include = [ "tests/fixtures/*.toon", "tests/fixtures/*.ndjson", "tests/fixtures/*.ndjson.gz", + "tests/fixtures/*.json", ] [dependencies] @@ -39,7 +40,7 @@ futures = "^0.3.34" glob = "0.3.4" log = "^0.4.33" reqwest = { version = "0.13.4", features = ["blocking"] } -serde_json = { version = "1.0.151", features = ["raw_value"] } +serde_json = { version = "1.0.151", features = ["arbitrary_precision", "raw_value"] } serde = { version = "^1.0.229", features = ["derive"] } serde_yaml = "0.9.34" serde_json5 = "0.2.1" diff --git a/README.md b/README.md index 95f9a03..c82cd4b 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,7 @@ To build and publish a multi-platform Docker Hub image: - `.ndjson` files - `.ndjson.gz` files - `.json` files +- JSON arrays or objects selected with `--split` - `.csv` files - `.csv.gz` files - local Markdown and text files @@ -94,6 +95,7 @@ Options: -q, --quiet Quiet mode, don't print runtime summary -z, --uncompressed Disable request body gzip compression --content Content subfield name for file imports [default: body] + --split JSON Pointer selecting an array or object to split --action Bulk action for Elasticsearch outputs [default: create] [possible values: create, index, update] --batch-size Documents per Elasticsearch bulk request [default: 5000] --max-requests Maximum concurrent Elasticsearch bulk requests [default: 16] @@ -171,12 +173,24 @@ When writing to Elasticsearch, the output path must include an index name. Remote `.json` inputs are treated as NDJSON. If the downloaded JSON payload does not match the required NDJSON shape, `espipe` exits with: `JSON payload does not look like required NDJSON input format.` +Passing `--split ` instead treats the single input as one JSON document and streams the children of the selected array or object. Split mode works with local paths, `file://` URIs, stdin, and HTTP/HTTPS JSON inputs; it accepts exactly one input source. + ## Data Format Rules ### NDJSON input Each line must be valid line-delimited JSON. For pass-through JSON inputs, `espipe` expects the first non-whitespace character on each line to be `{`. +### Split JSON input + +Use `--split /` to split a root JSON array or object. Use a JSON Pointer to drop wrappers and select a nested collection; for example, `--split /hits` and `--split /hits/` both select `hits`. One trailing slash is optional. Final empty-name members are not addressable, so paths with two trailing slashes such as `/hits//` are rejected. Pointer tokens use JSON Pointer escaping: `~1` represents `/` and `~0` represents `~`. Numeric tokens traverse intermediate arrays by zero-based index. + +Each selected array element is emitted as one JSON object without a generated identifier. Each selected object value is emitted as one JSON object with its property name added as a string `id` field. Object values that already contain `id`, non-object children, missing paths, and selected scalar or null values are errors. + +Split parsing is incremental and applies bounded backpressure through the existing output pipeline. Selected children are transformed in parallel batches using the machine's available CPU parallelism. Completed batches are forwarded immediately, so split mode does not guarantee source order for either arrays or objects. Include a sortable field in the source documents if downstream order matters. + +The complete input, wrapper, and selected collection are never materialized, but bounded batches and their individual documents are. JSON parsing is still streaming rather than transactional, so an error late in the input does not roll back documents already sent; documents in concurrently running batches may already have reached the output. + ### CSV input The first row must be a header row. Each subsequent row is converted into a JSON object using the CSV headers as field names. @@ -308,6 +322,18 @@ espipe docs.ndjson http://localhost:9200/my-index espipe users.csv http://localhost:9200/users ``` +### Split a root JSON object into documents + +```bash +espipe games.json http://localhost:9200/games --split / +``` + +### Split a wrapped JSON array into documents + +```bash +espipe response.json output.ndjson --split /hits/ +``` + ### Read NDJSON from stdin ```bash diff --git a/examples/steam-games/readme.md b/examples/steam-games/readme.md index b25b1df..7f19536 100644 --- a/examples/steam-games/readme.md +++ b/examples/steam-games/readme.md @@ -31,3 +31,27 @@ espipe ~/Downloads/steam-games-dataset-march-2026/games.csv \ ``` The pipeline splits comma-delimited `Tags` and `Screenshots` values into arrays and converts `Windows`, `Mac`, and `Linux` from title-case strings into booleans. + +### JSON object catalogue + +The archive also contains `games.json`, whose root object uses Steam application IDs as property names. Stream its values into a separate index with: + +```bash +espipe games.json \ + http://localhost:9200/steam-games-json \ + --split / +``` + +Each root property becomes one document, and its property name is added as the string `id` field. The JSON records already contain structured values, so this command does not use the CSV-specific pipeline or template above. + +Split batches are transformed in parallel and may be emitted in any order. The generated string `id` preserves each root map key for downstream sorting or identity. + +For a JSON export wrapped as `{"hits":[...]}`, select and drop the wrapper with: + +```bash +espipe wrapped-games.json \ + http://localhost:9200/steam-games-json \ + --split /hits/ +``` + +Array elements are emitted unchanged; unlike object properties, array positions do not generate `id` fields. Array output order is also unspecified. diff --git a/openspec/changes/add-streaming-json-split/.openspec.yaml b/openspec/changes/add-streaming-json-split/.openspec.yaml new file mode 100644 index 0000000..f774115 --- /dev/null +++ b/openspec/changes/add-streaming-json-split/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-08-20 diff --git a/openspec/changes/add-streaming-json-split/design.md b/openspec/changes/add-streaming-json-split/design.md new file mode 100644 index 0000000..b22a47b --- /dev/null +++ b/openspec/changes/add-streaming-json-split/design.md @@ -0,0 +1,99 @@ +## Context + +See `proposal.md` for motivation. The main ingest loop pulls one owned `Box` at a time from `Input::read_next` and awaits `Output::send`; Elasticsearch output already provides bounded batching and concurrency. Local and downloaded JSON currently use a line reader, with a special case that reads a pretty-printed object beginning with a line containing only `{` into memory. Automatically treating a root object or array as a collection would change existing single-document JSON behavior, so splitting must remain explicit. + +The useful pattern from `toon-rust` commit `9c7007a358e25a0c453fd54e02e287ac465ea824` is a `serde_json::Deserializer` driven by `DeserializeSeed` and `Visitor`: `MapAccess` and `SeqAccess` allow selected structure to be consumed incrementally while only individual document subtrees are materialized. Unlike the Toon encoder, `espipe` must navigate to an arbitrary selected collection and yield each child to an async consumer. + +## Goals / Non-Goals + +**Goals:** + +- Select a root or nested collection with predictable JSON Pointer-compatible syntax. +- Stream selected arrays and maps with memory proportional to bounded batches plus existing bounded output buffers. +- Maximize throughput with parallel document transformation and explicitly avoid the cost of restoring source order. +- Preserve the pull-like `Input::read_next` contract and owned raw-document output contract. +- Preserve a selected map's keys as authoritative document IDs without silently destroying an existing `id`. +- Propagate pointer, parser, and validation failures after any previously produced documents, consistent with streaming NDJSON behavior. +- Support the single input forms that can already provide JSON bytes: local paths and `file://` URIs, stdin, and HTTP/HTTPS JSON sources. + +**Non-Goals:** + +- Automatically infer a split path from a suffix or payload shape. +- Select multiple paths, use JSONPath filters/wildcards, recursively flatten descendants, or retain dropped wrapper fields. +- Add synthetic IDs to array elements or make the map identifier field configurable. +- Add compressed JSON suffixes beyond combinations already supported by the repository. +- Make streaming ingestion transactional or roll back documents sent before a later parse error. +- Avoid materializing an individual emitted document. + +## Decisions + +### Use one explicit JSON Pointer-derived option + +Add `--split ` and thread the optional value into `Input::try_new`. Split mode requires exactly one input. It treats that source as one JSON document independently of line boundaries; without the option, ordinary `.json`, `.ndjson`, stdin, and remote JSON retain current behavior. + +RFC 6901 defines the empty string as the root pointer and `/` as a member with an empty name. For command-line ergonomics and the requested examples, split mode defines `/` as a root alias and ignores one trailing slash on non-root paths, making `/hits` and `/hits/` equivalent. All remaining tokens follow RFC 6901 rules: decode `~1` before `~0`, compare object keys without Unicode normalization, and treat a canonical decimal token as an array index while traversing an array. The trade-off is that split mode cannot address a final empty-name member; document this intentional deviation and reject malformed escapes and non-absolute paths during CLI validation. + +Automatic sniffing was rejected because an intentional one-document object/array and a collection share the same root shapes. A generic input-format enum was rejected because `--split` describes a transformation orthogonal to file extension and can coexist with a future format override. + +### Compile the pointer and navigate incrementally with Serde seeds + +Parse the CLI value once into decoded reference tokens. A navigation `DeserializeSeed` consumes one token at a time: + +- On an object, iterate `MapAccess`, send the matching value through the next navigation seed, and consume unmatched values with `IgnoredAny`. +- On an array, validate the token as a canonical zero-based index, consume preceding and following elements with `IgnoredAny`, and send the matching element through the next navigation seed. +- At the terminal target, invoke a collection visitor rather than deserialize a `Value`. + +Missing keys/indices, traversal through scalars, and a terminal scalar/null become contextual split-path errors. Consuming siblings after the selected value and calling `Deserializer::end()` ensures malformed suffixes or trailing root values are still detected. This approach reads but does not materialize skipped wrapper branches. + +Deserializing the root into `Value` and applying `Value::pointer` was rejected because it defeats bounded memory. A custom JSON token parser was rejected because Serde already provides correct syntax, escaping, and line/column diagnostics. JSONPath was rejected because filters and multiple matches introduce broader semantics and buffering questions not needed here. + +### Split the terminal collection into bounded raw batches + +The terminal visitor accepts `MapAccess` or `SeqAccess` and captures each child as `Box` so the single JSON reader only finds document boundaries and validates the enclosing JSON. It groups children into small bounded batches: + +- For a map, retain each property name with its raw value. A CPU worker requires an object, rejects a pre-existing `id`, inserts `id: Value::String(property_name)`, then serializes it into `Box`. +- For an array, retain each zero-based index with its raw value. A CPU worker requires an object and serializes it compactly without adding an ID. + +Workers use Serde's arbitrary-precision number representation when materializing documents, so compact serialization does not round large integers or high-precision decimals. Nested objects and arrays inside an emitted document are preserved. Empty selected arrays/maps complete with no documents. A selected non-collection and non-object children fail rather than being wrapped because all downstream documents must remain JSON objects. + +Map keys become IDs to retain issue #14's catalogue semantics. Array indices are positional rather than domain identifiers, so synthesizing them as IDs was rejected. Rejecting an existing map-value `id` was chosen over overwrite/preserve behavior because either silently makes one identifier non-authoritative. + +### Bridge Serde's visitor lifetime with bounded, unordered parallel batches + +Create a split-input variant backed by two bounded `std::sync::mpsc::sync_channel` stages. A blocking parser worker owns the `Read + Send` source and sends raw document batches into a worker pool sized from available parallelism. Workers validate and transform batches concurrently, then send completed document batches to `Input::read_next`. The input variant drains its current batch before receiving another, preserving the existing one-document pull contract at the output boundary. + +The result channel does not resequence completed batches. Documents within a batch retain their local order, but batches can complete in any order, so split mode makes no output-order guarantee for maps or arrays. This is deliberate: restoring source order would introduce a serial coordination point and retain completed batches behind a slower predecessor. + +A worker boundary is necessary because `MapAccess` and `SeqAccess` are scoped to one visitor call and cannot be stored between repeated `read_next` calls. Both queues apply backpressure, bounding retained raw and transformed documents independently of collection length. If the consumer disconnects or any worker reports an invalid child, shared cancellation stops new work; already-running batches may finish before cancellation is observed. + +### Reuse source preparation and output dispatch + +Split mode reuses existing local readers, remote download/temp-file handling, URI origin metadata, and the main output loop. Remote JSON split mode bypasses only NDJSON-shape validation that conflicts with a single wrapped document. Each emitted child receives existing origin metadata where that source type already supplies it, then enters the same stdout, file, or Elasticsearch output. + +Split worker batches are fixed internal scheduling units rather than collection-sized buffers. Existing `--batch-size` and `--max-requests` settings independently continue to define Elasticsearch bulk buffering, pipeline execution, and request concurrency. + +### Prove streaming with gated and generated inputs + +Check in small fixtures for root/nested arrays and maps plus invalid shapes. Add instrumented/gated reader tests proving the first bounded batch crosses the handoff before EOF and that bounded channels stop further reading when the consumer pauses. Generated inputs larger than a bulk batch cover unordered count and ID/value parity without committing a production dataset. + +This deterministic evidence is preferred to an RSS threshold, which is platform-sensitive and unreliable in normal CI. Include pointer escaping, intermediate array traversal, missing paths, malformed suffixes, and unchanged no-flag behavior in focused tests. + +## Risks / Trade-offs + +- [The root/trailing-slash conveniences deviate from strict RFC 6901 empty-key semantics] → Document the normalization prominently and test it alongside standard `~0`/`~1` decoding. +- [A single emitted document or skipped sibling can itself be very large] → Document that emitted documents are materialized individually; `IgnoredAny` keeps skipped values out of `Value` trees even though their bytes must be parsed. +- [The parser remains sequential because one JSON byte stream must be tokenized in order] → Keep that stage allocation-light by capturing raw values, then parallelize object materialization, validation, ID insertion, and serialization across all remaining available CPUs. +- [Parallel batches change observable ordering] → Define split order as unspecified and test document-set parity rather than sequence. Users requiring order can include an explicit sortable field in their documents. +- [A worker pool adds threads for split input] → Split mode permits exactly one input, sizes the pool from available parallelism, and uses bounded queues. +- [An output error may leave the worker parsing its current document briefly] → Treat receiver disconnection as cancellation at the next handoff and let the worker-owned reader/temp file drop when it exits. +- [Late malformed content can be discovered after other batches were sent or are in flight] → Preserve streaming semantics, cancel pending work, and report clearly without implying rollback or a strict error boundary. +- [Remote input is downloaded fully to a temporary file before split parsing] → This preserves current HTTP behavior and bounded memory; direct response streaming is a separate transport optimization. + +## Migration Plan + +1. Add split-path parsing and CLI validation while leaving the default path untouched. +2. Add the pointer navigation seeds, terminal collection visitor, and bounded worker handoff. +3. Route supported existing source readers into split mode and verify output parity. +4. Add fixtures, unit/integration coverage, and documentation before release. + +Rollback consists of removing the opt-in option and split input variant; no stored data, configuration, or default behavior requires migration. diff --git a/openspec/changes/add-streaming-json-split/proposal.md b/openspec/changes/add-streaming-json-split/proposal.md new file mode 100644 index 0000000..1bc5480 --- /dev/null +++ b/openspec/changes/add-streaming-json-split/proposal.md @@ -0,0 +1,33 @@ +## Why + +Large JSON inputs are often one array or object, sometimes nested beneath response wrappers, and cannot be ingested as NDJSON without first materializing and reshaping the payload. `espipe` needs a general way to select a collection and stream its children as documents while preserving bounded memory and the existing output pipeline. + +## What Changes + +- Add an explicit `--split ` input mode for a single JSON source. +- Use `/` as an ergonomic root split and accept an optional trailing slash, so `/hits` and `/hits/` both select the collection beneath `hits`. +- Resolve nested pointer tokens incrementally, including RFC 6901 `~0` and `~1` escapes and numeric array indices, without materializing skipped wrappers or the selected collection. +- Stream both selected JSON arrays and selected JSON objects through bounded, parallel batches; output order is intentionally unspecified to maximize throughput. +- Emit each array element as one document. Emit each object property value as one document with the property name added as its string `id`, rejecting conflicts instead of silently overwriting data. +- Feed emitted documents through the existing stdout, file, and Elasticsearch output paths, including bulk batching, ingest pipelines, and index templates. +- Preserve existing NDJSON, ordinary `.json`, stdin, CSV, Toon, remote-input, glob, and multi-file behavior when `--split` is not supplied. +- Add fixtures and coverage for root and nested arrays/maps, pointer resolution, malformed inputs, bounded streaming, and output integration. +- Document split JSON ingestion alongside the Steam Games CSV example. + +## Capabilities + +### New Capabilities + +- `json-split-input`: Explicit pointer-selected JSON collection streaming, array/map document conversion, map-key identifiers, validation, errors, and existing-output integration. + +### Modified Capabilities + +None. + +## Impact + +- CLI: `src/main.rs` gains `--split ` plus single-input validation. +- Input pipeline: `src/input.rs` gains an incremental pointer navigator and collection reader built on `serde_json::Deserializer`, `DeserializeSeed`, `Visitor`, `MapAccess`, and `SeqAccess`, adapting the approach introduced in `toon-rust` commit `9c7007a358e25a0c453fd54e02e287ac465ea824`. Bounded batches follow esdiag's streaming-data-source pattern and are transformed by a parallel worker pool before entering the existing output pipeline. +- Tests and fixtures: input unit tests and CLI/output integration tests gain representative and large synthetic arrays/maps, nested wrappers, pointer errors, malformed JSON, and conflicting IDs. +- Documentation: `README.md` and `examples/steam-games/readme.md` gain split syntax, examples, and format rules. +- Dependencies: no new runtime crate is expected; the implementation uses the existing `serde`/`serde_json` stack. diff --git a/openspec/changes/add-streaming-json-split/specs/json-split-input/spec.md b/openspec/changes/add-streaming-json-split/specs/json-split-input/spec.md new file mode 100644 index 0000000..6723f8b --- /dev/null +++ b/openspec/changes/add-streaming-json-split/specs/json-split-input/spec.md @@ -0,0 +1,162 @@ +## Purpose + +Define how users incrementally select and split a JSON array or object into individual documents while retaining bounded processing and existing output behavior. + +## ADDED Requirements + +### Requirement: JSON splitting is explicitly selected +The system SHALL provide a `--split ` option that interprets the single input source as one JSON document and splits the selected collection, without changing default JSON or NDJSON parsing. + +#### Scenario: Root split is selected +- **WHEN** the user runs `espipe --split / games.json output.ndjson` +- **THEN** the system selects the root JSON value for splitting +- **AND** it does not apply line-delimited JSON record boundaries + +#### Scenario: Standard JSON input is used without split mode +- **WHEN** the user provides a `.json` or `.ndjson` input without `--split` +- **THEN** the system preserves the existing line-delimited and whole-object compatibility behavior + +#### Scenario: Split mode receives multiple inputs +- **WHEN** the user combines `--split` with more than one input source +- **THEN** startup fails before any document is sent +- **AND** the error states that split mode accepts exactly one input source + +### Requirement: Split paths select nested collections +The system SHALL resolve the split path as JSON Pointer tokens from the root, with `/` as a root alias and one trailing slash ignored for non-root paths. + +#### Scenario: Wrapped collection is selected +- **WHEN** the input contains `{"hits":[{"name":"Alpha"},{"name":"Beta"}]}` +- **AND** the user passes `--split /hits/` +- **THEN** the system splits the array stored beneath `hits` +- **AND** the `hits` wrapper is not present in emitted documents + +#### Scenario: Trailing slash is omitted +- **WHEN** the user passes `--split /hits` +- **THEN** the selected value is the same as for `--split /hits/` + +#### Scenario: Final empty-name member is requested +- **WHEN** the user passes a path with two trailing slashes such as `--split /hits//` +- **THEN** startup fails with an error stating that final empty-name members are not supported + +#### Scenario: Escaped object key is selected +- **WHEN** a path token contains `~1` or `~0` +- **THEN** the system resolves those sequences as `/` or `~` respectively + +#### Scenario: Pointer traverses an array +- **WHEN** an intermediate selected value is an array and the next token is a valid zero-based index +- **THEN** the system continues pointer evaluation through that array element + +#### Scenario: Pointer does not resolve +- **WHEN** a path token names a missing object member, uses an invalid array index, or traverses a scalar +- **THEN** ingestion fails with an error identifying the split path and failing token + +### Requirement: Selected objects stream property values as documents +The system SHALL emit each property value of a selected JSON object as one JSON object document, preserve its JSON values without numeric precision loss, add the property name as a string-valued `id` field, and make no guarantee about output order. + +#### Scenario: Root object is split +- **WHEN** the selected object is `{"10":{"name":"Alpha"},"20":{"name":"Beta"}}` +- **THEN** the system emits `{"id":"10","name":"Alpha"}` and `{"id":"20","name":"Beta"}` +- **AND** either document may be emitted first + +#### Scenario: Numeric-looking object key is used +- **WHEN** a selected object property name is `730` +- **THEN** the emitted document contains `"id":"730"` + +#### Scenario: Map values contain arbitrary-precision numbers +- **WHEN** a selected object property contains a valid JSON number beyond native integer or floating-point precision +- **THEN** the emitted document preserves that number without rounding + +#### Scenario: Object property already contains an id field +- **WHEN** a selected object property value already contains an `id` field +- **THEN** ingestion fails when that property is reached +- **AND** the error identifies the property key and conflicting `id` field +- **AND** the system does not overwrite the existing value + +#### Scenario: Selected object is empty +- **WHEN** the selected collection is an empty object +- **THEN** the system emits zero documents and completes successfully + +### Requirement: Selected arrays stream elements as documents +The system SHALL emit each element of a selected JSON array as one JSON object document without adding a synthetic identifier or losing numeric precision and SHALL NOT promise source-order preservation. + +#### Scenario: Root array is split +- **WHEN** the selected array is `[{"id":"10","name":"Alpha"},{"id":"20","name":"Beta"}]` +- **THEN** the system emits both objects in an unspecified order +- **AND** it preserves each object's fields and JSON types + +#### Scenario: Array objects do not contain ids +- **WHEN** a selected array contains object elements without an `id` field +- **THEN** the system emits those objects unchanged +- **AND** it does not add an array-index identifier + +#### Scenario: Array values contain arbitrary-precision numbers +- **WHEN** a selected array object contains a valid JSON number beyond native integer or floating-point precision +- **THEN** the emitted document preserves that number without rounding + +#### Scenario: Selected array is empty +- **WHEN** the selected collection is an empty array +- **THEN** the system emits zero documents and completes successfully + +### Requirement: Split documents must be JSON objects +The system SHALL reject any selected map value or array element that cannot be represented directly as a JSON object document. + +#### Scenario: Object property value is not an object +- **WHEN** a selected object's property value is an array, scalar, or null +- **THEN** ingestion fails when that property is reached +- **AND** the error identifies its property key + +#### Scenario: Array element is not an object +- **WHEN** a selected array element is an array, scalar, or null +- **THEN** ingestion fails when that element is reached +- **AND** the error identifies its zero-based array index + +#### Scenario: Selected value is not a collection +- **WHEN** the split path resolves to a scalar or null +- **THEN** ingestion fails with an error stating that the selected value must be an array or object + +### Requirement: JSON splitting remains bounded +The system SHALL navigate and deserialize split documents incrementally without materializing the complete input, skipped wrapper subtrees, or an unbounded collection of emitted documents in memory. + +#### Scenario: Large selected collection is ingested +- **WHEN** a selected array or object contains more documents than the configured Elasticsearch bulk batch size +- **THEN** documents flow incrementally through the existing bounded output pipeline +- **AND** split-input memory is bounded independently of the collection's total length + +#### Scenario: Batches complete at different speeds +- **WHEN** parallel split workers complete document batches in a different order from the source +- **THEN** each valid selected child is emitted exactly once +- **AND** the system does not buffer completed batches to restore source order + +#### Scenario: Output applies backpressure +- **WHEN** the selected output consumes documents more slowly than the split parser produces them +- **THEN** parsing waits on a bounded handoff +- **AND** it does not accumulate an unbounded queue of parsed documents + +#### Scenario: First batch precedes end of input +- **WHEN** the selected collection has a full valid worker batch followed by substantial remaining content +- **THEN** documents from that batch can reach the output before the complete input has been read + +### Requirement: Split documents use existing outputs +The system SHALL pass each emitted split document through the same output dispatch used by other input formats. + +#### Scenario: Split documents are written to NDJSON +- **WHEN** split mode targets stdout or a local NDJSON output +- **THEN** each selected child is written as one JSON line with the required map-key transformation, if any + +#### Scenario: Split documents are sent to Elasticsearch +- **WHEN** split mode targets Elasticsearch +- **THEN** documents use the configured bulk action, batch size, and maximum in-flight request limit +- **AND** configured ingest pipelines and index templates retain their existing behavior + +### Requirement: Invalid split inputs produce contextual errors +The system SHALL stop split ingestion on invalid pointer syntax, unresolved paths, malformed JSON, trailing JSON content, or invalid documents and SHALL report the source plus available path and child context. + +#### Scenario: JSON is malformed +- **WHEN** split mode encounters malformed or trailing JSON content +- **THEN** ingestion fails with an error that identifies the input source and JSON parse location + +#### Scenario: Error follows valid documents +- **WHEN** malformed or invalid content occurs after valid selected children +- **THEN** the system cancels new batch work after observing the error +- **AND** documents from concurrently in-flight batches may already have been emitted +- **AND** the diagnostic does not claim that already-sent output was rolled back diff --git a/openspec/changes/add-streaming-json-split/tasks.md b/openspec/changes/add-streaming-json-split/tasks.md new file mode 100644 index 0000000..c4a7c3e --- /dev/null +++ b/openspec/changes/add-streaming-json-split/tasks.md @@ -0,0 +1,52 @@ +## 1. Split option and path parsing + +- [x] 1.1 Add `--split `, pass the optional path into input construction, and reject split mode with more than one input before opening the output. +- [x] 1.2 Parse split paths into decoded tokens, implementing `/` as the root alias, optional non-root trailing slash normalization, RFC 6901 `~1`/`~0` decoding, and actionable validation errors. +- [x] 1.3 Route local paths, `file://` URIs, stdin, and HTTP/HTTPS JSON sources through split parsing when selected while preserving all existing input routes when the option is absent. + +## 2. Incremental pointer navigation + +- [x] 2.1 Add navigation `DeserializeSeed`/`Visitor` types that traverse object tokens with `MapAccess`, discard unmatched values with `IgnoredAny`, and report missing keys with full split-path context. +- [x] 2.2 Add array-token navigation with `SeqAccess`, canonical zero-based index validation, bounded skipping before and after the selected element, and contextual missing/invalid-index errors. +- [x] 2.3 Invoke the terminal collection visitor at the resolved value and validate the remainder of enclosing containers plus `Deserializer::end()` so malformed suffixes and trailing JSON are detected. + +## 3. Streaming collection handoff + +- [x] 3.1 Add a split `Input` variant and bounded document/failure/completion handoff whose blocking parser worker owns the source reader and exits when the consumer disconnects. +- [x] 3.2 Stream a terminal map incrementally, materializing one value at a time, requiring an object, rejecting an existing `id`, inserting the property key as a string `id`, and producing `Box`. +- [x] 3.3 Stream a terminal array incrementally, materializing one element at a time, requiring an object, preserving its fields without a synthetic ID, and producing `Box`. +- [x] 3.4 Preserve applicable origin metadata and propagate source-, path-, key-, index-, and parse-location-aware failures through `Input::read_next`, including clean empty-collection completion. + +## 4. Correctness and bounded-streaming coverage + +- [x] 4.1 Add checked-in root and nested map/array fixtures and tests for map-key string IDs, unchanged array objects, nested value preservation, and empty collections. +- [x] 4.2 Add split-path tests for `/`, equivalent `/hits` and `/hits/`, escaped `~0`/`~1` keys, intermediate array indices, malformed pointers, missing keys/indices, and scalar traversal. +- [x] 4.3 Add child-shape failure tests for scalar/null/nested-array children, selected scalar/null values, map `id` conflicts, malformed JSON, and trailing JSON with contextual diagnostics. +- [x] 4.4 Add an instrumented reader and bounded-handoff test proving the first batch is available before EOF and parsing waits rather than accumulating the complete collection when the consumer is gated. +- [x] 4.5 Add generated maps and arrays larger than an Elasticsearch batch and verify document-count, ID, and value parity without a platform-specific RSS assertion. +- [x] 4.6 Add CLI regressions for stdout and NDJSON file output, multiple-input rejection, remote origin metadata, and unchanged `.json`, `.ndjson`, and stdin behavior without `--split`. +- [x] 4.7 Extend Elasticsearch integration coverage to verify split documents reuse bulk action/configuration, ingest-pipeline, and index-template paths. + +## 5. Documentation and verification + +- [x] 5.1 Update the README CLI reference, supported inputs, pointer normalization/escaping, array-versus-map behavior, errors, examples, and one-document memory-bound limitation for `--split`. +- [x] 5.2 Add root-object and wrapped-array JSON commands beside the CSV command in `examples/steam-games/readme.md`, including map-key `id` behavior. +- [x] 5.3 Run Rust formatting, linting, the complete test suite, and focused CLI/integration tests; confirm existing input/output behavior remains unchanged. + +## 6. Unordered parallel split throughput + +- [x] 6.1 Revise the split contract and documentation to state that output order is unspecified and throughput takes priority over resequencing. +- [x] 6.2 Capture selected children as raw JSON into bounded batches so the parser can continue while CPU workers transform prior batches. +- [x] 6.3 Process batches concurrently using available parallelism, publish completed batches without restoring source order, and propagate cancellation/failures safely. +- [x] 6.4 Adapt `Input::read_next` to drain completed batches while retaining its one-document interface and existing output behavior. +- [x] 6.5 Replace order-sensitive tests with exact document-set parity and add coverage for bounded batch streaming, invalid children, and collections spanning many worker batches. +- [x] 6.6 Document parallel batching, run formatting/lint/full tests, and benchmark the large local JSON input against a local file output without persisting its path. +- [x] 6.7 Benchmark the large local JSON input against localhost Elasticsearch and verify the indexed document count. + +## 7. Verification fixes + +- [x] 7.1 Preserve arbitrary-precision JSON numbers through map and array split transformation and add regression coverage. +- [x] 7.2 Reject double trailing slashes that would address a final empty-name member, then document and test the path rule. +- [x] 7.3 Strengthen coverage for consumer backpressure, exact large-collection parity, late errors after emitted batches, missing array indices, and scalar traversal. +- [x] 7.4 Make the localhost Elasticsearch integration test explicitly opt-in so an unavailable node cannot produce a false passing result. +- [x] 7.5 Run formatting, the full test suite, linting, strict OpenSpec validation, and focused regression commands. diff --git a/openspec/specs/json-split-input/spec.md b/openspec/specs/json-split-input/spec.md new file mode 100644 index 0000000..9aaf198 --- /dev/null +++ b/openspec/specs/json-split-input/spec.md @@ -0,0 +1,162 @@ +## Purpose + +Define how users incrementally select and split a JSON array or object into individual documents while retaining bounded processing and existing output behavior. + +## Requirements + +### Requirement: JSON splitting is explicitly selected +The system SHALL provide a `--split ` option that interprets the single input source as one JSON document and splits the selected collection, without changing default JSON or NDJSON parsing. + +#### Scenario: Root split is selected +- **WHEN** the user runs `espipe --split / games.json output.ndjson` +- **THEN** the system selects the root JSON value for splitting +- **AND** it does not apply line-delimited JSON record boundaries + +#### Scenario: Standard JSON input is used without split mode +- **WHEN** the user provides a `.json` or `.ndjson` input without `--split` +- **THEN** the system preserves the existing line-delimited and whole-object compatibility behavior + +#### Scenario: Split mode receives multiple inputs +- **WHEN** the user combines `--split` with more than one input source +- **THEN** startup fails before any document is sent +- **AND** the error states that split mode accepts exactly one input source + +### Requirement: Split paths select nested collections +The system SHALL resolve the split path as JSON Pointer tokens from the root, with `/` as a root alias and one trailing slash ignored for non-root paths. + +#### Scenario: Wrapped collection is selected +- **WHEN** the input contains `{"hits":[{"name":"Alpha"},{"name":"Beta"}]}` +- **AND** the user passes `--split /hits/` +- **THEN** the system splits the array stored beneath `hits` +- **AND** the `hits` wrapper is not present in emitted documents + +#### Scenario: Trailing slash is omitted +- **WHEN** the user passes `--split /hits` +- **THEN** the selected value is the same as for `--split /hits/` + +#### Scenario: Final empty-name member is requested +- **WHEN** the user passes a path with two trailing slashes such as `--split /hits//` +- **THEN** startup fails with an error stating that final empty-name members are not supported + +#### Scenario: Escaped object key is selected +- **WHEN** a path token contains `~1` or `~0` +- **THEN** the system resolves those sequences as `/` or `~` respectively + +#### Scenario: Pointer traverses an array +- **WHEN** an intermediate selected value is an array and the next token is a valid zero-based index +- **THEN** the system continues pointer evaluation through that array element + +#### Scenario: Pointer does not resolve +- **WHEN** a path token names a missing object member, uses an invalid array index, or traverses a scalar +- **THEN** ingestion fails with an error identifying the split path and failing token + +### Requirement: Selected objects stream property values as documents +The system SHALL emit each property value of a selected JSON object as one JSON object document, preserve its JSON values without numeric precision loss, add the property name as a string-valued `id` field, and make no guarantee about output order. + +#### Scenario: Root object is split +- **WHEN** the selected object is `{"10":{"name":"Alpha"},"20":{"name":"Beta"}}` +- **THEN** the system emits `{"id":"10","name":"Alpha"}` and `{"id":"20","name":"Beta"}` +- **AND** either document may be emitted first + +#### Scenario: Numeric-looking object key is used +- **WHEN** a selected object property name is `730` +- **THEN** the emitted document contains `"id":"730"` + +#### Scenario: Map values contain arbitrary-precision numbers +- **WHEN** a selected object property contains a valid JSON number beyond native integer or floating-point precision +- **THEN** the emitted document preserves that number without rounding + +#### Scenario: Object property already contains an id field +- **WHEN** a selected object property value already contains an `id` field +- **THEN** ingestion fails when that property is reached +- **AND** the error identifies the property key and conflicting `id` field +- **AND** the system does not overwrite the existing value + +#### Scenario: Selected object is empty +- **WHEN** the selected collection is an empty object +- **THEN** the system emits zero documents and completes successfully + +### Requirement: Selected arrays stream elements as documents +The system SHALL emit each element of a selected JSON array as one JSON object document without adding a synthetic identifier or losing numeric precision and SHALL NOT promise source-order preservation. + +#### Scenario: Root array is split +- **WHEN** the selected array is `[{"id":"10","name":"Alpha"},{"id":"20","name":"Beta"}]` +- **THEN** the system emits both objects in an unspecified order +- **AND** it preserves each object's fields and JSON types + +#### Scenario: Array objects do not contain ids +- **WHEN** a selected array contains object elements without an `id` field +- **THEN** the system emits those objects unchanged +- **AND** it does not add an array-index identifier + +#### Scenario: Array values contain arbitrary-precision numbers +- **WHEN** a selected array object contains a valid JSON number beyond native integer or floating-point precision +- **THEN** the emitted document preserves that number without rounding + +#### Scenario: Selected array is empty +- **WHEN** the selected collection is an empty array +- **THEN** the system emits zero documents and completes successfully + +### Requirement: Split documents must be JSON objects +The system SHALL reject any selected map value or array element that cannot be represented directly as a JSON object document. + +#### Scenario: Object property value is not an object +- **WHEN** a selected object's property value is an array, scalar, or null +- **THEN** ingestion fails when that property is reached +- **AND** the error identifies its property key + +#### Scenario: Array element is not an object +- **WHEN** a selected array element is an array, scalar, or null +- **THEN** ingestion fails when that element is reached +- **AND** the error identifies its zero-based array index + +#### Scenario: Selected value is not a collection +- **WHEN** the split path resolves to a scalar or null +- **THEN** ingestion fails with an error stating that the selected value must be an array or object + +### Requirement: JSON splitting remains bounded +The system SHALL navigate and deserialize split documents incrementally without materializing the complete input, skipped wrapper subtrees, or an unbounded collection of emitted documents in memory. + +#### Scenario: Large selected collection is ingested +- **WHEN** a selected array or object contains more documents than the configured Elasticsearch bulk batch size +- **THEN** documents flow incrementally through the existing bounded output pipeline +- **AND** split-input memory is bounded independently of the collection's total length + +#### Scenario: Batches complete at different speeds +- **WHEN** parallel split workers complete document batches in a different order from the source +- **THEN** each valid selected child is emitted exactly once +- **AND** the system does not buffer completed batches to restore source order + +#### Scenario: Output applies backpressure +- **WHEN** the selected output consumes documents more slowly than the split parser produces them +- **THEN** parsing waits on a bounded handoff +- **AND** it does not accumulate an unbounded queue of parsed documents + +#### Scenario: First batch precedes end of input +- **WHEN** the selected collection has a full valid worker batch followed by substantial remaining content +- **THEN** documents from that batch can reach the output before the complete input has been read + +### Requirement: Split documents use existing outputs +The system SHALL pass each emitted split document through the same output dispatch used by other input formats. + +#### Scenario: Split documents are written to NDJSON +- **WHEN** split mode targets stdout or a local NDJSON output +- **THEN** each selected child is written as one JSON line with the required map-key transformation, if any + +#### Scenario: Split documents are sent to Elasticsearch +- **WHEN** split mode targets Elasticsearch +- **THEN** documents use the configured bulk action, batch size, and maximum in-flight request limit +- **AND** configured ingest pipelines and index templates retain their existing behavior + +### Requirement: Invalid split inputs produce contextual errors +The system SHALL stop split ingestion on invalid pointer syntax, unresolved paths, malformed JSON, trailing JSON content, or invalid documents and SHALL report the source plus available path and child context. + +#### Scenario: JSON is malformed +- **WHEN** split mode encounters malformed or trailing JSON content +- **THEN** ingestion fails with an error that identifies the input source and JSON parse location + +#### Scenario: Error follows valid documents +- **WHEN** malformed or invalid content occurs after valid selected children +- **THEN** the system cancels new batch work after observing the error +- **AND** documents from concurrently in-flight batches may already have been emitted +- **AND** the diagnostic does not claim that already-sent output was rolled back diff --git a/src/input.rs b/src/input.rs index a73d27e..a6e8b8e 100644 --- a/src/input.rs +++ b/src/input.rs @@ -1,3 +1,4 @@ +use crate::json_split::{SplitEvent, SplitPath, start_split_reader}; use eyre::{Report, Result, eyre}; use flate2::read::GzDecoder; use fluent_uri::UriRef; @@ -8,11 +9,12 @@ use reqwest::{ }; use serde_json::{Map, Value, value::RawValue}; use std::{ - collections::BTreeMap, + collections::{BTreeMap, VecDeque}, ffi::OsStr, fs::{self, File}, io::{BufRead, BufReader, Read, Seek, SeekFrom, Stdin, Write, stdin}, path::{Path, PathBuf}, + sync::mpsc::Receiver, time::Duration, }; use tempfile::{Builder, NamedTempFile}; @@ -41,6 +43,14 @@ pub enum Input { origin: Option, _temp_file: Option, }, + JsonSplit { + source: String, + receiver: Receiver, + pending_documents: VecDeque>, + finished: bool, + origin: Option, + _temp_file: Option, + }, Stdin { reader: Box>, }, @@ -82,11 +92,29 @@ enum InputKind { } impl Input { - pub async fn try_new(uris: Vec>, content_field: String) -> Result { + pub async fn try_new( + uris: Vec>, + content_field: String, + split: Option, + ) -> Result { validate_content_field(&content_field)?; if uris.is_empty() { return Err(eyre!("At least one input is required")); } + if let Some(split) = split { + if uris.len() != 1 { + return Err(eyre!("--split accepts exactly one input source")); + } + let uri = uris.into_iter().next().unwrap(); + return match uri.scheme().map(|scheme| scheme.as_str()) { + Some("http" | "https") => { + tokio::task::spawn_blocking(move || fetch_remote_split_input(uri, split)) + .await + .map_err(|err| eyre!("Remote input fetch task failed: {err}"))? + } + _ => open_split_input(uri, split), + }; + } if uris.len() == 1 { let uri = uris.into_iter().next().unwrap(); return match uri.scheme().map(|scheme| scheme.as_str()) { @@ -137,6 +165,40 @@ impl Input { )?; add_origin_to_raw(raw, origin.as_ref()) } + Input::JsonSplit { + source, + receiver, + pending_documents, + finished, + origin, + .. + } => { + if *finished { + return Err(eyre!("No split document")); + } + loop { + if let Some(raw) = pending_documents.pop_front() { + return add_origin_to_raw(raw, origin.as_ref()); + } + match receiver.recv() { + Ok(SplitEvent::Documents(documents)) => { + pending_documents.extend(documents); + } + Ok(SplitEvent::Failure(error)) => { + *finished = true; + return Err(eyre!(error)); + } + Ok(SplitEvent::Complete) => { + *finished = true; + return Err(eyre!("No split document")); + } + Err(_) => { + *finished = true; + return Err(eyre!("{source}: JSON split parser stopped unexpectedly")); + } + } + } + } Input::Stdin { reader, .. } => read_json_line(reader, line_buffer, false), Input::FileDocuments { .. } => read_file_document_line(self), } @@ -168,6 +230,7 @@ impl std::fmt::Display for Input { Input::FileJson { source, .. } => write!(f, "{source}"), Input::FileCsv { source, .. } => write!(f, "{source}"), Input::FileToon { source, .. } => write!(f, "{source}"), + Input::JsonSplit { source, .. } => write!(f, "{source}"), Input::Stdin { .. } => write!(f, "stdin"), Input::FileDocuments { source, .. } => write!(f, "{source}"), } @@ -227,6 +290,47 @@ fn open_input_values(uris: Vec>, content_field: &str) -> Result, split: SplitPath) -> Result { + match uri.scheme().map(|scheme| scheme.as_str()) { + Some("file") | None => {} + Some(scheme) => return Err(eyre!("Unsupported input scheme: {scheme}")), + } + + let path_str = uri.path().as_str(); + if uri.scheme().is_none() && path_str == "-" { + return open_json_split(Box::new(stdin()), "stdin".to_string(), split, None, None); + } + if has_glob_metachar(path_str) { + return Err(eyre!("--split does not support glob input: {path_str}")); + } + if is_unsupported_compressed_input(path_str) { + return Err(eyre!("Unsupported compressed input format: {path_str}")); + } + + let path = PathBuf::from(path_str); + let source = path.display().to_string(); + let file = File::open(&path)?; + open_json_split(local_file_reader(file, &path), source, split, None, None) +} + +fn open_json_split( + reader: Box, + source: String, + split: SplitPath, + origin: Option, + temp_file: Option, +) -> Result { + let receiver = start_split_reader(reader, source.clone(), split)?; + Ok(Input::JsonSplit { + source, + receiver, + pending_documents: VecDeque::new(), + finished: false, + origin, + _temp_file: temp_file, + }) +} + fn read_json_line( reader: &mut R, line_buffer: &mut String, @@ -597,7 +701,11 @@ fn split_markdown_frontmatter(text: &str) -> (Option<&str>, &str) { fn is_end_of_input(err: &eyre::Report) -> bool { matches!( err.to_string().as_str(), - "No JSON record" | "No CSV record" | "No file document" | "No Toon document" + "No JSON record" + | "No CSV record" + | "No file document" + | "No Toon document" + | "No split document" ) } @@ -886,7 +994,23 @@ fn fetch_remote_input(uri: UriRef) -> Result { fetch_remote_input_with_client(uri, &client) } +fn fetch_remote_split_input(uri: UriRef, split: SplitPath) -> Result { + let client = Client::builder() + .connect_timeout(REMOTE_CONNECT_TIMEOUT) + .timeout(REMOTE_REQUEST_TIMEOUT) + .build()?; + fetch_remote_input_with_client_and_split(uri, &client, Some(split)) +} + fn fetch_remote_input_with_client(uri: UriRef, client: &Client) -> Result { + fetch_remote_input_with_client_and_split(uri, client, None) +} + +fn fetch_remote_input_with_client_and_split( + uri: UriRef, + client: &Client, + split: Option, +) -> Result { let mut response = client .get(uri.as_str()) .header( @@ -903,6 +1027,9 @@ fn fetch_remote_input_with_client(uri: UriRef, client: &Client) -> Resul } let kind = remote_input_kind(&uri, &response)?; + if split.is_some() && !matches!(kind, InputKind::Ndjson | InputKind::Json) { + return Err(eyre!("--split requires a JSON input source")); + } let suffix = match kind { InputKind::Csv => ".csv", InputKind::Ndjson => ".ndjson", @@ -915,7 +1042,7 @@ fn fetch_remote_input_with_client(uri: UriRef, client: &Client) -> Resul std::io::copy(&mut response, temp_file.as_file_mut())?; temp_file.as_file_mut().flush()?; - if kind == InputKind::Json { + if kind == InputKind::Json && split.is_none() { validate_ndjson_file(temp_file.as_file_mut())?; } @@ -923,6 +1050,16 @@ fn fetch_remote_input_with_client(uri: UriRef, client: &Client) -> Resul let source = uri.to_string(); let origin = Some(origin_from_uri(&uri)); + if let Some(split) = split { + return open_json_split( + Box::new(reader_file), + source, + split, + origin, + Some(temp_file), + ); + } + match kind { InputKind::Csv => Ok(Input::FileCsv { source, @@ -1081,9 +1218,11 @@ fn ensure_json_opening(input: &str, error_message: &str) -> Result<()> { mod tests { use super::{ Input, InputKind, JSON_LINE_OPENING_ERROR, REMOTE_NDJSON_ERROR, - fetch_remote_input_with_client, input_kind_from_path, local_input_kind, open_input_values, - origin_from_local_path, origin_from_uri, validate_content_field, validate_ndjson_file, + fetch_remote_input_with_client, fetch_remote_input_with_client_and_split, + input_kind_from_path, local_input_kind, open_input_values, origin_from_local_path, + origin_from_uri, validate_content_field, validate_ndjson_file, }; + use crate::json_split::SplitPath; use base64::Engine as _; use flate2::{Compression, write::GzEncoder}; use fluent_uri::UriRef; @@ -1954,6 +2093,39 @@ mod tests { handle.join().unwrap(); } + #[test] + fn remote_http_json_split_preserves_origin_uri_components() { + let (base_url, _requests, handle) = spawn_http_server( + "200 OK", + "application/json", + r#"{"hits":[{"name":"alpha"},{"name":"beta"}]}"#, + ); + let uri = UriRef::parse(format!("{base_url}/docs/data.json?download=1#hits").to_string()) + .unwrap(); + let authority = uri.authority().unwrap().as_str().to_string(); + let values = collect_values( + fetch_remote_input_with_client_and_split( + uri, + &Client::builder().build().unwrap(), + Some(SplitPath::parse("/hits/").unwrap()), + ) + .unwrap(), + ); + + assert_eq!(values.len(), 2); + assert!(values.iter().any(|value| value["name"] == "alpha")); + assert!(values.iter().any(|value| value["name"] == "beta")); + for value in values { + assert_eq!(value["origin"]["scheme"], "http"); + assert_eq!(value["origin"]["authority"], authority); + assert_eq!(value["origin"]["path"], "/docs"); + assert_eq!(value["origin"]["filename"], "data.json"); + assert_eq!(value["origin"]["query"], "download=1"); + assert_eq!(value["origin"]["fragment"], "hits"); + } + handle.join().unwrap(); + } + #[test] fn json_extension_is_accepted_for_local_input_detection() { let path = PathBuf::from("/tmp/example.json"); diff --git a/src/json_split.rs b/src/json_split.rs new file mode 100644 index 0000000..f3101f7 --- /dev/null +++ b/src/json_split.rs @@ -0,0 +1,895 @@ +use eyre::{Result, eyre}; +use serde::de::{self, DeserializeSeed, Error as _, IgnoredAny, MapAccess, SeqAccess, Visitor}; +use serde_json::{Value, value::RawValue}; +use std::{ + fmt, + io::{BufReader, Read}, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, + mpsc::{Receiver, SyncSender, sync_channel}, + }, + thread, +}; + +const SPLIT_BATCH_SIZE: usize = 128; +const QUEUED_BATCHES_PER_WORKER: usize = 2; +const SPLIT_READER_CAPACITY: usize = 64 * 1024; + +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct SplitPath { + display: String, + tokens: Vec, +} + +impl SplitPath { + pub(crate) fn parse(input: &str) -> Result { + if input.is_empty() || input == "/" { + return Ok(Self { + display: "/".to_string(), + tokens: Vec::new(), + }); + } + if !input.starts_with('/') { + return Err(eyre!( + "Invalid --split path '{input}': JSON Pointer paths must start with '/'" + )); + } + + let normalized = input.strip_suffix('/').unwrap_or(input); + if normalized.is_empty() || normalized == "/" { + return Err(eyre!( + "Invalid --split path '{input}': use '/' to split the root collection" + )); + } + if normalized.ends_with('/') { + return Err(eyre!( + "Invalid --split path '{input}': final empty-name members are not supported; remove the extra trailing slash" + )); + } + + let tokens = normalized[1..] + .split('/') + .map(|token| decode_token(input, token)) + .collect::>>()?; + + Ok(Self { + display: input.to_string(), + tokens, + }) + } + + pub(crate) fn display(&self) -> &str { + &self.display + } +} + +fn decode_token(path: &str, token: &str) -> Result { + let mut decoded = String::with_capacity(token.len()); + let mut chars = token.chars(); + while let Some(character) = chars.next() { + if character != '~' { + decoded.push(character); + continue; + } + match chars.next() { + Some('0') => decoded.push('~'), + Some('1') => decoded.push('/'), + Some(other) => { + return Err(eyre!( + "Invalid --split path '{path}': unsupported escape '~{other}'" + )); + } + None => { + return Err(eyre!( + "Invalid --split path '{path}': incomplete '~' escape" + )); + } + } + } + Ok(decoded) +} + +pub(crate) enum SplitEvent { + Documents(Vec>), + Failure(String), + Complete, +} + +enum PendingDocument { + Map { key: String, raw: Box }, + Array { index: usize, raw: Box }, +} + +pub(crate) fn start_split_reader( + reader: R, + source: String, + path: SplitPath, +) -> Result> +where + R: Read + Send + 'static, +{ + start_split_reader_with_worker_count(reader, source, path, split_worker_count()) +} + +fn start_split_reader_with_worker_count( + reader: R, + source: String, + path: SplitPath, + worker_count: usize, +) -> Result> +where + R: Read + Send + 'static, +{ + debug_assert!(worker_count > 0); + let result_capacity = worker_count + .saturating_mul(QUEUED_BATCHES_PER_WORKER) + .max(1); + let (sender, receiver) = sync_channel(result_capacity); + let parser_source = source.clone(); + thread::Builder::new() + .name("espipe-json-split".to_string()) + .spawn(move || run_split_reader(reader, parser_source, path, sender, worker_count)) + .map_err(|error| eyre!("Could not start JSON split parser for {source}: {error}"))?; + Ok(receiver) +} + +fn split_worker_count() -> usize { + thread::available_parallelism() + .map(usize::from) + .unwrap_or(1) + .saturating_sub(1) + .max(1) +} + +fn run_split_reader( + reader: R, + source: String, + path: SplitPath, + sender: SyncSender, + worker_count: usize, +) where + R: Read, +{ + let queue_capacity = worker_count + .saturating_mul(QUEUED_BATCHES_PER_WORKER) + .max(1); + let (batch_sender, batch_receiver) = sync_channel(queue_capacity); + let batch_receiver = Arc::new(Mutex::new(batch_receiver)); + let cancelled = Arc::new(AtomicBool::new(false)); + let context = Arc::new(format!( + "{source}: error splitting JSON at '{}'", + path.display() + )); + + thread::scope(|scope| { + for _ in 0..worker_count { + let batch_receiver = Arc::clone(&batch_receiver); + let sender = sender.clone(); + let cancelled = Arc::clone(&cancelled); + let context = Arc::clone(&context); + scope.spawn(move || run_transform_worker(batch_receiver, sender, cancelled, context)); + } + + let reader = BufReader::with_capacity(SPLIT_READER_CAPACITY, reader); + let mut deserializer = serde_json::Deserializer::from_reader(reader); + let result = NavigateSeed { + tokens: &path.tokens, + full_path: path.display(), + sender: &batch_sender, + cancelled: &cancelled, + } + .deserialize(&mut deserializer) + .and_then(|()| deserializer.end()); + + if let Err(error) = result + && !cancelled.load(Ordering::Acquire) + { + publish_failure(&sender, &cancelled, format!("{context}: {error}")); + } + drop(batch_sender); + }); + + if !cancelled.load(Ordering::Acquire) { + let _ = sender.send(SplitEvent::Complete); + } +} + +fn run_transform_worker( + receiver: Arc>>>, + sender: SyncSender, + cancelled: Arc, + context: Arc, +) { + loop { + let batch = match receiver.lock() { + Ok(receiver) => receiver.recv(), + Err(_) => return, + }; + let Ok(batch) = batch else { + return; + }; + + if cancelled.load(Ordering::Acquire) { + continue; + } + + match transform_batch(batch) { + Ok(documents) => { + if sender.send(SplitEvent::Documents(documents)).is_err() { + cancelled.store(true, Ordering::Release); + } + } + Err(error) => publish_failure(&sender, &cancelled, format!("{context}: {error}")), + } + } +} + +fn publish_failure(sender: &SyncSender, cancelled: &AtomicBool, error: String) { + if cancelled + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_ok() + { + let _ = sender.send(SplitEvent::Failure(error)); + } +} + +fn transform_batch(batch: Vec) -> std::result::Result>, String> { + batch.into_iter().map(transform_document).collect() +} + +fn transform_document(document: PendingDocument) -> std::result::Result, String> { + let (context, key, raw) = match document { + PendingDocument::Map { key, raw } => { + let context = format!("object property '{key}'"); + (context, Some(key), raw) + } + PendingDocument::Array { index, raw } => (format!("array element {index}"), None, raw), + }; + + let value = serde_json::from_str::(raw.get()) + .map_err(|error| format!("{context} could not be deserialized: {error}"))?; + let Value::Object(mut object) = value else { + return Err(format!("{context} must contain a JSON object document")); + }; + + if let Some(key) = key { + if object.contains_key("id") { + return Err(format!("{context} conflicts with generated 'id' field")); + } + object.insert("id".to_string(), Value::String(key)); + } + + let json = serde_json::to_string(&Value::Object(object)) + .map_err(|error| format!("could not serialize {context}: {error}"))?; + RawValue::from_string(json).map_err(|error| format!("could not create {context}: {error}")) +} + +struct NavigateSeed<'a> { + tokens: &'a [String], + full_path: &'a str, + sender: &'a SyncSender>, + cancelled: &'a AtomicBool, +} + +impl<'de> DeserializeSeed<'de> for NavigateSeed<'_> { + type Value = (); + + fn deserialize(self, deserializer: D) -> Result + where + D: de::Deserializer<'de>, + { + if self.tokens.is_empty() { + return deserializer.deserialize_any(SplitVisitor { + full_path: self.full_path, + sender: self.sender, + cancelled: self.cancelled, + }); + } + deserializer.deserialize_any(NavigateVisitor { + tokens: self.tokens, + full_path: self.full_path, + sender: self.sender, + cancelled: self.cancelled, + }) + } +} + +struct NavigateVisitor<'a> { + tokens: &'a [String], + full_path: &'a str, + sender: &'a SyncSender>, + cancelled: &'a AtomicBool, +} + +impl<'de> Visitor<'de> for NavigateVisitor<'_> { + type Value = (); + + fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + formatter, + "an object or array while resolving token '{}' in split path '{}'", + self.tokens[0], self.full_path + ) + } + + fn visit_map(self, mut map: A) -> Result + where + A: MapAccess<'de>, + { + let token = &self.tokens[0]; + let mut found = false; + while let Some(key) = map.next_key::()? { + if !found && key == *token { + map.next_value_seed(NavigateSeed { + tokens: &self.tokens[1..], + full_path: self.full_path, + sender: self.sender, + cancelled: self.cancelled, + })?; + found = true; + } else { + map.next_value::()?; + } + } + if found { + Ok(()) + } else { + Err(A::Error::custom(format!( + "split path '{}' did not resolve object token '{token}'", + self.full_path + ))) + } + } + + fn visit_seq(self, mut sequence: A) -> Result + where + A: SeqAccess<'de>, + { + let token = &self.tokens[0]; + let target = parse_array_index(token).map_err(A::Error::custom)?; + let mut index = 0usize; + let mut found = false; + loop { + let has_element = if index == target { + sequence + .next_element_seed(NavigateSeed { + tokens: &self.tokens[1..], + full_path: self.full_path, + sender: self.sender, + cancelled: self.cancelled, + })? + .is_some() + } else { + sequence.next_element::()?.is_some() + }; + if !has_element { + break; + } + if index == target { + found = true; + } + index += 1; + } + if found { + Ok(()) + } else { + Err(A::Error::custom(format!( + "split path '{}' did not resolve array index '{token}'", + self.full_path + ))) + } + } +} + +fn parse_array_index(token: &str) -> std::result::Result { + if token.is_empty() + || (token.len() > 1 && token.starts_with('0')) + || !token.bytes().all(|byte| byte.is_ascii_digit()) + { + return Err(format!( + "split path array token '{token}' must be a canonical zero-based index" + )); + } + token + .parse::() + .map_err(|_| format!("split path array index '{token}' is too large for this platform")) +} + +struct SplitVisitor<'a> { + full_path: &'a str, + sender: &'a SyncSender>, + cancelled: &'a AtomicBool, +} + +impl<'de> Visitor<'de> for SplitVisitor<'_> { + type Value = (); + + fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + formatter, + "an array or object selected by split path '{}'", + self.full_path + ) + } + + fn visit_map(self, mut map: A) -> Result + where + A: MapAccess<'de>, + { + let mut batch = Vec::with_capacity(SPLIT_BATCH_SIZE); + while let Some(key) = map.next_key::()? { + if self.cancelled.load(Ordering::Acquire) { + return Err(A::Error::custom("split document consumer disconnected")); + } + let raw = map.next_value::>().map_err(|error| { + A::Error::custom(format!( + "object property '{key}' could not be deserialized: {error}" + )) + })?; + batch.push(PendingDocument::Map { key, raw }); + send_full_batch(self.sender, &mut batch).map_err(A::Error::custom)?; + } + send_pending_batch(self.sender, batch).map_err(A::Error::custom)?; + Ok(()) + } + + fn visit_seq(self, mut sequence: A) -> Result + where + A: SeqAccess<'de>, + { + let mut batch = Vec::with_capacity(SPLIT_BATCH_SIZE); + let mut index = 0usize; + loop { + let raw = sequence.next_element::>().map_err(|error| { + A::Error::custom(format!( + "array element {index} could not be deserialized: {error}" + )) + })?; + let Some(raw) = raw else { + break; + }; + if self.cancelled.load(Ordering::Acquire) { + return Err(A::Error::custom("split document consumer disconnected")); + } + batch.push(PendingDocument::Array { index, raw }); + send_full_batch(self.sender, &mut batch).map_err(A::Error::custom)?; + index += 1; + } + send_pending_batch(self.sender, batch).map_err(A::Error::custom)?; + Ok(()) + } +} + +fn send_full_batch( + sender: &SyncSender>, + batch: &mut Vec, +) -> std::result::Result<(), String> { + if batch.len() < SPLIT_BATCH_SIZE { + return Ok(()); + } + let full = std::mem::replace(batch, Vec::with_capacity(SPLIT_BATCH_SIZE)); + send_pending_batch(sender, full) +} + +fn send_pending_batch( + sender: &SyncSender>, + batch: Vec, +) -> std::result::Result<(), String> { + if batch.is_empty() { + return Ok(()); + } + sender + .send(batch) + .map_err(|_| "split document consumer disconnected".to_string()) +} + +#[cfg(test)] +mod tests { + use super::{ + SPLIT_BATCH_SIZE, SplitEvent, SplitPath, start_split_reader, + start_split_reader_with_worker_count, + }; + use serde_json::Value; + use std::{ + io::{Cursor, Read}, + sync::{ + Arc, Condvar, Mutex, + atomic::{AtomicUsize, Ordering}, + }, + time::{Duration, Instant}, + }; + + fn collect_raw(input: &str, path: &str) -> std::result::Result, String> { + let receiver = start_split_reader( + Cursor::new(input.as_bytes().to_vec()), + "fixture.json".to_string(), + SplitPath::parse(path).unwrap(), + ) + .unwrap(); + let mut documents = Vec::new(); + loop { + match receiver.recv().unwrap() { + SplitEvent::Documents(batch) => { + documents.extend(batch.into_iter().map(|raw| raw.get().to_string())); + } + SplitEvent::Failure(error) => return Err(error), + SplitEvent::Complete => return Ok(documents), + } + } + } + + fn collect(input: &str, path: &str) -> std::result::Result, String> { + collect_raw(input, path).map(|documents| { + documents + .into_iter() + .map(|raw| serde_json::from_str(&raw).unwrap()) + .collect() + }) + } + + #[test] + fn parses_root_and_normalized_paths() { + assert_eq!(SplitPath::parse("").unwrap().tokens, Vec::::new()); + assert_eq!(SplitPath::parse("/").unwrap().tokens, Vec::::new()); + assert_eq!(SplitPath::parse("/hits").unwrap().tokens, vec!["hits"]); + assert_eq!(SplitPath::parse("/hits/").unwrap().tokens, vec!["hits"]); + assert_eq!( + SplitPath::parse("/a~1b/m~0n").unwrap().tokens, + vec!["a/b", "m~n"] + ); + } + + #[test] + fn rejects_invalid_paths() { + assert!( + SplitPath::parse("hits") + .unwrap_err() + .to_string() + .contains("start") + ); + assert!(SplitPath::parse("//").is_err()); + assert!( + SplitPath::parse("/hits//") + .unwrap_err() + .to_string() + .contains("final empty-name") + ); + assert!( + SplitPath::parse("/hits/~2") + .unwrap_err() + .to_string() + .contains("~2") + ); + assert!( + SplitPath::parse("/hits/~") + .unwrap_err() + .to_string() + .contains("incomplete") + ); + } + + #[test] + fn splits_root_map_with_string_ids() { + let documents = collect( + r#"{"20":{"name":"Beta"},"10":{"name":"Alpha","nested":[1,true]}}"#, + "/", + ) + .unwrap(); + assert_eq!(documents.len(), 2); + let beta = documents.iter().find(|doc| doc["id"] == "20").unwrap(); + let alpha = documents.iter().find(|doc| doc["id"] == "10").unwrap(); + assert_eq!(beta["name"], "Beta"); + assert_eq!(alpha["nested"], serde_json::json!([1, true])); + } + + #[test] + fn splits_nested_array_without_synthetic_ids() { + let documents = collect( + r#"{"hits":[{"name":"Alpha"},{"name":"Beta","nested":{"ok":true}}]}"#, + "/hits/", + ) + .unwrap(); + assert_eq!(documents.len(), 2); + assert!(documents.contains(&serde_json::json!({"name": "Alpha"}))); + assert!(documents.contains(&serde_json::json!({"name": "Beta", "nested": {"ok": true}}))); + } + + #[test] + fn preserves_arbitrary_precision_numbers() { + const LARGE_INTEGER: &str = "123456789012345678901234567890"; + const PRECISE_DECIMAL: &str = "0.123456789012345678901234567890"; + + let array = collect_raw( + &format!(r#"[{{"large":{LARGE_INTEGER},"decimal":{PRECISE_DECIMAL}}}]"#), + "/", + ) + .unwrap(); + assert_eq!(array.len(), 1); + assert!(array[0].contains(LARGE_INTEGER)); + assert!(array[0].contains(PRECISE_DECIMAL)); + + let map = collect_raw( + &format!(r#"{{"730":{{"large":{LARGE_INTEGER},"decimal":{PRECISE_DECIMAL}}}}}"#), + "/", + ) + .unwrap(); + assert_eq!(map.len(), 1); + assert!(map[0].contains(r#""id":"730""#)); + assert!(map[0].contains(LARGE_INTEGER)); + assert!(map[0].contains(PRECISE_DECIMAL)); + } + + #[test] + fn traverses_escaped_keys_and_array_indices() { + let documents = collect( + r#"{"a/b":[{"skip":[]},{"m~n":{"x":{"value":1},"y":{"value":2}}}]}"#, + "/a~1b/1/m~0n", + ) + .unwrap(); + assert!(documents.contains(&serde_json::json!({"id": "x", "value": 1}))); + assert!(documents.contains(&serde_json::json!({"id": "y", "value": 2}))); + } + + #[test] + fn accepts_empty_collections() { + assert!(collect("{}", "/").unwrap().is_empty()); + assert!(collect("[]", "/").unwrap().is_empty()); + } + + #[test] + fn reports_pointer_and_document_shape_errors() { + let missing = collect(r#"{"hits":[]}"#, "/missing").unwrap_err(); + assert!(missing.contains("fixture.json")); + assert!(missing.contains("missing")); + + let bad_index = collect(r#"{"hits":[[]]}"#, "/hits/01").unwrap_err(); + assert!(bad_index.contains("canonical zero-based index")); + + let missing_index = collect(r#"{"hits":[{}]}"#, "/hits/2").unwrap_err(); + assert!(missing_index.contains("array index '2'")); + + let scalar_traversal = collect(r#"{"hits":1}"#, "/hits/name").unwrap_err(); + assert!(scalar_traversal.contains("token 'name'")); + assert!(scalar_traversal.contains("/hits/name")); + + let scalar = collect(r#"{"hits":1}"#, "/hits").unwrap_err(); + assert!(scalar.contains("array or object selected")); + + let map_child = collect(r#"{"bad":null}"#, "/").unwrap_err(); + assert!(map_child.contains("property 'bad'")); + + let array_child = collect(r#"[{} , 1]"#, "/").unwrap_err(); + assert!(array_child.contains("element 1")); + + let conflict = collect(r#"{"10":{"id":"existing"}}"#, "/").unwrap_err(); + assert!(conflict.contains("property '10'")); + assert!(conflict.contains("generated 'id'")); + } + + #[test] + fn reports_malformed_and_trailing_json_after_prior_documents() { + let malformed = collect(r#"[{"ok":true},{"bad":}]"#, "/").unwrap_err(); + assert!(malformed.contains("array element 1")); + assert!(malformed.contains("line 1 column")); + + let malformed_map = collect(r#"{"good":{"ok":true},"bad":{"broken":}}"#, "/").unwrap_err(); + assert!(malformed_map.contains("object property 'bad'")); + assert!(malformed_map.contains("line 1 column")); + + let trailing = collect(r#"[{"ok":true}] {}"#, "/").unwrap_err(); + assert!(trailing.contains("trailing characters")); + } + + struct GatedReader { + bytes: Cursor>, + gate_at: usize, + gate: Arc<(Mutex, Condvar)>, + } + + impl Read for GatedReader { + fn read(&mut self, buffer: &mut [u8]) -> std::io::Result { + let position = self.bytes.position() as usize; + if position >= self.gate_at { + let (released, condition) = &*self.gate; + let mut released = released.lock().unwrap(); + while !*released { + released = condition.wait(released).unwrap(); + } + return self.bytes.read(buffer); + } + let allowed = buffer.len().min(self.gate_at - position); + self.bytes.read(&mut buffer[..allowed]) + } + } + + struct TrackingReader { + bytes: Cursor>, + bytes_read: Arc, + } + + impl Read for TrackingReader { + fn read(&mut self, buffer: &mut [u8]) -> std::io::Result { + let count = self.bytes.read(buffer)?; + self.bytes_read + .store(self.bytes.position() as usize, Ordering::Release); + Ok(count) + } + } + + #[test] + fn emits_before_reading_the_complete_collection() { + let documents = (0..=SPLIT_BATCH_SIZE) + .map(|index| format!(r#"{{"value":{index}}}"#)) + .collect::>(); + let input = format!("[{}]", documents.join(",")); + let gate_at = input + .match_indices(',') + .nth(SPLIT_BATCH_SIZE - 1) + .map(|(index, _)| index) + .unwrap(); + let gate = Arc::new((Mutex::new(false), Condvar::new())); + let receiver = start_split_reader( + GatedReader { + bytes: Cursor::new(input.into_bytes()), + gate_at, + gate: Arc::clone(&gate), + }, + "gated.json".to_string(), + SplitPath::parse("/").unwrap(), + ) + .unwrap(); + match receiver.recv().unwrap() { + SplitEvent::Documents(batch) => { + assert_eq!(batch.len(), SPLIT_BATCH_SIZE); + let values = batch + .into_iter() + .map(|raw| serde_json::from_str::(raw.get()).unwrap()) + .collect::>(); + assert!(values.iter().any(|value| value["value"] == 0)); + } + SplitEvent::Failure(error) => panic!("unexpected failure: {error}"), + SplitEvent::Complete => panic!("expected a document batch"), + } + let (released, condition) = &*gate; + *released.lock().unwrap() = true; + condition.notify_all(); + match receiver.recv().unwrap() { + SplitEvent::Documents(batch) => { + assert_eq!(batch.len(), 1); + let value: Value = serde_json::from_str(batch[0].get()).unwrap(); + assert_eq!(value["value"], SPLIT_BATCH_SIZE); + } + SplitEvent::Failure(error) => panic!("unexpected failure: {error}"), + SplitEvent::Complete => panic!("expected a second document batch"), + } + assert!(matches!(receiver.recv().unwrap(), SplitEvent::Complete)); + } + + #[test] + fn bounded_handoff_stops_reading_when_consumer_stalls() { + const DOCUMENTS: usize = 10_000; + let padding = "x".repeat(128); + let documents = (0..DOCUMENTS) + .map(|index| format!(r#"{{"value":{index},"padding":"{padding}"}}"#)) + .collect::>(); + let input = format!("[{}]", documents.join(",")); + let input_len = input.len(); + let bytes_read = Arc::new(AtomicUsize::new(0)); + let receiver = start_split_reader_with_worker_count( + TrackingReader { + bytes: Cursor::new(input.into_bytes()), + bytes_read: Arc::clone(&bytes_read), + }, + "backpressure.json".to_string(), + SplitPath::parse("/").unwrap(), + 1, + ) + .unwrap(); + + let deadline = Instant::now() + Duration::from_secs(5); + let mut previous = usize::MAX; + let mut stable_samples = 0; + while Instant::now() < deadline && stable_samples < 5 { + std::thread::sleep(Duration::from_millis(10)); + let current = bytes_read.load(Ordering::Acquire); + if current > 0 && current == previous { + stable_samples += 1; + } else { + stable_samples = 0; + previous = current; + } + } + + let stalled_at = bytes_read.load(Ordering::Acquire); + assert_eq!(stable_samples, 5, "split parser did not reach backpressure"); + assert!( + stalled_at < input_len, + "split parser read the complete input while its consumer was stalled" + ); + + let mut emitted = 0; + loop { + match receiver.recv().unwrap() { + SplitEvent::Documents(batch) => emitted += batch.len(), + SplitEvent::Failure(error) => panic!("unexpected failure: {error}"), + SplitEvent::Complete => break, + } + } + assert_eq!(emitted, DOCUMENTS); + } + + #[test] + fn reports_late_error_after_an_emitted_batch() { + let documents = (0..SPLIT_BATCH_SIZE) + .map(|index| format!(r#"{{"value":{index}}}"#)) + .collect::>(); + let input = format!(r#"[{},{{"bad":}}]"#, documents.join(",")); + let gate_at = input + .match_indices(',') + .nth(SPLIT_BATCH_SIZE - 1) + .map(|(index, _)| index) + .unwrap(); + let gate = Arc::new((Mutex::new(false), Condvar::new())); + let receiver = start_split_reader( + GatedReader { + bytes: Cursor::new(input.into_bytes()), + gate_at, + gate: Arc::clone(&gate), + }, + "late-error.json".to_string(), + SplitPath::parse("/").unwrap(), + ) + .unwrap(); + + match receiver.recv().unwrap() { + SplitEvent::Documents(batch) => assert_eq!(batch.len(), SPLIT_BATCH_SIZE), + SplitEvent::Failure(error) => panic!("unexpected early failure: {error}"), + SplitEvent::Complete => panic!("expected a document batch"), + } + + let (released, condition) = &*gate; + *released.lock().unwrap() = true; + condition.notify_all(); + match receiver.recv().unwrap() { + SplitEvent::Failure(error) => { + assert!(error.contains("late-error.json")); + assert!(error.contains("line 1 column")); + assert!(!error.contains("roll")); + } + SplitEvent::Documents(_) => panic!("unexpected document batch after malformed JSON"), + SplitEvent::Complete => panic!("expected malformed JSON failure"), + } + } + + #[test] + fn splits_collections_larger_than_the_default_bulk_batch() { + const DOCUMENTS: usize = 5_001; + + let mut map = String::from("{"); + let mut array = String::from("["); + for index in 0..DOCUMENTS { + if index > 0 { + map.push(','); + array.push(','); + } + map.push_str(&format!(r#""{index}":{{"value":{index}}}"#)); + array.push_str(&format!(r#"{{"value":{index}}}"#)); + } + map.push('}'); + array.push(']'); + + let mut map_documents = collect(&map, "/").unwrap(); + let mut array_documents = collect(&array, "/").unwrap(); + assert_eq!(map_documents.len(), DOCUMENTS); + assert_eq!(array_documents.len(), DOCUMENTS); + map_documents + .sort_by_key(|document| document["id"].as_str().unwrap().parse::().unwrap()); + array_documents.sort_by_key(|document| document["value"].as_u64().unwrap()); + for index in 0..DOCUMENTS { + assert_eq!(map_documents[index]["id"], index.to_string()); + assert_eq!(map_documents[index]["value"], index); + assert_eq!(array_documents[index]["value"], index); + } + } +} diff --git a/src/main.rs b/src/main.rs index 615d696..4c5497f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,11 +1,13 @@ mod client; mod input; +mod json_split; mod output; use clap::Parser; use client::Auth; use fluent_uri::UriRef; use input::Input; +use json_split::SplitPath; use output::{BulkAction, ElasticsearchOutputConfig, Output, OutputPreflightConfig}; use std::{env, path::PathBuf, process::ExitCode}; @@ -26,6 +28,13 @@ struct Cli { default_value = "body" )] content: String, + /// Split a JSON array or object selected by JSON Pointer + #[arg( + help = "JSON Pointer selecting an array or object to split", + long, + value_name = "JSON_POINTER" + )] + split: Option, /// Accept invalid certificates #[arg( help = "Ignore certificate validation", @@ -127,6 +136,7 @@ async fn main() -> ExitCode { let Cli { mut paths, content, + split, quiet, insecure, apikey, @@ -144,6 +154,16 @@ async fn main() -> ExitCode { } = args; let output = paths.pop().expect("clap requires at least two paths"); let inputs = paths; + let split = match split { + Some(_) if inputs.len() != 1 => { + return exit_with_error(eyre::eyre!("--split accepts exactly one input source")); + } + Some(path) => match SplitPath::parse(&path) { + Ok(path) => Some(path), + Err(err) => return exit_with_error(err), + }, + None => None, + }; if let Err(err) = validate_multi_input_output(&inputs, &output) { return exit_with_error(err); } @@ -192,14 +212,14 @@ async fn main() -> ExitCode { }; log::debug!("output: {output}"); - let input = match Input::try_new(inputs, content).await { + let input = match Input::try_new(inputs, content, split).await { Ok(input) => input, Err(err) => return exit_with_error(err), }; log::debug!("input: {input}"); (input, output) } else { - let input = match Input::try_new(inputs, content).await { + let input = match Input::try_new(inputs, content, split).await { Ok(input) => input, Err(err) => return exit_with_error(err), }; @@ -232,7 +252,12 @@ async fn main() -> ExitCode { let line = match input.read_next(&mut line_buffer) { Ok(Some(line)) => line, Ok(None) => break, - Err(err) => return exit_with_error(err), + Err(err) => { + if let Err(close_err) = output.close().await { + eprintln!("Could not close output after input error: {close_err}"); + } + return exit_with_error(err); + } }; input_line += 1; match output.send(line).await { diff --git a/tests/elasticsearch.rs b/tests/elasticsearch.rs index ae3e8f4..c91fe89 100644 --- a/tests/elasticsearch.rs +++ b/tests/elasticsearch.rs @@ -96,6 +96,106 @@ async fn cli_ingests_into_elasticsearch_if_available() -> Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires an unauthenticated local Elasticsearch node at http://localhost:9200"] +async fn cli_split_reuses_elasticsearch_batch_pipeline_and_template() -> Result<()> { + let base_url = Url::parse("http://localhost:9200")?; + let transport = + TransportBuilder::new(SingleNodeConnectionPool::new(base_url.clone())).build()?; + let client = Elasticsearch::new(transport); + + let temp_dir = temp_dir("espipe-es-split-it"); + let input_path = fixture_path("split_nested_array.json"); + let index = test_index_name(); + let pipeline_name = format!("{index}-pipeline"); + let template_name = format!("{index}-template"); + let output_url = format!("{}/{}", base_url.as_str().trim_end_matches('/'), index); + let pipeline_path = temp_dir.join("split-pipeline.json"); + let template_path = temp_dir.join("split-template.json"); + + fs::write( + &pipeline_path, + r#"{"processors":[{"set":{"field":"ingested_by","value":"espipe-localhost-pipeline"}}]}"#, + )?; + fs::write( + &template_path, + format!( + r#"{{ + "index_patterns": ["{index}"], + "template": {{ + "settings": {{ + "index.default_pipeline": "{pipeline_name}", + "number_of_shards": 1, + "number_of_replicas": 0 + }}, + "mappings": {{ + "properties": {{ + "id": {{"type": "keyword"}}, + "name": {{"type": "keyword"}}, + "ingested_by": {{"type": "keyword"}} + }} + }} + }} +}}"# + ), + )?; + + let output = Command::new(env!("CARGO_BIN_EXE_espipe")) + .arg(&input_path) + .arg(&output_url) + .arg("--split") + .arg("/hits/") + .arg("--action") + .arg("index") + .arg("--batch-size") + .arg("1") + .arg("--max-requests") + .arg("1") + .arg("--pipeline") + .arg(&pipeline_path) + .arg("--pipeline-name") + .arg(&pipeline_name) + .arg("--template") + .arg(&template_path) + .arg("--template-name") + .arg(&template_name) + .output() + .expect("run espipe"); + + assert!( + output.status.success(), + "stdout: {}\nstderr: {}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + + client + .indices() + .refresh(IndicesRefreshParts::Index(&[&index])) + .send() + .await?; + let response = client.count(CountParts::Index(&[&index])).send().await?; + let body: Value = response.json().await?; + assert_eq!(body.get("count").and_then(Value::as_u64), Some(2)); + assert_eq!(count_pipeline_field(&client, &index).await?, 2); + + cleanup_elasticsearch_resource(&client, Method::Delete, &format!("/{index}")).await?; + cleanup_elasticsearch_resource( + &client, + Method::Delete, + &format!("/_index_template/{template_name}"), + ) + .await?; + cleanup_elasticsearch_resource( + &client, + Method::Delete, + &format!("/_ingest/pipeline/{pipeline_name}"), + ) + .await?; + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[ignore = "requires a local Elasticsearch node at http://localhost:9200"] async fn cli_ingests_gzip_ndjson_fixture_into_localhost() -> Result<()> { diff --git a/tests/file_output.rs b/tests/file_output.rs index 70ba2e3..5e27143 100644 --- a/tests/file_output.rs +++ b/tests/file_output.rs @@ -3,12 +3,21 @@ use flate2::read::GzDecoder; use serde_json::Value; use std::{ fs, - io::Read, + io::{Read, Write}, path::{Path, PathBuf}, - process::Command, - time::{SystemTime, UNIX_EPOCH}, + process::{Command, Stdio}, + thread, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, }; +fn json_lines(output: &[u8]) -> Vec { + String::from_utf8_lossy(output) + .lines() + .filter(|line| !line.is_empty()) + .map(|line| serde_json::from_str(line).expect("JSON output line")) + .collect() +} + fn fixture_path(name: &str) -> PathBuf { PathBuf::from(env!("CARGO_MANIFEST_DIR")) .join("tests") @@ -66,6 +75,249 @@ fn validate_bulk_schema(lines: &[&str]) { } } +#[test] +fn cli_splits_root_map_to_ndjson_file() { + let input_path = fixture_path("split_root_map.json"); + let output_path = temp_output_path("split-root-map.ndjson"); + + let output = Command::new(env!("CARGO_BIN_EXE_espipe")) + .arg(&input_path) + .arg(&output_path) + .arg("--split") + .arg("/") + .output() + .expect("run espipe"); + + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + let documents = json_lines(&fs::read(&output_path).expect("read split output")); + assert_eq!(documents.len(), 2); + assert!(documents.contains(&serde_json::json!({"id": "20", "name": "Beta"}))); + assert!( + documents.contains( + &serde_json::json!({"id": "10", "name": "Alpha", "nested": {"enabled": true}}) + ) + ); +} + +#[test] +fn cli_splits_wrapped_array_to_stdout() { + let input_path = fixture_path("split_nested_array.json"); + let output = Command::new(env!("CARGO_BIN_EXE_espipe")) + .arg(&input_path) + .arg("-") + .arg("--split") + .arg("/hits/") + .arg("--quiet") + .output() + .expect("run espipe"); + + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + let documents = json_lines(&output.stdout); + assert_eq!(documents.len(), 2); + assert!(documents.contains(&serde_json::json!({"id": "alpha", "name": "Alpha"}))); + assert!( + documents + .contains(&serde_json::json!({"id": "beta", "name": "Beta", "tags": ["featured"]})) + ); +} + +#[test] +fn cli_splits_file_uri_and_stdin_sources() { + let input_path = fixture_path("split_nested_array.json"); + let file_uri = format!("file://{}", input_path.display()); + let file_output = Command::new(env!("CARGO_BIN_EXE_espipe")) + .arg(file_uri) + .arg("-") + .arg("--split") + .arg("/hits") + .arg("--quiet") + .output() + .expect("run espipe with file URI"); + assert!(file_output.status.success()); + assert_eq!(json_lines(&file_output.stdout).len(), 2); + + let mut child = Command::new(env!("CARGO_BIN_EXE_espipe")) + .arg("-") + .arg("-") + .arg("--split") + .arg("/") + .arg("--quiet") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .spawn() + .expect("run espipe with split stdin"); + child + .stdin + .take() + .unwrap() + .write_all(br#"[{"name":"stdin"}]"#) + .unwrap(); + let stdin_output = child.wait_with_output().unwrap(); + assert!(stdin_output.status.success()); + assert_eq!( + json_lines(&stdin_output.stdout), + vec![serde_json::json!({"name": "stdin"})] + ); +} + +#[test] +fn cli_closes_gzip_output_after_late_split_error() { + const COMPLETE_DOCUMENTS: usize = 128; + + let output_path = temp_output_path("late-split-error.ndjson.gz"); + let mut child = Command::new(env!("CARGO_BIN_EXE_espipe")) + .arg("-") + .arg(&output_path) + .arg("--split") + .arg("/") + .arg("--quiet") + .stdin(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("run espipe with split stdin"); + + let mut stdin = child.stdin.take().expect("open espipe stdin"); + let mut prefix = String::from("["); + for index in 0..COMPLETE_DOCUMENTS { + if index > 0 { + prefix.push(','); + } + let mut state = index as u64 + 1; + let padding = (0..1024) + .map(|_| { + state = state + .wrapping_mul(6_364_136_223_846_793_005) + .wrapping_add(1); + const ALPHANUMERIC: &[u8] = + b"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"; + char::from(ALPHANUMERIC[(state as usize) % ALPHANUMERIC.len()]) + }) + .collect::(); + prefix.push_str(&format!(r#"{{"value":{index},"padding":"{padding}"}}"#)); + } + prefix.push(','); + stdin + .write_all(prefix.as_bytes()) + .expect("write complete split batch"); + stdin.flush().expect("flush complete split batch"); + + let deadline = Instant::now() + Duration::from_secs(5); + while !fs::metadata(&output_path).is_ok_and(|metadata| metadata.len() > 0) + && Instant::now() < deadline + { + thread::sleep(Duration::from_millis(10)); + } + assert!( + fs::metadata(&output_path).is_ok_and(|metadata| metadata.len() > 0), + "espipe did not write the complete split batch before the deadline" + ); + + stdin + .write_all(br#"{"bad":}]"#) + .expect("write malformed split suffix"); + drop(stdin); + + let output = child.wait_with_output().expect("wait for espipe"); + assert!(!output.status.success(), "malformed JSON should fail"); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + stderr.contains("array element 128"), + "stderr should retain the split error: {stderr}" + ); + + let compressed = fs::read(&output_path).expect("read gzip split output"); + let mut decoder = GzDecoder::new(compressed.as_slice()); + let mut decoded = Vec::new(); + decoder + .read_to_end(&mut decoded) + .expect("late-error output should be a complete gzip stream"); + assert_eq!(json_lines(&decoded).len(), COMPLETE_DOCUMENTS); +} + +#[test] +fn cli_rejects_split_with_multiple_inputs_before_writing() { + let output_path = temp_output_path("split-multiple.ndjson"); + fs::write(&output_path, "preserve me").expect("write output sentinel"); + + let output = Command::new(env!("CARGO_BIN_EXE_espipe")) + .arg(fixture_path("split_root_map.json")) + .arg(fixture_path("split_nested_array.json")) + .arg(&output_path) + .arg("--split") + .arg("/") + .output() + .expect("run espipe"); + + assert!(!output.status.success()); + assert!(String::from_utf8_lossy(&output.stderr).contains("exactly one input source")); + assert_eq!(fs::read_to_string(output_path).unwrap(), "preserve me"); +} + +#[test] +fn cli_rejects_invalid_split_path_before_writing() { + let output_path = temp_output_path("split-invalid-path.ndjson"); + fs::write(&output_path, "preserve me").expect("write output sentinel"); + + let output = Command::new(env!("CARGO_BIN_EXE_espipe")) + .arg(fixture_path("split_root_map.json")) + .arg(&output_path) + .arg("--split") + .arg("hits") + .output() + .expect("run espipe"); + + assert!(!output.status.success()); + assert!(String::from_utf8_lossy(&output.stderr).contains("must start with '/'")); + assert_eq!(fs::read_to_string(output_path).unwrap(), "preserve me"); +} + +#[test] +fn cli_preserves_json_behavior_without_split() { + let input_path = fixture_path("split_root_map.json"); + let output_path = temp_output_path("unsplit-json.ndjson"); + + let output = Command::new(env!("CARGO_BIN_EXE_espipe")) + .arg(&input_path) + .arg(&output_path) + .output() + .expect("run espipe"); + + assert!(output.status.success()); + let document: Value = serde_json::from_slice(&fs::read(output_path).unwrap()).unwrap(); + assert!(document.get("20").is_some()); + assert!(document.get("10").is_some()); +} + +#[test] +fn cli_preserves_ndjson_stdin_without_split() { + let mut child = Command::new(env!("CARGO_BIN_EXE_espipe")) + .arg("-") + .arg("-") + .arg("--quiet") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .spawn() + .expect("run espipe"); + child + .stdin + .take() + .unwrap() + .write_all(b"{\"message\":\"hello\"}\n{\"message\":\"world\"}\n") + .unwrap(); + let output = child.wait_with_output().unwrap(); + + assert!(output.status.success()); + assert_eq!(json_lines(&output.stdout).len(), 2); +} + #[test] fn cli_writes_bulk_output_to_file() { let input_path = fixture_path("bulk_input.ndjson"); diff --git a/tests/fixtures/split_nested_array.json b/tests/fixtures/split_nested_array.json new file mode 100644 index 0000000..5fb39d1 --- /dev/null +++ b/tests/fixtures/split_nested_array.json @@ -0,0 +1,7 @@ +{ + "metadata": {"count": 2}, + "hits": [ + {"id": "alpha", "name": "Alpha"}, + {"id": "beta", "name": "Beta", "tags": ["featured"]} + ] +} diff --git a/tests/fixtures/split_root_map.json b/tests/fixtures/split_root_map.json new file mode 100644 index 0000000..95e3525 --- /dev/null +++ b/tests/fixtures/split_root_map.json @@ -0,0 +1,4 @@ +{ + "20": {"name": "Beta"}, + "10": {"name": "Alpha", "nested": {"enabled": true}} +}