fix all imports

Signed-off-by: Richard Chien <[email protected]>

src/batch/Cargo.toml

@@ -34,6 +34,7 @@ parking_lot = { version = "0.12", features = ["arc_lock"] }
 paste = "1"
 prometheus = { version = "0.13", features = ["process"] }
 risingwave_common = { workspace = true }
+risingwave_common_estimate_size = { workspace = true }
 risingwave_connector = { workspace = true }
 risingwave_dml = { workspace = true }
 risingwave_expr = { workspace = true }

src/batch/src/executor/aggregation/distinct.rs

@@ -17,9 +17,9 @@ use std::ops::Range;
 
 use risingwave_common::array::StreamChunk;
 use risingwave_common::buffer::BitmapBuilder;
-use risingwave_common::estimate_size::EstimateSize;
 use risingwave_common::row::{OwnedRow, Row};
 use risingwave_common::types::{DataType, Datum};
+use risingwave_common_estimate_size::EstimateSize;
 use risingwave_expr::aggregate::{
     AggStateDyn, AggregateFunction, AggregateState, BoxedAggregateFunction,
 };

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

@@ -16,12 +16,12 @@ use std::ops::Range;
 
 use anyhow::Context;
 use risingwave_common::array::{Op, RowRef, StreamChunk};
-use risingwave_common::estimate_size::EstimateSize;
 use risingwave_common::row::{OwnedRow, Row, RowExt};
 use risingwave_common::types::{DataType, Datum};
 use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
 use risingwave_common::util::memcmp_encoding;
 use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
+use risingwave_common_estimate_size::EstimateSize;
 use risingwave_expr::aggregate::{
     AggStateDyn, AggregateFunction, AggregateState, BoxedAggregateFunction,
 };

src/batch/src/executor/hash_agg.rs

@@ -18,11 +18,11 @@ use futures_async_stream::try_stream;
 use itertools::Itertools;
 use risingwave_common::array::{DataChunk, StreamChunk};
 use risingwave_common::catalog::{Field, Schema};
-use risingwave_common::estimate_size::EstimateSize;
 use risingwave_common::hash::{HashKey, HashKeyDispatcher, PrecomputedBuildHasher};
 use risingwave_common::memory::MemoryContext;
 use risingwave_common::types::DataType;
 use risingwave_common::util::iter_util::ZipEqFast;
+use risingwave_common_estimate_size::EstimateSize;
 use risingwave_expr::aggregate::{AggCall, AggregateState, BoxedAggregateFunction};
 use risingwave_pb::batch_plan::plan_node::NodeBody;
 use risingwave_pb::batch_plan::HashAggNode;

src/batch/src/executor/join/hash_join.rs

@@ -28,6 +28,7 @@ use risingwave_common::row::{repeat_n, RowExt};
 use risingwave_common::types::{DataType, Datum};
 use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
 use risingwave_common::util::iter_util::ZipEqFast;
+use risingwave_common_estimate_size::EstimateSize;
 use risingwave_expr::expr::{build_from_prost, BoxedExpression, Expression};
 use risingwave_pb::batch_plan::plan_node::NodeBody;
 
@@ -36,7 +37,6 @@ use crate::error::{BatchError, Result};
 use crate::executor::{
     BoxedDataChunkStream, BoxedExecutor, BoxedExecutorBuilder, Executor, ExecutorBuilder,
 };
-use crate::risingwave_common::estimate_size::EstimateSize;
 use crate::risingwave_common::hash::NullBitmap;
 use crate::task::{BatchTaskContext, ShutdownToken};
 

src/batch/src/executor/join/lookup_join_base.rs

@@ -26,6 +26,7 @@ use risingwave_common::types::{DataType, ToOwnedDatum};
 use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
 use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_common::util::sort_util::{cmp_datum_iter, OrderType};
+use risingwave_common_estimate_size::EstimateSize;
 use risingwave_expr::expr::BoxedExpression;
 
 use crate::error::BatchError;
@@ -34,7 +35,6 @@ use crate::executor::{
     utils, BoxedDataChunkListStream, BoxedExecutor, BufferChunkExecutor, EquiJoinParams,
     HashJoinExecutor, JoinHashMap, JoinType, LookupExecutorBuilder, RowId,
 };
-use crate::risingwave_common::estimate_size::EstimateSize;
 use crate::task::ShutdownToken;
 
 /// Lookup Join Base.

src/batch/src/executor/join/nested_loop_join.rs

@@ -17,12 +17,12 @@ use risingwave_common::array::data_chunk_iter::RowRef;
 use risingwave_common::array::{Array, DataChunk};
 use risingwave_common::buffer::BitmapBuilder;
 use risingwave_common::catalog::Schema;
-use risingwave_common::estimate_size::EstimateSize;
 use risingwave_common::memory::MemoryContext;
 use risingwave_common::row::{repeat_n, RowExt};
 use risingwave_common::types::{DataType, Datum};
 use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
 use risingwave_common::util::iter_util::ZipEqDebug;
+use risingwave_common_estimate_size::EstimateSize;
 use risingwave_expr::expr::{
     build_from_prost as expr_build_from_prost, BoxedExpression, Expression,
 };

src/batch/src/executor/merge_sort_exchange.rs

@@ -17,10 +17,10 @@ use std::sync::Arc;
 use futures_async_stream::try_stream;
 use risingwave_common::array::DataChunk;
 use risingwave_common::catalog::{Field, Schema};
-use risingwave_common::estimate_size::EstimateSize;
 use risingwave_common::memory::{MemMonitoredHeap, MemoryContext, MonitoredGlobalAlloc};
 use risingwave_common::types::ToOwnedDatum;
 use risingwave_common::util::sort_util::{ColumnOrder, HeapElem};
+use risingwave_common_estimate_size::EstimateSize;
 use risingwave_pb::batch_plan::plan_node::NodeBody;
 use risingwave_pb::batch_plan::PbExchangeSource;
 

src/batch/src/executor/top_n.rs

@@ -19,12 +19,12 @@ use std::vec::Vec;
 use futures_async_stream::try_stream;
 use risingwave_common::array::DataChunk;
 use risingwave_common::catalog::Schema;
-use risingwave_common::estimate_size::EstimateSize;
 use risingwave_common::memory::{MemMonitoredHeap, MemoryContext};
 use risingwave_common::row::{OwnedRow, Row};
 use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
 use risingwave_common::util::memcmp_encoding::{encode_chunk, MemcmpEncoded};
 use risingwave_common::util::sort_util::ColumnOrder;
+use risingwave_common_estimate_size::EstimateSize;
 use risingwave_pb::batch_plan::plan_node::NodeBody;
 
 use crate::error::{BatchError, Result};

src/common/estimate_size/Cargo.toml

@@ -17,11 +17,13 @@ normal = ["workspace-hack"]
 [dependencies]
 bytes = "1"
 educe = "0.5"
+ethnum = { version = "1", features = ["serde"] }
 fixedbitset = "0.5"
+jsonbb = "0.1.2"
 lru = { workspace = true }
+risingwave_common_proc_macro = { workspace = true }
 rust_decimal = "1"
 serde_json = "1"
-risingwave_common_proc_macro = { workspace = true }
 
 [lints]
 workspace = true

src/common/estimate_size/src/lib.rs

@@ -17,13 +17,13 @@
 
 pub mod collections;
 
+use std::cmp::Reverse;
 use std::marker::PhantomData;
 use std::sync::atomic::{AtomicUsize, Ordering};
 
 use bytes::Bytes;
 use fixedbitset::FixedBitSet;
 pub use risingwave_common_proc_macro::EstimateSize;
-use rust_decimal::Decimal as RustDecimal;
 
 /// The trait for estimating the actual memory usage of a struct.
 ///
@@ -93,6 +93,18 @@ impl EstimateSize for serde_json::Value {
     }
 }
 
+impl EstimateSize for jsonbb::Value {
+    fn estimated_heap_size(&self) -> usize {
+        self.capacity()
+    }
+}
+
+impl EstimateSize for jsonbb::Builder {
+    fn estimated_heap_size(&self) -> usize {
+        self.capacity()
+    }
+}
+
 impl<T1: EstimateSize, T2: EstimateSize> EstimateSize for (T1, T2) {
     fn estimated_heap_size(&self) -> usize {
         self.0.estimated_heap_size() + self.1.estimated_heap_size()
@@ -127,13 +139,21 @@ impl<T: ZeroHeapSize> EstimateSize for Box<[T]> {
     }
 }
 
+impl<T: EstimateSize> EstimateSize for Reverse<T> {
+    fn estimated_heap_size(&self) -> usize {
+        self.0.estimated_heap_size()
+    }
+}
+
 impl<T: ZeroHeapSize, const LEN: usize> EstimateSize for [T; LEN] {
     fn estimated_heap_size(&self) -> usize {
         0
     }
 }
 
-impl ZeroHeapSize for RustDecimal {}
+impl ZeroHeapSize for rust_decimal::Decimal {}
+
+impl ZeroHeapSize for ethnum::I256 {}
 
 impl<T> ZeroHeapSize for PhantomData<T> {}
 

src/common/src/array/bool_array.rs

@@ -12,11 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use risingwave_common_estimate_size::EstimateSize;
 use risingwave_pb::data::{ArrayType, PbArray};
 
 use super::{Array, ArrayBuilder, DataType};
 use crate::buffer::{Bitmap, BitmapBuilder};
-use crate::estimate_size::EstimateSize;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct BoolArray {

src/common/src/array/bytes_array.rs

@@ -15,13 +15,13 @@
 use std::iter;
 use std::mem::size_of;
 
+use risingwave_common_estimate_size::EstimateSize;
 use risingwave_pb::common::buffer::CompressionType;
 use risingwave_pb::common::Buffer;
 use risingwave_pb::data::{ArrayType, PbArray};
 
 use super::{Array, ArrayBuilder, DataType};
 use crate::buffer::{Bitmap, BitmapBuilder};
-use crate::estimate_size::EstimateSize;
 use crate::util::iter_util::ZipEqDebug;
 
 /// `BytesArray` is a collection of Rust `[u8]`s.

src/common/src/array/data_chunk.rs

@@ -23,13 +23,13 @@ use either::Either;
 use itertools::Itertools;
 use rand::rngs::SmallRng;
 use rand::{Rng, SeedableRng};
+use risingwave_common_estimate_size::EstimateSize;
 use risingwave_pb::data::PbDataChunk;
 
 use super::{Array, ArrayImpl, ArrayRef, ArrayResult, StructArray};
 use crate::array::data_chunk_iter::RowRef;
 use crate::array::ArrayBuilderImpl;
 use crate::buffer::{Bitmap, BitmapBuilder};
-use crate::estimate_size::EstimateSize;
 use crate::field_generator::{FieldGeneratorImpl, VarcharProperty};
 use crate::hash::HashCode;
 use crate::row::Row;

src/common/src/array/jsonb_array.rs

@@ -12,11 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use risingwave_common_estimate_size::EstimateSize;
 use risingwave_pb::data::{PbArray, PbArrayType};
 
 use super::{Array, ArrayBuilder, ArrayImpl, ArrayResult};
 use crate::buffer::{Bitmap, BitmapBuilder};
-use crate::estimate_size::EstimateSize;
 use crate::types::{DataType, JsonbRef, JsonbVal, Scalar};
 
 #[derive(Debug, Clone, EstimateSize)]
@@ -176,15 +176,3 @@ impl FromIterator<JsonbVal> for JsonbArray {
         iter.into_iter().map(Some).collect()
     }
 }
-
-impl EstimateSize for jsonbb::Value {
-    fn estimated_heap_size(&self) -> usize {
-        self.capacity()
-    }
-}
-
-impl EstimateSize for jsonbb::Builder {
-    fn estimated_heap_size(&self) -> usize {
-        self.capacity()
-    }
-}

src/common/src/array/list_array.rs

@@ -20,6 +20,7 @@ use std::mem::size_of;
 
 use bytes::{Buf, BufMut};
 use itertools::Itertools;
+use risingwave_common_estimate_size::EstimateSize;
 use risingwave_pb::data::{ListArrayData, PbArray, PbArrayType};
 use serde::{Deserialize, Serializer};
 use thiserror_ext::AsReport;
@@ -29,7 +30,6 @@ use super::{
     PrimitiveArrayItemType, RowRef, Utf8Array,
 };
 use crate::buffer::{Bitmap, BitmapBuilder};
-use crate::estimate_size::EstimateSize;
 use crate::row::Row;
 use crate::types::{
     hash_datum, DataType, Datum, DatumRef, DefaultOrd, Scalar, ScalarImpl, ScalarRefImpl,

src/common/src/array/mod.rs

@@ -62,6 +62,7 @@ pub use jsonb_array::{JsonbArray, JsonbArrayBuilder};
 pub use list_array::{ListArray, ListArrayBuilder, ListRef, ListValue};
 use paste::paste;
 pub use primitive_array::{PrimitiveArray, PrimitiveArrayBuilder, PrimitiveArrayItemType};
+use risingwave_common_estimate_size::EstimateSize;
 use risingwave_pb::data::PbArray;
 pub use stream_chunk::{Op, StreamChunk, StreamChunkTestExt};
 pub use struct_array::{StructArray, StructArrayBuilder, StructRef, StructValue};
@@ -70,7 +71,6 @@ pub use utf8_array::*;
 pub use self::error::ArrayError;
 pub use crate::array::num256_array::{Int256Array, Int256ArrayBuilder};
 use crate::buffer::Bitmap;
-use crate::estimate_size::EstimateSize;
 use crate::types::*;
 use crate::{dispatch_array_builder_variants, dispatch_array_variants, for_all_array_variants};
 pub type ArrayResult<T> = Result<T, ArrayError>;

src/common/src/array/num256_array.rs

@@ -15,13 +15,13 @@
 use std::io::{Cursor, Read};
 
 use ethnum::I256;
+use risingwave_common_estimate_size::EstimateSize;
 use risingwave_pb::common::buffer::CompressionType;
 use risingwave_pb::common::Buffer;
 use risingwave_pb::data::PbArray;
 
 use crate::array::{Array, ArrayBuilder, ArrayImpl, ArrayResult};
 use crate::buffer::{Bitmap, BitmapBuilder};
-use crate::estimate_size::{EstimateSize, ZeroHeapSize};
 use crate::types::{DataType, Int256, Int256Ref, Scalar};
 
 #[derive(Debug, Clone, EstimateSize)]
@@ -36,8 +36,6 @@ pub struct Int256Array {
     data: Box<[I256]>,
 }
 
-impl ZeroHeapSize for I256 {}
-
 #[rustfmt::skip]
 macro_rules! impl_array_for_num256 {
     (

src/common/src/array/primitive_array.rs

@@ -16,13 +16,13 @@ use std::fmt::Debug;
 use std::io::Write;
 use std::mem::size_of;
 
+use risingwave_common_estimate_size::{EstimateSize, ZeroHeapSize};
 use risingwave_pb::common::buffer::CompressionType;
 use risingwave_pb::common::Buffer;
 use risingwave_pb::data::{ArrayType, PbArray};
 
 use super::{Array, ArrayBuilder, ArrayImpl, ArrayResult};
 use crate::buffer::{Bitmap, BitmapBuilder};
-use crate::estimate_size::{EstimateSize, ZeroHeapSize};
 use crate::for_all_native_types;
 use crate::types::*;