[Analytics-Engine] Lower LITERAL_AGG and accept all join condition shapes

[songkant-aws]

Description

Two complementary marking-phase fixes that build on top of #21795. Both keep Calcite as the source of truth and unblock end-to-end correlated-subquery execution on the analytics-engine route.

1. LITERAL_AGG lowering — after RelDecorrelator runs in PlannerImpl.removeSubQueries, a new Hep pass invokes ExtractLiteralAggRule (ported from Apache Impala with attribution) to rewrite LITERAL_AGG(literal) aggregate calls into an Aggregate + Project shape. SubQueryRemoveRule.rewriteIn emits LITERAL_AGG for NOT IN's null-aware semantics; Calcite ships no rule to remove it, and each downstream engine handles it differently (Druid registers a native impl, Ignite wires an accumulator, Impala lowers, Flink forks SubQueryRemoveRule). The Impala approach is a clean Calcite-rule-style rewrite — port it.

2. OpenSearchJoinRule.matches() relaxed — accept all join condition shapes, not just equi or mixed equi+non-equi. RelDecorrelator's output for correlated subqueries with non-equi correlation predicates (e.g. WHERE id > uid) is a pure non-equi LogicalJoin. DataFusion handles non-equi via NestedLoopJoin downstream, so there is no capability-level reason to reject these in marking.

PlannerImpl is also refactored to a fluent HepPhase builder so each Hep pass attaches the existing RuleProfilingListener consistently and the phase boundary is recorded in the planner profile.

Dependency on #21795

This PR is draft until #21795 merges. The branch contains all of #21795's commits plus the three commits in this layer; once #21795 lands, this PR will rebase onto main and only show the layered diff.

Test results (on this branch = #21795 + these fixes)

Suite	Result
ExistsSubqueryCommandIT	11/11
ScalarSubqueryCommandIT	13/13
InSubqueryCommandIT	11/13
StreamstatsCommandIT.testStreamstatsReset	pass
JoinCommandIT	12/12 (no regression from join relaxation)
Total	36/38 ≈ 95%

The two remaining InSubqueryCommandIT failures are unrelated to subquery decorrelation:

• testTwoExpressionsInSubquery — row-ordering assertion vs. hash-join bucketing. Test issue.
• testTwoExpressionsNotInSubquery — OpenSearchAggregateSplitRule rejects nullable filter expressions ("filter must be BOOLEAN NOT NULL") on a multi-column NOT IN's COUNT FILTER. Independent split-rule bug.

Related Issues

Builds on #21795.

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.

[github-actions]

PR Code Analyzer ❗

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

Path	Line	Severity	Description
sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/PlannerImpl.java	87	medium	Debug logging statement logs full query plan trees at INFO level in production code ('After removeSubQueries:\n{}' + RelOptUtil.toString). This can expose sensitive query structure, table names, filter predicates, and join conditions in log files. Appears to be development debug code inadvertently left in, but could leak schema or data access patterns from production query plans.

The table above displays the top 10 most important findings.

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

---

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

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

---

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

Thanks.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 021cce9)

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 The rule does not verify that the LITERAL_AGG's literal type matches the aggregate call's declared return type. If a future rule or frontend constructs LITERAL_AGG with a mismatched type (e.g., LITERAL_AGG(INTEGER_LITERAL) but the AggregateCall declares BOOLEAN return type), the rewritten Project will emit a literal of the wrong type at that output position, causing a schema mismatch downstream. The Preconditions check on line 117-120 only asserts that exactly one RexLiteral exists, not that its type is compatible with ac.getType(). if (ac.getAggregation().getName().equals("LITERAL_AGG")) { Preconditions.checkState( ac.rexList.size() == 1 && ac.rexList.get(0) instanceof RexLiteral, "LITERAL_AGG must carry exactly one RexLiteral preOperand" ); literalAtSlot.add(ac.rexList.get(0)); } else {

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 021cce9

Explore these optional code suggestions:

Category	Suggestion	Impact
General	Add bounds check for index If literalAtSlot contains only nulls (no LITERAL_AGG calls), the early return at line 128 should prevent reaching this code. However, if that check is bypassed or modified, rebuiltCallIdx could exceed keptCalls.size(), causing makeInputRef to reference a non-existent column. Add a bounds check or assertion to ensure rebuiltCallIdx stays within valid range. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/ExtractLiteralAggRule.java [155-163] int rebuiltCallIdx = 0; for (RexNode literal : literalAtSlot) { if (literal != null) { projects.add(literal); } else { + assert rebuiltCallIdx < keptCalls.size() : "rebuiltCallIdx out of bounds"; projects.add(rexBuilder.makeInputRef(newAggregate, groupCount + rebuiltCallIdx)); rebuiltCallIdx++; } } Suggestion importance[1-10]: 5 __ Why: The suggestion adds a defensive assertion to ensure rebuiltCallIdx stays within bounds. While the early return at line 128 should prevent this scenario, the assertion provides additional safety. However, the logic is already protected by the partition logic that ensures rebuiltCallIdx increments only for non-literal slots, making this a minor defensive improvement.	Low
	Handle malformed LITERAL_AGG gracefully The Preconditions.checkState will throw an IllegalStateException if the condition fails, but this happens during rule execution. If malformed LITERAL_AGG calls reach this point, the planner will crash. Consider adding a guard in matches() to reject aggregates with malformed LITERAL_AGG calls before onMatch() executes. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/ExtractLiteralAggRule.java [116-122] if (ac.getAggregation().getName().equals("LITERAL_AGG")) { - Preconditions.checkState( - ac.rexList.size() == 1 && ac.rexList.get(0) instanceof RexLiteral, - "LITERAL_AGG must carry exactly one RexLiteral preOperand" - ); + if (ac.rexList.size() != 1 || !(ac.rexList.get(0) instanceof RexLiteral)) { + return; // Skip malformed LITERAL_AGG instead of crashing + } literalAtSlot.add(ac.rexList.get(0)); } Suggestion importance[1-10]: 4 __ Why: The suggestion proposes replacing Preconditions.checkState with a silent return to handle malformed LITERAL_AGG calls. While this prevents crashes, silently skipping malformed aggregates could hide bugs. The current approach of failing fast with a clear error message is more appropriate for detecting plan-time issues.	Low
	Avoid unnecessary defensive copy The List.copyOf(rules) creates a defensive copy, but if rules is already immutable (e.g., from List.of(...)), this creates an unnecessary allocation. Consider checking if the collection is already immutable before copying, or document that callers should pass mutable collections if performance is critical. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/HepPhase.java [111-115] public HepPhase addRuleCollection(Collection<? extends RelOptRule> rules) { - List<RelOptRule> snapshot = List.copyOf(rules); + List<RelOptRule> snapshot = rules instanceof List && ((List<?>) rules).getClass().getName().contains("ImmutableCollections") + ? (List<RelOptRule>) rules + : List.copyOf(rules); instructions.add(builder -> builder.addRuleCollection(snapshot)); return this; } Suggestion importance[1-10]: 3 __ Why: The suggestion addresses a minor performance optimization by avoiding unnecessary copies of immutable collections. However, the proposed implementation using getClass().getName().contains("ImmutableCollections") is fragile and relies on internal JDK implementation details. The current code is correct and the performance impact is negligible in typical usage.	Low

---

Previous suggestions

Suggestions up to commit a71924d

Category	Suggestion	Impact
General	Validate kept calls count matches expectations If the number of non-null entries in literalAtSlot doesn't match keptCalls.size(), the rebuiltCallIdx could exceed the bounds of the rebuilt aggregate's output columns, causing makeInputRef() to reference an invalid column position. Add a validation check to ensure consistency between keptCalls size and the count of null entries in literalAtSlot. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/ExtractLiteralAggRule.java [155-163] int rebuiltCallIdx = 0; +int expectedKeptCalls = (int) literalAtSlot.stream().filter(Objects::isNull).count(); +Preconditions.checkState(expectedKeptCalls == keptCalls.size(), + "Mismatch between kept calls and non-literal slots"); for (RexNode literal : literalAtSlot) { if (literal != null) { projects.add(literal); } else { projects.add(rexBuilder.makeInputRef(newAggregate, groupCount + rebuiltCallIdx)); rebuiltCallIdx++; } } Suggestion importance[1-10]: 8 __ Why: This validation check would catch a critical logic error where the count of non-literal slots doesn't match keptCalls.size(), which could lead to invalid column references. The precondition ensures internal consistency and prevents potential runtime errors from invalid makeInputRef() calls.	Medium
	Add null check before accessing list The precondition check occurs after the equals() comparison succeeds, but if ac.rexList is empty or null, calling size() or get(0) will throw an exception before the precondition message is displayed. Verify rexList is non-null and non-empty before accessing its elements to provide clearer error messages. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/ExtractLiteralAggRule.java [116-122] if (ac.getAggregation().getName().equals("LITERAL_AGG")) { Preconditions.checkState( - ac.rexList.size() == 1 && ac.rexList.get(0) instanceof RexLiteral, + ac.rexList != null && ac.rexList.size() == 1 && ac.rexList.get(0) instanceof RexLiteral, "LITERAL_AGG must carry exactly one RexLiteral preOperand" ); literalAtSlot.add(ac.rexList.get(0)); } Suggestion importance[1-10]: 7 __ Why: Adding a null check for ac.rexList before accessing its elements would prevent potential NullPointerException and provide clearer error messages. This is a defensive programming practice that improves robustness, though the likelihood of rexList being null in practice may be low.	Medium
	Compose multiple post-processors instead of overwriting The postProcess field can be overwritten by multiple calls to postProcess(), potentially losing previously registered transformations. Consider either documenting that only the last call takes effect, or composing multiple post-processors using andThen() to preserve all registered transformations. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/HepPhase.java [121-124] public HepPhase postProcess(UnaryOperator<RelNode> postProcess) { - this.postProcess = postProcess; + this.postProcess = this.postProcess.andThen(postProcess); return this; } Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies that multiple postProcess() calls will overwrite the previous transformation. Using andThen() to compose transformations is a valid improvement that preserves all registered post-processors, though the current usage pattern in the PR shows only single calls.	Low

Suggestions up to commit 9c79999

Category	Suggestion	Impact
General	Add null check before accessing rexList The precondition check occurs after the name comparison, but if rexList is null or empty, calling size() or get(0) will throw NullPointerException or IndexOutOfBoundsException before the checkState message is shown. Verify rexList is non-null and non-empty before accessing it to ensure the precondition message is meaningful. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/ExtractLiteralAggRule.java [116-121] if (ac.getAggregation().getName().equals("LITERAL_AGG")) { Preconditions.checkState( - ac.rexList.size() == 1 && ac.rexList.get(0) instanceof RexLiteral, + ac.rexList != null && ac.rexList.size() == 1 && ac.rexList.get(0) instanceof RexLiteral, "LITERAL_AGG must carry exactly one RexLiteral preOperand" ); literalAtSlot.add(ac.rexList.get(0)); } Suggestion importance[1-10]: 7 __ Why: Valid defensive programming suggestion. Adding a null check before accessing rexList ensures the precondition error message is meaningful if rexList is null, improving debuggability.	Medium
	Add bounds check for rebuiltCallIdx If literalAtSlot contains more non-null entries than expected, rebuiltCallIdx could exceed the bounds of keptCalls, causing makeInputRef to reference a non-existent column in newAggregate. Add a bounds check or assertion to ensure rebuiltCallIdx stays within valid range. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/ExtractLiteralAggRule.java [155-163] int rebuiltCallIdx = 0; for (RexNode literal : literalAtSlot) { if (literal != null) { projects.add(literal); } else { + Preconditions.checkState(rebuiltCallIdx < keptCalls.size(), "rebuiltCallIdx out of bounds"); projects.add(rexBuilder.makeInputRef(newAggregate, groupCount + rebuiltCallIdx)); rebuiltCallIdx++; } } Suggestion importance[1-10]: 6 __ Why: The suggestion adds defensive validation. However, the logic ensures rebuiltCallIdx increments only for non-literal slots, matching keptCalls.size() by construction, making this primarily a safety assertion rather than fixing an actual bug.	Low
	Compose multiple post-processors instead of overwriting The postProcess field can be overwritten by multiple calls to postProcess(), potentially losing previously registered transformations. Consider either documenting that only the last call takes effect, or composing multiple post-processors using andThen() to preserve all registered transformations. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/HepPhase.java [121-124] public HepPhase postProcess(UnaryOperator<RelNode> postProcess) { - this.postProcess = postProcess; + this.postProcess = this.postProcess.andThen(postProcess); return this; } Suggestion importance[1-10]: 5 __ Why: The suggestion correctly identifies that multiple postProcess() calls overwrite the field. However, the current usage pattern (single call per phase) makes this a minor enhancement rather than a bug fix.	Low

Suggestions up to commit ac27ee0

Category	Suggestion	Impact
Possible issue	Add null check for rexList The precondition check should validate that rexList is not null before accessing its size. If rexList is null, calling size() will throw a NullPointerException, bypassing the intended validation message. Add a null check before the size validation. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/ExtractLiteralAggRule.java [117-120] Preconditions.checkState( - ac.rexList.size() == 1 && ac.rexList.get(0) instanceof RexLiteral, + ac.rexList != null && ac.rexList.size() == 1 && ac.rexList.get(0) instanceof RexLiteral, "LITERAL_AGG must carry exactly one RexLiteral preOperand" ); Suggestion importance[1-10]: 7 __ Why: Adding a null check for ac.rexList before accessing its size() method prevents a potential NullPointerException. However, in Calcite's API, AggregateCall.rexList is typically non-null (though possibly empty), so this is more of a defensive programming practice than addressing a likely bug.	Medium
General	Handle post-processing exceptions properly If postProcess.apply() throws an exception, the listener's endPhase() will still be called in the finally block, but the exception from postProcess will propagate. Consider wrapping the post-processing in a try-catch to ensure proper error handling and phase tracking even when post-processing fails. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/HepPhase.java [125-130] try { planner.setRoot(input); - return postProcess.apply(planner.findBestExp()); + RelNode result = planner.findBestExp(); + return postProcess.apply(result); +} catch (Exception e) { + throw new RuntimeException("Phase '" + name + "' failed during post-processing", e); } finally { if (listener != null) listener.endPhase(name); } Suggestion importance[1-10]: 5 __ Why: The suggestion adds exception handling for postProcess.apply() with a more descriptive error message. While this improves error reporting, the current code already propagates exceptions correctly and the finally block ensures endPhase() is called. The improvement is marginal—better error messages but no functional correctness issue.	Low

Suggestions up to commit 5c54903

Category	Suggestion	Impact
Possible issue	Add null check for rexList Add null-safety check before accessing ac.rexList. If rexList is null or empty, the current code will throw a NullPointerException or IndexOutOfBoundsException instead of the intended precondition failure with a clear message. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/ExtractLiteralAggRule.java [117-120] Preconditions.checkState( - ac.rexList.size() == 1 && ac.rexList.get(0) instanceof RexLiteral, + ac.rexList != null && ac.rexList.size() == 1 && ac.rexList.get(0) instanceof RexLiteral, "LITERAL_AGG must carry exactly one RexLiteral preOperand" ); Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies a potential NullPointerException if ac.rexList is null. Adding the null check improves robustness and provides clearer error messaging, though the likelihood of rexList being null in practice may be low given Calcite's internal construction patterns.	Medium

Suggestions up to commit de43ffb

Category	Suggestion	Impact
Possible issue	Add null check for rexList The precondition check should validate that ac.rexList is not null before accessing its size. If rexList is null, calling size() will throw a NullPointerException instead of the intended IllegalStateException with a descriptive message. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/ExtractLiteralAggRule.java [116-119] Preconditions.checkState( - ac.rexList.size() == 1 && ac.rexList.get(0) instanceof RexLiteral, + ac.rexList != null && ac.rexList.size() == 1 && ac.rexList.get(0) instanceof RexLiteral, "LITERAL_AGG must carry exactly one RexLiteral preOperand" ); Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies that ac.rexList could be null, which would cause a NullPointerException before the Preconditions.checkState can provide a meaningful error message. However, this is a defensive programming improvement rather than a critical bug fix, as Calcite's AggregateCall typically ensures rexList is non-null.	Low
	Validate rebuiltCallIdx bounds during iteration The loop assumes keptCalls.size() matches the number of null entries in literalAtSlot, but if this invariant is violated (e.g., due to a bug in the partitioning logic), rebuiltCallIdx could exceed the bounds of the rebuilt aggregate's row type. Add a bounds check or assertion to prevent invalid RexInputRef construction. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/ExtractLiteralAggRule.java [154-162] int rebuiltCallIdx = 0; for (RexNode literal : literalAtSlot) { if (literal != null) { projects.add(literal); } else { + Preconditions.checkState(rebuiltCallIdx < keptCalls.size(), + "rebuiltCallIdx out of bounds: %s >= %s", rebuiltCallIdx, keptCalls.size()); projects.add(rexBuilder.makeInputRef(newAggregate, groupCount + rebuiltCallIdx)); rebuiltCallIdx++; } } Suggestion importance[1-10]: 5 __ Why: The suggestion adds a defensive bounds check for rebuiltCallIdx, which could prevent an invalid RexInputRef if the partitioning logic has a bug. However, the partitioning logic appears correct (each null in literalAtSlot corresponds to an entry in keptCalls), making this a low-probability edge case. The check adds safety but may be overly defensive.	Low
General	Track phase completion for listener If planner.findBestExp() or postProcess.apply() throws an exception, the finally block still calls listener.endPhase(name), but listener.beginPhase(name) was already invoked. This creates a phase-end event without a successful phase completion, which may corrupt profiling metrics. Consider wrapping the exception or resetting listener state. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/HepPhase.java [110-131] public RelNode run(RelNode input, RuleProfilingListener listener) { ... + boolean phaseCompleted = false; try { planner.setRoot(input); - return postProcess.apply(planner.findBestExp()); + RelNode result = postProcess.apply(planner.findBestExp()); + phaseCompleted = true; + return result; } finally { - if (listener != null) listener.endPhase(name); + if (listener != null) { + if (!phaseCompleted) { + // Phase failed; optionally log or mark as incomplete + } + listener.endPhase(name); + } } } Suggestion importance[1-10]: 4 __ Why: The suggestion raises a valid concern about listener.endPhase(name) being called even when the phase fails. However, the current design appears intentional—the finally block ensures phase boundaries are always matched for profiling consistency. The suggested change adds complexity without clear evidence that the current behavior is incorrect or that profiling metrics are corrupted by exceptions.	Low

[github-actions]

❕ Gradle check result for 6c34247: UNSTABLE

Please review all flaky tests that succeeded after retry and create an issue if one does not already exist to track the flaky failure.

[codecov]

Codecov Report

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

Additional details and impacted files

@@            Coverage Diff            @@
##               main   #21804   +/-   ##
=========================================
  Coverage     73.50%   73.51%           
- Complexity    75546    75583   +37     
=========================================
  Files          6034     6034           
  Lines        342661   342661           
  Branches      49294    49294           
=========================================
+ Hits         251879   251907   +28     
+ Misses        70790    70749   -41     
- Partials      19992    20005   +13     

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

Persistent review updated to latest commit aa33e4d

[github-actions]

❌ Gradle check result for aa33e4d: FAILURE

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

[github-actions]

Persistent review updated to latest commit 02e8b22

[github-actions]

✅ Gradle check result for 02e8b22: SUCCESS

[github-actions]

Persistent review updated to latest commit 31df763

[github-actions]

✅ Gradle check result for 31df763: SUCCESS

[github-actions]

Persistent review updated to latest commit 8fba98f

[github-actions]

❌ Gradle check result for 8fba98f: FAILURE

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

[github-actions]

Persistent review updated to latest commit de43ffb

[github-actions]

❌ Gradle check result for de43ffb: FAILURE

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

[github-actions]

Persistent review updated to latest commit 5c54903

[github-actions]

Persistent review updated to latest commit ac27ee0

[github-actions]

❌ Gradle check result for ac27ee0: FAILURE

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

[github-actions]

Persistent review updated to latest commit 9c79999

[github-actions]

❌ Gradle check result for 9c79999: FAILURE

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

[github-actions]

Persistent review updated to latest commit a71924d

[github-actions]

❌ Gradle check result for a71924d: FAILURE

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

[github-actions]

Persistent review updated to latest commit 021cce9

[github-actions]

✅ Gradle check result for 021cce9: SUCCESS