This is an automated email from the ASF dual-hosted git repository. agrove pushed a commit to branch comet-parquet-exec in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 22b648d26798c2d885e4a4fcff6a02a53de38dbe Author: Matt Butrovich <[email protected]> AuthorDate: Fri Nov 8 16:21:48 2024 -0500 filters? --- native/core/src/execution/datafusion/planner.rs | 43 ++++++++++++++++++---- native/proto/src/proto/operator.proto | 1 + .../org/apache/comet/serde/QueryPlanSerde.scala | 11 ++++-- 3 files changed, 43 insertions(+), 12 deletions(-) diff --git a/native/core/src/execution/datafusion/planner.rs b/native/core/src/execution/datafusion/planner.rs index 65ad73f8..aa858e44 100644 --- a/native/core/src/execution/datafusion/planner.rs +++ b/native/core/src/execution/datafusion/planner.rs @@ -119,7 +119,6 @@ use datafusion_physical_expr::window::WindowExpr; use itertools::Itertools; use jni::objects::GlobalRef; use num::{BigInt, ToPrimitive}; -use object_store::path::Path; use parquet::schema::parser::parse_message_type; use std::cmp::max; use std::{collections::HashMap, sync::Arc}; @@ -957,11 +956,13 @@ impl PhysicalPlanner { let data_schema_descriptor = parquet::schema::types::SchemaDescriptor::new(Arc::new(data_schema)); - let data_schema_arrow = parquet::arrow::schema::parquet_to_arrow_schema( - &data_schema_descriptor, - None, - ) - .unwrap(); + let data_schema_arrow = Arc::new( + parquet::arrow::schema::parquet_to_arrow_schema( + &data_schema_descriptor, + None, + ) + .unwrap(), + ); println!("data_schema_arrow: {:?}", data_schema_arrow); let required_schema_descriptor = @@ -985,6 +986,31 @@ impl PhysicalPlanner { assert_eq!(projection_vector.len(), required_schema_arrow.fields.len()); + // Convert the Spark expressions to Physical expressions + let data_filters: Result<Vec<Arc<dyn PhysicalExpr>>, ExecutionError> = scan + .data_filters + .iter() + .map(|expr| self.create_expr(expr, data_schema_arrow.clone())) + .collect(); + + // Create a conjunctive form of the vector because ParquetExecBuilder takes + // a single expression + let data_filters = data_filters?; + let test_data_filters = data_filters + .clone() + .into_iter() + .reduce(|left, right| { + Arc::new(BinaryExpr::new( + left, + datafusion::logical_expr::Operator::And, + right, + )) + }) + .unwrap(); + + println!("data_filters: {:?}", data_filters); + println!("test_data_filters: {:?}", test_data_filters); + let object_store_url = ObjectStoreUrl::local_filesystem(); let path = Url::parse(&*scan.path).unwrap(); @@ -998,11 +1024,12 @@ impl PhysicalPlanner { let file = PartitionedFile::from_path(path.path().to_string())?; let file_scan_config = - FileScanConfig::new(object_store_url, Arc::new(data_schema_arrow)) + FileScanConfig::new(object_store_url, data_schema_arrow.clone()) .with_file(file) .with_projection(Some(projection_vector)); - let builder = ParquetExecBuilder::new(file_scan_config); + let builder = + ParquetExecBuilder::new(file_scan_config).with_predicate(test_data_filters); let scan = builder.build(); return Ok((vec![], Arc::new(scan))); } diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto index 950b35ae..67070ee6 100644 --- a/native/proto/src/proto/operator.proto +++ b/native/proto/src/proto/operator.proto @@ -55,6 +55,7 @@ message Scan { string path = 3; string required_schema = 4; string data_schema = 5; + repeated spark.spark_expression.Expr data_filters = 6; } message Projection { diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala index 3775c7b2..e5929261 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -3157,11 +3157,14 @@ object QueryPlanSerde extends Logging with ShimQueryPlanSerde with CometExprShim // scalastyle:off println // System.out.println(op.simpleStringWithNodeId()) // System.out.println(scanTypes.asJava) // Spark types for output. -// System.out.println(cometScan.output) // This is the names of the output columns. + System.out.println(cometScan.output) // This is the names of the output columns. // System.out.println(cometScan.requiredSchema); // This is the projected columns. -// System.out.println( -// cometScan.dataFilters -// ); // This is the filter expressions that have been pushed down. + System.out.println( + cometScan.dataFilters + ); // This is the filter expressions that have been pushed down. + + val dataFilters = cometScan.dataFilters.map(exprToProto(_, cometScan.output)) + scanBuilder.addAllDataFilters(dataFilters.map(_.get).asJava) // System.out.println(cometScan.relation.location.inputFiles(0)) // System.out.println(cometScan.partitionFilters); // System.out.println(cometScan.relation.partitionSchema) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
