Introduce a Parquet based doc values codec

[manaslohani]

Description

Introduces a Parquet-based doc-values codec at plugin scope. This is the read-path extraction from #22752, split out so the codec can be reviewed and merged independently of the server-side searcher integration.

What this adds

• A FilterDirectoryReader/FilterLeafReader pair that synthesizes FieldInfos for supported numeric fields present only in Parquet and serves their doc values through a Parquet-backed producer. Fields mapped doc_values: false are excluded from synthesis.
• ParquetDocValuesProducer with fail-closed gates: an opensearch.format_version footer stamp check and a numRows == maxDoc check; each doc-values iterator gets a dedicated forward-only native cursor, opened lazily and released with the leaf reader.
• Supported field types: long, integer, short, byte, double, float, date, date_nanos, unsigned_long, scaled_float, half_float, boolean.
• Writer side stamps the format version into the Parquet footer; the version constants are shared between writer and reader through native-bridge-common, with a build-time equality test so the two cannot drift.
• Rust cursor support for the bit-packed boolean and float16 borrow kinds.

What this intentionally does not add
The server-module integration that triggers this codec end to end (composite searcher acquisition) stays in #22752. Until that lands, this path is dormant: the plugin wires its reader wrapper via IndexModule#setReaderWrapper, but no search on main reaches it.

Testing
Unit tests at plugin scope: field admission, producer format-version and row-count gates, iterator contracts (null-presence bitmaps, backward-advance cursor reopen, multi-batch reloads), boolean reads at byte boundaries, half_float pinned to Lucene's exact sortable-short encoding, and a writer/reader format-version round trip. Rust-side cursor tests cover the new borrow kinds. End-to-end integration tests live with #22752.

Related Issues

Related to #22752 (server-side integration and end-to-end tests).

Check List

• Functionality includes testing.
• API changes companion pull request created, if applicable.
• Public documentation issue/PR created, if applicable.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 1757a40)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

📝 TODO sections sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesProducer.java [167]: multi-value: no repeated read path; write path emits single values only.

🔀 Multiple PR themes Sub-PR theme: Introduce shared format-version constant and expose it in Parquet writer metadata Relevant files: sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetDataFormatPlugin.java sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/bridge/ParquetFileMetadata.java sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/bridge/RustBridge.java sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/writer/ParquetWriterTests.java sandbox/libs/dataformat-native/rust/common/src/format_version.rs sandbox/libs/dataformat-native/rust/common/src/lib.rs sandbox/plugins/parquet-data-format/src/main/rust/src/ffm.rs sandbox/plugins/parquet-data-format/src/main/rust/src/writer_properties_builder.rs Sub-PR theme: Add bit-packed Boolean and Float16 borrow kinds to the Parquet cursor bridge Relevant files: sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/bridge/DecodedBatch.java sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/bridge/ParquetCodecBridge.java sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/bridge/ParquetColumnReader.java sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/docvalues/bridge/DecodedBatchTests.java sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/docvalues/bridge/ParquetColumnReaderTests.java sandbox/plugins/analytics-backend-datafusion/rust/src/doc_values_cursor.rs Sub-PR theme: Parquet DocValues reader wrapper, producer, and Lucene DV iterator Relevant files: sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/CursorRegistry.java sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/FieldTypeMapping.java sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesDirectoryReader.java sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesLeafReader.java sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesProducer.java sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesProducerRegistry.java sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetSegmentLayout.java sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/iter/ParquetNumericDocValues.java sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionPlugin.java

⚡ Recommended focus areas for review Possible Issue In nextPresentRow, the first-byte start mask (0xFF << (bit & 7)) is computed as an int shift. This is fine here since bit & 7 is 0..7, but the mask is then AND'ed with the byte value; the loop's byteIdx << 3 comparison against lastBit is correct only when byteIdx stays within the bitmap's allocated bytes. On the very last byte, if bits has trailing zeros past lastBit, foundBit > lastBit guards correctly, but the subsequent iteration increments byteIdx and reads presenceBits.get(..., byteIdx) which may read one byte past the allocated presenceBytes length. The bitmap's allocation is sized to ((bitOffset + batchRows + 7) >>> 3), so reading byteIdx that satisfies (byteIdx << 3) <= lastBit should be safe, but confirm the loop-exit condition (byteIdx << 3) > lastBit runs before the next get. public long nextPresentRow(long fromRow) { if (contains(fromRow) == false) { throw new IndexOutOfBoundsException("row " + fromRow + " outside batch [" + firstRow + ", " + lastRow + "]"); } if (presenceBits == null) { return fromRow; } final long lastBit = lastRow - firstRow + presenceBitOffset; long bit = fromRow - firstRow + presenceBitOffset; // First byte: mask off bits below the starting row so an earlier present row is not reported. int bits = (presenceBits.get(ValueLayout.JAVA_BYTE, bit >>> 3) & 0xFF) & (0xFF << (bit & 7)); for (long byteIdx = bit >>> 3;;) { if (bits != 0) { long foundBit = (byteIdx << 3) + Integer.numberOfTrailingZeros(bits); if (foundBit > lastBit) { return -1; } return firstRow + (foundBit - presenceBitOffset); } byteIdx++; if ((byteIdx << 3) > lastBit) { return -1; } bits = presenceBits.get(ValueLayout.JAVA_BYTE, byteIdx) & 0xFF; } } Cross-Index Leak Risk PRODUCERS is a static (JVM-global) map keyed by Lucene CacheKey, but the producer captures a MapperService and index settings from one specific index. Cache keys are unique per segment core, so there is no functional collision, but the static registry means that if a plugin is reloaded or on a shared classloader test setup, previously-cached producers remain reachable and their closed-listener path depends on the core actually being dropped. Consider whether a node-scoped (rather than static) registry would be safer for lifecycle isolation. final class ParquetDocValuesProducerRegistry { private static final Map<IndexReader.CacheKey, ParquetDocValuesProducer> PRODUCERS = new ConcurrentHashMap<>(); private ParquetDocValuesProducerRegistry() {} /** * The producer for the segment core identified by {@code coreHelper}, creating it on first use * via {@code factory} and registering a closed-listener that closes it when Lucene drops the * core. */ static ParquetDocValuesProducer getOrCreate( IndexReader.CacheHelper coreHelper, CheckedSupplier<ParquetDocValuesProducer, IOException> factory ) throws IOException { IndexReader.CacheKey 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 = factory.get(); PRODUCERS.put(key, created); // Registered once, in the create branch only, so a core carries exactly one listener no // matter how many requests wrap its leaf. coreHelper.addClosedListener(ParquetDocValuesProducerRegistry::onCoreClosed); return created; } } /** Closes and drops the producer bound to a segment core when Lucene drops the core. */ private static void onCoreClosed(IndexReader.CacheKey key) throws IOException { ParquetDocValuesProducer removed = PRODUCERS.remove(key); if (removed != null) { removed.close(); } } Performance Concern assertRowIdsAreIdentity scans the entire segment on the first getSortedNumericDocValues call under assertions. While memoized and assertions-only, on large segments this adds a full O(maxDoc) walk per leaf on the first request. The synchronized method also serializes all first-time callers. This is acceptable for assertions but worth noting: any test/dev environment with large segments will see a first-request latency spike. private synchronized boolean assertRowIdsAreIdentity() throws IOException { if (rowIdsChecked) { return rowIdsAreIdentity; } rowIdsChecked = true; rowIdsAreIdentity = computeRowIdsAreIdentity(); return rowIdsAreIdentity; } private boolean computeRowIdsAreIdentity() throws IOException { SortedNumericDocValues rowId = in.getSortedNumericDocValues(DocumentInput.ROW_ID_FIELD); if (rowId == null) { return true; // no row-id field => identity by definition } for (int docId = 0; docId < maxDoc(); docId++) { if (rowId.advanceExact(docId) == false || rowId.nextValue() != docId) { return false; } } return true; } Exception Swallowed In close(), cursor.close() is called inside the loop with the comment "close never throws checked exceptions", but if any cursor's close throws an unchecked exception (RuntimeException from native side), remaining cursors will not be closed and will leak. Consider collecting exceptions and continuing to close all cursors, similar to the pattern used in ParquetDocValuesDirectoryReader.doClose. public void close() throws IOException { synchronized (cursors) { if (closed) { return; } closed = true; for (ParquetColumnReader cursor : cursors) { // Idempotent through NativeHandle; close never throws checked exceptions. cursor.close(); } cursors.clear(); } }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 1757a40

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Prevent producer leak on listener failure If addClosedListener throws after the producer has been put into the map, the producer will leak (never closed, never removed). Additionally, if the core is already closed at registration time, the listener may never fire. Register the closed listener before inserting into the map, or close the producer and remove the map entry on listener registration failure. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesProducerRegistry.java [47-58] synchronized (PRODUCERS) { existing = PRODUCERS.get(key); if (existing != null) { return existing; } ParquetDocValuesProducer created = factory.get(); + try { + coreHelper.addClosedListener(ParquetDocValuesProducerRegistry::onCoreClosed); + } catch (RuntimeException e) { + created.close(); + throw e; + } PRODUCERS.put(key, created); - // Registered once, in the create branch only, so a core carries exactly one listener no - // matter how many requests wrap its leaf. - coreHelper.addClosedListener(ParquetDocValuesProducerRegistry::onCoreClosed); return created; } Suggestion importance[1-10]: 5 __ Why: Valid defensive concern about resource leak if addClosedListener throws after PRODUCERS.put. The scenario is unlikely in practice, but the fix improves robustness of the lifecycle handling.	Low
General	Verify sign-extension of half-float encoding The result of the XOR is cast to short but returned via yield into a long switch expression, which will sign-extend the negative sortable-short values. valueAt returns long, so a negative sortable short will be returned as a negative long — but callers likely compare against unsigned 16-bit patterns for sorting semantics. Verify sign extension is intended; otherwise mask with & 0xFFFFL to preserve the unsigned 16-bit representation as the other unsigned kinds do. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/bridge/DecodedBatch.java [120-125] +case KIND_HALF_FLOAT -> { + // fp16-width sign-flip; exhaustively verified equal to HalfFloatPoint.halfFloatToSortableShort, + // avoiding the lucene-sandbox dependency. + short bits = values.getAtIndex(ValueLayout.JAVA_SHORT, idx); + yield (short) (bits ^ ((bits >> 15) & 0x7fff)); +} - Suggestion importance[1-10]: 3 __ Why: The suggestion questions the sign-extension behavior but the code correctly matches Lucene's HalfFloatPoint.halfFloatToSortableShort (verified by tests). The improved_code is identical to existing_code, making this a low-value verification-only suggestion.	Low
	Tighten bitmap scan bounds check The loop reads the next byte via presenceBits.get(..., byteIdx) without bounds-checking against the buffer's byte length. The buffer is sized to ((bitOffset + batchRows + 7) >>> 3) bytes, so once byteIdx advances past lastBit >>> 3 a read could exceed the reinterpreted segment. The termination check (byteIdx << 3) > lastBit runs before the read but permits reading the byte containing lastBit, which is fine — but ensure the final byte's trailing bits past lastBit cannot spuriously return a bit index > lastBit; the foundBit > lastBit check handles this correctly. Consider adding an assertion that byteIdx stays within the segment size for safety. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/bridge/DecodedBatch.java [152-169] final long lastBit = lastRow - firstRow + presenceBitOffset; +final long lastByteIdx = lastBit >>> 3; long bit = fromRow - firstRow + presenceBitOffset; -// First byte: mask off bits below the starting row so an earlier present row is not reported. int bits = (presenceBits.get(ValueLayout.JAVA_BYTE, bit >>> 3) & 0xFF) & (0xFF << (bit & 7)); for (long byteIdx = bit >>> 3;;) { if (bits != 0) { long foundBit = (byteIdx << 3) + Integer.numberOfTrailingZeros(bits); if (foundBit > lastBit) { return -1; } return firstRow + (foundBit - presenceBitOffset); } byteIdx++; - if ((byteIdx << 3) > lastBit) { + if (byteIdx > lastByteIdx) { return -1; } bits = presenceBits.get(ValueLayout.JAVA_BYTE, byteIdx) & 0xFF; } Suggestion importance[1-10]: 3 __ Why: A minor readability improvement using lastByteIdx instead of shifting each iteration. The original bounds check (byteIdx << 3) > lastBit is functionally equivalent and safe; no actual bug is fixed.	Low
	Document silent null return for numeric accessor Returning null for a synthesized Parquet field on getNumericDocValues is correct per DV-type contract, but any consumer that expected NUMERIC (single-valued path) will silently see no values instead of getting the singleton via SORTED_NUMERIC unwrap. Consider logging at debug level or documenting more prominently that callers must request SORTED_NUMERIC for Parquet fields; otherwise this becomes a silent failure mode when numeric-only aggregations bypass the sorted-numeric path. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesLeafReader.java [223-230] @Override public NumericDocValues getNumericDocValues(String field) throws IOException { if (parquetFieldInfo(field) != null) { - // Synthesized Parquet fields are SORTED_NUMERIC; like CodecReader, an accessor whose DV - // type does not match the FieldInfo returns null rather than serving the field. + // Parquet-served fields are exposed as SORTED_NUMERIC only; callers must go through + // getSortedNumericDocValues + DocValues.unwrapSingleton. Returning null here matches + // Lucene's CodecReader contract when the requested DV type does not match FieldInfo. return null; } return in.getNumericDocValues(field); } Suggestion importance[1-10]: 2 __ Why: The suggestion only asks to improve comments/documentation, which is low-impact. The existing comment already explains the CodecReader contract behavior adequately.	Low

---

Previous suggestions

Suggestions up to commit 14aeb50

Category	Suggestion	Impact
Possible issue	Set closed flag under the monitor Setting closed = true before entering the synchronized block leaves a window where dedicatedReaderFor has already opened and added a reader (holding the monitor) after the drain loop finishes but before the flag is set — actually the reverse: a dedicatedReaderFor call already past ensureOpen() but not yet in the synchronized block can add its reader to the list after this method clears it, leaking the cursor. Move closed = true inside the synchronized block, before the drain loop, so any concurrent registration under the same monitor observes closed=true and closes its own reader. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesProducer.java [196-213] public void close() throws IOException { - if (closed) { - return; - } - closed = true; synchronized (dedicatedReaders) { + if (closed) { + return; + } + closed = true; for (ParquetColumnReader reader : dedicatedReaders) { Suggestion importance[1-10]: 5 __ Why: Identifies a plausible race: closed is set before entering the synchronized block, so a concurrent dedicatedReaderFor could add a reader after the drain but before observing closed=true. Moving the flag assignment inside the monitor tightens the invariant, though the volatile flag and lock combination make the actual window narrow.	Low
	Fix close/open race on closed flag The closed field is volatile but is set outside the dedicatedReaders monitor in close(), so the race the comment claims to guard against is not actually prevented: a thread can pass ensureOpen(), then close() runs to completion (draining the list and setting closed=true), then this method acquires the lock, sees closed=true and closes its own reader — but if instead close() runs between the open() call and the synchronized block acquiring the monitor, the check works. However, if close() starts and finishes between ensureOpen() and ParquetColumnReader.open(), the newly opened reader is closed here — that is fine. The real bug is that close() sets closed=true before acquiring the monitor, so a concurrent dedicatedReaderFor may already have added its reader to dedicatedReaders after the loop drained it. Set closed=true inside the synchronized (dedicatedReaders) block to close this window. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesProducer.java [268-280] private ParquetColumnReader dedicatedReaderFor(FieldInfo field) throws IOException { ParquetColumnReader reader = ParquetColumnReader.open(parquetFile, field.getName(), indexSettings, storePointer); - // Register under the same lock close() clears the list under: an open that races a - // concurrent close would otherwise add to an already-drained list and leak the cursor. synchronized (dedicatedReaders) { if (closed) { reader.close(); throw new IllegalStateException("producer for " + parquetFile + " is closed"); } dedicatedReaders.add(reader); } return reader; } Suggestion importance[1-10]: 3 __ Why: The suggestion's improved_code is essentially identical to the existing_code (just removes a comment). The actual claim about the race is addressed in suggestion 2, not here. Low impact as presented.	Low
General	Verify advance/advanceExact iterator contract advanceExact already sets doc = target on every call inside the loop, even when it returns false — so after this loop scans past several null docs, doc has been mutated to point at those null docs, then reset to NO_MORE_DOCS only if no live doc is found. When a live doc is found, doc = d is redundant but harmless. More importantly, advanceExact sets doc = NO_MORE_DOCS when target >= maxDoc, so calling it in a loop that already checks d < maxDoc is fine — but advanceExact is contractually meant for random access, not iteration, and using it inside advance may violate Lucene's expectation that advance monotonically moves forward. Consider iterating batches directly or documenting this reuse; at minimum verify that advanceExact returning false on a null doc still leaves the iterator in a state where the next advance call works correctly. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/iter/ParquetNumericDocValues.java [76-85] @Override public int advance(int target) throws IOException { for (int d = target; d < maxDoc; d++) { if (advanceExact(d)) { - doc = d; return d; } } doc = NO_MORE_DOCS; return NO_MORE_DOCS; } Suggestion importance[1-10]: 3 __ Why: Mostly a request to verify behavior, with a minor cleanup of a redundant doc = d assignment. The redundancy is harmless and the underlying concern about the iterator contract is speculative.	Low
	Make half_float sign-extension explicit Returning a short from a switch expression whose result type is long will sign-extend the sortable short, but callers of valueAt treat the result as an unsigned bit pattern in some kinds (e.g., KIND_USHORT uses & 0xFFFFL). For KIND_HALF_FLOAT, sign-extending is intended (the sortable short must compare like a signed 16-bit integer when widened to long), so this is correct — but be explicit with a cast to (long) for clarity and to match the sign-extension contract documented for other kinds like KIND_SHORT. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/bridge/DecodedBatch.java [120-125] case KIND_HALF_FLOAT -> { // fp16-width sign-flip; exhaustively verified equal to HalfFloatPoint.halfFloatToSortableShort, // avoiding the lucene-sandbox dependency. short bits = values.getAtIndex(ValueLayout.JAVA_SHORT, idx); - yield (short) (bits ^ ((bits >> 15) & 0x7fff)); + yield (long) (short) (bits ^ ((bits >> 15) & 0x7fff)); } Suggestion importance[1-10]: 2 __ Why: A stylistic clarification; the existing code already implicitly widens the short to long via the switch expression's result type with sign extension, which is the intended behavior. Marginal improvement.	Low

Suggestions up to commit d06196f

Category	Suggestion	Impact
Possible issue	Fix race between close and reader registration The closed flag is volatile and set outside the dedicatedReaders lock in close(), so the check-then-add here is not actually protected against a concurrent close: close() can drain the list, then this method acquires the lock and adds a new reader after closed was already observed as false earlier. Set closed = true inside the synchronized (dedicatedReaders) block in close() so the check-and-add here is fully mutually exclusive with the drain. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesProducer.java [280-292] private ParquetColumnReader dedicatedReaderFor(FieldInfo field) throws IOException { ParquetColumnReader reader = ParquetColumnReader.open(parquetFile, field.getName(), indexSettings, storePointer); - // Register under the same lock close() clears the list under: an open that races a - // concurrent close would otherwise add to an already-drained list and leak the cursor. synchronized (dedicatedReaders) { if (closed) { reader.close(); throw new IllegalStateException("producer for " + parquetFile + " is closed"); } dedicatedReaders.add(reader); } return reader; } Suggestion importance[1-10]: 5 __ Why: The suggestion identifies a subtle race, but the existing code already re-checks closed inside the synchronized block after acquiring the lock, so a concurrent close that drains the list first would be observed here. The improved code is essentially identical to the existing code (only removes a comment), so the actual impact is minimal.	Low
General	Avoid serializing first concurrent DV requests This assertion scan is invoked from getNumericDocValues/getSortedNumericDocValues, which are commonly called on many concurrent slices. Although memoized per-leaf, the first caller pays an O(maxDoc) scan while holding the synchronized monitor on assertRowIdsAreIdentity, blocking every other numeric DV request on this leaf. Perform the scan without holding the monitor (compute into a local, then set the memoized fields under the lock) to avoid serializing the first wave of concurrent requests. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesLeafReader.java [203-214] -private boolean computeRowIdsAreIdentity() throws IOException { - SortedNumericDocValues rowId = in.getSortedNumericDocValues(DocumentInput.ROW_ID_FIELD); - if (rowId == null) { - return true; // no row-id field => identity by definition +private boolean assertRowIdsAreIdentity() throws IOException { + if (rowIdsChecked) { + return rowIdsAreIdentity; } - for (int docId = 0; docId < maxDoc(); docId++) { - if (rowId.advanceExact(docId) == false || rowId.nextValue() != docId) { - return false; + boolean identity = computeRowIdsAreIdentity(); + synchronized (this) { + if (rowIdsChecked == false) { + rowIdsAreIdentity = identity; + rowIdsChecked = true; } + return rowIdsAreIdentity; } - return true; } Suggestion importance[1-10]: 4 __ Why: Valid concurrency concern about serializing the first wave of requests during the O(maxDoc) scan; the fix is reasonable but the assertion path only runs with assertions enabled, limiting real-world impact.	Low
	Simplify advance and rely on advanceExact state advanceExact already sets doc = target (or NO_MORE_DOCS) on every call, so the loop sets and then overwrites the state repeatedly, and importantly a caller expects advance(target) to return the first doc >= target for which values exist — but per the DocValues iterator contract, advance/nextDoc should position on the next document that has a value. Currently for a null row advanceExact returns false but leaves doc = d; the loop then continues fine, but note that docID() between advance returning and next call may be inconsistent. Confirm the semantics match Lucene's expectation that advance skips null docs; if the underlying column is sparse this is O(rows) per advance. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/iter/ParquetNumericDocValues.java [76-85] @Override public int advance(int target) throws IOException { for (int d = target; d < maxDoc; d++) { if (advanceExact(d)) { - doc = d; return d; } } doc = NO_MORE_DOCS; return NO_MORE_DOCS; } Suggestion importance[1-10]: 3 __ Why: A minor readability improvement (removing a redundant doc = d assignment since advanceExact already sets it). The commentary is speculative and mostly asks for confirmation rather than pointing to a definite bug.	Low
	Document sign-extension of half_float return value The method returns long, but this arm yields a short which gets sign-extended to long. For a negative sortable-short encoding, sign-extension will produce a negative long, which is fine for comparison but may surprise callers that treat the returned long as unsigned bits. Since NumericDocValues.longValue() for half_float is typically consumed by casting back to short (via (short) longValue()), this works, but consider masking with & 0xFFFFL to make the returned value a zero-extended 16-bit pattern for consistency with the other unsigned kinds — or document explicitly that the caller must cast back to short. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/bridge/DecodedBatch.java [120-125] case KIND_HALF_FLOAT -> { - // fp16-width sign-flip; exhaustively verified equal to HalfFloatPoint.halfFloatToSortableShort, - // avoiding the lucene-sandbox dependency. + // fp16-width sign-flip; exhaustively verified equal to HalfFloatPoint.halfFloatToSortableShort. + // Returned sign-extended so (short) longValue() recovers the sortable short Lucene stores. short bits = values.getAtIndex(ValueLayout.JAVA_SHORT, idx); yield (short) (bits ^ ((bits >> 15) & 0x7fff)); } Suggestion importance[1-10]: 2 __ Why: This is essentially a documentation-only change; the code behavior is unchanged and consumers already cast back to short.	Low

Suggestions up to commit a8b6817

Category	Suggestion	Impact
Possible issue	Avoid sign-extending half_float sortable short The KIND_HALF_FLOAT result is cast to short before yielding a long, which sign-extends negative sortable shorts to negative longs. Lucene's half_float doc values store the sortable short in the low 16 bits of a NumericDocValues long; downstream code (e.g. HalfFloatPoint.sortableShortToHalfFloat) typically expects the raw short bits without sign extension, and this differs from the KIND_FLOAT path which explicitly masks. Verify whether sign-extension or zero-extension of the sortable short is intended; if the latter, mask to 16 bits. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesProducer.java [120-125] -case DecodedBatch.KIND_HALF_FLOAT -> { +case KIND_HALF_FLOAT -> { // fp16-width sign-flip; exhaustively verified equal to HalfFloatPoint.halfFloatToSortableShort, // avoiding the lucene-sandbox dependency. short bits = values.getAtIndex(ValueLayout.JAVA_SHORT, idx); - yield (short) (bits ^ ((bits >> 15) & 0x7fff)); + yield (long) ((short) (bits ^ ((bits >> 15) & 0x7fff))) & 0xffffL; } Suggestion importance[1-10]: 6 __ Why: The concern about sign-extension of the sortable short to a long is valid: (short) cast then implicit widening to long sign-extends, whereas Lucene's HalfFloatPoint value source may expect only the low 16 bits. However, the suggestion is hedged ("Verify whether...") and the tests appear to pass, so the impact is uncertain but worth investigating.	Low
General	Handle fourth version component explicitly FORMAT_VERSION is "1.0.0.0" (four components) but this encoder only consumes three, silently ignoring the fourth. If a future version bump changes the fourth component only (e.g. "1.0.0.1"), the encoded value stays the same, defeating the version-gating intent. Either reject a fourth non-zero component or include it in the encoding to keep the Java constant and stamp comparable. sandbox/libs/dataformat-native/rust/common/src/format_version.rs [32-51] pub fn encode_format_version(raw: &str) -> i64 { if raw.is_empty() { return FORMAT_VERSION_UNKNOWN; } let mut parts = raw.split('.'); let mut encoded = 0i64; for scale in [1_000_000i64, 1_000, 1] { - // A missing minor/patch reads as 0 ("1" and "1.0.0" encode identically); a present but - // non-numeric or negative component makes the whole version unusable rather than partial. let part = match parts.next() { None => break, Some(part) => part, }; match part.parse::<i64>() { Ok(value) if value >= 0 => encoded += value * scale, _ => return FORMAT_VERSION_UNKNOWN, } } + // Reject a non-zero fourth component so a bump there cannot silently encode identically. + if let Some(extra) = parts.next() { + match extra.parse::<i64>() { + Ok(0) => {} + _ => return FORMAT_VERSION_UNKNOWN, + } + } encoded } Suggestion importance[1-10]: 5 __ Why: Valid observation: FORMAT_VERSION = "1.0.0.0" has four components but only three are encoded, so a future bump of the fourth component would silently produce the same encoded value, defeating the version-gating purpose.	Low
	Prevent leaked cursor on close race If reader.close() inside the race-guard throws, the caller sees a RuntimeException from close masking the IllegalStateException, and the cursor is leaked. Wrap the close in try/finally (or suppress) so the IllegalStateException is always thrown even when close fails. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesProducer.java [280-292] private ParquetColumnReader dedicatedReaderFor(FieldInfo field) throws IOException { ParquetColumnReader reader = ParquetColumnReader.open(parquetFile, field.getName(), indexSettings, storePointer); - // Register under the same lock close() clears the list under: an open that races a - // concurrent close would otherwise add to an already-drained list and leak the cursor. synchronized (dedicatedReaders) { if (closed) { - reader.close(); + try { + reader.close(); + } catch (RuntimeException | IOException e) { + logger.warn("Failed to close racing Parquet column reader for [{}]", parquetFile, e); + } throw new IllegalStateException("producer for " + parquetFile + " is closed"); } dedicatedReaders.add(reader); } return reader; } Suggestion importance[1-10]: 4 __ Why: Minor robustness improvement in an edge-case race condition where the close of a racing reader could throw. The improved code correctly ensures the IllegalStateException is still thrown and logs the close failure.	Low
	Guard against FieldInfo number overflow Assigning synthetic FieldInfos ++maxNumber starting from the max existing field number can collide with numbers Lucene has reserved but that are not present in existing (e.g. gaps left by deletes) or, more importantly, produce field numbers that overlap with global field number allocation semantics used elsewhere. Since these synthetic FieldInfos live only in this wrapper's FieldInfos, collision within existing is what matters — the current loop is safe against that, but consider guarding against Integer.MAX_VALUE overflow when many fields are synthesized. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesLeafReader.java [107-114] FieldInfos existing = in.getFieldInfos(); Map<String, FieldInfo> parquetFields = new LinkedHashMap<>(); List<FieldInfo> combined = new ArrayList<>(); int maxNumber = -1; for (FieldInfo fi : existing) { combined.add(fi); - maxNumber = Math.max(maxNumber, fi.number); + if (fi.number > maxNumber) maxNumber = fi.number; +} +if (maxNumber == Integer.MAX_VALUE) { + throw new IllegalStateException("cannot allocate synthetic FieldInfo numbers: existing max is Integer.MAX_VALUE"); } Suggestion importance[1-10]: 2 __ Why: Guarding against Integer.MAX_VALUE overflow is extremely unlikely to occur in practice, and the suggestion itself notes the current code is safe. Low practical impact.	Low

Suggestions up to commit 46dd573

Category	Suggestion	Impact
Possible issue	Ensure all readers close on IOException ParquetColumnReader.close() is declared to throw IOException, but the catch here only handles RuntimeException, so a checked IOException from one reader will abort the loop and leak the remaining readers. Catch IOException (or Exception) as well to guarantee every reader is closed. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesProducer.java [201-212] public void close() throws IOException { if (closed) { return; } closed = true; synchronized (dedicatedReaders) { for (ParquetColumnReader reader : dedicatedReaders) { try { - // A teardown failure is logged by the reader itself; this only guards the loop - // so one bad reader cannot leave the rest open. reader.close(); - } catch (RuntimeException e) { + } catch (Exception e) { logger.warn("Failed to close Parquet column reader for [{}]", parquetFile, e); } } dedicatedReaders.clear(); } } Suggestion importance[1-10]: 7 __ Why: Valid observation: ParquetColumnReader.close() throws IOException, so catching only RuntimeException could abort the loop and leak resources. Broadening the catch improves robustness during shutdown.	Medium
	Avoid sign extension on encoded half_float valueAt returns long, but this arm yields a short which is implicitly sign-extended to long, producing negative long values for encoded shorts with the high bit set. Since callers (e.g. Lucene sortable-short decode) expect the raw unsigned 16-bit pattern, mask to 16 bits before yielding to avoid sign-extension corrupting the value. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/bridge/DecodedBatch.java [120-125] case KIND_HALF_FLOAT -> { // fp16-width sign-flip; exhaustively verified equal to HalfFloatPoint.halfFloatToSortableShort, // avoiding the lucene-sandbox dependency. short bits = values.getAtIndex(ValueLayout.JAVA_SHORT, idx); - yield (short) (bits ^ ((bits >> 15) & 0x7fff)); + yield (long) ((short) (bits ^ ((bits >> 15) & 0x7fff))) & 0xFFFFL; } Suggestion importance[1-10]: 3 __ Why: The test testHalfFloatSortableShortsMatchLucene shows callers cast back to short, so sign extension is expected and matches Lucene's sortable short representation. The suggested mask would break the sortable ordering by making negative-encoded values appear positive.	Low
General	Prevent unsafe query cache sharing Delegating getReaderCacheHelper to the underlying reader causes query/aggregation cache entries to be keyed on the wrapped reader, so cached results computed against Parquet-resident doc values will be served even after this wrapper is replaced or closed, and cached entries from before the wrapper was installed will be returned to consumers that now see extra fields. Return null (or a wrapper-owned CacheHelper) to prevent unsafe cache sharing across differing field views. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesDirectoryReader.java [49-53] @Override public CacheHelper getReaderCacheHelper() { - // This reader does not change the set of live docs, so it stays cache-coherent with the - // wrapped reader by delegating to its cache helper. - return in.getReaderCacheHelper(); + // Field set differs from the delegate (synthetic Parquet FieldInfos), so results cached + // against the delegate's key must not be reused here. + return null; } Suggestion importance[1-10]: 6 __ Why: Reasonable concern: since the wrapper adds synthetic FieldInfos, sharing the cache key with the delegate could allow cache entries to be reused across differing field views. Worth evaluating, though the comment claims coherence.	Low

Suggestions up to commit 4cfe273

Category	Suggestion	Impact
Possible issue	Fix close/open race that leaks cursors The closed flag is set outside the dedicatedReaders lock in close(), so a race is still possible: close() can set closed=true and drain the list before dedicatedReaderFor enters the synchronized block, then this method sees closed=true and correctly rejects; but if close() sets closed=true after this method's check but before dedicatedReaders.add, the reader is added to a list that was already drained and will leak. Move the closed=true assignment inside the synchronized (dedicatedReaders) block in close() to make the check-and-add atomic with the drain. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesProducer.java [280-292] private ParquetColumnReader dedicatedReaderFor(FieldInfo field) throws IOException { ParquetColumnReader reader = ParquetColumnReader.open(parquetFile, field.getName(), indexSettings, storePointer); - // Register under the same lock close() clears the list under: an open that races a - // concurrent close would otherwise add to an already-drained list and leak the cursor. synchronized (dedicatedReaders) { if (closed) { reader.close(); throw new IllegalStateException("producer for " + parquetFile + " is closed"); } dedicatedReaders.add(reader); } return reader; } Suggestion importance[1-10]: 3 __ Why: The suggestion misreads the code: close() already sets closed=true before entering the synchronized(dedicatedReaders) block, and dedicatedReaderFor checks closed inside the same lock after adding, so the race described is not actually present. The improved_code is essentially identical to the existing code (only a comment removed).	Low
General	Simplify advance and preserve doc state advance scans doc-by-doc calling advanceExact for each, which is O(maxDoc) per advance and defeats the purpose of advance skipping to the next live doc. More importantly, advanceExact(d) for a null row returns false but sets doc = d, so after the loop finds a live doc the intermediate doc field has been mutated repeatedly. Also, when advanceExact returns false at the last iteration, doc may be left as some intermediate value rather than NO_MORE_DOCS when the loop naturally terminates due to the exit-condition being met later; ensure doc is correctly maintained. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/iter/ParquetNumericDocValues.java [76-85] @Override public int advance(int target) throws IOException { for (int d = target; d < maxDoc; d++) { if (advanceExact(d)) { - doc = d; return d; } } doc = NO_MORE_DOCS; return NO_MORE_DOCS; } Suggestion importance[1-10]: 3 __ Why: The suggestion removes a redundant doc = d assignment (since advanceExact already sets doc), which is a minor cleanup. The concern about O(maxDoc) scanning is valid but the alternative would require more substantial changes; the suggestion itself is a minor readability improvement.	Low
	Avoid serializing search slices on assertion Even under assertions, this scans every doc in the segment on the first doc-values request, which for a large segment can be a substantial one-time stall on the first search. Additionally, advanceExact on a SortedNumericDocValues typically requires monotonic ascending targets - iterating 0..maxDoc is fine, but this scan runs while holding the leaf reader's synchronization (via synchronized assertRowIdsAreIdentity), potentially serializing concurrent search slices on the same leaf during the first check. Consider dropping the synchronization on the read path (the memoized fields can use volatile writes or a double-checked pattern), since this is assertions-only. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/ParquetDocValuesLeafReader.java [203-214] private boolean computeRowIdsAreIdentity() throws IOException { SortedNumericDocValues rowId = in.getSortedNumericDocValues(DocumentInput.ROW_ID_FIELD); if (rowId == null) { - return true; // no row-id field => identity by definition + return true; } for (int docId = 0; docId < maxDoc(); docId++) { if (rowId.advanceExact(docId) == false || rowId.nextValue() != docId) { return false; } } return true; } Suggestion importance[1-10]: 3 __ Why: Valid concern about assertion-time serialization/overhead, but since this is assertions-only (dev/test), the impact is limited. The improved_code is essentially identical to existing_code (just a comment removed), not actually addressing the concern raised.	Low
	Make half-float sign extension explicit The yield returns a short that is implicitly widened to long, sign-extending negative values. For a negative sortable-short encoding this yields a large negative long, but downstream code that treats longValue() as an unsigned-short-encoded value could observe unexpected sign extension. Verify this is intended: HalfFloatPoint.halfFloatToSortableShort returns a short, so callers reading it via NumericDocValues.longValue() will see the sign-extended long. This matches the float arm's (long) (int) cast pattern, but the explicit widening to long (as done for KIND_FLOAT via (long)) would make the intent clearer and match existing style. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/docvalues/bridge/DecodedBatch.java [121-127] case KIND_HALF_FLOAT -> { - // Same sign-flip the float/double arms use, at fp16 width. Verified exhaustively to - // equal HalfFloatPoint.halfFloatToSortableShort over every canonical non-NaN fp16 - // bit pattern, so this needs no lucene-sandbox dependency. short bits = values.getAtIndex(ValueLayout.JAVA_SHORT, idx); - yield (short) (bits ^ ((bits >> 15) & 0x7fff)); + yield (long) (short) (bits ^ ((bits >> 15) & 0x7fff)); } Suggestion importance[1-10]: 2 __ Why: A minor stylistic suggestion; the existing implicit widening of short to long produces the same sign-extended result as the explicit cast. Impact is very low.	Low

[github-actions]

✅ Gradle check result for 1f5ddd9: SUCCESS

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 71.85%. Comparing base (9a6d3c3) to head (1757a40).
⚠️ Report is 5 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #23055      +/-   ##
============================================
+ Coverage     71.83%   71.85%   +0.01%     
- Complexity    77801    77826      +25     
============================================
  Files          6173     6173              
  Lines        360563   360578      +15     
  Branches      52479    52483       +4     
============================================
+ Hits         259016   259081      +65     
+ Misses        81049    80927     -122     
- Partials      20498    20570      +72     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[github-actions]

Persistent review updated to latest commit 4cfe273

[github-actions]

Persistent review updated to latest commit 46dd573

[github-actions]

Persistent review updated to latest commit a8b6817

[github-actions]

✅ Gradle check result for a8b6817: SUCCESS

[github-actions]

Persistent review updated to latest commit d06196f

[github-actions]

✅ Gradle check result for d06196f: SUCCESS

[bharath-techie]

everywhere why do we need a long ? i think we just need a version for the file format, we don't need to tie it to a release. A simple integer starting from 1 will work. Take Lucene90DocValuesConsumer for example.

[github-actions]

Persistent review updated to latest commit 14aeb50

[github-actions]

✅ Gradle check result for 14aeb50: SUCCESS

[github-actions]

Persistent review updated to latest commit 1757a40

[github-actions]

✅ Gradle check result for 1757a40: SUCCESS