Monomorphic processing of TopN queries with simple double aggregators over historical segments (part of #3798)

common/src/main/java/io/druid/annotations/SubclassesMustBePublic.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.annotations;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Documented
+@Inherited
+@Retention(RetentionPolicy.SOURCE)
+@Target(ElementType.TYPE)
+public @interface SubclassesMustBePublic
+{
+}

common/src/main/java/io/druid/annotations/SubclassesMustBePublicAnnotationProcessor.java

@@ -0,0 +1,51 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.annotations;
+
+import javax.annotation.processing.AbstractProcessor;
+import javax.annotation.processing.RoundEnvironment;
+import javax.annotation.processing.SupportedAnnotationTypes;
+import javax.lang.model.element.Element;
+import javax.lang.model.element.Modifier;
+import javax.lang.model.element.TypeElement;
+import javax.tools.Diagnostic;
+import java.util.Set;
+
+@SupportedAnnotationTypes("io.druid.annotations.SubclassesMustBePublic")
+public class SubclassesMustBePublicAnnotationProcessor extends AbstractProcessor
+{
+  @Override
+  public boolean process(Set<? extends TypeElement> annotations, RoundEnvironment roundEnv)
+  {
+    for (TypeElement annotation : annotations) {
+      Set<? extends Element> elementsAnnotatedWith = roundEnv.getElementsAnnotatedWith(annotation);
+      for (Element element : elementsAnnotatedWith) {
+        if (!element.getModifiers().contains(Modifier.PUBLIC)) {
+          processingEnv.getMessager().printMessage(
+              Diagnostic.Kind.ERROR,
+              element.getSimpleName() + " must be public",
+              element
+          );
+        }
+      }
+    }
+    return false;
+  }
+}

common/src/main/resources/services/javax.annotation.processing.Processor

@@ -0,0 +1 @@
+io.druid.annotations.SubclassesMustBePublicAnnotationProcessor

extensions-contrib/distinctcount/src/main/java/io/druid/query/aggregation/distinctcount/EmptyDistinctCountBufferAggregator.java

@@ -77,5 +77,6 @@ public void close()
   @Override
   public void inspectRuntimeShape(RuntimeShapeInspector inspector)
   {
+    // nothing to inspect
   }
 }

extensions-core/datasketches/src/main/java/io/druid/query/aggregation/datasketches/theta/EmptySketchBufferAggregator.java

@@ -73,5 +73,6 @@ public void close()
   @Override
   public void inspectRuntimeShape(RuntimeShapeInspector inspector)
   {
+    // nothing to inspect
   }
 }

processing/pom.xml

@@ -132,6 +132,22 @@
 
     <build>
         <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <annotationProcessorPaths>
+                        <path>
+                            <groupId>io.druid</groupId>
+                            <artifactId>druid-common</artifactId>
+                            <version>${project.parent.version}</version>
+                        </path>
+                    </annotationProcessorPaths>
+                    <annotationProcessors>
+                        <annotationProcessor>io.druid.annotations.SubclassesMustBePublicAnnotationProcessor</annotationProcessor>
+                    </annotationProcessors>
+                </configuration>
+            </plugin>
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-jar-plugin</artifactId>

processing/src/main/java/io/druid/query/aggregation/BufferAggregator.java

@@ -126,6 +126,7 @@ public interface BufferAggregator extends HotLoopCallee
   @Override
   default void inspectRuntimeShape(RuntimeShapeInspector inspector)
   {
+    // nothing to inspect
   }
 
   /**

processing/src/main/java/io/druid/query/aggregation/CountBufferAggregator.java

@@ -68,5 +68,6 @@ public void close()
   @Override
   public void inspectRuntimeShape(RuntimeShapeInspector inspector)
   {
+    // nothing to inspect
   }
 }

processing/src/main/java/io/druid/query/aggregation/DoubleMaxBufferAggregator.java

@@ -25,7 +25,7 @@
 
 /**
  */
-public class DoubleMaxBufferAggregator extends DoubleBufferAggregator
+public class DoubleMaxBufferAggregator extends SimpleDoubleBufferAggregator
 {
 
   DoubleMaxBufferAggregator(FloatColumnSelector selector)
@@ -40,8 +40,18 @@ public void init(ByteBuffer buf, int position)
   }
 
   @Override
-  public void aggregate(ByteBuffer buf, int position)
+  public void putFirst(ByteBuffer buf, int position, double value)
   {
-    buf.putDouble(position, Math.max(buf.getDouble(position), (double) selector.get()));
+    if (!Double.isNaN(value)) {
+      buf.putDouble(position, value);
+    } else {
+      init(buf, position);
+    }
+  }
+
+  @Override
+  public void aggregate(ByteBuffer buf, int position, double value)
+  {
+    buf.putDouble(position, Math.max(buf.getDouble(position), value));
   }
 }

processing/src/main/java/io/druid/query/aggregation/DoubleMinBufferAggregator.java

@@ -25,7 +25,7 @@
 
 /**
  */
-public class DoubleMinBufferAggregator extends DoubleBufferAggregator
+public class DoubleMinBufferAggregator extends SimpleDoubleBufferAggregator
 {
 
   DoubleMinBufferAggregator(FloatColumnSelector selector)
@@ -40,8 +40,18 @@ public void init(ByteBuffer buf, int position)
   }
 
   @Override
-  public void aggregate(ByteBuffer buf, int position)
+  public void putFirst(ByteBuffer buf, int position, double value)
   {
-    buf.putDouble(position, Math.min(buf.getDouble(position), (double) selector.get()));
+    if (!Double.isNaN(value)) {
+      buf.putDouble(position, value);
+    } else {
+      init(buf, position);
+    }
+  }
+
+  @Override
+  public void aggregate(ByteBuffer buf, int position, double value)
+  {
+    buf.putDouble(position, Math.min(buf.getDouble(position), value));
   }
 }

processing/src/main/java/io/druid/query/aggregation/DoubleSumBufferAggregator.java

@@ -25,7 +25,7 @@
 
 /**
  */
-public class DoubleSumBufferAggregator extends DoubleBufferAggregator
+public class DoubleSumBufferAggregator extends SimpleDoubleBufferAggregator
 {
 
   DoubleSumBufferAggregator(FloatColumnSelector selector)
@@ -40,8 +40,14 @@ public void init(ByteBuffer buf, int position)
   }
 
   @Override
-  public void aggregate(ByteBuffer buf, int position)
+  public void putFirst(ByteBuffer buf, int position, double value)
   {
-    buf.putDouble(position, buf.getDouble(position) + (double) selector.get());
+    buf.putDouble(position, value);
+  }
+
+  @Override
+  public void aggregate(ByteBuffer buf, int position, double value)
+  {
+    buf.putDouble(position, buf.getDouble(position) + value);
   }
 }

processing/src/main/java/io/druid/query/aggregation/NoopBufferAggregator.java

@@ -73,5 +73,6 @@ public void close()
   @Override
   public void inspectRuntimeShape(RuntimeShapeInspector inspector)
   {
+    // nothing to inspect
   }
 }

processing/src/main/java/io/druid/query/aggregation/DoubleBufferAggregator.java → processing/src/main/java/io/druid/query/aggregation/SimpleDoubleBufferAggregator.java

@@ -19,34 +19,57 @@
 
 package io.druid.query.aggregation;
 
+import io.druid.query.monomorphicprocessing.CalledFromHotLoop;
 import io.druid.query.monomorphicprocessing.RuntimeShapeInspector;
 import io.druid.segment.FloatColumnSelector;
 
 import java.nio.ByteBuffer;
 
-public abstract class DoubleBufferAggregator implements BufferAggregator
+public abstract class SimpleDoubleBufferAggregator implements BufferAggregator
 {
   protected final FloatColumnSelector selector;
 
-  DoubleBufferAggregator(FloatColumnSelector selector)
+  SimpleDoubleBufferAggregator(FloatColumnSelector selector)
   {
     this.selector = selector;
   }
 
+  public FloatColumnSelector getSelector()
+  {
+    return selector;
+  }
+
+  /**
+   * Faster equivalent to
+   * aggregator.init(buf, position);
+   * aggregator.aggregate(buf, position, value);
+   */
+  @CalledFromHotLoop
+  public abstract void putFirst(ByteBuffer buf, int position, double value);
+
+  @CalledFromHotLoop
+  public abstract void aggregate(ByteBuffer buf, int position, double value);
+
+  @Override
+  public final void aggregate(ByteBuffer buf, int position)
+  {
+    aggregate(buf, position, (double) selector.get());
+  }
+
   @Override
-  public Object get(ByteBuffer buf, int position)
+  public final Object get(ByteBuffer buf, int position)
   {
     return buf.getDouble(position);
   }
 
   @Override
-  public float getFloat(ByteBuffer buf, int position)
+  public final float getFloat(ByteBuffer buf, int position)
   {
     return (float) buf.getDouble(position);
   }
 
   @Override
-  public long getLong(ByteBuffer buf, int position)
+  public final long getLong(ByteBuffer buf, int position)
   {
     return (long) buf.getDouble(position);
   }

processing/src/main/java/io/druid/query/dimension/ForwardingFilteredDimensionSelector.java

@@ -31,7 +31,7 @@
 import io.druid.segment.data.ArrayBasedIndexedInts;
 import io.druid.segment.data.IndexedInts;
 import io.druid.segment.filter.BooleanValueMatcher;
-import it.unimi.dsi.fastutil.ints.Int2IntMap;
+import it.unimi.dsi.fastutil.ints.Int2IntOpenHashMap;
 
 import javax.annotation.Nullable;
 import java.util.BitSet;
@@ -40,14 +40,18 @@ final class ForwardingFilteredDimensionSelector implements DimensionSelector, Id
 {
   private final DimensionSelector selector;
   private final IdLookup baseIdLookup;
-  private final Int2IntMap forwardMapping;
+  private final Int2IntOpenHashMap forwardMapping;
   private final int[] reverseMapping;
 
   /**
    * @param selector must return true from {@link DimensionSelector#nameLookupPossibleInAdvance()}
-   * @param forwardMapping must have {@link Int2IntMap#defaultReturnValue(int)} configured to -1.
+   * @param forwardMapping must have {@link Int2IntOpenHashMap#defaultReturnValue(int)} configured to -1.
    */
-  ForwardingFilteredDimensionSelector(DimensionSelector selector, Int2IntMap forwardMapping, int[] reverseMapping)
+  ForwardingFilteredDimensionSelector(
+      DimensionSelector selector,
+      Int2IntOpenHashMap forwardMapping,
+      int[] reverseMapping
+  )
   {
     this.selector = Preconditions.checkNotNull(selector);
     if (!selector.nameLookupPossibleInAdvance()) {
@@ -106,6 +110,12 @@ public boolean matches()
             // null should match empty rows in multi-value columns
             return nullRow && value == null;
           }
+
+          @Override
+          public void inspectRuntimeShape(RuntimeShapeInspector inspector)
+          {
+            inspector.visit("selector", selector);
+          }
         };
       } else {
         return BooleanValueMatcher.of(false);
@@ -141,6 +151,12 @@ public boolean matches()
         // null should match empty rows in multi-value columns
         return nullRow && matchNull;
       }
+
+      @Override
+      public void inspectRuntimeShape(RuntimeShapeInspector inspector)
+      {
+        inspector.visit("selector", selector);
+      }
     };
   }
 
@@ -179,6 +195,5 @@ public int lookupId(String name)
   public void inspectRuntimeShape(RuntimeShapeInspector inspector)
   {
     inspector.visit("selector", selector);
-    inspector.visit("forwardMapping", forwardMapping);
   }
 }

processing/src/main/java/io/druid/query/dimension/ListFilteredDimensionSpec.java

@@ -28,7 +28,6 @@
 import io.druid.query.filter.DimFilterUtils;
 import io.druid.segment.DimensionSelector;
 import io.druid.segment.IdLookup;
-import it.unimi.dsi.fastutil.ints.Int2IntMap;
 import it.unimi.dsi.fastutil.ints.Int2IntOpenHashMap;
 
 import javax.annotation.Nullable;
@@ -93,7 +92,7 @@ private DimensionSelector filterWhiteList(DimensionSelector selector)
     }
     final int maxPossibleFilteredCardinality = values.size();
     int count = 0;
-    final Int2IntMap forwardMapping = new Int2IntOpenHashMap(maxPossibleFilteredCardinality);
+    final Int2IntOpenHashMap forwardMapping = new Int2IntOpenHashMap(maxPossibleFilteredCardinality);
     forwardMapping.defaultReturnValue(-1);
     final int[] reverseMapping = new int[maxPossibleFilteredCardinality];
     IdLookup idLookup = selector.idLookup();
@@ -134,7 +133,7 @@ public boolean apply(@Nullable String input)
     }
     final int maxPossibleFilteredCardinality = selectorCardinality;
     int count = 0;
-    final Int2IntMap forwardMapping = new Int2IntOpenHashMap(maxPossibleFilteredCardinality);
+    final Int2IntOpenHashMap forwardMapping = new Int2IntOpenHashMap(maxPossibleFilteredCardinality);
     forwardMapping.defaultReturnValue(-1);
     final int[] reverseMapping = new int[maxPossibleFilteredCardinality];
     for (int i = 0; i < selectorCardinality; i++) {