Package org.elasticsearch.xpack.esql.datasources

Class AsyncExternalSourceBuffer

java.lang.Object
org.elasticsearch.xpack.esql.datasources.AsyncExternalSourceBuffer
public final class AsyncExternalSourceBuffer extends Object
Thread-safe buffer for async external source data. Modeled after ExchangeBuffer. This buffer provides: - Thread-safe page queue for cross-thread communication - Byte-based backpressure control proportional to actual memory usage - Notification via SubscribableListener when data becomes available - Lifecycle management (finished state tracking)

  • Field Details

    • DEFAULT_MAX_BUFFER_BYTES

      public static final long DEFAULT_MAX_BUFFER_BYTES
      Default byte limit for the buffer, preserving the original "10 normal-sized pages" intent.

  • Constructor Details

    • AsyncExternalSourceBuffer

      public AsyncExternalSourceBuffer(long maxBufferBytes)

  • Method Details

    • capturedSourceMetadataSink

      public ConcurrentMap<String,List<Map<String,Object>>> capturedSourceMetadataSink()
      The mutable per-file capture sink shared with the iterator wrapping.

    • recordWarning

      public void recordWarning(String warning)
      Records a client-visible partial-results warning to be re-emitted on the driver thread when the operator closes, and flips partial. Thread-safe: called from the background reader / parse-worker thread.

      This sink is wired exclusively to the lenient max_record_size truncation path (see StreamingParallelParsingCoordinator#emitTruncationWarning): a recorded warning here always means the read returned fewer records than the source held. Per-record SkipWarnings warnings (row skipped or field null-filled under a lenient ErrorPolicy) must use recordInformationalWarning(java.lang.String) instead — not because a skipped row is never a "real" partial result, but because partial has never tracked that case (this predates warning-sink relaying entirely: on the driver thread such warnings always emitted straight to HeaderWarning without touching this flag). Overloading partial's meaning to also cover SKIP_ROW drops is a separate, pre-existing question and out of scope here.

    • recordInformationalWarning

      public void recordInformationalWarning(String warning)
      Records a client-visible warning to be re-emitted on the driver thread when the operator closes, without affecting partial. Thread-safe: called from the background reader / parse-worker thread.

      Use this for warnings relayed from format-reader SkipWarnings sinks (see FormatReadContext#informationalWarningSink() / RangeReadContext#informationalWarningSink()) — e.g. CSV/NDJSON per-record skip/null-fill handling or Parquet on-disk/planner type mismatches. This preserves these warnings' pre-existing behavior of never flipping partial (previously they only ever reached HeaderWarning directly, which has no notion of partial either); this method only fixes their delivery when the read runs off the driver thread, without changing what they signal. See recordWarning(java.lang.String) for the one warning that has always mapped to partial.

      Each SkipWarnings instance caps its own per-event details at SkipWarnings.MAX_ADDED_WARNINGS (20), but that cap is per reader instance, not per query: a parallel or macro-split read constructs one SkipWarnings per chunk/segment, so a single read can add well more than 20 entries to pendingWarnings here — this queue itself is unbounded.

    • pollWarning

      public String pollWarning()
      Removes and returns the next recorded warning, or null if none remain.

    • isPartial

      public boolean isPartial()
      Whether the background read dropped data under a lenient policy (see partial). Read on the driver thread when assembling the operator Status.

    • addPage

      public void addPage(Page page)
      Add a page to the buffer. Called by the background reader thread.

    • pollPage

      public Page pollPage()
      Poll a page from the buffer. Called by the operator (driver thread).
      Returns:
      the next page, or null if no pages available

    • waitForSpace

      public SubscribableListener<Void> waitForSpace()
      Returns a SubscribableListener that completes when the buffer has space for writing. This is the method producers use for backpressure coordination: it integrates directly with ES async patterns and the producer drain loops.
      Returns:
      a listener that completes when space is available, or an already-completed listener if space exists

    • waitForReading

      public IsBlockedResult waitForReading()
      Returns an IsBlockedResult that completes when the buffer has data for reading. Used by operator to signal driver when waiting for data.

    • finish

      public boolean finish(boolean drainingPages)
      Mark the buffer as finished. Called when reading is done or an error occurs.

      drainingPages is honored regardless of whether this call wins the noMoreInputs transition: AsyncExternalSourceOperator.close() always calls finish(true), and by the time a driver closes its operator noMoreInputs has very often already been set by the producer's own onFailure(java.lang.Throwable) or an earlier finish(false) — e.g. the producer reached natural EOF, or the read failed, before the driver got a chance to drain every page via getOutput()/pollPage(). Gating discardPages() behind the transition used to skip it entirely in that (common) case, leaking whatever the producer had already buffered when the driver's close is not preceded by a full drain (e.g. cross-driver task cancellation cutting this operator before its own poll loop ever ran).

      Returns:
      true if this call performed the running→finishing transition; false if the buffer had already been finished (e.g. producer reached natural EOF, or a concurrent finish/onFailure beat us to it). The stop-hook path in AsyncExternalSourceOperatorFactory uses this to distinguish "STOP genuinely cut a running producer" (partial result) from "STOP raced with natural completion" (honestly complete result).

    • onFailure

      public void onFailure(Throwable t)
      Mark the buffer as failed. Called when the background reader encounters an error.

      Queued pages are retained so the driver can drain them before AsyncExternalSourceOperator surfaces the failure via Operator.getOutput().

    • isFinished

      public boolean isFinished()

    • noMoreInputs

      public boolean noMoreInputs()

    • size

      public int size()

    • addCompletionListener

      public void addCompletionListener(ActionListener<Void> listener)
      Adds a listener that will be notified when this buffer is finished.

    • failure

      public Throwable failure()

    • bytesInBuffer

      public long bytesInBuffer()
      Returns the current number of bytes buffered, as measured by Page.ramBytesUsedByBlocks().

    • recordFormatReaderStatus

      public void recordFormatReaderStatus(FormatReaderStatus snapshot)
      Records the latest format-reader counter snapshot for the operator's status view.

    • addBytesRead

      public void addBytesRead(long delta)
      Adds delta cumulative pre-decompression bytes read from the storage layer.

    • setSplitsTotal

      public void setSplitsTotal(int total)
      Sets the total number of splits the producer expects to process; callable once when known.

    • incSplitsProcessed

      public void incSplitsProcessed()
      Increments the count of splits the producer has finished processing.

    • setCurrentSplit

      public void setCurrentSplit(int idx)
      Records the 1-based index of the split currently being processed by the producer.

    • formatReaderStatus

      public FormatReaderStatus formatReaderStatus()
      Returns the latest format-reader counter snapshot, or null if none recorded yet.

    • bytesRead

      public long bytesRead()
      Returns cumulative pre-decompression bytes read from the storage layer.

    • splitsTotal

      public int splitsTotal()
      Returns the total number of splits the producer expects to process.

    • splitsProcessed

      public int splitsProcessed()
      Returns the number of splits the producer has finished processing.

    • currentSplit

      public int currentSplit()
      Returns the 1-based index of the split currently being processed by the producer.

      Semantics differ slightly between producer paths and the value should not be compared across them: the slice-queue path counts top-level splits pulled from the queue (a coalesced split with N leaves still increments the index by 1, not N), while the file-list / multi-file path uses the absolute file index. Use this for "where am I in the work" UX in a single-query profile, not for cross-query comparison or rate math.