ESQL: INLINESTATS

[nik9000]

This implements INLINESTATS. Most of the heavy lifting is done by LOOKUP, with this change mostly adding a new abstraction to logical plans, an interface I'm calling Phased. Implementing this interface allows a logical plan node to cut the query into phases. INLINESTATS implements it by asking for a "first phase" that's the same query, up to INLINESTATS, but with INLINESTATS replaced with STATS. The next phase replaces the INLINESTATS with a hash join on the results of the first phase.

So, this query:

FROM foo
| EVAL bar = a * b
| INLINESTATS m = MAX(bar) BY b
| WHERE m = bar
| LIMIT 1

gets split into

FROM foo
| EVAL bar = a * b
| STATS m = MAX(bar) BY b

followed by

FROM foo
| EVAL bar = a * b
| LOOKUP (results of m = MAX(bar) BY b) ON b
| WHERE m = bar
| LIMIT 1

Here's an example of the syntax:

$ curl -k -XDELETE -u'elastic:password' 'http://localhost:9200/test'
$ for a in {0..99}; do
    echo -n $a
    rm -f /tmp/bulk
    for b in {0..999}; do
        echo '{"index": {}}' >> /tmp/bulk
        echo '{"a": '$a', "b": '$b'}' >> /tmp/bulk
    done
    curl -s -k -XPOST -u'elastic:password' -HContent-Type:application/json 'http://localhost:9200/test/_bulk?pretty' --data-binary @/tmp/bulk | grep errors
done
$ curl -s -k -XPOST -u'elastic:password' -HContent-Type:application/json 'http://localhost:9200/test/_forcemerge?max_num_segments=1&pretty'
$ curl -s -k -XPOST -u'elastic:password' -HContent-Type:application/json 'http://localhost:9200/test/_refresh?pretty'
$ curl -k -XPOST -u'elastic:password' -HContent-Type:application/json http://localhost:9200/_query?pretty -d'{
    "query": "FROM test | INLINESTATS m=MAX(a * b) BY b | WHERE m == a * b | SORT a DESC, b DESC | LIMIT 1",
    "profile": true
}'
{
  "columns" : [
    {
      "name" : "a",
      "type" : "long"
    },
    {
      "name" : "b",
      "type" : "long"
    },
    {
      "name" : "m",
      "type" : "long"
    }
  ],
  "values" : [
    [
      99,
      999,
      98901
    ]

Closes #107589

[nik9000]

This is riddled with NOCOMMITs, as any good prototype deserves to be.

It's also not clear that this is the right way to go about this. I mean, cutting the request into two is the right way to do it, I think. The reworking of the callbacks to make that possible seems like the sane way to do that.

But from a planning side, this is an "interesting" choice. We analyze the INLINESTATS call, but then don't run the optimizer on it at all - that seems bad. OTOH, we do run the optimizer on the two halves of the query. That seems good.

[nik9000]

I've extracted #110445 out of this one so I can get it and not have to deal with merge conflicts.

[nik9000]

[nik9000]

There's a fun bug that I'm going to probably leave for a followup:

FROM airports
| INLINESTATS min_scalerank=MIN(scalerank) BY type
| MV_EXPAND type
| WHERE scalerank == MV_MIN(scalerank);

Hits the rule execution limit. It started from a typo - I meant to do scalerank == MV_MIN(min_scalerank) but didn't. And then it hit the rule execution limit. Then I removed extra commands from the query until I got to more minimal recreation.

[dnhatn]

@dnhatn I'm wondering if this Phased strategy would also work in case of rate function.

I encountered a similar issue while implementing metrics aggregations. To address it, I chose to execute mixed aggregations, wrapped in to_partial and from_partial, in a single phase along with other aggregations. I made this decision for two reasons:

• Data consistency: As others have noted, we need to ensure data consistency between phases, which can be challenging because we open and close readers of target shards in batches on data nodes to avoid holding excessive resources (e.g., file descriptors).
• Execution redundancy: We might need to execute the initial part of each phase multiple times, such as LuceneQuery and FieldExact.

An alternative approach I explored was implementing a multiplexed pipeline. This method involves broadcasting/scattering pages into multiple sub-plans and then gathering the pages. With InlineStats, we can gather and join pages using a HashJoin (or Eval). I did not spend enough time to make this alternative approach ready. However, we can revisit it if we encounter issues with maintaining data consistency across phases.

However, this PR looks good. Great work. Thank you, Nik!

[dnhatn]

Nice examples ❤️ . It would be lovely if we could push down the filter in the second phase to Lucene instead of using HashJoin, so that we can avoid scanning the entire dataset. Let's address this later.

[nik9000]

Yeah! I'm pretty sure we can push these down. That's on the followup list!

[astefan]

LGTM

I am ok with the code as is now and the awesome usecases it's unlocking, if some aspects will definitely be addressed later (some missing use cases and some performance improvements missing) and if we are ok with the "experimental" label until then.

[nik9000]

NOCOMMIT: Link the new javadocs for Phased and EsqlSession into the package level javadoc.

[nik9000]

NOCOMMIT: Link the new javadocs for Phased and EsqlSession into the package level javadoc.

Pushed a link.

[alex-spies]

I don't think we need the limit0 tests added here; the limit0 tests are only present in enrich.csv-spec so we can run at least the logical optimizer against tests that otherwise would need enrich_load.

[alex-spies]

Let's also add a shadowingSelf test!

| INLINESTATS city = COUNT(city)

[nik9000]

Oh. That.... Probably isn't going to work. I guess we do want it to work....

[alex-spies]

That'd be required to be consistent with enrich, eval, dissect and grok. I think we can add this to the list of follow ups, as this probably requires work on LOOKUP resp. JOIN.

[alex-spies]

Yeahp, this should work and is conceptually a bit nicer (IMHO) than doing that down in the physical mapping!

[alex-spies]

I think using the schema from the first phase is reasonable, but throwing an IllegalArgumentException if the schema doesn't line up with what we expected will make our lives much easier, esp. if this should blow up in production.

[alex-spies]

<3

[nik9000]

I've put this behind a feature flag and removed release highlight because it's not super clear that this'll get un-feature flagged in 8.16.

There are a bunch of extra follow ups. I had a conversation with @costin about moving around how we trigger the Phased nature. The plan now is to merge this as is and rework some things.

I'm going to push a few more tests and see if we can land this. Folks can experiment with it some more while we finish up.

[alex-spies]

Thanks Nik, this is much better now!

I have one last nit: the check is slightly insufficient because

firstPhase().output().equals(schema) == false

will not look at the name ids (they are not checked in NamedExpression.equals(), nor in NamedExpression's descendants). The plan may become inconsistent if the name ids do not line up, even if the first phase produces the correct names and data types. I think what we need is a little helper that we should call here.

public static equalsAndSemanticEquals(List<Attribute> left, List<Attribute> right) {
    if (left.equals(right) == false) {
        return false;
    }
    for (int i = 0; i < left.size(); i++) {
        if (left.get(i).semanticEquals(right.get(i)) == false) {
            return false;
        }
    }
    return true;
}

We could put that into the Expressions class (not Expression).

Either this, or we ignore name ids from the first phase: in ungroupedNextPhase we already do this, because we obtain the name ids from the aggregates. In groupedNextPhase, however, we put the schema's attributes directly as the attributes of the local relation.

[nik9000]

Release tests look unrelated. We'll fight with them as fight comes up.

[costin]

FTR, I've raised an meta ticket to track my progress at #112266

[maggieghamry]

@costin would it be possible to update this blog https://www.elastic.co/search-labs/blog/esql-piped-query-language-goes-ga to note that INLINESTATS is not yet available/will only be available in 8.16.0?