diff --git a/src/query/expression/src/block.rs b/src/query/expression/src/block.rs
index 8af901f49cbbc..cbeede2fec333 100644
--- a/src/query/expression/src/block.rs
+++ b/src/query/expression/src/block.rs
@@ -182,13 +182,13 @@ impl BlockEntry {
         }
     }
 
-    pub fn downcast<T: AccessType>(&self) -> Option<ColumnView<T>> {
+    pub fn downcast<T: AccessType>(&self) -> Result<ColumnView<T>> {
         match self {
-            BlockEntry::Const(scalar, _, num_rows) => T::try_downcast_scalar(&scalar.as_ref())
-                .map(|s| ColumnView::Const(T::to_owned_scalar(s), *num_rows)),
-            BlockEntry::Column(column) => {
-                T::try_downcast_column(column).map(|c| ColumnView::Column(c))
-            }
+            BlockEntry::Const(scalar, _, num_rows) => Ok(ColumnView::Const(
+                T::to_owned_scalar(T::try_downcast_scalar(&scalar.as_ref())?),
+                *num_rows,
+            )),
+            BlockEntry::Column(column) => Ok(ColumnView::Column(T::try_downcast_column(column)?)),
         }
     }
 }
diff --git a/src/query/expression/src/register.rs b/src/query/expression/src/register.rs
index d123d89542ab9..02ba968c98f04 100755
--- a/src/query/expression/src/register.rs
+++ b/src/query/expression/src/register.rs
@@ -912,21 +912,18 @@ fn erase_function_generic_0_arg<O: ArgType>(
 fn erase_function_generic_1_arg<I1: ArgType, O: ArgType>(
     func: impl for<'a> Fn(Value<I1>, &mut EvalContext) -> Value<O>,
 ) -> impl Fn(&[Value<AnyType>], &mut EvalContext) -> Value<AnyType> {
-    move |args, ctx| {
-        let arg1 = match args[0].try_downcast() {
-            Some(arg1) => arg1,
-            None => match &args[0] {
-                Value::Scalar(scalar) => {
-                    unreachable!("can't downcast from {scalar:?} to {}", I1::data_type())
-                }
-                Value::Column(column) => unreachable!(
-                    "can't downcast from {} to {}",
-                    column.data_type(),
-                    I1::data_type()
-                ),
-            },
-        };
-        Value::upcast(func(arg1, ctx))
+    move |args, ctx| match args[0].try_downcast() {
+        Ok(arg1) => Value::upcast(func(arg1, ctx)),
+        Err(_) => match &args[0] {
+            Value::Scalar(scalar) => {
+                unreachable!("can't downcast from {scalar:?} to {}", I1::data_type())
+            }
+            Value::Column(column) => unreachable!(
+                "can't downcast from {} to {}",
+                column.data_type(),
+                I1::data_type()
+            ),
+        },
     }
 }
 
diff --git a/src/query/expression/src/types.rs b/src/query/expression/src/types.rs
index 0c124067182dc..79d873e2456a6 100755
--- a/src/query/expression/src/types.rs
+++ b/src/query/expression/src/types.rs
@@ -277,7 +277,7 @@ impl DataType {
             DataType::Number(NumberDataType::Int64) => {
                 Ok(Scalar::Number(NumberScalar::Int64(i64::MAX)))
             }
-            _ => Result::Err(format!(
+            _ => Err(format!(
                 "only support numeric types and time types, but got {:?}",
                 self
             )),
@@ -305,7 +305,7 @@ impl DataType {
             DataType::Number(NumberDataType::Int64) => {
                 Ok(Scalar::Number(NumberScalar::Int64(i64::MIN)))
             }
-            _ => Result::Err(format!(
+            _ => Err(format!(
                 "only support numeric types and time types, but got {:?}",
                 self
             )),
@@ -579,10 +579,10 @@ pub trait AccessType: Debug + Clone + PartialEq + Sized + 'static {
     fn to_owned_scalar(scalar: Self::ScalarRef<'_>) -> Self::Scalar;
     fn to_scalar_ref(scalar: &Self::Scalar) -> Self::ScalarRef<'_>;
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>>;
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain>;
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>, ErrorCode>;
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain, ErrorCode>;
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column>;
+    fn try_downcast_column(col: &Column) -> Result<Self::Column, ErrorCode>;
 
     fn column_len(col: &Self::Column) -> usize;
     fn index_column(col: &Self::Column, index: usize) -> Option<Self::ScalarRef<'_>>;
@@ -650,6 +650,30 @@ pub trait AccessType: Debug + Clone + PartialEq + Sized + 'static {
     }
 }
 
+fn scalar_type_error<T>(value: &ScalarRef<'_>) -> ErrorCode {
+    ErrorCode::BadDataValueType(format!(
+        "failed to downcast scalar {:?} into {}",
+        value,
+        std::any::type_name::<T>()
+    ))
+}
+
+fn column_type_error<T>(value: &Column) -> ErrorCode {
+    ErrorCode::BadDataValueType(format!(
+        "failed to downcast column {:?} into {}",
+        value,
+        std::any::type_name::<T>()
+    ))
+}
+
+fn domain_type_error<T>(value: &Domain) -> ErrorCode {
+    ErrorCode::BadDataValueType(format!(
+        "failed to downcast domain {:?} into {}",
+        value,
+        std::any::type_name::<T>()
+    ))
+}
+
 /// [ValueType] includes the builder method of a data type based on [AccessType].
 pub trait ValueType: AccessType {
     type ColumnBuilder: Debug + Clone;
diff --git a/src/query/expression/src/types/any.rs b/src/query/expression/src/types/any.rs
index 0b54ae1b94205..2bb3d245f6eb5 100755
--- a/src/query/expression/src/types/any.rs
+++ b/src/query/expression/src/types/any.rs
@@ -15,11 +15,13 @@
 use std::cmp::Ordering;
 use std::ops::Range;
 
+use databend_common_exception::Result;
+
+use super::AccessType;
+use super::BuilderMut;
+use super::DataType;
+use super::ValueType;
 use crate::property::Domain;
-use crate::types::AccessType;
-use crate::types::BuilderMut;
-use crate::types::DataType;
-use crate::types::ValueType;
 use crate::values::Column;
 use crate::values::Scalar;
 use crate::ColumnBuilder;
@@ -44,16 +46,16 @@ impl AccessType for AnyType {
         scalar.as_ref()
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        Some(scalar.clone())
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        Ok(scalar.clone())
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        Some(col.clone())
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        Ok(col.clone())
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
-        Some(domain.clone())
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
+        Ok(domain.clone())
     }
 
     fn column_len(col: &Self::Column) -> usize {
diff --git a/src/query/expression/src/types/array.rs b/src/query/expression/src/types/array.rs
index cdf459dee9897..8481b4eb3525e 100755
--- a/src/query/expression/src/types/array.rs
+++ b/src/query/expression/src/types/array.rs
@@ -22,18 +22,21 @@ use databend_common_column::buffer::Buffer;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 
+use super::column_type_error;
+use super::domain_type_error;
+use super::scalar_type_error;
+use super::AccessType;
 use super::AnyType;
+use super::ArgType;
+use super::BuilderExt;
+use super::Column;
+use super::DataType;
+use super::GenericMap;
 use super::ReturnType;
+use super::Scalar;
+use super::ScalarRef;
+use super::ValueType;
 use crate::property::Domain;
-use crate::types::AccessType;
-use crate::types::ArgType;
-use crate::types::BuilderExt;
-use crate::types::DataType;
-use crate::types::GenericMap;
-use crate::types::Scalar;
-use crate::types::ScalarRef;
-use crate::types::ValueType;
-use crate::values::Column;
 use crate::ColumnBuilder;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -54,22 +57,25 @@ impl<T: AccessType> AccessType for ArrayType<T> {
         scalar.clone()
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
         match scalar {
             ScalarRef::Array(array) => T::try_downcast_column(array),
-            _ => None,
+            _ => Err(scalar_type_error::<Self>(scalar)),
         }
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        ArrayColumn::try_downcast(col.as_array()?)
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        let array = col
+            .as_array()
+            .ok_or_else(|| column_type_error::<Self>(col))?;
+        ArrayColumn::try_downcast(array)
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
         match domain {
-            Domain::Array(Some(domain)) => Some(Some(T::try_downcast_domain(domain)?)),
-            Domain::Array(None) => Some(None),
-            _ => None,
+            Domain::Array(Some(domain)) => Ok(Some(T::try_downcast_domain(domain)?)),
+            Domain::Array(None) => Ok(None),
+            _ => Err(domain_type_error::<Self>(domain)),
         }
     }
 
@@ -333,8 +339,8 @@ impl<T: ValueType> ArrayColumn<T> {
 }
 
 impl ArrayColumn<AnyType> {
-    pub fn try_downcast<T: AccessType>(&self) -> Option<ArrayColumn<T>> {
-        Some(ArrayColumn {
+    pub fn try_downcast<T: AccessType>(&self) -> Result<ArrayColumn<T>> {
+        Ok(ArrayColumn {
             values: T::try_downcast_column(&self.values)?,
             offsets: self.offsets.clone(),
         })
diff --git a/src/query/expression/src/types/binary.rs b/src/query/expression/src/types/binary.rs
index 64b747a3b5408..0c4c7b10b6218 100644
--- a/src/query/expression/src/types/binary.rs
+++ b/src/query/expression/src/types/binary.rs
@@ -15,15 +15,20 @@
 use std::cmp::Ordering;
 use std::ops::Range;
 
+use databend_common_exception::Result;
+
+use super::column_type_error;
+use super::domain_type_error;
+use super::scalar_type_error;
 use super::AccessType;
+use super::ArgType;
+use super::BuilderMut;
+use super::DataType;
+use super::GenericMap;
+use super::ReturnType;
+use super::ScalarRef;
+use super::ValueType;
 use crate::property::Domain;
-use crate::types::ArgType;
-use crate::types::BuilderMut;
-use crate::types::DataType;
-use crate::types::GenericMap;
-use crate::types::ReturnType;
-use crate::types::ScalarRef;
-use crate::types::ValueType;
 use crate::values::Column;
 use crate::values::Scalar;
 use crate::ColumnBuilder;
@@ -50,19 +55,24 @@ impl AccessType for BinaryType {
         scalar
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        scalar.as_binary().cloned()
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        scalar
+            .as_binary()
+            .cloned()
+            .ok_or_else(|| scalar_type_error::<Self>(scalar))
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        col.as_binary().cloned()
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        col.as_binary()
+            .cloned()
+            .ok_or_else(|| column_type_error::<Self>(col))
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
         if domain.is_undefined() {
-            Some(())
+            Ok(())
         } else {
-            None
+            Err(domain_type_error::<Self>(domain))
         }
     }
 
diff --git a/src/query/expression/src/types/bitmap.rs b/src/query/expression/src/types/bitmap.rs
index 9a65222d95c8f..841558bbfc43c 100644
--- a/src/query/expression/src/types/bitmap.rs
+++ b/src/query/expression/src/types/bitmap.rs
@@ -15,17 +15,22 @@
 use std::cmp::Ordering;
 use std::ops::Range;
 
+use databend_common_exception::Result;
+
 use super::binary::BinaryColumn;
 use super::binary::BinaryColumnBuilder;
 use super::binary::BinaryColumnIter;
+use super::column_type_error;
+use super::domain_type_error;
+use super::scalar_type_error;
+use super::AccessType;
+use super::ArgType;
+use super::BuilderMut;
+use super::DataType;
+use super::GenericMap;
+use super::ReturnType;
+use super::ValueType;
 use crate::property::Domain;
-use crate::types::AccessType;
-use crate::types::ArgType;
-use crate::types::BuilderMut;
-use crate::types::DataType;
-use crate::types::GenericMap;
-use crate::types::ReturnType;
-use crate::types::ValueType;
 use crate::values::Column;
 use crate::values::Scalar;
 use crate::ColumnBuilder;
@@ -49,19 +54,24 @@ impl AccessType for BitmapType {
         scalar
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        scalar.as_bitmap().cloned()
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        scalar
+            .as_bitmap()
+            .cloned()
+            .ok_or_else(|| scalar_type_error::<Self>(scalar))
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        col.as_bitmap().cloned()
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        col.as_bitmap()
+            .cloned()
+            .ok_or_else(|| column_type_error::<Self>(col))
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
         if domain.is_undefined() {
-            Some(())
+            Ok(())
         } else {
-            None
+            Err(domain_type_error::<Self>(domain))
         }
     }
 
diff --git a/src/query/expression/src/types/boolean.rs b/src/query/expression/src/types/boolean.rs
index 8635454287f0c..3ac9f5189ea74 100644
--- a/src/query/expression/src/types/boolean.rs
+++ b/src/query/expression/src/types/boolean.rs
@@ -17,15 +17,19 @@ use std::fmt::Debug;
 use std::ops::Range;
 
 pub use databend_common_column::bitmap::*;
+use databend_common_exception::Result;
 
+use super::column_type_error;
+use super::domain_type_error;
+use super::scalar_type_error;
 use super::AccessType;
+use super::ArgType;
+use super::BuilderMut;
+use super::DataType;
+use super::GenericMap;
+use super::ReturnType;
+use super::ValueType;
 use crate::property::Domain;
-use crate::types::ArgType;
-use crate::types::BuilderMut;
-use crate::types::DataType;
-use crate::types::GenericMap;
-use crate::types::ReturnType;
-use crate::types::ValueType;
 use crate::utils::arrow::bitmap_into_mut;
 use crate::values::Column;
 use crate::values::Scalar;
@@ -50,22 +54,25 @@ impl AccessType for BooleanType {
         *scalar
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
         match scalar {
-            ScalarRef::Boolean(scalar) => Some(*scalar),
-            _ => None,
+            ScalarRef::Boolean(scalar) => Ok(*scalar),
+            _ => Err(scalar_type_error::<Self>(scalar)),
         }
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
         match col {
-            Column::Boolean(column) => Some(column.clone()),
-            _ => None,
+            Column::Boolean(column) => Ok(column.clone()),
+            _ => Err(column_type_error::<Self>(col)),
         }
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
-        domain.as_boolean().cloned()
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
+        domain
+            .as_boolean()
+            .cloned()
+            .ok_or_else(|| domain_type_error::<Self>(domain))
     }
 
     fn column_len(col: &Self::Column) -> usize {
diff --git a/src/query/expression/src/types/compute_view.rs b/src/query/expression/src/types/compute_view.rs
index 99d30f5fc189d..1b2a27af355d3 100644
--- a/src/query/expression/src/types/compute_view.rs
+++ b/src/query/expression/src/types/compute_view.rs
@@ -17,19 +17,23 @@ use std::fmt::Debug;
 use std::marker::PhantomData;
 use std::ops::Range;
 
+use databend_common_exception::Result;
 use num_traits::AsPrimitive;
 
+use super::column_type_error;
+use super::domain_type_error;
+use super::scalar_type_error;
+use super::string::StringDomain;
+use super::string::StringIterator;
 use super::AccessType;
 use super::AnyType;
 use super::ArgType;
+use super::Number;
 use super::NumberType;
+use super::SimpleDomain;
 use super::StringColumn;
 use super::StringType;
 use crate::display::scalar_ref_to_string;
-use crate::types::string::StringDomain;
-use crate::types::string::StringIterator;
-use crate::types::Number;
-use crate::types::SimpleDomain;
 use crate::Column;
 use crate::Domain;
 use crate::ScalarRef;
@@ -80,15 +84,15 @@ where
         T::to_scalar_ref(scalar)
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        F::try_downcast_scalar(scalar).map(|v| C::compute(v))
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        F::try_downcast_scalar(scalar).map(C::compute)
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
         F::try_downcast_column(col)
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
         F::try_downcast_domain(domain).map(|domain| C::compute_domain(&domain))
     }
 
@@ -195,16 +199,24 @@ impl AccessType for OwnedStringType {
         scalar.clone()
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        scalar.as_string().map(|s| s.to_string())
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        scalar
+            .as_string()
+            .map(|s| s.to_string())
+            .ok_or_else(|| scalar_type_error::<Self>(scalar))
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        col.as_string().cloned()
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        col.as_string()
+            .cloned()
+            .ok_or_else(|| column_type_error::<Self>(col))
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
-        domain.as_string().cloned()
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
+        domain
+            .as_string()
+            .cloned()
+            .ok_or_else(|| domain_type_error::<Self>(domain))
     }
 
     fn column_len(col: &Self::Column) -> usize {
diff --git a/src/query/expression/src/types/decimal.rs b/src/query/expression/src/types/decimal.rs
index 3458a70979432..3f068eef61683 100644
--- a/src/query/expression/src/types/decimal.rs
+++ b/src/query/expression/src/types/decimal.rs
@@ -49,17 +49,20 @@ use num_traits::ToPrimitive;
 use serde::Deserialize;
 use serde::Serialize;
 
+use super::column_type_error;
 use super::compute_view::Compute;
 use super::compute_view::ComputeView;
+use super::domain_type_error;
+use super::scalar_type_error;
 use super::AccessType;
 use super::AnyType;
+use super::DataType;
 use super::NumberType;
 use super::SimpleDomain;
 use super::SimpleType;
 use super::SimpleValueType;
 use super::ValueType;
-use crate::types::DataType;
-use crate::types::F64;
+use super::F64;
 use crate::utils::arrow::buffer_into_mut;
 use crate::with_decimal_mapped_type;
 use crate::with_decimal_type;
@@ -98,16 +101,24 @@ impl<Num: Decimal> AccessType for CoreDecimal<Num> {
         *scalar
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        Num::try_downcast_scalar(scalar.as_decimal()?)
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        let decimal = scalar
+            .as_decimal()
+            .ok_or_else(|| scalar_type_error::<Self>(scalar))?;
+        Num::try_downcast_scalar(decimal).ok_or_else(|| scalar_type_error::<Self>(scalar))
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
-        Num::try_downcast_domain(domain.as_decimal()?)
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
+        let decimal = domain
+            .as_decimal()
+            .ok_or_else(|| domain_type_error::<Self>(domain))?;
+        Num::try_downcast_domain(decimal).ok_or_else(|| domain_type_error::<Self>(domain))
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        Num::try_downcast_column(col).map(|(col, _)| col)
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        Num::try_downcast_column(col)
+            .map(|(col, _)| col)
+            .ok_or_else(|| column_type_error::<Self>(col))
     }
 
     fn column_len(col: &Self::Column) -> usize {
@@ -222,22 +233,32 @@ impl<Num: Decimal> AccessType for CoreScalarDecimal<Num> {
         *scalar
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        let scalar = scalar.as_decimal()?;
-        Num::try_downcast_scalar(scalar).map(|v| (v, scalar.size()))
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        let decimal = scalar
+            .as_decimal()
+            .ok_or_else(|| scalar_type_error::<Self>(scalar))?;
+        Num::try_downcast_scalar(decimal)
+            .map(|v| (v, decimal.size()))
+            .ok_or_else(|| scalar_type_error::<Self>(scalar))
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
-        let size = domain.as_decimal()?.decimal_size();
-        let domain = Num::try_downcast_domain(domain.as_decimal()?)?;
-        Some(SimpleDomain {
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
+        let decimal = domain
+            .as_decimal()
+            .ok_or_else(|| domain_type_error::<Self>(domain))?;
+        let size = decimal.decimal_size();
+        let domain =
+            Num::try_downcast_domain(decimal).ok_or_else(|| domain_type_error::<Self>(domain))?;
+        Ok(SimpleDomain {
             min: (domain.min, size),
             max: (domain.max, size),
         })
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        Num::try_downcast_column(col).map(|(col, x)| col.into_iter().map(|v| (v, x)).collect())
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        Num::try_downcast_column(col)
+            .map(|(col, x)| col.into_iter().map(|v| (v, x)).collect())
+            .ok_or_else(|| column_type_error::<Self>(col))
     }
 
     fn column_len(col: &Self::Column) -> usize {
@@ -3132,8 +3153,8 @@ decimal_convert_type!(i256, i64, Decimal256As64Type, I256ToI64);
 
 #[cfg(test)]
 mod tests {
+    use super::ValueType;
     use super::*;
-    use crate::types::ValueType;
 
     #[test]
     fn test_decimal_cast() {
diff --git a/src/query/expression/src/types/empty_array.rs b/src/query/expression/src/types/empty_array.rs
index 4f35f7dceb605..d865c24976dc9 100644
--- a/src/query/expression/src/types/empty_array.rs
+++ b/src/query/expression/src/types/empty_array.rs
@@ -12,11 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use super::ArgType;
+use super::DataType;
 use super::ZeroSizeType;
 use super::ZeroSizeValueType;
 use crate::property::Domain;
-use crate::types::ArgType;
-use crate::types::DataType;
 use crate::values::Column;
 use crate::values::Scalar;
 use crate::ColumnBuilder;
diff --git a/src/query/expression/src/types/empty_map.rs b/src/query/expression/src/types/empty_map.rs
index 2a647ed8b0dcb..88a965a4492c2 100644
--- a/src/query/expression/src/types/empty_map.rs
+++ b/src/query/expression/src/types/empty_map.rs
@@ -12,11 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use super::ArgType;
+use super::DataType;
 use super::ZeroSizeType;
 use super::ZeroSizeValueType;
 use crate::property::Domain;
-use crate::types::ArgType;
-use crate::types::DataType;
 use crate::values::Column;
 use crate::values::Scalar;
 use crate::ColumnBuilder;
diff --git a/src/query/expression/src/types/generic.rs b/src/query/expression/src/types/generic.rs
index 438b4fd9fe587..a3bec9057894c 100755
--- a/src/query/expression/src/types/generic.rs
+++ b/src/query/expression/src/types/generic.rs
@@ -15,14 +15,16 @@
 use std::cmp::Ordering;
 use std::ops::Range;
 
+use databend_common_exception::Result;
+
+use super::AccessType;
+use super::ArgType;
+use super::BuilderMut;
+use super::DataType;
+use super::GenericMap;
 use super::ReturnType;
+use super::ValueType;
 use crate::property::Domain;
-use crate::types::AccessType;
-use crate::types::ArgType;
-use crate::types::BuilderMut;
-use crate::types::DataType;
-use crate::types::GenericMap;
-use crate::types::ValueType;
 use crate::values::Column;
 use crate::values::ColumnBuilder;
 use crate::values::ColumnIterator;
@@ -47,16 +49,16 @@ impl<const INDEX: usize> AccessType for GenericType<INDEX> {
         scalar.as_ref()
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        Some(scalar.clone())
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        Ok(scalar.clone())
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        Some(col.clone())
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        Ok(col.clone())
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
-        Some(domain.clone())
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
+        Ok(domain.clone())
     }
 
     fn column_len(col: &Self::Column) -> usize {
diff --git a/src/query/expression/src/types/geography.rs b/src/query/expression/src/types/geography.rs
index e01a428899ae5..d5a9ea98ce118 100644
--- a/src/query/expression/src/types/geography.rs
+++ b/src/query/expression/src/types/geography.rs
@@ -29,17 +29,20 @@ use geozero::ToWkt;
 use serde::Deserialize;
 use serde::Serialize;
 
+use super::binary::BinaryColumn;
+use super::binary::BinaryColumnBuilder;
 use super::binary::BinaryColumnIter;
+use super::column_type_error;
+use super::domain_type_error;
+use super::scalar_type_error;
 use super::AccessType;
+use super::ArgType;
+use super::BuilderMut;
+use super::DataType;
+use super::GenericMap;
+use super::ReturnType;
+use super::ValueType;
 use crate::property::Domain;
-use crate::types::binary::BinaryColumn;
-use crate::types::binary::BinaryColumnBuilder;
-use crate::types::ArgType;
-use crate::types::BuilderMut;
-use crate::types::DataType;
-use crate::types::GenericMap;
-use crate::types::ReturnType;
-use crate::types::ValueType;
 use crate::values::Column;
 use crate::values::Scalar;
 use crate::values::ScalarRef;
@@ -126,19 +129,24 @@ impl AccessType for GeographyType {
         scalar.as_ref()
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        scalar.as_geography().cloned()
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        scalar
+            .as_geography()
+            .cloned()
+            .ok_or_else(|| scalar_type_error::<Self>(scalar))
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        col.as_geography().cloned()
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        col.as_geography()
+            .cloned()
+            .ok_or_else(|| column_type_error::<Self>(col))
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
         if domain.is_undefined() {
-            Some(())
+            Ok(())
         } else {
-            None
+            Err(domain_type_error::<Self>(domain))
         }
     }
 
diff --git a/src/query/expression/src/types/geometry.rs b/src/query/expression/src/types/geometry.rs
index 1fb9194fc222d..0ad07ed49269e 100644
--- a/src/query/expression/src/types/geometry.rs
+++ b/src/query/expression/src/types/geometry.rs
@@ -15,6 +15,7 @@
 use std::cmp::Ordering;
 use std::ops::Range;
 
+use databend_common_exception::Result;
 use geozero::wkb::FromWkb;
 use geozero::wkb::WkbDialect;
 use geozero::wkt::Ewkt;
@@ -22,16 +23,19 @@ use geozero::wkt::Ewkt;
 use super::binary::BinaryColumn;
 use super::binary::BinaryColumnBuilder;
 use super::binary::BinaryColumnIter;
+use super::column_type_error;
+use super::domain_type_error;
+use super::scalar_type_error;
 use super::AccessType;
+use super::ArgType;
+use super::BuilderMut;
+use super::DataType;
+use super::GenericMap;
 use super::ReturnType;
+use super::Scalar;
+use super::ScalarRef;
+use super::ValueType;
 use crate::property::Domain;
-use crate::types::ArgType;
-use crate::types::BuilderMut;
-use crate::types::DataType;
-use crate::types::GenericMap;
-use crate::types::Scalar;
-use crate::types::ScalarRef;
-use crate::types::ValueType;
 use crate::values::Column;
 use crate::ColumnBuilder;
 
@@ -53,19 +57,24 @@ impl AccessType for GeometryType {
         scalar
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        scalar.as_geometry().cloned()
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        scalar
+            .as_geometry()
+            .cloned()
+            .ok_or_else(|| scalar_type_error::<Self>(scalar))
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        col.as_geometry().cloned()
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        col.as_geometry()
+            .cloned()
+            .ok_or_else(|| column_type_error::<Self>(col))
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
         if domain.is_undefined() {
-            Some(())
+            Ok(())
         } else {
-            None
+            Err(domain_type_error::<Self>(domain))
         }
     }
 
diff --git a/src/query/expression/src/types/interval.rs b/src/query/expression/src/types/interval.rs
index d4931b05e5c8b..d990e45f4a7f6 100644
--- a/src/query/expression/src/types/interval.rs
+++ b/src/query/expression/src/types/interval.rs
@@ -20,11 +20,11 @@ use databend_common_column::types::months_days_micros;
 use databend_common_io::Interval;
 
 use super::number::SimpleDomain;
+use super::ArgType;
+use super::DataType;
+use super::SimpleType;
+use super::SimpleValueType;
 use crate::property::Domain;
-use crate::types::ArgType;
-use crate::types::DataType;
-use crate::types::SimpleType;
-use crate::types::SimpleValueType;
 use crate::values::Column;
 use crate::values::Scalar;
 use crate::ColumnBuilder;
diff --git a/src/query/expression/src/types/map.rs b/src/query/expression/src/types/map.rs
index 1f5b090b15885..8777ff6efca56 100755
--- a/src/query/expression/src/types/map.rs
+++ b/src/query/expression/src/types/map.rs
@@ -17,19 +17,24 @@ use std::iter::TrustedLen;
 use std::marker::PhantomData;
 use std::ops::Range;
 
+use databend_common_exception::Result;
+
+use super::array::ArrayColumn;
+use super::array::ArrayColumnBuilderMut;
+use super::column_type_error;
+use super::domain_type_error;
+use super::scalar_type_error;
 use super::AccessType;
+use super::ArgType;
 use super::ArrayType;
+use super::BuilderExt;
+use super::DataType;
+use super::GenericMap;
 use super::ReturnType;
+use super::Scalar;
+use super::ScalarRef;
+use super::ValueType;
 use crate::property::Domain;
-use crate::types::array::ArrayColumn;
-use crate::types::array::ArrayColumnBuilderMut;
-use crate::types::ArgType;
-use crate::types::BuilderExt;
-use crate::types::DataType;
-use crate::types::GenericMap;
-use crate::types::Scalar;
-use crate::types::ScalarRef;
-use crate::types::ValueType;
 use crate::values::Column;
 use crate::ColumnBuilder;
 
@@ -52,33 +57,33 @@ impl<K: AccessType, V: AccessType> AccessType for KvPair<K, V> {
         (K::to_scalar_ref(k), V::to_scalar_ref(v))
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
         match scalar {
-            ScalarRef::Tuple(fields) if fields.len() == 2 => Some((
+            ScalarRef::Tuple(fields) if fields.len() == 2 => Ok((
                 K::try_downcast_scalar(&fields[0])?,
                 V::try_downcast_scalar(&fields[1])?,
             )),
-            _ => None,
+            _ => Err(scalar_type_error::<Self>(scalar)),
         }
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
         match col {
-            Column::Tuple(fields) if fields.len() == 2 => Some(KvColumn {
+            Column::Tuple(fields) if fields.len() == 2 => Ok(KvColumn {
                 keys: K::try_downcast_column(&fields[0])?,
                 values: V::try_downcast_column(&fields[1])?,
             }),
-            _ => None,
+            _ => Err(column_type_error::<Self>(col)),
         }
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
         match domain {
-            Domain::Tuple(fields) if fields.len() == 2 => Some((
+            Domain::Tuple(fields) if fields.len() == 2 => Ok((
                 K::try_downcast_domain(&fields[0])?,
                 V::try_downcast_domain(&fields[1])?,
             )),
-            _ => None,
+            _ => Err(domain_type_error::<Self>(domain)),
         }
     }
 
@@ -451,22 +456,23 @@ impl<K: AccessType, V: AccessType> AccessType for MapType<K, V> {
         MapInternal::<K, V>::to_scalar_ref(scalar)
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
         match scalar {
             ScalarRef::Map(array) => KvPair::<K, V>::try_downcast_column(array),
-            _ => None,
+            _ => Err(scalar_type_error::<Self>(scalar)),
         }
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        ArrayColumn::try_downcast(col.as_map()?)
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        let map = col.as_map().ok_or_else(|| column_type_error::<Self>(col))?;
+        ArrayColumn::try_downcast(map)
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
         match domain {
-            Domain::Map(Some(domain)) => Some(Some(KvPair::<K, V>::try_downcast_domain(domain)?)),
-            Domain::Map(None) => Some(None),
-            _ => None,
+            Domain::Map(Some(domain)) => Ok(Some(KvPair::<K, V>::try_downcast_domain(domain)?)),
+            Domain::Map(None) => Ok(None),
+            _ => Err(domain_type_error::<Self>(domain)),
         }
     }
 
diff --git a/src/query/expression/src/types/null.rs b/src/query/expression/src/types/null.rs
index 5f3a76fe50d6c..094c114778882 100644
--- a/src/query/expression/src/types/null.rs
+++ b/src/query/expression/src/types/null.rs
@@ -13,11 +13,11 @@
 // limitations under the License.
 
 use super::nullable::NullableDomain;
+use super::ArgType;
+use super::DataType;
 use super::ZeroSizeType;
 use super::ZeroSizeValueType;
 use crate::property::Domain;
-use crate::types::ArgType;
-use crate::types::DataType;
 use crate::values::Column;
 use crate::values::Scalar;
 use crate::ColumnBuilder;
diff --git a/src/query/expression/src/types/nullable.rs b/src/query/expression/src/types/nullable.rs
index 6ffa24ada037a..ed4263fe07091 100755
--- a/src/query/expression/src/types/nullable.rs
+++ b/src/query/expression/src/types/nullable.rs
@@ -19,16 +19,19 @@ use std::ops::Range;
 
 use databend_common_column::bitmap::Bitmap;
 use databend_common_column::bitmap::MutableBitmap;
+use databend_common_exception::Result;
 
+use super::column_type_error;
+use super::domain_type_error;
 use super::AccessType;
 use super::AnyType;
+use super::ArgType;
+use super::BuilderExt;
+use super::DataType;
+use super::GenericMap;
 use super::ReturnType;
+use super::ValueType;
 use crate::property::Domain;
-use crate::types::ArgType;
-use crate::types::BuilderExt;
-use crate::types::DataType;
-use crate::types::GenericMap;
-use crate::types::ValueType;
 use crate::utils::arrow::bitmap_into_mut;
 use crate::values::Column;
 use crate::values::Scalar;
@@ -54,34 +57,37 @@ impl<T: AccessType> AccessType for NullableType<T> {
         scalar.as_ref().map(T::to_scalar_ref)
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
         match scalar {
-            ScalarRef::Null => Some(None),
-            scalar => Some(Some(T::try_downcast_scalar(scalar)?)),
+            ScalarRef::Null => Ok(None),
+            scalar => Ok(Some(T::try_downcast_scalar(scalar)?)),
         }
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        NullableColumn::try_downcast(col.as_nullable()?)
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        let nullable = col
+            .as_nullable()
+            .ok_or_else(|| column_type_error::<Self>(col))?;
+        NullableColumn::try_downcast(nullable)
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
         match domain {
             Domain::Nullable(NullableDomain {
                 has_null,
                 value: Some(value),
-            }) => Some(NullableDomain {
+            }) => Ok(NullableDomain {
                 has_null: *has_null,
                 value: Some(Box::new(T::try_downcast_domain(value)?)),
             }),
             Domain::Nullable(NullableDomain {
                 has_null,
                 value: None,
-            }) => Some(NullableDomain {
+            }) => Ok(NullableDomain {
                 has_null: *has_null,
                 value: None,
             }),
-            _ => None,
+            _ => Err(domain_type_error::<Self>(domain)),
         }
     }
 
@@ -381,10 +387,10 @@ impl<T: ArgType> NullableColumn<T> {
 }
 
 impl NullableColumn<AnyType> {
-    pub fn try_downcast<T: AccessType>(&self) -> Option<NullableColumn<T>> {
+    pub fn try_downcast<T: AccessType>(&self) -> Result<NullableColumn<T>> {
         let inner = T::try_downcast_column(&self.column)?;
         debug_assert_eq!(T::column_len(&inner), self.validity.len());
-        Some(NullableColumn {
+        Ok(NullableColumn {
             column: inner,
             validity: self.validity.clone(),
         })
diff --git a/src/query/expression/src/types/number.rs b/src/query/expression/src/types/number.rs
index 0bd92a7d526c2..5845b63a7d4d5 100644
--- a/src/query/expression/src/types/number.rs
+++ b/src/query/expression/src/types/number.rs
@@ -33,11 +33,11 @@ use serde::Deserialize;
 use serde::Serialize;
 
 use super::decimal::DecimalSize;
+use super::ArgType;
+use super::DataType;
+use super::SimpleType;
+use super::SimpleValueType;
 use crate::property::Domain;
-use crate::types::ArgType;
-use crate::types::DataType;
-use crate::types::SimpleType;
-use crate::types::SimpleValueType;
 use crate::utils::arrow::buffer_into_mut;
 use crate::values::Column;
 use crate::values::Scalar;
diff --git a/src/query/expression/src/types/number_class.rs b/src/query/expression/src/types/number_class.rs
index 12d3d0ecb0fc2..2a555548433f3 100644
--- a/src/query/expression/src/types/number_class.rs
+++ b/src/query/expression/src/types/number_class.rs
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::types::NumberDataType;
+use super::NumberDataType;
 
 pub enum NumberClass {
     UInt8,
diff --git a/src/query/expression/src/types/opaque.rs b/src/query/expression/src/types/opaque.rs
index 0c4b682306818..02d9145564e43 100644
--- a/src/query/expression/src/types/opaque.rs
+++ b/src/query/expression/src/types/opaque.rs
@@ -20,10 +20,13 @@ use databend_common_column::buffer::Buffer;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 
-use crate::types::AccessType;
-use crate::types::BuilderMut;
-use crate::types::DataType;
-use crate::types::ValueType;
+use super::column_type_error;
+use super::domain_type_error;
+use super::scalar_type_error;
+use super::AccessType;
+use super::BuilderMut;
+use super::DataType;
+use super::ValueType;
 use crate::Column;
 use crate::ColumnBuilder;
 use crate::Domain;
@@ -107,32 +110,42 @@ impl<const N: usize> AccessType for OpaqueType<N> {
         scalar
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        let scalar = scalar.as_opaque()?;
-        with_opaque_type!(|T| match scalar {
-            OpaqueScalarRef::T(v) => Some(reinterpret_ref(v)),
-        })
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        let opaque = scalar
+            .as_opaque()
+            .ok_or_else(|| scalar_type_error::<Self>(scalar))?;
+        match (opaque, N) {
+            (OpaqueScalarRef::Opaque1(v), 1) => Ok(reinterpret_ref(v)),
+            (OpaqueScalarRef::Opaque2(v), 2) => Ok(reinterpret_ref(v)),
+            (OpaqueScalarRef::Opaque3(v), 3) => Ok(reinterpret_ref(v)),
+            (OpaqueScalarRef::Opaque4(v), 4) => Ok(reinterpret_ref(v)),
+            (OpaqueScalarRef::Opaque5(v), 5) => Ok(reinterpret_ref(v)),
+            (OpaqueScalarRef::Opaque6(v), 6) => Ok(reinterpret_ref(v)),
+            _ => Err(scalar_type_error::<Self>(scalar)),
+        }
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
         if domain.is_undefined() {
-            Some(())
+            Ok(())
         } else {
-            None
+            Err(domain_type_error::<Self>(domain))
         }
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        let opaque_col = col.as_opaque()?;
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        let opaque_col = col
+            .as_opaque()
+            .ok_or_else(|| column_type_error::<Self>(col))?;
         unsafe {
             match (opaque_col, N) {
-                (OpaqueColumn::Opaque1(buffer), 1) => Some(std::mem::transmute(buffer.clone())),
-                (OpaqueColumn::Opaque2(buffer), 2) => Some(std::mem::transmute(buffer.clone())),
-                (OpaqueColumn::Opaque3(buffer), 3) => Some(std::mem::transmute(buffer.clone())),
-                (OpaqueColumn::Opaque4(buffer), 4) => Some(std::mem::transmute(buffer.clone())),
-                (OpaqueColumn::Opaque5(buffer), 5) => Some(std::mem::transmute(buffer.clone())),
-                (OpaqueColumn::Opaque6(buffer), 6) => Some(std::mem::transmute(buffer.clone())),
-                _ => None,
+                (OpaqueColumn::Opaque1(buffer), 1) => Ok(std::mem::transmute(buffer.clone())),
+                (OpaqueColumn::Opaque2(buffer), 2) => Ok(std::mem::transmute(buffer.clone())),
+                (OpaqueColumn::Opaque3(buffer), 3) => Ok(std::mem::transmute(buffer.clone())),
+                (OpaqueColumn::Opaque4(buffer), 4) => Ok(std::mem::transmute(buffer.clone())),
+                (OpaqueColumn::Opaque5(buffer), 5) => Ok(std::mem::transmute(buffer.clone())),
+                (OpaqueColumn::Opaque6(buffer), 6) => Ok(std::mem::transmute(buffer.clone())),
+                _ => Err(column_type_error::<Self>(col)),
             }
         }
     }
@@ -527,13 +540,13 @@ impl OpaqueColumnBuilder {
     }
 }
 
-impl<const N: usize> crate::types::ReturnType for OpaqueType<N> {
-    fn create_builder(capacity: usize, _: &crate::types::GenericMap) -> Self::ColumnBuilder {
+impl<const N: usize> super::ReturnType for OpaqueType<N> {
+    fn create_builder(capacity: usize, _: &super::GenericMap) -> Self::ColumnBuilder {
         Vec::with_capacity(capacity)
     }
 }
 
-impl<const N: usize> crate::types::ArgType for OpaqueType<N> {
+impl<const N: usize> super::ArgType for OpaqueType<N> {
     fn data_type() -> DataType {
         DataType::Opaque(N)
     }
diff --git a/src/query/expression/src/types/simple_type.rs b/src/query/expression/src/types/simple_type.rs
index e7fcd8cc89720..ad7ea3f81de39 100644
--- a/src/query/expression/src/types/simple_type.rs
+++ b/src/query/expression/src/types/simple_type.rs
@@ -18,15 +18,19 @@ use std::marker::PhantomData;
 use std::ops::Range;
 
 use databend_common_column::buffer::Buffer;
+use databend_common_exception::Result;
 
+use super::column_type_error;
+use super::domain_type_error;
+use super::scalar_type_error;
 use super::AccessType;
+use super::BuilderMut;
+use super::DataType;
 use super::GenericMap;
 use super::ReturnType;
 use super::Scalar;
 use super::ValueType;
 use crate::arrow::buffer_into_mut;
-use crate::types::BuilderMut;
-use crate::types::DataType;
 use crate::Column;
 use crate::ColumnBuilder;
 use crate::Domain;
@@ -96,16 +100,16 @@ impl<T: SimpleType> AccessType for SimpleValueType<T> {
         *scalar
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        T::downcast_scalar(scalar)
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        T::downcast_scalar(scalar).ok_or_else(|| scalar_type_error::<Self>(scalar))
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Buffer<Self::Scalar>> {
-        T::downcast_column(col)
+    fn try_downcast_column(col: &Column) -> Result<Buffer<Self::Scalar>> {
+        T::downcast_column(col).ok_or_else(|| column_type_error::<Self>(col))
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
-        T::downcast_domain(domain)
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
+        T::downcast_domain(domain).ok_or_else(|| domain_type_error::<Self>(domain))
     }
 
     fn column_len(buffer: &Buffer<Self::Scalar>) -> usize {
diff --git a/src/query/expression/src/types/string.rs b/src/query/expression/src/types/string.rs
index 9785743b9fcd6..0b302ba0a4c0c 100644
--- a/src/query/expression/src/types/string.rs
+++ b/src/query/expression/src/types/string.rs
@@ -20,16 +20,19 @@ use databend_common_column::binview::BinaryViewColumnIter;
 use databend_common_column::binview::Utf8ViewColumn;
 use databend_common_exception::Result;
 
+use super::binary::BinaryColumn;
+use super::column_type_error;
+use super::domain_type_error;
+use super::scalar_type_error;
 use super::AccessType;
+use super::ArgType;
+use super::BuilderMut;
+use super::ColumnBuilder;
+use super::DataType;
+use super::GenericMap;
+use super::ReturnType;
+use super::ValueType;
 use crate::property::Domain;
-use crate::types::binary::BinaryColumn;
-use crate::types::ArgType;
-use crate::types::BuilderMut;
-use crate::types::ColumnBuilder;
-use crate::types::DataType;
-use crate::types::GenericMap;
-use crate::types::ReturnType;
-use crate::types::ValueType;
 use crate::values::Column;
 use crate::values::Scalar;
 use crate::ScalarRef;
@@ -52,16 +55,24 @@ impl AccessType for StringType {
         scalar
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        scalar.as_string().cloned()
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        scalar
+            .as_string()
+            .cloned()
+            .ok_or_else(|| scalar_type_error::<Self>(scalar))
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        col.as_string().cloned()
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        col.as_string()
+            .cloned()
+            .ok_or_else(|| column_type_error::<Self>(col))
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
-        domain.as_string().cloned()
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
+        domain
+            .as_string()
+            .cloned()
+            .ok_or_else(|| domain_type_error::<Self>(domain))
     }
 
     fn column_len(col: &Self::Column) -> usize {
diff --git a/src/query/expression/src/types/timestamp.rs b/src/query/expression/src/types/timestamp.rs
index 21b8996f9ad29..da2275c501d98 100644
--- a/src/query/expression/src/types/timestamp.rs
+++ b/src/query/expression/src/types/timestamp.rs
@@ -26,11 +26,11 @@ use jiff::tz::TimeZone;
 use jiff::Zoned;
 
 use super::number::SimpleDomain;
+use super::ArgType;
+use super::DataType;
+use super::SimpleType;
+use super::SimpleValueType;
 use crate::property::Domain;
-use crate::types::ArgType;
-use crate::types::DataType;
-use crate::types::SimpleType;
-use crate::types::SimpleValueType;
 use crate::utils::date_helper::DateConverter;
 use crate::values::Column;
 use crate::values::Scalar;
diff --git a/src/query/expression/src/types/timestamp_tz.rs b/src/query/expression/src/types/timestamp_tz.rs
index 7a7828767e004..33ba974d76e9e 100644
--- a/src/query/expression/src/types/timestamp_tz.rs
+++ b/src/query/expression/src/types/timestamp_tz.rs
@@ -20,13 +20,13 @@ use jiff::fmt;
 use jiff::tz;
 use jiff::tz::TimeZone;
 
-use crate::types::simple_type::SimpleType;
-use crate::types::simple_type::SimpleValueType;
-use crate::types::timestamp::TIMESTAMP_MAX;
-use crate::types::timestamp::TIMESTAMP_MIN;
-use crate::types::ArgType;
-use crate::types::DataType;
-use crate::types::SimpleDomain;
+use super::simple_type::SimpleType;
+use super::simple_type::SimpleValueType;
+use super::timestamp::TIMESTAMP_MAX;
+use super::timestamp::TIMESTAMP_MIN;
+use super::ArgType;
+use super::DataType;
+use super::SimpleDomain;
 use crate::Column;
 use crate::ColumnBuilder;
 use crate::Domain;
diff --git a/src/query/expression/src/types/tuple.rs b/src/query/expression/src/types/tuple.rs
index 8100da50504f6..a3b013e0d8aea 100644
--- a/src/query/expression/src/types/tuple.rs
+++ b/src/query/expression/src/types/tuple.rs
@@ -17,6 +17,11 @@ use std::iter::Map;
 use std::iter::Zip;
 use std::marker::PhantomData;
 
+use databend_common_exception::Result;
+
+use super::column_type_error;
+use super::domain_type_error;
+use super::scalar_type_error;
 use super::AccessType;
 use super::AnyType;
 use crate::Column;
@@ -52,23 +57,32 @@ where A: AccessType
         A::to_scalar_ref(scalar)
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        let [a] = scalar.as_tuple()?.as_slice() else {
-            return None;
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        let Some(tuple) = scalar.as_tuple() else {
+            return Err(scalar_type_error::<Self>(scalar));
+        };
+        let [a] = tuple.as_slice() else {
+            return Err(scalar_type_error::<Self>(scalar));
         };
         A::try_downcast_scalar(a)
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
-        let [a] = domain.as_tuple()?.as_slice() else {
-            return None;
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
+        let Some(tuple) = domain.as_tuple() else {
+            return Err(domain_type_error::<Self>(domain));
+        };
+        let [a] = tuple.as_slice() else {
+            return Err(domain_type_error::<Self>(domain));
         };
         A::try_downcast_domain(a)
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        let [a] = col.as_tuple()?.as_slice() else {
-            return None;
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        let Some(tuple) = col.as_tuple() else {
+            return Err(column_type_error::<Self>(col));
+        };
+        let [a] = tuple.as_slice() else {
+            return Err(column_type_error::<Self>(col));
         };
         A::try_downcast_column(a)
     }
@@ -120,25 +134,34 @@ where
         (A::to_scalar_ref(a), B::to_scalar_ref(b))
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        let [a, b] = scalar.as_tuple()?.as_slice() else {
-            return None;
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        let Some(tuple) = scalar.as_tuple() else {
+            return Err(scalar_type_error::<Self>(scalar));
+        };
+        let [a, b] = tuple.as_slice() else {
+            return Err(scalar_type_error::<Self>(scalar));
         };
-        Some((A::try_downcast_scalar(a)?, B::try_downcast_scalar(b)?))
+        Ok((A::try_downcast_scalar(a)?, B::try_downcast_scalar(b)?))
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
-        let [a, b] = domain.as_tuple()?.as_slice() else {
-            return None;
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
+        let Some(tuple) = domain.as_tuple() else {
+            return Err(domain_type_error::<Self>(domain));
         };
-        Some((A::try_downcast_domain(a)?, B::try_downcast_domain(b)?))
+        let [a, b] = tuple.as_slice() else {
+            return Err(domain_type_error::<Self>(domain));
+        };
+        Ok((A::try_downcast_domain(a)?, B::try_downcast_domain(b)?))
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        let [a, b] = col.as_tuple()?.as_slice() else {
-            return None;
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        let Some(tuple) = col.as_tuple() else {
+            return Err(column_type_error::<Self>(col));
+        };
+        let [a, b] = tuple.as_slice() else {
+            return Err(column_type_error::<Self>(col));
         };
-        Some((A::try_downcast_column(a)?, B::try_downcast_column(b)?))
+        Ok((A::try_downcast_column(a)?, B::try_downcast_column(b)?))
     }
 
     fn column_len((a, _b): &Self::Column) -> usize {
@@ -207,33 +230,42 @@ where
         )
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        let [a, b, c] = scalar.as_tuple()?.as_slice() else {
-            return None;
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        let Some(tuple) = scalar.as_tuple() else {
+            return Err(scalar_type_error::<Self>(scalar));
         };
-        Some((
+        let [a, b, c] = tuple.as_slice() else {
+            return Err(scalar_type_error::<Self>(scalar));
+        };
+        Ok((
             A::try_downcast_scalar(a)?,
             B::try_downcast_scalar(b)?,
             C::try_downcast_scalar(c)?,
         ))
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
-        let [a, b, c] = domain.as_tuple()?.as_slice() else {
-            return None;
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
+        let Some(tuple) = domain.as_tuple() else {
+            return Err(domain_type_error::<Self>(domain));
         };
-        Some((
+        let [a, b, c] = tuple.as_slice() else {
+            return Err(domain_type_error::<Self>(domain));
+        };
+        Ok((
             A::try_downcast_domain(a)?,
             B::try_downcast_domain(b)?,
             C::try_downcast_domain(c)?,
         ))
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        let [a, b, c] = col.as_tuple()?.as_slice() else {
-            return None;
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        let Some(tuple) = col.as_tuple() else {
+            return Err(column_type_error::<Self>(col));
         };
-        Some((
+        let [a, b, c] = tuple.as_slice() else {
+            return Err(column_type_error::<Self>(col));
+        };
+        Ok((
             A::try_downcast_column(a)?,
             B::try_downcast_column(b)?,
             C::try_downcast_column(c)?,
@@ -327,11 +359,14 @@ where
         )
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        let [a, b, c, d] = scalar.as_tuple()?.as_slice() else {
-            return None;
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        let Some(tuple) = scalar.as_tuple() else {
+            return Err(scalar_type_error::<Self>(scalar));
         };
-        Some((
+        let [a, b, c, d] = tuple.as_slice() else {
+            return Err(scalar_type_error::<Self>(scalar));
+        };
+        Ok((
             A::try_downcast_scalar(a)?,
             B::try_downcast_scalar(b)?,
             C::try_downcast_scalar(c)?,
@@ -339,11 +374,14 @@ where
         ))
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
-        let [a, b, c, d] = domain.as_tuple()?.as_slice() else {
-            return None;
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
+        let Some(tuple) = domain.as_tuple() else {
+            return Err(domain_type_error::<Self>(domain));
         };
-        Some((
+        let [a, b, c, d] = tuple.as_slice() else {
+            return Err(domain_type_error::<Self>(domain));
+        };
+        Ok((
             A::try_downcast_domain(a)?,
             B::try_downcast_domain(b)?,
             C::try_downcast_domain(c)?,
@@ -351,11 +389,14 @@ where
         ))
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        let [a, b, c, d] = col.as_tuple()?.as_slice() else {
-            return None;
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        let Some(tuple) = col.as_tuple() else {
+            return Err(column_type_error::<Self>(col));
         };
-        Some((
+        let [a, b, c, d] = tuple.as_slice() else {
+            return Err(column_type_error::<Self>(col));
+        };
+        Ok((
             A::try_downcast_column(a)?,
             B::try_downcast_column(b)?,
             C::try_downcast_column(c)?,
diff --git a/src/query/expression/src/types/variant.rs b/src/query/expression/src/types/variant.rs
index 05f18df07cda4..58b705b7cf79d 100644
--- a/src/query/expression/src/types/variant.rs
+++ b/src/query/expression/src/types/variant.rs
@@ -16,6 +16,7 @@ use core::cmp::Ordering;
 use std::collections::BTreeMap;
 use std::ops::Range;
 
+use databend_common_exception::Result;
 use databend_common_io::deserialize_bitmap;
 use geozero::wkb::Ewkb;
 use geozero::ToJson;
@@ -27,21 +28,24 @@ use jsonb::Value;
 use super::binary::BinaryColumn;
 use super::binary::BinaryColumnBuilder;
 use super::binary::BinaryColumnIter;
+use super::column_type_error;
 use super::date::date_to_string;
+use super::domain_type_error;
+use super::map::KvPair;
 use super::number::NumberScalar;
+use super::scalar_type_error;
 use super::timestamp::timestamp_to_string;
 use super::AccessType;
+use super::AnyType;
+use super::ArgType;
+use super::BuilderMut;
+use super::DataType;
+use super::DecimalScalar;
+use super::GenericMap;
+use super::ReturnType;
+use super::ValueType;
+use super::VectorScalarRef;
 use crate::property::Domain;
-use crate::types::map::KvPair;
-use crate::types::AnyType;
-use crate::types::ArgType;
-use crate::types::BuilderMut;
-use crate::types::DataType;
-use crate::types::DecimalScalar;
-use crate::types::GenericMap;
-use crate::types::ReturnType;
-use crate::types::ValueType;
-use crate::types::VectorScalarRef;
 use crate::values::Column;
 use crate::values::Scalar;
 use crate::values::ScalarRef;
@@ -70,19 +74,24 @@ impl AccessType for VariantType {
         scalar
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        scalar.as_variant().cloned()
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        scalar
+            .as_variant()
+            .cloned()
+            .ok_or_else(|| scalar_type_error::<Self>(scalar))
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        col.as_variant().cloned()
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        col.as_variant()
+            .cloned()
+            .ok_or_else(|| column_type_error::<Self>(col))
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
         if domain.is_undefined() {
-            Some(())
+            Ok(())
         } else {
-            None
+            Err(domain_type_error::<Self>(domain))
         }
     }
 
diff --git a/src/query/expression/src/types/vector.rs b/src/query/expression/src/types/vector.rs
index bba986d4597e7..8db61c286cd38 100644
--- a/src/query/expression/src/types/vector.rs
+++ b/src/query/expression/src/types/vector.rs
@@ -19,18 +19,22 @@ use std::ops::Range;
 use borsh::BorshDeserialize;
 use borsh::BorshSerialize;
 use databend_common_column::buffer::Buffer;
+use databend_common_exception::Result;
 use enum_as_inner::EnumAsInner;
 use serde::Deserialize;
 use serde::Serialize;
 
+use super::column_type_error;
+use super::domain_type_error;
+use super::scalar_type_error;
+use super::AccessType;
+use super::BuilderMut;
+use super::DataType;
+use super::NumberColumn;
+use super::NumberDataType;
+use super::ValueType;
+use super::F32;
 use crate::property::Domain;
-use crate::types::AccessType;
-use crate::types::BuilderMut;
-use crate::types::DataType;
-use crate::types::NumberColumn;
-use crate::types::NumberDataType;
-use crate::types::ValueType;
-use crate::types::F32;
 use crate::values::Column;
 use crate::values::Scalar;
 use crate::ColumnBuilder;
@@ -55,19 +59,24 @@ impl AccessType for VectorType {
         scalar.as_ref()
     }
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        scalar.as_vector().cloned()
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        scalar
+            .as_vector()
+            .cloned()
+            .ok_or_else(|| scalar_type_error::<Self>(scalar))
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        col.as_vector().cloned()
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        col.as_vector()
+            .cloned()
+            .ok_or_else(|| column_type_error::<Self>(col))
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<Self::Domain> {
+    fn try_downcast_domain(domain: &Domain) -> Result<Self::Domain> {
         if domain.is_undefined() {
-            Some(())
+            Ok(())
         } else {
-            None
+            Err(domain_type_error::<Self>(domain))
         }
     }
 
@@ -197,7 +206,7 @@ macro_rules! with_vector_number_mapped_type {
         match_template::match_template! {
             $t = [
                 Int8 => i8,
-                Float32 => $crate::types::number::F32,
+                Float32 => $super::number::F32,
             ],
             $($tail)*
         }
diff --git a/src/query/expression/src/types/zero_size_type.rs b/src/query/expression/src/types/zero_size_type.rs
index 33642372f34b1..5e1ce7f653083 100644
--- a/src/query/expression/src/types/zero_size_type.rs
+++ b/src/query/expression/src/types/zero_size_type.rs
@@ -17,13 +17,18 @@ use std::fmt::Debug;
 use std::marker::PhantomData;
 use std::ops::Range;
 
+use databend_common_exception::Result;
+
+use super::column_type_error;
+use super::domain_type_error;
+use super::scalar_type_error;
 use super::AccessType;
+use super::BuilderMut;
+use super::DataType;
 use super::GenericMap;
 use super::ReturnType;
 use super::Scalar;
 use super::ValueType;
-use crate::types::BuilderMut;
-use crate::types::DataType;
 use crate::Column;
 use crate::ColumnBuilder;
 use crate::Domain;
@@ -57,16 +62,16 @@ impl<T: ZeroSizeType> AccessType for ZeroSizeValueType<T> {
     fn to_owned_scalar(_: Self::ScalarRef<'_>) -> Self::Scalar {}
     fn to_scalar_ref(_: &Self::Scalar) -> Self::ScalarRef<'_> {}
 
-    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Option<Self::ScalarRef<'a>> {
-        T::downcast_scalar(scalar)
+    fn try_downcast_scalar<'a>(scalar: &ScalarRef<'a>) -> Result<Self::ScalarRef<'a>> {
+        T::downcast_scalar(scalar).ok_or_else(|| scalar_type_error::<Self>(scalar))
     }
 
-    fn try_downcast_column(col: &Column) -> Option<Self::Column> {
-        T::downcast_column(col)
+    fn try_downcast_column(col: &Column) -> Result<Self::Column> {
+        T::downcast_column(col).ok_or_else(|| column_type_error::<Self>(col))
     }
 
-    fn try_downcast_domain(domain: &Domain) -> Option<()> {
-        T::downcast_domain(domain)
+    fn try_downcast_domain(domain: &Domain) -> Result<()> {
+        T::downcast_domain(domain).ok_or_else(|| domain_type_error::<Self>(domain))
     }
 
     fn column_len(len: &usize) -> usize {
diff --git a/src/query/expression/src/values.rs b/src/query/expression/src/values.rs
index 3e7b194da9b9e..ca3004ba8e514 100755
--- a/src/query/expression/src/values.rs
+++ b/src/query/expression/src/values.rs
@@ -380,13 +380,13 @@ impl Value<AnyType> {
         }
     }
 
-    pub fn try_downcast<T: AccessType>(&self) -> Option<Value<T>> {
-        Some(match self {
-            Value::Scalar(scalar) => Value::Scalar(T::to_owned_scalar(T::try_downcast_scalar(
+    pub fn try_downcast<T: AccessType>(&self) -> Result<Value<T>> {
+        match self {
+            Value::Scalar(scalar) => Ok(Value::Scalar(T::to_owned_scalar(T::try_downcast_scalar(
                 &scalar.as_ref(),
-            )?)),
-            Value::Column(col) => Value::Column(T::try_downcast_column(col)?),
-        })
+            )?))),
+            Value::Column(col) => Ok(Value::Column(T::try_downcast_column(col)?)),
+        }
     }
 
     pub fn wrap_nullable(self, validity: Option<Bitmap>) -> Self {
diff --git a/src/query/functions/src/aggregates/aggregate_bitmap.rs b/src/query/functions/src/aggregates/aggregate_bitmap.rs
index ba4e2914bf345..9486e6d06e5e4 100644
--- a/src/query/functions/src/aggregates/aggregate_bitmap.rs
+++ b/src/query/functions/src/aggregates/aggregate_bitmap.rs
@@ -672,10 +672,9 @@ fn extract_params<T: AccessType>(
 ) -> Result<Vec<T::Scalar>> {
     let mut filter_values = Vec::with_capacity(params.len());
     for (i, param) in params.iter().enumerate() {
-        let val_opt = T::try_downcast_scalar(&param.as_ref()).map(|s| T::to_owned_scalar(s));
-        match val_opt {
-            Some(val) => filter_values.push(val),
-            None => {
+        match T::try_downcast_scalar(&param.as_ref()) {
+            Ok(scalar) => filter_values.push(T::to_owned_scalar(scalar)),
+            Err(_) => {
                 return Err(ErrorCode::BadDataValueType(format!(
                     "{} param({}) type mismatch, expect: '{:?}', but got: '{:?}'",
                     display_name,
diff --git a/src/query/functions/src/aggregates/aggregate_st_collect.rs b/src/query/functions/src/aggregates/aggregate_st_collect.rs
index ab5e68a830a15..7ed5f0e95f5fb 100644
--- a/src/query/functions/src/aggregates/aggregate_st_collect.rs
+++ b/src/query/functions/src/aggregates/aggregate_st_collect.rs
@@ -259,16 +259,14 @@ where
     ) -> Result<()> {
         let state = place.get::<State>();
         match &columns[0].to_column() {
+            Column::Null { .. } => Ok(()),
             Column::Nullable(box nullable_column) => {
                 let column = T::try_downcast_column(&nullable_column.column).unwrap();
                 state.add_batch(&column, Some(&nullable_column.validity))
             }
             _ => {
-                if let Some(column) = T::try_downcast_column(&columns[0].to_column()) {
-                    state.add_batch(&column, None)
-                } else {
-                    Ok(())
-                }
+                let column = T::try_downcast_column(&columns[0].to_column()).unwrap();
+                state.add_batch(&column, None)
             }
         }
     }
@@ -281,6 +279,7 @@ where
         _input_rows: usize,
     ) -> Result<()> {
         match &columns[0].to_column() {
+            Column::Null { .. } => (),
             Column::Nullable(box nullable_column) => {
                 let column = T::try_downcast_column(&nullable_column.column).unwrap();
                 let column_iter = T::iter_column(&column);
@@ -296,13 +295,12 @@ where
                     });
             }
             _ => {
-                if let Some(column) = T::try_downcast_column(&columns[0].to_column()) {
-                    let column_iter = T::iter_column(&column);
-                    column_iter.zip(places.iter()).for_each(|(v, place)| {
-                        let state = AggrState::new(*place, loc).get::<State>();
-                        state.add(Some(v.clone()))
-                    });
-                }
+                let column = T::try_downcast_column(&columns[0].to_column()).unwrap();
+                let column_iter = T::iter_column(&column);
+                column_iter.zip(places.iter()).for_each(|(v, place)| {
+                    let state = AggrState::new(*place, loc).get::<State>();
+                    state.add(Some(v.clone()))
+                });
             }
         }
 
@@ -312,6 +310,7 @@ where
     fn accumulate_row(&self, place: AggrState, columns: ProjectedBlock, row: usize) -> Result<()> {
         let state = place.get::<State>();
         match &columns[0].to_column() {
+            Column::Null { .. } => (),
             Column::Nullable(box nullable_column) => {
                 let valid = nullable_column.validity.get_bit(row);
                 if valid {
@@ -323,10 +322,9 @@ where
                 }
             }
             _ => {
-                if let Some(column) = T::try_downcast_column(&columns[0].to_column()) {
-                    let v = T::index_column(&column, row);
-                    state.add(v);
-                }
+                let column = T::try_downcast_column(&columns[0].to_column()).unwrap();
+                let v = T::index_column(&column, row);
+                state.add(v);
             }
         }
 
diff --git a/src/query/functions/src/scalars/comparison.rs b/src/query/functions/src/scalars/comparison.rs
index 389ccdb7b784a..a23c6d8a28981 100644
--- a/src/query/functions/src/scalars/comparison.rs
+++ b/src/query/functions/src/scalars/comparison.rs
@@ -919,13 +919,17 @@ fn register_like(registry: &mut FunctionRegistry) {
                             return Value::Scalar(Scalar::Boolean(Default::default()));
                         }
                     };
-                    let escape: Value<StringType> = args.get(2).cloned().and_then(|value| value.try_downcast()).unwrap_or(Value::Scalar("".to_string()));
+                    let escape: Value<StringType> = args
+                        .get(2)
+                        .cloned()
+                        .and_then(|value| value.try_downcast().ok())
+                        .unwrap_or(Value::Scalar("".to_string()));
 
-                    let result = if let Some(value) = arg.try_downcast::<StringType>() {
+                    let result = if let Ok(value) = arg.try_downcast::<StringType>() {
                         let like = vectorize_like(|str, pattern_type| pattern_type.compare(str));
                         patterns.iter().map(|pattern| like(value.clone(), Value::<StringType>::Scalar(pattern.as_string().unwrap().clone()), escape.clone(), ctx))
                             .collect::<Vec<_>>()
-                    } else if let Some(value) = arg.try_downcast::<VariantType>() {
+                    } else if let Ok(value) = arg.try_downcast::<VariantType>() {
                         let like = variant_vectorize_like_jsonb();
                         patterns.iter().map(|pattern| like(value.clone(), Value::<StringType>::Scalar(pattern.as_string().unwrap().clone()), escape.clone(), ctx))
                             .collect::<Vec<_>>()
diff --git a/src/query/functions/src/scalars/decimal/src/cast.rs b/src/query/functions/src/scalars/decimal/src/cast.rs
index 57e78e94192f0..7bdaac6aac8d7 100644
--- a/src/query/functions/src/scalars/decimal/src/cast.rs
+++ b/src/query/functions/src/scalars/decimal/src/cast.rs
@@ -416,15 +416,15 @@ pub fn register_decimal_to_string(registry: &mut FunctionRegistry) {
             eval: FunctionEval::Scalar {
                 calc_domain: Box::new(|_, _| FunctionDomain::Full),
                 eval: Box::new(move |args, ctx| {
-                    if let Some(arg) = args[0].try_downcast::<Decimal64Type>() {
+                    if let Ok(arg) = args[0].try_downcast::<Decimal64Type>() {
                         let arg_type = DecimalDataType::Decimal64(size);
                         return decimal_to_string(arg, arg_type, ctx).upcast();
                     };
-                    if let Some(arg) = args[0].try_downcast::<Decimal128Type>() {
+                    if let Ok(arg) = args[0].try_downcast::<Decimal128Type>() {
                         let arg_type = DecimalDataType::Decimal128(size);
                         return decimal_to_string(arg, arg_type, ctx).upcast();
                     };
-                    if let Some(arg) = args[0].try_downcast::<Decimal256Type>() {
+                    if let Ok(arg) = args[0].try_downcast::<Decimal256Type>() {
                         let arg_type = DecimalDataType::Decimal256(size);
                         return decimal_to_string(arg, arg_type, ctx).upcast();
                     };
diff --git a/src/query/functions/src/scalars/obfuscator.rs b/src/query/functions/src/scalars/obfuscator.rs
index fdb193738f5fd..0dac03718b50a 100644
--- a/src/query/functions/src/scalars/obfuscator.rs
+++ b/src/query/functions/src/scalars/obfuscator.rs
@@ -93,7 +93,7 @@ impl ColumnTable {
             return None;
         }
         let buckets = tuple.pop()?;
-        let buckets = MapType::<UInt32Type, UInt32Type>::try_downcast_column(&buckets)?;
+        let buckets = MapType::<UInt32Type, UInt32Type>::try_downcast_column(&buckets).ok()?;
         let Some(Column::Number(NumberColumn::UInt32(count_end))) = tuple.pop() else {
             return None;
         };
diff --git a/src/query/pipeline/transforms/src/processors/transforms/sorts/core/row_convert/fixed.rs b/src/query/pipeline/transforms/src/processors/transforms/sorts/core/row_convert/fixed.rs
index d483e7073f91d..fd1f35ca899e5 100644
--- a/src/query/pipeline/transforms/src/processors/transforms/sorts/core/row_convert/fixed.rs
+++ b/src/query/pipeline/transforms/src/processors/transforms/sorts/core/row_convert/fixed.rs
@@ -101,8 +101,11 @@ impl<const N: usize> Rows for FixedRows<N> {
         OpaqueType::upcast_column(self.0.clone())
     }
 
-    fn try_from_column(col: &Column) -> Option<Self> {
-        OpaqueType::<N>::try_downcast_column(col).map(FixedRows)
+    fn from_column(col: &Column) -> Result<Self> {
+        match OpaqueType::<N>::try_downcast_column(col) {
+            Ok(col) => Ok(FixedRows(col)),
+            Err(err) => Err(err.add_message("Order column type mismatched.")),
+        }
     }
 
     fn slice(&self, range: Range<usize>) -> Self {
diff --git a/src/query/pipeline/transforms/src/processors/transforms/sorts/core/row_convert/simple.rs b/src/query/pipeline/transforms/src/processors/transforms/sorts/core/row_convert/simple.rs
index 99bc1b8d2a0cc..b76f5ad14034f 100644
--- a/src/query/pipeline/transforms/src/processors/transforms/sorts/core/row_convert/simple.rs
+++ b/src/query/pipeline/transforms/src/processors/transforms/sorts/core/row_convert/simple.rs
@@ -60,9 +60,11 @@ where
         T::upcast_column_with_type(self.inner.clone(), &T::data_type())
     }
 
-    fn try_from_column(col: &Column) -> Option<Self> {
-        let inner = T::try_downcast_column(col)?;
-        Some(Self { inner })
+    fn from_column(col: &Column) -> Result<Self> {
+        match T::try_downcast_column(col) {
+            Ok(inner) => Ok(Self { inner }),
+            Err(err) => Err(err.add_message("Order column type mismatched.")),
+        }
     }
 
     fn slice(&self, range: Range<usize>) -> Self {
@@ -112,9 +114,11 @@ where
         T::upcast_column_with_type(self.inner.clone(), &T::data_type())
     }
 
-    fn try_from_column(col: &Column) -> Option<Self> {
-        let inner = T::try_downcast_column(col)?;
-        Some(Self { inner })
+    fn from_column(col: &Column) -> Result<Self> {
+        match T::try_downcast_column(col) {
+            Ok(inner) => Ok(Self { inner }),
+            Err(err) => Err(err.add_message("Order column type mismatched.")),
+        }
     }
 
     fn slice(&self, range: Range<usize>) -> Self {
diff --git a/src/query/pipeline/transforms/src/processors/transforms/sorts/core/row_convert/variable.rs b/src/query/pipeline/transforms/src/processors/transforms/sorts/core/row_convert/variable.rs
index a2c7f2e32971b..ca5e044531140 100644
--- a/src/query/pipeline/transforms/src/processors/transforms/sorts/core/row_convert/variable.rs
+++ b/src/query/pipeline/transforms/src/processors/transforms/sorts/core/row_convert/variable.rs
@@ -78,8 +78,15 @@ impl Rows for VariableRows {
         Column::Binary(self.clone())
     }
 
-    fn try_from_column(col: &Column) -> Option<Self> {
-        col.as_binary().cloned()
+    fn from_column(col: &Column) -> Result<Self> {
+        match col.as_binary().cloned() {
+            Some(col) => Ok(col),
+            None => Err(ErrorCode::BadDataValueType(format!(
+                "Order column type mismatched. Expected {} but got {}",
+                Self::data_type(),
+                col.data_type()
+            ))),
+        }
     }
 
     fn slice(&self, range: Range<usize>) -> Self {
diff --git a/src/query/pipeline/transforms/src/processors/transforms/sorts/core/rows.rs b/src/query/pipeline/transforms/src/processors/transforms/sorts/core/rows.rs
index a444440eedd98..25d324a2a3cef 100644
--- a/src/query/pipeline/transforms/src/processors/transforms/sorts/core/rows.rs
+++ b/src/query/pipeline/transforms/src/processors/transforms/sorts/core/rows.rs
@@ -15,7 +15,6 @@
 use std::fmt::Debug;
 use std::ops::Range;
 
-use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::types::ArgType;
 use databend_common_expression::types::DataType;
@@ -35,17 +34,7 @@ where Self: Sized + Clone + Debug + Send
     fn row(&self, index: usize) -> Self::Item<'_>;
     fn to_column(&self) -> Column;
 
-    fn from_column(col: &Column) -> Result<Self> {
-        Self::try_from_column(col).ok_or_else(|| {
-            ErrorCode::BadDataValueType(format!(
-                "Order column type mismatched. Expected {} but got {}",
-                Self::data_type(),
-                col.data_type()
-            ))
-        })
-    }
-
-    fn try_from_column(col: &Column) -> Option<Self>;
+    fn from_column(col: &Column) -> Result<Self>;
 
     fn data_type() -> DataType {
         Self::Type::data_type()
diff --git a/src/query/service/src/pipelines/processors/transforms/transform_dictionary.rs b/src/query/service/src/pipelines/processors/transforms/transform_dictionary.rs
index 78d911201cbce..2a7bce3d5876c 100644
--- a/src/query/service/src/pipelines/processors/transforms/transform_dictionary.rs
+++ b/src/query/service/src/pipelines/processors/transforms/transform_dictionary.rs
@@ -180,8 +180,8 @@ impl DictionaryOperator {
                 Value::Column(column) => {
                     let (_, validity) = column.validity();
                     let column = match StringType::try_downcast_column(&column.remove_nullable()) {
-                        Some(col) => col,
-                        None => {
+                        Ok(col) => col,
+                        Err(_) => {
                             return Err(ErrorCode::DictionarySourceError(format!(
                                 "Redis dictionary operator currently does not support column type {}",
                                 column.data_type(),
diff --git a/src/query/service/src/servers/flight/v1/scatter/flight_scatter_hash.rs b/src/query/service/src/servers/flight/v1/scatter/flight_scatter_hash.rs
index e6c03d09abd29..35e6656bd94b9 100644
--- a/src/query/service/src/servers/flight/v1/scatter/flight_scatter_hash.rs
+++ b/src/query/service/src/servers/flight/v1/scatter/flight_scatter_hash.rs
@@ -246,45 +246,43 @@ fn get_hash_values(
             _ => unreachable!(),
         },
         Value::Column(c) => {
-            if let Some(column) = NumberType::<u64>::try_downcast_column(&c) {
-                Ok(column)
-            } else if let Some(mut column) =
-                NullableType::<NumberType<u64>>::try_downcast_column(&c)
-            {
-                let null_map = column.validity;
-                if null_map.null_count() == 0 {
-                    Ok(column.column)
-                } else if null_map.null_count() == null_map.len() {
-                    Ok(vec![default_scatter_index; rows].into())
-                } else {
-                    let mut need_new_vec = true;
-                    if let Some(column) = unsafe { column.column.get_mut() } {
-                        column
-                            .iter_mut()
-                            .zip(null_map.iter())
-                            .for_each(|(x, valid)| {
-                                if valid {
-                                    *x *= valid as u64;
-                                } else {
-                                    *x = default_scatter_index;
-                                }
-                            });
-                        need_new_vec = false;
-                    }
+            if let Ok(column) = NumberType::<u64>::try_downcast_column(&c) {
+                return Ok(column);
+            }
 
-                    if !need_new_vec {
-                        Ok(column.column)
-                    } else {
-                        Ok(column
-                            .column
-                            .iter()
-                            .zip(null_map.iter())
-                            .map(|(x, b)| if b { *x } else { default_scatter_index })
-                            .collect())
-                    }
-                }
+            let mut column = NullableType::<NumberType<u64>>::try_downcast_column(&c).unwrap();
+            let null_map = column.validity;
+            if null_map.null_count() == 0 {
+                return Ok(column.column);
+            }
+            if null_map.null_count() == null_map.len() {
+                return Ok(vec![default_scatter_index; rows].into());
+            }
+
+            let mut need_new_vec = true;
+            if let Some(column) = unsafe { column.column.get_mut() } {
+                column
+                    .iter_mut()
+                    .zip(null_map.iter())
+                    .for_each(|(x, valid)| {
+                        if valid {
+                            *x *= valid as u64;
+                        } else {
+                            *x = default_scatter_index;
+                        }
+                    });
+                need_new_vec = false;
+            }
+
+            if !need_new_vec {
+                Ok(column.column)
             } else {
-                unreachable!()
+                Ok(column
+                    .column
+                    .iter()
+                    .zip(null_map.iter())
+                    .map(|(x, b)| if b { *x } else { default_scatter_index })
+                    .collect())
             }
         }
     }