code cleanup, use tempdir crate in tests

rust/datafusion/src/execution/physical_plan/expressions.rs

@@ -23,6 +23,7 @@ use crate::error::Result;
 use crate::execution::physical_plan::PhysicalExpr;
 use crate::logicalplan::{Operator, ScalarValue};
 
+use crate::error::ExecutionError;
 use arrow::array::{ArrayRef, UInt32Array, UInt32Builder};
 use arrow::compute::lt;
 use arrow::datatypes::{DataType, Schema};
@@ -87,7 +88,8 @@ impl PhysicalExpr for Literal {
                 }
                 Ok(Arc::new(builder.finish()))
             }
-            _ => unimplemented!(),
+            // TODO add other types
+            _ => Err(ExecutionError::General("Unsupported data type".to_string())),
         }
     }
 }
@@ -122,14 +124,25 @@ impl PhysicalExpr for BinaryExpr {
     fn evaluate(&self, batch: &RecordBatch) -> Result<ArrayRef> {
         let left = self.left.evaluate(batch)?;
         let right = self.right.evaluate(batch)?;
+        if left.data_type() != right.data_type() {
+            return Err(ExecutionError::General("Type mismatch".to_string()));
+        }
         match self.op {
             Operator::Lt => {
-                // assume primitive array for now
-                let left = left.as_any().downcast_ref::<UInt32Array>().unwrap();
-                let right = right.as_any().downcast_ref::<UInt32Array>().unwrap();
-                Ok(Arc::new(lt(left, right)?))
+                match left.data_type() {
+                    DataType::UInt32 => {
+                        let left = left.as_any().downcast_ref::<UInt32Array>().unwrap();
+                        let right = right.as_any().downcast_ref::<UInt32Array>().unwrap();
+                        Ok(Arc::new(lt(left, right)?))
+                    }
+                    // TODO add other types
+                    _ => {
+                        Err(ExecutionError::General("Unsupported data type".to_string()))
+                    }
+                }
             }
-            _ => unimplemented!(),
+            // TODO add other operators
+            _ => Err(ExecutionError::General("Unsupported operator".to_string())),
         }
     }
 }

rust/datafusion/src/test/mod.rs

@@ -23,12 +23,11 @@ use crate::execution::physical_plan::ExecutionPlan;
 use arrow::datatypes::{DataType, Field, Schema};
 use arrow::record_batch::RecordBatch;
 use std::env;
-use std::fs;
 use std::fs::File;
 use std::io::prelude::*;
 use std::io::{BufReader, BufWriter};
-use std::path::Path;
 use std::sync::Arc;
+use tempdir::TempDir;
 
 /// Get the value of the ARROW_TEST_DATA environment variable
 pub fn arrow_testdata_path() -> String {
@@ -46,19 +45,13 @@ pub fn create_partitioned_csv(filename: &str, partitions: usize) -> Result<Strin
     let testdata = arrow_testdata_path();
     let path = format!("{}/csv/{}", testdata, filename);
 
-    let mut dir = env::temp_dir();
-    dir.push(&format!("{}-{}", filename, partitions));
-
-    if Path::new(&dir).exists() {
-        fs::remove_dir_all(&dir).unwrap();
-    }
-    fs::create_dir(dir.clone()).unwrap();
+    let tmp_dir = TempDir::new("datafusion")?;
 
     let mut writers = vec![];
     for i in 0..partitions {
-        let mut filename = dir.clone();
-        filename.push(format!("part{}.csv", i));
-        let writer = BufWriter::new(File::create(&filename).unwrap());
+        let filename = format!("partition-{}.csv", i);
+        let path = tmp_dir.path().join(&filename);
+        let writer = BufWriter::new(File::create(path).unwrap());
         writers.push(writer);
     }
 
@@ -87,7 +80,7 @@ pub fn create_partitioned_csv(filename: &str, partitions: usize) -> Result<Strin
         w.flush().unwrap();
     }
 
-    Ok(dir.as_os_str().to_str().unwrap().to_string())
+    Ok(path.to_string())
 }
 
 /// Get the schema for the aggregate_test_* csv files