[fix](new_json_reader)fix core for new_json_reader (#45905)

fix coredump with new_json_reader if we read invalid data will core
which source from #43469

be/src/vec/exec/format/json/new_json_reader.cpp

@@ -1658,9 +1658,7 @@ Status NewJsonReader::_simdjson_write_data_to_column(simdjson::ondemand::value&
         data_serde = serde->get_nested_serdes()[0];
 
         // kNullType will put 1 into the Null map, so there is no need to push 0 for kNullType.
-        if (value.type() != simdjson::ondemand::json_type::null) {
-            nullable_column->get_null_map_data().push_back(0);
-        } else {
+        if (value.type() == simdjson::ondemand::json_type::null) {
             nullable_column->insert_default();
             *valid = true;
             return Status::OK();
@@ -1818,6 +1816,10 @@ Status NewJsonReader::_simdjson_write_data_to_column(simdjson::ondemand::value&
     } else {
         return Status::InternalError("Not support load to complex column.");
     }
+    //We need to finally set the nullmap of column_nullable to keep the size consistent with data_column
+    if (nullable_column && value.type() != simdjson::ondemand::json_type::null) {
+        nullable_column->get_null_map_data().push_back(0);
+    }
     *valid = true;
     return Status::OK();
 }

regression-test/data/load_p0/stream_load/load_json_parse_failed/test_error_json.json

@@ -0,0 +1 @@
+[{"id":3,"time":{"2024-06-27 22:52:40.41845+08:00"}}]

regression-test/suites/load_p0/stream_load/load_json_parse_failed/load_error_json.groovy

@@ -0,0 +1,53 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.codehaus.groovy.runtime.IOGroovyMethods
+
+suite ("load_error_json") {
+
+    sql """ DROP TABLE IF EXISTS test_error_json; """
+
+    sql """
+        CREATE TABLE `test_error_json` (
+                id INT,
+                time STRUCT<timestamp:DATETIME(6)>
+            ) ENGINE=OLAP
+            DISTRIBUTED BY HASH(`id`) BUCKETS AUTO
+            PROPERTIES (
+            "replication_allocation" = "tag.location.default: 1"
+        ); 
+        """
+
+    // load should return fail not core	
+    streamLoad {
+        table "test_error_json"
+        set 'format', 'json'
+        set 'read_json_by_line', 'true'
+        set 'strip_outer_array', 'true'
+        file 'test_error_json.json'
+
+        check { result, exception, startTime, endTime ->
+            if (exception != null) {
+                throw exception
+            }
+            log.info("Stream load result: ${result}".toString())
+            def json = parseJson(result)
+            assertEquals("fail", json.Status.toLowerCase())
+        }
+    }
+}
+