[feature](scan) Implement parallel scanning by dividing the tablets based on the row range

be/src/olap/iterators.h

@@ -24,6 +24,7 @@
 #include "olap/block_column_predicate.h"
 #include "olap/column_predicate.h"
 #include "olap/olap_common.h"
+#include "olap/rowset/segment_v2/row_ranges.h"
 #include "olap/tablet_schema.h"
 #include "runtime/runtime_state.h"
 #include "vec/core/block.h"
@@ -115,6 +116,7 @@ class StorageReadOptions {
     int32_t tablet_id = 0;
     // slots that cast may be eliminated in storage layer
     std::map<std::string, PrimitiveType> target_cast_type_for_variants;
+    RowRanges row_ranges;
 };
 
 class RowwiseIterator;

be/src/olap/parallel_scanner_builder.cpp

@@ -0,0 +1,201 @@
+// 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 "parallel_scanner_builder.h"
+
+#include "olap/rowset/beta_rowset.h"
+#include "vec/exec/scan/new_olap_scanner.h"
+
+namespace doris {
+
+using namespace vectorized;
+
+template <typename ParentType>
+Status ParallelScannerBuilder<ParentType>::build_scanners(std::list<VScannerSPtr>& scanners) {
+    RETURN_IF_ERROR(_load());
+    if (_is_dup_mow_key) {
+        return _build_scanners_by_rowid(scanners);
+    } else {
+        // TODO: support to split by key range
+        return Status::NotSupported("split by key range not supported yet.");
+    }
+}
+
+template <typename ParentType>
+Status ParallelScannerBuilder<ParentType>::_build_scanners_by_rowid(
+        std::list<VScannerSPtr>& scanners) {
+    DCHECK_GE(_rows_per_scanner, _min_rows_per_scanner);
+    for (auto&& [tablet, version] : _tablets) {
+        DCHECK(_all_rowsets.contains(tablet->tablet_id()));
+        auto& rowsets = _all_rowsets[tablet->tablet_id()];
+
+        TabletReader::ReadSource reade_source_with_delete_info;
+        if (!_state->skip_delete_predicate()) {
+            RETURN_IF_ERROR(tablet->capture_rs_readers(
+                    {0, version}, &reade_source_with_delete_info.rs_splits, false));
+            reade_source_with_delete_info.fill_delete_predicates();
+        }
+
+        TabletReader::ReadSource read_source;
+
+        int64_t rows_collected = 0;
+        for (auto& rowset : rowsets) {
+            auto beta_rowset = std::dynamic_pointer_cast<BetaRowset>(rowset);
+            RowsetReaderSharedPtr reader;
+            RETURN_IF_ERROR(beta_rowset->create_reader(&reader));
+            const auto rowset_id = beta_rowset->rowset_id();
+
+            DCHECK(_segment_cache_handles.contains(rowset_id));
+            auto& segment_cache_handle = _segment_cache_handles[rowset_id];
+
+            if (beta_rowset->num_rows() == 0) {
+                continue;
+            }
+
+            const auto& segments = segment_cache_handle.get_segments();
+            int segment_start = 0;
+            auto split = RowSetSplits(reader->clone());
+
+            for (size_t i = 0; i != segments.size(); ++i) {
+                const auto& segment = segments[i];
+                RowRanges row_ranges;
+                const size_t rows_of_segment = segment->num_rows();
+                int64_t offset_in_segment = 0;
+
+                // try to split large segments into RowRanges
+                while (offset_in_segment < rows_of_segment) {
+                    const int64_t remaining_rows = rows_of_segment - offset_in_segment;
+                    auto rows_need = _rows_per_scanner - rows_collected;
+
+                    // 0.9: try to avoid splitting the segments into excessively small parts.
+                    if (rows_need >= remaining_rows * 0.9) {
+                        rows_need = remaining_rows;
+                    }
+                    DCHECK_LE(rows_need, remaining_rows);
+
+                    // RowRange stands for range: [From, To), From is inclusive, To is exclusive.
+                    row_ranges.add({offset_in_segment,
+                                    offset_in_segment + static_cast<int64_t>(rows_need)});
+                    rows_collected += rows_need;
+                    offset_in_segment += rows_need;
+
+                    // If collected enough rows, build a new scanner
+                    if (rows_collected >= _rows_per_scanner) {
+                        split.segment_offsets.first = segment_start,
+                        split.segment_offsets.second = i + 1;
+                        split.segment_row_ranges.emplace_back(std::move(row_ranges));
+
+                        DCHECK_EQ(split.segment_offsets.second - split.segment_offsets.first,
+                                  split.segment_row_ranges.size());
+
+                        read_source.rs_splits.emplace_back(std::move(split));
+
+                        scanners.emplace_back(
+                                _build_scanner(tablet, version, _key_ranges,
+                                               {std::move(read_source.rs_splits),
+                                                reade_source_with_delete_info.delete_predicates}));
+
+                        read_source = TabletReader::ReadSource();
+                        split = RowSetSplits(reader->clone());
+                        row_ranges = RowRanges();
+
+                        segment_start = offset_in_segment < rows_of_segment ? i : i + 1;
+                        rows_collected = 0;
+                    }
+                }
+
+                // The non-empty `row_ranges` means there are some rows left in this segment not added into `split`.
+                if (!row_ranges.is_empty()) {
+                    DCHECK_GT(rows_collected, 0);
+                    DCHECK_EQ(row_ranges.to(), segment->num_rows());
+                    split.segment_row_ranges.emplace_back(std::move(row_ranges));
+                }
+            }
+
+            DCHECK_LE(rows_collected, _rows_per_scanner);
+            if (rows_collected > 0) {
+                split.segment_offsets.first = segment_start;
+                split.segment_offsets.second = segments.size();
+                DCHECK_GT(split.segment_offsets.second, split.segment_offsets.first);
+                DCHECK_EQ(split.segment_row_ranges.size(),
+                          split.segment_offsets.second - split.segment_offsets.first);
+                read_source.rs_splits.emplace_back(std::move(split));
+            }
+        } // end `for (auto& rowset : rowsets)`
+
+        DCHECK_LE(rows_collected, _rows_per_scanner);
+        if (rows_collected > 0) {
+            DCHECK_GT(read_source.rs_splits.size(), 0);
+#ifndef NDEBUG
+            for (auto& split : read_source.rs_splits) {
+                DCHECK(split.rs_reader != nullptr);
+                DCHECK_LT(split.segment_offsets.first, split.segment_offsets.second);
+                DCHECK_EQ(split.segment_row_ranges.size(),
+                          split.segment_offsets.second - split.segment_offsets.first);
+            }
+#endif
+            scanners.emplace_back(
+                    _build_scanner(tablet, version, _key_ranges,
+                                   {std::move(read_source.rs_splits),
+                                    reade_source_with_delete_info.delete_predicates}));
+        }
+    }
+
+    return Status::OK();
+}
+
+/**
+ * Load rowsets of each tablet with specified version, segments of each rowset.
+ */
+template <typename ParentType>
+Status ParallelScannerBuilder<ParentType>::_load() {
+    _total_rows = 0;
+    for (auto&& [tablet, version] : _tablets) {
+        const auto tablet_id = tablet->tablet_id();
+        auto& rowsets = _all_rowsets[tablet_id];
+        RETURN_IF_ERROR(tablet->capture_consistent_rowsets({0, version}, &rowsets));
+
+        for (auto& rowset : rowsets) {
+            RETURN_IF_ERROR(rowset->load());
+            const auto rowset_id = rowset->rowset_id();
+            auto& segment_cache_handle = _segment_cache_handles[rowset_id];
+
+            RETURN_IF_ERROR(SegmentLoader::instance()->load_segments(
+                    std::dynamic_pointer_cast<BetaRowset>(rowset), &segment_cache_handle, true));
+            _total_rows += rowset->num_rows();
+        }
+    }
+
+    _rows_per_scanner = _total_rows / _max_scanners_count;
+    _rows_per_scanner = std::max<size_t>(_rows_per_scanner, _min_rows_per_scanner);
+
+    return Status::OK();
+}
+
+template <typename ParentType>
+std::shared_ptr<NewOlapScanner> ParallelScannerBuilder<ParentType>::_build_scanner(
+        BaseTabletSPtr tablet, int64_t version, const std::vector<OlapScanRange*>& key_ranges,
+        TabletReader::ReadSource&& read_source) {
+    NewOlapScanner::Params params {
+            _state,  _scanner_profile.get(), key_ranges,         std::move(tablet),
+            version, std::move(read_source), _limit_per_scanner, _is_preaggregation};
+    return NewOlapScanner::create_shared(_parent, std::move(params));
+}
+
+template class ParallelScannerBuilder<NewOlapScanNode>;
+
+} // namespace doris

be/src/olap/parallel_scanner_builder.h

@@ -0,0 +1,98 @@
+// 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.
+
+#pragma once
+
+#include <memory>
+#include <string>
+#include <utility>
+
+#include "olap/rowset/segment_v2/row_ranges.h"
+#include "olap/segment_loader.h"
+#include "olap/tablet.h"
+#include "vec/exec/scan/new_olap_scanner.h"
+
+namespace doris {
+
+namespace vectorized {
+class VScanner;
+}
+
+using TabletSPtr = std::shared_ptr<Tablet>;
+using VScannerSPtr = std::shared_ptr<vectorized::VScanner>;
+
+struct TabletWithVersion {
+    TabletSPtr tablet;
+    int64_t version;
+};
+
+template <typename ParentType>
+class ParallelScannerBuilder {
+public:
+    ParallelScannerBuilder(ParentType* parent, const std::vector<TabletWithVersion>& tablets,
+                           const std::shared_ptr<RuntimeProfile>& profile,
+                           const std::vector<OlapScanRange*>& key_ranges, RuntimeState* state,
+                           int64_t limit_per_scanner, bool is_dup_mow_key, bool is_preaggregation)
+            : _parent(parent),
+              _scanner_profile(profile),
+              _state(state),
+              _limit_per_scanner(limit_per_scanner),
+              _is_dup_mow_key(is_dup_mow_key),
+              _is_preaggregation(is_preaggregation),
+              _tablets(tablets.cbegin(), tablets.cend()),
+              _key_ranges(key_ranges.cbegin(), key_ranges.cend()) {}
+
+    Status build_scanners(std::list<VScannerSPtr>& scanners);
+
+    void set_max_scanners_count(size_t count) { _max_scanners_count = count; }
+
+    void set_min_rows_per_scanner(int64_t size) { _min_rows_per_scanner = size; }
+
+private:
+    Status _load();
+
+    Status _build_scanners_by_rowid(std::list<VScannerSPtr>& scanners);
+
+    std::shared_ptr<vectorized::NewOlapScanner> _build_scanner(
+            BaseTabletSPtr tablet, int64_t version, const std::vector<OlapScanRange*>& key_ranges,
+            TabletReader::ReadSource&& read_source);
+
+    ParentType* _parent;
+
+    /// Max scanners count limit to build
+    size_t _max_scanners_count {16};
+
+    /// Min rows per scanner
+    size_t _min_rows_per_scanner {2 * 1024 * 1024};
+
+    size_t _total_rows {};
+
+    size_t _rows_per_scanner {_min_rows_per_scanner};
+
+    std::map<RowsetId, SegmentCacheHandle> _segment_cache_handles;
+
+    std::shared_ptr<RuntimeProfile> _scanner_profile;
+    RuntimeState* _state;
+    int64_t _limit_per_scanner;
+    bool _is_dup_mow_key;
+    bool _is_preaggregation;
+    std::vector<TabletWithVersion> _tablets;
+    std::vector<OlapScanRange*> _key_ranges;
+    std::unordered_map<int64_t, std::vector<RowsetSharedPtr>> _all_rowsets;
+};
+
+} // namespace doris

be/src/olap/rowset/beta_rowset_reader.cpp

@@ -237,10 +237,23 @@ Status BetaRowsetReader::get_segment_iterators(RowsetReaderContext* read_context
     for (int i = seg_start; i < seg_end; i++) {
         auto& seg_ptr = segments[i];
         std::unique_ptr<RowwiseIterator> iter;
-        auto s = seg_ptr->new_iterator(_input_schema, _read_options, &iter);
-        if (!s.ok()) {
-            LOG(WARNING) << "failed to create iterator[" << seg_ptr->id() << "]: " << s.to_string();
-            return Status::Error<ROWSET_READER_INIT>(s.to_string());
+        Status status;
+
+        /// If `_segment_row_ranges` is empty, the segment is not split.
+        if (_segment_row_ranges.empty()) {
+            _read_options.row_ranges.clear();
+            status = seg_ptr->new_iterator(_input_schema, _read_options, &iter);
+        } else {
+            DCHECK_EQ(seg_end - seg_start, _segment_row_ranges.size());
+            auto local_options = _read_options;
+            local_options.row_ranges = _segment_row_ranges[i - seg_start];
+            status = seg_ptr->new_iterator(_input_schema, local_options, &iter);
+        }
+
+        if (!status.ok()) {
+            LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
+                         << "]: " << status.to_string();
+            return Status::Error<ROWSET_READER_INIT>(status.to_string());
         }
         if (iter->empty()) {
             continue;
@@ -255,6 +268,7 @@ Status BetaRowsetReader::init(RowsetReaderContext* read_context, const RowSetSpl
     _read_context = read_context;
     _read_context->rowset_id = _rowset->rowset_id();
     _segment_offsets = rs_splits.segment_offsets;
+    _segment_row_ranges = rs_splits.segment_row_ranges;
     return Status::OK();
 }
 

be/src/olap/rowset/beta_rowset_reader.h

@@ -96,6 +96,7 @@ class BetaRowsetReader : public RowsetReader {
     DorisCallOnce<Status> _init_iter_once;
 
     std::pair<int, int> _segment_offsets;
+    std::vector<RowRanges> _segment_row_ranges;
 
     SchemaSPtr _input_schema;
     RowsetReaderContext* _read_context = nullptr;

be/src/olap/rowset/rowset_reader.h

@@ -25,6 +25,7 @@
 #include "olap/iterators.h"
 #include "olap/rowset/rowset_fwd.h"
 #include "olap/rowset/rowset_reader_context.h"
+#include "olap/rowset/segment_v2/row_ranges.h"
 #include "vec/core/block.h"
 
 namespace doris {
@@ -41,6 +42,9 @@ struct RowSetSplits {
     // and pipeline
     std::pair<int, int> segment_offsets;
 
+    // RowRanges of each segment.
+    std::vector<RowRanges> segment_row_ranges;
+
     RowSetSplits(RowsetReaderSharedPtr rs_reader_)
             : rs_reader(rs_reader_), segment_offsets({0, 0}) {}
     RowSetSplits() = default;

be/src/olap/rowset/segment_v2/segment_iterator.cpp

@@ -381,6 +381,10 @@ Status SegmentIterator::_lazy_init() {
                    << _opts.delete_bitmap.at(segment_id())->cardinality() << ", "
                    << _opts.stats->rows_del_by_bitmap << " rows deleted by bitmap";
     }
+
+    if (!_opts.row_ranges.is_empty()) {
+        _row_bitmap &= RowRanges::ranges_to_roaring(_opts.row_ranges);
+    }
     if (_opts.read_orderby_key_reverse) {
         _range_iter.reset(new BackwardBitmapRangeIterator(_row_bitmap));
     } else {