Remove TypeManager.resolveOperator

This API is a hack. TypeManager.resolveOperator internally will call
FunctionManager.resolveOperator. However, FunctionManager is not initialized
in TypeRegistry constructor, but set in FunctionManager's constructo, due to
cyclic dependency. So to be able to call this API, we see following hacky code
in test everywhere:
    TypeManager typeManager = new TypeRegistr();
    // associate typeManager with a function manager
    new FunctionManager(typeManager, ....);

The only reason that this API exists is because MapParametricType.createType
needs to use it. So we did
1) adding TypeManager to ParamatricType.createType
2) adding TypeManager.resolveOperator

just so MapParametricType could provide MethodHandle that would be used to
build hashtable in Map block, while no other API needs
TypeManager.resolveOperator, and no other ParametricType needs access to
TypeManager. Another problem is that the API directly returns a MethodHandle,
which means this API will not work once we support user defined types with
operator implementation that's not a Java MethodHandle.

Since the solution will not work for future generalized caase, removing it for
now, and directly check for MapParametricType in TypeRegistry to special handle
map types.

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

@@ -17,7 +17,7 @@
 
 import java.util.List;
 
-import static com.facebook.presto.common.type.StatisticalDigestParametricType.checkArgument;
+import static java.lang.String.format;
 
 public final class LongEnumParametricType
         implements ParametricType
@@ -38,7 +38,7 @@ public String getName()
     }
 
     @Override
-    public Type createType(TypeManager typeManager, List<TypeParameter> parameters)
+    public Type createType(List<TypeParameter> parameters)
     {
         if (parameters.isEmpty()) {
             return new LongEnumType(name, enumMap);
@@ -50,4 +50,11 @@ public Type createType(TypeManager typeManager, List<TypeParameter> parameters)
                 parameters);
         return new LongEnumType(name, parameters.get(0).getLongEnumMap());
     }
+
+    private static void checkArgument(boolean argument, String format, Object... args)
+    {
+        if (!argument) {
+            throw new IllegalArgumentException(format(format, args));
+        }
+    }
 }

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

@@ -19,5 +19,5 @@ public interface ParametricType
 {
     String getName();
 
-    Type createType(TypeManager typeManager, List<TypeParameter> parameters);
+    Type createType(List<TypeParameter> parameters);
 }

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

@@ -21,7 +21,7 @@ public abstract class StatisticalDigestParametricType
         implements ParametricType
 {
     @Override
-    public Type createType(TypeManager typeManager, List<TypeParameter> parameters)
+    public Type createType(List<TypeParameter> parameters)
     {
         checkArgument(parameters.size() == 1, "%s type expects exactly one type as a parameter, got %s", this.getName(), parameters);
         checkArgument(

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

@@ -13,9 +13,6 @@
  */
 package com.facebook.presto.common.type;
 
-import com.facebook.presto.common.function.OperatorType;
-
-import java.lang.invoke.MethodHandle;
 import java.util.Collection;
 import java.util.List;
 import java.util.Optional;
@@ -49,6 +46,4 @@ public interface TypeManager
     boolean isTypeOnlyCoercion(Type actualType, Type expectedType);
 
     Optional<Type> coerceTypeBase(Type sourceType, String resultTypeBase);
-
-    MethodHandle resolveOperator(OperatorType operatorType, List<? extends Type> argumentTypes);
 }

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

@@ -17,7 +17,7 @@
 
 import java.util.List;
 
-import static com.facebook.presto.common.type.StatisticalDigestParametricType.checkArgument;
+import static java.lang.String.format;
 
 public final class VarcharEnumParametricType
         implements ParametricType
@@ -38,7 +38,7 @@ public String getName()
     }
 
     @Override
-    public Type createType(TypeManager typeManager, List<TypeParameter> parameters)
+    public Type createType(List<TypeParameter> parameters)
     {
         if (parameters.isEmpty()) {
             return new VarcharEnumType(name, enumMap);
@@ -50,4 +50,11 @@ public Type createType(TypeManager typeManager, List<TypeParameter> parameters)
                 parameters);
         return new VarcharEnumType(name, parameters.get(0).getVarcharEnumMap());
     }
+
+    private static void checkArgument(boolean argument, String format, Object... args)
+    {
+        if (!argument) {
+            throw new IllegalArgumentException(format(format, args));
+        }
+    }
 }

presto-common/src/test/java/com/facebook/presto/common/type/TestingTypeManager.java

@@ -13,10 +13,8 @@
  */
 package com.facebook.presto.common.type;
 
-import com.facebook.presto.common.function.OperatorType;
 import com.google.common.collect.ImmutableList;
 
-import java.lang.invoke.MethodHandle;
 import java.util.Collection;
 import java.util.List;
 import java.util.Optional;
@@ -86,10 +84,4 @@ public Optional<Type> coerceTypeBase(Type sourceType, String resultTypeBase)
     {
         throw new UnsupportedOperationException();
     }
-
-    @Override
-    public MethodHandle resolveOperator(OperatorType operatorType, List<? extends Type> argumentTypes)
-    {
-        throw new UnsupportedOperationException();
-    }
 }

presto-main/src/main/java/com/facebook/presto/testing/TestingEnvironment.java

@@ -14,19 +14,26 @@
 package com.facebook.presto.testing;
 
 import com.facebook.presto.block.BlockEncodingManager;
+import com.facebook.presto.common.function.OperatorType;
+import com.facebook.presto.common.type.Type;
 import com.facebook.presto.common.type.TypeManager;
 import com.facebook.presto.metadata.FunctionManager;
 import com.facebook.presto.sql.analyzer.FeaturesConfig;
 import com.facebook.presto.type.TypeRegistry;
 
+import java.lang.invoke.MethodHandle;
+
+import static com.facebook.presto.sql.analyzer.TypeSignatureProvider.fromTypes;
+
 public class TestingEnvironment
 {
     private TestingEnvironment() {}
 
     public static final TypeManager TYPE_MANAGER = new TypeRegistry();
+    public static final FunctionManager FUNCTION_MANAGER = new FunctionManager(TYPE_MANAGER, new BlockEncodingManager(), new FeaturesConfig());
 
-    static {
-        // wire TYPE_MANAGER with function manager
-        new FunctionManager(TYPE_MANAGER, new BlockEncodingManager(), new FeaturesConfig());
+    public static MethodHandle getOperatorMethodHandle(OperatorType operatorType, Type... parameterTypes)
+    {
+        return FUNCTION_MANAGER.getBuiltInScalarFunctionImplementation(FUNCTION_MANAGER.resolveOperator(operatorType, fromTypes(parameterTypes))).getMethodHandle();
     }
 }

presto-main/src/main/java/com/facebook/presto/type/ArrayParametricType.java

@@ -18,7 +18,6 @@
 import com.facebook.presto.common.type.ParametricType;
 import com.facebook.presto.common.type.StandardTypes;
 import com.facebook.presto.common.type.Type;
-import com.facebook.presto.common.type.TypeManager;
 import com.facebook.presto.common.type.TypeParameter;
 
 import java.util.List;
@@ -41,7 +40,7 @@ public String getName()
     }
 
     @Override
-    public Type createType(TypeManager typeManager, List<TypeParameter> parameters)
+    public Type createType(List<TypeParameter> parameters)
     {
         checkArgument(parameters.size() == 1, "Array type expects exactly one type as a parameter, got %s", parameters);
         checkArgument(

presto-main/src/main/java/com/facebook/presto/type/CharParametricType.java

@@ -17,7 +17,6 @@
 import com.facebook.presto.common.type.ParametricType;
 import com.facebook.presto.common.type.StandardTypes;
 import com.facebook.presto.common.type.Type;
-import com.facebook.presto.common.type.TypeManager;
 import com.facebook.presto.common.type.TypeParameter;
 import com.facebook.presto.spi.PrestoException;
 
@@ -38,7 +37,7 @@ public String getName()
     }
 
     @Override
-    public Type createType(TypeManager typeManager, List<TypeParameter> parameters)
+    public Type createType(List<TypeParameter> parameters)
     {
         if (parameters.isEmpty()) {
             return createCharType(1);

presto-main/src/main/java/com/facebook/presto/type/DecimalParametricType.java

@@ -18,7 +18,6 @@
 import com.facebook.presto.common.type.ParametricType;
 import com.facebook.presto.common.type.StandardTypes;
 import com.facebook.presto.common.type.Type;
-import com.facebook.presto.common.type.TypeManager;
 import com.facebook.presto.common.type.TypeParameter;
 import com.facebook.presto.spi.PrestoException;
 
@@ -38,7 +37,7 @@ public String getName()
     }
 
     @Override
-    public Type createType(TypeManager typeManager, List<TypeParameter> parameters)
+    public Type createType(List<TypeParameter> parameters)
     {
         try {
             switch (parameters.size()) {

presto-main/src/main/java/com/facebook/presto/type/FunctionParametricType.java

@@ -17,7 +17,6 @@
 import com.facebook.presto.common.type.ParameterKind;
 import com.facebook.presto.common.type.ParametricType;
 import com.facebook.presto.common.type.Type;
-import com.facebook.presto.common.type.TypeManager;
 import com.facebook.presto.common.type.TypeParameter;
 
 import java.util.List;
@@ -42,7 +41,7 @@ public String getName()
     }
 
     @Override
-    public Type createType(TypeManager typeManager, List<TypeParameter> parameters)
+    public Type createType(List<TypeParameter> parameters)
     {
         checkArgument(parameters.size() >= 1, "Function type must have at least one parameter, got %s", parameters);
         checkArgument(

presto-main/src/main/java/com/facebook/presto/type/MapParametricType.java

@@ -19,15 +19,15 @@
 import com.facebook.presto.common.type.ParametricType;
 import com.facebook.presto.common.type.StandardTypes;
 import com.facebook.presto.common.type.Type;
-import com.facebook.presto.common.type.TypeManager;
 import com.facebook.presto.common.type.TypeParameter;
-import com.google.common.collect.ImmutableList;
+import com.facebook.presto.metadata.FunctionManager;
 
 import java.lang.invoke.MethodHandle;
 import java.util.List;
 
 import static com.facebook.presto.common.block.MethodHandleUtil.compose;
 import static com.facebook.presto.common.block.MethodHandleUtil.nativeValueGetter;
+import static com.facebook.presto.sql.analyzer.TypeSignatureProvider.fromTypes;
 import static com.google.common.base.Preconditions.checkArgument;
 
 public final class MapParametricType
@@ -42,7 +42,12 @@ public String getName()
     }
 
     @Override
-    public Type createType(TypeManager typeManager, List<TypeParameter> parameters)
+    public Type createType(List<TypeParameter> parameters)
+    {
+        throw new IllegalStateException("MapType creation requires map operator MethodHandle. Please use `createType(functionManager, parameters)` instead");
+    }
+
+    public Type createType(FunctionManager functionManager, List<TypeParameter> parameters)
     {
         checkArgument(parameters.size() == 2, "Expected two parameters, got %s", parameters);
         TypeParameter firstParameter = parameters.get(0);
@@ -54,9 +59,9 @@ public Type createType(TypeManager typeManager, List<TypeParameter> parameters)
 
         Type keyType = firstParameter.getType();
         Type valueType = secondParameter.getType();
-        MethodHandle keyNativeEquals = typeManager.resolveOperator(OperatorType.EQUAL, ImmutableList.of(keyType, keyType));
+        MethodHandle keyNativeEquals = functionManager.getBuiltInScalarFunctionImplementation(functionManager.resolveOperator(OperatorType.EQUAL, fromTypes(keyType, keyType))).getMethodHandle();
         MethodHandle keyBlockEquals = compose(keyNativeEquals, nativeValueGetter(keyType), nativeValueGetter(keyType));
-        MethodHandle keyNativeHashCode = typeManager.resolveOperator(OperatorType.HASH_CODE, ImmutableList.of(keyType));
+        MethodHandle keyNativeHashCode = functionManager.getBuiltInScalarFunctionImplementation(functionManager.resolveOperator(OperatorType.HASH_CODE, fromTypes(keyType))).getMethodHandle();
         MethodHandle keyBlockHashCode = compose(keyNativeHashCode, nativeValueGetter(keyType));
         return new MapType(
                 keyType,

presto-main/src/main/java/com/facebook/presto/type/RowParametricType.java

@@ -20,7 +20,6 @@
 import com.facebook.presto.common.type.RowType;
 import com.facebook.presto.common.type.StandardTypes;
 import com.facebook.presto.common.type.Type;
-import com.facebook.presto.common.type.TypeManager;
 import com.facebook.presto.common.type.TypeParameter;
 import com.facebook.presto.common.type.TypeSignature;
 import com.facebook.presto.common.type.TypeSignatureParameter;
@@ -46,7 +45,7 @@ public String getName()
     }
 
     @Override
-    public Type createType(TypeManager typeManager, List<TypeParameter> parameters)
+    public Type createType(List<TypeParameter> parameters)
     {
         checkArgument(!parameters.isEmpty(), "Row type must have at least one parameter");
         checkArgument(

presto-main/src/main/java/com/facebook/presto/type/TypeRegistry.java

@@ -13,7 +13,6 @@
  */
 package com.facebook.presto.type;
 
-import com.facebook.presto.common.function.OperatorType;
 import com.facebook.presto.common.type.ArrayType;
 import com.facebook.presto.common.type.CharType;
 import com.facebook.presto.common.type.DecimalType;
@@ -42,7 +41,6 @@
 import javax.annotation.concurrent.ThreadSafe;
 import javax.inject.Inject;
 
-import java.lang.invoke.MethodHandle;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -75,7 +73,6 @@
 import static com.facebook.presto.common.type.VarbinaryType.VARBINARY;
 import static com.facebook.presto.common.type.VarcharType.createUnboundedVarcharType;
 import static com.facebook.presto.common.type.VarcharType.createVarcharType;
-import static com.facebook.presto.sql.analyzer.TypeSignatureProvider.fromTypes;
 import static com.facebook.presto.type.ArrayParametricType.ARRAY;
 import static com.facebook.presto.type.CodePointsType.CODE_POINTS;
 import static com.facebook.presto.type.ColorType.COLOR;
@@ -213,8 +210,11 @@ private Type instantiateParametricType(TypeSignature signature)
         if (parametricType == null) {
             throw new IllegalArgumentException("Unknown type " + signature);
         }
+        else if (parametricType instanceof MapParametricType) {
+            return ((MapParametricType) parametricType).createType(functionManager, parameters);
+        }
 
-        Type instantiatedType = parametricType.createType(this, parameters);
+        Type instantiatedType = parametricType.createType(parameters);
 
         // TODO: reimplement this check? Currently "varchar(Integer.MAX_VALUE)" fails with "varchar"
         //checkState(instantiatedType.equalsSignature(signature), "Instantiated parametric type name (%s) does not match expected name (%s)", instantiatedType, signature);
@@ -678,13 +678,6 @@ public static boolean isCovariantTypeBase(String typeBase)
         return typeBase.equals(StandardTypes.ARRAY) || typeBase.equals(StandardTypes.MAP);
     }
 
-    @Override
-    public MethodHandle resolveOperator(OperatorType operatorType, List<? extends Type> argumentTypes)
-    {
-        requireNonNull(functionManager, "functionManager is null");
-        return functionManager.getBuiltInScalarFunctionImplementation(functionManager.resolveOperator(operatorType, fromTypes(argumentTypes))).getMethodHandle();
-    }
-
     public static class TypeCompatibility
     {
         private final Optional<Type> commonSuperType;

presto-main/src/main/java/com/facebook/presto/type/VarcharParametricType.java

@@ -16,7 +16,6 @@
 import com.facebook.presto.common.type.ParametricType;
 import com.facebook.presto.common.type.StandardTypes;
 import com.facebook.presto.common.type.Type;
-import com.facebook.presto.common.type.TypeManager;
 import com.facebook.presto.common.type.TypeParameter;
 import com.facebook.presto.common.type.VarcharType;
 
@@ -36,7 +35,7 @@ public String getName()
     }
 
     @Override
-    public Type createType(TypeManager typeManager, List<TypeParameter> parameters)
+    public Type createType(List<TypeParameter> parameters)
     {
         if (parameters.isEmpty()) {
             return createUnboundedVarcharType();

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

@@ -24,7 +24,6 @@
 import com.facebook.presto.common.type.MapType;
 import com.facebook.presto.common.type.RowType;
 import com.facebook.presto.common.type.Type;
-import com.google.common.collect.ImmutableList;
 import io.airlift.slice.Slice;
 
 import java.lang.invoke.MethodHandle;
@@ -62,7 +61,7 @@
 import static com.facebook.presto.common.type.VarbinaryType.VARBINARY;
 import static com.facebook.presto.common.type.VarcharType.VARCHAR;
 import static com.facebook.presto.testing.TestingConnectorSession.SESSION;
-import static com.facebook.presto.testing.TestingEnvironment.TYPE_MANAGER;
+import static com.facebook.presto.testing.TestingEnvironment.getOperatorMethodHandle;
 import static com.facebook.presto.util.StructuralTestUtil.appendToBlockBuilder;
 import static com.google.common.base.Preconditions.checkArgument;
 import static io.airlift.slice.Slices.utf8Slice;
@@ -849,10 +848,9 @@ public static Block wrapBlock(Block block, int positionCount, List<Encoding> wra
 
     public static MapType createMapType(Type keyType, Type valueType)
     {
-        MethodHandle keyNativeEquals = TYPE_MANAGER.resolveOperator(OperatorType.EQUAL, ImmutableList.of(keyType, keyType));
-        MethodHandle keyBlockNativeEquals = compose(keyNativeEquals, nativeValueGetter(keyType));
+        MethodHandle keyNativeEquals = getOperatorMethodHandle(OperatorType.EQUAL, keyType, keyType);
         MethodHandle keyBlockEquals = compose(keyNativeEquals, nativeValueGetter(keyType), nativeValueGetter(keyType));
-        MethodHandle keyNativeHashCode = TYPE_MANAGER.resolveOperator(OperatorType.HASH_CODE, ImmutableList.of(keyType));
+        MethodHandle keyNativeHashCode = getOperatorMethodHandle(OperatorType.HASH_CODE, keyType);
         MethodHandle keyBlockHashCode = compose(keyNativeHashCode, nativeValueGetter(keyType));
         return new MapType(
                 keyType,