fix: preserve literal metadata and destructure leaf and unary proto hooks

datafusion/physical-expr/src/expressions/column.rs

@@ -153,26 +153,34 @@ 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,
+                },
+            )),
         }))
     }
 }
 
 #[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)))
     }
 }
 

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)))
     }

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)))
     }

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().clone(),
+                },
+            )
+        };
         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");

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)))
     }

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)))
     }

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)))
     }
 }
 

datafusion/proto-models/proto/datafusion.proto

@@ -1074,9 +1074,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;