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 2022/10/08 13:41:17 UTC
[GitHub] [arrow-datafusion] thinkharderdev opened a new pull request, #3769: Add benchmarks for testing row filtering
thinkharderdev opened a new pull request, #3769:
URL: https://github.com/apache/arrow-datafusion/pull/3769
# Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Closes #3457
# Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
Need a set benchmarks for evaluating performance implications of parquet predicate pushdown. This sets up some very basic benchmarks which can be used for that purpose. Thanks to @tustvold for cooking up a script to generate synthetic datasets for this purpose.
# What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
Add new benchmark script `parquet_filter_pushdown` which will execute a series of `ParquetExec` plans with different filter predicates. For each predicate in the suite, we will execute the plan with all three different `ParquetScanOptions` configurations:
1. Disable filter pushdown entirely (currently the default behavior)
2. Enable filter pushdown and predicate reordering
3. Enable filter pushdown without predicate reordering.
# Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
No
<!--
If there are any breaking changes to public APIs, please add the `api change` label.
-->
--
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
[GitHub] [arrow-datafusion] Ted-Jiang commented on a diff in pull request #3769: Add benchmarks for testing row filtering
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on code in PR #3769:
URL: https://github.com/apache/arrow-datafusion/pull/3769#discussion_r990726743
##########
benchmarks/README.md:
##########
@@ -126,3 +126,37 @@ h2o groupby query 1 took 1669 ms
[1]: http://www.tpc.org/tpch/
[2]: https://www1.nyc.gov/site/tlc/about/tlc-trip-record-data.page
+
+## Parquet filter pushdown benchmarks
+
+This is a set of benchmarks for testing and verifying performance of parquet filter pushdown. The queries are executed on
+a synthetic dataset generated during the benchmark execution and designed to simulate web server access logs.
+
+```base
+cargo run --release --bin parquet_filter_pushdown --query --path ./data --scale-factor 1.0
Review Comment:
i think the `--query` is redundant
##########
benchmarks/src/bin/parquet_filter_pushdown.rs:
##########
@@ -0,0 +1,462 @@
+// 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::array::{
+ Int32Builder, StringBuilder, StringDictionaryBuilder, TimestampNanosecondBuilder,
+ UInt16Builder,
+};
+use arrow::datatypes::{DataType, Field, Int32Type, Schema, SchemaRef, TimeUnit};
+use arrow::record_batch::RecordBatch;
+use arrow::util::pretty;
+use datafusion::common::Result;
+use datafusion::datasource::listing::{ListingTableUrl, PartitionedFile};
+use datafusion::datasource::object_store::ObjectStoreUrl;
+use datafusion::execution::context::ExecutionProps;
+use datafusion::logical_expr::{lit, or, Expr};
+use datafusion::logical_plan::ToDFSchema;
+use datafusion::physical_expr::create_physical_expr;
+use datafusion::physical_plan::collect;
+use datafusion::physical_plan::file_format::{
+ FileScanConfig, ParquetExec, ParquetScanOptions,
+};
+use datafusion::physical_plan::filter::FilterExec;
+use datafusion::prelude::{col, combine_filters, SessionConfig, SessionContext};
+use object_store::path::Path;
+use object_store::ObjectMeta;
+use parquet::arrow::ArrowWriter;
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
+use std::fs::File;
+use std::ops::Range;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Instant;
+use structopt::StructOpt;
+
+#[cfg(feature = "snmalloc")]
+#[global_allocator]
+static ALLOC: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc;
+
+#[derive(Debug, StructOpt)]
+#[structopt(name = "Benchmarks", about = "Apache Arrow Rust Benchmarks.")]
+struct Opt {
+ /// Activate debug mode to see query results
+ #[structopt(short, long)]
+ debug: bool,
+
+ /// Number of iterations of each test run
+ #[structopt(short = "i", long = "iterations", default_value = "3")]
+ iterations: usize,
+
+ /// Number of partitions to process in parallel
+ #[structopt(long = "partitions", default_value = "2")]
+ partitions: usize,
+
+ /// Path to folder where access log file will be generated
+ #[structopt(parse(from_os_str), required = true, short = "p", long = "path")]
+ path: PathBuf,
+
+ /// Batch size when reading Parquet files
+ #[structopt(short = "s", long = "batch-size", default_value = "8192")]
+ batch_size: usize,
+
+ /// Total size of generated dataset. The default scale factor of 1.0 will generate a roughly 1GB parquet file
+ #[structopt(short = "s", long = "scale-factor", default_value = "1.0")]
+ scale_factor: f32,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+ let opt: Opt = Opt::from_args();
+ println!("Running benchmarks with the following options: {:?}", opt);
+
+ let config = SessionConfig::new()
+ .with_target_partitions(opt.partitions)
+ .with_batch_size(opt.batch_size);
+ let mut ctx = SessionContext::with_config(config);
+
+ let path = opt.path.join("logs.parquet");
+
+ let (object_store_url, object_meta) = gen_data(path, opt.scale_factor)?;
+
+ run_benchmarks(
+ &mut ctx,
+ object_store_url.clone(),
+ object_meta.clone(),
+ opt.iterations,
+ opt.debug,
+ )
+ .await?;
+
+ Ok(())
+}
+
+async fn run_benchmarks(
+ ctx: &mut SessionContext,
+ object_store_url: ObjectStoreUrl,
+ object_meta: ObjectMeta,
+ iterations: usize,
+ debug: bool,
+) -> Result<()> {
+ let scan_options_matrix = vec![
+ ParquetScanOptions::default(),
+ ParquetScanOptions::default()
+ .with_page_index(true)
+ .with_pushdown_filters(true)
+ .with_reorder_predicates(true),
+ ParquetScanOptions::default()
+ .with_page_index(true)
+ .with_pushdown_filters(true)
+ .with_reorder_predicates(false),
+ ];
+
+ let filter_matrix = vec![
+ // Selective-ish filter
Review Comment:
well-defined test case and test data! 👍
##########
benchmarks/src/bin/parquet_filter_pushdown.rs:
##########
@@ -0,0 +1,462 @@
+// 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::array::{
+ Int32Builder, StringBuilder, StringDictionaryBuilder, TimestampNanosecondBuilder,
+ UInt16Builder,
+};
+use arrow::datatypes::{DataType, Field, Int32Type, Schema, SchemaRef, TimeUnit};
+use arrow::record_batch::RecordBatch;
+use arrow::util::pretty;
+use datafusion::common::Result;
+use datafusion::datasource::listing::{ListingTableUrl, PartitionedFile};
+use datafusion::datasource::object_store::ObjectStoreUrl;
+use datafusion::execution::context::ExecutionProps;
+use datafusion::logical_expr::{lit, or, Expr};
+use datafusion::logical_plan::ToDFSchema;
+use datafusion::physical_expr::create_physical_expr;
+use datafusion::physical_plan::collect;
+use datafusion::physical_plan::file_format::{
+ FileScanConfig, ParquetExec, ParquetScanOptions,
+};
+use datafusion::physical_plan::filter::FilterExec;
+use datafusion::prelude::{col, combine_filters, SessionConfig, SessionContext};
+use object_store::path::Path;
+use object_store::ObjectMeta;
+use parquet::arrow::ArrowWriter;
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
+use std::fs::File;
+use std::ops::Range;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Instant;
+use structopt::StructOpt;
+
+#[cfg(feature = "snmalloc")]
+#[global_allocator]
+static ALLOC: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc;
+
+#[derive(Debug, StructOpt)]
+#[structopt(name = "Benchmarks", about = "Apache Arrow Rust Benchmarks.")]
+struct Opt {
+ /// Activate debug mode to see query results
+ #[structopt(short, long)]
+ debug: bool,
+
+ /// Number of iterations of each test run
+ #[structopt(short = "i", long = "iterations", default_value = "3")]
+ iterations: usize,
+
+ /// Number of partitions to process in parallel
+ #[structopt(long = "partitions", default_value = "2")]
+ partitions: usize,
+
+ /// Path to folder where access log file will be generated
+ #[structopt(parse(from_os_str), required = true, short = "p", long = "path")]
+ path: PathBuf,
+
+ /// Batch size when reading Parquet files
+ #[structopt(short = "s", long = "batch-size", default_value = "8192")]
Review Comment:
i think there two short options 's'
--
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
[GitHub] [arrow-datafusion] thinkharderdev commented on a diff in pull request #3769: Add benchmarks for testing row filtering
Posted by GitBox <gi...@apache.org>.
thinkharderdev commented on code in PR #3769:
URL: https://github.com/apache/arrow-datafusion/pull/3769#discussion_r992727637
##########
benchmarks/src/bin/parquet_filter_pushdown.rs:
##########
@@ -0,0 +1,462 @@
+// 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::array::{
+ Int32Builder, StringBuilder, StringDictionaryBuilder, TimestampNanosecondBuilder,
+ UInt16Builder,
+};
+use arrow::datatypes::{DataType, Field, Int32Type, Schema, SchemaRef, TimeUnit};
+use arrow::record_batch::RecordBatch;
+use arrow::util::pretty;
+use datafusion::common::Result;
+use datafusion::datasource::listing::{ListingTableUrl, PartitionedFile};
+use datafusion::datasource::object_store::ObjectStoreUrl;
+use datafusion::execution::context::ExecutionProps;
+use datafusion::logical_expr::{lit, or, Expr};
+use datafusion::logical_plan::ToDFSchema;
+use datafusion::physical_expr::create_physical_expr;
+use datafusion::physical_plan::collect;
+use datafusion::physical_plan::file_format::{
+ FileScanConfig, ParquetExec, ParquetScanOptions,
+};
+use datafusion::physical_plan::filter::FilterExec;
+use datafusion::prelude::{col, combine_filters, SessionConfig, SessionContext};
+use object_store::path::Path;
+use object_store::ObjectMeta;
+use parquet::arrow::ArrowWriter;
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
+use std::fs::File;
+use std::ops::Range;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Instant;
+use structopt::StructOpt;
+
+#[cfg(feature = "snmalloc")]
+#[global_allocator]
+static ALLOC: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc;
+
+#[derive(Debug, StructOpt)]
+#[structopt(name = "Benchmarks", about = "Apache Arrow Rust Benchmarks.")]
+struct Opt {
+ /// Activate debug mode to see query results
+ #[structopt(short, long)]
+ debug: bool,
+
+ /// Number of iterations of each test run
+ #[structopt(short = "i", long = "iterations", default_value = "3")]
+ iterations: usize,
+
+ /// Number of partitions to process in parallel
+ #[structopt(long = "partitions", default_value = "2")]
+ partitions: usize,
+
+ /// Path to folder where access log file will be generated
+ #[structopt(parse(from_os_str), required = true, short = "p", long = "path")]
+ path: PathBuf,
+
+ /// Batch size when reading Parquet files
+ #[structopt(short = "s", long = "batch-size", default_value = "8192")]
+ batch_size: usize,
+
+ /// Total size of generated dataset. The default scale factor of 1.0 will generate a roughly 1GB parquet file
+ #[structopt(short = "s", long = "scale-factor", default_value = "1.0")]
+ scale_factor: f32,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+ let opt: Opt = Opt::from_args();
+ println!("Running benchmarks with the following options: {:?}", opt);
+
+ let config = SessionConfig::new()
+ .with_target_partitions(opt.partitions)
+ .with_batch_size(opt.batch_size);
+ let mut ctx = SessionContext::with_config(config);
+
+ let path = opt.path.join("logs.parquet");
+
+ let (object_store_url, object_meta) = gen_data(path, opt.scale_factor)?;
+
+ run_benchmarks(
+ &mut ctx,
+ object_store_url.clone(),
+ object_meta.clone(),
+ opt.iterations,
+ opt.debug,
+ )
+ .await?;
+
+ Ok(())
+}
+
+async fn run_benchmarks(
+ ctx: &mut SessionContext,
+ object_store_url: ObjectStoreUrl,
+ object_meta: ObjectMeta,
+ iterations: usize,
+ debug: bool,
+) -> Result<()> {
+ let scan_options_matrix = vec![
+ ParquetScanOptions::default(),
+ ParquetScanOptions::default()
+ .with_page_index(true)
+ .with_pushdown_filters(true)
+ .with_reorder_predicates(true),
+ ParquetScanOptions::default()
+ .with_page_index(true)
+ .with_pushdown_filters(true)
+ .with_reorder_predicates(false),
+ ];
+
+ let filter_matrix = vec![
+ // Selective-ish filter
+ col("request_method").eq(lit("GET")),
+ // Non-selective filter
+ col("request_method").not_eq(lit("GET")),
+ // Basic conjunction
+ col("request_method")
+ .eq(lit("POST"))
+ .and(col("response_status").eq(lit(503_u16))),
+ // Nested filters
+ col("request_method").eq(lit("POST")).and(or(
+ col("response_status").eq(lit(503_u16)),
+ col("response_status").eq(lit(403_u16)),
+ )),
+ // Many filters
+ combine_filters(&[
+ col("request_method").not_eq(lit("GET")),
+ col("response_status").eq(lit(400_u16)),
+ // TODO this fails in the FilterExec with Error: Internal("The type of Dictionary(Int32, Utf8) = Utf8 of binary physical should be same")
+ // col("service").eq(lit("backend")),
+ ])
+ .unwrap(),
+ // Filter everything
+ col("response_status").eq(lit(429_u16)),
+ // Filter nothing
+ col("response_status").gt(lit(0_u16)),
+ ];
+
+ for filter_expr in &filter_matrix {
+ println!("Executing with filter '{}'", filter_expr);
+ for scan_options in &scan_options_matrix {
+ println!("Using scan options {:?}", scan_options);
+ for i in 0..iterations {
+ let start = Instant::now();
+ let rows = exec_scan(
+ ctx,
+ object_store_url.clone(),
+ object_meta.clone(),
+ filter_expr.clone(),
+ scan_options.clone(),
+ debug,
+ )
+ .await?;
+ println!(
+ "Iteration {} returned {} rows in {} ms",
+ i,
+ rows,
+ start.elapsed().as_millis()
+ );
+ }
+ }
+ println!("\n");
+ }
+ Ok(())
+}
+
+async fn exec_scan(
+ ctx: &SessionContext,
+ object_store_url: ObjectStoreUrl,
+ object_meta: ObjectMeta,
+ filter: Expr,
+ scan_options: ParquetScanOptions,
+ debug: bool,
+) -> Result<usize> {
+ let schema = BatchBuilder::schema();
+ let scan_config = FileScanConfig {
+ object_store_url,
+ file_schema: schema.clone(),
+ file_groups: vec![vec![PartitionedFile {
+ object_meta,
+ partition_values: vec![],
+ range: None,
+ extensions: None,
+ }]],
+ statistics: Default::default(),
+ projection: None,
+ limit: None,
+ table_partition_cols: vec![],
+ };
+
+ let df_schema = schema.clone().to_dfschema()?;
+
+ let physical_filter_expr = create_physical_expr(
+ &filter,
+ &df_schema,
+ schema.as_ref(),
+ &ExecutionProps::default(),
+ )?;
+
+ let parquet_exec = Arc::new(
+ ParquetExec::new(scan_config, Some(filter), None).with_scan_options(scan_options),
+ );
+
+ let exec = Arc::new(FilterExec::try_new(physical_filter_expr, parquet_exec)?);
+
+ let task_ctx = ctx.task_ctx();
+ let result = collect(exec, task_ctx).await?;
+
+ if debug {
+ pretty::print_batches(&result)?;
+ }
+ Ok(result.iter().map(|b| b.num_rows()).sum())
+}
+
+fn gen_data(path: PathBuf, scale_factor: f32) -> Result<(ObjectStoreUrl, ObjectMeta)> {
+ let generator = Generator::new();
+
+ let file = File::create(&path).unwrap();
+ let mut writer = ArrowWriter::try_new(file, generator.schema.clone(), None).unwrap();
Review Comment:
Yeah, I'll need to revisit this again once https://github.com/apache/arrow-rs/pull/2854 is released and pulled in so we cam generate the files with proper page sizes (which should make a significant difference)
--
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
[GitHub] [arrow-datafusion] thinkharderdev commented on pull request #3769: Add benchmarks for testing row filtering
Posted by GitBox <gi...@apache.org>.
thinkharderdev commented on PR #3769:
URL: https://github.com/apache/arrow-datafusion/pull/3769#issuecomment-1272332472
@alamb @tustvold @Ted-Jiang
--
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
[GitHub] [arrow-datafusion] thinkharderdev commented on a diff in pull request #3769: Add benchmarks for testing row filtering
Posted by GitBox <gi...@apache.org>.
thinkharderdev commented on code in PR #3769:
URL: https://github.com/apache/arrow-datafusion/pull/3769#discussion_r992723442
##########
benchmarks/src/bin/parquet_filter_pushdown.rs:
##########
@@ -0,0 +1,462 @@
+// 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::array::{
+ Int32Builder, StringBuilder, StringDictionaryBuilder, TimestampNanosecondBuilder,
+ UInt16Builder,
+};
+use arrow::datatypes::{DataType, Field, Int32Type, Schema, SchemaRef, TimeUnit};
+use arrow::record_batch::RecordBatch;
+use arrow::util::pretty;
+use datafusion::common::Result;
+use datafusion::datasource::listing::{ListingTableUrl, PartitionedFile};
+use datafusion::datasource::object_store::ObjectStoreUrl;
+use datafusion::execution::context::ExecutionProps;
+use datafusion::logical_expr::{lit, or, Expr};
+use datafusion::logical_plan::ToDFSchema;
+use datafusion::physical_expr::create_physical_expr;
+use datafusion::physical_plan::collect;
+use datafusion::physical_plan::file_format::{
+ FileScanConfig, ParquetExec, ParquetScanOptions,
+};
+use datafusion::physical_plan::filter::FilterExec;
+use datafusion::prelude::{col, combine_filters, SessionConfig, SessionContext};
+use object_store::path::Path;
+use object_store::ObjectMeta;
+use parquet::arrow::ArrowWriter;
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
+use std::fs::File;
+use std::ops::Range;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Instant;
+use structopt::StructOpt;
+
+#[cfg(feature = "snmalloc")]
+#[global_allocator]
+static ALLOC: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc;
+
+#[derive(Debug, StructOpt)]
+#[structopt(name = "Benchmarks", about = "Apache Arrow Rust Benchmarks.")]
+struct Opt {
+ /// Activate debug mode to see query results
+ #[structopt(short, long)]
+ debug: bool,
+
+ /// Number of iterations of each test run
+ #[structopt(short = "i", long = "iterations", default_value = "3")]
+ iterations: usize,
+
+ /// Number of partitions to process in parallel
+ #[structopt(long = "partitions", default_value = "2")]
+ partitions: usize,
+
+ /// Path to folder where access log file will be generated
+ #[structopt(parse(from_os_str), required = true, short = "p", long = "path")]
+ path: PathBuf,
+
+ /// Batch size when reading Parquet files
+ #[structopt(short = "s", long = "batch-size", default_value = "8192")]
Review Comment:
Yeah, just removed one of them. I don't think batch size needs to be a cli option in this benchmark.
--
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
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #3769: Add benchmarks for testing row filtering
Posted by GitBox <gi...@apache.org>.
alamb commented on code in PR #3769:
URL: https://github.com/apache/arrow-datafusion/pull/3769#discussion_r992715070
##########
benchmarks/src/bin/parquet_filter_pushdown.rs:
##########
@@ -0,0 +1,462 @@
+// 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::array::{
+ Int32Builder, StringBuilder, StringDictionaryBuilder, TimestampNanosecondBuilder,
+ UInt16Builder,
+};
+use arrow::datatypes::{DataType, Field, Int32Type, Schema, SchemaRef, TimeUnit};
+use arrow::record_batch::RecordBatch;
+use arrow::util::pretty;
+use datafusion::common::Result;
+use datafusion::datasource::listing::{ListingTableUrl, PartitionedFile};
+use datafusion::datasource::object_store::ObjectStoreUrl;
+use datafusion::execution::context::ExecutionProps;
+use datafusion::logical_expr::{lit, or, Expr};
+use datafusion::logical_plan::ToDFSchema;
+use datafusion::physical_expr::create_physical_expr;
+use datafusion::physical_plan::collect;
+use datafusion::physical_plan::file_format::{
+ FileScanConfig, ParquetExec, ParquetScanOptions,
+};
+use datafusion::physical_plan::filter::FilterExec;
+use datafusion::prelude::{col, combine_filters, SessionConfig, SessionContext};
+use object_store::path::Path;
+use object_store::ObjectMeta;
+use parquet::arrow::ArrowWriter;
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
+use std::fs::File;
+use std::ops::Range;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Instant;
+use structopt::StructOpt;
+
+#[cfg(feature = "snmalloc")]
+#[global_allocator]
+static ALLOC: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc;
+
+#[derive(Debug, StructOpt)]
+#[structopt(name = "Benchmarks", about = "Apache Arrow Rust Benchmarks.")]
+struct Opt {
+ /// Activate debug mode to see query results
+ #[structopt(short, long)]
+ debug: bool,
+
+ /// Number of iterations of each test run
+ #[structopt(short = "i", long = "iterations", default_value = "3")]
+ iterations: usize,
+
+ /// Number of partitions to process in parallel
+ #[structopt(long = "partitions", default_value = "2")]
+ partitions: usize,
+
+ /// Path to folder where access log file will be generated
+ #[structopt(parse(from_os_str), required = true, short = "p", long = "path")]
+ path: PathBuf,
+
+ /// Batch size when reading Parquet files
+ #[structopt(short = "s", long = "batch-size", default_value = "8192")]
Review Comment:
in fact when you run the example in debug mode it asserts on exactly this problem:
```shell
cargo run --bin parquet_filter_pushdown -- --path ./data --\|alamb@aal-dev:~/arrow-datafusion$
scale-factor 1.0
...
Running `target/debug/parquet_filter_pushdown --path ./data --scale-factor 1.0` |error[E0433]: failed to resolve: use of undeclared type `WriterProperties`
thread 'main' panicked at 'Argument short must be unique | --> benchmarks/src/bin/parquet_filter_pushdown.rs:235:17
| |
-s is already in use', /home/alamb/.cargo/registry/src/github.com-1ecc6299db9ec823/cla\|235 | let props = WriterProperties::builder()
p-2.34.0/src/app/parser.rs:190:13 | | ^^^^^^^^^^^^^^^^ not found in this scope
note: run with `RUST_BACKTRACE=1` environment variable to display a ba
```
--
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
[GitHub] [arrow-datafusion] Ted-Jiang commented on pull request #3769: Add benchmarks for testing row filtering
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on PR #3769:
URL: https://github.com/apache/arrow-datafusion/pull/3769#issuecomment-1272454236
@thinkharderdev thanks for your great bench.
I run parquet tools in local get (1.0 GB)
```
(venv) yangjiang@LM-SHC-15009782 data % parquet-tools column-index ./logs.parquet
row group 0:
column index for column service:
Boudary order: UNORDERED
null count min max
page-0 0 backend frontend
offset index for column service:
offset compressed size first row index
page-0 62 117 0
column index for column host:
Boudary order: UNORDERED
null count min max
page-0 0 i-1ec3ca3151468928.ec2.internal i-1ec408f54dbd3750.ec2.internal
offset index for column host:
offset compressed size first row index
page-0 566 125 0
column index for column pod:
Boudary order: UNORDERED
null count min max
page-0 0 aejowuublavflbbsvlfozigwpmrxldvhaollk zxxlzhdrucrhpicpdgxtfpyuknvviimggtq
offset index for column pod:
offset compressed size first row index
page-0 6689 602 0
column index for column container:
Boudary order: UNORDERED
null count min max
page-0 0 backend_container_0 frontend_container_1
offset index for column container:
offset compressed size first row index
page-0 7602 593 0
```
There are at most two pages in one col, I think if we adjust to get more pages in one col (like reduce the page size), it will get greater performance in enable `enable_page_index `, we can get more opportunitys to skip whole pages without decoding! 🤔
FYI, i see impala choose to use fixed row number in one page to do benchmark for getting good performance.
--
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
[GitHub] [arrow-datafusion] alamb merged pull request #3769: Add benchmarks for testing row filtering
Posted by GitBox <gi...@apache.org>.
alamb merged PR #3769:
URL: https://github.com/apache/arrow-datafusion/pull/3769
--
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
[GitHub] [arrow-datafusion] thinkharderdev commented on pull request #3769: Add benchmarks for testing row filtering
Posted by GitBox <gi...@apache.org>.
thinkharderdev commented on PR #3769:
URL: https://github.com/apache/arrow-datafusion/pull/3769#issuecomment-1275207603
> This looks great -- thank you @thinkharderdev
>
> I also verified the parquet file that was created
>
> ```shell
> $ du -s -h /tmp/data/logs.parquet
> 988M /tmp/data/logs.parquet
> ```
>
> It looks good to me (using the neat [`pqrs` ](https://github.com/manojkarthick/pqrs)tool from @manojkarthick)
>
> ```shell
> alamb@aal-dev:~/2022-10-05-slow-query-high-cardinality$ pqrs cat --csv /tmp/data/logs.parquet | head
>
> ############################
> File: /tmp/data/logs.parquet
> ############################
>
> service,host,pod,container,image,time,client_addr,request_duration_ns,request_user_agent,request_method,request_host,request_bytes,response_bytes,response_status
> backend,i-1ec3ca3151468928.ec2.internal,aqcathnxqsphdhgjtgvxsfyiwbmhlmg,backend_container_0,backend_container_0@sha256:30375999bf03beec2187843017b10c9e88d8b1a91615df4eb6350fb39472edd9,1970-01\
> -01T00:00:00.000000000+00:00,127.216.178.64,-1261239112,rkxttrfiiietlsaygzphhwlqcgngnumuphliejmxfdznuurswhdcicrlprbnocibvsbukiohjjbjdygwbfhxqvurm,PUT,https://backend.mydomain.com,-312099516,1\
> 448834362,200
> backend,i-1ec3ca3151468928.ec2.internal,aqcathnxqsphdhgjtgvxsfyiwbmhlmg,backend_container_0,backend_container_0@sha256:30375999bf03beec2187843017b10c9e88d8b1a91615df4eb6350fb39472edd9,1970-01\
> -01T00:00:00.000001024+00:00,187.49.24.179,1374800428,sdxkctvmvuqxhwigrhjaouwdzvasqlqphymcgqvfmsbjswswnzgvanmalnmvsvruakcudmqvzateabhlya,PATCH,https://backend.mydomain.com,-1363067408,1111765\
> 98,204
> backend,i-1ec3ca3151468928.ec2.internal,aqcathnxqsphdhgjtgvxsfyiwbmhlmg,backend_container_0,backend_container_0@sha256:30375999bf03beec2187843017b10c9e88d8b1a91615df4eb6350fb39472edd9,1970-01\
> -01T00:00:00.000002048+00:00,14.29.229.168,-1795280692,bhlvymbbtgcqrwzujukyotusnsoidygnklhx,GET,https://backend.mydomain.com,-1323615082,-705662117,400
> backend,i-1ec3ca3151468928.ec2.internal,aqcathnxqsphdhgjtgvxsfyiwbmhlmg,backend_container_0,backend_container_0@sha256:30375999bf03beec2187843017b10c9e88d8b1a91615df4eb6350fb39472edd9,1970-01\
> -01T00:00:00.000003072+00:00,180.188.29.17,-717290117,hjaynltdswdekcguqmrkucsepzqjhasklmimkibabijihitimmsglgettywifdzmraipvyvekczuwxettayslrffyz,HEAD,https://backend.mydomain.com,-1847395296,\
> 1206750179,200
> backend,i-1ec3ca3151468928.ec2.internal,aqcathnxqsphdhgjtgvxsfyiwbmhlmg,backend_container_0,backend_container_0@sha256:30375999bf03beec2187843017b10c9e88d8b1a91615df4eb6350fb39472edd9,1970-01\
> -01T00:00:00.000004096+00:00,68.92.115.208,759902764,yupopowlaqbwskdwvtlitugpzzxoajhvnmndhca,DELETE,https://backend.mydomain.com,-50170254,-415949533,403
> backend,i-1ec3ca3151468928.ec2.internal,aqcathnxqsphdhgjtgvxsfyiwbmhlmg,backend_container_0,backend_container_0@sha256:30375999bf03beec2187843017b10c9e88d8b1a91615df4eb6350fb39472edd9,1970-01\
> -01T00:00:00.000005120+00:00,230.160.203.201,-1271567754,pwbruedgdgtsavjuksxwkecxulbnjbsaltuvcjxcmblhnraawouvrunwwsmvjbq,GET,https://backend.mydomain.com,-1193079450,1281912293,204
> backend,i-1ec3ca3151468928.ec2.internal,aqcathnxqsphdhgjtgvxsfyiwbmhlmg,backend_container_0,backend_container_0@sha256:30375999bf03beec2187843017b10c9e88d8b1a91615df4eb6350fb39472edd9,1970-01\
> -01T00:00:00.000006144+00:00,249.254.50.191,-971196614,amtuqookzibtvrtqfnyzuyesikbrafhcfnjhoaoedvmlwpkypfsedtbbwlbnzigwgjpzcwdxtwhrykcibmhlxnkckynvgli,PATCH,https://backend.mydomain.com,-2627\
> 74709,-1695212300,204
> backend,i-1ec3ca3151468928.ec2.internal,aqcathnxqsphdhgjtgvxsfyiwbmhlmg,backend_container_0,backend_container_0@sha256:30375999bf03beec2187843017b10c9e88d8b1a91615df4eb6350fb39472edd9,1970-01\
> -01T00:00:00.000007168+00:00,77.183.81.164,-547300163,ogkufdxssjqzjphxwvegwvofchpsgntbyslgarcyqcawokzfoppdftoctmtlwcvikazwrujlgrzrlqueaaceibxvdicfhp,HEAD,https://backend.mydomain.com,-1349820\
> 595,-327759246,204
> backend,i-1ec3ca3151468928.ec2.internal,aqcathnxqsphdhgjtgvxsfyiwbmhlmg,backend_container_0,backend_container_0@sha256:30375999bf03beec2187843017b10c9e88d8b1a91615df4eb6350fb39472edd9,1970-01\
> -01T00:00:00.000008192+00:00,63.17.88.115,-88404773,ogardohhoorttptpnkxmvyenqfzvvkjabcrfwapoywttjdunvmlgwgstmsjbefxqta,HEAD,https://backend.mydomain.com,1830978558,,200
> Error: ArrowReadWriteError(CsvError("Broken pipe (os error 32)"))
> ```
I stole the gen code from @tustvold so you know it works :)
--
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
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #3769: Add benchmarks for testing row filtering
Posted by GitBox <gi...@apache.org>.
alamb commented on code in PR #3769:
URL: https://github.com/apache/arrow-datafusion/pull/3769#discussion_r993296160
##########
benchmarks/src/bin/parquet_filter_pushdown.rs:
##########
@@ -0,0 +1,487 @@
+// 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::array::{
+ Int32Builder, StringBuilder, StringDictionaryBuilder, TimestampNanosecondBuilder,
+ UInt16Builder,
+};
+use arrow::datatypes::{DataType, Field, Int32Type, Schema, SchemaRef, TimeUnit};
+use arrow::record_batch::RecordBatch;
+use arrow::util::pretty;
+use datafusion::common::Result;
+use datafusion::datasource::listing::{ListingTableUrl, PartitionedFile};
+use datafusion::datasource::object_store::ObjectStoreUrl;
+use datafusion::execution::context::ExecutionProps;
+use datafusion::logical_expr::{lit, or, Expr};
+use datafusion::logical_plan::ToDFSchema;
+use datafusion::physical_expr::create_physical_expr;
+use datafusion::physical_plan::collect;
+use datafusion::physical_plan::file_format::{
+ FileScanConfig, ParquetExec, ParquetScanOptions,
+};
+use datafusion::physical_plan::filter::FilterExec;
+use datafusion::prelude::{col, combine_filters, SessionConfig, SessionContext};
+use object_store::path::Path;
+use object_store::ObjectMeta;
+use parquet::arrow::ArrowWriter;
+use parquet::file::properties::WriterProperties;
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
+use std::fs::File;
+use std::ops::Range;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Instant;
+use structopt::StructOpt;
+
+#[cfg(feature = "snmalloc")]
+#[global_allocator]
+static ALLOC: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc;
+
+#[derive(Debug, StructOpt)]
+#[structopt(name = "Benchmarks", about = "Apache Arrow Rust Benchmarks.")]
+struct Opt {
+ /// Activate debug mode to see query results
+ #[structopt(short, long)]
+ debug: bool,
+
+ /// Number of iterations of each test run
+ #[structopt(short = "i", long = "iterations", default_value = "3")]
+ iterations: usize,
+
+ /// Number of partitions to process in parallel
+ #[structopt(long = "partitions", default_value = "2")]
+ partitions: usize,
+
+ /// Path to folder where access log file will be generated
+ #[structopt(parse(from_os_str), required = true, short = "p", long = "path")]
+ path: PathBuf,
+
+ /// Data page size of the generated parquet file
+ #[structopt(long = "page-size")]
+ page_size: Option<usize>,
+
+ /// Data page size of the generated parquet file
+ #[structopt(long = "row-group-size")]
+ row_group_size: Option<usize>,
+
+ /// Total size of generated dataset. The default scale factor of 1.0 will generate a roughly 1GB parquet file
+ #[structopt(short = "s", long = "scale-factor", default_value = "1.0")]
+ scale_factor: f32,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+ let opt: Opt = Opt::from_args();
+ println!("Running benchmarks with the following options: {:?}", opt);
+
+ let config = SessionConfig::new().with_target_partitions(opt.partitions);
+ let mut ctx = SessionContext::with_config(config);
+
+ let path = opt.path.join("logs.parquet");
+
+ let (object_store_url, object_meta) =
+ gen_data(path, opt.scale_factor, opt.page_size, opt.row_group_size)?;
+
+ run_benchmarks(
+ &mut ctx,
+ object_store_url.clone(),
+ object_meta.clone(),
+ opt.iterations,
+ opt.debug,
+ )
+ .await?;
+
+ Ok(())
+}
+
+async fn run_benchmarks(
+ ctx: &mut SessionContext,
+ object_store_url: ObjectStoreUrl,
+ object_meta: ObjectMeta,
+ iterations: usize,
+ debug: bool,
+) -> Result<()> {
+ let scan_options_matrix = vec![
+ ParquetScanOptions::default(),
+ ParquetScanOptions::default()
+ .with_page_index(true)
+ .with_pushdown_filters(true)
+ .with_reorder_predicates(true),
+ ParquetScanOptions::default()
+ .with_page_index(true)
+ .with_pushdown_filters(true)
+ .with_reorder_predicates(false),
+ ];
+
+ let filter_matrix = vec![
+ // Selective-ish filter
+ col("request_method").eq(lit("GET")),
+ // Non-selective filter
+ col("request_method").not_eq(lit("GET")),
+ // Basic conjunction
+ col("request_method")
+ .eq(lit("POST"))
+ .and(col("response_status").eq(lit(503_u16))),
+ // Nested filters
+ col("request_method").eq(lit("POST")).and(or(
+ col("response_status").eq(lit(503_u16)),
+ col("response_status").eq(lit(403_u16)),
+ )),
+ // Many filters
+ combine_filters(&[
+ col("request_method").not_eq(lit("GET")),
+ col("response_status").eq(lit(400_u16)),
+ // TODO this fails in the FilterExec with Error: Internal("The type of Dictionary(Int32, Utf8) = Utf8 of binary physical should be same")
Review Comment:
Coercion!
--
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
[GitHub] [arrow-datafusion] alamb commented on pull request #3769: Add benchmarks for testing row filtering
Posted by GitBox <gi...@apache.org>.
alamb commented on PR #3769:
URL: https://github.com/apache/arrow-datafusion/pull/3769#issuecomment-1275960252
CI check is unrelated : https://github.com/apache/arrow-datafusion/issues/3798
--
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
[GitHub] [arrow-datafusion] ursabot commented on pull request #3769: Add benchmarks for testing row filtering
Posted by GitBox <gi...@apache.org>.
ursabot commented on PR #3769:
URL: https://github.com/apache/arrow-datafusion/pull/3769#issuecomment-1275969922
Benchmark runs are scheduled for baseline = ae5b23ec3c9b22e917c6e1b0a15e20396c82b8bd and contender = fb39d5d9fcab5cdab69e35505b877048a73a2e2e. fb39d5d9fcab5cdab69e35505b877048a73a2e2e is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on ec2-t3-xlarge-us-east-2] [ec2-t3-xlarge-us-east-2](https://conbench.ursa.dev/compare/runs/c8667d7c0ab047fa93b18eb3a1a19686...7228d5c0f059428da11ffc863a8d536c/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on test-mac-arm] [test-mac-arm](https://conbench.ursa.dev/compare/runs/28b0551374414edca232b8200b0f9f24...fb6dbbe047934e67a2ddfc0bd84db71d/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on ursa-i9-9960x] [ursa-i9-9960x](https://conbench.ursa.dev/compare/runs/873b6c8764f2478b8827abd55ee0401a...bbedffab261d4aa5a453790e5585ec60/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on ursa-thinkcentre-m75q] [ursa-thinkcentre-m75q](https://conbench.ursa.dev/compare/runs/be860d950b3640bfa96f3ff27315398b...2cbefedba463423cab380457f6ba28f6/)
Buildkite builds:
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java
--
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
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #3769: Add benchmarks for testing row filtering
Posted by GitBox <gi...@apache.org>.
alamb commented on code in PR #3769:
URL: https://github.com/apache/arrow-datafusion/pull/3769#discussion_r992689413
##########
benchmarks/README.md:
##########
@@ -126,3 +126,37 @@ h2o groupby query 1 took 1669 ms
[1]: http://www.tpc.org/tpch/
[2]: https://www1.nyc.gov/site/tlc/about/tlc-trip-record-data.page
+
+## Parquet filter pushdown benchmarks
+
+This is a set of benchmarks for testing and verifying performance of parquet filter pushdown. The queries are executed on
+a synthetic dataset generated during the benchmark execution and designed to simulate web server access logs.
+
+```base
+cargo run --release --bin parquet_filter_pushdown --query --path ./data --scale-factor 1.0
Review Comment:
When I tried this command, I got
```shell
alamb@aal-dev:~/arrow-datafusion$ cargo run --release --bin parquet_filter_pushdown --query --path ./data --scale-facto
error: Found argument '--query' which wasn't expected, or isn't valid in this context
If you tried to supply `--query` as a value rather than a flag, use `-- --query`
USAGE:
cargo run --release --bin [<NAME>]
```
Also, as @Ted-Jiang mentions, the `--query` argument is not correct and I get an error if it is included:
```
error: Found argument '--query' which wasn't expected, or isn't valid in this context
```
```suggestion
cargo run --release --bin parquet_filter_pushdown -- --path ./data --scale-factor 1.0
```
##########
benchmarks/src/bin/parquet_filter_pushdown.rs:
##########
@@ -0,0 +1,462 @@
+// 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::array::{
+ Int32Builder, StringBuilder, StringDictionaryBuilder, TimestampNanosecondBuilder,
+ UInt16Builder,
+};
+use arrow::datatypes::{DataType, Field, Int32Type, Schema, SchemaRef, TimeUnit};
+use arrow::record_batch::RecordBatch;
+use arrow::util::pretty;
+use datafusion::common::Result;
+use datafusion::datasource::listing::{ListingTableUrl, PartitionedFile};
+use datafusion::datasource::object_store::ObjectStoreUrl;
+use datafusion::execution::context::ExecutionProps;
+use datafusion::logical_expr::{lit, or, Expr};
+use datafusion::logical_plan::ToDFSchema;
+use datafusion::physical_expr::create_physical_expr;
+use datafusion::physical_plan::collect;
+use datafusion::physical_plan::file_format::{
+ FileScanConfig, ParquetExec, ParquetScanOptions,
+};
+use datafusion::physical_plan::filter::FilterExec;
+use datafusion::prelude::{col, combine_filters, SessionConfig, SessionContext};
+use object_store::path::Path;
+use object_store::ObjectMeta;
+use parquet::arrow::ArrowWriter;
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
+use std::fs::File;
+use std::ops::Range;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Instant;
+use structopt::StructOpt;
+
+#[cfg(feature = "snmalloc")]
+#[global_allocator]
+static ALLOC: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc;
+
+#[derive(Debug, StructOpt)]
+#[structopt(name = "Benchmarks", about = "Apache Arrow Rust Benchmarks.")]
+struct Opt {
+ /// Activate debug mode to see query results
+ #[structopt(short, long)]
+ debug: bool,
+
+ /// Number of iterations of each test run
+ #[structopt(short = "i", long = "iterations", default_value = "3")]
+ iterations: usize,
+
+ /// Number of partitions to process in parallel
+ #[structopt(long = "partitions", default_value = "2")]
+ partitions: usize,
+
+ /// Path to folder where access log file will be generated
+ #[structopt(parse(from_os_str), required = true, short = "p", long = "path")]
+ path: PathBuf,
+
+ /// Batch size when reading Parquet files
+ #[structopt(short = "s", long = "batch-size", default_value = "8192")]
+ batch_size: usize,
+
+ /// Total size of generated dataset. The default scale factor of 1.0 will generate a roughly 1GB parquet file
+ #[structopt(short = "s", long = "scale-factor", default_value = "1.0")]
+ scale_factor: f32,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+ let opt: Opt = Opt::from_args();
+ println!("Running benchmarks with the following options: {:?}", opt);
+
+ let config = SessionConfig::new()
+ .with_target_partitions(opt.partitions)
+ .with_batch_size(opt.batch_size);
+ let mut ctx = SessionContext::with_config(config);
+
+ let path = opt.path.join("logs.parquet");
+
+ let (object_store_url, object_meta) = gen_data(path, opt.scale_factor)?;
+
+ run_benchmarks(
+ &mut ctx,
+ object_store_url.clone(),
+ object_meta.clone(),
+ opt.iterations,
+ opt.debug,
+ )
+ .await?;
+
+ Ok(())
+}
+
+async fn run_benchmarks(
+ ctx: &mut SessionContext,
+ object_store_url: ObjectStoreUrl,
+ object_meta: ObjectMeta,
+ iterations: usize,
+ debug: bool,
+) -> Result<()> {
+ let scan_options_matrix = vec![
+ ParquetScanOptions::default(),
+ ParquetScanOptions::default()
+ .with_page_index(true)
+ .with_pushdown_filters(true)
+ .with_reorder_predicates(true),
+ ParquetScanOptions::default()
+ .with_page_index(true)
+ .with_pushdown_filters(true)
+ .with_reorder_predicates(false),
+ ];
+
+ let filter_matrix = vec![
+ // Selective-ish filter
+ col("request_method").eq(lit("GET")),
+ // Non-selective filter
+ col("request_method").not_eq(lit("GET")),
+ // Basic conjunction
+ col("request_method")
+ .eq(lit("POST"))
+ .and(col("response_status").eq(lit(503_u16))),
+ // Nested filters
+ col("request_method").eq(lit("POST")).and(or(
+ col("response_status").eq(lit(503_u16)),
+ col("response_status").eq(lit(403_u16)),
+ )),
+ // Many filters
+ combine_filters(&[
+ col("request_method").not_eq(lit("GET")),
+ col("response_status").eq(lit(400_u16)),
+ // TODO this fails in the FilterExec with Error: Internal("The type of Dictionary(Int32, Utf8) = Utf8 of binary physical should be same")
+ // col("service").eq(lit("backend")),
+ ])
+ .unwrap(),
+ // Filter everything
+ col("response_status").eq(lit(429_u16)),
+ // Filter nothing
+ col("response_status").gt(lit(0_u16)),
+ ];
+
+ for filter_expr in &filter_matrix {
+ println!("Executing with filter '{}'", filter_expr);
+ for scan_options in &scan_options_matrix {
+ println!("Using scan options {:?}", scan_options);
+ for i in 0..iterations {
+ let start = Instant::now();
+ let rows = exec_scan(
+ ctx,
+ object_store_url.clone(),
+ object_meta.clone(),
+ filter_expr.clone(),
+ scan_options.clone(),
+ debug,
+ )
+ .await?;
+ println!(
+ "Iteration {} returned {} rows in {} ms",
+ i,
+ rows,
+ start.elapsed().as_millis()
+ );
+ }
+ }
+ println!("\n");
+ }
+ Ok(())
+}
+
+async fn exec_scan(
+ ctx: &SessionContext,
+ object_store_url: ObjectStoreUrl,
+ object_meta: ObjectMeta,
+ filter: Expr,
+ scan_options: ParquetScanOptions,
+ debug: bool,
+) -> Result<usize> {
+ let schema = BatchBuilder::schema();
+ let scan_config = FileScanConfig {
+ object_store_url,
+ file_schema: schema.clone(),
+ file_groups: vec![vec![PartitionedFile {
+ object_meta,
+ partition_values: vec![],
+ range: None,
+ extensions: None,
+ }]],
+ statistics: Default::default(),
+ projection: None,
+ limit: None,
+ table_partition_cols: vec![],
+ };
+
+ let df_schema = schema.clone().to_dfschema()?;
+
+ let physical_filter_expr = create_physical_expr(
+ &filter,
+ &df_schema,
+ schema.as_ref(),
+ &ExecutionProps::default(),
+ )?;
+
+ let parquet_exec = Arc::new(
+ ParquetExec::new(scan_config, Some(filter), None).with_scan_options(scan_options),
+ );
+
+ let exec = Arc::new(FilterExec::try_new(physical_filter_expr, parquet_exec)?);
+
+ let task_ctx = ctx.task_ctx();
+ let result = collect(exec, task_ctx).await?;
+
+ if debug {
+ pretty::print_batches(&result)?;
+ }
+ Ok(result.iter().map(|b| b.num_rows()).sum())
+}
+
+fn gen_data(path: PathBuf, scale_factor: f32) -> Result<(ObjectStoreUrl, ObjectMeta)> {
+ let generator = Generator::new();
+
+ let file = File::create(&path).unwrap();
+ let mut writer = ArrowWriter::try_new(file, generator.schema.clone(), None).unwrap();
Review Comment:
I wonder if we should make the properties used here explicit?
Like maybe explicitly set what type of statistics are created as well as potentially setting ZSTD compression
https://docs.rs/parquet/24.0.0/parquet/file/properties/struct.WriterPropertiesBuilder.html
--
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
[GitHub] [arrow-datafusion] alamb commented on pull request #3769: Add benchmarks for testing row filtering
Posted by GitBox <gi...@apache.org>.
alamb commented on PR #3769:
URL: https://github.com/apache/arrow-datafusion/pull/3769#issuecomment-1274786395
Thank you @thinkharderdev -- I plan to review this PR in detail later today
--
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