fix p0

Author: csun5285

be/src/http/action/stream_load.cpp

@@ -429,6 +429,7 @@ Status StreamLoadAction::_process_put(HttpRequest* http_req,
         if (ctx->is_chunked_transfer) {
             pipe = std::make_shared<io::StreamLoadPipe>(
                     io::kMaxPipeBufferedBytes /* max_buffered_bytes */);
+            pipe->set_is_chunked_transfer(true);
         } else {
             pipe = std::make_shared<io::StreamLoadPipe>(
                     io::kMaxPipeBufferedBytes /* max_buffered_bytes */,

be/src/io/fs/stream_load_pipe.h

@@ -85,7 +85,11 @@ class StreamLoadPipe : public MessageBodySink, public FileReader {
 
     size_t current_capacity();
 
-    int64_t total_length() const { return _total_length; }
+    bool is_chunked_transfer() const { return _is_chunked_transfer; }
+
+    void set_is_chunked_transfer(bool is_chunked_transfer) {
+        _is_chunked_transfer = is_chunked_transfer;
+    }
 
 protected:
     Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
@@ -121,6 +125,10 @@ class StreamLoadPipe : public MessageBodySink, public FileReader {
 
     // no use, only for compatibility with the `Path` interface
     Path _path = "";
+
+    // When importing JSON data and using chunked transfer encoding,
+    // the data needs to be completely read before it can be parsed.
+    bool _is_chunked_transfer = false;
 };
 } // namespace io
 } // namespace doris

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

@@ -1178,12 +1178,10 @@ Status NewJsonReader::_read_one_message_from_pipe(std::unique_ptr<uint8_t[]>* fi
                                                   size_t* read_size) {
     auto* stream_load_pipe = dynamic_cast<io::StreamLoadPipe*>(_file_reader.get());
 
-    // With the total length known, the complete data can now be retrieved
-    if (stream_load_pipe->total_length() != -1) {
+    if (!stream_load_pipe->is_chunked_transfer()) {
         return stream_load_pipe->read_one_message(file_buf, read_size);
     }
 
-    // The total_length is -1, which means that the data arrives in a stream and the length is unknown.
     // 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;