You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/06 07:01:49 UTC
(datafusion) branch main updated: fix: preserve literal metadata and destructure leaf and unary proto hooks (#25382)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 d9a93219ae fix: preserve literal metadata and destructure leaf and unary proto hooks (#25382)
d9a93219ae is described below
commit d9a93219aef11d9482bf8dc0495e99cb0858b30d
Author: Peter Lee <pe...@gmail.com>
AuthorDate: Tue Oct 6 06:52:10 2026 +0000
fix: preserve literal metadata and destructure leaf and unary proto hooks (#25382)
## Which issue does this PR close?
Closes #24613. Literal field metadata is lost when physical expressions
are serialized.
## Rationale for this change
Losing this metadata can strip Arrow extension types and custom metadata
from projected columns.
## What changes are included in this PR?
Use exhaustive destructuring in all seven leaf and unary expression
encoders and decoders. Add a literal protobuf message that carries
metadata and regenerate the Rust and JSON bindings.
## What is the testing strategy for this PR?
A projection round-trip test covers null and non-null literals through
protobuf and JSON and was verified to fail before the fix. Existing
tests cover plain literals, and a new test rejects messages missing a
value. The extended workspace tests and all-feature Clippy pass.
## Are there any user-facing changes?
Literal metadata survives physical plan serialization. Literals with
metadata use a new protobuf variant that requires an updated reader.
---
datafusion/physical-expr/src/expressions/column.rs | 20 +++-
.../physical-expr/src/expressions/is_not_null.rs | 11 +-
.../physical-expr/src/expressions/is_null.rs | 6 +-
.../physical-expr/src/expressions/literal.rs | 76 +++++++++++--
.../physical-expr/src/expressions/negative.rs | 6 +-
datafusion/physical-expr/src/expressions/not.rs | 7 +-
.../src/expressions/unknown_column.rs | 9 +-
datafusion/proto-models/proto/datafusion.proto | 6 +
datafusion/proto-models/src/generated/pbjson.rs | 124 +++++++++++++++++++++
datafusion/proto-models/src/generated/prost.rs | 14 ++-
datafusion/proto/src/physical_plan/from_proto.rs | 4 +-
datafusion/proto/tests/cases/plans/exprs.rs | 48 ++++++++
12 files changed, 293 insertions(+), 38 deletions(-)
diff --git a/datafusion/physical-expr/src/expressions/column.rs b/datafusion/physical-expr/src/expressions/column.rs
index 482ab6ef1e..ea00110ecc 100644
--- a/datafusion/physical-expr/src/expressions/column.rs
+++ b/datafusion/physical-expr/src/expressions/column.rs
@@ -153,9 +153,15 @@ impl PhysicalExpr for Column {
_ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
use datafusion_proto_models::protobuf;
+ let Self { name, index } = self;
Ok(Some(protobuf::PhysicalExprNode {
expr_id: None,
- expr_type: Some(protobuf::physical_expr_node::ExprType::Column(self.into())),
+ expr_type: Some(protobuf::physical_expr_node::ExprType::Column(
+ protobuf::PhysicalColumn {
+ name: name.clone(),
+ index: *index as u32,
+ },
+ )),
}))
}
}
@@ -163,16 +169,18 @@ impl PhysicalExpr for Column {
#[cfg(feature = "proto")]
impl From<&datafusion_proto_models::protobuf::PhysicalColumn> for Column {
fn from(c: &datafusion_proto_models::protobuf::PhysicalColumn) -> Self {
- Column::new(&c.name, c.index as usize)
+ let datafusion_proto_models::protobuf::PhysicalColumn { name, index } = c;
+ Column::new(name, *index as usize)
}
}
#[cfg(feature = "proto")]
impl From<&Column> for datafusion_proto_models::protobuf::PhysicalColumn {
fn from(c: &Column) -> Self {
+ let Column { name, index } = c;
Self {
- name: c.name.clone(),
- index: c.index as u32,
+ name: name.clone(),
+ index: *index as u32,
}
}
}
@@ -196,12 +204,12 @@ impl Column {
) -> Result<Arc<dyn PhysicalExpr>> {
use datafusion_physical_expr_common::expect_expr_variant;
use datafusion_proto_models::protobuf;
- let column = expect_expr_variant!(
+ let protobuf::PhysicalColumn { name, index } = expect_expr_variant!(
node,
protobuf::physical_expr_node::ExprType::Column,
"Column",
);
- Ok(Arc::new(Column::from(column)))
+ Ok(Arc::new(Column::new(name, *index as usize)))
}
}
diff --git a/datafusion/physical-expr/src/expressions/is_not_null.rs b/datafusion/physical-expr/src/expressions/is_not_null.rs
index 3f3b7d16e5..0444ecb724 100644
--- a/datafusion/physical-expr/src/expressions/is_not_null.rs
+++ b/datafusion/physical-expr/src/expressions/is_not_null.rs
@@ -110,11 +110,12 @@ impl PhysicalExpr for IsNotNullExpr {
) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
use datafusion_proto_models::protobuf;
+ let Self { arg } = self;
Ok(Some(protobuf::PhysicalExprNode {
expr_id: None,
expr_type: Some(protobuf::physical_expr_node::ExprType::IsNotNullExpr(
Box::new(protobuf::PhysicalIsNotNull {
- expr: Some(Box::new(ctx.encode_child(&self.arg)?)),
+ expr: Some(Box::new(ctx.encode_child(arg)?)),
}),
)),
}))
@@ -136,11 +137,9 @@ impl IsNotNullExpr {
protobuf::physical_expr_node::ExprType::IsNotNullExpr,
"IsNotNullExpr",
);
- let expr = ctx.decode_required_expression(
- node.expr.as_deref(),
- "IsNotNullExpr",
- "expr",
- )?;
+ let protobuf::PhysicalIsNotNull { expr } = node.as_ref();
+ let expr =
+ ctx.decode_required_expression(expr.as_deref(), "IsNotNullExpr", "expr")?;
Ok(Arc::new(IsNotNullExpr::new(expr)))
}
diff --git a/datafusion/physical-expr/src/expressions/is_null.rs b/datafusion/physical-expr/src/expressions/is_null.rs
index da008a1cfb..08748af2f3 100644
--- a/datafusion/physical-expr/src/expressions/is_null.rs
+++ b/datafusion/physical-expr/src/expressions/is_null.rs
@@ -109,11 +109,12 @@ impl PhysicalExpr for IsNullExpr {
) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
use datafusion_proto_models::protobuf;
+ let Self { arg } = self;
Ok(Some(protobuf::PhysicalExprNode {
expr_id: None,
expr_type: Some(protobuf::physical_expr_node::ExprType::IsNullExpr(
Box::new(protobuf::PhysicalIsNull {
- expr: Some(Box::new(ctx.encode_child(&self.arg)?)),
+ expr: Some(Box::new(ctx.encode_child(arg)?)),
}),
)),
}))
@@ -135,8 +136,9 @@ impl IsNullExpr {
protobuf::physical_expr_node::ExprType::IsNullExpr,
"IsNullExpr",
);
+ let protobuf::PhysicalIsNull { expr } = node.as_ref();
let expr =
- ctx.decode_required_expression(node.expr.as_deref(), "IsNullExpr", "expr")?;
+ ctx.decode_required_expression(expr.as_deref(), "IsNullExpr", "expr")?;
Ok(Arc::new(IsNullExpr::new(expr)))
}
diff --git a/datafusion/physical-expr/src/expressions/literal.rs b/datafusion/physical-expr/src/expressions/literal.rs
index a7af824230..ebdca432c7 100644
--- a/datafusion/physical-expr/src/expressions/literal.rs
+++ b/datafusion/physical-expr/src/expressions/literal.rs
@@ -143,11 +143,21 @@ impl PhysicalExpr for Literal {
) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
use datafusion_proto_models::protobuf;
+ let Self { value, field } = self;
+ // The field name, type, and nullability are reconstructed by new_with_metadata.
+ let expr_type = if field.metadata().is_empty() {
+ protobuf::physical_expr_node::ExprType::Literal(value.try_into()?)
+ } else {
+ protobuf::physical_expr_node::ExprType::LiteralWithMetadata(
+ protobuf::PhysicalLiteralNode {
+ value: Some(value.try_into()?),
+ metadata: field.metadata().into(),
+ },
+ )
+ };
Ok(Some(protobuf::PhysicalExprNode {
expr_id: None,
- expr_type: Some(protobuf::physical_expr_node::ExprType::Literal(
- (&self.value).try_into()?,
- )),
+ expr_type: Some(expr_type),
}))
}
}
@@ -159,16 +169,38 @@ impl Literal {
node: &datafusion_proto_models::protobuf::PhysicalExprNode,
_ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
) -> Result<Arc<dyn PhysicalExpr>> {
- use datafusion_physical_expr_common::expect_expr_variant;
+ use datafusion_common::{internal_datafusion_err, internal_err};
use datafusion_proto_models::protobuf;
-
- let scalar_proto = expect_expr_variant!(
- node,
- protobuf::physical_expr_node::ExprType::Literal,
- "Literal",
- );
- let value = ScalarValue::try_from(scalar_proto)?;
- Ok(Arc::new(Literal::new(value)))
+ use protobuf::physical_expr_node::ExprType;
+
+ let protobuf::PhysicalExprNode {
+ expr_type,
+ // Expression IDs are handled by the enclosing proto converter.
+ expr_id: _,
+ } = node;
+ let (value, metadata) = match expr_type {
+ Some(ExprType::Literal(scalar)) => {
+ let datafusion_proto_models::datafusion_common::ScalarValue {
+ // The scalar payload is decoded by ScalarValue::try_from.
+ value: _,
+ } = scalar;
+ (ScalarValue::try_from(scalar)?, None)
+ }
+ Some(ExprType::LiteralWithMetadata(protobuf::PhysicalLiteralNode {
+ value,
+ metadata,
+ })) => {
+ let value = value.as_ref().ok_or_else(|| {
+ internal_datafusion_err!("Literal is missing required field 'value'")
+ })?;
+ (
+ ScalarValue::try_from(value)?,
+ Some(FieldMetadata::from(metadata)),
+ )
+ }
+ _ => return internal_err!("PhysicalExprNode is not a Literal"),
+ };
+ Ok(Arc::new(Literal::new_with_metadata(value, metadata)))
}
}
@@ -315,6 +347,26 @@ mod proto_tests {
assert_eq!(lit.value(), &ScalarValue::Int32(Some(42)));
}
+ #[test]
+ fn try_from_proto_rejects_missing_value() {
+ let node = datafusion_proto_models::protobuf::PhysicalExprNode {
+ expr_id: None,
+ expr_type: Some(physical_expr_node::ExprType::LiteralWithMetadata(
+ datafusion_proto_models::protobuf::PhysicalLiteralNode {
+ value: None,
+ metadata: Default::default(),
+ },
+ )),
+ };
+ let schema = Schema::empty();
+ let decoder = UnreachableDecoder;
+ let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+ let err = Literal::try_from_proto(&node, &ctx).unwrap_err();
+ assert!(
+ matches!(err, DataFusionError::Internal(msg) if msg.contains("Literal is missing required field 'value'"))
+ );
+ }
+
#[test]
fn try_from_proto_rejects_non_literal_node() {
let node = column_node("a");
diff --git a/datafusion/physical-expr/src/expressions/negative.rs b/datafusion/physical-expr/src/expressions/negative.rs
index c894c12784..a932c8d522 100644
--- a/datafusion/physical-expr/src/expressions/negative.rs
+++ b/datafusion/physical-expr/src/expressions/negative.rs
@@ -184,11 +184,12 @@ impl PhysicalExpr for NegativeExpr {
) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
use datafusion_proto_models::protobuf;
+ let Self { arg } = self;
Ok(Some(protobuf::PhysicalExprNode {
expr_id: None,
expr_type: Some(protobuf::physical_expr_node::ExprType::Negative(Box::new(
protobuf::PhysicalNegativeNode {
- expr: Some(Box::new(ctx.encode_child(&self.arg)?)),
+ expr: Some(Box::new(ctx.encode_child(arg)?)),
},
))),
}))
@@ -210,8 +211,9 @@ impl NegativeExpr {
protobuf::physical_expr_node::ExprType::Negative,
"Negative",
);
+ let protobuf::PhysicalNegativeNode { expr } = n.as_ref();
let expr =
- ctx.decode_required_expression(n.expr.as_deref(), "NegativeExpr", "expr")?;
+ ctx.decode_required_expression(expr.as_deref(), "NegativeExpr", "expr")?;
Ok(Arc::new(NegativeExpr::new(expr)))
}
diff --git a/datafusion/physical-expr/src/expressions/not.rs b/datafusion/physical-expr/src/expressions/not.rs
index f856dd568a..7699c8944e 100644
--- a/datafusion/physical-expr/src/expressions/not.rs
+++ b/datafusion/physical-expr/src/expressions/not.rs
@@ -189,11 +189,12 @@ impl PhysicalExpr for NotExpr {
) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
use datafusion_proto_models::protobuf;
+ let Self { arg } = self;
Ok(Some(protobuf::PhysicalExprNode {
expr_id: None,
expr_type: Some(protobuf::physical_expr_node::ExprType::NotExpr(Box::new(
protobuf::PhysicalNot {
- expr: Some(Box::new(ctx.encode_child(&self.arg)?)),
+ expr: Some(Box::new(ctx.encode_child(arg)?)),
},
))),
}))
@@ -215,8 +216,8 @@ impl NotExpr {
protobuf::physical_expr_node::ExprType::NotExpr,
"NotExpr",
);
- let expr =
- ctx.decode_required_expression(not_expr.expr.as_deref(), "NotExpr", "expr")?;
+ let protobuf::PhysicalNot { expr } = not_expr.as_ref();
+ let expr = ctx.decode_required_expression(expr.as_deref(), "NotExpr", "expr")?;
Ok(Arc::new(NotExpr::new(expr)))
}
diff --git a/datafusion/physical-expr/src/expressions/unknown_column.rs b/datafusion/physical-expr/src/expressions/unknown_column.rs
index ed85f20dd2..c3b6dd9daf 100644
--- a/datafusion/physical-expr/src/expressions/unknown_column.rs
+++ b/datafusion/physical-expr/src/expressions/unknown_column.rs
@@ -93,12 +93,11 @@ impl PhysicalExpr for UnKnownColumn {
) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
use datafusion_proto_models::protobuf;
+ let Self { name } = self;
Ok(Some(protobuf::PhysicalExprNode {
expr_id: None,
expr_type: Some(protobuf::physical_expr_node::ExprType::UnknownColumn(
- protobuf::UnknownColumn {
- name: self.name.clone(),
- },
+ protobuf::UnknownColumn { name: name.clone() },
)),
}))
}
@@ -114,12 +113,12 @@ impl UnKnownColumn {
use datafusion_physical_expr_common::expect_expr_variant;
use datafusion_proto_models::protobuf;
- let unknown_col = expect_expr_variant!(
+ let protobuf::UnknownColumn { name } = expect_expr_variant!(
node,
protobuf::physical_expr_node::ExprType::UnknownColumn,
"UnKnownColumn",
);
- Ok(Arc::new(UnKnownColumn::new(&unknown_col.name)))
+ Ok(Arc::new(UnKnownColumn::new(name)))
}
}
diff --git a/datafusion/proto-models/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto
index 04b9464c9b..38abf7c203 100644
--- a/datafusion/proto-models/proto/datafusion.proto
+++ b/datafusion/proto-models/proto/datafusion.proto
@@ -1105,9 +1105,15 @@ message PhysicalExprNode {
PhysicalLambdaVariableExprNode lambda_variable = 26;
PhysicalRangeExprNode range_expr = 27;
PhysicalSqlSimilarToPatternNode sql_similar_to_pattern = 28;
+ PhysicalLiteralNode literal_with_metadata = 29;
}
}
+message PhysicalLiteralNode {
+ datafusion_common.ScalarValue value = 1;
+ map<string, string> metadata = 2;
+}
+
message PhysicalDynamicFilterNode {
repeated PhysicalExprNode children = 1;
repeated PhysicalExprNode remapped_children = 2;
diff --git a/datafusion/proto-models/src/generated/pbjson.rs b/datafusion/proto-models/src/generated/pbjson.rs
index c3b5d1d374..4dfcc8a0d8 100644
--- a/datafusion/proto-models/src/generated/pbjson.rs
+++ b/datafusion/proto-models/src/generated/pbjson.rs
@@ -19633,6 +19633,9 @@ impl serde::Serialize for PhysicalExprNode {
physical_expr_node::ExprType::SqlSimilarToPattern(v) => {
struct_ser.serialize_field("sqlSimilarToPattern", v)?;
}
+ physical_expr_node::ExprType::LiteralWithMetadata(v) => {
+ struct_ser.serialize_field("literalWithMetadata", v)?;
+ }
}
}
struct_ser.end()
@@ -19692,6 +19695,8 @@ impl<'de> serde::Deserialize<'de> for PhysicalExprNode {
"rangeExpr",
"sql_similar_to_pattern",
"sqlSimilarToPattern",
+ "literal_with_metadata",
+ "literalWithMetadata",
];
#[allow(clippy::enum_variant_names)]
@@ -19723,6 +19728,7 @@ impl<'de> serde::Deserialize<'de> for PhysicalExprNode {
LambdaVariable,
RangeExpr,
SqlSimilarToPattern,
+ LiteralWithMetadata,
}
impl<'de> serde::Deserialize<'de> for GeneratedField {
fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
@@ -19771,6 +19777,7 @@ impl<'de> serde::Deserialize<'de> for PhysicalExprNode {
"lambdaVariable" | "lambda_variable" => Ok(GeneratedField::LambdaVariable),
"rangeExpr" | "range_expr" => Ok(GeneratedField::RangeExpr),
"sqlSimilarToPattern" | "sql_similar_to_pattern" => Ok(GeneratedField::SqlSimilarToPattern),
+ "literalWithMetadata" | "literal_with_metadata" => Ok(GeneratedField::LiteralWithMetadata),
_ => Err(serde::de::Error::unknown_field(value, FIELDS)),
}
}
@@ -19982,6 +19989,13 @@ impl<'de> serde::Deserialize<'de> for PhysicalExprNode {
return Err(serde::de::Error::duplicate_field("sqlSimilarToPattern"));
}
expr_type__ = map_.next_value::<::std::option::Option<_>>()?.map(physical_expr_node::ExprType::SqlSimilarToPattern)
+;
+ }
+ GeneratedField::LiteralWithMetadata => {
+ if expr_type__.is_some() {
+ return Err(serde::de::Error::duplicate_field("literalWithMetadata"));
+ }
+ expr_type__ = map_.next_value::<::std::option::Option<_>>()?.map(physical_expr_node::ExprType::LiteralWithMetadata)
;
}
}
@@ -21261,6 +21275,116 @@ impl<'de> serde::Deserialize<'de> for PhysicalLikeExprNode {
deserializer.deserialize_struct("datafusion.PhysicalLikeExprNode", FIELDS, GeneratedVisitor)
}
}
+impl serde::Serialize for PhysicalLiteralNode {
+ #[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.value.is_some() {
+ len += 1;
+ }
+ if !self.metadata.is_empty() {
+ len += 1;
+ }
+ let mut struct_ser = serializer.serialize_struct("datafusion.PhysicalLiteralNode", len)?;
+ if let Some(v) = self.value.as_ref() {
+ struct_ser.serialize_field("value", v)?;
+ }
+ if !self.metadata.is_empty() {
+ struct_ser.serialize_field("metadata", &self.metadata)?;
+ }
+ struct_ser.end()
+ }
+}
+impl<'de> serde::Deserialize<'de> for PhysicalLiteralNode {
+ #[allow(deprecated)]
+ fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ const FIELDS: &[&str] = &[
+ "value",
+ "metadata",
+ ];
+
+ #[allow(clippy::enum_variant_names)]
+ enum GeneratedField {
+ Value,
+ Metadata,
+ }
+ 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 serde::de::Visitor<'_> 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 {
+ "value" => Ok(GeneratedField::Value),
+ "metadata" => Ok(GeneratedField::Metadata),
+ _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
+ }
+ }
+ }
+ deserializer.deserialize_identifier(GeneratedVisitor)
+ }
+ }
+ struct GeneratedVisitor;
+ impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+ type Value = PhysicalLiteralNode;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ formatter.write_str("struct datafusion.PhysicalLiteralNode")
+ }
+
+ fn visit_map<V>(self, mut map_: V) -> std::result::Result<PhysicalLiteralNode, V::Error>
+ where
+ V: serde::de::MapAccess<'de>,
+ {
+ let mut value__ = None;
+ let mut metadata__ = None;
+ while let Some(k) = map_.next_key()? {
+ match k {
+ GeneratedField::Value => {
+ if value__.is_some() {
+ return Err(serde::de::Error::duplicate_field("value"));
+ }
+ value__ = map_.next_value()?;
+ }
+ GeneratedField::Metadata => {
+ if metadata__.is_some() {
+ return Err(serde::de::Error::duplicate_field("metadata"));
+ }
+ metadata__ = Some(
+ map_.next_value::<std::collections::HashMap<_, _>>()?
+ );
+ }
+ }
+ }
+ Ok(PhysicalLiteralNode {
+ value: value__,
+ metadata: metadata__.unwrap_or_default(),
+ })
+ }
+ }
+ deserializer.deserialize_struct("datafusion.PhysicalLiteralNode", FIELDS, GeneratedVisitor)
+ }
+}
impl serde::Serialize for PhysicalNegativeNode {
#[allow(deprecated)]
fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
diff --git a/datafusion/proto-models/src/generated/prost.rs b/datafusion/proto-models/src/generated/prost.rs
index ec01eb1e67..f30a5d640b 100644
--- a/datafusion/proto-models/src/generated/prost.rs
+++ b/datafusion/proto-models/src/generated/prost.rs
@@ -1631,7 +1631,7 @@ pub struct PhysicalExprNode {
pub expr_id: ::core::option::Option<u64>,
#[prost(
oneof = "physical_expr_node::ExprType",
- tags = "1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 14, 15, 16, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28"
+ tags = "1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 14, 15, 16, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29"
)]
pub expr_type: ::core::option::Option<physical_expr_node::ExprType>,
}
@@ -1700,9 +1700,21 @@ pub mod physical_expr_node {
SqlSimilarToPattern(
::prost::alloc::boxed::Box<super::PhysicalSqlSimilarToPatternNode>,
),
+ #[prost(message, tag = "29")]
+ LiteralWithMetadata(super::PhysicalLiteralNode),
}
}
#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct PhysicalLiteralNode {
+ #[prost(message, optional, tag = "1")]
+ pub value: ::core::option::Option<super::datafusion_common::ScalarValue>,
+ #[prost(map = "string, string", tag = "2")]
+ pub metadata: ::std::collections::HashMap<
+ ::prost::alloc::string::String,
+ ::prost::alloc::string::String,
+ >,
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
pub struct PhysicalDynamicFilterNode {
#[prost(message, repeated, tag = "1")]
pub children: ::prost::alloc::vec::Vec<PhysicalExprNode>,
diff --git a/datafusion/proto/src/physical_plan/from_proto.rs b/datafusion/proto/src/physical_plan/from_proto.rs
index bc443149df..8c3416ec98 100644
--- a/datafusion/proto/src/physical_plan/from_proto.rs
+++ b/datafusion/proto/src/physical_plan/from_proto.rs
@@ -269,7 +269,9 @@ pub fn parse_physical_expr_with_converter(
// to the right constructor.
ExprType::Column(_) => Column::try_from_proto(proto, &decode_ctx)?,
ExprType::UnknownColumn(_) => UnKnownColumn::try_from_proto(proto, &decode_ctx)?,
- ExprType::Literal(_) => Literal::try_from_proto(proto, &decode_ctx)?,
+ ExprType::Literal(_) | ExprType::LiteralWithMetadata(_) => {
+ Literal::try_from_proto(proto, &decode_ctx)?
+ }
ExprType::BinaryExpr(_) => BinaryExpr::try_from_proto(proto, &decode_ctx)?,
ExprType::AggregateExpr(_) => {
return not_impl_err!(
diff --git a/datafusion/proto/tests/cases/plans/exprs.rs b/datafusion/proto/tests/cases/plans/exprs.rs
index a7f6aa7200..f0828b5a66 100644
--- a/datafusion/proto/tests/cases/plans/exprs.rs
+++ b/datafusion/proto/tests/cases/plans/exprs.rs
@@ -46,6 +46,54 @@ use datafusion_proto::protobuf::PhysicalPlanNode;
use std::sync::Arc;
use std::vec;
+#[test]
+fn roundtrip_literal_metadata() -> Result<()> {
+ use datafusion_common::metadata::FieldMetadata;
+ use datafusion_proto::bytes::{physical_plan_from_bytes, physical_plan_to_bytes};
+
+ let schema = Arc::new(Schema::empty());
+ let metadata = FieldMetadata::from(std::collections::HashMap::from([
+ ("ARROW:extension:name".to_string(), "example.id".to_string()),
+ ("ARROW:extension:metadata".to_string(), "{}".to_string()),
+ ("description".to_string(), "identifier".to_string()),
+ ]));
+ let ctx = SessionContext::new();
+ for value in [ScalarValue::Int32(Some(42)), ScalarValue::Int32(None)] {
+ let literal: Arc<dyn PhysicalExpr> = Arc::new(Literal::new_with_metadata(
+ value.clone(),
+ Some(metadata.clone()),
+ ));
+ let plan: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
+ vec![ProjectionExpr::new(Arc::clone(&literal), "result")],
+ Arc::new(EmptyExec::new(Arc::clone(&schema))),
+ )?);
+ let bytes = physical_plan_to_bytes(Arc::clone(&plan))?;
+ let decoded = physical_plan_from_bytes(&bytes, ctx.task_ctx().as_ref())?;
+ assert_eq!(decoded.schema(), plan.schema());
+ let projection = decoded.downcast_ref::<ProjectionExec>().unwrap();
+ let decoded_literal = &projection.expr()[0].expr;
+ assert_eq!(
+ decoded_literal.return_field(&schema)?,
+ literal.return_field(&schema)?
+ );
+ assert_eq!(
+ decoded_literal.downcast_ref::<Literal>().unwrap().value(),
+ &value
+ );
+
+ #[cfg(feature = "json")]
+ {
+ use datafusion_proto::bytes::{
+ physical_plan_from_json, physical_plan_to_json,
+ };
+ let json = physical_plan_to_json(Arc::clone(&plan))?;
+ let decoded = physical_plan_from_json(&json, ctx.task_ctx().as_ref())?;
+ assert_eq!(decoded.schema(), plan.schema());
+ }
+ }
+ Ok(())
+}
+
#[test]
fn roundtrip_date_time_interval() -> Result<()> {
let schema = Schema::new(vec![
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org