Repository: risingwavelabs/risingwave
File: src/stream/src/executor/mview/materialize.rs (Rust)

Code before:
        let lru_cache: ManagedLruCache<Vec<u8>, CacheValue> =
            ManagedLruCache::unbounded(watermark_sequence, metrics_info.clone());
        Self {
            lru_cache,
            row_serde,
            version_column_indices,
            _serde: PhantomData,
        }
    }

    async fn handle<S: StateStore>(
        &mut self,
        row_ops: Vec<(Op, Vec<u8>, Bytes)>,
        table: &StateTableInner<S, SD>,
        conflict_behavior: ConflictBehavior,
        metrics: &MaterializeMetrics,
        toastable_column_indices: Option<&[usize]>,
    ) -> StreamExecutorResult<ChangeBuffer> {
        assert_matches!(conflict_behavior, checked_conflict_behaviors!());

        let key_set: HashSet<Box<[u8]>> = row_ops
            .iter()
            .map(|(_, k, _)| k.as_slice().into())
            .collect();

        // Populate the LRU cache with the keys in input chunk.
        // For new keys, row values are set to None.
        self.fetch_keys(
            key_set.iter().map(|v| v.deref()),
            table,
            conflict_behavior,
            metrics,
        )
        .await?;

        let mut change_buffer = ChangeBuffer::new();
        let row_serde = self.row_serde.clone();
        let version_column_indices = self.version_column_indices.clone();
        for (op, key, row) in row_ops {
            match op {
                Op::Insert | Op::UpdateInsert => {
                    let Some(old_row) = self.get_expected(&key) else {
                        // not exists before, meaning no conflict, simply insert
                        let new_row_deserialized =
                            row_serde.deserializer.deserialize(row.clone())?;
                        change_buffer.insert(key.clone(), new_row_deserialized);
                        self.lru_cache.put(key, Some(CompactedRow { row }));
                        continue;
                    };

                    // now conflict happens, handle it according to the specified behavior

Review comment (suggestion):
nits: This is not the subject of this PR. **PrefetchOptions** should be an Optional type. PrefetchOptions::new(false, false) is not intuitive, as it means prefetching is disabled. Alternatively, we could construct it via a dedicated function, such as Prefetch::without_prefetch()....

Code after:
        metrics_info: MetricsInfo,
        row_serde: BasicSerde,
        version_column_indices: Vec<u32>,
    ) -> Self {
        let lru_cache: ManagedLruCache<Vec<u8>, CacheValue> =
            ManagedLruCache::unbounded(watermark_sequence, metrics_info.clone());
        Self {
            lru_cache,
            row_serde,
            version_column_indices,
            vnode_max_keys: BTreeMap::new(),
            _serde: PhantomData,
        }
    }

    async fn init_vnode_max_keys<S: StateStore>(
        &mut self,
        state_table: &StateTableInner<S, SD>,
    ) -> StreamExecutorResult<()> {
        assert!(self.vnode_max_keys.is_empty());
        'vnode_loop: for vnode in state_table.vnodes().iter_vnodes() {
            let table_iter = state_table
                .rev_iter_keyed_row_with_vnode(
                    vnode,
                    &(Bound::<OwnedRow>::Unbounded, Bound::<OwnedRow>::Unbounded),
                    PrefetchOptions::new(false, false),
                )
                .await?;

            #[for_await]
            for res in table_iter {
                let key_row = res?;
                assert_eq!(vnode, key_row.vnode());
                let prev = self
                    .vnode_max_keys
                    .insert(vnode, Some(key_row.key().to_vec()));
                assert_eq!(prev, None);
                continue 'vnode_loop;
            }
            let prev = self.vnode_max_keys.insert(vnode, None);
            assert_eq!(prev, None);
        }
        tracing::trace!("Initialized vnode max keys: {:?}", self.vnode_max_keys);
        assert_eq!(self.vnode_max_keys.len(), state_table.vnodes().count_ones());
        Ok(())
    }

    async fn handle<S: StateStore>(
        &mut self,
        row_ops: Vec<(Op, Vec<u8>, Bytes)>,
        table: &StateTableInner<S, SD>,