Package org.elasticsearch.xpack.esql.datasources

Class StreamingParallelParsingCoordinator

java.lang.Object

org.elasticsearch.xpack.esql.datasources.StreamingParallelParsingCoordinator

public final class StreamingParallelParsingCoordinator
extends Object

Coordinates parallel parsing of a sequential (non-seekable) decompressed stream by chunking it at record boundaries and dispatching each chunk to a parser thread.

This is the streaming counterpart to ParallelParsingCoordinator, designed for stream-only compression codecs (gzip, zstd without seekable frames) where the decompressed data is only available as a sequential InputStream.

Architecture:

1. A segmentator thread reads the decompressed stream into byte buffers from a pool
2. Each buffer is split at the last record boundary to form a complete chunk
3. Chunks are dispatched to parser threads via a bounded queue
4. Parser threads parse each chunk independently using the format reader
5. Results are yielded in chunk order via per-slot page queues

Shutdown blocking sites: when the iterator's close() is invoked the segmentator may be parked on (a) the upstream InputStream.read(byte[], int, int), (b) bufferPool.take(), (c) chunkQueue.put(), or (d) dispatchPermits.acquire(). Close sets closed=true, releases one permit on dispatchPermits (covers (d)), and drains both the chunk queue and page queues (covers (b) and (c) by freeing slots so put/take either succeeds or completes after the post-acquire closed re-check). Case (a) is the responsibility of the upstream stream wrapper — most stream-only codecs return on close; if the upstream blocks indefinitely on read, close will time out after the iterator's close-timeout and log a warning.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static final record	StreamingParallelParsingCoordinator.WarningSinks	Bundles the two independent, easily-confused warning relays a streaming parallel read may need to thread onto worker threads, as a single typed parameter — two bare adjacent Consumer<String> parameters/fields invite silent transposition at a call site.

Method Summary

Modifier and Type	Method	Description
static CloseableIterator<Page>	parallelRead(SegmentableFormatReader reader, InputStream decompressedStream, List<String> projectedColumns, int batchSize, int parallelism, Executor executor, ErrorPolicy errorPolicy)	Creates a parallel-parsing iterator over a sequential decompressed stream.
static CloseableIterator<Page>	parallelRead(SegmentableFormatReader reader, InputStream decompressedStream, StorageObject storageObject, List<String> projectedColumns, int batchSize, int parallelism, Executor executor, ErrorPolicy errorPolicy, List<Attribute> readSchema, long baseFileOffset, int maxRecordBytes, ConcurrentMap<String,List<Map<String,Object>>> captureSink, long statsStripeSize, StripeColumnScope statsColumnScope, Consumer<String> partialResultsWarningSink, org.elasticsearch.xpack.esql.datasources.StreamingSegmentatorAdmission admission)	Variant that propagates the planner-resolved readSchema.
static CloseableIterator<Page>	parallelRead(SegmentableFormatReader reader, InputStream decompressedStream, StorageObject storageObject, List<String> projectedColumns, int batchSize, int parallelism, Executor executor, ErrorPolicy errorPolicy, List<Attribute> readSchema, long baseFileOffset, int maxRecordBytes, ConcurrentMap<String,List<Map<String,Object>>> captureSink, long statsStripeSize, StripeColumnScope statsColumnScope, StreamingParallelParsingCoordinator.WarningSinks warningSinks)	As the above, plus warningSinks.informationalWarningSink() — see StreamingParallelParsingCoordinator.WarningSinks' Javadoc.
static CloseableIterator<Page>	parallelRead(SegmentableFormatReader reader, InputStream decompressedStream, StorageObject storageObject, List<String> projectedColumns, int batchSize, int parallelism, Executor executor, ErrorPolicy errorPolicy, List<Attribute> readSchema, long baseFileOffset, int maxRecordBytes, ConcurrentMap<String,List<Map<String,Object>>> captureSink, long statsStripeSize, StripeColumnScope statsColumnScope, StreamingParallelParsingCoordinator.WarningSinks warningSinks, org.elasticsearch.xpack.esql.datasources.StreamingSegmentatorAdmission admission, CircuitBreaker breaker)	Full-control overload that also accepts an explicit StreamingSegmentatorAdmission controller, bounding how many segmentators occupy the shared executor at once (see StreamingSegmentatorAdmission).

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Details

parallelRead

public static CloseableIterator<Page> parallelRead(SegmentableFormatReader reader,
 InputStream decompressedStream,
 List<String> projectedColumns,
 int batchSize,
 int parallelism,
 Executor executor,
 ErrorPolicy errorPolicy)
                                            throws IOException

Creates a parallel-parsing iterator over a sequential decompressed stream.

Parameters:

reader - the segmentable format reader (provides record boundary semantics)

decompressedStream - the sequential decompressed input stream

projectedColumns - columns to project

batchSize - rows per page

parallelism - number of parallel parser threads

executor - executor for segmentator and parser threads

errorPolicy - error handling policy

Returns:

an iterator that yields pages in chunk order

Throws:

IOException

parallelRead

public static CloseableIterator<Page> parallelRead(SegmentableFormatReader reader,
 InputStream decompressedStream,
 @Nullable
 StorageObject storageObject,
 List<String> projectedColumns,
 int batchSize,
 int parallelism,
 Executor executor,
 ErrorPolicy errorPolicy,
 @Nullable
 List<Attribute> readSchema,
 long baseFileOffset,
 int maxRecordBytes,
 @Nullable
 ConcurrentMap<String,List<Map<String,Object>>> captureSink,
 long statsStripeSize,
 StripeColumnScope statsColumnScope,
 @Nullable
 Consumer<String> partialResultsWarningSink,
 org.elasticsearch.xpack.esql.datasources.StreamingSegmentatorAdmission admission)
                                            throws IOException

Variant that propagates the planner-resolved readSchema. Mirrors the same parameter on ParallelParsingCoordinator.parallelRead(org.elasticsearch.xpack.esql.datasources.spi.SegmentableFormatReader, org.elasticsearch.xpack.esql.datasources.spi.StorageObject, java.util.List<java.lang.String>, int, int, java.util.concurrent.Executor); the streaming path must thread it so multi-file globs over gzip/zstd/bz2 inputs honor the planner's typing instead of re-inferring per file. Pass null when no read schema is bound.

Parameters:

baseFileOffset - file-global byte offset added to each chunk's decompressed start byte before it is handed to the reader as FormatReadContext.splitStartByte(). Stream-only compressed inputs are not macro-split, so this is 0; the decompressed cumulative offset is the logical file-global offset on its own.

Full-control overload that takes both the max_record_size grow-loop bound and an explicit consumer-owned captureSink for per-chunk source-stats contributions. Each chunk is parsed on a worker thread; this coordinator binds captureSink on that worker around the per-chunk Closeable.close() so text-format readers' close hooks publish into the same map the consumer-thread wrapper sees. Pass null for the sink when no capture is desired (tests, benchmarks). When storageObject is non-null its path is stamped on every chunk and each chunk carries its decompressed-stream coverage range, so the coordinator reconciler can union the chunks under the real file key (see ParallelParsingCoordinator).

partialResultsWarningSink receives a single client-visible message if a non-strict ErrorPolicy truncates the read at a max_record_size cap-hit — a genuine partial- results signal. Production passes AsyncExternalSourceBuffer.recordWarning(java.lang.String) so the operator can re-emit it on the driver thread (the segmentator runs on a forked worker whose response headers never reach the client — see #835). Pass null to fall back to a direct HeaderWarning on the current thread (tests, benchmarks).

Throws:

IOException

parallelRead

public static CloseableIterator<Page> parallelRead(SegmentableFormatReader reader,
 InputStream decompressedStream,
 @Nullable
 StorageObject storageObject,
 List<String> projectedColumns,
 int batchSize,
 int parallelism,
 Executor executor,
 ErrorPolicy errorPolicy,
 @Nullable
 List<Attribute> readSchema,
 long baseFileOffset,
 int maxRecordBytes,
 @Nullable
 ConcurrentMap<String,List<Map<String,Object>>> captureSink,
 long statsStripeSize,
 StripeColumnScope statsColumnScope,
 StreamingParallelParsingCoordinator.WarningSinks warningSinks)
                                            throws IOException

As the above, plus warningSinks.informationalWarningSink() — see StreamingParallelParsingCoordinator.WarningSinks' Javadoc. Kept as a separate overload so the many existing truncation-focused callers (tests, benchmarks) that don't care about generic per-format warnings are unaffected.

Supplies an StreamingSegmentatorAdmission.unbounded() controller: the segmentator is dispatched immediately, matching the pre-admission behavior. Retained for tests and benchmarks running on an isolated, generously-sized pool where segmentator saturation cannot arise; production always threads the per-node admission gate via the overload below.

Throws:

IOException

parallelRead

public static CloseableIterator<Page> parallelRead(SegmentableFormatReader reader,
 InputStream decompressedStream,
 @Nullable
 StorageObject storageObject,
 List<String> projectedColumns,
 int batchSize,
 int parallelism,
 Executor executor,
 ErrorPolicy errorPolicy,
 @Nullable
 List<Attribute> readSchema,
 long baseFileOffset,
 int maxRecordBytes,
 @Nullable
 ConcurrentMap<String,List<Map<String,Object>>> captureSink,
 long statsStripeSize,
 StripeColumnScope statsColumnScope,
 StreamingParallelParsingCoordinator.WarningSinks warningSinks,
 org.elasticsearch.xpack.esql.datasources.StreamingSegmentatorAdmission admission,
 CircuitBreaker breaker)
                                            throws IOException

Full-control overload that also accepts an explicit StreamingSegmentatorAdmission controller, bounding how many segmentators occupy the shared executor at once (see StreamingSegmentatorAdmission). Production always threads the per-node admission gate; the overload above supplies StreamingSegmentatorAdmission.unbounded() for tests and benchmarks running on an isolated, generously-sized pool where segmentator saturation cannot arise.

Throws:

IOException