You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by GitBox <gi...@apache.org> on 2021/09/24 05:41:49 UTC
[GitHub] [arrow-datafusion] jiangzhx commented on issue #1041: does datafusion master version default enable simd?
jiangzhx commented on issue #1041:
URL: https://github.com/apache/arrow-datafusion/issues/1041#issuecomment-926361164
i did more test, load parquet file to memtable first then query sql;
SELECT sum(LO_EXTENDEDPRICE * LO_DISCOUNT) AS revenue FROM lineorder_flat;
with simd and mimalloc command:
RUSTFLAGS='-C target-cpu=native' cargo +nightly run --release --features "simd mimalloc" --example ssb_memtable_sum_example
prepare memtable usage millis: 130
sql execution usage millis: 26
with mimalloc command:
RUSTFLAGS='-C target-cpu=native' cargo +nightly run --release --features "simd mimalloc" --example ssb_memtable_sum_example
prepare memtable usage millis: 130
sql execution usage millis: 25
data rows: 59,986,052 rows
i guess the reson should be like @jorgecarleitao and @alamb comments;
rust already compiler code to use SIMD automatically;
code:
use datafusion::arrow::record_batch::RecordBatch;
use datafusion::arrow::util::pretty::print_batches;
use chrono::prelude::*;
use datafusion::prelude::*;
use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use datafusion::datasource::parquet::ParquetTable;
use datafusion::datasource::{MemTable, TableProvider};
use datafusion::logical_plan::Expr;
use datafusion::physical_plan::display::DisplayableExecutionPlan;
use datafusion::physical_plan::{common, displayable};
use futures::StreamExt;
use std::sync::Arc;
#[cfg(feature = "snmalloc")]
#[global_allocator]
static ALLOC: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc;
#[cfg(feature = "mimalloc")]
#[global_allocator]
static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc;
#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
let parquet_path = "/Users/sylar/workspace/opensource/ssb-dbgen/output/parquet_10";
let lineorder_flat_schema = Schema::new(vec![
Field::new("LO_ORDERKEY", DataType::Int64, false),
Field::new("LO_LINENUMBER", DataType::Int64, false),
Field::new("LO_CUSTKEY", DataType::Int64, false),
Field::new("LO_PARTKEY", DataType::Int64, false),
Field::new("LO_SUPPKEY", DataType::Int64, false),
Field::new("LO_ORDERDATE", DataType::Int64, false),
Field::new("LO_ORDERPRIORITY", DataType::Utf8, false),
Field::new("LO_SHIPPRIOTITY", DataType::Int64, false),
Field::new("LO_QUANTITY", DataType::Int64, false),
Field::new("LO_EXTENDEDPRICE", DataType::Int64, false),
Field::new("LO_ORDTOTALPRICE", DataType::Int64, false),
Field::new("LO_DISCOUNT", DataType::Int64, false),
Field::new("LO_REVENUE", DataType::Int64, false),
Field::new("LO_SUPPLYCOST", DataType::Int64, false),
Field::new("LO_TAX", DataType::Int64, false),
Field::new("LO_COMMITDATE", DataType::Int64, false),
Field::new("LO_SHIPMODE", DataType::Utf8, false),
Field::new("C_NAME", DataType::Utf8, false),
Field::new("C_ADDRESS", DataType::Utf8, false),
Field::new("C_CITY", DataType::Utf8, false),
Field::new("C_NATION", DataType::Utf8, false),
Field::new("C_REGION", DataType::Utf8, false),
Field::new("C_PHONE", DataType::Utf8, false),
Field::new("C_MKTSEGMENT", DataType::Utf8, false),
Field::new("S_NAME", DataType::Utf8, false),
Field::new("S_ADDRESS", DataType::Utf8, false),
Field::new("S_CITY", DataType::Utf8, false),
Field::new("S_NATION", DataType::Utf8, false),
Field::new("S_REGION", DataType::Utf8, false),
Field::new("S_PHONE", DataType::Utf8, false),
Field::new("P_NAME", DataType::Utf8, false),
Field::new("P_MFGR", DataType::Utf8, false),
Field::new("P_CATEGORY", DataType::Utf8, false),
Field::new("P_BRAND", DataType::Utf8, false),
Field::new("P_COLOR", DataType::Utf8, false),
Field::new("P_TYPE", DataType::Utf8, false),
Field::new("P_SIZE", DataType::Int64, false),
Field::new("P_CONTAINER", DataType::Utf8, false),
]);
let project_schema = Schema::new(vec![
Field::new("LO_EXTENDEDPRICE", DataType::Int64, false),
Field::new("LO_DISCOUNT", DataType::Int64, false),
]);
let dt = Local::now();
let table_provider = Arc::new(ParquetTable::try_new_with_schema(
parquet_path,
lineorder_flat_schema,
32,
false,
)?);
let exec = table_provider
.scan(&Option::Some(vec![9,11]), 8192, &[], None)
.await?;
let partition_count = exec.output_partitioning().partition_count();
let tasks = (0..partition_count)
.map(|part_i| {
let exec = exec.clone();
tokio::spawn(async move {
let stream = exec.execute(part_i).await?;
common::collect(stream).await
})
})
.collect::<Vec<_>>();
let mut data: Vec<Vec<RecordBatch>> = Vec::new();
for task in tasks {
let result = task.await.expect("MemTable::load could not join task")?;
data.push(result);
}
let memtable = MemTable::try_new(SchemaRef::new(project_schema).clone(), data).unwrap();
println!(
"prepare memtable usage millis: {}",
Local::now().timestamp_millis() - dt.timestamp_millis()
);
let execution_config = ExecutionConfig::new();
let mut ctx = ExecutionContext::with_config(execution_config);
ctx.register_table("lineorder_flat", Arc::new(memtable))?;
let dt = Local::now();
let sql = "SELECT sum(LO_EXTENDEDPRICE * LO_DISCOUNT) AS revenue FROM lineorder_flat";
let df = ctx.sql(sql)?;
let results: Vec<RecordBatch> = df.collect().await?;
print_batches(&results)?;
println!(
"sql execution usage millis: {}",
Local::now().timestamp_millis() - dt.timestamp_millis()
);
Ok(())
}
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org