From 3fcf00cf6ecd69408dcc559ec94b214196ca5743 Mon Sep 17 00:00:00 2001
From: st1page <1245835950@qq.com>
Date: Tue, 12 Sep 2023 13:41:16 +0800
Subject: [PATCH] change interface

---
 src/common/src/array/compact_chunk.rs | 12 ++++--------
 1 file changed, 4 insertions(+), 8 deletions(-)

diff --git a/src/common/src/array/compact_chunk.rs b/src/common/src/array/compact_chunk.rs
index a4e0fa0b81f25..8634b5d3ca1a4 100644
--- a/src/common/src/array/compact_chunk.rs
+++ b/src/common/src/array/compact_chunk.rs
@@ -102,7 +102,7 @@ impl StreamChunkCompactor {
     /// and UPDATE DELETE will be converted to INSERT and DELETE, and dropped according to
     /// certain rules (see `merge_insert` and `merge_delete` for more details).
 
-    pub fn into_compacted_chunks(self) -> Vec<StreamChunk> {
+    pub fn into_compacted_chunks(self) -> impl Iterator<Item = StreamChunk> {
         let (chunks, key_indices) = self.into_inner();
 
         let estimate_size = chunks.iter().map(|c| c.cardinality()).sum();
@@ -155,18 +155,14 @@ impl StreamChunkCompactor {
                 }
             }
         }
-        chunks.into_iter().map(|(_, c)| c.into()).collect_vec()
+        chunks.into_iter().map(|(_, c)| c.into())
     }
 }
 
 pub fn merge_chunk_row(stream_chunk: StreamChunk, pk_indices: &[usize]) -> StreamChunk {
     let mut compactor = StreamChunkCompactor::new(pk_indices.to_vec());
     compactor.push_chunk(stream_chunk);
-    compactor
-        .into_compacted_chunks()
-        .into_iter()
-        .next()
-        .unwrap()
+    compactor.into_compacted_chunks().next().unwrap()
 }
 
 #[cfg(test)]
@@ -199,7 +195,7 @@ mod tests {
             + 2 2 2
             + 9 9 1",
         ));
-        let mut iter = compactor.into_compacted_chunks().into_iter();
+        let mut iter = compactor.into_compacted_chunks();
         assert_eq!(
             iter.next().unwrap().compact(),
             StreamChunk::from_pretty(