Add nan definition field to DoubleType and RealType

Add a boolean field to DoubleType and RealType to determine whether to
use the new nan definition.  If useNewNanDefinition is set to true in
the configuration property, then only doubles/reals with that property
set to true will be created, and if it is false, then only doubles/reals
with that property set to false will be created.  This will be used in
later commits to make decisions about how to handle nans.

Because DoubleType and RealType are now parametrized types, it is no
longer correct to use type == DOUBLE for type checking, as it is no
longer a singleton instance.  All code that was using type == DOUBLE or
type == REAL has been updated to use .equals() comparison

presto-clickhouse/src/main/java/com/facebook/presto/plugin/clickhouse/ClickHouseClient.java

@@ -854,10 +854,10 @@ private String toWriteMapping(Type type)
         if (type == BIGINT) {
             return "Int64";
         }
-        if (type == REAL) {
+        if (type.equals(REAL)) {
             return "Float32";
         }
-        if (type == DOUBLE) {
+        if (type.equals(DOUBLE)) {
             return "Float64";
         }
         if (type instanceof DecimalType) {

presto-clickhouse/src/main/java/com/facebook/presto/plugin/clickhouse/ClickHouseColumnHandle.java

@@ -195,10 +195,10 @@ public static ClickHouseTypeHandle toMappingDefaultJdbcHandle(Type type)
         if (type == BIGINT) {
             return new ClickHouseTypeHandle(Types.BIGINT, Optional.of("bigint"), 8, 0, Optional.empty(), Optional.empty());
         }
-        if (type == REAL) {
+        if (type.equals(REAL)) {
             return new ClickHouseTypeHandle(Types.REAL, Optional.of("real"), 16, 4, Optional.empty(), Optional.empty());
         }
-        if (type == DOUBLE) {
+        if (type.equals(DOUBLE)) {
             return new ClickHouseTypeHandle(Types.DOUBLE, Optional.of("double precision"), 32, 4, Optional.empty(), Optional.empty());
         }
         if (type instanceof CharType || type instanceof VarcharType) {

presto-common/src/main/java/com/facebook/presto/common/predicate/TupleDomainFilterUtils.java

@@ -148,7 +148,7 @@ public static TupleDomainFilter toFilter(Domain domain)
 
     /**
      * Returns true is ranges represent != or NOT IN filter for double, float or string column.
-     *
+     * <p>
      * The logic is to return true if ranges are next to each other, but don't include the touch value.
      */
     private static boolean isNotIn(List<Range> ranges)
@@ -161,7 +161,7 @@ private static boolean isNotIn(List<Range> ranges)
         Marker previousHigh = firstRange.getHigh();
 
         Type type = previousHigh.getType();
-        if (type != DOUBLE && type != REAL && !isVarcharType(type) && !(type instanceof CharType)) {
+        if (!type.equals(DOUBLE) && !type.equals(REAL) && !isVarcharType(type) && !(type instanceof CharType)) {
             return false;
         }
 
@@ -219,10 +219,10 @@ private static TupleDomainFilter createRangeFilter(Type type, Range range, boole
             checkArgument(range.isSingleValue(), "Unexpected range of boolean values");
             return BooleanValue.of(((Boolean) range.getSingleValue()).booleanValue(), nullAllowed);
         }
-        if (type == DOUBLE) {
+        if (type.equals(DOUBLE)) {
             return doubleRangeToFilter(range, nullAllowed);
         }
-        if (type == REAL) {
+        if (type.equals(REAL)) {
             return floatRangeToFilter(range, nullAllowed);
         }
         if (type instanceof DecimalType) {

presto-common/src/main/java/com/facebook/presto/common/type/DoubleType.java

@@ -29,11 +29,15 @@ public final class DoubleType
         extends AbstractPrimitiveType
         implements FixedWidthType
 {
-    public static final DoubleType DOUBLE = new DoubleType();
+    public static final DoubleType DOUBLE = new DoubleType(true);
+    public static final DoubleType OLD_NAN_DOUBLE = new DoubleType(false);
 
-    private DoubleType()
+    private final boolean useNewNanDefintion;
+
+    private DoubleType(boolean useNewNanDefintion)
     {
         super(parseTypeSignature(StandardTypes.DOUBLE), double.class);
+        this.useNewNanDefintion = useNewNanDefintion;
     }
 
     @Override
@@ -149,7 +153,7 @@ public final BlockBuilder createFixedSizeBlockBuilder(int positionCount)
     @SuppressWarnings("EqualsWhichDoesntCheckParameterClass")
     public boolean equals(Object other)
     {
-        return other == DOUBLE;
+        return other instanceof DoubleType;
     }
 
     @Override

presto-common/src/main/java/com/facebook/presto/common/type/RealType.java

@@ -27,11 +27,14 @@
 public final class RealType
         extends AbstractIntType
 {
-    public static final RealType REAL = new RealType();
+    public static final RealType REAL = new RealType(true);
+    public static final RealType OLD_NAN_REAL = new RealType(false);
+    private final boolean useNewNanDefinition;
 
-    private RealType()
+    private RealType(boolean useNewNanDefinition)
     {
         super(parseTypeSignature(StandardTypes.REAL));
+        this.useNewNanDefinition = useNewNanDefinition;
     }
 
     @Override
@@ -86,7 +89,7 @@ public void writeLong(BlockBuilder blockBuilder, long value)
     @Override
     public boolean equals(Object other)
     {
-        return other == REAL;
+        return other instanceof RealType;
     }
 
     @Override

presto-common/src/main/java/com/facebook/presto/common/type/TypeUtils.java

@@ -175,10 +175,10 @@ public static boolean isFloatingPointNaN(Type type, Object value)
         requireNonNull(type, "type is null");
         requireNonNull(value, "value is null");
 
-        if (type == DOUBLE) {
+        if (type.equals(DOUBLE)) {
             return Double.isNaN((double) value);
         }
-        if (type == REAL) {
+        if (type.equals(REAL)) {
             return Float.isNaN(intBitsToFloat(toIntExact((long) value)));
         }
         return false;

presto-common/src/test/java/com/facebook/presto/common/block/TestBlockRetainedSizeBreakdown.java

@@ -176,7 +176,7 @@ private static Object castIntegerToObject(int value, Type type)
         if (type == VARCHAR) {
             return String.valueOf(value);
         }
-        if (type == DOUBLE) {
+        if (type.equals(DOUBLE)) {
             return (double) value;
         }
         throw new UnsupportedOperationException();

presto-druid/src/main/java/com/facebook/presto/druid/column/ColumnReader.java

@@ -35,13 +35,13 @@ static ColumnReader createColumnReader(Type type, ColumnValueSelector valueSelec
         if (type == VARCHAR) {
             return new StringColumnReader(valueSelector);
         }
-        if (type == DOUBLE) {
+        if (type.equals(DOUBLE)) {
             return new DoubleColumnReader(valueSelector);
         }
         if (type == BIGINT) {
             return new LongColumnReader(valueSelector);
         }
-        if (type == REAL) {
+        if (type.equals(REAL)) {
             return new FloatColumnReader(valueSelector);
         }
         if (type == TIMESTAMP) {

presto-druid/src/main/java/com/facebook/presto/druid/column/DoubleColumnReader.java

@@ -35,7 +35,7 @@ public DoubleColumnReader(ColumnValueSelector valueSelector)
     @Override
     public Block readBlock(Type type, int batchSize)
     {
-        checkArgument(type == DOUBLE);
+        checkArgument(type.equals(DOUBLE));
         BlockBuilder builder = type.createBlockBuilder(null, batchSize);
         for (int i = 0; i < batchSize; i++) {
             type.writeDouble(builder, valueSelector.getDouble());

presto-druid/src/main/java/com/facebook/presto/druid/column/FloatColumnReader.java

@@ -36,7 +36,7 @@ public FloatColumnReader(ColumnValueSelector valueSelector)
     @Override
     public Block readBlock(Type type, int batchSize)
     {
-        checkArgument(type == REAL);
+        checkArgument(type.equals(REAL));
         BlockBuilder builder = type.createBlockBuilder(null, batchSize);
         for (int i = 0; i < batchSize; i++) {
             type.writeLong(builder, floatToRawIntBits(valueSelector.getFloat()));

presto-elasticsearch/src/test/java/com/facebook/presto/elasticsearch/ElasticsearchLoader.java

@@ -138,7 +138,7 @@ private Object convertValue(Object value, Type type)
             if (type == INTEGER) {
                 return ((Number) value).intValue();
             }
-            if (type == DOUBLE) {
+            if (type.equals(DOUBLE)) {
                 return ((Number) value).doubleValue();
             }
             throw new IllegalArgumentException("Unhandled type: " + type);

presto-hive-function-namespace/src/test/java/com/facebook/presto/hive/functions/AbstractTestHiveFunctions.java

@@ -108,7 +108,7 @@ public void assertQuery(@Language("SQL") String sql, Column... cols)
             for (int j = 0; j < numColumns; j++) {
                 Object actual = rows.get(i).getField(j);
                 Object expected = cols[j].values[i];
-                if (cols[j].type == DOUBLE) {
+                if (cols[j].type.equals(DOUBLE)) {
                     assertEquals(((Number) actual).doubleValue(), ((double) expected), 0.000001);
                 }
                 else {

presto-hive-function-namespace/src/test/java/com/facebook/presto/hive/functions/TestHiveAggregationFunctions.java

@@ -128,7 +128,7 @@ public void check(@Language("SQL") String query, Column... expectedColumns)
             for (int j = 0; j < numColumns; j++) {
                 Object actual = rows.get(i).getField(j);
                 Object expected = expectedColumns[j].values[i];
-                if (expectedColumns[j].type == DOUBLE) {
+                if (expectedColumns[j].type.equals(DOUBLE)) {
                     assertEquals(((Number) actual).doubleValue(), ((double) expected), 0.000001);
                 }
                 else {

presto-hive/src/main/java/com/facebook/presto/hive/FilteringPageSource.java

@@ -246,11 +246,11 @@ private static boolean testNonNullPosition(Block block, int position, Type type,
             return filter.testBoolean(type.getBoolean(block, position));
         }
 
-        if (type == DOUBLE) {
+        if (type.equals(DOUBLE)) {
             return filter.testDouble(longBitsToDouble(block.getLong(position)));
         }
 
-        if (type == REAL) {
+        if (type.equals(REAL)) {
             return filter.testFloat(intBitsToFloat(block.getInt(position)));
         }
 

presto-kafka/src/main/java/com/facebook/presto/kafka/encoder/AbstractRowEncoder.java

@@ -91,10 +91,10 @@ else if (type == SMALLINT) {
         else if (type == TINYINT) {
             appendByte(SignedBytes.checkedCast(type.getLong(block, position)));
         }
-        else if (type == DOUBLE) {
+        else if (type.equals(DOUBLE)) {
             appendDouble(type.getDouble(block, position));
         }
-        else if (type == REAL) {
+        else if (type.equals(REAL)) {
             appendFloat(intBitsToFloat(toIntExact(type.getLong(block, position))));
         }
         else if (isVarcharType(type)) {

presto-kafka/src/main/java/com/facebook/presto/kafka/encoder/raw/RawRowEncoder.java

@@ -196,7 +196,7 @@ else if (columnType == TINYINT) {
             else if (columnType == BOOLEAN) {
                 checkFieldTypeOneOf(fieldType, columnName, columnType, FieldType.BYTE, FieldType.SHORT, FieldType.INT, FieldType.LONG);
             }
-            else if (columnType == DOUBLE) {
+            else if (columnType.equals(DOUBLE)) {
                 checkFieldTypeOneOf(fieldType, columnName, columnType, FieldType.DOUBLE, FieldType.FLOAT);
             }
             else if (isVarcharType(columnType)) {

presto-kudu/src/main/java/com/facebook/presto/kudu/TypeHelper.java

@@ -65,10 +65,10 @@ else if (type == SmallintType.SMALLINT) {
         else if (type == TinyintType.TINYINT) {
             return org.apache.kudu.Type.INT8;
         }
-        else if (type == RealType.REAL) {
+        else if (type.equals(RealType.REAL)) {
             return org.apache.kudu.Type.FLOAT;
         }
-        else if (type == DoubleType.DOUBLE) {
+        else if (type.equals(DoubleType.DOUBLE)) {
             return org.apache.kudu.Type.DOUBLE;
         }
         else if (type == BooleanType.BOOLEAN) {
@@ -146,10 +146,10 @@ else if (type == SmallintType.SMALLINT) {
         else if (type == TinyintType.TINYINT) {
             return ((Long) nativeValue).byteValue();
         }
-        else if (type == DoubleType.DOUBLE) {
+        else if (type.equals(DoubleType.DOUBLE)) {
             return nativeValue;
         }
-        else if (type == RealType.REAL) {
+        else if (type.equals(RealType.REAL)) {
             // conversion can result in precision lost
             return intBitsToFloat(((Long) nativeValue).intValue());
         }
@@ -191,10 +191,10 @@ else if (type == SmallintType.SMALLINT) {
             else if (type == TinyintType.TINYINT) {
                 return row.getByte(field);
             }
-            else if (type == DoubleType.DOUBLE) {
+            else if (type.equals(DoubleType.DOUBLE)) {
                 return row.getDouble(field);
             }
-            else if (type == RealType.REAL) {
+            else if (type.equals(RealType.REAL)) {
                 return row.getFloat(field);
             }
             else if (type == BooleanType.BOOLEAN) {
@@ -229,7 +229,7 @@ else if (type == SmallintType.SMALLINT) {
         else if (type == TinyintType.TINYINT) {
             return row.getByte(field);
         }
-        else if (type == RealType.REAL) {
+        else if (type.equals(RealType.REAL)) {
             return floatToRawIntBits(row.getFloat(field));
         }
         else if (type instanceof DecimalType) {
@@ -258,7 +258,7 @@ public static boolean getBoolean(Type type, RowResult row, int field)
 
     public static double getDouble(Type type, RowResult row, int field)
     {
-        if (type == DoubleType.DOUBLE) {
+        if (type.equals(DoubleType.DOUBLE)) {
             return row.getDouble(field);
         }
         else {

presto-main/src/main/java/com/facebook/presto/metadata/BuiltInTypeAndFunctionNamespaceManager.java

@@ -316,13 +316,15 @@
 import static com.facebook.presto.common.type.BooleanType.BOOLEAN;
 import static com.facebook.presto.common.type.DateType.DATE;
 import static com.facebook.presto.common.type.DoubleType.DOUBLE;
+import static com.facebook.presto.common.type.DoubleType.OLD_NAN_DOUBLE;
 import static com.facebook.presto.common.type.HyperLogLogType.HYPER_LOG_LOG;
 import static com.facebook.presto.common.type.IntegerType.INTEGER;
 import static com.facebook.presto.common.type.JsonType.JSON;
 import static com.facebook.presto.common.type.KdbTreeType.KDB_TREE;
 import static com.facebook.presto.common.type.KllSketchParametricType.KLL_SKETCH;
 import static com.facebook.presto.common.type.P4HyperLogLogType.P4_HYPER_LOG_LOG;
 import static com.facebook.presto.common.type.QuantileDigestParametricType.QDIGEST;
+import static com.facebook.presto.common.type.RealType.OLD_NAN_REAL;
 import static com.facebook.presto.common.type.RealType.REAL;
 import static com.facebook.presto.common.type.SmallintType.SMALLINT;
 import static com.facebook.presto.common.type.TDigestParametricType.TDIGEST;
@@ -605,14 +607,14 @@ public BuiltInTypeAndFunctionNamespaceManager(
                 .build(CacheLoader.from(this::instantiateParametricType));
 
         registerBuiltInFunctions(getBuiltInFunctions(featuresConfig));
-        registerBuiltInTypes();
+        registerBuiltInTypes(featuresConfig);
 
         for (Type type : requireNonNull(types, "types is null")) {
             addType(type);
         }
     }
 
-    private void registerBuiltInTypes()
+    private void registerBuiltInTypes(FeaturesConfig featuresConfig)
     {
         // always add the built-in types; Presto will not function without these
         addType(UNKNOWN);
@@ -621,8 +623,14 @@ private void registerBuiltInTypes()
         addType(INTEGER);
         addType(SMALLINT);
         addType(TINYINT);
-        addType(DOUBLE);
-        addType(REAL);
+        if(!featuresConfig.getUseNewNanDefinition()) {
+            addType(OLD_NAN_DOUBLE);
+            addType(OLD_NAN_REAL);
+
+        } else {
+            addType(DOUBLE);
+            addType(REAL);
+        }
         addType(VARBINARY);
         addType(DATE);
         addType(TIME);

presto-main/src/main/java/com/facebook/presto/operator/DynamicFilterSourceOperator.java

@@ -191,7 +191,7 @@ private DynamicFilterSourceOperator(
         for (int channelIndex = 0; channelIndex < channels.size(); ++channelIndex) {
             Type type = channels.get(channelIndex).getType();
             // Skipping DOUBLE and REAL in collectMinMaxValues to avoid dealing with NaN values
-            if (minMaxCollectionLimit > 0 && type.isOrderable() && type != DOUBLE && type != REAL) {
+            if (minMaxCollectionLimit > 0 && type.isOrderable() && !type.equals(DOUBLE) && !type.equals(REAL)) {
                 minMaxChannelsBuilder.add(channelIndex);
             }
             this.blockBuilders[channelIndex] = type.createBlockBuilder(null, EXPECTED_BLOCK_BUILDER_SIZE);

presto-main/src/main/java/com/facebook/presto/sql/analyzer/ExpressionAnalyzer.java

@@ -844,7 +844,7 @@ protected Type visitLongLiteral(LongLiteral node, StackableAstVisitorContext<Con
         @Override
         protected Type visitDoubleLiteral(DoubleLiteral node, StackableAstVisitorContext<Context> context)
         {
-            return setExpressionType(node, DOUBLE);
+            return setExpressionType(node, functionAndTypeResolver.getType(DOUBLE.getTypeSignature()));
         }
 
         @Override

presto-main/src/main/java/com/facebook/presto/sql/relational/SqlToRowExpressionTranslator.java

@@ -358,7 +358,7 @@ protected RowExpression visitLongLiteral(LongLiteral node, Context context)
         @Override
         protected RowExpression visitDoubleLiteral(DoubleLiteral node, Context context)
         {
-            return constant(node.getValue(), DOUBLE);
+            return constant(node.getValue(), functionAndTypeManager.getType(DOUBLE.getTypeSignature()));
         }
 
         @Override

presto-main/src/test/java/com/facebook/presto/block/BlockAssertions.java

@@ -781,7 +781,7 @@ public static Block createRandomBlockForType(
         else if (type == BIGINT) {
             block = createRandomLongsBlock(positionCount, primitiveNullRate);
         }
-        else if (type == INTEGER || type == REAL) {
+        else if (type == INTEGER || type.equals(REAL)) {
             block = createRandomIntsBlock(positionCount, primitiveNullRate);
         }
         else if (type == SMALLINT) {

presto-main/src/test/java/com/facebook/presto/operator/scalar/BenchmarkJsonToArrayCast.java

@@ -151,7 +151,7 @@ private static String generateRandomJsonValue(Type valueType)
             if (valueType == BIGINT) {
                 return Long.toString(ThreadLocalRandom.current().nextLong());
             }
-            else if (valueType == DOUBLE) {
+            else if (valueType.equals(DOUBLE)) {
                 return Double.toString(ThreadLocalRandom.current().nextDouble());
             }
             else if (valueType == VARCHAR) {

presto-main/src/test/java/com/facebook/presto/operator/scalar/BenchmarkJsonToMapCast.java

@@ -155,7 +155,7 @@ private static String generateRandomJsonValue(Type valueType)
             if (valueType == BIGINT) {
                 return Long.toString(ThreadLocalRandom.current().nextLong());
             }
-            else if (valueType == DOUBLE) {
+            else if (valueType.equals(DOUBLE)) {
                 return Double.toString(ThreadLocalRandom.current().nextDouble());
             }
             else if (valueType == VARCHAR) {

presto-main/src/test/java/com/facebook/presto/sql/planner/TestExpressionDomainTranslator.java

@@ -1556,7 +1556,7 @@ public T getMax()
 
         public boolean isFractional()
         {
-            return type == DOUBLE || type == REAL || (type instanceof DecimalType && ((DecimalType) type).getScale() > 0);
+            return type.equals(DOUBLE) || type.equals(REAL) || (type instanceof DecimalType && ((DecimalType) type).getScale() > 0);
         }
     }
 }