| // 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/olap/vcollect_iterator.h" |
| |
| #include <gen_cpp/olap_file.pb.h> |
| #include <glog/logging.h> |
| |
| #include <algorithm> |
| #include <iterator> |
| #include <ostream> |
| #include <set> |
| #include <utility> |
| |
| #include "common/compiler_util.h" // IWYU pragma: keep |
| #include "common/status.h" |
| #include "io/io_common.h" |
| #include "olap/olap_common.h" |
| #include "olap/olap_define.h" |
| #include "olap/rowset/rowset.h" |
| #include "olap/rowset/rowset_meta.h" |
| #include "olap/tablet.h" |
| #include "olap/tablet_schema.h" |
| #include "runtime/query_context.h" |
| #include "runtime/runtime_predicate.h" |
| #include "runtime/runtime_state.h" |
| #include "vec/columns/column.h" |
| #include "vec/core/column_with_type_and_name.h" |
| #include "vec/core/field.h" |
| #include "vec/data_types/data_type.h" |
| #include "vec/exprs/vexpr_context.h" |
| |
| namespace doris { |
| using namespace ErrorCode; |
| |
| namespace vectorized { |
| |
| #define RETURN_IF_NOT_EOF_AND_OK(stmt) \ |
| do { \ |
| const Status& _status_ = (stmt); \ |
| if (UNLIKELY(!_status_.ok() && !_status_.is<END_OF_FILE>())) { \ |
| return _status_; \ |
| } \ |
| } while (false) |
| |
| VCollectIterator::~VCollectIterator() = default; |
| |
| void VCollectIterator::init(TabletReader* reader, bool ori_data_overlapping, bool force_merge, |
| bool is_reverse) { |
| _reader = reader; |
| |
| // when aggregate is enabled or key_type is DUP_KEYS, we don't merge |
| // multiple data to aggregate for better performance |
| if (_reader->_reader_type == ReaderType::READER_QUERY && |
| (_reader->_direct_mode || _reader->_tablet->keys_type() == KeysType::DUP_KEYS || |
| (_reader->_tablet->keys_type() == KeysType::UNIQUE_KEYS && |
| _reader->_tablet->enable_unique_key_merge_on_write()))) { |
| _merge = false; |
| } |
| |
| // When data is none overlapping, no need to build heap to traverse data |
| if (!ori_data_overlapping) { |
| _merge = false; |
| } else if (force_merge) { |
| _merge = true; |
| } |
| _is_reverse = is_reverse; |
| |
| // use topn_next opt only for DUP_KEYS and UNIQUE_KEYS with MOW |
| if (_reader->_reader_context.read_orderby_key_limit > 0 && |
| (_reader->_tablet->keys_type() == KeysType::DUP_KEYS || |
| (_reader->_tablet->keys_type() == KeysType::UNIQUE_KEYS && |
| _reader->_tablet->enable_unique_key_merge_on_write()))) { |
| _topn_limit = _reader->_reader_context.read_orderby_key_limit; |
| } else { |
| _topn_limit = 0; |
| } |
| } |
| |
| Status VCollectIterator::add_child(const RowSetSplits& rs_splits) { |
| if (use_topn_next()) { |
| _rs_splits.push_back(rs_splits); |
| return Status::OK(); |
| } |
| |
| _children.push_back(std::make_unique<Level0Iterator>(rs_splits.rs_reader, _reader)); |
| return Status::OK(); |
| } |
| |
| // Build a merge heap. If _merge is true, a rowset with the max rownum |
| // status will be used as the base rowset, and the other rowsets will be merged first and |
| // then merged with the base rowset. |
| Status VCollectIterator::build_heap(std::vector<RowsetReaderSharedPtr>& rs_readers) { |
| if (use_topn_next()) { |
| return Status::OK(); |
| } |
| |
| DCHECK(rs_readers.size() == _children.size()); |
| _skip_same = _reader->_tablet_schema->keys_type() == KeysType::UNIQUE_KEYS; |
| if (_children.empty()) { |
| _inner_iter.reset(nullptr); |
| return Status::OK(); |
| } else if (_merge) { |
| DCHECK(!rs_readers.empty()); |
| bool have_multiple_child = false; |
| for (auto [c_iter, r_iter] = std::pair {_children.begin(), rs_readers.begin()}; |
| c_iter != _children.end();) { |
| auto s = (*c_iter)->init(have_multiple_child); |
| if (!s.ok()) { |
| c_iter = _children.erase(c_iter); |
| r_iter = rs_readers.erase(r_iter); |
| if (!s.is<END_OF_FILE>()) { |
| return s; |
| } |
| } else { |
| have_multiple_child = true; |
| ++c_iter; |
| ++r_iter; |
| } |
| } |
| |
| // build merge heap with two children, a base rowset as level0iterator and |
| // other cumulative rowsets as a level1iterator |
| if (_children.size() > 1) { |
| // find 'base rowset', 'base rowset' is the rowset which contains the max row number |
| int64_t max_row_num = 0; |
| int base_reader_idx = 0; |
| for (size_t i = 0; i < rs_readers.size(); ++i) { |
| int64_t cur_row_num = rs_readers[i]->rowset()->rowset_meta()->num_rows(); |
| if (cur_row_num > max_row_num) { |
| max_row_num = cur_row_num; |
| base_reader_idx = i; |
| } |
| } |
| auto base_reader_child = _children.begin(); |
| std::advance(base_reader_child, base_reader_idx); |
| |
| std::list<std::unique_ptr<LevelIterator>> cumu_children; |
| for (auto iter = _children.begin(); iter != _children.end();) { |
| if (iter != base_reader_child) { |
| cumu_children.push_back(std::move(*iter)); |
| iter = _children.erase(iter); |
| } else { |
| ++iter; |
| } |
| } |
| bool is_merge = cumu_children.size() > 1; |
| std::unique_ptr<LevelIterator> cumu_iter = std::make_unique<Level1Iterator>( |
| std::move(cumu_children), _reader, is_merge, _is_reverse, _skip_same); |
| RETURN_IF_NOT_EOF_AND_OK(cumu_iter->init()); |
| std::list<std::unique_ptr<LevelIterator>> children; |
| children.push_back(std::move(*base_reader_child)); |
| children.push_back(std::move(cumu_iter)); |
| _inner_iter.reset(new Level1Iterator(std::move(children), _reader, _merge, _is_reverse, |
| _skip_same)); |
| // need to clear _children here, or else if the following _inner_iter->init() return early |
| // base_reader_child will be double deleted |
| _children.clear(); |
| } else { |
| // _children.size() == 1 |
| _inner_iter.reset(new Level1Iterator(std::move(_children), _reader, _merge, _is_reverse, |
| _skip_same)); |
| } |
| } else { |
| auto level1_iter = std::make_unique<Level1Iterator>(std::move(_children), _reader, _merge, |
| _is_reverse, _skip_same); |
| _children.clear(); |
| level1_iter->init_level0_iterators_for_union(); |
| RETURN_IF_ERROR(level1_iter->ensure_first_row_ref()); |
| _inner_iter = std::move(level1_iter); |
| } |
| RETURN_IF_NOT_EOF_AND_OK(_inner_iter->init()); |
| // Clear _children earlier to release any related references |
| _children.clear(); |
| return Status::OK(); |
| } |
| |
| bool VCollectIterator::LevelIteratorComparator::operator()(LevelIterator* lhs, LevelIterator* rhs) { |
| const IteratorRowRef& lhs_ref = *lhs->current_row_ref(); |
| const IteratorRowRef& rhs_ref = *rhs->current_row_ref(); |
| |
| int cmp_res = UNLIKELY(lhs->compare_columns()) |
| ? lhs_ref.compare(rhs_ref, lhs->compare_columns()) |
| : lhs_ref.compare(rhs_ref, lhs->tablet_schema().num_key_columns()); |
| if (cmp_res != 0) { |
| return UNLIKELY(_is_reverse) ? cmp_res < 0 : cmp_res > 0; |
| } |
| |
| if (_sequence != -1) { |
| cmp_res = lhs_ref.block->get_by_position(_sequence).column->compare_at( |
| lhs_ref.row_pos, rhs_ref.row_pos, |
| *(rhs_ref.block->get_by_position(_sequence).column), -1); |
| } |
| |
| // if row cursors equal, compare data version. |
| // read data from higher version to lower version. |
| // for UNIQUE_KEYS just read the highest version and no need agg_update. |
| // for AGG_KEYS if a version is deleted, the lower version no need to agg_update |
| bool lower = (cmp_res != 0) ? (cmp_res < 0) : (lhs->version() < rhs->version()); |
| lower ? lhs->set_same(true) : rhs->set_same(true); |
| |
| return lower; |
| } |
| |
| Status VCollectIterator::current_row(IteratorRowRef* ref) const { |
| if (LIKELY(_inner_iter)) { |
| *ref = *_inner_iter->current_row_ref(); |
| if (ref->row_pos == -1) { |
| return Status::Error<END_OF_FILE>(""); |
| } else { |
| return Status::OK(); |
| } |
| } |
| return Status::Error<DATA_ROW_BLOCK_ERROR>("inner iter is nullptr"); |
| } |
| |
| Status VCollectIterator::next(IteratorRowRef* ref) { |
| if (LIKELY(_inner_iter)) { |
| return _inner_iter->next(ref); |
| } else { |
| return Status::Error<END_OF_FILE>(""); |
| } |
| } |
| |
| Status VCollectIterator::next(Block* block) { |
| if (use_topn_next()) { |
| return _topn_next(block); |
| } |
| |
| if (LIKELY(_inner_iter)) { |
| return _inner_iter->next(block); |
| } else { |
| return Status::Error<END_OF_FILE>(""); |
| } |
| } |
| |
| Status VCollectIterator::_topn_next(Block* block) { |
| if (_topn_eof) { |
| return Status::Error<END_OF_FILE>(""); |
| } |
| |
| auto clone_block = block->clone_empty(); |
| MutableBlock mutable_block = vectorized::MutableBlock::build_mutable_block(&clone_block); |
| // clear TEMP columns to avoid column align problem in mutable_block.add_rows bellow |
| auto all_column_names = mutable_block.get_names(); |
| for (auto& name : all_column_names) { |
| if (name.rfind(BeConsts::BLOCK_TEMP_COLUMN_PREFIX, 0) == 0) { |
| mutable_block.erase(name); |
| // clear TEMP columns from block to prevent from storage engine merge with this |
| // fake column |
| block->erase(name); |
| } |
| } |
| |
| if (!_reader->_reader_context.read_orderby_key_columns) { |
| return Status::Error<ErrorCode::INTERNAL_ERROR>( |
| "read_orderby_key_columns should not be nullptr"); |
| } |
| |
| size_t first_sort_column_idx = (*_reader->_reader_context.read_orderby_key_columns)[0]; |
| const std::vector<uint32_t>* sort_columns = _reader->_reader_context.read_orderby_key_columns; |
| |
| BlockRowPosComparator row_pos_comparator(&mutable_block, sort_columns, |
| _reader->_reader_context.read_orderby_key_reverse); |
| std::multiset<size_t, BlockRowPosComparator, std::allocator<size_t>> sorted_row_pos( |
| row_pos_comparator); |
| |
| if (_is_reverse) { |
| std::reverse(_rs_splits.begin(), _rs_splits.end()); |
| } |
| |
| for (size_t i = 0; i < _rs_splits.size(); i++) { |
| const auto& rs_split = _rs_splits[i]; |
| // init will prune segment by _reader_context.conditions and _reader_context.runtime_conditions |
| RETURN_IF_ERROR(rs_split.rs_reader->init(&_reader->_reader_context, rs_split)); |
| |
| // read _topn_limit rows from this rs |
| size_t read_rows = 0; |
| bool eof = false; |
| while (read_rows < _topn_limit && !eof) { |
| block->clear_column_data(); |
| auto status = rs_split.rs_reader->next_block(block); |
| if (!status.ok()) { |
| if (status.is<END_OF_FILE>()) { |
| eof = true; |
| if (block->rows() == 0) { |
| break; |
| } |
| } else { |
| return status; |
| } |
| } |
| |
| auto col_name = block->get_names()[first_sort_column_idx]; |
| |
| // filter block |
| RETURN_IF_ERROR(VExprContext::filter_block( |
| _reader->_reader_context.filter_block_conjuncts, block, block->columns())); |
| // clear TMPE columns to avoid column align problem in mutable_block.add_rows bellow |
| auto all_column_names = block->get_names(); |
| for (auto& name : all_column_names) { |
| if (name.rfind(BeConsts::BLOCK_TEMP_COLUMN_PREFIX, 0) == 0) { |
| block->erase(name); |
| } |
| } |
| |
| // update read rows |
| read_rows += block->rows(); |
| |
| // insert block rows to mutable_block and adjust sorted_row_pos |
| bool changed = false; |
| |
| size_t rows_to_copy = 0; |
| if (sorted_row_pos.empty()) { |
| rows_to_copy = std::min(block->rows(), _topn_limit); |
| } else { |
| // _is_reverse == true last_row_pos is the pos of smallest row |
| // _is_reverse == false last_row_pos is biggest row |
| size_t last_row_pos = *sorted_row_pos.rbegin(); |
| |
| // find the how many rows which is less than the last row in mutable_block |
| for (size_t i = 0; i < block->rows(); i++) { |
| // if there is not enough rows in sorted_row_pos, just copy new rows |
| if (sorted_row_pos.size() + rows_to_copy < _topn_limit) { |
| rows_to_copy++; |
| continue; |
| } |
| |
| DCHECK_GE(block->columns(), sort_columns->size()); |
| DCHECK_GE(mutable_block.columns(), sort_columns->size()); |
| |
| int res = 0; |
| for (auto j : *sort_columns) { |
| DCHECK(block->get_by_position(j).type->equals( |
| *mutable_block.get_datatype_by_position(j))); |
| res = block->get_by_position(j).column->compare_at( |
| i, last_row_pos, *(mutable_block.get_column_by_position(j)), -1); |
| if (res) { |
| break; |
| } |
| } |
| |
| // only copy needed rows |
| // _is_reverse == true > smallest is ok |
| // _is_reverse == false < biggest is ok |
| if ((_is_reverse && res > 0) || (!_is_reverse && res < 0)) { |
| rows_to_copy++; |
| } else { |
| break; |
| } |
| } |
| } |
| |
| if (rows_to_copy > 0) { |
| // create column that is not in mutable_block but in block |
| for (size_t i = mutable_block.columns(); i < block->columns(); ++i) { |
| auto col = block->get_by_position(i).clone_empty(); |
| mutable_block.mutable_columns().push_back(col.column->assume_mutable()); |
| mutable_block.data_types().push_back(std::move(col.type)); |
| mutable_block.get_names().push_back(std::move(col.name)); |
| } |
| |
| size_t base = mutable_block.rows(); |
| // append block to mutable_block |
| mutable_block.add_rows(block, 0, rows_to_copy); |
| // insert appended rows pos in mutable_block to sorted_row_pos and sort it |
| for (size_t i = 0; i < rows_to_copy; i++) { |
| sorted_row_pos.insert(base + i); |
| changed = true; |
| } |
| } |
| |
| // delete to keep _topn_limit row pos |
| if (sorted_row_pos.size() > _topn_limit) { |
| auto first = sorted_row_pos.begin(); |
| for (size_t i = 0; i < _topn_limit; i++) { |
| first++; |
| } |
| sorted_row_pos.erase(first, sorted_row_pos.end()); |
| |
| // shrink mutable_block to save memory when rows > _topn_limit * 2 |
| if (mutable_block.rows() > _topn_limit * 2) { |
| VLOG_DEBUG << "topn debug start shrink mutable_block from " |
| << mutable_block.rows() << " rows"; |
| Block tmp_block = mutable_block.to_block(); |
| clone_block = tmp_block.clone_empty(); |
| mutable_block = vectorized::MutableBlock::build_mutable_block(&clone_block); |
| for (auto it = sorted_row_pos.begin(); it != sorted_row_pos.end(); it++) { |
| mutable_block.add_row(&tmp_block, *it); |
| } |
| |
| sorted_row_pos.clear(); |
| for (size_t i = 0; i < _topn_limit; i++) { |
| sorted_row_pos.insert(i); |
| } |
| VLOG_DEBUG << "topn debug finish shrink mutable_block to " |
| << mutable_block.rows() << " rows"; |
| } |
| } |
| |
| // update runtime_predicate |
| if (_reader->_reader_context.use_topn_opt && changed && |
| sorted_row_pos.size() >= _topn_limit) { |
| // get field value from column |
| size_t last_sorted_row = *sorted_row_pos.rbegin(); |
| auto col_ptr = mutable_block.get_column_by_position(first_sort_column_idx).get(); |
| Field new_top; |
| col_ptr->get(last_sorted_row, new_top); |
| |
| // update orderby_extrems in query global context |
| auto query_ctx = _reader->_reader_context.runtime_state->get_query_ctx(); |
| RETURN_IF_ERROR( |
| query_ctx->get_runtime_predicate().update(new_top, col_name, _is_reverse)); |
| } |
| } // end of while (read_rows < _topn_limit && !eof) |
| VLOG_DEBUG << "topn debug rowset " << i << " read_rows=" << read_rows << " eof=" << eof |
| << " _topn_limit=" << _topn_limit |
| << " sorted_row_pos.size()=" << sorted_row_pos.size() |
| << " mutable_block.rows()=" << mutable_block.rows(); |
| } // end of for (auto rs_reader : _rs_readers) |
| |
| // copy result_block to block |
| VLOG_DEBUG << "topn debug result _topn_limit=" << _topn_limit |
| << " sorted_row_pos.size()=" << sorted_row_pos.size() |
| << " mutable_block.rows()=" << mutable_block.rows(); |
| *block = mutable_block.to_block(); |
| // append a column to indicate scanner filter_block is already done |
| auto filtered_datatype = std::make_shared<DataTypeUInt8>(); |
| auto filtered_column = filtered_datatype->create_column_const(block->rows(), (uint8_t)1); |
| block->insert( |
| {filtered_column, filtered_datatype, BeConsts::BLOCK_TEMP_COLUMN_SCANNER_FILTERED}); |
| |
| _topn_eof = true; |
| return block->rows() > 0 ? Status::OK() : Status::Error<END_OF_FILE>(""); |
| } |
| |
| bool VCollectIterator::BlockRowPosComparator::operator()(const size_t& lpos, |
| const size_t& rpos) const { |
| int ret = _mutable_block->compare_at(lpos, rpos, _compare_columns, *_mutable_block, -1); |
| return _is_reverse ? ret > 0 : ret < 0; |
| } |
| |
| VCollectIterator::Level0Iterator::Level0Iterator(RowsetReaderSharedPtr rs_reader, |
| TabletReader* reader) |
| : LevelIterator(reader), _rs_reader(rs_reader), _reader(reader) { |
| DCHECK_EQ(RowsetTypePB::BETA_ROWSET, rs_reader->type()); |
| } |
| |
| Status VCollectIterator::Level0Iterator::init(bool get_data_by_ref) { |
| _get_data_by_ref = get_data_by_ref && _rs_reader->support_return_data_by_ref(); |
| if (!_get_data_by_ref) { |
| _block = std::make_shared<Block>(_schema.create_block( |
| _reader->_return_columns, _reader->_tablet_columns_convert_to_null_set)); |
| } |
| |
| auto st = refresh_current_row(); |
| if (_get_data_by_ref && _block_view.size()) { |
| _ref = _block_view[0]; |
| } else { |
| _ref = {_block, 0, false}; |
| } |
| return st; |
| } |
| |
| // if is_first_child = true, return first row in block。Unique keys and agg keys will |
| // read a line first and then start loop : |
| // while (!eof) { |
| // collect_iter->next(&_next_row); |
| // } |
| // so first child load first row and other child row_pos = -1 |
| void VCollectIterator::Level0Iterator::init_for_union(bool get_data_by_ref) { |
| _get_data_by_ref = get_data_by_ref && _rs_reader->support_return_data_by_ref(); |
| } |
| |
| Status VCollectIterator::Level0Iterator::ensure_first_row_ref() { |
| DCHECK(!_get_data_by_ref); |
| auto s = refresh_current_row(); |
| _ref = {_block, 0, false}; |
| |
| return s; |
| } |
| |
| int64_t VCollectIterator::Level0Iterator::version() const { |
| return _rs_reader->version().second; |
| } |
| |
| Status VCollectIterator::Level0Iterator::refresh_current_row() { |
| do { |
| if (_block == nullptr && !_get_data_by_ref) { |
| _block = std::make_shared<Block>(_schema.create_block( |
| _reader->_return_columns, _reader->_tablet_columns_convert_to_null_set)); |
| } |
| |
| if (!_is_empty() && _current_valid()) { |
| return Status::OK(); |
| } else { |
| _reset(); |
| auto res = _refresh(); |
| if (!res.ok() && !res.is<END_OF_FILE>()) { |
| return res; |
| } |
| if (res.is<END_OF_FILE>() && _is_empty()) { |
| break; |
| } |
| |
| if (UNLIKELY(_reader->_reader_context.record_rowids)) { |
| RETURN_IF_ERROR(_rs_reader->current_block_row_locations(&_block_row_locations)); |
| } |
| } |
| } while (!_is_empty()); |
| _ref.row_pos = -1; |
| _current = -1; |
| _rs_reader = nullptr; |
| return Status::Error<END_OF_FILE>(""); |
| } |
| |
| Status VCollectIterator::Level0Iterator::next(IteratorRowRef* ref) { |
| if (_get_data_by_ref) { |
| _current++; |
| } else { |
| _ref.row_pos++; |
| } |
| |
| RETURN_IF_ERROR(refresh_current_row()); |
| |
| if (_get_data_by_ref) { |
| _ref = _block_view[_current]; |
| } |
| |
| *ref = _ref; |
| return Status::OK(); |
| } |
| |
| Status VCollectIterator::Level0Iterator::next(Block* block) { |
| CHECK(!_get_data_by_ref); |
| if (_ref.row_pos <= 0 && _ref.block != nullptr && UNLIKELY(_ref.block->rows() > 0)) { |
| block->swap(*_ref.block); |
| _ref.reset(); |
| return Status::OK(); |
| } else { |
| if (_rs_reader == nullptr) { |
| return Status::Error<END_OF_FILE>(""); |
| } |
| auto res = _rs_reader->next_block(block); |
| if (!res.ok() && !res.is<END_OF_FILE>()) { |
| return res; |
| } |
| if (res.is<END_OF_FILE>() && block->rows() == 0) { |
| return Status::Error<END_OF_FILE>(""); |
| } |
| if (UNLIKELY(_reader->_reader_context.record_rowids)) { |
| RETURN_IF_ERROR(_rs_reader->current_block_row_locations(&_block_row_locations)); |
| } |
| return Status::OK(); |
| } |
| } |
| |
| RowLocation VCollectIterator::Level0Iterator::current_row_location() { |
| RowLocation& segment_row_id = _block_row_locations[_get_data_by_ref ? _current : _ref.row_pos]; |
| return RowLocation(_rs_reader->rowset()->rowset_id(), segment_row_id.segment_id, |
| segment_row_id.row_id); |
| } |
| |
| Status VCollectIterator::Level0Iterator::current_block_row_locations( |
| std::vector<RowLocation>* block_row_locations) { |
| block_row_locations->resize(_block_row_locations.size()); |
| for (auto i = 0; i < _block_row_locations.size(); i++) { |
| RowLocation& row_location = _block_row_locations[i]; |
| (*block_row_locations)[i] = RowLocation(_rs_reader->rowset()->rowset_id(), |
| row_location.segment_id, row_location.row_id); |
| } |
| return Status::OK(); |
| } |
| |
| VCollectIterator::Level1Iterator::Level1Iterator( |
| std::list<std::unique_ptr<VCollectIterator::LevelIterator>> children, TabletReader* reader, |
| bool merge, bool is_reverse, bool skip_same) |
| : LevelIterator(reader), |
| _children(std::move(children)), |
| _reader(reader), |
| _merge(merge), |
| _is_reverse(is_reverse), |
| _skip_same(skip_same) { |
| _ref.reset(); |
| // !_merge means that data are in order, so we just reverse children to return data in reverse |
| if (!_merge && _is_reverse) { |
| _children.reverse(); |
| } |
| } |
| |
| VCollectIterator::Level1Iterator::~Level1Iterator() { |
| if (_heap) { |
| while (!_heap->empty()) { |
| delete _heap->top(); |
| _heap->pop(); |
| } |
| } |
| } |
| |
| // Read next row into *row. |
| // Returns |
| // OK when read successfully. |
| // Status::Error<END_OF_FILE>("") and set *row to nullptr when EOF is reached. |
| // Others when error happens |
| Status VCollectIterator::Level1Iterator::next(IteratorRowRef* ref) { |
| if (UNLIKELY(_cur_child == nullptr)) { |
| _ref.reset(); |
| return Status::Error<END_OF_FILE>(""); |
| } |
| if (_merge) { |
| return _merge_next(ref); |
| } else { |
| return _normal_next(ref); |
| } |
| } |
| |
| // Read next block |
| // Returns |
| // OK when read successfully. |
| // Status::Error<END_OF_FILE>("") and set *row to nullptr when EOF is reached. |
| // Others when error happens |
| Status VCollectIterator::Level1Iterator::next(Block* block) { |
| if (UNLIKELY(_cur_child == nullptr)) { |
| return Status::Error<END_OF_FILE>(""); |
| } |
| if (_merge) { |
| return _merge_next(block); |
| } else { |
| return _normal_next(block); |
| } |
| } |
| |
| int64_t VCollectIterator::Level1Iterator::version() const { |
| if (_cur_child != nullptr) { |
| return _cur_child->version(); |
| } |
| return -1; |
| } |
| |
| Status VCollectIterator::Level1Iterator::init(bool get_data_by_ref) { |
| if (_children.empty()) { |
| return Status::OK(); |
| } |
| |
| // Only when there are multiple children that need to be merged |
| if (_merge && _children.size() > 1) { |
| auto sequence_loc = -1; |
| for (int loc = 0; loc < _reader->_return_columns.size(); loc++) { |
| if (_reader->_return_columns[loc] == _reader->_sequence_col_idx) { |
| sequence_loc = loc; |
| break; |
| } |
| } |
| _heap.reset(new MergeHeap {LevelIteratorComparator(sequence_loc, _is_reverse)}); |
| for (auto&& child : _children) { |
| DCHECK(child != nullptr); |
| //DCHECK(child->current_row().ok()); |
| _heap->push(child.release()); |
| } |
| _cur_child.reset(_heap->top()); |
| _heap->pop(); |
| // Clear _children earlier to release any related references |
| _children.clear(); |
| } else { |
| _merge = false; |
| _heap.reset(nullptr); |
| _cur_child = std::move(*_children.begin()); |
| _children.pop_front(); |
| } |
| _ref = *_cur_child->current_row_ref(); |
| return Status::OK(); |
| } |
| |
| Status VCollectIterator::Level1Iterator::ensure_first_row_ref() { |
| for (auto iter = _children.begin(); iter != _children.end();) { |
| auto s = (*iter)->ensure_first_row_ref(); |
| if (!s.ok()) { |
| iter = _children.erase(iter); |
| if (!s.is<END_OF_FILE>()) { |
| return s; |
| } |
| } else { |
| // we get a real row |
| break; |
| } |
| } |
| |
| return Status::OK(); |
| } |
| |
| void VCollectIterator::Level1Iterator::init_level0_iterators_for_union() { |
| bool have_multiple_child = false; |
| for (auto iter = _children.begin(); iter != _children.end();) { |
| (*iter)->init_for_union(have_multiple_child); |
| have_multiple_child = true; |
| ++iter; |
| } |
| } |
| |
| Status VCollectIterator::Level1Iterator::_merge_next(IteratorRowRef* ref) { |
| auto res = _cur_child->next(ref); |
| if (LIKELY(res.ok())) { |
| _heap->push(_cur_child.release()); |
| _cur_child.reset(_heap->top()); |
| _heap->pop(); |
| } else if (res.is<END_OF_FILE>()) { |
| // current child has been read, to read next |
| if (!_heap->empty()) { |
| _cur_child.reset(_heap->top()); |
| _heap->pop(); |
| } else { |
| _ref.reset(); |
| _cur_child.reset(); |
| return Status::Error<END_OF_FILE>(""); |
| } |
| } else { |
| _ref.reset(); |
| _cur_child.reset(); |
| LOG(WARNING) << "failed to get next from child, res=" << res; |
| return res; |
| } |
| |
| if (_skip_same && _cur_child->is_same()) { |
| _reader->_merged_rows++; |
| _cur_child->set_same(false); |
| return _merge_next(ref); |
| } |
| |
| _ref = *_cur_child->current_row_ref(); |
| *ref = _ref; |
| |
| _cur_child->set_same(false); |
| |
| return Status::OK(); |
| } |
| |
| Status VCollectIterator::Level1Iterator::_normal_next(IteratorRowRef* ref) { |
| auto res = _cur_child->next(ref); |
| if (LIKELY(res.ok())) { |
| _ref = *ref; |
| return Status::OK(); |
| } else if (res.is<END_OF_FILE>()) { |
| // current child has been read, to read next |
| if (!_children.empty()) { |
| _cur_child = std::move(*(_children.begin())); |
| _children.pop_front(); |
| return _normal_next(ref); |
| } else { |
| _cur_child.reset(); |
| return Status::Error<END_OF_FILE>(""); |
| } |
| } else { |
| _cur_child.reset(); |
| LOG(WARNING) << "failed to get next from child, res=" << res; |
| return res; |
| } |
| } |
| |
| Status VCollectIterator::Level1Iterator::_merge_next(Block* block) { |
| int target_block_row = 0; |
| auto target_columns = block->mutate_columns(); |
| size_t column_count = target_columns.size(); |
| IteratorRowRef cur_row = _ref; |
| IteratorRowRef pre_row_ref = _ref; |
| |
| // append extra columns (eg. MATCH pred result column) from src_block to block |
| for (size_t i = block->columns(); i < cur_row.block->columns(); ++i) { |
| block->insert(cur_row.block->get_by_position(i).clone_empty()); |
| } |
| |
| auto batch_size = _reader->batch_size(); |
| if (UNLIKELY(_reader->_reader_context.record_rowids)) { |
| _block_row_locations.resize(batch_size); |
| } |
| int continuous_row_in_block = 0; |
| do { |
| if (UNLIKELY(_reader->_reader_context.record_rowids)) { |
| _block_row_locations[target_block_row] = _cur_child->current_row_location(); |
| } |
| ++target_block_row; |
| ++continuous_row_in_block; |
| // cur block finished, copy before merge_next cause merge_next will |
| // clear block column data |
| if (pre_row_ref.row_pos + continuous_row_in_block == pre_row_ref.block->rows()) { |
| const auto& src_block = pre_row_ref.block; |
| RETURN_IF_CATCH_EXCEPTION({ |
| for (size_t i = 0; i < column_count; ++i) { |
| target_columns[i]->insert_range_from(*(src_block->get_by_position(i).column), |
| pre_row_ref.row_pos, |
| continuous_row_in_block); |
| } |
| }); |
| continuous_row_in_block = 0; |
| pre_row_ref.reset(); |
| } |
| auto res = _merge_next(&cur_row); |
| if (UNLIKELY(res.is<END_OF_FILE>())) { |
| if (UNLIKELY(_reader->_reader_context.record_rowids)) { |
| _block_row_locations.resize(target_block_row); |
| } |
| block->set_columns(std::move(target_columns)); |
| return res; |
| } |
| |
| if (UNLIKELY(!res.ok())) { |
| LOG(WARNING) << "next failed: " << res; |
| return res; |
| } |
| if (target_block_row >= batch_size) { |
| if (continuous_row_in_block > 0) { |
| const auto& src_block = pre_row_ref.block; |
| for (size_t i = 0; i < column_count; ++i) { |
| target_columns[i]->insert_range_from(*(src_block->get_by_position(i).column), |
| pre_row_ref.row_pos, |
| continuous_row_in_block); |
| } |
| } |
| block->set_columns(std::move(target_columns)); |
| return Status::OK(); |
| } |
| if (continuous_row_in_block == 0) { |
| pre_row_ref = _ref; |
| continue; |
| } |
| // copy row if meet a new block |
| if (cur_row.block != pre_row_ref.block) { |
| const auto& src_block = pre_row_ref.block; |
| for (size_t i = 0; i < column_count; ++i) { |
| target_columns[i]->insert_range_from(*(src_block->get_by_position(i).column), |
| pre_row_ref.row_pos, continuous_row_in_block); |
| } |
| continuous_row_in_block = 0; |
| pre_row_ref = cur_row; |
| } |
| } while (true); |
| |
| return Status::OK(); |
| } |
| |
| Status VCollectIterator::Level1Iterator::_normal_next(Block* block) { |
| auto res = _cur_child->next(block); |
| if (LIKELY(res.ok())) { |
| return Status::OK(); |
| } else if (res.is<END_OF_FILE>()) { |
| // current child has been read, to read next |
| if (!_children.empty()) { |
| _cur_child = std::move(*(_children.begin())); |
| _children.pop_front(); |
| return _normal_next(block); |
| } else { |
| _cur_child.reset(); |
| return Status::Error<END_OF_FILE>(""); |
| } |
| } else { |
| _cur_child.reset(); |
| LOG(WARNING) << "failed to get next from child, res=" << res; |
| return res; |
| } |
| } |
| |
| Status VCollectIterator::Level1Iterator::current_block_row_locations( |
| std::vector<RowLocation>* block_row_locations) { |
| if (!_merge) { |
| if (UNLIKELY(_cur_child == nullptr)) { |
| block_row_locations->clear(); |
| return Status::Error<END_OF_FILE>(""); |
| } |
| return _cur_child->current_block_row_locations(block_row_locations); |
| } else { |
| DCHECK(_reader->_reader_context.record_rowids); |
| *block_row_locations = _block_row_locations; |
| return Status::OK(); |
| } |
| } |
| |
| } // namespace vectorized |
| } // namespace doris |
