refactor numeric primitive aggregators in sql compatible mode

extensions-contrib/materialized-view-maintenance/src/test/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisorSpecTest.java

@@ -23,6 +23,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.jsontype.NamedType;
 import com.google.common.collect.Lists;
+import org.apache.druid.common.config.NullHandling;
 import org.apache.druid.data.input.impl.DimensionsSpec;
 import org.apache.druid.data.input.impl.StringDimensionSchema;
 import org.apache.druid.indexer.HadoopTuningConfig;
@@ -56,6 +57,10 @@
 
 public class MaterializedViewSupervisorSpecTest
 {
+  static {
+    NullHandling.initializeForTests();
+  }
+
   @Rule
   public ExpectedException expectedException = ExpectedException.none();
 

extensions-contrib/materialized-view-maintenance/src/test/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisorTest.java

@@ -25,6 +25,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Sets;
+import org.apache.druid.common.config.NullHandling;
 import org.apache.druid.data.input.impl.DimensionsSpec;
 import org.apache.druid.data.input.impl.StringDimensionSchema;
 import org.apache.druid.indexer.HadoopIOConfig;
@@ -70,6 +71,10 @@
 
 public class MaterializedViewSupervisorTest
 {
+  static {
+    NullHandling.initializeForTests();
+  }
+
   @Rule
   public final TestDerbyConnector.DerbyConnectorRule derbyConnectorRule = new TestDerbyConnector.DerbyConnectorRule();
 

extensions-contrib/materialized-view-selection/src/test/java/org/apache/druid/query/materializedview/MaterializedViewQueryQueryToolChestTest.java

@@ -23,6 +23,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Function;
 import com.google.common.collect.ImmutableMap;
+import org.apache.druid.common.config.NullHandling;
 import org.apache.druid.data.input.MapBasedRow;
 import org.apache.druid.jackson.DefaultObjectMapper;
 import org.apache.druid.java.util.common.DateTimes;
@@ -57,6 +58,10 @@ public class MaterializedViewQueryQueryToolChestTest
 {
   private static final ObjectMapper JSON_MAPPER = new DefaultObjectMapper();
 
+  static {
+    NullHandling.initializeForTests();
+  }
+
   @Test
   public void testMakePostComputeManipulatorFn()
   {

extensions-contrib/materialized-view-selection/src/test/java/org/apache/druid/query/materializedview/MaterializedViewQueryTest.java

@@ -24,6 +24,7 @@
 import com.fasterxml.jackson.databind.jsontype.NamedType;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
+import org.apache.druid.common.config.NullHandling;
 import org.apache.druid.math.expr.ExprMacroTable;
 import org.apache.druid.query.Query;
 import org.apache.druid.query.QueryRunnerTestHelper;
@@ -44,6 +45,11 @@
 public class MaterializedViewQueryTest
 {
   private static final ObjectMapper JSON_MAPPER = TestHelper.makeJsonMapper();
+
+  static {
+    NullHandling.initializeForTests();
+  }
+
   private DataSourceOptimizer optimizer;
 
   @Before

extensions-core/druid-bloom-filter/src/test/java/org/apache/druid/query/filter/BloomDimFilterTest.java

@@ -59,39 +59,45 @@ public class BloomDimFilterTest extends BaseFilterTest
 {
   private static final String TIMESTAMP_COLUMN = "timestamp";
 
-  private static final InputRowParser<Map<String, Object>> PARSER = new MapInputRowParser(
-      new TimeAndDimsParseSpec(
-          new TimestampSpec(TIMESTAMP_COLUMN, "iso", DateTimes.of("2000")),
-          new DimensionsSpec(
-              DimensionsSpec.getDefaultSchemas(ImmutableList.of("dim0", "dim1", "dim2", "dim3", "dim6")),
-              null,
-              null
-          )
-      )
-  );
-
-  private static final List<InputRow> ROWS = ImmutableList.of(
-      PARSER.parseBatch(ImmutableMap.of(
-          "dim0",
-          "0",
-          "dim1",
-          "",
-          "dim2",
-          ImmutableList.of("a", "b"),
-          "dim6",
-          "2017-07-25"
-      )).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "1", "dim1", "10", "dim2", ImmutableList.of(), "dim6", "2017-07-25"))
-            .get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "2", "dim1", "2", "dim2", ImmutableList.of(""), "dim6", "2017-05-25"))
-            .get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "3", "dim1", "1", "dim2", ImmutableList.of("a"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "4", "dim1", "def", "dim2", ImmutableList.of("c"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "5", "dim1", "abc")).get(0)
-  );
+  private static final InputRowParser<Map<String, Object>> PARSER;
+
+  private static final List<InputRow> ROWS;
 
   private static DefaultObjectMapper mapper = new DefaultObjectMapper();
 
+  static {
+    NullHandling.initializeForTests();
+    PARSER = new MapInputRowParser(
+        new TimeAndDimsParseSpec(
+            new TimestampSpec(TIMESTAMP_COLUMN, "iso", DateTimes.of("2000")),
+            new DimensionsSpec(
+                DimensionsSpec.getDefaultSchemas(ImmutableList.of("dim0", "dim1", "dim2", "dim3", "dim6")),
+                null,
+                null
+            )
+        )
+    );
+    ROWS = ImmutableList.of(
+        PARSER.parseBatch(ImmutableMap.of(
+            "dim0",
+            "0",
+            "dim1",
+            "",
+            "dim2",
+            ImmutableList.of("a", "b"),
+            "dim6",
+            "2017-07-25"
+        )).get(0),
+        PARSER.parseBatch(ImmutableMap.of("dim0", "1", "dim1", "10", "dim2", ImmutableList.of(), "dim6", "2017-07-25"))
+              .get(0),
+        PARSER.parseBatch(ImmutableMap.of("dim0", "2", "dim1", "2", "dim2", ImmutableList.of(""), "dim6", "2017-05-25"))
+              .get(0),
+        PARSER.parseBatch(ImmutableMap.of("dim0", "3", "dim1", "1", "dim2", ImmutableList.of("a"))).get(0),
+        PARSER.parseBatch(ImmutableMap.of("dim0", "4", "dim1", "def", "dim2", ImmutableList.of("c"))).get(0),
+        PARSER.parseBatch(ImmutableMap.of("dim0", "5", "dim1", "abc")).get(0)
+    );
+  }
+
   public BloomDimFilterTest(
       String testName,
       IndexBuilder indexBuilder,

extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java

@@ -40,6 +40,7 @@
 import org.apache.druid.client.cache.CachePopulatorStats;
 import org.apache.druid.client.cache.MapCache;
 import org.apache.druid.client.indexing.NoopIndexingServiceClient;
+import org.apache.druid.common.config.NullHandling;
 import org.apache.druid.data.input.InputEntity;
 import org.apache.druid.data.input.InputEntityReader;
 import org.apache.druid.data.input.InputFormat;
@@ -211,6 +212,7 @@ public byte[] value()
   );
 
   static {
+    NullHandling.initializeForTests();
     Stream.concat(
         new KafkaIndexTaskModule().getJacksonModules().stream(),
         Stream.of(TEST_MODULE)

extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisSamplerSpecTest.java

@@ -62,30 +62,31 @@ public class KinesisSamplerSpecTest extends EasyMockSupport
 {
   private static final String STREAM = "sampling";
   private static final String SHARD_ID = "1";
-  private static final DataSchema DATA_SCHEMA = new DataSchema(
-      "test_ds",
-      new TimestampSpec("timestamp", "iso", null),
-      new DimensionsSpec(
-          Arrays.asList(
-              new StringDimensionSchema("dim1"),
-              new StringDimensionSchema("dim1t"),
-              new StringDimensionSchema("dim2"),
-              new LongDimensionSchema("dimLong"),
-              new FloatDimensionSchema("dimFloat")
-          ),
-          null,
-          null
-      ),
-      new AggregatorFactory[]{
-          new DoubleSumAggregatorFactory("met1sum", "met1"),
-          new CountAggregatorFactory("rows")
-      },
-      new UniformGranularitySpec(Granularities.DAY, Granularities.NONE, null),
-      null
-  );
+  private static final DataSchema DATA_SCHEMA;
 
   static {
     NullHandling.initializeForTests();
+    DATA_SCHEMA = new DataSchema(
+        "test_ds",
+        new TimestampSpec("timestamp", "iso", null),
+        new DimensionsSpec(
+            Arrays.asList(
+                new StringDimensionSchema("dim1"),
+                new StringDimensionSchema("dim1t"),
+                new StringDimensionSchema("dim2"),
+                new LongDimensionSchema("dimLong"),
+                new FloatDimensionSchema("dimFloat")
+            ),
+            null,
+            null
+        ),
+        new AggregatorFactory[]{
+            new DoubleSumAggregatorFactory("met1sum", "met1"),
+            new CountAggregatorFactory("rows")
+        },
+        new UniformGranularitySpec(Granularities.DAY, Granularities.NONE, null),
+        null
+    );
   }
 
   private final KinesisRecordSupplier recordSupplier = mock(KinesisRecordSupplier.class);

processing/src/main/java/org/apache/druid/query/aggregation/NonnullNumericAggregator.java

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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 org.apache.druid.query.aggregation;
+
+import org.apache.druid.segment.BaseNullableColumnValueSelector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+
+/**
+ * Null-aware numeric {@link Aggregator} whose input is not nullable, but which should be null valued if no
+ * values are aggregated at all.
+ *
+ * The result of this aggregator will be null only if no values are aggregated at all, otherwise the result will
+ * be the aggregated value of the delegate aggregator. This class is only used when SQL compatible null handling
+ * is enabled.
+ *
+ * @see NullableNumericAggregatorFactory#factorize(ColumnSelectorFactory)
+ * @see NonnullNumericBufferAggregator for the non-vectorized buffer version.
+ * @see NonnullNumericVectorAggregator the vectorized version.
+ */
+public final class NonnullNumericAggregator extends NullAwareNumericAggregator
+{
+  public NonnullNumericAggregator(Aggregator delegate, BaseNullableColumnValueSelector selector)
+  {
+    super(delegate, selector);
+  }
+
+  @Override
+  public void aggregate()
+  {
+    doAggregate();
+  }
+}

processing/src/main/java/org/apache/druid/query/aggregation/NonnullNumericBufferAggregator.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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 org.apache.druid.query.aggregation;
+
+import org.apache.druid.segment.BaseNullableColumnValueSelector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Null-aware numeric {@link BufferAggregator} whose input is not nullable, but which should be null valued if no
+ * values are aggregated at all.
+ *
+ * The result of this aggregator will only be null if no values are aggregated at all, otherwise the result will
+ * be the aggregated value of the delegate aggregator. This class is only used when SQL compatible null handling
+ * is enabled.
+ *
+ * @see NullableNumericAggregatorFactory#factorizeBuffered(ColumnSelectorFactory)
+ * @see NullableNumericAggregator for the non-vectorized heap version.
+ * @see NullableNumericVectorAggregator the vectorized version.
+ */
+public final class NonnullNumericBufferAggregator extends NullAwareNumericBufferAggregator
+{
+  public NonnullNumericBufferAggregator(
+      BufferAggregator delegate,
+      BaseNullableColumnValueSelector nullSelector
+  )
+  {
+    super(delegate, nullSelector);
+  }
+
+  @Override
+  public void aggregate(ByteBuffer buf, int position)
+  {
+    doAggregate(buf, position);
+  }
+}

processing/src/main/java/org/apache/druid/query/aggregation/NonnullNumericVectorAggregator.java

@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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 org.apache.druid.query.aggregation;
+
+import org.apache.druid.segment.vector.VectorColumnSelectorFactory;
+import org.apache.druid.segment.vector.VectorValueSelector;
+
+import javax.annotation.Nullable;
+import java.nio.ByteBuffer;
+
+/**
+ * A wrapper around a non-null-aware {@link VectorAggregator} whose input is not nullable, but which
+ * should be null valued if no values are aggregated at all.
+ *
+ * The result of this aggregator will be null only if no values are aggregated at all, otherwise the result
+ * would be the aggregated value of the delegate aggregator. This class is only used when SQL compatible null
+ * handling is enabled.
+ *
+ * @see NullableNumericAggregatorFactory#factorizeVector(VectorColumnSelectorFactory)
+ * @see NonnullNumericAggregator for the non-vectorized heap version.
+ * @see NonnullNumericBufferAggregator for the non-vectorized version.
+ */
+public final class NonnullNumericVectorAggregator extends NullAwareNumericVectorAggregator
+{
+  NonnullNumericVectorAggregator(
+      VectorAggregator delegate,
+      VectorValueSelector selector
+  )
+  {
+    super(delegate, selector);
+  }
+
+  @Override
+  public void aggregate(ByteBuffer buf, int position, int startRow, int endRow)
+  {
+    doAggregate(buf, position, startRow, endRow);
+  }
+
+  @Override
+  public void aggregate(ByteBuffer buf, int numRows, int[] positions, @Nullable int[] rows, int positionOffset)
+  {
+    doAggregate(buf, numRows, positions, rows, positionOffset);
+  }
+}