fix(json): Bug with duplicate keys leading to incorrect writes (facebookincubator#13445)

Summary:

Currently when we cast json to row's if we encounter a duplicate key, we throw, however this throw isnt caught in the json cast row calling function , instead the throw is subsumed by the applyToselectedNoThrow call. This causes a problem because the previous write operation is not cancelled and is still valid, often leading to invalid data being read. This diff fix this bug. This bug is very evident when the key duplicated has a string value  , it might be harder to replicate if the key duplicated is against a complex type , however the fix should work for all types.

Reviewed By: kevinwilfong

Differential Revision: D75251964

Author: Krishna Pai

velox/functions/prestosql/tests/JsonCastTest.cpp

@@ -1347,6 +1347,27 @@ TEST_F(JsonCastTest, toRowDuplicateKey) {
   expected->setNull(2, true);
 
   testCast(data, expected, true /*try_cast*/);
+
+  // Duplicate keys with strings
+  jsonStrings = {
+      R"({"c0": "abc", "c1": "xyz", "c0": "mno"})",
+      R"({"c0": "123", "c1": "hjk"})",
+  };
+
+  expected = makeRowVector({
+      makeFlatVector<std::string>({"", "123"}),
+      makeFlatVector<std::string>({"", "hjk"}),
+  });
+  expected->setNull(0, true);
+
+  data = makeNullableFlatVector<std::string>(jsonStrings, JSON());
+  testCast(data, expected, true);
+
+  testThrow<std::string>(
+      JSON(),
+      ROW({"c0", "c1"}, {VARCHAR(), VARCHAR()}),
+      jsonStrings,
+      "Duplicate field: c0");
 }
 
 TEST_F(JsonCastTest, toRow) {

velox/functions/prestosql/types/JsonCastOperator.cpp

@@ -1229,21 +1229,25 @@ void JsonCastOperator::castFromJson(
     maxSize = std::max(maxSize, input.size());
   });
   paddedInput_.resize(maxSize + simdjson::SIMDJSON_PADDING);
-  context.applyToSelectedNoThrow(rows, [&](auto row) {
-    writer.setOffset(row);
-    if (inputVector->isNullAt(row)) {
-      writer.commitNull();
-      return;
-    }
-    auto& input = inputVector->valueAt(row);
-    memcpy(paddedInput_.data(), input.data(), input.size());
-    simdjson::padded_string_view paddedInput(
-        paddedInput_.data(), input.size(), paddedInput_.size());
-    if (auto error = castFromJsonOneRow<kind>(paddedInput, writer)) {
-      context.setVeloxExceptionError(row, errors_[error]);
-      writer.commitNull();
-    }
-  });
+  context.applyToSelectedNoThrow(
+      rows,
+      [&](auto row) INLINE_LAMBDA {
+        writer.setOffset(row);
+        if (inputVector->isNullAt(row)) {
+          writer.commitNull();
+          return;
+        }
+        auto& input = inputVector->valueAt(row);
+        memcpy(paddedInput_.data(), input.data(), input.size());
+        simdjson::padded_string_view paddedInput(
+            paddedInput_.data(), input.size(), paddedInput_.size());
+        if (auto error = castFromJsonOneRow<kind>(paddedInput, writer)) {
+          context.setVeloxExceptionError(row, errors_[error]);
+          writer.commitNull();
+        }
+      },
+      [&](vector_size_t row) INLINE_LAMBDA { writer.commitNull(); });
+
   writer.finish();
 }