Rename and fix bug

Author: nik9000

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

@@ -121,7 +121,7 @@ protected Page getCheckedOutput() throws IOException {
                 remainingDocs = 0;
             } else {
                 if (scorer.tags().isEmpty() == false) {
-                    throw new UnsupportedOperationException("extra not supported by " + getClass());
+                    throw new UnsupportedOperationException("tags not supported by " + getClass());
                 }
                 Weight weight = scorer.weight();
                 var leafReaderContext = scorer.leafReaderContext();

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

@@ -103,7 +103,7 @@ public Page getCheckedOutput() throws IOException {
                 remainingDocs = 0;
             } else {
                 if (scorer.tags().isEmpty() == false) {
-                    throw new UnsupportedOperationException("extra not supported by " + getClass());
+                    throw new UnsupportedOperationException("tags not supported by " + getClass());
                 }
                 final LeafReader reader = scorer.leafReaderContext().reader();
                 final Query query = scorer.weight().getQuery();

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

@@ -155,10 +155,13 @@ LuceneScorer getCurrentOrLoadNextScorer() {
             final PartialLeafReaderContext partialLeaf = currentSlice.getLeaf(sliceIndex++);
             logger.trace("Starting {}", partialLeaf);
             final LeafReaderContext leaf = partialLeaf.leafReaderContext();
-            if (currentScorer == null || currentScorer.leafReaderContext() != leaf) {
+            if (currentScorer == null // First time
+                || currentScorer.leafReaderContext() != leaf // Moved to a new leaf
+                || currentScorer.weight != currentSlice.weight() // Moved to a new query
+            ) {
                 final Weight weight = currentSlice.weight();
                 processedQueries.add(weight.getQuery());
-                currentScorer = new LuceneScorer(currentSlice.shardContext(), weight, currentSlice.extra(), leaf);
+                currentScorer = new LuceneScorer(currentSlice.shardContext(), weight, currentSlice.tags(), leaf);
             }
             assert currentScorer.maxPosition <= partialLeaf.maxDoc() : currentScorer.maxPosition + ">" + partialLeaf.maxDoc();
             currentScorer.maxPosition = partialLeaf.maxDoc();

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

@@ -14,7 +14,7 @@
 /**
  * Holds a list of multiple partial Lucene segments
  */
-public record LuceneSlice(ShardContext shardContext, List<PartialLeafReaderContext> leaves, Weight weight, List<Object> extra) {
+public record LuceneSlice(ShardContext shardContext, List<PartialLeafReaderContext> leaves, Weight weight, List<Object> tags) {
     int numLeaves() {
         return leaves.size();
     }

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

@@ -318,7 +318,6 @@ public Page getCheckedOutput() throws IOException {
             final int discardedDocs = collectedDocs - limiter.tryAccumulateHits(collectedDocs);
             Page page = null;
             if (currentPagePos >= minPageSize || scorer.isDone() || (remainingDocs = limiter.remaining()) == 0) {
-                log.error("SADADSF {}", remainingDocs);
                 IntVector shard = null;
                 IntVector leaf = null;
                 IntVector docs = null;

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

@@ -172,7 +172,7 @@ private Page collect() throws IOException {
         }
         try {
             if (scorer.tags().isEmpty() == false) {
-                throw new UnsupportedOperationException("extra not supported by " + getClass());
+                throw new UnsupportedOperationException("tags not supported by " + getClass());
             }
             if (perShardCollector == null || perShardCollector.shardContext.index() != scorer.shardContext().index()) {
                 // TODO: share the bottom between shardCollectors

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

@@ -116,8 +116,8 @@ public Page getCheckedOutput() throws IOException {
                     doneCollecting = true;
                     return null;
                 }
-                if (slice.extra().isEmpty() == false) {
-                    throw new UnsupportedOperationException("extra not supported by " + getClass());
+                if (slice.tags().isEmpty() == false) {
+                    throw new UnsupportedOperationException("tags not supported by " + getClass());
                 }
                 Releasables.close(fieldsReader);
                 fieldsReader = new ShardLevelFieldsReader(blockFactory, slice.shardContext(), fieldsToExtracts);

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/lucene/LuceneSourceOperatorTests.java

@@ -9,10 +9,6 @@
 
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 
-import com.carrotsearch.randomizedtesting.annotations.Repeat;
-
-import com.carrotsearch.randomizedtesting.annotations.Seed;
-
 import org.apache.lucene.document.SortedNumericDocValuesField;
 import org.apache.lucene.index.IndexReader;
 import org.apache.lucene.index.IndexableField;
@@ -67,8 +63,6 @@
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 import static org.hamcrest.Matchers.matchesRegex;
 
-//@Seed("4FF2CB98F60FD89D:EBBB701671985F2B")
-@Repeat(iterations = 100)
 public class LuceneSourceOperatorTests extends AnyOperatorTestCase {
     private static final MappedFieldType S_FIELD = new NumberFieldMapper.NumberFieldType("s", NumberFieldMapper.NumberType.LONG);
 
@@ -210,14 +204,21 @@ private LuceneSourceOperator.Factory simple(DataPartitioning dataPartitioning, i
                 }
             }
             reader = writer.getReader();
+
+            IndexSearcher searcher = new IndexSearcher(reader);
+            int count = 0;
+            for (LuceneSliceQueue.QueryAndTags q : testCase.queryAndExtra()) {
+                count += searcher.count(q.query());
+            }
+            assertThat(count, equalTo(testCase.numResults(numDocs)));
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
 
         ShardContext ctx = new MockShardContext(reader, 0);
         Function<ShardContext, List<LuceneSliceQueue.QueryAndTags>> queryFunction = c -> testCase.queryAndExtra();
         int maxPageSize = between(10, Math.max(10, numDocs));
-        int taskConcurrency = 4; // randomIntBetween(1, 4); NOCOMMIT
+        int taskConcurrency = randomIntBetween(1, 4);
         return new LuceneSourceOperator.Factory(
             List.of(ctx),
             queryFunction,
@@ -268,17 +269,17 @@ private void testSimple(DataPartitioning partitioning) {
     }
 
     public void testEarlyTermination() {
-        int numDocs = 20_000; //between(1_000, 20_000); NOCOMMIT
-        int limit = 100_000; //between(0, numDocs * 2); NOCOMMIT
-        LuceneSourceOperator.Factory factory = simple(DataPartitioning.DOC
-            // randomFrom(DataPartitioning.values()) NOCOMMIT
-            , numDocs, limit, scoring);
+        int numDocs = between(1_000, 20_000);
+        int limit = between(0, numDocs * 2);
+        LuceneSourceOperator.Factory factory = simple(randomFrom(DataPartitioning.values()), numDocs, limit, scoring);
         int taskConcurrency = factory.taskConcurrency();
         final AtomicInteger receivedRows = new AtomicInteger();
+        List<SourceOperator> sources = new ArrayList<>();
         List<Driver> drivers = new ArrayList<>();
         for (int i = 0; i < taskConcurrency; i++) {
             DriverContext driverContext = driverContext();
             SourceOperator sourceOperator = factory.get(driverContext);
+            sources.add(sourceOperator);
             SinkOperator sinkOperator = new PageConsumerOperator(p -> {
                 receivedRows.addAndGet(p.getPositionCount());
                 p.releaseBlocks();
@@ -301,6 +302,9 @@ public void testEarlyTermination() {
             drivers.add(driver);
         }
         OperatorTestCase.runDriver(drivers);
+        for (SourceOperator source : sources) {
+            logger.info("source status {}", source.status());
+        }
         logger.info(
             "{} received={} limit={} numResults={}",
             factory.dataPartitioning,
@@ -376,6 +380,9 @@ private void testSimple(DriverContext ctx, DataPartitioning partitioning, int nu
         }
 
         testCase.checkPages(numDocs, limit, factory.maxPageSize(), results);
+        int count = results.stream().mapToInt(Page::getPositionCount).sum();
+        logger.info("{} received={} limit={} numResults={}", factory.dataPartitioning, count, limit, testCase.numResults(numDocs));
+        assertThat(count, equalTo(Math.min(limit, testCase.numResults(numDocs))));
     }
 
     // Returns the initial block index, ignoring the score block if scoring is enabled