Mv_expand and projections and filters

[dnhatn]

A user reported an issue where an ES|QL query takes several minutes to execute. I don't have the query, but I do have a task that includes a list of operators. I suspect the issue is related to mv_expand combined with projections.

from test | MV_EXPAND count_d | where count_d > 20

produces

ExchangeSinkExec[[color{f}#34, count{f}#35, count_d{r}#36, data{f}#37, data_d{f}#38, tag{f}#39, time{f}#40],false]
\_ProjectExec[[color{f}#34, count{f}#35, count_d{r}#36, data{f}#37, data_d{f}#38, tag{f}#39, time{f}#40]]
  \_FieldExtractExec[color{f}#34, count{f}#35, data{f}#37, data_d{f}#38, ..]<[],[]>
    \_LimitExec[1000[INTEGER],28]
      \_FilterExec[count_d{r}#36 > 20[INTEGER]]
        \_MvExpandExec[count_d{f}#41,count_d{r}#36]
          \_FieldExtractExec[count_d{f}#41]<[],[]>
            \_EsQueryExec[test], indexMode[standard], [_doc{f}#42], limit[], sort[] estimatedRowSize[144] queryBuilderAndTags [[QueryBuilderAndTags{queryBuilder=[null], tags=[]}]]

from test | KEEP * | MV_EXPAND count_d | where count_d > 20

produces

ExchangeSinkExec[[color{f}#14, count{f}#15, count_d{r}#16, data{f}#17, data_d{f}#18, tag{f}#19, time{f}#20],false]
\_ProjectExec[[color{f}#14, count{f}#15, count_d{r}#16, data{f}#17, data_d{f}#18, tag{f}#19, time{f}#20]]
  \_LimitExec[1000[INTEGER],140]
    \_FilterExec[count_d{r}#16 > 20[INTEGER]]
      \_MvExpandExec[count_d{f}#21,count_d{r}#16]
        \_ProjectExec[[color{f}#14, count{f}#15, count_d{f}#21, data{f}#17, data_d{f}#18, tag{f}#19, time{f}#20]]
          \_FieldExtractExec[color{f}#14, count{f}#15, count_d{f}#21, data{f}#17, ..]<[],[]>
            \_EsQueryExec[test], indexMode[standard], [_doc{f}#22], limit[], sort[] estimatedRowSize[144] queryBuilderAndTags [[QueryBuilderAndTags{queryBuilder=[null], tags=[]}]]

With KEEP * before mv_expand, all fields are eagerly loaded before the filter. I believe both queries should produce the same physical plan that loading fields after the filter.

I think we could push the filter (without single-value-match) to Lucene in this scenario, but that's a separate issue.

[elasticsearchmachine]

Pinging @elastic/es-analytical-engine (Team:Analytics)

[dnhatn]

Per the user's report, zero rows are emitted after the filter.

"active_operators":
[
    {
        "operator": "LuceneSourceOperator[shards = [, maxPageSize = 32, remainingDocs = 2147483615]",
        "status":
        {
            "processed_slices": 1,
            "processed_queries":
            [
                "*:*"
            ],
            "processed_shards":
            [
                ""
            ],
            "process_nanos": 95643549,
            "process_time": "95.6ms",
            "slice_index": 1,
            "total_slices": 98,
            "pages_emitted": 8336,
            "slice_min": 0,
            "slice_max": 0,
            "current": 344659,
            "rows_emitted": 266752,
        }
    },
    {
        "operator": "EvalOperator[evaluator=LiteralsEvaluator[lit=null]]",
        "status":
        {
            "process_nanos": 18468754,
            "process_time": "18.4ms",
            "pages_processed": 8336,
            "rows_received": 266752,
            "rows_emitted": 266752
        }
    },
    {
        "operator": "EvalOperator[evaluator=LiteralsEvaluator[lit=null]]",
        "status":
        {
            "process_nanos": 1830136,
            "process_time": "1.8ms",
            "pages_processed": 8336,
            "rows_received": 266752,
            "rows_emitted": 266752
        }
    },
    {
        "operator": "EvalOperator[evaluator=LiteralsEvaluator[lit=null]]",
        "status":
        {
            "process_nanos": 1501899,
            "process_time": "1.5ms",
            "pages_processed": 8336,
            "rows_received": 266752,
            "rows_emitted": 266752
        }
    },
    {
        "operator": "EvalOperator[evaluator=LiteralsEvaluator[lit=null]]",
        "status":
        {
            "process_nanos": 1557285,
            "process_time": "1.5ms",
            "pages_processed": 8336,
            "rows_received": 266752,
            "rows_emitted": 266752
        }
    },
    {
        "operator": "EvalOperator[evaluator=LiteralsEvaluator[lit=null]]",
        "status":
        {
            "process_nanos": 1522002,
            "process_time": "1.5ms",
            "pages_processed": 8336,
            "rows_received": 266752,
            "rows_emitted": 266752
        }
    },
    {
        "operator": "EvalOperator[evaluator=LiteralsEvaluator[lit=null]]",
        "status":
        {
            "process_nanos": 1434007,
            "process_time": "1.4ms",
            "pages_processed": 8336,
            "rows_received": 266752,
            "rows_emitted": 266752
        }
    },
    {
        "operator": "EvalOperator[evaluator=LiteralsEvaluator[lit=null]]",
        "status":
        {
            "process_nanos": 1665803,
            "process_time": "1.6ms",
            "pages_processed": 8336,
            "rows_received": 266752,
            "rows_emitted": 266752
        }
    },
    {
        "operator": "ValuesSourceReaderOperator[fields = [880 fields]]",
        "status":
        {
            "readers_built":{},
            "values_loaded": 23686504,
            "process_nanos": 597928577332,
            "process_time": "9.9m",
            "pages_received": 8336,
            "pages_emitted": 8336,
            "rows_received": 266752,
            "rows_emitted": 266752
        }
    },
    {
        "operator": "ProjectOperator[projection = [1486 fields]]",
        "status":
        {
            "process_nanos": 792017765,
            "process_time": "792ms",
            "pages_processed": 8336,
            "rows_received": 266752,
            "rows_emitted": 266752
        }
    },
    {
        "operator": "MvExpandOperator[channel=1139]",
        "status":
        {
            "pages_received": 8336,
            "pages_emitted": 8336,
            "noops": 76,
            "rows_received": 266752,
            "rows_emitted": 1508802
        }
    },
    {
        "operator": "FilterOperator[evaluator=EqualsKeywordsEvaluator[lhs=Attribute[channel=1139], rhs=LiteralsEvaluator[lit=DELETE]]]",
        "status":
        {
            "process_nanos": 5699057382,
            "process_time": "5.6s",
            "pages_processed": 8336,
            "rows_received": 1508802,
            "rows_emitted": 0
        }
    },
    {
        "operator": "LimitOperator[limit = 1000/1000]",
        "status":
        {
            "limit": 1000,
            "limit_remaining": 1000,
            "pages_processed": 0,
            "rows_received": 0,
            "rows_emitted": 0
        }
    },
    {
        "operator": "ProjectOperator[projection = [1486 fields]]",
        "status":
        {
            "process_nanos": 0,
            "process_time": "0s",
            "pages_processed": 0,
            "rows_received": 0,
            "rows_emitted": 0
        }
    },
    {
        "operator": "ExchangeSinkOperator",
        "status":
        {
            "pages_received": 0,
            "rows_received": 0
        }
    }
]

[dnhatn]

We can also eagerly load all fields for the query FROM test | MV_EXPAND count_d | where count_d > 20 if some fields are missing in the target shards.

[dnhatn]

Two primary issues here:

1. Projections cause fields to be loaded too early.

2. Filters are not being passed over MV_EXPAND to be pushed down to Lucene. We can pass a filter over MV_EXPAND by wrapping it in PUSH_DOWN_TO_LUCENE_OR_NOOP, then push it down to Lucene if possible. This filter will either translate to a query for Lucene, or act as a no-op (doesn't filter anything) in the compute engine.

For example, this query:

FROM logs*
| MV_EXPAND message
| WHERE message LIKE "DELETE"

becomes:

FROM logs*
| WHERE PUSH_DOWN_TO_LUCENE_OR_NOOP(message LIKE "DELETE")
| MV_EXPAND message
| WHERE message LIKE "DELETE"

It is important to keep the original filter after MV_EXPAND, as the pushed-down filter in Lucene can return false positives. The filter after MV_EXPAND ensures unmatched rows are excluded.

For PUSH_DOWN_TO_LUCENE_OR_NOOP, handle two cases:

1. Non-negative filters (e.g., message LIKE "DELETE", color == "red"): can be pushed down directly, as matching documents are returned regardless of single or multi-value fields.

2. Negative filters (e.g., message != "DELETE"): must be written with OR multi-value to avoid excluding documents with multi-valued fields. For example, message != "DELETE" can be translated as message != "DELETE" OR (message == "DELETE" AND multi-value) or message != "DELETE" OR multi-value.

We should not wrap these queries with single_match_query.

[alex-spies]

Pushing down things past Project is fairly standard, and it should be easy enough to write a similar pushdown rule for MV_EXPAND.

Pushing/duplicating filters past MV_EXPAND is a bit more effort. Internally,

| MV_EXPAND message
| WHERE message LIKE "DELETE"

should probably become

| (optional) WHERE ANY message LIKE "DELETE"
| MV_EXPAND message
| WHERE message LIKE "DELETE"

and we need to properly push down the ANY message LIKE "DELETE", like you described.

The no-op idea is interesting, although there's probably value in actually evaluating the WHERE ANY message LIKE "DELETE", because this filter will generally be pushed down even further and might e.g. reduce the row count for any lookup joins and enriches that come before the MV_EXPAND.

We thought about multivalue predicates like ANY message LIKE "DELETE". @luigidellaquila had some great thoughts on the topic.

[alex-spies]

Let me split out the project optimization into a separate issue: #136292