Repository: risingwavelabs/risingwave
File: src/stream/src/common/log_store_impl/kv_log_store/buffer.rs (Rust)

Code before:
            LogStoreBufferItem::Barrier { .. } => 0,
        }
    }

    fn estimated_size(&self) -> usize {
        match self {
            LogStoreBufferItem::StreamChunk {
                chunk,
                start_seq_id: _,
                end_seq_id: _,
                flushed: _,
                chunk_id: _,
            } => {
                chunk.estimated_size()
                    + size_of::<SeqId>() * 2
                    + size_of::<bool>()
                    + size_of::<ChunkId>()
            }
            LogStoreBufferItem::Flushed {
                vnode_bitmap,
                start_seq_id: _,
                end_seq_id: _,
                chunk_id: _,
            } => vnode_bitmap.estimated_size() + size_of::<SeqId>() * 2 + size_of::<ChunkId>(),
            LogStoreBufferItem::Barrier { .. } => 0,
        }
    }
}

struct LogStoreBufferInner {
    /// Items not read by log reader. Newer item at the front
    unconsumed_queue: VecDeque<(u64, LogStoreBufferItem)>,
    /// Items already read by log reader by not truncated. Newer item at the front
    consumed_queue: VecDeque<(u64, LogStoreBufferItem)>,
    row_count: usize,
    max_row_count: usize,
    chunk_size: usize,

    truncation_list: VecDeque<ReaderTruncationOffsetType>,

    next_chunk_id: ChunkId,

    metrics: KvLogStoreMetrics,
}

impl LogStoreBufferInner {
    fn update_buffer_metrics(&self) {
        let mut epoch_count = 0;
        let mut row_count = 0;
        let mut memory_bytes = 0;
        for (_, item) in &self.unconsumed_queue {

Review comment (suggestion):
The `estimated_size` implementation includes stack-allocated fields (SeqId, bool, ChunkId) which are already accounted for in the enum's stack size. The `EstimateSize` trait should typically only count heap-allocated memory. Consider whether these `size_of` additions are necessary or if `estimated_heap_size` should be used instead.
```suggestion
        size_of::<Self>() + self.estimated_heap_size()
```

Code after:
            LogStoreBufferItem::Barrier { .. } => 0,
        }
    }

    fn estimated_size(&self) -> usize {
        size_of::<Self>() + self.estimated_heap_size()
    }
}

struct LogStoreBufferInner {
    /// Items not read by log reader. Newer item at the front
    unconsumed_queue: VecDeque<(u64, LogStoreBufferItem)>,
    /// Items already read by log reader by not truncated. Newer item at the front
    consumed_queue: VecDeque<(u64, LogStoreBufferItem)>,
    row_count: usize,
    max_row_count: usize,
    chunk_size: usize,

    truncation_list: VecDeque<ReaderTruncationOffsetType>,

    next_chunk_id: ChunkId,

    metrics: KvLogStoreMetrics,
}

impl LogStoreBufferInner {
    fn update_buffer_metrics(&self) {
        let mut epoch_count = 0;
        let mut row_count = 0;
        let mut memory_bytes = 0;
        for (_, item) in &self.unconsumed_queue {
            match item {
                LogStoreBufferItem::StreamChunk { chunk, .. } => {
                    row_count += chunk.cardinality();
                }
                LogStoreBufferItem::Flushed {
                    start_seq_id,
                    end_seq_id,
                    ..
                } => {
                    row_count += (end_seq_id - start_seq_id) as usize;
                }
                LogStoreBufferItem::Barrier { .. } => {
                    epoch_count += 1;
                }
            }
            memory_bytes += item.estimated_size();
        }
        for (_, item) in &self.consumed_queue {
            memory_bytes += item.estimated_size();
        }