Add Parquet DocValues read foundation: types + native bridge

[manaslohani]

Description

#22598

Introduce the foundational pieces the Parquet DocValues codec read path builds on, so a Parquet-primary composite index can later serve its columns as Lucene doc values for DSL aggregations.

This change adds only the leaf dependencies; nothing wires them into the query path yet, so there is no functional change at runtime.

Java (parquet-data-format):

• ParquetPhysicalType: enum of Parquet physical types whose code() (0-5) is the discriminant exchanged with the native reader; pinned by test to the ffm.rs TYPE_* constants.
• FieldTypeMapping: OpenSearch mapping type -> (single/multi Lucene DocValuesType, ParquetPhysicalType), with forType/isSupported/validate. Unmapped types throw rather than serve wrong values.

Native (ffm.rs, additive to the write/merge path):

• Column-reader read functions: open/close/count, single-row and repeated reads, page-count, page-index, and whole-page decode into caller buffers with a packed presence bitset. A retained forward cursor keeps an ascending scan from reopening the row group per page.
• Per-file metadata cache (footer + page layouts) shared across readers.

Bridge:

• RustBridge: MethodHandle binds + wrappers for the read functions.
• NativeCall: invokeStatic/invokeIOStatic helpers for allocation-free native invocation with status checking.

Test Coverage

• Unit (this PR):
  • ParquetPhysicalTypeTests — pins each code() (0–5) to the ffm.rs TYPE_*
    constants so the Java<->Rust discriminant contract cannot drift.
  • FieldTypeMappingTests — mapped types resolve to the expected DocValuesType +
    physical type; validate accepts single/multi forms; unmapped types throw.
• Native (ffm.rs): cargo check (with --cfg tokio_unstable, matching the
  build) passes with no errors/warnings. The read-path FFI is exercised end-to-end
  (write-then-read) by ParquetColumnReaderTests, which lands with the column-reader
  change in the next PR (it depends on ParquetColumnReader, not present here).
• Not applicable yet: no query-path wiring in this PR, so there is no
  aggregation/integration coverage to add here; that arrives with the reader-wrapper PR.
• Build: spotlessJavaCheck, compileJava, and compileTestJava all pass on the
  touched modules.

Notes for reviewers

• RustBridge.openColumnReaderCount() is a test-only leak check; its caller
  (ParquetColumnReaderTests) arrives in the next PR.
• Two TODO(consider optimisation) markers are intentional: the second File::open at
  reader open (re-reads the footer already parsed into the cache) and the linear
  locate() scan over row groups (off the aggregation path).

Related Issues

Resolves #[Issue number to be closed when this PR is merged]

#22598

Check List

• [X ] 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 91c6d64)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

📝 TODO sections sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/codec/FieldTypeMapping.java [45]: Several writable OpenSearch types not yet mapped; codec serves no doc values for them. sandbox/plugins/parquet-data-format/src/main/rust/src/ffm.rs [1253]: Reader could be constructed from already-parsed ParquetMetaData to skip second footer read. sandbox/plugins/parquet-data-format/src/main/rust/src/ffm.rs [1279]: Replace linear row-group scan in `locate` with binary search like `page_for_row`.

🔀 No multiple PR themes

⚡ Recommended focus areas for review Possible Issue parquet_open_column_reader_count is declared pub unsafe extern "C" but is NOT annotated with #[ffm_safe] (unlike its neighbors). Any panic inside (e.g. poisoned.into_inner() behavior aside, or future changes) would unwind across the FFI boundary, which is UB. Given the comment says it "never errors; recovers from a poisoned mutex", the #[ffm_safe] guard should still be applied for consistency and panic isolation. #[no_mangle] pub unsafe extern "C" fn parquet_open_column_reader_count() -> i64 { match COLUMN_READERS.lock() { Ok(guard) => guard.len() as i64, Err(poisoned) => poisoned.into_inner().len() as i64, } } Possible Issue In expand_to_outbuf, the null-path zeroes num_rows * 8 bytes via write_bytes and then scatters using presence_bits. However, if the packed presence has more set bits than dense.len() (e.g. due to a decoder mismatch), dense.get_unchecked(di) will read out of bounds. Consider a debug_assert! on the popcount of presence bits vs dense.len(), or bounds-check di < dense.len() in the loop, to catch decoder/presence mismatches instead of silently reading OOB. unsafe fn expand_to_outbuf<T: Copy>( dense: &[T], to_bits: impl Fn(T) -> i64, null_count: i64, num_rows: usize, out: *mut u8, presence_bits: *const i64, ) { let out_i64 = out as *mut i64; if null_count == 0 { // All rows present — tight conversion loop, no branching, SIMD-friendly. for i in 0..num_rows { *out_i64.add(i) = to_bits(*dense.get_unchecked(i)); } } else { // Scatter using the packed presence bits. Read one word at a time and use // trailing_zeros to jump to set bits (pop-and-scatter pattern). // First zero-fill so null slots hold 0 without an explicit branch. std::ptr::write_bytes(out, 0, num_rows * 8); let mut di = 0usize; let words = (num_rows + 63) / 64; for w in 0..words { let mut bits = *presence_bits.add(w) as u64; while bits != 0 { let b = bits.trailing_zeros() as usize; let row = w * 64 + b; *out_i64.add(row) = to_bits(*dense.get_unchecked(di)); di += 1; bits &= bits - 1; // clear lowest set bit } } } } Cache Growth FILE_METADATA_CACHE is a Mutex<HashMap<String, Arc<...>>> with no size bound and eviction only via explicit parquet_evict_file_metadata. Over a long-running node with many segment files this can grow unboundedly (each entry holds schema descriptor + per-column page layouts). If callers don't reliably invoke evict on shard close / file delete, this is a slow memory leak. Consider a size cap with LRU eviction or documenting the strict eviction requirement. lazy_static! { /// Node-level file metadata cache. Keyed by absolute file path. Entries are never /// invalidated (immutable files) — evicted only on explicit `parquet_evict_file_metadata`. static ref FILE_METADATA_CACHE: Mutex<HashMap<String, std::sync::Arc<FileMetadataCache>>> = Mutex::new(HashMap::new()); /// Per-handle registry of open column readers, keyed by an opaque i64 handle. /// Mirrors the writer-side handle pattern; serialised behind a single mutex /// since column readers are not shared across threads. static ref COLUMN_READERS: Mutex<HashMap<i64, ColumnReaderState>> = Mutex::new(HashMap::new()); } Contention COLUMN_READERS is a single global Mutex<HashMap<i64, ColumnReaderState>>, and every read/decode call takes lock_readers() for the entire operation (including the actual Parquet decode in parquet_decode_page_at_row, which can be expensive). This serializes all column-reader operations across the node onto one mutex, defeating parallel per-segment/per-shard scans. Consider sharding the registry or holding the handle-to-state lookup briefly and moving decoding outside the map lock (e.g., via Arc<Mutex<ColumnReaderState>> per handle). let mut guard = lock_readers()?; let state = guard .get_mut(&handle) .ok_or_else(|| format!("parquet_decode_page_at_row: unknown handle {}", handle))?;

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 91c6d64
Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Protect FFI boundary from panics This exported extern "C" function lacks the #[ffm_safe] wrapper other exports use, meaning a panic inside HashMap::len or the mutex handling would unwind across the FFI boundary and cause undefined behavior. Wrap the body in catch_unwind or apply #[ffm_safe] even though this function "never errors". sandbox/plugins/parquet-data-format/src/main/rust/src/ffm.rs [1576-1582] -/// Debug-only symbol: returns the number of currently open column-reader -/// handles. Used by Property 7 (native handle non-leakage). Never errors; -/// recovers from a poisoned mutex rather than panicking. +#[ffm_safe] #[no_mangle] pub unsafe extern "C" fn parquet_open_column_reader_count() -> i64 { - match COLUMN_READERS.lock() { + let count = match COLUMN_READERS.lock() { Ok(guard) => guard.len() as i64, Err(poisoned) => poisoned.into_inner().len() as i64, - } + }; + Ok(count) } Suggestion importance[1-10]: 5 __ Why: Valid concern about panics unwinding across the FFI boundary. Applying #[ffm_safe] for consistency with other exports would prevent UB in the unlikely case of a panic, though HashMap::len and the shown lock handling are unlikely to panic in practice.	Low
	Replace debug_assert with runtime check The precondition def_levels.len() >= num_rows is only checked via debug_assert!, but the optional-column branch uses get_unchecked in release builds. If a corrupt page or decoder-state bug produces fewer def_levels than num_rows, this reads out-of-bounds in production. Promote this to a runtime check returning an error, since an incorrect page decode is not a fundamental invariant of this helper's callers but of parquet library behavior. sandbox/plugins/parquet-data-format/src/main/rust/src/ffm.rs [2044-2065] #[inline] unsafe fn pack_presence_from_def_levels( def_levels: &[i16], max_def_level: i16, num_rows: usize, out: *mut i64, -) { - // The optional-column branch reads def_levels[0..num_rows] unchecked; assert the precondition - // so a corrupt page index or decoder state fails loudly in tests rather than reading OOB. - debug_assert!(def_levels.len() >= num_rows || max_def_level == 0); +) -> Result<(), String> { + if max_def_level != 0 && def_levels.len() < num_rows { + return Err(format!( + "pack_presence_from_def_levels: def_levels len {} < num_rows {}", + def_levels.len(), num_rows + )); + } let words = (num_rows + 63) / 64; if max_def_level == 0 { - // Required column: every row is present → all-ones, mask tail. for w in 0..words { let remaining = num_rows - w * 64; if remaining >= 64 { - *out.add(w) = -1i64; // all bits set + *out.add(w) = -1i64; } else { *out.add(w) = ((1u64 << remaining) - 1) as i64; } } } else { Suggestion importance[1-10]: 5 __ Why: Reasonable defensive suggestion since get_unchecked in release builds could read OOB if the parquet decoder produced fewer def_levels than expected. However, changing the signature to return Result requires updating all callers, and the invariant is well-controlled by the decoder contract. tokens: 502 </details></details></td><td align=center>Low </td></tr><tr><td> <details><summary>Guard against null offsets buffer</summary> ___ **<code>write_bytes_repeated</code> writes <code>out_byte_offsets[0..=count]</code> (count+1 entries) but only <br>checks that <code>count <= out_long_cap</code>. The Java side declares <code>out_byte_offsets</code> with <br>capacity <code>out_long_cap + 1</code>, but this is not asserted or documented as invariant in <br>the Rust code, and a mismatch would corrupt memory. Add an explicit guard rejecting <br><code>out_byte_offsets</code> being null when <code>count > 0</code>, and consider treating offsets capacity <br>as <code>out_long_cap + 1</code> inputs from the caller for defense-in-depth.** [sandbox/plugins/parquet-data-format/src/main/rust/src/ffm.rs [1834-1855]](https://github.com/opensearch-project/OpenSearch/pull/22599/files#diff-6cc1a9b5887afe1678e9b9ea440ca83baf0b92c1c36f6a3be50bed3ed74d916aR1834-R1855) ```diff unsafe fn write_bytes_repeated( slices: &[&[u8]], out_count: *mut i64, out_long_cap: i64, out_byte_buf: *mut u8, out_byte_offsets: *mut i64, out_byte_buf_cap: i64, ) -> Result<i64, String> { let count = slices.len(); let total_bytes: usize = slices.iter().map(|s| s.len()).sum(); if !out_count.is_null() { *out_count = count as i64; } - // Element-count overflow: cannot safely write offsets (capacity is count+1). if (count as i64) > out_long_cap { return Ok(RC_OVERFLOW); } + if count > 0 && out_byte_offsets.is_null() { + return Ok(RC_OVERFLOW); + } - // Element count fits: write the full CSR offsets so that, even on a byte - // overflow, out_byte_offsets[count] == total_bytes reports the required size. if !out_byte_offsets.is_null() { Suggestion importance[1-10]: 4 __ Why: The suggestion adds a defensive null check for out_byte_offsets, which is reasonable defense-in-depth but the existing code already guards writes with !out_byte_offsets.is_null(). However, the subsequent write at line 1861 (*out_byte_offsets.add(count) = acc) is only inside the null-check block, so the null check is already handled. Minor improvement.	Low
	Reject null required out-parameters explicitly The overflow check treats a null out pointer as an overflow condition, but on the RC_OVERFLOW path the caller relies on out_value_actual_len, out_first_row, and out_last_row to size buffers. If any of those output pointers are null, callers cannot retry correctly. Validate that all required out-parameter pointers are non-null before touching them, and return an error (not overflow) when required outputs are null, to prevent silent misuse. sandbox/plugins/parquet-data-format/src/main/rust/src/ffm.rs [2218-2244] -if !out_first_row.is_null() { - *out_first_row = first_global; +if out_value_actual_len.is_null() || out_first_row.is_null() || out_last_row.is_null() { + return Err("parquet_decode_page_at_row: required out-parameter is null".to_string()); } -if !out_last_row.is_null() { - *out_last_row = first_global + num_rows - 1; -} +*out_first_row = first_global; +*out_last_row = first_global + num_rows - 1; -// For primitive types the value byte length is known before decode (num_rows * 8), so check -// capacity up front and write straight into the out-buffers. BYTE_ARRAY defers its capacity -// check to after decode, since its total byte length is data-dependent. let is_primitive = physical_type != PhysicalType::BYTE_ARRAY && physical_type != PhysicalType::FIXED_LEN_BYTE_ARRAY; if is_primitive { let value_bytes = (num_rows_usize * 8) as i64; - if !out_value_actual_len.is_null() { - *out_value_actual_len = value_bytes; - } + *out_value_actual_len = value_bytes; let presence_words = ((num_rows_usize + 63) / 64) as i64; if value_bytes > out_value_buf_cap || out_value_buf.is_null() || presence_words > out_presence_bits_cap || out_presence_bitset.is_null() { return Ok(RC_OVERFLOW); } } Suggestion importance[1-10]: 3 __ Why: The existing code defensively checks for null pointers before dereferencing, which is safe. Rejecting null out-parameters as errors is a defensible API choice but is not strictly necessary for correctness, and the caller (ParquetColumnReader) already always passes non-null out-parameters.	Low

---

Previous suggestions

Suggestions up to commit 3e0079d

Category	Suggestion	Impact
Possible issue	Prevent i32 offset accumulator overflow The CSR offset accumulator is i32, so acc += s.len() as i32 will silently wrap when the total bytes on a page exceed i32::MAX (~2 GiB), producing corrupt offsets and out-of-bounds writes into out_value_buf. Use a checked/i64 accumulator and validate against i32::MAX before storing, returning an error for pages that overflow the offset width. sandbox/plugins/parquet-data-format/src/main/rust/src/ffm.rs [2436-2444] -let mut acc: i32 = 0; +let mut acc: i64 = 0; for (i, s) in slices.iter().enumerate() { - *out_byte_offsets.add(i) = acc; + if acc > i32::MAX as i64 { + return Err(format!("byte page offset {} exceeds i32::MAX", acc)); + } + *out_byte_offsets.add(i) = acc as i32; if !s.is_empty() { std::ptr::copy_nonoverlapping(s.as_ptr(), out_value_buf.add(acc as usize), s.len()); } - acc += s.len() as i32; + acc += s.len() as i64; } -*out_byte_offsets.add(slices.len()) = acc; +if acc > i32::MAX as i64 { + return Err(format!("byte page total {} exceeds i32::MAX", acc)); +} +*out_byte_offsets.add(slices.len()) = acc as i32; Suggestion importance[1-10]: 7 __ Why: Correctly identifies that acc: i32 can silently overflow for pages > 2 GiB, leading to corrupt offsets and OOB writes. Detecting the overflow explicitly improves correctness/safety at the FFI boundary.	Medium
General	Only retain cursor on successful decode On RC_OVERFLOW the byte-path advances the cursor to next_position and reinstalls it, but the primitive path does NOT advance the cursor on overflow (it early-returns before any decode call). This asymmetry means a byte-column overflow-retry needlessly discards the cursor and reopens the row group, defeating the retained-cursor optimization exactly when the caller is re-issuing the same row. Consider only installing the cursor on RC_OK, or documenting that the retry path is expected to open a fresh reader. sandbox/plugins/parquet-data-format/src/main/rust/src/ffm.rs [2323-2344] ColumnReader::ByteArrayColumnReader(mut r) => { let rc = decode_byte_page( &mut r, skip, num_rows_usize, max_def_level, &mut state.scratch.def_levels, out_value_buf, out_value_buf_cap, out_value_actual_len, out_byte_offsets, out_byte_offsets_cap, out_presence_bitset, out_presence_bits_cap, )?; - // The page was fully consumed even on RC_OVERFLOW, so the cursor position is - // next_position either way; the overflow retry of the same row then simply - // misses the cursor (position > first_global) and opens a fresh reader. - state.cursor = Some(CursorState { rg_idx, col_reader: ColumnReader::ByteArrayColumnReader(r), position: next_position }); + if rc == RC_OK { + state.cursor = Some(CursorState { rg_idx, col_reader: ColumnReader::ByteArrayColumnReader(r), position: next_position }); + } return Ok(rc); } Suggestion importance[1-10]: 5 __ Why: Valid observation about cursor asymmetry between byte and primitive paths on overflow; installing cursor only on RC_OK matches the primitive path's behavior and could improve retry performance, though the existing comment acknowledges this trade-off.	Low
	Assert input length before unchecked access def_levels.get_unchecked(b) skips bounds checks; if num_rows is ever passed larger than def_levels.len() (e.g. due to a page-index/records mismatch) this becomes UB. The function is unsafe so this is a soundness precondition rather than a bug per se — but adding a debug assertion (or a checked bounds check at function entry: assert!(def_levels.len() >= num_rows)) would catch a corrupt page-index / decoder state early rather than silently reading OOB memory. sandbox/plugins/parquet-data-format/src/main/rust/src/ffm.rs [2068-2070] +debug_assert!(def_levels.len() >= num_rows); for b in base..end { bits |= ((*def_levels.get_unchecked(b) == max_def_level) as u64) << (b - base); } Suggestion importance[1-10]: 3 __ Why: Adding a debug_assert is a minor defensive improvement for an internal invariant, but the function is already unsafe with documented preconditions, so impact is low.	Low
	Guard against null output pointers The overflow check compares n (page count) against out_buf_capacity, but the caller may pass null pointers for arrays it doesn't care about. Any non-null array pointer with capacity >= n is then written to via .add(i) without a null check, which is fine, but there is no null-safety check like other functions in the file. Add null checks per-array (mirroring the pattern used in parquet_read_value_at_row) to avoid dereferencing null when a caller opts out of some outputs. sandbox/plugins/parquet-data-format/src/main/rust/src/ffm.rs [1945-1964] +if (n as i64) > out_buf_capacity { + return Ok(RC_OVERFLOW); +} +for (i, e) in state.pages.iter().enumerate() { + if !out_first_row.is_null() { + *out_first_row.add(i) = e.global_first_row; Suggestion importance[1-10]: 2 __ Why: The existing code already has null checks for each output pointer before writing (e.g. if !out_first_row.is_null()), and the existing_code is identical to the improved_code. The suggestion adds no value.	Low

[github-actions]

✅ Gradle check result for 3e0079d: SUCCESS

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 71.44%. Comparing base (d19f68e) to head (91c6d64).
⚠️ Report is 8 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff            @@
##               main   #22599   +/-   ##
=========================================
  Coverage     71.43%   71.44%           
+ Complexity    76760    76746   -14     
=========================================
  Files          6142     6142           
  Lines        357766   357794   +28     
  Branches      52148    52162   +14     
=========================================
+ Hits         255581   255631   +50     
+ Misses        81861    81801   -60     
- Partials      20324    20362   +38     

☔ 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.

[aggarwalmayank]

In Parquet's own type taxonomy, BYTE_ARRAY is a primitive type, should we call it IsFixedWidthPrimitive?

[aggarwalmayank]

why do we need static methods? why can't we use non static versions of these which are already there?

[github-actions]

Persistent review updated to latest commit 91c6d64

[github-actions]

✅ Gradle check result for 91c6d64: SUCCESS