Fix time-series geo_line to include reduce phase in MergedGeoLines (#96953)

Recently we added support for line-simplification in geo-lines when in time-series aggregations: #94954. This work did not, however, cover the case of `MergedGeoLines`, which occurs when the same geo_line covers multiple shards on data nodes, and needs to be merged on the coordinating node. The original work would use line-simplification and time-ordering on the data node, but then revert to re-sorting (unnecessary) and truncation (incorrect) on the coordinating node. This PR rectifies that, and adds two fields to the InternalGeoLine: * nonOverlapping * re-sorting is *NOT* required because the sort value ranges do not overlap, so we only sort the set of incoming geo_lines (easy/fast) instead of every single point (complex/slow). * simplified * The approach to memory limiting is by line-simplification instead of truncation. * If the data nodes are doing line-simplification, we want the coordinating node to do that as well For time-series, both of the above are true, and at this point all other cases will have both false. The reason for two booleans is: * They are not necessarily related concepts * We consider supporting line-simplification as a user-requested option in the non-time-series geo_line in future (much nicer than truncation) Fixes #96983

Author: craigtaverner

docs/changelog/96953.yaml

@@ -0,0 +1,6 @@
+pr: 96953
+summary: Fix time-series geo_line to include reduce phase in MergedGeoLines
+area: Geo
+type: bug
+issues:
+ - 96983

server/src/main/java/org/elasticsearch/TransportVersion.java

@@ -146,8 +146,10 @@ private static TransportVersion registerTransportVersion(int id, String uniqueId
  // 8.10.0
  public static final TransportVersion V_8_500_019 = registerTransportVersion(8_500_019, "09bae57f-cab8-423c-aab3-c9778509ffe3");
 
+ public static final TransportVersion V_8_500_020 = registerTransportVersion(8_500_020, "ECB42C26-B258-42E5-A835-E31AF84A76DE");
+
  private static class CurrentHolder {
- private static final TransportVersion CURRENT = findCurrent(V_8_500_019);
+ private static final TransportVersion CURRENT = findCurrent(V_8_500_020);
 
  // finds the pluggable current version, or uses the given fallback
  private static TransportVersion findCurrent(TransportVersion fallback) {

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

@@ -61,7 +61,7 @@ public ScoreMode scoreMode() {
 
  @Override
  public InternalAggregation buildEmptyAggregation() {
- return new InternalGeoLine(name, new long[0], new double[0], metadata(), true, includeSorts, sortOrder, size);
+ return new InternalGeoLine(name, new long[0], new double[0], metadata(), true, includeSorts, sortOrder, size, true, false);
  }
 
  static class Empty extends GeoLineAggregator {

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

@@ -68,7 +68,7 @@ InternalGeoLine buildAggregation(
  double[] sortVals = getSortValues(bucket);
  long[] bucketLine = getPoints(bucket);
  PathArraySorter.forOrder(sortOrder).apply(bucketLine, sortVals).sort();
- return new InternalGeoLine(name, bucketLine, sortVals, metadata, complete, includeSorts, sortOrder, size);
+ return new InternalGeoLine(name, bucketLine, sortVals, metadata, complete, includeSorts, sortOrder, size, false, false);
  }
 
  long sizeOf(long bucket) {

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

@@ -7,6 +7,7 @@
 package org.elasticsearch.xpack.spatial.search.aggregations;
 
 import org.apache.lucene.geo.GeoEncodingUtils;
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -30,12 +31,14 @@
 public class InternalGeoLine extends InternalAggregation implements GeoShapeMetricAggregation {
  private static final double SCALE = Math.pow(10, 6);
 
- private long[] line;
- private double[] sortVals;
- private boolean complete;
- private boolean includeSorts;
- private SortOrder sortOrder;
- private int size;
+ private final long[] line;
+ private final double[] sortVals;
+ private final boolean complete;
+ private final boolean includeSorts;
+ private final SortOrder sortOrder;
+ private final int size;
+ private final boolean nonOverlapping;
+ private final boolean simplified;
 
  /**
  * A geo_line representing the bucket for a {@link GeoLineAggregationBuilder}. The values of <code>line</code> and <code>sortVals</code>
@@ -49,6 +52,8 @@ public class InternalGeoLine extends InternalAggregation implements GeoShapeMetr
  * @param includeSorts true iff the sort-values should be rendered in xContent as properties of the line-string. False otherwise.
  * @param sortOrder the {@link SortOrder} for the line. Whether the points are to be plotted in asc or desc order
  * @param size the max length of the line-string.
+ * @param nonOverlapping true iff the geo_line will not overlap with other geo_lines at reduce phase, allowing a simpler reduce.
+ * @param simplified true iff the geo_line was created by line simplification (not truncation) so we should do so in reduce also.
  */
  InternalGeoLine(
  String name,
@@ -58,7 +63,9 @@ public class InternalGeoLine extends InternalAggregation implements GeoShapeMetr
  boolean complete,
  boolean includeSorts,
  SortOrder sortOrder,
- int size
+ int size,
+ boolean nonOverlapping,
+ boolean simplified
  ) {
  super(name, metadata);
  this.line = line;
@@ -67,6 +74,8 @@ public class InternalGeoLine extends InternalAggregation implements GeoShapeMetr
  this.includeSorts = includeSorts;
  this.sortOrder = sortOrder;
  this.size = size;
+ this.nonOverlapping = nonOverlapping;
+ this.simplified = simplified;
  }
 
  /**
@@ -80,6 +89,13 @@ public InternalGeoLine(StreamInput in) throws IOException {
  this.includeSorts = in.readBoolean();
  this.sortOrder = SortOrder.readFromStream(in);
  this.size = in.readVInt();
+ if (in.getTransportVersion().onOrAfter(TransportVersion.V_8_500_020)) {
+ nonOverlapping = in.readBoolean();
+ simplified = in.readBoolean();
+ } else {
+ nonOverlapping = false;
+ simplified = false;
+ }
  }
 
  @Override
@@ -90,6 +106,10 @@ protected void doWriteTo(StreamOutput out) throws IOException {
  out.writeBoolean(includeSorts);
  sortOrder.writeTo(out);
  out.writeVInt(size);
+ if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_500_020)) {
+ out.writeBoolean(nonOverlapping);
+ out.writeBoolean(simplified);
+ }
  }
 
  @Override
@@ -108,7 +128,9 @@ public InternalAggregation reduce(List<InternalAggregation> aggregations, Aggreg
  reducedComplete &= mergedSize <= size;
  int finalSize = Math.min(mergedSize, size);
 
- MergedGeoLines mergedGeoLines = new MergedGeoLines(internalGeoLines, finalSize, sortOrder);
+ MergedGeoLines mergedGeoLines = nonOverlapping
+ ? new MergedGeoLines.NonOverlapping(internalGeoLines, finalSize, sortOrder, simplified)
+ : new MergedGeoLines.Overlapping(internalGeoLines, finalSize, sortOrder, simplified);
  mergedGeoLines.merge();
  return new InternalGeoLine(
  name,
@@ -118,7 +140,9 @@ public InternalAggregation reduce(List<InternalAggregation> aggregations, Aggreg
  reducedComplete,
  reducedIncludeSorts,
  sortOrder,
- size
+ size,
+ nonOverlapping,
+ simplified
  );
  }