diff --git a/sandbox/plugins/parquet-data-format/benchmarks/src/main/java/org/opensearch/parquet/benchmark/DocValuesRandomAccessBenchmark.java b/sandbox/plugins/parquet-data-format/benchmarks/src/main/java/org/opensearch/parquet/benchmark/DocValuesRandomAccessBenchmark.java
index f09d64c92a890..3c67f93b0de21 100644
--- a/sandbox/plugins/parquet-data-format/benchmarks/src/main/java/org/opensearch/parquet/benchmark/DocValuesRandomAccessBenchmark.java
+++ b/sandbox/plugins/parquet-data-format/benchmarks/src/main/java/org/opensearch/parquet/benchmark/DocValuesRandomAccessBenchmark.java
@@ -21,8 +21,8 @@
import org.apache.arrow.vector.types.pojo.FieldType;
import org.apache.arrow.vector.types.pojo.Schema;
import org.opensearch.nativebridge.spi.ArrowExport;
+import org.opensearch.parquet.bridge.DataFusionColumnReader;
import org.opensearch.parquet.bridge.NativeParquetWriter;
-import org.opensearch.parquet.bridge.ParquetColumnReader;
import org.opensearch.parquet.bridge.ParquetSortConfig;
import org.opensearch.parquet.bridge.RustBridge;
import org.opensearch.parquet.codec.ParquetPhysicalType;
@@ -52,26 +52,30 @@
import java.util.concurrent.TimeUnit;
/**
- * JMH benchmark for {@link ParquetNumericDocValues#advanceExact} under different access
- * patterns, isolating the native page-decode path (`parquet_decode_page_at_row`).
+ * JMH benchmark for {@link ParquetNumericDocValues#advanceExact} over the DataFusion decode
+ * path ({@link DataFusionColumnReader}: retained Arrow cursor, AIMD batch window, page-level
+ * OffsetIndex skips, PageCache-resident batches).
+ *
+ * Successor to the retired benchmark of the same name that measured the deleted
+ * codec-native {@code ParquetColumnReader}; access patterns are kept identical so historical
+ * numbers remain roughly comparable.
*
*
Access patterns:
*
- * {@code sequentialScan} — ascending doc IDs: almost all L1/L2 PageCache hits, one FFM
- * decode per page (~rows/20k decodes total). Baseline; largely unaffected by decode-path
- * changes.
- * {@code randomAccess} — uniform random doc IDs across the whole file: with ~50 pages
- * resident-page hit probability is ~2%, so nearly every call is a cold page decode.
- * This is the pattern the zero-alloc/branchless decode work targets.
- * {@code pageMissPingPong} — alternates between the first and last page: every call
- * evicts the resident page, giving a pure worst-case decode measurement.
+ * {@code sequentialScan} — ascending doc IDs: the cursor's home turf; the AIMD window
+ * grows and almost every call is a resident-batch hit.
+ * {@code randomAccess} — uniform random doc IDs across the whole file: forward targets
+ * ride cheap skips, backward targets exercise {@code parquet_df_reset_iter}'s cheap
+ * cursor rewind (the fix that killed the reopen storm).
+ * {@code pageMissPingPong} — alternates between the first and last page: every call is
+ * a worst-case long-distance reposition.
+ * {@code openReadClose} — per-query lifecycle: open a fresh cursor, touch a few
+ * scattered docs, close. Measures the metadata/open overhead a real search pays once
+ * per (field, producer).
*
*
- * Parameters cover both presence-pack paths (required-column memset vs branchless
- * def-level compare) and both value-expand paths (all-present tight widening loop vs
- * bitset pop-and-scatter): {@code nullFraction=0.0} exercises the former of each pair,
- * {@code nullFraction=0.3} the latter. {@code columnType} covers the i64 passthrough
- * (INT64) and the SIMD-widening (INT32) conversions.
+ *
{@code nullFraction} covers the required-column and nullable presence paths;
+ * {@code columnType} covers the i64 passthrough (INT64) and widening (INT32) conversions.
*
*
Run with:
*
@@ -91,7 +95,7 @@ public class DocValuesRandomAccessBenchmark {
@Param({ "1000000" })
private int rows;
- /** Fraction of null rows. 0.0 → required-column fast paths; 0.3 → nullable scatter paths. */
+ /** Fraction of null rows. 0.0 → required-column fast paths; 0.3 → nullable presence paths. */
@Param({ "0.0", "0.3" })
private double nullFraction;
@@ -103,11 +107,13 @@ public class DocValuesRandomAccessBenchmark {
private static final int TARGETS = 100_000;
/** Rows per Arrow batch handed to the native writer during setup. */
private static final int BATCH_ROWS = 100_000;
+ /** Starting AIMD decode window, matching the production default. */
+ private static final int INITIAL_BATCH_SIZE = 32;
private BufferAllocator allocator;
private Path file;
private BufferPool bufferPool;
- private ParquetColumnReader reader;
+ private DataFusionColumnReader reader;
private ParquetNumericDocValues docValues;
private int[] randomTargets;
private int[] pingPongTargets;
@@ -120,9 +126,7 @@ public void setupTrial() throws Exception {
writeFile();
bufferPool = new BufferPool();
- String column = columnType.equals("INT64") ? "val_i64" : "val_i32";
- ParquetPhysicalType physical = columnType.equals("INT64") ? ParquetPhysicalType.INT64 : ParquetPhysicalType.INT32;
- reader = ParquetColumnReader.open(file, column, physical, false, bufferPool);
+ reader = DataFusionColumnReader.open(file, column(), physical(), false, bufferPool, INITIAL_BATCH_SIZE);
docValues = new ParquetNumericDocValues(reader, rows);
// Fixed seed so every fork/param combination replays the identical target sequence.
@@ -131,13 +135,21 @@ public void setupTrial() throws Exception {
for (int i = 0; i < TARGETS; i++) {
randomTargets[i] = random.nextInt(rows);
}
- // Alternate between the first and last page so every call is a page miss.
+ // Alternate between the first and last page so every call is a long reposition.
pingPongTargets = new int[TARGETS];
for (int i = 0; i < TARGETS; i++) {
pingPongTargets[i] = (i % 2 == 0) ? (i / 2) % 1000 : rows - 1 - ((i / 2) % 1000);
}
}
+ private String column() {
+ return columnType.equals("INT64") ? "val_i64" : "val_i32";
+ }
+
+ private ParquetPhysicalType physical() {
+ return columnType.equals("INT64") ? ParquetPhysicalType.INT64 : ParquetPhysicalType.INT32;
+ }
+
@TearDown(Level.Trial)
public void tearDownTrial() throws Exception {
if (reader != null) {
@@ -152,8 +164,8 @@ public void tearDownTrial() throws Exception {
}
/**
- * Baseline: ascending scan of the whole column. L1/L2 hit rate ≈ (1 - pages/rows);
- * decode cost is amortized over ~20k rows per FFM call.
+ * Baseline: ascending scan of the whole column. The AIMD window converges to its ceiling
+ * and per-call cost is a resident-batch array read.
*/
@Benchmark
@OperationsPerInvocation(1_000_000)
@@ -166,9 +178,8 @@ public void sequentialScan(Blackhole bh) throws IOException {
}
/**
- * The optimized case: uniform random targets, ~98% resident-page misses, so throughput
- * is dominated by the native page decode (scratch reuse + branchless presence pack +
- * direct-to-outbuf expand).
+ * Uniform random targets: mostly resident misses. Forward jumps use page skips; backward
+ * jumps use the cheap cursor rewind instead of a file reopen.
*/
@Benchmark
@OperationsPerInvocation(TARGETS)
@@ -180,7 +191,7 @@ public void randomAccess(Blackhole bh) throws IOException {
}
}
- /** Worst case: every advanceExact evicts the resident page — a pure cold-decode measurement. */
+ /** Worst case: every advanceExact is a full-length reposition (first page ⇄ last page). */
@Benchmark
@OperationsPerInvocation(TARGETS)
public void pageMissPingPong(Blackhole bh) throws IOException {
@@ -192,23 +203,21 @@ public void pageMissPingPong(Blackhole bh) throws IOException {
}
/**
- * Per-query lifecycle cost: open a fresh column reader (the once-per-field-per-query step a
- * real search pays for every producer), read a handful of scattered docs, close. Dominated
- * by {@link ParquetColumnReader#open}'s metadata work — schema resolution + page-layout
- * (OffsetIndex/ColumnIndex) computation + ColumnPageIndex marshal — which the node-level
- * file-metadata cache converts from a per-open parse into an Arc-clone lookup. This is the
- * benchmark that shows the "dvm-equivalent" win; the decode benchmarks above open once per
- * trial and cannot see it.
+ * Per-query lifecycle cost: open a fresh cursor (the once-per-field-per-query step a real
+ * search pays for every producer), read a handful of scattered docs, close. Dominated by
+ * {@code parquet_df_open_iter}'s metadata work, which the node-level metadata caches
+ * convert from a per-open parse into a lookup.
*/
@Benchmark
public long openReadClose(Blackhole bh) throws IOException {
- String column = columnType.equals("INT64") ? "val_i64" : "val_i32";
- ParquetPhysicalType physical = columnType.equals("INT64") ? ParquetPhysicalType.INT64 : ParquetPhysicalType.INT32;
- try (BufferPool pool = new BufferPool(); ParquetColumnReader r = ParquetColumnReader.open(file, column, physical, false, pool)) {
+ try (
+ BufferPool pool = new BufferPool();
+ DataFusionColumnReader r = DataFusionColumnReader.open(file, column(), physical(), false, pool, INITIAL_BATCH_SIZE)
+ ) {
ParquetNumericDocValues dv = new ParquetNumericDocValues(r, rows);
long sum = 0;
// Touch a few scattered docs so the open isn't dead-code-eliminated and the reader
- // exercises a realistic first-access pattern (a couple of page decodes).
+ // exercises a realistic first-access pattern.
for (int i = 0; i < 8; i++) {
int t = randomTargets[i * (TARGETS / 8)];
if (dv.advanceExact(t)) {
diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetDataFormatPlugin.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetDataFormatPlugin.java
index 56266ff538fdd..fe91cf30c4f96 100644
--- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetDataFormatPlugin.java
+++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetDataFormatPlugin.java
@@ -158,12 +158,22 @@ public Collection createComponents(
ParquetDocValuesProducer.setDecodePath(ParquetSettings.DOCVALUES_DECODE_PATH.get(this.settings));
ParquetDocValuesProducer.setInitialBatchSize(ParquetSettings.DOCVALUES_INITIAL_BATCH_SIZE.get(this.settings));
ParquetDocValuesProducer.setDiagnostics(ParquetSettings.DOCVALUES_DIAGNOSTICS.get(this.settings));
+ ParquetDocValuesProducer.setDictionaryMaxTerms(ParquetSettings.DOCVALUES_DICTIONARY_MAX_TERMS.get(this.settings));
+ ParquetDocValuesProducer.setDictionaryCacheBytes(ParquetSettings.DOCVALUES_DICTIONARY_CACHE_BYTES.get(this.settings));
+ ParquetDocValuesProducer.setUninvertMaxDiskBytes(ParquetSettings.DOCVALUES_UNINVERT_MAX_DISK_BYTES.get(this.settings));
+ org.opensearch.parquet.codec.UninvertedOrdinalsCache.setOrdsDir(environment.dataFiles()[0].resolve("parquet-ords"));
clusterService.getClusterSettings()
.addSettingsUpdateConsumer(ParquetSettings.DOCVALUES_DECODE_PATH, ParquetDocValuesProducer::setDecodePath);
clusterService.getClusterSettings()
.addSettingsUpdateConsumer(ParquetSettings.DOCVALUES_INITIAL_BATCH_SIZE, ParquetDocValuesProducer::setInitialBatchSize);
clusterService.getClusterSettings()
.addSettingsUpdateConsumer(ParquetSettings.DOCVALUES_DIAGNOSTICS, ParquetDocValuesProducer::setDiagnostics);
+ clusterService.getClusterSettings()
+ .addSettingsUpdateConsumer(ParquetSettings.DOCVALUES_DICTIONARY_MAX_TERMS, ParquetDocValuesProducer::setDictionaryMaxTerms);
+ clusterService.getClusterSettings()
+ .addSettingsUpdateConsumer(ParquetSettings.DOCVALUES_DICTIONARY_CACHE_BYTES, ParquetDocValuesProducer::setDictionaryCacheBytes);
+ clusterService.getClusterSettings()
+ .addSettingsUpdateConsumer(ParquetSettings.DOCVALUES_UNINVERT_MAX_DISK_BYTES, ParquetDocValuesProducer::setUninvertMaxDiskBytes);
// Register virtual pools if allocator is available (arrow-base loaded)
if (nativeAllocator != null) {
@@ -253,6 +263,13 @@ public Map getStoreStrategies(IndexSettings indexSett
return Map.of(parquetFormat, storeStrategy);
}
+ @Override
+ public void close() throws java.io.IOException {
+ // Abort any in-flight uninverted-ordinal builds so node shutdown is not delayed.
+ org.opensearch.parquet.codec.UninvertedOrdinalsCache.shutdown();
+ super.close();
+ }
+
@Override
public List> getSettings() {
return ParquetSettings.getSettings();
diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetSettings.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetSettings.java
index d19932d090681..ceae3808b96af 100644
--- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetSettings.java
+++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetSettings.java
@@ -264,6 +264,41 @@ private ParquetSettings() {}
Setting.Property.Dynamic
);
+ /**
+ * Cardinality budget for dictionary-rank keyword ordinals: fields whose distinct-term count
+ * (from the Lucene sidecar's terms index) is at most this many get fully contract-compliant
+ * segment ordinals; larger fields stay on the streaming fail-fast path.
+ */
+ public static final Setting DOCVALUES_DICTIONARY_MAX_TERMS = Setting.intSetting(
+ "parquet.docvalues.dictionary.max_terms",
+ 65536,
+ 0,
+ Setting.Property.NodeScope,
+ Setting.Property.Dynamic
+ );
+
+ /** Node-wide heap budget for cached keyword term dictionaries. */
+ public static final Setting DOCVALUES_DICTIONARY_CACHE_BYTES = Setting.longSetting(
+ "parquet.docvalues.dictionary.cache_bytes",
+ 64 * 1024 * 1024,
+ 0,
+ Setting.Property.NodeScope,
+ Setting.Property.Dynamic
+ );
+
+ /**
+ * Node-wide disk budget for uninverted-ordinal files. When a new build would exceed it,
+ * unreferenced ord files are reclaimed oldest-first; if it still does not fit, the tier is
+ * refused for that field (consumers fall back to the streaming fail-fast path).
+ */
+ public static final Setting DOCVALUES_UNINVERT_MAX_DISK_BYTES = Setting.longSetting(
+ "parquet.docvalues.uninvert.max_disk_bytes",
+ 2L * 1024 * 1024 * 1024,
+ 0,
+ Setting.Property.NodeScope,
+ Setting.Property.Dynamic
+ );
+
/** Emits DataFusion cursor diagnostics for DocValues decoder benchmarking. */
public static final Setting DOCVALUES_DIAGNOSTICS = Setting.boolSetting(
"parquet.docvalues.diagnostics",
@@ -917,6 +952,9 @@ public static List> getSettings() {
DOCVALUES_DECODE_PATH,
DOCVALUES_INITIAL_BATCH_SIZE,
DOCVALUES_DIAGNOSTICS,
+ DOCVALUES_DICTIONARY_MAX_TERMS,
+ DOCVALUES_DICTIONARY_CACHE_BYTES,
+ DOCVALUES_UNINVERT_MAX_DISK_BYTES,
MERGE_DEFERRED_COLUMN_THRESHOLD,
WRITE_POOL_MIN,
WRITE_POOL_MAX,
diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/bridge/DataFusionColumnReader.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/bridge/DataFusionColumnReader.java
index fb2c00f83d88a..57d67657a9aef 100644
--- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/bridge/DataFusionColumnReader.java
+++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/bridge/DataFusionColumnReader.java
@@ -47,6 +47,39 @@ public final class DataFusionColumnReader implements Closeable, NumericPageReade
private static final byte[] EMPTY_BYTES = new byte[0];
+ /** Distinguishes pool slots across reader instances: several dedicated readers may serve the
+ * same column concurrently (one per search slice), and slots must never be shared. */
+ private static final java.util.concurrent.atomic.AtomicLong INSTANCE_IDS = new java.util.concurrent.atomic.AtomicLong();
+
+ /**
+ * GC backstop for native cursors. Readers opened by segment-lifetime shared producers are
+ * handed to cache-retained iterators with no close hook; when such an iterator becomes
+ * unreachable, the cleaner releases its cursor instead of waiting for segment close.
+ * Explicit {@link #close()} remains the primary path and unregisters the action.
+ */
+ private static final java.lang.ref.Cleaner CLEANER = java.lang.ref.Cleaner.create();
+
+ /** Cursor handle shared with the cleaner action; cleared on explicit close. */
+ private static final class CursorState implements Runnable {
+ private final java.util.concurrent.atomic.AtomicLong handle;
+
+ CursorState(long handle) {
+ this.handle = new java.util.concurrent.atomic.AtomicLong(handle);
+ }
+
+ @Override
+ public void run() {
+ long stale = handle.getAndSet(CLOSED_HANDLE);
+ if (stale != CLOSED_HANDLE) {
+ try {
+ RustBridge.dfCloseIter(stale);
+ } catch (java.io.IOException e) {
+ // Nothing actionable during GC-driven cleanup.
+ }
+ }
+ }
+ }
+
private final BufferPool bufferPool;
private final Path file;
private final String column;
@@ -71,6 +104,7 @@ public final class DataFusionColumnReader implements Closeable, NumericPageReade
private final String slotPrefix;
private long handle;
+ private final CursorState cursorState;
private ColumnPageIndex pageIndex;
private PageCache cache;
private int outputRowsCapacity;
@@ -87,13 +121,15 @@ private DataFusionColumnReader(
int initialBatchSize
) {
this.handle = handle;
+ this.cursorState = new CursorState(handle);
+ CLEANER.register(this, cursorState);
this.file = file;
this.column = column;
this.type = type;
this.repeated = repeated;
this.bufferPool = bufferPool;
this.initialBatchSize = initialBatchSize;
- this.slotPrefix = "df:" + column + ":";
+ this.slotPrefix = "df:" + INSTANCE_IDS.incrementAndGet() + ":" + column + ":";
this.firstRowSlot = slotPrefix + "firstRow";
this.lastRowSlot = slotPrefix + "lastRow";
this.valueLenSlot = slotPrefix + "valueLen";
@@ -484,6 +520,7 @@ public void close() throws IOException {
long current = handle;
handle = CLOSED_HANDLE;
cache = null;
+ cursorState.handle.set(CLOSED_HANDLE);
RustBridge.dfCloseIter(current);
}
diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/OrdinalTable.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/OrdinalTable.java
deleted file mode 100644
index 48aa0b4cf2e11..0000000000000
--- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/OrdinalTable.java
+++ /dev/null
@@ -1,165 +0,0 @@
-/*
- * SPDX-License-Identifier: Apache-2.0
- *
- * The OpenSearch Contributors require contributions made to
- * this file be licensed under the Apache-2.0 license or a
- * compatible open source license.
- */
-
-package org.opensearch.parquet.codec;
-
-import org.apache.lucene.util.BytesRef;
-import org.opensearch.parquet.bridge.BinaryPageReader;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-/**
- * Per-segment ordinal table for {@code SortedDocValues} / {@code SortedSetDocValues} over a
- * Parquet {@code BYTE_ARRAY} (keyword/ip) column.
- *
- * Built lazily on first access via the naive multi-pass column scan described in the
- * design: collect distinct terms + per-row occurrences, sort terms lex-byte ascending, assign
- * ordinals, then materialise the per-row ordinal mapping. The single-valued layout uses
- * {@code rowOrdinals} ({@code -1} = missing); the multi-valued layout uses CSR
- * ({@code csrOffsets}/{@code csrOrdinals}) with each row's slice sorted ascending so
- * {@code nextOrd()} yields ascending ordinals.
- *
- *
Lexicographic byte ordering matches Lucene's {@code BytesRef} natural order (unsigned
- * byte comparison), so the resulting ordinals agree with Lucene's term-ordinal contract.
- */
-public final class OrdinalTable {
-
- /** Distinct terms in lex-byte order; {@code sortedTerms[ord]} is the term for ordinal {@code ord}. */
- private final byte[][] sortedTerms;
-
- /** Single-valued: ordinal per row, or {@code -1} when the row is null. */
- private final int[] rowOrdinals;
-
- /** Multi-valued: CSR offsets (length {@code N + 1}) into {@link #csrOrdinals}. */
- private final int[] csrOffsets;
-
- /** Multi-valued: flattened, per-row-ascending ordinals. */
- private final int[] csrOrdinals;
-
- private OrdinalTable(byte[][] sortedTerms, int[] rowOrdinals, int[] csrOffsets, int[] csrOrdinals) {
- this.sortedTerms = sortedTerms;
- this.rowOrdinals = rowOrdinals;
- this.csrOffsets = csrOffsets;
- this.csrOrdinals = csrOrdinals;
- }
-
- /** Number of distinct terms (ordinal count). */
- public int valueCount() {
- return sortedTerms.length;
- }
-
- /** Single-valued ordinal for {@code row}, or {@code -1} when missing. */
- public int ordForRow(int row) {
- return rowOrdinals[row];
- }
-
- /** Number of multi-valued ordinals for {@code row}. */
- public int countForRow(int row) {
- return csrOffsets[row + 1] - csrOffsets[row];
- }
-
- /** The {@code i}-th (0-based, ascending) multi-valued ordinal for {@code row}. */
- public int ordForRow(int row, int i) {
- return csrOrdinals[csrOffsets[row] + i];
- }
-
- /** Returns a fresh {@link BytesRef} over the term for {@code ord}. */
- public BytesRef lookupOrd(int ord) {
- return new BytesRef(sortedTerms[ord]);
- }
-
- /**
- * Builds the single-valued ordinal table by scanning {@code rows} of a keyword/ip column
- * through the configured binary batch reader.
- */
- public static OrdinalTable buildSingleValued(BinaryPageReader reader, int numRows) throws IOException {
- // Pass 1: collect distinct terms and per-row occurrences.
- Map> occurrences = new HashMap<>();
- for (int row = 0; row < numRows; row++) {
- byte[] v = reader.readBytesAtRow(row);
- if (v != null) {
- occurrences.computeIfAbsent(new BytesRef(v), k -> new ArrayList<>()).add(row);
- }
- }
- Build b = assignOrdinals(occurrences);
-
- // Pass 3: per-row single ordinal (-1 = missing).
- int[] rowOrdinals = new int[numRows];
- Arrays.fill(rowOrdinals, -1);
- for (Map.Entry> e : occurrences.entrySet()) {
- int ord = b.termToOrd.get(e.getKey());
- for (int r : e.getValue()) {
- rowOrdinals[r] = ord;
- }
- }
- return new OrdinalTable(b.sortedTerms, rowOrdinals, null, null);
- }
-
- /**
- * Builds the multi-valued ordinal table by scanning {@code rows} of a multi-valued
- * keyword/ip column through the configured repeated-binary batch reader. The per-row CSR
- * slice is sorted ascending and de-duplicated (set semantics).
- */
- public static OrdinalTable buildMultiValued(BinaryPageReader reader, int numRows) throws IOException {
- // Pass 1: collect distinct terms and, per row, the distinct set of terms present.
- Map> occurrences = new HashMap<>();
- // Track per-row term sets so we can de-duplicate within a row (SortedSet semantics).
- List> perRowTerms = new ArrayList<>(numRows);
- for (int row = 0; row < numRows; row++) {
- byte[][] vals = reader.readRepeatedBytesAtRow(row);
- List rowTerms = new ArrayList<>();
- if (vals != null) {
- for (byte[] v : vals) {
- BytesRef term = new BytesRef(v);
- occurrences.computeIfAbsent(term, k -> new ArrayList<>());
- rowTerms.add(term);
- }
- }
- perRowTerms.add(rowTerms);
- }
- Build b = assignOrdinals(occurrences);
-
- // Pass 3: build CSR with per-row de-duplicated, ascending ordinals.
- int[] csrOffsets = new int[numRows + 1];
- int[][] perRowOrds = new int[numRows][];
- for (int row = 0; row < numRows; row++) {
- // Distinct ordinals for this row.
- int[] ords = perRowTerms.get(row).stream().mapToInt(t -> b.termToOrd.get(t)).distinct().sorted().toArray();
- perRowOrds[row] = ords;
- csrOffsets[row + 1] = csrOffsets[row] + ords.length;
- }
- int[] csrOrdinals = new int[csrOffsets[numRows]];
- for (int row = 0; row < numRows; row++) {
- System.arraycopy(perRowOrds[row], 0, csrOrdinals, csrOffsets[row], perRowOrds[row].length);
- }
- return new OrdinalTable(b.sortedTerms, null, csrOffsets, csrOrdinals);
- }
-
- /** Intermediate build state: lex-sorted distinct terms + term→ordinal map. */
- private record Build(byte[][] sortedTerms, Map termToOrd) {
- }
-
- private static Build assignOrdinals(Map> occurrences) {
- List terms = new ArrayList<>(occurrences.keySet());
- terms.sort(Comparator.naturalOrder()); // BytesRef natural order = unsigned lex-byte.
- byte[][] sortedTerms = new byte[terms.size()][];
- Map termToOrd = new HashMap<>(terms.size() * 2);
- for (int ord = 0; ord < terms.size(); ord++) {
- BytesRef t = terms.get(ord);
- sortedTerms[ord] = Arrays.copyOfRange(t.bytes, t.offset, t.offset + t.length);
- termToOrd.put(t, ord);
- }
- return new Build(sortedTerms, termToOrd);
- }
-}
diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/ParquetDocValuesLeafReader.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/ParquetDocValuesLeafReader.java
index cde669728195b..8a9d506b1f6c5 100644
--- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/ParquetDocValuesLeafReader.java
+++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/ParquetDocValuesLeafReader.java
@@ -35,6 +35,9 @@
import org.opensearch.index.mapper.MappedFieldType;
import org.opensearch.index.mapper.MapperService;
import org.opensearch.parquet.codec.cache.QueryParquetStats;
+import org.opensearch.parquet.codec.iter.ParquetDictionarySortedDocValues;
+import org.opensearch.parquet.codec.iter.ParquetSortedDocValues;
+import org.opensearch.parquet.codec.iter.ParquetUninvertedSortedDocValues;
import java.io.IOException;
import java.util.ArrayList;
@@ -230,9 +233,28 @@ private synchronized ParquetDocValuesProducer producer() throws IOException {
producer.setQueryStats(queryStats);
producerInitialized = true;
}
+ if (producer != null && producer.isClosed()) {
+ // This wrapper outlived its request: a cache (fielddata, global ordinals) retained it
+ // and is calling back after the search closed the request producer. Serve through the
+ // segment-lifetime shared producer so cached consumers stay valid until the segment
+ // itself closes — the contract every reader-keyed cache in Lucene/OpenSearch assumes.
+ ParquetDocValuesProducer shared = SharedProducerRegistry.get(in.getCoreCacheHelper(), segmentReadState, mapperService);
+ if (shared == null) {
+ throw new IllegalStateException(
+ "doc values requested after the search closed and the segment has no core cache identity"
+ );
+ }
+ return shared;
+ }
return producer;
}
+ /** Whether the mapper types this field (or subfield) as keyword — values indexed verbatim. */
+ private boolean isKeywordField(String field) {
+ org.opensearch.index.mapper.MappedFieldType fieldType = mapperService.fieldType(field);
+ return fieldType != null && "keyword".equals(fieldType.typeName());
+ }
+
/** Returns the synthetic FieldInfo if the given field is served from Parquet, else null. */
private FieldInfo parquetFieldInfo(String field) {
return parquetFields.get(field);
@@ -338,12 +360,48 @@ public SortedDocValues getSortedDocValues(String field) throws IOException {
FieldInfo fi = parquetFieldInfo(field);
if (fi != null && fi.getDocValuesType() == DocValuesType.SORTED) {
RowIdResolver resolver = newRowIdResolver();
- SortedDocValues sorted = producer().getSorted(fi);
+ SortedDocValues sorted = withDictionaryOrdinals(field, producer().getSorted(fi));
return resolver == RowIdResolver.IDENTITY ? sorted : RowIdRemappingDocValues.sorted(sorted, resolver, maxDoc());
}
return in.getSortedDocValues(field);
}
+ /**
+ * Upgrades a streaming sorted iterator to fully contract-compliant segment ordinals when the
+ * field's cardinality fits the dictionary budget. The sorted term dictionary is read from
+ * the composite index's Lucene sidecar (O(distinct), cached per segment) — never from a row
+ * scan. Above-budget fields keep the streaming iterator, whose global-ordinal operations
+ * fail fast rather than materialize.
+ */
+ private SortedDocValues withDictionaryOrdinals(String field, SortedDocValues sorted) throws IOException {
+ // Ordinal tiers rank Parquet VALUES against the Lucene sidecar's TERMS, which only
+ // coincide for untokenized (keyword) fields. A text field's terms are analyzer tokens:
+ // ranking values against tokens would produce silently wrong ordinals. Text fields stay
+ // on the streaming iterator, whose global operations fail fast toward execution_hint:map.
+ if (isKeywordField(field) == false) {
+ return sorted;
+ }
+ if (sorted instanceof ParquetSortedDocValues streaming) {
+ TermDictionary dictionary = TermDictionaryCache.get(
+ in,
+ field,
+ ParquetDocValuesProducer.dictionaryMaxTerms(),
+ ParquetDocValuesProducer.dictionaryCacheBytes()
+ );
+ if (dictionary != null) {
+ return new ParquetDictionarySortedDocValues(streaming, dictionary);
+ }
+ // Above the dictionary budget: disk-backed uninverted ordinals (built once per
+ // segment from the sidecar's postings, memory-mapped, working-set resident).
+ long expectedNonNull = producer().nonNullRowCount(parquetFieldInfo(field));
+ UninvertedOrdinals uninverted = UninvertedOrdinalsCache.get(in, segmentReadState.segmentInfo, field, expectedNonNull);
+ if (uninverted != null) {
+ return new ParquetUninvertedSortedDocValues(uninverted, streaming, maxDoc());
+ }
+ }
+ return sorted;
+ }
+
@Override
public SortedSetDocValues getSortedSetDocValues(String field) throws IOException {
FieldInfo fi = parquetFieldInfo(field);
@@ -361,7 +419,7 @@ public SortedSetDocValues getSortedSetDocValues(String field) throws IOException
FieldInfo asSorted = fi.getDocValuesType() == DocValuesType.SORTED
? fi
: newDocValuesFieldInfo(field, fi.number, DocValuesType.SORTED, fi.docValuesSkipIndexType());
- SortedDocValues sorted = producer().getSorted(asSorted);
+ SortedDocValues sorted = withDictionaryOrdinals(field, producer().getSorted(asSorted));
RowIdResolver resolver = newRowIdResolver();
SortedDocValues remapped = resolver == RowIdResolver.IDENTITY
? sorted
@@ -427,6 +485,9 @@ protected StoredFieldsReader doGetSequentialStoredFieldsReader(StoredFieldsReade
// Cache helpers must delegate to the underlying reader so query/segment caches stay coherent.
@Override
public CacheHelper getCoreCacheHelper() {
+ // Full cache identity restored: filter cache, fielddata and global-ordinals caches all key
+ // off this. Consumers cached beyond the request remain valid because producer() reroutes
+ // post-close access to the segment-lifetime shared producer (SharedProducerRegistry).
return in.getCoreCacheHelper();
}
diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/ParquetDocValuesProducer.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/ParquetDocValuesProducer.java
index f7235aaad4055..3b42bb48a681a 100644
--- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/ParquetDocValuesProducer.java
+++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/ParquetDocValuesProducer.java
@@ -23,6 +23,7 @@
import org.opensearch.index.mapper.MappedFieldType;
import org.opensearch.index.mapper.MapperService;
import org.opensearch.parquet.ParquetSettings;
+import org.opensearch.parquet.bridge.BinaryPageReader;
import org.opensearch.parquet.bridge.DataFusionColumnReader;
import org.opensearch.parquet.bridge.ParquetColumnReader;
import org.opensearch.parquet.bridge.ParquetFileMetadata;
@@ -56,8 +57,8 @@
* Laziness and lifecycle
* The constructor resolves the Parquet file and checks the invariant but opens no column-reader
* handle. Each {@code getX(field)} lazily opens and caches the reader selected by the configured
- * decode path; {@code getSorted}/{@code getSortedSet} additionally build and cache an
- * {@link OrdinalTable} on first access. {@link #close()} releases every reader, ordinal table,
+ * decode path; {@code getSorted}/{@code getSortedSet} serve streaming
+ * per-document ordinals with no segment-wide ordinal structure (see ParquetSortedSetDocValues). {@link #close()} releases every reader, ordinal table,
* and the shared {@link BufferPool}, and is idempotent.
*
* Not thread-safe: one producer serves one segment on one query thread.
@@ -68,12 +69,41 @@ public final class ParquetDocValuesProducer extends DocValuesProducer {
private static volatile boolean useDataFusionDecodePath;
private static volatile int dataFusionInitialBatchSize = 32;
private static volatile boolean dataFusionDiagnostics;
+ private static volatile int dictionaryMaxTerms = 65536;
+ private static volatile long dictionaryCacheBytes = 64 * 1024 * 1024;
+ private static volatile long uninvertMaxDiskBytes = 2L * 1024 * 1024 * 1024;
/** Updates the node-wide DocValues decode path. */
public static void setDecodePath(String decodePath) {
useDataFusionDecodePath = ParquetSettings.DECODE_PATH_DATAFUSION.equals(decodePath);
}
+ /** Updates the cardinality budget for dictionary-rank keyword ordinals. */
+ public static void setDictionaryMaxTerms(int maxTerms) {
+ dictionaryMaxTerms = maxTerms;
+ }
+
+ /** Updates the node-wide heap budget for cached term dictionaries. */
+ public static void setDictionaryCacheBytes(long bytes) {
+ dictionaryCacheBytes = bytes;
+ }
+
+ static int dictionaryMaxTerms() {
+ return dictionaryMaxTerms;
+ }
+
+ static long dictionaryCacheBytes() {
+ return dictionaryCacheBytes;
+ }
+
+ public static void setUninvertMaxDiskBytes(long bytes) {
+ uninvertMaxDiskBytes = bytes;
+ }
+
+ static long uninvertMaxDiskBytes() {
+ return uninvertMaxDiskBytes;
+ }
+
/** Updates the starting window used by newly opened DataFusion cursors. */
public static void setInitialBatchSize(int initialBatchSize) {
dataFusionInitialBatchSize = initialBatchSize;
@@ -120,9 +150,9 @@ public static synchronized void setDiagnostics(boolean diagnostics) {
private final long parquetRowCount;
private final BufferPool bufferPool = new BufferPool();
- private final Map columnReaders = new HashMap<>();
- private final Map dataFusionColumnReaders = new HashMap<>();
- private final Map ordinalTables = new HashMap<>();
+ private final Map columnReaders = new java.util.concurrent.ConcurrentHashMap<>();
+ private final Map dataFusionColumnReaders = new java.util.concurrent.ConcurrentHashMap<>();
+ private final java.util.List dedicatedReaders = java.util.Collections.synchronizedList(new java.util.ArrayList<>());
/** Optional per-query accumulator; propagated to each column reader so its stats roll up at close. */
private QueryParquetStats queryStats;
@@ -224,16 +254,17 @@ public BinaryDocValues getBinary(FieldInfo field) throws IOException {
public SortedDocValues getSorted(FieldInfo field) throws IOException {
ensureOpen();
validate(field, DocValuesType.SORTED);
- OrdinalTable table = ordinalTableFor(field, false);
- return new ParquetSortedDocValues(table, maxDoc);
+ return new ParquetSortedDocValues(binaryReaderFor(field, false), maxDoc);
}
@Override
public SortedSetDocValues getSortedSet(FieldInfo field) throws IOException {
ensureOpen();
validate(field, DocValuesType.SORTED_SET);
- OrdinalTable table = ordinalTableFor(field, true);
- return new ParquetSortedSetDocValues(table, maxDoc);
+ // Convention (mirrors the ordinal-table era and the leaf reader's routing): SORTED_SET
+ // reaches this producer only for genuinely repeated columns; single-valued keywords are
+ // served through getSorted and wrapped with DocValues.singleton by the leaf reader.
+ return new ParquetSortedSetDocValues(binaryReaderFor(field, true), true, maxDoc);
}
/**
@@ -320,9 +351,18 @@ public void close() throws IOException {
logger.warn("Failed to close DataFusion column reader for [{}]", parquetFile, e);
}
}
+ for (java.io.Closeable reader : dedicatedReaders) {
+ try {
+ reader.close();
+ } catch (IOException | RuntimeException e) {
+ if (first == null && e instanceof IOException io) {
+ first = io;
+ }
+ }
+ }
+ dedicatedReaders.clear();
columnReaders.clear();
dataFusionColumnReaders.clear();
- ordinalTables.clear();
bufferPool.close();
if (first != null) {
throw first;
@@ -361,7 +401,7 @@ private ParquetPhysicalType physicalType(FieldInfo field) {
};
}
- private ParquetColumnReader readerFor(FieldInfo field, boolean repeated) throws IOException {
+ private synchronized ParquetColumnReader readerFor(FieldInfo field, boolean repeated) throws IOException {
ParquetColumnReader reader = columnReaders.get(field.getName());
if (reader == null) {
reader = ParquetColumnReader.open(parquetFile, field.getName(), physicalType(field), repeated, bufferPool);
@@ -371,7 +411,51 @@ private ParquetColumnReader readerFor(FieldInfo field, boolean repeated) throws
return reader;
}
- private DataFusionColumnReader dataFusionReaderFor(FieldInfo field, boolean repeated) throws IOException {
+
+ /**
+ * A dedicated (non-shared) binary reader for one streaming sorted iterator. Concurrent
+ * segment-search slices each obtain their own DocValues instance; sharing one forward
+ * cursor between them turns every access into a resident-page miss as the slices ping-pong
+ * the shared PageCache. Dedicated readers keep each slice's scan sequential. Registered for
+ * close with the producer.
+ */
+ /**
+ * Number of rows with a non-null value in this column, from the Parquet page index's
+ * per-page null counts; {@code -1} when any page lacks the statistic. Used to verify that
+ * postings-derived ordinal tables cover every stored value.
+ */
+ long nonNullRowCount(FieldInfo field) throws IOException {
+ org.opensearch.parquet.codec.cache.ColumnPageIndex idx = dataFusionReaderFor(field, false).pageIndex();
+ long nonNull = 0;
+ for (int page = 0; page < idx.pageCount(); page++) {
+ long nulls = idx.nullCountOf(page);
+ if (nulls < 0) {
+ return -1;
+ }
+ nonNull += idx.numRowsOf(page) - nulls;
+ }
+ return nonNull;
+ }
+
+ private synchronized BinaryPageReader binaryReaderFor(FieldInfo field, boolean repeated) throws IOException {
+ if (useDataFusionDecodePath) {
+ DataFusionColumnReader reader = DataFusionColumnReader.open(
+ parquetFile,
+ field.getName(),
+ physicalType(field),
+ repeated,
+ bufferPool,
+ dataFusionInitialBatchSize
+ );
+ dedicatedReaders.add(reader);
+ return reader;
+ }
+ // codec_native path: keep the shared per-field reader (its iterators tolerate sharing,
+ // and its pool slots are not instance-scoped).
+ return readerFor(field, repeated);
+ }
+
+ private synchronized DataFusionColumnReader dataFusionReaderFor(FieldInfo field, boolean repeated) throws IOException {
DataFusionColumnReader reader = dataFusionColumnReaders.get(field.getName());
if (reader == null) {
reader = DataFusionColumnReader.open(
@@ -387,20 +471,9 @@ private DataFusionColumnReader dataFusionReaderFor(FieldInfo field, boolean repe
return reader;
}
- private OrdinalTable ordinalTableFor(FieldInfo field, boolean multiValued) throws IOException {
- OrdinalTable table = ordinalTables.get(field.getName());
- if (table == null) {
- if (useDataFusionDecodePath) {
- DataFusionColumnReader reader = dataFusionReaderFor(field, multiValued);
- table = multiValued ? OrdinalTable.buildMultiValued(reader, maxDoc) : OrdinalTable.buildSingleValued(reader, maxDoc);
- ordinalTables.put(field.getName(), table);
- return table;
- }
- ParquetColumnReader reader = readerFor(field, multiValued);
- table = multiValued ? OrdinalTable.buildMultiValued(reader, maxDoc) : OrdinalTable.buildSingleValued(reader, maxDoc);
- ordinalTables.put(field.getName(), table);
- }
- return table;
+ /** Whether {@link #close()} has run (leaf wrappers reroute to the shared producer then). */
+ boolean isClosed() {
+ return closed;
}
private void ensureOpen() {
diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/SharedProducerRegistry.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/SharedProducerRegistry.java
new file mode 100644
index 0000000000000..1f448cdb2254b
--- /dev/null
+++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/SharedProducerRegistry.java
@@ -0,0 +1,84 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.parquet.codec;
+
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.SegmentReadState;
+import org.opensearch.index.mapper.MapperService;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Node-level registry of segment-lifetime {@link ParquetDocValuesProducer}s.
+ *
+ * Search-time wrappers ({@link ParquetDocValuesLeafReader}) are request-scoped, but Lucene and
+ * OpenSearch caches — fielddata, global ordinals — legitimately retain them beyond the request
+ * and call doc-values accessors later (composite aggregations resolve global ordinals through a
+ * reader-keyed cache, for example). Those late calls cannot be served by the request's producer,
+ * which closes with its search. They are routed here instead: one shared producer per segment
+ * core, created on first use and closed by the segment core's closed-listener.
+ *
+ *
Shared producers are accessed concurrently (any query may race a cached consumer), which the
+ * producer supports for the accessors reachable from caches: sorted/sorted-set iterators use
+ * dedicated per-instance readers with instance-scoped buffer slots, and the producer's lazy maps
+ * are thread-safe. Native cursors opened on this path are reclaimed by {@code Cleaner} when their
+ * iterators become unreachable, with the producer close (segment close) as the final backstop.
+ */
+final class SharedProducerRegistry {
+
+ private static final Map PRODUCERS = new ConcurrentHashMap<>();
+
+ private SharedProducerRegistry() {}
+
+ /**
+ * The segment-lifetime producer for the segment identified by {@code coreHelper}, creating it
+ * on first use. Returns {@code null} when the segment exposes no core cache helper (no safe
+ * lifecycle to attach to — callers must fail rather than leak).
+ */
+ static ParquetDocValuesProducer get(
+ IndexReader.CacheHelper coreHelper,
+ SegmentReadState segmentReadState,
+ MapperService mapperService
+ ) throws IOException {
+ if (coreHelper == null) {
+ return null;
+ }
+ Object key = coreHelper.getKey();
+ ParquetDocValuesProducer existing = PRODUCERS.get(key);
+ if (existing != null) {
+ return existing;
+ }
+ synchronized (PRODUCERS) {
+ existing = PRODUCERS.get(key);
+ if (existing != null) {
+ return existing;
+ }
+ ParquetDocValuesProducer created = new ParquetDocValuesProducer(segmentReadState, mapperService);
+ PRODUCERS.put(key, created);
+ coreHelper.addClosedListener(k -> {
+ ParquetDocValuesProducer removed = PRODUCERS.remove(k);
+ if (removed != null) {
+ try {
+ removed.close();
+ } catch (IOException e) {
+ // Segment is going away; nothing actionable.
+ }
+ }
+ });
+ return created;
+ }
+ }
+
+ /** Number of live shared producers (tests / diagnostics). */
+ static int size() {
+ return PRODUCERS.size();
+ }
+}
diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/TermDictionary.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/TermDictionary.java
new file mode 100644
index 0000000000000..7e1bc91ce815a
--- /dev/null
+++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/TermDictionary.java
@@ -0,0 +1,106 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.parquet.codec;
+
+import org.apache.lucene.index.Terms;
+import org.apache.lucene.index.TermsEnum;
+import org.apache.lucene.util.BytesRef;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A per-segment sorted term dictionary for one keyword field, loaded from the composite index's
+ * Lucene sidecar — the inverted index already stores every distinct term in sorted order, so
+ * reading it costs O(distinct terms) and never scans rows.
+ *
+ * Backs dictionary-rank ordinals: a document's ordinal is computed on access by binary
+ * search of its value against this dictionary, giving fully contract-compliant
+ * segment-global ordinals for fields whose cardinality fits the configured budget. Fields
+ * above the budget (or whose term count is unknown) are not eligible and stay on the
+ * streaming fail-fast path.
+ *
+ *
Instances are immutable and cached per (segment core key, field) — see
+ * {@link TermDictionaryCache}.
+ */
+public final class TermDictionary {
+
+ private final BytesRef[] terms;
+ private final long sizeInBytes;
+
+ private TermDictionary(BytesRef[] terms, long sizeInBytes) {
+ this.terms = terms;
+ this.sizeInBytes = sizeInBytes;
+ }
+
+ /**
+ * Loads the sorted dictionary, or returns {@code null} when the field is not eligible:
+ * no terms, unknown term count, or cardinality above {@code maxTerms}.
+ */
+ public static TermDictionary load(Terms terms, int maxTerms) throws IOException {
+ if (terms == null) {
+ return null;
+ }
+ long size = terms.size();
+ if (size < 0 || size > maxTerms) {
+ return null;
+ }
+ List collected = new ArrayList<>((int) size);
+ long bytes = 0;
+ TermsEnum termsEnum = terms.iterator();
+ for (BytesRef term = termsEnum.next(); term != null; term = termsEnum.next()) {
+ BytesRef copy = BytesRef.deepCopyOf(term);
+ collected.add(copy);
+ bytes += copy.length + 32; // value bytes + object/array overhead estimate
+ }
+ return new TermDictionary(collected.toArray(new BytesRef[0]), bytes);
+ }
+
+ /** Cache sentinel marking a field as checked-and-ineligible. */
+ static TermDictionary sentinel() {
+ return new TermDictionary(new BytesRef[0], 0);
+ }
+
+ /** Number of distinct terms. */
+ public int size() {
+ return terms.length;
+ }
+
+ /** Estimated heap footprint, for cache accounting. */
+ public long sizeInBytes() {
+ return sizeInBytes;
+ }
+
+ /** The term for a segment ordinal. */
+ public BytesRef term(int ord) {
+ return terms[ord];
+ }
+
+ /**
+ * The segment ordinal of {@code value}, or {@code -insertionPoint - 1} when absent
+ * (the {@code lookupTerm} contract).
+ */
+ public int rank(BytesRef value) {
+ int low = 0;
+ int high = terms.length - 1;
+ while (low <= high) {
+ int mid = (low + high) >>> 1;
+ int cmp = terms[mid].compareTo(value);
+ if (cmp < 0) {
+ low = mid + 1;
+ } else if (cmp > 0) {
+ high = mid - 1;
+ } else {
+ return mid;
+ }
+ }
+ return -(low + 1);
+ }
+}
diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/TermDictionaryCache.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/TermDictionaryCache.java
new file mode 100644
index 0000000000000..42b510d6339c5
--- /dev/null
+++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/TermDictionaryCache.java
@@ -0,0 +1,87 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.parquet.codec;
+
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.LeafReader;
+import org.apache.lucene.index.Terms;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Node-level cache of {@link TermDictionary} instances, keyed by (segment core key, field).
+ *
+ * Segments are immutable, so a loaded dictionary is valid for the segment's lifetime and is
+ * released via the core's closed-listener when the segment goes away. Total heap is bounded:
+ * when the budget is exhausted, dictionaries are still served but not cached (each producer
+ * pays the O(distinct) load), which degrades latency, never correctness or memory.
+ */
+public final class TermDictionaryCache {
+
+ /** Sentinel for "checked and not eligible" so ineligible fields are not re-probed. */
+ private static final TermDictionary INELIGIBLE = TermDictionary.sentinel();
+
+ private static final Map> CACHE = new ConcurrentHashMap<>();
+ private static final AtomicLong CACHED_BYTES = new AtomicLong();
+
+ private TermDictionaryCache() {}
+
+ /**
+ * The dictionary for {@code field} in {@code leaf}'s segment, or {@code null} when the
+ * field is above the term budget (or has no usable terms index).
+ */
+ public static TermDictionary get(LeafReader leaf, String field, int maxTerms, long maxCacheBytes) throws IOException {
+ IndexReader.CacheHelper helper = leaf.getCoreCacheHelper();
+ if (helper == null) {
+ Terms terms = leaf.terms(field);
+ return TermDictionary.load(terms, maxTerms);
+ }
+ Object key = helper.getKey();
+ Map perSegment = CACHE.get(key);
+ if (perSegment == null) {
+ perSegment = new ConcurrentHashMap<>();
+ Map existing = CACHE.putIfAbsent(key, perSegment);
+ if (existing != null) {
+ perSegment = existing;
+ } else {
+ helper.addClosedListener(k -> {
+ Map removed = CACHE.remove(k);
+ if (removed != null) {
+ long freed = removed.values().stream().filter(d -> d != INELIGIBLE).mapToLong(TermDictionary::sizeInBytes).sum();
+ CACHED_BYTES.addAndGet(-freed);
+ }
+ });
+ }
+ }
+ TermDictionary cached = perSegment.get(field);
+ if (cached != null) {
+ return cached == INELIGIBLE ? null : cached;
+ }
+ TermDictionary loaded = TermDictionary.load(leaf.terms(field), maxTerms);
+ if (loaded == null) {
+ perSegment.put(field, INELIGIBLE);
+ return null;
+ }
+ if (CACHED_BYTES.addAndGet(loaded.sizeInBytes()) <= maxCacheBytes) {
+ perSegment.put(field, loaded);
+ } else {
+ // Over budget: serve uncached; the producer keeps its own reference for the query.
+ CACHED_BYTES.addAndGet(-loaded.sizeInBytes());
+ }
+ return loaded;
+ }
+
+ /** Currently cached bytes (tests / diagnostics). */
+ public static long cachedBytes() {
+ return CACHED_BYTES.get();
+ }
+}
diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/UninvertedOrdinals.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/UninvertedOrdinals.java
new file mode 100644
index 0000000000000..c105b844e497f
--- /dev/null
+++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/UninvertedOrdinals.java
@@ -0,0 +1,388 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.parquet.codec;
+
+import org.apache.lucene.index.PostingsEnum;
+import org.apache.lucene.index.Terms;
+import org.apache.lucene.index.TermsEnum;
+import org.apache.lucene.search.DocIdSetIterator;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.store.IOContext;
+import org.apache.lucene.store.IndexInput;
+import org.apache.lucene.store.IndexOutput;
+import org.apache.lucene.store.MMapDirectory;
+import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.LongValues;
+import org.apache.lucene.util.packed.DirectWriter;
+import org.apache.lucene.util.packed.DirectReader;
+import org.apache.lucene.util.packed.PackedInts;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Segment-global ordinals for a high-cardinality keyword field, uninverted once from the Lucene
+ * sidecar's postings and spilled to a memory-mapped node-local file — read-side only, with
+ * Lucene's own storage economics: the packed doc→ord array lives on disk and only touched pages
+ * are resident.
+ *
+ * Build
+ * One sequential sweep of the field's terms (already sorted on disk) and their postings assigns
+ * each document its term's rank. The transient in-heap packed buffer is released after the spill;
+ * builds are serialized node-wide and check the cancellation flag between terms. Deleted
+ * documents keep their ordinals (collectors never visit them), matching Lucene's own doc-values
+ * semantics until merge.
+ *
+ * Read
+ * {@code ordinal(doc)} is one packed read from the mapped file (0 = missing; stored values are
+ * ord + 1). {@code lookupOrd} uses sparse in-heap checkpoints (every {@value #CHECKPOINT_INTERVAL}
+ * terms) plus a bounded {@code TermsEnum} advance — only final buckets and sort bounds resolve
+ * terms, never per-document access.
+ */
+public final class UninvertedOrdinals implements Closeable {
+
+ static final int CHECKPOINT_INTERVAL = 1024;
+ private static final String CODEC_PREFIX = "parquet-ords";
+
+ private final Directory directory;
+ private final IndexInput input;
+ private final LongValues ords;
+ private final BytesRef[] checkpoints;
+ private final Terms terms;
+ private final int valueCount;
+ private final long sizeInBytes;
+ private final String fileName;
+
+ private UninvertedOrdinals(
+ Directory directory,
+ IndexInput input,
+ LongValues ords,
+ BytesRef[] checkpoints,
+ Terms terms,
+ int valueCount,
+ long sizeInBytes,
+ String fileName
+ ) {
+ this.directory = directory;
+ this.input = input;
+ this.ords = ords;
+ this.checkpoints = checkpoints;
+ this.terms = terms;
+ this.valueCount = valueCount;
+ this.sizeInBytes = sizeInBytes;
+ this.fileName = fileName;
+ }
+
+ /**
+ * Builds (or maps an existing) ordinal file for the field. {@code cancelled} is polled
+ * between terms during the sweep so runaway builds die with their task.
+ */
+ static UninvertedOrdinals build(
+ Path ordsDir,
+ String fileKey,
+ Terms terms,
+ int maxDoc,
+ long expectedNonNullDocs,
+ java.util.function.BooleanSupplier cancelled
+ ) throws IOException {
+ if (expectedNonNullDocs < 0) {
+ throw new IllegalStateException(
+ "cannot verify ordinal coverage (column null statistics unavailable); refusing to "
+ + "serve postings-derived ordinals that may silently drop unindexed values"
+ );
+ }
+ long termCount = terms.size();
+ if (termCount < 0) {
+ throw new IllegalStateException("terms index reports unknown size; cannot uninvert");
+ }
+ Files.createDirectories(ordsDir);
+ Directory directory = new MMapDirectory(ordsDir);
+ String fileName = CODEC_PREFIX + "-" + fileKey + ".ord";
+ // +1 shifted encoding (0 = missing). DirectWriter supports only specific widths;
+ // its bitsRequired rounds up to the nearest supported one.
+ int bits = DirectWriter.bitsRequired(termCount + 1);
+ List checkpoints = new ArrayList<>((int) (termCount / CHECKPOINT_INTERVAL) + 1);
+
+ boolean exists;
+ try {
+ directory.fileLength(fileName);
+ exists = true;
+ } catch (java.io.FileNotFoundException | java.nio.file.NoSuchFileException e) {
+ exists = false;
+ }
+
+ if (exists == false) {
+ // Sweep: heap-transient packed buffer (released after spill), then one sequential write.
+ PackedInts.Mutable building = PackedInts.getMutable(maxDoc, bits, PackedInts.COMPACT);
+ TermsEnum termsEnum = terms.iterator();
+ PostingsEnum postings = null;
+ long ord = 0;
+ for (BytesRef term = termsEnum.next(); term != null; term = termsEnum.next(), ord++) {
+ if ((ord & (CHECKPOINT_INTERVAL - 1)) == 0) {
+ if (cancelled.getAsBoolean()) {
+ throw new IOException("ordinal build cancelled for " + fileKey);
+ }
+ }
+ postings = termsEnum.postings(postings, PostingsEnum.NONE);
+ for (int doc = postings.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = postings.nextDoc()) {
+ building.set(doc, ord + 1);
+ }
+ }
+ String tempName = fileName + ".tmp";
+ try {
+ directory.deleteFile(tempName); // stale leftover from an interrupted build
+ } catch (java.io.FileNotFoundException | java.nio.file.NoSuchFileException e) {
+ // normal case
+ }
+ try (IndexOutput out = directory.createOutput(tempName, IOContext.DEFAULT)) {
+ DirectWriter writer = DirectWriter.getInstance(out, maxDoc, bits);
+ for (int doc = 0; doc < maxDoc; doc++) {
+ writer.add(building.get(doc));
+ }
+ writer.finish();
+ }
+ directory.rename(tempName, fileName);
+ }
+
+ // Checkpoints are cheap relative to the sweep; collect them on every load.
+ TermsEnum termsEnum = terms.iterator();
+ long ord = 0;
+ for (BytesRef term = termsEnum.next(); term != null; term = termsEnum.next(), ord++) {
+ if ((ord % CHECKPOINT_INTERVAL) == 0) {
+ checkpoints.add(BytesRef.deepCopyOf(term));
+ }
+ }
+
+ IndexInput input = directory.openInput(fileName, IOContext.DEFAULT);
+ LongValues ords = DirectReader.getInstance(input.randomAccessSlice(0, input.length()), bits);
+ long size = directory.fileLength(fileName);
+ // Coverage verification: postings only contain INDEXED values. A stored value that was
+ // never indexed (ignore_above truncation, analyzer drops) would silently become
+ // "missing" and undercount every aggregation on this field. Count assigned ordinals and
+ // require exact agreement with the Parquet column's non-null row count; refuse loudly
+ // otherwise. One sequential mapped read per segment load (~100ms per 100M rows).
+ long assigned = 0;
+ for (int doc = 0; doc < maxDoc; doc++) {
+ if (ords.get(doc) != 0) {
+ assigned++;
+ }
+ }
+ if (assigned != expectedNonNullDocs) {
+ input.close();
+ directory.close();
+ throw new IllegalStateException(
+ "ordinal coverage mismatch for "
+ + fileKey
+ + ": postings assign "
+ + assigned
+ + " documents but the column stores "
+ + expectedNonNullDocs
+ + " non-null values — some stored values are not indexed (ignore_above?); "
+ + "refusing uninverted ordinals to avoid silent undercounts"
+ );
+ }
+ return new UninvertedOrdinals(
+ directory,
+ input,
+ ords,
+ checkpoints.toArray(new BytesRef[0]),
+ terms,
+ (int) termCount,
+ size,
+ fileName
+ );
+ }
+
+ /** The segment ordinal for {@code doc}, or -1 when the document has no value. */
+ public int ordinal(int doc) {
+ return (int) ords.get(doc) - 1;
+ }
+
+ /** Number of distinct terms. */
+ public int valueCount() {
+ return valueCount;
+ }
+
+ /** On-disk footprint (cache accounting). */
+ public long sizeInBytes() {
+ return sizeInBytes;
+ }
+
+ /** The ord file's name within the ords directory (disk-budget pinning). */
+ public String fileName() {
+ return fileName;
+ }
+
+ /**
+ * The field's real terms enumeration — the exact sorted term space these ordinals rank —
+ * wrapped with ordinal tracking, because consumers like {@code OrdinalMap} require
+ * {@link TermsEnum#ord()} which BlockTree does not implement. Ord seeks use the sparse
+ * checkpoints; byte seeks re-derive the position via {@link #rank}.
+ */
+ public TermsEnum termsEnum() throws IOException {
+ return new OrdTrackingTermsEnum(terms.iterator());
+ }
+
+ private final class OrdTrackingTermsEnum extends org.apache.lucene.index.FilterLeafReader.FilterTermsEnum {
+ private long position = -1;
+
+ OrdTrackingTermsEnum(TermsEnum in) {
+ super(in);
+ }
+
+ @Override
+ public BytesRef next() throws IOException {
+ BytesRef term = in.next();
+ if (term != null) {
+ position++;
+ } else {
+ position = valueCount;
+ }
+ return term;
+ }
+
+ @Override
+ public long ord() {
+ return position;
+ }
+
+ @Override
+ public void seekExact(long ord) throws IOException {
+ int checkpoint = (int) (ord / CHECKPOINT_INTERVAL);
+ in.seekCeil(checkpoints[checkpoint]);
+ position = (long) checkpoint * CHECKPOINT_INTERVAL;
+ while (position < ord) {
+ in.next();
+ position++;
+ }
+ }
+
+ @Override
+ public boolean seekExact(BytesRef text) throws IOException {
+ boolean found = in.seekExact(text);
+ position = found ? rank(text) : -1;
+ return found;
+ }
+
+ @Override
+ public SeekStatus seekCeil(BytesRef text) throws IOException {
+ SeekStatus status = in.seekCeil(text);
+ if (status == SeekStatus.END) {
+ position = valueCount;
+ } else {
+ int r = rank(in.term());
+ position = r >= 0 ? r : -(r + 1);
+ }
+ return status;
+ }
+ }
+
+ /** Resolves an ordinal to its term: checkpoint seek plus a bounded enum advance. */
+ public BytesRef term(int ord) {
+ try {
+ TermsEnum termsEnum = terms.iterator();
+ int checkpoint = ord / CHECKPOINT_INTERVAL;
+ termsEnum.seekCeil(checkpoints[checkpoint]);
+ for (int i = checkpoint * CHECKPOINT_INTERVAL; i < ord; i++) {
+ termsEnum.next();
+ }
+ return BytesRef.deepCopyOf(termsEnum.term());
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ /** A single-consumer stateful term resolver: ascending ordinal walks cost one enum pass. */
+ public TermCursor newTermCursor() {
+ return new TermCursor();
+ }
+
+ /**
+ * Stateful ord→term resolution for one consumer (not thread-safe, like doc-values
+ * iterators). A stateless resolver pays a checkpoint seek plus up to
+ * {@value #CHECKPOINT_INTERVAL} enum steps on EVERY call, which turns full-column walks
+ * quadratic; this cursor advances forward from its last position when the requested
+ * ordinal is ahead, so monotonic access — bucket resolution, ordinal-map style walks —
+ * amortizes to a single sequential pass over the terms file.
+ */
+ public final class TermCursor {
+ private TermsEnum cursorEnum;
+ private long cursorOrd = -1;
+
+ public BytesRef term(int ord) {
+ try {
+ long behind = cursorEnum == null ? Long.MAX_VALUE : ord - cursorOrd;
+ if (behind < 0 || behind > CHECKPOINT_INTERVAL) {
+ // Behind us, or far ahead: re-seek to the nearest checkpoint.
+ cursorEnum = terms.iterator();
+ int checkpoint = ord / CHECKPOINT_INTERVAL;
+ cursorEnum.seekCeil(checkpoints[checkpoint]);
+ cursorOrd = (long) checkpoint * CHECKPOINT_INTERVAL;
+ }
+ while (cursorOrd < ord) {
+ cursorEnum.next();
+ cursorOrd++;
+ }
+ return cursorEnum.term();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+ }
+
+ /** The ordinal of {@code key}, or {@code -insertionPoint - 1} (the lookupTerm contract). */
+ public int rank(BytesRef key) {
+ try {
+ // Binary search over checkpoints, then a bounded linear scan with the enum.
+ int low = 0;
+ int high = checkpoints.length - 1;
+ while (low <= high) {
+ int mid = (low + high) >>> 1;
+ int cmp = checkpoints[mid].compareTo(key);
+ if (cmp < 0) {
+ low = mid + 1;
+ } else if (cmp > 0) {
+ high = mid - 1;
+ } else {
+ return mid * CHECKPOINT_INTERVAL;
+ }
+ }
+ int checkpoint = Math.max(low - 1, 0);
+ TermsEnum termsEnum = terms.iterator();
+ termsEnum.seekCeil(checkpoints[checkpoint]);
+ int ord = checkpoint * CHECKPOINT_INTERVAL;
+ BytesRef term = termsEnum.term();
+ while (term != null) {
+ int cmp = term.compareTo(key);
+ if (cmp == 0) {
+ return ord;
+ }
+ if (cmp > 0) {
+ return -(ord + 1);
+ }
+ term = termsEnum.next();
+ ord++;
+ }
+ return -(ord + 1);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ input.close();
+ directory.close();
+ }
+}
diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/UninvertedOrdinalsCache.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/UninvertedOrdinalsCache.java
new file mode 100644
index 0000000000000..ac429710e3daf
--- /dev/null
+++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/UninvertedOrdinalsCache.java
@@ -0,0 +1,281 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.parquet.codec;
+
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.LeafReader;
+import org.apache.lucene.index.SegmentInfo;
+import org.apache.lucene.index.Terms;
+import org.apache.lucene.util.StringHelper;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Node-level cache of {@link UninvertedOrdinals}, keyed by (segment core key, field).
+ *
+ * Builds are serialized node-wide (one postings sweep at a time — the transient packed buffer
+ * and the sweep's CPU never stack). Entries are evicted (and their mapped files closed) by the
+ * segment core's closed-listener; the on-disk artifact is keyed by the segment's stable id and
+ * survives restarts, so a re-opened segment maps the existing file instead of rebuilding.
+ */
+public final class UninvertedOrdinalsCache {
+
+ private static final org.apache.logging.log4j.Logger LOGGER = org.apache.logging.log4j.LogManager.getLogger(
+ UninvertedOrdinalsCache.class
+ );
+ /** Marks a (segment, field) whose ordinals failed coverage verification — do not retry. */
+ private static final Map> INELIGIBLE = new ConcurrentHashMap<>();
+
+ private static final Map> CACHE = new ConcurrentHashMap<>();
+ private static final Object BUILD_LOCK = new Object();
+ /** Default under java.io.tmpdir (unit tests); the plugin points this at the node data path. */
+ private static volatile Path ORDS_DIR = Path.of(System.getProperty("java.io.tmpdir"), "opensearch-parquet-ords");
+
+ private static volatile boolean shuttingDown = false;
+
+ /**
+ * Called once at plugin init: ord files live with the node's data, not in tmp. Also performs
+ * crash hygiene: interrupted builds' {@code .tmp} files are deleted, and the directory is
+ * trimmed to the disk budget oldest-first (nothing is pinned yet at startup).
+ */
+ public static void setOrdsDir(Path dir) {
+ ORDS_DIR = dir;
+ shuttingDown = false;
+ cleanupAtStartup(dir);
+ }
+
+ /** Called at plugin close: aborts in-flight builds so node shutdown is not held hostage. */
+ public static void shutdown() {
+ shuttingDown = true;
+ }
+
+ /**
+ * Builds ordinals, retrying ONCE after deleting the on-disk file when verification fails on
+ * a pre-existing file: a file left by a crashed or killed process may be stale for reasons a
+ * rebuild fixes (segment data moved on after an unclean stop). Only a failure on a FRESH
+ * build is genuine (unindexed stored values) and latches the field ineligible.
+ */
+ private static UninvertedOrdinals buildWithRetry(String fileKey, Terms terms, int maxDoc, long expectedNonNullDocs)
+ throws IOException {
+ String fileName = "parquet-ords-" + fileKey + ".ord";
+ boolean preExisting = java.nio.file.Files.exists(ORDS_DIR.resolve(fileName));
+ try {
+ return UninvertedOrdinals.build(
+ ORDS_DIR,
+ fileKey,
+ terms,
+ maxDoc,
+ expectedNonNullDocs,
+ () -> shuttingDown || Thread.currentThread().isInterrupted()
+ );
+ } catch (IllegalStateException e) {
+ if (preExisting == false) {
+ throw e;
+ }
+ LOGGER.warn("ord file [{}] failed verification ({}); deleting and rebuilding once", fileName, e.getMessage());
+ java.nio.file.Files.deleteIfExists(ORDS_DIR.resolve(fileName));
+ return UninvertedOrdinals.build(
+ ORDS_DIR,
+ fileKey,
+ terms,
+ maxDoc,
+ expectedNonNullDocs,
+ () -> shuttingDown || Thread.currentThread().isInterrupted()
+ );
+ }
+ }
+
+ private static void cleanupAtStartup(Path dir) {
+ long budget = ParquetDocValuesProducer.uninvertMaxDiskBytes();
+ long used = 0;
+ List files = new java.util.ArrayList<>();
+ try (java.util.stream.Stream listing = java.nio.file.Files.list(dir)) {
+ for (Path file : (Iterable) listing::iterator) {
+ if (file.getFileName().toString().endsWith(".tmp")) {
+ java.nio.file.Files.deleteIfExists(file); // interrupted build leftovers
+ } else {
+ used += java.nio.file.Files.size(file);
+ files.add(file);
+ }
+ }
+ } catch (java.nio.file.NoSuchFileException e) {
+ return;
+ } catch (IOException e) {
+ LOGGER.warn("ords directory startup cleanup failed for [{}]: {}", dir, e.getMessage());
+ return;
+ }
+ if (used <= budget) {
+ return;
+ }
+ files.sort(java.util.Comparator.comparingLong(f -> {
+ try {
+ return java.nio.file.Files.getLastModifiedTime(f).toMillis();
+ } catch (IOException e) {
+ return Long.MAX_VALUE;
+ }
+ }));
+ for (Path victim : files) {
+ if (used <= budget) {
+ break;
+ }
+ try {
+ long size = java.nio.file.Files.size(victim);
+ java.nio.file.Files.deleteIfExists(victim);
+ used -= size;
+ LOGGER.info("reclaimed ord file [{}] at startup (over budget)", victim.getFileName());
+ } catch (IOException e) {
+ // skip
+ }
+ }
+ }
+
+ private UninvertedOrdinalsCache() {}
+
+ /** Transient refusal: budget can be raised or freed, so it is never latched as INELIGIBLE. */
+ private static final class BudgetExceededException extends IllegalStateException {
+ BudgetExceededException(String message) {
+ super(message);
+ }
+ }
+
+ /**
+ * Keeps the ords directory within {@code parquet.docvalues.uninvert.max_disk_bytes}. Files
+ * belonging to live cache entries are pinned; everything else (closed segments, merged-away
+ * segments, other fields' leftovers) is reclaimable oldest-mtime-first. If the new file
+ * still does not fit after reclaim, the build is refused — bounded disk, loud fallback.
+ */
+ private static void enforceDiskBudget(String fileKey, Terms terms, int maxDoc) throws IOException {
+ String fileName = "parquet-ords-" + fileKey + ".ord";
+ if (java.nio.file.Files.exists(ORDS_DIR.resolve(fileName))) {
+ return; // reusing an existing file adds no disk
+ }
+ long budget = ParquetDocValuesProducer.uninvertMaxDiskBytes();
+ long termCount = Math.max(terms.size(), 0);
+ long bits = org.apache.lucene.util.packed.DirectWriter.bitsRequired(termCount + 1);
+ long estimate = (maxDoc * bits + 7) / 8 + 1024;
+ java.util.Set pinned = new java.util.HashSet<>();
+ for (Map perSegment : CACHE.values()) {
+ for (UninvertedOrdinals live : perSegment.values()) {
+ pinned.add(live.fileName());
+ }
+ }
+ long used = 0;
+ List reclaimable = new java.util.ArrayList<>();
+ try (java.util.stream.Stream listing = java.nio.file.Files.list(ORDS_DIR)) {
+ for (Path file : (Iterable) listing::iterator) {
+ used += java.nio.file.Files.size(file);
+ if (pinned.contains(file.getFileName().toString()) == false) {
+ reclaimable.add(file);
+ }
+ }
+ } catch (java.nio.file.NoSuchFileException e) {
+ return; // directory not created yet: nothing used
+ }
+ if (used + estimate <= budget) {
+ return;
+ }
+ reclaimable.sort(java.util.Comparator.comparingLong(f -> {
+ try {
+ return java.nio.file.Files.getLastModifiedTime(f).toMillis();
+ } catch (IOException e) {
+ return Long.MAX_VALUE;
+ }
+ }));
+ for (Path victim : reclaimable) {
+ if (used + estimate <= budget) {
+ break;
+ }
+ try {
+ long size = java.nio.file.Files.size(victim);
+ java.nio.file.Files.deleteIfExists(victim);
+ used -= size;
+ } catch (IOException e) {
+ // still referenced by an mmap on some platforms or raced; skip
+ }
+ }
+ if (used + estimate > budget) {
+ throw new BudgetExceededException(
+ "uninverted ordinals disk budget exceeded: "
+ + used
+ + "B used + "
+ + estimate
+ + "B needed > "
+ + budget
+ + "B (parquet.docvalues.uninvert.max_disk_bytes)"
+ );
+ }
+ }
+
+ /**
+ * The uninverted ordinals for {@code field}, building (or re-mapping) on first use.
+ * Returns {@code null} when the segment lacks a core cache identity or a terms index.
+ */
+ static UninvertedOrdinals get(LeafReader leaf, SegmentInfo segmentInfo, String field, long expectedNonNullDocs) throws IOException {
+ IndexReader.CacheHelper helper = leaf.getCoreCacheHelper();
+ Terms terms = leaf.terms(field);
+ if (helper == null || terms == null) {
+ return null;
+ }
+ Object key = helper.getKey();
+ java.util.Set ineligible = INELIGIBLE.get(key);
+ if (ineligible != null && ineligible.contains(field)) {
+ return null;
+ }
+ Map perSegment = CACHE.computeIfAbsent(key, k -> {
+ helper.addClosedListener(closedKey -> {
+ INELIGIBLE.remove(closedKey);
+ Map removed = CACHE.remove(closedKey);
+ if (removed != null) {
+ for (UninvertedOrdinals ords : removed.values()) {
+ try {
+ ords.close();
+ } catch (IOException e) {
+ // Segment is going away; nothing actionable.
+ }
+ }
+ }
+ });
+ return new ConcurrentHashMap<>();
+ });
+ UninvertedOrdinals cached = perSegment.get(field);
+ if (cached != null) {
+ return cached;
+ }
+ synchronized (BUILD_LOCK) {
+ cached = perSegment.get(field);
+ if (cached != null) {
+ return cached;
+ }
+ String fileKey = StringHelper.idToString(segmentInfo.getId()) + "-" + field;
+ try {
+ enforceDiskBudget(fileKey, terms, leaf.maxDoc());
+ UninvertedOrdinals built = buildWithRetry(fileKey, terms, leaf.maxDoc(), expectedNonNullDocs);
+ perSegment.put(field, built);
+ return built;
+ } catch (BudgetExceededException e) {
+ // Disk budget refusals are transient (budget can be raised, files can be
+ // reclaimed): log and fall back WITHOUT latching, so the next query retries.
+ LOGGER.warn("refusing uninverted ordinals for field [{}]: {}", field, e.getMessage());
+ return null;
+ } catch (IllegalStateException e) {
+ // Coverage verification failed: postings do not represent every stored value
+ // (ignore_above truncation and the like). Serving them would silently
+ // undercount. Remember the refusal and let global-ordinal consumers hit the
+ // streaming iterator's loud fail-fast toward execution_hint:map.
+ LOGGER.warn("refusing uninverted ordinals for field [{}]: {}", field, e.getMessage());
+ INELIGIBLE.computeIfAbsent(key, k -> java.util.concurrent.ConcurrentHashMap.newKeySet()).add(field);
+ return null;
+ }
+ }
+ }
+}
diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/iter/ParquetDictionarySortedDocValues.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/iter/ParquetDictionarySortedDocValues.java
new file mode 100644
index 0000000000000..3cdee374e909f
--- /dev/null
+++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/iter/ParquetDictionarySortedDocValues.java
@@ -0,0 +1,116 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.parquet.codec.iter;
+
+import org.apache.lucene.index.SortedDocValues;
+import org.apache.lucene.util.BytesRef;
+import org.opensearch.parquet.codec.TermDictionary;
+
+import java.io.IOException;
+
+/**
+ * Fully contract-compliant {@link SortedDocValues} for keyword fields whose cardinality fits
+ * the dictionary budget: segment-global ordinals are computed on access as the rank of the
+ * document's value in the segment's sorted {@link TermDictionary} (loaded O(distinct) from the
+ * composite index's Lucene sidecar — never from a row scan).
+ *
+ * Values come from the streaming iterator's zero-copy page path; the only added per-document
+ * cost is one binary search over the heap-resident dictionary. All global operations —
+ * cross-document ordinal comparison (sorting), {@link #getValueCount()} (cardinality,
+ * composite, global-ordinals terms aggregations), {@link #lookupTerm} — are exact.
+ */
+public final class ParquetDictionarySortedDocValues extends SortedDocValues {
+
+ private final ParquetSortedDocValues stream;
+ private final TermDictionary dictionary;
+
+ private int currentOrd = -1;
+
+ public ParquetDictionarySortedDocValues(ParquetSortedDocValues stream, TermDictionary dictionary) {
+ this.stream = stream;
+ this.dictionary = dictionary;
+ }
+
+ @Override
+ public boolean advanceExact(int target) throws IOException {
+ if (stream.advanceExact(target) == false) {
+ currentOrd = -1;
+ return false;
+ }
+ computeOrd();
+ return true;
+ }
+
+ /**
+ * Ranks the positioned document's value. The streaming iterator's ord is transient and
+ * immediately resolved — the exact access pattern it supports; the dictionary rank is the
+ * real segment ordinal.
+ */
+ private void computeOrd() {
+ BytesRef value = stream.lookupOrd(stream.ordValue());
+ int ord = dictionary.rank(value);
+ if (ord < 0) {
+ throw new IllegalStateException(
+ "value ["
+ + value.utf8ToString()
+ + "] present in doc values but absent from the field's terms index; "
+ + "dictionary ordinals require every stored value to be indexed (no ignore_above)"
+ );
+ }
+ currentOrd = ord;
+ }
+
+ @Override
+ public int ordValue() {
+ return currentOrd;
+ }
+
+ @Override
+ public BytesRef lookupOrd(int ord) {
+ return dictionary.term(ord);
+ }
+
+ @Override
+ public int getValueCount() {
+ return dictionary.size();
+ }
+
+ @Override
+ public int lookupTerm(BytesRef key) {
+ return dictionary.rank(key);
+ }
+
+ @Override
+ public int docID() {
+ return stream.docID();
+ }
+
+ @Override
+ public int nextDoc() throws IOException {
+ int doc = stream.nextDoc();
+ if (doc != NO_MORE_DOCS) {
+ computeOrd();
+ }
+ return doc;
+ }
+
+ @Override
+ public int advance(int target) throws IOException {
+ int doc = stream.advance(target);
+ if (doc != NO_MORE_DOCS) {
+ computeOrd();
+ }
+ return doc;
+ }
+
+ @Override
+ public long cost() {
+ return stream.cost();
+ }
+}
diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/iter/ParquetSortedDocValues.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/iter/ParquetSortedDocValues.java
index 432317246cf70..54bcc7db54094 100644
--- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/iter/ParquetSortedDocValues.java
+++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/iter/ParquetSortedDocValues.java
@@ -10,26 +10,34 @@
import org.apache.lucene.index.SortedDocValues;
import org.apache.lucene.util.BytesRef;
-import org.opensearch.parquet.codec.OrdinalTable;
+import org.opensearch.parquet.bridge.BinaryPageReader;
+import org.opensearch.parquet.codec.cache.PageCache;
import java.io.IOException;
/**
- * {@link SortedDocValues} backed by a per-segment {@link OrdinalTable} for a single-valued
- * Parquet keyword/ip column. The ordinal table is built once (lazily) by the producer; this
- * iterator only walks the per-row ordinal array and serves {@code lookupOrd} from the sorted
- * term dictionary.
+ * Streaming single-valued {@link SortedDocValues} over a Parquet keyword column — sequential
+ * access only, with no segment-wide ordinal structure.
+ *
+ *
Same capability contract as {@link ParquetSortedSetDocValues}: transient per-document
+ * ordinals (the ordinal is the docId, which keeps it inside {@code int} range),
+ * resolved immediately via {@link #lookupOrd}; segment-global operations
+ * ({@link #getValueCount()}, {@link #lookupTerm}) and stale-ordinal resolution throw rather
+ * than return wrong results. Serves the fetch phase and bytes-view consumers at O(rows
+ * visited); ordinal-comparing consumers (global-ordinals aggregations) must use
+ * {@code execution_hint: map}.
*/
public final class ParquetSortedDocValues extends SortedDocValues {
- private final OrdinalTable table;
+ private final BinaryPageReader reader;
private final int maxDoc;
+ private final BytesRef scratch = new BytesRef();
private int doc = -1;
- private int currentOrd = -1;
+ private boolean currentPresent;
- public ParquetSortedDocValues(OrdinalTable table, int maxDoc) {
- this.table = table;
+ public ParquetSortedDocValues(BinaryPageReader reader, int maxDoc) {
+ this.reader = reader;
this.maxDoc = maxDoc;
}
@@ -37,27 +45,71 @@ public ParquetSortedDocValues(OrdinalTable table, int maxDoc) {
public boolean advanceExact(int target) throws IOException {
if (target >= maxDoc) {
doc = NO_MORE_DOCS;
- currentOrd = -1;
+ currentPresent = false;
return false;
}
doc = target;
- currentOrd = table.ordForRow(target);
- return currentOrd != -1;
+ // Zero-copy hot path (mirrors ParquetBinaryDocValues): serve the value as a view into
+ // the resident page buffer — no per-document allocation on 100M-doc scans.
+ PageCache cache = reader.cache();
+ if (cache == null || target > cache.lastRow || target < cache.firstRow) {
+ reader.loadPageContaining(target);
+ cache = reader.cache();
+ if (cache == null) {
+ currentPresent = false;
+ return false;
+ }
+ }
+ currentPresent = cache.isPresent(target);
+ if (currentPresent) {
+ int rel = (int) (target - cache.firstRow);
+ int start = cache.byteOffsets[rel];
+ int end = cache.byteOffsets[rel + 1];
+ scratch.bytes = cache.byteBuf;
+ scratch.offset = start;
+ scratch.length = end - start;
+ }
+ return currentPresent;
}
@Override
public int ordValue() {
- return currentOrd;
+ // The document id doubles as the transient ordinal: unique per positioned doc,
+ // int-ranged, and verifiable in lookupOrd.
+ return doc;
}
@Override
public BytesRef lookupOrd(int ord) {
- return table.lookupOrd(ord);
+ if (ord != doc || currentPresent == false) {
+ throw new UnsupportedOperationException(
+ "ordinal "
+ + ord
+ + " was issued for another document (current doc "
+ + doc
+ + "): composite Parquet keyword fields serve per-document streaming ordinals "
+ + "only; consumers requiring segment-global ordinals must use execution_hint:map"
+ );
+ }
+ return scratch;
}
@Override
public int getValueCount() {
- return table.valueCount();
+ throw new UnsupportedOperationException(
+ "getValueCount requires segment-global ordinals, which composite Parquet keyword "
+ + "fields do not materialize at read time; aggregations on these fields must use "
+ + "execution_hint:map"
+ );
+ }
+
+ @Override
+ public int lookupTerm(BytesRef key) {
+ throw new UnsupportedOperationException(
+ "lookupTerm requires segment-global ordinals, which composite Parquet keyword "
+ + "fields do not materialize at read time; aggregations on these fields must use "
+ + "execution_hint:map"
+ );
}
@Override
@@ -73,14 +125,12 @@ public int nextDoc() throws IOException {
@Override
public int advance(int target) throws IOException {
for (int d = target; d < maxDoc; d++) {
- if (table.ordForRow(d) != -1) {
- doc = d;
- currentOrd = table.ordForRow(d);
+ if (advanceExact(d)) {
return d;
}
}
doc = NO_MORE_DOCS;
- currentOrd = -1;
+ currentPresent = false;
return NO_MORE_DOCS;
}
diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/iter/ParquetUninvertedSortedDocValues.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/iter/ParquetUninvertedSortedDocValues.java
new file mode 100644
index 0000000000000..232d62e9ced98
--- /dev/null
+++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/iter/ParquetUninvertedSortedDocValues.java
@@ -0,0 +1,128 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.parquet.codec.iter;
+
+import org.apache.lucene.index.SortedDocValues;
+import org.apache.lucene.index.TermsEnum;
+import org.apache.lucene.util.BytesRef;
+import org.opensearch.parquet.codec.UninvertedOrdinals;
+
+import java.io.IOException;
+
+/**
+ * Fully contract-compliant {@link SortedDocValues} for high-cardinality keyword fields, backed by
+ * disk-resident {@link UninvertedOrdinals}. Access-path economics:
+ *
+ *
+ * {@code ordValue()} — one packed read from the memory-mapped ordinal file; no Parquet
+ * decode at all (sorting, global-ordinal collection).
+ * {@code lookupOrd(currentOrd)} — the per-document value pattern (map-hint terms,
+ * cardinality hashing): served zero-copy from the streaming reader's resident page, never
+ * through the terms index.
+ * {@code lookupOrd(otherOrd)} — bucket-key resolution: a stateful cursor over the sidecar's
+ * terms enum; ascending walks amortize to one sequential pass.
+ *
+ */
+public final class ParquetUninvertedSortedDocValues extends SortedDocValues {
+
+ private final UninvertedOrdinals ordinals;
+ private final ParquetSortedDocValues streaming;
+ private final int maxDoc;
+
+ private UninvertedOrdinals.TermCursor termCursor;
+ private int doc = -1;
+ private int currentOrd = -1;
+ private boolean streamingPositioned = false;
+
+ public ParquetUninvertedSortedDocValues(UninvertedOrdinals ordinals, ParquetSortedDocValues streaming, int maxDoc) {
+ this.ordinals = ordinals;
+ this.streaming = streaming;
+ this.maxDoc = maxDoc;
+ }
+
+ @Override
+ public boolean advanceExact(int target) {
+ if (target >= maxDoc) {
+ doc = NO_MORE_DOCS;
+ currentOrd = -1;
+ return false;
+ }
+ doc = target;
+ streamingPositioned = false; // value read is lazy; most consumers never need it
+ currentOrd = ordinals.ordinal(target);
+ return currentOrd >= 0;
+ }
+
+ @Override
+ public int ordValue() {
+ return currentOrd;
+ }
+
+ @Override
+ public BytesRef lookupOrd(int ord) throws IOException {
+ if (ord == currentOrd && doc >= 0 && doc != NO_MORE_DOCS) {
+ // Per-document value access: the streaming reader serves the CURRENT document's
+ // bytes from its resident page — O(1), not a terms-index walk.
+ if (streamingPositioned == false) {
+ streaming.advanceExact(doc);
+ streamingPositioned = true;
+ }
+ return streaming.lookupOrd(streaming.ordValue());
+ }
+ if (termCursor == null) {
+ termCursor = ordinals.newTermCursor();
+ }
+ return termCursor.term(ord);
+ }
+
+ @Override
+ public int getValueCount() {
+ return ordinals.valueCount();
+ }
+
+ @Override
+ public int lookupTerm(BytesRef key) {
+ return ordinals.rank(key);
+ }
+
+ @Override
+ public TermsEnum termsEnum() throws IOException {
+ // The default implementation resolves every ordinal through lookupOrd — quadratic over
+ // millions of terms when OrdinalMap walks the enum. The sidecar's own enum IS this
+ // ordinal space, in order, streamed off disk.
+ return ordinals.termsEnum();
+ }
+
+ @Override
+ public int docID() {
+ return doc;
+ }
+
+ @Override
+ public int nextDoc() throws IOException {
+ return advance(doc + 1);
+ }
+
+ @Override
+ public int advance(int target) throws IOException {
+ for (int d = target; d < maxDoc; d++) {
+ if (advanceExact(d)) {
+ return d;
+ }
+ }
+ doc = NO_MORE_DOCS;
+ currentOrd = -1;
+ return NO_MORE_DOCS;
+ }
+
+ @Override
+ public long cost() {
+ return maxDoc;
+ }
+}
diff --git a/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/codec/ParquetRoundTripDocValuesFormatTests.java b/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/codec/ParquetRoundTripDocValuesFormatTests.java
new file mode 100644
index 0000000000000..7886b79165421
--- /dev/null
+++ b/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/codec/ParquetRoundTripDocValuesFormatTests.java
@@ -0,0 +1,54 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.parquet.codec;
+
+import org.apache.lucene.codecs.Codec;
+import org.apache.lucene.tests.index.BaseDocValuesFormatTestCase;
+import org.apache.lucene.tests.util.TestUtil;
+import org.opensearch.parquet.bridge.RustBridge;
+
+import java.nio.file.Files;
+
+/**
+ * Runs Lucene's exhaustive {@link BaseDocValuesFormatTestCase} contract battery (randomized
+ * values, missing docs, advance/advanceExact semantics, merges, huge segments) against the REAL
+ * Parquet doc-values read stack via {@link RoundTripParquetDocValuesFormat}: NUMERIC, BINARY and
+ * SORTED_NUMERIC fields round-trip through a genuine Parquet file and are read back through
+ * {@link ParquetDocValuesProducer} on the DataFusion decode path.
+ *
+ * SORTED / SORTED_SET fields delegate to Lucene90 inside the round-trip format: without the
+ * composite engine's sidecar terms index the Parquet sorted path is deliberately fail-fast
+ * (see the tiered-ordinals design doc), so those tests exercise the delegate, not our stack.
+ */
+public class ParquetRoundTripDocValuesFormatTests extends BaseDocValuesFormatTestCase {
+
+ private Codec codec;
+
+ @Override
+ protected Codec getCodec() {
+ if (codec == null) {
+ RustBridge.initLogger();
+ // This branch still defaults to the legacy codec_native decode path; the production
+ // configuration under test is the DataFusion path.
+ ParquetDocValuesProducer.setDecodePath(org.opensearch.parquet.ParquetSettings.DECODE_PATH_DATAFUSION);
+ try {
+ RoundTripParquetDocValuesFormat.SPILL_DIR = Files.createTempDirectory("parquet-dv-roundtrip");
+ } catch (java.io.IOException e) {
+ throw new java.io.UncheckedIOException(e);
+ }
+ // Randomize the tier boundary so the battery exercises BOTH ordinal tiers:
+ // 0 forces every sorted field onto disk-backed uninverted ordinals; larger values
+ // keep low-cardinality fields on the heap dictionary tier.
+ RoundTripParquetDocValuesFormat.DICTIONARY_MAX_TERMS = org.apache.lucene.tests.util.LuceneTestCase.random()
+ .nextBoolean() ? 0 : 65536;
+ codec = TestUtil.alwaysDocValuesFormat(new RoundTripParquetDocValuesFormat(RoundTripParquetDocValuesFormat.SPILL_DIR));
+ }
+ return codec;
+ }
+}
diff --git a/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/codec/RoundTripParquetDocValuesFormat.java b/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/codec/RoundTripParquetDocValuesFormat.java
new file mode 100644
index 0000000000000..46bf47e400788
--- /dev/null
+++ b/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/codec/RoundTripParquetDocValuesFormat.java
@@ -0,0 +1,526 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.parquet.codec;
+
+import org.apache.arrow.c.ArrowArray;
+import org.apache.arrow.c.ArrowSchema;
+import org.apache.arrow.c.Data;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.BigIntVector;
+import org.apache.arrow.vector.VarBinaryVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.complex.ListVector;
+import org.apache.arrow.vector.complex.impl.UnionListWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.lucene.codecs.DocValuesConsumer;
+import org.apache.lucene.codecs.DocValuesFormat;
+import org.apache.lucene.codecs.DocValuesProducer;
+import org.apache.lucene.codecs.lucene90.Lucene90DocValuesFormat;
+import org.apache.lucene.index.BinaryDocValues;
+import org.apache.lucene.index.DocValuesSkipper;
+import org.apache.lucene.index.DocValuesType;
+import org.apache.lucene.index.FieldInfo;
+import org.apache.lucene.index.NumericDocValues;
+import org.apache.lucene.index.SegmentReadState;
+import org.apache.lucene.index.SegmentWriteState;
+import org.apache.lucene.index.SortedDocValues;
+import org.apache.lucene.index.SortedNumericDocValues;
+import org.apache.lucene.index.SortedSetDocValues;
+import org.apache.lucene.search.DocIdSetIterator;
+import org.apache.lucene.util.BytesRef;
+import org.opensearch.nativebridge.spi.ArrowExport;
+import org.opensearch.parquet.bridge.NativeParquetWriter;
+import org.opensearch.parquet.bridge.ParquetSortConfig;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ * Test-only round-trip {@link DocValuesFormat} that funnels Lucene's exhaustive
+ * {@code BaseDocValuesFormatTestCase} battery through the REAL Parquet read stack.
+ *
+ *
Write side: buffers NUMERIC / BINARY / SORTED_NUMERIC doc values handed over by
+ * {@code IndexWriter} (including merges) and spills them to a genuine Parquet file via
+ * {@link NativeParquetWriter}, stamping the file's path into the segment attributes exactly like
+ * the composite engine does. SORTED / SORTED_SET fields delegate to {@link Lucene90DocValuesFormat}:
+ * without the composite engine's sidecar terms index, the Parquet sorted path is deliberately
+ * non-contractual (streaming fail-fast), so running the sorted battery against it would only
+ * measure known refusals rather than find bugs.
+ *
+ *
Read side: {@link ParquetDocValuesProducer} — the production producer, DataFusion decode
+ * path, page cache, skipper and all.
+ */
+public final class RoundTripParquetDocValuesFormat extends DocValuesFormat {
+
+ public static final String NAME = "RoundTripParquet";
+
+ /** Spill directory for the current test run; set by the test before indexing. */
+ public static volatile Path SPILL_DIR;
+
+ /**
+ * Dictionary-tier cardinality budget for the current test run (mirrors
+ * {@code parquet.docvalues.dictionary.max_terms}). Randomized by the test so the battery
+ * exercises BOTH ordinal tiers: tiny values force the disk-backed uninverted tier, large
+ * values keep fields on the heap dictionary tier.
+ */
+ public static volatile int DICTIONARY_MAX_TERMS = 65536;
+
+ private final Lucene90DocValuesFormat sortedDelegate = new Lucene90DocValuesFormat();
+ private final Path spillDir;
+
+ /** No-arg constructor for Lucene SPI (read path resolves the format by name). */
+ public RoundTripParquetDocValuesFormat() {
+ super(NAME);
+ this.spillDir = null;
+ }
+
+ private Path spillDir() {
+ Path dir = spillDir != null ? spillDir : SPILL_DIR;
+ if (dir == null) {
+ throw new IllegalStateException("RoundTripParquetDocValuesFormat.SPILL_DIR not set by the test");
+ }
+ return dir;
+ }
+
+ public RoundTripParquetDocValuesFormat(Path spillDir) {
+ super(NAME);
+ this.spillDir = spillDir;
+ }
+
+ @Override
+ public DocValuesConsumer fieldsConsumer(SegmentWriteState state) throws IOException {
+ return new RecordingConsumer(state, sortedDelegate.fieldsConsumer(state), spillDir());
+ }
+
+ @Override
+ public DocValuesProducer fieldsProducer(SegmentReadState state) throws IOException {
+ DocValuesProducer sorted = null;
+ boolean hasSorted = false;
+ for (FieldInfo fi : state.fieldInfos) {
+ DocValuesType t = fi.getDocValuesType();
+ if (t == DocValuesType.SORTED || t == DocValuesType.SORTED_SET) {
+ hasSorted = true;
+ }
+ }
+ if (hasSorted) {
+ // The delegate consumer always ran (its files exist even with zero entries), so
+ // opening it is safe; routing decides per field.
+ sorted = sortedDelegate.fieldsProducer(state);
+ }
+ // The consumer stamps the attribute whenever it spilled ANY field — numeric shapes or
+ // sidecar-backed sorted fields — so attribute presence alone decides.
+ ParquetDocValuesProducer parquet = null;
+ Path parquetFile = null;
+ String attr = state.segmentInfo.getAttribute(ParquetSegmentLayout.PARQUET_FILE_ATTRIBUTE);
+ if (attr != null) {
+ parquet = new ParquetDocValuesProducer(state, null);
+ parquetFile = Path.of(attr);
+ }
+ return new RoutingProducer(parquet, sorted, parquetFile, state, spillDir());
+ }
+
+ static Path sidecarPath(Path parquetFile, String field) {
+ return parquetFile.resolveSibling(parquetFile.getFileName() + "." + field + ".terms");
+ }
+
+ /**
+ * Routes reads: numeric shapes to the parquet stack; sorted shapes with a terms sidecar to
+ * the REAL ordinal tiers ({@link TermDictionary} within budget, disk-backed
+ * {@link UninvertedOrdinals} above it — coverage verification included); sorted shapes
+ * without a sidecar (multi-valued sortedset, skipper fields) to the Lucene90 delegate.
+ */
+ private static final class RoutingProducer extends DocValuesProducer {
+ private final ParquetDocValuesProducer parquet;
+ private final DocValuesProducer sorted;
+ private final Path parquetFile;
+ private final SegmentReadState state;
+ private final Path ordsDir;
+ private final List openedOrdinals = new ArrayList<>();
+
+ RoutingProducer(ParquetDocValuesProducer parquet, DocValuesProducer sorted, Path parquetFile, SegmentReadState state, Path ordsDir) {
+ this.parquet = parquet;
+ this.sorted = sorted;
+ this.parquetFile = parquetFile;
+ this.state = state;
+ this.ordsDir = ordsDir;
+ }
+
+ /** The production tier ladder, minus the node-level caches (fresh per producer). */
+ private SortedDocValues tieredSorted(FieldInfo field) throws IOException {
+ Path sidecar = sidecarPath(parquetFile, field.name);
+ org.apache.lucene.index.Terms terms = SidecarTerms.read(sidecar);
+ org.opensearch.parquet.codec.iter.ParquetSortedDocValues streaming =
+ (org.opensearch.parquet.codec.iter.ParquetSortedDocValues) parquet.getSorted(field);
+ TermDictionary dictionary = TermDictionary.load(terms, DICTIONARY_MAX_TERMS);
+ if (dictionary != null) {
+ return new org.opensearch.parquet.codec.iter.ParquetDictionarySortedDocValues(streaming, dictionary);
+ }
+ long expectedNonNull = parquet.nonNullRowCount(field);
+ UninvertedOrdinals ordinals = UninvertedOrdinals.build(
+ ordsDir.resolve("ords"),
+ org.apache.lucene.util.StringHelper.idToString(state.segmentInfo.getId()) + "-" + field.name,
+ terms,
+ state.segmentInfo.maxDoc(),
+ expectedNonNull,
+ () -> false
+ );
+ openedOrdinals.add(ordinals);
+ return new org.opensearch.parquet.codec.iter.ParquetUninvertedSortedDocValues(
+ ordinals,
+ streaming,
+ state.segmentInfo.maxDoc()
+ );
+ }
+
+ private boolean hasSidecar(FieldInfo field) {
+ return parquetFile != null && java.nio.file.Files.exists(sidecarPath(parquetFile, field.name));
+ }
+
+ @Override
+ public NumericDocValues getNumeric(FieldInfo field) throws IOException {
+ return parquet.getNumeric(field);
+ }
+
+ @Override
+ public BinaryDocValues getBinary(FieldInfo field) throws IOException {
+ return parquet.getBinary(field);
+ }
+
+ @Override
+ public SortedNumericDocValues getSortedNumeric(FieldInfo field) throws IOException {
+ return parquet.getSortedNumeric(field);
+ }
+
+ @Override
+ public SortedDocValues getSorted(FieldInfo field) throws IOException {
+ return hasSidecar(field) ? tieredSorted(field) : sorted.getSorted(field);
+ }
+
+ @Override
+ public SortedSetDocValues getSortedSet(FieldInfo field) throws IOException {
+ // Single-valued fields round-trip through our tiers with the production singleton
+ // convention; multi-valued fields (no sidecar) stay on the delegate.
+ return hasSidecar(field)
+ ? org.apache.lucene.index.DocValues.singleton(tieredSorted(field))
+ : sorted.getSortedSet(field);
+ }
+
+ @Override
+ public DocValuesSkipper getSkipper(FieldInfo field) throws IOException {
+ DocValuesType t = field.getDocValuesType();
+ if (t == DocValuesType.SORTED || t == DocValuesType.SORTED_SET) {
+ return sorted.getSkipper(field);
+ }
+ return parquet.getSkipper(field);
+ }
+
+ @Override
+ public void checkIntegrity() throws IOException {
+ if (parquet != null) {
+ parquet.checkIntegrity();
+ }
+ if (sorted != null) {
+ sorted.checkIntegrity();
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ IOException first = null;
+ for (UninvertedOrdinals ordinals : openedOrdinals) {
+ try {
+ ordinals.close();
+ } catch (IOException e) {
+ if (first == null) {
+ first = e;
+ }
+ }
+ }
+ if (parquet != null) {
+ try {
+ parquet.close();
+ } catch (IOException e) {
+ if (first == null) {
+ first = e;
+ }
+ }
+ }
+ if (sorted != null) {
+ sorted.close();
+ }
+ if (first != null) {
+ throw first;
+ }
+ }
+ }
+
+ /** Buffers parquet-bound fields; delegates sorted shapes; spills parquet on close. */
+ private static final class RecordingConsumer extends DocValuesConsumer {
+
+ private final SegmentWriteState state;
+ private final DocValuesConsumer sortedDelegate;
+ private final Path spillDir;
+ private final int maxDoc;
+
+ // field name → per-doc values; null element = missing document.
+ private final Map numericFields = new LinkedHashMap<>();
+ private final Map binaryFields = new LinkedHashMap<>();
+ private final Map sortedNumericFields = new LinkedHashMap<>();
+ // Sorted fields served by OUR tiers: flat binary column + terms sidecar.
+ private final Map sortedValueFields = new LinkedHashMap<>();
+ private final Map> sortedTerms = new LinkedHashMap<>();
+ private final Map> sortedPostings = new LinkedHashMap<>();
+
+ RecordingConsumer(SegmentWriteState state, DocValuesConsumer sortedDelegate, Path spillDir) {
+ this.state = state;
+ this.sortedDelegate = sortedDelegate;
+ this.spillDir = spillDir;
+ this.maxDoc = state.segmentInfo.maxDoc();
+ }
+
+ @Override
+ public void addNumericField(FieldInfo field, DocValuesProducer valuesProducer) throws IOException {
+ NumericDocValues values = valuesProducer.getNumeric(field);
+ long[][] perDoc = new long[maxDoc][];
+ for (int doc = values.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = values.nextDoc()) {
+ perDoc[doc] = new long[] { values.longValue() };
+ }
+ numericFields.put(field.name, perDoc);
+ }
+
+ @Override
+ public void addBinaryField(FieldInfo field, DocValuesProducer valuesProducer) throws IOException {
+ BinaryDocValues values = valuesProducer.getBinary(field);
+ byte[][] perDoc = new byte[maxDoc][];
+ for (int doc = values.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = values.nextDoc()) {
+ BytesRef v = values.binaryValue();
+ perDoc[doc] = new byte[v.length];
+ System.arraycopy(v.bytes, v.offset, perDoc[doc], 0, v.length);
+ }
+ binaryFields.put(field.name, perDoc);
+ }
+
+ @Override
+ public void addSortedNumericField(FieldInfo field, DocValuesProducer valuesProducer) throws IOException {
+ SortedNumericDocValues values = valuesProducer.getSortedNumeric(field);
+ long[][] perDoc = new long[maxDoc][];
+ for (int doc = values.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = values.nextDoc()) {
+ long[] docValues = new long[values.docValueCount()];
+ for (int i = 0; i < docValues.length; i++) {
+ docValues[i] = values.nextValue();
+ }
+ perDoc[doc] = docValues;
+ }
+ sortedNumericFields.put(field.name, perDoc);
+ }
+
+ @Override
+ public void addSortedField(FieldInfo field, DocValuesProducer valuesProducer) throws IOException {
+ if (field.docValuesSkipIndexType() != org.apache.lucene.index.DocValuesSkipIndexType.NONE) {
+ // Production declares skip indexes only for numeric shapes; sorted-with-skipper
+ // is not our feature, keep it on the delegate for a compliant skipper.
+ sortedDelegate.addSortedField(field, valuesProducer);
+ return;
+ }
+ captureSorted(field, valuesProducer.getSorted(field));
+ }
+
+ @Override
+ public void addSortedSetField(FieldInfo field, DocValuesProducer valuesProducer) throws IOException {
+ if (field.docValuesSkipIndexType() != org.apache.lucene.index.DocValuesSkipIndexType.NONE) {
+ sortedDelegate.addSortedSetField(field, valuesProducer);
+ return;
+ }
+ // Our production convention supports single-valued keyword fields (singleton
+ // sortedset). Multi-valued sortedset ordinals are a documented unsupported feature:
+ // those fields stay on the delegate.
+ SortedSetDocValues probe = valuesProducer.getSortedSet(field);
+ boolean singleValued = true;
+ for (int doc = probe.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = probe.nextDoc()) {
+ if (probe.docValueCount() > 1) {
+ singleValued = false;
+ break;
+ }
+ }
+ if (singleValued == false) {
+ sortedDelegate.addSortedSetField(field, valuesProducer);
+ return;
+ }
+ SortedSetDocValues values = valuesProducer.getSortedSet(field);
+ captureSortedOrds(field, values.getValueCount(), values::lookupOrd, new OrdIterator() {
+ @Override
+ public int nextDoc() throws IOException {
+ return values.nextDoc();
+ }
+
+ @Override
+ public int ord() throws IOException {
+ return (int) values.nextOrd();
+ }
+ });
+ }
+
+ private void captureSorted(FieldInfo field, SortedDocValues values) throws IOException {
+ captureSortedOrds(field, values.getValueCount(), values::lookupOrd, new OrdIterator() {
+ @Override
+ public int nextDoc() throws IOException {
+ return values.nextDoc();
+ }
+
+ @Override
+ public int ord() throws IOException {
+ return values.ordValue();
+ }
+ });
+ }
+
+ private interface OrdIterator {
+ int nextDoc() throws IOException;
+
+ int ord() throws IOException;
+ }
+
+ private void captureSortedOrds(FieldInfo field, long valueCount, OrdToTerm lookupOrd, OrdIterator it) throws IOException {
+ List terms = new ArrayList<>((int) valueCount);
+ for (int ord = 0; ord < valueCount; ord++) {
+ terms.add(BytesRef.deepCopyOf(lookupOrd.term(ord)));
+ }
+ byte[][] perDoc = new byte[maxDoc][];
+ List> postings = new ArrayList<>((int) valueCount);
+ for (int ord = 0; ord < valueCount; ord++) {
+ postings.add(new ArrayList<>());
+ }
+ for (int doc = it.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = it.nextDoc()) {
+ int ord = it.ord();
+ BytesRef term = terms.get(ord);
+ perDoc[doc] = new byte[term.length];
+ System.arraycopy(term.bytes, term.offset, perDoc[doc], 0, term.length);
+ postings.get(ord).add(doc);
+ }
+ List postingArrays = new ArrayList<>((int) valueCount);
+ for (List docs : postings) {
+ postingArrays.add(docs.stream().mapToInt(Integer::intValue).toArray());
+ }
+ sortedValueFields.put(field.name, perDoc);
+ sortedTerms.put(field.name, terms);
+ sortedPostings.put(field.name, postingArrays);
+ }
+
+ private interface OrdToTerm {
+ BytesRef term(int ord) throws IOException;
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ boolean any = numericFields.isEmpty() == false
+ || binaryFields.isEmpty() == false
+ || sortedNumericFields.isEmpty() == false
+ || sortedValueFields.isEmpty() == false;
+ if (any) {
+ Path file = spillDir.resolve(state.segmentInfo.name + "_" + UUID.randomUUID() + ".parquet");
+ writeParquet(file);
+ for (String field : sortedTerms.keySet()) {
+ SidecarTerms.write(sidecarPath(file, field), sortedTerms.get(field), sortedPostings.get(field));
+ }
+ state.segmentInfo.putAttribute(ParquetSegmentLayout.PARQUET_FILE_ATTRIBUTE, file.toString());
+ }
+ } finally {
+ sortedDelegate.close();
+ }
+ }
+
+ private void writeParquet(Path file) throws IOException {
+ List arrowFields = new ArrayList<>();
+ for (String name : numericFields.keySet()) {
+ arrowFields.add(new Field(name, FieldType.nullable(new ArrowType.Int(64, true)), null));
+ }
+ for (String name : binaryFields.keySet()) {
+ arrowFields.add(new Field(name, FieldType.nullable(new ArrowType.Binary()), null));
+ }
+ for (String name : sortedValueFields.keySet()) {
+ arrowFields.add(new Field(name, FieldType.nullable(new ArrowType.Binary()), null));
+ }
+ for (String name : sortedNumericFields.keySet()) {
+ Field item = new Field("item", FieldType.nullable(new ArrowType.Int(64, true)), null);
+ arrowFields.add(new Field(name, FieldType.nullable(new ArrowType.List()), List.of(item)));
+ }
+ Schema schema = new Schema(arrowFields);
+
+ try (BufferAllocator allocator = new RootAllocator()) {
+ NativeParquetWriter writer = new NativeParquetWriter(file.toString());
+ ArrowSchema schemaExport = ArrowSchema.allocateNew(allocator);
+ Data.exportSchema(allocator, schema, null, schemaExport);
+ try (ArrowExport export = new ArrowExport(null, schemaExport)) {
+ writer.initialize("dv-roundtrip-test", export.getSchemaAddress(), ParquetSortConfig.empty(), 0L);
+ }
+ try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
+ for (Map.Entry e : numericFields.entrySet()) {
+ BigIntVector vec = (BigIntVector) root.getVector(e.getKey());
+ for (int doc = 0; doc < maxDoc; doc++) {
+ long[] v = e.getValue()[doc];
+ if (v == null) {
+ vec.setNull(doc);
+ } else {
+ vec.setSafe(doc, v[0]);
+ }
+ }
+ }
+ Map allBinary = new LinkedHashMap<>(binaryFields);
+ allBinary.putAll(sortedValueFields);
+ for (Map.Entry e : allBinary.entrySet()) {
+ VarBinaryVector vec = (VarBinaryVector) root.getVector(e.getKey());
+ for (int doc = 0; doc < maxDoc; doc++) {
+ byte[] v = e.getValue()[doc];
+ if (v == null) {
+ vec.setNull(doc);
+ } else {
+ vec.setSafe(doc, v);
+ }
+ }
+ }
+ for (Map.Entry e : sortedNumericFields.entrySet()) {
+ ListVector vec = (ListVector) root.getVector(e.getKey());
+ UnionListWriter listWriter = vec.getWriter();
+ for (int doc = 0; doc < maxDoc; doc++) {
+ long[] v = e.getValue()[doc];
+ listWriter.setPosition(doc);
+ if (v != null) {
+ listWriter.startList();
+ for (long value : v) {
+ listWriter.writeBigInt(value);
+ }
+ listWriter.endList();
+ }
+ }
+ vec.setValueCount(maxDoc);
+ }
+ root.setRowCount(maxDoc);
+
+ ArrowArray array = ArrowArray.allocateNew(allocator);
+ ArrowSchema arrowSchema = ArrowSchema.allocateNew(allocator);
+ Data.exportVectorSchemaRoot(allocator, root, null, array, arrowSchema);
+ try (ArrowExport export = new ArrowExport(array, arrowSchema)) {
+ writer.write(export.getArrayAddress(), export.getSchemaAddress());
+ }
+ }
+ writer.flush();
+ }
+ }
+ }
+}
diff --git a/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/codec/SidecarTerms.java b/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/codec/SidecarTerms.java
new file mode 100644
index 0000000000000..547f6c2603345
--- /dev/null
+++ b/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/codec/SidecarTerms.java
@@ -0,0 +1,251 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.parquet.codec;
+
+import org.apache.lucene.index.BaseTermsEnum;
+import org.apache.lucene.index.ImpactsEnum;
+import org.apache.lucene.index.IndexOptions;
+import org.apache.lucene.index.PostingsEnum;
+import org.apache.lucene.index.Terms;
+import org.apache.lucene.index.TermsEnum;
+import org.apache.lucene.search.DocIdSetIterator;
+import org.apache.lucene.util.BytesRef;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Test-only stand-in for the composite engine's Lucene sidecar terms index: sorted distinct
+ * terms with postings, serialized to a simple file at write time and served through the real
+ * {@link Terms}/{@link TermsEnum}/{@link PostingsEnum} contract at read time. This is what
+ * {@link TermDictionary} and {@link UninvertedOrdinals} rank against in production; feeding
+ * them through the same abstraction lets Lucene's doc-values contract battery exercise the
+ * ordinal tiers for real.
+ */
+final class SidecarTerms extends Terms {
+
+ private final BytesRef[] terms;
+ private final int[][] postings;
+ private final long sumDocFreq;
+ private final int docCount;
+
+ private SidecarTerms(BytesRef[] terms, int[][] postings, long sumDocFreq, int docCount) {
+ this.terms = terms;
+ this.postings = postings;
+ this.sumDocFreq = sumDocFreq;
+ this.docCount = docCount;
+ }
+
+ /** Serializes sorted terms + postings. {@code termToDocs[i]} must be ascending doc ids. */
+ static void write(Path file, List sortedTerms, List termToDocs) throws IOException {
+ try (DataOutputStream out = new DataOutputStream(Files.newOutputStream(file))) {
+ out.writeInt(sortedTerms.size());
+ for (int i = 0; i < sortedTerms.size(); i++) {
+ BytesRef term = sortedTerms.get(i);
+ out.writeInt(term.length);
+ out.write(term.bytes, term.offset, term.length);
+ int[] docs = termToDocs.get(i);
+ out.writeInt(docs.length);
+ for (int doc : docs) {
+ out.writeInt(doc);
+ }
+ }
+ }
+ }
+
+ static SidecarTerms read(Path file) throws IOException {
+ try (DataInputStream in = new DataInputStream(Files.newInputStream(file))) {
+ int termCount = in.readInt();
+ BytesRef[] terms = new BytesRef[termCount];
+ int[][] postings = new int[termCount][];
+ long sumDocFreq = 0;
+ java.util.BitSet docsWithValue = new java.util.BitSet();
+ for (int i = 0; i < termCount; i++) {
+ byte[] bytes = new byte[in.readInt()];
+ in.readFully(bytes);
+ terms[i] = new BytesRef(bytes);
+ int[] docs = new int[in.readInt()];
+ for (int d = 0; d < docs.length; d++) {
+ docs[d] = in.readInt();
+ docsWithValue.set(docs[d]);
+ }
+ postings[i] = docs;
+ sumDocFreq += docs.length;
+ }
+ return new SidecarTerms(terms, postings, sumDocFreq, docsWithValue.cardinality());
+ }
+ }
+
+ @Override
+ public TermsEnum iterator() {
+ return new SidecarTermsEnum();
+ }
+
+ @Override
+ public long size() {
+ return terms.length;
+ }
+
+ @Override
+ public long getSumTotalTermFreq() {
+ return sumDocFreq;
+ }
+
+ @Override
+ public long getSumDocFreq() {
+ return sumDocFreq;
+ }
+
+ @Override
+ public int getDocCount() {
+ return docCount;
+ }
+
+ @Override
+ public boolean hasFreqs() {
+ return false;
+ }
+
+ @Override
+ public boolean hasOffsets() {
+ return false;
+ }
+
+ @Override
+ public boolean hasPositions() {
+ return false;
+ }
+
+ @Override
+ public boolean hasPayloads() {
+ return false;
+ }
+
+ /**
+ * Deliberately mirrors BlockTree's behavior: {@code ord()} is unsupported, so consumers
+ * (OrdinalMap!) must go through {@link UninvertedOrdinals}'s ord-tracking wrapper —
+ * exactly the production constraint the battery should exercise.
+ */
+ private final class SidecarTermsEnum extends BaseTermsEnum {
+ private int position = -1;
+
+ @Override
+ public BytesRef next() {
+ position++;
+ return position < terms.length ? terms[position] : null;
+ }
+
+ @Override
+ public SeekStatus seekCeil(BytesRef text) {
+ int idx = Arrays.binarySearch(terms, text);
+ if (idx >= 0) {
+ position = idx;
+ return SeekStatus.FOUND;
+ }
+ position = -idx - 1;
+ return position >= terms.length ? SeekStatus.END : SeekStatus.NOT_FOUND;
+ }
+
+ @Override
+ public void seekExact(long ord) {
+ throw new UnsupportedOperationException("sidecar terms have no ord index (mirrors BlockTree)");
+ }
+
+ @Override
+ public BytesRef term() {
+ return terms[position];
+ }
+
+ @Override
+ public long ord() {
+ throw new UnsupportedOperationException("sidecar terms have no ord index (mirrors BlockTree)");
+ }
+
+ @Override
+ public int docFreq() {
+ return postings[position].length;
+ }
+
+ @Override
+ public long totalTermFreq() {
+ return postings[position].length;
+ }
+
+ @Override
+ public PostingsEnum postings(PostingsEnum reuse, int flags) {
+ int[] docs = postings[position];
+ return new PostingsEnum() {
+ private int idx = -1;
+
+ @Override
+ public int docID() {
+ if (idx < 0) {
+ return -1;
+ }
+ return idx < docs.length ? docs[idx] : DocIdSetIterator.NO_MORE_DOCS;
+ }
+
+ @Override
+ public int nextDoc() {
+ idx++;
+ return docID();
+ }
+
+ @Override
+ public int advance(int target) {
+ do {
+ idx++;
+ } while (idx < docs.length && docs[idx] < target);
+ return docID();
+ }
+
+ @Override
+ public long cost() {
+ return docs.length;
+ }
+
+ @Override
+ public int freq() {
+ return 1;
+ }
+
+ @Override
+ public int nextPosition() {
+ return -1;
+ }
+
+ @Override
+ public int startOffset() {
+ return -1;
+ }
+
+ @Override
+ public int endOffset() {
+ return -1;
+ }
+
+ @Override
+ public BytesRef getPayload() {
+ return null;
+ }
+ };
+ }
+
+ @Override
+ public ImpactsEnum impacts(int flags) {
+ throw new UnsupportedOperationException();
+ }
+ }
+}
diff --git a/sandbox/plugins/parquet-data-format/src/test/resources/META-INF/services/org.apache.lucene.codecs.DocValuesFormat b/sandbox/plugins/parquet-data-format/src/test/resources/META-INF/services/org.apache.lucene.codecs.DocValuesFormat
new file mode 100644
index 0000000000000..45d8866b98085
--- /dev/null
+++ b/sandbox/plugins/parquet-data-format/src/test/resources/META-INF/services/org.apache.lucene.codecs.DocValuesFormat
@@ -0,0 +1 @@
+org.opensearch.parquet.codec.RoundTripParquetDocValuesFormat