Fix read counters isolation bugs

[smalyshev]

We have the following problems in the read counters:

• Cross-query contamination: counters from previous queries carry into new ones
• Cross-driver contamination: within one query, multiple drivers mix their readings into the
  same counter instance
• Wrong reader snapshotted: snapshotFormatReaderStatus reads from the factory-level reader,
  not the per-split reader that actually did the work
• Overwrite instead of accumulate: recordFormatReaderStatus replaces the previous snapshot,
  discarding earlier splits' timing data

This patch makes sure every operator gets a fresh instance of format reader (sharing caches but not counters) to avoid double-counting and cross-counting.

[elasticsearchmachine]

Pinging @elastic/es-analytical-engine (Team:Analytics)

[julian-elastic]

CPU time should not live on FormatReader

The current approach has many issues, see my comments. Instead of addressing them one by one, I think we need a new safer approach.

Please redo the isolation of read_nanos / read_cpu_nanos. freshCounters() plus snapshotting currentFileReader does not work: we routinely replace the reader, and the object we snapshot is then not the object that incremented.

Why the reader is a bad owner

A FormatReader is a wither-style config object. withSchema, withReadConfig, withPushedFilter, and freshCounters() each may return a new instance. CSV and NDJSON withSchema already allocate new counters (null / default). The factory, ParallelParsingCoordinator (leading COUNT(*)), and StreamingParallelParsingCoordinator.bindInferredSchema (including nonempty projections) all do this swap after we have pinned currentFileReader.

The registry reader is also shared across queries. Putting billing adders on it is how we got cross-query contamination in the first place. Copying the reader per split (freshCounters) only multiplies the wither problem: every swap orphans another counter object.

What happens today

graph TD
    R[Registry FormatReader]
    A[Reader A currentFileReader]
    B[Reader B withSchema copy]
    S[statusSnapshot of A]
    BUF[Buffer accRead]
    M[B increments never snapshotted]
    R -->|with or freshCounters| A
    A -->|withSchema| B
    A --> S
    S -->|delta| BUF
    B --> M

Loading

A is what we snapshot. B is what parsed. B's increments never reach the buffer. On the streaming path the coordinator also times the same parser calls and later does originalReader.acceptReadCpuNanos(...) on A, so sharing A/B counters without changing that close path would double-count parser CPU.

Where it should live

Owner: AsyncExternalSourceBuffer (already per operator, per query, already has accReadNanos / accReadCpuNanos). Not the reader. Not the registry.

Carrier: a small timing sink the buffer owns, passed through each FormatReadContext (same pattern as informationalWarningSink) and RangeReadContext. Context is built per read(), is not shared across queries, and already reaches every remapped reader inside both coordinators.

graph TD
    BUF[Buffer owns timing sink]
    CTX[FormatReadContext holds sink ref]
    A[Reader A]
    B[Reader B after withSchema]
    BUF --> CTX
    CTX --> A
    CTX --> B
    A -->|addReadNanos addReadCpuNanos| BUF
    B -->|addReadNanos addReadCpuNanos| BUF

Loading

Reader swaps become irrelevant. freshCounters(), currentFileReader for timing, and baseline-delta snapshots are unnecessary for these two fields. Footer-cache / row-group / rows-emitted can stay on the reader if useful for diagnostics.

Final accounting also needs to observe producer completion, including early LIMIT. Reader-local counters can remain for diagnostics.

[julian-elastic]

Why return this? I add a new reader, it gets wrong counters by default? Should we make this abstract so it is always implemented?
Why did you overwrite it for other readers if this is just for parquet?

[smalyshev]

It's not just for parquet, but there's a ton of test readers which don't need this part and if I made it abstract I'd have to implement it in all of them.

[julian-elastic]

Maybe add some debug assertion to fail if the delta is negative? This is indicative of a problem we need to debug and fix.

[julian-elastic]

This does not work properly right now. What happens today:

1. This drain decides it must park (waitForReady or waitForSpace is not done).
2. It registers a resume listener first. That listener can immediately (or a moment later) submit runProducerLoop on another esql_worker thread.
3. Only after that does it return BLOCKED and then call recordFormatReaderStatus.

So ownership of the buffer's baseline has already been handed to a successor, while this drain still has one more write in its pocket. Two threads can then do the same non-atomic "read baseline, subtract, add delta, write baseline." A late write can also land after the next split's resetBufferBaseline and charge the new split for the old reader.

Please fix the order:

1. Record this drain's telemetry first, while it still owns the baseline.
2. Then register the resume listener.
3. Return BLOCKED without recording again.

The successor may start as soon as the listener runs, but there is no trailing snapshot left to collide with it.

[julian-elastic]

This does not work properly right now. On EOF and DONE the factory snapshots currentFileReader, then closes the iterator, then drops the reader.

On the streaming parallel path, segmentator and parser CPU is accumulated during the read and moved onto originalReader only inside StreamingParallelIterator.close(). hasNext() at EOF does not close, and StatsCapturingIterator only forwards close(). The snapshot therefore misses the actual read CPU, not just close overhead. With per-split freshCounters() that increment is then discarded forever; the old shared counters could recover it on a later snapshot of the same object.

Close the iterator first, snapshot while currentFileReader still names this split, then clear it. The single-file rails already use close-then-record.

[julian-elastic]

Super buggy here, both CSV and NDJSON allocate separate counters in withSchema, so the instance that performs the read is not the instance being snapshotted. Please fix.

Long term I am not sure the reader is the right place for those flags.