Merge pull request ESQL-1408 from elastic/main

🤖 ESQL: Merge upstream

docs/changelog/97509.yaml

@@ -0,0 +1,6 @@
+pr: 97509
+summary: Fix bug when creating empty `geo_lines`
+area: Geo
+type: bug
+issues:
+ - 97311

server/src/test/java/org/elasticsearch/reservedstate/service/FileSettingsServiceTests.java

@@ -166,7 +166,6 @@ public class FileSettingsServiceTests extends ESTestCase {
     }
 
     @SuppressWarnings("unchecked")
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/95436")
     public void testInitialFileWorks() throws Exception {
         ReservedClusterStateService stateService = mock(ReservedClusterStateService.class);
 
@@ -176,18 +175,11 @@ public class FileSettingsServiceTests extends ESTestCase {
             return null;
         }).when(stateService).process(any(), (XContentParser) any(), any());
 
-        AtomicBoolean settingsChanged = new AtomicBoolean(false);
         CountDownLatch latch = new CountDownLatch(1);
 
         final FileSettingsService service = spy(new FileSettingsService(clusterService, stateService, env));
 
-        service.addFileChangedListener(() -> settingsChanged.set(true));
-
-        doAnswer((Answer<Void>) invocation -> {
-            invocation.callRealMethod();
-            latch.countDown();
-            return null;
-        }).when(service).processFileChanges();
+        service.addFileChangedListener(latch::countDown);
 
         Files.createDirectories(service.watchedFileDir());
         // contents of the JSON don't matter, we just need a file to exist
@@ -196,12 +188,10 @@ public class FileSettingsServiceTests extends ESTestCase {
         service.start();
         service.clusterChanged(new ClusterChangedEvent("test", clusterService.state(), ClusterState.EMPTY_STATE));
 
-        // wait until the watcher thread has started, and it has discovered the file
+        // wait for listener to be called
         assertTrue(latch.await(20, TimeUnit.SECONDS));
 
         verify(service, times(1)).processFileChanges();
-        // assert we notified the listeners the file settings have changed, they were successfully applied
-        assertTrue(settingsChanged.get());
 
         service.stop();
         service.close();

x-pack/plugin/spatial/src/main/java/org/elasticsearch/xpack/spatial/search/aggregations/InternalGeoLine.java

@@ -116,6 +116,8 @@ public class InternalGeoLine extends InternalAggregation implements GeoShapeMetr
         int mergedSize = 0;
         boolean reducedComplete = true;
         boolean reducedIncludeSorts = true;
+        boolean reducedNonOverlapping = this.nonOverlapping;
+        boolean reducedSimplified = this.simplified;
         List<InternalGeoLine> internalGeoLines = new ArrayList<>(aggregations.size());
         for (InternalAggregation aggregation : aggregations) {
             InternalGeoLine geoLine = (InternalGeoLine) aggregation;
@@ -123,13 +125,16 @@ public class InternalGeoLine extends InternalAggregation implements GeoShapeMetr
             mergedSize += geoLine.line.length;
             reducedComplete &= geoLine.complete;
             reducedIncludeSorts &= geoLine.includeSorts;
+            reducedNonOverlapping &= geoLine.nonOverlapping;
+            reducedSimplified |= geoLine.simplified;
         }
         reducedComplete &= mergedSize <= size;
         int finalSize = Math.min(mergedSize, size);
 
-        MergedGeoLines mergedGeoLines = nonOverlapping
-            ? new MergedGeoLines.NonOverlapping(internalGeoLines, finalSize, sortOrder, simplified)
-            : new MergedGeoLines.Overlapping(internalGeoLines, finalSize, sortOrder, simplified);
+        // If all geo_lines are marked as non-overlapping, then we can optimize the merge-sort
+        MergedGeoLines mergedGeoLines = reducedNonOverlapping
+            ? new MergedGeoLines.NonOverlapping(internalGeoLines, finalSize, sortOrder, reducedSimplified)
+            : new MergedGeoLines.Overlapping(internalGeoLines, finalSize, sortOrder, reducedSimplified);
         mergedGeoLines.merge();
         return new InternalGeoLine(
             name,

x-pack/plugin/spatial/src/test/java/org/elasticsearch/xpack/spatial/search/aggregations/MergedGeoLinesTests.java

@@ -8,16 +8,19 @@ package org.elasticsearch.xpack.spatial.search.aggregations;
 
 import org.apache.lucene.geo.GeoEncodingUtils;
 import org.elasticsearch.common.util.ArrayUtils;
+import org.elasticsearch.search.aggregations.InternalAggregation;
 import org.elasticsearch.search.sort.SortOrder;
 import org.elasticsearch.test.ESTestCase;
 import org.hamcrest.CoreMatchers;
 import org.hamcrest.Matcher;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
+import java.util.Map;
 import java.util.TreeSet;
 
 import static org.hamcrest.Matchers.equalTo;
@@ -145,6 +148,27 @@ public class MergedGeoLinesTests extends ESTestCase {
         }
     }
 
+    public void testEmptyAndNonOverlappingGeoLines() throws IOException {
+        int docsPerLine = 10;
+        int numLines = 10;
+        int finalLength = 25;  // should get entire 100 points simplified down to 25
+        boolean simplify = true;
+        for (SortOrder sortOrder : new SortOrder[] { SortOrder.ASC, SortOrder.DESC }) {
+            InternalAggregation empty = makeEmptyGeoLine(sortOrder, finalLength);
+            List<InternalGeoLine> sorted = makeGeoLines(docsPerLine, numLines, simplify, sortOrder);
+            // Shuffle to ensure the tests cover geo_lines coming from data nodes in random order
+            List<InternalGeoLine> shuffled = shuffleGeoLines(sorted);
+            ArrayList<InternalAggregation> aggregations = new ArrayList<>(shuffled);
+            InternalGeoLine reduced = (InternalGeoLine) empty.reduce(aggregations, null);
+            assertLinesSimplified(sorted, sortOrder, finalLength, reduced.sortVals(), reduced.line());
+        }
+    }
+
+    private InternalGeoLine makeEmptyGeoLine(SortOrder sortOrder, int size) throws IOException {
+        // Make sure this matches the contents of 'GeoLineAggregator.buildEmptyAggregation'
+        return new InternalGeoLine("test", new long[0], new double[0], Map.of(), true, randomBoolean(), sortOrder, size, true, false);
+    }
+
     private void assertLinesTruncated(List<InternalGeoLine> lines, int docsPerLine, int finalLength, MergedGeoLines mergedGeoLines) {
         double[] values = mergedGeoLines.getFinalSortValues();
         long[] points = mergedGeoLines.getFinalPoints();
@@ -162,8 +186,10 @@ public class MergedGeoLinesTests extends ESTestCase {
     }
 
     private void assertLinesSimplified(List<InternalGeoLine> lines, SortOrder sortOrder, int finalLength, MergedGeoLines mergedGeoLines) {
-        double[] values = mergedGeoLines.getFinalSortValues();
-        long[] points = mergedGeoLines.getFinalPoints();
+        assertLinesSimplified(lines, sortOrder, finalLength, mergedGeoLines.getFinalSortValues(), mergedGeoLines.getFinalPoints());
+    }
+
+    private void assertLinesSimplified(List<InternalGeoLine> lines, SortOrder sortOrder, int finalLength, double[] values, long[] points) {
         assertThat("Same length arrays", values.length, equalTo(points.length));
         assertThat("Geo-line is simplified", values.length, equalTo(finalLength));
         GeoLineAggregatorTests.TestLine simplified = makeSimplifiedLine(lines, sortOrder, finalLength);