diff --git a/Cargo.lock b/Cargo.lock
index 2036e15085..b66d9254c6 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4949,6 +4949,7 @@ dependencies = [
  "common-time",
  "datafusion",
  "datafusion-common",
+ "datafusion-expr",
  "datafusion-physical-expr",
  "datatypes",
  "format_num",
diff --git a/src/query/Cargo.toml b/src/query/Cargo.toml
index 34c3977d90..6d92c2e9d8 100644
--- a/src/query/Cargo.toml
+++ b/src/query/Cargo.toml
@@ -17,6 +17,7 @@ common-telemetry = { path = "../common/telemetry" }
 common-time = { path = "../common/time" }
 datafusion = "14.0.0"
 datafusion-common = "14.0.0"
+datafusion-expr = "14.0.0"
 datafusion-physical-expr = "14.0.0"
 datatypes = { path = "../datatypes" }
 futures = "0.3"
diff --git a/src/query/src/optimizer.rs b/src/query/src/optimizer.rs
index cee0c2727a..c50046cd51 100644
--- a/src/query/src/optimizer.rs
+++ b/src/query/src/optimizer.rs
@@ -17,15 +17,12 @@ use std::sync::Arc;
 
 use common_time::timestamp::{TimeUnit, Timestamp};
 use datafusion::execution::context::ExecutionProps;
-use datafusion::logical_plan::plan::Filter;
-use datafusion::logical_plan::{
-    Expr, ExprRewritable, ExprRewriter, ExprSchemable, LogicalPlan, Operator, TableScan,
-};
+use datafusion_expr::{Expr, ExprSchemable, LogicalPlan, Operator, TableScan};
+use datafusion_expr::expr_rewriter::{ExprRewriter, ExprRewritable};
 use datafusion::optimizer::optimizer::OptimizerRule;
 use datafusion::optimizer::utils;
 use datafusion_common::{DFSchemaRef, DataFusionError, Result, ScalarValue};
 use datatypes::arrow::compute;
-use datatypes::arrow::compute::cast::CastOptions;
 use datatypes::arrow::datatypes::DataType;
 
 /// TypeConversionRule converts some literal values in logical plan to other types according
@@ -139,7 +136,7 @@ impl<'a> TypeConverter<'a> {
             (target_type, value) => {
                 let value_arr = value.to_array();
                 let arr =
-                    compute::cast::cast(value_arr.as_ref(), target_type, CastOptions::default())
+                    compute::cast(&value_arr, target_type)
                         .map_err(DataFusionError::ArrowError)?;
 
                 ScalarValue::try_from_array(
diff --git a/src/query/src/plan.rs b/src/query/src/plan.rs
index c7e337c0e9..5182db4f6a 100644
--- a/src/query/src/plan.rs
+++ b/src/query/src/plan.rs
@@ -14,7 +14,7 @@
 
 use std::fmt::Debug;
 
-use datafusion::logical_plan::LogicalPlan as DfLogicalPlan;
+use datafusion_expr::LogicalPlan as DfLogicalPlan;
 
 /// A LogicalPlan represents the different types of relational
 /// operators (such as Projection, Filter, etc) and can be created by
diff --git a/src/query/src/query_engine/state.rs b/src/query/src/query_engine/state.rs
index 36bd331b36..b638ead920 100644
--- a/src/query/src/query_engine/state.rs
+++ b/src/query/src/query_engine/state.rs
@@ -19,7 +19,7 @@ use std::sync::{Arc, RwLock};
 use catalog::CatalogListRef;
 use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_function::scalars::aggregate::AggregateFunctionMetaRef;
-use common_query::physical_plan::RuntimeEnv;
+use common_query::physical_plan::{SessionContext, TaskContext};
 use common_query::prelude::ScalarUdf;
 use datafusion::optimizer::common_subexpr_eliminate::CommonSubexprEliminate;
 use datafusion::optimizer::eliminate_limit::EliminateLimit;
@@ -28,8 +28,7 @@ use datafusion::optimizer::limit_push_down::LimitPushDown;
 use datafusion::optimizer::projection_push_down::ProjectionPushDown;
 use datafusion::optimizer::single_distinct_to_groupby::SingleDistinctToGroupBy;
 use datafusion::optimizer::to_approx_perc::ToApproxPerc;
-use datafusion::prelude::{ExecutionConfig, ExecutionContext};
-
+use datafusion::execution::context::SessionConfig;
 use crate::datafusion::DfCatalogListAdapter;
 use crate::optimizer::TypeConversionRule;