diff --git a/src/expr/src/lib.rs b/src/expr/src/lib.rs index 44d5a6987d7f8..e8e266726f5ed 100644 --- a/src/expr/src/lib.rs +++ b/src/expr/src/lib.rs @@ -42,9 +42,9 @@ pub use relation::func::{ pub use relation::join_input_mapper::JoinInputMapper; pub use relation::{ AccessStrategy, AggregateExpr, CollectionPlan, ColumnOrder, JoinImplementation, - JoinInputCharacteristics, LetRecLimit, MirRelationExpr, RECURSION_LIMIT, RowComparator, - RowSetFinishing, RowSetFinishingIncremental, WindowFrame, WindowFrameBound, WindowFrameUnits, - canonicalize, compare_columns, non_nullable_columns, + JoinInputCharacteristics, JoinInputCharacteristicsVersion, LetRecLimit, MirRelationExpr, + RECURSION_LIMIT, RowComparator, RowSetFinishing, RowSetFinishingIncremental, WindowFrame, + WindowFrameBound, WindowFrameUnits, canonicalize, compare_columns, non_nullable_columns, }; pub use scalar::func::{self, BinaryFunc, UnaryFunc, UnmaterializableFunc, VariadicFunc}; pub use scalar::{ diff --git a/src/expr/src/relation.rs b/src/expr/src/relation.rs index 567388969d963..0646c71b328bf 100644 --- a/src/expr/src/relation.rs +++ b/src/expr/src/relation.rs @@ -3315,37 +3315,66 @@ pub enum JoinInputCharacteristics { V1(JoinInputCharacteristicsV1), /// Newer version, with `enable_join_prioritize_arranged` turned on. V2(JoinInputCharacteristicsV2), + /// Newest version, with `enable_join_reverse_edge_scoring` turned on. + V3(JoinInputCharacteristicsV3), +} + +/// Selects among the versions of [`JoinInputCharacteristics`]. +/// +/// Callers resolve the active feature flags into a version once, and then +/// construct every characteristic of a planning run with the same version. +/// Comparisons across versions are not meaningful. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum JoinInputCharacteristicsVersion { + /// All scoring flags off. + V1, + /// `enable_join_prioritize_arranged` on. + V2, + /// `enable_join_reverse_edge_scoring` on (subsumes V2). + V3, } impl JoinInputCharacteristics { /// Creates a new instance with the given characteristics. + /// + /// `prefix_unique` participates only in + /// [`JoinInputCharacteristicsVersion::V3`]. Earlier versions ignore it. pub fn new( + version: JoinInputCharacteristicsVersion, unique_key: bool, + prefix_unique: bool, key_length: usize, arranged: bool, cardinality: Option, filters: FilterCharacteristics, input: usize, - enable_join_prioritize_arranged: bool, ) -> Self { - if enable_join_prioritize_arranged { - Self::V2(JoinInputCharacteristicsV2::new( + match version { + JoinInputCharacteristicsVersion::V3 => Self::V3(JoinInputCharacteristicsV3::new( unique_key, + prefix_unique, key_length, arranged, cardinality, filters, input, - )) - } else { - Self::V1(JoinInputCharacteristicsV1::new( + )), + JoinInputCharacteristicsVersion::V2 => Self::V2(JoinInputCharacteristicsV2::new( + unique_key, + key_length, + arranged, + cardinality, + filters, + input, + )), + JoinInputCharacteristicsVersion::V1 => Self::V1(JoinInputCharacteristicsV1::new( unique_key, key_length, arranged, cardinality, filters, input, - )) + )), } } @@ -3354,6 +3383,7 @@ impl JoinInputCharacteristics { match self { Self::V1(jic) => jic.explain(), Self::V2(jic) => jic.explain(), + Self::V3(jic) => jic.explain(), } } @@ -3362,6 +3392,7 @@ impl JoinInputCharacteristics { match self { Self::V1(jic) => jic.arranged, Self::V2(jic) => jic.arranged, + Self::V3(jic) => jic.arranged, } } @@ -3370,7 +3401,102 @@ impl JoinInputCharacteristics { match self { Self::V1(jic) => &mut jic.filters, Self::V2(jic) => &mut jic.filters, + Self::V3(jic) => &mut jic.filters, + } + } +} + +/// Newest version of `JoinInputCharacteristics`, with +/// `enable_join_reverse_edge_scoring` turned on. +/// +/// Extends [`JoinInputCharacteristicsV2`] with `prefix_unique`, which credits +/// join order candidates whose equivalences are traversed "backwards": the +/// already-placed prefix is unique on the equated expressions, so each +/// incoming row matches at most one prefix row, and the extended result is +/// bounded by the incoming relation's size. +#[derive( + Eq, + PartialEq, + Ord, + PartialOrd, + Debug, + Clone, + Serialize, + Deserialize, + Hash, + MzReflect +)] +pub struct JoinInputCharacteristicsV3 { + /// An excellent indication that record count will not increase. + pub unique_key: bool, + /// The placed prefix is unique on the expressions equated with this + /// input's bound key, bounding the record count by this input's size. + /// Weaker than `unique_key` (which bounds by the prefix we carry), but + /// still a structural no-blowup guarantee. + pub prefix_unique: bool, + /// Cross joins are bad. + /// (`key_length > 0` also implies that it is not a cross join. However, we need to note cross + /// joins in a separate field, because not being a cross join is more important than `arranged`, + /// but otherwise `key_length` is less important than `arranged`.) + /// NOTE: a cross join against a prefix with at most one row can set + /// `prefix_unique`, which deliberately outranks this field. + pub not_cross: bool, + /// Indicates that there will be no additional in-memory footprint. + pub arranged: bool, + /// A weaker signal that record count will not increase. + pub key_length: usize, + /// Estimated cardinality (lower is better) + pub cardinality: Option>, + /// Characteristics of the filter that is applied at this input. + pub filters: FilterCharacteristics, + /// We want to prefer input earlier in the input list, for stability of ordering. + pub input: std::cmp::Reverse, +} + +impl JoinInputCharacteristicsV3 { + /// Creates a new instance with the given characteristics. + pub fn new( + unique_key: bool, + prefix_unique: bool, + key_length: usize, + arranged: bool, + cardinality: Option, + filters: FilterCharacteristics, + input: usize, + ) -> Self { + Self { + unique_key, + prefix_unique, + not_cross: key_length > 0, + arranged, + key_length, + cardinality: cardinality.map(std::cmp::Reverse), + filters, + input: std::cmp::Reverse(input), + } + } + + /// Turns the instance into a String to be printed in EXPLAIN. + pub fn explain(&self) -> String { + let mut e = "".to_owned(); + if self.unique_key { + e.push_str("U"); + } + if self.prefix_unique { + e.push_str("P"); + } + // Don't need to print `not_cross`, because that is visible in the printed key. + for _ in 0..self.key_length { + e.push_str("K"); + } + if self.arranged { + e.push_str("A"); } + if let Some(std::cmp::Reverse(cardinality)) = self.cardinality { + e.push_str(&format!("|{cardinality}|")); + } + e.push_str(&self.filters.explain()); + e } } diff --git a/src/repr/src/optimize.rs b/src/repr/src/optimize.rs index f919eb0027762..665a585869f1b 100644 --- a/src/repr/src/optimize.rs +++ b/src/repr/src/optimize.rs @@ -120,6 +120,8 @@ optimizer_feature_flags!({ // See the feature flag of the same name. enable_join_prioritize_arranged: bool, // See the feature flag of the same name. + enable_join_reverse_edge_scoring: bool, + // See the feature flag of the same name. enable_projection_pushdown_after_relation_cse: bool, // See the feature flag of the same name. enable_less_reduce_in_eqprop: bool, diff --git a/src/sql-lexer/src/keywords.txt b/src/sql-lexer/src/keywords.txt index 7a1d427d27d7b..4c1a1acba435f 100644 --- a/src/sql-lexer/src/keywords.txt +++ b/src/sql-lexer/src/keywords.txt @@ -156,6 +156,7 @@ Dot Double Drop Eager +Edge Element Else Enable @@ -408,6 +409,7 @@ Restrict Retain Return Returning +Reverse Revoke Right Role @@ -424,6 +426,7 @@ Schedule Schema Schemas Scope +Scoring Second Seconds Secret diff --git a/src/sql-parser/src/ast/defs/statement.rs b/src/sql-parser/src/ast/defs/statement.rs index 1e1d53a21d2f0..a48522caad279 100644 --- a/src/sql-parser/src/ast/defs/statement.rs +++ b/src/sql-parser/src/ast/defs/statement.rs @@ -4089,6 +4089,7 @@ pub enum ExplainPlanOptionName { EnableVariadicLeftJoinLowering, EnableLetrecFixpointAnalysis, EnableJoinPrioritizeArranged, + EnableJoinReverseEdgeScoring, EnableProjectionPushdownAfterRelationCse, } @@ -4126,6 +4127,7 @@ impl WithOptionName for ExplainPlanOptionName { | Self::EnableVariadicLeftJoinLowering | Self::EnableLetrecFixpointAnalysis | Self::EnableJoinPrioritizeArranged + | Self::EnableJoinReverseEdgeScoring | Self::EnableProjectionPushdownAfterRelationCse => false, } } diff --git a/src/sql/src/plan/statement/ddl.rs b/src/sql/src/plan/statement/ddl.rs index 5ffda71c9a48e..4e5a5fadeaa7c 100644 --- a/src/sql/src/plan/statement/ddl.rs +++ b/src/sql/src/plan/statement/ddl.rs @@ -4947,6 +4947,7 @@ pub fn unplan_create_cluster( enable_variadic_left_join_lowering, enable_letrec_fixpoint_analysis, enable_join_prioritize_arranged, + enable_join_reverse_edge_scoring: _, enable_projection_pushdown_after_relation_cse, enable_less_reduce_in_eqprop: _, enable_dequadratic_eqprop_map: _, diff --git a/src/sql/src/plan/statement/dml.rs b/src/sql/src/plan/statement/dml.rs index f86aea68c6bd3..8d31598c8fe5d 100644 --- a/src/sql/src/plan/statement/dml.rs +++ b/src/sql/src/plan/statement/dml.rs @@ -573,6 +573,7 @@ generate_extracted_config!( (EnableVariadicLeftJoinLowering, Option, Default(None)), (EnableLetrecFixpointAnalysis, Option, Default(None)), (EnableJoinPrioritizeArranged, Option, Default(None)), + (EnableJoinReverseEdgeScoring, Option, Default(None)), ( EnableProjectionPushdownAfterRelationCse, Option, @@ -627,6 +628,7 @@ impl TryFrom for ExplainConfig { persist_fast_path_limit: Default::default(), reoptimize_imported_views: v.reoptimize_imported_views, enable_join_prioritize_arranged: v.enable_join_prioritize_arranged, + enable_join_reverse_edge_scoring: v.enable_join_reverse_edge_scoring, enable_projection_pushdown_after_relation_cse: v .enable_projection_pushdown_after_relation_cse, enable_less_reduce_in_eqprop: Default::default(), diff --git a/src/sql/src/session/vars/definitions.rs b/src/sql/src/session/vars/definitions.rs index b0fa1ee4bc1fd..bb100b510ecd1 100644 --- a/src/sql/src/session/vars/definitions.rs +++ b/src/sql/src/session/vars/definitions.rs @@ -2200,6 +2200,13 @@ feature_flags!( enable_for_item_parsing: false, scope: ParameterScope::Cluster, }, + { + name: enable_join_reverse_edge_scoring, + desc: "Whether join planning should credit join order candidates on whose join keys the already-placed prefix is unique, bounding the result by the candidate's size.", + default: false, + enable_for_item_parsing: false, + scope: ParameterScope::Cluster, + }, { name: enable_projection_pushdown_after_relation_cse, desc: "Run ProjectionPushdown one more time after the last RelationCSE.", @@ -2309,6 +2316,7 @@ impl From<&super::SystemVars> for OptimizerFeatures { persist_fast_path_limit: vars.persist_fast_path_limit(), reoptimize_imported_views: false, enable_join_prioritize_arranged: vars.enable_join_prioritize_arranged(), + enable_join_reverse_edge_scoring: vars.enable_join_reverse_edge_scoring(), enable_projection_pushdown_after_relation_cse: vars .enable_projection_pushdown_after_relation_cse(), enable_less_reduce_in_eqprop: vars.enable_less_reduce_in_eqprop(), @@ -2353,6 +2361,7 @@ mod tests { persist_fast_path_limit, reoptimize_imported_views, enable_join_prioritize_arranged, + enable_join_reverse_edge_scoring, enable_projection_pushdown_after_relation_cse, enable_less_reduce_in_eqprop, enable_dequadratic_eqprop_map, @@ -2383,6 +2392,7 @@ mod tests { set_var!(persist_fast_path_limit); let _ = reoptimize_imported_views; // no corresponding var set_var!(enable_join_prioritize_arranged); + set_var!(enable_join_reverse_edge_scoring); set_var!(enable_projection_pushdown_after_relation_cse); set_var!(enable_less_reduce_in_eqprop); set_var!(enable_dequadratic_eqprop_map); diff --git a/src/transform/src/join_implementation.rs b/src/transform/src/join_implementation.rs index 480e0f3181f69..03df027bdd257 100644 --- a/src/transform/src/join_implementation.rs +++ b/src/transform/src/join_implementation.rs @@ -16,14 +16,14 @@ //! determining the orders of collections, lifting predicates if useful arrangements exist, //! and identifying opportunities to use indexes to replace filters. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use itertools::Itertools; use mz_expr::JoinImplementation::{Differential, IndexedFilter, Unimplemented}; use mz_expr::visit::{Visit, VisitChildren}; use mz_expr::{ - Columns, FilterCharacteristics, Id, JoinInputCharacteristics, JoinInputMapper, - MapFilterProject, MirRelationExpr, MirScalarExpr, RECURSION_LIMIT, + Columns, FilterCharacteristics, Id, JoinInputCharacteristics, JoinInputCharacteristicsVersion, + JoinInputMapper, MapFilterProject, MirRelationExpr, MirScalarExpr, RECURSION_LIMIT, }; use mz_ore::stack::{CheckedRecursion, RecursionGuard}; use mz_ore::{soft_assert_or_log, soft_panic_or_log}; @@ -375,15 +375,17 @@ impl JoinImplementation { } // Only plan a delta join if it's no new arrangements (beyond what differential planned). - if let Ok((delta_query_plan, 0)) = delta_queries::plan( - relation, - &input_mapper, + let delta_query_plan = optimize_orders( + equivalences, &available_arrangements, &unique_keys, &cardinalities, &filters, + &input_mapper, features, - ) { + ) + .and_then(|orders| delta_queries::plan(relation, &available_arrangements, orders)); + if let Ok((delta_query_plan, 0)) = delta_query_plan { tracing::debug!(plan = ?delta_query_plan, "replacing differential join with delta join"); *relation = delta_query_plan; } @@ -393,17 +395,25 @@ impl JoinImplementation { // To have reached here, we must be in our first run of join planning. // - // We plan a differential join first. - let (differential_query_plan, differential_new_arrangements) = differential::plan( - relation, - &input_mapper, + // We compute one order for each possible starting point, and both the + // differential and the delta planner choose from these. + // + // It is an invariant that the orders are in input order: the ith order begins with the ith input. + let orders = optimize_orders( + equivalences, &available_arrangements, &unique_keys, &cardinalities, &filters, + &input_mapper, features, ) - .expect("Failed to produce a differential join plan"); + .expect("Failed to produce join orders"); + + // We plan a differential join first. + let (differential_query_plan, differential_new_arrangements) = + differential::plan(relation, &available_arrangements, orders.clone()) + .expect("Failed to produce a differential join plan"); // Binary joins _must_ be differential. We won't plan a delta join. if num_inputs <= 2 { @@ -456,18 +466,7 @@ impl JoinImplementation { // ⨝ // // At the two internal joins, the differential join will need two new arrangements. - // - // TODO(mgree): with this refactoring, we should compute `orders` once---both joins - // call `optimize_orders` and we can save some work. - match delta_queries::plan( - relation, - &input_mapper, - &available_arrangements, - &unique_keys, - &cardinalities, - &filters, - features, - ) { + match delta_queries::plan(relation, &available_arrangements, orders) { // If delta plan's inputs need no new arrangements, pick the delta plan. Ok((delta_query_plan, 0)) => { soft_assert_or_log!( @@ -579,45 +578,29 @@ mod delta_queries { use std::collections::BTreeSet; - use mz_expr::{ - FilterCharacteristics, JoinImplementation, JoinInputMapper, MirRelationExpr, MirScalarExpr, - }; - use mz_repr::optimize::OptimizerFeatures; + use mz_expr::{JoinImplementation, JoinInputCharacteristics, MirRelationExpr, MirScalarExpr}; use crate::TransformError; /// Creates a delta query plan, and any predicates that need to be lifted. /// It also returns the number of new arrangements necessary for this plan. /// + /// `orders` are the per-starting-input orders from `optimize_orders`. + /// /// The method returns `Err` if any errors occur during planning. pub fn plan( join: &MirRelationExpr, - input_mapper: &JoinInputMapper, available: &[Vec>], - unique_keys: &[Vec>], - cardinalities: &[Option], - filters: &[FilterCharacteristics], - optimizer_features: &OptimizerFeatures, + orders: Vec, usize)>>, ) -> Result<(MirRelationExpr, usize), TransformError> { let mut new_join = join.clone(); if let MirRelationExpr::Join { inputs, - equivalences, + equivalences: _, implementation, } = &mut new_join { - // Determine a viable order for each relation, or return `Err` if none found. - let orders = super::optimize_orders( - equivalences, - available, - unique_keys, - cardinalities, - filters, - input_mapper, - optimizer_features.enable_join_prioritize_arranged, - )?; - // Count new arrangements. let new_arrangements: usize = orders .iter() @@ -671,51 +654,37 @@ mod delta_queries { mod differential { use std::collections::BTreeSet; - use mz_expr::{Columns, JoinImplementation, JoinInputMapper, MirRelationExpr, MirScalarExpr}; + use mz_expr::{ + Columns, JoinImplementation, JoinInputCharacteristics, MirRelationExpr, MirScalarExpr, + }; use mz_ore::soft_assert_eq_or_log; - use mz_repr::optimize::OptimizerFeatures; use crate::TransformError; use crate::join_implementation::FilterCharacteristics; /// Creates a linear differential plan, and any predicates that need to be lifted. /// It also returns the number of new arrangements necessary for this plan. + /// + /// `orders` are the per-starting-input orders from `optimize_orders`, and we + /// will choose one from these. + /// + /// We could change this preference at any point, but the list of orders should still inform. + /// Important, we should choose something stable under re-ordering, to converge under fixed + /// point iteration; we choose to start with the first input optimizing our criteria, which + /// should remain stable even when promoted to the first position. pub fn plan( join: &MirRelationExpr, - input_mapper: &JoinInputMapper, available: &[Vec>], - unique_keys: &[Vec>], - cardinalities: &[Option], - filters: &[FilterCharacteristics], - optimizer_features: &OptimizerFeatures, + mut orders: Vec, usize)>>, ) -> Result<(MirRelationExpr, usize), TransformError> { let mut new_join = join.clone(); if let MirRelationExpr::Join { inputs, - equivalences, + equivalences: _, implementation, } = &mut new_join { - // We compute one order for each possible starting point, and we will choose one from - // these. - // - // It is an invariant that the orders are in input order: the ith order begins with the ith input. - // - // We could change this preference at any point, but the list of orders should still inform. - // Important, we should choose something stable under re-ordering, to converge under fixed - // point iteration; we choose to start with the first input optimizing our criteria, which - // should remain stable even when promoted to the first position. - let mut orders = super::optimize_orders( - equivalences, - available, - unique_keys, - cardinalities, - filters, - input_mapper, - optimizer_features.enable_join_prioritize_arranged, - )?; - // Count new arrangements. // // We collect the count for each input, to be used to calculate `new_arrangements` below. @@ -767,7 +736,7 @@ mod differential { // worst `Characteristic`: we inspect the entire `Characteristic` vector of each // of these orders, and choose the best among these. This pushes bad stuff to // happen later, by which time we might have applied some filters. - .max_by_key(|o| o.clone()) + .max_by(|a, b| a.cmp(b)) .ok_or_else(|| { TransformError::Internal(String::from( "could not find max-min characteristics", @@ -989,6 +958,25 @@ fn install_lifted_mfp(new_join: &mut MirRelationExpr, mfp: MapFilterProject) { } } +/// Deduplicates prefix keys, drops keys that are supersets of other keys, +/// and truncates to [`MAX_PREFIX_KEYS`]. +/// +/// The sort makes the truncation deterministic, which order stability under +/// re-planning requires. +fn minimize_prefix_keys(keys: &mut Vec>) { + keys.sort_by(|a, b| (a.len(), a).cmp(&(b.len(), b))); + keys.dedup(); + let mut minimized: Vec> = Vec::new(); + for key in keys.drain(..) { + // Sorted by size, so any subset of `key` already kept suffices. + if !minimized.iter().any(|kept| kept.is_subset(&key)) { + minimized.push(key); + } + } + minimized.truncate(MAX_PREFIX_KEYS); + *keys = minimized; +} + /// Permute the keys in `order` to compensate for projections being lifted from inputs. /// `lifted_projections` has an optional projection for each input. fn permute_order( @@ -1014,8 +1002,15 @@ fn optimize_orders( cardinalities: &[Option], // cardinalities of input relations filters: &[FilterCharacteristics], // filter characteristics per input input_mapper: &JoinInputMapper, // join helper - enable_join_prioritize_arranged: bool, + optimizer_features: &OptimizerFeatures, ) -> Result, usize)>>, TransformError> { + let version = if optimizer_features.enable_join_reverse_edge_scoring { + JoinInputCharacteristicsVersion::V3 + } else if optimizer_features.enable_join_prioritize_arranged { + JoinInputCharacteristicsVersion::V2 + } else { + JoinInputCharacteristicsVersion::V1 + }; let mut orderer = Orderer::new( equivalences, available, @@ -1023,13 +1018,23 @@ fn optimize_orders( cardinalities, filters, input_mapper, - enable_join_prioritize_arranged, + version, ); (0..available.len()) .map(move |i| orderer.optimize_order_for(i)) .collect::, _>>() } +/// The maximum number of prefix unique keys tracked while ordering. +/// +/// Placements that are unique in neither direction pair every prefix key with +/// every key of the incoming relation, so the tracked set can grow +/// quadratically. Truncating only weakens the `prefix_unique` signal (fewer +/// uniqueness claims), it never overclaims. The bound is arbitrary, chosen +/// small because keys beyond the first few come from repeated non-unique +/// pairings and rarely get covered later. +const MAX_PREFIX_KEYS: usize = 8; + struct Orderer<'a> { inputs: usize, equivalences: &'a [Vec], @@ -1040,6 +1045,9 @@ struct Orderer<'a> { input_mapper: &'a JoinInputMapper, reverse_equivalences: Vec>, unique_arrangement: Vec>, + // A map from global expressions to the equivalence classes containing + // them, used to test prefix key coverage through the equivalences. + expr_classes: BTreeMap<&'a MirScalarExpr, Vec>, order: Vec<(JoinInputCharacteristics, Vec, usize)>, placed: Vec, @@ -1048,8 +1056,12 @@ struct Orderer<'a> { arrangement_active: Vec>, priority_queue: std::collections::BinaryHeap<(JoinInputCharacteristics, Vec, usize)>, + // Unique keys (as sets of global expressions) of the join of the placed + // inputs, `None` until the starting input is placed. Maintained only for + // `JoinInputCharacteristicsVersion::V3`, and always `None` otherwise. + prefix_keys: Option>>, - enable_join_prioritize_arranged: bool, + version: JoinInputCharacteristicsVersion, } impl<'a> Orderer<'a> { @@ -1060,7 +1072,7 @@ impl<'a> Orderer<'a> { cardinalities: &'a [Option], filters: &'a [FilterCharacteristics], input_mapper: &'a JoinInputMapper, - enable_join_prioritize_arranged: bool, + version: JoinInputCharacteristicsVersion, ) -> Self { let inputs = arrangements.len(); // A map from inputs to the equivalence classes in which they are referenced. @@ -1082,6 +1094,15 @@ impl<'a> Orderer<'a> { })); } } + let mut expr_classes = BTreeMap::new(); + for (index, equivalence) in equivalences.iter().enumerate() { + for expr in equivalence.iter() { + expr_classes + .entry(expr) + .or_insert_with(Vec::new) + .push(index); + } + } let order = Vec::with_capacity(inputs); let placed = vec![false; inputs]; @@ -1099,16 +1120,105 @@ impl<'a> Orderer<'a> { input_mapper, reverse_equivalences, unique_arrangement, + expr_classes, order, placed, bound, equivalences_active, arrangement_active, priority_queue, - enable_join_prioritize_arranged, + prefix_keys: None, + version, } } + /// Whether the placed prefix is unique on the expressions equated with + /// the given key of `rel`, which bounds the extended join result by + /// `rel`'s size (each row of `rel` matches at most one prefix row). + /// + /// Each key component reaches the prefix through the equivalence classes + /// containing it. A prefix key is covered when each of its elements + /// shares a class with some key component. An empty key covers exactly + /// the empty prefix key, i.e. a cross join against an at-most-one-row + /// prefix counts as prefix unique. + fn candidate_prefix_unique(&self, rel: usize, key: &[MirScalarExpr]) -> bool { + match &self.prefix_keys { + Some(prefix_keys) => self.prefix_covered(prefix_keys, rel, key), + None => false, + } + } + + fn prefix_covered( + &self, + prefix_keys: &[BTreeSet], + rel: usize, + key: &[MirScalarExpr], + ) -> bool { + let mut key_classes = BTreeSet::new(); + for k in key.iter() { + let global = self.input_mapper.map_expr_to_global(k.clone(), rel); + if let Some(classes) = self.expr_classes.get(&global) { + key_classes.extend(classes.iter().copied()); + } + } + prefix_keys.iter().any(|prefix_key| { + prefix_key.iter().all(|expr| { + self.expr_classes + .get(expr) + .is_some_and(|classes| classes.iter().any(|c| key_classes.contains(c))) + }) + }) + } + + /// Folds a newly placed input into `prefix_keys`, using the standard + /// join key inference: when the equated expressions cover a unique key + /// of the incoming relation the old prefix keys remain keys, when they + /// cover a prefix key the incoming relation's keys become keys, and + /// otherwise the pairwise unions are keys. The pairwise unions are what + /// lets `prefix_unique` compose along chains of placements. + fn update_prefix_keys(&mut self, input: usize) { + if !matches!(self.version, JoinInputCharacteristicsVersion::V3) { + return; + } + let rel_keys: Vec> = self.unique_keys[input] + .iter() + .map(|cols| { + cols.iter() + .map(|c| { + self.input_mapper + .map_expr_to_global(MirScalarExpr::column(*c), input) + }) + .collect() + }) + .collect(); + let Some(prefix_keys) = &self.prefix_keys else { + // `input` is the starting input, whose keys seed the prefix keys. + self.prefix_keys = Some(rel_keys); + return; + }; + let forward = self.unique_keys[input].iter().any(|cols| { + cols.iter() + .all(|c| self.bound[input].contains(&MirScalarExpr::column(*c))) + }); + let reverse = self.prefix_covered(prefix_keys, input, &self.bound[input]); + let mut new_keys = Vec::new(); + if forward { + new_keys.extend(prefix_keys.iter().cloned()); + } + if reverse { + new_keys.extend(rel_keys.iter().cloned()); + } + if !forward && !reverse { + for pk in prefix_keys.iter() { + for rk in rel_keys.iter() { + new_keys.push(pk.union(rk).cloned().collect()); + } + } + } + minimize_prefix_keys(&mut new_keys); + self.prefix_keys = Some(new_keys); + } + fn optimize_order_for( &mut self, start: usize, @@ -1123,6 +1233,7 @@ impl<'a> Orderer<'a> { for index in 0..self.equivalences.len() { self.equivalences_active[index] = false; } + self.prefix_keys = None; // Introduce cross joins as a possibility. for input in 0..self.inputs { @@ -1136,13 +1247,16 @@ impl<'a> Orderer<'a> { self.arrangement_active[input].push(pos); self.priority_queue.push(( JoinInputCharacteristics::new( + self.version, is_unique, + // No prefix has formed yet against which to claim + // uniqueness. + false, 0, true, cardinality, self.filters[input].clone(), input, - self.enable_join_prioritize_arranged, ), vec![], input, @@ -1150,13 +1264,14 @@ impl<'a> Orderer<'a> { } else { self.priority_queue.push(( JoinInputCharacteristics::new( + self.version, is_unique, + false, 0, false, cardinality, self.filters[input].clone(), input, - self.enable_join_prioritize_arranged, ), vec![], input, @@ -1186,13 +1301,17 @@ impl<'a> Orderer<'a> { // We start with some default values: let mut start_tuple = ( JoinInputCharacteristics::new( + self.version, false, + // The starting input is trivially bounded by itself. All + // starts get the same treatment, so this cannot skew the + // cross-order comparison. + true, 0, false, self.cardinalities[start], self.filters[start].clone(), start, - self.enable_join_prioritize_arranged, ), vec![], start, @@ -1223,13 +1342,14 @@ impl<'a> Orderer<'a> { .is_some(); start_tuple = ( JoinInputCharacteristics::new( + self.version, is_unique, + true, candidate_start_key.len(), arranged, cardinality, self.filters[start].clone(), start, - self.enable_join_prioritize_arranged, ), candidate_start_key, start, @@ -1264,6 +1384,9 @@ impl<'a> Orderer<'a> { /// keys are available to consider (both arranged, and unarranged). fn order_input(&mut self, input: usize) { self.placed[input] = true; + // Fold the newly placed input into the prefix keys before scoring + // new candidates against the extended prefix. + self.update_prefix_keys(input); for (equivalence, expr_index) in self.reverse_equivalences[input].iter() { if !self.equivalences_active[*equivalence] { // Placing `input` *may* activate the equivalence. Each of its columns @@ -1312,15 +1435,18 @@ impl<'a> Orderer<'a> { self.arrangement_active[rel].push(pos); // TODO: This could be pre-computed, as it is independent of the order. let is_unique = self.unique_arrangement[rel][pos]; + let prefix_unique = + self.candidate_prefix_unique(rel, key); self.priority_queue.push(( JoinInputCharacteristics::new( + self.version, is_unique, + prefix_unique, key.len(), true, self.cardinalities[rel], self.filters[rel].clone(), rel, - self.enable_join_prioritize_arranged, ), key.clone(), rel, @@ -1335,15 +1461,18 @@ impl<'a> Orderer<'a> { self.bound[rel].contains(&MirScalarExpr::column(*c)) }) }); + let prefix_unique = + self.candidate_prefix_unique(rel, &self.bound[rel]); self.priority_queue.push(( JoinInputCharacteristics::new( + self.version, is_unique, + prefix_unique, self.bound[rel].len(), false, self.cardinalities[rel], self.filters[rel].clone(), rel, - self.enable_join_prioritize_arranged, ), self.bound[rel].clone(), rel, diff --git a/test/sqllogictest/transform/join_reverse_edges.slt b/test/sqllogictest/transform/join_reverse_edges.slt new file mode 100644 index 0000000000000..a4404fe9badfc --- /dev/null +++ b/test/sqllogictest/transform/join_reverse_edges.slt @@ -0,0 +1,511 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +# Join orders over foreign-key shapes, where some good orders traverse +# key equivalences "backwards" (the unique side is already placed, the +# foreign-key side is being added). Such edges bound the join output by +# the size of the incoming relation, and these plans document how join +# ordering scores them. +# +# The dimension views use GROUP BY so that key inference derives a +# unique key, which is what `unique_keys` in join ordering consumes. + +mode cockroach + +# Chain: a.b_fk -> b.b_pk, b.c_fk -> c.c_pk. + +statement ok +CREATE TABLE a_raw (b_fk int NOT NULL, a_x int NOT NULL); + +statement ok +CREATE TABLE b_raw (b_pk int NOT NULL, c_fk int NOT NULL, b_x int NOT NULL); + +statement ok +CREATE TABLE c_raw (c_pk int NOT NULL, c_x int NOT NULL); + +statement ok +CREATE VIEW b AS SELECT b_pk, max(c_fk) AS c_fk, max(b_x) AS b_x FROM b_raw GROUP BY b_pk; + +statement ok +CREATE VIEW c AS SELECT c_pk, max(c_x) AS c_x FROM c_raw GROUP BY c_pk; + +# Chain, no indexes. Forward orders (a » b » c) use the dimensions' +# unique keys. Reverse orders are bounded but score as unknown. +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(join implementations, humanized expressions) AS VERBOSE TEXT FOR +SELECT a.a_x, b.b_x, c.c_x +FROM a_raw a, b, c +WHERE a.b_fk = b.b_pk AND b.c_fk = c.c_pk; +---- +Explained Query: + Project (#1{a_x}, #4{max_b_x}, #6{max_c_x}) + Join on=(#0{b_fk} = #2{b_pk} AND #3{max_c_fk} = #5{c_pk}) type=delta + implementation + %0:a_raw » %1[#0]UKA » %2[#0]UKA + %1 » %2[#0]UKA » %0:a_raw[#0{b_fk}]K + %2 » %1[#1{c_fk}]K » %0:a_raw[#0{b_fk}]K + ArrangeBy keys=[[#0{b_fk}]] + ReadStorage materialize.public.a_raw + ArrangeBy keys=[[#0{b_pk}], [#1{max_c_fk}]] + Reduce group_by=[#0{b_pk}] aggregates=[max(#1{c_fk}), max(#2{b_x})] + ReadStorage materialize.public.b_raw + ArrangeBy keys=[[#0{c_pk}]] + Reduce group_by=[#0{c_pk}] aggregates=[max(#1{c_x})] + ReadStorage materialize.public.c_raw + +Source materialize.public.a_raw +Source materialize.public.b_raw +Source materialize.public.c_raw + +Target cluster: quickstart + +EOF + +# Chain with a selective filter on the far end of the chain. Starting +# at c and walking backwards (c » b » a) is bounded at every step, and +# benefits from the filter first. +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(join implementations, humanized expressions) AS VERBOSE TEXT FOR +SELECT a.a_x, b.b_x, c.c_x +FROM a_raw a, b, c +WHERE a.b_fk = b.b_pk AND b.c_fk = c.c_pk AND c.c_x = 42; +---- +Explained Query: + Project (#1{a_x}, #4{max_b_x}, #6{max_c_x}) + Filter (#6{max_c_x} = 42) + Join on=(#0{b_fk} = #2{b_pk} AND #3{max_c_fk} = #5{c_pk}) type=delta + implementation + %0:a_raw » %1[#0]UKA » %2[#0]UKAef + %1 » %2[#0]UKAef » %0:a_raw[#0{b_fk}]K + %2 » %1[#1{c_fk}]K » %0:a_raw[#0{b_fk}]K + ArrangeBy keys=[[#0{b_fk}]] + ReadStorage materialize.public.a_raw + ArrangeBy keys=[[#0{b_pk}], [#1{max_c_fk}]] + Reduce group_by=[#0{b_pk}] aggregates=[max(#1{c_fk}), max(#2{b_x})] + ReadStorage materialize.public.b_raw + ArrangeBy keys=[[#0{c_pk}]] + Reduce group_by=[#0{c_pk}] aggregates=[max(#1{c_x})] + ReadStorage materialize.public.c_raw + +Source materialize.public.a_raw +Source materialize.public.b_raw +Source materialize.public.c_raw + +Target cluster: quickstart + +EOF + +# Index everything, so that a delta join is available. Delta joins need +# an order starting from every input, so the path starting at c must +# traverse both edges backwards. + +statement ok +CREATE INDEX a_raw_bfk_idx ON a_raw (b_fk); + +statement ok +CREATE INDEX b_bpk_idx ON b (b_pk); + +statement ok +CREATE INDEX b_cfk_idx ON b (c_fk); + +statement ok +CREATE INDEX c_cpk_idx ON c (c_pk); + +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(join implementations, humanized expressions) AS VERBOSE TEXT FOR +SELECT a.a_x, b.b_x, c.c_x +FROM a_raw a, b, c +WHERE a.b_fk = b.b_pk AND b.c_fk = c.c_pk; +---- +Explained Query: + Project (#1{a_x}, #4{b_x}, #6{c_x}) + Join on=(#0{b_fk} = #2{b_pk} AND #3{c_fk} = #5{c_pk}) type=delta + implementation + %0:a_raw » %1:b[#0{b_pk}]UKA » %2:c[#0{c_pk}]UKA + %1:b » %2:c[#0{c_pk}]UKA » %0:a_raw[#0{b_fk}]KA + %2:c » %1:b[#1{c_fk}]KA » %0:a_raw[#0{b_fk}]KA + ArrangeBy keys=[[#0{b_fk}]] + ReadIndex on=a_raw a_raw_bfk_idx=[delta join 1st input (full scan)] + ArrangeBy keys=[[#0{b_pk}], [#1{c_fk}]] + ReadIndex on=b b_bpk_idx=[delta join lookup] b_cfk_idx=[delta join lookup] + ArrangeBy keys=[[#0{c_pk}]] + ReadIndex on=c c_cpk_idx=[delta join lookup] + +Used Indexes: + - materialize.public.a_raw_bfk_idx (delta join 1st input (full scan)) + - materialize.public.b_bpk_idx (delta join lookup) + - materialize.public.b_cfk_idx (delta join lookup) + - materialize.public.c_cpk_idx (delta join lookup) + +Target cluster: quickstart + +EOF + +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(join implementations, humanized expressions) AS VERBOSE TEXT FOR +SELECT a.a_x, b.b_x, c.c_x +FROM a_raw a, b, c +WHERE a.b_fk = b.b_pk AND b.c_fk = c.c_pk AND c.c_x = 42; +---- +Explained Query: + Project (#1{a_x}, #4{b_x}, #6{c_x}) + Filter (#6{c_x} = 42) + Join on=(#0{b_fk} = #2{b_pk} AND #3{c_fk} = #5{c_pk}) type=delta + implementation + %0:a_raw » %1:b[#0{b_pk}]UKA » %2:c[#0{c_pk}]UKAef + %1:b » %2:c[#0{c_pk}]UKAef » %0:a_raw[#0{b_fk}]KA + %2:c » %1:b[#1{c_fk}]KA » %0:a_raw[#0{b_fk}]KA + ArrangeBy keys=[[#0{b_fk}]] + ReadIndex on=a_raw a_raw_bfk_idx=[delta join 1st input (full scan)] + ArrangeBy keys=[[#0{b_pk}], [#1{c_fk}]] + ReadIndex on=b b_bpk_idx=[delta join lookup] b_cfk_idx=[delta join lookup] + ArrangeBy keys=[[#0{c_pk}]] + ReadIndex on=c c_cpk_idx=[delta join lookup] + +Used Indexes: + - materialize.public.a_raw_bfk_idx (delta join 1st input (full scan)) + - materialize.public.b_bpk_idx (delta join lookup) + - materialize.public.b_cfk_idx (delta join lookup) + - materialize.public.c_cpk_idx (delta join lookup) + +Target cluster: quickstart + +EOF + +# Star: fact f references two keyed dimensions. The delta path starting +# at a dimension enters the fact through a reverse edge (bounded by +# |f|), then proceeds forward into the other dimension. + +statement ok +CREATE TABLE f_raw (d1_fk int NOT NULL, d2_fk int NOT NULL, f_x int NOT NULL); + +statement ok +CREATE TABLE d1_raw (d1_pk int NOT NULL, d1_x int NOT NULL); + +statement ok +CREATE TABLE d2_raw (d2_pk int NOT NULL, d2_x int NOT NULL); + +statement ok +CREATE VIEW d1 AS SELECT d1_pk, max(d1_x) AS d1_x FROM d1_raw GROUP BY d1_pk; + +statement ok +CREATE VIEW d2 AS SELECT d2_pk, max(d2_x) AS d2_x FROM d2_raw GROUP BY d2_pk; + +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(join implementations, humanized expressions) AS VERBOSE TEXT FOR +SELECT f.f_x, d1.d1_x, d2.d2_x +FROM f_raw f, d1, d2 +WHERE f.d1_fk = d1.d1_pk AND f.d2_fk = d2.d2_pk; +---- +Explained Query: + Project (#2{f_x}, #4{max_d1_x}, #6{max_d2_x}) + Join on=(#0{d1_fk} = #3{d1_pk} AND #1{d2_fk} = #5{d2_pk}) type=delta + implementation + %0:f_raw » %1[#0]UKA » %2[#0]UKA + %1 » %0:f_raw[#0{d1_fk}]K » %2[#0]UKA + %2 » %0:f_raw[#1{d2_fk}]K » %1[#0]UKA + ArrangeBy keys=[[#0{d1_fk}], [#1{d2_fk}]] + ReadStorage materialize.public.f_raw + ArrangeBy keys=[[#0{d1_pk}]] + Reduce group_by=[#0{d1_pk}] aggregates=[max(#1{d1_x})] + ReadStorage materialize.public.d1_raw + ArrangeBy keys=[[#0{d2_pk}]] + Reduce group_by=[#0{d2_pk}] aggregates=[max(#1{d2_x})] + ReadStorage materialize.public.d2_raw + +Source materialize.public.f_raw +Source materialize.public.d1_raw +Source materialize.public.d2_raw + +Target cluster: quickstart + +EOF + +statement ok +CREATE INDEX f_raw_d1_idx ON f_raw (d1_fk); + +statement ok +CREATE INDEX f_raw_d2_idx ON f_raw (d2_fk); + +statement ok +CREATE INDEX d1_pk_idx ON d1 (d1_pk); + +statement ok +CREATE INDEX d2_pk_idx ON d2 (d2_pk); + +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(join implementations, humanized expressions) AS VERBOSE TEXT FOR +SELECT f.f_x, d1.d1_x, d2.d2_x +FROM f_raw f, d1, d2 +WHERE f.d1_fk = d1.d1_pk AND f.d2_fk = d2.d2_pk; +---- +Explained Query: + Project (#2{f_x}, #4{d1_x}, #6{d2_x}) + Join on=(#0{d1_fk} = #3{d1_pk} AND #1{d2_fk} = #5{d2_pk}) type=delta + implementation + %0:f_raw » %1:d1[#0{d1_pk}]UKA » %2:d2[#0{d2_pk}]UKA + %1:d1 » %0:f_raw[#0{d1_fk}]KA » %2:d2[#0{d2_pk}]UKA + %2:d2 » %0:f_raw[#1{d2_fk}]KA » %1:d1[#0{d1_pk}]UKA + ArrangeBy keys=[[#0{d1_fk}], [#1{d2_fk}]] + ReadIndex on=f_raw f_raw_d1_idx=[delta join 1st input (full scan)] f_raw_d2_idx=[delta join lookup] + ArrangeBy keys=[[#0{d1_pk}]] + ReadIndex on=d1 d1_pk_idx=[delta join lookup] + ArrangeBy keys=[[#0{d2_pk}]] + ReadIndex on=d2 d2_pk_idx=[delta join lookup] + +Used Indexes: + - materialize.public.f_raw_d1_idx (delta join 1st input (full scan)) + - materialize.public.f_raw_d2_idx (delta join lookup) + - materialize.public.d1_pk_idx (delta join lookup) + - materialize.public.d2_pk_idx (delta join lookup) + +Target cluster: quickstart + +EOF + +# Star with a selective dimension filter. If reverse edges were scored +# as bounded, d2 » f » d1 would be an attractive differential order. +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(join implementations, humanized expressions) AS VERBOSE TEXT FOR +SELECT f.f_x, d1.d1_x, d2.d2_x +FROM f_raw f, d1, d2 +WHERE f.d1_fk = d1.d1_pk AND f.d2_fk = d2.d2_pk AND d2.d2_x = 7; +---- +Explained Query: + Project (#2{f_x}, #4{d1_x}, #6{d2_x}) + Filter (#6{d2_x} = 7) + Join on=(#0{d1_fk} = #3{d1_pk} AND #1{d2_fk} = #5{d2_pk}) type=delta + implementation + %0:f_raw » %2:d2[#0{d2_pk}]UKAef » %1:d1[#0{d1_pk}]UKA + %1:d1 » %0:f_raw[#0{d1_fk}]KA » %2:d2[#0{d2_pk}]UKAef + %2:d2 » %0:f_raw[#1{d2_fk}]KA » %1:d1[#0{d1_pk}]UKA + ArrangeBy keys=[[#0{d1_fk}], [#1{d2_fk}]] + ReadIndex on=f_raw f_raw_d1_idx=[delta join 1st input (full scan)] f_raw_d2_idx=[delta join lookup] + ArrangeBy keys=[[#0{d1_pk}]] + ReadIndex on=d1 d1_pk_idx=[delta join lookup] + ArrangeBy keys=[[#0{d2_pk}]] + ReadIndex on=d2 d2_pk_idx=[delta join lookup] + +Used Indexes: + - materialize.public.f_raw_d1_idx (delta join 1st input (full scan)) + - materialize.public.f_raw_d2_idx (delta join lookup) + - materialize.public.d1_pk_idx (delta join lookup) + - materialize.public.d2_pk_idx (delta join lookup) + +Target cluster: quickstart + +EOF + +# Snowflake: fact -> dimension -> sub-dimension, with the selective +# filter on the sub-dimension. The bounded reverse walk is +# s » d » f, two backwards edges after a filtered start. + +statement ok +CREATE TABLE s_raw (s_pk int NOT NULL, s_x int NOT NULL); + +statement ok +CREATE VIEW s AS SELECT s_pk, max(s_x) AS s_x FROM s_raw GROUP BY s_pk; + +statement ok +CREATE TABLE d_raw (d_pk int NOT NULL, s_fk int NOT NULL, d_x int NOT NULL); + +statement ok +CREATE VIEW d AS SELECT d_pk, max(s_fk) AS s_fk, max(d_x) AS d_x FROM d_raw GROUP BY d_pk; + +# The same shapes with reverse edge scoring enabled. Bounded reverse +# edges now carry a `P` (prefix unique) mark, and order choices may +# shift toward bounded walks. + +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(join implementations, humanized expressions, enable join reverse edge scoring) AS VERBOSE TEXT FOR +SELECT a.a_x, b.b_x, c.c_x +FROM a_raw a, b, c +WHERE a.b_fk = b.b_pk AND b.c_fk = c.c_pk; +---- +Explained Query: + Project (#1{a_x}, #4{b_x}, #6{c_x}) + Join on=(#0{b_fk} = #2{b_pk} AND #3{c_fk} = #5{c_pk}) type=delta + implementation + %0:a_raw » %1:b[#0{b_pk}]UKA » %2:c[#0{c_pk}]UKA + %1:b » %2:c[#0{c_pk}]UKA » %0:a_raw[#0{b_fk}]PKA + %2:c » %1:b[#1{c_fk}]PKA » %0:a_raw[#0{b_fk}]PKA + ArrangeBy keys=[[#0{b_fk}]] + ReadIndex on=a_raw a_raw_bfk_idx=[delta join 1st input (full scan)] + ArrangeBy keys=[[#0{b_pk}], [#1{c_fk}]] + ReadIndex on=b b_bpk_idx=[delta join lookup] b_cfk_idx=[delta join lookup] + ArrangeBy keys=[[#0{c_pk}]] + ReadIndex on=c c_cpk_idx=[delta join lookup] + +Used Indexes: + - materialize.public.a_raw_bfk_idx (delta join 1st input (full scan)) + - materialize.public.b_bpk_idx (delta join lookup) + - materialize.public.b_cfk_idx (delta join lookup) + - materialize.public.c_cpk_idx (delta join lookup) + +Target cluster: quickstart + +EOF + +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(join implementations, humanized expressions, enable join reverse edge scoring) AS VERBOSE TEXT FOR +SELECT a.a_x, b.b_x, c.c_x +FROM a_raw a, b, c +WHERE a.b_fk = b.b_pk AND b.c_fk = c.c_pk AND c.c_x = 42; +---- +Explained Query: + Project (#1{a_x}, #4{b_x}, #6{c_x}) + Filter (#6{c_x} = 42) + Join on=(#0{b_fk} = #2{b_pk} AND #3{c_fk} = #5{c_pk}) type=delta + implementation + %0:a_raw » %1:b[#0{b_pk}]UKA » %2:c[#0{c_pk}]UKAef + %1:b » %2:c[#0{c_pk}]UKAef » %0:a_raw[#0{b_fk}]PKA + %2:c » %1:b[#1{c_fk}]PKA » %0:a_raw[#0{b_fk}]PKA + ArrangeBy keys=[[#0{b_fk}]] + ReadIndex on=a_raw a_raw_bfk_idx=[delta join 1st input (full scan)] + ArrangeBy keys=[[#0{b_pk}], [#1{c_fk}]] + ReadIndex on=b b_bpk_idx=[delta join lookup] b_cfk_idx=[delta join lookup] + ArrangeBy keys=[[#0{c_pk}]] + ReadIndex on=c c_cpk_idx=[delta join lookup] + +Used Indexes: + - materialize.public.a_raw_bfk_idx (delta join 1st input (full scan)) + - materialize.public.b_bpk_idx (delta join lookup) + - materialize.public.b_cfk_idx (delta join lookup) + - materialize.public.c_cpk_idx (delta join lookup) + +Target cluster: quickstart + +EOF + +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(join implementations, humanized expressions, enable join reverse edge scoring) AS VERBOSE TEXT FOR +SELECT f.f_x, d1.d1_x, d2.d2_x +FROM f_raw f, d1, d2 +WHERE f.d1_fk = d1.d1_pk AND f.d2_fk = d2.d2_pk; +---- +Explained Query: + Project (#2{f_x}, #4{d1_x}, #6{d2_x}) + Join on=(#0{d1_fk} = #3{d1_pk} AND #1{d2_fk} = #5{d2_pk}) type=delta + implementation + %0:f_raw » %1:d1[#0{d1_pk}]UKA » %2:d2[#0{d2_pk}]UKA + %1:d1 » %0:f_raw[#0{d1_fk}]PKA » %2:d2[#0{d2_pk}]UKA + %2:d2 » %0:f_raw[#1{d2_fk}]PKA » %1:d1[#0{d1_pk}]UKA + ArrangeBy keys=[[#0{d1_fk}], [#1{d2_fk}]] + ReadIndex on=f_raw f_raw_d1_idx=[delta join 1st input (full scan)] f_raw_d2_idx=[delta join lookup] + ArrangeBy keys=[[#0{d1_pk}]] + ReadIndex on=d1 d1_pk_idx=[delta join lookup] + ArrangeBy keys=[[#0{d2_pk}]] + ReadIndex on=d2 d2_pk_idx=[delta join lookup] + +Used Indexes: + - materialize.public.f_raw_d1_idx (delta join 1st input (full scan)) + - materialize.public.f_raw_d2_idx (delta join lookup) + - materialize.public.d1_pk_idx (delta join lookup) + - materialize.public.d2_pk_idx (delta join lookup) + +Target cluster: quickstart + +EOF + +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(join implementations, humanized expressions, enable join reverse edge scoring) AS VERBOSE TEXT FOR +SELECT f.f_x, d1.d1_x, d2.d2_x +FROM f_raw f, d1, d2 +WHERE f.d1_fk = d1.d1_pk AND f.d2_fk = d2.d2_pk AND d2.d2_x = 7; +---- +Explained Query: + Project (#2{f_x}, #4{d1_x}, #6{d2_x}) + Filter (#6{d2_x} = 7) + Join on=(#0{d1_fk} = #3{d1_pk} AND #1{d2_fk} = #5{d2_pk}) type=delta + implementation + %0:f_raw » %2:d2[#0{d2_pk}]UKAef » %1:d1[#0{d1_pk}]UKA + %1:d1 » %0:f_raw[#0{d1_fk}]PKA » %2:d2[#0{d2_pk}]UKAef + %2:d2 » %0:f_raw[#1{d2_fk}]PKA » %1:d1[#0{d1_pk}]UKA + ArrangeBy keys=[[#0{d1_fk}], [#1{d2_fk}]] + ReadIndex on=f_raw f_raw_d1_idx=[delta join 1st input (full scan)] f_raw_d2_idx=[delta join lookup] + ArrangeBy keys=[[#0{d1_pk}]] + ReadIndex on=d1 d1_pk_idx=[delta join lookup] + ArrangeBy keys=[[#0{d2_pk}]] + ReadIndex on=d2 d2_pk_idx=[delta join lookup] + +Used Indexes: + - materialize.public.f_raw_d1_idx (delta join 1st input (full scan)) + - materialize.public.f_raw_d2_idx (delta join lookup) + - materialize.public.d1_pk_idx (delta join lookup) + - materialize.public.d2_pk_idx (delta join lookup) + +Target cluster: quickstart + +EOF + +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(join implementations, humanized expressions) AS VERBOSE TEXT FOR +SELECT f.f_x, d.d_x, s.s_x +FROM f_raw f, d, s +WHERE f.d1_fk = d.d_pk AND d.s_fk = s.s_pk AND s.s_x = 11; +---- +Explained Query: + Project (#2{f_x}, #5{max_d_x}, #7{max_s_x}) + Filter (#7{max_s_x} = 11) + Join on=(#0{d1_fk} = #3{d_pk} AND #4{max_s_fk} = #6{s_pk}) type=delta + implementation + %0:f_raw » %1[#0]UKA » %2[#0]UKAef + %1 » %2[#0]UKAef » %0:f_raw[#0{d1_fk}]KA + %2 » %1[#1{s_fk}]K » %0:f_raw[#0{d1_fk}]KA + ArrangeBy keys=[[#0{d1_fk}]] + ReadIndex on=f_raw f_raw_d1_idx=[delta join 1st input (full scan)] + ArrangeBy keys=[[#0{d_pk}], [#1{max_s_fk}]] + Reduce group_by=[#0{d_pk}] aggregates=[max(#1{s_fk}), max(#2{d_x})] + ReadStorage materialize.public.d_raw + ArrangeBy keys=[[#0{s_pk}]] + Reduce group_by=[#0{s_pk}] aggregates=[max(#1{s_x})] + ReadStorage materialize.public.s_raw + +Source materialize.public.s_raw +Source materialize.public.d_raw + +Used Indexes: + - materialize.public.f_raw_d1_idx (delta join 1st input (full scan)) + +Target cluster: quickstart + +EOF + +query T multiline +EXPLAIN OPTIMIZED PLAN WITH(join implementations, humanized expressions, enable join reverse edge scoring) AS VERBOSE TEXT FOR +SELECT f.f_x, d.d_x, s.s_x +FROM f_raw f, d, s +WHERE f.d1_fk = d.d_pk AND d.s_fk = s.s_pk AND s.s_x = 11; +---- +Explained Query: + Project (#2{f_x}, #5{max_d_x}, #7{max_s_x}) + Filter (#7{max_s_x} = 11) + Join on=(#0{d1_fk} = #3{d_pk} AND #4{max_s_fk} = #6{s_pk}) type=delta + implementation + %0:f_raw » %1[#0]UKA » %2[#0]UKAef + %1 » %2[#0]UKAef » %0:f_raw[#0{d1_fk}]PKA + %2 » %1[#1{s_fk}]PK » %0:f_raw[#0{d1_fk}]PKA + ArrangeBy keys=[[#0{d1_fk}]] + ReadIndex on=f_raw f_raw_d1_idx=[delta join 1st input (full scan)] + ArrangeBy keys=[[#0{d_pk}], [#1{max_s_fk}]] + Reduce group_by=[#0{d_pk}] aggregates=[max(#1{s_fk}), max(#2{d_x})] + ReadStorage materialize.public.d_raw + ArrangeBy keys=[[#0{s_pk}]] + Reduce group_by=[#0{s_pk}] aggregates=[max(#1{s_x})] + ReadStorage materialize.public.s_raw + +Source materialize.public.s_raw +Source materialize.public.d_raw + +Used Indexes: + - materialize.public.f_raw_d1_idx (delta join 1st input (full scan)) + +Target cluster: quickstart + +EOF