Support load json-data into Doris by RoutineLoad or StreamLoad

.gitignore

@@ -6,6 +6,7 @@ output
 docs/.temp
 docs/.vuepress/dist
 docs/node_modules
+docs/build
 gensrc/build
 fe/target
 thirdparty/src

be/src/exec/CMakeLists.txt

@@ -93,6 +93,7 @@ set(EXEC_FILES
  parquet_scanner.cpp
  parquet_reader.cpp
  orc_scanner.cpp
+ json_scanner.cpp
  assert_num_rows_node.cpp
 )
 

be/src/exec/broker_reader.cpp

@@ -121,6 +121,11 @@ Status BrokerReader::open() {
  return Status::OK();
 }
 
+//not support
+Status BrokerReader::read_all(uint8_t** buf, size_t *length) {
+ return Status::NotSupported("Not support");
+}
+
 Status BrokerReader::read(uint8_t* buf, size_t* buf_len, bool* eof) {
  DCHECK_NE(*buf_len, 0);
  RETURN_IF_ERROR(readat(_cur_offset, (int64_t)*buf_len, (int64_t*)buf_len, buf));

be/src/exec/broker_reader.h

@@ -52,6 +52,7 @@ class BrokerReader : public FileReader {
  // Read 
  virtual Status read(uint8_t* buf, size_t* buf_len, bool* eof) override;
  virtual Status readat(int64_t position, int64_t nbytes, int64_t* bytes_read, void* out) override;
+ virtual Status read_all(uint8_t** buf, size_t *length) override;
  virtual int64_t size() override;
  virtual Status seek(int64_t position) override;
  virtual Status tell(int64_t* position) override;

be/src/exec/broker_scan_node.cpp

@@ -27,6 +27,7 @@
 #include "exec/broker_scanner.h"
 #include "exec/parquet_scanner.h"
 #include "exec/orc_scanner.h"
+#include "exec/json_scanner.h"
 #include "exprs/expr.h"
 #include "util/runtime_profile.h"
 
@@ -289,6 +290,14 @@ std::unique_ptr<BaseScanner> BrokerScanNode::create_scanner(const TBrokerScanRan
  scan_range.broker_addresses,
  counter);
  break;
+ case TFileFormatType::FORMAT_JSON:
+ scan = new JsonScanner(_runtime_state,
+ runtime_profile(),
+ scan_range.params,
+ scan_range.ranges,
+ scan_range.broker_addresses,
+ counter);
+ break;
  default:
  scan = new BrokerScanner(
  _runtime_state,

be/src/exec/broker_scanner.cpp

@@ -34,6 +34,7 @@
 #include "exec/local_file_reader.h"
 #include "exec/broker_reader.h"
 #include "exec/decompressor.h"
+#include "json_scanner.h"
 #include "util/utf8_check.h"
 
 namespace doris {
@@ -180,6 +181,7 @@ Status BrokerScanner::create_decompressor(TFileFormatType::type type) {
  CompressType compress_type;
  switch (type) {
  case TFileFormatType::FORMAT_CSV_PLAIN:
+ case TFileFormatType::FORMAT_JSON:
  compress_type = CompressType::UNCOMPRESSED;
  break;
  case TFileFormatType::FORMAT_CSV_GZ:

be/src/exec/file_reader.h

@@ -34,6 +34,16 @@ class FileReader {
  // is set to zero.
  virtual Status read(uint8_t* buf, size_t* buf_len, bool* eof) = 0;
  virtual Status readat(int64_t position, int64_t nbytes, int64_t* bytes_read, void* out) = 0;
+
+ /**
+ * if read eof then return Status::OK and length is set 0 and buf is set NULL,
+ * other return readed bytes.
+ *
+ * !! Important !!
+ * the buf must be deleted by user, otherwise leak memory
+ * !! Important !!
+ */
+ virtual Status read_all(uint8_t** buf, size_t *length) = 0;
  virtual int64_t size () = 0;
  virtual Status seek(int64_t position) = 0;
  virtual Status tell(int64_t* position) = 0;