egglog-core-relations 3.0.0

egglog is a language that combines the benefits of equality saturation and datalog. It can be used for analysis, optimization, and synthesis of programs. It is the successor to the popular rust library egg.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
//! Support for containers
//!
//! Containers behave a lot like base values. They are implemented differently because
//! their ids share a space with other Ids in the egraph and as a result, their ids need to be
//! sparse.
//!
//! This is a relatively "eagler" implementation of containers, reflecting egglog's current
//! semantics. One could imagine a variant of containers in which they behave more like egglog
//! functions than base values.

use std::{
    any::{Any, TypeId},
    hash::{Hash, Hasher},
    ops::Deref,
};

use crate::numeric_id::{DenseIdMap, IdVec, NumericId, define_id};
use crossbeam_queue::SegQueue;
use dashmap::SharedValue;
use rustc_hash::FxHasher;

use crate::{
    ColumnId, CounterId, ExecutionState, Offset, SubsetRef, TableId, TaggedRowBuffer, Value,
    WrappedTable,
    common::{DashMap, IndexSet, SubsetTracker},
    parallel,
    parallel_heuristics::{parallelize_inter_container_op, parallelize_intra_container_op},
    table_spec::{Rebuilder, ValueRebuilder},
};

#[cfg(test)]
mod tests;

define_id!(pub ContainerValueId, u32, "an identifier for containers");

pub trait MergeFn:
    Fn(&mut ExecutionState, Value, Value) -> Value + dyn_clone::DynClone + Send + Sync
{
}
impl<T: Fn(&mut ExecutionState, Value, Value) -> Value + Clone + Send + Sync> MergeFn for T {}

// Implements `Clone` for `Box<dyn MergeFn>`.
dyn_clone::clone_trait_object!(MergeFn);

#[derive(Clone, Default)]
struct ContainerIds {
    ids: IndexSet<TypeId>,
}

impl ContainerIds {
    fn insert(&mut self, ty: TypeId) -> ContainerValueId {
        if let Some(idx) = self.ids.get_index_of(&ty) {
            ContainerValueId::from_usize(idx)
        } else {
            let idx = self.ids.len();
            self.ids.insert(ty);
            ContainerValueId::from_usize(idx)
        }
    }

    fn get(&self, ty: &TypeId) -> Option<ContainerValueId> {
        self.ids.get_index_of(ty).map(ContainerValueId::from_usize)
    }
}

#[derive(Clone, Default)]
pub struct ContainerValues {
    subset_tracker: SubsetTracker,
    container_ids: ContainerIds,
    data: DenseIdMap<ContainerValueId, Box<dyn DynamicContainerEnv + Send + Sync>>,
}

/// Summary returned by container rebuild.
///
/// `changed` means some container entry changed during rebuild, either because
/// its contents changed or because its outer id canonicalized.
///
/// `dirty_ids` is narrower: it records container ids whose semantics changed
/// while their stored outer id stayed stable. Ordinary table rebuild already
/// handles changed-id cases; these ids need a follow-up parent-row refresh.
/// This includes containers that changed directly and containers whose
/// contained containers changed in place.
///
/// For example, `l(vec-of(w(k(b))))` can rebuild to `l(vec-of(k(b)))` without
/// changing the `Vec` id. The row is now newly matchable, but seminaive will
/// miss it unless the parent row is retimestamped.
#[derive(Clone, Default)]
pub struct ContainerRebuildSummary {
    changed: bool,
    // Container ids whose semantics changed in a way that may not produce a
    // fresh parent-row delta during ordinary table rebuild.
    dirty_ids: IndexSet<Value>,
}

impl ContainerRebuildSummary {
    /// Returns whether any container entry changed during rebuild.
    pub fn changed(&self) -> bool {
        self.changed
    }

    /// Returns the container ids whose parent rows may need retimestamping.
    pub fn dirty_ids(&self) -> &IndexSet<Value> {
        &self.dirty_ids
    }

    fn note_change(&mut self) {
        self.changed = true;
    }

    fn note_dirty_id(&mut self, value: Value) {
        self.changed = true;
        self.dirty_ids.insert(value);
    }

    fn extend(&mut self, other: Self) {
        self.changed |= other.changed;
        self.dirty_ids.extend(other.dirty_ids);
    }
}

impl ContainerValues {
    pub fn new() -> Self {
        Default::default()
    }

    fn get<C: ContainerValue>(&self) -> Option<&ContainerEnv<C>> {
        let id = self.container_ids.get(&TypeId::of::<C>())?;
        let res = self.data.get(id)?.as_any();
        Some(res.downcast_ref::<ContainerEnv<C>>().unwrap())
    }

    /// Iterate over the containers of the given type.
    pub fn for_each<C: ContainerValue>(&self, mut f: impl FnMut(&C, Value)) {
        let Some(env) = self.get::<C>() else {
            return;
        };
        for ent in env.to_id.iter() {
            f(ent.key(), *ent.value());
        }
    }

    /// Get the container associated with the value `val` in the database. The caller must know the
    /// type of the container.
    ///
    /// The return type of this function may contain lock guards. Attempts to modify the contents
    /// of the containers database may deadlock if the given guard has not been dropped.
    pub fn get_val<C: ContainerValue>(&self, val: Value) -> Option<impl Deref<Target = C> + '_> {
        self.get::<C>()?.get_container(val)
    }

    pub fn register_val<C: ContainerValue>(
        &self,
        container: C,
        exec_state: &mut ExecutionState,
    ) -> Value {
        let env = self
            .get::<C>()
            .expect("must register container type before registering a value");
        env.get_or_insert(&container, exec_state)
    }

    /// Rebuild a single container value by remapping each contained value
    /// through `remap`, returning the (possibly new) interned value, or `None`
    /// if `value` is not a registered container of the type behind `type_id`.
    ///
    /// The original container is left alone; the result is interned separately.
    ///
    /// Unlike [`ContainerValues::rebuild_all`], which drives rebuilds off the
    /// backend union-find, the caller supplies the remapping explicitly and
    /// identifies the container type dynamically by its [`TypeId`].
    pub fn rebuild_val_with(
        &self,
        type_id: TypeId,
        value: Value,
        exec_state: &mut ExecutionState,
        remap: &(dyn Fn(Value) -> Value + Send + Sync),
    ) -> Option<Value> {
        let id = self.container_ids.get(&type_id)?;
        let env = self.data.get(id)?;
        env.rebuild_val_with(value, exec_state, remap)
    }

    /// Apply the given rebuild to the contents of each container.
    pub fn rebuild_all(
        &mut self,
        table_id: TableId,
        table: &WrappedTable,
        exec_state: &mut ExecutionState,
    ) -> ContainerRebuildSummary {
        let Some(rebuilder) = table.rebuilder(&[]) else {
            return Default::default();
        };
        let to_scan = rebuilder.hint_col().map(|_| {
            // We may attempt an incremental rebuild.
            self.subset_tracker.recent_updates(table_id, table)
        });
        let mut summary = if parallelize_inter_container_op(self.data.next_id().index()) {
            parallel::map_dense_id_map_mut(&mut self.data, |_, env| {
                let mut exec_state = exec_state.clone();
                env.apply_rebuild(
                    table,
                    &*rebuilder,
                    to_scan.as_ref().map(|x| x.as_ref()),
                    &mut exec_state,
                )
            })
            .into_iter()
            .fold(ContainerRebuildSummary::default(), |mut acc, summary| {
                acc.extend(summary);
                acc
            })
        } else {
            let mut summary = ContainerRebuildSummary::default();
            for (_, env) in self.data.iter_mut() {
                summary.extend(env.apply_rebuild(
                    table,
                    &*rebuilder,
                    to_scan.as_ref().map(|x| x.as_ref()),
                    exec_state,
                ));
            }
            summary
        };
        self.expand_dirty_id_closure(&mut summary);
        summary
    }

    /// Add ancestor containers to the dirty-id set until it is transitively closed.
    ///
    /// A rebuild can change a container's semantics in place without changing
    /// its id. If that container is itself stored inside another container,
    /// the parent container has also changed semantically even though no direct
    /// rebuild touched its contents. For example, with
    /// `(p (vec-of (vec-of (w (b)))))` and `(rewrite (w x) x)`, the inner
    /// `Vec` rebuilds in place to `vec-of (b)`. Without this closure, only the
    /// inner `Vec` id is dirty; the outer `Vec` row is not retimestamped, so a
    /// later rule like `(rewrite (p (vec-of (vec-of (b)))) (b))` can miss the
    /// newly matchable parent row.
    fn expand_dirty_id_closure(&self, summary: &mut ContainerRebuildSummary) {
        let mut frontier = summary.dirty_ids.clone();
        let mut seen = frontier.iter().copied().collect::<IndexSet<_>>();

        while !frontier.is_empty() {
            let mut next = IndexSet::default();
            for (_, env) in self.data.iter() {
                env.extend_containers_containing(&frontier, &mut next);
            }
            frontier.clear();
            for value in next {
                if seen.insert(value) {
                    summary.note_dirty_id(value);
                    frontier.insert(value);
                }
            }
        }
    }

    /// Add a new container type to the given [`ContainerValue`] instance.
    ///
    /// Container types need a meaans of generating fresh ids (`id_counter`) along with a means of
    /// merging conflicting ids (`merge_fn`).
    pub fn register_type<C: ContainerValue>(
        &mut self,
        id_counter: CounterId,
        merge_fn: impl MergeFn + 'static,
    ) -> ContainerValueId {
        let id = self.container_ids.insert(TypeId::of::<C>());
        self.data.get_or_insert(id, || {
            Box::new(ContainerEnv::<C>::new(Box::new(merge_fn), id_counter))
        });
        id
    }
}

/// A trait implemented by container types.
///
/// Containers behave a lot like base values, but they include extra trait methods to support
/// rebuilding of container contents and merging containers that become equal after a rebuild pass
/// has taken place.
pub trait ContainerValue: Hash + Eq + Clone + Send + Sync + 'static {
    /// Rebuild an additional container in place according the the given [`ValueRebuilder`].
    ///
    /// If this method returns `false` then the container must not have been modified (i.e. it must
    /// hash to the same value, and compare equal to a copy of itself before the call).
    fn rebuild_contents(&mut self, rebuilder: &dyn ValueRebuilder) -> bool;

    /// Iterate over the contents of the container.
    ///
    /// Note that containers can be more structured than just a sequence of values. This iterator
    /// is used to populate an index that in turn is used to speed up rebuilds. If a value in the
    /// container is eligible for a rebuild and it is not mentioned by this iterator, the outer
    /// container registry may skip rebuilding this container.
    fn iter(&self) -> impl Iterator<Item = Value> + '_;
}

pub trait DynamicContainerEnv: Any + dyn_clone::DynClone + Send + Sync {
    fn as_any(&self) -> &dyn Any;
    fn apply_rebuild(
        &mut self,
        table: &WrappedTable,
        rebuilder: &dyn Rebuilder,
        subset: Option<SubsetRef>,
        exec_state: &mut ExecutionState,
    ) -> ContainerRebuildSummary;
    /// Add ids for containers in this environment that contain any `values`.
    ///
    /// This uses the container content index populated from
    /// [`ContainerValue::iter`] and lets callers climb from dirty child ids to
    /// all directly containing parent container ids.
    fn extend_containers_containing(&self, values: &IndexSet<Value>, out: &mut IndexSet<Value>);
    /// Rebuild the single container `value` by remapping each contained value
    /// through `remap`, returning the (possibly new) interned value, or `None`
    /// if `value` is not registered in this environment.
    fn rebuild_val_with(
        &self,
        value: Value,
        exec_state: &mut ExecutionState,
        remap: &(dyn Fn(Value) -> Value + Send + Sync),
    ) -> Option<Value>;
}

// Implements `Clone` for `Box<dyn DynamicContainerEnv>`.
dyn_clone::clone_trait_object!(DynamicContainerEnv);

fn hash_container(container: &impl ContainerValue) -> u64 {
    let mut hasher = FxHasher::default();
    container.hash(&mut hasher);
    hasher.finish()
}

#[derive(Clone)]
struct ContainerEnv<C: Eq + Hash> {
    merge_fn: Box<dyn MergeFn>,
    counter: CounterId,
    to_id: DashMap<C, Value>,
    to_container: DashMap<Value, (usize /* hash code */, usize /* map */)>,
    /// Map from a Value to the set of ids of containers that contain that value.
    val_index: DashMap<Value, IndexSet<Value>>,
}

impl<C: ContainerValue> DynamicContainerEnv for ContainerEnv<C> {
    fn as_any(&self) -> &dyn Any {
        self
    }

    fn apply_rebuild(
        &mut self,
        table: &WrappedTable,
        rebuilder: &dyn Rebuilder,
        subset: Option<SubsetRef>,
        exec_state: &mut ExecutionState,
    ) -> ContainerRebuildSummary {
        if let Some(subset) = subset
            && incremental_rebuild(
                subset.size(),
                self.to_id.len(),
                parallelize_intra_container_op(self.to_id.len()),
            )
        {
            return self.apply_rebuild_incremental(
                table,
                rebuilder,
                exec_state,
                subset,
                rebuilder.hint_col().unwrap(),
            );
        }
        self.apply_rebuild_nonincremental(rebuilder, exec_state)
    }

    fn extend_containers_containing(&self, values: &IndexSet<Value>, out: &mut IndexSet<Value>) {
        for value in values {
            if let Some(containers) = self.val_index.get(value) {
                out.extend(containers.iter().copied());
            }
        }
    }

    fn rebuild_val_with(
        &self,
        value: Value,
        exec_state: &mut ExecutionState,
        remap: &(dyn Fn(Value) -> Value + Send + Sync),
    ) -> Option<Value> {
        // Clone out of the guard before re-interning to avoid deadlocking on
        // the underlying map.
        let mut container = self.get_container(value)?.clone();
        container.rebuild_contents(&ClosureRebuilder { remap });
        Some(self.get_or_insert(&container, exec_state))
    }
}

impl<C: ContainerValue> ContainerEnv<C> {
    pub fn new(merge_fn: Box<dyn MergeFn>, counter: CounterId) -> Self {
        Self {
            merge_fn,
            counter,
            to_id: DashMap::default(),
            to_container: DashMap::default(),
            val_index: DashMap::default(),
        }
    }

    fn get_or_insert(&self, container: &C, exec_state: &mut ExecutionState) -> Value {
        if let Some(value) = self.to_id.get(container) {
            return *value;
        }

        // Time to insert a new mapping. First, insert into `to_container`: the moment that we
        // insert a new value into `to_id`, someone else can return it from another call to
        // `get_or_insert` and then feed that value to `get_container`.

        let value = Value::from_usize(exec_state.inc_counter(self.counter));
        let target_map = self.to_id.determine_map(container);
        // This assertion is here because in parallel rebuilding we use `to_container` to
        // compute the intended shard for to_id, because we have a mutable borrow of
        // `to_container` that means we cannot call `determine_map` on `to_id`.
        debug_assert_eq!(
            target_map,
            self.to_container
                .determine_shard(hash_container(container) as usize)
        );
        self.to_container
            .insert(value, (hash_container(container) as usize, target_map));

        // Now insert into `to_id`, handling the case where a different thread is doing the same
        // thing.
        match self.to_id.entry(container.clone()) {
            dashmap::Entry::Vacant(vac) => {
                // Common case: insert the mapping in to_id and update the index.
                vac.insert(value);
                for val in container.iter() {
                    self.val_index.entry(val).or_default().insert(value);
                }
                value
            }
            dashmap::Entry::Occupied(occ) => {
                // Someone inserted `container` into the mapping since we looked it up. Remove the
                // mapping that we inserted into `to_container` (we won't use it), and instead
                // return the "winning" value.
                let res = *occ.get();
                std::mem::drop(occ); // drop the lock.
                self.to_container.remove(&value);
                res
            }
        }
    }

    fn insert_owned(&self, container: C, value: Value, exec_state: &mut ExecutionState) -> Value {
        let hc = hash_container(&container);
        let target_map = self.to_id.determine_map(&container);
        match self.to_id.entry(container) {
            dashmap::Entry::Occupied(mut occ) => {
                let result = (self.merge_fn)(exec_state, *occ.get(), value);
                let old_val = *occ.get();
                if result != old_val {
                    self.to_container.remove(&old_val);
                    self.to_container.insert(result, (hc as usize, target_map));
                    *occ.get_mut() = result;
                    for val in occ.key().iter() {
                        let mut index = self.val_index.entry(val).or_default();
                        index.swap_remove(&old_val);
                        index.insert(result);
                    }
                }
                result
            }
            dashmap::Entry::Vacant(vacant_entry) => {
                self.to_container.insert(value, (hc as usize, target_map));
                for val in vacant_entry.key().iter() {
                    self.val_index.entry(val).or_default().insert(value);
                }
                vacant_entry.insert(value);
                value
            }
        }
    }

    fn reinsert_incremental(
        &self,
        container: C,
        old_id: Value,
        rebuilt_id: Value,
        container_changed: bool,
        exec_state: &mut ExecutionState,
        summary: &mut ContainerRebuildSummary,
    ) {
        if container_changed || rebuilt_id != old_id {
            summary.note_change();
        }
        if rebuilt_id != old_id {
            // Parent rows will get a real delta from ordinary table rebuild, so
            // we only need an explicit refresh when the outer id stayed stable.
            self.to_container.remove(&old_id);
        }
        let actual = self.insert_owned(container, rebuilt_id, exec_state);
        if container_changed && rebuilt_id == old_id && actual == old_id {
            summary.note_dirty_id(old_id);
        }
    }

    fn apply_rebuild_incremental(
        &mut self,
        table: &WrappedTable,
        rebuilder: &dyn Rebuilder,
        exec_state: &mut ExecutionState,
        to_scan: SubsetRef,
        search_col: ColumnId,
    ) -> ContainerRebuildSummary {
        // NB: there is no parallel implementation as of now.
        //
        // Implementing one should be straightforward, but we should wait for a real benchmark that
        // requires it. It's possible that incremental rebuilding will only be profitable when the
        // total number of ids to rebuild is small, in which case the overhead of parallelism may
        // not be worth it in the first place.
        let mut summary = ContainerRebuildSummary::default();
        let mut buf = TaggedRowBuffer::new(1);
        table.scan_project(
            to_scan,
            &[search_col],
            Offset::new(0),
            usize::MAX,
            &[],
            &mut buf,
        );
        // For each value in the buffer, rebuild all containers that mention it.
        let mut to_rebuild = IndexSet::<Value>::default();
        for (_, row) in buf.iter() {
            to_rebuild.insert(row[0]);
            let Some(ids) = self.val_index.get(&row[0]) else {
                continue;
            };
            to_rebuild.extend(&*ids);
        }
        for id in to_rebuild {
            let Some((hc, target_map)) = self.to_container.get(&id).map(|x| *x) else {
                continue;
            };
            let shard_mut = self.to_id.shards_mut()[target_map].get_mut();
            let Some((mut container, _)) =
                shard_mut.remove_entry(hc as u64, |(_, v)| *v.get() == id)
            else {
                continue;
            };
            let rebuilt_id = rebuilder.rebuild_val(id);
            let container_changed = container.rebuild_contents(rebuilder);
            self.reinsert_incremental(
                container,
                id,
                rebuilt_id,
                container_changed,
                exec_state,
                &mut summary,
            );
        }
        summary
    }

    fn apply_rebuild_nonincremental(
        &mut self,
        rebuilder: &dyn Rebuilder,
        exec_state: &mut ExecutionState,
    ) -> ContainerRebuildSummary {
        if parallelize_inter_container_op(self.to_id.len()) {
            return self.apply_rebuild_nonincremental_parallel(rebuilder, exec_state);
        }
        let mut summary = ContainerRebuildSummary::default();
        let mut to_reinsert = Vec::new();
        let shards = self.to_id.shards_mut();
        for shard in shards.iter_mut() {
            let shard = shard.get_mut();
            // SAFETY: the iterator does not outlive `shard`.
            for bucket in unsafe { shard.iter() } {
                // SAFETY: the bucket is valid; we just got it from the iterator.
                let (container, val) = unsafe { bucket.as_mut() };
                let old_val = *val.get();
                let new_val = rebuilder.rebuild_val(old_val);
                let container_changed = container.rebuild_contents(rebuilder);
                if !container_changed && new_val == old_val {
                    // Nothing changed about this entry. Leave it in place.
                    continue;
                }
                summary.note_change();
                if container_changed {
                    // The container changed. Remove both map entries then reinsert.
                    // SAFETY: This is a valid bucket. Furthermore, iterators remain valid if
                    // buckets they have already yielded have been removed.
                    let ((container, _), _) = unsafe { shard.remove(bucket) };
                    self.to_container.remove(&old_val);
                    to_reinsert.push((container, new_val, new_val == old_val));
                } else {
                    // Just the value changed. Leave the container in place.
                    *val.get_mut() = new_val;
                    let prev = self.to_container.remove(&old_val).unwrap().1;
                    self.to_container.insert(new_val, prev);
                }
            }
        }
        for (container, val, stable_id) in to_reinsert {
            let actual = self.insert_owned(container, val, exec_state);
            // Refresh only when rebuild changed container semantics in place.
            // If the outer id changed, ordinary table rebuild already creates a
            // fresh parent-row delta for seminaive to follow.
            if stable_id && actual == val {
                summary.note_dirty_id(val);
            }
        }
        summary
    }

    fn apply_rebuild_nonincremental_parallel(
        &mut self,
        rebuilder: &dyn Rebuilder,
        exec_state: &mut ExecutionState,
    ) -> ContainerRebuildSummary {
        // This is very similar to the serial variant. The main difference is that
        // `to_reinsert` isn't a flat vector. It's instead a vector of queues - one per
        // destination map shard. This lets us do a bulk insertion in parallel without having
        // to grab a lock per container.
        let mut to_reinsert =
            IdVec::<usize /* to_id shard */, SegQueue<(C, Value, bool)>>::default();
        to_reinsert.resize_with(self.to_id.shards().len(), Default::default);

        let shards = self.to_id.shards_mut();
        let changed = parallel::map_mut(shards, |_, shard| {
            let mut changed = false;
            let shard = shard.get_mut();
            // SAFETY: the iterator does not outlive `shard`.
            for bucket in unsafe { shard.iter() } {
                // SAFETY: the bucket is valid; we just got it from the iterator.
                let (container, val) = unsafe { bucket.as_mut() };
                let old_val = *val.get();
                let new_val = rebuilder.rebuild_val(old_val);
                let container_changed = container.rebuild_contents(rebuilder);
                if !container_changed && new_val == old_val {
                    // Nothing changed about this entry. Leave it in place.
                    continue;
                }
                changed = true;
                if container_changed {
                    // The container changed. Remove both map entries then reinsert.
                    // SAFETY: This is a valid bucket. Furthermore, iterators remain valid if
                    // buckets they have already yielded have been removed.
                    let ((container, _), _) = unsafe { shard.remove(bucket) };
                    self.to_container.remove(&old_val);
                    // Spooky: we're using `to_container` to determine the shard for
                    // `to_id`. We are assuming that the # shards determination is
                    // deterministic here. There is a debug assertion in `get_or_insert`
                    // that attempts to verify this.
                    let shard = self
                        .to_container
                        .determine_shard(hash_container(&container) as usize);
                    to_reinsert[shard].push((container, new_val, new_val == old_val));
                } else {
                    // Just the value changed. Leave the container in place.
                    *val.get_mut() = new_val;
                    let prev = self.to_container.remove(&old_val).unwrap().1;
                    self.to_container.insert(new_val, prev);
                }
            }
            changed
        })
        .into_iter()
        .any(|changed| changed);

        let dirty_ids = SegQueue::new();
        parallel::for_each_mut(shards, |shard_id, shard| {
            let mut exec_state = exec_state.clone();
            // This bit is a real slog. Once Dashmap updates from RawTable to HashTable for
            // the underlying shard, this will get a little better.
            //
            // NB: We are probably leaving some paralellism on the floor with these calls
            // to `to_container` and `val_index`.
            let shard = shard.get_mut();
            let queue = &to_reinsert[shard_id];
            while let Some((container, val, stable_id)) = queue.pop() {
                let hc = hash_container(&container);
                let target_map = self.to_container.determine_shard(hc as usize);
                match shard.find_or_find_insert_slot(
                    hc,
                    |(c, _)| c == &container,
                    |(c, _)| hash_container(c),
                ) {
                    Ok(bucket) => {
                        // SAFETY: the bucket is valid; we just got it from the shard and
                        // we have not done any operations that can invalidate the bucket.
                        let (container, val_slot) = unsafe { bucket.as_mut() };
                        let old_val = *val_slot.get();
                        let result = (self.merge_fn)(&mut exec_state, old_val, val);
                        if result != old_val {
                            self.to_container.remove(&old_val);
                            self.to_container.insert(result, (hc as usize, target_map));
                            *val_slot.get_mut() = result;
                            for val in container.iter() {
                                let mut index = self.val_index.entry(val).or_default();
                                index.swap_remove(&old_val);
                                index.insert(result);
                            }
                        }
                        // As in the serial path, only same-id semantic
                        // changes need an explicit parent-row refresh.
                        if stable_id && result == val {
                            dirty_ids.push(val);
                        }
                    }
                    Err(slot) => {
                        self.to_container.insert(val, (hc as usize, target_map));
                        for v in container.iter() {
                            self.val_index.entry(v).or_default().insert(val);
                        }
                        // SAFETY: We just got this slot from `find_or_find_insert_slot`
                        // and we have not mutated the map at all since then.
                        unsafe {
                            shard.insert_in_slot(hc, slot, (container, SharedValue::new(val)));
                        }
                        if stable_id {
                            dirty_ids.push(val);
                        }
                    }
                }
            }
        });
        let mut summary = ContainerRebuildSummary::default();
        if changed {
            summary.note_change();
        }
        while let Some(value) = dirty_ids.pop() {
            summary.note_dirty_id(value);
        }
        summary
    }

    fn get_container(&self, value: Value) -> Option<impl Deref<Target = C> + '_> {
        let (hc, target_map) = *self.to_container.get(&value)?;
        let shard = &self.to_id.shards()[target_map];
        let read_guard = shard.read();
        let val_ptr: *const (C, _) = shard
            .read()
            .find(hc as u64, |(_, v)| *v.get() == value)?
            .as_ptr();
        struct ValueDeref<'a, T, Guard> {
            _guard: Guard,
            data: &'a T,
        }

        impl<T, Guard> Deref for ValueDeref<'_, T, Guard> {
            type Target = T;

            fn deref(&self) -> &T {
                self.data
            }
        }

        Some(ValueDeref {
            _guard: read_guard,
            // SAFETY: the value will remain valid for as long as `read_guard` is in scope.
            data: unsafe {
                let unwrapped: &(C, _) = &*val_ptr;
                &unwrapped.0
            },
        })
    }
}

fn incremental_rebuild(uf_size: usize, table_size: usize, parallel: bool) -> bool {
    if parallel {
        table_size > 1000 && uf_size * 512 <= table_size
    } else {
        table_size > 1000 && uf_size * 8 <= table_size
    }
}

/// A [`ValueRebuilder`] that remaps individual values through a caller-supplied
/// closure. Used by [`ContainerValues::rebuild_val_with`] to rebuild a single
/// container against an explicit value mapping rather than a backend union-find.
struct ClosureRebuilder<'a> {
    remap: &'a (dyn Fn(Value) -> Value + Send + Sync),
}

impl ValueRebuilder for ClosureRebuilder<'_> {
    fn rebuild_val(&self, val: Value) -> Value {
        (self.remap)(val)
    }
}