ES|QL: FORK memory management

[luigidellaquila]

I got a failure from Generative tests that is a bit suspicious:

On CSV dataset (load with ./gradlew :x-pack:plugin:esql:qa:testFixtures:loadCsvSpecData --args="http://elastic-admin:elastic-password@localhost:9200"):

from airp*
| rename scalerank as language_code 
| lookup join languages_lookup on language_code 
| stats  `name` = count_distinct(language_code), dJAukFBDtW = max(language_code), `location` = min(language_code), EtWxTktW = min(language_code) by abbrev 
| mv_expand name 
| mv_expand `location` 
| FORK ( WHERE true ) ( WHERE true ) ( WHERE true ) ( WHERE true ) ( WHERE true ) 
| WHERE _fork == "fork2" 
| DROP _fork 
| limit 8169

"type": "circuit_breaking_exception",
"reason": "[request] Data too large, data for [<reused_arrays>] would be [322134656/307.2mb], which is larger than the limit of [322122547/307.1mb];

What is strange is that the query, without FORK, returns only 889 records and 5 columns with very small values

from airp*
| rename scalerank as language_code 
| lookup join languages_lookup on language_code 
| stats  `name` = count_distinct(language_code), dJAukFBDtW = max(language_code), `location` = min(language_code), EtWxTktW = min(language_code) by abbrev 
| mv_expand name 
| mv_expand `location` 
| STATS count(*)

   count(*)    
---------------
889   

from airp*
| rename scalerank as language_code 
| lookup join languages_lookup on language_code 
| stats  `name` = count_distinct(language_code), dJAukFBDtW = max(language_code), `location` = min(language_code), EtWxTktW = min(language_code) by abbrev 
| mv_expand name 
| mv_expand `location` 

     name      |  dJAukFBDtW   |   location    |   EtWxTktW    |    abbrev     
---------------+---------------+---------------+---------------+---------------
1              |8              |8              |8              |null           
1              |9              |9              |9              |AWZ            
1              |9              |9              |9              |GWL            
1              |9              |9              |9              |HOD            
1              |9              |9              |9              |IXR            
1              |9              |9              |9              |LUH           
...

The failure is not completely deterministic, but it's very frequent, and it happens also with a smaller query, with only three FORK branches

from airp*
| rename scalerank as language_code 
| lookup join languages_lookup on language_code 
| stats  `name` = count_distinct(language_code), dJAukFBDtW = max(language_code), `location` = min(language_code), EtWxTktW = min(language_code) by abbrev 
| FORK ( WHERE true ) ( WHERE true ) ( WHERE true )

300MB seems too much for such small query, the source indices contain only 5189 records and 11 columns.

I'm labeling it as a bug for now.

[elasticsearchmachine]

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

[alex-spies]

@ioanatia , this looks like it could be very relevant/interesting to you!

[nik9000]

The default precision on count_distinct is quite expensive. This might be reasonable. We could, for example, initialize count_distinct to a smaller memory footprint at first rather than it's entire precision.

[ioanatia]

I can't replicate the issue anymore - I used to be able too with the examples provided here, but not anymore.

What I did:

• obviously, tested the queries from this issue - can't replicate even if run them multiple times
• I have run GenerativeForkIT test with 8 branches, by modifying this line here:
  

  elasticsearch/x-pack/plugin/esql/qa/server/src/main/java/org/elasticsearch/xpack/esql/qa/rest/generative/GenerativeForkRestTest.java

  Lines 40 to 43 in 564b509

  	protected void doTest() throws Throwable {
  	String query = testCase.query + " | FORK (WHERE true) (WHERE true) | WHERE _fork == \"fork1\" | DROP _fork";
  	doTest(query);
  	}

  
  The GenerativeForkIT run all the existing CSV specs by appending FORK (WHERE true) (WHERE true) | WHERE _fork == \"fork1\" | DROP _fork". These did not fail even with 8 branches.
• I have run GenerativeIT after modifying the ForkGenerator with a fix number of branches set to 8. I increased the number of iterations (queries to run) to absurdly high number e.g. 100k. I haven't seen any failures, stopped the tests after some time.

I will make sure the ForkGenerator is no longer restricted to 2 or 3 branches.
If we don't see any errors in the next weeks, I will close this issue 🤞 .

[ioanatia]

Okay - I managed to replicate this locally if I run the following query with 8 branches, running this multiple times will eventually lead to a circuit breaker exception:

from airp*
| rename scalerank as language_code 
| lookup join languages_lookup on language_code 
| stats  `name` = count_distinct(language_code), dJAukFBDtW = max(language_code), `location` = min(language_code), EtWxTktW = min(language_code) by abbrev 
| mv_expand name 
| mv_expand `location` 
| FORK ( WHERE true ) ( WHERE true ) ( WHERE true ) ( WHERE true ) ( WHERE true ) ( WHERE true )  ( WHERE true )  ( WHERE true ) 
| WHERE _fork == "fork2" 
| DROP _fork 
| limit 8169

I also see this when running with 7 branches
It's definitely an improvement from what we have seen initially.

@nik9000 pointed me to #131133 and #131307 as the potential reasons why this issue is harder to replicate.

On the other hand if we do:

from airp*
| rename scalerank as language_code 
| lookup join languages_lookup on language_code 
| EVAL a = MV_APPEND(language_code, 3)
| EVAL b = MV_APPEND(language_code, 3)
| EVAL c = MV_APPEND(language_code, 3)
| EVAL d= MV_APPEND(language_code, 3)
| EVAL d= MV_APPEND(language_code, 3)
| EVAL e= MV_APPEND(language_code, 3)
| EVAL f= MV_APPEND(language_code, 3)
| EVAL g = MV_APPEND(language_code, 3)
| EVAL g = MV_APPEND(language_code, 3)
| EVAL h = MV_APPEND(language_code, 3)
| EVAL i = MV_APPEND(language_code, 3)
| EVAL j = MV_APPEND(language_code, 3)
| EVAL k = MV_APPEND(language_code, 3)
| EVAL l = MV_APPEND(language_code, 3)
| EVAL m = MV_APPEND(language_code, 3)
| EVAL n = MV_APPEND(language_code, 3)
| stats  `name` = count_distinct(language_code), dJAukFBDtW = max(language_code), `location` = min(language_code), EtWxTktW = min(language_code),
 a = count_distinct(a), b = count_distinct(b), c = count_distinct(c), d = count_distinct(d), e = count_distinct(language_code),  f = count_distinct(f),  g = count_distinct(g), h = count_distinct(h), i = count_distinct(i),  j = count_distinct(j),  k = count_distinct(k), l = count_distinct(l),  m = count_distinct(m), n = count_distinct(n)   by abbrev 
| mv_expand name 
| mv_expand `location` 
| limit 8169

This is not a query that uses FORK but it results in partial failures with the same circuit breaker exception.

So while the problem is not solved entirely, we do see that it becomes much harder to replicate.