diff --git a/crates/core/src/delta_datafusion/cdf/scan.rs b/crates/core/src/delta_datafusion/cdf/scan.rs
index ea34855b77..bd7488899f 100644
--- a/crates/core/src/delta_datafusion/cdf/scan.rs
+++ b/crates/core/src/delta_datafusion/cdf/scan.rs
@@ -26,6 +26,10 @@ impl DisplayAs for DeltaCdfScan {
 }
 
 impl ExecutionPlan for DeltaCdfScan {
+    fn name(&self) -> &str {
+        Self::static_name()
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }
diff --git a/crates/core/src/delta_datafusion/find_files/mod.rs b/crates/core/src/delta_datafusion/find_files/mod.rs
index 347925f31f..d25d0765ee 100644
--- a/crates/core/src/delta_datafusion/find_files/mod.rs
+++ b/crates/core/src/delta_datafusion/find_files/mod.rs
@@ -141,7 +141,7 @@ async fn scan_table_by_files(
 
     // Identify which columns we need to project
     let mut used_columns = expression
-        .to_columns()?
+        .column_refs()
         .into_iter()
         .map(|column| logical_schema.index_of(&column.name))
         .collect::<std::result::Result<Vec<usize>, ArrowError>>()?;
diff --git a/crates/core/src/delta_datafusion/find_files/physical.rs b/crates/core/src/delta_datafusion/find_files/physical.rs
index eb295912a2..e23a561e5b 100644
--- a/crates/core/src/delta_datafusion/find_files/physical.rs
+++ b/crates/core/src/delta_datafusion/find_files/physical.rs
@@ -85,6 +85,10 @@ impl DisplayAs for FindFilesExec {
 }
 
 impl ExecutionPlan for FindFilesExec {
+    fn name(&self) -> &str {
+        Self::static_name()
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }
diff --git a/crates/core/src/delta_datafusion/mod.rs b/crates/core/src/delta_datafusion/mod.rs
index 860a02be56..fa0c8b97d9 100644
--- a/crates/core/src/delta_datafusion/mod.rs
+++ b/crates/core/src/delta_datafusion/mod.rs
@@ -60,12 +60,11 @@ use datafusion_common::scalar::ScalarValue;
 use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion, TreeNodeVisitor};
 use datafusion_common::{
     config::ConfigOptions, Column, DFSchema, DataFusionError, Result as DataFusionResult,
-    ToDFSchema,
+    TableReference, ToDFSchema,
 };
 use datafusion_expr::logical_plan::CreateExternalTable;
 use datafusion_expr::utils::conjunction;
 use datafusion_expr::{col, Expr, Extension, LogicalPlan, TableProviderFilterPushDown, Volatility};
-use datafusion_physical_expr::PhysicalExpr;
 use datafusion_proto::logical_plan::LogicalExtensionCodec;
 use datafusion_proto::physical_plan::PhysicalExtensionCodec;
 use datafusion_sql::planner::ParserOptions;
@@ -846,6 +845,10 @@ impl DisplayAs for DeltaScan {
 }
 
 impl ExecutionPlan for DeltaScan {
+    fn name(&self) -> &str {
+        Self::static_name()
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }
@@ -1317,6 +1320,7 @@ impl LogicalExtensionCodec for DeltaLogicalCodec {
     fn try_decode_table_provider(
         &self,
         buf: &[u8],
+        _table_ref: &TableReference,
         _schema: SchemaRef,
         _ctx: &SessionContext,
     ) -> Result<Arc<dyn TableProvider>, DataFusionError> {
@@ -1327,6 +1331,7 @@ impl LogicalExtensionCodec for DeltaLogicalCodec {
 
     fn try_encode_table_provider(
         &self,
+        _table_ref: &TableReference,
         node: Arc<dyn TableProvider>,
         buf: &mut Vec<u8>,
     ) -> Result<(), DataFusionError> {
@@ -1507,7 +1512,7 @@ pub(crate) async fn find_files_scan<'a>(
 
     // Identify which columns we need to project
     let mut used_columns = expression
-        .to_columns()?
+        .column_refs()
         .into_iter()
         .map(|column| logical_schema.index_of(&column.name))
         .collect::<Result<Vec<usize>, ArrowError>>()?;
diff --git a/crates/core/src/delta_datafusion/physical.rs b/crates/core/src/delta_datafusion/physical.rs
index bfc220cf86..adbb7fb4fe 100644
--- a/crates/core/src/delta_datafusion/physical.rs
+++ b/crates/core/src/delta_datafusion/physical.rs
@@ -74,6 +74,10 @@ impl DisplayAs for MetricObserverExec {
 }
 
 impl ExecutionPlan for MetricObserverExec {
+    fn name(&self) -> &str {
+        Self::static_name()
+    }
+
     fn as_any(&self) -> &dyn std::any::Any {
         self
     }
diff --git a/crates/core/src/operations/cdc.rs b/crates/core/src/operations/cdc.rs
index cc8bff2359..5c4b3be0f5 100644
--- a/crates/core/src/operations/cdc.rs
+++ b/crates/core/src/operations/cdc.rs
@@ -196,6 +196,10 @@ impl DisplayAs for CDCObserver {
 }
 
 impl ExecutionPlan for CDCObserver {
+    fn name(&self) -> &str {
+        Self::static_name()
+    }
+
     fn as_any(&self) -> &dyn std::any::Any {
         self
     }
diff --git a/crates/core/src/operations/merge/barrier.rs b/crates/core/src/operations/merge/barrier.rs
index 04cde87a19..e9b2f8fd00 100644
--- a/crates/core/src/operations/merge/barrier.rs
+++ b/crates/core/src/operations/merge/barrier.rs
@@ -67,6 +67,10 @@ impl MergeBarrierExec {
 }
 
 impl ExecutionPlan for MergeBarrierExec {
+    fn name(&self) -> &str {
+        Self::static_name()
+    }
+
     fn as_any(&self) -> &dyn std::any::Any {
         self
     }
diff --git a/crates/sql/src/planner.rs b/crates/sql/src/planner.rs
index e2c76e68fd..90b7827575 100644
--- a/crates/sql/src/planner.rs
+++ b/crates/sql/src/planner.rs
@@ -42,6 +42,7 @@ impl<'a, S: ContextProvider> DeltaSqlToRel<'a, S> {
                     ParserOptions {
                         parse_float_as_decimal: self.options.parse_float_as_decimal,
                         enable_ident_normalization: self.options.enable_ident_normalization,
+                        support_varchar_with_length: false,
                     },
                 );
                 planner.statement_to_plan(s)