Package org.elasticsearch.xpack.esql.datasources

Class ParallelParsingCoordinator

java.lang.Object
org.elasticsearch.xpack.esql.datasources.ParallelParsingCoordinator
public final class ParallelParsingCoordinator extends Object
Coordinates parallel parsing of a single file by splitting it into byte-range segments and dispatching each segment to a parser thread. Pages are emitted to the consumer in completion (as-ready) order, not segment order: any segment whose pages are ready drains immediately, so a fully-parsed later segment never waits behind a slower earlier one.

Inspired by ClickHouse's ParallelParsingInputFormat. The approach:

  1. A segmentator divides the file into N byte-range segments at record boundaries
  2. Each segment is parsed independently on a separate executor thread
  3. The coordinator yields pages as they become ready via a CloseableIterator

Row ordering. Cross-segment row order is intentionally not preserved: an external scan has no row-order guarantee absent an explicit SORT, and the read schema is bound up-front (see 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)) so segment 0 has no obligation to emit first. Holding pages back to reconstruct segment order is what created the bug this design fixes: a parsed-but-not-yet-emitted segment kept its object-store socket open and idle until the in-order cursor reached it; on S3 that idle socket exceeds the server idle timeout and is reset, surfacing as an HTTP 500. Emitting as-ready lets each segment's socket close as soon as its pages are consumed.

This coordinator only works with SegmentableFormatReader implementations (line-oriented formats like CSV and NDJSON). Columnar formats have their own row-group-level parallelism.

  • Method Details

    • parallelRead

      public static CloseableIterator<Page> parallelRead(SegmentableFormatReader reader, StorageObject storageObject, List<String> projectedColumns, int batchSize, int parallelism, Executor executor) throws IOException
      Creates a parallel-parsing iterator over a single storage object.

      The file is divided into parallelism segments at record boundaries. Each segment is parsed independently and pages are yielded as they become ready (completion order, not segment order). If the file is too small for meaningful parallelism (below the reader's SegmentableFormatReader.minimumSegmentSize() per segment), falls back to single-threaded reading.

      Parameters:
      reader - the segmentable format reader
      storageObject - the file to read
      projectedColumns - columns to project
      batchSize - rows per page
      parallelism - number of parallel parser threads
      executor - executor for parser threads
      Returns:
      an iterator that yields pages as they become ready (cross-segment row order not preserved)
      Throws:
      IOException

    • parallelRead

      public static CloseableIterator<Page> parallelRead(SegmentableFormatReader reader, StorageObject storageObject, List<String> projectedColumns, int batchSize, int parallelism, Executor executor, ErrorPolicy errorPolicy) throws IOException
      Creates a parallel-parsing iterator with an explicit error policy.
      Parameters:
      errorPolicy - error handling policy for per-segment parsing, or null for defaults
      Throws:
      IOException

    • parallelRead

      public static CloseableIterator<Page> parallelRead(SegmentableFormatReader reader, StorageObject storageObject, List<String> projectedColumns, int batchSize, int parallelism, Executor executor, ErrorPolicy errorPolicy, boolean splitStartsAtRecordBoundary) throws IOException
      Convenience overload that forwards splitIncludesFileLeader=true. This assumes the storage object includes the file's leading bytes (header row). For non-leading macro-splits, use the nine-argument overload with explicit splitIncludesFileLeader=false.
      Parameters:
      splitStartsAtRecordBoundary - when true, storageObject is a byte range that already begins on a record boundary (e.g. newline-aligned macro FileSplit); single-threaded fallback reads must set FormatReadContext.recordAligned().
      Throws:
      IOException

    • parallelRead

      public static CloseableIterator<Page> parallelRead(SegmentableFormatReader reader, StorageObject storageObject, List<String> projectedColumns, int batchSize, int parallelism, Executor executor, ErrorPolicy errorPolicy, boolean splitStartsAtRecordBoundary, boolean splitIncludesFileLeader) throws IOException
      Parameters:
      splitStartsAtRecordBoundary - when true, storageObject is a byte range that already begins on a record boundary (e.g. newline-aligned macro FileSplit); single-threaded fallback reads must set FormatReadContext.recordAligned().
      splitIncludesFileLeader - whether this split contains the file-leading bytes (and therefore file header for header-bearing formats). For whole-file reads this is true; for non-leading macro-splits this is false.
      Throws:
      IOException

    • parallelRead

      public static CloseableIterator<Page> parallelRead(SegmentableFormatReader reader, StorageObject storageObject, List<String> projectedColumns, int batchSize, int parallelism, Executor executor, ErrorPolicy errorPolicy, boolean splitStartsAtRecordBoundary, boolean splitIncludesFileLeader, List<Attribute> readSchema, long baseFileOffset) throws IOException
      Forwards to the full-control overload with the default open-segment cap and captureSink=null (text-format readers' close hooks then publish into whatever ExternalStatsCapture sink — if any — happens to be bound on the calling thread). Callers that resolve the max_concurrent_open_segments pragma or care about cross-thread stats capture use the 12-arg overload.
      Parameters:
      readSchema - planner-bound read schema, or null for per-file inference
      baseFileOffset - file-global byte offset of storageObject's first byte (i.e. the macro split's FileSplit.offset()). 0 for whole-file reads. Added to each segment's split-relative offset so readers emit file-global, split-invariant record positions on the _rowPosition channel.
      Throws:
      IOException

    • parallelRead

      public static CloseableIterator<Page> parallelRead(SegmentableFormatReader reader, StorageObject storageObject, List<String> projectedColumns, int batchSize, int parallelism, Executor executor, ErrorPolicy errorPolicy, boolean splitStartsAtRecordBoundary, boolean splitIncludesFileLeader, List<Attribute> readSchema, int maxConcurrentOpenSegments) throws IOException
      Convenience overload that adds the max_concurrent_open_segments cap without per-chunk source-stats capture (equivalent to passing captureSink == null).
      Throws:
      IOException

    • parallelRead

      public static CloseableIterator<Page> parallelRead(SegmentableFormatReader reader, StorageObject storageObject, List<String> projectedColumns, int batchSize, int parallelism, Executor executor, ErrorPolicy errorPolicy, boolean splitStartsAtRecordBoundary, boolean splitIncludesFileLeader, List<Attribute> readSchema, int maxConcurrentOpenSegments, @Nullable ConcurrentMap<String,List<Map<String,Object>>> captureSink) throws IOException
      Full-control overload that takes both the max_concurrent_open_segments cap and an explicit captureSink for per-chunk source-stats contributions.

      maxConcurrentOpenSegments is the per-file limit on byte-range segments whose read streams are open at once. A sliding window dispatches at most maxConcurrentOpenSegments segments at a time; each completion opens the next. This caps the open-stream / buffer count independent of file count and length. See ParallelParsingCoordinator.AsReadyParallelIterator.

      Each segment is parsed on a worker thread; the worker binds captureSink around the per-segment Closeable.close() so text-format readers' close hooks publish their _stats.* contribution into the same sink the consumer-thread wrapper sees. Pass null when no capture is desired (tests, benchmarks).

      Parameters:
      maxConcurrentOpenSegments - per-file cap on concurrently-open segment streams (>= 1)
      captureSink - consumer-owned per-file stats sink to bind on each parser worker, or null to disable capture
      Throws:
      IOException

    • parallelRead

      public static CloseableIterator<Page> parallelRead(SegmentableFormatReader reader, StorageObject storageObject, List<String> projectedColumns, int batchSize, int parallelism, Executor executor, ErrorPolicy errorPolicy, boolean splitStartsAtRecordBoundary, boolean splitIncludesFileLeader, List<Attribute> readSchema, long baseFileOffset, int maxConcurrentOpenSegments, @Nullable ConcurrentMap<String,List<Map<String,Object>>> captureSink, int maxRecordBytes, long statsStripeSize, StripeColumnScope statsColumnScope, boolean splitIsFileFinal) throws IOException
      Full-control overload that also takes the max_record_size cap used by record splitters, the file-global byte base offset, and the canonical-stripe grid for per-stripe stats attribution (<= 0 disables; pure overlay).
      Throws:
      IOException

    • parallelRead

      public static CloseableIterator<Page> parallelRead(SegmentableFormatReader reader, StorageObject storageObject, List<String> projectedColumns, int batchSize, int parallelism, Executor executor, ErrorPolicy errorPolicy, boolean splitStartsAtRecordBoundary, boolean splitIncludesFileLeader, List<Attribute> readSchema, long baseFileOffset, int maxConcurrentOpenSegments, @Nullable ConcurrentMap<String,List<Map<String,Object>>> captureSink, int maxRecordBytes, long statsStripeSize, StripeColumnScope statsColumnScope, boolean splitIsFileFinal, ExternalSourceMetrics metrics) throws IOException
      As the captureSink+maxRecordBytes overload, plus a node telemetry sink used to record a reader.pool.rejected event when a parser-segment submission is rejected (executor saturated / shutting down). Pass ExternalSourceMetrics.NOOP to disable (all narrower overloads do).
      Throws:
      IOException

    • parallelRead

      public static CloseableIterator<Page> parallelRead(SegmentableFormatReader reader, StorageObject storageObject, List<String> projectedColumns, int batchSize, int parallelism, Executor executor, ErrorPolicy errorPolicy, boolean splitStartsAtRecordBoundary, boolean splitIncludesFileLeader, List<Attribute> readSchema, long baseFileOffset, int maxConcurrentOpenSegments, @Nullable ConcurrentMap<String,List<Map<String,Object>>> captureSink, int maxRecordBytes, long statsStripeSize, StripeColumnScope statsColumnScope, boolean splitIsFileFinal, ExternalSourceMetrics metrics, @Nullable Consumer<String> warningSink) throws IOException
      As the metrics overload, plus a relay for client-visible lenient-policy warnings raised while parsing a segment (see SkipWarnings). Segment parsing runs on executor worker threads, so a direct HeaderWarning call there would attach to the wrong thread's response headers and be dropped; production passes AsyncExternalSourceBuffer::recordInformationalWarning so the operator can re-emit it on the driver thread without flipping the response's is_partial flag (these are per-record skip/null-fill warnings, not a truncation of the whole read — see AsyncExternalSourceBuffer#recordWarning for that case). Pass null to fall back to a direct HeaderWarning call on the parsing thread (tests, benchmarks).
      Throws:
      IOException

    • computeSegments

      public static List<long[]> computeSegments(SegmentableFormatReader reader, StorageObject storageObject, long fileLength, int parallelism, long minSegment) throws IOException
      Computes byte-range segments for the file by probing record boundaries. Each segment is a [offset, length] pair. The first segment starts at offset 0; subsequent segments start at the record boundary found after the nominal split point.
      Throws:
      IOException

    • computeSegments

      public static List<long[]> computeSegments(SegmentableFormatReader reader, StorageObject storageObject, long fileLength, int parallelism, long minSegment, int maxRecordBytes) throws IOException
      Computes byte-range segments using a splitter capped by maxRecordBytes.
      Throws:
      IOException