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

Code before:
            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
            .get_stream_for_blocks(
                self.sstable_info.object_id,
                &self.block_metas[self.block_idx..],
            )
            .instrument_await("stream_iter_get_stream".verbose())
            .await?;
        self.block_stream = Some(block_stream);
        Ok(())
    }

    async fn prune_from_valid_block_iter(&mut self) -> HummockResult<()> {
        while let Some(block_iter) = self.block_iter.as_mut() {
            if self.sstable_table_ids.contains(&block_iter.table_id()) {
                return Ok(());
            } else {
                self.next_block().await?;
            }
        }
        Ok(())
    }

    /// Initialises the iterator by moving it to the first KV-pair in the stream's first block where
    /// key >= `seek_key`. If that block does not contain such a KV-pair, the iterator continues to
    /// the first KV-pair of the next block. If `seek_key` is not given, the iterator will move to
    /// the very first KV-pair of the stream's first block.
    pub async fn seek(&mut self, seek_key: Option<FullKey<&[u8]>>) -> HummockResult<()> {
        // Load first block.
        self.next_block().await?;

        // We assume that a block always contains at least one KV pair. Subsequently, if
        // `next_block()` loads a new block (i.e., `block_iter` is not `None`), then `block_iter` is
        // also valid and pointing on the block's first KV-pair.

        let seek_key = if let Some(seek_key) = seek_key {
            if seek_key.cmp(&self.key_range_left.to_ref()).is_lt() {
                Some(self.key_range_left.to_ref())
            } else {

Review comment (suggestion):
The `block_metas_range.clone()` is unnecessary here as `Range` implements `Copy`. Remove `.clone()` to avoid redundant operations.

Code after:
        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,
            io_retry_times: 0,
            max_io_retry_times,
            key_range_left,
            key_range_right,
            key_range_right_exclusive,
        }
    }

    /// Returns the block metas slice for this iterator.
    #[inline]
    fn block_metas(&self) -> &[BlockMeta] {
        &self.sstable.meta.block_metas[self.block_metas_range.clone()]
    }

    /// Returns the number of blocks in this iterator.
    #[inline]
    fn block_count(&self) -> usize {
        self.block_metas_range.len()
    }

    async fn create_stream(&mut self) -> HummockResult<()> {
        let block_stream = self
            .sstable_store
            .get_stream_for_blocks(
                self.sstable_info.object_id,
                &self.block_metas()[self.block_idx..],
            )
            .instrument_await("stream_iter_get_stream".verbose())
            .await?;
        self.block_stream = Some(block_stream);
        Ok(())
    }

    async fn prune_from_valid_block_iter(&mut self) -> HummockResult<()> {
        while let Some(block_iter) = self.block_iter.as_mut() {
            if self.sstable_table_ids.contains(&block_iter.table_id()) {
                return Ok(());