Skip to content

Commit 8d7b355

Browse files
committed
refactor(silo): do not overly rely on std::ostream when writing to http buffer
1 parent 51fbf91 commit 8d7b355

1 file changed

Lines changed: 14 additions & 2 deletions

File tree

src/silo/query_engine/exec_node/ndjson_sink.cpp

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
#include "silo/query_engine/exec_node/ndjson_sink.h"
22

3+
#include <ios>
4+
35
#include <arrow/acero/options.h>
46
#include <arrow/array.h>
57
#include <arrow/array/array_binary.h>
@@ -14,6 +16,16 @@ namespace silo::query_engine::exec_node {
1416

1517
namespace {
1618

19+
void writeChunked(std::ostream& output, std::string_view content) {
20+
const size_t chunk_size = 8192;
21+
for (size_t pos = 0; pos < content.size(); pos += chunk_size) {
22+
size_t remaining_size = content.size() - pos;
23+
size_t write_size = std::min(chunk_size, remaining_size);
24+
output.write(content.data() + pos, static_cast<std::streamsize>(write_size));
25+
output.flush(); // Flush after each small chunk
26+
}
27+
}
28+
1729
template <size_t BATCH_SIZE>
1830
struct BatchedStringStream {
1931
std::array<std::stringstream, BATCH_SIZE> streams;
@@ -26,7 +38,7 @@ struct BatchedStringStream {
2638

2739
void operator>>(std::ostream& output) {
2840
for (size_t i = 0; i < BATCH_SIZE; ++i) {
29-
output << streams[i].rdbuf();
41+
writeChunked(output, std::move(streams[i]).str());
3042
}
3143
}
3244
};
@@ -205,7 +217,7 @@ arrow::Result<arrow::acero::BackpressureMonitor*> createGenerator(
205217
auto node, arrow::acero::MakeExecNode(std::string{"sink"}, plan, {input}, options)
206218
);
207219
node->SetLabel("final sink of the plan");
208-
return backpressure_monitor;
220+
return arrow::Result<arrow::acero::BackpressureMonitor*>{backpressure_monitor};
209221
}
210222

211223
} // namespace silo::query_engine::exec_node

0 commit comments

Comments
 (0)