feat: implement question/question_pipe/question_and

Author: sunng87

datafusion/physical-expr/src/expressions/binary.rs

@@ -17,7 +17,12 @@
 
 mod kernels;
 
-use crate::expressions::binary::kernels::contains::collection_contains_dyn;
+use crate::expressions::binary::kernels::contains::{
+    collection_contains_all_strings_dyn, collection_contains_all_strings_dyn_scalar,
+    collection_contains_any_string_dyn, collection_contains_any_string_dyn_scalar,
+    collection_contains_dyn, collection_contains_dyn_scalar,
+    collection_contains_string_dyn, collection_contains_string_dyn_scalar,
+};
 use crate::expressions::binary::kernels::select::{
     cast_to_string_array, collection_select_dyn_scalar, collection_select_path_dyn_scalar,
 };
@@ -599,6 +604,11 @@ impl BinaryExpr {
             HashArrow => collection_select_path_dyn_scalar(array, scalar),
             HashLongArrow => collection_select_path_dyn_scalar(array, scalar)
                 .map(|arr| arr.and_then(cast_to_string_array)),
+            AtArrow => collection_contains_dyn_scalar(&array, scalar),
+            // TODO: ArrowAt
+            Question => collection_contains_string_dyn_scalar(&array, scalar),
+            QuestionPipe => collection_contains_any_string_dyn_scalar(&array, scalar),
+            QuestionAnd => collection_contains_all_strings_dyn_scalar(&array, scalar),
             // if scalar operation is not supported - fallback to array implementation
             _ => None,
         };
@@ -654,8 +664,11 @@ impl BinaryExpr {
             StringConcat => concat_elements(left, right),
             AtArrow => collection_contains_dyn(left, right),
             ArrowAt => collection_contains_dyn(right, left),
+            Question => collection_contains_string_dyn(left, right),
+            QuestionPipe => collection_contains_any_string_dyn(left, right),
+            QuestionAnd => collection_contains_all_strings_dyn(left, right),
             Arrow | LongArrow | HashArrow | HashLongArrow | AtAt | HashMinus
-            | AtQuestion | Question | QuestionAnd | QuestionPipe | IntegerDivide => {
+            | AtQuestion | IntegerDivide => {
                 not_impl_err!(
                     "Binary operator '{:?}' is not supported in the physical expr",
                     self.op