Package org.elasticsearch.xpack.esql.datasources

Class SourceExtractors

java.lang.Object
org.elasticsearch.xpack.esql.datasources.SourceExtractors
All Implemented Interfaces:
Closeable, AutoCloseable, Releasable
public final class SourceExtractors extends Object implements Releasable
Driver-scoped lookup table of per-file ColumnExtractors, addressed by a compact int id allocated at registration time. Acts as the rendezvous between an AsyncExternalSourceOperatorFactory (which registers a new extractor each time it opens a file) and an ExternalFieldExtractOperator (which decodes the synthetic _rowPosition carried on pages and routes extraction requests to the correct extractor).

_rowPosition encoding

The synthetic column flowing between source and extract carries a long per row that packs the extractor id and the file-local position into a single value: 
   bit  63       reserved zero (keeps the encoded value non-negative for signed long order)
   bits 62..48   extractor id (15 bits, 0..32767)
   bits 47..0    file-local row position (48 bits, 0..2^48-1)
Concretely: encoded = ((long) id << 48) | (localPos & MAX_LOCAL_POSITION). This keeps the row identifier a primitive long, so the existing LongBlock type carries it end-to-end without any new block-type plumbing. Bit 63 is deliberately reserved so every encoded value is non-negative; that way Java's signed-long ordering (used by ESQL's TopN encoder, which calls NumericUtils.longToSortableBytes) coincides with (id, localPos) lexicographic order, and the encoding remains usable as a stable tiebreaker should anything downstream sort by it.

Lifecycle

One SourceExtractors per driver. The source operator registers extractors during the async producer loop; the extract operator reads from it on the driver thread. Both sides hold the same instance via AsyncExternalSourceOperatorFactory#sourceExtractorsFor(DriverContext). The extract operator owns close(); closing is idempotent.

Threading

register(org.elasticsearch.xpack.esql.datasources.spi.ColumnExtractor) is invoked from source executor threads (one at a time per driver; the source loop is single-producer within a driver). get(int) and materialize(long[], int, java.util.List<java.lang.String>, java.util.List<org.elasticsearch.xpack.esql.core.type.DataType>, org.elasticsearch.compute.data.BlockFactory) are invoked from the driver thread. The data-flow invariant — the source publishes a page only after the file's extractor has been registered — establishes the necessary happens-before via the page buffer's internal synchronization. We use a synchronized ArrayList as a defensive choice; the cost is negligible for the few-dozen registrations per driver in practice.

  • Field Details

    • LOCAL_POSITION_BITS

      public static final int LOCAL_POSITION_BITS
      Number of bits reserved for the file-local row position; the rest go to the extractor id. Mirrors ColumnExtractor.LOCAL_POSITION_BITS (the SPI-side constant iterator implementations OR against when they pre-encode _rowPosition values); kept aligned via the assert below so a divergent edit surfaces in tests instead of as a wire- format mismatch.
      See Also:

    • MAX_LOCAL_POSITION

      public static final long MAX_LOCAL_POSITION
      Maximum representable file-local row position (inclusive): 2^48 - 1.
      See Also:

    • MAX_EXTRACTOR_ID

      public static final int MAX_EXTRACTOR_ID
      Maximum representable extractor id (inclusive): 2^15 - 1 = 32 767. We use 15 bits (not the 16 available between LOCAL_POSITION_BITS and Long.SIZE) so that bit 63 stays clear and every encoded value is non-negative — see class-level javadoc.
      See Also:

  • Constructor Details

    • SourceExtractors

      public SourceExtractors()

  • Method Details

    • encode

      public static long encode(int extractorId, long localPosition)
      Pack an (extractorId, localPosition) pair into a single long. The packed value is the on-the-wire representation of _rowPosition downstream of the source.

    • decodeExtractorId

      public static int decodeExtractorId(long encoded)
      Extract the extractor id (high 15 bits) from an encoded row reference.

    • decodeLocalPosition

      public static long decodeLocalPosition(long encoded)
      Extract the file-local position (low 48 bits) from an encoded row reference.

    • register

      public int register(ColumnExtractor extractor)
      Register a new ColumnExtractor and return the id assigned to it. Ids are allocated sequentially starting at 0, in registration order. The returned id is what callers must pack into the _rowPosition values they emit for rows coming from this extractor's file.
      Throws:
      IllegalStateException - if the table is closed or if the id space is exhausted

    • get

      public ColumnExtractor get(int id)
      Look up a registered extractor by id. Throws IllegalArgumentException for ids that were never assigned — this should never happen in normal operation because the id always comes from a previously-encoded row reference, so failure here indicates a wiring bug.

    • size

      public int size()
      Number of extractors currently registered.

    • registerTrailingCloseable

      public void registerTrailingCloseable(Closeable closeable)
      Attach a closeable whose lifecycle must extend beyond the source operator's. The closeable is closed during close() after all registered extractors, so any storage objects the extractors hold get released before the trailing resource (the per-query concurrency budget) is torn down.

      If the registry has already been closed, the closeable is closed immediately to preserve the invariant that registered resources are eventually closed exactly once.

    • materialize

      public Block[] materialize(long[] encodedRefs, int count, List<String> columns, List<DataType> targetTypes, BlockFactory factory)
      Materialize columns for the rows referenced by encodedRefs[0..count). Output blocks preserve the input order: row i of each output block corresponds to encodedRefs[i].

      Algorithm: classify each input ref by its extractor id, build per-extractor packed arrays of file-local positions, dispatch one ColumnExtractor.extract(java.lang.String[], org.elasticsearch.xpack.esql.core.type.DataType[], long[], org.elasticsearch.compute.data.BlockFactory) call per (column, id) pair, then reassemble the output by walking the original slot order. The grouping makes each underlying extractor see a single batched call per column rather than one call per row, which matters for parquet because each call walks the file forward.

      Parameters:
      encodedRefs - encoded row references (as emitted via _rowPosition)
      count - number of valid entries in encodedRefs
      columns - column names to materialize
      targetTypes - planner/declared type per column, aligned with columns; threads the declared-type coercion contract into ColumnExtractor.extract(java.lang.String[], org.elasticsearch.xpack.esql.core.type.DataType[], long[], org.elasticsearch.compute.data.BlockFactory) (null — the list or an entry — disables coercion for that column)
      factory - block factory for memory accounting
      Returns:
      one Block per column, each with exactly count rows; in caller-supplied order

    • close

      public void close()
      Specified by:
      close in interface AutoCloseable
      Specified by:
      close in interface Closeable
      Specified by:
      close in interface Releasable