Wire doc_values collection into the planner

[harshavamsi]

Description

This is part 2 and must be merged after #22844. The first commit is part of that PR.

This PR integrates Lucene doc-values sources into normal PPL execution through DataFusion. It adds planner routing, standard Lucene shard-reader support, cross-backend coordinator reduction, and safe lifecycle/cancellation handling without global holders.

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

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 Multiple PR themes Sub-PR theme: Lucene doc-values Arrow batch source reader Relevant files: sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/DocValuesBatchSource.java sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/DocValuesBatchSourceFactory.java sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/LuceneDocValuesBatchSourceTests.java Sub-PR theme: DataFusion Arrow batch source native bridge and executor Relevant files: sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/arrow/ArrowBatchSourceCallbacks.java sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DatafusionArrowBatchSourceExecutor.java sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/nativelib/NativeBridge.java sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/arrow/ArrowBatchSourceCallbacksTests.java sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/DatafusionArrowBatchSourceExecutorTests.java Sub-PR theme: Planner routing and Lucene capability wiring for doc-values Relevant files: sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneFragmentPlanner.java sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneFragmentConvertor.java sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneAnalyticsBackendPlugin.java sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneShardPreference.java sandbox/plugins/analytics-backend-lucene/src/test/java/org/opensearch/be/lucene/PlanAlternativeSelectorTests.java sandbox/plugins/analytics-backend-lucene/src/test/java/org/opensearch/be/lucene/LuceneArrowSourcePlanTests.java

⚡ Recommended focus areas for review Row index overflow at 65536 batch size decodeSorted packs the (ord, row) pair into a long with ((long)(ord+1) << 20) | row, using only 20 bits for the row index (max 0xFFFFF = 1,048,575). BATCH_SIZE is 65,536 so this fits today, but the mask (int)(ordRowScratch[i] & 0xFFFFF) also caps the row bits at 20. If BATCH_SIZE is ever raised above ~1M, row indices will silently collide/truncate producing corrupt results. Consider using 21+ bits or a constant derived from BATCH_SIZE and asserting the invariant. for (int i = 0; i < size; i++) { ordRowScratch[i] = ((long) (ordScratch[i] + 1) << 20) | i; } Arrays.sort(ordRowScratch, 0, size); int lastOrd = Integer.MIN_VALUE; BytesRef term = null; for (int i = 0; i < size; i++) { int row = (int) (ordRowScratch[i] & 0xFFFFF); int ord = (int) (ordRowScratch[i] >>> 20) - 1; if (ord < 0) { vector.setNull(row); } else { if (ord != lastOrd) { term = values.lookupOrd(ord); lastOrd = ord; } setBytes(vector, row, term); } } vector.setValueCount(size); Missing decRef on partially opened leaf close() calls searcher.getIndexReader().decRef() unconditionally once. If advanceLeaf throws after weight.scorer(leaf) returned non-null but before iterator/readers are fully initialized (e.g. openColumn throws), state is inconsistent but the reader ref count is still balanced by close(). However, if the constructor completes and the caller never calls close() (e.g. factory transfer path fails after DocValuesBatchSource construction), the reader ref acquired in the factory leaks. Verify all failure paths downstream of construction reliably invoke close(). public void close() { cancel(); if (closed.compareAndSet(false, true)) { try { searcher.getIndexReader().decRef(); } catch (IOException e) { LOGGER.warn("failed to release doc-values source reader", e); } } } Concurrency: source map access outside lock nextBatch calls lease.binding.source(sourceKey) (synchronized get) then invokes source.exportNextBatch without holding the binding lock. Meanwhile requestClose/release may call source.close() concurrently. SourceEntry.exportNextBatch and close are both synchronized on the entry, so they serialize correctly, but note that after close sets closed=true a concurrent nextBatch that already acquired the entry lock later will throw IllegalStateException("Arrow batch source is closed"), which is mapped to ERROR rather than CANCELLED. Consumers observing a shutdown race will see a generic error instead of cancellation. CallbackLease lease = acquire(bindingId, false); if (lease == null) { writeError(errorPointer, errorCapacity, "Arrow batch source binding is closed: " + bindingId); return ERROR; } try (lease) { SourceEntry source = lease.binding.source(sourceKey); if (source == null) { writeError(errorPointer, errorCapacity, "Unknown Arrow batch source key: " + sourceKey); return ERROR; } return source.exportNextBatch(arrayPointer, schemaPointer); } catch (TaskCancelledException cancelled) { return CANCELLED; } catch (Throwable throwable) { writeError(errorPointer, errorCapacity, throwable.toString()); LOGGER.debug("Failed to read Arrow batch source binding {} key {}", bindingId, sourceKey, throwable); return ERROR; } } Filter not preserved in rebased fragment In extractArrowSourceShape, when an Aggregate has a Filter between it and the scan, the filter is extracted into filter but the rebased fragment (rebasedAggregate over input.scan() or over a rebased Project of input.scan()) never re-attaches the filter. The ArrowSourceShape.filter() is serialized separately as a Lucene QueryBuilder, but the referenced columns collected from the aggregate/project may miss columns referenced only in the filter condition — since the filter is applied at the Lucene layer before batches, any filter-only column doesn't need to be in the Arrow source, but the rebased fragment's row type must still match what DataFusion expects. Verify the filter's column references are not needed in the Arrow input schema, or that the Lucene shard applies the filter before doc-values decoding (which appears to be the intent via state.filterQuery() in LuceneSearchExecEngine). private static ArrowSourceShape extractArrowSourceShape(RelNode fragment) { RelNode originalFragment = fragment; List<RelNode> wrappers = new ArrayList<>(); while (fragment instanceof Aggregate == false) { if (fragment.getInputs().size() != 1 || (fragment instanceof Project == false && fragment instanceof Filter == false && fragment instanceof org.apache.calcite.rel.core.Sort == false)) { return extractRowArrowSourceShape(originalFragment); } wrappers.add(fragment); fragment = fragment.getInput(0); } Aggregate aggregate = (Aggregate) fragment; if (aggregate.getGroupSets() != null && aggregate.getGroupSets().size() > 1) { return null; } RelNode below = aggregate.getInput(); Project project = null; if (below instanceof Project candidate) { project = candidate; below = candidate.getInput(); } Filter filter = null; if (below instanceof Filter candidate) { filter = candidate; below = candidate.getInput(); } if (below instanceof OpenSearchRelNode == false || below.getInputs().isEmpty() == false) { return null; } List<FieldStorageInfo> storage = ((OpenSearchRelNode) below).getOutputFieldStorage(); if (storage == null) { return null; } TreeSet<Integer> referenced = new TreeSet<>(); if (project != null) { RexShuttle collector = inputReferenceCollector(referenced); for (RexNode expression : project.getProjects()) { expression.accept(collector); } } else { for (int ordinal : aggregate.getGroupSet()) { referenced.add(ordinal); } for (AggregateCall call : aggregate.getAggCallList()) { referenced.addAll(call.getArgList()); if (call.filterArg >= 0) { return null; } } } // Preserve the metadata count fast path for COUNT(*). if (referenced.isEmpty()) { return null; } RebasedInput input = rebaseInput(aggregate, below, storage, referenced); if (input == null) { return null; } RexShuttle remap = inputRemapper(input.oldToNew()); RelNode rebasedInput; Aggregate rebasedAggregate; if (project != null) { List<RexNode> remapped = new ArrayList<>(project.getProjects().size()); for (RexNode expression : project.getProjects()) { remapped.add(expression.accept(remap)); } rebasedInput = LogicalProject.create(input.scan(), project.getHints(), remapped, project.getRowType().getFieldNames()); rebasedAggregate = aggregate.copy( aggregate.getTraitSet(), rebasedInput, aggregate.getGroupSet(), aggregate.getGroupSets(), aggregate.getAggCallList() ); } else { ImmutableBitSet.Builder newGroupSet = ImmutableBitSet.builder(); for (int ordinal : aggregate.getGroupSet()) { newGroupSet.set(input.oldToNew()[ordinal]); } List<AggregateCall> remappedCalls = new ArrayList<>(aggregate.getAggCallList().size()); for (AggregateCall call : aggregate.getAggCallList()) { List<Integer> remappedArguments = new ArrayList<>(call.getArgList().size()); for (int argument : call.getArgList()) { remappedArguments.add(input.oldToNew()[argument]); } remappedCalls.add(call.copy(remappedArguments, -1, call.collation)); } rebasedAggregate = aggregate.copy(aggregate.getTraitSet(), input.scan(), newGroupSet.build(), null, remappedCalls); } RelNode rebased = rebasedAggregate; for (int i = wrappers.size() - 1; i >= 0; i--) { RelNode wrapper = wrappers.get(i); rebased = wrapper.copy(wrapper.getTraitSet(), List.of(rebased)); } return new ArrowSourceShape(SOURCE_INPUT_ID, rebased, input.columns(), filter, resultNames(originalFragment)); }

---

⚠️ Review coverage: The following files were not included in this review because of the token budget:

• sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/AnnotatedPredicate.java
• sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneFragmentWirePlan.java
• sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/AnalyticsPlugin.java
• sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ArrowBatchSourcePlan.java
• sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/AnalyticsSearchBackendPlugin.java
• sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/DocValuesBatchSourceFactory.java
• sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneSearcherState.java
• sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/PplAggregateCallRewriter.java
• sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/PlanAlternativeSelector.java
• sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/BackendPlanAdapter.java
• sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/DataFusionFragmentConvertorTests.java
• sandbox/libs/analytics-framework/src/test/java/org/opensearch/analytics/spi/ArrowBatchSourcePlanTests.java
• sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/backend/ShardScanExecutionContext.java
• sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ArrowBatchSourceFactory.java
• sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchAggregateSplitRule.java
• sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ArrowBatchSource.java
• sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DatafusionLocalSession.java
• sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/PlannerImpl.java
• sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/CapabilityRegistry.java
• sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchFilterRule.java
• sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchSortRule.java
• sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/PlanForker.java
• sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchLateMaterializationRewriter.java
• sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/PlannerContext.java
• sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/DAGBuilder.java
• sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/CapabilityResolutionUtils.java
• sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/MockLuceneBackend.java
• sandbox/plugins/analytics-backend-lucene/src/test/java/org/opensearch/be/lucene/LuceneQueryConversionUtilsTests.java
• sandbox/plugins/analytics-backend-datafusion/rust/src/arrow_batch_source/table_provider.rs
• sandbox/plugins/analytics-backend-datafusion/rust/src/arrow_batch_source/callbacks.rs
• sandbox/plugins/analytics-backend-datafusion/rust/src/ffm.rs
• sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs
• sandbox/plugins/analytics-backend-datafusion/rust/src/local_executor.rs
• sandbox/plugins/analytics-backend-datafusion/rust/src/arrow_batch_source/mod.rs
• sandbox/plugins/analytics-backend-datafusion/rust/src/lib.rs
• sandbox/plugins/analytics-backend-datafusion/licenses/jackson-datatype-jdk8-2.22.2.jar.sha1
• sandbox/plugins/analytics-engine/licenses/jackson-dataformat-yaml-2.22.2.jar.sha1
• sandbox/libs/analytics-framework/licenses/jackson-databind-2.22.2.jar.sha1
• sandbox/libs/analytics-framework/licenses/jackson-core-2.22.2.jar.sha1

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to ecd8739

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Fix ord/row bit-packing overflow for high-cardinality fields The packed ord/row encoding uses only 20 bits for both the row index and the ord value, but BATCH_SIZE is 65_536 (fits in 17 bits) while ord values can exceed ~1M (2^20-1) for high-cardinality keyword fields. When ordScratch[i] + 1 >= (1 << 20), the shift will overflow into the row bits and corrupt results. Use a wider split (e.g., 32 bits for ord, 32 bits for row) since the value is already a long. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/DocValuesBatchSource.java [271-280] for (int i = 0; i < size; i++) { - ordRowScratch[i] = ((long) (ordScratch[i] + 1) << 20) | i; + ordRowScratch[i] = ((long) (ordScratch[i] + 1) << 32) | i; } Arrays.sort(ordRowScratch, 0, size); int lastOrd = Integer.MIN_VALUE; BytesRef term = null; for (int i = 0; i < size; i++) { - int row = (int) (ordRowScratch[i] & 0xFFFFF); - int ord = (int) (ordRowScratch[i] >>> 20) - 1; + int row = (int) (ordRowScratch[i] & 0xFFFFFFFFL); + int ord = (int) (ordRowScratch[i] >>> 32) - 1; Suggestion importance[1-10]: 9 __ Why: Correctly identifies a real bug: BATCH_SIZE is 65_536 but ord values for high-cardinality keyword fields can easily exceed 2^20-1 (~1M), causing bit overflow that corrupts the row index. Since the packed value is already a long, widening the split to 32/32 is a valid fix.	High
	Fix QTF gating when Lucene is a candidate backend The predicate returns true (allowing QTF) whenever there is more than one viable backend, even if Lucene is among them and is ultimately selected as the scan backend. This can allow QTF to fire against a Lucene scan that cannot serve the fetch phase. The check should skip QTF whenever Lucene is one of the viable backends for the scan, unless it is guaranteed not to be chosen. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/PlannerImpl.java [152-155] Optional<RelNode> lateMat = OpenSearchLateMaterializationRewriter.rewrite( modifiedRelNode, - scan -> scan.getViableBackends().size() != 1 || scan.getViableBackends().contains("lucene") == false + scan -> scan.getViableBackends().contains("lucene") == false ); Suggestion importance[1-10]: 6 __ Why: The suggestion raises a valid concern: if Lucene is among multiple viable backends, QTF could still fire and the scan may resolve to Lucene which cannot serve fetch. However, the original logic may intentionally allow QTF when non-Lucene alternatives exist. The concern is legitimate but the fix may be overly conservative.	Low
	Verify Arrow buffer ownership after C Data export exportVectorSchemaRoot transfers ownership of the Arrow buffers to the C Data consumer, but root.close() immediately after export may release/decrement refcounts of those buffers before the native side has imported them. Depending on Arrow's export semantics, closing the root here can corrupt the exported array. Verify whether Arrow Java's Data.exportVectorSchemaRoot retains buffers independently; if not, do not close root after successful export (only on failure). sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/arrow/ArrowBatchSourceCallbacks.java [401-407] ArrowArray array = ArrowArray.wrap(arrayPointer.address()); ArrowSchema schema = ArrowSchema.wrap(schemaPointer.address()); - Data.exportVectorSchemaRoot(source.allocator(), root, null, array, schema); - return root.getRowCount() == 0 ? EMPTY_BATCH : root.getRowCount(); - } finally { - root.close(); - } + boolean exported = false; + try { + Data.exportVectorSchemaRoot(source.allocator(), root, null, array, schema); + exported = true; + return root.getRowCount() == 0 ? EMPTY_BATCH : root.getRowCount(); + } finally { + if (!exported) { + root.close(); + } else { + root.close(); // confirm Arrow retains buffers after export; otherwise remove + } + } Suggestion importance[1-10]: 4 __ Why: The concern is speculative and asks for verification. Arrow Java's Data.exportVectorSchemaRoot is documented to retain buffer references, so closing the root after export is standard practice. The improved_code is nearly identical to existing_code.	Low
General	Stop recursion at exchange or scan nodes bindToBackend recurses through all children including any OpenSearchTableScan beneath the fragment, but the intent (per comment) is only to bind coordinator-local wrappers above the exchange. If an exchange boundary or scan is embedded in the subtree, this will narrow its annotations too and may drop the original scan-backend context. Consider stopping the recursion at exchange nodes or at leaf scans to avoid inadvertently rewriting scan-side operators. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/PlanForker.java [48-55] static RelNode bindToBackend(RelNode node, String backend) { + if (node instanceof OpenSearchTableScan || node instanceof OpenSearchExchange) { + return node; + } List<RelNode> children = node.getInputs().stream().map(child -> bindToBackend(child, backend)).toList(); if (node instanceof OpenSearchRelNode openSearchNode) { List<OperatorAnnotation> annotations = openSearchNode.getAnnotations().stream().map(a -> a.narrowTo(backend)).toList(); return openSearchNode.copyResolved(backend, children, annotations); } return children.isEmpty() ? node : node.copy(node.getTraitSet(), children); } Suggestion importance[1-10]: 6 __ Why: The concern is valid: bindToBackend recurses unconditionally and could rewrite scan-side operators if any are reachable in the subtree. However, the caller applies this only to the root fragment above the exchange, so the practical impact depends on fragment structure. Worth considering as a safety guard.	Low
	Deduplicate backend list before returning result may contain duplicates when a backend declares filter capability across multiple formats (both doc-values and index). The prior version had the same aggregation but returning duplicates to callers could cause downstream logic (e.g. viability set intersections) to misbehave. Consider deduplicating the result before returning. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/CapabilityRegistry.java [259-265] for (String format : field.getIndexFormats()) { result.addAll(filterBackends(function, fieldType, format)); } List<String> docValueReaders = scanBackendsForField(field); List<String> indexReaders = indexScanBackendsForField(field); result.removeIf(backend -> docValueReaders.contains(backend) == false && indexReaders.contains(backend) == false); -return result; +return result.stream().distinct().toList(); Suggestion importance[1-10]: 4 __ Why: Deduplication is a reasonable minor improvement, but the pre-existing behavior already had this property, so it's not directly introduced by this PR. Marginal improvement.	Low
	Avoid overflow when capping error message length Math.toIntExact throws when capacity - 1 exceeds Integer.MAX_VALUE, but the intent is to cap at buffer size. Since bytes.length is already an int, use a plain Math.min cast on the small side. Also, when capacity == 1, length becomes 0 and only the null terminator is written — that's fine, but the current toIntExact on a possibly large capacity - 1L will throw ArithmeticException for very large capacities. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/arrow/ArrowBatchSourceCallbacks.java [199-202] - int length = Math.toIntExact(Math.min(bytes.length, capacity - 1L)); + int length = (int) Math.min(bytes.length, capacity - 1L); + if (length < 0) length = 0; MemorySegment view = output.reinterpret(capacity); MemorySegment.copy(bytes, 0, view, ValueLayout.JAVA_BYTE, 0L, length); view.set(ValueLayout.JAVA_BYTE, length, (byte) 0); Suggestion importance[1-10]: 3 __ Why: Minor edge case; error capacities in this code path are small buffers (256 bytes typically), so Math.toIntExact will not realistically overflow. Low practical impact.	Low
	Verify dual-iterator numeric doc-values semantics numericValues is called twice for the same field to populate values and validity. For SORTED_NUMERIC types this creates two independent iterators, but the NumericColumn code path calls column.values().longValues(...) followed by column.validity().advanceExact(docs[i]) per row — using two separate iterators is intentional here. However, ensure that both iterators are advanced consistently; if longValues returns fewer values than docs.length (e.g., for missing docs), the fallback slot may hold stale data. Confirm longValues guarantees a value per requested doc (or zero-fill), otherwise nullable rows may read wrong data before the validity mask blanks them. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/DocValuesBatchSource.java [407-416] + case KEYWORD, IP -> new SortedColumn(sortedValues(leaf, column, type), column.kind()); + case BINARY -> type == DocValuesType.BINARY + ? new BinaryColumn(leaf.reader().getBinaryDocValues(column.name())) + : new SortedColumn(sortedValues(leaf, column, type), column.kind()); + case LONG, TIMESTAMP, BOOLEAN, FLOAT, DOUBLE -> new NumericColumn( + numericValues(leaf, column.name(), type), + numericValues(leaf, column.name(), type), + column.kind() + ); - Suggestion importance[1-10]: 2 __ Why: The existing_code and improved_code are identical; the suggestion only asks to verify semantics without proposing a concrete change.	Low
	Validate imported batch structure more strictly import_batch compares batch.num_rows() to expected_rows, but for EMPTY_BATCH (-3) and status codes that report row counts, an empty struct with no columns has num_rows() == 0 regardless of the sentinel. The row-count validation may be wrong for zero-column projections, and more importantly, the sentinel EMPTY_BATCH value passed as expected_rows = 0 is fine but nonzero-row batches with mismatched counts will fail. Consider validating that the returned batch actually has the schema's field count as well, to catch mismatched imports early. sandbox/plugins/analytics-backend-datafusion/rust/src/arrow_batch_source/callbacks.rs [129-141] +match status { + 0 => Ok(None), + CANCELLED => Err(DataFusionError::Execution("query cancelled".into())), + ERROR => Err(callback_error( + &error, + "Java Arrow batch source failed".into(), + )), + EMPTY_BATCH => import_batch(array, schema, 0), + rows if rows > 0 => import_batch(array, schema, rows as usize), - Suggestion importance[1-10]: 2 __ Why: The improved_code is identical to existing_code, so no actual change is proposed. The commentary is vague and doesn't provide an actionable fix.	Low

---

Previous suggestions

Suggestions up to commit 89e0da6

Category	Suggestion	Impact
Possible issue	Do not block cancel behind nextBatch Holding the SourceEntry monitor across source.nextBatch() means a concurrent cancel() upcall from native blocks until the current batch is fully produced and exported, defeating the cooperative-cancellation contract advertised on ArrowBatchSource.cancel(). Move cancel() outside synchronization (it already has its own cancellationLock) so cancellation can interrupt an in-flight nextBatch promptly. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/arrow/ArrowBatchSourceCallbacks.java [392-408] -private synchronized long exportNextBatch(MemorySegment arrayPointer, MemorySegment schemaPointer) throws Exception { - if (closed) { - throw new IllegalStateException("Arrow batch source is closed"); +private long exportNextBatch(MemorySegment arrayPointer, MemorySegment schemaPointer) throws Exception { + synchronized (this) { + if (closed) { + throw new IllegalStateException("Arrow batch source is closed"); + } } VectorSchemaRoot root = source.nextBatch(); if (root == null) { return 0L; } try { ArrowArray array = ArrowArray.wrap(arrayPointer.address()); ArrowSchema schema = ArrowSchema.wrap(schemaPointer.address()); Data.exportVectorSchemaRoot(source.allocator(), root, null, array, schema); return root.getRowCount() == 0 ? EMPTY_BATCH : root.getRowCount(); } finally { root.close(); } Suggestion importance[1-10]: 7 __ Why: Holding the monitor across nextBatch() can indeed delay cooperative cancellation, defeating the documented contract; releasing the lock during the blocking call is a valid correctness/behavior improvement.	Medium
	Guard FFI array lifetime on error paths When status is 0 (end-of-stream), array and schema are dropped without being imported, but if Java populated them (or on other error/cancelled paths) their FFI release callbacks may not have been invoked correctly. More critically, on the CANCELLED/ERROR branches the FFI_ArrowArray/FFI_ArrowSchema are dropped: if Java partially initialized them with a release callback, that's fine; if not, their release pointer must be null. Ensure the Java side guarantees these are either fully released-callback-bearing or left zero-initialized on non-success returns to avoid leaking or invalid frees. sandbox/plugins/analytics-backend-datafusion/rust/src/arrow_batch_source/callbacks.rs [129-141] +match status { + 0 => Ok(None), + CANCELLED => Err(DataFusionError::Execution("query cancelled".into())), + ERROR => Err(callback_error( + &error, + "Java Arrow batch source failed".into(), + )), + EMPTY_BATCH => import_batch(array, schema, 0), + rows if rows > 0 => import_batch(array, schema, rows as usize), + other => Err(DataFusionError::Execution(format!( + "unexpected Arrow batch callback status {other}" + ))), +} - Suggestion importance[1-10]: 4 __ Why: The suggestion raises a valid concern about FFI array lifetime on error paths, but it does not propose a concrete code change (existing_code equals improved_code), only asking for verification of Java-side behavior.	Low
	Address drop/next_batch race on release Drop calls self.cancel() and then invokes release_callback unconditionally. If cancel has already been called before drop, cancelled.swap returns true and the cancel callback is skipped — good — but if Drop runs while a next_batch call is in flight on another thread (e.g., blocking task not yet joined), releasing the source key mid-call can race. Ensure Drop waits for or is only reachable after outstanding calls complete; alternatively document that callers must guarantee no concurrent next_batch at drop time. sandbox/plugins/analytics-backend-datafusion/rust/src/arrow_batch_source/callbacks.rs [146-150] -if self.cancelled.swap(true, Ordering::AcqRel) == false { +if !self.cancelled.swap(true, Ordering::AcqRel) { if let Some(callback) = cancel_callback() { unsafe { callback(self.binding_id, self.source_key) }; } } Suggestion importance[1-10]: 3 __ Why: Raises a potential race condition concern, but the improved code only changes == false to ! (a style change) and does not actually address the race. The Drop implementation aborts the pending handle anyway.	Low
General	Guard packed sort against BATCH_SIZE growth The packed encoding ((long)(ord+1) << 20) | i reserves only 20 bits for the row index, which supports at most 2^20 = 1,048,576 rows — larger than BATCH_SIZE=65_536 today, but any future increase to BATCH_SIZE beyond 1M will silently corrupt sort ordering. Assert or derive the shift from BATCH_SIZE to prevent regressions. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/DocValuesBatchSource.java [265-274] private void decodeSorted(SortedColumn column, FieldVector vector, int size) throws IOException { SortedDocValues values = column.values(); + assert size <= (1 << 20) : "row index does not fit in 20 bits; increase shift or reduce BATCH_SIZE"; for (int i = 0; i < size; i++) { ordScratch[i] = values.advanceExact(docs[i]) ? values.ordValue() : -1; } allocateBytes(vector, size); for (int i = 0; i < size; i++) { ordRowScratch[i] = ((long) (ordScratch[i] + 1) << 20) | i; } Suggestion importance[1-10]: 5 __ Why: Adding an assert makes the 20-bit row-index limit explicit and prevents silent corruption if BATCH_SIZE is increased later; a reasonable defensive check but low current-impact.	Low
	Gate sort fallback on exchange presence This fallback admits any sort-capable backend with an exchange sink even when the child produces no exchange, which risks selecting a backend disconnected from the actual data flow. Verify the child subtree actually terminates in an exchange before broadening viableBackends; otherwise the planner may pick a backend that cannot receive the scan output. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchSortRule.java [61-67] if (viableBackends.isEmpty()) { // A coordinator sort above an exchange is scan-free. It can run on any // sort-capable sink backend even when Lucene drove the shard scan. - viableBackends = sortCapable.stream() - .filter(backend -> context.getCapabilityRegistry().getBackend(backend).getExchangeSinkProvider() != null) - .toList(); + // Only broaden when the child subtree actually crosses an exchange. + if (hasExchangeBelow(call.rel(0))) { + viableBackends = sortCapable.stream() + .filter(backend -> context.getCapabilityRegistry().getBackend(backend).getExchangeSinkProvider() != null) + .toList(); + } } Suggestion importance[1-10]: 5 __ Why: Valid concern about the fallback broadening backend selection without verifying an exchange exists below, but relies on a hypothetical hasExchangeBelow helper and may contradict the PR's stated intent per the comment.	Low
	Race between open and requestClose factory.open() runs while holding openLock but without the binding monitor; if requestClose() runs concurrently between the second closing check and factory.open(), the newly opened source is registered into sources after requestClose() snapshotted sources.values(), so it will not be cancelled and only gets closed via tryFinishClose() after all callback leases drop. Verify that tryFinishClose() reliably collects such races, or hold openLock inside requestClose() to serialize with in-flight opens. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/arrow/ArrowBatchSourceCallbacks.java [252-290] private int open(int[] projection) throws Exception { - synchronized (this) { - if (closing) { - throw new IllegalStateException("Arrow batch source binding is closing: " + bindingId); - } - } ArrowBatchSource source; synchronized (openLock) { synchronized (this) { if (closing) { throw new IllegalStateException("Arrow batch source binding is closing: " + bindingId); } } source = factory.open(projection); } Suggestion importance[1-10]: 4 __ Why: Highlights a plausible race, but the code already handles late-opened sources via tryFinishClose() cleaning up remaining entries; the suggestion is more of a verification request than a concrete fix.	Low
	Fix misleading row-count error wording The error message swaps the labels: expected_rows is the value declared by Java, while batch.num_rows() is what was actually imported. As written, the message reads as if the imported count is the expected one. Swap the wording so operators can diagnose the mismatch correctly. sandbox/plugins/analytics-backend-datafusion/rust/src/arrow_batch_source/callbacks.rs [173-178] if batch.num_rows() != expected_rows { return Err(DataFusionError::Execution(format!( - "Arrow batch callback returned row count {expected_rows}, imported {}", + "Arrow batch callback declared row count {expected_rows}, imported {}", batch.num_rows() ))); } Suggestion importance[1-10]: 3 __ Why: Minor wording clarification for an error message; low impact and the original message is arguably understandable.	Low
	Document ord/row bit-packing invariants When iterator reaches NO_MORE_DOCS, per-leaf state (readers, liveDocs, leaf) is left pointing at the finished segment, so the next call re-enters decodeBatch for size > 0 from the just-finished leaf but the following iteration will re-advance to a new leaf via advanceLeaf(), which reallocates readers. That is fine, but if size == 0 after exhausting the last leaf, continue correctly re-enters advanceLeaf; however if size > 0, readers still references the current (now finished) leaf and is correctly used. The bug is subtler: decodeSorted uses a bitmask (ordScratch[i] + 1) << 20 | i which requires size <= 0xFFFFF (1,048,575). With BATCH_SIZE = 65_536 this is safe, but the ord encoding assumes ord fits in 43 bits — Lucene ord can exceed 1<<43 only in pathological cases, but ord + 1 may overflow into the row bits if ord is negative or huge. Guard the encoding or document the constraint explicitly. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/DocValuesBatchSource.java [265-274] +while (size < BATCH_SIZE) { + int doc = iterator.nextDoc(); + if (doc == DocIdSetIterator.NO_MORE_DOCS) { + iterator = null; + break; + } + if (liveDocs == null || liveDocs.get(doc)) { + docs[size++] = doc; + } - Suggestion importance[1-10]: 2 __ Why: The existing_code equals improved_code; the suggestion only asks to document/guard an invariant without proposing a concrete change.	Low

Suggestions up to commit b7a7333

Category	Suggestion	Impact
Possible issue	Ensure snapshot reference is released The success path's cleanup closes searcher and reader but not snapshotRef, whereas the failure path closes searcher and snapshotRef. If DataFormatAwareReader does not itself close the passed-in snapshotRef on its own close(), the snapshot reference will leak on the happy path. Ensure the cleanup lambda also closes snapshotRef (or confirm DataFormatAwareReader.close() releases it) to keep reference counts balanced. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneAnalyticsBackendPlugin.java [320-329] try { DataFormatAwareReader reader = new DataFormatAwareReader( snapshotRef, Map.of(plugin.getDataFormat(), new LuceneReader(searcher.getDirectoryReader(), Map.of())) ); - return new GatedCloseable<>(reader, () -> IOUtils.close(searcher, reader)); + return new GatedCloseable<>(reader, () -> IOUtils.close(searcher, reader, snapshotRef)); } catch (RuntimeException | Error e) { IOUtils.closeWhileHandlingException(searcher, snapshotRef); throw e; } Suggestion importance[1-10]: 6 __ Why: Valid concern about potential snapshot leak on the happy path. The DataFormatAwareReader constructor takes snapshotRef and likely releases it on close, but this should be verified. If not released, it constitutes a resource leak.	Low
	Verify duplicate numeric iterators are independent NumericDocValues iterators are stateful and single-pass; obtaining two independent instances here means one is used for value reads and one for validity checks, but calling advanceExact on the validity iterator inside decodeNumeric/decodeLong after longValues consumed the values iterator may still work only because they are separate iterators. However, allocating two iterators per leaf doubles I/O and can be avoided by using a single iterator and tracking advanceExact results during the initial pass. At minimum, verify both calls actually return distinct iterators for all codecs; otherwise validity checks will misbehave. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/DocValuesBatchSource.java [412-416] -case LONG, TIMESTAMP, BOOLEAN, FLOAT, DOUBLE -> new NumericColumn( - numericValues(leaf, column.name(), type), - numericValues(leaf, column.name(), type), - column.kind() -); +case LONG, TIMESTAMP, BOOLEAN, FLOAT, DOUBLE -> { + NumericDocValues values = numericValues(leaf, column.name(), type); + NumericDocValues validity = numericValues(leaf, column.name(), type); + yield new NumericColumn(values, validity, column.kind()); +} Suggestion importance[1-10]: 5 __ Why: The suggestion correctly notes that two iterators are created for values and validity, but the code already does allocate two iterators via two calls to numericValues. The concern about doubling I/O is valid but the improved_code is essentially equivalent to existing code. Moderate value in flagging the pattern.	Low
	Drop metadata driver when it covers nothing When admitMetadataDriver is true and metadataDriverCoversAny is false, the metadata driver is not removed if it was already in viableBackends — but earlier per-field pruning may have kept it via the delegation-supporter branch. Confirm this is intentional; otherwise, when the metadata driver covers no field, it should be dropped even under admitMetadataDriver=true to match the previous behavior's permissive-gate semantics. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchTableScanRule.java [126-133] if (admitMetadataDriver) { - if (metadataDriverCoversAny && viableBackends.contains(metadataDriver) == false) { - viableBackends.add(metadataDriver); + if (metadataDriverCoversAny) { + if (viableBackends.contains(metadataDriver) == false) { + viableBackends.add(metadataDriver); + } + } else { + viableBackends.remove(metadataDriver); } } else { // This setting controls both Lucene's metadata count path and its doc-values path. viableBackends.remove(metadataDriver); } Suggestion importance[1-10]: 5 __ Why: Identifies a plausible behavioral gap where the metadata driver could remain viable via delegation despite covering no field; however, correctness depends on planner semantics that may intentionally allow this path.	Low
General	Validate projection length before narrowing Math.toIntExact throws ArithmeticException (not IllegalArgumentException) for values > Integer.MAX_VALUE, and the subsequent negative check is unreachable since toIntExact on a non-negative long cannot yield a negative int. Validate the raw long before narrowing so oversized/negative lengths produce a controlled error message written to errorPointer via the outer catch. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/arrow/ArrowBatchSourceCallbacks.java [104-107] -int length = Math.toIntExact(projectionLength); -if (length < 0) { - throw new IllegalArgumentException("projection length must be non-negative"); +if (projectionLength < 0L || projectionLength > Integer.MAX_VALUE) { + throw new IllegalArgumentException("projection length out of range: " + projectionLength); } +int length = (int) projectionLength; Suggestion importance[1-10]: 4 __ Why: Valid observation that Math.toIntExact throws ArithmeticException and the negative check is unreachable. However, the outer catch still handles the exception and writes to errorPointer, so functional impact is minor.	Low
	Guard packed ord/row bit layout invariant Packing row into the low 20 bits limits size to 2^20 = 1,048,576, but BATCH_SIZE is only 65,536 today so this is safe now — however, encoding also assumes the ord fits in the remaining 44 bits. More critically, if BATCH_SIZE is ever raised above 1M, rows will silently collide with ords. Add an assertion or use higher bit-shift (e.g., 32) to make the invariant explicit and future-proof. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/DocValuesBatchSource.java [271-273] +assert size <= (1 << 20) : "row index does not fit in packed ord/row scratch"; for (int i = 0; i < size; i++) { ordRowScratch[i] = ((long) (ordScratch[i] + 1) << 20) | i; } Suggestion importance[1-10]: 4 __ Why: A defensive assertion for future-proofing the bit-packing invariant. Current BATCH_SIZE (65,536) fits well within 20 bits, so this is a minor hardening suggestion.	Low
	Fix lost cancel when callback unregistered self.cancelled.swap(true, ...) == false is idiomatic-inverted; more importantly, if cancel_callback() returns None on the first call, we've set cancelled=true but never invoked cancellation. Later registration of a cancel callback would then be silently skipped. Consider only setting the flag when the callback ran, or checking callback availability first. sandbox/plugins/analytics-backend-datafusion/rust/src/arrow_batch_source/callbacks.rs [145-151] pub fn cancel(&self) { - if self.cancelled.swap(true, Ordering::AcqRel) == false { + if !self.cancelled.swap(true, Ordering::AcqRel) { if let Some(callback) = cancel_callback() { unsafe { callback(self.binding_id, self.source_key) }; } } } Suggestion importance[1-10]: 4 __ Why: The improved_code only changes == false to ! idiom without addressing the actual concern raised (setting the flag without calling callback). The suggestion identifies a real edge case but the improved code does not fix it.	Low
	Document registry-less deserialization constraint Record component evaluation order in Java is left-to-right for the constructor arguments, but nested method calls inside new are evaluated left-to-right at the call site — this is fine. However, fromBytes reconstructs a plan without a NamedWriteableRegistry, and ArrowBatchSourcePlan's StreamInput constructor cannot deserialize any nested named-writeables. This is correct only if ArrowBatchSourcePlan never contains them; consider asserting or documenting this constraint to prevent silent misuse. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneFragmentWirePlan.java [38-44] static LuceneFragmentWirePlan fromBytes(byte[] bytes) { + // Note: ArrowBatchSourcePlan must not contain NamedWriteables since no registry is threaded here. try (StreamInput input = StreamInput.wrap(bytes)) { LuceneFragmentWirePlan plan = new LuceneFragmentWirePlan( input.readStringList(), readOptionalBytes(input), input.readOptionalWriteable(ArrowBatchSourcePlan::new) ); Suggestion importance[1-10]: 3 __ Why: Adding a comment about a constraint is a minor documentation improvement; it does not fix a bug and the current code works correctly for the intended usage.	Low
	Clarify shutdown ordering for native bindings On failure, session::close runs before registration::close. If the session's native side still holds a reference to the callback binding (via the registered Arrow batch source provider), releasing the registration after the session is closed is fine, but any in-flight native call could touch the binding after the Java-side registration is closed. Consider closing registration last only if you're sure no native thread still references it after session::close completes; otherwise reverse the order or add a synchronization barrier. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DatafusionArrowBatchSourceExecutor.java [81-96] } catch (RuntimeException | Error throwable) { if (shardTask != null) { shardTask.clearCancellationListener(); } if (taskId != 0L) { closeAfterFailure(throwable, () -> NativeBridge.cancelQuery(taskId)); } if (output != null) { closeAfterFailure(throwable, output::close); } if (session != null) { closeAfterFailure(throwable, session::close); } + // registration closed after session so native side cannot re-enter released bindings closeAfterFailure(throwable, registration::close); throw throwable; } Suggestion importance[1-10]: 2 __ Why: The suggestion only adds a comment and asks for verification of the ordering, providing no concrete fix.	Low

Suggestions up to commit b00a2ac

Category	Suggestion	Impact
Possible issue	Fix ord/row bit-packing to avoid truncation The 20-bit packing of the row index into ordRowScratch is inconsistent with BATCH_SIZE = 65_536, which requires only 16 bits, but more importantly it silently truncates ordinals larger than (1 << 43) - 2 and rows above 1 << 20. Since BATCH_SIZE is a compile-time constant that could easily be increased (or an ord could exceed the implicit budget in unusual segments), the packing should use bit widths derived from the actual maxima (e.g. shift by 32 and mask with 0xFFFFFFFFL) to avoid a data-corruption bug. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/DocValuesBatchSource.java [271-280] for (int i = 0; i < size; i++) { - ordRowScratch[i] = ((long) (ordScratch[i] + 1) << 20) | i; + ordRowScratch[i] = ((long) (ordScratch[i] + 1) << 32) | (i & 0xFFFFFFFFL); } Arrays.sort(ordRowScratch, 0, size); int lastOrd = Integer.MIN_VALUE; BytesRef term = null; for (int i = 0; i < size; i++) { - int row = (int) (ordRowScratch[i] & 0xFFFFF); - int ord = (int) (ordRowScratch[i] >>> 20) - 1; + int row = (int) (ordRowScratch[i] & 0xFFFFFFFFL); + int ord = (int) (ordRowScratch[i] >>> 32) - 1; Suggestion importance[1-10]: 8 __ Why: Valid correctness concern: with BATCH_SIZE = 65_536, rows can be up to 65535 which fits in 16 bits but the code uses 20 bits leaving 44 bits for ord+1 - this is actually sufficient for practical ordinals. However, the suggestion's 32/32 split is safer and future-proof against BATCH_SIZE increases, and mirrors more standard bit-packing patterns.	Medium
	Close source factory on synchronous failure The SPI contract states ownership of sourceFactory transfers to executeArrowBatchSource including on failure, but if the call itself throws synchronously before taking ownership (e.g., NPE, IllegalStateException), sourceFactory will leak (holds an incRef on the reader). Wrap the call so the factory is closed on synchronous throw. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneSearchExecEngine.java [86-99] DocValuesBatchSourceFactory sourceFactory = new DocValuesBatchSourceFactory( state.searcher(), state.filterQuery(), sourcePlan.inputColumns(), allocator, context.getTask() ); -return arrowSourceBackend.executeArrowBatchSource( - allocator, - sourcePlan, - sourceFactory, - context.getTask(), - context.getDelegationThreadTracker() -); +boolean handedOff = false; +try { + EngineResultStream stream = arrowSourceBackend.executeArrowBatchSource( + allocator, + sourcePlan, + sourceFactory, + context.getTask(), + context.getDelegationThreadTracker() + ); + handedOff = true; + return stream; +} finally { + if (handedOff == false) { + sourceFactory.close(); + } +} Suggestion importance[1-10]: 7 __ Why: Valid concern: the SPI transfers ownership on entry, so a synchronous throw before ownership transfer could leak the factory and its reader ref, causing resource leaks.	Medium
	Ensure paired reader decRef on source close Each open() calls incRef() on the reader, but DocValuesBatchSource must be responsible for calling the matching decRef() on its close. If the source class does not perform that decrement (only the factory decRefs on its own close), the reader will leak a ref count per opened source. Verify DocValuesBatchSource.close() calls searcher.getIndexReader().decRef(), otherwise this incRef will leak references on every open. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/DocValuesBatchSourceFactory.java [74-84] searcher.getIndexReader().incRef(); boolean success = false; try { + // NOTE: DocValuesBatchSource.close() must call decRef() on the reader to + // release the reference acquired here. DocValuesBatchSource source = new DocValuesBatchSource(searcher, weight, projected, allocator, task); success = true; return source; } finally { if (success == false) { searcher.getIndexReader().decRef(); } } Suggestion importance[1-10]: 5 __ Why: Highlights a real concern about ref-count balance but only asks the author to verify behavior in a class not shown in the diff, limiting confidence.	Low
General	Validate negative length before narrowing conversion Math.toIntExact already throws ArithmeticException for values that exceed Integer.MAX_VALUE, but it does not reject negative long inputs — a negative projectionLength will convert to a negative int and only then hit the length < 0 check. However, if projectionLength is a large negative value like Long.MIN_VALUE, toIntExact will throw before your explicit check. Consider validating negativity on the long value first for clarity and consistent error messages. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/arrow/ArrowBatchSourceCallbacks.java [104-107] -int length = Math.toIntExact(projectionLength); -if (length < 0) { +if (projectionLength < 0L) { throw new IllegalArgumentException("projection length must be non-negative"); } +int length = Math.toIntExact(projectionLength); Suggestion importance[1-10]: 3 __ Why: Minor code style improvement for clearer error messages. Math.toIntExact handles overflow but the negative check ordering is a small readability concern with negligible functional impact.	Low
	Fix boolean comparison in cancel guard swap(...) == false uses the previous value: when previous was false (not yet cancelled), the expression is true, so cancel fires — this is correct, but the code reads as inverted logic. More importantly, swap returns the prior value; the intended guard should ensure cancel is only invoked on the first call. Use !self.cancelled.swap(true, Ordering::AcqRel) for readability, and consider that the current comparison-to-false pattern is a Clippy warning (bool_comparison). sandbox/plugins/analytics-backend-datafusion/rust/src/arrow_batch_source/callbacks.rs [146-151] pub fn cancel(&self) { - if self.cancelled.swap(true, Ordering::AcqRel) == false { + if !self.cancelled.swap(true, Ordering::AcqRel) { if let Some(callback) = cancel_callback() { unsafe { callback(self.binding_id, self.source_key) }; } } } Suggestion importance[1-10]: 3 __ Why: Minor readability/style improvement replacing == false with !; the existing logic is functionally correct.	Low
	Avoid duplicate doc-values iterator creation numericValues is called twice, opening two independent NumericDocValues iterators against the same field. Both values and validity share the same underlying iterator API which advances state, so decodeNumeric correctly uses one for bulk longValues and one for per-doc advanceExact. However, sortedNumericValues unwrapping in numericValues will throw for multi-valued sorted-numeric fields, and calling it twice doubles the exception risk and I/O cost. Consider fetching the values once and deriving validity from the same iterator, or documenting why two iterators are necessary. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/DocValuesBatchSource.java [407-417] -return switch (column.kind()) { - case KEYWORD, IP -> new SortedColumn(sortedValues(leaf, column, type), column.kind()); - case BINARY -> type == DocValuesType.BINARY - ? new BinaryColumn(leaf.reader().getBinaryDocValues(column.name())) - : new SortedColumn(sortedValues(leaf, column, type), column.kind()); - case LONG, TIMESTAMP, BOOLEAN, FLOAT, DOUBLE -> new NumericColumn( - numericValues(leaf, column.name(), type), - numericValues(leaf, column.name(), type), - column.kind() - ); -}; +case LONG, TIMESTAMP, BOOLEAN, FLOAT, DOUBLE -> { + NumericDocValues valuesIter = numericValues(leaf, column.name(), type); + NumericDocValues validityIter = numericValues(leaf, column.name(), type); + yield new NumericColumn(valuesIter, validityIter, column.kind()); +} Suggestion importance[1-10]: 2 __ Why: The two iterators are intentional - one is consumed by longValues bulk API and the other used for per-doc advanceExact validity checks. The suggested change doesn't actually improve anything since two independent iterators are still needed.	Low
	Clarify factory ownership on registration failure ArrowBatchSourceCallbacks.register takes ownership of the sourceFactory on success (it will close it when the registration closes). On failure of register itself, we correctly close the factory here. However, if register succeeds and later cleanup runs, registration.close() triggers Binding.tryFinishClose() which calls factory.close() — good. But if register throws AFTER partially populating internal state, the factory could be double-closed. Consider making factory.close() idempotent or documenting the ownership handoff explicitly. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DatafusionArrowBatchSourceExecutor.java [53-59] ArrowBatchSourceCallbacks.Registration registration; try { registration = ArrowBatchSourceCallbacks.register(sourceFactory, threadTracker); } catch (RuntimeException | Error throwable) { + // register() did not take ownership; close the factory here. closeAfterFailure(throwable, sourceFactory::close); throw throwable; } Suggestion importance[1-10]: 2 __ Why: The improved code is essentially identical to the existing code with just an added comment. Low impact suggestion that only adds a documentation comment without actual code change.	Low
	Avoid redundant byte-array cloning fromBytes decodes filterBytes without a NamedWriteableRegistry, but the outer filterQuery(registry) uses one later. That's fine — however, calling new LuceneFragmentWirePlan(...) here invokes the compact constructor which clones filterBytes, then the record's filterBytes() accessor clones again on each call. Consider inlining without the extra clone in fromBytes (e.g., store the bytes directly), since deserialization already owns the buffer, to avoid unnecessary allocations on every shard. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneFragmentWirePlan.java [39-44] try (StreamInput input = StreamInput.wrap(bytes)) { - LuceneFragmentWirePlan plan = new LuceneFragmentWirePlan( - input.readStringList(), - readOptionalBytes(input), - input.readOptionalWriteable(ArrowBatchSourcePlan::new) - ); + List<String> names = input.readStringList(); + byte[] filter = readOptionalBytes(input); + ArrowBatchSourcePlan arrow = input.readOptionalWriteable(ArrowBatchSourcePlan::new); + if (input.available() != 0) { + throw new IllegalStateException("Unexpected trailing Lucene fragment bytes"); + } + return new LuceneFragmentWirePlan(names, filter, arrow); Suggestion importance[1-10]: 2 __ Why: Minor micro-optimization; the compact constructor still clones so the suggested change alone doesn't eliminate the clone, offering limited value.	Low

[github-actions]

Persistent review updated to latest commit b7a7333

[github-actions]

✅ Gradle check result for b7a7333: SUCCESS

[codecov]

Codecov Report

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

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #22852      +/-   ##
============================================
+ Coverage     71.53%   71.63%   +0.10%     
- Complexity    77249    77400     +151     
============================================
  Files          6170     6170              
  Lines        359710   359710              
  Branches      52460    52460              
============================================
+ Hits         257318   257686     +368     
+ Misses        81965    81629     -336     
+ Partials      20427    20395      -32     

☔ 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 89e0da6

[github-actions]

PR Code Analyzer ❗

AI-powered 'Code-Diff-Analyzer' found issues on commit ecd8739.

⛔ Hard block: Issues at Medium severity or above will block this PR from merging.

Path	Line	Severity	Description
sandbox/libs/analytics-framework/licenses/jackson-core-2.22.2.jar.sha1	1	high	Dependency version change: jackson-core bumped from 2.22.1 to 2.22.2. New SHA1 is 13a748ea3e329fa220076e021b45c8391b32420c. Artifact authenticity cannot be verified from this diff alone; maintainers must confirm the new artifact matches the expected release from the upstream registry.
sandbox/libs/analytics-framework/licenses/jackson-databind-2.22.2.jar.sha1	1	high	Dependency version change: jackson-databind bumped from 2.22.1 to 2.22.2. New SHA1 is 921bd2092b0c539b2876de7063d55c72edcd05d3. Maintainers must verify the artifact against the official release.
sandbox/plugins/analytics-backend-datafusion/licenses/jackson-datatype-jdk8-2.22.2.jar.sha1	1	high	Dependency version change: jackson-datatype-jdk8 bumped from 2.22.1 to 2.22.2. New SHA1 is 0b900dd7125fa16cfdf46135d3ffb3243d0f8b88. Maintainers must verify the artifact against the official release.
sandbox/plugins/analytics-engine/licenses/jackson-dataformat-yaml-2.22.2.jar.sha1	1	high	Dependency version change: jackson-dataformat-yaml bumped from 2.22.1 to 2.22.2. New SHA1 is a4f075bf4cc1ee814ab98d69c1612786c2f42bc3. Maintainers must verify the artifact against the official release.

The table above displays the top 10 most important findings.

Total: 4 | Critical: 0 | High: 4 | Medium: 0 | Low: 0

---

Pull Requests Author(s): Please update your Pull Request according to the report above.

Repository Maintainer(s): You can bypass diff analyzer by adding label skip-diff-analyzer after reviewing the changes carefully, then re-run failed actions. To re-enable the analyzer, remove the label, then re-run all actions.

---

⚠️ Note: The Code-Diff-Analyzer helps protect against potentially harmful code patterns. Please ensure you have thoroughly reviewed the changes beforehand.

Thanks.

[github-actions]

Persistent review updated to latest commit ecd8739

[github-actions]

✅ Gradle check result for ecd8739: SUCCESS