Package org.elasticsearch.xpack.esql.datasources.spi

Interface FormatReader

All Superinterfaces:
AutoCloseable, Closeable
All Known Subinterfaces:
NoConfigFormatReader, RangeAwareFormatReader, SegmentableFormatReader
public interface FormatReader extends Closeable
Unified interface for reading data formats.

Simple formats: implement only read(StorageObject, FormatReadContext) (sync) - async wrapping is automatic. Async-capable formats: override readAsync(StorageObject, FormatReadContext, Executor, ActionListener) for native async behavior.

The output is ESQL's native Page format rather than Arrow to avoid mandating Arrow as a dependency for all format implementations.

Implementations should provide metadata discovery via metadata(StorageObject) which returns a unified SourceMetadata containing schema and source information.

Per-query format configuration (delimiter, encoding, etc.) is set on the reader instance via withConfig(Map). Per-query optimizer hints (pushed filters for row-group or stripe skipping) are set via withPushedFilter(Object). Per-read execution parameters (projection, batch size, limit, error policy, split config) are bundled in FormatReadContext.

  • Field Details

    • NO_LIMIT

      static final int NO_LIMIT
      See Also:

    • DEFAULT_SCHEMA_RESOLUTION

      static final FormatReader.SchemaResolution DEFAULT_SCHEMA_RESOLUTION
      Cluster-wide default schema resolution strategy when a query does not specify one.

      This is the single source of truth: it is consulted both by this SPI's defaultSchemaResolution() and by ExternalSourceResolver.parseSchemaResolution when no schema_resolution key is present in the per-query config. The format detected at glob-expansion time is not yet known when the resolver decides whether to take the read-all-and-reconcile path versus the FFW fast path, so there is no format dispatch here today; if per-format defaults become desirable in the future the resolver will need to peek at the lex-smallest file's format first, and this constant becomes the fallback only.

  • Method Details

    • defaultSchemaResolution

      default FormatReader.SchemaResolution defaultSchemaResolution()
      Returns the cluster-wide default schema resolution for this reader. Format implementations may override this to advertise a different preferred default, but the resolver does not consult it today (see DEFAULT_SCHEMA_RESOLUTION for the rationale). Override is effectively informational until that wiring exists.

    • defaultErrorPolicy

      default ErrorPolicy defaultErrorPolicy()
      Returns the default error policy for this format. The base default is ErrorPolicy.STRICT (fail_fast) and every format inherits it, so a bad per-value coercion fails the read across all formats unless the user opts into error_mode: null_field. Pinned per reader by a testDefaultErrorPolicyIsStrict guard; do not override to a lenient default (that would silently diverge one format from the others).

    • metadata

      SourceMetadata metadata(StorageObject object) throws IOException
      Throws:
      IOException

    • metadataAsync

      default void metadataAsync(StorageObject object, Executor executor, ActionListener<SourceMetadata> listener)
      Asynchronously resolves metadata for the given storage object.

      The default wraps the synchronous metadata(StorageObject) in the provided executor, mirroring readAsync(org.elasticsearch.xpack.esql.datasources.spi.StorageObject, org.elasticsearch.xpack.esql.datasources.spi.FormatReadContext, java.util.concurrent.Executor, org.elasticsearch.action.ActionListener<org.elasticsearch.compute.operator.CloseableIterator<org.elasticsearch.compute.data.Page>>). Formats whose footer/metadata read can be issued without holding an executor thread across the network round-trip (e.g. Parquet via StorageObject.readBytesAsync(long, long, org.elasticsearch.xpack.esql.datasources.spi.DirectBufferFactory, java.util.concurrent.Executor, org.elasticsearch.action.ActionListener<org.elasticsearch.xpack.esql.datasources.spi.DirectReadBuffer>)) should override this so that a wide discovery fan-out is bounded by an in-flight permit rather than by the number of executor threads it pins.

    • schema

      default List<Attribute> schema(StorageObject object) throws IOException
      Throws:
      IOException

    • read

      CloseableIterator<Page> read(StorageObject object, FormatReadContext context) throws IOException
      Reads data from the given storage object using the provided context.

      This is the primary read method. All implementations must override this method.

      Throws:
      IOException

    • read

      default CloseableIterator<Page> read(StorageObject object, List<String> projectedColumns, int batchSize) throws IOException
      Convenience overload that delegates to read(StorageObject, FormatReadContext). Keeps test code and simple call sites working without constructing a context.
      Throws:
      IOException

    • readAsync

      default void readAsync(StorageObject object, FormatReadContext context, Executor executor, ActionListener<CloseableIterator<Page>> listener)
      Asynchronously reads data from the given storage object using the provided context.

      The default wraps the synchronous read(StorageObject, FormatReadContext) in the provided executor. Formats with native async support should override this.

    • formatName

      String formatName()

    • fileExtensions

      List<String> fileExtensions()

    • withConfig

      default FormatReader withConfig(Map<String,Object> config)
      Returns a reader configured from the input config map. Default delegates to withConfigTrackingConsumedKeys(Map) and discards the consumed-keys set; use this overload when the caller does not need to validate against the consumed keys.

      Override target: implementations must override withConfigTrackingConsumedKeys(Map), NOT this method. The default withConfig delegates through the tracking variant, so an override here alone would be silently bypassed by every caller. The tracking variant is the single configuration entry point for the SPI.

    • withConfigTrackingConsumedKeys

      Configured<FormatReader> withConfigTrackingConsumedKeys(Map<String,Object> config)
      Returns a reader configured from the input config map, paired with the keys consumed from it.

      Required override. Every reader must explicitly declare which keys it claims, even if the answer is "none" (return Configured.empty(this)). The previous default silently dropped any unknown keys; that footgun is the reason this is no longer optional. Implementations that read configuration from the map should override this method (not withConfig(Map)); the consumed-keys set is required by ConfigKeyValidator for unknown-key rejection at planning time.

    • withPushedFilter

      default FormatReader withPushedFilter(Object pushedFilter)
      Returns a format reader configured with the given pushed filter from the optimizer.

      The pushed filter is an opaque object produced by FilterPushdownSupport during local physical optimization. Only format readers that support predicate pushdown (e.g., Parquet row-group skipping, ORC stripe-level predicates) need to override this.

      The filter is per-query: it applies identically to every file/split in the query. Implementations should cast the filter to their expected type and return a new reader instance with the filter stored as an instance field.

      Parameters:
      pushedFilter - opaque filter object, or null if no filter was pushed
      Returns:
      a new reader with the filter applied, or this if the filter is not applicable

    • aggregatePushdownSupport

      default AggregatePushdownSupport aggregatePushdownSupport()
      Returns the aggregate pushdown support for this format. Only format readers with column statistics in their metadata (Parquet, ORC) override this.

    • withSchema

      default FormatReader withSchema(List<Attribute> schema)
      Returns a format reader configured with the schema attributes.

      The schema is determined during the planning phase (via metadata(StorageObject)) and is constant for all files/splits in a query. Passing it here allows the reader to skip re-reading/inferring the schema from the file header on every read, which is especially important for split-based reads where the split may start mid-file (no header available).

      Formats with embedded schemas (Parquet, ORC) may ignore this since they always read the schema from the file metadata.

      Parameters:
      schema - the planning-phase schema attributes, or null to clear
      Returns:
      a new reader with the schema set, or this if the schema is not needed

    • withDeclaredDateFormats

      default FormatReader withDeclaredDateFormats(Map<String,String> physicalNameToPattern)
      Returns a format reader that parses the given columns' dates with the given patterns instead of the ISO default / file-level datetime_format. Keyed by physical (file) column name — the caller (FileSourceFactory) has already applied any declared path rename. The patterns are ES DateFormatter patterns (named formats and || chains included), matching the dataset-put validation.

      Only the text formats (CSV/TSV, NDJSON) parse dates from text and override this. Columnar formats (Parquet, ORC) carry native typed values and keep the no-op default — a declared format on a columnar column is rejected upstream at query resolution, so it never reaches here.

      Parameters:
      physicalNameToPattern - per-column date patterns keyed by physical column name; empty for no declared formats
      Returns:
      a new reader applying the per-column formats, or this when none apply

    • withDeclaredTypeColumns

      default FormatReader withDeclaredTypeColumns(Set<String> physicalDeclaredColumns)
      Returns a format reader that treats the given columns as declared-type columns: their target type came from an explicit declaration rather than inference, which licenses a lossy read-time coercion toward it (e.g. a declared integer over an int64 file column narrows per value, null on overflow). An inferred target must never narrow — a cross-file clash widens-or-nulls. Keyed by physical (file) column name; the caller (FileSourceFactory) has already applied any declared path rename.

      Only the by-name columnar formats (Parquet, ORC) make a whole-column incompatibility null-fill decision and override this — a declared column keeps the coercion escape, an inferred column null-fills whenever the file type is not widening-compatible. The text formats (CSV/TSV, NDJSON) parse straight into the target and keep the no-op default (their per-field failures are governed by the ErrorPolicy, not a whole-column type check).

      Parameters:
      physicalDeclaredColumns - physical names of the declared-type columns; empty when no column type was declared
      Returns:
      a new reader honoring the declared-type set, or this when none apply

    • filterPushdownSupport

      default FilterPushdownSupport filterPushdownSupport()
      Returns the filter pushdown support for this format, or null if not supported.

      When non-null, the optimizer can translate ESQL filter expressions into format-specific predicates (e.g., Parquet FilterPredicate) that enable row-group skipping via statistics, dictionary, and bloom filter checks.

      Returns:
      FilterPushdownSupport for this format, or null if not supported

    • supportsNativeAsync

      default boolean supportsNativeAsync()

    • supportsWholeFileCompression

      default boolean supportsWholeFileCompression()
      Whether this format supports being wrapped in a whole-file, stream-only decompressor (e.g. .parquet.zst or .orc.gz). Sequential formats (CSV, NDJSON) return the default true. Tail/footer-based formats (Parquet, ORC) must override to false because they require random access and a known decompressed length. This flag does NOT affect a format's own internal compression (e.g. Parquet column-chunk zstd).

    • statusSnapshot

      default FormatReaderStatus statusSnapshot()
      Returns a typed snapshot of format-reader I/O counters, or null when the reader tracks none. The snapshot is folded into the format_reader field of the external-source operator status.

    • rowPositionStrategy

      RowPositionStrategy rowPositionStrategy()
      Returns this reader's RowPositionStrategy — the dispatcher applies it polymorphically to wrap (or pass through) the reader's emitted page iterator so each page has the _rowPosition slot populated. Every reader must explicitly declare a strategy: a PassThroughRowPositionStrategy when the reader natively fills the slot in its own iterator (parquet-mr, ORC, CSV, NDJSON), a NullSpliceRowPositionStrategy when the reader has no row-position channel and the slot must surface NULL (parquet-rs), or a future strategy that injects the column from per-page reader state. There is no default — readers that "don't care" still participate, by returning PassThroughRowPositionStrategy.