diff --git a/datafusion/common/src/column.rs b/datafusion/common/src/column.rs
index c7f0b5a4f4881..71253891f7d3e 100644
--- a/datafusion/common/src/column.rs
+++ b/datafusion/common/src/column.rs
@@ -327,26 +327,6 @@ impl Column {
     }
 }
 
-impl From<&str> for Column {
-    fn from(c: &str) -> Self {
-        Self::from_qualified_name(c)
-    }
-}
-
-/// Create a column, cloning the string
-impl From<&String> for Column {
-    fn from(c: &String) -> Self {
-        Self::from_qualified_name(c)
-    }
-}
-
-/// Create a column, reusing the existing string
-impl From<String> for Column {
-    fn from(c: String) -> Self {
-        Self::from_qualified_name(c)
-    }
-}
-
 /// Create a column, use qualifier and field name
 impl From<(Option<&TableReference>, &Field)> for Column {
     fn from((relation, field): (Option<&TableReference>, &Field)) -> Self {
@@ -366,7 +346,7 @@ impl std::str::FromStr for Column {
     type Err = std::convert::Infallible;
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
-        Ok(s.into())
+        Ok(Self::from_qualified_name(s))
     }
 }
 
diff --git a/datafusion/core/src/dataframe/mod.rs b/datafusion/core/src/dataframe/mod.rs
index fe760760eef3f..c5d27af4319ea 100644
--- a/datafusion/core/src/dataframe/mod.rs
+++ b/datafusion/core/src/dataframe/mod.rs
@@ -514,7 +514,10 @@ impl DataFrame {
         columns: &[&str],
         options: UnnestOptions,
     ) -> Result<DataFrame> {
-        let columns = columns.iter().map(|c| Column::from(*c)).collect();
+        let columns = columns
+            .iter()
+            .map(|c| Column::from_qualified_name(*c))
+            .collect();
         let plan = LogicalPlanBuilder::from(self.plan)
             .unnest_columns_with_options(columns, options)?
             .build()?;
@@ -1257,7 +1260,16 @@ impl DataFrame {
             .join(
                 right.plan,
                 join_type,
-                (left_cols.to_vec(), right_cols.to_vec()),
+                (
+                    left_cols
+                        .iter()
+                        .map(|c| Column::from_qualified_name(*c))
+                        .collect(),
+                    right_cols
+                        .iter()
+                        .map(|c| Column::from_qualified_name(*c))
+                        .collect(),
+                ),
                 filter,
             )?
             .build()?;
@@ -2167,7 +2179,7 @@ impl DataFrame {
                     col_exists = true;
                     Some((new_column.clone(), true))
                 } else {
-                    let e = col(Column::from((qualifier, field)));
+                    let e = Expr::Column(Column::from((qualifier, field)));
                     Some((e, self.projection_requires_validation))
                 }
             })
@@ -2245,12 +2257,12 @@ impl DataFrame {
             .map(|(qualifier, field)| {
                 if qualifier.eq(&qualifier_rename) && field == field_rename {
                     (
-                        col(Column::from((qualifier, field)))
+                        Expr::Column(Column::from((qualifier, field)))
                             .alias_qualified(qualifier.cloned(), new_name),
                         false,
                     )
                 } else {
-                    (col(Column::from((qualifier, field))), false)
+                    (Expr::Column(Column::from((qualifier, field))), false)
                 }
             })
             .collect::<Vec<_>>();
diff --git a/datafusion/core/src/datasource/listing/table.rs b/datafusion/core/src/datasource/listing/table.rs
index 4e33f3cad51a4..32516cf5a7823 100644
--- a/datafusion/core/src/datasource/listing/table.rs
+++ b/datafusion/core/src/datasource/listing/table.rs
@@ -129,7 +129,7 @@ mod tests {
         ListingOptions, ListingTable, ListingTableConfig, SchemaSource,
     };
     use datafusion_common::{
-        DataFusionError, Result, ScalarValue, assert_contains,
+        Column, DataFusionError, Result, ScalarValue, assert_contains,
         stats::Precision,
         test_util::{batches_to_string, datafusion_test_data},
     };
@@ -776,7 +776,7 @@ mod tests {
         )]));
 
         let filter_predicate = Expr::BinaryExpr(BinaryExpr::new(
-            Box::new(Expr::Column("column1".into())),
+            Box::new(Expr::Column(Column::from_qualified_name("column1"))),
             Operator::GtEq,
             Box::new(Expr::Literal(ScalarValue::Int32(Some(0)), None)),
         ));
diff --git a/datafusion/expr/src/expr.rs b/datafusion/expr/src/expr.rs
index 8234247c92db6..14a69d2653a67 100644
--- a/datafusion/expr/src/expr.rs
+++ b/datafusion/expr/src/expr.rs
@@ -1852,7 +1852,7 @@ impl Expr {
     /// # use datafusion_common::Column;
     /// use datafusion_expr::{col, Expr};
     /// let expr = col("foo");
-    /// assert_eq!(expr.try_as_col(), Some(&Column::from("foo")));
+    /// assert_eq!(expr.try_as_col(), Some(&Column::from_qualified_name("foo")));
     ///
     /// let expr = col("foo").alias("bar");
     /// assert_eq!(expr.try_as_col(), None);
diff --git a/datafusion/expr/src/expr_fn.rs b/datafusion/expr/src/expr_fn.rs
index 4254602d7c555..13d624b783e34 100644
--- a/datafusion/expr/src/expr_fn.rs
+++ b/datafusion/expr/src/expr_fn.rs
@@ -65,15 +65,15 @@ use std::sync::Arc;
 /// let c3 = col(r#""A""#);
 /// assert_ne!(c1, c3);
 /// ```
-pub fn col(ident: impl Into<Column>) -> Expr {
-    Expr::Column(ident.into())
+pub fn col(ident: impl Into<String>) -> Expr {
+    Expr::Column(Column::from_qualified_name(ident))
 }
 
 /// Create an out reference column which hold a reference that has been resolved to a field
 /// outside of the current plan.
 /// The expression created by this function does not preserve the metadata of the outer column.
 /// Please use `out_ref_col_with_metadata` if you want to preserve the metadata.
-pub fn out_ref_col(dt: DataType, ident: impl Into<Column>) -> Expr {
+pub fn out_ref_col(dt: DataType, ident: impl Into<String>) -> Expr {
     out_ref_col_with_metadata(dt, HashMap::new(), ident)
 }
 
@@ -81,9 +81,9 @@ pub fn out_ref_col(dt: DataType, ident: impl Into<Column>) -> Expr {
 pub fn out_ref_col_with_metadata(
     dt: DataType,
     metadata: HashMap<String, String>,
-    ident: impl Into<Column>,
+    ident: impl Into<String>,
 ) -> Expr {
-    let column = ident.into();
+    let column = Column::from_qualified_name(ident);
     let field: FieldRef =
         Arc::new(Field::new(column.name(), dt, true).with_metadata(metadata));
     Expr::OuterReferenceColumn(field, column)
diff --git a/datafusion/expr/src/expr_schema.rs b/datafusion/expr/src/expr_schema.rs
index 854e907d68b1a..6ba54a1b9028a 100644
--- a/datafusion/expr/src/expr_schema.rs
+++ b/datafusion/expr/src/expr_schema.rs
@@ -1080,11 +1080,8 @@ mod tests {
         assert_eq!(meta, expr.metadata(&schema).unwrap());
 
         // outer ref constructed by `out_ref_col_with_metadata` should be metadata-preserving
-        let outer_ref = out_ref_col_with_metadata(
-            DataType::Int32,
-            meta.to_hashmap(),
-            Column::from_name("foo"),
-        );
+        let outer_ref =
+            out_ref_col_with_metadata(DataType::Int32, meta.to_hashmap(), "foo");
         assert_eq!(meta, outer_ref.metadata(&schema).unwrap());
     }
 
diff --git a/datafusion/expr/src/logical_plan/builder.rs b/datafusion/expr/src/logical_plan/builder.rs
index edf989a6df596..8b12bd6e475ca 100644
--- a/datafusion/expr/src/logical_plan/builder.rs
+++ b/datafusion/expr/src/logical_plan/builder.rs
@@ -1500,19 +1500,23 @@ impl LogicalPlanBuilder {
     }
 
     /// Unnest the given column.
-    pub fn unnest_column(self, column: impl Into<Column>) -> Result<Self> {
-        unnest(Arc::unwrap_or_clone(self.plan), vec![column.into()]).map(Self::new)
+    pub fn unnest_column(self, column: impl Into<String>) -> Result<Self> {
+        unnest(
+            Arc::unwrap_or_clone(self.plan),
+            vec![Column::from_qualified_name(column)],
+        )
+        .map(Self::new)
     }
 
     /// Unnest the given column given [`UnnestOptions`]
     pub fn unnest_column_with_options(
         self,
-        column: impl Into<Column>,
+        column: impl Into<String>,
         options: UnnestOptions,
     ) -> Result<Self> {
         unnest_with_options(
             Arc::unwrap_or_clone(self.plan),
-            vec![column.into()],
+            vec![Column::from_qualified_name(column)],
             options,
         )
         .map(Self::new)
@@ -2664,7 +2668,7 @@ mod tests {
         // Unnesting multiple fields at the same time, using infer syntax
         let cols = vec!["strings", "structs", "struct_singular"]
             .into_iter()
-            .map(|c| c.into())
+            .map(Column::from_qualified_name)
             .collect();
 
         let plan = nested_table_scan("test_table")?
@@ -2683,16 +2687,19 @@ mod tests {
         // Simultaneously unnesting a list (with different depth) and a struct column
         let plan = nested_table_scan("test_table")?
             .unnest_columns_with_options(
-                vec!["stringss".into(), "struct_singular".into()],
+                vec![
+                    Column::from_qualified_name("stringss"),
+                    Column::from_qualified_name("struct_singular"),
+                ],
                 UnnestOptions::default()
                     .with_recursions(RecursionUnnestOption {
-                        input_column: "stringss".into(),
-                        output_column: "stringss_depth_1".into(),
+                        input_column: Column::from_qualified_name("stringss"),
+                        output_column: Column::from_qualified_name("stringss_depth_1"),
                         depth: 1,
                     })
                     .with_recursions(RecursionUnnestOption {
-                        input_column: "stringss".into(),
-                        output_column: "stringss_depth_2".into(),
+                        input_column: Column::from_qualified_name("stringss"),
+                        output_column: Column::from_qualified_name("stringss_depth_2"),
                         depth: 2,
                     }),
             )?
diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs
index 5b143ef135eda..af6b2ecd57411 100644
--- a/datafusion/expr/src/logical_plan/plan.rs
+++ b/datafusion/expr/src/logical_plan/plan.rs
@@ -4989,7 +4989,8 @@ mod tests {
         let col = schema.field_names()[0].clone();
 
         let filter = Filter::try_new(
-            Expr::Column(col.into()).eq(Expr::Literal(ScalarValue::Int32(Some(1)), None)),
+            Expr::Column(Column::from_qualified_name(col))
+                .eq(Expr::Literal(ScalarValue::Int32(Some(1)), None)),
             scan,
         )
         .unwrap();
@@ -5018,8 +5019,11 @@ mod tests {
         }));
         let col = schema.field_names()[0].clone();
 
-        let filter =
-            Filter::try_new(Expr::Column(col.into()).eq(lit(1i32)), scan).unwrap();
+        let filter = Filter::try_new(
+            Expr::Column(Column::from_qualified_name(col)).eq(lit(1i32)),
+            scan,
+        )
+        .unwrap();
         assert!(filter.is_scalar());
     }
 
diff --git a/datafusion/optimizer/src/analyzer/resolve_grouping_function.rs b/datafusion/optimizer/src/analyzer/resolve_grouping_function.rs
index 747c54e2cd26d..a89b5f67619ec 100644
--- a/datafusion/optimizer/src/analyzer/resolve_grouping_function.rs
+++ b/datafusion/optimizer/src/analyzer/resolve_grouping_function.rs
@@ -205,7 +205,8 @@ fn grouping_function_on_id(
         }
     };
 
-    let grouping_id_column = Expr::Column(Column::from(Aggregate::INTERNAL_GROUPING_ID));
+    let grouping_id_column =
+        Expr::Column(Column::from_qualified_name(Aggregate::INTERNAL_GROUPING_ID));
     // The grouping call is exactly our internal grouping id
     if args.len() == group_by_expr_count
         && args
diff --git a/datafusion/optimizer/src/common_subexpr_eliminate.rs b/datafusion/optimizer/src/common_subexpr_eliminate.rs
index d9273a8f60fb2..bc2711a1922d8 100644
--- a/datafusion/optimizer/src/common_subexpr_eliminate.rs
+++ b/datafusion/optimizer/src/common_subexpr_eliminate.rs
@@ -1167,7 +1167,15 @@ mod test {
         let table_scan_1 = test_table_scan_with_name("test1").unwrap();
         let table_scan_2 = test_table_scan_with_name("test2").unwrap();
         let join = LogicalPlanBuilder::from(table_scan_1)
-            .join(table_scan_2, JoinType::Inner, (vec!["a"], vec!["a"]), None)
+            .join(
+                table_scan_2,
+                JoinType::Inner,
+                (
+                    vec![Column::from_qualified_name("a")],
+                    vec![Column::from_qualified_name("a")],
+                ),
+                None,
+            )
             .unwrap()
             .build()
             .unwrap();
diff --git a/datafusion/optimizer/src/eliminate_cross_join.rs b/datafusion/optimizer/src/eliminate_cross_join.rs
index c5a3a7d96ce89..4c9d5cb1961bb 100644
--- a/datafusion/optimizer/src/eliminate_cross_join.rs
+++ b/datafusion/optimizer/src/eliminate_cross_join.rs
@@ -449,6 +449,7 @@ mod tests {
     use crate::optimizer::OptimizerContext;
     use crate::test::*;
 
+    use datafusion_common::Column;
     use datafusion_expr::{
         Operator::{And, Or},
         binary_expr, col, lit,
@@ -654,10 +655,21 @@ mod tests {
             .join(
                 t3,
                 JoinType::Inner,
-                (vec!["t1.a"], vec!["t3.a"]),
+                (
+                    vec![Column::from_qualified_name("t1.a")],
+                    vec![Column::from_qualified_name("t3.a")],
+                ),
                 Some(col("t1.a").gt(lit(20u32))),
             )?
-            .join(t2, JoinType::Inner, (vec!["t1.a"], vec!["t2.a"]), None)?
+            .join(
+                t2,
+                JoinType::Inner,
+                (
+                    vec![Column::from_qualified_name("t1.a")],
+                    vec![Column::from_qualified_name("t2.a")],
+                ),
+                None,
+            )?
             .filter(col("t1.a").gt(lit(15u32)))?
             .build()?;
 
diff --git a/datafusion/optimizer/src/filter_null_join_keys.rs b/datafusion/optimizer/src/filter_null_join_keys.rs
index c8f419d3e543e..aa4d952577e3f 100644
--- a/datafusion/optimizer/src/filter_null_join_keys.rs
+++ b/datafusion/optimizer/src/filter_null_join_keys.rs
@@ -308,7 +308,10 @@ mod tests {
             .join(
                 t2,
                 JoinType::Inner,
-                (vec!["optional_id"], vec!["t2.optional_id"]),
+                (
+                    vec![Column::from_qualified_name("optional_id")],
+                    vec![Column::from_qualified_name("t2.optional_id")],
+                ),
                 None,
             )?
             .build()?;
diff --git a/datafusion/optimizer/src/optimize_projections/mod.rs b/datafusion/optimizer/src/optimize_projections/mod.rs
index 548eadffa242e..e943cca90cc0d 100644
--- a/datafusion/optimizer/src/optimize_projections/mod.rs
+++ b/datafusion/optimizer/src/optimize_projections/mod.rs
@@ -1734,9 +1734,12 @@ mod tests {
         let plan = table_scan(Some("m4"), &schema, None)?
             .aggregate(
                 Vec::<Expr>::new(),
-                vec![max(col(Column::new_unqualified("tag.one"))).alias("tag.one")],
+                vec![
+                    max(Expr::Column(Column::new_unqualified("tag.one")))
+                        .alias("tag.one"),
+                ],
             )?
-            .project([col(Column::new_unqualified("tag.one"))])?
+            .project([Expr::Column(Column::new_unqualified("tag.one"))])?
             .build()?;
 
         assert_optimized_plan_equal!(
@@ -1842,7 +1845,15 @@ mod tests {
         let table2_scan = scan_empty(Some("test2"), &schema, None)?.build()?;
 
         let plan = LogicalPlanBuilder::from(table_scan)
-            .join(table2_scan, JoinType::Left, (vec!["a"], vec!["c1"]), None)?
+            .join(
+                table2_scan,
+                JoinType::Left,
+                (
+                    vec![Column::from_qualified_name("a")],
+                    vec![Column::from_qualified_name("c1")],
+                ),
+                None,
+            )?
             .project(vec![col("a"), col("b"), col("c1")])?
             .build()?;
 
@@ -1894,7 +1905,15 @@ mod tests {
         let table2_scan = scan_empty(Some("test2"), &schema, None)?.build()?;
 
         let plan = LogicalPlanBuilder::from(table_scan)
-            .join(table2_scan, JoinType::Left, (vec!["a"], vec!["c1"]), None)?
+            .join(
+                table2_scan,
+                JoinType::Left,
+                (
+                    vec![Column::from_qualified_name("a")],
+                    vec![Column::from_qualified_name("c1")],
+                ),
+                None,
+            )?
             // projecting joined column `a` should push the right side column `c1` projection as
             // well into test2 table even though `c1` is not referenced in projection.
             .project(vec![col("a"), col("b")])?
@@ -1949,7 +1968,11 @@ mod tests {
         let table2_scan = scan_empty(Some("test2"), &schema, None)?.build()?;
 
         let plan = LogicalPlanBuilder::from(table_scan)
-            .join_using(table2_scan, JoinType::Left, vec!["a".into()])?
+            .join_using(
+                table2_scan,
+                JoinType::Left,
+                vec![Column::from_qualified_name("a")],
+            )?
             .project(vec![col("a"), col("b")])?
             .build()?;
 
diff --git a/datafusion/optimizer/src/push_down_limit.rs b/datafusion/optimizer/src/push_down_limit.rs
index 7b302adf22acc..17c78200625f7 100644
--- a/datafusion/optimizer/src/push_down_limit.rs
+++ b/datafusion/optimizer/src/push_down_limit.rs
@@ -279,7 +279,7 @@ mod test {
     use crate::test::*;
 
     use crate::OptimizerContext;
-    use datafusion_common::DFSchemaRef;
+    use datafusion_common::{Column, DFSchemaRef};
     use datafusion_expr::{
         Expr, Extension, UserDefinedLogicalNodeCore, col, exists,
         logical_plan::builder::LogicalPlanBuilder,
@@ -837,7 +837,10 @@ mod test {
             .join(
                 LogicalPlanBuilder::from(table_scan_2).build()?,
                 JoinType::Inner,
-                (vec!["a"], vec!["a"]),
+                (
+                    vec![Column::from_qualified_name("a")],
+                    vec![Column::from_qualified_name("a")],
+                ),
                 None,
             )?
             .limit(10, Some(1000))?
@@ -864,7 +867,10 @@ mod test {
             .join(
                 LogicalPlanBuilder::from(table_scan_2).build()?,
                 JoinType::Inner,
-                (vec!["a"], vec!["a"]),
+                (
+                    vec![Column::from_qualified_name("a")],
+                    vec![Column::from_qualified_name("a")],
+                ),
                 None,
             )?
             .limit(10, Some(1000))?
@@ -955,7 +961,10 @@ mod test {
             .join(
                 LogicalPlanBuilder::from(table_scan_2).build()?,
                 JoinType::Left,
-                (vec!["a"], vec!["a"]),
+                (
+                    vec![Column::from_qualified_name("a")],
+                    vec![Column::from_qualified_name("a")],
+                ),
                 None,
             )?
             .limit(10, Some(1000))?
@@ -983,7 +992,10 @@ mod test {
             .join(
                 LogicalPlanBuilder::from(table_scan_2).build()?,
                 JoinType::Right,
-                (vec!["a"], vec!["a"]),
+                (
+                    vec![Column::from_qualified_name("a")],
+                    vec![Column::from_qualified_name("a")],
+                ),
                 None,
             )?
             .limit(0, Some(1000))?
@@ -1011,7 +1023,10 @@ mod test {
             .join(
                 LogicalPlanBuilder::from(table_scan_2).build()?,
                 JoinType::Right,
-                (vec!["a"], vec!["a"]),
+                (
+                    vec![Column::from_qualified_name("a")],
+                    vec![Column::from_qualified_name("a")],
+                ),
                 None,
             )?
             .limit(10, Some(1000))?
diff --git a/datafusion/optimizer/src/replace_distinct_aggregate.rs b/datafusion/optimizer/src/replace_distinct_aggregate.rs
index f24ca1f119b61..e554110178c34 100644
--- a/datafusion/optimizer/src/replace_distinct_aggregate.rs
+++ b/datafusion/optimizer/src/replace_distinct_aggregate.rs
@@ -26,7 +26,7 @@ use datafusion_common::{Column, Result};
 use datafusion_expr::expr_rewriter::normalize_cols;
 use datafusion_expr::utils::expand_wildcard;
 use datafusion_expr::{Aggregate, Distinct, DistinctOn, Expr, LogicalPlan};
-use datafusion_expr::{ExprFunctionExt, Limit, LogicalPlanBuilder, col, lit};
+use datafusion_expr::{ExprFunctionExt, Limit, LogicalPlanBuilder, lit};
 
 /// Optimizer that replaces logical [[Distinct]] with a logical [[Aggregate]]
 ///
@@ -179,7 +179,7 @@ impl OptimizerRule for ReplaceDistinctWithAggregate {
                     .skip(expr_cnt)
                     .zip(schema.iter())
                     .map(|((new_qualifier, new_field), (old_qualifier, old_field))| {
-                        col(Column::from((new_qualifier, new_field)))
+                        Expr::Column(Column::from((new_qualifier, new_field)))
                             .alias_qualified(old_qualifier.cloned(), old_field.name())
                     })
                     .collect::<Vec<Expr>>();
diff --git a/datafusion/optimizer/src/simplify_expressions/simplify_predicates.rs b/datafusion/optimizer/src/simplify_expressions/simplify_predicates.rs
index e811ce7313102..584d2d468bd9f 100644
--- a/datafusion/optimizer/src/simplify_expressions/simplify_predicates.rs
+++ b/datafusion/optimizer/src/simplify_expressions/simplify_predicates.rs
@@ -300,7 +300,10 @@ mod tests {
 
         // Test that it still extracts from direct column references
         let col_expr = col("a");
-        assert_eq!(extract_column_from_expr(&col_expr), Some(Column::from("a")));
+        assert_eq!(
+            extract_column_from_expr(&col_expr),
+            Some(Column::from_qualified_name("a"))
+        );
     }
 
     #[test]
@@ -322,8 +325,8 @@ mod tests {
         let has_a_predicate = result.iter().any(|p| {
             matches!(p, Expr::BinaryExpr(BinaryExpr { 
                 left, 
-                op: Operator::Lt, 
-                right 
+                op: Operator::Lt,
+                right
             }) if left == &Box::new(col("a")) && right == &Box::new(lit(3i32)))
         });
         assert!(has_a_predicate, "Should have a < 3 predicate");
@@ -332,8 +335,8 @@ mod tests {
         let has_b_predicate = result.iter().any(|p| {
             matches!(p, Expr::BinaryExpr(BinaryExpr { 
                 left, 
-                op: Operator::Gt, 
-                right 
+                op: Operator::Gt,
+                right
             }) if left == &Box::new(col("b")) && right == &Box::new(lit(20i32)))
         });
         assert!(has_b_predicate, "Should have b > 20 predicate");
diff --git a/datafusion/sql/src/unparser/rewrite.rs b/datafusion/sql/src/unparser/rewrite.rs
index ec1b17cd28a91..7df265462055a 100644
--- a/datafusion/sql/src/unparser/rewrite.rs
+++ b/datafusion/sql/src/unparser/rewrite.rs
@@ -202,19 +202,23 @@ pub(super) fn rewrite_plan_for_sort_on_non_projected_fields(
         .enumerate()
         .map(|(i, f)| match f {
             Expr::Alias(alias) => {
-                let a = Expr::Column(alias.name.clone().into());
+                let a = Expr::Column(Column::from_qualified_name(alias.name.clone()));
                 map.insert(a.clone(), f.clone());
                 a
             }
             Expr::Column(_) => {
                 map.insert(
-                    Expr::Column(inner_p.schema.field(i).name().into()),
+                    Expr::Column(Column::from_qualified_name(
+                        inner_p.schema.field(i).name(),
+                    )),
                     f.clone(),
                 );
                 f.clone()
             }
             _ => {
-                let a = Expr::Column(inner_p.schema.field(i).name().into());
+                let a = Expr::Column(Column::from_qualified_name(
+                    inner_p.schema.field(i).name(),
+                ));
                 map.insert(a.clone(), f.clone());
                 a
             }
diff --git a/datafusion/sql/tests/cases/plan_to_sql.rs b/datafusion/sql/tests/cases/plan_to_sql.rs
index 46a42ae534af0..b8eed270dd0f3 100644
--- a/datafusion/sql/tests/cases/plan_to_sql.rs
+++ b/datafusion/sql/tests/cases/plan_to_sql.rs
@@ -1773,7 +1773,10 @@ fn test_join_with_table_scan_filters() -> Result<()> {
         .join(
             right_plan.clone(),
             datafusion_expr::JoinType::Inner,
-            (vec!["left.id"], vec!["right_table.id"]),
+            (
+                vec![Column::from_qualified_name("left.id")],
+                vec![Column::from_qualified_name("right_table.id")],
+            ),
             Some(col("left.id").gt(lit(5))),
         )?
         .build()?;
@@ -1788,7 +1791,10 @@ fn test_join_with_table_scan_filters() -> Result<()> {
         .join(
             right_plan,
             datafusion_expr::JoinType::Inner,
-            (vec!["left.id"], vec!["right_table.id"]),
+            (
+                vec![Column::from_qualified_name("left.id")],
+                vec![Column::from_qualified_name("right_table.id")],
+            ),
             None,
         )?
         .build()?;
@@ -1812,7 +1818,10 @@ fn test_join_with_table_scan_filters() -> Result<()> {
         .join(
             right_plan_with_filter,
             datafusion_expr::JoinType::Inner,
-            (vec!["left.id"], vec!["right_table.id"]),
+            (
+                vec![Column::from_qualified_name("left.id")],
+                vec![Column::from_qualified_name("right_table.id")],
+            ),
             Some(col("left.id").gt(lit(5))),
         )?
         .filter(col("left.name").eq(lit("after_join_filter_val")))?
@@ -1843,7 +1852,10 @@ fn test_join_with_table_scan_filters() -> Result<()> {
         .join(
             right_plan_with_duplicated_filter,
             datafusion_expr::JoinType::Inner,
-            (vec!["left.id"], vec!["right_table.id"]),
+            (
+                vec![Column::from_qualified_name("left.id")],
+                vec![Column::from_qualified_name("right_table.id")],
+            ),
             Some(col("left.id").gt(lit(5))),
         )?
         .build()?;
diff --git a/datafusion/substrait/src/logical_plan/consumer/plan.rs b/datafusion/substrait/src/logical_plan/consumer/plan.rs
index d5e10fb604017..26a19e0223e27 100644
--- a/datafusion/substrait/src/logical_plan/consumer/plan.rs
+++ b/datafusion/substrait/src/logical_plan/consumer/plan.rs
@@ -20,7 +20,7 @@ use super::{DefaultSubstraitConsumer, SubstraitConsumer};
 use crate::extensions::Extensions;
 use datafusion::common::{not_impl_err, plan_err};
 use datafusion::execution::SessionState;
-use datafusion::logical_expr::{Aggregate, LogicalPlan, Projection, col};
+use datafusion::logical_expr::{Aggregate, Expr, LogicalPlan, Projection};
 use std::sync::Arc;
 use substrait::proto::{Plan, plan_rel};
 
@@ -107,7 +107,7 @@ pub async fn from_substrait_plan_with_consumer(
                                     plan.schema()
                                         .columns()
                                         .iter()
-                                        .map(|c| col(c.to_owned())),
+                                        .map(|c| Expr::Column(c.clone())),
                                     plan.schema(),
                                     renamed_schema.fields(),
                                 )?,
diff --git a/datafusion/substrait/src/logical_plan/consumer/rel/join_rel.rs b/datafusion/substrait/src/logical_plan/consumer/rel/join_rel.rs
index 3604630d6f0bb..37dfe5fc521cd 100644
--- a/datafusion/substrait/src/logical_plan/consumer/rel/join_rel.rs
+++ b/datafusion/substrait/src/logical_plan/consumer/rel/join_rel.rs
@@ -75,7 +75,7 @@ pub async fn from_join_rel(
             .build()
         }
         None => {
-            let on: Vec<String> = vec![];
+            let on: Vec<Column> = vec![];
             left.join_detailed(
                 right.build()?,
                 join_type,
diff --git a/datafusion/substrait/src/logical_plan/producer/expr/mod.rs b/datafusion/substrait/src/logical_plan/producer/expr/mod.rs
index 5057564d370cf..35771e676f988 100644
--- a/datafusion/substrait/src/logical_plan/producer/expr/mod.rs
+++ b/datafusion/substrait/src/logical_plan/producer/expr/mod.rs
@@ -168,7 +168,7 @@ mod tests {
     use super::*;
     use crate::logical_plan::consumer::from_substrait_extended_expr;
     use datafusion::arrow::datatypes::{DataType, Schema};
-    use datafusion::common::{DFSchema, DataFusionError, ScalarValue};
+    use datafusion::common::{Column, DFSchema, DataFusionError, ScalarValue};
     use datafusion::execution::SessionStateBuilder;
 
     #[tokio::test]
@@ -191,8 +191,8 @@ mod tests {
         assert_eq!(rt_expr, &expr);
 
         // Multiple expressions, with column references
-        let expr1 = Expr::Column("c0".into());
-        let expr2 = Expr::Column("c1".into());
+        let expr1 = Expr::Column(Column::from_qualified_name("c0"));
+        let expr2 = Expr::Column(Column::from_qualified_name("c1"));
         let out1 = Field::new("out1", DataType::Int32, true);
         let out2 = Field::new("out2", DataType::Utf8, true);
         let input_schema = DFSchemaRef::new(DFSchema::try_from(Schema::new(vec![
@@ -228,7 +228,7 @@ mod tests {
         let state = SessionStateBuilder::default().build();
 
         // Not ok if input schema is missing field referenced by expr
-        let expr = Expr::Column("missing".into());
+        let expr = Expr::Column(Column::from_qualified_name("missing"));
         let field = Field::new("out", DataType::Int32, false);
         let empty_schema = DFSchemaRef::new(DFSchema::empty());