[Enhancement] Make the segment iterator reusable across scans

Add the segment-layer machinery for reusing a SegmentIterator across multiple
scans instead of building a fresh one each time, plus the prepared-scan-range
plumbing that a reused iterator consumes:

- SegmentIterator::reset_for_reuse() / _init_reused_scan() and a one-time-setup
  split (ScanContext::reset, _init_scan_range_and_context, _one_time_setup_done)
  so per-scan state is reset while the expensive one-time setup is preserved;
- Segment::new_reusable_iterator() + new_reusable_segment_iterator();
- NonClosingChunkIterator so a finishing driver does not tear down the shared
  reused iterator;
- SegmentReadStateCache on SegmentReadOptions and _apply_precomputed_scan_range()
  so a reused iterator can skip re-pruning when a precomputed scan range is
  supplied, plus the _get_prepared_pruned_row_ranges() helper that computes it;
- BitmapIndexEvaluator::init() idempotency reset so it can be re-initialized on
  reuse.

Infrastructure only: nothing populates the read-state cache or calls
reset_for_reuse yet (the lake prepared physical split scan follow-ups do). No
behavior change on the existing scan path.

Signed-off-by: stephen <stephen5217@163.com>

Author: stephen-shelby

be/src/storage/CMakeLists.txt

@@ -46,6 +46,7 @@ set(STORAGE_FILES
     segment_flush_executor.cpp
     segment_replicate_executor.cpp
     metadata_util.cpp
+    non_closing_chunk_iterator.cpp
     kv_store.cpp
     local_tablet_reader.cpp
     olap_server.cpp

be/src/storage/non_closing_chunk_iterator.cpp

@@ -0,0 +1,58 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed 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
+//
+//     https://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 "storage/non_closing_chunk_iterator.h"
+
+#include <memory>
+#include <unordered_set>
+#include <utility>
+#include <vector>
+
+namespace starrocks {
+
+class NonClosingChunkIterator final : public ChunkIterator {
+public:
+    explicit NonClosingChunkIterator(ChunkIteratorPtr child)
+            : ChunkIterator(child->schema(), child->chunk_size()), _child(std::move(child)) {}
+
+    void close() override {}
+
+    size_t merged_rows() const override { return _child->merged_rows(); }
+
+    Status init_encoded_schema(ColumnIdToGlobalDictMap& dict_maps) override {
+        RETURN_IF_ERROR(ChunkIterator::init_encoded_schema(dict_maps));
+        return _child->init_encoded_schema(dict_maps);
+    }
+
+    Status init_output_schema(const std::unordered_set<uint32_t>& unused_output_column_ids) override {
+        RETURN_IF_ERROR(ChunkIterator::init_output_schema(unused_output_column_ids));
+        return _child->init_output_schema(unused_output_column_ids);
+    }
+
+protected:
+    Status do_get_next(Chunk* chunk) override { return _child->get_next(chunk); }
+    Status do_get_next(Chunk* chunk, std::vector<uint32_t>* rowid) override { return _child->get_next(chunk, rowid); }
+    Status do_get_next(Chunk* chunk, std::vector<uint64_t>* rssid_rowids) override {
+        return _child->get_next(chunk, rssid_rowids);
+    }
+
+private:
+    ChunkIteratorPtr _child;
+};
+
+ChunkIteratorPtr new_non_closing_chunk_iterator(const ChunkIteratorPtr& child) {
+    return std::make_shared<NonClosingChunkIterator>(child);
+}
+
+} // namespace starrocks

be/src/storage/non_closing_chunk_iterator.h

@@ -0,0 +1,25 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed 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
+//
+//     https://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 "storage_primitive/chunk_iterator.h"
+
+namespace starrocks {
+
+// Wraps |child| as a per-scan iterator without taking ownership of its close
+// lifecycle. The slot owner must close |child| explicitly.
+ChunkIteratorPtr new_non_closing_chunk_iterator(const ChunkIteratorPtr& child);
+
+} // namespace starrocks

be/src/storage/rowset/bitmap_index_evaluator.cpp

@@ -414,6 +414,11 @@ void BitmapIndexEvaluator::close() {
 }
 
 Status BitmapIndexEvaluator::init(BitmapIndexIteratorSupplier&& supplier) {
+    close();
+    _closed = false;
+    _ctx = BitmapContext{};
+    _has_bitmap_index = false;
+    _bitmap_index_iterators.clear();
     _bitmap_index_iterators.resize(ChunkSchemaHelper::max_column_id(_schema) + 1, nullptr);
     ASSIGN_OR_RETURN(_has_bitmap_index,
                      _pred_tree.visit(BitmapIndexInitializer{this, supplier}, nullptr, CompoundNodeType::AND));

be/src/storage/rowset/segment.cpp

@@ -338,9 +338,7 @@ struct SegmentZoneMapPruner {
     const SegmentReadOptions& read_options;
 };
 
-StatusOr<ChunkIteratorPtr> Segment::_new_iterator(const Schema& schema, const SegmentReadOptions& read_options) {
-    DCHECK(read_options.stats != nullptr);
-
+Status Segment::_prune_by_segment_zone_map(const SegmentReadOptions& read_options) {
     const auto pruned = config::enable_index_segment_level_zonemap_filter &&
                         read_options.pred_tree_for_zone_map.visit(SegmentZoneMapPruner{this, read_options});
     if (pruned) {
@@ -349,7 +347,13 @@ StatusOr<ChunkIteratorPtr> Segment::_new_iterator(const Schema& schema, const Se
         }
         return Status::EndOfFile(strings::Substitute("End of file $0, empty iterator", _segment_file_info.path));
     }
+    return Status::OK();
+}
 
+StatusOr<ChunkIteratorPtr> Segment::_new_iterator(const Schema& schema, const SegmentReadOptions& read_options) {
+    DCHECK(read_options.stats != nullptr);
+
+    RETURN_IF_ERROR(_prune_by_segment_zone_map(read_options));
     return new_segment_iterator(shared_from_this(), schema, read_options);
 }
 
@@ -360,6 +364,17 @@ StatusOr<ChunkIteratorPtr> Segment::new_iterator(const Schema& schema, const Seg
     return _new_iterator(schema, read_options);
 }
 
+StatusOr<ChunkIteratorPtr> Segment::new_reusable_iterator(const Schema& iterator_schema, const Schema& output_schema,
+                                                          const SegmentReadOptions& read_options,
+                                                          ChunkIteratorPtr* reusable_slot) {
+    if (read_options.stats == nullptr) {
+        return Status::InvalidArgument("stats is null pointer");
+    }
+    RETURN_IF_ERROR(_prune_by_segment_zone_map(read_options));
+    return new_reusable_segment_iterator(shared_from_this(), iterator_schema, output_schema, read_options,
+                                         reusable_slot);
+}
+
 Status Segment::new_inverted_index_iterator(uint32_t ucid, InvertedIndexIterator** iter, const SegmentReadOptions& opts,
                                             const IndexReadOptions& index_opt) {
     auto column_reader_iter = _column_readers.find(ucid);

be/src/storage/rowset/segment.h

@@ -108,6 +108,9 @@ class Segment : public std::enable_shared_from_this<Segment> {
 
     // may return EndOfFile
     StatusOr<ChunkIteratorPtr> new_iterator(const Schema& schema, const SegmentReadOptions& read_options);
+    StatusOr<ChunkIteratorPtr> new_reusable_iterator(const Schema& iterator_schema, const Schema& output_schema,
+                                                     const SegmentReadOptions& read_options,
+                                                     ChunkIteratorPtr* reusable_slot);
 
     StatusOr<std::shared_ptr<Segment>> new_dcg_segment(const DeltaColumnGroup& dcg, uint32_t idx,
                                                        const TabletSchemaCSPtr& read_tablet_schema);
@@ -302,6 +305,7 @@ class Segment : public std::enable_shared_from_this<Segment> {
                                               std::unordered_map<uint32_t, uint32_t>& column_id_to_footer_ordinal);
 
     StatusOr<ChunkIteratorPtr> _new_iterator(const Schema& schema, const SegmentReadOptions& read_options);
+    Status _prune_by_segment_zone_map(const SegmentReadOptions& read_options);
 
     bool _use_segment_zone_map_filter(const SegmentReadOptions& read_options);