Eliminate FFM round-trips via nextDoc piggyback on collectDocs

[aasom143]

collectDocs now returns packed i64: upper 32 bits = next matching docId beyond maxDoc, lower 32 bits = wordsWritten. No new FFM calls needed.

Rust consumers use nextDoc for three levels of optimization:

1. Full RG skip: nextDoc >= rgMax → no collectDocs call
2. Tightened range: nextDoc > rgMin → start from nextDoc, smaller bitset
3. No benefit: nextDoc <= rgMin → call as before

Implemented in both SingleCollectorEvaluator (AtomicI32 across RGs) and CollectorLeafBitmaps/BitmapTree (per-leaf HashMap keyed by Arc identity).

Description

[Describe what this change achieves]

Related Issues

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

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 edb60a7)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Incorrect nextDoc semantics nextDoc is computed as Math.max(handle.currentDoc, maxDoc), which forces the returned nextDoc to be at least maxDoc even when the underlying iterator's currentDoc is inside the range (e.g. equal to maxDoc-1 or less if the scanTo was clamped by partitionMaxDoc). The Rust side uses next_doc > call_max to skip and last_next.max(*r_min) to tighten; if currentDoc < maxDoc (which happens when scanTo = min(maxDoc, partitionMaxDoc) < maxDoc), lying and returning maxDoc as nextDoc could cause the next RG's effective_min to be raised past valid matches or cause an unwarranted skip. The nextDoc should reflect the actual next matching docId (which is handle.currentDoc when iteration stopped past scanTo), not artificially clamped to maxDoc. nextDoc = (handle.currentDoc == DocIdSetIterator.NO_MORE_DOCS) ? Integer.MAX_VALUE : Math.max(handle.currentDoc, maxDoc); } catch (IOException exception) { LOGGER.warn("IOException during collectDocs, returning partial bitset", exception); } } else { nextDoc = (handle.currentDoc == DocIdSetIterator.NO_MORE_DOCS) ? Integer.MAX_VALUE : Math.max(handle.currentDoc, maxDoc); } Race condition on last_next_doc last_next_doc is an AtomicI32 shared across RGs, but the load-then-store pattern (load at start of prefetch_rg, use last_next locally, store at end) is not atomic. When RGs are processed concurrently (max_collector_parallelism > 1 for SingleCollectorEvaluator), two threads can load the same stale value, each perform their own collectDocs at the same effective_min, and race on the store — losing skip/tighten benefits and potentially causing one thread to overwrite a higher nextDoc from a later RG with a lower one from an earlier RG. Also, since RGs may be processed out of order, storing unconditionally can regress last_next_doc. Consider ordering guarantees or a fetch_max for the store. // Skip RG if the previous collectDocs told us the next match is beyond this RG. let mut last_next = self .last_next_doc .load(std::sync::atomic::Ordering::Acquire); if last_next > max_doc { native_bridge_common::log_debug!( "SingleCollector: skipping RG {} — nextDoc={} > maxDoc={}", rg.index, last_next, max_doc ); return Ok(None); } // Page-prune to discover which row ranges survive. let page_ranges: Option<Vec<(i32, i32)>> = self.pruning_predicate.as_ref().and_then(|pp| { self.page_pruner .prune_rg(pp, rg.index, self.page_prune_metrics.as_ref()) .map(|sel| { let mut ranges = Vec::new(); let mut rg_pos: i64 = 0; for s in sel.iter() { if s.skip { rg_pos += s.row_count as i64; } else { let abs_min = min_doc + rg_pos as i32; let abs_max = min_doc + rg_pos as i32 + s.row_count as i32; ranges.push((abs_min, abs_max)); rg_pos += s.row_count as i64; } } ranges }) }); // All pages pruned by stats → skip bloom + collector entirely. if let Some(ref ranges) = page_ranges { if ranges.is_empty() { return Ok(None); } } // Bloom filter pruning: runs after page pruning (free) but before // the expensive FFM collector call. Uses the IO runtime handle from // the RuntimeManager to drive the async object-store read. if let (Some(bloom), Some(pp)) = (&self.bloom_config, &self.pruning_predicate) { let _timer = bloom.bloom_filter_eval_time.as_ref().map(|t| t.timer()); let pruned = bloom .io_handle .block_on(crate::indexed_table::bloom_pruner::bloom_prune_rg( &*bloom.store, &bloom.object_path, &bloom.metadata, &bloom.arrow_schema, rg.index, pp.as_ref(), )); if pruned { if let Some(ref c) = bloom.rg_bloom_pruned { c.add(1); } return Ok(None); } } // Build candidates either from the always-call correctness collector OR, when // the query is performance-only (no Collector leaves), from the page-pruned // universe. Performance leaves are AND'd in below if the selectivity gate fires. let mut candidates = match self.collector.as_ref() { Some(collector) => { // Dispatch collector call strategy. let call_ranges: Vec<(i32, i32)> = match self.call_strategy { CollectorCallStrategy::FullRange => vec![(min_doc, max_doc)], CollectorCallStrategy::TightenOuterBounds => match &page_ranges { Some(r) if r.is_empty() => return Ok(None), Some(r) => vec![(r.first().unwrap().0, r.last().unwrap().1)], None => vec![(min_doc, max_doc)], }, CollectorCallStrategy::PageRangeSplit => match &page_ranges { Some(r) if r.is_empty() => return Ok(None), Some(r) => r.clone(), None => vec![(min_doc, max_doc)], }, }; // Call collector for each range, merge into one RG-relative bitmap. let mut bm = RoaringBitmap::new(); for (r_min, r_max) in &call_ranges { if last_next > *r_max { continue; } let effective_min = last_next.max(*r_min); let result = collector .collect_packed_u64_bitset(effective_min, *r_max) .map_err(|e| { format!( "collector.collect_packed_u64_bitset(rg={}, [{}, {})): {}", rg.index, r_min, r_max, e ) })?; if let Some(ref c) = self.ffm_collector_calls { c.add(1); } last_next = result.next_doc; let offset = (effective_min as i64 - rg.first_row) as u32; let num_docs = (*r_max - effective_min) as u32; let bytes: &[u8] = unsafe { std::slice::from_raw_parts( result.words.as_ptr() as *const u8, result.words.len() * 8, ) }; let mut chunk = RoaringBitmap::from_lsb0_bytes(offset, bytes); let upper = offset.saturating_add(num_docs); if upper < u32::MAX { chunk.remove_range(upper..); } bm |= chunk; } self.last_next_doc .store(last_next, std::sync::atomic::Ordering::Release); next_doc skip ignores call ranges The early-skip check if last_next > max_doc { return Ok(None); } at the RG level bypasses the entire RG based on max_doc, but when CollectorCallStrategy is PageRangeSplit or TightenOuterBounds, the actual collector calls occur over sub-ranges (page_ranges) with their own bounds. The RG-level skip is fine (nextDoc past max_doc means no matches in the RG), but the correctness depends on last_next being the true next match. If prior code path stored an inflated nextDoc (see Lucene side), a legitimate RG can be entirely skipped. Verify this invariant holds under all collector implementations. // Skip RG if the previous collectDocs told us the next match is beyond this RG. let mut last_next = self .last_next_doc .load(std::sync::atomic::Ordering::Acquire); if last_next > max_doc { native_bridge_common::log_debug!( "SingleCollector: skipping RG {} — nextDoc={} > maxDoc={}", rg.index, last_next, max_doc ); return Ok(None); } Leaf identity via Arc pointer leaf_key = Arc::as_ptr(collector) as usize uses pointer identity to key leaf_next_doc. If the same logical leaf produces different Arc allocations across calls (e.g., via Arc::clone vs. rebuilding the resolved tree per RG), the map will grow unbounded and skip/tighten state will not be reused. Additionally, if an Arc is dropped and its address reused by a different collector, state could be misattributed. Confirm collector Arcs are stable across the RG-iteration lifetime of this CollectorLeafBitmaps. let leaf_key = Arc::as_ptr(collector) as *const () as usize; let mut last_next_doc = { let map = self.leaf_next_doc.lock().unwrap(); map.get(&leaf_key).copied().unwrap_or(i32::MIN) };

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to edb60a7

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible bug	Avoid pointer-identity keys for leaves Using an Arc pointer as a stable identity key is unsafe: once an Arc is dropped its allocation can be reused by an unrelated collector, causing spurious next_doc reuse across leaves and potentially skipping valid RGs. Prefer a stable identifier stored on the collector/node (e.g., an annotation_id, provider key, or leaf index) or key on the DFS leaf index that's already passed in. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/eval/bitmap_tree.rs [1020] -let leaf_key = Arc::as_ptr(collector) as *const () as usize; +let leaf_key = _leaf_dfs_index; Suggestion importance[1-10]: 8 __ Why: Legitimate correctness concern: using an Arc raw pointer as a stable identity key is fragile because allocations can be reused after drop, potentially causing incorrect next_doc reuse across unrelated leaves. Using a stable identifier like _leaf_dfs_index is safer.	Medium
	Report accurate nextDoc without clamping Clamping nextDoc to Math.max(handle.currentDoc, maxDoc) is incorrect: if the iterator already advanced past scanTo but still lies within [minDoc, maxDoc) (e.g. scanTo < maxDoc due to partitionMaxDoc), the true next matching doc can be less than maxDoc. Return handle.currentDoc directly (only replacing with MAX_VALUE when exhausted) so callers get an accurate skip hint and don't miss subsequent matches. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneFilterDelegationHandle.java [245-247] nextDoc = (handle.currentDoc == DocIdSetIterator.NO_MORE_DOCS) ? Integer.MAX_VALUE - : Math.max(handle.currentDoc, maxDoc); + : handle.currentDoc; Suggestion importance[1-10]: 7 __ Why: Valid concern: clamping nextDoc to Math.max(handle.currentDoc, maxDoc) can misreport the actual next matching doc when the iterator advanced within the range but stopped before maxDoc (e.g., due to partitionMaxDoc). This could cause incorrect skip decisions downstream.	Medium
	Guard against concurrent nextDoc races SingleCollectorEvaluator stores a single last_next_doc per evaluator, but prefetch_rg can be invoked concurrently from multiple partitions/threads on the same evaluator instance. Interleaved reads/writes of this atomic can cause one partition's next_doc from a later RG to incorrectly skip an earlier RG on another partition. Consider keying last_next_doc per-partition or per-thread, or documenting/enforcing single-threaded use. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/eval/single_collector.rs [292-295] let mut last_next = self .last_next_doc .load(std::sync::atomic::Ordering::Acquire); +// NOTE: assumes prefetch_rg is called serially per evaluator instance. if last_next > max_doc { Suggestion importance[1-10]: 7 __ Why: Valid concern about thread-safety: a single AtomicI32 shared across partitions can cause cross-partition interference, potentially skipping RGs incorrectly. Worth investigating whether partitions share the evaluator instance.	Medium
General	Decode packed nextDoc more defensively packed is i64; the sign-check packed < 0 only rules out the whole value being negative, but a legitimate nextDoc = Integer.MAX_VALUE with a valid wordsWritten results in a positive i64, while any nextDoc with the high-bit set (unlikely but possible if future callers ever encode differently) would be treated as error. More importantly, cast the low-32 correctly and consider validating next_doc >= min_doc to catch callback bugs early. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/ffm_callbacks.rs [243-245] let packed_u = packed as u64; let words_written = (packed_u & 0xFFFF_FFFF) as usize; -let next_doc = (packed_u >> 32) as i32; +let next_doc = ((packed_u >> 32) & 0xFFFF_FFFF) as u32 as i32; Suggestion importance[1-10]: 3 __ Why: The current decoding (packed_u >> 32) as i32 already correctly extracts the upper 32 bits; the suggested alternative is functionally equivalent. The remark about additional validation is minor.	Low

---

Previous suggestions

Suggestions up to commit 772f4f3

Category	Suggestion	Impact
Possible issue	Use exact sentinel for error check The packed < 0 error check is unreliable because valid packed values with nextDoc >= 0x80000000 (e.g., Integer.MAX_VALUE in Java is 0x7FFFFFFF, but any negative-signed high-bit interpretation) may not trigger it — however Java returns -1L on error, which as i64 is 0xFFFF_FFFF_FFFF_FFFF, and as packed this would decode to next_doc = -1, words = MAX. The check packed < 0 correctly catches -1L, but a valid packed value where the upper 32 bits form a negative i32 (any nextDoc with bit 31 set) would also make packed negative. Explicitly compare to -1 to distinguish the error sentinel from valid packed data. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/ffm_callbacks.rs [237-245] -if packed < 0 { +if packed == -1 { return Err(format!( "collectDocs(context_id={}, key={}) failed: {}", self.context_id, self.key, packed )); } let packed_u = packed as u64; let words_written = (packed_u & 0xFFFF_FFFF) as usize; let next_doc = (packed_u >> 32) as i32; Suggestion importance[1-10]: 7 __ Why: Valid concern: a valid packed value with nextDoc having bit 31 set (negative i32) would make the entire i64 negative and be misinterpreted as an error. However, in practice nextDoc values are non-negative doc IDs bounded by Integer.MAX_VALUE, so the risk is limited but the fix improves robustness.	Medium
	Return actual next matching docId Using Math.max(handle.currentDoc, maxDoc) masks the real next-matching docId when currentDoc < maxDoc (e.g., when the scanner stopped due to scanTo < maxDoc because of partition bounds). This can cause callers to under-skip. Return handle.currentDoc directly (it is already >= scanTo after the loop when not exhausted, and is the true next matching doc otherwise). sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneFilterDelegationHandle.java [245-247] nextDoc = (handle.currentDoc == DocIdSetIterator.NO_MORE_DOCS) ? Integer.MAX_VALUE - : Math.max(handle.currentDoc, maxDoc); + : handle.currentDoc; Suggestion importance[1-10]: 6 __ Why: Valid observation: Math.max(handle.currentDoc, maxDoc) inflates nextDoc when the scanner stops early due to partition bounds (scanTo < maxDoc), potentially causing incorrect skip decisions. However, the impact depends on whether callers rely on the exact next-doc semantics for correctness or just for skip optimization.	Low
General	Fix off-by-one in skip condition **The skip condition last_next_doc > *call_max combined with effective_min = last_next_doc.max(call_min) can produce effective_min == call_max (empty range) when last_next_doc == call_max, which is wasteful. Also, if last_next_doc >= call_max the range is empty and should be skipped. Use >= to match the doc-file semantics ("next matching docId beyond maxDoc" — an exclusive upper bound). sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/eval/bitmap_tree.rs [1034-1038] -if last_next_doc > *call_max { +if last_next_doc >= *call_max { continue; } let effective_min = last_next_doc.max(*call_min); let result = collector.collect_packed_u64_bitset(effective_min, *call_max)?; Suggestion importance[1-10]: 5 __ Why: Reasonable minor optimization: when last_next_doc == call_max, the range [call_max, call_max) is empty and calling the collector is wasteful. Using >= avoids the unnecessary call, though it's a boundary optimization rather than a correctness bug.	Low
	Use inclusive comparison for skip The same off-by-one issue exists here: next_doc is documented as "next matching docId beyond max_doc", so when last_next == max_doc there is still no match inside [min_doc, max_doc) and the RG can be skipped. Use >= for consistency with the FFI contract and to avoid an unnecessary FFM call in the boundary case. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/eval/single_collector.rs [292-295] let mut last_next = self .last_next_doc .load(std::sync::atomic::Ordering::Acquire); -if last_next > max_doc { +if last_next >= max_doc { Suggestion importance[1-10]: 5 __ Why: Same boundary optimization for consistency: when last_next == max_doc, no match can exist in [min_doc, max_doc), so >= is more precise. Minor efficiency improvement.	Low

Suggestions up to commit f2ced7a

Category	Suggestion	Impact
Possible issue	Avoid over-skipping on partial IOException When an IOException occurs during scanning, nextDoc remains at Integer.MAX_VALUE (its initial value), which would cause callers to incorrectly skip all subsequent RGs even though the scan was only partial. Set nextDoc to maxDoc in the catch block so callers retry the next range rather than treating the iterator as exhausted. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneFilterDelegationHandle.java [245-252] nextDoc = (handle.currentDoc == DocIdSetIterator.NO_MORE_DOCS) ? Integer.MAX_VALUE : Math.max(handle.currentDoc, maxDoc); } catch (IOException exception) { LOGGER.warn("IOException during collectDocs, returning partial bitset", exception); + nextDoc = maxDoc; } } else { nextDoc = (handle.currentDoc == DocIdSetIterator.NO_MORE_DOCS) ? Integer.MAX_VALUE : Math.max(handle.currentDoc, maxDoc); } Suggestion importance[1-10]: 8 __ Why: Valid correctness concern: on IOException, nextDoc stays at Integer.MAX_VALUE, causing subsequent RGs to be incorrectly skipped after only a partial scan. Setting nextDoc = maxDoc in the catch is a meaningful bug fix.	Medium
General	Safer default for next_doc field *Defaulting next_doc to i32::MIN in the From impl is dangerous: any real caller that used .into() (e.g., a stub returning Vec::new().into()) would produce a next_doc that, per the trait contract, is meant to be "next matching doc beyond max_doc". Callers using last_next.max(r_min) are safe, but a stale i32::MIN cached across RGs can also mask logic bugs. Consider defaulting to i32::MAX (exhausted) or making the field mandatory to prevent silently-wrong values from test-shim .into() conversions leaking behavior. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/index.rs [58-65] impl From<Vec<u64>> for CollectDocsResult { fn from(words: Vec<u64>) -> Self { CollectDocsResult { words, - next_doc: i32::MIN, + next_doc: i32::MAX, } } } Suggestion importance[1-10]: 6 __ Why: Using i32::MIN as default causes last_next.max(*r_min) to behave like no skip info, which is intentional. Changing to i32::MAX would break the "unknown" semantics and cause all RGs to be skipped in test stubs, so the suggestion is partly misguided but raises a valid clarity concern.	Low
	Fix off-by-one in skip comparison Using > here is inconsistent with the RG-level skip check above which uses if last_next > max_doc. Since next_doc is defined as the next matching doc beyond max_doc (i.e. next_doc >= max_doc means no matches in [..max_doc)), the comparison should be >= to correctly skip a range when last_next == r_max (no matches possible in [..r_max)). Same inconsistency exists in bitmap_tree.rs. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/eval/single_collector.rs [380-383] for (r_min, r_max) in &call_ranges { - if last_next > *r_max { + if last_next >= *r_max { continue; } let effective_min = last_next.max(*r_min); Suggestion importance[1-10]: 5 __ Why: Reasonable consistency point regarding > vs >= for skip check, though the exact semantic (next_doc is beyond maxDoc, could equal maxDoc) makes correctness debatable; the RG-level check uses > too, so calling it inconsistent is not quite right.	Low
	Guard against sentinel/packed collision The -1 error sentinel is checked as packed < 0, but packed_u = packed as u64 treats the same value as unsigned. If the Java side ever returns a legitimately packed value whose high bit is set (e.g., nextDoc encoded such that the whole long is negative when interpreted as i64), it will be misclassified as an error. Since next_doc is up to Integer.MAX_VALUE, the upper 32 bits are non-negative, so today it works — but explicitly documenting or asserting next_doc >= 0 after unpacking would prevent silent regressions. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/ffm_callbacks.rs [243-245] let packed_u = packed as u64; let words_written = (packed_u & 0xFFFF_FFFF) as usize; let next_doc = (packed_u >> 32) as i32; + debug_assert!(next_doc >= 0, "next_doc must be non-negative; got {}", next_doc); Suggestion importance[1-10]: 4 __ Why: Adding a debug assertion for next_doc >= 0 is a minor defensive improvement; the current code is already correct given the Java contract.	Low

Suggestions up to commit 2fb7234

Category	Suggestion	Impact
Possible issue	Document/guard cross-partition ordering assumption Concurrent partitions calling prefetch_rg share the same AtomicI32 without synchronization: two threads can each load the same last_next, then race on the final store, silently losing skip information or (worse) reordering RGs. Additionally, if RGs are processed out-of-order across partitions, a stale next_doc from a later RG could cause an earlier RG to be incorrectly skipped. Either document the single-threaded contract, gate the skip on ordered RG processing, or protect the state per-RG-sequence. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/eval/single_collector.rs [292-295] let mut last_next = self .last_next_doc .load(std::sync::atomic::Ordering::Acquire); - if last_next > max_doc { + // NOTE: assumes RGs are processed in ascending order on a single partition. + // Callers must not share this evaluator across partitions concurrently. + if last_next >= max_doc { Suggestion importance[1-10]: 7 __ Why: Valid concern about concurrent access to the AtomicI32 state — races between partitions could silently lose skip info or cause incorrect skips. Documenting or guarding the ordering assumption is important for correctness.	Medium
	Report true next docId, not clamped The nextDoc computation uses Math.max(handle.currentDoc, maxDoc), which forces nextDoc >= maxDoc even when the iterator's actual position is inside [minDoc, maxDoc) (e.g., when scan was skipped or bounded by scanTo < maxDoc). This can cause callers to skip subsequent RGs whose range overlaps the real next doc. Report handle.currentDoc directly (guarding only for NO_MORE_DOCS) so the caller sees the true next matching docId. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneFilterDelegationHandle.java [245-252] nextDoc = (handle.currentDoc == DocIdSetIterator.NO_MORE_DOCS) ? Integer.MAX_VALUE - : Math.max(handle.currentDoc, maxDoc); + : handle.currentDoc; } catch (IOException exception) { LOGGER.warn("IOException during collectDocs, returning partial bitset", exception); } } else { - nextDoc = (handle.currentDoc == DocIdSetIterator.NO_MORE_DOCS) ? Integer.MAX_VALUE : Math.max(handle.currentDoc, maxDoc); + nextDoc = (handle.currentDoc == DocIdSetIterator.NO_MORE_DOCS) ? Integer.MAX_VALUE : handle.currentDoc; } Suggestion importance[1-10]: 6 __ Why: The suggestion raises a valid concern about Math.max(handle.currentDoc, maxDoc) potentially clamping nextDoc upward, but since the contract says nextDoc is the next matching docId beyond maxDoc, the current implementation may actually be intentionally correct. Moderate impact and requires deeper context to verify.	Low
General	Use explicit sentinel for error encoding Because the Java side already checks packed < 0 for error, but packed is i64 and the upper 32 bits can be a large positive nextDoc (e.g., Integer.MAX_VALUE), a valid response with nextDoc having its sign bit set would make packed negative and be misinterpreted as an error. Since nextDoc is documented as an i32, its sign bit could realistically become 1 for Integer.MAX_VALUE shifted (it won't, but for any nextDoc >= 0x80000000 it would). Reserve a distinct sentinel (e.g., all-ones) for errors instead of relying on the sign bit of the packed long. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/ffm_callbacks.rs [237-245] + // Contract: Java returns -1 (all-ones) as the error sentinel; any other + // value is a valid packed(nextDoc:i32, wordsWritten:u32). + if packed == -1 { + return Err(format!( + "collectDocs(context_id={}, key={}) failed", + self.context_id, self.key + )); + } let packed_u = packed as u64; let words_written = (packed_u & 0xFFFF_FFFF) as usize; let next_doc = (packed_u >> 32) as i32; if words_written > word_count { Suggestion importance[1-10]: 6 __ Why: Valid concern: since nextDoc is packed in the upper 32 bits and can have its sign bit set (e.g., large positive values), the packed < 0 check could misinterpret valid results as errors. However, Integer.MAX_VALUE has sign bit 0, so realistic values are safe unless nextDoc exceeds that.	Low
	Fix off-by-one in skip predicate **The skip predicate last_next > r_max is inconsistent with the RG-level skip above (last_next > max_doc) and with the contract that next_doc is the next match >= some position. If last_next == r_max, the range [, r_max) cannot contain a match, so it should also be skipped. Use last_next >= r_max to avoid an unnecessary FFM call. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/eval/single_collector.rs [380-385] for (r_min, r_max) in &call_ranges { - if last_next > *r_max { + if last_next >= *r_max { continue; } let effective_min = last_next.max(*r_min); let result = collector .collect_packed_u64_bitset(effective_min, *r_max) Suggestion importance[1-10]: 5 __ Why: Using >= instead of > would save an unnecessary FFM call when last_next == r_max, since the range [*, r_max) is exclusive. Minor optimization but logically valid.	Low

Suggestions up to commit b7c9979

Category	Suggestion	Impact
Possible issue	Avoid using raw Arc pointer as identity Using the raw Arc pointer as a stable identity key is unsafe: if the referenced collector is dropped and a new one is allocated at the same address, the stale next_doc entry in leaf_next_doc will incorrectly be reused, causing wrong skipping. Either clear the map between queries, key it by something more stable (e.g., the resolved leaf DFS index or provider_key), or store Weak/keep an Arc alive with the entry. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/eval/bitmap_tree.rs [1020] -let leaf_key = Arc::as_ptr(collector) as *const () as usize; +// Use the DFS index (stable per query/tree) instead of Arc pointer identity. +let leaf_key = _leaf_dfs_index; Suggestion importance[1-10]: 8 __ Why: Legitimate concern: using raw Arc::as_ptr as a stable key is fragile because addresses can be reused after drop, leading to stale next_doc reuse and incorrect skipping. Suggesting a more stable key (like DFS index) improves correctness.	Medium
	Do not clamp nextDoc upward to maxDoc Using Math.max(handle.currentDoc, maxDoc) will report nextDoc >= maxDoc even when the iterator's current position is still inside the just-scanned range (e.g., when scanTo < maxDoc due to partition bounds). This can incorrectly cause callers to skip subsequent RGs that actually contain matches. Return handle.currentDoc directly (it is guaranteed to be >= scanTo after the loop), and only clamp to Integer.MAX_VALUE on exhaustion. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneFilterDelegationHandle.java [245-247] nextDoc = (handle.currentDoc == DocIdSetIterator.NO_MORE_DOCS) ? Integer.MAX_VALUE - : Math.max(handle.currentDoc, maxDoc); + : handle.currentDoc; Suggestion importance[1-10]: 7 __ Why: Valid concern: when scanTo < maxDoc due to partition bounds, clamping handle.currentDoc up to maxDoc can misrepresent the true next matching doc position and cause incorrect skipping of subsequent RGs. Returning handle.currentDoc directly is more accurate.	Medium
	Use >= for next_doc skip comparison **When last_next >= r_max the sub-range should be skipped (not just when strictly greater), since next_doc is the next matching docId beyond the previous max and is compared against an exclusive upper bound. Using > can cause a redundant FFM call for the boundary case. Also, effective_min may end up equal to r_max in edge cases when a previous range wasn't skipped; guard against that to avoid an empty-but-nonzero-length call. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/eval/single_collector.rs [380-385] -if last_next > *r_max { +if last_next >= *r_max { continue; } let effective_min = last_next.max(*r_min); +if effective_min >= *r_max { + continue; +} let result = collector .collect_packed_u64_bitset(effective_min, *r_max) Suggestion importance[1-10]: 5 __ Why: Reasonable minor optimization: since max_doc/r_max is exclusive and next_doc is the next matching doc, using >= avoids a wasted FFM call at the boundary. Impact is minor but correctness-oriented.	Low
General	Skip RG when next_doc equals max_doc The RG-level skip check uses > against the exclusive max_doc; since next_doc is the next matching docId beyond the last range, it should skip when next_doc >= max_doc. Otherwise a next_doc equal to max_doc will still trigger a full RG evaluation that yields nothing. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/eval/single_collector.rs [295] let mut last_next = self .last_next_doc .load(std::sync::atomic::Ordering::Acquire); -if last_next > max_doc { +if last_next >= max_doc { Suggestion importance[1-10]: 5 __ Why: Valid boundary correctness improvement — using >= avoids a redundant full RG evaluation when next_doc == max_doc. Minor but correct.	Low

Suggestions up to commit f24bb98

Category	Suggestion	Impact
Possible issue	Return accurate nextDoc without clamping The nextDoc is clamped with Math.max(handle.currentDoc, maxDoc), which discards the true next matching docId when it lies within [minDoc, maxDoc) (e.g., when scanTo < maxDoc due to partitionMaxDoc). This can incorrectly cause callers to skip a subsequent RG whose max is greater than the actual next match. Return handle.currentDoc directly so the caller sees the accurate next-match position. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneFilterDelegationHandle.java [245-247] nextDoc = (handle.currentDoc == DocIdSetIterator.NO_MORE_DOCS) ? Integer.MAX_VALUE - : Math.max(handle.currentDoc, maxDoc); + : handle.currentDoc; Suggestion importance[1-10]: 7 __ Why: Valid concern: clamping nextDoc with Math.max(handle.currentDoc, maxDoc) can misrepresent the true next-match position when the scan was bounded by partitionMaxDoc < maxDoc, potentially causing incorrect skipping of subsequent RGs. Returning handle.currentDoc directly aligns with the documented contract.	Medium
	Fix off-by-one in skip condition The skip condition uses last_next > max_doc, but max_doc is exclusive; the documented semantics of next_doc are "next matching docId beyond maxDoc" and skipping should occur when next_doc >= rg_max. Using > may fail to skip an RG whose max_doc equals next_doc. Change to >= to match the contract and the doc comment above. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/eval/single_collector.rs [295] let mut last_next = self .last_next_doc .load(std::sync::atomic::Ordering::Acquire); -if last_next > max_doc { +if last_next >= max_doc { Suggestion importance[1-10]: 6 __ Why: Given max_doc is exclusive and the API doc says skip when next_doc >= rg_max, using > may fail to skip an RG when equality holds, allowing an unnecessary FFM call. The fix aligns with the stated contract.	Low
	Reconcile empty-range next_doc sentinel The error branch above already handles packed < 0, but note that packed as u64 on a negative i64 sign-extends and would decode next_doc from the upper bits. That's fine since we early-return, but the empty-range branch at the top returns next_doc: i32::MAX while From<Vec> uses i32::MIN; ensure the empty-range value is consistent with what callers expect (callers use i32::MIN as "no info", and returning i32::MAX here will cause all subsequent RGs to be skipped after an empty-range call). sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/ffm_callbacks.rs [218-221] +let packed_u = packed as u64; +let words_written = (packed_u & 0xFFFF_FFFF) as usize; +let next_doc = (packed_u >> 32) as i32; - Suggestion importance[1-10]: 5 __ Why: Raises a legitimate consistency concern: the empty-range branch returns next_doc: i32::MAX which would trigger skips of subsequent RGs, while other paths use i32::MIN as the "no info" sentinel. However, the improved_code is identical to existing_code, so the suggestion only flags the issue without proposing a concrete fix.	Low
General	Use inclusive skip bound for ranges *Same off-by-one as above: r_max is exclusive, so ranges should be skipped when last_next >= r_max, not >. Otherwise a range with r_max == last_next will still trigger an unnecessary FFM call whose result is guaranteed empty (since effective_min == r_max). sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/eval/single_collector.rs [380-383] for (r_min, r_max) in &call_ranges { - if last_next > *r_max { + if last_next >= *r_max { continue; } let effective_min = last_next.max(*r_min); Suggestion importance[1-10]: 6 __ Why: Same off-by-one applies to the per-range skip check; using >= avoids a useless FFM call when last_next == r_max. Minor correctness/efficiency improvement.	Low

[github-actions]

Persistent review updated to latest commit 55cac5b

[github-actions]

Persistent review updated to latest commit 454eafc

[github-actions]

❌ Gradle check result for 454eafc: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit e7cda94

[github-actions]

✅ Gradle check result for e7cda94: SUCCESS

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 71.43%. Comparing base (dbde5fb) to head (edb60a7).
⚠️ Report is 1 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #22493      +/-   ##
============================================
- Coverage     73.41%   71.43%   -1.98%     
- Complexity    76419    76625     +206     
============================================
  Files          6104     6132      +28     
  Lines        346628   357308   +10680     
  Branches      49886    52076    +2190     
============================================
+ Hits         254469   255241     +772     
- Misses        71860    81748    +9888     
- Partials      20299    20319      +20     

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

[github-actions]

Persistent review updated to latest commit f06f05c

[github-actions]

Persistent review updated to latest commit 4051f8a

[github-actions]

❌ Gradle check result for 4051f8a: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit c8c76ac

[github-actions]

❌ Gradle check result for c8c76ac: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit 15a465a

[github-actions]

❌ Gradle check result for 15a465a: null

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit f24bb98

[github-actions]

✅ Gradle check result for f24bb98: SUCCESS

[github-actions]

Persistent review updated to latest commit b7c9979

[github-actions]

❌ Gradle check result for b7c9979: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit 2fb7234

[github-actions]

Persistent review updated to latest commit f2ced7a

[github-actions]

❌ Gradle check result for f2ced7a: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit 772f4f3

[github-actions]

❌ Gradle check result for 772f4f3: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit edb60a7

[github-actions]

✅ Gradle check result for edb60a7: SUCCESS