You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by al...@apache.org on 2025/02/05 12:45:45 UTC
(datafusion) branch main updated: Serialize `parquet_options` in `datafusion-proto` (#14465)
This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 168fe49be4 Serialize `parquet_options` in `datafusion-proto` (#14465)
168fe49be4 is described below
commit 168fe49be4590f11c02b00d4b13bdd48b01ad544
Author: Dmitrii Blaginin <dm...@blaginin.me>
AuthorDate: Wed Feb 5 12:45:38 2025 +0000
Serialize `parquet_options` in `datafusion-proto` (#14465)
* Serialize `parquet_options`
* Fix format
---
datafusion/proto/proto/datafusion.proto | 2 ++
datafusion/proto/src/generated/pbjson.rs | 18 ++++++++++++++++++
datafusion/proto/src/generated/prost.rs | 4 ++++
datafusion/proto/src/physical_plan/mod.rs | 6 ++++++
.../proto/tests/cases/roundtrip_physical_plan.rs | 5 +++++
5 files changed, 35 insertions(+)
diff --git a/datafusion/proto/proto/datafusion.proto b/datafusion/proto/proto/datafusion.proto
index 57c2f5f51b..3bc884257d 100644
--- a/datafusion/proto/proto/datafusion.proto
+++ b/datafusion/proto/proto/datafusion.proto
@@ -1005,6 +1005,8 @@ message ParquetScanExecNode {
reserved 2;
PhysicalExprNode predicate = 3;
+
+ datafusion_common.TableParquetOptions parquet_options = 4;
}
message CsvScanExecNode {
diff --git a/datafusion/proto/src/generated/pbjson.rs b/datafusion/proto/src/generated/pbjson.rs
index 0cf893cbc5..add72e4f77 100644
--- a/datafusion/proto/src/generated/pbjson.rs
+++ b/datafusion/proto/src/generated/pbjson.rs
@@ -12129,6 +12129,9 @@ impl serde::Serialize for ParquetScanExecNode {
if self.predicate.is_some() {
len += 1;
}
+ if self.parquet_options.is_some() {
+ len += 1;
+ }
let mut struct_ser = serializer.serialize_struct("datafusion.ParquetScanExecNode", len)?;
if let Some(v) = self.base_conf.as_ref() {
struct_ser.serialize_field("baseConf", v)?;
@@ -12136,6 +12139,9 @@ impl serde::Serialize for ParquetScanExecNode {
if let Some(v) = self.predicate.as_ref() {
struct_ser.serialize_field("predicate", v)?;
}
+ if let Some(v) = self.parquet_options.as_ref() {
+ struct_ser.serialize_field("parquetOptions", v)?;
+ }
struct_ser.end()
}
}
@@ -12149,12 +12155,15 @@ impl<'de> serde::Deserialize<'de> for ParquetScanExecNode {
"base_conf",
"baseConf",
"predicate",
+ "parquet_options",
+ "parquetOptions",
];
#[allow(clippy::enum_variant_names)]
enum GeneratedField {
BaseConf,
Predicate,
+ ParquetOptions,
}
impl<'de> serde::Deserialize<'de> for GeneratedField {
fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
@@ -12178,6 +12187,7 @@ impl<'de> serde::Deserialize<'de> for ParquetScanExecNode {
match value {
"baseConf" | "base_conf" => Ok(GeneratedField::BaseConf),
"predicate" => Ok(GeneratedField::Predicate),
+ "parquetOptions" | "parquet_options" => Ok(GeneratedField::ParquetOptions),
_ => Err(serde::de::Error::unknown_field(value, FIELDS)),
}
}
@@ -12199,6 +12209,7 @@ impl<'de> serde::Deserialize<'de> for ParquetScanExecNode {
{
let mut base_conf__ = None;
let mut predicate__ = None;
+ let mut parquet_options__ = None;
while let Some(k) = map_.next_key()? {
match k {
GeneratedField::BaseConf => {
@@ -12213,11 +12224,18 @@ impl<'de> serde::Deserialize<'de> for ParquetScanExecNode {
}
predicate__ = map_.next_value()?;
}
+ GeneratedField::ParquetOptions => {
+ if parquet_options__.is_some() {
+ return Err(serde::de::Error::duplicate_field("parquetOptions"));
+ }
+ parquet_options__ = map_.next_value()?;
+ }
}
}
Ok(ParquetScanExecNode {
base_conf: base_conf__,
predicate: predicate__,
+ parquet_options: parquet_options__,
})
}
}
diff --git a/datafusion/proto/src/generated/prost.rs b/datafusion/proto/src/generated/prost.rs
index 26efb617e0..df32c1a70d 100644
--- a/datafusion/proto/src/generated/prost.rs
+++ b/datafusion/proto/src/generated/prost.rs
@@ -1517,6 +1517,10 @@ pub struct ParquetScanExecNode {
pub base_conf: ::core::option::Option<FileScanExecConf>,
#[prost(message, optional, tag = "3")]
pub predicate: ::core::option::Option<PhysicalExprNode>,
+ #[prost(message, optional, tag = "4")]
+ pub parquet_options: ::core::option::Option<
+ super::datafusion_common::TableParquetOptions,
+ >,
}
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct CsvScanExecNode {
diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs
index a266d55b46..9ded9122b7 100644
--- a/datafusion/proto/src/physical_plan/mod.rs
+++ b/datafusion/proto/src/physical_plan/mod.rs
@@ -258,6 +258,11 @@ impl AsExecutionPlan for protobuf::PhysicalPlanNode {
})
.transpose()?;
let mut builder = ParquetExec::builder(base_config);
+
+ if let Some(options) = scan.parquet_options.as_ref() {
+ builder = builder.with_table_parquet_options(options.try_into()?)
+ }
+
if let Some(predicate) = predicate {
builder = builder.with_predicate(predicate)
}
@@ -1654,6 +1659,7 @@ impl AsExecutionPlan for protobuf::PhysicalPlanNode {
extension_codec,
)?),
predicate,
+ parquet_options: Some(exec.table_parquet_options().try_into()?),
},
)),
});
diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
index 7c34fe0680..50c0802446 100644
--- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
@@ -730,9 +730,14 @@ fn roundtrip_parquet_exec_with_pruning_predicate() -> Result<()> {
Operator::Eq,
lit("1"),
));
+
+ let mut options = TableParquetOptions::new();
+ options.global.pushdown_filters = true;
+
roundtrip_test(
ParquetExec::builder(scan_config)
.with_predicate(predicate)
+ .with_table_parquet_options(options)
.build_arc(),
)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org