Merge branch 'master' into 'master'

基准测试数据和数据量统计留存

See merge request noah/argo_engine!28

Author: 刘黎

benchmarks/examples/ssb_query_example.rs

@@ -2,17 +2,17 @@ use chrono::prelude::*;
 use datafusion::arrow::util::pretty::print_batches;
 use datafusion::datasource::listing::ListingTable;
 use datafusion::logical_plan::{count_distinct, Expr};
-use datafusion::optimizer::common_subexpr_eliminate::CommonSubexprEliminate;
-use datafusion::optimizer::constant_folding::ConstantFolding;
-use datafusion::optimizer::eliminate_limit::EliminateLimit;
-use datafusion::optimizer::filter_push_down::FilterPushDown;
-use datafusion::optimizer::limit_push_down::LimitPushDown;
-use datafusion::optimizer::projection_push_down::ProjectionPushDown;
-use datafusion::optimizer::simplify_expressions::SimplifyExpressions;
-use datafusion::physical_optimizer::aggregate_statistics::AggregateStatistics;
-use datafusion::physical_optimizer::coalesce_batches::CoalesceBatches;
-use datafusion::physical_optimizer::hash_build_probe_order::HashBuildProbeOrder;
-use datafusion::physical_optimizer::merge_exec::AddCoalescePartitionsExec;
+// use datafusion::optimizer::common_subexpr_eliminate::CommonSubexprEliminate;
+// use datafusion::optimizer::constant_folding::ConstantFolding;
+// use datafusion::optimizer::eliminate_limit::EliminateLimit;
+// use datafusion::optimizer::filter_push_down::FilterPushDown;
+// use datafusion::optimizer::limit_push_down::LimitPushDown;
+// use datafusion::optimizer::projection_push_down::ProjectionPushDown;
+// use datafusion::optimizer::simplify_expressions::SimplifyExpressions;
+// use datafusion::physical_optimizer::aggregate_statistics::AggregateStatistics;
+// use datafusion::physical_optimizer::coalesce_batches::CoalesceBatches;
+// use datafusion::physical_optimizer::hash_build_probe_order::HashBuildProbeOrder;
+// use datafusion::physical_optimizer::merge_exec::AddCoalescePartitionsExec;
 use datafusion::physical_optimizer::repartition::Repartition;
 use datafusion::physical_plan::{collect, displayable};
 use datafusion::prelude::*;
@@ -34,22 +34,22 @@ async fn main() -> datafusion::error::Result<()> {
     let execution_config = ExecutionConfig::new()
         .with_optimizer_rules(
             vec![
-                Arc::new(ConstantFolding::new()),
-                Arc::new(CommonSubexprEliminate::new()),
-                Arc::new(EliminateLimit::new()),
-                Arc::new(ProjectionPushDown::new()),
-                Arc::new(FilterPushDown::new()),
-                Arc::new(SimplifyExpressions::new()),
-                Arc::new(LimitPushDown::new()),
+                // Arc::new(ConstantFolding::new()),
+                // Arc::new(CommonSubexprEliminate::new()),
+                // Arc::new(EliminateLimit::new()),
+                // Arc::new(ProjectionPushDown::new()),
+                // Arc::new(FilterPushDown::new()),
+                // Arc::new(SimplifyExpressions::new()),
+                // Arc::new(LimitPushDown::new()),
             ]
         )
         .with_physical_optimizer_rules(
             vec![
-                Arc::new(AggregateStatistics::new()),
-                Arc::new(HashBuildProbeOrder::new()),
-                Arc::new(CoalesceBatches::new()),
-                Arc::new(Repartition::new()),
-                Arc::new(AddCoalescePartitionsExec::new()),
+                // Arc::new(AggregateStatistics::new()),
+                // Arc::new(HashBuildProbeOrder::new()),
+                // Arc::new(CoalesceBatches::new()),
+                // Arc::new(Repartition::new()),
+                // Arc::new(AddCoalescePartitionsExec::new()),
             ]
         )
         //.with_target_partitions(8)

benchmarks/examples/ssb_sql_example.rs

@@ -22,17 +22,17 @@ use datafusion::arrow::util::pretty::print_batches;
 use std::sync::Arc;
 
 use datafusion::error::Result;
-use datafusion::optimizer::common_subexpr_eliminate::CommonSubexprEliminate;
-use datafusion::optimizer::constant_folding::ConstantFolding;
-use datafusion::optimizer::eliminate_limit::EliminateLimit;
-use datafusion::optimizer::filter_push_down::FilterPushDown;
-use datafusion::optimizer::limit_push_down::LimitPushDown;
-use datafusion::optimizer::projection_push_down::ProjectionPushDown;
-use datafusion::optimizer::simplify_expressions::SimplifyExpressions;
-use datafusion::physical_optimizer::aggregate_statistics::AggregateStatistics;
-use datafusion::physical_optimizer::coalesce_batches::CoalesceBatches;
-use datafusion::physical_optimizer::hash_build_probe_order::HashBuildProbeOrder;
-use datafusion::physical_optimizer::merge_exec::AddCoalescePartitionsExec;
+// use datafusion::optimizer::common_subexpr_eliminate::CommonSubexprEliminate;
+// use datafusion::optimizer::constant_folding::ConstantFolding;
+// use datafusion::optimizer::eliminate_limit::EliminateLimit;
+// use datafusion::optimizer::filter_push_down::FilterPushDown;
+// use datafusion::optimizer::limit_push_down::LimitPushDown;
+// use datafusion::optimizer::projection_push_down::ProjectionPushDown;
+// use datafusion::optimizer::simplify_expressions::SimplifyExpressions;
+// use datafusion::physical_optimizer::aggregate_statistics::AggregateStatistics;
+// use datafusion::physical_optimizer::coalesce_batches::CoalesceBatches;
+// use datafusion::physical_optimizer::hash_build_probe_order::HashBuildProbeOrder;
+// use datafusion::physical_optimizer::merge_exec::AddCoalescePartitionsExec;
 use datafusion::prelude::*;
 use utils::test_util;
 
@@ -51,23 +51,23 @@ async fn main() -> Result<()> {
     let execution_config = ExecutionConfig::new()
         .with_optimizer_rules(
             vec![
-                Arc::new(ConstantFolding::new()),
-                Arc::new(CommonSubexprEliminate::new()),
-                Arc::new(EliminateLimit::new()),
-                Arc::new(ProjectionPushDown::new()),
-                Arc::new(FilterPushDown::new()),
-                Arc::new(SimplifyExpressions::new()),
-                Arc::new(LimitPushDown::new()),
+                // Arc::new(ConstantFolding::new()),
+                // Arc::new(CommonSubexprEliminate::new()),
+                // Arc::new(EliminateLimit::new()),
+                // Arc::new(ProjectionPushDown::new()),
+                // Arc::new(FilterPushDown::new()),
+                // Arc::new(SimplifyExpressions::new()),
+                // Arc::new(LimitPushDown::new()),
                 //Arc::new(SingleDistinctAggregationToGroupBy::new()),
             ]
         )
         .with_physical_optimizer_rules(
             vec![
-                Arc::new(AggregateStatistics::new()),
-                Arc::new(HashBuildProbeOrder::new()),
-                Arc::new(CoalesceBatches::new()),
-                // Arc::new(Repartition::new()),
-                Arc::new(AddCoalescePartitionsExec::new()),
+                // Arc::new(AggregateStatistics::new()),
+                // Arc::new(HashBuildProbeOrder::new()),
+                // Arc::new(CoalesceBatches::new()),
+                // // Arc::new(Repartition::new()),
+                // Arc::new(AddCoalescePartitionsExec::new()),
             ]
         )
         //.with_target_partitions(8)

benchmarks/src/bin/bench_count.rs

@@ -0,0 +1,246 @@
+use datafusion::execution::context::ExecutionContext;
+use datafusion::arrow::record_batch::RecordBatch;
+use datafusion::arrow::util::display::array_value_to_string;
+use std::convert::From;
+use std::error::Error;
+use utils::test_util::{BenchType, BenchDataType};
+use utils::test_util;
+
+
+#[tokio::main]
+async fn main() -> datafusion::error::Result<()> {
+    // 读取event count的answer
+    let answer_vec: Vec<CountAnswer> = read_count_answer().unwrap();
+
+    // 加载数据 执行查询 并返回 Vec<CountAnswer>
+    let query_vec: Vec<CountAnswer> = query_count().await.unwrap();
+
+    // 比对查询结果
+    let err_info_vec = test_result(&answer_vec, &query_vec).unwrap();
+
+    // 打印测试的结果
+    if err_info_vec.len() > 0 {
+        println!("事件量统计测试未通过\n错误详情如下：");
+        for x in err_info_vec {
+            println!("{}", x);
+        }
+    } else {
+        println!("事件量统计测试通过!!!");
+    }
+
+    Ok(())
+}
+
+// 读取event count的answer
+fn read_count_answer() -> Result<Vec<CountAnswer>, Box<dyn Error>> {
+    let answer_file = test_util::bench_answer_file(BenchType::EventCount);
+    let answer_content: Vec<String> = *(test_util::read_answer_file(answer_file.as_ref()));
+    let mut answer_vec: Vec<CountAnswer> = Vec::new();
+    for line in answer_content {
+        let answer = CountAnswer::from(line);
+        answer_vec.push(answer);
+    }
+
+    Ok(answer_vec)
+}
+
+// 加载数据 执行查询 并返回 Vec<CountAnswer>
+async fn query_count() -> Result<Vec<CountAnswer>, Box<dyn Error>> {
+    let mut ctx = ExecutionContext::new();
+
+    ctx.register_parquet("event", test_util::bench_data_file(BenchDataType::Event).as_str()).await?;
+    ctx.register_parquet("profile", test_util::bench_data_file(BenchDataType::Profile).as_str()).await?;
+
+    let detail_sql = r#"select xwhat,count(1) as cn ,count(distinct distinct_id) as uv
+                             from event
+                             join profile
+                             on event.distinct_id = profile.distinct_id
+                             group by xwhat
+                             order by xwhat"#;
+    let detail_f = ctx.sql(detail_sql).await?;
+    let detail_results = detail_f.collect().await?;
+    let mut query_vec = CountAnswer::collect_answer(&detail_results);
+
+    let total_sql = r#"select 'null' as xwhat,count(1) as cn ,count(distinct distinct_id) as uv
+                             from event
+                             join profile
+                             on event.distinct_id = profile.distinct_id"#;
+    let total_df = ctx.sql(total_sql).await?;
+    let total_results = total_df.collect().await?;
+    let mut total_query_vec = CountAnswer::collect_answer(&total_results);
+
+    query_vec.append(&mut total_query_vec);
+
+    Ok(query_vec)
+}
+
+// 比对查询结果
+fn test_result(answer_vec: &[CountAnswer], query_vec: &[CountAnswer]) -> Result<Vec<String>, Box<dyn Error>> {
+    let mut err_info_vec: Vec<String> = Vec::new();
+    for index in 0..answer_vec.len() {
+        let answer_item: &CountAnswer = answer_vec.get(index).unwrap();
+        let query_item: &CountAnswer = query_vec.get(index).unwrap();
+
+        if query_item != answer_item {
+            err_info_vec.push(format!("期望结果：{}\n查询结果：{}\n", answer_item.to_string(), query_item.to_string()));
+        }
+    }
+
+    Ok(err_info_vec)
+}
+
+// 封装事件数量统计的结果
+struct CountAnswer {
+    xwhat: String,
+    count: u32,
+    uv: u32,
+}
+
+impl PartialEq for CountAnswer {
+    fn eq(&self, other: &Self) -> bool {
+        other.xwhat == self.xwhat
+            && other.count == self.count
+            && other.uv == self.uv
+    }
+
+    fn ne(&self, other: &Self) -> bool {
+        !self.eq(other)
+    }
+}
+
+impl From<String> for CountAnswer {
+    fn from(content: String) -> Self {
+        let words: Vec<&str> = content.split(r#"|"#).collect();
+        if words.len() != 3 {
+            panic!("size of spliting {}  by | is not 3", content);
+        }
+
+        let tmp_xwhat = String::from(words[0]);
+        let tmp_count = words[1].parse::<u32>().unwrap();
+        let tmp_uv = words[2].parse::<u32>().unwrap();
+        let answer = CountAnswer {
+            xwhat: tmp_xwhat,
+            count: tmp_count,
+            uv: tmp_uv,
+        };
+
+        answer
+    }
+}
+
+impl CountAnswer {
+    fn to_string(&self) -> String {
+        format!("{}|{}|{}", self.xwhat, self.count, self.uv)
+    }
+
+    pub fn collect_answer(records: &[RecordBatch]) -> Vec<CountAnswer> {
+        let mut result = Vec::new();
+
+        if !records.is_empty() {
+            for batch in records {
+                for row in 0..batch.num_rows() {
+                    let mut cells = Vec::new();
+                    for col in 0..batch.num_columns() {
+                        let column = batch.column(col);
+                        cells.push(array_value_to_string(&column, row).unwrap());
+                    }
+
+                    result.push(CountAnswer::from(cells.join("|")));
+                }
+            }
+        }
+
+        result
+    }
+}
+
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn event_count_answer_eq() {
+        let answer = CountAnswer { xwhat: String::from("startUp"), count: 12, uv: 100 };
+
+        let other = CountAnswer { xwhat: String::from("startUp"), count: 12, uv: 100 };
+        if answer != other {
+            panic!("event_count_answer_eq")
+        }
+
+        let other = CountAnswer { xwhat: String::from("startUp"), count: 12, uv: 200 };
+        if answer == other {
+            panic!("event_count_answer_eq")
+        }
+
+        let other = CountAnswer { xwhat: String::from("startUp"), count: 2, uv: 100 };
+        if answer == other {
+            panic!("event_count_answer_eq")
+        }
+
+        let other = CountAnswer { xwhat: String::from("startUp"), count: 2, uv: 100 };
+        if answer == other {
+            panic!("event_count_answer_eq")
+        }
+
+        let other = CountAnswer { xwhat: String::from("ABC"), count: 12, uv: 100 };
+        if answer == other {
+            panic!("event_count_answer_eq")
+        }
+    }
+
+    #[test]
+    fn from_test() {
+        let answer = CountAnswer::from(String::from("SignIn_isSuccess|26488|3387"));
+        let count_answer = CountAnswer {
+            xwhat: String::from("SignIn_isSuccess"),
+            count: 26488,
+            uv: 3387,
+        };
+
+        if answer != count_answer {
+            panic!();
+        }
+    }
+
+    #[test]
+    fn to_string_test() {
+        let count_answer = CountAnswer {
+            xwhat: String::from("SignIn_isSuccess"),
+            count: 26488,
+            uv: 3387,
+        };
+
+        let string = count_answer.to_string();
+        if string != String::from("SignIn_isSuccess|26488|3387") {
+            panic!();
+        }
+    }
+
+    #[tokio::test]
+    async fn query_count_test(){
+        let vec = query_count().await.unwrap();
+        for x in vec {
+            println!("{}", x.to_string());
+        }
+    }
+
+    #[test]
+     fn read_count_answer_test(){
+        let vec = read_count_answer().unwrap();
+        for x in vec {
+            println!("{}", x.to_string());
+        }
+    }
+
+    #[tokio::test]
+    async fn test_result_test(){
+        let query_vec = query_count().await.unwrap();
+        let answer_vec = read_count_answer().unwrap();
+        let vec = test_result(&answer_vec, &query_vec).unwrap();
+        for x in vec {
+            println!("{}", x);
+        }
+    }
+}
+