Package org.elasticsearch.xpack.esql.datasources.cache

Class StripeStatsHarvester

java.lang.Object
org.elasticsearch.xpack.esql.datasources.cache.StripeStatsHarvester
public final class StripeStatsHarvester extends Object
Shared canonical-stripe statistics harvester for the row-format readers (CSV / TSV / NDJSON). Both readers attribute each record to its canonical stripe by the record's OWN file-global start offset (ordinal = floor(recordStartOffset / B)) and feed this harvester; at close the harvester emits one stripe-addressed contribution per stripe the chunk's byte range overlaps, using the byte-range cover model.

The byte-range cover model is the one proven correct at multi-stripe / multi-chunk scale: the emit loop walks every stripe ordinal the chunk's half-open byte range [splitStartByte, chunkAbsEnd) overlaps — including stripes with no records (a stripe whose first record lands in the next chunk; this chunk still owns that stripe's left edge and must anchor it). Each fragment's byte sub-range is the chunk range clamped to the stripe's grid cell ([max(splitStartByte, k*B), min(chunkAbsEnd, (k+1)*B))), so sibling chunks' fragments for a split stripe tile contiguously and the coordinator's per-stripe interval-cover fold reaches whole-file completeness. Cover anchors are pure byte-range-overlap predicates:

  • atStart — this chunk covers the stripe's left grid line (splitStartByte <= k*B).
  • atEnd — this chunk covers the right grid line (chunkAbsEnd >= (k+1)*B) or this is the file-final chunk.
  • eof — file-final chunk and this is its last stripe.
Record attribution is by each record's own start, so per-stripe row counts are scan-invariant; misaligned sibling tilings collapse to one answer.

The harvester owns the per-stripe StripeStatsHarvester.StripeAccum (rows + projected cols + all-schema allCols) and the emit loop. The readers own the per-format extraction of typed values into those accumulators (CSV from the output page / raw record, NDJSON from the output page / widened decoded page), because the value-extraction differs by format; everything downstream of attribution is shared here.

Not thread-safe: each instance is owned by exactly one batch iterator over that iterator's lifetime.

  • Constructor Details

    • StripeStatsHarvester

      public StripeStatsHarvester(long stripeSize, boolean fileFinal)

  • Method Details

    • ordinalOf

      public long ordinalOf(long recordStartOffset)
      Stripe ordinal a record starting at recordStartOffset is attributed to.

    • isEmpty

      public boolean isEmpty()
      Whether any stripe has been touched (drives the close-time emit-vs-skip decision).

    • getOrCreate

      public StripeStatsHarvester.StripeAccum getOrCreate(long ordinal)
      Returns the accumulator for ordinal, creating it (zero rows, no columns) on first touch.

    • forEachRun

      public boolean forEachRun(long[] offsets, int recordCount, int positionCount, StripeStatsHarvester.StripeRunConsumer consumer)
      Walks a page's positions, splits them into maximal runs of consecutive same-stripe rows (by each row's own start offset), and calls consumer once per run with the run's stripe accumulator. This is the ONE run-bucketing walk both row-format readers share -- CSV and NDJSON used to inline the identical loop, which is where the offset/page alignment invariant drifted between them (the A1/F1 class).

      Fail loud: the caller only reaches here on a tracked page, so it MUST present exactly one record offset per emitted position. A missing array or a count mismatch is a page-builder/decoder bug -- assert in dev/test builds; in production return false so the caller safe-misses (disables capture) rather than mis-attributing. The consumer decides what to fold (rows and/or columns): the walk itself counts nothing, so a reader may run it twice over the same page (e.g. once for all-schema columns, once for projected + rows) without double-counting.

      Parameters:
      offsets - per-record file-global start offsets, indexed 0..recordCount-1 (may be null)
      recordCount - the number of valid offsets (== the emitted record count)
      positionCount - the page's position count
      Returns:
      true if the walk ran (offsets aligned); false if alignment was lost (caller safe-misses)

    • mergeColumnStats

      public static Map<String,ExternalStats.ColumnStats> mergeColumnStats(ColumnStatsAccumulator projected, ColumnStatsAccumulator all)
      Folds the projected accumulator and the ALL-scope full-schema accumulator into one committed column-stats map. The full-schema map (when present) is a superset of the projected one and the two agree on shared columns, so the projected map is overlaid first and the full-schema map fills in (and re-affirms) the rest. Either may be null: COUNT commits neither, PROJECTED only the projected one, ALL both.

      Invariant the merged.putAll(allSnapshot) relies on: all must be a strict superset of projected over the SAME rows, so the two accumulators agree on every shared column and ALL simply wins on the overlap. Passing two accumulators that disagree on a shared column (e.g. computed over different row sets) would silently let the ALL value clobber the PROJECTED one — never do that.

    • emit

      public void emit(String sourceLocation, long splitStartByte, long chunkBytes, long pinnedMtimeMillis, String fingerprint, List<Attribute> schema)
      Emits one stripe-addressed fragment for every stripe the chunk's byte range [splitStartByte, splitStartByte + chunkBytes) overlaps. A chunkBytes <= 0 (unknown or empty byte range) is a safe miss — nothing is emitted, so a warm aggregate re-scans rather than serving a wrong answer. Each emitted contribution carries the well-known _stats.* addressing keys the coordinator's interval-cover fold consumes.
      Parameters:
      sourceLocation - file path key for the capture sink
      splitStartByte - file-global byte offset of this chunk's first byte (decompressed-stream coordinate)
      chunkBytes - bytes this chunk consumed (decompressed coordinate); <= 0 ⇒ safe miss
      pinnedMtimeMillis - mtime pinned at iterator open
      fingerprint - config fingerprint over the full file schema
      schema - full file schema (drives the per-column serialization)