This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-1.2-lts by this push:
new 1d26fea38e [fix](vec)crashing caused by parallel output file (#17384)
(#17477)
1d26fea38e is described below
commit 1d26fea38e330caff452cf8d3ef653f2a3287679
Author: Jerry Hu <[email protected]>
AuthorDate: Tue Mar 7 08:46:56 2023 +0800
[fix](vec)crashing caused by parallel output file (#17384) (#17477)
cherry-pick #17384
---
be/src/vec/sink/vdata_stream_sender.cpp | 19 ++++++-
be/src/vec/sink/vdata_stream_sender.h | 1 +
be/src/vec/sink/vresult_file_sink.cpp | 61 +++++++---------------
be/src/vec/sink/vresult_file_sink.h | 7 ++-
.../suites/export_p0/test_outfile.groovy | 34 ++++++++++++
5 files changed, 77 insertions(+), 45 deletions(-)
diff --git a/be/src/vec/sink/vdata_stream_sender.cpp
b/be/src/vec/sink/vdata_stream_sender.cpp
index 00dcb68756..9768f61244 100644
--- a/be/src/vec/sink/vdata_stream_sender.cpp
+++ b/be/src/vec/sink/vdata_stream_sender.cpp
@@ -303,7 +303,6 @@ VDataStreamSender::VDataStreamSender(ObjectPool* pool, int
sender_id, const RowD
sink.output_partition.type ==
TPartitionType::BUCKET_SHFFULE_HASH_PARTITIONED);
//
std::map<int64_t, int64_t> fragment_id_to_channel_index;
-
for (int i = 0; i < destinations.size(); ++i) {
// Select first dest as transfer chain.
bool is_transfer_chain = (i == 0);
@@ -326,6 +325,7 @@ VDataStreamSender::VDataStreamSender(ObjectPool* pool, int
sender_id, const RowD
}
VDataStreamSender::VDataStreamSender(ObjectPool* pool, int sender_id, const
RowDescriptor& row_desc,
+ PlanNodeId dest_node_id,
const
std::vector<TPlanFragmentDestination>& destinations,
int per_channel_buffer_size,
bool
send_query_statistics_with_every_batch)
@@ -333,6 +333,7 @@ VDataStreamSender::VDataStreamSender(ObjectPool* pool, int
sender_id, const RowD
_pool(pool),
_row_desc(row_desc),
_current_channel_idx(0),
+ _part_type(TPartitionType::UNPARTITIONED),
_ignore_not_found(true),
_cur_pb_block(&_pb_block1),
_profile(nullptr),
@@ -346,8 +347,22 @@ VDataStreamSender::VDataStreamSender(ObjectPool* pool, int
sender_id, const RowD
_split_block_distribute_by_channel_timer(nullptr),
_blocks_sent_counter(nullptr),
_local_bytes_send_counter(nullptr),
- _dest_node_id(0) {
+ _dest_node_id(dest_node_id) {
_name = "VDataStreamSender";
+ std::map<int64_t, int64_t> fragment_id_to_channel_index;
+ for (int i = 0; i < destinations.size(); ++i) {
+ const auto& fragment_instance_id =
destinations[i].fragment_instance_id;
+ if (fragment_id_to_channel_index.find(fragment_instance_id.lo) ==
+ fragment_id_to_channel_index.end()) {
+ _channel_shared_ptrs.emplace_back(
+ new Channel(this, row_desc, destinations[i].brpc_server,
fragment_instance_id,
+ _dest_node_id, per_channel_buffer_size, false,
+ send_query_statistics_with_every_batch));
+ }
+ fragment_id_to_channel_index.emplace(fragment_instance_id.lo,
+ _channel_shared_ptrs.size() - 1);
+ _channels.push_back(_channel_shared_ptrs.back().get());
+ }
}
VDataStreamSender::VDataStreamSender(ObjectPool* pool, const RowDescriptor&
row_desc,
diff --git a/be/src/vec/sink/vdata_stream_sender.h
b/be/src/vec/sink/vdata_stream_sender.h
index efa5a5bad1..3e17170367 100644
--- a/be/src/vec/sink/vdata_stream_sender.h
+++ b/be/src/vec/sink/vdata_stream_sender.h
@@ -50,6 +50,7 @@ public:
int per_channel_buffer_size, bool
send_query_statistics_with_every_batch);
VDataStreamSender(ObjectPool* pool, int sender_id, const RowDescriptor&
row_desc,
+ PlanNodeId dest_node_id,
const std::vector<TPlanFragmentDestination>&
destinations,
int per_channel_buffer_size, bool
send_query_statistics_with_every_batch);
diff --git a/be/src/vec/sink/vresult_file_sink.cpp
b/be/src/vec/sink/vresult_file_sink.cpp
index d1d2cfaa72..bceaa910f5 100644
--- a/be/src/vec/sink/vresult_file_sink.cpp
+++ b/be/src/vec/sink/vresult_file_sink.cpp
@@ -34,9 +34,7 @@ VResultFileSink::VResultFileSink(ObjectPool* pool, const
RowDescriptor& row_desc
const TResultFileSink& sink, int
per_channel_buffer_size,
bool send_query_statistics_with_every_batch,
const std::vector<TExpr>& t_output_expr)
- : VDataStreamSender(pool, row_desc, per_channel_buffer_size,
- send_query_statistics_with_every_batch),
- _t_output_expr(t_output_expr) {
+ : _t_output_expr(t_output_expr), _row_desc(row_desc) {
CHECK(sink.__isset.file_options);
_file_opts.reset(new ResultFileOptions(sink.file_options));
CHECK(sink.__isset.storage_backend_type);
@@ -55,20 +53,18 @@ VResultFileSink::VResultFileSink(ObjectPool* pool, int
sender_id, const RowDescr
int per_channel_buffer_size,
bool send_query_statistics_with_every_batch,
const std::vector<TExpr>& t_output_expr,
DescriptorTbl& descs)
- : VDataStreamSender(pool, sender_id, row_desc, destinations,
per_channel_buffer_size,
- send_query_statistics_with_every_batch),
- _t_output_expr(t_output_expr),
-
_output_row_descriptor(descs.get_tuple_descriptor(sink.output_tuple_id), false)
{
+ : _t_output_expr(t_output_expr),
+
_output_row_descriptor(descs.get_tuple_descriptor(sink.output_tuple_id), false),
+ _row_desc(row_desc) {
CHECK(sink.__isset.file_options);
_file_opts.reset(new ResultFileOptions(sink.file_options));
CHECK(sink.__isset.storage_backend_type);
_storage_type = sink.storage_backend_type;
_is_top_sink = false;
- DCHECK_EQ(destinations.size(), 1);
- _channel_shared_ptrs.emplace_back(new Channel(
- this, _output_row_descriptor, destinations[0].brpc_server,
- destinations[0].fragment_instance_id, sink.dest_node_id,
_buf_size, true, true));
- _channels.push_back(_channel_shared_ptrs.back().get());
+ CHECK_EQ(destinations.size(), 1);
+ _stream_sender.reset(new VDataStreamSender(pool, sender_id, row_desc,
sink.dest_node_id,
+ destinations,
per_channel_buffer_size,
+
send_query_statistics_with_every_batch));
_name = "VResultFileSink";
//for impl csv_with_name and csv_with_names_and_types
@@ -77,6 +73,9 @@ VResultFileSink::VResultFileSink(ObjectPool* pool, int
sender_id, const RowDescr
}
Status VResultFileSink::init(const TDataSink& tsink) {
+ if (!_is_top_sink) {
+ RETURN_IF_ERROR(_stream_sender->init(tsink));
+ }
return Status::OK();
}
@@ -111,30 +110,24 @@ Status VResultFileSink::prepare(RuntimeState* state) {
_output_row_descriptor));
} else {
// init channel
- _profile = _pool->add(new RuntimeProfile(title.str()));
- _state = state;
- _serialize_batch_timer = ADD_TIMER(profile(), "SerializeBatchTime");
- _bytes_sent_counter = ADD_COUNTER(profile(), "BytesSent",
TUnit::BYTES);
- _local_bytes_send_counter = ADD_COUNTER(profile(), "LocalBytesSent",
TUnit::BYTES);
- _uncompressed_bytes_counter =
- ADD_COUNTER(profile(), "UncompressedRowBatchSize",
TUnit::BYTES);
- // create writer
_output_block.reset(new
Block(_output_row_descriptor.tuple_descriptors()[0]->slots(), 1));
_writer.reset(new (std::nothrow) VFileResultWriter(
_file_opts.get(), _storage_type,
state->fragment_instance_id(), _output_vexpr_ctxs,
_profile, nullptr, _output_block.get(),
state->return_object_data_as_binary(),
_output_row_descriptor));
+ RETURN_IF_ERROR(_stream_sender->prepare(state));
+ _profile->add_child(_stream_sender->profile(), true, nullptr);
}
_writer->set_header_info(_header_type, _header);
RETURN_IF_ERROR(_writer->init(state));
- for (int i = 0; i < _channels.size(); ++i) {
- RETURN_IF_ERROR(_channels[i]->init(state));
- }
return Status::OK();
}
Status VResultFileSink::open(RuntimeState* state) {
START_AND_SCOPE_SPAN(state->get_tracer(), span, "VResultFileSink::open");
+ if (!_is_top_sink) {
+ RETURN_IF_ERROR(_stream_sender->open(state));
+ }
return VExpr::open(_output_vexpr_ctxs, state);
}
@@ -174,25 +167,9 @@ Status VResultFileSink::close(RuntimeState* state, Status
exec_status) {
state->fragment_instance_id());
} else {
if (final_status.ok()) {
- RETURN_IF_ERROR(serialize_block(_output_block.get(),
_cur_pb_block, _channels.size()));
- for (auto channel : _channels) {
- RETURN_IF_ERROR(channel->send_block(_cur_pb_block));
- }
- }
- Status final_st = Status::OK();
- for (int i = 0; i < _channels.size(); ++i) {
- Status st = _channels[i]->close(state);
- if (!st.ok() && final_st.ok()) {
- final_st = st;
- }
- }
- // wait all channels to finish
- for (int i = 0; i < _channels.size(); ++i) {
- Status st = _channels[i]->close_wait(state);
- if (!st.ok() && final_st.ok()) {
- final_st = st;
- }
+ RETURN_IF_ERROR(_stream_sender->send(state, _output_block.get()));
}
+ RETURN_IF_ERROR(_stream_sender->close(state, final_status));
_output_block->clear();
}
@@ -206,7 +183,7 @@ void
VResultFileSink::set_query_statistics(std::shared_ptr<QueryStatistics> stat
if (_is_top_sink) {
_sender->set_query_statistics(statistics);
} else {
- _query_statistics = statistics;
+ _stream_sender->set_query_statistics(statistics);
}
}
diff --git a/be/src/vec/sink/vresult_file_sink.h
b/be/src/vec/sink/vresult_file_sink.h
index c5c509caf7..be3c2926bb 100644
--- a/be/src/vec/sink/vresult_file_sink.h
+++ b/be/src/vec/sink/vresult_file_sink.h
@@ -24,7 +24,7 @@ namespace doris {
namespace vectorized {
class VResultWriter;
-class VResultFileSink : public VDataStreamSender {
+class VResultFileSink : public DataSink {
public:
VResultFileSink(ObjectPool* pool, const RowDescriptor& row_desc, const
TResultFileSink& sink,
int per_channel_buffer_size, bool
send_query_statistics_with_every_batch,
@@ -48,6 +48,7 @@ public:
RuntimeProfile* profile() override { return _profile; }
void set_query_statistics(std::shared_ptr<QueryStatistics> statistics)
override;
+ const RowDescriptor& row_desc() const { return _row_desc; }
private:
Status prepare_exprs(RuntimeState* state);
@@ -62,11 +63,15 @@ private:
std::unique_ptr<Block> _output_block = nullptr;
std::shared_ptr<BufferControlBlock> _sender;
+ std::unique_ptr<VDataStreamSender> _stream_sender;
std::shared_ptr<VResultWriter> _writer;
int _buf_size = 1024; // Allocated from _pool
bool _is_top_sink = true;
std::string _header;
std::string _header_type;
+
+ RowDescriptor _row_desc;
+ RuntimeProfile* _profile;
};
} // namespace vectorized
} // namespace doris
diff --git a/regression-test/suites/export_p0/test_outfile.groovy
b/regression-test/suites/export_p0/test_outfile.groovy
index 4fa3350268..d2ceb46c5e 100644
--- a/regression-test/suites/export_p0/test_outfile.groovy
+++ b/regression-test/suites/export_p0/test_outfile.groovy
@@ -186,4 +186,38 @@ suite("test_outfile") {
path.delete();
}
}
+
+ // test parallel output
+ try {
+ File path = new File(outFilePath)
+ if (!path.exists()) {
+ assert path.mkdirs()
+ } else {
+ throw new IllegalStateException("""${outFilePath} already exists!
""")
+ }
+ sql """drop table if exists select_into_file"""
+ sql """CREATE TABLE `select_into_file` (
+ `id` int,
+ `name` varchar(30)
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`id`)
+ DISTRIBUTED BY HASH(`id`) BUCKETS 2
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1"
+ );"""
+ sql """insert into select_into_file values(1, "b"),(2, "z"),(3, "a"),
+ (4, "c"), (5, "睿"), (6, "多"), (7, "丝"), (8, "test"),
+ (100, "aa"), (111, "bb"), (123, "cc"), (222, "dd");"""
+ sql "set enable_parallel_outfile = true;"
+ sql """select * from select_into_file into outfile
"file://${outFilePath}/" properties("success_file_name" = "SUCCESS");"""
+ } finally {
+ try_sql("DROP TABLE IF EXISTS select_into_file")
+ File path = new File(outFilePath)
+ if (path.exists()) {
+ for (File f: path.listFiles()) {
+ f.delete();
+ }
+ path.delete();
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
