[Analytics Engine] Wire broadcast join end-to-end via DataFusion backend

[LantaoJin]

Description

Lands the production path for broadcast-join (milestone 1) execution on the DataFusion backend on top
of PR #21639 (distributed join planning). The JoinStrategyAdvisor already picked BROADCAST when one join side fit inside the broadcast gates, but DefaultPlanExecutor fell back to coordinator-centric.
This PR replaces that fallback with a two-pass dispatcher that:

1. Pass 1: build only. Runs the BROADCAST_BUILD child stage against a new BroadcastCaptureSink that buffers its Arrow batches into a single Arrow-IPC byte buffer.
2. Pass 2: probe + root. Appends a BroadcastInjectionInstructionNode carrying the IPC bytes to every probe-stage plan alternative, then dispatches through the normal QueryScheduler path. Each probe data node decodes the payload, registers it as a MemTable under "broadcast-<buildStageId>" on its shard-scan SessionContextHandle, and the native engine executes Join(ShardScan, NamedScan("broadcast-<buildStageId>")).

Scope: The join runs on each probe data node in parallel, against its local shards plus the injected memtable. The coordinator only gathers joined rows -- there is no coord-side join in this M1 path.

End-to-end verified. BroadcastJoinIT runs INNER and LEFT OUTER broadcast joins on a 2-node cluster against parquet-backed indices and asserts row-multiset parity with the coord-centric baseline plus a strategy-counter delta proving BROADCAST actually fired.

PR#21639 landed a Volcano split-rule architecture that changed planner-shape contracts our
M0 work was built on. This PR rebases on top of it and reconciles:

M0/M1 <--> PR#21639 reconciliation

• OpenSearchHashJoinRule removed. M0's HASH-shuffle Volcano rule was incompatible with PR#21639's split-rule design — registering it caused a memo explosion (verified empirically: PlanShapeTests.testJoinThenSort_2shard hit a 16-minute suite timeout before this fix). M2 hash-shuffle support needs to be redesigned as a sibling split rule. Tracked as a follow-up.
• ExchangeInfo reverted to PR#21639's 2-field shape. M0 added partitionCount for the M2 hash-shuffle work; auto-generated record toString mismatch broke ~11 PR#21639-introduced *PlanShapeTests. The field comes back when M2 hash-shuffle lands.
• OpenSearchJoin SEMI/ANTI fix preserved. M0 patched getOutputFieldStorage() to return left-only storage for SEMI/ANTI; PR#21639's reset of the file dropped that. Re-applied (if (getJoinType().projectsRight())).
• CapabilityRegistry join indexing removed. M0's per-format Equi/Theta indexing was built on the M0 JoinCapability API; PR#21639 replaced that API with a simpler record JoinCapability(Set<JoinKind>, Set<String>). The indexing was dead code under the new API.
• Stale fixtures updated. Six tests asserted M0's coord-centric shape onshardCount=1; under PR#21639's split rule, single-shard same-table joins go SHARD-local (no ERs, no separate child stages). Switched fixtures to shardCount=2 or 3 so they exercise the COORDINATOR-localized path they were designed for.
• Theta-join tests now assert PR#21639's failure mode. PR#21639 rejects non-equi joins at the rule level (!info.isEqui() → unmarked LogicalJoin survives, Volcano's trait converter crashes). The M0/M1 tests that asserted "theta routes coord-centric" now use expectThrows(RuntimeException.class, ...) to pin the failure mode. M2 follow-up: re-enable theta joins through a coord-centric fallback consistent with the new split-rule architecture.

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 Code Analyzer ❗

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

Path	Line	Severity	Description
sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/join/BroadcastDispatch.java	284	medium	Reflection is used to invoke `ipcBytesFuture()` on an arbitrary ExchangeSink implementation by method name string. While the author documents this as an intentional architectural decision to avoid circular module dependencies, reflective invocation by name is generally fragile and could be exploited if an attacker-controlled ExchangeSink implementation is ever reachable through the factory parameter. The unchecked cast on the return value also bypasses generics-level type safety.
sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/action/TransportAnalyticsShuffleDataAction.java	62	low	ShuffleBufferManager.getOrCreateBuffer() creates new buffers keyed by attacker-supplied (queryId, targetStageId, partitionIndex) values from incoming transport requests. There is no apparent eviction or cleanup path for orphaned buffers in this diff, which could allow an authorized but misbehaving node to exhaust coordinator memory by issuing requests with many distinct queryId values without ever completing the buffers.
sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/rest/RestJoinStrategyStatsAction.java	46	low	A new unauthenticated REST endpoint GET /_analytics/_strategies is registered that exposes internal per-node dispatch counters. While the data itself is non-sensitive (strategy counts), adding new REST endpoints without explicit access-control annotations may bypass or circumvent cluster-level security policies depending on the deployment's security plugin configuration.

The table above displays the top 10 most important findings.

Total: 3 | Critical: 0 | High: 0 | Medium: 1 | Low: 2

---

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.

[LantaoJin]

#21677 (comment)

Real bug: fixed
Finding 5 (Low)
Finding 1 (Medium)

False alarm: no action
Finding 2 (Medium)
Finding 3 (Medium)
Finding 4 (Low)

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 11e4c7a

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Fix backpressure race condition The backpressure check has a race condition. Between addAndGet(size) and the cap check, another thread could push currentBytes over maxBytes, causing both threads to accept data that collectively exceeds the cap. Use getAndAdd with a post-check or a CAS loop to enforce the cap atomically. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/shuffle/ShuffleBufferManager.java [108-122] public boolean tryAddData(String side, byte[] data) { int size = data == null ? 0 : data.length; - long newTotal = currentBytes.addAndGet(size); + long prev = currentBytes.getAndAdd(size); + long newTotal = prev + size; if (newTotal > maxBytes) { currentBytes.addAndGet(-size); rejectedCount.incrementAndGet(); return false; } if ("left".equals(side)) { leftData.add(data); } else { rightData.add(data); } return true; } Suggestion importance[1-10]: 9 __ Why: Critical race condition in tryAddData. Between addAndGet(size) and the cap check, concurrent threads can push currentBytes over maxBytes, allowing multiple threads to accept data that collectively exceeds the cap. Using getAndAdd with a post-check ensures atomic enforcement of the byte cap.	High
	Protect allocator close with synchronization The allocator close operation happens outside the synchronized block. If two threads call close() concurrently on the same context instance, both could read !closed as true before either sets closed = true, causing both to attempt closing the allocator. While Arrow's allocator close may handle this, the pattern is fragile. Consider moving the allocator close inside the synchronized block or using a compare-and-set pattern. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/QueryContext.java [206-213] -boolean closeAllocator; synchronized (sharedState) { - closeAllocator = !closed && ownsAllocator; - closed = true; -} -if (closeAllocator) { - allocator.close(); + if (closed) { + // Already closed by this instance, skip allocator close + } else { + closed = true; + if (ownsAllocator) { + allocator.close(); + } + } } Suggestion importance[1-10]: 7 __ Why: The suggestion identifies a potential race condition where two threads could both attempt to close the allocator if they read !closed before either sets closed = true. Moving the allocator close inside the synchronized block would eliminate this race. This is a valid concurrency issue that could lead to double-close attempts on the allocator, though the impact depends on Arrow's allocator implementation.	Medium
General	Prevent exponential backoff overflow The left-shift can overflow when attempt is large (e.g. attempt=64 shifts by 63 bits, wrapping to negative). Cap the shift operand to prevent overflow before the Math.min clamp, ensuring backoff remains positive and bounded. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/shuffle/ShuffleSenderRetry.java [90] -long backoff = Math.min(DEFAULT_MAX_BACKOFF_MILLIS, initialBackoffMillis << (attempt - 1)); +int shiftAmount = Math.min(attempt - 1, 30); +long backoff = Math.min(DEFAULT_MAX_BACKOFF_MILLIS, initialBackoffMillis << shiftAmount); Suggestion importance[1-10]: 7 __ Why: The left-shift initialBackoffMillis << (attempt - 1) can overflow when attempt is large (e.g., attempt=64 shifts by 63 bits), wrapping to negative values. Capping the shift operand prevents overflow and ensures backoff remains positive and bounded.	Medium
	Validate null-data shuffle requests When data is null and isLast is true, the handler marks the sender done without adding any data. If a sender never sends data (only an isLast marker), the buffer's currentBytes remains at zero but completion is signaled. Verify that zero-data senders are intentional; if not, reject requests where data==null && !isLast. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/action/TransportAnalyticsShuffleDataAction.java [52-72] ShuffleBufferManager.ShuffleBuffer buffer = shuffleBufferManager.getOrCreateBuffer( request.getQueryId(), request.getTargetStageId(), request.getPartitionIndex() ); +if (request.getData() == null && !request.isLast()) { + listener.onFailure(new IllegalArgumentException("Shuffle request with null data must have isLast=true")); + return; +} if (request.getData() != null) { boolean accepted = buffer.tryAddData(request.getSide(), request.getData()); if (!accepted) { ... listener.onResponse(AnalyticsShuffleDataResponse.backpressureReject()); return; } } if (request.isLast()) { buffer.senderDone(request.getSide()); ... } Suggestion importance[1-10]: 6 __ Why: When data is null and isLast is false, the handler silently skips adding data without marking completion. This could indicate a protocol violation. Rejecting such requests explicitly prevents silent failures and clarifies the expected wire contract.	Low
	Preserve timeout exception type The extractIpcBytes method can throw TimeoutException when the future doesn't complete within the timeout window. The current catch block wraps all throwables in a generic RuntimeException, losing the specific timeout context. Consider preserving the original exception type or adding specific handling for TimeoutException to provide clearer diagnostics when the capture sink violates its contract. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/join/BroadcastDispatch.java [136-142] byte[] ipcBytes; try { ipcBytes = extractIpcBytes(captureSink); +} catch (TimeoutException te) { + LOGGER.warn("[BroadcastDispatch] capture sink did not complete ipcBytesFuture within timeout", te); + terminal.onFailure(te); + return; } catch (Throwable t) { LOGGER.warn("[BroadcastDispatch] failed to extract IPC bytes from capture sink", t); terminal.onFailure(new RuntimeException("BroadcastDispatch: capture sink did not produce IPC bytes", t)); return; } Suggestion importance[1-10]: 5 __ Why: The suggestion correctly identifies that extractIpcBytes can throw TimeoutException and proposes preserving the exception type for clearer diagnostics. However, the existing code already wraps all throwables in a RuntimeException with a descriptive message, which is a common pattern. The improvement is marginal since the timeout context is already logged via the LOGGER.warn call.	Low
	Protect late-install callback replay The late-install replay runs the callback inline on the caller's thread. If the callback blocks or throws, it can stall the dispatcher. Wrap the replay in a try-finally or schedule it asynchronously to prevent blocking the install path, ensuring the method returns promptly even if the callback misbehaves. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/task/AnalyticsQueryTask.java [95-107] public void setOnCancelCallback(Runnable callback) { onCancelCallback.set(callback); if (callback != null && isCancelled()) { try { callback.run(); } catch (Exception e) { logger.warn( new ParameterizedMessage("[AnalyticsQueryTask] late-install onCancel callback failed for queryId={}", queryId), e ); + } finally { + // Ensure method returns even if callback throws } } } Suggestion importance[1-10]: 5 __ Why: The late-install replay runs the callback inline on the caller's thread. If the callback blocks or throws, it can stall the dispatcher. The existing try-catch already logs exceptions, but wrapping in try-finally or scheduling asynchronously would further protect the install path from misbehaving callbacks.	Low
	Check cancellation before sink close The cancel-after-success race guard checks parentTask.isCancelled() after pass 1 succeeds but before starting pass 2. However, the capture sink has already been closed at this point, and if the task was cancelled, the IPC bytes extraction may have already consumed resources. Consider moving this cancellation check earlier, immediately after the build execution transitions to SUCCEEDED and before closing the capture sink, to avoid unnecessary work. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/join/BroadcastDispatch.java [148-153] +// Guard against the cancel-after-success race before closing the sink if (parentTask != null && parentTask.isCancelled()) { String reason = parentTask.getReasonCancelled() != null ? parentTask.getReasonCancelled() : "unknown"; LOGGER.debug("[BroadcastDispatch] task cancelled after build SUCCEEDED, reason={}; aborting pass 2", reason); terminal.onFailure(new org.opensearch.core.tasks.TaskCancelledException("query cancelled: " + reason)); return; } +// P1 — close the capture sink... +try { + captureSink.close(); +} catch (Throwable closeErr) { + ... +} + Suggestion importance[1-10]: 4 __ Why: The suggestion proposes moving the cancellation check before closing the capture sink to avoid unnecessary work. While this is a minor optimization, the current code is correct and the sink close is a lightweight operation. The improvement is marginal and the existing order (close sink, then check cancellation) is defensible since the sink must be closed regardless of cancellation status to release resources.	Low
	Clarify byte counter for dropped batches When the byte cap is exceeded, the batch is closed and dropped, but the accumulated bytes counter is still incremented. This means the error message in close() will report a total that includes dropped batches, which could be misleading for diagnostics. Consider either not incrementing the counter for dropped batches, or clearly documenting that the reported size includes rejected data. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/BroadcastCaptureSink.java [119-124] if (exceeded || accumulatedBytes + batchBytes > maxBytes) { - exceeded = true; - accumulatedBytes += batchBytes; + if (!exceeded) { + exceeded = true; + // Record the first-exceeding batch size for error reporting + accumulatedBytes += batchBytes; + } batch.close(); return; } Suggestion importance[1-10]: 3 __ Why: The suggestion addresses a potential confusion in error reporting where the accumulated bytes counter includes dropped batches. The proposed change only increments the counter once when the cap is first exceeded, which could make error messages clearer. However, the current behavior (tracking the total would-be size) is also reasonable for diagnostics, and the improvement is minor.	Low

---

Previous suggestions

Suggestions up to commit 568a099

Category	Suggestion	Impact
Possible issue	Close capture sink on cancel The cancel-after-success race guard checks parentTask.isCancelled() but does not close the captureSink before returning. If cancellation lands between build SUCCEEDED and this check, the sink remains open, leaking its Arrow allocator buffers. Close the sink in this path to match the cleanup done in the FAILED and CANCELLED state-listener branches. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/join/BroadcastDispatch.java [148-153] if (parentTask != null && parentTask.isCancelled()) { String reason = parentTask.getReasonCancelled() != null ? parentTask.getReasonCancelled() : "unknown"; LOGGER.debug("[BroadcastDispatch] task cancelled after build SUCCEEDED, reason={}; aborting pass 2", reason); + try { + captureSink.close(); + } catch (Throwable ignore) { + // Best-effort cleanup before surfacing the cancellation. + } terminal.onFailure(new org.opensearch.core.tasks.TaskCancelledException("query cancelled: " + reason)); return; } Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies a resource leak: when cancellation lands after build SUCCEEDED but before pass 2 starts, captureSink is not closed, leaking Arrow allocator buffers. The fix matches the cleanup pattern in the FAILED/CANCELLED branches and is critical for memory safety.	Medium
	Close sink on extraction failure When extractIpcBytes throws, the dispatcher calls terminal.onFailure and returns without closing captureSink. The sink's Arrow buffers remain allocated, leaking memory. Add a try-finally around the extraction so the sink is closed even when the future-get fails or times out. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/join/BroadcastDispatch.java [136-142] byte[] ipcBytes; try { ipcBytes = extractIpcBytes(captureSink); } catch (Throwable t) { LOGGER.warn("[BroadcastDispatch] failed to extract IPC bytes from capture sink", t); + try { + captureSink.close(); + } catch (Throwable ignore) { + // Best-effort cleanup before surfacing the primary error. + } terminal.onFailure(new RuntimeException("BroadcastDispatch: capture sink did not produce IPC bytes", t)); return; } Suggestion importance[1-10]: 8 __ Why: Valid resource-leak fix. When extractIpcBytes throws, the dispatcher returns without closing captureSink, leaking its Arrow buffers. The suggested try-finally ensures cleanup on all failure paths, matching the pattern used in the state-listener branches.	Medium
	Validate schemaIpc before native call The method validates pointer and array length but does not validate schemaIpc for null or empty. If schemaIpc is null, call.bytes(schemaIpc) will throw NPE. If empty, the Rust side's schema_from_ipc_bytes may fail with a cryptic error. Add explicit validation before the native call. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/nativelib/NativeBridge.java [907-934] public static long registerMemtableOnSessionContext( long sessionContextHandlePtr, String inputId, byte[] schemaIpc, long[] arrayPtrs, long[] schemaPtrs ) { NativeHandle.validatePointer(sessionContextHandlePtr, "sessionContextHandle"); + if (schemaIpc == null || schemaIpc.length == 0) { + throw new IllegalArgumentException("schemaIpc must not be null or empty"); + } if (arrayPtrs.length != schemaPtrs.length) { throw new IllegalArgumentException( "arrayPtrs.length (" + arrayPtrs.length + ") != schemaPtrs.length (" + schemaPtrs.length + ")" ); } ... } Suggestion importance[1-10]: 5 __ Why: The suggestion correctly identifies that schemaIpc is not validated for null or empty before being passed to the native call. Adding this validation would prevent a potential NPE and provide clearer error messages. This is a reasonable defensive programming improvement, though the impact is moderate since the caller (BroadcastInjectionHandler) is expected to provide valid input.	Low
	Fix callback replay race condition The late-install replay logic has a race condition. Between checking isCancelled() and running callback.run(), another thread could invoke onCancelled() and run the old callback. This creates a window where the callback runs twice or not at all. Use atomic compare-and-set or synchronization to ensure exactly-once semantics. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/task/AnalyticsQueryTask.java [95-107] -public void setOnCancelCallback(Runnable callback) { +public synchronized void setOnCancelCallback(Runnable callback) { onCancelCallback.set(callback); if (callback != null && isCancelled()) { try { callback.run(); } catch (Exception e) { logger.warn( new ParameterizedMessage("[AnalyticsQueryTask] late-install onCancel callback failed for queryId={}", queryId), e ); } } } Suggestion importance[1-10]: 3 __ Why: The suggestion identifies a potential race condition in the late-install callback replay logic. However, adding synchronized to setOnCancelCallback may introduce contention if called frequently. The existing code already uses AtomicReference for onCancelCallback, and the race window is narrow. The suggestion is valid but the impact is moderate, as the race is unlikely in typical usage patterns.	Low
General	Add timeout to IndicesStats fetch The actionGet() call blocks indefinitely if the cluster is unresponsive. For a per-query statistics fetch on the SEARCH executor, this can hang the thread pool. Use actionGet(timeout) with a reasonable timeout (e.g. 5 seconds) to fail fast and fall back to the zero-row-count fail-safe. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/join/StatisticsCollector.java [88-116] private static Map<String, Long> fetchPrimaryDocCounts(Client client, Collection<String> indexNames) { if (indexNames.isEmpty()) { return Map.of(); } try { IndicesStatsRequest request = new IndicesStatsRequest(); request.indices(indexNames.toArray(new String[0])); request.clear(); request.docs(true); - IndicesStatsResponse response = client.admin().indices().stats(request).actionGet(); + IndicesStatsResponse response = client.admin().indices().stats(request).actionGet(java.time.Duration.ofSeconds(5)); ... } catch (Exception e) { LOGGER.warn(...); return Map.of(); } } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that actionGet() blocks indefinitely, which can hang the SEARCH executor thread if the cluster is unresponsive. Adding a timeout (e.g., 5 seconds) would allow the query to fail fast and fall back to the zero-row-count fail-safe, improving resilience. This is a practical improvement with moderate impact on query robustness.	Medium
	Fail future immediately on cap violation When the byte cap is exceeded, feed closes the incoming batch and returns without failing the future. The sink remains open and subsequent batches are silently dropped. The failure is only surfaced from close(), but if a sender never calls close() (e.g. a hung shard handler), the dispatcher waits indefinitely on ipcBytesFuture(). Fail the future immediately on the first cap violation so the dispatcher can route the error through the terminal listener without waiting for close(). sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/BroadcastCaptureSink.java [119-124] if (exceeded || accumulatedBytes + batchBytes > maxBytes) { exceeded = true; accumulatedBytes += batchBytes; batch.close(); + if (!ipcBytesFuture.isDone()) { + ipcBytesFuture.completeExceptionally(new BroadcastSizeExceededException(accumulatedBytes, maxBytes)); + } return; } Suggestion importance[1-10]: 6 __ Why: The suggestion improves responsiveness by failing the future immediately when the byte cap is exceeded, rather than waiting for close(). However, the current design is intentional: close() is the single point where the future is settled, and the dispatcher always calls close() before awaiting the future. Failing early would complicate the state machine (multiple completion paths) without a strong correctness benefit.	Low
	Cache build-side decision for consistency For INNER joins, when both sides fit the broadcast gates, the selector returns BROADCAST but does not record which side was chosen as the build. The caller (selectBuildSide) re-runs the same logic to pick the smaller side, but if the statistics change between the two calls (e.g. concurrent index updates), the build-side decision could flip. Cache the chosen build side in selectStrategy and return it alongside the strategy, or ensure selectBuildSide is called with the same statistics snapshot. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/join/JoinStrategySelector.java [103-111] +// Option: return a record (JoinStrategy, String buildSide) from selectStrategy +// so the caller doesn't re-run the selection logic. Alternatively, pass the +// statistics snapshot to selectBuildSide to ensure consistency. if ("either".equals(eligibleBuild)) { boolean leftFits = fitsBroadcastGates(leftShards, leftRows); boolean rightFits = fitsBroadcastGates(rightShards, rightRows); if (leftFits || rightFits) { + // Cache the chosen side here and expose it via a getter, or return + // a tuple (JoinStrategy, buildSide) so the caller doesn't re-decide. return JoinStrategy.BROADCAST; } ... } Suggestion importance[1-10]: 4 __ Why: The concern about statistics changing between selectStrategy and selectBuildSide is theoretical—both are called synchronously within the same advisor pass, and the statistics snapshot is immutable for that pass. The current design is intentional: selectStrategy decides the strategy, selectBuildSide decides the side, both using the same snapshot. Caching would add complexity without addressing a real race.	Low

Suggestions up to commit 8178ef0

Category	Suggestion	Impact
General	Add timeout to IndicesStats fetch The actionGet() call blocks indefinitely if the cluster is unresponsive or the request hangs. This can stall the SEARCH executor thread and cascade to query timeouts. Use actionGet(timeout) with a bounded wait (e.g., 5 seconds) so the method fails fast and returns the fail-safe empty map, preserving query responsiveness. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/join/StatisticsCollector.java [97] private static Map<String, Long> fetchPrimaryDocCounts(Client client, Collection<String> indexNames) { if (indexNames.isEmpty()) { return Map.of(); } try { IndicesStatsRequest request = new IndicesStatsRequest(); request.indices(indexNames.toArray(new String[0])); request.clear(); request.docs(true); - IndicesStatsResponse response = client.admin().indices().stats(request).actionGet(); + IndicesStatsResponse response = client.admin().indices().stats(request).actionGet(java.time.Duration.ofSeconds(5)); ... } catch (Exception e) { LOGGER.warn(...); return Map.of(); } } Suggestion importance[1-10]: 8 __ Why: Excellent catch. The unbounded actionGet() can block the SEARCH executor indefinitely if the cluster is unresponsive, cascading to query timeouts and thread exhaustion. Adding a bounded timeout (5s is reasonable for stats fetch) ensures fail-fast behavior and preserves the fail-safe rowCount=0 contract. High impact for production resilience.	Medium
	Log broadcast refusal for any unknown rows The log statement inside the conditional is only emitted when both sides have unknown row counts, but the method returns HASH_SHUFFLE unconditionally afterward. If only one side has unknown rows, the fallback happens silently without logging. Move the log outside the conditional or add a separate log for the single-unknown-side case to improve observability. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/join/JoinStrategySelector.java [108-111] -if (leftRows <= 0 && rightRows <= 0) { - LOGGER.info("Join strategy: row count unknown for both inner-join sides; refusing broadcast fail-safe"); +if (leftRows <= 0 || rightRows <= 0) { + LOGGER.info("Join strategy: row count unknown for at least one inner-join side; refusing broadcast fail-safe"); } return JoinStrategy.HASH_SHUFFLE; Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies that the log statement only fires when both sides have unknown rows, but the method falls back to HASH_SHUFFLE for any unknown-row case. Broadening the condition to log whenever at least one side is unknown improves observability without changing behavior.	Low
	Validate schema IPC before native call The method validates pointer and array lengths but does not validate that schemaIpc is non-null or non-empty before passing it to native code. A null or empty schema blob would cause a native-side error that surfaces as a cryptic runtime exception. Add an explicit null/empty check with a clear error message before the FFI call. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/nativelib/NativeBridge.java [907-937] public static long registerMemtableOnSessionContext( long sessionContextHandlePtr, String inputId, byte[] schemaIpc, long[] arrayPtrs, long[] schemaPtrs ) { NativeHandle.validatePointer(sessionContextHandlePtr, "sessionContextHandle"); + if (schemaIpc == null || schemaIpc.length == 0) { + throw new IllegalArgumentException("schemaIpc must be non-null and non-empty"); + } if (arrayPtrs.length != schemaPtrs.length) { throw new IllegalArgumentException( "arrayPtrs.length (" + arrayPtrs.length + ") != schemaPtrs.length (" + schemaPtrs.length + ")" ); } try (var call = new NativeCall()) { ... } } Suggestion importance[1-10]: 6 __ Why: Valid point that schemaIpc null/empty validation is missing, which could cause cryptic native errors. However, the caller (BroadcastInjectionHandler) is expected to provide valid IPC bytes from the coordinator's capture sink, so this is defensive programming rather than fixing a likely bug. The check improves error clarity at minimal cost.	Low
	Fail fast when byte cap exceeded When the byte cap is exceeded, subsequent batches are dropped and closed, but the sink continues accepting feed() calls until close() is invoked. This can silently discard data without immediate feedback to the caller. Consider throwing an exception on the first exceeded batch so the dispatcher fails fast rather than accumulating phantom byte counts. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/BroadcastCaptureSink.java [119-124] if (exceeded || accumulatedBytes + batchBytes > maxBytes) { exceeded = true; accumulatedBytes += batchBytes; batch.close(); - return; + throw new BroadcastSizeExceededException(accumulatedBytes, maxBytes); } Suggestion importance[1-10]: 4 __ Why: The suggestion proposes throwing immediately when the cap is exceeded, but the current design intentionally defers the failure to close() so the dispatcher can route it through the terminal listener. Throwing from feed() would require every caller to handle the exception, breaking the sink's contract. The design choice is deliberate, not a bug.	Low
	Prevent integer overflow in backoff The left-shift operation initialBackoffMillis << (attempt - 1) can overflow when attempt is large (e.g., attempt=8 shifts by 7, multiplying 50ms by 128 = 6400ms, which is safe, but the pattern is fragile). For attempt=31, the shift would overflow into negative territory. Cap the shift amount explicitly before the operation to prevent integer overflow. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/shuffle/ShuffleSenderRetry.java [90] -long backoff = Math.min(DEFAULT_MAX_BACKOFF_MILLIS, initialBackoffMillis << (attempt - 1)); +int shiftAmount = Math.min(attempt - 1, 30); +long backoff = Math.min(DEFAULT_MAX_BACKOFF_MILLIS, initialBackoffMillis << shiftAmount); Suggestion importance[1-10]: 4 __ Why: The concern about overflow is theoretically valid but practically unlikely given DEFAULT_MAX_ATTEMPTS=8 caps attempts well below overflow territory (shift by 7 yields 6400ms, safely under the cap). The suggestion adds defensive code that doesn't address a real bug in the current configuration, though it could prevent future issues if constants change.	Low
	Ensure sink cleanup on extraction failure When extractIpcBytes throws, the capture sink remains open and its buffers are not released. The sink was already closed earlier in the SUCCEEDED branch, but if extraction fails after close, the allocator resources may still be held. Ensure the sink is closed in a finally block or catch to prevent resource leaks. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/join/BroadcastDispatch.java [135-142] byte[] ipcBytes; try { ipcBytes = extractIpcBytes(captureSink); } catch (Throwable t) { LOGGER.warn("[BroadcastDispatch] failed to extract IPC bytes from capture sink", t); + try { + captureSink.close(); + } catch (Throwable ignore) {} terminal.onFailure(new RuntimeException("BroadcastDispatch: capture sink did not produce IPC bytes", t)); return; } Suggestion importance[1-10]: 3 __ Why: The suggestion is partially valid but misses that captureSink.close() was already called at line 128 before extractIpcBytes is invoked. The sink is already closed at this point, so adding another close in the catch block is redundant defensive code rather than fixing an actual leak.	Low
Possible issue	Close capture sink on cancel race The cancel-after-success race guard checks task cancellation but doesn't close the capture sink before returning. This leaks the IPC bytes buffer and any allocator resources held by the sink. Add captureSink.close() in a try-catch before the early return to ensure cleanup. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/join/BroadcastDispatch.java [148-153] if (parentTask != null && parentTask.isCancelled()) { String reason = parentTask.getReasonCancelled() != null ? parentTask.getReasonCancelled() : "unknown"; LOGGER.debug("[BroadcastDispatch] task cancelled after build SUCCEEDED, reason={}; aborting pass 2", reason); + try { + captureSink.close(); + } catch (Throwable ignore) { + // Best-effort cleanup before failing terminal listener + } terminal.onFailure(new org.opensearch.core.tasks.TaskCancelledException("query cancelled: " + reason)); return; } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies a resource leak when the task is cancelled after build success. The captureSink should be closed before returning to prevent buffer leaks. However, the sink was already closed earlier in the SUCCEEDED branch (line 128), so this is a defensive cleanup for the race window.	Medium
	Prevent callback double-execution race The late-install replay logic creates a race condition. Between checking isCancelled() and running callback.run(), another thread could invoke onCancelled() and execute the same callback, causing it to run twice. Synchronize the check-and-run block with the onCancelled() method to ensure exactly-once execution semantics. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/task/AnalyticsQueryTask.java [95-106] public void setOnCancelCallback(Runnable callback) { onCancelCallback.set(callback); - if (callback != null && isCancelled()) { - try { - callback.run(); - } catch (Exception e) { - logger.warn( - new ParameterizedMessage("[AnalyticsQueryTask] late-install onCancel callback failed for queryId={}", queryId), - e - ); + synchronized (this) { + if (callback != null && isCancelled()) { + try { + callback.run(); + } catch (Exception e) { + logger.warn( + new ParameterizedMessage("[AnalyticsQueryTask] late-install onCancel callback failed for queryId={}", queryId), + e + ); + } } } } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies a race condition where the callback could execute twice if onCancelled() runs between the isCancelled() check and callback.run(). However, the fix requires coordination with onCancelled() method (not shown in the diff), and the impact depends on whether callbacks are idempotent. The synchronization approach is sound but may need broader locking strategy verification.	Medium

Suggestions up to commit ece1d02

Category	Suggestion	Impact
Possible issue	Close capture sink on cancel The cancel-after-success race guard checks parentTask.isCancelled() but does not close the captureSink before returning. If the task was cancelled between build SUCCEEDED and this check, the sink remains open and its allocated buffers are never released, causing a resource leak. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/join/BroadcastDispatch.java [147-152] if (parentTask != null && parentTask.isCancelled()) { String reason = parentTask.getReasonCancelled() != null ? parentTask.getReasonCancelled() : "unknown"; LOGGER.debug("[BroadcastDispatch] task cancelled after build SUCCEEDED, reason={}; aborting pass 2", reason); + try { + captureSink.close(); + } catch (Throwable ignore) { + // Best-effort cleanup + } terminal.onFailure(new org.opensearch.core.tasks.TaskCancelledException("query cancelled: " + reason)); return; } Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies a resource leak: when the task is cancelled after build SUCCEEDED, the captureSink remains open. The fix properly closes the sink before returning, preventing buffer leaks. This is a real correctness issue in the cancel-after-success race guard.	Medium
	Fix race in byte cap check The check-then-act sequence for the byte cap is not atomic. Between addAndGet(size) and the if (newTotal > maxBytes) check, another thread could add more data, causing the buffer to exceed maxBytes before the rollback addAndGet(-size) executes. Use compare-and-set to ensure atomicity. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/shuffle/ShuffleBufferManager.java [108-122] public boolean tryAddData(String side, byte[] data) { int size = data == null ? 0 : data.length; - long newTotal = currentBytes.addAndGet(size); - if (newTotal > maxBytes) { - currentBytes.addAndGet(-size); - rejectedCount.incrementAndGet(); - return false; - } + long current; + long newTotal; + do { + current = currentBytes.get(); + newTotal = current + size; + if (newTotal > maxBytes) { + rejectedCount.incrementAndGet(); + return false; + } + } while (!currentBytes.compareAndSet(current, newTotal)); if ("left".equals(side)) { leftData.add(data); } else { rightData.add(data); } return true; } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies a race condition in the byte cap check. The current addAndGet followed by a rollback is not atomic, allowing multiple threads to exceed maxBytes before any rollback occurs. The proposed compare-and-set loop ensures atomicity and prevents the buffer from growing beyond the cap.	Medium
	Fix allocator close race condition The closeBufferAllocator() method can be called from multiple threads when multiple QueryContext instances share the same AllocatorHolder. After setting closed = true, a race condition exists where another thread could call bufferAllocator() between the null check and the close operation, creating a new allocator that will never be closed. Move the closed flag check inside bufferAllocator() to occur after acquiring the lock. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/QueryContext.java [183-190] synchronized (allocatorHolder) { if (allocatorHolder.closed) return; - allocatorHolder.closed = true; if (allocatorHolder.bufferAllocator != null) { allocatorHolder.bufferAllocator.close(); allocatorHolder.bufferAllocator = null; } + allocatorHolder.closed = true; } Suggestion importance[1-10]: 7 __ Why: The suggestion identifies a potential race condition where bufferAllocator() could create a new allocator after closed is set but before the existing allocator is closed. However, the proposed fix (moving closed = true to the end) doesn't fully address the issue since bufferAllocator() checks closed before acquiring the lock. The real fix would require checking closed after acquiring the lock in bufferAllocator().	Medium
General	Clarify null-data isLast protocol When request.getData() is null and request.isLast() is true, the handler calls buffer.senderDone() without ever adding data. If a sender consistently sends only isLast=true markers with null data, the buffer's senderDone count increments but getCurrentBytes() remains zero. Verify that the buffer's completion logic (awaitReady) correctly handles senders that never send data, or document that null-data + isLast is a valid protocol pattern. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/action/TransportAnalyticsShuffleDataAction.java [56-82] ShuffleBufferManager.ShuffleBuffer buffer = shuffleBufferManager.getOrCreateBuffer( request.getQueryId(), request.getTargetStageId(), request.getPartitionIndex() ); if (request.getData() != null) { boolean accepted = buffer.tryAddData(request.getSide(), request.getData()); if (!accepted) { - ... + logger.debug(...); listener.onResponse(AnalyticsShuffleDataResponse.backpressureReject()); return; } } +// Null data + isLast is valid: sender signals completion without payload (e.g., empty partition). if (request.isLast()) { buffer.senderDone(request.getSide()); ... } Suggestion importance[1-10]: 5 __ Why: The suggestion correctly identifies that null data with isLast=true is a valid protocol pattern (empty partition completion). However, the 'improved_code' only adds a comment without changing behavior. This is more of a documentation suggestion than a code improvement. The existing code already handles this case correctly.	Low
	Cancel future on extraction failure When extractIpcBytes throws, the captureSink has already been closed (line 127) but the exception path does not verify that the future was properly cancelled. If the timeout exception from extractIpcBytes leaves the future in a non-terminal state, subsequent code waiting on it could hang indefinitely. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/join/BroadcastDispatch.java [134-141] byte[] ipcBytes; try { ipcBytes = extractIpcBytes(captureSink); } catch (Throwable t) { LOGGER.warn("[BroadcastDispatch] failed to extract IPC bytes from capture sink", t); + try { + captureSink.ipcBytesFuture().cancel(false); + } catch (Throwable ignore) { + // Best-effort cleanup + } terminal.onFailure(new RuntimeException("BroadcastDispatch: capture sink did not produce IPC bytes", t)); return; } Suggestion importance[1-10]: 4 __ Why: The suggestion addresses a potential edge case where extractIpcBytes throws but the future isn't cancelled. However, extractIpcBytes already cancels the future on timeout (line 254), and other exceptions from fut.get() leave the future in a terminal state. The added cancel is defensive but not strictly necessary given the existing contract.	Low
	Prevent exponential backoff overflow The bit-shift operation initialBackoffMillis << (attempt - 1) can overflow for large attempt values, producing negative or incorrect backoff delays. Since attempt can reach DEFAULT_MAX_ATTEMPTS (8), the shift amount can be 7, which for a 50ms initial backoff yields 6400ms before capping—safe. However, if constants change or the method is reused with different parameters, overflow becomes possible. Use Math.multiplyExact() or explicit overflow guards. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/shuffle/ShuffleSenderRetry.java [90] -long backoff = Math.min(DEFAULT_MAX_BACKOFF_MILLIS, initialBackoffMillis << (attempt - 1)); +long uncapped = attempt == 1 ? initialBackoffMillis : Math.min(Long.MAX_VALUE / 2, initialBackoffMillis << (attempt - 1)); +long backoff = Math.min(DEFAULT_MAX_BACKOFF_MILLIS, uncapped); Suggestion importance[1-10]: 4 __ Why: While overflow protection is generally good practice, the current implementation with DEFAULT_MAX_ATTEMPTS=8 and DEFAULT_INITIAL_BACKOFF_MILLIS=50 will never overflow (max shift is 7, yielding 6400ms before the Math.min cap). The suggestion adds complexity for a theoretical issue that won't occur with current constants. The Math.min with DEFAULT_MAX_BACKOFF_MILLIS already provides sufficient protection.	Low
	Fail fast when cap exceeded When the byte cap is exceeded, feed() closes the incoming batch but continues accepting subsequent batches (they are also closed and dropped). The sink should transition to a terminal failure state immediately to prevent further resource allocation and provide faster feedback to the caller that the broadcast has failed. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/BroadcastCaptureSink.java [119-124] if (exceeded || accumulatedBytes + batchBytes > maxBytes) { - exceeded = true; + if (!exceeded) { + exceeded = true; + ipcBytesFuture.completeExceptionally( + new BroadcastSizeExceededException(accumulatedBytes + batchBytes, maxBytes) + ); + } accumulatedBytes += batchBytes; batch.close(); - return; + throw new IllegalStateException("BroadcastCaptureSink: byte cap exceeded, sink is closed"); } Suggestion importance[1-10]: 3 __ Why: The suggestion proposes failing fast by throwing an exception when the cap is exceeded. However, the current design intentionally allows feed() to return false so the caller can handle backpressure gracefully. Throwing an exception would break the contract and force callers to catch exceptions instead of checking the return value. The failure is already surfaced from close() (line 168).	Low
	Guard against shared subgraph rewrites The recursive rewrite does not handle cycles or shared subgraphs. If the RelNode tree contains a cycle (unlikely in Calcite's immutable design but possible via custom nodes), the recursion will stack-overflow. If a node appears multiple times in the tree (shared subgraph), it will be rewritten multiple times, potentially creating duplicate StageInputTableScan instances. Consider adding a visited set or documenting that the input tree is guaranteed acyclic and non-shared. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java [473-489] private static RelNode rewriteStageInputScans(RelNode node) { + return rewriteStageInputScansImpl(node, new java.util.IdentityHashMap<>()); +} + +private static RelNode rewriteStageInputScansImpl(RelNode node, java.util.IdentityHashMap<RelNode, RelNode> visited) { + RelNode cached = visited.get(node); + if (cached != null) return cached; + if (node instanceof OpenSearchStageInputScan scan) { - return new StageInputTableScan(scan.getCluster(), scan.getTraitSet(), "input-" + scan.getChildStageId(), scan.getRowType()); + RelNode result = new StageInputTableScan(scan.getCluster(), scan.getTraitSet(), "input-" + scan.getChildStageId(), scan.getRowType()); + visited.put(node, result); + return result; } if (node instanceof OpenSearchBroadcastScan scan) { - return new StageInputTableScan(scan.getCluster(), scan.getTraitSet(), scan.getNamedInputId(), scan.getRowType()); + RelNode result = new StageInputTableScan(scan.getCluster(), scan.getTraitSet(), scan.getNamedInputId(), scan.getRowType()); + visited.put(node, result); + return result; } - List<RelNode> newInputs = new ArrayList<>(node.getInputs().size()); - boolean changed = false; - for (RelNode input : node.getInputs()) { - RelNode rewritten = rewriteStageInputScans(input); - newInputs.add(rewritten); - if (rewritten != input) { - changed = true; - } - } - return changed ? node.copy(node.getTraitSet(), newInputs) : node; + ... } Suggestion importance[1-10]: 3 __ Why: The suggestion raises a theoretical concern about cycles and shared subgraphs in the RelNode tree. However, Calcite's RelNode trees are designed to be DAGs (directed acyclic graphs) by convention, and shared subgraphs are intentionally allowed and handled by the immutable copy pattern. Adding an IdentityHashMap would add overhead for a problem that doesn't exist in practice and could break legitimate shared-subgraph optimizations.	Low

Suggestions up to commit 183897c

Category	Suggestion	Impact
Possible issue	Prevent integer overflow in backoff calculation The exponential backoff calculation initialBackoffMillis << (attempt - 1) can overflow for large attempt counts. When attempt reaches 31 or higher (depending on initialBackoffMillis), the left shift produces a negative value due to integer overflow, which Math.min then selects over DEFAULT_MAX_BACKOFF_MILLIS, resulting in a negative delay passed to the scheduler. Cast to long before the shift to prevent overflow. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/shuffle/ShuffleSenderRetry.java [90] -long backoff = Math.min(DEFAULT_MAX_BACKOFF_MILLIS, initialBackoffMillis << (attempt - 1)); +long backoff = Math.min(DEFAULT_MAX_BACKOFF_MILLIS, ((long) initialBackoffMillis) << (attempt - 1)); Suggestion importance[1-10]: 8 __ Why: Identifies a real overflow risk in the exponential backoff calculation. With DEFAULT_MAX_ATTEMPTS = 8 and DEFAULT_INITIAL_BACKOFF_MILLIS = 50, overflow won't occur in practice (50 << 7 = 6400), but the cast to long is a cheap safety measure that prevents potential issues if these constants are changed.	Medium
	Close capture sink on cancel The cancel-after-success race guard checks parentTask.isCancelled() but does not close the captureSink before returning. If the task was cancelled between pass 1 SUCCEEDED and this check, the sink remains open and its allocated buffers leak. Close the sink in this path to release resources. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/join/BroadcastDispatch.java [147-152] if (parentTask != null && parentTask.isCancelled()) { String reason = parentTask.getReasonCancelled() != null ? parentTask.getReasonCancelled() : "unknown"; LOGGER.debug("[BroadcastDispatch] task cancelled after build SUCCEEDED, reason={}; aborting pass 2", reason); + try { + captureSink.close(); + } catch (Throwable ignore) { + // Best-effort cleanup + } terminal.onFailure(new org.opensearch.core.tasks.TaskCancelledException("query cancelled: " + reason)); return; } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies a resource leak when the task is cancelled after build SUCCEEDED but before pass 2 starts. The captureSink should be closed to release allocated buffers. However, the impact is moderate since this is a rare race condition and the buffers would eventually be released when the allocator is closed.	Medium
	Validate allocator state before derivation The withDag method creates a new context sharing the same allocatorHolder, but the javadoc states "both phases belong to the same query and must share a single per-query allocator." However, there's no validation that prevents calling withDag on an already-closed context. If allocatorHolder.closed is true when withDag is called, the new context will inherit a closed allocator state, causing bufferAllocator() calls to throw immediately. Add a check to fail fast if the allocator is already closed. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/QueryContext.java [112-122] public QueryContext withDag(QueryDAG newDag) { + synchronized (allocatorHolder) { + if (allocatorHolder.closed) { + throw new IllegalStateException("Cannot create derived context from closed QueryContext for query " + dag.queryId()); + } + } return new QueryContext( newDag, searchExecutor, parentTask, maxConcurrentShardRequests, perQueryMemoryLimit, operationListeners, allocatorHolder ); } Suggestion importance[1-10]: 7 __ Why: Valid concern about creating a derived context from a closed allocator. The check prevents silent failures and provides clear error messaging. However, this is a defensive check for an edge case that shouldn't occur in normal operation if closeBufferAllocator() is called correctly.	Medium
General	Fix byte counter for dropped batches When the byte cap is exceeded, feed closes the incoming batch and returns without adding it to the buffer. However, the accumulatedBytes counter is incremented even though the batch was dropped. This inflates the reported total in the exception message. Only increment accumulatedBytes for batches that are actually retained. [sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/BroadcastCaptureSink.java [119-124]](https://github.com/opensearch-project/OpenSearch/pull/21677/files#diff-a11aeda156afad84848fa57b21b88e4ecffaa0f9018ee417689713874ee3...

[LantaoJin]

BTW, here is the new configurations added. Any thoughts? @mch2

1. analytics.mpp.enabled — boolean, default true. Master kill switch. When false, every join routes through coordinator-centric regardless of the advisor's decision.
2. analytics.mpp.broadcast_max_rows — long, default 1,000,000. Pre-flight row gate; the eligible build side's IndicesStats.primaries.docs.count must be > 0 AND ≤ this value for BROADCAST to fire.
3. analytics.mpp.broadcast_max_bytes — ByteSize, default 32MB. Runtime byte cap on the accumulated build-side payload at BroadcastCaptureSink. When exceeded, close() fails the future with BroadcastSizeExceededException.

[github-actions]

❌ Gradle check result for ece1d02: 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]

❌ Gradle check result for 8178ef0: 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]

✅ Gradle check result for 568a099: SUCCESS

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 73.45%. Comparing base (8f2d058) to head (568a099).
⚠️ Report is 4 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #21677      +/-   ##
============================================
- Coverage     73.46%   73.45%   -0.02%     
- Complexity    74825    74835      +10     
============================================
  Files          5997     6005       +8     
  Lines        339688   339767      +79     
  Branches      48961    48969       +8     
============================================
+ Hits         249558   249560       +2     
- Misses        70272    70371      +99     
+ Partials      19858    19836      -22     

☔ View full report in Codecov by Sentry.
📢 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]

❌ Gradle check result for 11e4c7a: 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?

[LantaoJin]

The CI failures not related, it introduced by #21703 which removed the try { root.close() } catch (IllegalStateException) { logger.warn(...) } block from DefaultArrowAllocatorService

[LantaoJin]

✅ Gradle check result for 568a099: SUCCESS

the previous CI passed, please take a review @mch2 @sandeshkr419

[LantaoJin]

@mch2 as discussed offline. I will continue developing the M2 and further works in my branch https://github.com/LantaoJin/OpenSearch/tree/feature/scaffold_mpp_stype_join. Close this PR and reopen later.