fix(function): Const(Nullable(Object)) column downcast to Nullable failed

common/datavalues/src/columns/array/mod.rs

@@ -91,6 +91,10 @@ impl Column for ArrayColumn {
         self.data_type.clone()
     }
 
+    fn column_type_name(&self) -> String {
+        "Array".to_string()
+    }
+
     fn len(&self) -> usize {
         self.offsets.len() - 1
     }

common/datavalues/src/columns/boolean/mod.rs

@@ -75,6 +75,10 @@ impl Column for BooleanColumn {
         BooleanType::arc()
     }
 
+    fn column_type_name(&self) -> String {
+        "Boolean".to_string()
+    }
+
     fn len(&self) -> usize {
         self.values.len()
     }

common/datavalues/src/columns/column.rs

@@ -39,6 +39,8 @@ pub trait Column: Send + Sync {
     }
     fn data_type(&self) -> DataTypePtr;
 
+    fn column_type_name(&self) -> String;
+
     fn is_nullable(&self) -> bool {
         false
     }

common/datavalues/src/columns/const_/mod.rs

@@ -50,6 +50,10 @@ impl Column for ConstColumn {
         self.column.data_type()
     }
 
+    fn column_type_name(&self) -> String {
+        format!("Const({})", self.column.column_type_name())
+    }
+
     fn is_nullable(&self) -> bool {
         self.column.is_nullable()
     }

common/datavalues/src/columns/null/mod.rs

@@ -55,6 +55,10 @@ impl Column for NullColumn {
         Arc::new(NullType {})
     }
 
+    fn column_type_name(&self) -> String {
+        "Null".to_string()
+    }
+
     fn is_null(&self) -> bool {
         true
     }

common/datavalues/src/columns/nullable/mod.rs

@@ -71,6 +71,10 @@ impl Column for NullableColumn {
         Arc::new(NullableType::create(nest))
     }
 
+    fn column_type_name(&self) -> String {
+        format!("Nullable({})", self.column.column_type_name())
+    }
+
     fn is_nullable(&self) -> bool {
         true
     }

common/datavalues/src/columns/object/mod.rs

@@ -101,6 +101,10 @@ impl<T: ObjectType> Column for ObjectColumn<T> {
         T::data_type()
     }
 
+    fn column_type_name(&self) -> String {
+        "Object".to_string()
+    }
+
     fn len(&self) -> usize {
         self.values.len()
     }

common/datavalues/src/columns/primitive/mod.rs

@@ -156,6 +156,10 @@ impl<T: PrimitiveType> Column for PrimitiveColumn<T> {
         create_primitive_datatype::<T>()
     }
 
+    fn column_type_name(&self) -> String {
+        "Primitive".to_string()
+    }
+
     fn len(&self) -> usize {
         self.values.len()
     }

common/datavalues/src/columns/series.rs

@@ -48,7 +48,7 @@ impl Series {
             .ok_or_else(|| {
                 ErrorCode::UnknownColumn(format!(
                     "downcast column error, column type: {:?}, expected column: {:?}",
-                    column.data_type(),
+                    column.column_type_name(),
                     std::any::type_name::<T>(),
                 ))
             });
@@ -59,7 +59,7 @@ impl Series {
         let arr = column.as_any().downcast_ref::<T>().ok_or_else(|| {
             ErrorCode::UnknownColumn(format!(
                 "downcast column error, column type: {:?}, expected column: {:?}",
-                column.data_type(),
+                column.column_type_name(),
                 std::any::type_name::<T>(),
             ))
         });

common/datavalues/src/columns/string/mod.rs

@@ -143,6 +143,10 @@ impl Column for StringColumn {
         StringType::arc()
     }
 
+    fn column_type_name(&self) -> String {
+        "String".to_string()
+    }
+
     fn len(&self) -> usize {
         self.offsets.len() - 1
     }

common/datavalues/src/columns/struct_/mod.rs

@@ -71,6 +71,10 @@ impl Column for StructColumn {
         self.data_type.clone()
     }
 
+    fn column_type_name(&self) -> String {
+        "Struct".to_string()
+    }
+
     fn len(&self) -> usize {
         self.values[0].len()
     }

common/functions/src/scalars/function_adapter.rs

@@ -128,6 +128,31 @@ impl Function for FunctionAdapter {
             return inner.eval(columns, input_rows);
         }
 
+        // is there nullable constant? Did not consider this case
+        // unwrap constant
+        if self.passthrough_constant() && columns.iter().all(|v| v.column().is_const()) {
+            let columns = columns
+                .iter()
+                .map(|v| {
+                    let c = v.column();
+                    let c: &ConstColumn = unsafe { Series::static_cast(c) };
+
+                    ColumnWithField::new(c.inner().clone(), v.field().clone())
+                })
+                .collect::<Vec<_>>();
+
+            let col = self.eval(&columns, 1)?;
+            let col = if col.is_const() && col.len() == 1 {
+                col.replicate(&[input_rows])
+            } else if col.is_null() {
+                NullColumn::new(input_rows).arc()
+            } else {
+                ConstColumn::new(col, input_rows).arc()
+            };
+
+            return Ok(col);
+        }
+
         // nullable or null
         if self.passthrough_null {
             if columns
@@ -180,7 +205,7 @@ impl Function for FunctionAdapter {
                 });
 
                 let col = if col.is_nullable() {
-                    let nullable_column: &NullableColumn = unsafe { Series::static_cast(&col) };
+                    let nullable_column: &NullableColumn = Series::check_get(&col)?;
                     NullableColumn::new(nullable_column.inner().clone(), validity)
                 } else {
                     NullableColumn::new(col, validity)
@@ -189,31 +214,6 @@ impl Function for FunctionAdapter {
             }
         }
 
-        // is there nullable constant? Did not consider this case
-        // unwrap constant
-        if self.passthrough_constant() && columns.iter().all(|v| v.column().is_const()) {
-            let columns = columns
-                .iter()
-                .map(|v| {
-                    let c = v.column();
-                    let c: &ConstColumn = unsafe { Series::static_cast(c) };
-
-                    ColumnWithField::new(c.inner().clone(), v.field().clone())
-                })
-                .collect::<Vec<_>>();
-
-            let col = self.eval(&columns, 1)?;
-            let col = if col.is_const() && col.len() == 1 {
-                col.replicate(&[input_rows])
-            } else if col.is_null() {
-                NullColumn::new(input_rows).arc()
-            } else {
-                ConstColumn::new(col, input_rows).arc()
-            };
-
-            return Ok(col);
-        }
-
         inner.eval(columns, input_rows)
     }