refactor(type): refactor ListValue using array

src/batch/src/executor/aggregation/orderby.rs

@@ -172,15 +172,7 @@ mod tests {
         agg.update(&mut state, &chunk).await.unwrap();
         assert_eq!(
             agg.get_result(&state).await.unwrap(),
-            Some(
-                ListValue::new(vec![
-                    Some(789.into()),
-                    Some(456.into()),
-                    Some(123.into()),
-                    Some(321.into()),
-                ])
-                .into()
-            )
+            Some(ListValue::from_iter([789, 456, 123, 321]).into())
         );
     }
 

src/batch/src/executor/filter.rs

@@ -137,17 +137,15 @@ mod tests {
 
     #[tokio::test]
     async fn test_list_filter_executor() {
-        use risingwave_common::array::{ArrayBuilder, ListArrayBuilder, ListRef, ListValue};
+        use risingwave_common::array::{ArrayBuilder, ListArrayBuilder, ListValue};
         use risingwave_common::types::Scalar;
 
         let mut builder = ListArrayBuilder::with_type(4, DataType::List(Box::new(DataType::Int32)));
 
         // Add 4 ListValues to ArrayBuilder
-        (1..=4).for_each(|i| {
-            builder.append(Some(ListRef::ValueRef {
-                val: &ListValue::new(vec![Some(i.to_scalar_value())]),
-            }));
-        });
+        for i in 1..=4 {
+            builder.append(Some(ListValue::from_iter([i]).as_scalar_ref()));
+        }
 
         // Use builder to obtain a single (List) column DataChunk
         let chunk = DataChunk::new(vec![builder.finish().into_ref()], 4);
@@ -185,15 +183,11 @@ mod tests {
             // Assert that values 3 and 4 are bigger than 2
             assert_eq!(
                 col1.value_at(0),
-                Some(ListRef::ValueRef {
-                    val: &ListValue::new(vec![Some(3.to_scalar_value())]),
-                })
+                Some(ListValue::from_iter([3]).as_scalar_ref())
             );
             assert_eq!(
                 col1.value_at(1),
-                Some(ListRef::ValueRef {
-                    val: &ListValue::new(vec![Some(4.to_scalar_value())]),
-                })
+                Some(ListValue::from_iter([4]).as_scalar_ref())
             );
         }
         let res = stream.next().await;

src/batch/src/executor/order_by.rs

@@ -613,17 +613,11 @@ mod tests {
                 },
                 {
                     list_builder.append(None);
-                    list_builder.append(Some(ListRef::ValueRef {
-                        val: &ListValue::new(vec![
-                            Some(1i64.to_scalar_value()),
-                            None,
-                            Some(3i64.to_scalar_value()),
-                        ]),
-                    }));
+                    list_builder.append(Some(
+                        ListValue::from_iter([Some(1i64), None, Some(3i64)]).as_scalar_ref(),
+                    ));
                     list_builder.append(None);
-                    list_builder.append(Some(ListRef::ValueRef {
-                        val: &ListValue::new(vec![Some(2i64.to_scalar_value())]),
-                    }));
+                    list_builder.append(Some(ListValue::from_iter([2i64]).as_scalar_ref()));
                     list_builder.append(None);
                     list_builder.finish().into_ref()
                 },
@@ -675,16 +669,10 @@ mod tests {
                 },
                 {
                     list_builder.append(None);
-                    list_builder.append(Some(ListRef::ValueRef {
-                        val: &ListValue::new(vec![Some(2i64.to_scalar_value())]),
-                    }));
-                    list_builder.append(Some(ListRef::ValueRef {
-                        val: &ListValue::new(vec![
-                            Some(1i64.to_scalar_value()),
-                            None,
-                            Some(3i64.to_scalar_value()),
-                        ]),
-                    }));
+                    list_builder.append(Some(ListValue::from_iter([2i64]).as_scalar_ref()));
+                    list_builder.append(Some(
+                        ListValue::from_iter([Some(1i64), None, Some(3i64)]).as_scalar_ref(),
+                    ));
                     list_builder.append(None);
                     list_builder.append(None);
                     list_builder.finish().into_ref()

src/common/benches/bench_encoding.rs

@@ -124,7 +124,7 @@ fn bench_encoding(c: &mut Criterion) {
         Case::new(
             "List of Bool (len = 100)",
             DataType::List(Box::new(DataType::Boolean)),
-            ScalarImpl::List(ListValue::new(vec![Some(ScalarImpl::Bool(true)); 100])),
+            ScalarImpl::List(ListValue::from_iter([true; 100])),
         ),
     ];
 

src/common/src/array/arrow.rs

@@ -674,11 +674,11 @@ impl TryFrom<&arrow_array::ListArray> for ListArray {
     type Error = ArrayError;
 
     fn try_from(array: &arrow_array::ListArray) -> Result<Self, Self::Error> {
-        let iter: Vec<_> = array
-            .iter()
-            .map(|o| o.map(|a| ArrayImpl::try_from(&a)).transpose())
-            .try_collect()?;
-        Ok(ListArray::from_iter(iter, (&array.value_type()).into()))
+        Ok(ListArray {
+            value: Box::new(ArrayImpl::try_from(array.values())?),
+            bitmap: array.nulls().expect("no null bitmap").iter().collect(),
+            offsets: array.offsets().iter().map(|o| *o as u32).collect(),
+        })
     }
 }
 
@@ -904,15 +904,7 @@ mod tests {
 
     #[test]
     fn list() {
-        let array = ListArray::from_iter(
-            [
-                Some(I32Array::from_iter([None, Some(-7), Some(25)]).into()),
-                None,
-                Some(I32Array::from_iter([Some(0), Some(-127), Some(127), Some(50)]).into()),
-                Some(I32Array::from_iter([0; 0]).into()),
-            ],
-            DataType::Int32,
-        );
+        let array = ListArray::from_iter([None, Some(vec![0, -127, 127, 50]), Some(vec![0; 0])]);
         let arrow = arrow_array::ListArray::from(&array);
         assert_eq!(ListArray::try_from(&arrow).unwrap(), array);
     }

src/common/src/array/bool_array.rs

@@ -125,7 +125,7 @@ impl Array for BoolArray {
 }
 
 /// `BoolArrayBuilder` constructs a `BoolArray` from `Option<Bool>`.
-#[derive(Debug)]
+#[derive(Debug, Clone, EstimateSize)]
 pub struct BoolArrayBuilder {
     bitmap: BitmapBuilder,
     data: BitmapBuilder,

src/common/src/array/bytes_array.rs

@@ -25,19 +25,11 @@ use crate::estimate_size::EstimateSize;
 use crate::util::iter_util::ZipEqDebug;
 
 /// `BytesArray` is a collection of Rust `[u8]`s.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, EstimateSize)]
 pub struct BytesArray {
-    offset: Vec<u32>,
+    offset: Box<[u32]>,
     bitmap: Bitmap,
-    data: Vec<u8>,
-}
-
-impl EstimateSize for BytesArray {
-    fn estimated_heap_size(&self) -> usize {
-        self.offset.capacity() * size_of::<u32>()
-            + self.bitmap.estimated_heap_size()
-            + self.data.capacity()
-    }
+    data: Box<[u8]>,
 }
 
 impl Array for BytesArray {
@@ -86,7 +78,7 @@ impl Array for BytesArray {
             },
             Buffer {
                 compression: CompressionType::None as i32,
-                body: data_buffer,
+                body: data_buffer.into(),
             },
         ];
         let null_bitmap = self.null_bitmap().to_protobuf();
@@ -146,7 +138,7 @@ impl<'a> FromIterator<&'a [u8]> for BytesArray {
 }
 
 /// `BytesArrayBuilder` use `&[u8]` to build an `BytesArray`.
-#[derive(Debug)]
+#[derive(Debug, Clone, EstimateSize)]
 pub struct BytesArrayBuilder {
     offset: Vec<u32>,
     bitmap: BitmapBuilder,
@@ -221,9 +213,9 @@ impl ArrayBuilder for BytesArrayBuilder {
 
     fn finish(self) -> BytesArray {
         BytesArray {
-            bitmap: (self.bitmap).finish(),
-            data: self.data,
-            offset: self.offset,
+            bitmap: self.bitmap.finish(),
+            data: self.data.into(),
+            offset: self.offset.into(),
         }
     }
 }

src/common/src/array/data_chunk.rs

@@ -1107,7 +1107,7 @@ mod tests {
     #[test]
     fn test_chunk_estimated_size() {
         assert_eq!(
-            96,
+            72,
             DataChunk::from_pretty(
                 "I I I
                  1 5 2
@@ -1117,7 +1117,7 @@ mod tests {
             .estimated_heap_size()
         );
         assert_eq!(
-            64,
+            48,
             DataChunk::from_pretty(
                 "I I
                  1 2

src/common/src/array/jsonb_array.rs

@@ -19,13 +19,13 @@ use crate::buffer::{Bitmap, BitmapBuilder};
 use crate::estimate_size::EstimateSize;
 use crate::types::{DataType, JsonbRef, JsonbVal, Scalar};
 
-#[derive(Debug)]
+#[derive(Debug, Clone, EstimateSize)]
 pub struct JsonbArrayBuilder {
     bitmap: BitmapBuilder,
     builder: jsonbb::Builder,
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, EstimateSize)]
 pub struct JsonbArray {
     bitmap: Bitmap,
     /// Elements are stored as a single JSONB array value.
@@ -177,8 +177,14 @@ impl FromIterator<JsonbVal> for JsonbArray {
     }
 }
 
-impl EstimateSize for JsonbArray {
+impl EstimateSize for jsonbb::Value {
     fn estimated_heap_size(&self) -> usize {
-        self.bitmap.estimated_heap_size() + self.data.capacity()
+        self.capacity()
+    }
+}
+
+impl EstimateSize for jsonbb::Builder {
+    fn estimated_heap_size(&self) -> usize {
+        self.capacity()
     }
 }