feat: add ANY/ALL/SOME(array) operator to support get_indexes in sqlalchemy

e2e_test/batch/basic/all_any_some.slt.part

@@ -0,0 +1,108 @@
+statement ok
+SET RW_IMPLICIT_FLUSH TO true;
+
+query T
+select 1 = SOME(null);
+----
+NULL
+
+query T
+select 2 < ANY(array[null::integer]);
+----
+NULL
+
+query T
+select 3 >= ALL(array[1, 2, null])
+----
+NULL
+
+query T
+select 'a' not like ANY(array['a', null]);
+----
+NULL
+
+query T
+select 'a' not like SOME(array['b', null]);
+----
+t
+
+query T
+select 1 < SOME(array[1] || array[2]);
+----
+t
+
+query T
+select 1 = ALL((select array[1]) || array[2]);
+----
+f
+
+query T
+select 1 = ANY((select array[1]) || array[2]);
+----
+t
+
+# `current_schemas(null)` will return null.
+query T
+select 1 < ANY(current_schemas(null)::integer[]);
+----
+NULL
+
+query T
+select 1 <> ANY(array[1, 2]);
+----
+t
+
+query T
+select 1 <> ALL(array[1, 2]);
+----
+f
+
+query T
+select 1 <> ANY(array[2, 3]);
+----
+t
+
+query T
+select 1 <> ALL(array[2, 3]);
+----
+t
+
+statement error
+select 1 = SOME(1);
+
+statement error
+select 1 < ANY(array[null])
+
+statement ok
+create table tmp(id int, vec int[]);
+
+statement ok
+insert into tmp values(1, array[2,3]);
+
+statement ok
+insert into tmp values(2, array[2,3]);
+
+statement ok
+create materialized view mv1 as select id, vec from tmp where id = any(vec);
+
+query TT
+select id, vec from mv1;
+----
+2 {2,3}
+
+statement ok
+create materialized view mv2 as select id, vec from tmp where id <> ALL(vec);
+
+query TT
+select id, vec from mv2;
+----
+1 {2,3}
+
+statement ok
+drop materialized view mv1;
+
+statement ok
+drop materialized view mv2;
+
+statement ok
+drop table tmp;

e2e_test/batch/catalog/pg_attribute.slt.part

@@ -19,9 +19,28 @@ select a.attname, a.atttypid, a.attlen, a.attnum from pg_catalog.pg_class t
 ----
 id2 23 4 1
 
-
 statement ok
 drop view view1;
 
 statement ok
 drop table tmp;
+
+statement ok
+create table tmp(id1 int, id2 int, id3 int, id4 int);
+
+statement ok
+create index tmp_idx on tmp(id2) include(id1, id3);
+
+query TT
+select i.relname, a.attname, ix.indkey from pg_catalog.pg_class t 
+    join pg_catalog.pg_index ix on t.oid = ix.indrelid
+    join pg_catalog.pg_class i on i.oid = ix.indexrelid
+    join pg_catalog.pg_attribute a on t.oid = a.attrelid and a.attnum = ANY(ix.indkey)
+    where t.relname = 'tmp' order by a.attnum;
+----
+tmp_idx id1 {2,1,3}
+tmp_idx id2 {2,1,3}
+tmp_idx id3 {2,1,3}
+
+statement ok
+drop table tmp;

proto/expr.proto

@@ -30,6 +30,8 @@ message ExprNode {
     OR = 22;
     NOT = 23;
     IN = 24;
+    SOME = 25;
+    ALL = 26;
     // bitwise operators
     BITWISE_AND = 31;
     BITWISE_OR = 32;

src/expr/src/expr/build_expr_from_prost.rs

@@ -14,9 +14,10 @@
 
 use risingwave_common::types::DataType;
 use risingwave_common::util::value_encoding::deserialize_datum;
-use risingwave_pb::expr::expr_node::RexNode;
-use risingwave_pb::expr::ExprNode;
+use risingwave_pb::expr::expr_node::{RexNode, Type};
+use risingwave_pb::expr::{ExprNode, FunctionCall, InputRefExpr};
 
+use super::expr_some_all::SomeAllExpression;
 use crate::expr::expr_binary_bytes::{
     new_ltrim_characters, new_repeat, new_rtrim_characters, new_substr_start, new_to_char,
     new_trim_characters,
@@ -246,6 +247,67 @@ pub fn build_to_char_expr(prost: &ExprNode) -> Result<BoxedExpression> {
     }
 }
 
+pub fn build_some_all_expr_prost(prost: &ExprNode) -> Result<BoxedExpression> {
+    let outer_expr_type = prost.get_expr_type().unwrap();
+    let (outer_children, outer_return_type) = get_children_and_return_type(prost)?;
+    ensure!(matches!(outer_return_type, DataType::Boolean));
+
+    let mut inner_expr_type = outer_children[0].get_expr_type().unwrap();
+    let (mut inner_children, mut inner_return_type) =
+        get_children_and_return_type(&outer_children[0])?;
+    let mut stack = vec![];
+    while inner_children.len() != 2 {
+        stack.push((inner_expr_type, inner_return_type));
+        inner_expr_type = inner_children[0].get_expr_type().unwrap();
+        (inner_children, inner_return_type) = get_children_and_return_type(&inner_children[0])?;
+    }
+
+    let [left_child, right_child]: [_; 2] = inner_children.try_into().unwrap();
+    let left_expr = expr_build_from_prost(&left_child)?;
+    let right_expr = expr_build_from_prost(&right_child)?;
+
+    let DataType::List { datatype: right_expr_return_type } = right_expr.return_type() else {
+        bail!("Expect Array Type");
+    };
+
+    let eval_func = {
+        let left_expr_input_ref = ExprNode {
+            expr_type: Type::InputRef as i32,
+            return_type: Some(left_expr.return_type().to_protobuf()),
+            rex_node: Some(RexNode::InputRef(InputRefExpr { column_idx: 0 })),
+        };
+        let right_expr_input_ref = ExprNode {
+            expr_type: Type::InputRef as i32,
+            return_type: Some(right_expr_return_type.to_protobuf()),
+            rex_node: Some(RexNode::InputRef(InputRefExpr { column_idx: 1 })),
+        };
+        let mut root_expr_node = ExprNode {
+            expr_type: inner_expr_type as i32,
+            return_type: Some(inner_return_type.to_protobuf()),
+            rex_node: Some(RexNode::FuncCall(FunctionCall {
+                children: vec![left_expr_input_ref, right_expr_input_ref],
+            })),
+        };
+        while let Some((expr_type, return_type)) = stack.pop() {
+            root_expr_node = ExprNode {
+                expr_type: expr_type as i32,
+                return_type: Some(return_type.to_protobuf()),
+                rex_node: Some(RexNode::FuncCall(FunctionCall {
+                    children: vec![root_expr_node],
+                })),
+            }
+        }
+        expr_build_from_prost(&root_expr_node)?
+    };
+
+    Ok(Box::new(SomeAllExpression::new(
+        left_expr,
+        right_expr,
+        outer_expr_type,
+        eval_func,
+    )))
+}
+
 #[cfg(test)]
 mod tests {
     use std::vec;