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 023b018311 support unknown col expr in proto (#13603)
023b018311 is described below
commit 023b0183119f858f9a355ab909805ec2f8647bfd
Author: Onur Satici <[email protected]>
AuthorDate: Sat Nov 30 12:43:14 2024 +0000
support unknown col expr in proto (#13603)
---
datafusion/proto/proto/datafusion.proto | 6 ++
datafusion/proto/src/generated/pbjson.rs | 105 +++++++++++++++++++++++
datafusion/proto/src/generated/prost.rs | 9 +-
datafusion/proto/src/physical_plan/from_proto.rs | 3 +-
datafusion/proto/src/physical_plan/to_proto.rs | 10 ++-
5 files changed, 130 insertions(+), 3 deletions(-)
diff --git a/datafusion/proto/proto/datafusion.proto
b/datafusion/proto/proto/datafusion.proto
index 504e5e1cee..cb0235f1f2 100644
--- a/datafusion/proto/proto/datafusion.proto
+++ b/datafusion/proto/proto/datafusion.proto
@@ -828,6 +828,8 @@ message PhysicalExprNode {
PhysicalLikeExprNode like_expr = 18;
PhysicalExtensionExprNode extension = 19;
+
+ UnknownColumn unknown_column = 20;
}
}
@@ -1067,6 +1069,10 @@ message PhysicalColumn {
uint32 index = 2;
}
+message UnknownColumn {
+ string name = 1;
+}
+
message JoinOn {
PhysicalExprNode left = 1;
PhysicalExprNode right = 2;
diff --git a/datafusion/proto/src/generated/pbjson.rs
b/datafusion/proto/src/generated/pbjson.rs
index 29920814a8..f920e16d0a 100644
--- a/datafusion/proto/src/generated/pbjson.rs
+++ b/datafusion/proto/src/generated/pbjson.rs
@@ -13991,6 +13991,9 @@ impl serde::Serialize for PhysicalExprNode {
physical_expr_node::ExprType::Extension(v) => {
struct_ser.serialize_field("extension", v)?;
}
+ physical_expr_node::ExprType::UnknownColumn(v) => {
+ struct_ser.serialize_field("unknownColumn", v)?;
+ }
}
}
struct_ser.end()
@@ -14031,6 +14034,8 @@ impl<'de> serde::Deserialize<'de> for PhysicalExprNode {
"like_expr",
"likeExpr",
"extension",
+ "unknown_column",
+ "unknownColumn",
];
#[allow(clippy::enum_variant_names)]
@@ -14052,6 +14057,7 @@ impl<'de> serde::Deserialize<'de> for PhysicalExprNode {
ScalarUdf,
LikeExpr,
Extension,
+ UnknownColumn,
}
impl<'de> serde::Deserialize<'de> for GeneratedField {
fn deserialize<D>(deserializer: D) ->
std::result::Result<GeneratedField, D::Error>
@@ -14090,6 +14096,7 @@ impl<'de> serde::Deserialize<'de> for PhysicalExprNode {
"scalarUdf" | "scalar_udf" =>
Ok(GeneratedField::ScalarUdf),
"likeExpr" | "like_expr" =>
Ok(GeneratedField::LikeExpr),
"extension" => Ok(GeneratedField::Extension),
+ "unknownColumn" | "unknown_column" =>
Ok(GeneratedField::UnknownColumn),
_ => Err(serde::de::Error::unknown_field(value,
FIELDS)),
}
}
@@ -14229,6 +14236,13 @@ impl<'de> serde::Deserialize<'de> for PhysicalExprNode
{
return
Err(serde::de::Error::duplicate_field("extension"));
}
expr_type__ =
map_.next_value::<::std::option::Option<_>>()?.map(physical_expr_node::ExprType::Extension)
+;
+ }
+ GeneratedField::UnknownColumn => {
+ if expr_type__.is_some() {
+ return
Err(serde::de::Error::duplicate_field("unknownColumn"));
+ }
+ expr_type__ =
map_.next_value::<::std::option::Option<_>>()?.map(physical_expr_node::ExprType::UnknownColumn)
;
}
}
@@ -20141,6 +20155,97 @@ impl<'de> serde::Deserialize<'de> for UnionNode {
deserializer.deserialize_struct("datafusion.UnionNode", FIELDS,
GeneratedVisitor)
}
}
+impl serde::Serialize for UnknownColumn {
+ #[allow(deprecated)]
+ fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok,
S::Error>
+ where
+ S: serde::Serializer,
+ {
+ use serde::ser::SerializeStruct;
+ let mut len = 0;
+ if !self.name.is_empty() {
+ len += 1;
+ }
+ let mut struct_ser =
serializer.serialize_struct("datafusion.UnknownColumn", len)?;
+ if !self.name.is_empty() {
+ struct_ser.serialize_field("name", &self.name)?;
+ }
+ struct_ser.end()
+ }
+}
+impl<'de> serde::Deserialize<'de> for UnknownColumn {
+ #[allow(deprecated)]
+ fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ const FIELDS: &[&str] = &[
+ "name",
+ ];
+
+ #[allow(clippy::enum_variant_names)]
+ enum GeneratedField {
+ Name,
+ }
+ impl<'de> serde::Deserialize<'de> for GeneratedField {
+ fn deserialize<D>(deserializer: D) ->
std::result::Result<GeneratedField, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ struct GeneratedVisitor;
+
+ impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+ type Value = GeneratedField;
+
+ fn expecting(&self, formatter: &mut
std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(formatter, "expected one of: {:?}", &FIELDS)
+ }
+
+ #[allow(unused_variables)]
+ fn visit_str<E>(self, value: &str) ->
std::result::Result<GeneratedField, E>
+ where
+ E: serde::de::Error,
+ {
+ match value {
+ "name" => Ok(GeneratedField::Name),
+ _ => Err(serde::de::Error::unknown_field(value,
FIELDS)),
+ }
+ }
+ }
+ deserializer.deserialize_identifier(GeneratedVisitor)
+ }
+ }
+ struct GeneratedVisitor;
+ impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+ type Value = UnknownColumn;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) ->
std::fmt::Result {
+ formatter.write_str("struct datafusion.UnknownColumn")
+ }
+
+ fn visit_map<V>(self, mut map_: V) ->
std::result::Result<UnknownColumn, V::Error>
+ where
+ V: serde::de::MapAccess<'de>,
+ {
+ let mut name__ = None;
+ while let Some(k) = map_.next_key()? {
+ match k {
+ GeneratedField::Name => {
+ if name__.is_some() {
+ return
Err(serde::de::Error::duplicate_field("name"));
+ }
+ name__ = Some(map_.next_value()?);
+ }
+ }
+ }
+ Ok(UnknownColumn {
+ name: name__.unwrap_or_default(),
+ })
+ }
+ }
+ deserializer.deserialize_struct("datafusion.UnknownColumn", FIELDS,
GeneratedVisitor)
+ }
+}
impl serde::Serialize for Unnest {
#[allow(deprecated)]
fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok,
S::Error>
diff --git a/datafusion/proto/src/generated/prost.rs
b/datafusion/proto/src/generated/prost.rs
index 07090b7cba..a2be3207ac 100644
--- a/datafusion/proto/src/generated/prost.rs
+++ b/datafusion/proto/src/generated/prost.rs
@@ -1165,7 +1165,7 @@ pub struct PhysicalExtensionNode {
pub struct PhysicalExprNode {
#[prost(
oneof = "physical_expr_node::ExprType",
- tags = "1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 14, 15, 16, 18, 19"
+ tags = "1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 14, 15, 16, 18, 19, 20"
)]
pub expr_type: ::core::option::Option<physical_expr_node::ExprType>,
}
@@ -1214,6 +1214,8 @@ pub mod physical_expr_node {
LikeExpr(::prost::alloc::boxed::Box<super::PhysicalLikeExprNode>),
#[prost(message, tag = "19")]
Extension(super::PhysicalExtensionExprNode),
+ #[prost(message, tag = "20")]
+ UnknownColumn(super::UnknownColumn),
}
}
#[derive(Clone, PartialEq, ::prost::Message)]
@@ -1566,6 +1568,11 @@ pub struct PhysicalColumn {
pub index: u32,
}
#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct UnknownColumn {
+ #[prost(string, tag = "1")]
+ pub name: ::prost::alloc::string::String,
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
pub struct JoinOn {
#[prost(message, optional, tag = "1")]
pub left: ::core::option::Option<PhysicalExprNode>,
diff --git a/datafusion/proto/src/physical_plan/from_proto.rs
b/datafusion/proto/src/physical_plan/from_proto.rs
index e528b38b84..5bf6218cb9 100644
--- a/datafusion/proto/src/physical_plan/from_proto.rs
+++ b/datafusion/proto/src/physical_plan/from_proto.rs
@@ -38,7 +38,7 @@ use datafusion::logical_expr::WindowFunctionDefinition;
use datafusion::physical_expr::{LexOrdering, PhysicalSortExpr,
ScalarFunctionExpr};
use datafusion::physical_plan::expressions::{
in_list, BinaryExpr, CaseExpr, CastExpr, Column, IsNotNullExpr,
IsNullExpr, LikeExpr,
- Literal, NegativeExpr, NotExpr, TryCastExpr,
+ Literal, NegativeExpr, NotExpr, TryCastExpr, UnKnownColumn,
};
use datafusion::physical_plan::windows::{create_window_expr,
schema_add_window_field};
use datafusion::physical_plan::{Partitioning, PhysicalExpr, WindowExpr};
@@ -219,6 +219,7 @@ pub fn parse_physical_expr(
let pcol: Column = c.into();
Arc::new(pcol)
}
+ ExprType::UnknownColumn(c) => Arc::new(UnKnownColumn::new(&c.name)),
ExprType::Literal(scalar) =>
Arc::new(Literal::new(scalar.try_into()?)),
ExprType::BinaryExpr(binary_expr) => Arc::new(BinaryExpr::new(
parse_required_physical_expr(
diff --git a/datafusion/proto/src/physical_plan/to_proto.rs
b/datafusion/proto/src/physical_plan/to_proto.rs
index c1bbb3ad26..3805b97059 100644
--- a/datafusion/proto/src/physical_plan/to_proto.rs
+++ b/datafusion/proto/src/physical_plan/to_proto.rs
@@ -23,7 +23,7 @@ use
datafusion::physical_expr::window::{SlidingAggregateWindowExpr, StandardWind
use datafusion::physical_expr::{LexOrdering, PhysicalSortExpr,
ScalarFunctionExpr};
use datafusion::physical_plan::expressions::{
BinaryExpr, CaseExpr, CastExpr, Column, InListExpr, IsNotNullExpr,
IsNullExpr,
- Literal, NegativeExpr, NotExpr, TryCastExpr,
+ Literal, NegativeExpr, NotExpr, TryCastExpr, UnKnownColumn,
};
use datafusion::physical_plan::udaf::AggregateFunctionExpr;
use datafusion::physical_plan::windows::{PlainAggregateWindowExpr,
WindowUDFExpr};
@@ -220,6 +220,14 @@ pub fn serialize_physical_expr(
},
)),
})
+ } else if let Some(expr) = expr.downcast_ref::<UnKnownColumn>() {
+ Ok(protobuf::PhysicalExprNode {
+ expr_type:
Some(protobuf::physical_expr_node::ExprType::UnknownColumn(
+ protobuf::UnknownColumn {
+ name: expr.name().to_string(),
+ },
+ )),
+ })
} else if let Some(expr) = expr.downcast_ref::<BinaryExpr>() {
let binary_expr = Box::new(protobuf::PhysicalBinaryExprNode {
l: Some(Box::new(serialize_physical_expr(expr.left(), codec)?)),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
