From 54003fff2d736049ca35dab7fe9554ae96c5aa50 Mon Sep 17 00:00:00 2001 From: jayzhan211 Date: Thu, 1 Aug 2024 20:22:01 +0800 Subject: [PATCH 1/6] convert to single Signed-off-by: jayzhan211 --- datafusion/core/src/physical_planner.rs | 118 +++++++++++++++--------- 1 file changed, 72 insertions(+), 46 deletions(-) diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs index 329d343f13fc1..d934576d0ed74 100644 --- a/datafusion/core/src/physical_planner.rs +++ b/datafusion/core/src/physical_planner.rs @@ -956,55 +956,81 @@ 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.clone(), - filters.clone(), - input_exec, - physical_input_schema.clone(), - )?); - - // update group column indices based on partial aggregate plan evaluation - let final_group: Vec> = - 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 - // `AggregateExpr`/`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 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 = + Arc::new(AggregateExec::try_new( + AggregateMode::Partial, + groups.clone(), + aggregates.clone(), + 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(), - ); + // update group column indices based on partial aggregate plan evaluation + let final_group: Vec> = + initial_aggr.output_group_expr(); - Arc::new(AggregateExec::try_new( - next_partition_mode, - final_grouping_set, - updated_aggregates, - filters, - initial_aggr, - physical_input_schema.clone(), - )?) + 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 + // `AggregateExpr`/`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( From c427e75d879962e2f82823a8941e578a5fa4c589 Mon Sep 17 00:00:00 2001 From: jayzhan211 Date: Thu, 1 Aug 2024 20:40:17 +0800 Subject: [PATCH 2/6] queries Signed-off-by: jayzhan211 --- benchmarks/queries/clickbench/queries.sql | 43 +---------------------- 1 file changed, 1 insertion(+), 42 deletions(-) diff --git a/benchmarks/queries/clickbench/queries.sql b/benchmarks/queries/clickbench/queries.sql index 52e72e02e1e0d..d8816dbf0a3b7 100644 --- a/benchmarks/queries/clickbench/queries.sql +++ b/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", concat("SearchPhrase", repeat('hello', 20)) as s, COUNT(*) FROM hits GROUP BY "UserID", s ORDER BY COUNT(*) DESC LIMIT 10; \ No newline at end of file From 1abd5e13ab6bf6bd5a7df2250a47fb3ef799d8a9 Mon Sep 17 00:00:00 2001 From: jayzhan211 Date: Thu, 1 Aug 2024 21:09:56 +0800 Subject: [PATCH 3/6] reuse-hash benchmark Signed-off-by: jayzhan211 --- datafusion/core/Cargo.toml | 4 + datafusion/core/benches/reuse_hash.rs | 120 ++++++++++++++++++++++++++ 2 files changed, 124 insertions(+) create mode 100644 datafusion/core/benches/reuse_hash.rs diff --git a/datafusion/core/Cargo.toml b/datafusion/core/Cargo.toml index 0415c3164b38c..d4896a4c03754 100644 --- a/datafusion/core/Cargo.toml +++ b/datafusion/core/Cargo.toml @@ -180,6 +180,10 @@ name = "sort_limit_query_sql" harness = false name = "math_query_sql" +[[bench]] +harness = false +name = "reuse_hash" + [[bench]] harness = false name = "filter_query_sql" diff --git a/datafusion/core/benches/reuse_hash.rs b/datafusion/core/benches/reuse_hash.rs new file mode 100644 index 0000000000000..04dad9ec8d6ba --- /dev/null +++ b/datafusion/core/benches/reuse_hash.rs @@ -0,0 +1,120 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 arrow::{ + datatypes::{DataType, Field, Schema}, + record_batch::RecordBatch, +}; +use arrow_array::{Int64Array, StringArray}; +use criterion::{criterion_group, criterion_main, Criterion}; +use datafusion::prelude::SessionContext; +use datafusion::{datasource::MemTable, error::Result}; +use futures::executor::block_on; +use std::sync::Arc; +use tokio::runtime::Runtime; + +async fn query(ctx: &mut SessionContext, sql: &str) { + let rt = Runtime::new().unwrap(); + + // execute the query + let df = rt.block_on(ctx.sql(sql)).unwrap(); + criterion::black_box(rt.block_on(df.collect()).unwrap()); +} + +fn create_context(array_len: usize, batch_size: usize) -> Result { + // define a schema. + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, false), + Field::new("b", DataType::Utf8, false), + ])); + + // define data. + let batches = (0..array_len / batch_size) + .map(|_i| { + let data1 = (0..batch_size) + .into_iter() + .map(|x| x as i64) + .collect::>(); + let data2 = (0..batch_size) + .into_iter() + .map(|j| format!("a{j}")) + .collect::>(); + + RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(Int64Array::from(data1)), + Arc::new(StringArray::from(data2)), + ], + ) + .unwrap() + }) + .collect::>(); + + let ctx = SessionContext::new(); + + // declare a table in memory. In spark API, this corresponds to createDataFrame(...). + let provider = MemTable::try_new(schema, vec![batches])?; + ctx.register_table("t", Arc::new(provider))?; + + Ok(ctx) +} + +fn criterion_benchmark(c: &mut Criterion) { + let array_len = 2000000; // 2M rows + let batch_size = array_len; + + c.bench_function("benchmark", |b| { + let mut ctx = create_context(array_len, batch_size).unwrap(); + b.iter(|| block_on(query(&mut ctx, "select a, b, count(*) from t group by a, b order by count(*) desc limit 10"))) + }); +} + +criterion_group! { + name = benches; + // This can be any expression that returns a `Criterion` object. + config = Criterion::default().sample_size(10); + targets = criterion_benchmark +} +criterion_main!(benches); + +// Previous result +// reuse-hash +// benchmark time: [2.5999 s 6.3132 s 11.062 s] +// Found 1 outliers among 10 measurements (10.00%) + +// main +// benchmark time: [4.1404 s 8.4601 s 13.226 s] + +// Latest reuslt +// single-multi-groupby +// Gnuplot not found, using plotters backend +// benchmark time: [82.970 ms 98.723 ms 111.32 ms] +// change: [-99.328% -98.932% -97.777%] (p = 0.00 < 0.05) +// Performance has improved. +// Found 2 outliers among 10 measurements (20.00%) +// 2 (20.00%) high mild + +// main (I guess this improves because of the change Check hashes first during probing the aggr hash table #11718, but I didn't verify) +// Gnuplot not found, using plotters backend +// Benchmarking benchmark: Warming up for 3.0000 s +// Warning: Unable to complete 10 samples in 5.0s. You may wish to increase target time to 23.3s. +// benchmark time: [660.82 ms 1.3354 s 2.1247 s] +// change: [+675.04% +1377.8% +2344.1%] (p = 0.00 < 0.05) +// Performance has regressed. +// Found 1 outliers among 10 measurements (10.00%) +// 1 (10.00%) high mild \ No newline at end of file From 5b5bf9d7389ac27c6d05bd14c387a744e0e497fb Mon Sep 17 00:00:00 2001 From: jayzhan211 Date: Fri, 2 Aug 2024 09:15:45 +0800 Subject: [PATCH 4/6] low cardinality Signed-off-by: jayzhan211 --- datafusion/core/Cargo.toml | 4 + datafusion/core/benches/low_cardinality.rs | 93 ++++++++++++++++++++++ 2 files changed, 97 insertions(+) create mode 100644 datafusion/core/benches/low_cardinality.rs diff --git a/datafusion/core/Cargo.toml b/datafusion/core/Cargo.toml index d4896a4c03754..7776147cf9455 100644 --- a/datafusion/core/Cargo.toml +++ b/datafusion/core/Cargo.toml @@ -184,6 +184,10 @@ name = "math_query_sql" harness = false name = "reuse_hash" +[[bench]] +harness = false +name = "low_cardinality" + [[bench]] harness = false name = "filter_query_sql" diff --git a/datafusion/core/benches/low_cardinality.rs b/datafusion/core/benches/low_cardinality.rs new file mode 100644 index 0000000000000..7e2a9c5e247e8 --- /dev/null +++ b/datafusion/core/benches/low_cardinality.rs @@ -0,0 +1,93 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 arrow::{ + datatypes::{DataType, Field, Schema}, + record_batch::RecordBatch, +}; +use arrow_array::{Int64Array, StringArray}; +use criterion::{criterion_group, criterion_main, Criterion}; +use datafusion::prelude::SessionContext; +use datafusion::{datasource::MemTable, error::Result}; +use futures::executor::block_on; +use std::sync::Arc; +use tokio::runtime::Runtime; + +async fn query(ctx: &mut SessionContext, sql: &str) { + let rt = Runtime::new().unwrap(); + + // execute the query + let df = rt.block_on(ctx.sql(sql)).unwrap(); + criterion::black_box(rt.block_on(df.collect()).unwrap()); +} + +fn create_context(array_len: usize, batch_size: usize) -> Result { + // define a schema. + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, false), + Field::new("b", DataType::Utf8, false), + ])); + + // define data. + let batches = (0..array_len / batch_size) + .map(|_i| { + let data1 = (0..batch_size) + .into_iter() + .map(|x| (x % 2) as i64) + .collect::>(); + let data2 = (0..batch_size) + .into_iter() + .map(|j| format!("a{}", j % 2)) + .collect::>(); + + RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(Int64Array::from(data1)), + Arc::new(StringArray::from(data2)), + ], + ) + .unwrap() + }) + .collect::>(); + + let ctx = SessionContext::new(); + + // declare a table in memory. In spark API, this corresponds to createDataFrame(...). + let provider = MemTable::try_new(schema, vec![batches])?; + ctx.register_table("t", Arc::new(provider))?; + + Ok(ctx) +} + +fn criterion_benchmark(c: &mut Criterion) { + let array_len = 2000000; // 2M rows + let batch_size = array_len; + + c.bench_function("benchmark", |b| { + let mut ctx = create_context(array_len, batch_size).unwrap(); + b.iter(|| block_on(query(&mut ctx, "select a, b, count(*) from t group by a, b order by count(*) desc limit 10"))) + }); +} + +criterion_group! { + name = benches; + // This can be any expression that returns a `Criterion` object. + config = Criterion::default().sample_size(10); + targets = criterion_benchmark +} +criterion_main!(benches); From f55b639c5c13e9d5e4105589d16ab62b50eb96aa Mon Sep 17 00:00:00 2001 From: jayzhan211 Date: Fri, 2 Aug 2024 16:52:52 +0800 Subject: [PATCH 5/6] v2 Signed-off-by: jayzhan211 --- datafusion/core/Cargo.toml | 2 +- .../{reuse_hash.rs => high_cardinality.rs} | 0 datafusion/core/benches/low_cardinality.rs | 18 +-- datafusion/core/src/physical_planner.rs | 28 +++-- datafusion/core/tests/dataframe/mod.rs | 104 +++++++++++++++++- .../physical-plan/src/aggregates/row_hash.rs | 2 + datafusion/sqllogictest/test_files/test1.slt | 26 +++++ 7 files changed, 158 insertions(+), 22 deletions(-) rename datafusion/core/benches/{reuse_hash.rs => high_cardinality.rs} (100%) create mode 100644 datafusion/sqllogictest/test_files/test1.slt diff --git a/datafusion/core/Cargo.toml b/datafusion/core/Cargo.toml index 7776147cf9455..498ae937a5a33 100644 --- a/datafusion/core/Cargo.toml +++ b/datafusion/core/Cargo.toml @@ -182,7 +182,7 @@ name = "math_query_sql" [[bench]] harness = false -name = "reuse_hash" +name = "high_cardinality" [[bench]] harness = false diff --git a/datafusion/core/benches/reuse_hash.rs b/datafusion/core/benches/high_cardinality.rs similarity index 100% rename from datafusion/core/benches/reuse_hash.rs rename to datafusion/core/benches/high_cardinality.rs diff --git a/datafusion/core/benches/low_cardinality.rs b/datafusion/core/benches/low_cardinality.rs index 7e2a9c5e247e8..947bfb2cec4ff 100644 --- a/datafusion/core/benches/low_cardinality.rs +++ b/datafusion/core/benches/low_cardinality.rs @@ -45,14 +45,8 @@ fn create_context(array_len: usize, batch_size: usize) -> Result // define data. let batches = (0..array_len / batch_size) .map(|_i| { - let data1 = (0..batch_size) - .into_iter() - .map(|x| (x % 2) as i64) - .collect::>(); - let data2 = (0..batch_size) - .into_iter() - .map(|j| format!("a{}", j % 2)) - .collect::>(); + let data1 = (0..batch_size).into_iter().map(|x| (x % 4 > 1) as i64).collect::>(); + let data2 = (0..batch_size).into_iter().map(|j| format!("a{}", (j % 2))).collect::>(); RecordBatch::try_new( schema.clone(), @@ -91,3 +85,11 @@ criterion_group! { targets = criterion_benchmark } criterion_main!(benches); + +// single-multi-groupby-v2 +// Gnuplot not found, using plotters backend +// benchmark time: [41.512 ms 41.931 ms 42.318 ms] +// change: [-6.5657% -5.6393% -4.6452%] (p = 0.00 < 0.05) +// Performance has improved. +// Found 1 outliers among 10 measurements (10.00%) +// 1 (10.00%) high mild \ No newline at end of file diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs index d934576d0ed74..159a5feebb745 100644 --- a/datafusion/core/src/physical_planner.rs +++ b/datafusion/core/src/physical_planner.rs @@ -957,18 +957,22 @@ impl DefaultPhysicalPlanner { multiunzip(agg_filter); 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 - }; + // if false { + // 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 + // }; + + // We don't do pre-partition + let mode = AggregateMode::Single; let aggr = AggregateExec::try_new( mode, diff --git a/datafusion/core/tests/dataframe/mod.rs b/datafusion/core/tests/dataframe/mod.rs index d83a47ceb0699..85a377c6beca6 100644 --- a/datafusion/core/tests/dataframe/mod.rs +++ b/datafusion/core/tests/dataframe/mod.rs @@ -29,7 +29,7 @@ use arrow::{ }, record_batch::RecordBatch, }; -use arrow_array::{Array, Float32Array, Float64Array, UnionArray}; +use arrow_array::{Array, Float32Array, Float64Array, Int64Array, UnionArray}; use arrow_buffer::ScalarBuffer; use arrow_schema::{ArrowError, UnionFields, UnionMode}; use datafusion_functions_aggregate::count::count_udaf; @@ -2363,3 +2363,105 @@ async fn dense_union_is_null() { ]; assert_batches_sorted_eq!(expected, &result_df.collect().await.unwrap()); } + +#[tokio::test] +async fn test_low_cardinality() -> Result<()> { + async fn query(ctx: &mut SessionContext, sql: &str) -> Result> { + // execute the query + ctx.sql(sql).await?.collect().await + } + + fn create_context(array_len: usize, batch_size: usize) -> Result { + // define a schema. + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, false), + Field::new("b", DataType::Utf8, false), + ])); + + // define data. + let batches = (0..array_len / batch_size) + .map(|i| { + let data1 = (0..batch_size).into_iter().map(|x| (x % 4 > 1) as i64).collect::>(); + let data2 = (0..batch_size).into_iter().map(|j| format!("a{}", (j % 2))).collect::>(); + + RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(Int64Array::from(data1)), + Arc::new(StringArray::from(data2)), + ], + ) + .unwrap() + }) + .collect::>(); + + // println!("batches: {:?}", batches); + + let ctx = SessionContext::new(); + + // declare a table in memory. In spark API, this corresponds to createDataFrame(...). + let provider = MemTable::try_new(schema, vec![batches])?; + ctx.register_table("t", Arc::new(provider))?; + + + Ok(ctx) + } + + let array_len = 2000000; // 2^19 + let batch_size = array_len; // 2^12 + let mut ctx = create_context(array_len, batch_size).unwrap(); + let res = query(&mut ctx, "select a, b, count(*) from t group by a, b order by count(*) desc limit 10").await.unwrap(); + // println!("res: {:?}", res); + Ok(()) +} + +#[tokio::test] +async fn test_high_cardinality() -> Result<()> { + async fn query(ctx: &mut SessionContext, sql: &str) -> Result> { + // execute the query + ctx.sql(sql).await?.collect().await + } + + fn create_context(array_len: usize, batch_size: usize) -> Result { + // define a schema. + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, false), + Field::new("b", DataType::Utf8, false), + ])); + + // define data. + let batches = (0..array_len / batch_size) + .map(|i| { + let data1 = (0..batch_size).into_iter().map(|x| x as i64).collect::>(); + let data2 = (0..batch_size).into_iter().map(|j| format!("a{}", j)).collect::>(); + + RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(Int64Array::from(data1)), + Arc::new(StringArray::from(data2)), + ], + ) + .unwrap() + }) + .collect::>(); + + // println!("batches: {:?}", batches); + + let ctx = SessionContext::new(); + + // declare a table in memory. In spark API, this corresponds to createDataFrame(...). + let provider = MemTable::try_new(schema, vec![batches])?; + ctx.register_table("t", Arc::new(provider))?; + + + Ok(ctx) + } + + let array_len = 2000000; // 2^19 + let batch_size = array_len; // 2^12 + let mut ctx = create_context(array_len, batch_size).unwrap(); + let res = query(&mut ctx, "select a, b, count(*) from t group by a, b order by count(*) desc limit 10").await.unwrap(); + // println!("res: {:?}", res); + Ok(()) +} \ No newline at end of file diff --git a/datafusion/physical-plan/src/aggregates/row_hash.rs b/datafusion/physical-plan/src/aggregates/row_hash.rs index 167ca72407503..8448d17c3f8f3 100644 --- a/datafusion/physical-plan/src/aggregates/row_hash.rs +++ b/datafusion/physical-plan/src/aggregates/row_hash.rs @@ -580,6 +580,7 @@ impl GroupedHashAggregateStream { | AggregateMode::SinglePartitioned if !self.spill_state.is_stream_merging => { + acc.update_batch( values, group_indices, @@ -658,6 +659,7 @@ impl GroupedHashAggregateStream { // emit reduces the memory usage. Ignore Err from update_memory_reservation. Even if it is // over the target memory size after emission, we can emit again rather than returning Err. let _ = self.update_memory_reservation(); + // println!("output: {:?}", output); let batch = RecordBatch::try_new(schema, output)?; Ok(batch) } diff --git a/datafusion/sqllogictest/test_files/test1.slt b/datafusion/sqllogictest/test_files/test1.slt new file mode 100644 index 0000000000000..3de2358f75444 --- /dev/null +++ b/datafusion/sqllogictest/test_files/test1.slt @@ -0,0 +1,26 @@ +statement ok +CREATE EXTERNAL TABLE hits +STORED AS PARQUET +LOCATION '../core/tests/data/clickbench_hits_10.parquet'; + +query ITI +SELECT "UserID", "SearchPhrase", COUNT(*) FROM hits GROUP BY "UserID", "SearchPhrase" ORDER BY COUNT(*) DESC LIMIT 10; +---- +-2461439046089301801 (empty) 5 +519640690937130534 (empty) 2 +7418527520126366595 (empty) 1 +427738049800818189 (empty) 1 +376160620089546609 (empty) 1 + +query TT +explain SELECT "UserID", "SearchPhrase", COUNT(*) FROM hits GROUP BY "UserID", "SearchPhrase" ORDER BY COUNT(*) DESC LIMIT 10; +---- +logical_plan +01)Limit: skip=0, fetch=10 +02)--Sort: count(*) DESC NULLS FIRST, fetch=10 +03)----Aggregate: groupBy=[[hits.UserID, hits.SearchPhrase]], aggr=[[count(Int64(1)) AS count(*)]] +04)------TableScan: hits projection=[UserID, SearchPhrase] +physical_plan +01)SortExec: TopK(fetch=10), expr=[count(*)@2 DESC], preserve_partitioning=[false] +02)--AggregateExec: mode=Single, gby=[UserID@0 as UserID, SearchPhrase@1 as SearchPhrase], aggr=[count(*)] +03)----ParquetExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/clickbench_hits_10.parquet]]}, projection=[UserID, SearchPhrase] From 40d6a0a0ed20c1be3d8098e776e5bb7b92dcd22d Mon Sep 17 00:00:00 2001 From: jayzhan211 Date: Fri, 2 Aug 2024 16:53:09 +0800 Subject: [PATCH 6/6] result Signed-off-by: jayzhan211 --- datafusion/core/benches/high_cardinality.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/datafusion/core/benches/high_cardinality.rs b/datafusion/core/benches/high_cardinality.rs index 04dad9ec8d6ba..b67983808109c 100644 --- a/datafusion/core/benches/high_cardinality.rs +++ b/datafusion/core/benches/high_cardinality.rs @@ -117,4 +117,13 @@ criterion_main!(benches); // change: [+675.04% +1377.8% +2344.1%] (p = 0.00 < 0.05) // Performance has regressed. // Found 1 outliers among 10 measurements (10.00%) -// 1 (10.00%) high mild \ No newline at end of file +// 1 (10.00%) high mild + + +// Gnuplot not found, using plotters backend +// Benchmarking benchmark: Warming up for 3.0000 s +// Warning: Unable to complete 10 samples in 5.0s. You may wish to increase target time to 8.2s or enable flat sampling. +// benchmark time: [150.22 ms 154.31 ms 158.06 ms] +// change: [+254.76% +261.92% +270.28%] (p = 0.00 < 0.05) +// Performance has regressed. +// Found 1 outliers among 10 measurements (10.00%) \ No newline at end of file