analytics-engine: add distributed join planning and execution

[mch2]

Description

Adds distributed planning and execution for Join and Union in the analytics engine. Multi-input shapes now plan through Volcano's cost model — OpenSearchExchangeReducer (ER) insertion is no longer hand-rolled at HEP marking time.

How ER insertion works now

1. HEP marker pass rewrites LogicalJoin/LogicalUnion → OpenSearchJoin/OpenSearchUnion. No ERs are inserted here, the marker only attaches viable backends and the required output distribution (SINGLETON).
2. Per-operator split rules (OpenSearchJoinSplitRule, OpenSearchUnionSplitRule) emit alternative plans for the optimizer:
   - COORDINATOR-local alternative: gather both sides to coord (always valid).
   - SHARD-local alternative: when inputs co-locate (same tableId, single shard), keep the operator shard-local with no gather — strictly cheaper, so Volcano picks it.
3. OpenSearchDistribution gained a Locality dimension (SHARD vs COORDINATOR) and carries tableId/shardCount on SHARD distributions; this is what the cost gate compares against.
4. DAGBuilder now recurses with sever() into ER child fragments, nested ERs (a Join's per-side ERs sitting below a top-level gather ER) get cut into their own child stages. Each Stage reads its ExchangeInfo directly off the ER instead of hard-coding SINGLETON.

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.

[expani]

Thanks @mch2 for getting this started.

I think we should explore Calcite's Volcano Trait mismatch to insert exchanges instead of deterministically doing it via HEP rule.

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to b01917a

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Check empty list before access When grandchildren is not empty, the code calls reduceViable.getFirst() without checking if reduceViable is empty. If filterByReduceCapability returns an empty list, getFirst() will throw NoSuchElementException. Add a validation check to ensure reduceViable is not empty before accessing its first element. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/DAGBuilder.java [110-113] -private static RelNode cutAtExchange( - OpenSearchExchangeReducer reducer, - int[] counter, - List<Stage> parentChildStages, - CapabilityRegistry registry, - ClusterService clusterService -) { - List<Stage> grandchildren = new ArrayList<>(); - RelNode childFragment = sever(reducer.getInput(), counter, grandchildren, registry, clusterService); - - int childStageId = counter[0]++; - TargetResolver targetResolver = grandchildren.isEmpty() ? new ShardTargetResolver(childFragment, clusterService) : null; - ExchangeSinkProvider childSinkProvider = null; - if (!grandchildren.isEmpty()) { - List<String> reduceViable = CapabilityResolutionUtils.filterByReduceCapability(registry, reducer.getViableBackends()); - childSinkProvider = registry.getBackend(reduceViable.getFirst()).getExchangeSinkProvider(); +if (!grandchildren.isEmpty()) { + List<String> reduceViable = CapabilityResolutionUtils.filterByReduceCapability(registry, reducer.getViableBackends()); + if (reduceViable.isEmpty()) { + throw new IllegalStateException("No reduce-capable backend found among viable backends: " + reducer.getViableBackends()); } - ... + childSinkProvider = registry.getBackend(reduceViable.getFirst()).getExchangeSinkProvider(); } Suggestion importance[1-10]: 8 __ Why: Calling getFirst() on reduceViable without checking if it's empty could throw NoSuchElementException. This is a potential runtime error that should be caught with proper validation before accessing the list element.	Medium
	Prevent infinite loop in traversal The loop may not terminate if the project chain contains a cycle or if unwrapHep returns the same node repeatedly. Add a visited set or iteration limit to prevent infinite loops in malformed plan trees. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchSortRule.java [158-173] private static int remapInputIndexThroughProjects(int index, RelNode node, OpenSearchSort inner) { RelNode current = RelNodeUtils.unwrapHep(node); int idx = index; + Set<RelNode> visited = new HashSet<>(); while (current instanceof OpenSearchProject project) { + if (!visited.add(current)) { + return -1; + } if (idx < 0 || idx >= project.getProjects().size()) { return -1; } RexNode expr = project.getProjects().get(idx); if (!(expr instanceof RexInputRef ref)) { return -1; } idx = ref.getIndex(); current = RelNodeUtils.unwrapHep(project.getInput()); } return current == inner ? idx : -1; } Suggestion importance[1-10]: 7 __ Why: Adding cycle detection to remapInputIndexThroughProjects is a valuable defensive measure. While malformed plan trees should not occur in normal operation, the visited set prevents potential infinite loops in edge cases, improving robustness without significant overhead.	Medium
	Validate null input distributions The cost gate skips validation when inputDist is null but continues iterating. If all inputs have null distributions, the method returns makeTinyCost() without verifying locality constraints. This could allow invalid join configurations to pass. Consider returning infinite cost when any input has a null distribution, or explicitly validate that at least one input satisfies the locality requirements. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/OpenSearchJoin.java [102-107] @Override public org.apache.calcite.plan.RelOptCost computeSelfCost( org.apache.calcite.plan.RelOptPlanner planner, org.apache.calcite.rel.metadata.RelMetadataQuery mq ) { OpenSearchDistribution selfDist = distributionOf(this); if (selfDist == null || selfDist.getType() != org.apache.calcite.rel.RelDistribution.Type.SINGLETON) { return planner.getCostFactory().makeInfiniteCost(); } for (RelNode input : getInputs()) { OpenSearchDistribution inputDist = distributionOf(input); - if (inputDist == null) continue; + if (inputDist == null) { + return planner.getCostFactory().makeInfiniteCost(); + } if (inputDist.getType() == org.apache.calcite.rel.RelDistribution.Type.ANY) continue; ... } return planner.getCostFactory().makeTinyCost(); } Suggestion importance[1-10]: 7 __ Why: The cost gate currently skips null inputDist values but continues iterating, potentially allowing invalid join configurations to pass. Returning infinite cost for null distributions would make the validation more robust and prevent edge cases where all inputs have null distributions.	Medium
General	Use shutdownNow for executor cleanup The executor pool should be shut down gracefully with shutdownNow() in the finally block to prevent thread leaks if an exception occurs before normal shutdown. Additionally, verify that all tasks complete or are cancelled before asserting termination. sandbox/plugins/analytics-backend-datafusion/src/internalClusterTest/java/org/opensearch/be/datafusion/CoordinatorJoinMultiNodeIT.java [293-316] public void testConcurrentJoinsAreIsolated() throws Exception { final int NUM_KEYS = 30; final int N_QUERIES = 4; String t1 = "join_mn_conc_t1"; String t2 = "join_mn_conc_t2"; createParquetIndex(t1, NUM_SHARDS, "v"); createParquetIndex(t2, NUM_SHARDS, "w"); indexUnique(t1, "v", 1, NUM_KEYS, k -> k * 11); indexUnique(t2, "w", 1, NUM_KEYS, k -> k * 113); ExecutorService pool = Executors.newFixedThreadPool(N_QUERIES); try { @SuppressWarnings("unchecked") CompletableFuture<PPLResponse>[] futures = new CompletableFuture[N_QUERIES]; for (int i = 0; i < N_QUERIES; i++) { futures[i] = CompletableFuture.supplyAsync( () -> executePPL("source=" + t1 + " | join on " + t1 + ".k = " + t2 + ".k " + t2), pool ); } for (int i = 0; i < N_QUERIES; i++) { PPLResponse response; try { response = futures[i].get(60, TimeUnit.SECONDS); } catch (ExecutionException e) { throw new AssertionError("query " + i + " threw", e.getCause()); } assertColumns(response, "k", "v", "w"); assertEquals("query " + i + " row count", NUM_KEYS, response.getRows().size()); } } finally { - pool.shutdown(); + pool.shutdownNow(); assertTrue("executor must terminate", pool.awaitTermination(10, TimeUnit.SECONDS)); } } Suggestion importance[1-10]: 6 __ Why: Using shutdownNow() instead of shutdown() in the finally block is a safer cleanup practice that prevents thread leaks if an exception occurs. However, the current code already has proper cleanup with shutdown() and awaitTermination(), so this is a minor improvement rather than a critical fix.	Low
	Use appropriate planning exception The code throws IllegalStateException when no backend supports the required join kind, but this exception is unchecked and may propagate unexpectedly during query planning. Consider whether this should be a checked exception or a more specific planning exception type that can be handled gracefully by the planner to provide better error messages to users. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchJoinRule.java [92-96] -viableBackends.removeIf(backend -> { - var caps = context.getCapabilityRegistry().getBackend(backend).getCapabilityProvider(); - for (JoinCapability cap : caps.joinCapabilities()) { - if (cap.kinds().contains(requiredKind)) return false; - } - return true; -}); if (viableBackends.isEmpty()) { - throw new IllegalStateException( + throw new org.apache.calcite.plan.RelOptPlanner.CannotPlanException( "No backend supports join kind [" + requiredKind + "] among viable backends " + candidateBackends ); } Suggestion importance[1-10]: 5 __ Why: While using a more specific exception type like CannotPlanException would be better for error handling, the current IllegalStateException is still functional. This is a minor improvement for better error categorization rather than a critical issue.	Low
	Guard against empty bulk requests The batch loop increments by batchSize but computes batchEnd as batchStart + batchSize - 1. When keyHi - keyLo + 1 is not a multiple of batchSize, the final batch may be skipped or processed incorrectly. Verify the loop condition handles the final partial batch correctly. sandbox/plugins/analytics-backend-datafusion/src/internalClusterTest/java/org/opensearch/be/datafusion/CoordinatorJoinMultiNodeIT.java [353-373] private void bulkIndexUnique(String indexName, String payloadField, int keyLo, int keyHi, IntUnaryOperator keyToPayload) { final int batchSize = 500; for (int batchStart = keyLo; batchStart <= keyHi; batchStart += batchSize) { int batchEnd = Math.min(batchStart + batchSize - 1, keyHi); org.opensearch.action.bulk.BulkRequestBuilder bulk = client().prepareBulk(); for (int key = batchStart; key <= batchEnd; key++) { bulk.add( client().prepareIndex(indexName) .setId(indexName + "_" + key) .setSource("k", key, payloadField, keyToPayload.applyAsInt(key)) ); } - org.opensearch.action.bulk.BulkResponse response = bulk.get(); - assertFalse( - "bulk index batch [" + batchStart + ", " + batchEnd + "] had failures: " + response.buildFailureMessage(), - response.hasFailures() - ); + if (bulk.numberOfActions() > 0) { + org.opensearch.action.bulk.BulkResponse response = bulk.get(); + assertFalse( + "bulk index batch [" + batchStart + ", " + batchEnd + "] had failures: " + response.buildFailureMessage(), + response.hasFailures() + ); + } } client().admin().indices().prepareRefresh(indexName).get(); client().admin().indices().prepareFlush(indexName).get(); } Suggestion importance[1-10]: 4 __ Why: The suggestion to check numberOfActions() > 0 before executing the bulk request is a minor defensive improvement. However, the existing logic with batchEnd = Math.min(batchStart + batchSize - 1, keyHi) already ensures at least one action per batch when batchStart <= keyHi, making this check largely redundant.	Low
	Clarify single-input condition logic The condition checks allChildrenAreExchangeReducer && node.getInputs().size() == 1, but then iterates over all inputs. This is logically inconsistent: if there's only one input, the loop executes once, but the comment suggests this path handles single-input ancestors. Verify whether this condition should be == 1 or if the loop logic needs adjustment for the multi-input case that follows. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/FragmentConversionDriver.java [283-291] -if (allChildrenAreExchangeReducer && node.getInputs().size() == 1) { - List<RelNode> finalAggInputs = new ArrayList<>(node.getInputs().size()); - for (RelNode input : node.getInputs()) { - finalAggInputs.add(strip(input.getInputs().getFirst(), delegationBytes)); +if (allChildrenAreExchangeReducer) { + if (node.getInputs().size() == 1) { + List<RelNode> finalAggInputs = new ArrayList<>(node.getInputs().size()); + for (RelNode input : node.getInputs()) { + finalAggInputs.add(strip(input.getInputs().getFirst(), delegationBytes)); + } + RelNode finalAggFragment = node.copy(node.getTraitSet(), finalAggInputs); + return convertor.convertFinalAggFragment(strip(finalAggFragment, delegationBytes)); } - RelNode finalAggFragment = node.copy(node.getTraitSet(), finalAggInputs); - return convertor.convertFinalAggFragment(strip(finalAggFragment, delegationBytes)); } Suggestion importance[1-10]: 4 __ Why: The condition allChildrenAreExchangeReducer && node.getInputs().size() == 1 followed by a loop over all inputs is logically consistent (the loop executes once for the single input). The suggested refactoring doesn't materially improve clarity and the existing code is correct.	Low

---

Previous suggestions

Suggestions up to commit 2575a64

Category	Suggestion	Impact
General	Preserve locality when degrading to ANY When a HASH key cannot be mapped, the method returns a new distribution with locality=null. Downstream code may assume locality is non-null for non-ANY distributions, leading to null pointer exceptions. Preserve the original locality or document that ANY distributions may have null locality. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/OpenSearchDistribution.java [130-146] @Override public RelDistribution apply(Mappings.TargetMapping mapping) { if (traitDef == null) { throw new IllegalStateException("OpenSearchDistribution has null traitDef"); } if (type != Type.HASH_DISTRIBUTED || keys.isEmpty()) { return this; } - ... List<Integer> newKeys = new java.util.ArrayList<>(keys.size()); for (int key : keys) { int target = mapping.getTargetOpt(key); if (target < 0) { - return new OpenSearchDistribution(traitDef, null, Type.ANY, List.of(), null, null); + return new OpenSearchDistribution(traitDef, locality, Type.ANY, List.of(), null, null); } newKeys.add(target); } return new OpenSearchDistribution(traitDef, locality, Type.HASH_DISTRIBUTED, newKeys, tableId, shardCount); } Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies that setting locality=null when degrading to Type.ANY may cause issues downstream if code assumes non-null locality for non-ANY distributions. Preserving the original locality maintains consistency and prevents potential null pointer exceptions.	Medium
	Ensure executor pool cleanup on timeout The executor pool is not forcibly terminated if awaitTermination times out or if an exception occurs before shutdown completes. Add pool.shutdownNow() in a finally block to ensure threads are interrupted and resources are released even when the test fails or times out. sandbox/plugins/analytics-backend-datafusion/src/internalClusterTest/java/org/opensearch/be/datafusion/CoordinatorJoinMultiNodeIT.java [293-317] public void testConcurrentJoinsAreIsolated() throws Exception { final int NUM_KEYS = 30; final int N_QUERIES = 4; String t1 = "join_mn_conc_t1"; String t2 = "join_mn_conc_t2"; createParquetIndex(t1, NUM_SHARDS, "v"); createParquetIndex(t2, NUM_SHARDS, "w"); indexUnique(t1, "v", 1, NUM_KEYS, k -> k * 11); indexUnique(t2, "w", 1, NUM_KEYS, k -> k * 113); ExecutorService pool = Executors.newFixedThreadPool(N_QUERIES); try { @SuppressWarnings("unchecked") CompletableFuture<PPLResponse>[] futures = new CompletableFuture[N_QUERIES]; for (int i = 0; i < N_QUERIES; i++) { futures[i] = CompletableFuture.supplyAsync( () -> executePPL("source=" + t1 + " | join on " + t1 + ".k = " + t2 + ".k " + t2), pool ); } for (int i = 0; i < N_QUERIES; i++) { PPLResponse response; try { response = futures[i].get(60, TimeUnit.SECONDS); } catch (ExecutionException e) { throw new AssertionError("query " + i + " threw", e.getCause()); } assertColumns(response, "k", "v", "w"); assertEquals("query " + i + " row count", NUM_KEYS, response.getRows().size()); } } finally { pool.shutdown(); - assertTrue("executor must terminate", pool.awaitTermination(10, TimeUnit.SECONDS)); + if (!pool.awaitTermination(10, TimeUnit.SECONDS)) { + pool.shutdownNow(); + } } } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that pool.shutdownNow() should be called if awaitTermination times out to forcibly interrupt threads and release resources. This prevents potential resource leaks in test failures.	Medium
	Prevent infinite loop in project traversal The loop may not terminate if the project chain contains a cycle or if unwrapHep returns the same node repeatedly. Add a visited-node check or a maximum iteration limit to prevent infinite loops in malformed plan trees. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchSortRule.java [158-173] private static int remapInputIndexThroughProjects(int index, RelNode node, OpenSearchSort inner) { RelNode current = RelNodeUtils.unwrapHep(node); int idx = index; + int depth = 0; + final int MAX_DEPTH = 100; while (current instanceof OpenSearchProject project) { + if (++depth > MAX_DEPTH) { + return -1; + } if (idx < 0 || idx >= project.getProjects().size()) { return -1; } RexNode expr = project.getProjects().get(idx); if (!(expr instanceof RexInputRef ref)) { return -1; } idx = ref.getIndex(); current = RelNodeUtils.unwrapHep(project.getInput()); } return current == inner ? idx : -1; } Suggestion importance[1-10]: 6 __ Why: Adding a depth limit is a reasonable defensive measure against malformed plan trees or cycles. While the current code may not encounter such cases in practice, the guard improves robustness with minimal overhead.	Low
	Remove unnecessary loop for single input The condition checks node.getInputs().size() == 1 but then iterates over all inputs to build finalAggInputs. This is inconsistent: if the size is guaranteed to be 1, the loop is unnecessary. If multi-input nodes are expected, the condition should allow size() >= 1. Align the logic to match the intended behavior. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/FragmentConversionDriver.java [283-291] if (allChildrenAreExchangeReducer && node.getInputs().size() == 1) { - List<RelNode> finalAggInputs = new ArrayList<>(node.getInputs().size()); - for (RelNode input : node.getInputs()) { - finalAggInputs.add(strip(input.getInputs().getFirst(), delegationBytes)); - } - return convertor.convertFinalAggFragment(strip(node.copyWithNewInputs(finalAggInputs), delegationBytes)); + RelNode input = node.getInputs().getFirst(); + RelNode finalAggInput = strip(input.getInputs().getFirst(), delegationBytes); + return convertor.convertFinalAggFragment(strip(node.copyWithNewInputs(List.of(finalAggInput)), delegationBytes)); } Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies that when node.getInputs().size() == 1, the loop over all inputs is redundant. The improved code simplifies this by directly accessing the single input via getFirst(), which is clearer and more efficient. This is a valid code quality improvement.	Low
	Clarify error message for debugging The code modifies viableBackends in-place via removeIf, then throws an exception if empty. However, candidateBackends is created as an immutable copy before the removal. If the exception is thrown, the error message references candidateBackends which still contains all original backends, potentially confusing debugging. Consider creating the snapshot after filtering or clarify the error message. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchJoinRule.java [85-96] viableBackends.removeIf(backend -> { var caps = context.getCapabilityRegistry().getBackend(backend).getCapabilityProvider(); for (JoinCapability cap : caps.joinCapabilities()) { if (cap.kinds().contains(requiredKind)) return false; } return true; }); if (viableBackends.isEmpty()) { throw new IllegalStateException( - "No backend supports join kind [" + requiredKind + "] among viable backends " + candidateBackends + "No backend supports join kind [" + requiredKind + "] among candidates " + candidateBackends + " (after capability filtering: " + viableBackends + ")" ); } Suggestion importance[1-10]: 4 __ Why: The suggestion points out that candidateBackends is a snapshot taken before filtering, so the error message could be misleading. However, the improved code's error message still references viableBackends which is empty at that point, making the message confusing. The suggestion is minor and doesn't significantly improve clarity.	Low
	Guard against empty bulk requests The batch loop increments by batchSize but computes batchEnd as batchStart + batchSize - 1. When keyHi - keyLo + 1 is not a multiple of batchSize, the final batch may be skipped if batchStart overshoots keyHi. Verify the loop condition handles the final partial batch correctly. sandbox/plugins/analytics-backend-datafusion/src/internalClusterTest/java/org/opensearch/be/datafusion/CoordinatorJoinMultiNodeIT.java [353-373] private void bulkIndexUnique(String indexName, String payloadField, int keyLo, int keyHi, IntUnaryOperator keyToPayload) { final int batchSize = 500; for (int batchStart = keyLo; batchStart <= keyHi; batchStart += batchSize) { int batchEnd = Math.min(batchStart + batchSize - 1, keyHi); org.opensearch.action.bulk.BulkRequestBuilder bulk = client().prepareBulk(); for (int key = batchStart; key <= batchEnd; key++) { bulk.add( client().prepareIndex(indexName) .setId(indexName + "_" + key) .setSource("k", key, payloadField, keyToPayload.applyAsInt(key)) ); } - org.opensearch.action.bulk.BulkResponse response = bulk.get(); - assertFalse( - "bulk index batch [" + batchStart + ", " + batchEnd + "] had failures: " + response.buildFailureMessage(), - response.hasFailures() - ); + if (bulk.numberOfActions() > 0) { + org.opensearch.action.bulk.BulkResponse response = bulk.get(); + assertFalse( + "bulk index batch [" + batchStart + ", " + batchEnd + "] had failures: " + response.buildFailureMessage(), + response.hasFailures() + ); + } } client().admin().indices().prepareRefresh(indexName).get(); client().admin().indices().prepareFlush(indexName).get(); } Suggestion importance[1-10]: 3 __ Why: The suggestion adds a check for numberOfActions() > 0 before executing the bulk request. However, the existing loop logic already ensures at least one action is added per batch (since batchStart <= batchEnd is guaranteed by the outer loop condition), making this check redundant.	Low
Possible issue	Guard against empty reduce-capable backend list Calling getFirst() on reduceViable without checking if the list is empty will throw NoSuchElementException when no backends support reduce capability. Verify that reduceViable is non-empty before accessing its first element, or handle the empty case explicitly. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/DAGBuilder.java [110-113] -private static RelNode cutAtExchange( - OpenSearchExchangeReducer reducer, - int[] counter, - List<Stage> parentChildStages, - CapabilityRegistry registry, - ClusterService clusterService -) { - ... - ExchangeSinkProvider childSinkProvider = null; - if (!grandchildren.isEmpty()) { - List<String> reduceViable = CapabilityResolutionUtils.filterByReduceCapability(registry, reducer.getViableBackends()); - childSinkProvider = registry.getBackend(reduceViable.getFirst()).getExchangeSinkProvider(); +ExchangeSinkProvider childSinkProvider = null; +if (!grandchildren.isEmpty()) { + List<String> reduceViable = CapabilityResolutionUtils.filterByReduceCapability(registry, reducer.getViableBackends()); + if (reduceViable.isEmpty()) { + throw new IllegalStateException("No reduce-capable backend found among viable backends: " + reducer.getViableBackends()); } - ... + childSinkProvider = registry.getBackend(reduceViable.getFirst()).getExchangeSinkProvider(); } Suggestion importance[1-10]: 8 __ Why: Valid concern: calling getFirst() on reduceViable without checking if it's empty will throw NoSuchElementException. The suggestion correctly identifies this potential runtime error and proposes an explicit check with a meaningful error message, improving robustness.	Medium
	Prevent potential null pointer exception The cost gate skips validation when inputDist is null but proceeds to check locality matching below. If selfDist.getLocality() is non-null and inputDist becomes null after the continue, the locality comparison at line 108 will throw a NullPointerException. Add an explicit null check before the locality comparison or ensure the loop only processes non-null distributions. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/OpenSearchJoin.java [101-122] -@Override -public org.apache.calcite.plan.RelOptCost computeSelfCost( - org.apache.calcite.plan.RelOptPlanner planner, - org.apache.calcite.rel.metadata.RelMetadataQuery mq -) { - OpenSearchDistribution selfDist = distributionOf(this); - if (selfDist == null || selfDist.getType() != org.apache.calcite.rel.RelDistribution.Type.SINGLETON) { +for (RelNode input : getInputs()) { + OpenSearchDistribution inputDist = distributionOf(input); + if (inputDist == null || inputDist.getType() == org.apache.calcite.rel.RelDistribution.Type.ANY) { + continue; + } + if (inputDist.getType() != org.apache.calcite.rel.RelDistribution.Type.SINGLETON) { return planner.getCostFactory().makeInfiniteCost(); } - for (RelNode input : getInputs()) { - OpenSearchDistribution inputDist = distributionOf(input); - if (inputDist == null) continue; - if (inputDist.getType() == org.apache.calcite.rel.RelDistribution.Type.ANY) continue; - ... - } - return planner.getCostFactory().makeTinyCost(); + ... } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that the loop continues when inputDist is null but then attempts to access inputDist.getLocality() at line 109 without re-checking for null. However, the code does check inputDist == null and continues, so the locality comparison only runs on non-null distributions. The improved code consolidates the checks more clearly, which is a minor improvement in readability and safety.	Medium

Suggestions up to commit 835c59a

Category	Suggestion	Impact
Possible issue	Validate join child type When a join child is not an OpenSearchRelNode, appendChildStorage silently skips it, leaving out incomplete. This can cause downstream operators to see a truncated field storage list, leading to incorrect field resolution. Add an else branch that throws an exception or logs a warning to surface this condition during development. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/OpenSearchJoin.java [68-73] private static void appendChildStorage(RelNode child, List<FieldStorageInfo> out) { RelNode unwrapped = RelNodeUtils.unwrapHep(child); if (unwrapped instanceof OpenSearchRelNode os) { out.addAll(os.getOutputFieldStorage()); + } else { + throw new IllegalStateException( + "Join child must be OpenSearchRelNode to derive field storage, got: " + unwrapped.getClass().getSimpleName() + ); } } Suggestion importance[1-10]: 8 __ Why: Valid concern that silently skipping non-OpenSearchRelNode children produces incomplete field storage lists. The suggestion to throw an exception is appropriate since join children must be marked nodes by construction, and failing fast would catch planner bugs earlier.	Medium
	Fail fast on unmarked children The comment "empty list forces the fallback path above" is misleading. When viableBackendsOf returns an empty list for an unmarked child, computeViableBackends performs intersection.retainAll(rightBackends) where one side is empty, resulting in an empty intersection. This causes the join to have zero viable backends, triggering the IllegalStateException in onMatch. Consider returning a sentinel or throwing an exception to fail fast when children aren't marked yet. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchJoinRule.java [128-134] private static List<String> viableBackendsOf(RelNode rel) { if (RelNodeUtils.unwrapHep(rel) instanceof OpenSearchRelNode osNode) { return osNode.getViableBackends(); } - // Not yet marked — empty list forces the fallback path above. - return List.of(); + throw new IllegalStateException( + "Join child not yet marked as OpenSearchRelNode: " + rel.getClass().getSimpleName() + ); } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that returning an empty list for unmarked children leads to an empty intersection and eventual IllegalStateException. Failing fast would improve debuggability, though the current behavior is technically correct since bottom-up HEP traversal guarantees children are marked before the join rule fires.	Medium
General	Ensure executor shutdown on timeout The executor pool is not forcibly terminated if awaitTermination times out or if an exception occurs before the shutdown completes. Add pool.shutdownNow() in a catch block or after the awaitTermination assertion to ensure threads are interrupted and resources are released even when the test fails or hangs. sandbox/plugins/analytics-backend-datafusion/src/internalClusterTest/java/org/opensearch/be/datafusion/CoordinatorJoinMultiNodeIT.java [293-316] -public void testConcurrentJoinsAreIsolated() throws Exception { - final int NUM_KEYS = 30; - final int N_QUERIES = 4; - String t1 = "join_mn_conc_t1"; - String t2 = "join_mn_conc_t2"; - createParquetIndex(t1, NUM_SHARDS, "v"); - createParquetIndex(t2, NUM_SHARDS, "w"); - indexUnique(t1, "v", 1, NUM_KEYS, k -> k * 11); - indexUnique(t2, "w", 1, NUM_KEYS, k -> k * 113); - - ExecutorService pool = Executors.newFixedThreadPool(N_QUERIES); - try { - @SuppressWarnings("unchecked") - CompletableFuture<PPLResponse>[] futures = new CompletableFuture[N_QUERIES]; - for (int i = 0; i < N_QUERIES; i++) { - futures[i] = CompletableFuture.supplyAsync( - () -> executePPL("source=" + t1 + " | join on " + t1 + ".k = " + t2 + ".k " + t2), - pool - ); +ExecutorService pool = Executors.newFixedThreadPool(N_QUERIES); +try { + @SuppressWarnings("unchecked") + CompletableFuture<PPLResponse>[] futures = new CompletableFuture[N_QUERIES]; + for (int i = 0; i < N_QUERIES; i++) { + futures[i] = CompletableFuture.supplyAsync( + () -> executePPL("source=" + t1 + " | join on " + t1 + ".k = " + t2 + ".k " + t2), + pool + ); + } + for (int i = 0; i < N_QUERIES; i++) { + PPLResponse response; + try { + response = futures[i].get(60, TimeUnit.SECONDS); + } catch (ExecutionException e) { + throw new AssertionError("query " + i + " threw", e.getCause()); } - for (int i = 0; i < N_QUERIES; i++) { - PPLResponse response; - try { - response = futures[i].get(60, TimeUnit.SECONDS); - } catch (ExecutionException e) { - throw new AssertionError("query " + i + " threw", e.getCause()); - } - assertColumns(response, "k", "v", "w"); - assertEquals("query " + i + " row count", NUM_KEYS, response.getRows().size()); - } - } finally { - pool.shutdown(); - assertTrue("executor must terminate", pool.awaitTermination(10, TimeUnit.SECONDS)); + assertColumns(response, "k", "v", "w"); + assertEquals("query " + i + " row count", NUM_KEYS, response.getRows().size()); + } +} finally { + pool.shutdown(); + if (!pool.awaitTermination(10, TimeUnit.SECONDS)) { + pool.shutdownNow(); } } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that pool.shutdownNow() should be called if awaitTermination times out to forcibly terminate threads and release resources. This prevents potential resource leaks in test failures.	Medium
	Prevent infinite loop in traversal The loop walks down through projects but does not guard against cycles in the RelNode graph. If a malformed plan contains a cycle, this method will loop indefinitely. Add a visited-set or depth limit to prevent infinite loops. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchSortRule.java [154-173] private static int remapInputIndexThroughProjects(int index, RelNode node, OpenSearchSort inner) { RelNode current = RelNodeUtils.unwrapHep(node); int idx = index; + int depth = 0; + final int MAX_DEPTH = 100; while (current instanceof OpenSearchProject project) { + if (++depth > MAX_DEPTH) { + return -1; + } if (idx < 0 || idx >= project.getProjects().size()) { return -1; } RexNode expr = project.getProjects().get(idx); if (!(expr instanceof RexInputRef ref)) { return -1; } idx = ref.getIndex(); current = RelNodeUtils.unwrapHep(project.getInput()); } return current == inner ? idx : -1; } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that the traversal loop lacks cycle detection. Adding a depth limit prevents infinite loops in malformed plans, improving robustness against edge cases.	Medium
	Validate traitDef is non-null The apply method returns this when traitDef is null, but the constructor allows traitDef to be set. If traitDef is null during apply, subsequent operations may fail with NullPointerException. Validate traitDef is non-null in the constructor or document that null is invalid. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/OpenSearchDistribution.java [60-74] -@Override -public RelDistribution apply(Mappings.TargetMapping mapping) { +OpenSearchDistribution( + OpenSearchDistributionTraitDef traitDef, + Locality locality, + Type type, + List<Integer> keys, + Integer tableId, + Integer shardCount +) { if (traitDef == null) { - return this; + throw new IllegalArgumentException("traitDef must not be null"); } - if (type != Type.HASH_DISTRIBUTED || keys.isEmpty()) { - return this; - } - // Calcite's contract on RelDistribution.apply (RelDistribution.java:53-67) is to - // silently degrade to ANY if any HASH key cannot be mapped through the projection. - // Mappings.apply2 throws on an unmapped key, which is the wrong behavior here — fall - // back to ANY when the mapping drops a key we depend on. - List<Integer> newKeys = new java.util.ArrayList<>(keys.size()); - for (int key : keys) { - int target = mapping.getTargetOpt(key); - if (target < 0) { - return new OpenSearchDistribution(traitDef, null, Type.ANY, List.of(), null, null); - } - newKeys.add(target); - } - return new OpenSearchDistribution(traitDef, locality, Type.HASH_DISTRIBUTED, newKeys, tableId, shardCount); + this.traitDef = traitDef; + this.locality = locality; + this.type = type; + this.keys = keys; + this.tableId = tableId; + this.shardCount = shardCount; } Suggestion importance[1-10]: 6 __ Why: The suggestion identifies a potential NullPointerException risk when traitDef is null. Adding validation in the constructor improves robustness, though the existing code already checks for null in apply().	Low
	Validate operator is non-null The method calls keyToPayload.applyAsInt(key) inside the loop without null-checking the operator. If keyToPayload is null, a NullPointerException will occur during indexing. Validate the operator is non-null at the method entry to fail fast with a clear message. sandbox/plugins/analytics-backend-datafusion/src/internalClusterTest/java/org/opensearch/be/datafusion/CoordinatorJoinMultiNodeIT.java [353-373] private void bulkIndexUnique(String indexName, String payloadField, int keyLo, int keyHi, IntUnaryOperator keyToPayload) { + if (keyToPayload == null) { + throw new IllegalArgumentException("keyToPayload must not be null"); + } final int batchSize = 500; for (int batchStart = keyLo; batchStart <= keyHi; batchStart += batchSize) { int batchEnd = Math.min(batchStart + batchSize - 1, keyHi); org.opensearch.action.bulk.BulkRequestBuilder bulk = client().prepareBulk(); for (int key = batchStart; key <= batchEnd; key++) { bulk.add( client().prepareIndex(indexName) .setId(indexName + "_" + key) .setSource("k", key, payloadField, keyToPayload.applyAsInt(key)) ); } org.opensearch.action.bulk.BulkResponse response = bulk.get(); assertFalse( "bulk index batch [" + batchStart + ", " + batchEnd + "] had failures: " + response.buildFailureMessage(), response.hasFailures() ); } client().admin().indices().prepareRefresh(indexName).get(); client().admin().indices().prepareFlush(indexName).get(); } Suggestion importance[1-10]: 5 __ Why: The suggestion adds a null check for keyToPayload to fail fast with a clear message. While this improves defensive programming, the method is private and all call sites in the PR pass non-null operators, making this a minor improvement.	Low
	Clarify single-input aggregate handling The condition allChildrenAreExchangeReducer && node.getInputs().size() == 1 is contradictory: if all children are ERs and there's only one input, the loop that follows iterates exactly once, making the list construction unnecessary. This suggests the intent was to handle single-input final aggregates differently from multi-input nodes. Clarify the logic or remove the size check if it's redundant. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/FragmentConversionDriver.java [283-291] -if (allChildrenAreExchangeReducer && node.getInputs().size() == 1) { - List<RelNode> finalAggInputs = new ArrayList<>(node.getInputs().size()); - for (RelNode input : node.getInputs()) { - // Skip the ER, keep StageInputScan below it as the leaf for schema inference. - finalAggInputs.add(strip(input.getInputs().getFirst(), delegationBytes)); +if (allChildrenAreExchangeReducer) { + if (node.getInputs().size() == 1) { + // Single-input final aggregate: strip ER, keep StageInputScan. + RelNode finalAggInput = strip(node.getInputs().getFirst().getInputs().getFirst(), delegationBytes); + return convertor.convertFinalAggFragment(strip(node.copyWithNewInputs(List.of(finalAggInput)), delegationBytes)); } - return convertor.convertFinalAggFragment(strip(node.copyWithNewInputs(finalAggInputs), delegationBytes)); + // Multi-input case handled below } Suggestion importance[1-10]: 5 __ Why: The observation about the redundant loop is correct, but the suggested refactoring doesn't materially improve clarity. The existing code is slightly verbose but functionally correct, and the multi-input case is handled separately below, so the impact is minor.	Low
	Handle empty input list When inputs is empty, the method returns null without iterating, which may not be the intended behavior. An empty input list could represent a degenerate join case that should be rejected explicitly. Add a guard at the start to handle empty inputs or document that null is the correct return for that case. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchJoinSplitRule.java [89-103] private static Integer commonColocatedTableId(List<RelNode> inputs) { + if (inputs.isEmpty()) { + return null; // or throw if empty inputs are invalid + } Integer commonId = null; for (RelNode input : inputs) { - OpenSearchDistribution dist = distributionOf(input); - if (dist == null) return null; - if (dist.getLocality() != OpenSearchDistribution.Locality.SHARD) return null; - if (dist.getType() != RelDistribution.Type.SINGLETON) return null; - if (!Integer.valueOf(1).equals(dist.getShardCount())) return null; - Integer tid = dist.getTableId(); - if (tid == null) return null; - if (commonId == null) commonId = tid; - else if (!commonId.equals(tid)) return null; + ... } return commonId; } Suggestion importance[1-10]: 3 __ Why: While technically correct that empty inputs return null, joins always have at least two inputs by definition, so this is a non-issue in practice. The suggestion adds defensive code for a case that cannot occur in valid join plans, providing minimal value.	Low

Suggestions up to commit a5fac52

Category	Suggestion	Impact
Possible issue	Fix contradictory multi-input condition The condition allChildrenAreExchangeReducer && node.getInputs().size() == 1 is contradictory: if all children are ERs and there's only one input, the loop that follows iterates once, making the multi-input logic unreachable. This breaks final-aggregate conversion for single-input nodes. Remove the size check to allow the loop to handle both single and multi-input cases. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/FragmentConversionDriver.java [283-291] -if (allChildrenAreExchangeReducer && node.getInputs().size() == 1) { +if (allChildrenAreExchangeReducer) { List<RelNode> finalAggInputs = new ArrayList<>(node.getInputs().size()); for (RelNode input : node.getInputs()) { // Skip the ER, keep StageInputScan below it as the leaf for schema inference. finalAggInputs.add(strip(input.getInputs().getFirst(), delegationBytes)); } RelNode finalAggFragment = node.copy(node.getTraitSet(), finalAggInputs); return convertor.convertFinalAggFragment(finalAggFragment); } Suggestion importance[1-10]: 9 __ Why: The condition allChildrenAreExchangeReducer && node.getInputs().size() == 1 is indeed contradictory and would prevent multi-input nodes (Join, Union) from being handled correctly. Removing the size check allows the code to handle both single and multi-input cases as intended by the PR's multi-input conversion logic.	High
	Reject joins with unresolved distributions The cost gate skips inputs with null distribution, allowing joins with unresolved traits to pass. This can lead to incorrect plans when Volcano explores partially-resolved subsets. Return infinite cost when inputDist is null to enforce that all inputs must have a resolved distribution before the join is considered valid. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/OpenSearchJoin.java [101-107] for (RelNode input : getInputs()) { OpenSearchDistribution inputDist = distributionOf(input); - if (inputDist == null) continue; + if (inputDist == null) { + return planner.getCostFactory().makeInfiniteCost(); + } if (inputDist.getType() == org.apache.calcite.rel.RelDistribution.Type.ANY) continue; if (inputDist.getType() != org.apache.calcite.rel.RelDistribution.Type.SINGLETON) { return planner.getCostFactory().makeInfiniteCost(); } ... } Suggestion importance[1-10]: 7 __ Why: The suggestion to return infinite cost when inputDist is null is valid for enforcing that all inputs have resolved distributions. However, the existing code's continue may be intentional to allow Volcano to explore partially-resolved states during optimization. The impact is moderate as it tightens the cost gate.	Medium
	Fail fast on unmarked inputs The method returns an empty list when the node isn't an OpenSearchRelNode, which causes computeViableBackends to produce an empty intersection. This breaks join planning when inputs aren't yet marked. Instead, return a sentinel or throw an exception to fail fast and surface the issue during development. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchJoinRule.java [128-138] private static List<String> viableBackendsOf(RelNode rel) { RelNode unwrapped = rel; if (unwrapped instanceof HepRelVertex vertex) { unwrapped = vertex.getCurrentRel(); } if (unwrapped instanceof OpenSearchRelNode osNode) { return osNode.getViableBackends(); } - // Not yet marked — empty list forces the fallback path above. - return List.of(); + throw new IllegalStateException( + "Join input not yet marked as OpenSearchRelNode: " + unwrapped.getClass().getSimpleName() + ); } Suggestion importance[1-10]: 3 __ Why: The suggestion to throw an exception instead of returning an empty list is reasonable for catching unmarked inputs during development. However, the comment in the existing code explicitly states "Not yet marked — empty list forces the fallback path above," suggesting this is intentional behavior. The suggestion may break the intended fallback mechanism.	Low
General	Ensure executor pool cleanup on timeout The executor pool is not forcibly terminated if awaitTermination times out or if an exception occurs before the shutdown completes. Add pool.shutdownNow() in a catch block or after the awaitTermination assertion to ensure threads are interrupted and resources are released even on failure. sandbox/plugins/analytics-backend-datafusion/src/internalClusterTest/java/org/opensearch/be/datafusion/CoordinatorJoinMultiNodeIT.java [293-316] -public void testConcurrentJoinsAreIsolated() throws Exception { - final int NUM_KEYS = 30; - final int N_QUERIES = 4; - String t1 = "join_mn_conc_t1"; - String t2 = "join_mn_conc_t2"; - createParquetIndex(t1, NUM_SHARDS, "v"); - createParquetIndex(t2, NUM_SHARDS, "w"); - indexUnique(t1, "v", 1, NUM_KEYS, k -> k * 11); - indexUnique(t2, "w", 1, NUM_KEYS, k -> k * 113); - - ExecutorService pool = Executors.newFixedThreadPool(N_QUERIES); - try { - @SuppressWarnings("unchecked") - CompletableFuture<PPLResponse>[] futures = new CompletableFuture[N_QUERIES]; - for (int i = 0; i < N_QUERIES; i++) { - futures[i] = CompletableFuture.supplyAsync( - () -> executePPL("source=" + t1 + " | join on " + t1 + ".k = " + t2 + ".k " + t2), - pool - ); +ExecutorService pool = Executors.newFixedThreadPool(N_QUERIES); +try { + @SuppressWarnings("unchecked") + CompletableFuture<PPLResponse>[] futures = new CompletableFuture[N_QUERIES]; + for (int i = 0; i < N_QUERIES; i++) { + futures[i] = CompletableFuture.supplyAsync( + () -> executePPL("source=" + t1 + " | join on " + t1 + ".k = " + t2 + ".k " + t2), + pool + ); + } + for (int i = 0; i < N_QUERIES; i++) { + PPLResponse response; + try { + response = futures[i].get(60, TimeUnit.SECONDS); + } catch (ExecutionException e) { + throw new AssertionError("query " + i + " threw", e.getCause()); } - for (int i = 0; i < N_QUERIES; i++) { - PPLResponse response; - try { - response = futures[i].get(60, TimeUnit.SECONDS); - } catch (ExecutionException e) { - throw new AssertionError("query " + i + " threw", e.getCause()); - } - assertColumns(response, "k", "v", "w"); - assertEquals("query " + i + " row count", NUM_KEYS, response.getRows().size()); - } - } finally { - pool.shutdown(); - assertTrue("executor must terminate", pool.awaitTermination(10, TimeUnit.SECONDS)); + assertColumns(response, "k", "v", "w"); + assertEquals("query " + i + " row count", NUM_KEYS, response.getRows().size()); + } +} finally { + pool.shutdown(); + if (!pool.awaitTermination(10, TimeUnit.SECONDS)) { + pool.shutdownNow(); } } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies a resource leak risk: if awaitTermination times out, the executor pool may not be forcibly shut down. Adding shutdownNow() ensures threads are interrupted and resources are released. This is a valid improvement for robustness in test cleanup.	Medium
	Preserve locality when degrading to ANY When a HASH key cannot be mapped, the method returns a new distribution with locality=null. This may cause downstream code expecting a non-null Locality to fail. Verify that callers handle null locality correctly, or consider preserving the original locality when degrading to Type.ANY. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/OpenSearchDistribution.java [130-146] public RelDistribution apply(Mappings.TargetMapping mapping) { ... List<Integer> newKeys = new java.util.ArrayList<>(keys.size()); for (int key : keys) { int target = mapping.getTargetOpt(key); if (target < 0) { - return new OpenSearchDistribution(traitDef, null, Type.ANY, List.of(), null, null); + return new OpenSearchDistribution(traitDef, locality, Type.ANY, List.of(), null, null); } newKeys.add(target); } return new OpenSearchDistribution(traitDef, locality, Type.HASH_DISTRIBUTED, newKeys, tableId, shardCount); } Suggestion importance[1-10]: 6 __ Why: The suggestion raises a valid concern: when degrading to Type.ANY, setting locality=null may cause issues if downstream code expects a non-null Locality. Preserving the original locality is a safer default. However, the impact depends on how callers handle null locality, which is not fully clear from the diff alone.	Low
	Prevent infinite loop on cyclic graphs The loop does not guard against cycles in the RelNode graph. If a malformed plan contains a cycle, this method will loop indefinitely. Add a visited-node set or a maximum iteration count to prevent infinite loops. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchSortRule.java [154-173] private static int remapInputIndexThroughProjects(int index, RelNode node, OpenSearchSort inner) { RelNode current = RelNodeUtils.unwrapHep(node); int idx = index; + Set<RelNode> visited = new HashSet<>(); while (current instanceof OpenSearchProject project) { + if (!visited.add(current)) { + return -1; + } if (idx < 0 || idx >= project.getProjects().size()) { return -1; } RexNode expr = project.getProjects().get(idx); if (!(expr instanceof RexInputRef ref)) { return -1; } idx = ref.getIndex(); current = RelNodeUtils.unwrapHep(project.getInput()); } return current == inner ? idx : -1; } Suggestion importance[1-10]: 5 __ Why: The suggestion correctly identifies a potential ...

[github-actions]

✅ Gradle check result for 4570b3f: SUCCESS

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 73.45%. Comparing base (446a1c9) to head (b01917a).
⚠️ Report is 2 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #21639      +/-   ##
============================================
- Coverage     73.49%   73.45%   -0.04%     
+ Complexity    74624    74570      -54     
============================================
  Files          5980     5980              
  Lines        338825   338825              
  Branches      48857    48857              
============================================
- Hits         249010   248889     -121     
- Misses        70041    70093      +52     
- Partials      19774    19843      +69     

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

[mch2]

@expani Thanks for the review - i've swapped this to be pure CBO, no hardcoded HEP wrapping - covers single shard case and multi nicely with two traits. Still polishing a few things and will open back up

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit b01917a)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Possible Issue In the apply method, when a HASH key cannot be mapped through a projection, the code degrades to ANY distribution. However, it creates a new OpenSearchDistribution with locality=null, which may cause issues downstream if code expects locality to be non-null for certain distribution types. The original distribution's locality is discarded without validation that null is acceptable in all contexts where ANY distributions are used. List<Integer> newKeys = new java.util.ArrayList<>(keys.size()); for (int key : keys) { int target = mapping.getTargetOpt(key); if (target < 0) { return new OpenSearchDistribution(traitDef, null, Type.ANY, List.of(), null, null); } Possible Issue The test creates indices with NUM_REPLICAS = 0 due to a known limitation (composite engine doesn't implement acquireSafeIndexCommit). The class javadoc states replicas should be enabled once that's fixed, but there's no tracking issue reference or TODO comment in the code itself. If someone enables replicas without fixing the underlying issue, tests will fail with RecoveryFailedException. Consider adding an explicit runtime check or a more prominent warning. /** See class javadoc — composite engine doesn't implement acquireSafeIndexCommit * yet, so replicas can't be recovered. Fixed at 0 until that lands. */ private static final int NUM_REPLICAS = 0; Possible Issue The findStagePlanByFragmentType helper walks the DAG to find a stage with a specific fragment type, but throws AssertionError if not found. In multi-stage plans where the structure changes (e.g., after optimization passes), this could fail unexpectedly if the expected fragment type is optimized away or moved. The error message doesn't indicate which test called it, making debugging harder when the DAG structure evolves. RelNode arm1 = buildSortedAggArm(); RelNode arm2 = buildSortedAggArm(); return LogicalUnion.create(List.of(arm1, arm2), true); } private RelNode buildSortedAggArm() { RelNode scan = stubScan(mockTable("test_index", "status", "size"));

[github-actions]

Persistent review updated to latest commit 35703cb

[github-actions]

Persistent review updated to latest commit 835c59a

[github-actions]

Persistent review updated to latest commit 2575a64

[github-actions]

✅ Gradle check result for 2575a64: SUCCESS

[github-actions]

Persistent review updated to latest commit b01917a

[github-actions]

✅ Gradle check result for b01917a: SUCCESS

[LantaoJin]

Basically LGTM, broadcast join (M1) was done in local and I will submit in tomorrow after merging and rebasing with this PR.