feat: support for AnyExpression

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

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

datafusion/core/src/optimizer/common_subexpr_eliminate.rs

@@ -404,6 +404,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

@@ -400,6 +400,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

@@ -61,6 +61,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, .. }
@@ -154,6 +157,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_expr::utils::generate_sort_key;
@@ -109,6 +116,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,
@@ -1182,7 +1194,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()
@@ -1245,6 +1256,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/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.
@@ -302,6 +311,7 @@ impl Expr {
             Expr::AggregateUDF { .. } => "AggregateUDF",
             Expr::Alias(..) => "Alias",
             Expr::Between { .. } => "Between",
+            Expr::AnyExpr { .. } => "AnyExpr",
             Expr::BinaryExpr { .. } => "BinaryExpr",
             Expr::Case { .. } => "Case",
             Expr::Cast { .. } => "Cast",
@@ -524,6 +534,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,
@@ -707,6 +720,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_rewriter.rs

@@ -127,6 +127,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/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

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