chore: remove local_stats_alloc

Cargo.toml

@@ -61,7 +61,6 @@ members = [
     "src/utils/delta_btree_map",
     "src/utils/futures_util",
     "src/utils/iter_util",
-    "src/utils/local_stats_alloc",
     "src/utils/openai_embedding_service",
     "src/utils/pgwire",
     "src/utils/resource_util",

docs/rustdoc/index.md

@@ -47,6 +47,5 @@ Common functionalities shared inside RisingWave.
 
 The crates under `src/utils` are several independent util crates which helps to simplify development. We plan to publish them to [crates.io](https://crates.io/) in future when they are more mature.
 
-- [local_stats_alloc](local_stats_alloc/index.html)
 - [memcomparable](memcomparable/index.html)
 - [pgwire](pgwire/index.html)

src/stream/Cargo.toml

@@ -33,7 +33,6 @@ hytra = "0.1.2"
 iceberg = { workspace = true }
 itertools = { workspace = true }
 jsonbb = { workspace = true }
-local_stats_alloc = { path = "../utils/local_stats_alloc" }
 lru = { workspace = true }
 maplit = "1.0.2"
 memcomparable = "0.2"

src/stream/src/executor/join/hash_join.rs

@@ -11,7 +11,7 @@
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
-use std::alloc::Global;
+
 use std::cmp::Ordering;
 use std::ops::{Bound, Deref, DerefMut, RangeBounds};
 use std::sync::Arc;
@@ -21,7 +21,6 @@ use futures::future::{join, try_join};
 use futures::{StreamExt, pin_mut, stream};
 use futures_async_stream::for_await;
 use join_row_set::JoinRowSet;
-use local_stats_alloc::{SharedStatsAlloc, StatsAlloc};
 use risingwave_common::bitmap::Bitmap;
 use risingwave_common::hash::{HashKey, PrecomputedBuildHasher};
 use risingwave_common::metrics::LabelGuardedIntCounter;
@@ -102,8 +101,7 @@ impl<E: JoinEncoding> DerefMut for HashValueWrapper<E> {
     }
 }
 
-type JoinHashMapInner<K, E> =
-    ManagedLruCache<K, HashValueWrapper<E>, PrecomputedBuildHasher, SharedStatsAlloc<Global>>;
+type JoinHashMapInner<K, E> = ManagedLruCache<K, HashValueWrapper<E>, PrecomputedBuildHasher>;
 
 pub struct JoinHashMapMetrics {
     /// Basic information
@@ -425,7 +423,6 @@ impl<K: HashKey, S: StateStore, E: JoinEncoding> JoinHashMap<K, S, E> {
         fragment_id: FragmentId,
         side: &'static str,
     ) -> Self {
-        let alloc = StatsAlloc::new(Global).shared();
         // TODO: unify pk encoding with state table.
         let pk_data_types = state_pk_indices
             .iter()
@@ -461,11 +458,10 @@ impl<K: HashKey, S: StateStore, E: JoinEncoding> JoinHashMap<K, S, E> {
             format!("hash join {}", side),
         );
 
-        let cache = ManagedLruCache::unbounded_with_hasher_in(
+        let cache = ManagedLruCache::unbounded_with_hasher(
             watermark_sequence,
             metrics_info,
             PrecomputedBuildHasher,
-            alloc,
         );
 
         Self {

src/stream/src/executor/temporal_join.rs

@@ -12,15 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::alloc::Global;
 use std::collections::HashMap;
 use std::collections::hash_map::Entry;
 
 use either::Either;
 use futures::TryStreamExt;
 use futures::stream::{self, PollNext};
 use itertools::Itertools;
-use local_stats_alloc::{SharedStatsAlloc, StatsAlloc};
 use lru::DefaultHasher;
 use risingwave_common::array::Op;
 use risingwave_common::bitmap::BitmapBuilder;
@@ -109,7 +107,7 @@ struct TemporalSide<K: HashKey, S: StateStore> {
     source: BatchTable<S>,
     table_stream_key_indices: Vec<usize>,
     table_output_indices: Vec<usize>,
-    cache: ManagedLruCache<K, JoinEntry, DefaultHasher, SharedStatsAlloc<Global>>,
+    cache: ManagedLruCache<K, JoinEntry, DefaultHasher>,
     join_key_data_types: Vec<DataType>,
 }
 
@@ -613,20 +611,17 @@ impl<K: HashKey, S: StateStore, const T: JoinTypePrimitive, const APPEND_ONLY: b
         join_key_data_types: Vec<DataType>,
         memo_table: Option<StateTable<S>>,
     ) -> Self {
-        let alloc = StatsAlloc::new(Global).shared();
-
         let metrics_info = MetricsInfo::new(
             metrics.clone(),
             table.table_id().table_id,
             ctx.id,
             "temporal join",
         );
 
-        let cache = ManagedLruCache::unbounded_with_hasher_in(
+        let cache = ManagedLruCache::unbounded_with_hasher(
             watermark_sequence,
             metrics_info,
             DefaultHasher::default(),
-            alloc,
         );
 
         let metrics = metrics.new_temporal_join_metrics(table.table_id(), ctx.id, ctx.fragment_id);