Skip to content

sorted set cache changes #68

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bharath-techie wants to merge 2 commits into
base: df-integration
Choose a base branch
from
Open

sorted set cache changes #68

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d47f13b
sorted set cache changes
bharath-techie Aug 3, 2026
9846e64
Lucene tests and benchmark
bharath-techie Aug 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 48 additions & 39 deletions ...hmarks/src/main/java/org/opensearch/parquet/benchmark/DocValuesRandomAccessBenchmark.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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).
*
* <p>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.
*
* <p>Access patterns:
* <ul>
* <li>{@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.</li>
* <li>{@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.</li>
* <li>{@code pageMissPingPong} — alternates between the first and last page: every call
* evicts the resident page, giving a pure worst-case decode measurement.</li>
* <li>{@code sequentialScan} — ascending doc IDs: the cursor's home turf; the AIMD window
* grows and almost every call is a resident-batch hit.</li>
* <li>{@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).</li>
* <li>{@code pageMissPingPong} — alternates between the first and last page: every call is
* a worst-case long-distance reposition.</li>
* <li>{@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).</li>
* </ul>
*
* <p>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.
* <p>{@code nullFraction} covers the required-column and nullable presence paths;
* {@code columnType} covers the i64 passthrough (INT64) and widening (INT32) conversions.
*
* <p>Run with:
* <pre>
Expand All @@ -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;

Expand All @@ -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;
Expand All @@ -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.
Expand All @@ -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) {
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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 {
Expand All @@ -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)) {
Expand Down
17 changes: 17 additions & 0 deletions ...ins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetDataFormatPlugin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -158,12 +158,22 @@ public Collection<Object> 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) {
Expand Down Expand Up @@ -253,6 +263,13 @@ public Map<DataFormat, StoreStrategy> 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<Setting<?>> getSettings() {
return ParquetSettings.getSettings();
Expand Down
38 changes: 38 additions & 0 deletions ...box/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetSettings.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Integer> 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<Long> 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<Long> 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<Boolean> DOCVALUES_DIAGNOSTICS = Setting.boolSetting(
"parquet.docvalues.diagnostics",
Expand Down Expand Up @@ -917,6 +952,9 @@ public static List<Setting<?>> 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,
Expand Down
39 changes: 38 additions & 1 deletion ...rquet-data-format/src/main/java/org/opensearch/parquet/bridge/DataFusionColumnReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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";
Expand Down Expand Up @@ -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);
}

Expand Down
Loading