Preserve case for RowType's field name and JSON content when CAST

Preserve the case of double quoted field names in RowType. When casting
from ``JSON`` to ``ROW``, treat the double quoted field name as case
sensitive and unquoted field name as case insensitive in field
matching.

Add a config property `legacy_json_cast` to control and support the
legacy behavior which do not enforce the case when matching.

presto-common/src/main/java/com/facebook/presto/common/function/SqlFunctionProperties.java

@@ -34,6 +34,7 @@ public class SqlFunctionProperties
     private final Locale sessionLocale;
     private final String sessionUser;
     private final boolean fieldNamesInJsonCastEnabled;
+    private final boolean legacyJsonCast;
     private final Map<String, String> extraCredentials;
 
     private SqlFunctionProperties(
@@ -46,6 +47,7 @@ private SqlFunctionProperties(
             Locale sessionLocale,
             String sessionUser,
             boolean fieldNamesInJsonCastEnabled,
+            boolean legacyJsonCast,
             Map<String, String> extraCredentials)
     {
         this.parseDecimalLiteralAsDouble = parseDecimalLiteralAsDouble;
@@ -57,6 +59,7 @@ private SqlFunctionProperties(
         this.sessionLocale = requireNonNull(sessionLocale, "sessionLocale is null");
         this.sessionUser = requireNonNull(sessionUser, "sessionUser is null");
         this.fieldNamesInJsonCastEnabled = fieldNamesInJsonCastEnabled;
+        this.legacyJsonCast = legacyJsonCast;
         this.extraCredentials = requireNonNull(extraCredentials, "extraCredentials is null");
     }
 
@@ -111,6 +114,11 @@ public boolean isFieldNamesInJsonCastEnabled()
         return fieldNamesInJsonCastEnabled;
     }
 
+    public boolean isLegacyJsonCast()
+    {
+        return legacyJsonCast;
+    }
+
     @Override
     public boolean equals(Object o)
     {
@@ -129,13 +137,16 @@ public boolean equals(Object o)
                 Objects.equals(sessionStartTime, that.sessionStartTime) &&
                 Objects.equals(sessionLocale, that.sessionLocale) &&
                 Objects.equals(sessionUser, that.sessionUser) &&
-                Objects.equals(extraCredentials, that.extraCredentials);
+                Objects.equals(extraCredentials, that.extraCredentials) &&
+                Objects.equals(legacyJsonCast, that.legacyJsonCast);
     }
 
     @Override
     public int hashCode()
     {
-        return Objects.hash(parseDecimalLiteralAsDouble, legacyRowFieldOrdinalAccessEnabled, timeZoneKey, legacyTimestamp, legacyMapSubscript, sessionStartTime, sessionLocale, sessionUser, extraCredentials);
+        return Objects.hash(parseDecimalLiteralAsDouble, legacyRowFieldOrdinalAccessEnabled, timeZoneKey,
+                legacyTimestamp, legacyMapSubscript, sessionStartTime, sessionLocale, sessionUser,
+                extraCredentials, legacyJsonCast);
     }
 
     public static Builder builder()
@@ -154,6 +165,7 @@ public static class Builder
         private Locale sessionLocale;
         private String sessionUser;
         private boolean fieldNamesInJsonCastEnabled;
+        private boolean legacyJsonCast;
         private Map<String, String> extraCredentials = emptyMap();
 
         private Builder() {}
@@ -218,9 +230,26 @@ public Builder setFieldNamesInJsonCastEnabled(boolean fieldNamesInJsonCastEnable
             return this;
         }
 
+        public Builder setLegacyJsonCast(boolean legacyJsonCast)
+        {
+            this.legacyJsonCast = legacyJsonCast;
+            return this;
+        }
+
         public SqlFunctionProperties build()
         {
-            return new SqlFunctionProperties(parseDecimalLiteralAsDouble, legacyRowFieldOrdinalAccessEnabled, timeZoneKey, legacyTimestamp, legacyMapSubscript, sessionStartTime, sessionLocale, sessionUser, fieldNamesInJsonCastEnabled, extraCredentials);
+            return new SqlFunctionProperties(
+                    parseDecimalLiteralAsDouble,
+                    legacyRowFieldOrdinalAccessEnabled,
+                    timeZoneKey,
+                    legacyTimestamp,
+                    legacyMapSubscript,
+                    sessionStartTime,
+                    sessionLocale,
+                    sessionUser,
+                    fieldNamesInJsonCastEnabled,
+                    legacyJsonCast,
+                    extraCredentials);
         }
     }
 }

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

@@ -83,6 +83,11 @@ public static Field field(Type type)
         return new Field(Optional.empty(), type);
     }
 
+    public static Field field(String name, Type type, boolean delimited)
+    {
+        return new Field(Optional.of(name), type, delimited);
+    }
+
     private static TypeSignature makeSignature(List<Field> fields)
     {
         int size = fields.size();

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

@@ -510,6 +510,7 @@ private static TypeSignature parseRowTypeSignature(String signature, Set<String>
 
         List<TypeSignatureParameter> fields = new ArrayList<>();
 
+        Set<String> distinctFieldNames = new HashSet<>();
         for (int i = StandardTypes.ROW.length() + 1; i < signature.length(); i++) {
             char c = signature.charAt(i);
             switch (state) {
@@ -556,13 +557,19 @@ else if (c == ')' && bracketLevel > 1) {
                     }
                     else if (c == ')') {
                         verify(tokenStart >= 0, "Expect tokenStart to be non-negative");
-                        fields.add(parseTypeOrNamedType(signature.substring(tokenStart, i).trim(), literalParameters));
+                        TypeSignatureParameter parameter = parseTypeOrNamedType(signature.substring(tokenStart, i).trim(), literalParameters);
+                        parameter.getNamedTypeSignature().getName()
+                                .ifPresent(fieldName -> checkDuplicateAndAdd(distinctFieldNames, fieldName));
+                        fields.add(parameter);
                         tokenStart = -1;
                         state = RowTypeSignatureParsingState.FINISHED;
                     }
                     else if (c == ',' && bracketLevel == 1) {
                         verify(tokenStart >= 0, "Expect tokenStart to be non-negative");
-                        fields.add(parseTypeOrNamedType(signature.substring(tokenStart, i).trim(), literalParameters));
+                        TypeSignatureParameter parameter = parseTypeOrNamedType(signature.substring(tokenStart, i).trim(), literalParameters);
+                        parameter.getNamedTypeSignature().getName()
+                                .ifPresent(fieldName -> checkDuplicateAndAdd(distinctFieldNames, fieldName));
+                        fields.add(parameter);
                         tokenStart = -1;
                         state = RowTypeSignatureParsingState.START_OF_FIELD;
                     }
@@ -578,6 +585,7 @@ else if (c == ')' && bracketLevel > 1) {
                     else if (c == ')') {
                         verify(tokenStart >= 0, "Expect tokenStart to be non-negative");
                         verify(delimitedColumnName != null, "Expect delimitedColumnName to be non-null");
+                        checkDuplicateAndAdd(distinctFieldNames, delimitedColumnName);
                         fields.add(TypeSignatureParameter.of(new NamedTypeSignature(
                                 Optional.of(new RowFieldName(delimitedColumnName, true)),
                                 parseTypeSignature(signature.substring(tokenStart, i).trim(), literalParameters))));
@@ -588,6 +596,7 @@ else if (c == ')') {
                     else if (c == ',' && bracketLevel == 1) {
                         verify(tokenStart >= 0, "Expect tokenStart to be non-negative");
                         verify(delimitedColumnName != null, "Expect delimitedColumnName to be non-null");
+                        checkDuplicateAndAdd(distinctFieldNames, delimitedColumnName);
                         fields.add(TypeSignatureParameter.of(new NamedTypeSignature(
                                 Optional.of(new RowFieldName(delimitedColumnName, true)),
                                 parseTypeSignature(signature.substring(tokenStart, i).trim(), literalParameters))));
@@ -609,6 +618,10 @@ else if (c == ',' && bracketLevel == 1) {
         return new TypeSignature(signature.substring(0, StandardTypes.ROW.length()), fields);
     }
 
+    private static void checkDuplicateAndAdd(Set<String> fieldNames, String fieldName)
+    {
+        verify(fieldNames.add(fieldName), "Duplicate field: " + fieldName);
+    }
     private static TypeSignatureParameter parseTypeOrNamedType(String typeOrNamedType, Set<String> literalParameters)
     {
         int split = typeOrNamedType.indexOf(' ');

presto-docs/src/main/sphinx/admin/properties.rst

@@ -981,3 +981,16 @@ Logging Properties
     * **Default value:** ``100MB``
 
     The maximum file size for the log file of the HTTP server.
+
+
+Legacy Compatible Properties
+------------------------------
+
+``legacy_json_cast``
+^^^^^^^^^^^^^^^^^^^^^
+
+    * **Type:** ``boolean``
+    * **Default value:** ``false``
+
+    When casting from ``JSON`` to ``ROW``, by default the case of double quoted field names in RowType is strictly enforced when matching, and unquoted field name in RowType is case-insensitive when matching.
+    To ignore the case of field names in RowType when casting from ``JSON`` to ``ROW`` (for legacy support), set config property ``legacy_json_cast`` to ``true``. After setting the property, the matching would be always case-insensitive.

presto-docs/src/main/sphinx/functions/json.rst

@@ -71,6 +71,31 @@ Cast from JSON
         SELECT CAST(JSON '{"k1": [1, 23], "k2": 456}' AS MAP(VARCHAR, JSON)); -- {k1 = JSON '[1,23]', k2 = JSON '456'}
         SELECT CAST(JSON '[null]' AS ARRAY(JSON)); -- [JSON 'null']
 
+.. note::
+
+    When casting from ``JSON`` to ``ROW``, by default the case of double quoted field names
+    in RowType is strictly enforced when matching, and unquoted field name in RowType is
+    case-insensitive when matching. For example::
+
+        SELECT CAST(JSON '{"v1":123,"V2":"abc","v3":true}' AS ROW(v1 BIGINT, v2 VARCHAR, v3 BOOLEAN)); -- {v1=123, v2=abc, v3=true}
+        SELECT CAST(JSON '{"v1":123,"V2":"abc","v3":true}' AS ROW(v1 BIGINT, "V2" VARCHAR, v3 BOOLEAN)); -- {v1=123, V2=abc, v3=true}
+        SELECT CAST(JSON '{"v1":123,"V2":"abc","v3":true}' AS ROW(v1 BIGINT, "v2" VARCHAR, v3 BOOLEAN)); -- {v1=123, v2=null, v3=true}
+        SELECT CAST(JSON '{"v1":123,"V2":"abc", "v2":"abc2","v3":true}' AS ROW(v1 BIGINT, v2 VARCHAR, "V2" VARCHAR, v3 BOOLEAN)); -- {v1=123, v2=abc2, V2=abc, v3=true}
+
+    If the name of a field does not match (including case sensitivity), the value is null.
+
+    To ignore the case of field names in RowType when casting from ``JSON`` to ``ROW``
+    (for example, for legacy support) set config property ``legacy_json_cast`` to ``true``
+    in the coordinator and the worker's `config properties <../admin/properties.html#legacy-compatible-properties>`_.
+    After setting the property, the matching would be case-insensitive as the following example::
+
+        SELECT CAST(JSON '{"v1":123,"V2":"abc","v3":true}' AS ROW(v1 BIGINT, v2 VARCHAR, v3 BOOLEAN)); -- {v1=123, v2=abc, v3=true}
+        SELECT CAST(JSON '{"v1":123,"V2":"abc","v3":true}' AS ROW(v1 BIGINT, "V2" VARCHAR, "V3" BOOLEAN)); -- {v1=123, V2=abc, V3=true}
+
+    The following statement returns an error due to duplicate field names in RowType::
+
+        SELECT CAST(JSON '{"v1":123,"V2":"abc","v2":"abc2","v3":true}' AS ROW(v1 BIGINT, "V2" VARCHAR, v2 VARCHAR, "V3" BOOLEAN));
+
 .. note:: When casting from ``JSON`` to ``ROW``, both JSON array and JSON object are supported.
 
 JSON Functions

presto-main/src/main/java/com/facebook/presto/Session.java

@@ -52,6 +52,7 @@
 import java.util.TimeZone;
 import java.util.stream.Collectors;
 
+import static com.facebook.presto.SystemSessionProperties.LEGACY_JSON_CAST;
 import static com.facebook.presto.SystemSessionProperties.isFieldNameInJsonCastEnabled;
 import static com.facebook.presto.SystemSessionProperties.isLegacyMapSubscript;
 import static com.facebook.presto.SystemSessionProperties.isLegacyRowFieldOrdinalAccessEnabled;
@@ -496,6 +497,7 @@ public ConnectorSession toConnectorSession()
 
     public SqlFunctionProperties getSqlFunctionProperties()
     {
+        boolean legacyJsonCast = this.sessionPropertyManager.decodeSystemPropertyValue(LEGACY_JSON_CAST, null, Boolean.class);
         return SqlFunctionProperties.builder()
                 .setTimeZoneKey(timeZoneKey)
                 .setLegacyRowFieldOrdinalAccessEnabled(isLegacyRowFieldOrdinalAccessEnabled(this))
@@ -506,6 +508,7 @@ public SqlFunctionProperties getSqlFunctionProperties()
                 .setSessionLocale(getLocale())
                 .setSessionUser(getUser())
                 .setFieldNamesInJsonCastEnabled(isFieldNameInJsonCastEnabled(this))
+                .setLegacyJsonCast(legacyJsonCast)
                 .setExtraCredentials(identity.getExtraCredentials())
                 .build();
     }

presto-main/src/main/java/com/facebook/presto/SystemSessionProperties.java

@@ -312,6 +312,7 @@ public final class SystemSessionProperties
     public static final String ADD_PARTIAL_NODE_FOR_ROW_NUMBER_WITH_LIMIT = "add_partial_node_for_row_number_with_limit";
     public static final String REWRITE_CASE_TO_MAP_ENABLED = "rewrite_case_to_map_enabled";
     public static final String FIELD_NAMES_IN_JSON_CAST_ENABLED = "field_names_in_json_cast_enabled";
+    public static final String LEGACY_JSON_CAST = "legacy_json_cast";
     public static final String PULL_EXPRESSION_FROM_LAMBDA_ENABLED = "pull_expression_from_lambda_enabled";
     public static final String REWRITE_CONSTANT_ARRAY_CONTAINS_TO_IN_EXPRESSION = "rewrite_constant_array_contains_to_in_expression";
     public static final String INFER_INEQUALITY_PREDICATES = "infer_inequality_predicates";
@@ -1754,6 +1755,11 @@ public SystemSessionProperties(
                         "Include field names in json output when casting rows",
                         featuresConfig.isFieldNamesInJsonCastEnabled(),
                         false),
+                booleanProperty(
+                        LEGACY_JSON_CAST,
+                        "Keep the legacy json cast behavior, do not reserve the case for field names when casting to row type",
+                        featuresConfig.isLegacyJsonCast(),
+                        false),
                 booleanProperty(
                         OPTIMIZE_JOIN_PROBE_FOR_EMPTY_BUILD_RUNTIME,
                         "Optimize join probe at runtime if build side is empty",

presto-main/src/main/java/com/facebook/presto/operator/scalar/JsonToArrayCast.java

@@ -95,7 +95,7 @@ public static Block toArray(ArrayType arrayType, BlockBuilderAppender elementApp
             }
             BlockBuilder blockBuilder = arrayType.getElementType().createBlockBuilder(null, 20);
             while (jsonParser.nextToken() != JsonToken.END_ARRAY) {
-                elementAppender.append(jsonParser, blockBuilder);
+                elementAppender.append(jsonParser, blockBuilder, properties);
             }
             if (jsonParser.nextToken() != null) {
                 throw new JsonCastException(format("Unexpected trailing token: %s", jsonParser.getText()));

presto-main/src/main/java/com/facebook/presto/operator/scalar/JsonToMapCast.java

@@ -103,9 +103,9 @@ public static Block toMap(MapType mapType, BlockBuilderAppender keyAppender, Blo
             HashTable hashTable = new HashTable(mapType.getKeyType(), singleMapBlockBuilder);
             int position = 0;
             while (jsonParser.nextToken() != JsonToken.END_OBJECT) {
-                keyAppender.append(jsonParser, singleMapBlockBuilder);
+                keyAppender.append(jsonParser, singleMapBlockBuilder, properties);
                 jsonParser.nextToken();
-                valueAppender.append(jsonParser, singleMapBlockBuilder);
+                valueAppender.append(jsonParser, singleMapBlockBuilder, properties);
 
                 // Duplicate key detection is required even if the JSON is valid.
                 // For example: CAST(JSON '{"1": 1, "01": 2}' AS MAP<INTEGER, INTEGER>).

presto-main/src/main/java/com/facebook/presto/operator/scalar/JsonToRowCast.java

@@ -35,8 +35,6 @@
 
 import java.lang.invoke.MethodHandle;
 import java.util.List;
-import java.util.Map;
-import java.util.Optional;
 
 import static com.facebook.presto.common.type.TypeSignature.parseTypeSignature;
 import static com.facebook.presto.operator.scalar.ScalarFunctionImplementationChoice.ArgumentProperty.valueTypeArgumentProperty;
@@ -48,7 +46,6 @@
 import static com.facebook.presto.util.JsonUtil.JSON_FACTORY;
 import static com.facebook.presto.util.JsonUtil.canCastFromJson;
 import static com.facebook.presto.util.JsonUtil.createJsonParser;
-import static com.facebook.presto.util.JsonUtil.getFieldNameToIndex;
 import static com.facebook.presto.util.JsonUtil.parseJsonToSingleRowBlock;
 import static com.facebook.presto.util.JsonUtil.truncateIfNecessaryForErrorMessage;
 import static com.facebook.presto.util.Reflection.methodHandle;
@@ -61,7 +58,7 @@ public class JsonToRowCast
         extends SqlOperator
 {
     public static final JsonToRowCast JSON_TO_ROW = new JsonToRowCast();
-    private static final MethodHandle METHOD_HANDLE = methodHandle(JsonToRowCast.class, "toRow", RowType.class, BlockBuilderAppender[].class, Optional.class, SqlFunctionProperties.class, Slice.class);
+    private static final MethodHandle METHOD_HANDLE = methodHandle(JsonToRowCast.class, "toRow", RowType.class, BlockBuilderAppender[].class, SqlFunctionProperties.class, Slice.class);
 
     private JsonToRowCast()
     {
@@ -83,7 +80,7 @@ public BuiltInScalarFunctionImplementation specialize(BoundVariables boundVariab
         BlockBuilderAppender[] fieldAppenders = rowFields.stream()
                 .map(rowField -> createBlockBuilderAppender(rowField.getType()))
                 .toArray(BlockBuilderAppender[]::new);
-        MethodHandle methodHandle = METHOD_HANDLE.bindTo(rowType).bindTo(fieldAppenders).bindTo(getFieldNameToIndex(rowFields));
+        MethodHandle methodHandle = METHOD_HANDLE.bindTo(rowType).bindTo(fieldAppenders);
         return new BuiltInScalarFunctionImplementation(
                 true,
                 ImmutableList.of(valueTypeArgumentProperty(RETURN_NULL_ON_NULL)),
@@ -94,7 +91,6 @@ public BuiltInScalarFunctionImplementation specialize(BoundVariables boundVariab
     public static Block toRow(
             RowType rowType,
             BlockBuilderAppender[] fieldAppenders,
-            Optional<Map<String, Integer>> fieldNameToIndex,
             SqlFunctionProperties properties,
             Slice json)
     {
@@ -113,7 +109,8 @@ public static Block toRow(
                     jsonParser,
                     (SingleRowBlockWriter) rowBlockBuilder.beginBlockEntry(),
                     fieldAppenders,
-                    fieldNameToIndex);
+                    rowType,
+                    properties);
             rowBlockBuilder.closeEntry();
 
             if (jsonParser.nextToken() != null) {

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

@@ -277,6 +277,7 @@ public class FeaturesConfig
     private boolean useDefaultsForCorrelatedAggregationPushdownThroughOuterJoins = true;
     private boolean mergeDuplicateAggregationsEnabled = true;
     private boolean fieldNamesInJsonCastEnabled;
+    private boolean legacyJsonCast;
     private boolean mergeAggregationsWithAndWithoutFilter;
     private boolean simplifyPlanWithEmptyInput = true;
     private PushDownFilterThroughCrossJoinStrategy pushDownFilterExpressionEvaluationThroughCrossJoin = PushDownFilterThroughCrossJoinStrategy.REWRITTEN_TO_INNER_JOIN;
@@ -652,6 +653,18 @@ public FeaturesConfig setFieldNamesInJsonCastEnabled(boolean fieldNamesInJsonCas
         return this;
     }
 
+    public boolean isLegacyJsonCast()
+    {
+        return legacyJsonCast;
+    }
+
+    @Config("legacy-json-cast")
+    public FeaturesConfig setLegacyJsonCast(boolean legacyJsonCast)
+    {
+        this.legacyJsonCast = legacyJsonCast;
+        return this;
+    }
+
     @Config("reduce-agg-for-complex-types-enabled")
     public FeaturesConfig setReduceAggForComplexTypesEnabled(boolean reduceAggForComplexTypesEnabled)
     {