fix from_json function

  * prevent from_json from concatenating duplicate case-insensitive fields.
  * make spark from_json case-sensitive

Author: howz97

bolt/functions/prestosql/tests/JsonCastTest.cpp

@@ -1016,7 +1016,7 @@ TEST_F(JsonCastTest, toRow) {
   auto map = makeNullableFlatVector<JsonNativeType>(
       {R"({"c0":123,"c1":"abc","c2":true})"_sv,
        R"({"c1":"abc","c2":true,"c0":123})"_sv,
-       R"({"c0":123,"c2":true,"c0":456})"_sv,
+       R"({"c10":123,"c2":true,"c0":456})"_sv,
        R"({"c3":123,"c4":"abc","c2":false})"_sv,
        R"({"c0":null,"c2":false})"_sv,
        R"({"c0":null,"c2":null,"c1":null})"_sv},
@@ -1037,17 +1037,17 @@ TEST_F(JsonCastTest, toRow) {
 
   // Use a mix of lower case and upper case JSON keys.
   map = makeNullableFlatVector<JsonNativeType>(
-      {R"({"c0":123,"c1":"abc","c2":true})"_sv,
-       R"({"c1":"abc","c2":true,"c0":123})"_sv,
-       R"({"c0":123,"c2":true,"c0":456})"_sv,
-       R"({"c3":123,"c4":"abc","c2":false})"_sv,
+      {R"({"C0":123,"C1":"abc","C2":true})"_sv,
+       R"({"c1":"abc","C2":true,"c0":123})"_sv,
+       R"({"C10":123,"C2":true,"c0":456})"_sv,
+       R"({"c3":123,"C4":"abc","c2":false})"_sv,
        R"({"c0":null,"c2":false})"_sv,
-       R"({"c0":null,"c2":null,"c1":null})"_sv},
+       R"({"c0":null,"c2":null,"C1":null})"_sv},
       JSON());
   testCast(map, makeRowVector({child4, child5, child6}));
 
   // Use a mix of lower case and upper case field names in target ROW type.
-  testCast(map, makeRowVector({child4, child5, child6}));
+  testCast(map, makeRowVector({"c0", "C1", "C2"}, {child4, child5, child6}));
 
   // Test casting to ROW from JSON null.
   auto null = makeNullableFlatVector<JsonNativeType>({"null"_sv}, JSON());
@@ -1308,3 +1308,35 @@ TEST_F(JsonCastTest, tryCastFromJson) {
   evaluateAndVerify(
       JSON(), ROW({REAL()}), makeRowVector({data}), expectedRow, true);
 }
+
+TEST_F(JsonCastTest, duplicateCaseInsensitiveFields) {
+  auto input = makeFlatVector<JsonNativeType>(
+      {R"({"Name":"first","name":"second"})"_sv}, JSON());
+
+  BOLT_ASSERT_USER_THROW(
+      evaluateCast(JSON(), ROW({"name"}, {VARCHAR()}), makeRowVector({input})),
+      "Duplicate field: name");
+}
+
+TEST_F(JsonCastTest, fromJsonIgnoresDuplicateCaseInsensitiveFields) {
+  auto input = makeFlatVector<StringView>(
+      {R"({"FieldA":["READ","BLUE"],"Foo":"Bar","fieldA":["red","blue"],"foo":"bar"})"},
+      JSON());
+  auto resultType = ROW({"FieldA", "Foo"}, {ARRAY(VARCHAR()), VARCHAR()});
+  auto inputRow = makeRowVector({input});
+  SelectivityVector rows(input->size());
+  exec::ExprSet exprSet({}, &execCtx_);
+  exec::EvalCtx context(&execCtx_, &exprSet, inputRow.get());
+  VectorPtr result;
+
+  auto* jsonCast =
+      dynamic_cast<const JsonCastOperator*>(JsonCastOperator::get().get());
+  ASSERT_NE(jsonCast, nullptr);
+  jsonCast->castFrom(
+      *input, context, rows, resultType, result, /*isFromJson=*/true);
+
+  auto fieldA = makeArrayVector<std::string>({{"READ", "BLUE"}});
+  auto foo = makeFlatVector<StringView>({"Bar"});
+  auto expected = makeRowVector({"FieldA", "Foo"}, {fieldA, foo});
+  bytedance::bolt::test::assertEqualVectors(expected, result);
+}

bolt/functions/prestosql/types/JsonType.cpp

@@ -1119,7 +1119,9 @@ struct CastFromJsonTypedImpl {
         for (column_index_t numFields = rowType.size(), i = 0; i < numFields;
              ++i) {
           rowKey = rowType.nameOf(i);
-          boost::algorithm::to_lower(rowKey);
+          if (!isFromJson) {
+            boost::algorithm::to_lower(rowKey);
+          }
           fieldKey2Idx[rowKey] = i;
           appearedIdxs.insert(i);
         }
@@ -1134,12 +1136,22 @@ struct CastFromJsonTypedImpl {
         for (auto fieldResult : object) {
           SIMDJSON_ASSIGN_OR_RAISE(auto field, fieldResult);
           SIMDJSON_ASSIGN_OR_RAISE(jsonKey, field.unescaped_key(true));
-          boost::algorithm::to_lower(jsonKey);
+          if (!isFromJson) {
+            boost::algorithm::to_lower(jsonKey);
+          }
           auto it = fieldKey2Idx.find(jsonKey);
           if (it == fieldKey2Idx.end()) {
             continue;
           }
           column_index_t jsonKeyIdx = it->second;
+          const bool isFirstOccurrence = appearedIdxs.erase(jsonKeyIdx) > 0;
+          if (!isFirstOccurrence) {
+            if (isFromJson) {
+              continue;
+            }
+            BOLT_USER_FAIL("Duplicate field: {}", jsonKey);
+          }
+
           if (!field.value().is_null()) {
             auto err = BOLT_DYNAMIC_TYPE_DISPATCH(
                 CastFromJsonTypedImpl<simdjson::ondemand::value>::apply,
@@ -1157,7 +1169,6 @@ struct CastFromJsonTypedImpl {
           } else {
             writerTyped.set_null_at(jsonKeyIdx);
           }
-          appearedIdxs.erase(jsonKeyIdx);
         }
 
         // set all non appearing columns to null
@@ -1509,7 +1520,7 @@ void JsonCastOperator::castFromJson(
   auto flinkCompatible =
       context.execCtx()->queryCtx()->queryConfig().enableFlinkCompatible();
 
-  rows.applyToSelected([&](auto row) {
+  context.applyToSelectedNoThrow(rows, [&](auto row) {
     writer.setOffset(row);
     if (inputVector->isNullAt(row)) {
       if (flinkCompatible &&

bolt/functions/sparksql/specialforms/FromJson.cpp

@@ -224,12 +224,7 @@ struct ExtractJsonTypeImpl {
         SIMDJSON_ASSIGN_OR_RAISE(auto object, value.get_object());
 
         const auto& names = rowType.names();
-        bool allFieldsAreAscii =
-            std::all_of(names.begin(), names.end(), [](const auto& name) {
-              return functions::stringCore::isAscii(name.data(), name.size());
-            });
-
-        auto fieldIndices = makeFieldIndicesMap(rowType, allFieldsAreAscii);
+        auto fieldIndices = makeFieldIndicesMap(rowType);
 
         std::string key;
         for (const auto& fieldResult : object) {
@@ -240,11 +235,6 @@ struct ExtractJsonTypeImpl {
           if (!field.value().is_null()) {
             SIMDJSON_ASSIGN_OR_RAISE(key, field.unescaped_key(true));
 
-            if (allFieldsAreAscii) {
-              folly::toLowerAscii(key);
-            } else {
-              boost::algorithm::to_lower(key);
-            }
             auto it = fieldIndices.find(key);
             if (it != fieldIndices.end() && it->second >= 0) {
               const auto index = it->second;
@@ -354,20 +344,13 @@ struct ExtractJsonTypeImpl {
     return simdjson::SUCCESS;
   }
 
-  // Creates a map of lower case field names to their indices in the row type.
+  // Creates a map of field names to their indices in the row type.
   static folly::F14FastMap<std::string, int32_t> makeFieldIndicesMap(
-      const RowType& rowType,
-      bool allFieldsAreAscii) {
+      const RowType& rowType) {
     folly::F14FastMap<std::string, int32_t> fieldIndices;
     const auto size = rowType.size();
     for (auto i = 0; i < size; ++i) {
       std::string key = rowType.nameOf(i);
-      if (allFieldsAreAscii) {
-        folly::toLowerAscii(key);
-      } else {
-        boost::algorithm::to_lower(key);
-      }
-
       fieldIndices[key] = i;
     }
 

bolt/functions/sparksql/tests/FromJsonTest.cpp

@@ -283,5 +283,45 @@ TEST_F(FromJsonTest, scalarRootDocumentDoesNotThrow) {
   testFromJson(scalarInput, expectedRow);
 }
 
+TEST_F(FromJsonTest, DuplicateCaseInsensitiveFields) {
+  // Regression test: when a JSON object contains two keys that differ only in
+  // case (e.g. "FieldA" and "fieldA"), from_json must honor only the first
+  // match (case-insensitive) and NOT aggregate the two values. A buggy
+  // implementation reuses the same child writer without resetting it, which
+  // causes array fields to grow by appending and string fields to be
+  // concatenated.
+  {
+    // json string -> struct<FieldA:array<string>,Foo:string>
+    const std::string inputJson = R"JSON(
+      {
+        "FieldA": ["READ", "BLUE"],
+        "Foo": "Bar",
+        "fieldA": ["red", "blue"],
+        "foo": "bar"
+      }
+    )JSON";
+    auto values =
+        makeFlatVector<std::string>({inputJson, inputJson, inputJson});
+
+    auto fieldAVector = makeArrayVector<std::string>(
+        {{"READ", "BLUE"}, {"READ", "BLUE"}, {"READ", "BLUE"}});
+    auto fooVector = makeFlatVector<StringView>({"Bar", "Bar", "Bar"});
+    auto result = makeRowVector({"FieldA", "Foo"}, {fieldAVector, fooVector});
+
+    testFromJson(values, result);
+  }
+}
+
+TEST_F(FromJsonTest, caseSensitiveFields) {
+  const std::string inputJson = R"({"NAME":"first","age":"second"})";
+  auto input = makeFlatVector<std::string>({inputJson, inputJson, inputJson});
+  auto expected = makeRowVector(
+      {"name"},
+      {makeNullableFlatVector<StringView>(
+          {std::nullopt, std::nullopt, std::nullopt})});
+
+  testFromJson(input, expected);
+}
+
 } // namespace
 } // namespace bytedance::bolt::functions::sparksql::test