ESQL: Expand HeapAttack for LOOKUP (#120754)

* ESQL: Expand HeapAttack for LOOKUP

This expands the heap attack tests for LOOKUP. Now there are three
flavors:
1. LOOKUP a single geo_point - about 30 bytes or so.
2. LOOKUP a one mb string.
3. LOOKUP no fields - just JOIN to alter cardinality.

Fetching a geo_point is fine with about 500 repeated docs before it
circuit breaks which works out to about 256mb of buffered results.
That's sensible on our 512mb heap and likely to work ok for most folks.
We'll flip to a streaming method eventually and this won't be a problem
any more. But for now, we buffer.

The no lookup fields is fine with like 7500 matches per incoming row.
That's quite a lot, really.

The 1mb string is trouble! We circuit break properly which is great and
safe, but if you join 1mb worth of columns in LOOKUP you are going to
need bigger heaps than our test. Again, we'll move from buffering these
results to streaming them and it'll work better, but for now we buffer.

* updates

Author: nik9000

test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackIT.java

@@ -628,24 +628,79 @@ private Response fetchMvLongs() throws IOException {
     }
 
     public void testLookupExplosion() throws IOException {
-        int sensorDataCount = 7500;
+        int sensorDataCount = 500;
         int lookupEntries = 10000;
         Map<?, ?> map = lookupExplosion(sensorDataCount, lookupEntries);
         assertMap(map, matchesMap().extraOk().entry("values", List.of(List.of(sensorDataCount * lookupEntries))));
     }
 
     public void testLookupExplosionManyMatches() throws IOException {
         assertCircuitBreaks(() -> {
-            Map<?, ?> result = lookupExplosion(8500, 10000);
+            Map<?, ?> result = lookupExplosion(1500, 10000);
+            logger.error("should have failed but got {}", result);
+        });
+    }
+
+    public void testLookupExplosionNoFetch() throws IOException {
+        int sensorDataCount = 7500;
+        int lookupEntries = 10000;
+        Map<?, ?> map = lookupExplosionNoFetch(sensorDataCount, lookupEntries);
+        assertMap(map, matchesMap().extraOk().entry("values", List.of(List.of(sensorDataCount * lookupEntries))));
+    }
+
+    public void testLookupExplosionNoFetchManyMatches() throws IOException {
+        assertCircuitBreaks(() -> {
+            Map<?, ?> result = lookupExplosionNoFetch(8500, 10000);
+            logger.error("should have failed but got {}", result);
+        });
+    }
+
+    public void testLookupExplosionBigString() throws IOException {
+        int sensorDataCount = 150;
+        int lookupEntries = 1;
+        Map<?, ?> map = lookupExplosionBigString(sensorDataCount, lookupEntries);
+        assertMap(map, matchesMap().extraOk().entry("values", List.of(List.of(sensorDataCount * lookupEntries))));
+    }
+
+    public void testLookupExplosionBigStringManyMatches() throws IOException {
+        assertCircuitBreaks(() -> {
+            Map<?, ?> result = lookupExplosionBigString(500, 1);
             logger.error("should have failed but got {}", result);
         });
     }
 
     private Map<?, ?> lookupExplosion(int sensorDataCount, int lookupEntries) throws IOException {
+        lookupExplosionData(sensorDataCount, lookupEntries);
+        StringBuilder query = startQuery();
+        query.append("FROM sensor_data | LOOKUP JOIN sensor_lookup ON id | STATS COUNT(location)\"}");
+        return responseAsMap(query(query.toString(), null));
+    }
+
+    private Map<?, ?> lookupExplosionNoFetch(int sensorDataCount, int lookupEntries) throws IOException {
+        lookupExplosionData(sensorDataCount, lookupEntries);
+        StringBuilder query = startQuery();
+        query.append("FROM sensor_data | LOOKUP JOIN sensor_lookup ON id | STATS COUNT(*)\"}");
+        return responseAsMap(query(query.toString(), null));
+    }
+
+    private void lookupExplosionData(int sensorDataCount, int lookupEntries) throws IOException {
         initSensorData(sensorDataCount, 1);
         initSensorLookup(lookupEntries, 1, i -> "73.9857 40.7484");
+    }
+
+    private Map<?, ?> lookupExplosionBigString(int sensorDataCount, int lookupEntries) throws IOException {
+        initSensorData(sensorDataCount, 1);
+        initSensorLookupString(lookupEntries, 1, i -> {
+            int target = Math.toIntExact(ByteSizeValue.ofMb(1).getBytes());
+            StringBuilder str = new StringBuilder(Math.toIntExact(ByteSizeValue.ofMb(2).getBytes()));
+            while (str.length() < target) {
+                str.append("Lorem ipsum dolor sit amet, consectetur adipiscing elit.");
+            }
+            logger.info("big string is {} characters", str.length());
+            return str.toString();
+        });
         StringBuilder query = startQuery();
-        query.append("FROM sensor_data | LOOKUP JOIN sensor_lookup ON id | STATS COUNT(*)\"}");
+        query.append("FROM sensor_data | LOOKUP JOIN sensor_lookup ON id | STATS COUNT(string)\"}");
         return responseAsMap(query(query.toString(), null));
     }
 
@@ -834,6 +889,31 @@ private void initSensorLookup(int lookupEntries, int sensorCount, IntFunction<St
         initIndex("sensor_lookup", data.toString());
     }
 
+    private void initSensorLookupString(int lookupEntries, int sensorCount, IntFunction<String> string) throws IOException {
+        logger.info("loading sensor lookup with huge strings");
+        createIndex("sensor_lookup", Settings.builder().put(IndexSettings.MODE.getKey(), IndexMode.LOOKUP.getName()).build(), """
+            {
+                "properties": {
+                    "id": { "type": "long" },
+                    "string": { "type": "text" }
+                }
+            }""");
+        int docsPerBulk = 10;
+        StringBuilder data = new StringBuilder();
+        for (int i = 0; i < lookupEntries; i++) {
+            int sensor = i % sensorCount;
+            data.append(String.format(Locale.ROOT, """
+                {"create":{}}
+                {"id": %d, "string": "%s"}
+                """, sensor, string.apply(sensor)));
+            if (i % docsPerBulk == docsPerBulk - 1) {
+                bulk("sensor_lookup", data.toString());
+                data.setLength(0);
+            }
+        }
+        initIndex("sensor_lookup", data.toString());
+    }
+
     private void initSensorEnrich(int lookupEntries, int sensorCount, IntFunction<String> location) throws IOException {
         initSensorLookup(lookupEntries, sensorCount, location);
         logger.info("loading sensor enrich");