feat: Initial support for AnyExpression

datafusion/core/src/datasource/listing/helpers.rs

@@ -90,6 +90,7 @@ impl ExpressionVisitor for ApplicabilityVisitor<'_> {
             | Expr::Cast { .. }
             | Expr::TryCast { .. }
             | Expr::BinaryExpr { .. }
+            | Expr::AnyExpr { .. }
             | Expr::Between { .. }
             | Expr::InList { .. }
             | Expr::Exists { .. }

datafusion/core/src/logical_plan/expr_rewriter.rs

@@ -130,6 +130,11 @@ impl ExprRewritable for Expr {
                 op,
                 right: rewrite_boxed(right, rewriter)?,
             },
+            Expr::AnyExpr { left, op, right } => Expr::AnyExpr {
+                left: rewrite_boxed(left, rewriter)?,
+                op,
+                right: rewrite_boxed(right, rewriter)?,
+            },
             Expr::Not(expr) => Expr::Not(rewrite_boxed(expr, rewriter)?),
             Expr::IsNotNull(expr) => Expr::IsNotNull(rewrite_boxed(expr, rewriter)?),
             Expr::IsNull(expr) => Expr::IsNull(rewrite_boxed(expr, rewriter)?),

datafusion/core/src/optimizer/common_subexpr_eliminate.rs

@@ -401,6 +401,10 @@ impl ExprIdentifierVisitor<'_> {
                 desc.push_str("BinaryExpr-");
                 desc.push_str(&op.to_string());
             }
+            Expr::AnyExpr { op, .. } => {
+                desc.push_str("AnyExpr-");
+                desc.push_str(&op.to_string());
+            }
             Expr::Not(_) => {
                 desc.push_str("Not-");
             }

datafusion/core/src/optimizer/simplify_expressions.rs

@@ -387,6 +387,7 @@ impl<'a> ConstEvaluator<'a> {
             Expr::ScalarUDF { fun, .. } => Self::volatility_ok(fun.signature.volatility),
             Expr::Literal(_)
             | Expr::BinaryExpr { .. }
+            | Expr::AnyExpr { .. }
             | Expr::Not(_)
             | Expr::IsNotNull(_)
             | Expr::IsNull(_)

datafusion/core/src/optimizer/utils.rs

@@ -254,6 +254,9 @@ pub fn expr_sub_expressions(expr: &Expr) -> Result<Vec<Expr>> {
         Expr::BinaryExpr { left, right, .. } => {
             Ok(vec![left.as_ref().to_owned(), right.as_ref().to_owned()])
         }
+        Expr::AnyExpr { left, right, .. } => {
+            Ok(vec![left.as_ref().to_owned(), right.as_ref().to_owned()])
+        }
         Expr::IsNull(expr)
         | Expr::IsNotNull(expr)
         | Expr::Cast { expr, .. }
@@ -347,6 +350,11 @@ pub fn rewrite_expression(expr: &Expr, expressions: &[Expr]) -> Result<Expr> {
             op: *op,
             right: Box::new(expressions[1].clone()),
         }),
+        Expr::AnyExpr { op, .. } => Ok(Expr::AnyExpr {
+            left: Box::new(expressions[0].clone()),
+            op: *op,
+            right: Box::new(expressions[1].clone()),
+        }),
         Expr::IsNull(_) => Ok(Expr::IsNull(Box::new(expressions[0].clone()))),
         Expr::IsNotNull(_) => Ok(Expr::IsNotNull(Box::new(expressions[0].clone()))),
         Expr::ScalarFunction { fun, .. } => Ok(Expr::ScalarFunction {

datafusion/core/src/physical_plan/planner.rs

@@ -19,8 +19,15 @@
 
 use super::analyze::AnalyzeExec;
 use super::{
-    aggregates, empty::EmptyExec, expressions::binary, functions,
-    hash_join::PartitionMode, udaf, union::UnionExec, values::ValuesExec, windows,
+    aggregates,
+    empty::EmptyExec,
+    expressions::{any, binary},
+    functions,
+    hash_join::PartitionMode,
+    udaf,
+    union::UnionExec,
+    values::ValuesExec,
+    windows,
 };
 use crate::execution::context::{ExecutionProps, SessionState};
 use crate::logical_plan::plan::{
@@ -107,6 +114,11 @@ fn create_physical_name(e: &Expr, is_first_expr: bool) -> Result<String> {
             let right = create_physical_name(right, false)?;
             Ok(format!("{} {:?} {}", left, op, right))
         }
+        Expr::AnyExpr { left, op, right } => {
+            let left = create_physical_name(left, false)?;
+            let right = create_physical_name(right, false)?;
+            Ok(format!("{} {:?} ANY({})", left, op, right))
+        }
         Expr::Case {
             expr,
             when_then_expr,
@@ -1113,7 +1125,6 @@ pub fn create_physical_expr(
             create_physical_expr(expr, input_dfschema, input_schema, execution_props)?,
             key.clone(),
         ))),
-
         Expr::ScalarFunction { fun, args } => {
             let physical_args = args
                 .iter()
@@ -1176,6 +1187,21 @@ pub fn create_physical_expr(
                 binary_expr
             }
         }
+        Expr::AnyExpr { left, op, right } => {
+            let lhs = create_physical_expr(
+                left,
+                input_dfschema,
+                input_schema,
+                execution_props,
+            )?;
+            let rhs = create_physical_expr(
+                right,
+                input_dfschema,
+                input_schema,
+                execution_props,
+            )?;
+            any(lhs, *op, rhs, input_schema)
+        }
         Expr::InList {
             expr,
             list,

datafusion/core/src/sql/planner.rs

@@ -1467,6 +1467,30 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
         }
     }
 
+    fn parse_sql_binary_any(
+        &self,
+        left: SQLExpr,
+        op: BinaryOperator,
+        right: SQLExpr,
+        schema: &DFSchema,
+        ctes: &mut HashMap<String, LogicalPlan>,
+    ) -> Result<Expr> {
+        let operator = match op {
+            BinaryOperator::Eq => Ok(Operator::Eq),
+            BinaryOperator::NotEq => Ok(Operator::NotEq),
+            _ => Err(DataFusionError::NotImplemented(format!(
+                "Unsupported SQL ANY operator {:?}",
+                op
+            ))),
+        }?;
+
+        Ok(Expr::AnyExpr {
+            left: Box::new(self.sql_expr_to_logical_expr(left, schema, ctes)?),
+            op: operator,
+            right: Box::new(self.sql_expr_to_logical_expr(right, schema, ctes)?),
+        })
+    }
+
     fn parse_sql_binary_op(
         &self,
         left: SQLExpr,
@@ -1475,6 +1499,19 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
         schema: &DFSchema,
         ctes: &mut HashMap<String, LogicalPlan>,
     ) -> Result<Expr> {
+        match right {
+            SQLExpr::AnyOp(any_expr) => {
+                return self.parse_sql_binary_any(left, op, *any_expr, schema, ctes);
+            }
+            SQLExpr::AllOp(_) => {
+                return Err(DataFusionError::NotImplemented(format!(
+                    "Unsupported SQL ALL operator {:?}",
+                    right
+                )));
+            }
+            _ => {}
+        };
+
         let operator = match op {
             BinaryOperator::Gt => Ok(Operator::Gt),
             BinaryOperator::GtEq => Ok(Operator::GtEq),

datafusion/core/src/sql/utils.rs

@@ -385,6 +385,11 @@ where
                 op: *op,
                 right: Box::new(clone_with_replacement(&**right, replacement_fn)?),
             }),
+            Expr::AnyExpr { left, right, op } => Ok(Expr::AnyExpr {
+                left: Box::new(clone_with_replacement(&**left, replacement_fn)?),
+                op: *op,
+                right: Box::new(clone_with_replacement(&**right, replacement_fn)?),
+            }),
             Expr::Case {
                 expr: case_expr_opt,
                 when_then_expr,

datafusion/core/tests/sql/expr.rs

@@ -955,6 +955,30 @@ async fn test_extract_date_part() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn test_binary_any() -> Result<()> {
+    // =
+    test_expression!("1 = ANY([1, 2])", "true");
+    test_expression!("3 = ANY([1, 2])", "false");
+    test_expression!("NULL = ANY([1, 2])", "NULL");
+    // utf8
+    test_expression!("'a' = ANY(['a', 'b'])", "true");
+    test_expression!("'c' = ANY(['a', 'b'])", "false");
+    // bool
+    test_expression!("true = ANY([true, false])", "true");
+    test_expression!("false = ANY([true, false])", "true");
+    test_expression!("false = ANY([true, true])", "false");
+    // <>
+    test_expression!("3 <> ANY([1, 2])", "true");
+    test_expression!("1 <> ANY([1, 2])", "false");
+    test_expression!("2 <> ANY([1, 2])", "false");
+    test_expression!("NULL = ANY([1, 2])", "NULL");
+    test_expression!("'c' <> ANY(['a', 'b'])", "true");
+    test_expression!("'a' <> ANY(['a', 'b'])", "false");
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_in_list_scalar() -> Result<()> {
     test_expression!("'a' IN ('a','b')", "true");

datafusion/expr/src/expr.rs

@@ -99,6 +99,15 @@ pub enum Expr {
         /// Right-hand side of the expression
         right: Box<Expr>,
     },
+    /// A binary expression such as "age > 21"
+    AnyExpr {
+        /// Left-hand side of the expression
+        left: Box<Expr>,
+        /// The comparison operator
+        op: Operator,
+        /// Right-hand side of the expression
+        right: Box<Expr>,
+    },
     /// Negation of an expression. The expression's type must be a boolean to make sense.
     Not(Box<Expr>),
     /// Whether an expression is not Null. This expression is never null.
@@ -490,6 +499,9 @@ impl fmt::Debug for Expr {
             Expr::BinaryExpr { left, op, right } => {
                 write!(f, "{:?} {} {:?}", left, op, right)
             }
+            Expr::AnyExpr { left, op, right } => {
+                write!(f, "{:?} {} ANY({:?})", left, op, right)
+            }
             Expr::Sort {
                 expr,
                 asc,
@@ -673,6 +685,11 @@ fn create_name(e: &Expr, input_schema: &DFSchema) -> Result<String> {
             let right = create_name(right, input_schema)?;
             Ok(format!("{} {} {}", left, op, right))
         }
+        Expr::AnyExpr { left, op, right } => {
+            let left = create_name(left, input_schema)?;
+            let right = create_name(right, input_schema)?;
+            Ok(format!("{} {} ANY({})", left, op, right))
+        }
         Expr::Case {
             expr,
             when_then_expr,

datafusion/expr/src/expr_schema.rs

@@ -104,6 +104,7 @@ impl ExprSchemable for Expr {
             | Expr::InSubquery { .. }
             | Expr::Between { .. }
             | Expr::InList { .. }
+            | Expr::AnyExpr { .. }
             | Expr::IsNotNull(_) => Ok(DataType::Boolean),
             Expr::ScalarSubquery(subquery) => {
                 Ok(subquery.subquery.schema().field(0).data_type().clone())
@@ -191,6 +192,11 @@ impl ExprSchemable for Expr {
                 ref right,
                 ..
             } => Ok(left.nullable(input_schema)? || right.nullable(input_schema)?),
+            Expr::AnyExpr {
+                ref left,
+                ref right,
+                ..
+            } => Ok(left.nullable(input_schema)? || right.nullable(input_schema)?),
             Expr::Wildcard => Err(DataFusionError::Internal(
                 "Wildcard expressions are not valid in a logical query plan".to_owned(),
             )),

datafusion/expr/src/expr_visitor.rs

@@ -128,6 +128,10 @@ impl ExprVisitable for Expr {
                 let visitor = left.accept(visitor)?;
                 right.accept(visitor)
             }
+            Expr::AnyExpr { left, right, .. } => {
+                let visitor = left.accept(visitor)?;
+                right.accept(visitor)
+            }
             Expr::Between {
                 expr, low, high, ..
             } => {

datafusion/expr/src/utils.rs

@@ -49,6 +49,7 @@ impl ExpressionVisitor for ColumnNameVisitor<'_> {
             Expr::Alias(_, _)
             | Expr::Literal(_)
             | Expr::BinaryExpr { .. }
+            | Expr::AnyExpr { .. }
             | Expr::Not(_)
             | Expr::IsNotNull(_)
             | Expr::IsNull(_)