Experiment: Single mode + hack for low cardinality

benchmarks/queries/clickbench/queries.sql

@@ -1,43 +1,2 @@
-SELECT COUNT(*) FROM hits;
-SELECT COUNT(*) FROM hits WHERE "AdvEngineID" <> 0;
-SELECT SUM("AdvEngineID"), COUNT(*), AVG("ResolutionWidth") FROM hits;
-SELECT AVG("UserID") FROM hits;
-SELECT COUNT(DISTINCT "UserID") FROM hits;
-SELECT COUNT(DISTINCT "SearchPhrase") FROM hits;
-SELECT MIN("EventDate"::INT::DATE), MAX("EventDate"::INT::DATE) FROM hits;
-SELECT "AdvEngineID", COUNT(*) FROM hits WHERE "AdvEngineID" <> 0 GROUP BY "AdvEngineID" ORDER BY COUNT(*) DESC;
-SELECT "RegionID", COUNT(DISTINCT "UserID") AS u FROM hits GROUP BY "RegionID" ORDER BY u DESC LIMIT 10;
-SELECT "RegionID", SUM("AdvEngineID"), COUNT(*) AS c, AVG("ResolutionWidth"), COUNT(DISTINCT "UserID") FROM hits GROUP BY "RegionID" ORDER BY c DESC LIMIT 10;
-SELECT "MobilePhoneModel", COUNT(DISTINCT "UserID") AS u FROM hits WHERE "MobilePhoneModel" <> '' GROUP BY "MobilePhoneModel" ORDER BY u DESC LIMIT 10;
-SELECT "MobilePhone", "MobilePhoneModel", COUNT(DISTINCT "UserID") AS u FROM hits WHERE "MobilePhoneModel" <> '' GROUP BY "MobilePhone", "MobilePhoneModel" ORDER BY u DESC LIMIT 10;
-SELECT "SearchPhrase", COUNT(*) AS c FROM hits WHERE "SearchPhrase" <> '' GROUP BY "SearchPhrase" ORDER BY c DESC LIMIT 10;
-SELECT "SearchPhrase", COUNT(DISTINCT "UserID") AS u FROM hits WHERE "SearchPhrase" <> '' GROUP BY "SearchPhrase" ORDER BY u DESC LIMIT 10;
-SELECT "SearchEngineID", "SearchPhrase", COUNT(*) AS c FROM hits WHERE "SearchPhrase" <> '' GROUP BY "SearchEngineID", "SearchPhrase" ORDER BY c DESC LIMIT 10;
-SELECT "UserID", COUNT(*) FROM hits GROUP BY "UserID" ORDER BY COUNT(*) DESC LIMIT 10;
 SELECT "UserID", "SearchPhrase", COUNT(*) FROM hits GROUP BY "UserID", "SearchPhrase" ORDER BY COUNT(*) DESC LIMIT 10;
-SELECT "UserID", "SearchPhrase", COUNT(*) FROM hits GROUP BY "UserID", "SearchPhrase" LIMIT 10;
-SELECT "UserID", extract(minute FROM to_timestamp_seconds("EventTime")) AS m, "SearchPhrase", COUNT(*) FROM hits GROUP BY "UserID", m, "SearchPhrase" ORDER BY COUNT(*) DESC LIMIT 10;
-SELECT "UserID" FROM hits WHERE "UserID" = 435090932899640449;
-SELECT COUNT(*) FROM hits WHERE "URL" LIKE '%google%';
-SELECT "SearchPhrase", MIN("URL"), COUNT(*) AS c FROM hits WHERE "URL" LIKE '%google%' AND "SearchPhrase" <> '' GROUP BY "SearchPhrase" ORDER BY c DESC LIMIT 10;
-SELECT "SearchPhrase", MIN("URL"), MIN("Title"), COUNT(*) AS c, COUNT(DISTINCT "UserID") FROM hits WHERE "Title" LIKE '%Google%' AND "URL" NOT LIKE '%.google.%' AND "SearchPhrase" <> '' GROUP BY "SearchPhrase" ORDER BY c DESC LIMIT 10;
-SELECT * FROM hits WHERE "URL" LIKE '%google%' ORDER BY to_timestamp_seconds("EventTime") LIMIT 10;
-SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY to_timestamp_seconds("EventTime") LIMIT 10;
-SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
-SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY to_timestamp_seconds("EventTime"), "SearchPhrase" LIMIT 10;
-SELECT "CounterID", AVG(length("URL")) AS l, COUNT(*) AS c FROM hits WHERE "URL" <> '' GROUP BY "CounterID" HAVING COUNT(*) > 100000 ORDER BY l DESC LIMIT 25;
-SELECT REGEXP_REPLACE("Referer", '^https?://(?:www\.)?([^/]+)/.*$', '\1') AS k, AVG(length("Referer")) AS l, COUNT(*) AS c, MIN("Referer") FROM hits WHERE "Referer" <> '' GROUP BY k HAVING COUNT(*) > 100000 ORDER BY l DESC LIMIT 25;
-SELECT SUM("ResolutionWidth"), SUM("ResolutionWidth" + 1), SUM("ResolutionWidth" + 2), SUM("ResolutionWidth" + 3), SUM("ResolutionWidth" + 4), SUM("ResolutionWidth" + 5), SUM("ResolutionWidth" + 6), SUM("ResolutionWidth" + 7), SUM("ResolutionWidth" + 8), SUM("ResolutionWidth" + 9), SUM("ResolutionWidth" + 10), SUM("ResolutionWidth" + 11), SUM("ResolutionWidth" + 12), SUM("ResolutionWidth" + 13), SUM("ResolutionWidth" + 14), SUM("ResolutionWidth" + 15), SUM("ResolutionWidth" + 16), SUM("ResolutionWidth" + 17), SUM("ResolutionWidth" + 18), SUM("ResolutionWidth" + 19), SUM("ResolutionWidth" + 20), SUM("ResolutionWidth" + 21), SUM("ResolutionWidth" + 22), SUM("ResolutionWidth" + 23), SUM("ResolutionWidth" + 24), SUM("ResolutionWidth" + 25), SUM("ResolutionWidth" + 26), SUM("ResolutionWidth" + 27), SUM("ResolutionWidth" + 28), SUM("ResolutionWidth" + 29), SUM("ResolutionWidth" + 30), SUM("ResolutionWidth" + 31), SUM("ResolutionWidth" + 32), SUM("ResolutionWidth" + 33), SUM("ResolutionWidth" + 34), SUM("ResolutionWidth" + 35), SUM("ResolutionWidth" + 36), SUM("ResolutionWidth" + 37), SUM("ResolutionWidth" + 38), SUM("ResolutionWidth" + 39), SUM("ResolutionWidth" + 40), SUM("ResolutionWidth" + 41), SUM("ResolutionWidth" + 42), SUM("ResolutionWidth" + 43), SUM("ResolutionWidth" + 44), SUM("ResolutionWidth" + 45), SUM("ResolutionWidth" + 46), SUM("ResolutionWidth" + 47), SUM("ResolutionWidth" + 48), SUM("ResolutionWidth" + 49), SUM("ResolutionWidth" + 50), SUM("ResolutionWidth" + 51), SUM("ResolutionWidth" + 52), SUM("ResolutionWidth" + 53), SUM("ResolutionWidth" + 54), SUM("ResolutionWidth" + 55), SUM("ResolutionWidth" + 56), SUM("ResolutionWidth" + 57), SUM("ResolutionWidth" + 58), SUM("ResolutionWidth" + 59), SUM("ResolutionWidth" + 60), SUM("ResolutionWidth" + 61), SUM("ResolutionWidth" + 62), SUM("ResolutionWidth" + 63), SUM("ResolutionWidth" + 64), SUM("ResolutionWidth" + 65), SUM("ResolutionWidth" + 66), SUM("ResolutionWidth" + 67), SUM("ResolutionWidth" + 68), SUM("ResolutionWidth" + 69), SUM("ResolutionWidth" + 70), SUM("ResolutionWidth" + 71), SUM("ResolutionWidth" + 72), SUM("ResolutionWidth" + 73), SUM("ResolutionWidth" + 74), SUM("ResolutionWidth" + 75), SUM("ResolutionWidth" + 76), SUM("ResolutionWidth" + 77), SUM("ResolutionWidth" + 78), SUM("ResolutionWidth" + 79), SUM("ResolutionWidth" + 80), SUM("ResolutionWidth" + 81), SUM("ResolutionWidth" + 82), SUM("ResolutionWidth" + 83), SUM("ResolutionWidth" + 84), SUM("ResolutionWidth" + 85), SUM("ResolutionWidth" + 86), SUM("ResolutionWidth" + 87), SUM("ResolutionWidth" + 88), SUM("ResolutionWidth" + 89) FROM hits;
-SELECT "SearchEngineID", "ClientIP", COUNT(*) AS c, SUM("IsRefresh"), AVG("ResolutionWidth") FROM hits WHERE "SearchPhrase" <> '' GROUP BY "SearchEngineID", "ClientIP" ORDER BY c DESC LIMIT 10;
-SELECT "WatchID", "ClientIP", COUNT(*) AS c, SUM("IsRefresh"), AVG("ResolutionWidth") FROM hits WHERE "SearchPhrase" <> '' GROUP BY "WatchID", "ClientIP" ORDER BY c DESC LIMIT 10;
-SELECT "WatchID", "ClientIP", COUNT(*) AS c, SUM("IsRefresh"), AVG("ResolutionWidth") FROM hits GROUP BY "WatchID", "ClientIP" ORDER BY c DESC LIMIT 10;
-SELECT "URL", COUNT(*) AS c FROM hits GROUP BY "URL" ORDER BY c DESC LIMIT 10;
-SELECT 1, "URL", COUNT(*) AS c FROM hits GROUP BY 1, "URL" ORDER BY c DESC LIMIT 10;
-SELECT "ClientIP", "ClientIP" - 1, "ClientIP" - 2, "ClientIP" - 3, COUNT(*) AS c FROM hits GROUP BY "ClientIP", "ClientIP" - 1, "ClientIP" - 2, "ClientIP" - 3 ORDER BY c DESC LIMIT 10;
-SELECT "URL", COUNT(*) AS PageViews FROM hits WHERE "CounterID" = 62 AND "EventDate"::INT::DATE >= '2013-07-01' AND "EventDate"::INT::DATE <= '2013-07-31' AND "DontCountHits" = 0 AND "IsRefresh" = 0 AND "URL" <> '' GROUP BY "URL" ORDER BY PageViews DESC LIMIT 10;
-SELECT "Title", COUNT(*) AS PageViews FROM hits WHERE "CounterID" = 62 AND "EventDate"::INT::DATE >= '2013-07-01' AND "EventDate"::INT::DATE <= '2013-07-31' AND "DontCountHits" = 0 AND "IsRefresh" = 0 AND "Title" <> '' GROUP BY "Title" ORDER BY PageViews DESC LIMIT 10;
-SELECT "URL", COUNT(*) AS PageViews FROM hits WHERE "CounterID" = 62 AND "EventDate"::INT::DATE >= '2013-07-01' AND "EventDate"::INT::DATE <= '2013-07-31' AND "IsRefresh" = 0 AND "IsLink" <> 0 AND "IsDownload" = 0 GROUP BY "URL" ORDER BY PageViews DESC LIMIT 10 OFFSET 1000;
-SELECT "TraficSourceID", "SearchEngineID", "AdvEngineID", CASE WHEN ("SearchEngineID" = 0 AND "AdvEngineID" = 0) THEN "Referer" ELSE '' END AS Src, "URL" AS Dst, COUNT(*) AS PageViews FROM hits WHERE "CounterID" = 62 AND "EventDate"::INT::DATE >= '2013-07-01' AND "EventDate"::INT::DATE <= '2013-07-31' AND "IsRefresh" = 0 GROUP BY "TraficSourceID", "SearchEngineID", "AdvEngineID", Src, Dst ORDER BY PageViews DESC LIMIT 10 OFFSET 1000;
-SELECT "URLHash", "EventDate"::INT::DATE, COUNT(*) AS PageViews FROM hits WHERE "CounterID" = 62 AND "EventDate"::INT::DATE >= '2013-07-01' AND "EventDate"::INT::DATE <= '2013-07-31' AND "IsRefresh" = 0 AND "TraficSourceID" IN (-1, 6) AND "RefererHash" = 3594120000172545465 GROUP BY "URLHash", "EventDate"::INT::DATE ORDER BY PageViews DESC LIMIT 10 OFFSET 100;
-SELECT "WindowClientWidth", "WindowClientHeight", COUNT(*) AS PageViews FROM hits WHERE "CounterID" = 62 AND "EventDate"::INT::DATE >= '2013-07-01' AND "EventDate"::INT::DATE <= '2013-07-31' AND "IsRefresh" = 0 AND "DontCountHits" = 0 AND "URLHash" = 2868770270353813622 GROUP BY "WindowClientWidth", "WindowClientHeight" ORDER BY PageViews DESC LIMIT 10 OFFSET 10000;
-SELECT DATE_TRUNC('minute', to_timestamp_seconds("EventTime")) AS M, COUNT(*) AS PageViews FROM hits WHERE "CounterID" = 62 AND "EventDate"::INT::DATE >= '2013-07-14' AND "EventDate"::INT::DATE <= '2013-07-15' AND "IsRefresh" = 0 AND "DontCountHits" = 0 GROUP BY DATE_TRUNC('minute', to_timestamp_seconds("EventTime")) ORDER BY DATE_TRUNC('minute', M) LIMIT 10 OFFSET 1000;
+SELECT "UserID", "SearchPhrase", COUNT(*) FROM hits GROUP BY "UserID", "SearchPhrase" LIMIT 10;

datafusion/core/src/physical_optimizer/optimizer.rs

@@ -25,7 +25,7 @@ use super::update_aggr_exprs::OptimizeAggregateOrder;
 use crate::physical_optimizer::aggregate_statistics::AggregateStatistics;
 use crate::physical_optimizer::coalesce_batches::CoalesceBatches;
 use crate::physical_optimizer::combine_partial_final_agg::CombinePartialFinalAggregate;
-use crate::physical_optimizer::enforce_distribution::EnforceDistribution;
+// use crate::physical_optimizer::enforce_distribution::EnforceDistribution;
 use crate::physical_optimizer::enforce_sorting::EnforceSorting;
 use crate::physical_optimizer::join_selection::JoinSelection;
 use crate::physical_optimizer::limit_pushdown::LimitPushdown;
@@ -69,7 +69,7 @@ impl PhysicalOptimizer {
  // requirements. Please make sure that the whole plan tree is determined before this rule.
  // This rule increases parallelism if doing so is beneficial to the physical plan; i.e. at
  // least one of the operators in the plan benefits from increased parallelism.
- Arc::new(EnforceDistribution::new()),
+ // Arc::new(EnforceDistribution::new()),
  // The CombinePartialFinalAggregate rule should be applied after the EnforceDistribution rule
  Arc::new(CombinePartialFinalAggregate::new()),
  // The EnforceSorting rule is for adding essential local sorting to satisfy the required

datafusion/core/src/physical_optimizer/sanity_checker.rs

@@ -137,17 +137,17 @@ pub fn check_plan_sanity(
  }
  }
 
- if !child
- .output_partitioning()
- .satisfy(child_dist_req, child_eq_props)
- {
- let child_plan_str = get_plan_string(child);
- return plan_err!(
- "Child: {:?} does not satisfy parent distribution requirements: {:?}",
- child_plan_str,
- child_dist_req
- );
- }
+ // if !child
+ //  .output_partitioning()
+ //  .satisfy(child_dist_req, child_eq_props)
+ // {
+ //  let child_plan_str = get_plan_string(child);
+ //  return plan_err!(
+ //  "Child: {:?} does not satisfy parent distribution requirements: {:?}",
+ //  child_plan_str,
+ //  child_dist_req
+ //  );
+ // }
  }
 
  Ok(Transformed::no(plan))

datafusion/core/src/physical_planner.rs

@@ -698,55 +698,79 @@ impl DefaultPhysicalPlanner {
  let (aggregates, filters, _order_bys): (Vec<_>, Vec<_>, Vec<_>) =
  multiunzip(agg_filter);
 
- let initial_aggr = Arc::new(AggregateExec::try_new(
- AggregateMode::Partial,
- groups.clone(),
- aggregates,
- filters.clone(),
- input_exec,
- physical_input_schema.clone(),
- )?);
-
- // update group column indices based on partial aggregate plan evaluation
- let final_group: Vec<Arc<dyn PhysicalExpr>> =
- initial_aggr.output_group_expr();
-
- let can_repartition = !groups.is_empty()
- && session_state.config().target_partitions() > 1
- && session_state.config().repartition_aggregations();
-
- // Some aggregators may be modified during initialization for
- // optimization purposes. For example, a FIRST_VALUE may turn
- // into a LAST_VALUE with the reverse ordering requirement.
- // To reflect such changes to subsequent stages, use the updated
- // `AggregateFunctionExpr`/`PhysicalSortExpr` objects.
- let updated_aggregates = initial_aggr.aggr_expr().to_vec();
-
- let next_partition_mode = if can_repartition {
- // construct a second aggregation with 'AggregateMode::FinalPartitioned'
- AggregateMode::FinalPartitioned
+ // if false {
+ if group_expr.len() > 1 {
+ let can_repartition = !groups.is_empty()
+ && session_state.config().target_partitions() > 1
+ && session_state.config().repartition_aggregations();
+ let mode = if can_repartition {
+ // construct a second aggregation with 'AggregateMode::FinalPartitioned'
+ AggregateMode::SinglePartitioned
+ } else {
+ // construct a second aggregation, keeping the final column name equal to the
+ // first aggregation and the expressions corresponding to the respective aggregate
+ AggregateMode::Single
+ };
+ let aggr = AggregateExec::try_new(
+ mode,
+ groups.clone(),
+ aggregates.clone(),
+ filters.clone(),
+ input_exec.clone(),
+ physical_input_schema.clone(),
+ )?;
+ Arc::new(aggr)
  } else {
- // construct a second aggregation, keeping the final column name equal to the
- // first aggregation and the expressions corresponding to the respective aggregate
- AggregateMode::Final
- };
+ let initial_aggr = Arc::new(AggregateExec::try_new(
+ AggregateMode::Partial,
+ groups.clone(),
+ aggregates,
+ filters.clone(),
+ input_exec,
+ physical_input_schema.clone(),
+ )?);
 
- let final_grouping_set = PhysicalGroupBy::new_single(
- final_group
- .iter()
- .enumerate()
- .map(|(i, expr)| (expr.clone(), groups.expr()[i].1.clone()))
- .collect(),
- );
-
- Arc::new(AggregateExec::try_new(
- next_partition_mode,
- final_grouping_set,
- updated_aggregates,
- filters,
- initial_aggr,
- physical_input_schema.clone(),
- )?)
+ // update group column indices based on partial aggregate plan evaluation
+ let final_group: Vec<Arc<dyn PhysicalExpr>> =
+ initial_aggr.output_group_expr();
+
+ let can_repartition = !groups.is_empty()
+ && session_state.config().target_partitions() > 1
+ && session_state.config().repartition_aggregations();
+
+ // Some aggregators may be modified during initialization for
+ // optimization purposes. For example, a FIRST_VALUE may turn
+ // into a LAST_VALUE with the reverse ordering requirement.
+ // To reflect such changes to subsequent stages, use the updated
+ // `AggregateFunctionExpr`/`PhysicalSortExpr` objects.
+ let updated_aggregates = initial_aggr.aggr_expr().to_vec();
+
+ let next_partition_mode = if can_repartition {
+ // construct a second aggregation with 'AggregateMode::FinalPartitioned'
+ AggregateMode::FinalPartitioned
+ } else {
+ // construct a second aggregation, keeping the final column name equal to the
+ // first aggregation and the expressions corresponding to the respective aggregate
+ AggregateMode::Final
+ };
+
+ let final_grouping_set = PhysicalGroupBy::new_single(
+ final_group
+ .iter()
+ .enumerate()
+ .map(|(i, expr)| (expr.clone(), groups.expr()[i].1.clone()))
+ .collect(),
+ );
+
+ Arc::new(AggregateExec::try_new(
+ next_partition_mode,
+ final_grouping_set,
+ updated_aggregates,
+ filters,
+ initial_aggr,
+ physical_input_schema.clone(),
+ )?)
+ }
  }
  LogicalPlan::Projection(Projection { input, expr, .. }) => self
  .create_project_physical_exec(

datafusion/physical-plan/src/aggregates/group_values/row.rs

@@ -100,6 +100,9 @@ impl GroupValuesRows {
 
 impl GroupValues for GroupValuesRows {
  fn intern(&mut self, cols: &[ArrayRef], groups: &mut Vec<usize>) -> Result<()> {
+
+ // println!("cols: {:?}", cols[0].len());
+
  // Convert the group keys into the row format
  let group_rows = &mut self.rows_buffer;
  group_rows.clear();

datafusion/physical-plan/src/aggregates/row_hash.rs

@@ -762,6 +762,7 @@ impl GroupedHashAggregateStream {
  evaluate_optional(&self.filter_expressions, &batch)?
  };
 
+ // println!("called: {:?}", self.mode);
  for group_values in &group_by_values {
  // calculate the group indices for each input row
  let starting_num_groups = self.group_values.len();

datafusion/physical-plan/src/projection.rs

@@ -318,6 +318,7 @@ impl ProjectionStream {
  RecordBatch::try_new_with_options(Arc::clone(&self.schema), arrays, &options)
  .map_err(Into::into)
  } else {
+ // println!("arrays.len(): {:?}", arrays[0].len());
  RecordBatch::try_new(Arc::clone(&self.schema), arrays).map_err(Into::into)
  }
  }