Metadata is kept in projections for non-derived columns

datafusion/src/physical_plan/projection.rs

@@ -37,7 +37,6 @@ use super::expressions::Column;
 use super::metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet};
 use super::{RecordBatchStream, SendableRecordBatchStream, Statistics};
 use async_trait::async_trait;
-
 use futures::stream::Stream;
 use futures::stream::StreamExt;
 
@@ -62,18 +61,21 @@ impl ProjectionExec {
     ) -> Result<Self> {
         let input_schema = input.schema();
 
-        let fields: Result<Vec<_>> = expr
+        let fields: Vec<Field> = expr
             .iter()
             .map(|(e, name)| {
-                Ok(Field::new(
-                    name,
-                    e.data_type(&input_schema)?,
-                    e.nullable(&input_schema)?,
-                ))
+                match input_schema.field_with_name(&name) {
+                    Ok(f) => f.clone(),
+                    Err(_) => Field::new(
+                        name,
+                        e.data_type(&input_schema).unwrap(),
+                        e.nullable(&input_schema).unwrap_or(true),
+                    )
+                }
             })
             .collect();
 
-        let schema = Arc::new(Schema::new(fields?));
+        let schema = Arc::new(Schema::new(fields));
 
         Ok(Self {
             expr,
@@ -178,7 +180,7 @@ impl ExecutionPlan for ProjectionExec {
 
 fn stats_projection(
     stats: Statistics,
-    exprs: impl Iterator<Item = Arc<dyn PhysicalExpr>>,
+    exprs: impl Iterator<Item=Arc<dyn PhysicalExpr>>,
 ) -> Statistics {
     let column_statistics = stats.column_statistics.map(|input_col_stats| {
         exprs
@@ -296,6 +298,11 @@ mod tests {
             Arc::new(csv),
         )?;
 
+        let col_field = projection.schema.field(0);
+        let col_metadata = col_field.metadata().clone().unwrap().clone();
+        let data: &str = &col_metadata["testing"];
+        assert_eq!(data, "test");
+
         let mut partition_count = 0;
         let mut row_count = 0;
         for partition in 0..projection.output_partitioning().partition_count() {

datafusion/src/test_util.rs

@@ -18,6 +18,7 @@
 //! Utility functions to make testing DataFusion based crates easier
 
 use std::{env, error::Error, path::PathBuf, sync::Arc};
+use std::collections::BTreeMap;
 
 use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
 
@@ -229,8 +230,10 @@ fn get_data_dir(udf_env: &str, submodule_data: &str) -> Result<PathBuf, Box<dyn
 
 /// Get the schema for the aggregate_test_* csv files
 pub fn aggr_test_schema() -> SchemaRef {
-    Arc::new(Schema::new(vec![
-        Field::new("c1", DataType::Utf8, false),
+    let mut f1 = Field::new("c1", DataType::Utf8, false);
+    f1.set_metadata(Some(BTreeMap::from_iter(vec![("testing".into(), "test".into())].into_iter())));
+    let schema = Schema::new(vec![
+        f1,
         Field::new("c2", DataType::UInt32, false),
         Field::new("c3", DataType::Int8, false),
         Field::new("c4", DataType::Int16, false),
@@ -243,7 +246,9 @@ pub fn aggr_test_schema() -> SchemaRef {
         Field::new("c11", DataType::Float32, false),
         Field::new("c12", DataType::Float64, false),
         Field::new("c13", DataType::Utf8, false),
-    ]))
+    ]);
+
+    Arc::new(schema)
 }
 
 #[cfg(test)]