ES|QL|DS: stop destroying a connection per record-boundary probe

[quackaplop]

What this is about

A FROM <dataset> | LIMIT 10 over a single ~9.8 GB ndjson file on S3 in another region takes ~73 seconds and returns ten rows. The same file read through a local file:// dataset returns in 0.35s. All of the time is spent in planning, before a row is read, in newline-aligned macro-split discovery.

Discovery walks the file in target_split_size strides — about 150 of them at the 64 MB default — and probes each stride for the next record boundary. Each probe opened a range covering everything from the stride offset to the end of the file, read the few hundred bytes to the next newline, and aborted.

That range is the defect. It is gigabytes long, so it can never be drained, so the only way to release it is abortStream — and an aborted response destroys its HTTP connection rather than returning it to the pool. Every probe therefore re-paid a TCP connect and a TLS handshake before it could ask its question. Cross-region that is most of the ~0.5s each probe cost. The probes also ran strictly serially: discovery fans out sixteen ways across files, and this dataset is one file, so the fan-out was structurally present and did nothing.

Worth noting the abort was not an oversight. It was the fix for a real problem — closing a partially-read stream used to drag the whole body across the wire — and it is correct when a large remainder genuinely needs discarding. It was the wrong trade only here, where we never wanted the remainder at all.

What this PR does

Stops asking for the remainder. ChunkedStorageInputStream reads a range as bounded, fully-consumed chunks via StorageObject.readBytes. Each chunk is a closed range read to completion, so closing it hands back a connection the provider can pool — the same reason S3StorageObject.fetchMetadata drains its one-byte length probe instead of aborting it. Chunk sizes grow geometrically from 32 KB to a 4 MB cap, so a normal record costs one request and a pathological one costs a logarithmic number, with total transfer within roughly twice the bytes consumed.

Chunk boundaries are invisible to the format code. A splitter peeking past a lone CR to test for LF simply reads the first byte of the next chunk, and a splitter's byte budget keeps counting across chunks, so RecordSplitter implementations are untouched.

Probes the strides concurrently when nothing else is using the pool. A strided splitter resolves a boundary at any offset without knowing the previous one, so the probes are independent. The strided path now fixes its grid up front at stride multiples and dispatches it through BoundedParallelGather, but only on the single-file branch of discovery — the multi-file branch already fans out across files, and the two branches are complements by construction, so a probe gather can never nest inside a file gather and block the pool waiting on itself.

Fixes the read-time copy too. ParallelParsingCoordinator.computeSegments carries a near-verbatim copy of the probe loop, run per macro-split on the data node to sub-segment it, ahead of any parsing dispatch. Same defect, roughly 15 probes per split, on the order of 2,000 connection setups across a full scan. Both loops now share the chunked stream.

Does it work?

Boundaries are unchanged, and that is the claim the tests are built around — an off-by-one here does not fail, it silently shifts where one split ends and the next begins and the query returns a wrong row count.

ChunkedProbeEquivalenceTests probes every offset of each corpus twice, once through the chunked stream and once through a plain whole-remainder stream, and asserts the boundaries are equal. Chunk sizes are randomised down to 1–17 bytes so chunk edges land inside records, on terminators, and between the CR and LF of a pair. Corpora cover LF, CRLF, lone-CR, mixed terminators with ragged records, an unterminated tail, and a record over the byte budget. The quoted-CSV proven probe and exact walk get the same treatment.

On top of that, the existing boundary-value and segment tests pass unmodified — those were not written for this change, which is what makes them worth more than the ones I wrote.

The four tests that pinned the old behaviour now assert its absence: they used to require that every probe abort, which was asserting the defect. Neither probe loop calls abortStream any more. abortStream itself stays in the SPI and its schema-detection test is untouched and green — this removes a misuse of the mechanism, not the mechanism.

:x-pack:plugin:esql:test --tests "...datasources.*": 140 classes, 2,312 tests, green. precommit and spotlessJavaCheck clean.

What this is worth

Modelled, not yet measured — stated plainly because the only measurement so far is the 73s in the report.

A probe today is a TCP connect, a TLS handshake, then the request and time-to-first-byte. Pooling removes the connection setup, leaving one request round trip, which the arithmetic puts at roughly 3–4x. That alone is not enough: ~150 sequential cross-region round trips is a floor no per-probe saving gets under, so the chunked read is the enabling change rather than the fix. Probing the grid sixteen ways is what takes the remainder to about a second, against the 0.35s local floor.

Validating that against the real repro is the next step, and the number here will be replaced with a measured one before this leaves draft.

What's out of scope

The remaining abortStream callers — schema detection in NdJsonFormatReader and CsvFormatReader, and early scan termination in AsyncExternalSourceOperatorFactory — have the same shape but fire once per file rather than once per 64 MB stride. Worth converting; not worth widening a diff whose safety argument is currently tight.

Split discovery still does work proportional to file size before anything has established the query needs it. For a limit-only plan the distribution strategy already routes to local execution and the pushed row limit already forces a single-stream read, so the splits this spends time computing are never used. That is a larger behavioural change and belongs on its own.

What's still to do

• Measure the real repro and replace the modelled numbers above.
• Apply the outstanding review findings: a false javadoc on one equivalence test that describes a comparison it does not perform, a stale class-level javadoc on the abort-chain suite, a leftover bare block in the proven-probe path, and documenting the no-nested-gather invariant on the callee rather than only at the call site.
• Decide whether to unify the two probe loops, which this PR deliberately did not do.

Why this is in draft

Adversarial review is still in flight, and the headline numbers are arithmetic rather than measurement. Neither should be taken on trust yet.