morningman commented on a change in pull request #7255: URL: https://github.com/apache/incubator-doris/pull/7255#discussion_r767156283
########## File path: be/src/exec/table_function_node.cpp ########## @@ -0,0 +1,341 @@ +// 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 "exec/table_function_node.h" + +#include "exprs/expr.h" +#include "exprs/expr_context.h" +#include "runtime/descriptors.h" +#include "runtime/raw_value.h" +#include "runtime/row_batch.h" +#include "runtime/runtime_state.h" +#include "runtime/tuple_row.h" +#include "exprs/table_function/table_function_factory.h" + +namespace doris { + +TableFunctionNode::TableFunctionNode(ObjectPool* pool, const TPlanNode& tnode, const DescriptorTbl& descs) + : ExecNode(pool, tnode, descs) { + +} + +TableFunctionNode::~TableFunctionNode() { + +} + +Status TableFunctionNode::init(const TPlanNode& tnode, RuntimeState* state) { + RETURN_IF_ERROR(ExecNode::init(tnode, state)); + + for (const TExpr& texpr : tnode.table_function_node.fnCallExprList) { + ExprContext* ctx = nullptr; + RETURN_IF_ERROR(Expr::create_expr_tree(_pool, texpr, &ctx)); + _fn_ctxs.push_back(ctx); + + Expr* root = ctx->root(); + const std::string& tf_name = root->fn().name.function_name; + TableFunction* fn; + RETURN_IF_ERROR(TableFunctionFactory::get_fn(tf_name, _pool, &fn)); + fn->set_expr_context(ctx); + _fns.push_back(fn); + } + _fn_num = _fns.size(); + _fn_values.resize(_fn_num); + + // Prepare output slot ids + RETURN_IF_ERROR(_prepare_output_slot_ids(tnode)); + return Status::OK(); +} + +Status TableFunctionNode::_prepare_output_slot_ids(const TPlanNode& tnode) { + // Prepare output slot ids + if (tnode.table_function_node.outputSlotIds.empty()) { + return Status::InternalError("Output slots of table function node is empty"); + } + SlotId max_id = -1; + for (auto slot_id : tnode.table_function_node.outputSlotIds) { + if (slot_id > max_id) { + max_id = slot_id; + } + } + _output_slot_ids = std::vector<bool>(max_id + 1, false); + for (auto slot_id : tnode.table_function_node.outputSlotIds) { + _output_slot_ids[slot_id] = true; + } + + return Status::OK(); +} + +Status TableFunctionNode::prepare(RuntimeState* state) { + RETURN_IF_ERROR(ExecNode::prepare(state)); + + RETURN_IF_ERROR(Expr::prepare(_fn_ctxs, state, _row_descriptor, expr_mem_tracker())); + for (auto fn : _fns) { + RETURN_IF_ERROR(fn->prepare()); + } + return Status::OK(); +} + +Status TableFunctionNode::open(RuntimeState* state) { + SCOPED_TIMER(_runtime_profile->total_time_counter()); + RETURN_IF_CANCELLED(state); + RETURN_IF_ERROR(ExecNode::open(state)); + + RETURN_IF_ERROR(Expr::open(_fn_ctxs, state)); + for (auto fn : _fns) { + RETURN_IF_ERROR(fn->open()); + } + + RETURN_IF_ERROR(_children[0]->open(state)); + return Status::OK(); +} + +Status TableFunctionNode::_process_next_child_row() { + if (_cur_child_offset == _cur_child_batch->num_rows()) { + _child_batch_exhausted = true; + return Status::OK(); + } + _cur_child_tuple_row = _cur_child_batch->get_row(_cur_child_offset++); + for (TableFunction* fn : _fns) { + RETURN_IF_ERROR(fn->process(_cur_child_tuple_row)); + } + + _child_batch_exhausted = false; + return Status::OK(); +} + +// Returns the index of fn of the last eos counted from back to front +// eg: there are 3 functions in `_fns` +// eos: false, true, true +// return: 1 +// +// eos: false, false, true +// return: 2 +// +// eos: false, false, false +// return: -1 +// +// eos: true, true, true +// return: 0 +// +// return: +// 0: all fns are eos +// -1: all fns are not eos +// >0: some of fns are eos +int TableFunctionNode::_find_last_fn_eos_idx() { + for (int i = _fn_num - 1; i >=0; --i) { + if (!_fns[i]->eos()) { + if (i == _fn_num - 1) { + return -1; + } else { + return i + 1; + } + } + } + // all eos + return 0; +} + +// Roll to reset the table function. +// Eg: +// There are 3 functions f1, f2 and f3 in `_fns`. +// If `last_eos_idx` is 1, which means f2 and f3 are eos. +// So we need to forward f1, and reset f2 and f3. +bool TableFunctionNode::_roll_table_functions(int last_eos_idx) { + bool fn_eos = false; + int i = last_eos_idx - 1; + for (; i >= 0; --i) { + _fns[i]->forward(&fn_eos); + if (!fn_eos) { + break; + } + } + if (i == -1) { + // after forward, all functions are eos. + // we should process next child row to get more table function results. + return false; + } + + for (int j = i + 1; j < _fn_num; ++j) { + _fns[j]->reset(); + } + + return true; +} + +Status TableFunctionNode::get_next(RuntimeState* state, RowBatch* row_batch, bool* eos) { + RETURN_IF_ERROR(exec_debug_action(TExecNodePhase::GETNEXT)); + SCOPED_TIMER(_runtime_profile->total_time_counter()); + + const RowDescriptor& parent_rowdesc = row_batch->row_desc(); + const RowDescriptor& child_rowdesc = _children[0]->row_desc(); + if (_parent_tuple_desc_size == -1) { + _parent_tuple_desc_size = parent_rowdesc.tuple_descriptors().size(); + _child_tuple_desc_size = child_rowdesc.tuple_descriptors().size(); + for (int i = 0; i < _child_tuple_desc_size; ++i) { + _child_slot_sizes.push_back(child_rowdesc.tuple_descriptors()[i]->slots().size()); + } + } + + uint8_t* tuple_buffer = nullptr; + Tuple* tuple_ptr = nullptr; + Tuple* pre_tuple_ptr = nullptr; + int row_idx = 0; + + while (true) { + RETURN_IF_CANCELLED(state); + RETURN_IF_ERROR(state->check_query_state("TableFunctionNode, while getting next batch.")); + + if (_cur_child_batch == nullptr) { + _cur_child_batch.reset(new RowBatch(child_rowdesc, state->batch_size(), mem_tracker().get())); + } + if (_child_batch_exhausted) { + // current child batch is exhausted, get next batch from child + RETURN_IF_ERROR(_children[0]->get_next(state, _cur_child_batch.get(), eos)); + if (*eos) { + break; + } + _cur_child_offset = 0; + RETURN_IF_ERROR(_process_next_child_row()); Review comment: It need to be called before we call `_find_last_fn_eos_idx()` at first time -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
