ARROW-7795: [Rust] Added support for NOT

Closes #6384 from jorgecarleitao/not and squashes the following commits:

af46e0e <Jorge C. Leitao> ARROW-7795  Added support for NOT

Authored-by: Jorge C. Leitao <jorgecarleitao@gmail.com>
Signed-off-by: Andy Grove <andygrove73@gmail.com>

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

@@ -987,6 +987,48 @@ pub fn binary(
     Arc::new(BinaryExpr::new(l, op, r))
 }
 
+/// Not expression
+pub struct NotExpr {
+    arg: Arc<dyn PhysicalExpr>,
+}
+
+impl NotExpr {
+    /// Create new not expression
+    pub fn new(arg: Arc<dyn PhysicalExpr>) -> Self {
+        Self { arg }
+    }
+}
+
+impl PhysicalExpr for NotExpr {
+    fn name(&self) -> String {
+        "NOT".to_string()
+    }
+
+    fn data_type(&self, _input_schema: &Schema) -> Result<DataType> {
+        return Ok(DataType::Boolean);
+    }
+
+    fn evaluate(&self, batch: &RecordBatch) -> Result<ArrayRef> {
+        let arg = self.arg.evaluate(batch)?;
+        if arg.data_type() != &DataType::Boolean {
+            return Err(ExecutionError::General(format!(
+                "Cannot evaluate \"not\" expression with type {:?}",
+                arg.data_type(),
+            )));
+        }
+        let arg = arg
+            .as_any()
+            .downcast_ref::<BooleanArray>()
+            .expect("boolean_op failed to downcast array");
+        return Ok(Arc::new(arrow::compute::kernels::boolean::not(arg)?));
+    }
+}
+
+/// Create a unary expression
+pub fn not(arg: Arc<dyn PhysicalExpr>) -> Arc<dyn PhysicalExpr> {
+    Arc::new(NotExpr::new(arg))
+}
+
 /// CAST expression casts an expression to a specific data type
 pub struct CastExpr {
     /// The expression to cast
@@ -1815,4 +1857,27 @@ mod tests {
             assert_eq!(expected.value(i), actual.value(i));
         }
     }
+
+    #[test]
+    fn neg_op() -> Result<()> {
+        let schema = Schema::new(vec![Field::new("a", DataType::Boolean, true)]);
+        let a = BooleanArray::from(vec![true, false]);
+        let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?;
+
+        // expression: "!a"
+        let lt = not(col(0));
+        let result = lt.evaluate(&batch)?;
+        assert_eq!(result.len(), 2);
+
+        let expected = vec![false, true];
+        let result = result
+            .as_any()
+            .downcast_ref::<BooleanArray>()
+            .expect("failed to downcast to BooleanArray");
+        for i in 0..2 {
+            assert_eq!(result.value(i), expected[i]);
+        }
+
+        Ok(())
+    }
 }

rust/datafusion/src/logicalplan.rs

@@ -192,6 +192,8 @@ pub enum Expr {
         /// Right-hand side of the expression
         right: Arc<Expr>,
     },
+    /// unary NOT
+    Not(Arc<Expr>),
     /// unary IS NOT NULL
     IsNotNull(Arc<Expr>),
     /// unary IS NULL
@@ -239,6 +241,7 @@ impl Expr {
             Expr::Cast { data_type, .. } => data_type.clone(),
             Expr::ScalarFunction { return_type, .. } => return_type.clone(),
             Expr::AggregateFunction { return_type, .. } => return_type.clone(),
+            Expr::Not(_) => DataType::Boolean,
             Expr::IsNull(_) => DataType::Boolean,
             Expr::IsNotNull(_) => DataType::Boolean,
             Expr::BinaryExpr {
@@ -333,6 +336,11 @@ impl Expr {
             right: Arc::new(other.clone()),
         }
     }
+
+    /// Not
+    pub fn not(&self) -> Expr {
+        Expr::Not(Arc::new(self.clone()))
+    }
 }
 
 impl fmt::Debug for Expr {
@@ -343,6 +351,7 @@ impl fmt::Debug for Expr {
             Expr::Cast { expr, data_type } => {
                 write!(f, "CAST({:?} AS {:?})", expr, data_type)
             }
+            Expr::Not(expr) => write!(f, "NOT {:?}", expr),
             Expr::IsNull(expr) => write!(f, "{:?} IS NULL", expr),
             Expr::IsNotNull(expr) => write!(f, "{:?} IS NOT NULL", expr),
             Expr::BinaryExpr { left, op, right } => {

rust/datafusion/src/optimizer/projection_push_down.rs

@@ -231,6 +231,7 @@ impl ProjectionPushDown {
         match expr {
             Expr::Column(i) => Ok(Expr::Column(self.new_index(mapping, i)?)),
             Expr::Literal(_) => Ok(expr.clone()),
+            Expr::Not(e) => Ok(Expr::Not(Arc::new(self.rewrite_expr(e, mapping)?))),
             Expr::IsNull(e) => Ok(Expr::IsNull(Arc::new(self.rewrite_expr(e, mapping)?))),
             Expr::IsNotNull(e) => {
                 Ok(Expr::IsNotNull(Arc::new(self.rewrite_expr(e, mapping)?)))

rust/datafusion/src/optimizer/utils.rs

@@ -38,6 +38,7 @@ pub fn expr_to_column_indices(expr: &Expr, accum: &mut HashSet<usize>) {
             accum.insert(*i);
         }
         Expr::Literal(_) => { /* not needed */ }
+        Expr::Not(e) => expr_to_column_indices(e, accum),
         Expr::IsNull(e) => expr_to_column_indices(e, accum),
         Expr::IsNotNull(e) => expr_to_column_indices(e, accum),
         Expr::BinaryExpr { left, right, .. } => {

rust/datafusion/src/sql/planner.rs

@@ -308,6 +308,18 @@ impl<S: SchemaProvider> SqlToRel<S> {
                 Ok(Expr::IsNotNull(Arc::new(self.sql_to_rex(expr, schema)?)))
             }
 
+            ASTNode::SQLUnary{
+                ref operator,
+                ref expr,
+            } => {
+                match *operator {
+                    SQLOperator::Not => Ok(Expr::Not(Arc::new(self.sql_to_rex(expr, schema)?))),
+                    _ => Err(ExecutionError::InternalError(format!(
+                        "SQL binary operator cannot be interpreted as a unary operator"
+                    ))),
+                }
+            }
+
             ASTNode::SQLBinaryExpr {
                 ref left,
                 ref op,
@@ -332,11 +344,16 @@ impl<S: SchemaProvider> SqlToRel<S> {
                     SQLOperator::NotLike => Operator::NotLike,
                 };
 
-                Ok(Expr::BinaryExpr {
-                    left: Arc::new(self.sql_to_rex(&left, &schema)?),
-                    op: operator,
-                    right: Arc::new(self.sql_to_rex(&right, &schema)?),
-                })
+                match operator {
+                    Operator::Not => Err(ExecutionError::InternalError(format!(
+                        "SQL unary operator \"NOT\" cannot be interpreted as a binary operator"
+                    ))),
+                    _ => Ok(Expr::BinaryExpr {
+                        left: Arc::new(self.sql_to_rex(&left, &schema)?),
+                        op: operator,
+                        right: Arc::new(self.sql_to_rex(&right, &schema)?),
+                    })
+                }
             }
 
             //            &ASTNode::SQLOrderBy { ref expr, asc } => Ok(Expr::Sort {
@@ -495,6 +512,16 @@ mod tests {
         quick_test(sql, expected);
     }
 
+    #[test]
+    fn select_neg_selection() {
+        let sql = "SELECT id, first_name, last_name \
+                   FROM person WHERE NOT state";
+        let expected = "Projection: #0, #1, #2\
+                        \n  Selection: NOT #4\
+                        \n    TableScan: person projection=None";
+        quick_test(sql, expected);
+    }
+
     #[test]
     fn select_compound_selection() {
         let sql = "SELECT id, first_name, last_name \