ARROW-6700: [Rust] [DataFusion] Use new Arrow Parquet reader

rust/arrow/src/compute/kernels/cast.rs

@@ -35,6 +35,7 @@
 //! assert_eq!(7.0, c.value(2));
 //! ```
 
+use std::str;
 use std::sync::Arc;
 
 use crate::array::*;
@@ -203,6 +204,22 @@ pub fn cast(array: &ArrayRef, to_type: &DataType) -> Result<ArrayRef> {
             Int64 => cast_numeric_to_string::<Int64Type>(array),
             Float32 => cast_numeric_to_string::<Float32Type>(array),
             Float64 => cast_numeric_to_string::<Float64Type>(array),
+            Binary => {
+                let from = array.as_any().downcast_ref::<BinaryArray>().unwrap();
+                let mut b = StringBuilder::new(array.len());
+                for i in 0..array.len() {
+                    if array.is_null(i) {
+                        b.append_null()?;
+                    } else {
+                        match str::from_utf8(from.value(i)) {
+                            Ok(s) => b.append_value(s)?,
+                            Err(_) => b.append_null()?, // not valid UTF8
+                        }
+                    }
+                }
+
+                Ok(Arc::new(b.finish()) as ArrayRef)
+            }
             _ => Err(ArrowError::ComputeError(format!(
                 "Casting from {:?} to {:?} not supported",
                 from_type, to_type,

rust/datafusion/examples/parquet_sql.rs

@@ -39,7 +39,7 @@ fn main() -> Result<()> {
     )?;
 
     // simple selection
-    let sql = "SELECT int_col, double_col, date_string_col FROM alltypes_plain WHERE id > 1 AND tinyint_col < double_col";
+    let sql = "SELECT int_col, double_col, CAST(date_string_col as VARCHAR) FROM alltypes_plain WHERE id > 1 AND tinyint_col < double_col";
 
     // create the query plan
     let plan = ctx.create_logical_plan(&sql)?;

rust/datafusion/src/datasource/csv.rs

@@ -50,8 +50,8 @@ impl CsvFile {
 }
 
 impl TableProvider for CsvFile {
-    fn schema(&self) -> &Arc<Schema> {
-        &self.schema
+    fn schema(&self) -> Arc<Schema> {
+        self.schema.clone()
     }
 
     fn scan(

rust/datafusion/src/datasource/datasource.rs

@@ -31,7 +31,7 @@ pub type ScanResult = Arc<Mutex<dyn BatchIterator>>;
 /// Source table
 pub trait TableProvider {
     /// Get a reference to the schema for this table
-    fn schema(&self) -> &Arc<Schema>;
+    fn schema(&self) -> Arc<Schema>;
 
     /// Perform a scan of a table and return a sequence of iterators over the data (one
     /// iterator per partition)

rust/datafusion/src/datasource/memory.rs

@@ -66,8 +66,8 @@ impl MemTable {
 }
 
 impl TableProvider for MemTable {
-    fn schema(&self) -> &Arc<Schema> {
-        &self.schema
+    fn schema(&self) -> Arc<Schema> {
+        self.schema.clone()
     }
 
     fn scan(