ESQL: Fix async operator warnings not always sent when blocking

[ivancea]

Fixes #128030
Fixes #130296
Fixes #130642
Fixes #131148
Fixes #132554
Fixes #132555
Fixes #132604
Fixes #131563
Fixes #132778

Extracted from #132738

An AsyncOperator listener misordering caused the warnings collection and status metrics updates to be executed after the onSeqNoCompleted()>notifyIfBlocked()>future.onResponse(null), which ends the processing in some cases.

Bug explanation

An example profile of one of the CI fails:

{
  planning={took_millis=8, start_millis=1754605547195, stop_millis=1754605547203, took_nanos=8612200},
  plans=[
    {
      cluster_name=test-cluster,
      node_name=test-cluster-1,
      description=single,
      plan=
        OutputExec[org.elasticsearch.xpack.esql.plugin.ComputeService$$Lambda/0x0000000095705068@4387ca3]	
        \_ProjectExec[[language_code{r}#25708, language_name{f}#25719, country{f}#25716]]	
          \_TopNExec[[Order[language_code{r}#25708,ASC,LAST], Order[language_name{f}#25719,ASC,LAST], Order[country{f}#25716,ASC,LAST]],1000[INTEGER],104]	
            \_LookupJoinExec[[language_code{r}#25708],[language_code{f}#25718],[country{f}#25716, language_name{f}#25719]]	
              |_LocalSourceExec[[language_code{r}#25708],org.elasticsearch.xpack.esql.plan.logical.local.CopyingLocalSupplier@1ef1bc4]	
              \_EsQueryExec[languages_lookup_non_unique_key], indexMode[lookup], query[][_doc{f}#25720], limit[], sort[] estimatedRowSize[104]
    }
  ],
  query={took_millis=14, start_millis=1754605547195, stop_millis=1754605547209, took_nanos=14013972},
  drivers=[
    {
      cluster_name=test-cluster, sleeps={last=[], counts={}, first=[]}, cpu_nanos=3189539,
      operators=[
        {operator=LocalSourceOperator},
        {
          operator=LookupOperator[
            index=languages_lookup_non_unique_key
            input_type=INTEGER
            match_field=language_code
            load_fields=[country{f}#25716, language_name{f}#25719]
            inputChannel=0
          ],
          status={received_pages=0, total_terms=4, emitted_pages=1, process_nanos=0, completed_pages=0}
        },
        {
          operator=TopNOperator[
            count=0/1000,
            elementTypes=[INT, BYTES_REF, BYTES_REF],
            encoders=[DefaultSortable, UTF8TopNEncoder, UTF8TopNEncoder],
            sortOrders=[SortOrder[channel=0, asc=true, nullsFirst=false], SortOrder[channel=2, asc=true, nullsFirst=false], SortOrder[channel=1, asc=true, nullsFirst=false]]
          ],
          status={receive_nanos=36566, emit_nanos=33132, occupied_rows=0, rows_emitted=1, pages_emitted=1, ram_used=4.2kb, rows_received=1, pages_received=1, ram_bytes_used=4368}
        },
        {operator=ProjectOperator[projection = [0, 2, 1]], status={pages_processed=1, rows_emitted=1, process_nanos=2856, rows_received=1}},
        {operator=OutputOperator[columns = [language_code, language_name, country]]}
      ],
      node_name=test-cluster-1,
      description=single,
      documents_found=0,
      start_millis=1754605547205,
      stop_millis=1754605547208,
      values_loaded=0,
      iterations=3124,
      took_nanos=3432530
    }
  ]
}

The LookupOperator has this strange status:
status={received_pages=0, total_terms=4, emitted_pages=1, process_nanos=0, completed_pages=0}

0 received pages, 0 completed pages, 0 prcess_nanos, but 4 total terms and 1 emitted page. All the 0s come from AsyncOperator, while the others come from LookupFromIndexOperator.

After some investigation, the received_pages and completed_pages were actually wrong (Fixed in #132738), but the process_nanos shows the issue: That part of the code didn't get to execute yet.

[elasticsearchmachine]

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

[elasticsearchmachine]

Hi @ivancea, I've created a changelog YAML for you.

[elasticsearchmachine]

Hi @ivancea, I've updated the changelog YAML for you.

[nik9000]

Looks better to me. Let @dnhatn look too please.

[nik9000]

@dnhatn , could you have a look at this bit?

[nik9000]

Actually, this whole PR to be honest. It's the kind of thing I don't trust myself with. Obviously, I broke it in the first place.

[ivancea]

Hey! This change feels a bit mystic. It's properly tested in the other PR, where I fix and test multiple things around the status() here.

I'll remove it from here, it doesn't make sense in this PR, and I wanted only the other bugfix to be here!

[dnhatn]

Thanks Ivan!

I think the problematic sequence is: onSeqNoCompleted(seqNo) → close() → responseHeadersCollector.finish() → responseHeadersCollector.collect() → driverContext.removeAsyncAction(), where responseHeadersCollector.finish() is called before responseHeadersCollector.collect(). This PR prevents that sequence. However, we should not increase the sequence number in onFailure() (call onSeqNoCompleted), as it may cause a NullPointerException later:

elasticsearch/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/AsyncOperator.java

Line 168 in b504514

Fetched result = buffers.remove(nextCheckpoint);

[elasticsearchmachine]

Hi @ivancea, I've updated the changelog YAML for you.

[ivancea]

(Answering you here @dnhatn, as to have a thread)

This PR prevents that sequence. However, we should not increase the sequence number in onFailure() (call onSeqNoCompleted), as it may cause a NullPointerException later

From what I see, there's already an if (fetched != null) there on line 170, so no possible NPE, unless you mean something else or I missed something 👀

elasticsearch/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/AsyncOperator.java

Lines 165 to 174 in 9f927b2

	private void discardResults() {
	long nextCheckpoint;
	while ((nextCheckpoint = checkpoint.getPersistedCheckpoint() + 1) <= checkpoint.getProcessedCheckpoint()) {
	Fetched result = buffers.remove(nextCheckpoint);
	checkpoint.markSeqNoAsPersisted(nextCheckpoint);
	if (result != null) {
	releaseFetchedOnAnyThread(result);
	}
	}
	}

[dnhatn]

Thanks Ivan. I took another closer look. I think we are all good.

[dnhatn]

LGTM. Thanks Ivan!

[dnhatn]

Thanks Ivan. I took another closer look. I think we are all good.