[Enhancement](stream-load-json) using simdjson to parse json

be/CMakeLists.txt

@@ -328,6 +328,9 @@ set_target_properties(aws-s2n PROPERTIES IMPORTED_LOCATION ${THIRDPARTY_DIR}/lib
 add_library(minizip STATIC IMPORTED)
 set_target_properties(minizip PROPERTIES IMPORTED_LOCATION ${THIRDPARTY_DIR}/lib64/libminizip.a)
 
+add_library(simdjson STATIC IMPORTED)
+set_target_properties(simdjson PROPERTIES IMPORTED_LOCATION ${THIRDPARTY_DIR}/lib64/libsimdjson.a)
+
 add_library(idn STATIC IMPORTED)
 set_target_properties(idn PROPERTIES IMPORTED_LOCATION ${THIRDPARTY_DIR}/lib64/libidn.a)
 
@@ -701,6 +704,7 @@ set(COMMON_THIRDPARTY
     hdfs3
     xml2
     lzma
+    simdjson
 )
 
 if (${MAKE_TEST} STREQUAL "ON")

be/src/common/config.h

@@ -827,6 +827,7 @@ CONF_Validator(file_cache_type, [](const std::string config) -> bool {
 CONF_Int32(s3_transfer_executor_pool_size, "2");
 
 CONF_Bool(enable_time_lut, "true");
+CONF_Bool(enable_simdjson_reader, "true");
 
 // number of s3 scanner thread pool size
 CONF_Int32(doris_remote_scanner_thread_pool_thread_num, "16");

be/src/exec/broker_scan_node.cpp

@@ -243,9 +243,15 @@ std::unique_ptr<BaseScanner> BrokerScanNode::create_scanner(const TBrokerScanRan
         break;
     case TFileFormatType::FORMAT_JSON:
         if (_vectorized) {
-            scan = new vectorized::VJsonScanner(
-                    _runtime_state, runtime_profile(), scan_range.params, scan_range.ranges,
-                    scan_range.broker_addresses, _pre_filter_texprs, counter);
+            if (config::enable_simdjson_reader) {
+                scan = new vectorized::VJsonScanner<vectorized::VSIMDJsonReader>(
+                        _runtime_state, runtime_profile(), scan_range.params, scan_range.ranges,
+                        scan_range.broker_addresses, _pre_filter_texprs, counter);
+            } else {
+                scan = new vectorized::VJsonScanner<vectorized::VJsonReader>(
+                        _runtime_state, runtime_profile(), scan_range.params, scan_range.ranges,
+                        scan_range.broker_addresses, _pre_filter_texprs, counter);
+            }
         } else {
             scan = new JsonScanner(_runtime_state, runtime_profile(), scan_range.params,
                                    scan_range.ranges, scan_range.broker_addresses,

be/src/exprs/json_functions.h

@@ -65,6 +65,26 @@ struct JsonPath {
         return ss.str();
     }
 
+    std::string to_simdjson_pointer(bool* valid) const {
+        std::stringstream ss;
+        if (!is_valid) {
+            *valid = false;
+            return "";
+        }
+        ss << "/";
+        if (!key.empty()) {
+            ss << key;
+        }
+        if (idx == -2) {
+            // not support [*]
+            *valid = false;
+            return "";
+        } else if (idx > -1) {
+            ss << "/" << idx;
+        }
+        return ss.str();
+    }
+
     std::string debug_string() const {
         return fmt::format("key:{}, idx:{}, valid:{}", key, idx, is_valid);
     }

be/src/vec/core/block.cpp

@@ -254,6 +254,22 @@ const ColumnWithTypeAndName& Block::get_by_name(const std::string& name) const {
     return data[it->second];
 }
 
+ColumnWithTypeAndName* Block::try_get_by_name(const std::string& name) {
+    auto it = index_by_name.find(name);
+    if (index_by_name.end() == it) {
+        return nullptr;
+    }
+    return &data[it->second];
+}
+
+const ColumnWithTypeAndName* Block::try_get_by_name(const std::string& name) const {
+    auto it = index_by_name.find(name);
+    if (index_by_name.end() == it) {
+        return nullptr;
+    }
+    return &data[it->second];
+}
+
 bool Block::has(const std::string& name) const {
     return index_by_name.end() != index_by_name.find(name);
 }

be/src/vec/core/block.h

@@ -150,6 +150,10 @@ class Block {
     ColumnWithTypeAndName& get_by_name(const std::string& name);
     const ColumnWithTypeAndName& get_by_name(const std::string& name) const;
 
+    // return nullptr when no such column name
+    ColumnWithTypeAndName* try_get_by_name(const std::string& name);
+    const ColumnWithTypeAndName* try_get_by_name(const std::string& name) const;
+
     Container::iterator begin() { return data.begin(); }
     Container::iterator end() { return data.end(); }
     Container::const_iterator begin() const { return data.begin(); }

be/src/vec/exec/vbroker_scan_node.cpp

@@ -326,9 +326,15 @@ std::unique_ptr<BaseScanner> VBrokerScanNode::create_scanner(const TBrokerScanRa
                                            _pre_filter_texprs, counter);
         break;
     case TFileFormatType::FORMAT_JSON:
-        scan = new vectorized::VJsonScanner(_runtime_state, runtime_profile(), scan_range.params,
-                                            scan_range.ranges, scan_range.broker_addresses,
-                                            _pre_filter_texprs, counter);
+        if (config::enable_simdjson_reader) {
+            scan = new vectorized::VJsonScanner<vectorized::VSIMDJsonReader>(
+                    _runtime_state, runtime_profile(), scan_range.params, scan_range.ranges,
+                    scan_range.broker_addresses, _pre_filter_texprs, counter);
+        } else {
+            scan = new vectorized::VJsonScanner<vectorized::VJsonReader>(
+                    _runtime_state, runtime_profile(), scan_range.params, scan_range.ranges,
+                    scan_range.broker_addresses, _pre_filter_texprs, counter);
+        }
         break;
     default:
         scan = new vectorized::VBrokerScanner(_runtime_state, runtime_profile(), scan_range.params,