Support qualified columns in queries

datafusion/src/dataframe.rs

@@ -20,7 +20,7 @@
 use crate::arrow::record_batch::RecordBatch;
 use crate::error::Result;
 use crate::logical_plan::{
-    DFSchema, Expr, FunctionRegistry, JoinType, LogicalPlan, Partitioning,
+    Column, DFSchema, Expr, FunctionRegistry, JoinType, LogicalPlan, Partitioning,
 };
 use std::sync::Arc;
 
@@ -175,7 +175,12 @@ pub trait DataFrame: Send + Sync {
     ///     col("a").alias("a2"),
     ///     col("b").alias("b2"),
     ///     col("c").alias("c2")])?;
-    /// let join = left.join(right, JoinType::Inner, &["a", "b"], &["a2", "b2"])?;
+    /// let join = left.join(
+    ///     right,
+    ///     JoinType::Inner,
+    ///     vec![Column::from_name("a".to_string()), Column::from_name("b".to_string())],
+    ///     vec![Column::from_name("a2".to_string()), Column::from_name("b2".to_string())],
+    /// )?;
     /// let batches = join.collect().await?;
     /// # Ok(())
     /// # }
@@ -184,8 +189,8 @@ pub trait DataFrame: Send + Sync {
         &self,
         right: Arc<dyn DataFrame>,
         join_type: JoinType,
-        left_cols: &[&str],
-        right_cols: &[&str],
+        left_cols: Vec<Column>,
+        right_cols: Vec<Column>,
     ) -> Result<Arc<dyn DataFrame>>;
 
     /// Repartition a DataFrame based on a logical partitioning scheme.

datafusion/src/datasource/csv.rs

@@ -33,7 +33,7 @@
 //! let schema = csvdata.schema();
 //! ```
 
-use arrow::datatypes::SchemaRef;
+use arrow::datatypes::{Schema, SchemaRef};
 use std::any::Any;
 use std::string::String;
 use std::sync::Arc;
@@ -123,10 +123,18 @@ impl TableProvider for CsvFile {
         _filters: &[Expr],
         limit: Option<usize>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
+        let fields = self
+            .schema
+            .fields()
+            .iter()
+            .map(|f| f.clone())
+            .collect::<Vec<_>>();
+        let schema = Schema::new(fields);
+
         Ok(Arc::new(CsvExec::try_new(
             &self.path,
             CsvReadOptions::new()
-                .schema(&self.schema)
+                .schema(&schema)
                 .has_header(self.has_header)
                 .delimiter(self.delimiter)
                 .file_extension(self.file_extension.as_str()),