optimizer: add prefix-unique join order scoring behind a flag

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::{

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<usize>,
         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<std::cmp::Reverse<usize>>,
+    /// 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<usize>,
+}
+
+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<usize>,
+        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
     }
 }
 

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,

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

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,
         }
     }

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: _,

src/sql/src/plan/statement/dml.rs

@@ -573,6 +573,7 @@ generate_extracted_config!(
     (EnableVariadicLeftJoinLowering, Option<bool>, Default(None)),
     (EnableLetrecFixpointAnalysis, Option<bool>, Default(None)),
     (EnableJoinPrioritizeArranged, Option<bool>, Default(None)),
+    (EnableJoinReverseEdgeScoring, Option<bool>, Default(None)),
     (
         EnableProjectionPushdownAfterRelationCse,
         Option<bool>,
@@ -627,6 +628,7 @@ impl TryFrom<ExplainPlanOptionExtracted> 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(),

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);