dataroaring commented on code in PR #48196:
URL: https://github.com/apache/doris/pull/48196#discussion_r1969575132
##########
be/src/vec/exec/format/json/new_json_reader.cpp:
##########
@@ -1174,6 +1173,41 @@ Status
NewJsonReader::_read_one_message(std::unique_ptr<uint8_t[]>* file_buf, si
}
return Status::OK();
}
+
+Status NewJsonReader::_read_one_message_from_pipe(std::unique_ptr<uint8_t[]>*
file_buf,
+ size_t* read_size) {
+ auto* stream_load_pipe =
dynamic_cast<io::StreamLoadPipe*>(_file_reader.get());
+
+ if (!stream_load_pipe->is_chunked_transfer()) {
+ return stream_load_pipe->read_one_message(file_buf, read_size);
+ }
+
+ // StreamLoadPipe::read_one_message only reads a portion of the data when
stream loading with a chunked transfer HTTP request.
+ // Need to read all the data before performing JSON parsing.
+ uint64_t buffer_size = 1024 * 1024;
+ std::vector<uint8_t> buf(buffer_size);
+
+ uint64_t cur_size = 0;
+ while (true) {
+ RETURN_IF_ERROR(stream_load_pipe->read_one_message(file_buf,
read_size));
+ if (*read_size == 0) {
+ break;
+ } else {
+ if (cur_size + (*read_size) > buf.size()) {
+ buffer_size = 2 * (cur_size + (*read_size));
+ buf.resize(buffer_size);
+ }
+ memcpy(buf.data() + cur_size, file_buf->get(), *read_size);
+ cur_size += *read_size;
+ }
+ }
+
+ file_buf->reset(new uint8_t[cur_size]);
+ memcpy(file_buf->get(), buf.data(), cur_size);
+ *read_size = cur_size;
+ return Status::OK();
+}
+
Review Comment:
Please add more tests for the function above in ut or regression test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
