[feature]add vectorized vjson_scanner

be/src/exec/base_scanner.cpp

@@ -156,7 +156,6 @@ Status BaseScanner::init_expr_ctxes() {
  RETURN_IF_ERROR(ctx->open(_state));
  _dest_expr_ctx.emplace_back(ctx);
  }
-
  if (has_slot_id_map) {
  auto it = _params.dest_sid_to_src_sid_without_trans.find(slot_desc->id());
  if (it == std::end(_params.dest_sid_to_src_sid_without_trans)) {
@@ -279,6 +278,58 @@ Status BaseScanner::_fill_dest_tuple(Tuple* dest_tuple, MemPool* mem_pool) {
  return Status::OK();
 }
 
+Status BaseScanner::filter_block(vectorized::Block* temp_block, size_t slot_num) {
+ // filter block
+ if (!_vpre_filter_ctxs.empty()) {
+ for (auto _vpre_filter_ctx : _vpre_filter_ctxs) {
+ auto old_rows = temp_block->rows();
+ RETURN_IF_ERROR(
+ vectorized::VExprContext::filter_block(_vpre_filter_ctx, temp_block, slot_num));
+ _counter->num_rows_unselected += old_rows - temp_block->rows();
+ }
+ }
+ return Status::OK();
+}
+
+Status BaseScanner::execute_exprs(vectorized::Block* output_block, vectorized::Block* temp_block) {
+ // Do vectorized expr here
+ Status status;
+ if (!_dest_vexpr_ctx.empty()) {
+ *output_block = vectorized::VExprContext::get_output_block_after_execute_exprs(
+ _dest_vexpr_ctx, *temp_block, status);
+ if (UNLIKELY(output_block->rows() == 0)) {
+ return status;
+ }
+ }
+
+ return Status::OK();
+}
+
+Status BaseScanner::fill_dest_block(vectorized::Block* dest_block,
+ std::vector<vectorized::MutableColumnPtr>& columns) {
+ if (columns.empty() || columns[0]->size() == 0) {
+ return Status::OK();
+ }
+
+ std::unique_ptr<vectorized::Block> temp_block(new vectorized::Block());
+ auto n_columns = 0;
+ for (const auto slot_desc : _src_slot_descs) {
+ temp_block->insert(vectorized::ColumnWithTypeAndName(std::move(columns[n_columns++]),
+ slot_desc->get_data_type_ptr(),
+ slot_desc->col_name()));
+ }
+
+ RETURN_IF_ERROR(BaseScanner::filter_block(temp_block.get(), _dest_tuple_desc->slots().size()));
+
+ if (_dest_vexpr_ctx.empty()) {
+ *dest_block = *temp_block;
+ } else {
+ RETURN_IF_ERROR(BaseScanner::execute_exprs(dest_block, temp_block.get()));
+ }
+
+ return Status::OK();
+}
+
 void BaseScanner::fill_slots_of_columns_from_path(
  int start, const std::vector<std::string>& columns_from_path) {
  // values of columns from path can not be null

be/src/exec/base_scanner.h

@@ -22,6 +22,7 @@
 #include "runtime/tuple.h"
 #include "util/runtime_profile.h"
 #include "vec/exprs/vexpr.h"
+#include "vec/exprs/vexpr_context.h"
 
 namespace doris {
 
@@ -75,11 +76,18 @@ class BaseScanner {
  virtual void close() = 0;
  Status fill_dest_tuple(Tuple* dest_tuple, MemPool* mem_pool, bool* fill_tuple);
 
+ Status fill_dest_block(vectorized::Block* dest_block,
+ std::vector<vectorized::MutableColumnPtr>& columns);
+
  void fill_slots_of_columns_from_path(int start,
  const std::vector<std::string>& columns_from_path);
 
  void free_expr_local_allocations();
 
+ Status filter_block(vectorized::Block* temp_block, size_t slot_num);
+
+ Status execute_exprs(vectorized::Block* output_block, vectorized::Block* temp_block);
+
 protected:
  RuntimeState* _state;
  const TBrokerScanRangeParams& _params;

be/src/exec/broker_scan_node.cpp

@@ -32,6 +32,7 @@
 #include "util/runtime_profile.h"
 #include "util/thread.h"
 #include "vec/exec/vbroker_scanner.h"
+#include "vec/exec/vjson_scanner.h"
 
 namespace doris {
 
@@ -234,9 +235,15 @@ std::unique_ptr<BaseScanner> BrokerScanNode::create_scanner(const TBrokerScanRan
  counter);
  break;
  case TFileFormatType::FORMAT_JSON:
- scan = new JsonScanner(_runtime_state, runtime_profile(), scan_range.params,
- scan_range.ranges, scan_range.broker_addresses, _pre_filter_texprs,
- counter);
+ if (_vectorized) {
+ scan = new vectorized::VJsonScanner(
+ _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,
+ _pre_filter_texprs, counter);
+ }
  break;
  default:
  if (_vectorized) {

be/src/exec/json_scanner.cpp

@@ -112,14 +112,17 @@ Status JsonScanner::open_next_reader() {
  _scanner_eof = true;
  return Status::OK();
  }
+ RETURN_IF_ERROR(open_based_reader());
+ RETURN_IF_ERROR(open_json_reader());
+ _next_range++;
+ return Status::OK();
+}
 
+Status JsonScanner::open_based_reader() {
  RETURN_IF_ERROR(open_file_reader());
  if (_read_json_by_line) {
  RETURN_IF_ERROR(open_line_reader());
  }
- RETURN_IF_ERROR(open_json_reader());
- _next_range++;
-
  return Status::OK();
 }
 
@@ -215,6 +218,25 @@ Status JsonScanner::open_json_reader() {
  bool num_as_string = false;
  bool fuzzy_parse = false;
 
+ RETURN_IF_ERROR(
+ get_range_params(jsonpath, json_root, strip_outer_array, num_as_string, fuzzy_parse));
+ if (_read_json_by_line) {
+ _cur_json_reader =
+ new JsonReader(_state, _counter, _profile, strip_outer_array, num_as_string,
+ fuzzy_parse, &_scanner_eof, nullptr, _cur_line_reader);
+ } else {
+ _cur_json_reader =
+ new JsonReader(_state, _counter, _profile, strip_outer_array, num_as_string,
+ fuzzy_parse, &_scanner_eof, _cur_file_reader);
+ }
+
+ RETURN_IF_ERROR(_cur_json_reader->init(jsonpath, json_root));
+ return Status::OK();
+}
+
+Status JsonScanner::get_range_params(std::string& jsonpath, std::string& json_root,
+ bool& strip_outer_array, bool& num_as_string,
+ bool& fuzzy_parse) {
  const TBrokerRangeDesc& range = _ranges[_next_range];
 
  if (range.__isset.jsonpaths) {
@@ -232,17 +254,6 @@ Status JsonScanner::open_json_reader() {
  if (range.__isset.fuzzy_parse) {
  fuzzy_parse = range.fuzzy_parse;
  }
- if (_read_json_by_line) {
- _cur_json_reader =
- new JsonReader(_state, _counter, _profile, strip_outer_array, num_as_string,
- fuzzy_parse, &_scanner_eof, nullptr, _cur_line_reader);
- } else {
- _cur_json_reader =
- new JsonReader(_state, _counter, _profile, strip_outer_array, num_as_string,
- fuzzy_parse, &_scanner_eof, _cur_file_reader);
- }
-
- RETURN_IF_ERROR(_cur_json_reader->init(jsonpath, json_root));
  return Status::OK();
 }
 
@@ -308,14 +319,8 @@ JsonReader::~JsonReader() {
 }
 
 Status JsonReader::init(const std::string& jsonpath, const std::string& json_root) {
- // parse jsonpath
- if (!jsonpath.empty()) {
- Status st = _generate_json_paths(jsonpath, &_parsed_jsonpaths);
- RETURN_IF_ERROR(st);
- }
- if (!json_root.empty()) {
- JsonFunctions::parse_json_paths(json_root, &_parsed_json_root);
- }
+ // generate _parsed_jsonpaths and _parsed_json_root
+ RETURN_IF_ERROR(_parse_jsonpath_and_json_root(jsonpath, json_root));
 
  //improve performance
  if (_parsed_jsonpaths.empty()) { // input is a simple json-string
@@ -330,6 +335,18 @@ Status JsonReader::init(const std::string& jsonpath, const std::string& json_roo
  return Status::OK();
 }
 
+Status JsonReader::_parse_jsonpath_and_json_root(const std::string& jsonpath,
+ const std::string& json_root) {
+ // parse jsonpath
+ if (!jsonpath.empty()) {
+ RETURN_IF_ERROR(_generate_json_paths(jsonpath, &_parsed_jsonpaths));
+ }
+ if (!json_root.empty()) {
+ JsonFunctions::parse_json_paths(json_root, &_parsed_json_root);
+ }
+ return Status::OK();
+}
+
 Status JsonReader::_generate_json_paths(const std::string& jsonpath,
  std::vector<std::vector<JsonPath>>* vect) {
  rapidjson::Document jsonpaths_doc;

be/src/exec/json_scanner.h

@@ -67,13 +67,17 @@ class JsonScanner : public BaseScanner {
  // Close this scanner
  void close() override;
 
-private:
+protected:
  Status open_file_reader();
  Status open_line_reader();
  Status open_json_reader();
  Status open_next_reader();
 
-private:
+ Status open_based_reader();
+ Status get_range_params(std::string& jsonpath, std::string& json_root, bool& strip_outer_array,
+ bool& num_as_string, bool& fuzzy_parse);
+
+protected:
  const std::vector<TBrokerRangeDesc>& _ranges;
  const std::vector<TNetworkAddress>& _broker_addresses;
 
@@ -129,7 +133,7 @@ class JsonReader {
  Status read_json_row(Tuple* tuple, const std::vector<SlotDescriptor*>& slot_descs,
  MemPool* tuple_pool, bool* is_empty_row, bool* eof);
 
-private:
+protected:
  Status (JsonReader::*_handle_json_callback)(Tuple* tuple,
  const std::vector<SlotDescriptor*>& slot_descs,
  MemPool* tuple_pool, bool* is_empty_row, bool* eof);
@@ -158,8 +162,9 @@ class JsonReader {
  void _close();
  Status _generate_json_paths(const std::string& jsonpath,
  std::vector<std::vector<JsonPath>>* vect);
+ Status _parse_jsonpath_and_json_root(const std::string& jsonpath, const std::string& json_root);
 
-private:
+protected:
  int _next_line;
  int _total_lines;
  RuntimeState* _state;

be/src/vec/CMakeLists.txt

@@ -101,6 +101,7 @@ set(VEC_FILES
  exec/vtable_function_node.cpp
  exec/vbroker_scan_node.cpp
  exec/vbroker_scanner.cpp
+ exec/vjson_scanner.cpp
  exec/join/vhash_join_node.cpp
  exprs/vectorized_agg_fn.cpp
  exprs/vectorized_fn_call.cpp