Cache value selectors in RowBasedColumnSelectorFactory.

There was already caching for dimension selectors. This patch adds caching
for value (object and number) selectors. It's helpful when the same field is
read multiple times during processing of a single row (for example, by being
an input to both MIN and MAX aggregations).

processing/src/main/java/org/apache/druid/common/config/NullHandling.java

@@ -182,10 +182,11 @@ public static <T> T defaultValueForClass(final Class<T> clazz)
    * May be null or non-null based on the current SQL-compatible null handling mode.
    */
   @Nullable
-  @SuppressWarnings("unchecked")
   public static Object defaultValueForType(ValueType type)
   {
-    if (type == ValueType.FLOAT) {
+    if (sqlCompatible()) {
+      return null;
+    } else if (type == ValueType.FLOAT) {
       return defaultFloatValue();
     } else if (type == ValueType.DOUBLE) {
       return defaultDoubleValue();

processing/src/main/java/org/apache/druid/data/input/ListBasedInputRow.java

@@ -0,0 +1,28 @@
+/*
+ * 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.data.input;
+
+import java.util.List;
+
+public class ListBasedRow implements InputRow
+{
+  private final List<String> dimensions;
+
+  }

processing/src/main/java/org/apache/druid/data/input/Rows.java

@@ -25,6 +25,7 @@
 import org.apache.druid.common.config.NullHandling;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.parsers.ParseException;
+import org.apache.druid.segment.column.ValueType;
 
 import javax.annotation.Nullable;
 import java.util.Arrays;
@@ -90,6 +91,7 @@ public static List<String> objectToStrings(final Object inputValue)
    *
    * @param name                 field name of the object being converted (may be used for exception messages)
    * @param inputValue           the actual object being converted
+   * @param expectedType         expected numeric type, or null if it should be automatically detected
    * @param throwParseExceptions whether this method should throw a {@link ParseException} or use a default/null value
    *                             when {@param inputValue} is not numeric
    *
@@ -98,45 +100,73 @@ public static List<String> objectToStrings(final Object inputValue)
    * @throws ParseException if the input cannot be converted to a number and {@code throwParseExceptions} is true
    */
   @Nullable
-  public static <T extends Number> Number objectToNumber(
+  public static Number objectToNumber(
       final String name,
       final Object inputValue,
+      @Nullable final ValueType expectedType,
       final boolean throwParseExceptions
   )
   {
     if (inputValue == null) {
-      return NullHandling.defaultLongValue();
+      return (Number) NullHandling.defaultValueForType(expectedType != null ? expectedType : ValueType.LONG);
     } else if (inputValue instanceof Number) {
       return (Number) inputValue;
     } else if (inputValue instanceof String) {
       try {
         String metricValueString = StringUtils.removeChar(((String) inputValue).trim(), ',');
         // Longs.tryParse() doesn't support leading '+', so we need to trim it ourselves
         metricValueString = trimLeadingPlusOfLongString(metricValueString);
-        Long v = Longs.tryParse(metricValueString);
-        // Do NOT use ternary operator here, because it makes Java to convert Long to Double
-        if (v != null) {
-          return v;
+
+        Number v = null;
+
+        // Try parsing as Long first, since it's significantly faster than Double parsing, and also there are various
+        // integer numbers that can be represented as Long but cannot be represented as Double.
+        if (expectedType == null || expectedType == ValueType.LONG) {
+          v = Longs.tryParse(metricValueString);
+        }
+
+        if ((expectedType == null && v == null) || expectedType == ValueType.DOUBLE) {
+          v = Double.valueOf(metricValueString);
+        }
+
+        if (v == null) {
+          if (throwParseExceptions) {
+            throw new ParseException(String.valueOf(inputValue), "Unable to parse value[%s] for field[%s] as type[%s]", inputValue.getClass(), name, expectedType);
+          } else {
+            return (Number) NullHandling.defaultValueForType(expectedType);
+          }
         } else {
-          return Double.valueOf(metricValueString);
+          return v;
         }
       }
       catch (Exception e) {
         if (throwParseExceptions) {
           throw new ParseException(String.valueOf(inputValue), e, "Unable to parse value[%s] for field[%s]", inputValue, name);
         } else {
-          return NullHandling.defaultLongValue();
+          return (Number) NullHandling.defaultValueForType(expectedType != null ? expectedType : ValueType.LONG);
         }
       }
     } else {
       if (throwParseExceptions) {
         throw new ParseException(String.valueOf(inputValue), "Unknown type[%s] for field[%s]", inputValue.getClass(), name);
       } else {
-        return NullHandling.defaultLongValue();
+        return (Number) NullHandling.defaultValueForType(expectedType != null ? expectedType : ValueType.LONG);
       }
     }
   }
 
+  /**
+   * Shorthand for {@link #objectToNumber(String, Object, ValueType, boolean)} with null expectedType.
+   */
+  public static Number objectToNumber(
+      final String name,
+      final Object inputValue,
+      final boolean throwParseExceptions
+  )
+  {
+    return objectToNumber(name, inputValue, null, throwParseExceptions);
+  }
+
   private static String trimLeadingPlusOfLongString(String metricValueString)
   {
     if (metricValueString.length() > 1 && metricValueString.charAt(0) == '+') {

processing/src/main/java/org/apache/druid/segment/RowBasedColumnSelectorFactory.java

@@ -32,6 +32,7 @@
 import org.apache.druid.segment.column.ColumnCapabilitiesImpl;
 import org.apache.druid.segment.column.ColumnHolder;
 import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.ValueType;
 import org.apache.druid.segment.data.IndexedInts;
 import org.apache.druid.segment.data.RangeIndexedInts;
 import org.apache.druid.segment.nested.StructuredData;
@@ -445,44 +446,55 @@ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
       return new TimeLongColumnSelector();
     } else {
       final Function<T, Object> columnFunction = adapter.columnFunction(columnName);
+      final ColumnCapabilities capabilities = columnInspector.getColumnCapabilities(columnName);
+      final ValueType expectedType = capabilities == null ? null : capabilities.getType();
 
       return new ColumnValueSelector<Object>()
       {
+        private long currentValueId = RowIdSupplier.INIT;
+        private long currentValueAsNumberId = RowIdSupplier.INIT;
+        @Nullable
+        private Object currentValue;
+        @Nullable
+        private Number currentValueAsNumber;
+
         @Override
         public boolean isNull()
         {
-          return !NullHandling.replaceWithDefault() && getCurrentValueAsNumber() == null;
+          updateCurrentValueAsNumber();
+          return !NullHandling.replaceWithDefault() && currentValueAsNumber == null;
         }
 
         @Override
         public double getDouble()
         {
-          final Number n = getCurrentValueAsNumber();
-          assert NullHandling.replaceWithDefault() || n != null;
-          return n != null ? n.doubleValue() : 0d;
+          updateCurrentValueAsNumber();
+          assert NullHandling.replaceWithDefault() || currentValueAsNumber != null;
+          return currentValueAsNumber != null ? currentValueAsNumber.doubleValue() : 0d;
         }
 
         @Override
         public float getFloat()
         {
-          final Number n = getCurrentValueAsNumber();
-          assert NullHandling.replaceWithDefault() || n != null;
-          return n != null ? n.floatValue() : 0f;
+          updateCurrentValueAsNumber();
+          assert NullHandling.replaceWithDefault() || currentValueAsNumber != null;
+          return currentValueAsNumber != null ? currentValueAsNumber.floatValue() : 0f;
         }
 
         @Override
         public long getLong()
         {
-          final Number n = getCurrentValueAsNumber();
-          assert NullHandling.replaceWithDefault() || n != null;
-          return n != null ? n.longValue() : 0L;
+          updateCurrentValueAsNumber();
+          assert NullHandling.replaceWithDefault() || currentValueAsNumber != null;
+          return currentValueAsNumber != null ? currentValueAsNumber.longValue() : 0L;
         }
 
         @Nullable
         @Override
         public Object getObject()
         {
-          return getCurrentValue();
+          updateCurrentValue();
+          return currentValue;
         }
 
         @Override
@@ -497,24 +509,42 @@ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
           inspector.visit("row", rowSupplier);
         }
 
-        @Nullable
-        private Object getCurrentValue()
+        private void updateCurrentValue()
         {
-          return columnFunction.apply(rowSupplier.get());
+          if (rowIdSupplier == null || rowIdSupplier.getRowId() != currentValueId) {
+            try {
+              currentValue = columnFunction.apply(rowSupplier.get());
+            }
+            catch (Throwable e) {
+              currentValueId = RowIdSupplier.INIT;
+              throw e;
+            }
+
+            if (rowIdSupplier != null) {
+              currentValueId = rowIdSupplier.getRowId();
+            }
+          }
         }
 
-        @Nullable
-        private Number getCurrentValueAsNumber()
+        private void updateCurrentValueAsNumber()
         {
-          final Object currentValue = getCurrentValue();
-          if (currentValue instanceof StructuredData) {
-            return Rows.objectToNumber(columnName, ((StructuredData) currentValue).getValue(), throwParseExceptions);
+          updateCurrentValue();
+
+          if (rowIdSupplier == null || rowIdSupplier.getRowId() != currentValueAsNumberId) {
+            try {
+              final Object valueToUse =
+                  currentValue instanceof StructuredData ? ((StructuredData) currentValue).getValue() : currentValue;
+              currentValueAsNumber = Rows.objectToNumber(columnName, valueToUse, expectedType, throwParseExceptions);
+            }
+            catch (Throwable e) {
+              currentValueAsNumberId = RowIdSupplier.INIT;
+              throw e;
+            }
+
+            if (rowIdSupplier != null) {
+              currentValueAsNumberId = rowIdSupplier.getRowId();
+            }
           }
-          return Rows.objectToNumber(
-              columnName,
-              currentValue,
-              throwParseExceptions
-          );
         }
       };
     }