diff --git a/src/common/statistics/src/stat_estimate.rs b/src/common/statistics/src/stat_estimate.rs index 83c3809457dce..35b8feb00cfa5 100644 --- a/src/common/statistics/src/stat_estimate.rs +++ b/src/common/statistics/src/stat_estimate.rs @@ -435,6 +435,18 @@ impl StatCount { } } } + + pub fn sum(left: Self, right: Self) -> Self { + match (left, right) { + (StatCount::Exact(left), StatCount::Exact(right)) => { + StatCount::exact(left.saturating_add(right)) + } + _ => StatCount::estimate( + left.expected() + right.expected(), + left.upper() + right.upper(), + ), + } + } } #[cfg(test)] diff --git a/src/query/expression/src/stat_evaluator.rs b/src/query/expression/src/stat_evaluator.rs index 30e297eb98d2e..714d791df8521 100644 --- a/src/query/expression/src/stat_evaluator.rs +++ b/src/query/expression/src/stat_evaluator.rs @@ -18,7 +18,10 @@ use std::collections::HashMap; use databend_common_exception::ErrorCode; use databend_common_exception::Result; +use super::Cast; use super::ColumnIndex; +use super::ColumnRef; +use super::ConstantFolder; use super::Expr; use super::FunctionCall; use super::FunctionContext; @@ -26,6 +29,7 @@ use super::FunctionDomain; use super::FunctionEval; use super::FunctionRegistry; use super::Scalar; +use super::conversion::classify_conversion; use super::function_stat::DeriveStat; use super::stat_distribution::ArgStat; use super::stat_distribution::BorrowedDistribution; @@ -93,10 +97,70 @@ impl<'a> StatEvaluator<'a> { Expr::FunctionCall(call) => Ok(self .eval_function_call(call, input_stats)? .map(CowStat::Owned)), - Expr::Cast(_) | Expr::LambdaFunctionCall(_) => Ok(None), + Expr::Cast(cast) => Ok(self.eval_cast(cast, input_stats)?.map(CowStat::Owned)), + Expr::LambdaFunctionCall(_) => Ok(None), } } + fn eval_cast<'s, I: ColumnIndex>( + &'a self, + cast: &Cast, + input_stats: &'s HashMap>, + ) -> Result> { + let src_type = cast.expr.data_type(); + if cast.is_try || !classify_conversion(src_type, &cast.dest_type).is_lossless_injective() { + return Ok(None); + } + + let Some(input) = self.eval(&cast.expr, input_stats)? else { + return Ok(None); + }; + let input = input.as_ref(); + + // Reuse the cast domain implementation without making statistics + // evaluation depend on the physical value evaluator. The synthetic + // column represents the already-derived statistics of the inner expr. + let expr = Expr::Cast(Cast { + span: cast.span, + is_try: false, + expr: Box::new(Expr::ColumnRef(ColumnRef { + span: cast.span, + id: 0, + data_type: src_type.clone(), + display_name: String::new(), + })), + dest_type: cast.dest_type.clone(), + }); + let input_domains = HashMap::from([(0, input.domain.clone())]); + let (_, Some(domain)) = ConstantFolder::fold_with_domain( + &expr, + &input_domains, + self.func_ctx, + self.fn_registry, + ) else { + return Ok(None); + }; + + let stat = ReturnStat { + domain, + // A lossless injective cast preserves distinct values and NULLs. + // Histograms are typed, so their boundaries cannot be reused. + ndv: input.ndv, + null_count: input.null_count, + distribution: OwnedDistribution::Unknown, + }; + if let Err(msg) = stat.check_consistency_with_type(Some(&cast.dest_type)) { + if cfg!(debug_assertions) { + return Err(ErrorCode::Internal(format!( + "Failed to derive statistics for cast: {msg}" + ))); + } + log::warn!(msg; "Derived invalid cast statistics"); + return Ok(None); + } + Ok(Some(stat)) + } + fn eval_function_call<'s, I: ColumnIndex>( &'a self, call: &FunctionCall, @@ -219,6 +283,8 @@ mod tests { use super::*; use crate::types::DataType; + use crate::types::NumberDataType; + use crate::types::number::NumberScalar; #[test] fn test_constant_null_uses_exact_input_cardinality() { @@ -237,4 +303,75 @@ mod tests { assert_eq!(stat.null_count, StatCount::exact(7)); } + + #[test] + fn test_lossless_cast_preserves_basic_statistics() { + let src_type = DataType::Number(NumberDataType::UInt8); + let dest_type = src_type.clone().wrap_nullable(); + let expr = Expr::Cast(Cast { + span: None, + is_try: false, + expr: Box::new(Expr::ColumnRef(ColumnRef { + span: None, + id: 0, + data_type: src_type.clone(), + display_name: "c0".to_string(), + })), + dest_type: dest_type.clone(), + }); + let domain = crate::Domain::from_min_max( + Scalar::Number(NumberScalar::UInt8(1)), + Scalar::Number(NumberScalar::UInt8(3)), + &src_type, + ); + let input_stats = HashMap::from([(0, ArgStat { + domain, + ndv: NdvEstimate::exact(3.0), + null_count: StatCount::exact(0), + distribution: BorrowedDistribution::Unknown, + })]); + + let stat = StatEvaluator::run( + &expr, + &FunctionContext::default(), + &FunctionRegistry::empty(), + StatCardinality::exact(3), + &input_stats, + ) + .unwrap() + .unwrap() + .into_owned(); + + assert!(stat.domain.matches_data_type(&dest_type)); + assert_eq!(stat.ndv, NdvEstimate::exact(3.0)); + assert_eq!(stat.null_count, StatCount::exact(0)); + assert!(matches!(stat.distribution, OwnedDistribution::Unknown)); + } + + #[test] + fn test_lossy_cast_is_not_derived() { + let expr = Expr::Cast(Cast { + span: None, + is_try: false, + expr: Box::new(Expr::ColumnRef(ColumnRef { + span: None, + id: 0, + data_type: DataType::Number(NumberDataType::Int64), + display_name: "c0".to_string(), + })), + dest_type: DataType::Number(NumberDataType::UInt8), + }); + + assert!( + StatEvaluator::run( + &expr, + &FunctionContext::default(), + &FunctionRegistry::empty(), + StatCardinality::exact(3), + &HashMap::new(), + ) + .unwrap() + .is_none() + ); + } } diff --git a/src/query/sql/src/planner/optimizer/ir/mod.rs b/src/query/sql/src/planner/optimizer/ir/mod.rs index 5a653cbd2780a..a9439533b5f6c 100644 --- a/src/query/sql/src/planner/optimizer/ir/mod.rs +++ b/src/query/sql/src/planner/optimizer/ir/mod.rs @@ -52,3 +52,4 @@ pub use stats::MAX_SELECTIVITY; pub use stats::SelectivityEstimator; pub use stats::TopNSet; pub use stats::UniformSampleSet; +pub(crate) use stats::finite_range_ndv_upper; diff --git a/src/query/sql/src/planner/optimizer/ir/stats/column_stat.rs b/src/query/sql/src/planner/optimizer/ir/stats/column_stat.rs index d3a631076bf87..3e8f5cd5e76ee 100644 --- a/src/query/sql/src/planner/optimizer/ir/stats/column_stat.rs +++ b/src/query/sql/src/planner/optimizer/ir/stats/column_stat.rs @@ -31,7 +31,7 @@ pub type ColumnStatSet = HashMap; pub type TopNSet = HashMap; pub type CountMinSketchSet = HashMap; -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq)] /// Statistics information of a column pub struct ColumnStat { /// Min value of the column diff --git a/src/query/sql/src/planner/optimizer/ir/stats/constraint.rs b/src/query/sql/src/planner/optimizer/ir/stats/constraint.rs index 0ec5b683e8f16..e85b5048cfc22 100644 --- a/src/query/sql/src/planner/optimizer/ir/stats/constraint.rs +++ b/src/query/sql/src/planner/optimizer/ir/stats/constraint.rs @@ -21,6 +21,7 @@ use databend_common_statistics::HistogramRangeBounds; use databend_common_statistics::NdvEstimate; use databend_common_statistics::StatCount; +use super::finite_range_ndv_upper; use crate::optimizer::ir::ColumnStat; use crate::plans::ComparisonOp; @@ -138,24 +139,6 @@ fn apply_range_bounds(column_stat: &mut ColumnStat, new_min: Datum, new_max: Dat Ok(()) } -fn finite_range_ndv_upper(min: &Datum, max: &Datum) -> Option { - if min == max { - return Some(1.0); - } - match (min, max) { - (Datum::Bool(false), Datum::Bool(true)) => Some(2.0), - (Datum::Int(min), Datum::Int(max)) => max - .checked_sub(*min) - .and_then(|diff| diff.checked_add(1)) - .map(|value| value as f64), - (Datum::UInt(min), Datum::UInt(max)) => max - .checked_sub(*min) - .and_then(|diff| diff.checked_add(1)) - .map(|value| value as f64), - _ => None, - } -} - pub(super) fn clear_for_empty_result(column_stat: &mut ColumnStat) { column_stat.ndv = NdvEstimate::exact(0.0); column_stat.null_count = StatCount::exact(0); diff --git a/src/query/sql/src/planner/optimizer/ir/stats/mod.rs b/src/query/sql/src/planner/optimizer/ir/stats/mod.rs index a3459db734582..cc66da1c74fcd 100644 --- a/src/query/sql/src/planner/optimizer/ir/stats/mod.rs +++ b/src/query/sql/src/planner/optimizer/ir/stats/mod.rs @@ -18,9 +18,28 @@ mod join; mod selectivity; pub use column_stat::*; +use databend_common_statistics::Datum; pub use databend_common_statistics::UniformSampleSet; pub(crate) use join::JoinConditionColumns; pub(crate) use join::JoinKeyStatUpdate; pub(crate) use join::JoinStatsEstimator; pub use selectivity::MAX_SELECTIVITY; pub use selectivity::SelectivityEstimator; + +pub(crate) fn finite_range_ndv_upper(min: &Datum, max: &Datum) -> Option { + if min == max { + return Some(1.0); + } + match (min, max) { + (Datum::Bool(false), Datum::Bool(true)) => Some(2.0), + (Datum::Int(min), Datum::Int(max)) => max + .checked_sub(*min) + .and_then(|diff| diff.checked_add(1)) + .map(|value| value as f64), + (Datum::UInt(min), Datum::UInt(max)) => max + .checked_sub(*min) + .and_then(|diff| diff.checked_add(1)) + .map(|value| value as f64), + _ => None, + } +} diff --git a/src/query/sql/src/planner/plans/eval_scalar.rs b/src/query/sql/src/planner/plans/eval_scalar.rs index f77ea40fd940c..0528ec31d0564 100644 --- a/src/query/sql/src/planner/plans/eval_scalar.rs +++ b/src/query/sql/src/planner/plans/eval_scalar.rs @@ -87,7 +87,7 @@ impl EvalScalar { Ok(used_columns) } - fn derive_item_stat( + pub(crate) fn derive_item_stat( scalar: &ScalarExpr, input_statistics: &Statistics, cardinality: StatCardinality, diff --git a/src/query/sql/src/planner/plans/union_all.rs b/src/query/sql/src/planner/plans/union_all.rs index 2f779929508d9..f9f69cf7357a6 100644 --- a/src/query/sql/src/planner/plans/union_all.rs +++ b/src/query/sql/src/planner/plans/union_all.rs @@ -12,14 +12,20 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::cmp::Ordering; use std::sync::Arc; use databend_common_catalog::table_context::TableContext; use databend_common_exception::Result; +use databend_common_expression::stat_distribution::NdvEstimate; +use databend_common_expression::stat_distribution::StatCardinality; +use databend_common_expression::stat_distribution::StatCount; use crate::ColumnSet; use crate::ScalarExpr; use crate::Symbol; +use crate::optimizer::ir::ColumnStat; +use crate::optimizer::ir::ColumnStatSet; use crate::optimizer::ir::Distribution; use crate::optimizer::ir::PhysicalProperty; use crate::optimizer::ir::RelExpr; @@ -27,15 +33,17 @@ use crate::optimizer::ir::RelationalProperty; use crate::optimizer::ir::RequiredProperty; use crate::optimizer::ir::StatInfo; use crate::optimizer::ir::Statistics; +use crate::optimizer::ir::finite_range_ndv_upper; +use crate::plans::EvalScalar; use crate::plans::Operator; use crate::plans::RelOp; #[derive(Clone, Debug, PartialEq, Eq, Hash)] pub struct UnionAll { // We'll cast the output of union to the expected data type by the cast expr at runtime. - // Left of union, output idx and the expected data type + // Left input symbol and its optional coercion expression, aligned by output position. pub left_outputs: Vec<(Symbol, Option)>, - // Right of union, output idx and the expected data type + // Right input symbol and its optional coercion expression, aligned by output position. pub right_outputs: Vec<(Symbol, Option)>, // Recursive cte scan names // For example: `with recursive t as (select 1 as x union all select m.x+f.x from t as m, t as f where m.x < 3) select * from t` @@ -57,34 +65,153 @@ impl UnionAll { Ok(used_columns) } - pub fn derive_union_stats( + fn derive_union_stats( &self, left_stat_info: Arc, right_stat_info: Arc, ) -> Result> { let cardinality = left_stat_info.cardinality + right_stat_info.cardinality; - let precise_cardinality = - left_stat_info - .statistics - .precise_cardinality - .and_then(|left_cardinality| { - right_stat_info - .statistics - .precise_cardinality - .map(|right_cardinality| left_cardinality + right_cardinality) - }); + let precise_cardinality = if let Some(left_cardinality) = + left_stat_info.statistics.precise_cardinality + && let Some(right_cardinality) = right_stat_info.statistics.precise_cardinality + { + Some(left_cardinality + right_cardinality) + } else { + None + }; + + let left_cardinality = left_stat_info + .statistics + .precise_cardinality + .map(StatCardinality::exact) + .unwrap_or_else(|| StatCardinality::estimate(left_stat_info.cardinality)); + let right_cardinality = right_stat_info + .statistics + .precise_cardinality + .map(StatCardinality::exact) + .unwrap_or_else(|| StatCardinality::estimate(right_stat_info.cardinality)); + + debug_assert_eq!(self.left_outputs.len(), self.right_outputs.len()); + debug_assert_eq!(self.left_outputs.len(), self.output_indexes.len()); + + let column_stats = self + .left_outputs + .iter() + .zip(&self.right_outputs) + .zip(self.output_indexes.iter().copied()) + .map( + |(((left_output, left_expr), (right_output, right_expr)), output)| { + let left = { + let statistics = &left_stat_info.statistics; + match left_expr.as_ref() { + Some(expr) => { + EvalScalar::derive_item_stat(expr, statistics, left_cardinality)? + } + None => statistics.column_stats.get(left_output).cloned(), + } + }; + let right = { + let statistics = &right_stat_info.statistics; + match right_expr.as_ref() { + Some(expr) => { + EvalScalar::derive_item_stat(expr, statistics, right_cardinality)? + } + None => statistics.column_stats.get(right_output).cloned(), + } + }; + + let (left, right) = match (left, right) { + (Some(left), Some(right)) => (left, right), + (Some(left), None) + if right_stat_info.statistics.precise_cardinality == Some(0) => + { + return Ok(Some((output, left))); + } + (None, Some(right)) + if left_stat_info.statistics.precise_cardinality == Some(0) => + { + return Ok(Some((output, right))); + } + // TODO: Distinguish all-NULL column statistics from unknown statistics so + // a non-empty all-NULL branch can contribute its NULL count without + // discarding the other branch's value statistics. + _ => return Ok(None), + }; + let mut ndv = Self::merge_ndv(&left, &right)?; + let min = if left.min.compare(&right.min)? == Ordering::Less { + left.min + } else { + right.min + }; + let max = if left.max.compare(&right.max)? == Ordering::Greater { + left.max + } else { + right.max + }; + + if let Some(upper) = finite_range_ndv_upper(&min, &max) { + ndv = ndv.reduce(upper); + } + let null_count = StatCount::sum(left.null_count, right.null_count); + Ok(Some((output, ColumnStat { + min, + max, + ndv, + null_count, + // Combining histograms requires aligning bucket boundaries and + // accounting for overlapping values. Dropping it is safer than + // exposing either child's distribution as the union distribution. + histogram: None, + }))) + }, + ) + .filter_map(Result::transpose) + .collect::>()?; Ok(Arc::new(StatInfo { cardinality, statistics: Statistics { precise_cardinality, - column_stats: Default::default(), + column_stats, top_n: Default::default(), count_min_sketch: Default::default(), }, })) } + + fn merge_ndv(left: &ColumnStat, right: &ColumnStat) -> Result { + let ndv_upper = left.ndv.upper + right.ndv.upper; + let ranges_disjoint = left.max.compare(&right.min)? == Ordering::Less + || right.max.compare(&left.min)? == Ordering::Less; + if ranges_disjoint + && let (Some(left), Some(right)) = (left.ndv.expected, right.ndv.expected) + { + return Ok(NdvEstimate::new(left + right, ndv_upper)); + } + + if left.min.is_numeric() + && let (Some(left), Some(left_ndv)) = (&left.histogram, left.ndv.expected) + && let (Some(right), Some(right_ndv)) = (&right.histogram, right.ndv.expected) + && let Some(intersection) = left.estimate_join_numeric_compatible(right)? + && let Some(intersection_ndv) = intersection.ndv.expected + { + let lower = left_ndv.max(right_ndv); + let expected = (left_ndv + right_ndv - intersection_ndv).clamp(lower, ndv_upper); + return Ok(NdvEstimate::new(expected, ndv_upper)); + } + + Ok(match (left.ndv.expected, right.ndv.expected) { + // Match join estimation's NDV fallback: use the larger expected + // NDV when both sides have one, preserve the known side when only + // one does, and become upper-only only when neither side has one. + (Some(left), Some(right)) => NdvEstimate::new(left.max(right), ndv_upper), + (Some(expected), None) | (None, Some(expected)) => { + NdvEstimate::new(expected, ndv_upper) + } + (None, None) => NdvEstimate::upper_bound(ndv_upper), + }) + } } impl Operator for UnionAll { diff --git a/src/query/sql/test-support/data/results/obfuscated/01_multi_join_avg_case_expression_optimized.txt b/src/query/sql/test-support/data/results/obfuscated/01_multi_join_avg_case_expression_optimized.txt index ead2cb9a734d2..4047c4affb5c9 100644 --- a/src/query/sql/test-support/data/results/obfuscated/01_multi_join_avg_case_expression_optimized.txt +++ b/src/query/sql/test-support/data/results/obfuscated/01_multi_join_avg_case_expression_optimized.txt @@ -14,27 +14,27 @@ Exchange(Merge) └── EvalScalar ├── scalars: [a0c.a0d (#0) AS (#0), a0c.a0k (#7) AS (#7), a0c.a0m (#9) AS (#9), a2x.a5m (#144) AS (#144), if(eq(a5r.a1v (#154), '603020'), 1, 0) AS (#166), a1z.a2t (#70) AS (#171), a2x.a4m (#118) AS (#172), a5r.a5w (#156) AS (#173)] └── Join(Inner) - ├── build keys: [a0m (#149)] - ├── probe keys: [a0c.a0m (#9)] + ├── build keys: [a5r.a5t (#151)] + ├── probe keys: [a0c.a0l (#8)] ├── other filters: [] ├── Exchange(Broadcast) - │ └── EvalScalar - │ ├── scalars: [CAST(a2x.a0m (#74) AS String NULL) AS (#149)] - │ └── Scan - │ ├── table: default.a2x (#2) - │ ├── filters: [eq(substring(a2x.a4m (#118), 20, 1), '1')] - │ ├── order by: [] - │ └── limit: NONE + │ └── Scan + │ ├── table: default.a5r (#3) + │ ├── filters: [eq(substring(a5r.a5w (#156), 1, 1), '1')] + │ ├── order by: [] + │ └── limit: NONE └── Join(Inner) - ├── build keys: [a5r.a5t (#151)] - ├── probe keys: [a0c.a0l (#8)] + ├── build keys: [a0m (#149)] + ├── probe keys: [a0c.a0m (#9)] ├── other filters: [] ├── Exchange(Broadcast) - │ └── Scan - │ ├── table: default.a5r (#3) - │ ├── filters: [eq(substring(a5r.a5w (#156), 1, 1), '1')] - │ ├── order by: [] - │ └── limit: NONE + │ └── EvalScalar + │ ├── scalars: [CAST(a2x.a0m (#74) AS String NULL) AS (#149)] + │ └── Scan + │ ├── table: default.a2x (#2) + │ ├── filters: [eq(substring(a2x.a4m (#118), 20, 1), '1')] + │ ├── order by: [] + │ └── limit: NONE └── Join(Inner) ├── build keys: [a1z.a0k (#48), a1z.a0n (#50)] ├── probe keys: [a0c.a0k (#7), a0c.a0n (#10)] diff --git a/src/query/sql/test-support/data/results/obfuscated/01_multi_join_avg_case_expression_physical.txt b/src/query/sql/test-support/data/results/obfuscated/01_multi_join_avg_case_expression_physical.txt index 1edb93957cf6b..e968e36f3c786 100644 --- a/src/query/sql/test-support/data/results/obfuscated/01_multi_join_avg_case_expression_physical.txt +++ b/src/query/sql/test-support/data/results/obfuscated/01_multi_join_avg_case_expression_physical.txt @@ -4,77 +4,77 @@ Exchange └── EvalScalar ├── output columns: [sell_mnt = 0 (#170)] ├── expressions: [t.sell_mnt (#169) = 0] - ├── estimated rows: 729765643534.22 + ├── estimated rows: 58807078.72 └── EvalScalar ├── output columns: [sell_mnt (#169)] ├── expressions: [sum(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) (#167) / CAST(if(CAST(count(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) (#168) = 0 AS Boolean NULL), 1, count(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) (#168)) AS UInt64 NULL) + 3] - ├── estimated rows: 729765643534.22 + ├── estimated rows: 58807078.72 └── AggregateFinal ├── output columns: [sum(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) (#167), count(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) (#168), a.a0d (#0), a.a0k (#7), a.a0m (#9), c.a5m (#144)] ├── group by: [a0d, a0k, a0m, a5m] ├── aggregate functions: [sum(sum_arg_0), count()] - ├── estimated rows: 729765643534.22 + ├── estimated rows: 58807078.72 └── Exchange ├── output columns: [sum(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) (#167), count(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) (#168), a.a0d (#0), a.a0k (#7), a.a0m (#9), c.a5m (#144)] ├── exchange type: Hash(0, 1, 2, 3) └── AggregatePartial ├── group by: [a0d, a0k, a0m, a5m] ├── aggregate functions: [sum(sum_arg_0), count()] - ├── estimated rows: 729765643534.22 + ├── estimated rows: 58807078.72 └── EvalScalar ├── output columns: [a.a0d (#0), a.a0k (#7), a.a0m (#9), c.a5m (#144), sum_arg_0 (#166)] ├── expressions: [if(d.a1v (#154) = '603020', 1, 0)] - ├── estimated rows: 3769669953985.67 + ├── estimated rows: 58807078.72 └── HashJoin - ├── output columns: [a.a0d (#0), a.a0k (#7), a.a0m (#9), d.a1v (#154), c.a5m (#144)] + ├── output columns: [a.a0d (#0), a.a0k (#7), a.a0m (#9), c.a5m (#144), d.a1v (#154)] ├── join type: INNER - ├── build keys: [c.a0m (#149)] - ├── probe keys: [a.a0m (#9)] + ├── build keys: [d.a5t (#151)] + ├── probe keys: [a.a0l (#8)] ├── keys is null equal: [false] ├── filters: [] ├── build join filters: - │ └── filter id:3, build key:c.a0m (#149), probe targets:[a.a0m (#9)@scan0], filter type:bloom,inlist,min_max - ├── estimated rows: 3769669953985.67 + │ └── filter id:3, build key:d.a5t (#151), probe targets:[a.a0l (#8)@scan0], filter type:bloom,inlist,min_max + ├── estimated rows: 58807078.72 ├── Exchange(Build) - │ ├── output columns: [c.a5m (#144), a0m (#149)] + │ ├── output columns: [d.a5t (#151), d.a1v (#154)] │ ├── exchange type: Broadcast - │ └── EvalScalar - │ ├── output columns: [c.a5m (#144), a0m (#149)] - │ ├── expressions: [CAST(c.a0m (#74) AS String NULL)] - │ ├── estimated rows: 63773.60 - │ └── TableScan - │ ├── table: default.default.a2x - │ ├── scan id: 2 - │ ├── output columns: [a0m (#74), a5m (#144)] - │ ├── read rows: 0 - │ ├── read size: 0 - │ ├── partitions total: 0 - │ ├── partitions scanned: 0 - │ ├── push downs: [filters: [is_true(substr(a2x.a4m (#118), 20, 1) = '1')], limit: NONE] - │ └── estimated rows: 63773.60 + │ └── TableScan + │ ├── table: default.default.a5r + │ ├── scan id: 3 + │ ├── output columns: [a5t (#151), a1v (#154)] + │ ├── read rows: 0 + │ ├── read size: 0 + │ ├── partitions total: 0 + │ ├── partitions scanned: 0 + │ ├── push downs: [filters: [is_true(substr(a5r.a5w (#156), 1, 1) = '1')], limit: NONE] + │ └── estimated rows: 806.60 └── HashJoin(Probe) - ├── output columns: [a.a0d (#0), a.a0k (#7), a.a0m (#9), d.a1v (#154)] + ├── output columns: [a.a0d (#0), a.a0k (#7), a.a0l (#8), a.a0m (#9), c.a5m (#144)] ├── join type: INNER - ├── build keys: [d.a5t (#151)] - ├── probe keys: [a.a0l (#8)] + ├── build keys: [c.a0m (#149)] + ├── probe keys: [a.a0m (#9)] ├── keys is null equal: [false] ├── filters: [] ├── build join filters: - │ └── filter id:2, build key:d.a5t (#151), probe targets:[a.a0l (#8)@scan0], filter type:bloom,inlist,min_max - ├── estimated rows: 59110195.35 + │ └── filter id:2, build key:c.a0m (#149), probe targets:[a.a0m (#9)@scan0], filter type:bloom,inlist,min_max + ├── estimated rows: 58716607.09 ├── Exchange(Build) - │ ├── output columns: [d.a5t (#151), d.a1v (#154)] + │ ├── output columns: [c.a5m (#144), a0m (#149)] │ ├── exchange type: Broadcast - │ └── TableScan - │ ├── table: default.default.a5r - │ ├── scan id: 3 - │ ├── output columns: [a5t (#151), a1v (#154)] - │ ├── read rows: 0 - │ ├── read size: 0 - │ ├── partitions total: 0 - │ ├── partitions scanned: 0 - │ ├── push downs: [filters: [is_true(substr(a5r.a5w (#156), 1, 1) = '1')], limit: NONE] - │ └── estimated rows: 806.60 + │ └── EvalScalar + │ ├── output columns: [c.a5m (#144), a0m (#149)] + │ ├── expressions: [CAST(c.a0m (#74) AS String NULL)] + │ ├── estimated rows: 63773.60 + │ └── TableScan + │ ├── table: default.default.a2x + │ ├── scan id: 2 + │ ├── output columns: [a0m (#74), a5m (#144)] + │ ├── read rows: 0 + │ ├── read size: 0 + │ ├── partitions total: 0 + │ ├── partitions scanned: 0 + │ ├── push downs: [filters: [is_true(substr(a2x.a4m (#118), 20, 1) = '1')], limit: NONE] + │ └── estimated rows: 63773.60 └── HashJoin(Probe) ├── output columns: [a.a0d (#0), a.a0k (#7), a.a0l (#8), a.a0m (#9)] ├── join type: INNER diff --git a/src/query/sql/test-support/data/results/obfuscated/01_multi_join_sum_case_expression_optimized.txt b/src/query/sql/test-support/data/results/obfuscated/01_multi_join_sum_case_expression_optimized.txt index 588e6c320b130..335dd4610ad11 100644 --- a/src/query/sql/test-support/data/results/obfuscated/01_multi_join_sum_case_expression_optimized.txt +++ b/src/query/sql/test-support/data/results/obfuscated/01_multi_join_sum_case_expression_optimized.txt @@ -3,44 +3,36 @@ Exchange(Merge) ├── scalars: [a0c.a0d (#0) AS (#0), a0c.a0k (#7) AS (#7), a0c.a0m (#9) AS (#9), a2x.a5m (#144) AS (#144), sum(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) (#167) AS (#167), eq(sum(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) (#167), 0) AS (#168)] └── Aggregate(Final) ├── group items: [a0c.a0d (#0) AS (#0), a0c.a0k (#7) AS (#7), a0c.a0m (#9) AS (#9), a2x.a5m (#144) AS (#144)] - ├── aggregate functions: [sum(sum(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) * _eager_count (#180)) AS (#167)] + ├── aggregate functions: [sum(sum_arg_0 (#166)) AS (#167)] └── Aggregate(Partial) ├── group items: [a0c.a0d (#0) AS (#0), a0c.a0k (#7) AS (#7), a0c.a0m (#9) AS (#9), a2x.a5m (#144) AS (#144)] - ├── aggregate functions: [sum(sum(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) * _eager_count (#180)) AS (#167)] + ├── aggregate functions: [sum(sum_arg_0 (#166)) AS (#167)] └── Exchange(Hash) ├── Exchange(Hash): keys: [a0c.a0d (#0)] └── EvalScalar - ├── scalars: [multiply(if(eq(a5r.a1v (#154), '603020'), 1, 0), CAST(count(*) (#179) AS UInt64)) AS (#180)] + ├── scalars: [a0c.a0d (#0) AS (#0), a0c.a0k (#7) AS (#7), a0c.a0m (#9) AS (#9), a2x.a5m (#144) AS (#144), if(eq(a5r.a1v (#154), '603020'), 1, 0) AS (#166), a1z.a2t (#70) AS (#169), a2x.a4m (#118) AS (#170), a5r.a5w (#156) AS (#171)] └── Join(Inner) - ├── build keys: [a0m (#149)] - ├── probe keys: [a0c.a0m (#9)] + ├── build keys: [a5r.a5t (#151)] + ├── probe keys: [a0c.a0l (#8)] ├── other filters: [] ├── Exchange(Broadcast) - │ └── Aggregate(Final) - │ ├── group items: [a2x.a5m (#144) AS (#144), a0m (#149) AS (#149)] - │ ├── aggregate functions: [count() AS (#179)] - │ └── Aggregate(Partial) - │ ├── group items: [a2x.a5m (#144) AS (#144), a0m (#149) AS (#149)] - │ ├── aggregate functions: [count() AS (#179)] - │ └── Exchange(Hash) - │ ├── Exchange(Hash): keys: [a2x.a5m (#144)] - │ └── EvalScalar - │ ├── scalars: [CAST(a2x.a0m (#74) AS String NULL) AS (#149)] - │ └── Scan - │ ├── table: default.a2x (#2) - │ ├── filters: [eq(substring(a2x.a4m (#118), 20, 1), '1')] - │ ├── order by: [] - │ └── limit: NONE + │ └── Scan + │ ├── table: default.a5r (#3) + │ ├── filters: [eq(substring(a5r.a5w (#156), 1, 1), '1')] + │ ├── order by: [] + │ └── limit: NONE └── Join(Inner) - ├── build keys: [a5r.a5t (#151)] - ├── probe keys: [a0c.a0l (#8)] + ├── build keys: [a0m (#149)] + ├── probe keys: [a0c.a0m (#9)] ├── other filters: [] ├── Exchange(Broadcast) - │ └── Scan - │ ├── table: default.a5r (#3) - │ ├── filters: [eq(substring(a5r.a5w (#156), 1, 1), '1')] - │ ├── order by: [] - │ └── limit: NONE + │ └── EvalScalar + │ ├── scalars: [CAST(a2x.a0m (#74) AS String NULL) AS (#149)] + │ └── Scan + │ ├── table: default.a2x (#2) + │ ├── filters: [eq(substring(a2x.a4m (#118), 20, 1), '1')] + │ ├── order by: [] + │ └── limit: NONE └── Join(Inner) ├── build keys: [a1z.a0k (#48), a1z.a0n (#50)] ├── probe keys: [a0c.a0k (#7), a0c.a0n (#10)] diff --git a/src/query/sql/test-support/data/results/obfuscated/01_multi_join_sum_case_expression_physical.txt b/src/query/sql/test-support/data/results/obfuscated/01_multi_join_sum_case_expression_physical.txt index dd15bb5fe54dc..8261d606287bc 100644 --- a/src/query/sql/test-support/data/results/obfuscated/01_multi_join_sum_case_expression_physical.txt +++ b/src/query/sql/test-support/data/results/obfuscated/01_multi_join_sum_case_expression_physical.txt @@ -4,85 +4,73 @@ Exchange └── EvalScalar ├── output columns: [sell_mnt = 0 (#168)] ├── expressions: [sum(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) (#167) = 0] - ├── estimated rows: 729765643534.22 + ├── estimated rows: 58807078.72 └── AggregateFinal ├── output columns: [sum(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) (#167), a.a0d (#0), a.a0k (#7), a.a0m (#9), c.a5m (#144)] ├── group by: [a0d, a0k, a0m, a5m] - ├── aggregate functions: [sum(sum(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) * _eager_count)] - ├── estimated rows: 729765643534.22 + ├── aggregate functions: [sum(sum_arg_0)] + ├── estimated rows: 58807078.72 └── Exchange ├── output columns: [sum(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) (#167), a.a0d (#0), a.a0k (#7), a.a0m (#9), c.a5m (#144)] ├── exchange type: Hash(0, 1, 2, 3) └── AggregatePartial ├── group by: [a0d, a0k, a0m, a5m] - ├── aggregate functions: [sum(sum(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) * _eager_count)] - ├── estimated rows: 729765643534.22 + ├── aggregate functions: [sum(sum_arg_0)] + ├── estimated rows: 58807078.72 └── EvalScalar - ├── output columns: [a.a0d (#0), a.a0k (#7), a.a0m (#9), c.a5m (#144), sum(CASE WHEN d.a1v = '603020' THEN 1 ELSE 0 END) * _eager_count (#180)] - ├── expressions: [if(d.a1v (#154) = '603020', 1, 0) * _eager_count (#179)] - ├── estimated rows: 1256556651328.56 + ├── output columns: [a.a0d (#0), a.a0k (#7), a.a0m (#9), c.a5m (#144), sum_arg_0 (#166)] + ├── expressions: [if(d.a1v (#154) = '603020', 1, 0)] + ├── estimated rows: 58807078.72 └── HashJoin - ├── output columns: [a.a0d (#0), a.a0k (#7), a.a0m (#9), d.a1v (#154), count(*) (#179), c.a5m (#144)] + ├── output columns: [a.a0d (#0), a.a0k (#7), a.a0m (#9), c.a5m (#144), d.a1v (#154)] ├── join type: INNER - ├── build keys: [c.a0m (#149)] - ├── probe keys: [a.a0m (#9)] + ├── build keys: [d.a5t (#151)] + ├── probe keys: [a.a0l (#8)] ├── keys is null equal: [false] ├── filters: [] ├── build join filters: - │ └── filter id:3, build key:c.a0m (#149), probe targets:[a.a0m (#9)@scan0], filter type:bloom,inlist,min_max - ├── estimated rows: 1256556651328.56 + │ └── filter id:3, build key:d.a5t (#151), probe targets:[a.a0l (#8)@scan0], filter type:bloom,inlist,min_max + ├── estimated rows: 58807078.72 ├── Exchange(Build) - │ ├── output columns: [count(*) (#179), c.a5m (#144), a0m (#149)] + │ ├── output columns: [d.a5t (#151), d.a1v (#154)] │ ├── exchange type: Broadcast - │ └── AggregateFinal - │ ├── output columns: [count(*) (#179), c.a5m (#144), a0m (#149)] - │ ├── group by: [a5m, a0m] - │ ├── aggregate functions: [count()] - │ ├── estimated rows: 21257.87 - │ └── Exchange - │ ├── output columns: [count(*) (#179), c.a5m (#144), a0m (#149)] - │ ├── exchange type: Hash(0, 1) - │ └── AggregatePartial - │ ├── group by: [a5m, a0m] - │ ├── aggregate functions: [count()] - │ ├── estimated rows: 21257.87 - │ └── EvalScalar - │ ├── output columns: [c.a5m (#144), a0m (#149)] - │ ├── expressions: [CAST(c.a0m (#74) AS String NULL)] - │ ├── estimated rows: 63773.60 - │ └── TableScan - │ ├── table: default.default.a2x - │ ├── scan id: 2 - │ ├── output columns: [a0m (#74), a5m (#144)] - │ ├── read rows: 0 - │ ├── read size: 0 - │ ├── partitions total: 0 - │ ├── partitions scanned: 0 - │ ├── push downs: [filters: [is_true(substr(a2x.a4m (#118), 20, 1) = '1')], limit: NONE] - │ └── estimated rows: 63773.60 + │ └── TableScan + │ ├── table: default.default.a5r + │ ├── scan id: 3 + │ ├── output columns: [a5t (#151), a1v (#154)] + │ ├── read rows: 0 + │ ├── read size: 0 + │ ├── partitions total: 0 + │ ├── partitions scanned: 0 + │ ├── push downs: [filters: [is_true(substr(a5r.a5w (#156), 1, 1) = '1')], limit: NONE] + │ └── estimated rows: 806.60 └── HashJoin(Probe) - ├── output columns: [a.a0d (#0), a.a0k (#7), a.a0m (#9), d.a1v (#154)] + ├── output columns: [a.a0d (#0), a.a0k (#7), a.a0l (#8), a.a0m (#9), c.a5m (#144)] ├── join type: INNER - ├── build keys: [d.a5t (#151)] - ├── probe keys: [a.a0l (#8)] + ├── build keys: [c.a0m (#149)] + ├── probe keys: [a.a0m (#9)] ├── keys is null equal: [false] ├── filters: [] ├── build join filters: - │ └── filter id:2, build key:d.a5t (#151), probe targets:[a.a0l (#8)@scan0], filter type:bloom,inlist,min_max - ├── estimated rows: 59110195.35 + │ └── filter id:2, build key:c.a0m (#149), probe targets:[a.a0m (#9)@scan0], filter type:bloom,inlist,min_max + ├── estimated rows: 58716607.09 ├── Exchange(Build) - │ ├── output columns: [d.a5t (#151), d.a1v (#154)] + │ ├── output columns: [c.a5m (#144), a0m (#149)] │ ├── exchange type: Broadcast - │ └── TableScan - │ ├── table: default.default.a5r - │ ├── scan id: 3 - │ ├── output columns: [a5t (#151), a1v (#154)] - │ ├── read rows: 0 - │ ├── read size: 0 - │ ├── partitions total: 0 - │ ├── partitions scanned: 0 - │ ├── push downs: [filters: [is_true(substr(a5r.a5w (#156), 1, 1) = '1')], limit: NONE] - │ └── estimated rows: 806.60 + │ └── EvalScalar + │ ├── output columns: [c.a5m (#144), a0m (#149)] + │ ├── expressions: [CAST(c.a0m (#74) AS String NULL)] + │ ├── estimated rows: 63773.60 + │ └── TableScan + │ ├── table: default.default.a2x + │ ├── scan id: 2 + │ ├── output columns: [a0m (#74), a5m (#144)] + │ ├── read rows: 0 + │ ├── read size: 0 + │ ├── partitions total: 0 + │ ├── partitions scanned: 0 + │ ├── push downs: [filters: [is_true(substr(a2x.a4m (#118), 20, 1) = '1')], limit: NONE] + │ └── estimated rows: 63773.60 └── HashJoin(Probe) ├── output columns: [a.a0d (#0), a.a0k (#7), a.a0l (#8), a.a0m (#9)] ├── join type: INNER diff --git a/src/query/sql/test-support/data/results/regressions/19574_correlated_exists_union_optimized.txt b/src/query/sql/test-support/data/results/regressions/19574_correlated_exists_union_optimized.txt index 4e94aeea7f641..0be42ca5e5d65 100644 --- a/src/query/sql/test-support/data/results/regressions/19574_correlated_exists_union_optimized.txt +++ b/src/query/sql/test-support/data/results/regressions/19574_correlated_exists_union_optimized.txt @@ -2,64 +2,64 @@ EvalScalar ├── scalars: [f1 (#0) AS (#0), marker (#7) AS (#8)] └── Filter ├── filters: [is_true(marker (#7))] - └── Join(RightMark) - ├── build keys: [f1 (#6)] - ├── probe keys: [f1 (#0)] + └── Join(LeftMark) + ├── build keys: [f1 (#0)] + ├── probe keys: [f1 (#6)] ├── other filters: [] - ├── Aggregate(Final) - │ ├── group items: [1 (#3) AS (#3), f1 (#6) AS (#6)] - │ ├── aggregate functions: [] - │ └── Aggregate(Partial) - │ ├── group items: [1 (#3) AS (#3), f1 (#6) AS (#6)] - │ ├── aggregate functions: [] - │ └── UnionAll - │ ├── output: [1 (#3), f1 (#6)] - │ ├── left: [1 (#1), f1 (#4)] - │ ├── right: [2 (#2), f1 (#5)] - │ ├── cte_scan_names: [] - │ ├── logical_recursive_cte_id: None - │ ├── Join(Cross) - │ │ ├── build keys: [] - │ │ ├── probe keys: [] - │ │ ├── other filters: [] - │ │ ├── EvalScalar - │ │ │ ├── scalars: [1 AS (#1)] - │ │ │ └── DummyTableScan(DummyTableScan { source_table_indexes: [] }) - │ │ └── Exchange(Merge) - │ │ └── Aggregate(Final) - │ │ ├── group items: [f1 (#4) AS (#4)] - │ │ ├── aggregate functions: [] - │ │ └── Aggregate(Partial) - │ │ ├── group items: [f1 (#4) AS (#4)] - │ │ ├── aggregate functions: [] - │ │ └── Exchange(Hash) - │ │ ├── Exchange(Hash): keys: [f1 (#4)] - │ │ └── ConstantTableScan - │ │ ├── columns: [f1 (#4)] - │ │ └── num_rows: [1] - │ └── EvalScalar - │ ├── scalars: [2 AS (#2), f1 (#5) AS (#5)] - │ └── Join(Cross) - │ ├── build keys: [] - │ ├── probe keys: [] - │ ├── other filters: [] - │ ├── DummyTableScan(DummyTableScan { source_table_indexes: [] }) - │ └── Exchange(Merge) - │ └── Aggregate(Final) - │ ├── group items: [f1 (#5) AS (#5)] - │ ├── aggregate functions: [] - │ └── Aggregate(Partial) - │ ├── group items: [f1 (#5) AS (#5)] - │ ├── aggregate functions: [] - │ └── Exchange(Hash) - │ ├── Exchange(Hash): keys: [f1 (#5)] - │ └── Filter - │ ├── filters: [eq(f1 (#5), 1)] - │ └── ConstantTableScan - │ ├── columns: [f1 (#5)] - │ └── num_rows: [1] - └── Exchange(Merge) - └── ConstantTableScan - ├── columns: [f1 (#0)] - └── num_rows: [1] + ├── Exchange(Merge) + │ └── ConstantTableScan + │ ├── columns: [f1 (#0)] + │ └── num_rows: [1] + └── Aggregate(Final) + ├── group items: [1 (#3) AS (#3), f1 (#6) AS (#6)] + ├── aggregate functions: [] + └── Aggregate(Partial) + ├── group items: [1 (#3) AS (#3), f1 (#6) AS (#6)] + ├── aggregate functions: [] + └── UnionAll + ├── output: [1 (#3), f1 (#6)] + ├── left: [1 (#1), f1 (#4)] + ├── right: [2 (#2), f1 (#5)] + ├── cte_scan_names: [] + ├── logical_recursive_cte_id: None + ├── Join(Cross) + │ ├── build keys: [] + │ ├── probe keys: [] + │ ├── other filters: [] + │ ├── EvalScalar + │ │ ├── scalars: [1 AS (#1)] + │ │ └── DummyTableScan(DummyTableScan { source_table_indexes: [] }) + │ └── Exchange(Merge) + │ └── Aggregate(Final) + │ ├── group items: [f1 (#4) AS (#4)] + │ ├── aggregate functions: [] + │ └── Aggregate(Partial) + │ ├── group items: [f1 (#4) AS (#4)] + │ ├── aggregate functions: [] + │ └── Exchange(Hash) + │ ├── Exchange(Hash): keys: [f1 (#4)] + │ └── ConstantTableScan + │ ├── columns: [f1 (#4)] + │ └── num_rows: [1] + └── EvalScalar + ├── scalars: [2 AS (#2), f1 (#5) AS (#5)] + └── Join(Cross) + ├── build keys: [] + ├── probe keys: [] + ├── other filters: [] + ├── DummyTableScan(DummyTableScan { source_table_indexes: [] }) + └── Exchange(Merge) + └── Aggregate(Final) + ├── group items: [f1 (#5) AS (#5)] + ├── aggregate functions: [] + └── Aggregate(Partial) + ├── group items: [f1 (#5) AS (#5)] + ├── aggregate functions: [] + └── Exchange(Hash) + ├── Exchange(Hash): keys: [f1 (#5)] + └── Filter + ├── filters: [eq(f1 (#5), 1)] + └── ConstantTableScan + ├── columns: [f1 (#5)] + └── num_rows: [1] diff --git a/src/query/sql/test-support/data/results/regressions/19574_correlated_exists_union_physical.txt b/src/query/sql/test-support/data/results/regressions/19574_correlated_exists_union_physical.txt index 08f8b9c5225a2..c0f8eecdefbbb 100644 --- a/src/query/sql/test-support/data/results/regressions/19574_correlated_exists_union_physical.txt +++ b/src/query/sql/test-support/data/results/regressions/19574_correlated_exists_union_physical.txt @@ -4,94 +4,94 @@ Filter ├── estimated rows: 0.20 └── HashJoin ├── output columns: [f1 (#0), marker (#7)] - ├── join type: RIGHT MARK - ├── build keys: [f1 (#6)] - ├── probe keys: [f1 (#0)] + ├── join type: LEFT MARK + ├── build keys: [f1 (#0)] + ├── probe keys: [f1 (#6)] ├── keys is null equal: [false] ├── filters: [] ├── estimated rows: 1.00 - ├── AggregateFinal(Build) - │ ├── output columns: [1 (#3), f1 (#6)] - │ ├── group by: [1, f1] - │ ├── aggregate functions: [] - │ ├── estimated rows: 1.00 - │ └── AggregatePartial - │ ├── group by: [1, f1] - │ ├── aggregate functions: [] - │ ├── estimated rows: 1.00 - │ └── UnionAll - │ ├── output columns: [1 (#3), f1 (#6)] - │ ├── estimated rows: 2.00 - │ ├── HashJoin - │ │ ├── output columns: [f1 (#4), 1 (#1)] - │ │ ├── join type: CROSS - │ │ ├── build keys: [] - │ │ ├── probe keys: [] - │ │ ├── keys is null equal: [] - │ │ ├── filters: [] - │ │ ├── estimated rows: 1.00 - │ │ ├── EvalScalar(Build) - │ │ │ ├── output columns: [1 (#1)] - │ │ │ ├── expressions: [1] - │ │ │ ├── estimated rows: 1.00 - │ │ │ └── DummyTableScan - │ │ └── Exchange(Probe) - │ │ ├── output columns: [f1 (#4)] - │ │ ├── exchange type: Merge - │ │ └── AggregateFinal - │ │ ├── output columns: [f1 (#4)] - │ │ ├── group by: [f1] - │ │ ├── aggregate functions: [] - │ │ ├── estimated rows: 1.00 - │ │ └── Exchange - │ │ ├── output columns: [f1 (#4)] - │ │ ├── exchange type: Hash(0) - │ │ └── AggregatePartial - │ │ ├── group by: [f1] - │ │ ├── aggregate functions: [] - │ │ ├── estimated rows: 1.00 - │ │ └── ConstantTableScan - │ │ ├── output columns: [f1 (#4)] - │ │ └── column 0: [1] - │ └── EvalScalar - │ ├── output columns: [f1 (#5), 2 (#2)] - │ ├── expressions: [2] - │ ├── estimated rows: 1.00 - │ └── HashJoin - │ ├── output columns: [f1 (#5)] - │ ├── join type: CROSS - │ ├── build keys: [] - │ ├── probe keys: [] - │ ├── keys is null equal: [] - │ ├── filters: [] - │ ├── estimated rows: 1.00 - │ ├── DummyTableScan(Build) - │ └── Exchange(Probe) - │ ├── output columns: [f1 (#5)] - │ ├── exchange type: Merge - │ └── AggregateFinal - │ ├── output columns: [f1 (#5)] - │ ├── group by: [f1] - │ ├── aggregate functions: [] - │ ├── estimated rows: 1.00 - │ └── Exchange - │ ├── output columns: [f1 (#5)] - │ ├── exchange type: Hash(0) - │ └── AggregatePartial - │ ├── group by: [f1] - │ ├── aggregate functions: [] - │ ├── estimated rows: 1.00 - │ └── Filter - │ ├── output columns: [f1 (#5)] - │ ├── filters: [outer.f1 (#5) = 1] - │ ├── estimated rows: 1.00 - │ └── ConstantTableScan - │ ├── output columns: [f1 (#5)] - │ └── column 0: [1] - └── Exchange(Probe) - ├── output columns: [f1 (#0)] - ├── exchange type: Merge - └── ConstantTableScan - ├── output columns: [f1 (#0)] - └── column 0: [1] + ├── Exchange(Build) + │ ├── output columns: [f1 (#0)] + │ ├── exchange type: Merge + │ └── ConstantTableScan + │ ├── output columns: [f1 (#0)] + │ └── column 0: [1] + └── AggregateFinal(Probe) + ├── output columns: [1 (#3), f1 (#6)] + ├── group by: [1, f1] + ├── aggregate functions: [] + ├── estimated rows: 2.00 + └── AggregatePartial + ├── group by: [1, f1] + ├── aggregate functions: [] + ├── estimated rows: 2.00 + └── UnionAll + ├── output columns: [1 (#3), f1 (#6)] + ├── estimated rows: 2.00 + ├── HashJoin + │ ├── output columns: [f1 (#4), 1 (#1)] + │ ├── join type: CROSS + │ ├── build keys: [] + │ ├── probe keys: [] + │ ├── keys is null equal: [] + │ ├── filters: [] + │ ├── estimated rows: 1.00 + │ ├── EvalScalar(Build) + │ │ ├── output columns: [1 (#1)] + │ │ ├── expressions: [1] + │ │ ├── estimated rows: 1.00 + │ │ └── DummyTableScan + │ └── Exchange(Probe) + │ ├── output columns: [f1 (#4)] + │ ├── exchange type: Merge + │ └── AggregateFinal + │ ├── output columns: [f1 (#4)] + │ ├── group by: [f1] + │ ├── aggregate functions: [] + │ ├── estimated rows: 1.00 + │ └── Exchange + │ ├── output columns: [f1 (#4)] + │ ├── exchange type: Hash(0) + │ └── AggregatePartial + │ ├── group by: [f1] + │ ├── aggregate functions: [] + │ ├── estimated rows: 1.00 + │ └── ConstantTableScan + │ ├── output columns: [f1 (#4)] + │ └── column 0: [1] + └── EvalScalar + ├── output columns: [f1 (#5), 2 (#2)] + ├── expressions: [2] + ├── estimated rows: 1.00 + └── HashJoin + ├── output columns: [f1 (#5)] + ├── join type: CROSS + ├── build keys: [] + ├── probe keys: [] + ├── keys is null equal: [] + ├── filters: [] + ├── estimated rows: 1.00 + ├── DummyTableScan(Build) + └── Exchange(Probe) + ├── output columns: [f1 (#5)] + ├── exchange type: Merge + └── AggregateFinal + ├── output columns: [f1 (#5)] + ├── group by: [f1] + ├── aggregate functions: [] + ├── estimated rows: 1.00 + └── Exchange + ├── output columns: [f1 (#5)] + ├── exchange type: Hash(0) + └── AggregatePartial + ├── group by: [f1] + ├── aggregate functions: [] + ├── estimated rows: 1.00 + └── Filter + ├── output columns: [f1 (#5)] + ├── filters: [outer.f1 (#5) = 1] + ├── estimated rows: 1.00 + └── ConstantTableScan + ├── output columns: [f1 (#5)] + └── column 0: [1] diff --git a/src/query/sql/test-support/src/optimizer/mod.rs b/src/query/sql/test-support/src/optimizer/mod.rs index 3ac438ed9fbf0..b716fd3f66956 100644 --- a/src/query/sql/test-support/src/optimizer/mod.rs +++ b/src/query/sql/test-support/src/optimizer/mod.rs @@ -93,6 +93,18 @@ pub struct ColumnStats { pub null_count: Option, } +impl ColumnStats { + pub fn to_basic_column_statistics(&self) -> BasicColumnStatistics { + BasicColumnStatistics { + min: to_datum(&self.min), + max: to_datum(&self.max), + ndv: self.ndv, + null_count: self.null_count.unwrap_or(0), + in_memory_size: 0, + } + } +} + #[derive(Debug, Serialize, Deserialize, Clone)] pub struct HistogramStats { pub accuracy: bool, @@ -100,6 +112,27 @@ pub struct HistogramStats { pub avg_spacing: Option, } +impl HistogramStats { + pub fn to_histogram(&self) -> Result { + let buckets = self + .buckets + .iter() + .map(|bucket| { + HistogramBucket::try_from_bounds( + bucket.lower_bound.clone(), + bucket.upper_bound.clone(), + bucket.num_values, + bucket.num_distinct, + ) + .map_err(|err| ErrorCode::Internal(format!("invalid histogram bucket: {err}"))) + }) + .collect::>>()?; + + Histogram::try_from_buckets(self.accuracy, buckets, self.avg_spacing) + .map_err(|err| ErrorCode::Internal(format!("invalid histogram: {err}"))) + } +} + #[derive(Debug, Serialize, Deserialize, Clone)] pub struct HistogramBucketStats { pub lower_bound: Datum, @@ -516,18 +549,10 @@ impl StatsApplier<'_> { { let full_name = format!("{table_name}.{column_name}"); if let Some(stats) = self.column_stats.get(&full_name) { - result.insert( - *column_index, - Some(BasicColumnStatistics { - min: to_datum(&stats.min) - .or_else(|| default_min_datum(&column.data_type())), - max: to_datum(&stats.max) - .or_else(|| default_max_datum(&column.data_type())), - ndv: stats.ndv, - null_count: stats.null_count.unwrap_or(0), - in_memory_size: 0, - }), - ); + let mut stats = stats.to_basic_column_statistics(); + stats.min = stats.min.or_else(|| default_min_datum(&column.data_type())); + stats.max = stats.max.or_else(|| default_max_datum(&column.data_type())); + result.insert(*column_index, Some(stats)); } } } @@ -552,7 +577,7 @@ impl StatsApplier<'_> { { let full_name = format!("{table_name}.{column_name}"); if let Some(stats) = self.histogram_stats.get(&full_name) { - result.insert(*column_index, Some(histogram_from_stats(stats)?)); + result.insert(*column_index, Some(stats.to_histogram()?)); } } } @@ -561,25 +586,6 @@ impl StatsApplier<'_> { } } -pub(crate) fn histogram_from_stats(stats: &HistogramStats) -> Result { - let buckets = stats - .buckets - .iter() - .map(|bucket| { - HistogramBucket::try_from_bounds( - bucket.lower_bound.clone(), - bucket.upper_bound.clone(), - bucket.num_values, - bucket.num_distinct, - ) - .map_err(|err| ErrorCode::Internal(format!("invalid histogram bucket: {err}"))) - }) - .collect::>>()?; - - Histogram::try_from_buckets(stats.accuracy, buckets, stats.avg_spacing) - .map_err(|err| ErrorCode::Internal(format!("invalid histogram: {err}"))) -} - fn write_result(mint: &mut Mint, name: &str, f: F) -> Result<()> where F: FnOnce(&mut dyn Write) -> Result<()> { let mut file = mint.new_goldenfile(name).unwrap(); diff --git a/src/query/sql/test-support/src/optimizer/replay.rs b/src/query/sql/test-support/src/optimizer/replay.rs index 20ebcc9655ff7..5b028de56e1dd 100644 --- a/src/query/sql/test-support/src/optimizer/replay.rs +++ b/src/query/sql/test-support/src/optimizer/replay.rs @@ -26,7 +26,6 @@ use databend_common_statistics::Histogram; use serde::Deserialize; use super::HistogramStats; -use super::histogram_from_stats; #[derive(Debug, Clone, Deserialize)] pub struct ReplayInput { @@ -159,7 +158,8 @@ impl ReplayInput { return None; } column.histogram.as_ref().map(|histogram| { - histogram_from_stats(histogram) + histogram + .to_histogram() .map(|histogram| (column.column_name.clone(), histogram)) }) }) diff --git a/src/query/sql/tests/it/optimizer/join_cardinality.rs b/src/query/sql/tests/it/optimizer/join_cardinality.rs index 3752d907c9ab0..8cf8039559936 100644 --- a/src/query/sql/tests/it/optimizer/join_cardinality.rs +++ b/src/query/sql/tests/it/optimizer/join_cardinality.rs @@ -17,7 +17,6 @@ use std::io::Write; use std::sync::Arc; use databend_common_catalog::BasicColumnStatistics; -use databend_common_catalog::TableStatistics; use databend_common_exception::ErrorCode; use databend_common_exception::Result; use databend_common_expression::stat_distribution::NdvEstimate; @@ -41,10 +40,11 @@ use databend_common_sql::plans::JoinType; use databend_common_sql::plans::Plan; use databend_common_sql::plans::RelOperator; use databend_common_sql::plans::ScalarExpr; -use databend_common_statistics::Datum; use databend_common_statistics::Histogram; -use databend_storages_common_table_meta::meta::TableSnapshotStatistics; +use super::column_stat; +use super::histogram_stat; +use super::table_statistics; use crate::framework::LiteTableContext; use crate::framework::golden::open_golden_file; use crate::framework::golden::write_case_title; @@ -52,12 +52,20 @@ use crate::framework::golden::write_case_title; #[derive(Clone, Copy)] struct TableStats { rows: u64, - min: i64, - max: i64, - ndv: u64, + column_json: &'static str, histogram_json: &'static str, } +impl TableStats { + fn column_stat(self) -> Result { + column_stat(self.column_json) + } + + fn histogram(self) -> Result { + histogram_stat(self.histogram_json) + } +} + #[derive(Clone, Copy)] struct JoinQueryCase { name: &'static str, @@ -86,55 +94,12 @@ struct JoinBehaviorGroup { cases: Vec, } -fn table_statistics(rows: u64) -> TableStatistics { - TableStatistics { - num_rows: Some(rows), - data_size: Some(rows.saturating_mul(8)), - data_size_compressed: None, - index_size: None, - bloom_index_size: None, - ngram_index_size: None, - inverted_index_size: None, - vector_index_size: None, - virtual_column_size: None, - number_of_blocks: Some(1), - number_of_segments: Some(1), - } -} - -fn column_statistics(stats: TableStats) -> HashMap { - HashMap::from([("k".to_string(), BasicColumnStatistics { - min: Some(Datum::Int(stats.min)), - max: Some(Datum::Int(stats.max)), - ndv: Some(stats.ndv), - null_count: 0, - in_memory_size: stats.rows.saturating_mul(8), - })]) -} - -fn histogram_from_json(json: &str) -> Result { - let statistics_json = format!( - r#"{{ - "format_version": 4, - "snapshot_id": "00000000-0000-0000-0000-000000000000", - "row_count": 0, - "hll": {{}}, - "histograms": {{"0": {json}}} - }}"# - ); - let mut statistics: TableSnapshotStatistics = serde_json::from_str(&statistics_json) - .map_err(|err| ErrorCode::Internal(format!("invalid histogram json: {err}")))?; - statistics - .histograms - .remove(&0) - .ok_or_else(|| ErrorCode::Internal("invalid histogram".to_string())) +fn column_statistics(stats: TableStats) -> Result> { + Ok(HashMap::from([("k".to_string(), stats.column_stat()?)])) } fn histogram_statistics(stats: TableStats) -> Result> { - Ok(HashMap::from([( - "k".to_string(), - histogram_from_json(stats.histogram_json)?, - )])) + Ok(HashMap::from([("k".to_string(), stats.histogram()?)])) } fn column_label(metadata: &Metadata, column: Symbol) -> String { @@ -271,16 +236,26 @@ fn direct_column(column: usize, table: &str) -> ScalarExpr { } fn direct_stat_info(column: usize, stats: TableStats) -> Result> { + let column_stat = stats.column_stat()?; + let min = column_stat + .min + .ok_or_else(|| ErrorCode::Internal("direct column statistics require min".to_string()))?; + let max = column_stat + .max + .ok_or_else(|| ErrorCode::Internal("direct column statistics require max".to_string()))?; + let ndv = column_stat + .ndv + .ok_or_else(|| ErrorCode::Internal("direct column statistics require ndv".to_string()))?; Ok(Arc::new(StatInfo { cardinality: stats.rows as f64, statistics: Statistics { precise_cardinality: None, column_stats: HashMap::from([(Symbol::new(column), ColumnStat { - min: Datum::Int(stats.min), - max: Datum::Int(stats.max), - ndv: NdvEstimate::exact(stats.ndv as f64), - null_count: StatCount::exact(0), - histogram: Some(histogram_from_json(stats.histogram_json)?), + min, + max, + ndv: NdvEstimate::exact(ndv as f64), + null_count: StatCount::exact(column_stat.null_count), + histogram: Some(stats.histogram()?), })]), top_n: Default::default(), count_min_sketch: Default::default(), @@ -329,14 +304,14 @@ async fn write_sql_join_input( ctx.register_table_sql_with_stats( "CREATE TABLE l(k BIGINT, t BIGINT)", Some(table_statistics(case.left.rows)), - column_statistics(case.left), + column_statistics(case.left)?, histogram_statistics(case.left)?, ) .await?; ctx.register_table_sql_with_stats( "CREATE TABLE r(k BIGINT, t BIGINT)", Some(table_statistics(case.right.rows)), - column_statistics(case.right), + column_statistics(case.right)?, histogram_statistics(case.right)?, ) .await?; @@ -360,15 +335,27 @@ async fn write_sql_join_input( fn write_stats_case_header(file: &mut impl Write, case: &JoinTestCase) -> Result<()> { writeln!(file, "case : {}", case.name)?; writeln!(file, "description : {}", case.description)?; + write_input_stats(file, "left", case.left)?; + write_input_stats(file, "right", case.right)?; + Ok(()) +} + +fn write_input_stats(file: &mut impl Write, side: &str, stats: TableStats) -> Result<()> { + let column_stat = stats.column_stat()?; + let min = column_stat + .min + .ok_or_else(|| ErrorCode::Internal(format!("{side} column statistics require min")))?; + let max = column_stat + .max + .ok_or_else(|| ErrorCode::Internal(format!("{side} column statistics require max")))?; + let ndv = column_stat + .ndv + .ok_or_else(|| ErrorCode::Internal(format!("{side} column statistics require ndv")))?; + let label = format!("{side} stats"); writeln!( file, - "left stats : rows={}, min={}, max={}, ndv={}", - case.left.rows, case.left.min, case.left.max, case.left.ndv - )?; - writeln!( - file, - "right stats : rows={}, min={}, max={}, ndv={}", - case.right.rows, case.right.min, case.right.max, case.right.ndv + "{label:<14}: rows={}, min={}, max={}, ndv={}", + stats.rows, min, max, ndv )?; Ok(()) } @@ -408,9 +395,7 @@ async fn test_join_cardinality_estimation_golden() -> Result<()> { fn overlap_left_stats() -> TableStats { TableStats { rows: 9, - min: 1, - max: 5, - ndv: 3, + column_json: r#"{"min": 1, "max": 5, "ndv": 3, "null_count": 0}"#, histogram_json: r#"{ "accuracy": true, "buckets": [ @@ -425,9 +410,7 @@ fn overlap_left_stats() -> TableStats { fn overlap_right_stats() -> TableStats { TableStats { rows: 26, - min: 1, - max: 5, - ndv: 4, + column_json: r#"{"min": 1, "max": 5, "ndv": 4, "null_count": 0}"#, histogram_json: r#"{ "accuracy": true, "buckets": [ @@ -443,9 +426,7 @@ fn overlap_right_stats() -> TableStats { fn no_overlap_right_stats() -> TableStats { TableStats { rows: 26, - min: 20, - max: 23, - ndv: 4, + column_json: r#"{"min": 20, "max": 23, "ndv": 4, "null_count": 0}"#, histogram_json: r#"{ "accuracy": true, "buckets": [ @@ -461,9 +442,7 @@ fn no_overlap_right_stats() -> TableStats { fn partial_overlap_right_stats() -> TableStats { TableStats { rows: 30, - min: 3, - max: 9, - ndv: 4, + column_json: r#"{"min": 3, "max": 9, "ndv": 4, "null_count": 0}"#, histogram_json: r#"{ "accuracy": true, "buckets": [ diff --git a/src/query/sql/tests/it/optimizer/mod.rs b/src/query/sql/tests/it/optimizer/mod.rs index 2c4c809277bce..49ccb7d34b994 100644 --- a/src/query/sql/tests/it/optimizer/mod.rs +++ b/src/query/sql/tests/it/optimizer/mod.rs @@ -12,6 +12,14 @@ // See the License for the specific language governing permissions and // limitations under the License. +use databend_common_catalog::BasicColumnStatistics; +use databend_common_catalog::TableStatistics; +use databend_common_exception::ErrorCode; +use databend_common_exception::Result; +use databend_common_sql_test_support::ColumnStats; +use databend_common_sql_test_support::HistogramStats; +use databend_common_statistics::Histogram; + mod collect_statistics; mod decorrelate_correlated_aliases; mod eager_aggregation; @@ -21,3 +29,32 @@ mod outer_join_to_anti; mod push_down_filter_project_set; mod selectivity; mod selectivity_smoke; +mod union_all; + +fn table_statistics(rows: u64) -> TableStatistics { + TableStatistics { + num_rows: Some(rows), + data_size: Some(rows.saturating_mul(8)), + data_size_compressed: None, + index_size: None, + bloom_index_size: None, + ngram_index_size: None, + inverted_index_size: None, + vector_index_size: None, + virtual_column_size: None, + number_of_blocks: Some(1), + number_of_segments: Some(1), + } +} + +fn column_stat(json: &str) -> Result { + let stats: ColumnStats = serde_json::from_str(json) + .map_err(|err| ErrorCode::Internal(format!("invalid column statistics JSON: {err}")))?; + Ok(stats.to_basic_column_statistics()) +} + +fn histogram_stat(json: &str) -> Result { + let stats: HistogramStats = serde_json::from_str(json) + .map_err(|err| ErrorCode::Internal(format!("invalid histogram statistics JSON: {err}")))?; + stats.to_histogram() +} diff --git a/src/query/sql/tests/it/optimizer/union_all.rs b/src/query/sql/tests/it/optimizer/union_all.rs new file mode 100644 index 0000000000000..4830f9398adbe --- /dev/null +++ b/src/query/sql/tests/it/optimizer/union_all.rs @@ -0,0 +1,479 @@ +// Copyright 2021 Datafuse Labs +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::HashMap; +use std::io::Write; +use std::sync::Arc; + +use databend_common_catalog::BasicColumnStatistics; +use databend_common_exception::ErrorCode; +use databend_common_exception::Result; +use databend_common_sql::ColumnEntry; +use databend_common_sql::FormatOptions; +use databend_common_sql::Metadata; +use databend_common_sql::MetadataRef; +use databend_common_sql::Symbol; +use databend_common_sql::optimizer::ir::RelExpr; +use databend_common_sql::optimizer::ir::SExpr; +use databend_common_sql::optimizer::ir::StatInfo; +use databend_common_sql::plans::Operator; +use databend_common_sql::plans::Plan; +use databend_common_sql::plans::RelOp; +use databend_common_statistics::Histogram; + +use super::column_stat; +use super::histogram_stat; +use super::table_statistics; +use crate::framework::LiteTableContext; +use crate::framework::golden::open_golden_file; +use crate::framework::golden::write_case_title; + +struct TableInput { + ddl: &'static str, + rows: u64, + column_stats: HashMap, + histograms: HashMap, +} + +struct UnionCase { + name: &'static str, + description: &'static str, + sql: &'static str, + tables: Vec, + targets: Vec, +} + +struct StatTarget { + // Child directions starting from the optimized Plan::Query SExpr root. + path: Vec, + operator: RelOp, +} + +#[derive(Clone, Copy, Debug)] +enum Child { + Left, + Right, +} + +impl Child { + fn index(self) -> usize { + match self { + Self::Left => 0, + Self::Right => 1, + } + } +} + +fn collect_operator_paths( + expr: &SExpr, + operator: &RelOp, + path: &mut Vec, + paths: &mut Vec>, +) { + if &expr.plan().rel_op() == operator { + paths.push(path.clone()); + } + + for (index, child) in expr.children().enumerate() { + path.push(match index { + 0 => Child::Left, + 1 => Child::Right, + _ => unreachable!("SExpr nodes have at most two children"), + }); + collect_operator_paths(child, operator, path, paths); + path.pop(); + } +} + +fn possible_operator_paths(root: &SExpr, operator: &RelOp) -> Vec> { + let mut paths = Vec::new(); + collect_operator_paths(root, operator, &mut Vec::new(), &mut paths); + paths +} + +fn invalid_target_path( + root: &SExpr, + target: &StatTarget, + reason: impl std::fmt::Display, +) -> ErrorCode { + ErrorCode::Internal(format!( + "cannot resolve {:?} target at path {:?}: {}; possible paths for this operator: {:?}", + target.operator, + target.path, + reason, + possible_operator_paths(root, &target.operator), + )) +} + +async fn register_table(ctx: &Arc, input: TableInput) -> Result<()> { + ctx.register_table_sql_with_stats( + input.ddl, + Some(table_statistics(input.rows)), + input.column_stats, + input.histograms, + ) + .await +} + +fn column_label(metadata: &Metadata, column: Symbol) -> String { + let id = column.as_usize(); + match metadata.column(column) { + ColumnEntry::BaseTableColumn(column) => { + let table = metadata.table(column.table_index); + format!("{}.{} (#{id})", table.name(), column.column_name) + } + entry => format!("{} (#{id})", entry.name()), + } +} + +fn write_stat_info(file: &mut impl Write, metadata: &Metadata, stat_info: &StatInfo) -> Result<()> { + writeln!(file, "cardinality: {:.3}", stat_info.cardinality)?; + writeln!( + file, + "precise_cardinality: {:?}", + stat_info.statistics.precise_cardinality + )?; + + let mut column_stats = stat_info.statistics.column_stats.iter().collect::>(); + column_stats.sort_by_key(|(column, _)| **column); + if column_stats.is_empty() { + writeln!(file, "column_stats: none")?; + return Ok(()); + } + + for (column, stat) in column_stats { + writeln!( + file, + "column_stat: {} min={}, max={}, ndv={:?}, null_count={:?}, histogram={}", + column_label(metadata, *column), + stat.min, + stat.max, + stat.ndv, + stat.null_count, + if stat.histogram.is_some() { + "some" + } else { + "none" + } + )?; + } + Ok(()) +} + +fn format_node(metadata: &MetadataRef, expr: &SExpr) -> Result { + Plan::Query { + s_expr: Box::new(expr.clone()), + metadata: metadata.clone(), + bind_context: Default::default(), + rewrite_kind: None, + formatted_ast: None, + ignore_result: false, + } + .format_indent(FormatOptions::default()) +} + +fn write_derived_stats( + file: &mut impl Write, + metadata: &MetadataRef, + root: &SExpr, + target: &StatTarget, +) -> Result<()> { + let mut expr = root; + for (depth, child) in target.path.iter().enumerate() { + let index = child.index(); + expr = expr.child(index).map_err(|_| { + invalid_target_path( + root, + target, + format!( + "child index {} does not exist at node {:?}, which has {} children", + index, + &target.path[..depth], + expr.arity() + ), + ) + })?; + } + + let actual = expr.plan().rel_op(); + if actual != target.operator { + return Err(invalid_target_path( + root, + target, + format!("found {actual:?} instead"), + )); + } + + writeln!(file, "path: {:?}", target.path)?; + writeln!(file, "node:")?; + writeln!(file, "{}", format_node(metadata, expr)?)?; + let stat_info = RelExpr::with_s_expr(expr).derive_cardinality()?; + write_stat_info(file, &metadata.read(), &stat_info)?; + Ok(()) +} + +async fn write_case(file: &mut impl Write, case: UnionCase) -> Result<()> { + let ctx = LiteTableContext::create().await?; + for table in case.tables { + register_table(&ctx, table).await?; + } + + let raw_plan = ctx.bind_sql(case.sql).await?; + let Plan::Query { + s_expr, metadata, .. + } = ctx.optimize_plan(raw_plan).await? + else { + unreachable!("UNION ALL query should bind to a query plan"); + }; + + write_case_title(file, case.name, case.description)?; + writeln!(file, "sql: {}", case.sql)?; + for target in &case.targets { + write_derived_stats(file, &metadata, &s_expr, target)?; + } + writeln!(file)?; + Ok(()) +} + +fn table(ddl: &'static str, rows: u64, column: Option) -> TableInput { + TableInput { + ddl, + rows, + column_stats: column + .map(|column| HashMap::from([("k".to_string(), column)])) + .unwrap_or_default(), + histograms: HashMap::new(), + } +} + +fn cases() -> Result> { + Ok(vec![ + UnionCase { + name: "lossless_coercion", + description: "SQL binding inserts a nullable INT to BIGINT coercion; after the cast drops the histogram, disjoint ranges still make the union NDV additive.", + sql: "SELECT k FROM l UNION ALL SELECT k FROM r", + tables: vec![ + table( + "CREATE TABLE l(k INT NULL)", + 10, + Some(column_stat( + r#"{"min": 1, "max": 3, "ndv": 3, "null_count": 1}"#, + )?), + ), + table( + "CREATE TABLE r(k BIGINT NULL)", + 20, + Some(column_stat( + r#"{"min": 4, "max": 8, "ndv": 5, "null_count": 2}"#, + )?), + ), + ], + targets: vec![StatTarget { + path: vec![], + operator: RelOp::UnionAll, + }], + }, + UnionCase { + name: "finite_range_reduces_union_ndv_bound", + description: "UnionAll reduces its conservative NDV upper bound to the merged finite range before an outer cast consumes the statistics.", + sql: "SELECT CAST(k AS BIGINT) AS k FROM (SELECT k FROM l UNION ALL SELECT k FROM r) AS u", + tables: vec![ + table( + "CREATE TABLE l(k INT NOT NULL)", + 2, + Some(column_stat( + r#"{"min": 1, "max": 2, "ndv": 2, "null_count": 0}"#, + )?), + ), + table( + "CREATE TABLE r(k INT NOT NULL)", + 1, + Some(column_stat( + r#"{"min": 1, "max": 1, "ndv": 1, "null_count": 0}"#, + )?), + ), + ], + targets: vec![StatTarget { + path: vec![], + operator: RelOp::EvalScalar, + }], + }, + UnionCase { + name: "nonempty_branch_without_column_stats", + description: "A non-empty branch with unknown column statistics makes the UnionAll output column statistics unknown.", + sql: "SELECT k FROM l UNION ALL SELECT k FROM r", + tables: vec![ + table( + "CREATE TABLE l(k BIGINT NOT NULL)", + 10, + Some(column_stat( + r#"{"min": 1, "max": 3, "ndv": 3, "null_count": 1}"#, + )?), + ), + table("CREATE TABLE r(k BIGINT NOT NULL)", 20, None), + ], + targets: vec![StatTarget { + path: vec![], + operator: RelOp::UnionAll, + }], + }, + UnionCase { + name: "empty_branch_without_column_stats", + description: "An exactly empty branch contributes no values, so UnionAll preserves the other branch statistics.", + sql: "SELECT k FROM l UNION ALL SELECT k FROM r", + tables: vec![ + table( + "CREATE TABLE l(k BIGINT NOT NULL)", + 10, + Some(column_stat( + r#"{"min": 1, "max": 3, "ndv": 3, "null_count": 1}"#, + )?), + ), + table("CREATE TABLE r(k BIGINT NOT NULL)", 0, None), + ], + targets: vec![StatTarget { + path: vec![], + operator: RelOp::UnionAll, + }], + }, + UnionCase { + name: "one_sided_expected_ndv", + description: "UnionAll preserves the known expected NDV when the other branch only has an upper bound.", + sql: "SELECT k FROM l UNION ALL SELECT k FROM r", + tables: vec![ + table( + "CREATE TABLE l(k BIGINT NOT NULL)", + 10, + Some(column_stat( + r#"{"min": 1, "max": 3, "ndv": null, "null_count": 0}"#, + )?), + ), + table( + "CREATE TABLE r(k BIGINT NOT NULL)", + 20, + Some(column_stat( + r#"{"min": 2, "max": 8, "ndv": 5, "null_count": 0}"#, + )?), + ), + ], + targets: vec![StatTarget { + path: vec![], + operator: RelOp::UnionAll, + }], + }, + UnionCase { + name: "histogram_overlap", + description: "Overlapping input histograms refine the merged NDV while the output histogram is dropped.", + sql: "SELECT k FROM l UNION ALL SELECT k FROM r", + tables: vec![ + TableInput { + ddl: "CREATE TABLE l(k BIGINT NOT NULL)", + rows: 100, + column_stats: HashMap::from([( + "k".to_string(), + column_stat(r#"{"min": 0, "max": 9, "ndv": 10, "null_count": 0}"#)?, + )]), + histograms: HashMap::from([( + "k".to_string(), + histogram_stat( + r#"{ + "accuracy": false, + "buckets": [ + {"lower_bound": {"Int": 0}, "upper_bound": {"Int": 4}, "num_values": 50.0, "num_distinct": 5.0}, + {"lower_bound": {"Int": 5}, "upper_bound": {"Int": 9}, "num_values": 50.0, "num_distinct": 5.0} + ], + "avg_spacing": 4.5 + }"#, + )?, + )]), + }, + TableInput { + ddl: "CREATE TABLE r(k BIGINT NOT NULL)", + rows: 100, + column_stats: HashMap::from([( + "k".to_string(), + column_stat(r#"{"min": 5, "max": 14, "ndv": 10, "null_count": 0}"#)?, + )]), + histograms: HashMap::from([( + "k".to_string(), + histogram_stat( + r#"{ + "accuracy": false, + "buckets": [ + {"lower_bound": {"Int": 5}, "upper_bound": {"Int": 9}, "num_values": 50.0, "num_distinct": 5.0}, + {"lower_bound": {"Int": 10}, "upper_bound": {"Int": 14}, "num_values": 50.0, "num_distinct": 5.0} + ], + "avg_spacing": 4.5 + }"#, + )?, + )]), + }, + ], + targets: vec![StatTarget { + path: vec![], + operator: RelOp::UnionAll, + }], + }, + UnionCase { + name: "join_consumes_union_stats", + description: "A join above the SQL-derived UnionAll consumes its merged output NDV instead of treating the key as unknown.", + sql: "SELECT u.k FROM (SELECT k FROM l UNION ALL SELECT k FROM r) AS u JOIN j ON u.k = j.k", + tables: vec![ + table( + "CREATE TABLE l(k BIGINT NOT NULL)", + 10, + Some(column_stat( + r#"{"min": 1, "max": 3, "ndv": 3, "null_count": 0}"#, + )?), + ), + table( + "CREATE TABLE r(k BIGINT NOT NULL)", + 20, + Some(column_stat( + r#"{"min": 4, "max": 8, "ndv": 5, "null_count": 0}"#, + )?), + ), + table( + "CREATE TABLE j(k BIGINT NOT NULL)", + 100, + Some(column_stat( + r#"{"min": 1, "max": 10, "ndv": 10, "null_count": 0}"#, + )?), + ), + ], + targets: vec![ + StatTarget { + path: vec![Child::Left], + operator: RelOp::Join, + }, + StatTarget { + path: vec![Child::Left, Child::Right], + operator: RelOp::UnionAll, + }, + ], + }, + ]) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn test_union_all_statistics_golden() -> Result<()> { + let mut output = Vec::new(); + for case in cases()? { + write_case(&mut output, case).await?; + } + + let mut file = open_golden_file("optimizer", "union_all.txt")?; + file.write_all(&output)?; + Ok(()) +} diff --git a/src/query/sql/tests/it/optimizer/union_all.txt b/src/query/sql/tests/it/optimizer/union_all.txt new file mode 100644 index 0000000000000..d80bd3ef2fabe --- /dev/null +++ b/src/query/sql/tests/it/optimizer/union_all.txt @@ -0,0 +1,217 @@ +=== lossless_coercion === +description: SQL binding inserts a nullable INT to BIGINT coercion; after the cast drops the histogram, disjoint ranges still make the union NDV additive. +sql: SELECT k FROM l UNION ALL SELECT k FROM r +path: [] +node: +UnionAll +├── output: [k (#2)] +├── left: [CAST(l.k (#0) AS Int64 NULL)] +├── right: [r.k (#1)] +├── cte_scan_names: [] +├── logical_recursive_cte_id: None +├── Scan +│ ├── table: default.l (#0) +│ ├── filters: [] +│ ├── order by: [] +│ └── limit: NONE +└── Scan + ├── table: default.r (#1) + ├── filters: [] + ├── order by: [] + └── limit: NONE + +cardinality: 30.000 +precise_cardinality: Some(30) +column_stat: k (#2) min=1, max=8, ndv=8.0, null_count=3, histogram=none + +=== finite_range_reduces_union_ndv_bound === +description: UnionAll reduces its conservative NDV upper bound to the merged finite range before an outer cast consumes the statistics. +sql: SELECT CAST(k AS BIGINT) AS k FROM (SELECT k FROM l UNION ALL SELECT k FROM r) AS u +path: [] +node: +EvalScalar +├── scalars: [CAST(k (#2) AS Int64) AS (#3)] +└── UnionAll + ├── output: [k (#2)] + ├── left: [l.k (#0)] + ├── right: [r.k (#1)] + ├── cte_scan_names: [] + ├── logical_recursive_cte_id: None + ├── Scan + │ ├── table: default.l (#0) + │ ├── filters: [] + │ ├── order by: [] + │ └── limit: NONE + └── Scan + ├── table: default.r (#1) + ├── filters: [] + ├── order by: [] + └── limit: NONE + +cardinality: 3.000 +precise_cardinality: Some(3) +column_stat: k (#2) min=1, max=2, ndv=2.0, null_count=0, histogram=none +column_stat: k (#3) min=1, max=2, ndv=2.0, null_count=0, histogram=none + +=== nonempty_branch_without_column_stats === +description: A non-empty branch with unknown column statistics makes the UnionAll output column statistics unknown. +sql: SELECT k FROM l UNION ALL SELECT k FROM r +path: [] +node: +UnionAll +├── output: [k (#2)] +├── left: [l.k (#0)] +├── right: [r.k (#1)] +├── cte_scan_names: [] +├── logical_recursive_cte_id: None +├── Scan +│ ├── table: default.l (#0) +│ ├── filters: [] +│ ├── order by: [] +│ └── limit: NONE +└── Scan + ├── table: default.r (#1) + ├── filters: [] + ├── order by: [] + └── limit: NONE + +cardinality: 30.000 +precise_cardinality: Some(30) +column_stats: none + +=== empty_branch_without_column_stats === +description: An exactly empty branch contributes no values, so UnionAll preserves the other branch statistics. +sql: SELECT k FROM l UNION ALL SELECT k FROM r +path: [] +node: +UnionAll +├── output: [k (#2)] +├── left: [l.k (#0)] +├── right: [r.k (#1)] +├── cte_scan_names: [] +├── logical_recursive_cte_id: None +├── Scan +│ ├── table: default.l (#0) +│ ├── filters: [] +│ ├── order by: [] +│ └── limit: NONE +└── Scan + ├── table: default.r (#1) + ├── filters: [] + ├── order by: [] + └── limit: NONE + +cardinality: 10.000 +precise_cardinality: Some(10) +column_stat: k (#2) min=1, max=3, ndv=3.0, null_count=1, histogram=some + +=== one_sided_expected_ndv === +description: UnionAll preserves the known expected NDV when the other branch only has an upper bound. +sql: SELECT k FROM l UNION ALL SELECT k FROM r +path: [] +node: +UnionAll +├── output: [k (#2)] +├── left: [l.k (#0)] +├── right: [r.k (#1)] +├── cte_scan_names: [] +├── logical_recursive_cte_id: None +├── Scan +│ ├── table: default.l (#0) +│ ├── filters: [] +│ ├── order by: [] +│ └── limit: NONE +└── Scan + ├── table: default.r (#1) + ├── filters: [] + ├── order by: [] + └── limit: NONE + +cardinality: 30.000 +precise_cardinality: Some(30) +column_stat: k (#2) min=1, max=8, ndv=~5.0[..8.0], null_count=0, histogram=none + +=== histogram_overlap === +description: Overlapping input histograms refine the merged NDV while the output histogram is dropped. +sql: SELECT k FROM l UNION ALL SELECT k FROM r +path: [] +node: +UnionAll +├── output: [k (#2)] +├── left: [l.k (#0)] +├── right: [r.k (#1)] +├── cte_scan_names: [] +├── logical_recursive_cte_id: None +├── Scan +│ ├── table: default.l (#0) +│ ├── filters: [] +│ ├── order by: [] +│ └── limit: NONE +└── Scan + ├── table: default.r (#1) + ├── filters: [] + ├── order by: [] + └── limit: NONE + +cardinality: 200.000 +precise_cardinality: Some(200) +column_stat: k (#2) min=0, max=14, ndv=15.0, null_count=0, histogram=none + +=== join_consumes_union_stats === +description: A join above the SQL-derived UnionAll consumes its merged output NDV instead of treating the key as unknown. +sql: SELECT u.k FROM (SELECT k FROM l UNION ALL SELECT k FROM r) AS u JOIN j ON u.k = j.k +path: [Left] +node: +Join(Inner) +├── build keys: [k (#2)] +├── probe keys: [j.k (#3)] +├── other filters: [] +├── UnionAll +│ ├── output: [k (#2)] +│ ├── left: [l.k (#0)] +│ ├── right: [r.k (#1)] +│ ├── cte_scan_names: [] +│ ├── logical_recursive_cte_id: None +│ ├── Scan +│ │ ├── table: default.l (#0) +│ │ ├── filters: [] +│ │ ├── order by: [] +│ │ └── limit: NONE +│ └── Scan +│ ├── table: default.r (#1) +│ ├── filters: [] +│ ├── order by: [] +│ └── limit: NONE +└── Scan + ├── table: default.j (#2) + ├── filters: [] + ├── order by: [] + └── limit: NONE + +cardinality: 300.000 +precise_cardinality: None +column_stat: k (#2) min=1, max=8, ndv=8.0, null_count=0, histogram=none +column_stat: j.k (#3) min=1, max=8, ndv=8.0, null_count=0, histogram=none +path: [Left, Right] +node: +UnionAll +├── output: [k (#2)] +├── left: [l.k (#0)] +├── right: [r.k (#1)] +├── cte_scan_names: [] +├── logical_recursive_cte_id: None +├── Scan +│ ├── table: default.l (#0) +│ ├── filters: [] +│ ├── order by: [] +│ └── limit: NONE +└── Scan + ├── table: default.r (#1) + ├── filters: [] + ├── order by: [] + └── limit: NONE + +cardinality: 30.000 +precise_cardinality: Some(30) +column_stat: k (#2) min=1, max=8, ndv=8.0, null_count=0, histogram=none + diff --git a/tests/sqllogictests/suites/mode/standalone/explain/join_reorder/chain.test b/tests/sqllogictests/suites/mode/standalone/explain/join_reorder/chain.test index 104e3be3adc72..6e722ccd3aa3d 100644 --- a/tests/sqllogictests/suites/mode/standalone/explain/join_reorder/chain.test +++ b/tests/sqllogictests/suites/mode/standalone/explain/join_reorder/chain.test @@ -451,7 +451,7 @@ HashJoin ├── probe keys: [t1.a (#2)] ├── keys is null equal: [false] ├── filters: [] -├── estimated rows: 10.00 +├── estimated rows: 1.00 ├── TableScan(Build) │ ├── table: default.join_reorder.t │ ├── scan id: 0