ESQL: Fix a bug in LuceneQueryExpressionEvaluator (#117252) (#117279) (#118078)

* ESQL: Fix a bug in LuceneQueryExpressionEvaluator

This fixes Lucene usage bug in `LuceneQueryExpressionEvaluator`, the
evaluator we plan to use to run things like `MATCH` when we *can't* push
it to a source operator. That'll be useful for things like:
```
FROM foo
| STATS COUNT(),
        COUNT() WHERE MATCH(message, "error")
```

Explanation:
When using Lucene's `Scorer` and `BulkScorer` you must stay on the same
thread. It's a rule. Most of the time nothing bad happens if you shift
threads, but sometimes things explode and Lucene doesn't work. Driver
can shift from one thread to another - that's just how it's designed.
It's a "yield after running a while" kind of thing.

In tests we sometimes get a version of the `Scorer` and `BulkScorer`
that assert that you don't shift threads. That is what caused this test
failure.

Anyway! This builds protection into `LuceneQueryExpressionEvaluator` so
that if it *does* shift threads then it'll rebuild the `Scorer` and
`BulkScorer`. That makes the test happy and makes even the most grump
Lucene object happy.

Closes #116879

Author: nik9000

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/lucene/LuceneQueryExpressionEvaluator.java

@@ -191,18 +191,29 @@ SegmentState segmentState(int segment) throws IOException {
     private class SegmentState {
         private final Weight weight;
         private final LeafReaderContext ctx;
+
         /**
          * Lazily initialed {@link Scorer} for this. {@code null} here means uninitialized
          * <strong>or</strong> that {@link #noMatch} is true.
          */
         private Scorer scorer;
 
+        /**
+         * Thread that initialized the {@link #scorer}.
+         */
+        private Thread scorerThread;
+
         /**
          * Lazily initialed {@link BulkScorer} for this. {@code null} here means uninitialized
          * <strong>or</strong> that {@link #noMatch} is true.
          */
         private BulkScorer bulkScorer;
 
+        /**
+         * Thread that initialized the {@link #bulkScorer}.
+         */
+        private Thread bulkScorerThread;
+
         /**
          * Set to {@code true} if, in the process of building a {@link Scorer} or {@link BulkScorer},
          * the {@link Weight} tells us there aren't any matches.
@@ -223,7 +234,10 @@ BooleanVector scoreDense(int min, int max) throws IOException {
             if (noMatch) {
                 return blockFactory.newConstantBooleanVector(false, length);
             }
-            if (bulkScorer == null) {
+            if (bulkScorer == null ||  // The bulkScorer wasn't initialized
+                Thread.currentThread() != bulkScorerThread // The bulkScorer was initialized on a different thread
+            ) {
+                bulkScorerThread = Thread.currentThread();
                 bulkScorer = weight.bulkScorer(ctx);
                 if (bulkScorer == null) {
                     noMatch = true;
@@ -257,8 +271,11 @@ private void initScorer(int minDocId) throws IOException {
             if (noMatch) {
                 return;
             }
-            if (scorer == null || scorer.iterator().docID() > minDocId) {
-                // The previous block might have been beyond this one, reset the scorer and try again.
+            if (scorer == null || // Scorer not initialized
+                scorerThread != Thread.currentThread() || // Scorer initialized on a different thread
+                scorer.iterator().docID() > minDocId // The previous block came "after" this one
+            ) {
+                scorerThread = Thread.currentThread();
                 scorer = weight.scorer(ctx);
                 if (scorer == null) {
                     noMatch = true;