Repository: risingwavelabs/risingwave
File: src/storage/src/hummock/compactor/iterator.rs (Rust)

Code before:
    // BlockIterator works as follows: After new(), we call seek(). That brings us
    // to the first element. Calling next() then brings us to the second element and does not
    // return anything.

    // BlockStream follows a different approach. After new(), we do not seek, instead next()
    // returns the first value.

    /// Initialises a new [`SstableStreamIterator`] which iterates over the given [`BlockDataStream`].
    /// The iterator reads at most `max_block_count` from the stream.
    pub fn new(
        block_metas: Vec<BlockMeta>,
        sstable_info: SstableInfo,
        stats: &StoreLocalStatistic,
        task_progress: Arc<TaskProgress>,
        sstable_store: SstableStoreRef,
        max_io_retry_times: usize,
    ) -> Self {
        let sstable_table_ids = HashSet::from_iter(sstable_info.table_ids.iter().cloned());

        // filter the block meta with key range
        let block_metas = filter_block_metas(
            &block_metas,
            &sstable_table_ids,
            sstable_info.key_range.clone(),
        );

        let key_range_left = FullKey::decode(&sstable_info.key_range.left).to_vec();
        let key_range_right = FullKey::decode(&sstable_info.key_range.right).to_vec();
        let key_range_right_exclusive = sstable_info.key_range.right_exclusive;

        Self {
            block_stream: None,
            block_iter: None,
            block_metas,
            block_idx: 0,
            stats_ptr: stats.remote_io_time.clone(),
            sstable_table_ids,
            sstable_info,
            sstable_store,
            task_progress,
            io_retry_times: 0,
            max_io_retry_times,
            key_range_left,
            key_range_right,
            key_range_right_exclusive,
        }
    }

    async fn create_stream(&mut self) -> HummockResult<()> {
        let block_stream = self
            .sstable_store

Review comment (performance):
The `block_metas_range.clone()` creates an unnecessary copy of the range for slicing. Since `Range` is `Copy`, you can use `block_metas_range` directly without `.clone()`.
```suggestion
            let block_metas = &sstable.meta.block_metas[block_metas_range];
```

Code after:
    // These iterators work in different ways.

    // BlockIterator works as follows: After new(), we call seek(). That brings us
    // to the first element. Calling next() then brings us to the second element and does not
    // return anything.

    // BlockStream follows a different approach. After new(), we do not seek, instead next()
    // returns the first value.

    /// Initialises a new [`SstableStreamIterator`] which iterates over the given [`BlockDataStream`].
    /// The iterator reads at most `max_block_count` from the stream.
    pub fn new(
        sstable: TableHolder,
        block_metas_range: Range<usize>,
        sstable_info: SstableInfo,
        stats: &StoreLocalStatistic,
        task_progress: Arc<TaskProgress>,
        sstable_store: SstableStoreRef,
        max_io_retry_times: usize,
    ) -> Self {
        let sstable_table_ids = HashSet::from_iter(sstable_info.table_ids.iter().cloned());

        // Further filter the block_metas_range with sstable_info.key_range
        // This is necessary when the SST is split into multiple SstableInfo with different key ranges
        let block_metas_range = {
            let block_metas = &sstable.meta.block_metas[block_metas_range.clone()];
            let inner_range = filter_block_metas(
                block_metas,
                &sstable_table_ids,
                sstable_info.key_range.clone(),
            );
            // Adjust the range to be relative to the original block_metas
            (block_metas_range.start + inner_range.start)
                ..(block_metas_range.start + inner_range.end)
        };

        let key_range_left = FullKey::decode(&sstable_info.key_range.left).to_vec();
        let key_range_right = FullKey::decode(&sstable_info.key_range.right).to_vec();
        let key_range_right_exclusive = sstable_info.key_range.right_exclusive;

        Self {
            block_stream: None,
            block_iter: None,
            sstable,
            block_metas_range,
            block_idx: 0,
            stats_ptr: stats.remote_io_time.clone(),
            sstable_table_ids,
            sstable_info,
            sstable_store,
            task_progress,