You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by ag...@apache.org on 2024/11/11 18:03:25 UTC
(datafusion-comet) 02/14: filters?
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 <mb...@gmail.com>
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: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org