Skip to content

Restrict remote ENRICH after FORK #131945

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
Jul 28, 2025
Merged

Restrict remote ENRICH after FORK #131945

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
8ac9dbe
Restrict remote LOOKUP JOIN after FORK
smalyshev Jul 25, 2025
98c334d
Update docs/changelog/131945.yaml
smalyshev Jul 25, 2025
d1d1971
Refactor and make For a pipeline breaker
smalyshev Jul 25, 2025
0acfc48
Add comment
smalyshev Jul 28, 2025
1b47591
Merge branch 'main' into remote-enrich-fork
smalyshev Jul 28, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/changelog/131945.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
pr: 131945
summary: Restrict remote ENRICH after FORK
area: ES|QL
type: bug
issues:
- 131445
25 changes: 8 additions & 17 deletions x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/Enrich.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
Expand Down Expand Up @@ -309,30 +310,20 @@ private static void checkForPlansForbiddenBeforeRemoteEnrich(Enrich enrich, Fail
return;
}

// TODO: shouldn't we also include FORK? Everything downstream from FORK should be coordinator-only.
// https://github.com/elastic/elasticsearch/issues/131445
boolean[] aggregate = { false };
boolean[] coordinatorOnlyEnrich = { false };
boolean[] lookupJoin = { false };
Set<String> badCommands = new HashSet<>();

enrich.forEachUp(LogicalPlan.class, u -> {
if (u instanceof Aggregate) {
aggregate[0] = true;
badCommands.add("STATS");
} else if (u instanceof Enrich upstreamEnrich && upstreamEnrich.mode() == Enrich.Mode.COORDINATOR) {
coordinatorOnlyEnrich[0] = true;
badCommands.add("another ENRICH with coordinator policy");
} else if (u instanceof LookupJoin) {
lookupJoin[0] = true;
badCommands.add("LOOKUP JOIN");
} else if (u instanceof Fork) {
badCommands.add("FORK");
}
});

if (aggregate[0]) {
failures.add(fail(enrich, "ENRICH with remote policy can't be executed after STATS"));
}
if (coordinatorOnlyEnrich[0]) {
failures.add(fail(enrich, "ENRICH with remote policy can't be executed after another ENRICH with coordinator policy"));
}
if (lookupJoin[0]) {
failures.add(fail(enrich, "ENRICH with remote policy can't be executed after LOOKUP JOIN"));
}
badCommands.forEach(c -> failures.add(fail(enrich, "ENRICH with remote policy can't be executed after " + c)));
}
}
2 changes: 1 addition & 1 deletion x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/Fork.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@
* A Fork is a n-ary {@code Plan} where each child is a sub plan, e.g.
* {@code FORK [WHERE content:"fox" ] [WHERE content:"dog"] }
*/
public class Fork extends LogicalPlan implements PostAnalysisPlanVerificationAware, TelemetryAware {
public class Fork extends LogicalPlan implements PostAnalysisPlanVerificationAware, TelemetryAware, PipelineBreaker {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, pipeline breakers are also being used in node-/cluster-level reduction here; I don't think this applies to FORK, as FORK cannot be applied in multiple steps (yet?); each FORK branch seems to be hooked up to the same exchange sink on the coordinator. @ioanatia , please correct me if I'm wrong.

We may be introducing a subtle bug here, or maybe not; in any case, it indicates that the PipelineBreaker interface needs further refinement, because for validation purposes FORK definitely is a pipeline breaker, even if it can't be used for node-/cluster-level reduction.

Maybe we should just add a method PipelineBreaker#isReducer or similar? It could default to true and return false for Fork.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am not sure adding PipelineBreaker has any effect at all.

PipelineBreaker is indeed used here - but I don't see in which case we would ever get to a point where a FragmentExec contains a logical plan that contains a Fork plan. Let me know if that's not the case.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for confirming @ioanatia, that makes sense ! If we're never pulling FORKS out of fragments, then this change should be safe.

If we're marking Fork as PipelineBreaker, maybe we could at least add a comment here to remind future us that Fork's never going to show up here.


public static final String FORK_FIELD = "_fork";
public static final int MAX_BRANCHES = 8;
Expand Down
2 changes: 2 additions & 0 deletions x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,8 @@ public static PhysicalPlan reductionPlan(PhysicalPlan plan) {
}
final FragmentExec fragment = (FragmentExec) fragments.getFirst();

// Though FORK is technically a pipeline breaker, it should never show up here.
// See also: https://github.com/elastic/elasticsearch/pull/131945/files#r2235572935
final var pipelineBreakers = fragment.fragment().collectFirstChildren(p -> p instanceof PipelineBreaker);
if (pipelineBreakers.isEmpty()) {
return null;
Expand Down
8 changes: 8 additions & 0 deletions x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/VerifierTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2457,6 +2457,14 @@ public void testRemoteEnrichAfterCoordinatorOnlyPlans() {
| ENRICH _remote:languages ON language_code
""", analyzer);
assertThat(err, containsString("7:3: ENRICH with remote policy can't be executed after another ENRICH with coordinator policy"));

err = error("""
FROM test
| FORK (WHERE languages == 1) (WHERE languages == 2)
| EVAL language_code = languages
| ENRICH _remote:languages ON language_code
""", analyzer);
assertThat(err, containsString("4:3: ENRICH with remote policy can't be executed after FORK"));
}

private void checkFullTextFunctionsInStats(String functionInvocation) {
Expand Down
Loading