diff --git a/core-relations/src/free_join/execute.rs b/core-relations/src/free_join/execute.rs index 8f32d5419..6b56a24fc 100644 --- a/core-relations/src/free_join/execute.rs +++ b/core-relations/src/free_join/execute.rs @@ -3353,6 +3353,23 @@ fn num_intersected_rels(join_stage: &JoinStage) -> i32 { } } +fn is_reorder_barrier(stage: &JoinStage) -> bool { + matches!( + stage, + JoinStage::FusedIntersectMat { + mode: MatScanMode::Lookup(_) | MatScanMode::Value(_) | MatScanMode::Full, + .. + } + ) +} + +/// Choose the physical execution order for a suffix of an already-planned +/// stage block. +/// +/// The cached logical order fixes stage identity and provides a stable +/// refinement anchor. At execution time, all-`Intersect` blocks additionally +/// follow the branch-local physical frontier; mixed and fused blocks retain +/// the stable-prefix bias to avoid producing many divergent probe paths. fn sort_plan_by_size( order: &mut InstrOrder, leaf_scans: &mut LeafScans, @@ -3362,14 +3379,7 @@ fn sort_plan_by_size( ) { let mut last_pos = start; for i in start..instrs.len() { - if matches!( - &instrs[order.get(i)], - // These nodes don't commute - JoinStage::FusedIntersectMat { - mode: MatScanMode::Lookup(_) | MatScanMode::Value(_) | MatScanMode::Full, - .. - } - ) { + if is_reorder_barrier(&instrs[order.get(i)]) { sort_plan_by_size_inner(order, last_pos..i, instrs, binding_info); last_pos = i + 1; } @@ -3478,98 +3488,93 @@ fn sort_plan_by_size_inner( } // How many times an atom has been intersected/joined let mut times_refined = with_pool_set(|ps| ps.get::>()); - - // Count how many times each atom has been refined so far. - for position in 0..range.start { - match &instrs[order.get(position)] { + let use_physical_refinements = instrs + .iter() + .all(|stage| matches!(stage, JoinStage::Intersect { .. })); + let mut physical_refinements = + use_physical_refinements.then(|| with_pool_set(|ps| ps.get::>())); + + let update_refinements = + |stage: &JoinStage, refinements: &mut DenseIdMap| match stage { JoinStage::Intersect { scans, .. } => scans.iter().for_each(|scan| { - *times_refined.get_or_default(scan.atom) += 1; + *refinements.get_or_default(scan.atom) += 1; }), JoinStage::FusedIntersect { cover, to_intersect, .. } => { - *times_refined.get_or_default(cover.to_index.atom) += + *refinements.get_or_default(cover.to_index.atom) += cover.to_index.vars.len() as i64; to_intersect.iter().for_each(|(spec, _)| { - *times_refined.get_or_default(spec.to_index.atom) += + *refinements.get_or_default(spec.to_index.atom) += spec.to_index.vars.len() as i64; }); } JoinStage::FusedIntersectMat { to_intersect, .. } => { to_intersect.iter().for_each(|(spec, _)| { - *times_refined.get_or_default(spec.to_index.atom) += + *refinements.get_or_default(spec.to_index.atom) += spec.to_index.vars.len() as i64; }); } + }; + + // The cached logical prefix remains the stable locality signal. + for stage in &instrs[..range.start] { + update_refinements(stage, &mut times_refined); + } + // Pure intersection blocks also track the stages actually selected on this + // branch, keeping recursive ordering connected to its physical frontier. + if let Some(physical_refinements) = &mut physical_refinements { + for position in 0..range.start { + update_refinements(&instrs[order.get(position)], physical_refinements); } } - // We prioritize variables by - // - // (1) how many times an atom with this variable has been refined, - // (2) then by the cardinality of the variable to be enumerated (smaller → earlier) - // (3) then by how many relations join on this variable (more → earlier) - // - // Estimate size is second so that stages with very small cardinality (e.g. FunDep - // consequents with exactly 1 value) are run before multi-relation stages that happen - // to have a larger current estimate. - let key_fn = |join_stage: &JoinStage, - binding_info: &BindingInfo, - times_refined: &DenseIdMap| { - let refine = match join_stage { + let refinement_count = + |join_stage: &JoinStage, refinements: &DenseIdMap| match join_stage { JoinStage::Intersect { scans, .. } => scans .iter() - .map(|scan| times_refined.get(scan.atom).copied().unwrap_or_default()) + .map(|scan| refinements.get(scan.atom).copied().unwrap_or_default()) .max() .unwrap(), - JoinStage::FusedIntersect { cover, .. } => times_refined + JoinStage::FusedIntersect { cover, .. } => refinements .get(cover.to_index.atom) .copied() .unwrap_or_default(), JoinStage::FusedIntersectMat { bind, .. } => bind.len() as _, }; - ( - -refine, - estimate_size(join_stage, binding_info), - -num_intersected_rels(join_stage), - ) - }; for i in range.clone() { - let mut key_i = key_fn(&instrs[order.get(i)], binding_info, ×_refined); - for j in (i + 1)..range.end { - let key_j = key_fn(&instrs[order.get(j)], binding_info, ×_refined); - if key_j < key_i { - order.data.swap(i, j); - key_i = key_j; + let mut best = None; + for j in i..range.end { + let stage = &instrs[order.get(j)]; + let stable_refinement = refinement_count(stage, ×_refined); + let size = estimate_size(stage, binding_info); + let relations = num_intersected_rels(stage); + let key = if let Some(physical_refinements) = &physical_refinements { + ( + -refinement_count(stage, physical_refinements), + size, + -stable_refinement, + -relations, + ) + } else { + (-stable_refinement, size, i64::from(-relations), 0) + }; + match best { + Some((_, best_key)) if key >= best_key => {} + _ => best = Some((j, key)), } } - // Update the counts after a new instruction is selected. - match &instrs[order.get(i)] { - JoinStage::Intersect { scans, .. } => scans.iter().for_each(|scan| { - *times_refined.get_or_default(scan.atom) += 1; - }), - JoinStage::FusedIntersect { - cover, - to_intersect, - .. - } => { - *times_refined.get_or_default(cover.to_index.atom) += - cover.to_index.vars.len() as i64; + let (best_position, _) = best.unwrap(); + order.data.swap(i, best_position); - to_intersect.iter().for_each(|(spec, _)| { - *times_refined.get_or_default(spec.to_index.atom) += - spec.to_index.vars.len() as i64; - }); - } - JoinStage::FusedIntersectMat { to_intersect, .. } => { - to_intersect.iter().for_each(|(spec, _)| { - *times_refined.get_or_default(spec.to_index.atom) += - spec.to_index.vars.len() as i64; - }); - } + // Update the counts after a new instruction is selected. + let selected = &instrs[order.get(i)]; + update_refinements(selected, &mut times_refined); + if let Some(physical_refinements) = &mut physical_refinements { + update_refinements(selected, physical_refinements); } } } @@ -3652,9 +3657,123 @@ mod top_index_tests { thread, }; - use crate::{common::Value, numeric_id::NumericId, offsets::Subset}; + use smallvec::{SmallVec, smallvec}; + + use crate::{ + common::Value, + free_join::{ + AtomId, Variable, + plan::{JoinStage, MatId, MatScanMode, SingleScanSpec}, + }, + numeric_id::NumericId, + offsets::Subset, + table_spec::ColumnId, + }; + + use super::{ + BindingInfo, ChildLock, InstrOrder, get_or_insert_child, sort_plan_by_size_inner, + top_index_shape_is_eligible, + }; + + fn intersect_stage(atom: usize, column: usize) -> JoinStage { + JoinStage::Intersect { + var: Variable::from_usize(column), + scans: smallvec![SingleScanSpec { + atom: AtomId::from_usize(atom), + column: ColumnId::from_usize(column), + cs: Vec::new(), + }], + } + } + + fn intersect_stage_atoms(atoms: &[usize], column: usize) -> JoinStage { + JoinStage::Intersect { + var: Variable::from_usize(column), + scans: atoms + .iter() + .map(|&atom| SingleScanSpec { + atom: AtomId::from_usize(atom), + column: ColumnId::from_usize(column), + cs: Vec::new(), + }) + .collect(), + } + } + + fn mat_stage(mat_id: usize) -> JoinStage { + JoinStage::FusedIntersectMat { + cover: MatId::from_usize(mat_id), + mode: MatScanMode::KeyOnly, + bind: SmallVec::new(), + to_intersect: Vec::new(), + } + } - use super::{ChildLock, get_or_insert_child, top_index_shape_is_eligible}; + fn binding_info(sizes: &[(usize, usize)]) -> BindingInfo { + let mut binding_info = BindingInfo::default(); + for &(atom, size) in sizes { + binding_info.insert_subset( + AtomId::from_usize(atom), + Subset::Dense(crate::OffsetRange::new( + crate::RowId::from_usize(0), + crate::RowId::from_usize(size), + )), + ); + } + binding_info + } + + #[test] + fn mixed_recursive_dvo_keeps_the_plan_prefix_as_its_refinement_anchor() { + let stages = vec![ + intersect_stage(0, 0), + intersect_stage(1, 0), + intersect_stage(1, 1), + mat_stage(0), + ]; + let mut binding_info = binding_info(&[(0, 100), (1, 100)]); + + let mut order = InstrOrder::from_iter([1, 0, 2, 3].into_iter()); + sort_plan_by_size_inner(&mut order, 1..3, &stages, &mut binding_info); + + assert_eq!(order.data.as_slice(), &[1, 0, 2, 3]); + } + + #[test] + fn pure_intersect_dvo_prioritizes_physical_refinement_depth() { + let stages = vec![ + intersect_stage(1, 0), + intersect_stage(1, 1), + intersect_stage(0, 2), + intersect_stage(1, 3), + intersect_stage_atoms(&[0, 1], 4), + intersect_stage(0, 5), + ]; + let mut binding_info = binding_info(&[(0, 1_000), (1, 1)]); + + let mut order = InstrOrder::from_iter([4, 5, 2, 3, 0, 1].into_iter()); + sort_plan_by_size_inner(&mut order, 2..4, &stages, &mut binding_info); + + assert_eq!(order.data.as_slice(), &[4, 5, 2, 3, 0, 1]); + } + + #[test] + fn pure_intersect_dvo_uses_cardinality_at_equal_physical_depth() { + let stages = vec![ + intersect_stage(0, 0), + intersect_stage(0, 1), + intersect_stage(0, 2), + intersect_stage(1, 3), + intersect_stage_atoms(&[0, 1], 4), + intersect_stage(2, 5), + ]; + let mut binding_info = binding_info(&[(0, 50), (1, 10), (2, 100)]); + + let mut order = InstrOrder::from_iter([4, 5, 2, 3, 0, 1].into_iter()); + sort_plan_by_size_inner(&mut order, 2..4, &stages, &mut binding_info); + + assert_eq!(order.data.as_slice(), &[4, 5, 3, 2, 0, 1]); + } #[test] fn top_index_partitioning_rejects_serial_tiny_and_skewed_shapes() { diff --git a/core-relations/src/free_join/plan.rs b/core-relations/src/free_join/plan.rs index a001f375f..b1f0f78d5 100644 --- a/core-relations/src/free_join/plan.rs +++ b/core-relations/src/free_join/plan.rs @@ -1526,9 +1526,12 @@ fn get_next_freejoin_stage( /// Plan generic join queries (one variable per stage). /// -/// Variables are visited in their natural id order. Runtime `sort_plan_by_size` reorders stages -/// anyway, so static ordering only needs to be deterministic; [`fuse_single_scans`] collapses -/// any same-atom single-scans afterwards regardless of where they ended up. +/// Variables are visited in their natural id order, establishing the +/// deterministic canonical order used as a stable refinement anchor by +/// runtime `sort_plan_by_size`. Runtime ordering supplies live residual +/// estimates and, for all-`Intersect` blocks, physical-prefix feedback without +/// mutating that cached order. [`fuse_single_scans`] then collapses compatible +/// same-atom scans. fn plan_gj( ctx: &PlanningContext, state: &mut PlanningState, diff --git a/core-relations/src/table/mod.rs b/core-relations/src/table/mod.rs index 35d3242b1..3d0b36f5f 100644 --- a/core-relations/src/table/mod.rs +++ b/core-relations/src/table/mod.rs @@ -627,7 +627,7 @@ impl SortedWritesTable { let (actual_shard, hc) = hash_code(shard_data, to_remove, n_keys); assert_eq!(actual_shard, shard_id); if let Ok(entry) = shard.find_entry(hc, |entry| { - entry.hashcode == (hc as _) + entry.hashcode == (hc as HashCode) && &data.get_row(entry.row)[0..n_keys] == to_remove }) { let (ent, _) = entry.remove(); @@ -665,7 +665,7 @@ impl SortedWritesTable { let (actual_shard, hc) = hash_code(shard_data, to_remove, self.n_keys); assert_eq!(actual_shard, shard_id); if let Ok(entry) = shard.find_entry(hc, |entry| { - entry.hashcode == (hc as _) + entry.hashcode == (hc as HashCode) && &self.data.get_row(entry.row).unwrap()[0..self.n_keys] == to_remove }) {