Support multivalued fields in ungrouped aggs (ESQL-1028)

This adds support for reading multivalued fields in ungrouped aggregations so `from foo | stats max(a)` will find the actual max of `a`, even if there are many values for `a`. I had to modify a method in `AbstractBlock` to make it useful for aggregations. It wasn't used anywhere else so that feels safe.
elastic · Apr 20, 2023 · 1431236 · 1431236
1 parent e05d9a5
commit 1431236
Show file tree

Hide file tree

Showing 52 changed files with 405 additions and 259 deletions.
diff --git a/...n/esql/compute/gen/src/main/java/org/elasticsearch/compute/gen/AggregatorImplementer.java b/...n/esql/compute/gen/src/main/java/org/elasticsearch/compute/gen/AggregatorImplementer.java
@@ -225,16 +225,20 @@ private MethodSpec addRawVector() {
     private MethodSpec addRawBlock() {
         MethodSpec.Builder builder = MethodSpec.methodBuilder("addRawBlock");
         builder.addModifiers(Modifier.PRIVATE).addParameter(valueBlockType(init, combine), "block");
-        builder.beginControlFlow("for (int p = 0; p < block.getTotalValueCount(); p++)");
+        builder.beginControlFlow("for (int p = 0; p < block.getPositionCount(); p++)");
         {
-            builder.beginControlFlow("if (block.isNull(p) == false)");
-            builder.addStatement("int i = block.getFirstValueIndex(p)");
+            builder.beginControlFlow("if (block.isNull(p))");
+            builder.addStatement("continue");
+            builder.endControlFlow();
+            builder.addStatement("int start = block.getFirstValueIndex(p)");
+            builder.addStatement("int end = start + block.getValueCount(p)");
+            builder.beginControlFlow("for (int i = start; i < end; i++)");
             combineRawInput(builder, "block");
             builder.endControlFlow();
         }
         builder.endControlFlow();
         if (combineValueCount != null) {
-            builder.addStatement("$T.combineValueCount(state, block.validPositionCount())", declarationType);
+            builder.addStatement("$T.combineValueCount(state, block.getTotalValueCount())", declarationType);
         }
         return builder.build();
     }

diff --git a/...src/main/generated/org/elasticsearch/compute/aggregation/AvgDoubleAggregatorFunction.java b/...src/main/generated/org/elasticsearch/compute/aggregation/AvgDoubleAggregatorFunction.java
@@ -58,13 +58,17 @@ private void addRawVector(DoubleVector vector) {
   }
 
   private void addRawBlock(DoubleBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         AvgDoubleAggregator.combine(state, block.getDouble(i));
       }
     }
-    AvgDoubleAggregator.combineValueCount(state, block.validPositionCount());
+    AvgDoubleAggregator.combineValueCount(state, block.getTotalValueCount());
   }
 
   @Override

diff --git a/...te/src/main/generated/org/elasticsearch/compute/aggregation/AvgIntAggregatorFunction.java b/...te/src/main/generated/org/elasticsearch/compute/aggregation/AvgIntAggregatorFunction.java
@@ -57,13 +57,17 @@ private void addRawVector(IntVector vector) {
   }
 
   private void addRawBlock(IntBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         AvgIntAggregator.combine(state, block.getInt(i));
       }
     }
-    AvgIntAggregator.combineValueCount(state, block.validPositionCount());
+    AvgIntAggregator.combineValueCount(state, block.getTotalValueCount());
   }
 
   @Override

diff --git a/...e/src/main/generated/org/elasticsearch/compute/aggregation/AvgLongAggregatorFunction.java b/...e/src/main/generated/org/elasticsearch/compute/aggregation/AvgLongAggregatorFunction.java
@@ -58,13 +58,17 @@ private void addRawVector(LongVector vector) {
   }
 
   private void addRawBlock(LongBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         AvgLongAggregator.combine(state, block.getLong(i));
       }
     }
-    AvgLongAggregator.combineValueCount(state, block.validPositionCount());
+    AvgLongAggregator.combineValueCount(state, block.getTotalValueCount());
   }
 
   @Override

diff --git a/...nerated/org/elasticsearch/compute/aggregation/CountDistinctBooleanAggregatorFunction.java b/...nerated/org/elasticsearch/compute/aggregation/CountDistinctBooleanAggregatorFunction.java
@@ -58,9 +58,13 @@ private void addRawVector(BooleanVector vector) {
   }
 
   private void addRawBlock(BooleanBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         CountDistinctBooleanAggregator.combine(state, block.getBoolean(i));
       }
     }

diff --git a/...enerated/org/elasticsearch/compute/aggregation/CountDistinctDoubleAggregatorFunction.java b/...enerated/org/elasticsearch/compute/aggregation/CountDistinctDoubleAggregatorFunction.java
@@ -57,9 +57,13 @@ private void addRawVector(DoubleVector vector) {
   }
 
   private void addRawBlock(DoubleBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         CountDistinctDoubleAggregator.combine(state, block.getDouble(i));
       }
     }

diff --git a/...n/generated/org/elasticsearch/compute/aggregation/CountDistinctIntAggregatorFunction.java b/...n/generated/org/elasticsearch/compute/aggregation/CountDistinctIntAggregatorFunction.java
@@ -56,9 +56,13 @@ private void addRawVector(IntVector vector) {
   }
 
   private void addRawBlock(IntBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         CountDistinctIntAggregator.combine(state, block.getInt(i));
       }
     }

diff --git a/.../generated/org/elasticsearch/compute/aggregation/CountDistinctLongAggregatorFunction.java b/.../generated/org/elasticsearch/compute/aggregation/CountDistinctLongAggregatorFunction.java
@@ -57,9 +57,13 @@ private void addRawVector(LongVector vector) {
   }
 
   private void addRawBlock(LongBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         CountDistinctLongAggregator.combine(state, block.getLong(i));
       }
     }

diff --git a/...src/main/generated/org/elasticsearch/compute/aggregation/MaxDoubleAggregatorFunction.java b/...src/main/generated/org/elasticsearch/compute/aggregation/MaxDoubleAggregatorFunction.java
@@ -57,9 +57,13 @@ private void addRawVector(DoubleVector vector) {
   }
 
   private void addRawBlock(DoubleBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         state.doubleValue(MaxDoubleAggregator.combine(state.doubleValue(), block.getDouble(i)));
       }
     }

diff --git a/...te/src/main/generated/org/elasticsearch/compute/aggregation/MaxIntAggregatorFunction.java b/...te/src/main/generated/org/elasticsearch/compute/aggregation/MaxIntAggregatorFunction.java
@@ -56,9 +56,13 @@ private void addRawVector(IntVector vector) {
   }
 
   private void addRawBlock(IntBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         state.intValue(MaxIntAggregator.combine(state.intValue(), block.getInt(i)));
       }
     }

diff --git a/...e/src/main/generated/org/elasticsearch/compute/aggregation/MaxLongAggregatorFunction.java b/...e/src/main/generated/org/elasticsearch/compute/aggregation/MaxLongAggregatorFunction.java
@@ -57,9 +57,13 @@ private void addRawVector(LongVector vector) {
   }
 
   private void addRawBlock(LongBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         state.longValue(MaxLongAggregator.combine(state.longValue(), block.getLong(i)));
       }
     }

diff --git a/...rg/elasticsearch/compute/aggregation/MedianAbsoluteDeviationDoubleAggregatorFunction.java b/...rg/elasticsearch/compute/aggregation/MedianAbsoluteDeviationDoubleAggregatorFunction.java
@@ -58,9 +58,13 @@ private void addRawVector(DoubleVector vector) {
   }
 
   private void addRawBlock(DoubleBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         MedianAbsoluteDeviationDoubleAggregator.combine(state, block.getDouble(i));
       }
     }

diff --git a/...d/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationIntAggregatorFunction.java b/...d/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationIntAggregatorFunction.java
@@ -57,9 +57,13 @@ private void addRawVector(IntVector vector) {
   }
 
   private void addRawBlock(IntBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         MedianAbsoluteDeviationIntAggregator.combine(state, block.getInt(i));
       }
     }

diff --git a/.../org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationLongAggregatorFunction.java b/.../org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationLongAggregatorFunction.java
@@ -58,9 +58,13 @@ private void addRawVector(LongVector vector) {
   }
 
   private void addRawBlock(LongBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         MedianAbsoluteDeviationLongAggregator.combine(state, block.getLong(i));
       }
     }

diff --git a/.../main/generated/org/elasticsearch/compute/aggregation/MedianDoubleAggregatorFunction.java b/.../main/generated/org/elasticsearch/compute/aggregation/MedianDoubleAggregatorFunction.java
@@ -57,9 +57,13 @@ private void addRawVector(DoubleVector vector) {
   }
 
   private void addRawBlock(DoubleBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         MedianDoubleAggregator.combine(state, block.getDouble(i));
       }
     }

diff --git a/...src/main/generated/org/elasticsearch/compute/aggregation/MedianIntAggregatorFunction.java b/...src/main/generated/org/elasticsearch/compute/aggregation/MedianIntAggregatorFunction.java
@@ -56,9 +56,13 @@ private void addRawVector(IntVector vector) {
   }
 
   private void addRawBlock(IntBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         MedianIntAggregator.combine(state, block.getInt(i));
       }
     }

diff --git a/...rc/main/generated/org/elasticsearch/compute/aggregation/MedianLongAggregatorFunction.java b/...rc/main/generated/org/elasticsearch/compute/aggregation/MedianLongAggregatorFunction.java
@@ -57,9 +57,13 @@ private void addRawVector(LongVector vector) {
   }
 
   private void addRawBlock(LongBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         MedianLongAggregator.combine(state, block.getLong(i));
       }
     }

diff --git a/...src/main/generated/org/elasticsearch/compute/aggregation/MinDoubleAggregatorFunction.java b/...src/main/generated/org/elasticsearch/compute/aggregation/MinDoubleAggregatorFunction.java
@@ -57,9 +57,13 @@ private void addRawVector(DoubleVector vector) {
   }
 
   private void addRawBlock(DoubleBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         state.doubleValue(MinDoubleAggregator.combine(state.doubleValue(), block.getDouble(i)));
       }
     }

diff --git a/...te/src/main/generated/org/elasticsearch/compute/aggregation/MinIntAggregatorFunction.java b/...te/src/main/generated/org/elasticsearch/compute/aggregation/MinIntAggregatorFunction.java
@@ -56,9 +56,13 @@ private void addRawVector(IntVector vector) {
   }
 
   private void addRawBlock(IntBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         state.intValue(MinIntAggregator.combine(state.intValue(), block.getInt(i)));
       }
     }

diff --git a/...e/src/main/generated/org/elasticsearch/compute/aggregation/MinLongAggregatorFunction.java b/...e/src/main/generated/org/elasticsearch/compute/aggregation/MinLongAggregatorFunction.java
@@ -57,9 +57,13 @@ private void addRawVector(LongVector vector) {
   }
 
   private void addRawBlock(LongBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         state.longValue(MinLongAggregator.combine(state.longValue(), block.getLong(i)));
       }
     }

diff --git a/...src/main/generated/org/elasticsearch/compute/aggregation/SumDoubleAggregatorFunction.java b/...src/main/generated/org/elasticsearch/compute/aggregation/SumDoubleAggregatorFunction.java
@@ -57,9 +57,13 @@ private void addRawVector(DoubleVector vector) {
   }
 
   private void addRawBlock(DoubleBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         SumDoubleAggregator.combine(state, block.getDouble(i));
       }
     }

diff --git a/...te/src/main/generated/org/elasticsearch/compute/aggregation/SumIntAggregatorFunction.java b/...te/src/main/generated/org/elasticsearch/compute/aggregation/SumIntAggregatorFunction.java
@@ -57,9 +57,13 @@ private void addRawVector(IntVector vector) {
   }
 
   private void addRawBlock(IntBlock block) {
-    for (int p = 0; p < block.getTotalValueCount(); p++) {
-      if (block.isNull(p) == false) {
-        int i = block.getFirstValueIndex(p);
+    for (int p = 0; p < block.getPositionCount(); p++) {
+      if (block.isNull(p)) {
+        continue;
+      }
+      int start = block.getFirstValueIndex(p);
+      int end = start + block.getValueCount(p);
+      for (int i = start; i < end; i++) {
         state.longValue(SumIntAggregator.combine(state.longValue(), block.getInt(i)));
       }
     }