yiguolei commented on code in PR #10013: URL: https://github.com/apache/incubator-doris/pull/10013#discussion_r893220089
########## be/src/vec/sink/vresult_file_sink.cpp: ########## @@ -0,0 +1,216 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "vec/sink/vresult_file_sink.h" + +#include "common/config.h" +#include "exprs/expr.h" +#include "runtime/buffer_control_block.h" +#include "runtime/exec_env.h" +#include "runtime/file_result_writer.h" +#include "runtime/result_buffer_mgr.h" +#include "runtime/result_file_sink.h" +#include "runtime/row_batch.h" +#include "runtime/runtime_state.h" +#include "util/uid_util.h" +#include "vec/runtime/vfile_result_writer.h" + +namespace doris::vectorized { + +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) { + 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 = true; + + _name = "VResultFileSink"; + //for impl csv_with_name and csv_with_names_and_types + _header_type = sink.header_type; + _header = sink.header; +} + +VResultFileSink::VResultFileSink(ObjectPool* pool, int sender_id, const RowDescriptor& row_desc, + const TResultFileSink& sink, + const std::vector<TPlanFragmentDestination>& destinations, + 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) { + 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()); + + _name = "VResultFileSink"; + //for impl csv_with_name and csv_with_names_and_types + _header_type = sink.header_type; + _header = sink.header; +} + +VResultFileSink::~VResultFileSink() { + if (_output_block != nullptr) { + delete _output_block; + } +} + +Status VResultFileSink::init(const TDataSink& tsink) { + return Status::OK(); +} + +Status VResultFileSink::prepare_exprs(RuntimeState* state) { + // From the thrift expressions create the real exprs. + RETURN_IF_ERROR(Expr::create_expr_trees(state->obj_pool(), _t_output_expr, &_output_expr_ctxs)); + // Prepare the exprs to run. + RETURN_IF_ERROR(Expr::prepare(_output_expr_ctxs, state, _row_desc, _expr_mem_tracker)); + return Status::OK(); +} + +Status VResultFileSink::prepare(RuntimeState* state) { + RETURN_IF_ERROR(DataSink::prepare(state)); + std::stringstream title; + title << "VResultFileSink (fragment_instance_id=" << print_id(state->fragment_instance_id()) + << ")"; + // create profile + _profile = state->obj_pool()->add(new RuntimeProfile(title.str())); + // prepare output_expr + RETURN_IF_ERROR(prepare_exprs(state)); + + CHECK(_file_opts.get() != nullptr); + if (_is_top_sink) { + // create sender + RETURN_IF_ERROR(state->exec_env()->result_mgr()->create_sender( + state->fragment_instance_id(), _buf_size, &_sender)); + // create writer + _writer.reset(new (std::nothrow) VFileResultWriter( + _file_opts.get(), _storage_type, state->fragment_instance_id(), _output_expr_ctxs, + _profile, _sender.get(), nullptr, state->return_object_data_as_binary(), + _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 = 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_expr_ctxs, + _profile, nullptr, _output_block, state->return_object_data_as_binary(), Review Comment: Use unique ptr for _output_block and call get() to pass the raw pointer to writer. -- 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
