Add geo_line aggregation (#41612) (#65442)

A metric aggregation that aggregates a set of points as
a GeoJSON LineString ordered by some sort parameter.

A `geo_line` aggregation request would specify a `geo_point` field, as well
as a `sort` field. `geo_point` represents the values used in the LineString,
while the `sort` values will be used as the total ordering of the points.

the `sort` field would support any numeric field, including date.

```
{
	"query": {
		"bool": {
			"must": [
				{ "term": { "person": "004" } },
				{ "term": { "trajectory": "20090131002206.plt" } }
			]
		}
	},
	"aggs": {
		"make_line": {
			"geo_line": {
				"point": {"field": "location"},
				"sort": { "field": "timestamp" },
                                "include_sort": true,
                                "sort_order": "desc",
                                "size": 15
			}
		}
	}
}
```

```
{
    "took": 21,
    "timed_out": false,
    "_shards": {...},
    "hits": {...},
    "aggregations": {
        "make_line": {
            "type": "LineString",
            "coordinates": [
                [
                    121.52926194481552,
                    38.92878997139633
                ],
                [
                    121.52922699227929,
                    38.92876998055726
                ],
             ]
        }
    }
}
```

Due to the cardinality of points, an initial max of 10k points
will be used. This should support many use-cases.

One solution to overcome this limitation is to keep a PriorityQueue of
points, and simplifying the line once it hits this max. If simplifying
makes sense, it may be a nice option, in general. The ability to use a parameter
to specify how aggressive one wants to simplify. This parameter could be
the number of points. Example algorithm one could use with a PriorityQueue:
https://bost.ocks.org/mike/simplify/. This would still require O(m) space, where m
is the number of points returned. And would also require heapifying triangles
sorted by their areas, which would be O(log(m)) operations. Since sorting is done,
anyways, simplifying would still be a O(n log(m)) operation, where n is the total number
of points to filter........... something to explore

closes #41649

Author: talevy

docs/reference/aggregations/metrics.asciidoc

@@ -23,6 +23,8 @@ include::metrics/geobounds-aggregation.asciidoc[]
 
 include::metrics/geocentroid-aggregation.asciidoc[]
 
+include::metrics/geoline-aggregation.asciidoc[]
+
 include::metrics/matrix-stats-aggregation.asciidoc[]
 
 include::metrics/max-aggregation.asciidoc[]

docs/reference/aggregations/metrics/geoline-aggregation.asciidoc

@@ -0,0 +1,143 @@
+[role="xpack"]
+[testenv="gold"]
+[[search-aggregations-metrics-geo-line]]
+=== Geo-Line Aggregation
+++++
+<titleabbrev>Geo-Line</titleabbrev>
+++++
+
+The `geo_line` aggregation aggregates all `geo_point` values within a bucket into a LineString ordered
+by the chosen `sort` field. This `sort` can be a date field, for example. The bucket returned is a valid
+https://tools.ietf.org/html/rfc7946#section-3.2[GeoJSON Feature] representing the line geometry.
+
+[source,console,id=search-aggregations-metrics-geo-line-simple]
+----
+PUT test
+{
+    "mappings": {
+        "dynamic": "strict",
+        "_source": {
+            "enabled": false
+        },
+        "properties": {
+            "my_location": {
+                "type": "geo_point"
+            },
+            "group": {
+                "type": "keyword"
+            },
+            "@timestamp": {
+                "type": "date"
+            }
+        }
+    }
+}
+
+POST /test/_bulk?refresh
+{"index": {}}
+{"my_location": {"lat":37.3450570, "lon": -122.0499820}, "@timestamp": "2013-09-06T16:00:36"}
+{"index": {}}
+{"my_location": {"lat": 37.3451320, "lon": -122.0499820}, "@timestamp": "2013-09-06T16:00:37Z"}
+{"index": {}}
+{"my_location": {"lat": 37.349283, "lon": -122.0505010}, "@timestamp": "2013-09-06T16:00:37Z"}
+
+POST /test/_search?filter_path=aggregations
+{
+  "aggs": {
+    "line": {
+      "geo_line": {
+        "point": {"field": "my_location"},
+        "sort": {"field": "@timestamp"}
+      }
+    }
+  }
+}
+----
+
+Which returns:
+
+[source,js]
+----
+{
+  "aggregations": {
+    "line": {
+      "type" : "Feature",
+      "geometry" : {
+        "type" : "LineString",
+        "coordinates" : [
+          [
+            -122.049982,
+            37.345057
+          ],
+          [
+            -122.050501,
+            37.349283
+          ],
+          [
+            -122.049982,
+            37.345132
+          ]
+        ]
+      },
+      "properties" : {
+        "complete" : true
+      }
+    }
+  }
+}
+----
+// TESTRESPONSE
+
+[[search-aggregations-metrics-geo-line-options]]
+==== Options
+
+`point`::
+(Required)
+
+This option specifies the name of the `geo_point` field
+
+Example usage configuring `my_location` as the point field:
+
+[source,js]
+----
+"point": {
+  "field": "my_location"
+}
+----
+// NOTCONSOLE
+
+`sort`::
+(Required)
+
+This option specifies the name of the numeric field to use as the sort key
+for ordering the points
+
+Example usage configuring `@timestamp` as the sort key:
+
+[source,js]
+----
+"point": {
+  "field": "@timestamp"
+}
+----
+// NOTCONSOLE
+
+`include_sort`::
+(Optional, boolean, default: `false`)
+
+This option includes, when true, an additional array of the sort values in the
+feature properties.
+
+`sort_order`::
+(Optional, string, default: `"ASC"`)
+
+This option accepts one of two values: "ASC", "DESC".
+
+The line is sorted in ascending order by the sort key when set to "ASC", and in descending
+with "DESC".
+
+`size`::
+(Optional, integer, default: `10000`)
+
+The maximum length of the line represented in the aggregation. Valid sizes are
+between one and 10000.

server/src/main/java/org/elasticsearch/search/sort/BucketedSort.java

@@ -66,7 +66,7 @@
  * worst case. Critically, it is a very fast {@code O(1)} to check if a value
  * is competitive at all which, so long as buckets aren't hit in reverse
  * order, they mostly won't be. Extracting results in sorted order is still
- * {@code O(n * log n)}. 
+ * {@code O(n * log n)}.
  * </p>
  * <p>
  * When we first collect a bucket we make sure that we've allocated enough
@@ -90,7 +90,7 @@ public interface ExtraData {
          * <p>
          * Both parameters will have previously been loaded by
          * {@link Loader#loadFromDoc(long, int)} so the implementer shouldn't
-         * need to grow the underlying storage to implement this. 
+         * need to grow the underlying storage to implement this.
          * </p>
          */
         void swap(long lhs, long rhs);
@@ -128,7 +128,7 @@ public Loader loader(LeafReaderContext ctx) throws IOException {
     private final SortOrder order;
     private final DocValueFormat format;
     private final int bucketSize;
-    private final ExtraData extra;
+    protected final ExtraData extra;
     /**
      * {@code true} if the bucket is in heap mode, {@code false} if
      * it is still gathering.
@@ -206,9 +206,9 @@ public final List<SortValue> getValues(long bucket) throws IOException {
     }
 
     /**
-     * Is this bucket a min heap {@code true} or in gathering mode {@code false}? 
+     * Is this bucket a min heap {@code true} or in gathering mode {@code false}?
      */
-    private boolean inHeapMode(long bucket) {
+    public boolean inHeapMode(long bucket) {
         return heapMode.get(bucket);
     }
 
@@ -254,7 +254,7 @@ private boolean inHeapMode(long bucket) {
     /**
      * {@code true} if the entry at index {@code lhs} is "better" than
      * the entry at {@code rhs}. "Better" in this means "lower" for
-     * {@link SortOrder#ASC} and "higher" for {@link SortOrder#DESC}. 
+     * {@link SortOrder#ASC} and "higher" for {@link SortOrder#DESC}.
      */
     protected abstract boolean betterThan(long lhs, long rhs);
 
@@ -283,7 +283,7 @@ protected final String debugFormat() {
 
     /**
      * Initialize the gather offsets after setting up values. Subclasses
-     * should call this once, after setting up their {@link #values()}.  
+     * should call this once, after setting up their {@link #values()}.
      */
     protected final void initGatherOffsets() {
         setNextGatherOffsets(0);
@@ -325,12 +325,12 @@ private void setNextGatherOffsets(long startingAt) {
      * case.
      * </p>
      * <ul>
-     * <li>Hayward, Ryan; McDiarmid, Colin (1991).  
+     * <li>Hayward, Ryan; McDiarmid, Colin (1991).
      * <a href="https://web.archive.org/web/20160205023201/http://www.stats.ox.ac.uk/__data/assets/pdf_file/0015/4173/heapbuildjalg.pdf">
      * Average Case Analysis of Heap Building byRepeated Insertion</a> J. Algorithms.
      * <li>D.E. Knuth, ”The Art of Computer Programming, Vol. 3, Sorting and Searching”</li>
      * </ul>
-     * @param rootIndex the index the start of the bucket 
+     * @param rootIndex the index the start of the bucket
      */
     private void heapify(long rootIndex) {
         int maxParent = bucketSize / 2 - 1;
@@ -344,7 +344,7 @@ private void heapify(long rootIndex) {
      * runs in {@code O(log n)} time.
      * @param rootIndex index of the start of the bucket
      * @param parent Index within the bucket of the parent to check.
-     *               For example, 0 is the "root". 
+     *               For example, 0 is the "root".
      */
     private void downHeap(long rootIndex, int parent) {
         while (true) {
@@ -443,7 +443,7 @@ public final void collect(int doc, long bucket) throws IOException {
         /**
          * {@code true} if the sort value for the doc is "better" than the
          * entry at {@code index}. "Better" in means is "lower" for
-         * {@link SortOrder#ASC} and "higher" for {@link SortOrder#DESC}. 
+         * {@link SortOrder#ASC} and "higher" for {@link SortOrder#DESC}.
          */
         protected abstract boolean docBetterThan(long index);
 
@@ -545,7 +545,7 @@ public abstract static class ForFloats extends BucketedSort {
          * The maximum size of buckets this can store. This is because we
          * store the next offset to write to in a float and floats only have
          * {@code 23} bits of mantissa so they can't accurate store values
-         * higher than {@code 2 ^ 24}. 
+         * higher than {@code 2 ^ 24}.
          */
         public static final int MAX_BUCKET_SIZE = (int) Math.pow(2, 24);
 

x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/topmetrics/TopMetricsAggregator.java

@@ -32,6 +32,7 @@
 import org.elasticsearch.search.sort.BucketedSort;
 import org.elasticsearch.search.sort.SortBuilder;
 import org.elasticsearch.search.sort.SortValue;
+import org.elasticsearch.xpack.core.common.search.aggregations.MissingHelper;
 import org.elasticsearch.xpack.analytics.topmetrics.InternalTopMetrics.MetricValue;
 
 import java.io.IOException;
@@ -495,62 +496,4 @@ public Loader loader(LeafReaderContext ctx) throws IOException {
         public void close() {}
     }
 
-    /**
-     * Helps {@link LongMetricValues} track "empty" slots. It attempts to have
-     * very low CPU overhead and no memory overhead when there *aren't* empty
-     * values.
-     */
-    private static class MissingHelper implements Releasable {
-        private final BigArrays bigArrays;
-        private BitArray tracker;
-
-        MissingHelper(BigArrays bigArrays) {
-            this.bigArrays = bigArrays;
-        }
-
-        void markMissing(long index) {
-            if (tracker == null) {
-                tracker = new BitArray(index, bigArrays);
-            }
-            tracker.set(index);
-        }
-
-        void markNotMissing(long index) {
-            if (tracker == null) {
-                return;
-            }
-            tracker.clear(index);
-        }
-
-        void swap(long lhs, long rhs) {
-            if (tracker == null) {
-                return;
-            }
-            boolean backup = tracker.get(lhs);
-            if (tracker.get(rhs)) {
-                tracker.set(lhs);
-            } else {
-                tracker.clear(lhs);
-            }
-            if (backup) {
-                tracker.set(rhs);
-            } else {
-                tracker.clear(rhs);
-            }
-        }
-
-        boolean isEmpty(long index) {
-            if (tracker == null) {
-                return false;
-            }
-            return tracker.get(index);
-        }
-
-        @Override
-        public void close() {
-            if (tracker != null) {
-                tracker.close();
-            }
-        }
-    }
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/license/XPackLicenseState.java

@@ -98,6 +98,8 @@ public enum Feature {
 
         SPATIAL_GEO_GRID(OperationMode.GOLD, true),
 
+        SPATIAL_GEO_LINE(OperationMode.GOLD, true),
+
         ANALYTICS(OperationMode.MISSING, true),
 
         AGGREGATE_METRIC(OperationMode.MISSING, true),