Fixes some edge cases and add a path array in serialization

Fixes min_depth parameter

Author: Clément Tourrière

src/main/java/org/opendatasoft/elasticsearch/search/aggregations/bucket/InternalPathHierarchy.java

@@ -17,6 +17,7 @@
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
@@ -31,6 +32,7 @@
 public class InternalPathHierarchy extends InternalMultiBucketAggregation<InternalPathHierarchy,
         InternalPathHierarchy.InternalBucket> implements PathHierarchy {
     protected static final ParseField SUM_OF_OTHER_HIERARCHY_NODES = new ParseField("sum_other_hierarchy_nodes");
+    protected static final ParseField PATHS = new ParseField("path");
 
     /**
      * The bucket class of InternalPathHierarchy.
@@ -45,13 +47,16 @@ public static class InternalBucket extends InternalMultiBucketAggregation.Intern
         protected long docCount;
         protected InternalAggregations aggregations;
         protected int level;
+        protected int minDepth;
         protected String basename;
 
-        public InternalBucket(long docCount, InternalAggregations aggregations, String basename, BytesRef term, int level, String[] paths) {
+        public InternalBucket(long docCount, InternalAggregations aggregations, String basename,
+                              BytesRef term, int level, int minDepth, String[] paths) {
             termBytes = term;
             this.docCount = docCount;
             this.aggregations = aggregations;
             this.level = level;
+            this.minDepth = minDepth;
             this.basename = basename;
             this.paths = paths;
         }
@@ -64,6 +69,7 @@ public InternalBucket(StreamInput in) throws IOException {
             docCount = in.readLong();
             aggregations = InternalAggregations.readAggregations(in);
             level = in.readInt();
+            minDepth = in.readInt();
             basename = in.readString();
             int pathsSize = in.readInt();
             paths = new String[pathsSize];
@@ -81,6 +87,7 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeLong(docCount);
             aggregations.writeTo(out);
             out.writeInt(level);
+            out.writeInt(minDepth);
             out.writeString(basename);
             out.writeInt(paths.length);
             for (String path: paths) {
@@ -229,7 +236,7 @@ public InternalPathHierarchy create(List<InternalBucket> buckets) {
     @Override
     public InternalBucket createBucket(InternalAggregations aggregations, InternalBucket prototype) {
         return new InternalBucket(prototype.docCount, aggregations, prototype.basename, prototype.termBytes,
-                prototype.level, prototype.paths);
+                prototype.level, prototype.minDepth, prototype.paths);
     }
 
     @Override
@@ -271,7 +278,13 @@ public InternalPathHierarchy doReduce(List<InternalAggregation> aggregations, Re
             final InternalBucket b = sameTermBuckets.get(0).reduce(sameTermBuckets, reduceContext);
             if (b.getDocCount() >= minDocCount || !reduceContext.isFinalReduce()) {
                 reduceContext.consumeBucketsAndMaybeBreak(1);
-                ordered.add(b.paths, b);
+                String [] pathsForTree;
+                if (b.minDepth > 0) {
+                    pathsForTree = Arrays.copyOfRange(b.paths, b.minDepth, b.paths.length);
+                } else {
+                    pathsForTree = b.paths;
+                }
+                ordered.add(pathsForTree, b);
             } else {
                 reduceContext.consumeBucketsAndMaybeBreak(-countInnerBucket(b));
             }
@@ -311,6 +324,7 @@ public XContentBuilder doXContentBody(XContentBuilder builder, Params params) th
             builder.startObject();
             builder.field(CommonFields.KEY.getPreferredName(), currentBucket.basename);
             builder.field(CommonFields.DOC_COUNT.getPreferredName(), currentBucket.docCount);
+            builder.field(PATHS.getPreferredName(), Arrays.copyOf(currentBucket.paths, currentBucket.paths.length -1));
             currentBucket.getAggregations().toXContentInternal(builder, params);
 
             prevBucket = currentBucket;
@@ -331,7 +345,7 @@ public XContentBuilder doXContentBody(XContentBuilder builder, Params params) th
 
     @Override
     protected int doHashCode() {
-        return Objects.hash(buckets, separator, order, requiredSize, shardSize, otherHierarchyNodes);
+        return Objects.hash(buckets, separator, order, requiredSize, shardSize, otherHierarchyNodes, minDocCount);
     }
 
     @Override
@@ -340,6 +354,7 @@ protected boolean doEquals(Object obj) {
         return Objects.equals(buckets, that.buckets)
                 && Objects.equals(separator, that.separator)
                 && Objects.equals(order, that.order)
+                && Objects.equals(minDocCount, that.minDocCount)
                 && Objects.equals(requiredSize, that.requiredSize)
                 && Objects.equals(shardSize, that.shardSize)
                 && Objects.equals(otherHierarchyNodes, that.otherHierarchyNodes);

src/main/java/org/opendatasoft/elasticsearch/search/aggregations/bucket/PathHierarchyAggregationBuilder.java

@@ -302,7 +302,7 @@ protected XContentBuilder doXContentBody(XContentBuilder builder, Params params)
      */
     @Override
     protected int innerHashCode() {
-        return Objects.hash(separator, minDepth, maxDepth, depth, order);
+        return Objects.hash(separator, minDepth, maxDepth, depth, order, minDocCount, bucketCountThresholds);
     }
 
     @Override

src/main/java/org/opendatasoft/elasticsearch/search/aggregations/bucket/PathHierarchyAggregator.java

@@ -25,6 +25,7 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -120,6 +121,7 @@ public boolean equals(Object obj) {
     private final BytesRefHash bucketOrds;
     private final BucketOrder order;
     private final long minDocCount;
+    private final int minDepth;
     private final BucketCountThresholds bucketCountThresholds;
     private final BytesRef separator;
 
@@ -132,6 +134,7 @@ public PathHierarchyAggregator(
             long minDocCount,
             BucketCountThresholds bucketCountThresholds,
             BytesRef separator,
+            int minDepth,
             Aggregator parent,
             List<PipelineAggregator> pipelineAggregators,
             Map<String, Object> metaData
@@ -143,6 +146,7 @@ public PathHierarchyAggregator(
         bucketOrds = new BytesRefHash(1, context.bigArrays());
         this.order = InternalOrder.validate(order, this);
         this.bucketCountThresholds = bucketCountThresholds;
+        this.minDepth = minDepth;
     }
 
     /**
@@ -173,7 +177,7 @@ public void collect(int doc, long owningBucketOrdinal) throws IOException {
                     // SortedBinaryDocValues don't guarantee uniqueness so we need to take care of dups
                     for (int i = 0; i < valuesCount; ++i) {
                         final BytesRef bytesValue = values.nextValue();
-                        if (previous.get().equals(bytesValue)) {
+                        if (i > 0 && previous.get().equals(bytesValue)) {
                             continue;
                         }
                         long bucketOrdinal = bucketOrds.add(bytesValue);
@@ -201,21 +205,32 @@ public InternalAggregation buildAggregation(long owningBucketOrdinal) throws IOE
 
         InternalPathHierarchy.InternalBucket spare = null;
         for (int i = 0; i < bucketOrds.size(); i++) {
-            spare = new InternalPathHierarchy.InternalBucket(0, null, null, new BytesRef(), 0, null);
+            spare = new InternalPathHierarchy.InternalBucket(0, null, null, new BytesRef(), 0, 0, null);
             BytesRef term = new BytesRef();
             bucketOrds.get(i, term);
 
-            String [] paths = term.utf8ToString().split(Pattern.quote(separator.utf8ToString()), -1);
+            String quotedPattern = Pattern.quote(separator.utf8ToString());
+
+            String [] paths = term.utf8ToString().split(quotedPattern, -1);
+
+            String [] pathsForTree;
+
+            if (minDepth > 0) {
+                pathsForTree = Arrays.copyOfRange(paths, minDepth, paths.length);
+            } else {
+                pathsForTree = paths;
+            }
 
             spare.termBytes = BytesRef.deepCopyOf(term);
             spare.aggregations = bucketAggregations(i);
-            spare.level = paths.length - 1;
+            spare.level = pathsForTree.length - 1;
             spare.docCount = bucketDocCount(i);
             spare.basename = paths[paths.length - 1];
+            spare.minDepth = minDepth;
             spare.bucketOrd = i;
             spare.paths = paths;
 
-            pathSortedTree.add(paths, spare);
+            pathSortedTree.add(pathsForTree, spare);
 
             consumeBucketsAndMaybeBreak(1);
         }

src/main/java/org/opendatasoft/elasticsearch/search/aggregations/bucket/PathHierarchyAggregatorFactory.java

@@ -1,6 +1,7 @@
 package org.opendatasoft.elasticsearch.search.aggregations.bucket;
 
 import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.util.ArrayUtil;
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.BytesRefBuilder;
 import org.apache.lucene.util.FutureArrays;
@@ -102,7 +103,7 @@ protected Aggregator doCreateInternal(
         bucketCountThresholds.ensureValidity();
         return new PathHierarchyAggregator(
                 name, factories, context,
-                valuesSourceBytes, order, minDocCount, bucketCountThresholds, separator,
+                valuesSourceBytes, order, minDocCount, bucketCountThresholds, separator, minDepth,
                 parent, pipelineAggregators, metaData);
     }
 
@@ -162,17 +163,17 @@ public boolean advanceExact(int docId) throws IOException {
 
                             // A new path needs to be add
                             if (startNewValOffset != -1) {
+                                cleanVal.append(val.bytes, val.offset + startNewValOffset, offset - startNewValOffset);
                                 if (depth >= minDepth) {
-                                    cleanVal.append(val.bytes, val.offset + startNewValOffset, offset - startNewValOffset);
                                     values[t++].copyBytes(cleanVal);
-                                    cleanVal.append(separator);
                                 }
                                 startNewValOffset = -1;
+                                cleanVal.append(separator);
                                 depth ++;
                             // two separators following each other
                             } else if (keepBlankPath) {
                                 count++;
-                                grow();
+                                growExact();
                                 values[t++].copyBytes(cleanVal);
                                 cleanVal.append(separator);
                                 depth ++;
@@ -185,8 +186,10 @@ public boolean advanceExact(int docId) throws IOException {
                         } else {
                             if (startNewValOffset == -1) {
                                 startNewValOffset = offset;
-                                count++;
-                                grow();
+                                if (depth >= minDepth) {
+                                    count++;
+                                    growExact();
+                                }
                             }
                         }
                     }
@@ -202,8 +205,17 @@ public boolean advanceExact(int docId) throws IOException {
             } else
                 return false;
         }
-    }
 
+        final void growExact() {
+            if (values.length < count) {
+                final int oldLen = values.length;
+                values = ArrayUtil.growExact(values, count);
+                for (int i = oldLen; i < count; ++i) {
+                    values[i] = new BytesRefBuilder();
+                }
+            }
+        }
+    }
 
     /**
      * To get ValuesSource as sorted bytes.