diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 20a7def8f3f80..5a5d25aee937a 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -33,7 +33,7 @@ use crate::joins::utils::{ OnceAsync, OnceFut, asymmetric_join_output_partitioning, emits_unmatched_left_rows, is_existence_join, reorder_output_after_swap, swap_join_projection, update_hash, }; -use crate::joins::{JoinOn, JoinOnRef, PartitionMode, SharedBitmapBuilder}; +use crate::joins::{JoinOn, JoinOnRef, PartitionMode, ReservedMap, SharedBitmapBuilder}; use crate::metrics::{Count, MetricBuilder, MetricCategory}; use crate::projection::{ EmbeddedProjection, JoinData, ProjectionExec, try_embed_projection, @@ -191,8 +191,9 @@ fn array_map_key_range( /// Immutable build buffers and their durable reservation. struct JoinBuildData { - /// Hash table with row indices into `batch`, also shared with dynamic filters. - map: Arc, + /// Hash table with row indices into `batch`, charged to its own + /// reservation because it is also shared with dynamic filters. + map: Arc, /// The input rows for the build side. batch: RecordBatch, /// Evaluated build-side key expressions, with float `-0.0` rewritten to @@ -3187,13 +3188,16 @@ fn build_in_memory( ) -> Result { let is_phj_candidate = is_perfect_hash_join_candidate(on_left, schema)?; + // Bound before the map so error paths release the map before its charge. + let mut map_reservation = reservation.new_empty(); + let (join_hash_map, batch, left_values) = if let Some((min_val, max_val)) = array_map_key_range( bounds.as_ref(), schema, batches, on_left, - &mut reservation, + &mut map_reservation, config.execution.perfect_hash_join_small_build_threshold, config.execution.perfect_hash_join_min_key_density, null_equality, @@ -3216,7 +3220,7 @@ fn build_in_memory( // Use `u32` indices for the JoinHashMap when num_rows ≤ u32::MAX, otherwise use the // `u64` indice variant // Arc is used instead of Box to allow sharing with SharedBuildAccumulator for hash map pushdown - let mut hashmap = new_join_hashmap(num_rows, &mut reservation, metrics)?; + let mut hashmap = new_join_hashmap(num_rows, &mut map_reservation, metrics)?; let scratch_reservation = reservation.new_empty(); if prepared { @@ -3313,7 +3317,7 @@ fn build_in_memory( }) .transpose()?; - let map = Arc::new(join_hash_map); + let map = Arc::new(ReservedMap::new(join_hash_map, map_reservation)); // For an ordinary build, nothing reads the strategy unless this join's own // dynamic filter accumulator exists. A prepared build is shared with joins @@ -3357,10 +3361,10 @@ fn build_in_memory( .map(Arc::new), _ => None, }; - // Held for the join's lifetime, so charge it like the maps; it is - // optional, so skip it rather than fail when the pool is full. + // Shared with the map, so charge it with the map; it is optional, + // so skip it rather than fail when the pool is full. let pruning_bitmap = pruning_bitmap.filter(|bitmap| { - let ok = reservation.try_grow(bitmap.size()).is_ok(); + let ok = map.try_grow(bitmap.size()).is_ok(); if ok { metrics.build_mem_used.add(bitmap.size()); } @@ -3540,14 +3544,21 @@ mod tests { Ok(()) } - /// Runs a join whose 200 keys spread over 2M would size a pruning bitmap - /// at the 128 KiB cap, and reports the bytes the build side reserved. - /// - /// Forces the IN-list threshold off: under `force_hash_collisions` every - /// key hashes to the same bucket, so `num_of_distinct_key()` no longer - /// reflects the real 200 distinct keys and would otherwise take the - /// IN-list branch instead of the one under test. - async fn build_mem_used(limit: usize, dynamic_filters: bool) -> Result { + /// 200 keys spread over 2M: sparse enough for a hash map rather than an + /// array map, and sizing a pruning bitmap at the 128 KiB cap. + fn sparse_keys() -> Vec { + (0..200).map(|i| i * 10_000).collect() + } + + /// A build side of [`sparse_keys`], a probe side with three of them, and + /// the join keys. + struct SparseKeyJoinInputs { + build: Arc, + probe: Arc, + on: JoinOn, + } + + fn sparse_key_join_inputs() -> Result { let schema = Arc::new(Schema::new(vec![Field::new("k", DataType::Int64, false)])); let batch = |keys: Vec| { let column = Arc::new(Int64Array::from(keys)) as ArrayRef; @@ -3558,8 +3569,22 @@ mod tests { Arc::new(Column::new_with_schema("k", &schema)?) as _, Arc::new(Column::new_with_schema("k", &schema)?) as _, )]; - let build = batch((0..200).map(|i| i * 10_000).collect())?; - let probe = batch(vec![0, 500_000, 1_500_000])?; + Ok(SparseKeyJoinInputs { + build: batch(sparse_keys())?, + probe: batch(vec![0, 500_000, 1_500_000])?, + on, + }) + } + + /// Runs a join over [`sparse_key_join_inputs`] and reports the bytes the + /// build side reserved. + /// + /// Forces the IN-list threshold off: under `force_hash_collisions` every + /// key hashes to the same bucket, so `num_of_distinct_key()` no longer + /// reflects the real 200 distinct keys and would otherwise take the + /// IN-list branch instead of the one under test. + async fn build_mem_used(limit: usize, dynamic_filters: bool) -> Result { + let SparseKeyJoinInputs { build, probe, on } = sparse_key_join_inputs()?; let join = if dynamic_filters { hash_join_with_dynamic_filter(build, probe, on, JoinType::Inner)?.0 } else { @@ -3605,6 +3630,180 @@ mod tests { Ok(()) } + /// The dynamic filter shares the build map and can outlive the join's + /// stream and plan, so the map stays reserved until the last holder drops + /// it, not only until the partition's stream ends. + #[rstest] + #[tokio::test] + async fn build_map_stays_reserved_with_dynamic_filter( + #[values(PartitionMode::CollectLeft, PartitionMode::Partitioned)] + mode: PartitionMode, + ) -> Result<()> { + let SparseKeyJoinInputs { build, probe, on } = sparse_key_join_inputs()?; + let (join, filter) = hash_join_with_dynamic_filter_and_mode( + build, + probe, + on, + JoinType::Inner, + mode, + )?; + let join: Arc = Arc::new(join); + + let pool: Arc = Arc::new(GreedyMemoryPool::new(1 << 20)); + let runtime = RuntimeEnvBuilder::new() + .with_memory_pool(Arc::clone(&pool)) + .build_arc()?; + let mut config = SessionConfig::new(); + config + .options_mut() + .optimizer + .hash_join_inlist_pushdown_max_size = 0; + let task_ctx = Arc::new( + TaskContext::default() + .with_runtime(runtime) + .with_session_config(config), + ); + let batches = common::collect(join.execute(0, task_ctx)?).await?; + assert_eq!(batches.iter().map(|b| b.num_rows()).sum::(), 3); + + let build_mem_used = join + .metrics() + .unwrap() + .sum_by_name("build_mem_used") + .unwrap() + .as_usize(); + let keys = sparse_keys(); + let map_bytes = + estimate_memory_size::<(u32, u64)>(keys.len(), size_of::())? + + keys.len() * size_of::(); + let keys_array: ArrayRef = Arc::new(Int64Array::from(keys.clone())); + let bitmap_bytes = KeyRangeBitmap::try_new( + &keys_array, + &ScalarValue::Int64(Some(0)), + &ScalarValue::Int64(Some(keys[keys.len() - 1])), + keys.len(), + ) + .map_or(0, |bitmap| bitmap.size()); + let expected = match mode { + // The stream is gone; only the map and its bitmap are still held. + PartitionMode::Partitioned => map_bytes + bitmap_bytes, + PartitionMode::CollectLeft => build_mem_used, + PartitionMode::Auto => unreachable!(), + }; + assert_eq!(pool.reserved(), expected); + + // The filter is the last holder. + drop(join); + assert_eq!(pool.reserved(), map_bytes + bitmap_bytes); + drop(filter); + assert_eq!(pool.reserved(), 0); + Ok(()) + } + + /// Dense keys build an `ArrayMap`, which is charged to the map's own + /// reservation in `array_map_key_range` rather than the build's. + #[rstest] + #[tokio::test] + async fn array_map_stays_reserved_with_dynamic_filter( + #[values(PartitionMode::CollectLeft, PartitionMode::Partitioned)] + mode: PartitionMode, + ) -> Result<()> { + let schema = Arc::new(Schema::new(vec![Field::new("k", DataType::Int64, false)])); + let batch = |keys: Vec| { + let column = Arc::new(Int64Array::from(keys)) as ArrayRef; + let batch = RecordBatch::try_new(Arc::clone(&schema), vec![column])?; + TestMemoryExec::try_new_exec(&[vec![batch]], Arc::clone(&schema), None) + }; + let on = vec![( + Arc::new(Column::new_with_schema("k", &schema)?) as _, + Arc::new(Column::new_with_schema("k", &schema)?) as _, + )]; + let (join, filter) = hash_join_with_dynamic_filter_and_mode( + batch((0..20_000).collect())?, + batch(vec![0, 5, 15])?, + on, + JoinType::Inner, + mode, + )?; + let join: Arc = Arc::new(join); + + let pool: Arc = Arc::new(GreedyMemoryPool::new(1 << 22)); + let runtime = RuntimeEnvBuilder::new() + .with_memory_pool(Arc::clone(&pool)) + .build_arc()?; + let mut config = SessionConfig::new(); + config + .options_mut() + .optimizer + .hash_join_inlist_pushdown_max_size = 0; + let task_ctx = Arc::new( + TaskContext::default() + .with_runtime(runtime) + .with_session_config(config), + ); + let batches = common::collect(join.execute(0, task_ctx)?).await?; + assert_eq!(batches.iter().map(|b| b.num_rows()).sum::(), 3); + assert_phj_used(&join.metrics().unwrap(), true); + + drop(join); + assert_eq!( + pool.reserved(), + ArrayMap::estimate_memory_size(0, 19_999, 20_000) + ); + drop(filter); + assert_eq!(pool.reserved(), 0); + Ok(()) + } + + /// With rows in more than one build partition, the filter is a routed lookup + /// that holds every partition's map, so the maps stay reserved until it is + /// dropped. + #[cfg(not(feature = "force_hash_collisions"))] + #[tokio::test] + async fn routed_maps_stay_reserved_with_dynamic_filter() -> Result<()> { + let SparseKeyJoinInputs { build, probe, on } = sparse_key_join_inputs()?; + let repartition = |input: Arc, key: &PhysicalExprRef| { + let partitioning = Partitioning::Hash(vec![Arc::clone(key)], 2); + RepartitionExec::try_new(input, partitioning) + }; + let (join, filter) = hash_join_with_dynamic_filter_and_mode( + Arc::new(repartition(build, &on[0].0)?), + Arc::new(repartition(probe, &on[0].1)?), + on, + JoinType::Inner, + PartitionMode::Partitioned, + )?; + let join: Arc = Arc::new(join); + + let pool: Arc = Arc::new(GreedyMemoryPool::new(1 << 20)); + let runtime = RuntimeEnvBuilder::new() + .with_memory_pool(Arc::clone(&pool)) + .build_arc()?; + let mut config = SessionConfig::new(); + config + .options_mut() + .optimizer + .hash_join_inlist_pushdown_max_size = 0; + let task_ctx = Arc::new( + TaskContext::default() + .with_runtime(runtime) + .with_session_config(config), + ); + let batches = crate::execution_plan::collect(Arc::clone(&join), task_ctx).await?; + assert_eq!(batches.iter().map(|b| b.num_rows()).sum::(), 3); + assert!(format!("{:?}", filter.current()?).contains("routed by")); + + // Every stream is gone; only the maps and their bitmaps are still held. + let reserved = pool.reserved(); + assert!(reserved > 0); + // The filter is the last holder. + drop(join); + assert_eq!(pool.reserved(), reserved); + drop(filter); + assert_eq!(pool.reserved(), 0); + Ok(()) + } + #[derive(Debug)] struct PartitionedTestExec { cache: Arc, diff --git a/datafusion/physical-plan/src/joins/hash_join/exec/prepared.rs b/datafusion/physical-plan/src/joins/hash_join/exec/prepared.rs index 602d6a5193b06..d20e1431e54df 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec/prepared.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec/prepared.rs @@ -45,7 +45,7 @@ impl fmt::Debug for PreparedHashJoinBuild { impl PreparedHashJoinBuild { /// Return the retained build reservation, excluding all per-probe state. pub fn reserved_bytes(&self) -> usize { - self.build.reservation.size() + self.build.reservation.size() + self.build.map.reserved_bytes() } /// Return the complete build row count, including duplicate and null keys. @@ -290,8 +290,8 @@ impl HashJoinExecBuilder { /// executions. An existing expression handle is preserved for its probe /// consumers, while the build-report accumulator is reset. /// * Retain a prepared-build lease if a dynamic-filter expression outlives - /// this plan. Its array/map references preserve allocation lifetime, but - /// the lease is what preserves the corresponding memory charge. + /// this plan. The map carries its own charge, but the IN-list arrays + /// share the build batch, and the lease is what preserves their charge. /// * Attach after physical optimizations that rewrite the build child. /// Probe-only rewrites must preserve the attached join's `left()`; replacing /// it or making incompatible key, mode, or join-type changes fails. diff --git a/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs b/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs index 66c7c75f616e6..ad52f23a52189 100644 --- a/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs +++ b/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs @@ -18,8 +18,8 @@ use datafusion_expr::ColumnarValue; use datafusion_expr_common::dyn_eq::DynHash; use datafusion_physical_expr_common::physical_expr::{PhysicalExpr, PhysicalExprRef}; -use crate::joins::Map; use crate::joins::key_range_bitmap::KeyRangeBitmap; +use crate::joins::{Map, ReservedMap}; use crate::repartition::RangeExpr; /// RandomState wrapper that preserves the seed used to create it. @@ -267,7 +267,7 @@ pub struct HashTableLookupExpr { on_columns: Vec, /// Random state for hashing (with seeds preserved for serialization) random_state: SeededRandomState, - /// Map(s) to check against + /// Map(s) to check against, with their reservations lookup: Lookup, /// Description for display description: String, @@ -279,14 +279,14 @@ pub struct HashTableLookupExpr { #[derive(Clone)] enum Lookup { /// One map holding every build key, probed directly with no routing. - Single(Arc), + Single(Arc), /// One hash table per build partition of a `Partitioned` join, `None` where /// the partition is empty. A key is only ever built into the partition its /// key routes to, so each probe row is looked up in that partition's table /// alone: no `CASE` over per-partition filters, and no merged table. Routed { router: PartitionRouter, - maps: Arc<[Option>]>, + maps: Arc<[Option>]>, }, } @@ -366,6 +366,30 @@ impl HashTableLookupExpr { maps: Vec>>, description: String, pruning_bitmap: Option>, + ) -> Result { + let maps = maps + .into_iter() + .map(|map| map.map(ReservedMap::unreserved).map(Arc::new)) + .collect(); + Self::new_reserved( + on_columns, + random_state, + router, + maps, + description, + pruning_bitmap, + ) + } + + /// [`Self::new`] for a join's maps, whose memory stays reserved while the + /// lookup holds them. + pub(crate) fn new_reserved( + on_columns: Vec, + random_state: SeededRandomState, + router: Option, + maps: Vec>>, + description: String, + pruning_bitmap: Option>, ) -> Result { let lookup = match (router, maps.as_slice()) { // Every row routes to the only partition, so skip routing. @@ -376,11 +400,10 @@ impl HashTableLookupExpr { ); } (Some(router), _) => { - if maps - .iter() - .flatten() - .any(|map| !matches!(map.as_ref(), Map::HashMap(_))) - { + if maps.iter().flatten().any(|map| { + let map: &Map = map; + !matches!(map, Map::HashMap(_)) + }) { return internal_err!( "routed lookups only support hash table partitions" ); @@ -410,7 +433,7 @@ impl HashTableLookupExpr { &self, keys: &[ArrayRef], router: &PartitionRouter, - maps: &[Option>], + maps: &[Option>], ) -> Result { let num_rows = keys[0].len(); let mut partition_ids = Vec::with_capacity(num_rows); @@ -441,7 +464,8 @@ impl HashTableLookupExpr { for (partition, map) in maps.iter().enumerate() { // An empty partition holds no key, so its rows stay unmatched. let Some(map) = map else { continue }; - let Map::HashMap(map) = map.as_ref() else { + let map: &Map = map; + let Map::HashMap(map) = map else { return internal_err!( "routed lookups only support hash table partitions" ); @@ -514,7 +538,7 @@ impl Lookup { /// created together with its maps, so the maps identify it as well. fn identity(&self) -> *const () { match self { - Lookup::Single(map) => Arc::as_ptr(map).cast(), + Lookup::Single(map) => std::ptr::from_ref::(map).cast(), Lookup::Routed { maps, .. } => Arc::as_ptr(maps).cast(), } } @@ -557,18 +581,23 @@ impl PhysicalExpr for HashTableLookupExpr { let join_keys = evaluate_columns(&self.on_columns, batch)?; match &self.lookup { - Lookup::Single(map) => match map.as_ref() { - Map::HashMap(map) => { - with_hashes(&join_keys, self.random_state.random_state(), |hashes| { - let array = map.contain_hashes(hashes); + Lookup::Single(map) => { + let map: &Map = map; + match map { + Map::HashMap(map) => with_hashes( + &join_keys, + self.random_state.random_state(), + |hashes| { + let array = map.contain_hashes(hashes); + Ok(ColumnarValue::Array(Arc::new(array))) + }, + ), + Map::ArrayMap(map) => { + let array = map.contain_keys(&join_keys)?; Ok(ColumnarValue::Array(Arc::new(array))) - }) - } - Map::ArrayMap(map) => { - let array = map.contain_keys(&join_keys)?; - Ok(ColumnarValue::Array(Arc::new(array))) + } } - }, + } Lookup::Routed { router, maps } => { let array = self.evaluate_routed(&join_keys, router, maps)?; Ok(ColumnarValue::Array(Arc::new(array))) diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs index 321ca4b10db59..395e8c92f8bc4 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -13,6 +13,7 @@ use crate::ExecutionPlanProperties; use crate::Partitioning; use crate::joins::Map; use crate::joins::PartitionMode; +use crate::joins::ReservedMap; use crate::joins::hash_join::exec::HASH_JOIN_SEED; use crate::joins::hash_join::inlist_builder::build_struct_fields; use crate::joins::hash_join::not_in::NotIn; @@ -160,7 +161,7 @@ fn create_membership_predicate( } // Use hash table lookup for large build sides PushdownStrategy::Map(hash_map, pruning_bitmap) => { - Ok(Some(Arc::new(HashTableLookupExpr::new( + Ok(Some(Arc::new(HashTableLookupExpr::new_reserved( on_right.to_vec(), random_state.clone(), None, @@ -312,7 +313,7 @@ pub(crate) enum PushdownStrategy { InList(ArrayRef), /// Use map lookup for large build sides. If `Some` the second field /// represents the build side's keys using a bucket bitmap. - Map(Arc, Option>), + Map(Arc, Option>), /// There was no data in this partition, do not build a dynamic filter for it Empty, } @@ -723,7 +724,8 @@ impl SharedBuildAccumulator { keys_have_null |= partition_keys_have_null; let mergeable = match &pushdown { PushdownStrategy::Map(map, _) => { - matches!(map.as_ref(), Map::HashMap(_)) + let map: &Map = map; + matches!(map, Map::HashMap(_)) } PushdownStrategy::InList(_) => true, PushdownStrategy::Empty => false, @@ -914,7 +916,7 @@ impl SharedBuildAccumulator { } }); } - Arc::new(HashTableLookupExpr::new( + Arc::new(HashTableLookupExpr::new_reserved( self.on_right.clone(), HASH_JOIN_SEED, Some(self.partition_router(num_partitions)?), @@ -953,7 +955,10 @@ impl SharedBuildAccumulator { /// A hash table of an IN list partition's keys, hashed as the probe side /// hashes its join keys: one column per join key. - fn in_list_hash_map(&self, values: &ArrayRef) -> Result> { + /// + /// The table is not reserved, same as the IN list it is built from. Its + /// size follows that list, which `hash_join_inlist_pushdown_max_size` caps. + fn in_list_hash_map(&self, values: &ArrayRef) -> Result> { let columns = match values.as_struct_opt() { Some(columns) if self.on_right.len() > 1 => columns.columns(), _ => std::slice::from_ref(values), @@ -962,7 +967,8 @@ impl SharedBuildAccumulator { create_hashes(columns, HASH_JOIN_SEED.random_state(), &mut hashes)?; let mut map = JoinHashMapU32::with_capacity(hashes.len()); map.update_from_iter(Box::new(hashes.iter().enumerate()), 0); - Ok(Arc::new(Map::HashMap(Box::new(map)))) + let map = Arc::new(Map::HashMap(Box::new(map))); + Ok(Arc::new(ReservedMap::unreserved(map))) } /// Container pruning reads one bitmap, so merge the partitions' key ranges: @@ -1341,10 +1347,8 @@ mod tests { } fn hash_map() -> PushdownStrategy { - PushdownStrategy::Map( - Arc::new(Map::HashMap(Box::new(JoinHashMapU32::with_capacity(1)))), - None, - ) + let map = Arc::new(Map::HashMap(Box::new(JoinHashMapU32::with_capacity(1)))); + PushdownStrategy::Map(Arc::new(ReservedMap::unreserved(map)), None) } #[test] @@ -1375,7 +1379,8 @@ mod tests { create_hashes([&keys], HASH_JOIN_SEED.random_state(), &mut hashes).unwrap(); let mut map = JoinHashMapU32::with_capacity(keys.len()); map.update_from_iter(Box::new(hashes.iter().enumerate()), 0); - PushdownStrategy::Map(Arc::new(Map::HashMap(Box::new(map))), None) + let map = Arc::new(Map::HashMap(Box::new(map))); + PushdownStrategy::Map(Arc::new(ReservedMap::unreserved(map)), None) } /// Splits `keys` across `num_partitions` build partitions as hash diff --git a/datafusion/physical-plan/src/joins/mod.rs b/datafusion/physical-plan/src/joins/mod.rs index a6bc4e63175b7..c4549d847f554 100644 --- a/datafusion/physical-plan/src/joins/mod.rs +++ b/datafusion/physical-plan/src/joins/mod.rs @@ -6,10 +6,14 @@ use core::fmt; use std::fmt::{Display, Formatter}; +use std::ops::Deref; +use std::sync::Arc; use arrow::array::BooleanBufferBuilder; pub use asof_join::{AsOfJoinExec, AsOfMatchExpr}; pub use cross_join::CrossJoinExec; +use datafusion_common::{Result, internal_err}; +use datafusion_execution::memory_pool::MemoryReservation; use datafusion_physical_expr::PhysicalExprRef; /// # Public Only for Internal Use: /// @@ -94,6 +98,56 @@ impl Map { } } +/// A [`Map`] and the reservation for its memory, which the dynamic filter +/// shares for as long as the plan lives. Held together, the memory stays +/// reserved until the last holder drops the map. +pub(crate) struct ReservedMap { + /// Shared, as [`HashTableLookupExpr::new`] takes one and compares by it. + map: Arc, + /// Also charged for data kept with the map, such as the pruning bitmap. + /// `None` for a map built outside a join; see [`HashTableLookupExpr::new`]. + // Keep the reservation after the map it accounts for. + reservation: Option, +} + +impl ReservedMap { + /// Wraps `map` with the reservation it was charged to. + pub(crate) fn new(map: Map, reservation: MemoryReservation) -> Self { + Self { + map: Arc::new(map), + reservation: Some(reservation), + } + } + + /// Wraps a map that was not charged to a pool. + pub(crate) fn unreserved(map: Arc) -> Self { + Self { + map, + reservation: None, + } + } + + /// Charges `bytes` for data kept with the map. + pub(crate) fn try_grow(&self, bytes: usize) -> Result<()> { + match &self.reservation { + Some(reservation) => reservation.try_grow(bytes), + None => internal_err!("ReservedMap has no reservation to grow"), + } + } + + pub(crate) fn reserved_bytes(&self) -> usize { + self.reservation.as_ref().map_or(0, MemoryReservation::size) + } +} + +impl Deref for ReservedMap { + type Target = Map; + + fn deref(&self) -> &Map { + &self.map + } +} + #[cfg(test)] pub mod test_utils;