Module org.elasticsearch.compute
Package org.elasticsearch.compute.operator.topn

Class ParallelTopNOperator

java.lang.Object
org.elasticsearch.compute.operator.topn.ParallelTopNOperator
All Implemented Interfaces:
Closeable, AutoCloseable, org.apache.lucene.util.Accountable, Operator, Releasable
public class ParallelTopNOperator extends Object implements Operator, org.apache.lucene.util.Accountable
A parallel variant of TopNOperator that dispatches incoming pages to multiple worker TopNOperators running on esql_worker threads. The driver thread is never blocked waiting for workers; it uses ExchangeBuffer as a bounded FIFO between the driver and the workers, relying on isBlocked() to back-pressure the upstream operator when the buffer is full.

Lifecycle

  1. Collecting – pages arrive via addInput(org.elasticsearch.compute.data.Page); they are enqueued in in and background workers drain that buffer concurrently.
  2. Finishing – finish() marks the buffer as complete; workers drain the remaining pages and build their local top-K queues.
  3. Merging – once all workers have signalled completion via allWorkersDone, the driver thread pops each worker's priority-queue directly into mergeTarget's queue (no page encode/decode), calls TopNOperator.finish() on it, and delegates subsequent getOutput() calls to it.

Memory accounting and LC ownership transfer

mergeTarget charges allocations directly to the driver's circuit breaker. Each background worker has its own LocalCircuitBreaker (LC) via DriverContext.createChildBlockFactory(). The LC backs every allocation the worker makes: the TopNQueue struct, the spare TopNRow, and the bytes inside each row (each TopNRow holds a direct reference to its creator's breaker).

Aborting workers clean up on the worker thread: TopNOperator.close() closes all rows and the queue struct, returning their bytes to the LC; DriverContext.releaseChildBlockFactory(org.elasticsearch.compute.data.BlockFactory) then closes the LC and returns its reserved bytes to the driver's breaker.

Non-aborting workers transfer ownership of their operator — including their LC — to the driver thread by signalling done without calling either method. The driver then:

  1. Pops the worker's queue into mergeTarget's queue in getOutput(). PriorityQueue.pop() nulls each heap slot, so the worker queue is empty after the drain; transferred rows still reference their worker LC. allWorkersDone guarantees the driver sees each LC's fully-updated state.
  2. Calls TopNOperator.close() on the worker. The queue is empty so no rows are closed; only the queue's struct bytes are returned to the LC.
  3. Calls DriverContext.releaseChildBlockFactory(org.elasticsearch.compute.data.BlockFactory) for all workers in close(), after mergeTarget.close() — which closes any surviving rows, each returning its bytes to its worker LC — so every LC is fully drained before being released to the driver's breaker.

Input pages have Page.allowPassingToDifferentDriver() called on the driver thread in addInput(org.elasticsearch.compute.data.Page) before being enqueued.

  • Constructor Details

    • ParallelTopNOperator

      public ParallelTopNOperator(TopNOperator.ParallelWorkerConfig config, DriverContext driverContext, TopNOperator initialWorker)
      Parameters:
      config - parallel worker configuration (executor, worker count, etc.)
      driverContext - driver context for block-factory and async-action tracking
      initialWorker - the promoted TopNOperator that holds pre-promotion rows; becomes mergeTarget and is never sent to a background thread

  • Method Details

    • needsInput

      public boolean needsInput()
      Description copied from interface: Operator
      whether the given operator can accept more input pages
      Specified by:
      needsInput in interface Operator

    • addInput

      public void addInput(Page page)
      Description copied from interface: Operator
      adds an input page to the operator. only called when needsInput() == true and isFinished() == false
      Specified by:
      addInput in interface Operator

    • finish

      public void finish()
      Description copied from interface: Operator
      notifies the operator that it won't receive any more input pages
      Specified by:
      finish in interface Operator

    • isFinished

      public boolean isFinished()
      Description copied from interface: Operator
      whether the operator has finished processing all input pages and made the corresponding output pages available
      Specified by:
      isFinished in interface Operator

    • canProduceMoreDataWithoutExtraInput

      public boolean canProduceMoreDataWithoutExtraInput()
      Description copied from interface: Operator
      Returns true if the operator can produce more output pages without requiring additional input pages. This is useful for operators that buffer data or have internal state that can produce multiple output pages.

      Operators that do not buffer data should return false - they cannot produce pages out of thin air. Examples of operators that may return true:

      • Operators with internal buffers (e.g., AsyncOperator with pending results)
      • Operators processing a single input page into multiple output pages
      • Aggregation operators that buffer partial results
      Specified by:
      canProduceMoreDataWithoutExtraInput in interface Operator
      Returns:
      true if the operator has buffered data that can produce output, false otherwise

    • isBlocked

      public IsBlockedResult isBlocked()
      Description copied from interface: Operator
      An operator can be blocked on some action (e.g. waiting for some resources to become available). If so, it returns a future that completes when the operator becomes unblocked. If the operator is not blocked, this method returns Operator.NOT_BLOCKED which is an already completed future.
      Specified by:
      isBlocked in interface Operator

    • getOutput

      public Page getOutput()
      Description copied from interface: Operator
      returns non-null if output page available. Only called when isFinished() == false
      Specified by:
      getOutput in interface Operator

    • close

      public void close()
      Description copied from interface: Operator
      notifies the operator that it won't be used anymore (i.e. none of the other methods called), and its resources can be cleaned up
      Specified by:
      close in interface AutoCloseable
      Specified by:
      close in interface Closeable
      Specified by:
      close in interface Operator
      Specified by:
      close in interface Releasable

    • ramBytesUsed

      public long ramBytesUsed()
      Specified by:
      ramBytesUsed in interface org.apache.lucene.util.Accountable

    • toString

      public String toString()
      Overrides:
      toString in class Object