Add Extension Type / Metadata support for Scalar UDFs (apache#15646)

* Add in plumbing to pass around metadata for physical expressions

* Adding argument metadata to scalar argument struct

* Since everywhere we use this we immediately clone, go ahead and returned an owned version of the metadata for simplicity

* Cargo fmt

* Benchmarks required args_metadata in tests

* Clippy warnings

* Switching over to passing Field around instead of metadata so we can handle extension types directly

* Switching return_type_from_args to return_field_from_args

* Updates to unit tests for switching to field instead of data_type

* Resolve unit test issues

* Update after rebase on main

* GetFieldFunc should return the field it finds instead of creating a new one

* Get metadata from scalar functions

* Change expr_schema to use to_field primarily instead of individual calls for getting data type, nullability, and schema

* Scalar function arguments should take return field instead of return data type now

* subquery should just get the field from below and not lose potential metadata

* Update comment

* Remove output_field now that we've determined it using return_field_from_args

* Change name to_field to field_from_column to be more consistent with the usage and prevent misconception about if we are doing some conversion

* Minor moving around of the explicit lifetimes in the struct definition

* Change physical expression to require to output a field which requires a lot of unit test updates, especially because the scalar arguments pass around borrowed values

* Change name from output_field to return_field to be more consistent

* Update migration guide for DF48 with user defined functions

* Whitespace

* Docstring correction

Author: timsaucer

Cargo.lock

@@ -472,7 +472,7 @@ impl DFSchema {
         let matches = self.qualified_fields_with_unqualified_name(name);
         match matches.len() {
             0 => Err(unqualified_field_not_found(name, self)),
-            1 => Ok((matches[0].0, (matches[0].1))),
+            1 => Ok((matches[0].0, matches[0].1)),
             _ => {
                 // When `matches` size > 1, it doesn't necessarily mean an `ambiguous name` problem.
                 // Because name may generate from Alias/... . It means that it don't own qualifier.
@@ -515,14 +515,6 @@ impl DFSchema {
         Ok(self.field(idx))
     }
 
-    /// Find the field with the given qualified column
-    pub fn field_from_column(&self, column: &Column) -> Result<&Field> {
-        match &column.relation {
-            Some(r) => self.field_with_qualified_name(r, &column.name),
-            None => self.field_with_unqualified_name(&column.name),
-        }
-    }
-
     /// Find the field with the given qualified column
     pub fn qualified_field_from_column(
         &self,
@@ -969,16 +961,28 @@ impl Display for DFSchema {
 /// widely used in the DataFusion codebase.
 pub trait ExprSchema: std::fmt::Debug {
     /// Is this column reference nullable?
-    fn nullable(&self, col: &Column) -> Result<bool>;
+    fn nullable(&self, col: &Column) -> Result<bool> {
+        Ok(self.field_from_column(col)?.is_nullable())
+    }
 
     /// What is the datatype of this column?
-    fn data_type(&self, col: &Column) -> Result<&DataType>;
+    fn data_type(&self, col: &Column) -> Result<&DataType> {
+        Ok(self.field_from_column(col)?.data_type())
+    }
 
     /// Returns the column's optional metadata.
-    fn metadata(&self, col: &Column) -> Result<&HashMap<String, String>>;
+    fn metadata(&self, col: &Column) -> Result<&HashMap<String, String>> {
+        Ok(self.field_from_column(col)?.metadata())
+    }
 
     /// Return the column's datatype and nullability
-    fn data_type_and_nullable(&self, col: &Column) -> Result<(&DataType, bool)>;
+    fn data_type_and_nullable(&self, col: &Column) -> Result<(&DataType, bool)> {
+        let field = self.field_from_column(col)?;
+        Ok((field.data_type(), field.is_nullable()))
+    }
+
+    // Return the column's field
+    fn field_from_column(&self, col: &Column) -> Result<&Field>;
 }
 
 // Implement `ExprSchema` for `Arc<DFSchema>`
@@ -998,24 +1002,18 @@ impl<P: AsRef<DFSchema> + std::fmt::Debug> ExprSchema for P {
     fn data_type_and_nullable(&self, col: &Column) -> Result<(&DataType, bool)> {
         self.as_ref().data_type_and_nullable(col)
     }
-}
-
-impl ExprSchema for DFSchema {
-    fn nullable(&self, col: &Column) -> Result<bool> {
-        Ok(self.field_from_column(col)?.is_nullable())
-    }
-
-    fn data_type(&self, col: &Column) -> Result<&DataType> {
-        Ok(self.field_from_column(col)?.data_type())
-    }
 
-    fn metadata(&self, col: &Column) -> Result<&HashMap<String, String>> {
-        Ok(self.field_from_column(col)?.metadata())
+    fn field_from_column(&self, col: &Column) -> Result<&Field> {
+        self.as_ref().field_from_column(col)
     }
+}
 
-    fn data_type_and_nullable(&self, col: &Column) -> Result<(&DataType, bool)> {
-        let field = self.field_from_column(col)?;
-        Ok((field.data_type(), field.is_nullable()))
+impl ExprSchema for DFSchema {
+    fn field_from_column(&self, col: &Column) -> Result<&Field> {
+        match &col.relation {
+            Some(r) => self.field_with_qualified_name(r, &col.name),
+            None => self.field_with_unqualified_name(&col.name),
+        }
     }
 }
 

datafusion/common/src/dfschema.rs

@@ -95,7 +95,7 @@ extended_tests = []
 [dependencies]
 arrow = { workspace = true }
 arrow-ipc = { workspace = true }
-arrow-schema = { workspace = true }
+arrow-schema = { workspace = true, features = ["canonical_extension_types"] }
 async-trait = { workspace = true }
 bytes = { workspace = true }
 bzip2 = { version = "0.5.2", optional = true }

datafusion/core/Cargo.toml

@@ -128,7 +128,7 @@ fn test_update_matching_exprs() -> Result<()> {
                     Arc::new(Column::new("b", 1)),
                 )),
             ],
-            DataType::Int32,
+            Field::new("f", DataType::Int32, true),
         )),
         Arc::new(CaseExpr::try_new(
             Some(Arc::new(Column::new("d", 2))),
@@ -193,7 +193,7 @@ fn test_update_matching_exprs() -> Result<()> {
                     Arc::new(Column::new("b", 1)),
                 )),
             ],
-            DataType::Int32,
+            Field::new("f", DataType::Int32, true),
         )),
         Arc::new(CaseExpr::try_new(
             Some(Arc::new(Column::new("d", 3))),
@@ -261,7 +261,7 @@ fn test_update_projected_exprs() -> Result<()> {
                     Arc::new(Column::new("b", 1)),
                 )),
             ],
-            DataType::Int32,
+            Field::new("f", DataType::Int32, true),
         )),
         Arc::new(CaseExpr::try_new(
             Some(Arc::new(Column::new("d", 2))),
@@ -326,7 +326,7 @@ fn test_update_projected_exprs() -> Result<()> {
                     Arc::new(Column::new("b_new", 1)),
                 )),
             ],
-            DataType::Int32,
+            Field::new("f", DataType::Int32, true),
         )),
         Arc::new(CaseExpr::try_new(
             Some(Arc::new(Column::new("d_new", 3))),

datafusion/core/tests/physical_optimizer/projection_pushdown.rs

@@ -110,7 +110,7 @@ select  channel, item, return_ratio, return_rank, currency_rank from
  		where 
  			sr.sr_return_amt > 10000 
  			and sts.ss_net_profit > 1
-                         and sts.ss_net_paid > 0 
+                         and sts.ss_net_paid > 0
                          and sts.ss_quantity > 0
                          and ss_sold_date_sk = d_date_sk
                          and d_year = 2000