[META] Parquet DocValues codec: serve Parquet-primary columns as Lucene doc values

[manaslohani]

Please describe the end goal of this project

Summary

This is the DSL-compatibility fallback for the DSL Query Executor Plugin
(#20914). That plugin intercepts _search and routes
supported queries to the analytics engine (Calcite -> analytics executor); .This codec is used whenever a query is not routed
through #20914, so it still executes correctly on the native Lucene path.

1. Problem & goal

Goal: serve Parquet-resident columns as Lucene doc values at query time, so the standard DSL
aggregation path works unchanged. It is read-only — the composite indexing engine remains the
sole writer of Parquet files — and it is a permanent DSL-compatibility path, not scaffolding.

---

2. Where it lives

All new code lives in parquet-data-format:

• Types + native bridge: ParquetPhysicalType, FieldTypeMapping, the read functions in ffm.rs,
  and their RustBridge / NativeCall binds (NativeCall is in dataformat-native).
• Codec: ParquetColumnReader, the page cache, the five DocValues iterators, OrdinalTable,
  ParquetDocValuesProducer, ParquetDocValuesFormat.
• Routing: ParquetDocValuesLeafReader + ParquetDocValuesDirectoryReader, installed by
  ParquetDataFormatPlugin.onIndexModule. The leaf reader synthesizes a FieldInfo per
  Parquet-resident field and serves its values from the producer.

---

3. Components

   Lucene search / aggregation
            │  asks for doc values of field "age"
            ▼
   ParquetDocValuesDirectoryReader        (wraps the DirectoryReader)
            │  per leaf
            ▼
   ParquetDocValuesLeafReader             (synthesizes FieldInfo; overrides the 5 DV accessors)
            │  getNumeric / getSortedSet / ...
            ▼
   ParquetDocValuesProducer               (one per segment; picks the iterator per field)
            │
      ┌─────┴───────────────┬───────────────────────────┐
      ▼                     ▼                            ▼
   iter/Parquet*DocValues  OrdinalTable (keyword/ip)   DocValuesSkipper (null; not served)
      │                     │
      └─────────┬───────────┘
                ▼
   bridge/ParquetColumnReader              (Java side of the FFM bridge; page cache)
                │  RustBridge.* (MethodHandles)
                ▼
   ffm.rs  (native)                        (opens & decodes the Parquet column)

Type/descriptor helpers used throughout: ParquetPhysicalType (the 0–5 code exchanged with
Rust) and FieldTypeMapping (OpenSearch type → Lucene DocValuesType + Parquet physical type).

---

4. Read flow

Aggregation over Parquet-only age (a long), size:0. There are two phases: setup (get the
iterator once per segment) and per-document reads (called for every matching doc).

Setup — the aggregator obtains a DocValues iterator once:

1. Aggregator calls leafReader.getSortedNumericDocValues("age").
2. Leaf reader sees age has a synthetic (Parquet) FieldInfo, so it calls
   producer.getNumeric(fieldInfo).
3. Producer opens a ParquetColumnReader (→ native open_column_reader, which validates the
   column's physical type) and constructs a ParquetNumericDocValues iterator over it.
4. The iterator is returned up to the aggregator (wrapped as singleton SortedNumeric).

Per-document — the aggregator drives the returned iterator:

5. For each matching doc, the aggregator calls iterator.advanceExact(doc) — this call lands on
   the ParquetNumericDocValues from step 4, not on the leaf reader or producer. doc is used
   directly as the Parquet row (Lucene doc N == Parquet row N in a finished segment, guaranteed
   by the write/merge path — no docId→row translation).
6. The iterator asks its ParquetColumnReader for that row. If the row's page is not already cached,
   the reader calls native decode_page_at_row, which decodes the whole page (~20k rows) into
   the Java-side PageCache.
7. iterator.longValue() returns the value for doc from the cached page.

The first read of a page pays one FFM crossing; every subsequent row in that page is served from the
PageCache with no crossing. A retained forward cursor on the native side keeps an ascending scan
from reopening the row group (and re-decoding the dictionary) on the next page.

---

6. Native read path (ffm.rs)

The native side exposes a small C-ABI surface, additive to the existing writer/merge functions. All
row indices are global (file-relative).

Return convention. 0 = success (RC_OK); 1 = RC_OVERFLOW (a caller buffer was too small;
required sizes are written to the out-parameters so the caller grows and retries once); < 0 = a
negated error pointer. Overflow is a positive sentinel so it can never be mistaken for an error
pointer.

Functions.

• parquet_open_column_reader / parquet_close_column_reader — open a per-column reader (validating
  the column exists and its physical type matches), returning an opaque i64 handle; close it.
• parquet_open_column_reader_count — number of open handles (leak diagnostics).
• parquet_read_value_at_row / parquet_read_repeated_at_row — single-value / multi-value point
  reads at one row.
• parquet_get_column_num_pages / parquet_get_column_page_index — page count, then the per-page
  row-range jump table + stats (first row, file offset, size, null count, min/max).
• parquet_decode_page_at_row — decode the whole page containing a row into caller buffers: a
  per-row value buffer plus a packed presence bitset (one bit per row; distinguishes a real 0 from
  a null). Manages the retained forward cursor.
• parquet_evict_file_metadata — drop a file's cached metadata on shard close.

State & caching. A node-level FileMetadataCache (keyed by path) parses the footer and every
column's page layout once; Parquet files are immutable, so entries never need invalidation. Each open
column reader owns a handle in a registry, its row-group layout, a retained decode cursor, and reused
scratch buffers so steady-state decoding is allocation-free.

Two outputs, two buffers. A page decode writes values and presence into separate caller-owned
(off-heap) buffers: the value buffer holds one i64 slot per row (nulls = 0); the presence bitset
holds one bit per row. Both are required because a stored 0 is otherwise indistinguishable from a
null. expand_to_outbuf writes the whole decoded page into the value buffer; the Java PageCache
then serves any row in that page by index with no further FFM crossing.

---

7. Key classes & methods (LLD)

ParquetPhysicalType (enum)

INT32(0) INT64(1) FLOAT(2) DOUBLE(3) BOOL(4) BYTE_ARRAY(5).
int code() — the discriminant sent to Rust (parquet_open_column_reader's expected_type); MUST
match ffm.rs TYPE_*. boolean isPrimitive() — false only for BYTE_ARRAY.

FieldTypeMapping

record Mapping(DocValuesType singleValued, DocValuesType multiValued, ParquetPhysicalType physical).

• boolean isSupported(String type) — is there a mapping? (leaf reader gate)
• Mapping forType(String type) — throws IllegalArgumentException for unmapped types
• void validate(String field, String type, DocValuesType requested) — throws on incompatible DV type

Mapped: boolean/byte/short/integer/long/float/double/date/date_nanos → NUMERIC(+SORTED_NUMERIC);
keyword/ip → SORTED(+SORTED_SET); text/binary → BINARY. Unmapped types (half_float, unsigned_long,
scaled_float, token_count, version, wildcard, constant_keyword, flat_object, match_only_text) throw
rather than serve wrong values. (scaled_float is stored as a raw scaled long, so it is not a plain
INT64 mapping.)

ParquetColumnReader

static ParquetColumnReader open(Path, String column, ParquetPhysicalType, boolean repeated, BufferPool).

• Value readValueAtRow(long row) — record Value(boolean present, long bits)
• byte[] readBytesAtRow(long row) / byte[][] readRepeatedBytesAtRow(long)
• RepeatedValues readRepeatedAtRow(long row) — record RepeatedValues(long[] bits)
• void loadPageContaining(long row) → PageCache cache() — decode-a-page + cache
• ColumnPageIndex pageIndex()

Owns the native handle and its lifecycle; the page cache and buffer pool live on the Java side.

OrdinalTable

static OrdinalTable buildSingleValued(reader, numRows) / buildMultiValued(...);
int ordForRow(int row) (and ordForRow(row, i) / countForRow for multi); BytesRef lookupOrd(int ord);
int valueCount(). Honors Lucene's per-segment-ordinal contract (ordinals ascend with term order).

The five iterators

ParquetNumericDocValues, ParquetSortedNumericDocValues, ParquetBinaryDocValues
(all take a ParquetColumnReader); ParquetSortedDocValues, ParquetSortedSetDocValues (take an
OrdinalTable). Each implements advanceExact / nextDoc / value accessor; nulls report absent.

ParquetDocValuesProducer extends DocValuesProducer

Built per segment; verifies numRows == maxDoc at construction (a mismatch throws
IllegalStateException). Accessors getNumeric, getSortedNumeric, getBinary, getSorted,
getSortedSet build the matching iterator; getSkipper returns null (no skipper is served);
checkIntegrity, close.

ParquetDocValuesFormat extends DocValuesFormat

fieldsProducer(SegmentReadState) → new ParquetDocValuesProducer(...);
fieldsConsumer(...) throws (read-only). SPI name "ParquetDocValues".

ParquetDocValuesLeafReader extends SequentialStoredFieldsLeafReader

• static LeafReader wrapIfApplicable(LeafReader in, MapperService, QueryParquetStats) — wraps only
  when a Parquet file resolves for the segment AND the mapping has a supported field missing Lucene DV.
• Synthesizes a FieldInfo per Parquet-resident field (DV type from FieldTypeMapping;
  skip-index type NONE, since no field claims a skipper the producer does not serve).
• Overrides the five DV accessors to serve from the producer (delegates all other fields);
  getDocValuesSkipper → null for Parquet fields.
• Extends SequentialStoredFieldsLeafReader so _source retrieval (size>0) works through the
  derived-source path.

ParquetDocValuesDirectoryReader extends FilterDirectoryReader

static wrap(DirectoryReader, MapperService); its SubReaderWrapper calls
ParquetDocValuesLeafReader.wrapIfApplicable per leaf.

---

8. Design decisions & rationale

• Read a whole page per FFM crossing, cache it on the Java side. Parquet values are only
  decodable in page-sized units, and the DocValues API is random-access. Decoding the page once and
  serving subsequent rows from an off-heap PageCache amortizes the crossing over ~20k rows.
• Retained forward cursor. A Parquet column reader is a forward-only stream. Keeping it across
  ascending page decodes lets a scan skip forward instead of reopening the row group and re-decoding
  the dictionary each page; a backward jump or row-group change transparently reopens.
• Values and presence are separate outputs. A stored 0 is a valid value, so nullability is
  carried in a parallel presence bitset rather than sentinel-encoded into the value.
• Unmapped types throw. The mapping table rejects types it cannot faithfully represent, so a
  query never silently reads wrong values for an unsupported field.
• Off-heap value buffers, heap page index. The large per-page value buffer stays off-heap (stable
  address for native writes, no GC copy); the small page-index arrays are copied to the Java heap
  once at open, where they are convenient and GC-managed.

---

Supporting References

Parent issue: #19902

Issues

[]

Related component

Search

[Bukhtawar]

DocValuesSkipper (null; not served)

I guess we still implement this backed by Parquet page index?

What about DocIdStream on the lines of #19324

[manaslohani]

DocValuesSkipper (null; not served)

I guess we still implement this backed by Parquet page index?

What about DocIdStream on the lines of #19324

Yes, will update this