1use std::hash::Hash;
2use std::marker::PhantomData;
3use std::ops::{Bound, ControlFlow};
4use std::sync::Arc;
5
6use crate::Key;
7
8use crate::compaction::CompactionIndex;
9use crate::config::Config;
10use crate::const_map::{ConstMap, ConstMapShard};
11use crate::durability::{Bitcask, Durability, Fixed};
12use crate::error::{DbError, DbResult};
13use crate::fixed::config::FixedConfig;
14use crate::hook::{NoHook, TypedWriteHook, WriteHook, ZeroHookAdapter};
15use crate::map_index::{Dir, KeyPage, MapIterView, Page, ShardSource, TryMapIter, dir_from};
16use crate::projection::IndexKeyed;
17use crate::upsert::Upsert;
18use crate::watch::{self, WatchOptions, WatchReceiver, WatchSpec, ZeroWatchDecoder};
19use crate::zero_tree::{from_value_bytes, to_bytes};
20
21pub struct ZeroMap<
78 K: Key + Send + Sync + Hash + Eq,
79 const V: usize,
80 T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable = [u8; V],
81 H: TypedWriteHook<K, T> = NoHook,
82 D: Durability = Bitcask,
83> {
84 inner: Arc<ConstMap<K, V, ZeroHookAdapter<K, T, H>, D>>,
85 _marker: PhantomData<T>,
86}
87
88impl<K: Key + Send + Sync + Hash + Eq, const V: usize, T, H, D> watch::sealed::Spec
89 for ZeroMap<K, V, T, H, D>
90where
91 T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
92 H: TypedWriteHook<K, T>,
93 D: Durability,
94{
95}
96impl<K: Key + Send + Sync + Hash + Eq, const V: usize, T, H, D> WatchSpec for ZeroMap<K, V, T, H, D>
97where
98 T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
99 H: TypedWriteHook<K, T>,
100 D: Durability,
101{
102 type Key = K;
103 type Value = T;
104 type Decoder = ZeroWatchDecoder<T>;
105}
106impl<K: Key + Send + Sync + Hash + Eq, const V: usize, T, H, D> ZeroMap<K, V, T, H, D>
107where
108 T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
109 H: TypedWriteHook<K, T> + watch::sealed::Enabled,
110 D: Durability,
111{
112 pub fn watch(&self, options: WatchOptions) -> WatchReceiver<Self> {
134 WatchReceiver::subscribe(
135 self.inner.watch_channel().expect("enabled watch hook"),
136 ZeroWatchDecoder::new(),
137 options,
138 )
139 }
140}
141
142impl<
147 K: Key + Send + Sync + Hash + Eq,
148 const V: usize,
149 T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
150> ZeroMap<K, V, T, NoHook, Bitcask>
151{
152 #[doc = include_str!("../docs/exclusive-access.md")]
156 pub fn open(path: impl AsRef<std::path::Path>, config: Config) -> DbResult<Self> {
157 const { assert!(size_of::<T>() == V) }
158 let adapter = ZeroHookAdapter {
159 inner: NoHook,
160 _marker: PhantomData,
161 };
162 Ok(Self {
163 inner: Arc::new(ConstMap::open_inner(path, config, adapter)?),
164 _marker: PhantomData,
165 })
166 }
167}
168
169impl<
170 K: Key + Send + Sync + Hash + Eq,
171 const V: usize,
172 T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
173 H: TypedWriteHook<K, T>,
174> ZeroMap<K, V, T, H, Bitcask>
175{
176 #[doc = include_str!("../docs/exclusive-access.md")]
179 pub fn open_hooked(
180 path: impl AsRef<std::path::Path>,
181 config: Config,
182 hook: H,
183 ) -> DbResult<Self> {
184 crate::counter_dirty::check_no_counters(&hook)?;
185 crate::unique::check_no_claims(&hook)?;
186 #[cfg(feature = "armour")]
187 if H::EMITS {
188 return Err(DbError::ChangeStreamRequiresManagedOpen);
189 }
190 Self::open_hooked_inner(path, config, hook)
191 }
192
193 pub(crate) fn open_hooked_inner(
197 path: impl AsRef<std::path::Path>,
198 config: Config,
199 hook: H,
200 ) -> DbResult<Self> {
201 const { assert!(size_of::<T>() == V) }
202 if H::WATCHES && !config.iterable {
203 return Err(DbError::WatchRequiresIterable);
204 }
205 let adapter = ZeroHookAdapter {
206 inner: hook,
207 _marker: PhantomData,
208 };
209 Ok(Self {
210 inner: Arc::new(ConstMap::open_inner(path, config, adapter)?),
211 _marker: PhantomData,
212 })
213 }
214
215 pub fn clean_shutdown(&self) -> DbResult<()> {
217 self.inner.clean_shutdown()
218 }
219
220 pub fn close(self) -> DbResult<()> {
221 self.clean_shutdown()
222 }
223
224 pub fn flush_buffers(&self) -> DbResult<()> {
226 self.inner.flush_buffers()
227 }
228
229 pub fn config(&self) -> &Config {
231 self.inner.config()
232 }
233
234 pub fn compact(&self) -> DbResult<usize> {
236 self.inner.compact()
237 }
238
239 pub fn sync_hints(&self) -> DbResult<()> {
241 self.inner.sync_hints()
242 }
243
244 pub fn as_inner(&self) -> &ConstMap<K, V, ZeroHookAdapter<K, T, H>, Bitcask> {
246 &self.inner
247 }
248}
249
250impl<
255 K: Key + Send + Sync + Hash + Eq,
256 const V: usize,
257 T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
258> ZeroMap<K, V, T, NoHook, Fixed>
259{
260 #[doc = include_str!("../docs/exclusive-access.md")]
264 pub fn open(path: impl AsRef<std::path::Path>, config: FixedConfig) -> DbResult<Self> {
265 const { assert!(size_of::<T>() == V) }
266 let adapter = ZeroHookAdapter {
267 inner: NoHook,
268 _marker: PhantomData,
269 };
270 Ok(Self {
271 inner: Arc::new(ConstMap::open_fixed_inner(path, config, adapter)?),
272 _marker: PhantomData,
273 })
274 }
275}
276
277impl<
278 K: Key + Send + Sync + Hash + Eq,
279 const V: usize,
280 T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
281 H: TypedWriteHook<K, T>,
282> ZeroMap<K, V, T, H, Fixed>
283{
284 #[doc = include_str!("../docs/exclusive-access.md")]
287 pub fn open_with_hook(
288 path: impl AsRef<std::path::Path>,
289 config: FixedConfig,
290 hook: H,
291 ) -> DbResult<Self> {
292 crate::counter_dirty::check_no_counters(&hook)?;
293 crate::unique::check_no_claims(&hook)?;
294 #[cfg(feature = "armour")]
295 if H::EMITS {
296 return Err(DbError::ChangeStreamRequiresManagedOpen);
297 }
298 Self::open_with_hook_inner(path, config, hook)
299 }
300
301 pub(crate) fn open_with_hook_inner(
305 path: impl AsRef<std::path::Path>,
306 config: FixedConfig,
307 hook: H,
308 ) -> DbResult<Self> {
309 const { assert!(size_of::<T>() == V) }
310 if H::WATCHES && !config.iterable {
311 return Err(DbError::WatchRequiresIterable);
312 }
313 let adapter = ZeroHookAdapter {
314 inner: hook,
315 _marker: PhantomData,
316 };
317 Ok(Self {
318 inner: Arc::new(ConstMap::open_fixed_inner(path, config, adapter)?),
319 _marker: PhantomData,
320 })
321 }
322
323 pub fn clean_shutdown(&self) -> DbResult<()> {
325 self.inner.clean_shutdown()
326 }
327
328 pub fn close(self) -> DbResult<()> {
329 self.clean_shutdown()
330 }
331}
332
333impl<
338 K: Key + Send + Sync + Hash + Eq,
339 const V: usize,
340 T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
341 H: TypedWriteHook<K, T>,
342 D: Durability,
343> ZeroMap<K, V, T, H, D>
344{
345 pub(crate) fn tx_inner(&self) -> &ConstMap<K, V, ZeroHookAdapter<K, T, H>, D> {
346 self.inner.as_ref()
347 }
348
349 pub(crate) fn bind_tx_self(arc: &Arc<Self>) {
351 ConstMap::bind_tx_self(&arc.inner);
352 }
353
354 pub(crate) fn tx_install_projection_metadata(&self, db: &crate::armour::Db) -> DbResult<()> {
355 self.tx_inner().tx_install_projection_metadata(db)
356 }
357
358 pub(crate) fn tx_preplan_projection_targets(
359 &self,
360 ) -> DbResult<Vec<Arc<dyn crate::txn::participant::TxParticipant>>> {
361 self.tx_inner().tx_preplan_projection_targets()
362 }
363
364 pub(crate) fn tx_project_writes(
365 &self,
366 key: &[u8],
367 old: Option<&[u8]>,
368 new: Option<&[u8]>,
369 ) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
370 self.tx_inner().tx_project_writes(key, old, new)
371 }
372
373 pub(crate) fn tx_counter_deltas_for(
374 &self,
375 key: &[u8],
376 old: Option<&[u8]>,
377 new: Option<&[u8]>,
378 ) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
379 self.tx_inner().tx_counter_deltas_for(key, old, new)
380 }
381
382 pub(crate) fn tx_claims_for(
386 &self,
387 name: &'static str,
388 key: &[u8],
389 old: Option<&[u8]>,
390 new: Option<&[u8]>,
391 ) -> DbResult<Vec<crate::txn::participant::ClaimOp>> {
392 self.tx_inner().tx_claims_for(name, key, old, new)
393 }
394
395 #[doc(hidden)]
404 pub fn build_ttl_index(&self) -> DbResult<()> {
405 self.inner.build_ttl_index()
406 }
407
408 #[doc(hidden)]
409 pub fn ttl_debug_len(&self) -> usize {
410 self.inner.ttl_debug_len()
411 }
412
413 #[doc(hidden)]
414 pub fn ttl_debug_deadline(&self, key: &K) -> Option<u64> {
415 self.inner.ttl_debug_deadline(key)
416 }
417
418 pub fn sweep_expired(&self, now_ms: u64, budget: usize) -> DbResult<crate::ttl::SweepOutcome> {
419 self.inner.sweep_expired(now_ms, budget)
420 }
421
422 pub fn set_ttl_sweep(&self, enabled: bool) -> DbResult<()> {
423 self.inner.set_ttl_sweep(enabled)
424 }
425
426 pub fn rebuild_ttl_index(&self) -> DbResult<()> {
427 self.inner.rebuild_ttl_index()
428 }
429
430 pub fn get(&self, key: &K) -> Option<T> {
435 let bytes = self.inner.get(key)?;
436 Some(from_value_bytes::<V, T>(&bytes))
437 }
438
439 pub fn get_or_err(&self, key: &K) -> DbResult<T> {
441 self.get(key).ok_or(DbError::KeyNotFound)
442 }
443
444 pub fn contains(&self, key: &K) -> bool {
446 self.inner.contains(key)
447 }
448
449 pub fn put(&self, key: &K, value: &T) -> DbResult<Option<T>> {
453 let bytes = to_bytes::<V, T>(value);
454 self.inner
455 .put(key, &bytes)
456 .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
457 }
458
459 pub fn insert(&self, key: &K, value: &T) -> DbResult<()> {
462 let bytes = to_bytes::<V, T>(value);
463 self.inner.insert(key, &bytes)
464 }
465
466 pub fn delete(&self, key: &K) -> DbResult<Option<T>> {
468 self.inner
469 .delete(key)
470 .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
471 }
472
473 pub fn cas(&self, key: &K, expected: &T, new_value: &T) -> DbResult<()> {
477 let exp_bytes = to_bytes::<V, T>(expected);
478 let new_bytes = to_bytes::<V, T>(new_value);
479 self.inner.cas(key, &exp_bytes, &new_bytes)
480 }
481
482 pub fn compare_delete(&self, key: &K, expected: &T) -> DbResult<()> {
486 let exp_bytes = to_bytes::<V, T>(expected);
487 self.inner.compare_delete(key, &exp_bytes)
488 }
489
490 pub fn update(&self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
494 self.inner
495 .update(key, |bytes| {
496 let val = from_value_bytes::<V, T>(bytes);
497 let new_val = f(&val);
498 to_bytes::<V, T>(&new_val)
499 })
500 .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
501 }
502
503 pub fn fetch_update(&self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
505 self.inner
506 .fetch_update(key, |bytes| {
507 let val = from_value_bytes::<V, T>(bytes);
508 let new_val = f(&val);
509 to_bytes::<V, T>(&new_val)
510 })
511 .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
512 }
513
514 pub fn upsert<U: crate::IntoUpsert<T>>(
519 &self,
520 key: &K,
521 f: impl FnOnce(Option<&T>) -> U,
522 ) -> crate::DbResult<U::Out> {
523 self.atomic(key, |s| crate::upsert::upsert_via_held(s, key, f))
524 }
525
526 pub fn atomic<R>(
532 &self,
533 shard_key: &K,
534 f: impl FnOnce(&mut ZeroMapShard<'_, K, V, T, D, ZeroHookAdapter<K, T, H>>) -> DbResult<R>,
535 ) -> DbResult<R> {
536 self.inner.atomic(shard_key, |const_shard| {
537 let shard = unsafe {
540 &mut *(const_shard as *mut ConstMapShard<'_, K, V, ZeroHookAdapter<K, T, H>, D>
541 as *mut ZeroMapShard<'_, K, V, T, D, ZeroHookAdapter<K, T, H>>)
542 };
543 f(shard)
544 })
545 }
546
547 pub fn len(&self) -> usize {
550 self.inner.len()
551 }
552
553 pub fn is_empty(&self) -> bool {
554 self.inner.is_empty()
555 }
556
557 pub fn shard_for(&self, key: &K) -> usize {
558 self.inner.shard_for(key)
559 }
560
561 pub fn flush(&self) -> DbResult<()> {
563 self.inner.flush()
564 }
565
566 pub fn migrate(&self, f: impl Fn(&K, &T) -> crate::MigrateAction<T>) -> DbResult<usize> {
578 self.migrate_inner(f, true)
579 }
580
581 pub(crate) fn migrate_inner(
582 &self,
583 f: impl Fn(&K, &T) -> crate::MigrateAction<T>,
584 fire_init: bool,
585 ) -> DbResult<usize> {
586 self.inner.migrate_inner(
587 |key, bytes| {
588 let val: T = from_value_bytes(bytes);
589 match f(key, &val) {
590 crate::MigrateAction::Keep => crate::MigrateAction::Keep,
591 crate::MigrateAction::Update(new) => {
592 crate::MigrateAction::Update(to_bytes::<V, T>(&new))
593 }
594 crate::MigrateAction::Delete => crate::MigrateAction::Delete,
595 }
596 },
597 fire_init,
598 )
599 }
600
601 pub fn for_each(&self, mut f: impl FnMut(&K, &T)) {
603 self.inner.for_each(|key, bytes| {
604 let val: T = from_value_bytes(bytes);
605 f(key, &val);
606 });
607 }
608
609 pub(crate) fn for_each_while(
611 &self,
612 mut f: impl FnMut(&K, &T) -> ControlFlow<()>,
613 ) -> ControlFlow<()> {
614 self.inner.for_each_while(|key, bytes| {
615 let val: T = from_value_bytes::<V, T>(bytes);
616 f(key, &val)
617 })
618 }
619
620 pub(crate) fn replay_init(&self) {
622 self.inner.replay_init();
623 }
624
625 pub(crate) fn reconcile_projections(&self) -> DbResult<()> {
626 self.inner.reconcile_projections()
627 }
628
629 pub(crate) fn effective_projection_limits(&self) -> (usize, usize) {
630 self.inner.effective_projection_limits()
631 }
632
633 pub fn iter_view(&self) -> Option<ZeroMapIterView<'_, K, V, T, H, D>> {
646 self.inner.iter_view()?;
647 Some(ZeroMapIterView {
648 inner: MapIterView::new(self, dir_from(self.inner.reversed())),
649 })
650 }
651
652 pub fn scope<const N: usize>(
654 &self,
655 key: &K,
656 ) -> Option<crate::scope::Scope<'_, crate::scope::MapScan<&Self>, N>>
657 where
658 T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
659 {
660 crate::scope::map_scope::<_, _, N>(
661 self,
662 key,
663 dir_from(self.reversed()),
664 self.iter_view().is_some(),
665 )
666 }
667
668 #[cfg(feature = "armour")]
671 pub fn tx_scope<const N: usize>(
672 self: &Arc<Self>,
673 key: &K,
674 ) -> Option<crate::TxScope<Self, K, N>> {
675 self.iter_view().is_some().then(|| {
676 crate::txn::scope::TxScope::from_parts(
677 self.clone(),
678 key,
679 dir_from(self.reversed()),
680 Self::tx_scope_keys,
681 )
682 })
683 }
684
685 fn tx_scope_keys<'a>(
686 collection: &'a Self,
687 start: Bound<&'a K>,
688 end: Bound<&'a K>,
689 dir: Dir,
690 ) -> Box<dyn Iterator<Item = K> + 'a> {
691 Box::new(MapIterView::new(collection, dir).keys_range_bounds(start, end))
692 }
693
694 pub fn first(&self) -> DbResult<Option<(K, T)>> {
699 let view = self.iter_view().ok_or(DbError::NotImplemented)?;
700 Ok(view.iter().next())
701 }
702
703 pub fn last(&self) -> DbResult<Option<(K, T)>> {
708 self.iter_view().ok_or(DbError::NotImplemented)?;
709 Ok(
710 MapIterView::new(self, dir_from(self.inner.reversed()).flip())
711 .iter()
712 .next(),
713 )
714 }
715
716 pub(crate) fn reversed(&self) -> bool {
718 self.inner.reversed()
719 }
720
721 pub(crate) fn inspection_shard_count(&self) -> usize {
723 ShardSource::shard_count(self)
724 }
725
726 pub(crate) fn inspection_range_bounds(
728 &self,
729 start: Bound<&K>,
730 end: Bound<&K>,
731 batch: usize,
732 ) -> TryMapIter<'_, ZeroMap<K, V, T, H, D>>
733 where
734 D: Durability,
735 {
736 MapIterView::with_batch(self, dir_from(self.inner.reversed()), batch)
737 .try_range_bounds(start, end)
738 }
739}
740
741impl<
742 K: Key + Send + Sync + Hash + Eq,
743 const V: usize,
744 T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
745 H: TypedWriteHook<K, T>,
746 D: Durability,
747> ShardSource for ZeroMap<K, V, T, H, D>
748{
749 type Key = K;
750 type Val = T;
751
752 fn shard_count(&self) -> usize {
753 ShardSource::shard_count(self.inner.as_ref())
754 }
755
756 fn next_batch(
757 &self,
758 shard: usize,
759 after: Option<&K>,
760 dir: Dir,
761 n: usize,
762 range_start: Bound<crate::map_index::OrdKey<K>>,
763 range_end: Bound<crate::map_index::OrdKey<K>>,
764 ) -> Vec<(K, T)> {
765 ShardSource::next_batch(
766 self.inner.as_ref(),
767 shard,
768 after,
769 dir,
770 n,
771 range_start,
772 range_end,
773 )
774 .into_iter()
775 .map(|(k, bytes)| (k, from_value_bytes::<V, T>(&bytes)))
776 .collect()
777 }
778
779 fn next_keys_batch(
780 &self,
781 shard: usize,
782 after: Option<&K>,
783 dir: Dir,
784 n: usize,
785 range_start: Bound<crate::map_index::OrdKey<K>>,
786 range_end: Bound<crate::map_index::OrdKey<K>>,
787 ) -> Vec<K> {
788 ShardSource::next_keys_batch(
789 self.inner.as_ref(),
790 shard,
791 after,
792 dir,
793 n,
794 range_start,
795 range_end,
796 )
797 }
798}
799
800pub struct ZeroMapIterView<'a, K, const V: usize, T, H = NoHook, D = Bitcask>
803where
804 K: Key + Send + Sync + Hash + Eq,
805 T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
806 H: TypedWriteHook<K, T>,
807 D: Durability,
808{
809 inner: MapIterView<'a, ZeroMap<K, V, T, H, D>>,
810}
811
812impl<K, const V: usize, T, H, D> ZeroMapIterView<'_, K, V, T, H, D>
813where
814 K: Key + Send + Sync + Hash + Eq,
815 T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
816 H: TypedWriteHook<K, T>,
817 D: Durability,
818{
819 pub fn iter(&self) -> impl Iterator<Item = (K, T)> + '_ {
821 self.inner.iter()
822 }
823
824 pub fn range(&self, start: &K, end: &K) -> impl Iterator<Item = (K, T)> + '_ {
826 self.inner.range(start, end)
827 }
828
829 pub fn range_bounds(
831 &self,
832 start: Bound<&K>,
833 end: Bound<&K>,
834 ) -> impl Iterator<Item = (K, T)> + '_ {
835 self.inner.range_bounds(start, end)
836 }
837
838 pub fn prefix_iter(&self, prefix: &[u8]) -> impl Iterator<Item = (K, T)> + '_ {
840 self.inner.prefix_iter(prefix)
841 }
842
843 pub fn paginate(&self, after: Option<&K>, limit: usize) -> Page<K, T> {
845 self.inner.paginate(after, limit)
846 }
847
848 pub fn keys(&self) -> impl Iterator<Item = K> + '_ {
850 self.inner.keys()
851 }
852
853 pub fn keys_range(&self, start: &K, end: &K) -> impl Iterator<Item = K> + '_ {
855 self.inner.keys_range(start, end)
856 }
857
858 pub fn keys_range_bounds(
860 &self,
861 start: Bound<&K>,
862 end: Bound<&K>,
863 ) -> impl Iterator<Item = K> + '_ {
864 self.inner.keys_range_bounds(start, end)
865 }
866
867 pub fn keys_prefix(&self, prefix: &[u8]) -> impl Iterator<Item = K> + '_ {
869 self.inner.keys_prefix(prefix)
870 }
871
872 pub fn keys_paginate(&self, after: Option<&K>, limit: usize) -> KeyPage<K> {
874 self.inner.keys_paginate(after, limit)
875 }
876
877 pub fn retain(&self, mut f: impl FnMut(&K, &T) -> bool) -> DbResult<()> {
884 for (key, value) in self.inner.iter() {
885 if f(&key, &value) {
886 continue;
887 }
888 match self.inner.src().compare_delete(&key, &value) {
889 Ok(()) => {}
890 Err(DbError::CasMismatch) | Err(DbError::KeyNotFound) => {
891 tracing::debug!("retain: compare_delete skipped (concurrent change)");
892 }
893 Err(e) => return Err(e),
894 }
895 }
896 Ok(())
897 }
898}
899
900impl<
901 K: Key + Send + Sync + Hash + Eq,
902 const V: usize,
903 T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable + Send + Sync,
904 H: TypedWriteHook<K, T>,
905> CompactionIndex<K> for ZeroMap<K, V, T, H, Bitcask>
906{
907 fn update_if_match(
908 &self,
909 key: &K,
910 old_loc: crate::disk_loc::DiskLoc,
911 new_loc: crate::disk_loc::DiskLoc,
912 ) -> bool {
913 self.inner.update_if_match(key, old_loc, new_loc)
914 }
915
916 fn contains_key(&self, key: &K) -> bool {
917 self.contains(key)
918 }
919
920 fn is_live(&self, shard_id: u8, key: &K, loc: crate::disk_loc::DiskLoc) -> bool {
921 self.inner.is_live(shard_id, key, loc)
922 }
923}
924
925#[repr(transparent)]
932pub struct ZeroMapShard<
933 'a,
934 K: Key + Send + Sync + Hash + Eq,
935 const V: usize,
936 T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable = [u8; V],
937 D: Durability = Bitcask,
938 WH: WriteHook<K> = NoHook,
939> {
940 inner: ConstMapShard<'a, K, V, WH, D>,
941 _marker: PhantomData<T>,
942}
943
944impl<
945 K: Key + Send + Sync + Hash + Eq,
946 const V: usize,
947 T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
948 D: Durability,
949 WH: WriteHook<K>,
950> ZeroMapShard<'_, K, V, T, D, WH>
951{
952 pub fn put(&mut self, key: &K, value: &T) -> DbResult<Option<T>> {
953 let bytes = to_bytes::<V, T>(value);
954 self.inner
955 .put(key, &bytes)
956 .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
957 }
958
959 pub fn insert(&mut self, key: &K, value: &T) -> DbResult<()> {
960 let bytes = to_bytes::<V, T>(value);
961 self.inner.insert(key, &bytes)
962 }
963
964 pub fn delete(&mut self, key: &K) -> DbResult<Option<T>> {
965 self.inner
966 .delete(key)
967 .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
968 }
969
970 pub fn try_get(&self, key: &K) -> DbResult<Option<T>> {
971 Ok(self
972 .inner
973 .try_get(key)?
974 .map(|bytes| from_value_bytes::<V, T>(&bytes)))
975 }
976
977 pub fn get_or_err(&self, key: &K) -> DbResult<T> {
978 self.try_get(key)?.ok_or(DbError::KeyNotFound)
979 }
980
981 pub fn try_contains(&self, key: &K) -> DbResult<bool> {
982 self.inner.try_contains(key)
983 }
984
985 pub fn update(&mut self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
988 self.inner
989 .update(key, |bytes| {
990 let val = from_value_bytes::<V, T>(bytes);
991 to_bytes::<V, T>(&f(&val))
992 })
993 .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
994 }
995
996 pub fn fetch_update(&mut self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
998 self.inner
999 .fetch_update(key, |bytes| {
1000 let val = from_value_bytes::<V, T>(bytes);
1001 to_bytes::<V, T>(&f(&val))
1002 })
1003 .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
1004 }
1005
1006 pub fn upsert<U: crate::IntoUpsert<T>>(
1008 &mut self,
1009 key: &K,
1010 f: impl FnOnce(Option<&T>) -> U,
1011 ) -> crate::DbResult<U::Out> {
1012 crate::upsert::upsert_via_held(self, key, f)
1013 }
1014}
1015
1016#[cfg(feature = "armour")]
1017impl<T, const V: usize, H> crate::armour::collection::Collection
1018 for ZeroMap<T::SelfId, V, T, H, Bitcask>
1019where
1020 T: crate::CollectionMeta
1021 + Copy
1022 + zerocopy::IntoBytes
1023 + zerocopy::FromBytes
1024 + zerocopy::Immutable
1025 + Send
1026 + Sync,
1027 H: crate::hook::TypedWriteHook<T::SelfId, T>,
1028 T::SelfId: crate::Key + Send + Sync + std::hash::Hash + Eq,
1029{
1030 fn name(&self) -> &str {
1031 T::NAME
1032 }
1033 fn len(&self) -> usize {
1034 self.len()
1035 }
1036 fn compact(&self) -> crate::DbResult<usize> {
1037 self.compact()
1038 }
1039 fn flush(&self) -> crate::DbResult<()> {
1040 crate::projection::collection_flush_fault()?;
1041 self.flush_buffers()?;
1042 if self.config().hints.unwrap_or(false) {
1043 self.sync_hints()?;
1044 }
1045 Ok(())
1046 }
1047 fn periodic_flush(&self) -> crate::DbResult<()> {
1048 self.flush_buffers()
1049 }
1050 fn durable_flush(&self) -> crate::DbResult<()> {
1051 self.inner.durable_flush()
1052 }
1053 fn clean_shutdown(&self) -> crate::DbResult<()> {
1054 ZeroMap::<T::SelfId, V, T, H, Bitcask>::clean_shutdown(self)
1055 }
1056 fn sweep_expired(
1057 &self,
1058 now_ms: u64,
1059 budget: usize,
1060 ) -> crate::DbResult<crate::ttl::SweepOutcome> {
1061 self.sweep_expired(now_ms, budget)
1062 }
1063 fn set_ttl_sweep(&self, enabled: bool) -> crate::DbResult<()> {
1064 self.set_ttl_sweep(enabled)
1065 }
1066 fn rebuild_ttl_index(&self) -> crate::DbResult<()> {
1067 self.rebuild_ttl_index()
1068 }
1069 fn reset_counter_dirty_marker(&self) -> bool {
1070 self.tx_inner()
1071 .counter_dirty
1072 .get()
1073 .is_some_and(|d| d.take_marked())
1074 }
1075 fn reconcile_counters(&self) -> crate::DbResult<u64> {
1076 self.tx_inner().reconcile_counters()
1077 }
1078 fn counter_target_names(&self) -> Vec<&'static str> {
1079 self.tx_inner().counter_target_names()
1080 }
1081 fn close_watches(&self) {
1082 if let Some(channel) = self.inner.watch_channel() {
1083 channel.close_terminal();
1084 }
1085 }
1086}
1087
1088#[cfg(feature = "armour")]
1089impl<T, const V: usize, H> crate::armour::collection::Collection
1090 for ZeroMap<T::SelfId, V, T, H, Fixed>
1091where
1092 T: crate::CollectionMeta
1093 + Copy
1094 + zerocopy::IntoBytes
1095 + zerocopy::FromBytes
1096 + zerocopy::Immutable
1097 + Send
1098 + Sync,
1099 H: crate::hook::TypedWriteHook<T::SelfId, T>,
1100 T::SelfId: crate::Key + Send + Sync + std::hash::Hash + Eq,
1101{
1102 fn name(&self) -> &str {
1103 T::NAME
1104 }
1105 fn len(&self) -> usize {
1106 self.len()
1107 }
1108 fn compact(&self) -> crate::DbResult<usize> {
1109 Ok(0)
1110 }
1111 fn flush(&self) -> crate::DbResult<()> {
1112 crate::projection::collection_flush_fault()?;
1113 self.flush()
1116 }
1117 fn periodic_flush(&self) -> crate::DbResult<()> {
1118 self.flush()
1122 }
1123 fn durable_flush(&self) -> crate::DbResult<()> {
1124 self.flush()
1125 }
1126 fn clean_shutdown(&self) -> crate::DbResult<()> {
1127 ZeroMap::<T::SelfId, V, T, H, Fixed>::clean_shutdown(self)
1128 }
1129 fn sweep_expired(
1130 &self,
1131 now_ms: u64,
1132 budget: usize,
1133 ) -> crate::DbResult<crate::ttl::SweepOutcome> {
1134 self.sweep_expired(now_ms, budget)
1135 }
1136 fn set_ttl_sweep(&self, enabled: bool) -> crate::DbResult<()> {
1137 self.set_ttl_sweep(enabled)
1138 }
1139 fn rebuild_ttl_index(&self) -> crate::DbResult<()> {
1140 self.rebuild_ttl_index()
1141 }
1142 fn reset_counter_dirty_marker(&self) -> bool {
1143 self.tx_inner()
1144 .counter_dirty
1145 .get()
1146 .is_some_and(|d| d.take_marked())
1147 }
1148 fn reconcile_counters(&self) -> crate::DbResult<u64> {
1149 self.tx_inner().reconcile_counters()
1150 }
1151 fn counter_target_names(&self) -> Vec<&'static str> {
1152 self.tx_inner().counter_target_names()
1153 }
1154 fn close_watches(&self) {
1155 if let Some(channel) = self.inner.watch_channel() {
1156 channel.close_terminal();
1157 }
1158 }
1159}
1160
1161#[cfg(feature = "typed-tree")]
1162impl<
1163 K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1164 const V: usize,
1165 T: crate::CollectionMeta
1166 + Copy
1167 + zerocopy::IntoBytes
1168 + zerocopy::FromBytes
1169 + zerocopy::Immutable
1170 + Default
1171 + PartialEq
1172 + Send
1173 + Sync
1174 + 'static,
1175 H: TypedWriteHook<K, T> + 'static,
1176> crate::projection::sealed::Sealed for ZeroMap<K, V, T, H, Bitcask>
1177{
1178}
1179
1180impl<
1181 K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1182 const V: usize,
1183 T: crate::CollectionMeta
1184 + Copy
1185 + zerocopy::IntoBytes
1186 + zerocopy::FromBytes
1187 + zerocopy::Immutable
1188 + Default
1189 + PartialEq
1190 + Send
1191 + Sync
1192 + 'static,
1193 H: TypedWriteHook<K, T> + 'static,
1194> crate::projection::ProjectableIndex for ZeroMap<K, V, T, H, Bitcask>
1195{
1196 fn collection_name(&self) -> &'static str {
1197 T::NAME
1198 }
1199
1200 fn put_bytes(&self, key: &[u8], value: &[u8]) -> DbResult<()> {
1201 if key.len() != std::mem::size_of::<K>() {
1202 return Err(DbError::CorruptedEntry { offset: 0 });
1203 }
1204 let k = K::from_bytes(key);
1205 let v: T = zerocopy::FromBytes::read_from_bytes(value)
1206 .map_err(|_| DbError::FormatMismatch("zero index value decode".into()))?;
1207 self.put(&k, &v)?;
1208 Ok(())
1209 }
1210
1211 fn delete_bytes(&self, key: &[u8]) -> DbResult<()> {
1212 if key.len() != std::mem::size_of::<K>() {
1213 return Err(DbError::CorruptedEntry { offset: 0 });
1214 }
1215 let k = K::from_bytes(key);
1216 self.delete(&k)?;
1217 Ok(())
1218 }
1219
1220 fn values_equal(&self, a: &[u8], b: &[u8]) -> DbResult<bool> {
1221 Ok(a == b)
1222 }
1223
1224 fn for_each_bytes(&self, f: &mut dyn FnMut(&[u8], &[u8]) -> DbResult<()>) -> DbResult<()> {
1225 let mut err: Option<DbError> = None;
1226 self.for_each(|k, v| {
1227 if err.is_some() {
1228 return;
1229 }
1230 if let Err(e) = f(k.as_bytes(), zerocopy::IntoBytes::as_bytes(v)) {
1231 err = Some(e);
1232 }
1233 });
1234 match err {
1235 Some(e) => Err(e),
1236 None => Ok(()),
1237 }
1238 }
1239
1240 fn durable_flush(&self) -> DbResult<()> {
1241 ZeroMap::flush(self)
1242 }
1243}
1244
1245#[cfg(feature = "typed-tree")]
1246impl<
1247 K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1248 const V: usize,
1249 T: crate::CollectionMeta
1250 + Copy
1251 + zerocopy::IntoBytes
1252 + zerocopy::FromBytes
1253 + zerocopy::Immutable
1254 + Default
1255 + PartialEq
1256 + Send
1257 + Sync
1258 + 'static,
1259 H: TypedWriteHook<K, T> + 'static,
1260> crate::projection::ProjectableIndexTx for ZeroMap<K, V, T, H, Bitcask>
1261{
1262 fn as_tx_participant(self: Arc<Self>) -> Arc<dyn crate::txn::participant::TxParticipant> {
1263 self
1264 }
1265}
1266
1267#[cfg(feature = "typed-tree")]
1268impl<
1269 K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1270 const V: usize,
1271 T: crate::CollectionMeta
1272 + Copy
1273 + zerocopy::IntoBytes
1274 + zerocopy::FromBytes
1275 + zerocopy::Immutable
1276 + Default
1277 + PartialEq
1278 + Send
1279 + Sync
1280 + 'static,
1281 H: TypedWriteHook<K, T> + 'static,
1282> crate::projection::sealed::Sealed for ZeroMap<K, V, T, H, Fixed>
1283{
1284}
1285
1286impl<
1287 K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1288 const V: usize,
1289 T: crate::CollectionMeta
1290 + Copy
1291 + zerocopy::IntoBytes
1292 + zerocopy::FromBytes
1293 + zerocopy::Immutable
1294 + Default
1295 + PartialEq
1296 + Send
1297 + Sync
1298 + 'static,
1299 H: TypedWriteHook<K, T> + 'static,
1300> crate::projection::ProjectableIndex for ZeroMap<K, V, T, H, Fixed>
1301{
1302 fn collection_name(&self) -> &'static str {
1303 T::NAME
1304 }
1305
1306 fn put_bytes(&self, key: &[u8], value: &[u8]) -> DbResult<()> {
1307 if key.len() != std::mem::size_of::<K>() {
1308 return Err(DbError::CorruptedEntry { offset: 0 });
1309 }
1310 let k = K::from_bytes(key);
1311 let v: T = zerocopy::FromBytes::read_from_bytes(value)
1312 .map_err(|_| DbError::FormatMismatch("zero index value decode".into()))?;
1313 self.put(&k, &v)?;
1314 Ok(())
1315 }
1316
1317 fn delete_bytes(&self, key: &[u8]) -> DbResult<()> {
1318 if key.len() != std::mem::size_of::<K>() {
1319 return Err(DbError::CorruptedEntry { offset: 0 });
1320 }
1321 let k = K::from_bytes(key);
1322 self.delete(&k)?;
1323 Ok(())
1324 }
1325
1326 fn values_equal(&self, a: &[u8], b: &[u8]) -> DbResult<bool> {
1327 Ok(a == b)
1328 }
1329
1330 fn for_each_bytes(&self, f: &mut dyn FnMut(&[u8], &[u8]) -> DbResult<()>) -> DbResult<()> {
1331 let mut err: Option<DbError> = None;
1332 self.for_each(|k, v| {
1333 if err.is_some() {
1334 return;
1335 }
1336 if let Err(e) = f(k.as_bytes(), zerocopy::IntoBytes::as_bytes(v)) {
1337 err = Some(e);
1338 }
1339 });
1340 match err {
1341 Some(e) => Err(e),
1342 None => Ok(()),
1343 }
1344 }
1345
1346 fn durable_flush(&self) -> DbResult<()> {
1347 ZeroMap::flush(self)
1348 }
1349}
1350
1351#[cfg(feature = "typed-tree")]
1352impl<
1353 K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1354 const V: usize,
1355 T: crate::CollectionMeta
1356 + Copy
1357 + zerocopy::IntoBytes
1358 + zerocopy::FromBytes
1359 + zerocopy::Immutable
1360 + Default
1361 + PartialEq
1362 + Send
1363 + Sync
1364 + 'static,
1365 H: TypedWriteHook<K, T> + 'static,
1366> crate::projection::ProjectableIndexTx for ZeroMap<K, V, T, H, Fixed>
1367{
1368 fn as_tx_participant(self: Arc<Self>) -> Arc<dyn crate::txn::participant::TxParticipant> {
1369 self
1370 }
1371}
1372
1373#[cfg(feature = "typed-tree")]
1374impl<
1375 K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1376 const V: usize,
1377 T: crate::CollectionMeta
1378 + Copy
1379 + zerocopy::IntoBytes
1380 + zerocopy::FromBytes
1381 + zerocopy::Immutable
1382 + Default
1383 + PartialEq
1384 + Send
1385 + Sync
1386 + 'static,
1387 H: TypedWriteHook<K, T> + 'static,
1388> crate::projection::IndexKeyed for ZeroMap<K, V, T, H, Bitcask>
1389{
1390 type K = K;
1391 type V = T;
1392
1393 fn encode_value(&self, v: &T) -> DbResult<Vec<u8>> {
1394 Ok(zerocopy::IntoBytes::as_bytes(v).to_vec())
1395 }
1396}
1397
1398#[cfg(feature = "typed-tree")]
1399impl<
1400 K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1401 const V: usize,
1402 T: crate::CollectionMeta
1403 + Copy
1404 + zerocopy::IntoBytes
1405 + zerocopy::FromBytes
1406 + zerocopy::Immutable
1407 + Default
1408 + PartialEq
1409 + Send
1410 + Sync
1411 + 'static,
1412 H: TypedWriteHook<K, T> + 'static,
1413> crate::counter::CounterCollection for ZeroMap<K, V, T, H, Bitcask>
1414{
1415 fn decode_value(&self, bytes: &[u8]) -> DbResult<T> {
1416 zerocopy::FromBytes::read_from_bytes(bytes)
1417 .map_err(|_| DbError::FormatMismatch("zero counter value decode".into()))
1418 }
1419
1420 fn read_bytes(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
1422 if key.len() != std::mem::size_of::<K>() {
1423 return Err(DbError::CorruptedEntry { offset: 0 });
1424 }
1425 let k = K::from_bytes(key);
1426 self.get(&k).map(|v| self.encode_value(&v)).transpose()
1427 }
1428
1429 fn contains_bytes(&self, key: &[u8]) -> DbResult<bool> {
1430 if key.len() != std::mem::size_of::<K>() {
1431 return Err(DbError::CorruptedEntry { offset: 0 });
1432 }
1433 let k = K::from_bytes(key);
1434 Ok(self.contains(&k))
1435 }
1436
1437 fn upsert_counter(
1438 &self,
1439 key: &[u8],
1440 f: &mut dyn FnMut(Option<&T>) -> Upsert<T>,
1441 ) -> DbResult<()> {
1442 if key.len() != std::mem::size_of::<K>() {
1443 return Err(DbError::CorruptedEntry { offset: 0 });
1444 }
1445 let k = K::from_bytes(key);
1446 self.upsert(&k, f)?;
1447 Ok(())
1448 }
1449}
1450
1451#[cfg(feature = "typed-tree")]
1452impl<
1453 K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1454 const V: usize,
1455 T: crate::CollectionMeta
1456 + Copy
1457 + zerocopy::IntoBytes
1458 + zerocopy::FromBytes
1459 + zerocopy::Immutable
1460 + Default
1461 + PartialEq
1462 + Send
1463 + Sync
1464 + 'static,
1465 H: TypedWriteHook<K, T> + 'static,
1466> crate::projection::IndexKeyed for ZeroMap<K, V, T, H, Fixed>
1467{
1468 type K = K;
1469 type V = T;
1470
1471 fn encode_value(&self, v: &T) -> DbResult<Vec<u8>> {
1472 Ok(zerocopy::IntoBytes::as_bytes(v).to_vec())
1473 }
1474}
1475
1476#[cfg(feature = "typed-tree")]
1477impl<
1478 K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1479 const V: usize,
1480 T: crate::CollectionMeta
1481 + Copy
1482 + zerocopy::IntoBytes
1483 + zerocopy::FromBytes
1484 + zerocopy::Immutable
1485 + Default
1486 + PartialEq
1487 + Send
1488 + Sync
1489 + 'static,
1490 H: TypedWriteHook<K, T> + 'static,
1491> crate::counter::CounterCollection for ZeroMap<K, V, T, H, Fixed>
1492{
1493 fn decode_value(&self, bytes: &[u8]) -> DbResult<T> {
1494 zerocopy::FromBytes::read_from_bytes(bytes)
1495 .map_err(|_| DbError::FormatMismatch("zero counter value decode".into()))
1496 }
1497
1498 fn read_bytes(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
1500 if key.len() != std::mem::size_of::<K>() {
1501 return Err(DbError::CorruptedEntry { offset: 0 });
1502 }
1503 let k = K::from_bytes(key);
1504 self.get(&k).map(|v| self.encode_value(&v)).transpose()
1505 }
1506
1507 fn contains_bytes(&self, key: &[u8]) -> DbResult<bool> {
1508 if key.len() != std::mem::size_of::<K>() {
1509 return Err(DbError::CorruptedEntry { offset: 0 });
1510 }
1511 let k = K::from_bytes(key);
1512 Ok(self.contains(&k))
1513 }
1514
1515 fn upsert_counter(
1516 &self,
1517 key: &[u8],
1518 f: &mut dyn FnMut(Option<&T>) -> Upsert<T>,
1519 ) -> DbResult<()> {
1520 if key.len() != std::mem::size_of::<K>() {
1521 return Err(DbError::CorruptedEntry { offset: 0 });
1522 }
1523 let k = K::from_bytes(key);
1524 self.upsert(&k, f)?;
1525 Ok(())
1526 }
1527}
1528
1529#[cfg(test)]
1530mod tests {
1531 use super::*;
1532 use crate::Config;
1533 use crate::hook::TypedWriteHook;
1534 use std::collections::HashMap as StdHashMap;
1535 use std::mem::size_of;
1536 use std::sync::Arc;
1537 use std::sync::atomic::{AtomicUsize, Ordering as AO};
1538 use tempfile::tempdir;
1539 use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout};
1540
1541 #[derive(
1542 FromBytes, IntoBytes, Immutable, KnownLayout, Clone, Copy, Debug, PartialEq, Eq, Default,
1543 )]
1544 #[repr(C)]
1545 struct Counters {
1546 views: u64,
1547 likes: u64,
1548 }
1549
1550 type Map = ZeroMap<[u8; 8], { size_of::<Counters>() }, Counters>;
1551
1552 fn open(dir: &std::path::Path) -> Map {
1553 ZeroMap::<[u8; 8], { size_of::<Counters>() }, Counters>::open(dir, Config::test()).unwrap()
1554 }
1555
1556 fn key_on_other_shard(map: &Map, anchor: &[u8; 8]) -> [u8; 8] {
1557 let anchor_shard = map.shard_for(anchor);
1558 (2u64..1000)
1559 .map(|i| i.to_be_bytes())
1560 .find(|k| map.shard_for(k) != anchor_shard)
1561 .expect("key on different shard")
1562 }
1563
1564 #[test]
1565 fn for_each_visits_all_entries() {
1566 let dir = tempdir().unwrap();
1567 let map = open(dir.path());
1568
1569 let key_a = 1u64.to_be_bytes();
1570 let key_b = key_on_other_shard(&map, &key_a);
1571 let key_c = key_on_other_shard(&map, &key_b);
1572
1573 let expected = StdHashMap::from([
1574 (
1575 key_a,
1576 Counters {
1577 views: 10,
1578 likes: 1,
1579 },
1580 ),
1581 (
1582 key_b,
1583 Counters {
1584 views: 20,
1585 likes: 2,
1586 },
1587 ),
1588 (
1589 key_c,
1590 Counters {
1591 views: 30,
1592 likes: 3,
1593 },
1594 ),
1595 ]);
1596 for (key, value) in &expected {
1597 map.put(key, value).unwrap();
1598 }
1599
1600 let mut seen = StdHashMap::new();
1601 map.for_each(|key, value| {
1602 seen.insert(*key, *value);
1603 });
1604
1605 assert_eq!(seen, expected);
1606 assert_ne!(map.shard_for(&key_a), map.shard_for(&key_b));
1607 }
1608
1609 struct ZHook {
1610 writes: Arc<AtomicUsize>,
1611 last_new: Arc<crate::sync::Mutex<Option<u32>>>,
1612 }
1613
1614 impl ZHook {
1615 fn new() -> Self {
1616 Self {
1617 writes: Arc::new(AtomicUsize::new(0)),
1618 last_new: Arc::new(crate::sync::Mutex::new(None)),
1619 }
1620 }
1621 }
1622
1623 impl TypedWriteHook<[u8; 8], u32> for ZHook {
1624 fn on_write(&self, _k: &[u8; 8], _old: Option<&u32>, new: Option<&u32>) {
1625 self.writes.fetch_add(1, AO::Relaxed);
1626 *crate::sync::lock(&self.last_new) = new.copied();
1627 }
1628 }
1629
1630 #[test]
1631 fn zero_map_atomic_fires_typed_hook() {
1632 let dir = tempfile::tempdir().unwrap();
1633 let mut cfg = Config::test();
1634 cfg.shard_count = 1;
1635 let hook = ZHook::new();
1636 let writes = Arc::clone(&hook.writes);
1637 let last_new = Arc::clone(&hook.last_new);
1638 let map = ZeroMap::<[u8; 8], 4, u32, _>::open_hooked(dir.path(), cfg, hook).unwrap();
1639 let k = 1u64.to_be_bytes();
1640 map.atomic(&k, |s| {
1641 s.put(&k, &42u32)?;
1642 Ok(())
1643 })
1644 .unwrap();
1645 assert_eq!(writes.load(AO::Relaxed), 1);
1646 assert_eq!(*crate::sync::lock(&last_new), Some(42));
1647 }
1648
1649 #[test]
1650 fn zero_map_first_last_contract_both_backends() {
1651 let bitcask_dir = tempdir().unwrap();
1652 let bitcask_cfg = Config::balanced()
1653 .shard_count(2)
1654 .hints(true)
1655 .iterable(true)
1656 .reversed(false)
1657 .build();
1658 let bitcask = ZeroMap::<[u8; 8], 8, u64>::open(bitcask_dir.path(), bitcask_cfg).unwrap();
1659 assert!(bitcask.first().unwrap().is_none());
1660 assert!(bitcask.last().unwrap().is_none());
1661 for id in [3_u64, 1, 2] {
1662 bitcask.put(&id.to_be_bytes(), &(id * 10)).unwrap();
1663 }
1664 assert_eq!(
1665 bitcask
1666 .first()
1667 .unwrap()
1668 .map(|(k, v)| (u64::from_be_bytes(k), v)),
1669 Some((1, 10))
1670 );
1671 assert_eq!(
1672 bitcask
1673 .last()
1674 .unwrap()
1675 .map(|(k, v)| (u64::from_be_bytes(k), v)),
1676 Some((3, 30))
1677 );
1678
1679 let fixed_dir = tempdir().unwrap();
1680 let fixed_cfg = crate::FixedConfig {
1681 iterable: true,
1682 reversed: false,
1683 ..crate::FixedConfig::test()
1684 };
1685 let fixed =
1686 ZeroMap::<[u8; 8], 8, u64, NoHook, Fixed>::open(fixed_dir.path(), fixed_cfg).unwrap();
1687 assert!(fixed.first().unwrap().is_none());
1688 assert!(fixed.last().unwrap().is_none());
1689 for id in [3_u64, 1, 2] {
1690 fixed.put(&id.to_be_bytes(), &(id * 10)).unwrap();
1691 }
1692 assert_eq!(
1693 fixed
1694 .first()
1695 .unwrap()
1696 .map(|(k, v)| (u64::from_be_bytes(k), v)),
1697 Some((1, 10))
1698 );
1699 assert_eq!(
1700 fixed
1701 .last()
1702 .unwrap()
1703 .map(|(k, v)| (u64::from_be_bytes(k), v)),
1704 Some((3, 30))
1705 );
1706
1707 let no_index_dir = tempdir().unwrap();
1708 let no_index =
1709 ZeroMap::<[u8; 8], 8, u64>::open(no_index_dir.path(), Config::test()).unwrap();
1710 assert!(matches!(no_index.first(), Err(DbError::NotImplemented)));
1711 assert!(matches!(no_index.last(), Err(DbError::NotImplemented)));
1712 }
1713
1714 #[test]
1715 fn zero_map_fixed_direction() {
1716 #[derive(Clone, Copy, FromBytes, IntoBytes, Immutable)]
1717 #[repr(C)]
1718 struct V {
1719 x: u32,
1720 }
1721 let dir = tempdir().unwrap();
1722 let cfg = crate::FixedConfig {
1723 iterable: true,
1724 reversed: false,
1725 ..crate::FixedConfig::test()
1726 };
1727 let m = ZeroMap::<[u8; 8], { size_of::<V>() }, V, NoHook, Fixed>::open(dir.path(), cfg)
1728 .unwrap();
1729 for i in 1u64..=3 {
1730 m.put(&i.to_be_bytes(), &V { x: i as u32 }).unwrap();
1731 }
1732 let keys: Vec<u64> = m
1733 .iter_view()
1734 .unwrap()
1735 .keys()
1736 .map(u64::from_be_bytes)
1737 .collect();
1738 assert_eq!(keys, [1, 2, 3], "ZeroMap<Fixed> reversed=false ASC");
1739 }
1740
1741 #[test]
1742 fn zero_map_iter_yields_decoded_values_desc() {
1743 #[derive(Clone, Copy, FromBytes, IntoBytes, Immutable)]
1744 #[repr(C)]
1745 struct Val {
1746 x: u32,
1747 }
1748 let dir = tempdir().unwrap();
1749 let cfg = Config::balanced()
1750 .shard_count(2)
1751 .hints(true)
1752 .iterable(true)
1753 .build();
1754 let m = ZeroMap::<[u8; 2], { size_of::<Val>() }, Val>::open(dir.path(), cfg).unwrap();
1755 m.put(&[1, 0], &Val { x: 11 }).unwrap();
1756 m.put(&[2, 0], &Val { x: 22 }).unwrap();
1757 let got: Vec<(u8, u32)> = m
1758 .iter_view()
1759 .unwrap()
1760 .iter()
1761 .map(|(k, v)| (k[0], v.x))
1762 .collect();
1763 assert_eq!(got, [(2, 22), (1, 11)]);
1764 let keys: Vec<u8> = m.iter_view().unwrap().keys().map(|k| k[0]).collect();
1766 assert_eq!(keys, [2, 1]);
1767 }
1768
1769 #[test]
1770 fn zero_map_shard_update_and_fetch_update() {
1771 let dir = tempdir().unwrap();
1772 let mut cfg = Config::test();
1773 cfg.shard_count = 1;
1774 let map = ZeroMap::<[u8; 8], 4, u32>::open(dir.path(), cfg).unwrap();
1775 let key = 1u64.to_be_bytes();
1776 map.put(&key, &100u32).unwrap();
1777
1778 let (upd, fetched, missing) = map
1779 .atomic(&key, |s| {
1780 let upd = s.update(&key, |old| old + 1)?; let fetched = s.fetch_update(&key, |old| old + 1)?; let missing = s.update(&2u64.to_be_bytes(), |old| *old)?;
1783 Ok((upd, fetched, missing))
1784 })
1785 .expect("atomic");
1786
1787 assert_eq!(upd, Some(101));
1788 assert_eq!(fetched, Some(101));
1789 assert_eq!(missing, None);
1790 assert_eq!(map.get(&key), Some(102));
1791 }
1792
1793 #[test]
1794 fn for_each_while_stops_after_break_without_visiting_later_shards() {
1795 use std::ops::ControlFlow;
1796 use std::sync::atomic::{AtomicUsize, Ordering};
1797
1798 let dir = tempdir().unwrap();
1799 let map = open(dir.path());
1800
1801 let key_a = 1u64.to_be_bytes();
1802 let key_b = key_on_other_shard(&map, &key_a);
1803 assert_ne!(map.shard_for(&key_a), map.shard_for(&key_b));
1804
1805 map.put(&key_a, &Counters { views: 1, likes: 0 }).unwrap();
1806 map.put(&key_b, &Counters { views: 2, likes: 0 }).unwrap();
1807
1808 let seen = AtomicUsize::new(0);
1809 let flow = map.for_each_while(|_k, _v| {
1810 seen.fetch_add(1, Ordering::Relaxed);
1811 ControlFlow::Break(())
1812 });
1813 assert_eq!(seen.load(Ordering::Relaxed), 1);
1814 assert!(matches!(flow, ControlFlow::Break(())));
1815 }
1816}