|
|
@@ -68,8 +68,10 @@ import org.elasticsearch.xpack.esql.plan.physical.EvalExec;
|
|
|
import org.elasticsearch.xpack.esql.plan.physical.ExchangeExec;
|
|
|
import org.elasticsearch.xpack.esql.plan.physical.FieldExtractExec;
|
|
|
import org.elasticsearch.xpack.esql.plan.physical.FilterExec;
|
|
|
+import org.elasticsearch.xpack.esql.plan.physical.GrokExec;
|
|
|
import org.elasticsearch.xpack.esql.plan.physical.LimitExec;
|
|
|
import org.elasticsearch.xpack.esql.plan.physical.LocalSourceExec;
|
|
|
+import org.elasticsearch.xpack.esql.plan.physical.LookupJoinExec;
|
|
|
import org.elasticsearch.xpack.esql.plan.physical.MvExpandExec;
|
|
|
import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan;
|
|
|
import org.elasticsearch.xpack.esql.plan.physical.ProjectExec;
|
|
|
@@ -109,6 +111,7 @@ import static org.elasticsearch.xpack.esql.EsqlTestUtils.configuration;
|
|
|
import static org.elasticsearch.xpack.esql.EsqlTestUtils.loadMapping;
|
|
|
import static org.elasticsearch.xpack.esql.EsqlTestUtils.unboundLogicalOptimizerContext;
|
|
|
import static org.elasticsearch.xpack.esql.EsqlTestUtils.withDefaultLimitWarning;
|
|
|
+import static org.elasticsearch.xpack.esql.analysis.AnalyzerTestUtils.defaultLookupResolution;
|
|
|
import static org.elasticsearch.xpack.esql.core.querydsl.query.Query.unscore;
|
|
|
import static org.elasticsearch.xpack.esql.plan.physical.EsStatsQueryExec.StatsType;
|
|
|
import static org.hamcrest.Matchers.contains;
|
|
|
@@ -201,7 +204,7 @@ public class LocalPhysicalPlanOptimizerTests extends MapperServiceTestCase {
|
|
|
IndexResolution getIndexResult = IndexResolution.valid(test);
|
|
|
|
|
|
return new Analyzer(
|
|
|
- new AnalyzerContext(config, new EsqlFunctionRegistry(), getIndexResult, enrichResolution),
|
|
|
+ new AnalyzerContext(config, new EsqlFunctionRegistry(), getIndexResult, defaultLookupResolution(), enrichResolution),
|
|
|
new Verifier(new Metrics(new EsqlFunctionRegistry()), new XPackLicenseState(() -> 0L))
|
|
|
);
|
|
|
}
|
|
|
@@ -1325,6 +1328,174 @@ public class LocalPhysicalPlanOptimizerTests extends MapperServiceTestCase {
|
|
|
assertThat(Expressions.names(fields), contains("_meta_field", "gender", "hire_date", "job", "job.raw", "languages", "long_noidx"));
|
|
|
}
|
|
|
|
|
|
+ /*
|
|
|
+ * LimitExec[1000[INTEGER]]
|
|
|
+ * \_AggregateExec[[language_code{r}#6],[COUNT(emp_no{f}#12,true[BOOLEAN]) AS c#11, language_code{r}#6],FINAL,[language_code{r}#6, $
|
|
|
+ * $c$count{r}#25, $$c$seen{r}#26],12]
|
|
|
+ * \_ExchangeExec[[language_code{r}#6, $$c$count{r}#25, $$c$seen{r}#26],true]
|
|
|
+ * \_AggregateExec[[languages{r}#15],[COUNT(emp_no{f}#12,true[BOOLEAN]) AS c#11, languages{r}#15 AS language_code#6],INITIAL,[langua
|
|
|
+ * ges{r}#15, $$c$count{r}#27, $$c$seen{r}#28],12]
|
|
|
+ * \_FieldExtractExec[emp_no{f}#12]<[],[]>
|
|
|
+ * \_EvalExec[[null[INTEGER] AS languages#15]]
|
|
|
+ * \_EsQueryExec[test], indexMode[standard], query[][_doc{f}#29], limit[], sort[] estimatedRowSize[12]
|
|
|
+ */
|
|
|
+ public void testMissingFieldsPurgesTheJoinLocally() {
|
|
|
+ var stats = EsqlTestUtils.statsForMissingField("languages");
|
|
|
+
|
|
|
+ var plan = plannerOptimizer.plan("""
|
|
|
+ from test
|
|
|
+ | keep emp_no, languages
|
|
|
+ | rename languages AS language_code
|
|
|
+ | lookup join languages_lookup ON language_code
|
|
|
+ | stats c = count(emp_no) by language_code
|
|
|
+ """, stats);
|
|
|
+
|
|
|
+ var limit = as(plan, LimitExec.class);
|
|
|
+ var agg = as(limit.child(), AggregateExec.class);
|
|
|
+ assertThat(Expressions.names(agg.output()), contains("c", "language_code"));
|
|
|
+
|
|
|
+ var exchange = as(agg.child(), ExchangeExec.class);
|
|
|
+ agg = as(exchange.child(), AggregateExec.class);
|
|
|
+ var extract = as(agg.child(), FieldExtractExec.class);
|
|
|
+ var eval = as(extract.child(), EvalExec.class);
|
|
|
+ var source = as(eval.child(), EsQueryExec.class);
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * LimitExec[1000[INTEGER]]
|
|
|
+ * \_AggregateExec[[language_code{r}#7],[COUNT(emp_no{r}#31,true[BOOLEAN]) AS c#17, language_code{r}#7],FINAL,[language_code{r}#7, $
|
|
|
+ * $c$count{r}#32, $$c$seen{r}#33],12]
|
|
|
+ * \_ExchangeExec[[language_code{r}#7, $$c$count{r}#32, $$c$seen{r}#33],true]
|
|
|
+ * \_AggregateExec[[language_code{r}#7],[COUNT(emp_no{r}#31,true[BOOLEAN]) AS c#17, language_code{r}#7],INITIAL,[language_code{r}#7,
|
|
|
+ * $$c$count{r}#34, $$c$seen{r}#35],12]
|
|
|
+ * \_GrokExec[first_name{f}#19,Parser[pattern=%{WORD:foo}, grok=org.elasticsearch.grok.Grok@75389ac1],[foo{r}#12]]
|
|
|
+ * \_MvExpandExec[emp_no{f}#18,emp_no{r}#31]
|
|
|
+ * \_ProjectExec[[emp_no{f}#18, languages{r}#21 AS language_code#7, first_name{f}#19]]
|
|
|
+ * \_FieldExtractExec[emp_no{f}#18, first_name{f}#19]<[],[]>
|
|
|
+ * \_EvalExec[[null[INTEGER] AS languages#21]]
|
|
|
+ * \_EsQueryExec[test], indexMode[standard], query[][_doc{f}#36], limit[], sort[] estimatedRowSize[112]
|
|
|
+ */
|
|
|
+ public void testMissingFieldsPurgesTheJoinLocallyThroughCommands() {
|
|
|
+ var stats = EsqlTestUtils.statsForMissingField("languages");
|
|
|
+
|
|
|
+ var plan = plannerOptimizer.plan("""
|
|
|
+ from test
|
|
|
+ | keep emp_no, languages, first_name
|
|
|
+ | rename languages AS language_code
|
|
|
+ | mv_expand emp_no
|
|
|
+ | grok first_name "%{WORD:foo}"
|
|
|
+ | lookup join languages_lookup ON language_code
|
|
|
+ | stats c = count(emp_no) by language_code
|
|
|
+ """, stats);
|
|
|
+
|
|
|
+ var limit = as(plan, LimitExec.class);
|
|
|
+ var agg = as(limit.child(), AggregateExec.class);
|
|
|
+ assertThat(Expressions.names(agg.output()), contains("c", "language_code"));
|
|
|
+
|
|
|
+ var exchange = as(agg.child(), ExchangeExec.class);
|
|
|
+ agg = as(exchange.child(), AggregateExec.class);
|
|
|
+ var grok = as(agg.child(), GrokExec.class);
|
|
|
+ var mvexpand = as(grok.child(), MvExpandExec.class);
|
|
|
+ var project = as(mvexpand.child(), ProjectExec.class);
|
|
|
+ var extract = as(project.child(), FieldExtractExec.class);
|
|
|
+ var eval = as(extract.child(), EvalExec.class);
|
|
|
+ var source = as(eval.child(), EsQueryExec.class);
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * LimitExec[1000[INTEGER]]
|
|
|
+ * \_AggregateExec[[language_code{r}#12],[COUNT(emp_no{r}#31,true[BOOLEAN]) AS c#17, language_code{r}#12],FINAL,[language_code{r}#12
|
|
|
+ * , $$c$count{r}#32, $$c$seen{r}#33],12]
|
|
|
+ * \_ExchangeExec[[language_code{r}#12, $$c$count{r}#32, $$c$seen{r}#33],true]
|
|
|
+ * \_AggregateExec[[language_code{r}#12],[COUNT(emp_no{r}#31,true[BOOLEAN]) AS c#17, language_code{r}#12],INITIAL,[language_code{r}#
|
|
|
+ * 12, $$c$count{r}#34, $$c$seen{r}#35],12]
|
|
|
+ * \_LookupJoinExec[[language_code{r}#12],[language_code{f}#29],[]]
|
|
|
+ * |_GrokExec[first_name{f}#19,Parser[pattern=%{NUMBER:language_code:int}, grok=org.elasticsearch.grok.Grok@764e5109],[languag
|
|
|
+ * e_code{r}#12]]
|
|
|
+ * | \_MvExpandExec[emp_no{f}#18,emp_no{r}#31]
|
|
|
+ * | \_ProjectExec[[emp_no{f}#18, languages{r}#21 AS language_code#7, first_name{f}#19]]
|
|
|
+ * | \_FieldExtractExec[emp_no{f}#18, first_name{f}#19]<[],[]>
|
|
|
+ * | \_EvalExec[[null[INTEGER] AS languages#21]]
|
|
|
+ * | \_EsQueryExec[test], indexMode[standard], query[][_doc{f}#36], limit[], sort[] estimatedRowSize[66]
|
|
|
+ * \_EsQueryExec[languages_lookup], indexMode[lookup], query[][_doc{f}#37], limit[], sort[] estimatedRowSize[4]
|
|
|
+ */
|
|
|
+ public void testMissingFieldsNotPurgingTheJoinLocally() {
|
|
|
+ var stats = EsqlTestUtils.statsForMissingField("languages");
|
|
|
+
|
|
|
+ var plan = plannerOptimizer.plan("""
|
|
|
+ from test
|
|
|
+ | keep emp_no, languages, first_name
|
|
|
+ | rename languages AS language_code
|
|
|
+ | mv_expand emp_no
|
|
|
+ | grok first_name "%{NUMBER:language_code:int}" // this reassigns language_code
|
|
|
+ | lookup join languages_lookup ON language_code
|
|
|
+ | stats c = count(emp_no) by language_code
|
|
|
+ """, stats);
|
|
|
+
|
|
|
+ var limit = as(plan, LimitExec.class);
|
|
|
+ var agg = as(limit.child(), AggregateExec.class);
|
|
|
+ assertThat(Expressions.names(agg.output()), contains("c", "language_code"));
|
|
|
+
|
|
|
+ var exchange = as(agg.child(), ExchangeExec.class);
|
|
|
+ agg = as(exchange.child(), AggregateExec.class);
|
|
|
+ var join = as(agg.child(), LookupJoinExec.class);
|
|
|
+ var grok = as(join.left(), GrokExec.class);
|
|
|
+ var mvexpand = as(grok.child(), MvExpandExec.class);
|
|
|
+ var project = as(mvexpand.child(), ProjectExec.class);
|
|
|
+ var extract = as(project.child(), FieldExtractExec.class);
|
|
|
+ var eval = as(extract.child(), EvalExec.class);
|
|
|
+ var source = as(eval.child(), EsQueryExec.class);
|
|
|
+ var right = as(join.right(), EsQueryExec.class);
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * LimitExec[1000[INTEGER]]
|
|
|
+ * \_LookupJoinExec[[language_code{r}#6],[language_code{f}#23],[language_name{f}#24]]
|
|
|
+ * |_LimitExec[1000[INTEGER]]
|
|
|
+ * | \_AggregateExec[[languages{f}#15],[COUNT(emp_no{f}#12,true[BOOLEAN]) AS c#10, languages{f}#15 AS language_code#6],FINAL,[language
|
|
|
+ * s{f}#15, $$c$count{r}#25, $$c$seen{r}#26],62]
|
|
|
+ * | \_ExchangeExec[[languages{f}#15, $$c$count{r}#25, $$c$seen{r}#26],true]
|
|
|
+ * | \_AggregateExec[[languages{r}#15],[COUNT(emp_no{f}#12,true[BOOLEAN]) AS c#10, languages{r}#15 AS language_code#6],INITIAL,
|
|
|
+ * [languages{r}#15, $$c$count{r}#27, $$c$seen{r}#28],12]
|
|
|
+ * | \_FieldExtractExec[emp_no{f}#12]<[],[]>
|
|
|
+ * | \_EvalExec[[null[INTEGER] AS languages#15]]
|
|
|
+ * | \_EsQueryExec[test], indexMode[standard], query[][_doc{f}#29], limit[], sort[] estimatedRowSize[12]
|
|
|
+ * \_EsQueryExec[languages_lookup], indexMode[lookup], query[][_doc{f}#30], limit[], sort[] estimatedRowSize[4]
|
|
|
+ */
|
|
|
+ public void testMissingFieldsDoesNotPurgeTheJoinOnCoordinator() {
|
|
|
+ var stats = EsqlTestUtils.statsForMissingField("languages");
|
|
|
+
|
|
|
+ // same as the query above, but with the last two lines swapped, so that the join is no longer pushed to the data nodes
|
|
|
+ var plan = plannerOptimizer.plan("""
|
|
|
+ from test
|
|
|
+ | keep emp_no, languages
|
|
|
+ | rename languages AS language_code
|
|
|
+ | stats c = count(emp_no) by language_code
|
|
|
+ | lookup join languages_lookup ON language_code
|
|
|
+ """, stats);
|
|
|
+
|
|
|
+ var limit = as(plan, LimitExec.class);
|
|
|
+ var join = as(limit.child(), LookupJoinExec.class);
|
|
|
+ limit = as(join.left(), LimitExec.class);
|
|
|
+ var agg = as(limit.child(), AggregateExec.class);
|
|
|
+ var exchange = as(agg.child(), ExchangeExec.class);
|
|
|
+ agg = as(exchange.child(), AggregateExec.class);
|
|
|
+ var extract = as(agg.child(), FieldExtractExec.class);
|
|
|
+ var eval = as(extract.child(), EvalExec.class);
|
|
|
+ assertThat(eval.fields().size(), is(1));
|
|
|
+ var alias = as(eval.fields().get(0), Alias.class);
|
|
|
+ assertThat(alias.name(), is("languages"));
|
|
|
+ var literal = as(alias.child(), Literal.class);
|
|
|
+ assertNull(literal.value());
|
|
|
+ var source = as(eval.child(), EsQueryExec.class);
|
|
|
+ assertThat(source.indexPattern(), is("test"));
|
|
|
+ assertThat(source.indexMode(), is(IndexMode.STANDARD));
|
|
|
+
|
|
|
+ source = as(join.right(), EsQueryExec.class);
|
|
|
+ assertThat(source.indexPattern(), is("languages_lookup"));
|
|
|
+ assertThat(source.indexMode(), is(IndexMode.LOOKUP));
|
|
|
+ }
|
|
|
+
|
|
|
/*
|
|
|
Checks that match filters are pushed down to Lucene when using no casting, for example:
|
|
|
WHERE first_name:"Anna")
|
Bogdan Pintea