Merge branch 'main' of https://github.com/risingwavelabs/risingwave into li0k/storage_intra_picker

Author: Li0k

Cargo.lock

@@ -0,0 +1,72 @@
+query I
+select array_max(array[1, 2, 3]);
+----
+3
+
+query I
+select array_max(array[2, 3, 5, 2, 4]);
+----
+5
+
+query I
+select array_max(array[114514, 114513]);
+----
+114514
+
+query I
+select array_max(array['a', 'b', 'c', 'a']);
+----
+c
+
+query I
+select array_max(array['e💩a', 'f🤔️b', 'c🥵c', 'd🥳d', 'e💩e']);
+----
+f🤔️b
+
+query I
+select array_max(array['2c😅🤔😅️c2', '114🥵514', '30🤣🥳03', '5🥵💩💩🥵5']);
+----
+5🥵💩💩🥵5
+
+query error invalid digit found in string
+select array_max(array['a', 'b', 'c', 114514]);
+
+query error invalid digit found in string
+select array_max(array[114514, 'a', 'b', 'c']);
+
+# i32::MIN & i32::MIN - 1 & i32::MAX
+query I
+select array_max(array[-2147483648, 2147483647, -2147483649]);
+----
+2147483647
+
+# i64::MIN & i64::MIN - 1 & i64::MAX
+query I
+select array_max(array[-9223372036854775808, 9223372036854775807, -9223372036854775809]);
+----
+9223372036854775807
+
+query I
+select array_max(array['a', '', 'c']);
+----
+c
+
+query I
+select array_max(array[3.14, 1.14, 1.14514]);
+----
+3.14
+
+query I
+select array_max(array[3.1415926, 191.14, 114514, 1313.1414]);
+----
+114514
+
+query I
+select array_max(array[1e-4, 1.14514e5, 1.14514e-5]);
+----
+114514
+
+query I
+select array_max(array[date'2002-10-30', date'2023-09-06', date'2017-06-18']);
+----
+2023-09-06

e2e_test/batch/functions/array_max.slt.part

@@ -12,7 +12,7 @@ statement ok
 create materialized view mv as select * from t emit on window close;
 
 statement ok
-insert into t values ('2023-05-06 16:51:00', 1);
+insert into t values ('2023-05-06 16:51:00', 1), ('2023-05-06 16:51:00', 2), ('2023-05-06 16:51:00', 3);
 
 statement ok
 insert into t values ('2023-05-06 16:56:01', 1);
@@ -25,6 +25,8 @@ query TI
 select * from mv;
 ----
 2023-05-06 16:51:00 1
+2023-05-06 16:51:00 2
+2023-05-06 16:51:00 3
 
 statement ok
 drop materialized view mv;

e2e_test/streaming/watermark.slt

@@ -198,6 +198,7 @@ message ExprNode {
     ARRAY_DIMS = 544;
     ARRAY_TRANSFORM = 545;
     ARRAY_MIN = 546;
+    ARRAY_MAX = 547;
 
     // Int256 functions
     HEX_TO_INT256 = 560;

proto/expr.proto

@@ -202,6 +202,9 @@ mod tests {
                 ArrayMin: [
                     "array_min(list) -> bytea/varchar/timestamptz/timestamp/time/date/int256/serial/decimal/float32/float64/int16/int32/int64",
                 ],
+                ArrayMax: [
+                    "array_max(list) -> bytea/varchar/timestamptz/timestamp/time/date/int256/serial/decimal/float32/float64/int16/int32/int64",
+                ],
             }
         "#]];
         expected.assert_debug_eq(&duplicated);

src/expr/src/sig/func.rs

@@ -18,6 +18,8 @@ use risingwave_expr_macro::function;
 
 use crate::Result;
 
+/// FIXME: #[`function("array_min(list`) -> any")] supports
+/// In this way we could avoid manual macro expansion
 #[function("array_min(list) -> *int")]
 #[function("array_min(list) -> *float")]
 #[function("array_min(list) -> decimal")]
@@ -36,3 +38,22 @@ pub fn array_min<T: Scalar>(list: ListRef<'_>) -> Result<Option<T>> {
         None => Ok(None),
     }
 }
+
+#[function("array_max(list) -> *int")]
+#[function("array_max(list) -> *float")]
+#[function("array_max(list) -> decimal")]
+#[function("array_max(list) -> serial")]
+#[function("array_max(list) -> int256")]
+#[function("array_max(list) -> date")]
+#[function("array_max(list) -> time")]
+#[function("array_max(list) -> timestamp")]
+#[function("array_max(list) -> timestamptz")]
+#[function("array_max(list) -> varchar")]
+#[function("array_max(list) -> bytea")]
+pub fn array_max<T: Scalar>(list: ListRef<'_>) -> Result<Option<T>> {
+    let max_value = list.iter().flatten().map(DefaultOrdered).max();
+    match max_value.map(|v| v.0).to_owned_datum() {
+        Some(s) => Ok(Some(s.try_into()?)),
+        None => Ok(None),
+    }
+}

src/expr/src/vector_op/array_min.rs renamed to src/expr/src/vector_op/array_min_max.rs

@@ -16,7 +16,7 @@ pub mod arithmetic_op;
 pub mod array_access;
 pub mod array_distinct;
 pub mod array_length;
-pub mod array_min;
+pub mod array_min_max;
 pub mod array_positions;
 pub mod array_range_access;
 pub mod array_remove;

src/expr/src/vector_op/mod.rs

@@ -793,6 +793,7 @@ impl Binder {
                 ("cardinality", raw_call(ExprType::Cardinality)),
                 ("array_remove", raw_call(ExprType::ArrayRemove)),
                 ("array_replace", raw_call(ExprType::ArrayReplace)),
+                ("array_max", raw_call(ExprType::ArrayMax)),
                 ("array_position", raw_call(ExprType::ArrayPosition)),
                 ("array_positions", raw_call(ExprType::ArrayPositions)),
                 ("trim_array", raw_call(ExprType::TrimArray)),

src/frontend/src/binder/expr/function.rs

@@ -153,6 +153,7 @@ impl ExprVisitor<bool> for ImpureAnalyzer {
             | expr_node::Type::Row
             | expr_node::Type::ArrayToString
             | expr_node::Type::ArrayCat
+            | expr_node::Type::ArrayMax
             | expr_node::Type::ArrayAppend
             | expr_node::Type::ArrayPrepend
             | expr_node::Type::FormatType

src/frontend/src/expr/pure.rs

@@ -613,6 +613,12 @@ fn infer_type_for_special(
             }
             Ok(Some(DataType::Varchar))
         }
+        ExprType::ArrayMax => {
+            ensure_arity!("array_max", | inputs | == 1);
+            inputs[0].ensure_array_type()?;
+
+            Ok(Some(inputs[0].return_type().as_list().clone()))
+        }
         ExprType::StringToArray => {
             ensure_arity!("string_to_array", 2 <= | inputs | <= 3);
 

src/frontend/src/expr/type_inference/func.rs

@@ -141,19 +141,19 @@ impl<S: StateStore> SortBuffer<S> {
         watermark: ScalarImpl,
         buffer_table: &'a mut StateTable<S>,
     ) {
-        let mut last_timestamp = None;
+        let mut last_table_pk = None;
         loop {
             if !self.cache.is_synced() {
                 // Refill the cache, then consume from the cache, to ensure strong row ordering
                 // and prefetch for the next watermark.
-                self.refill_cache(last_timestamp.take(), buffer_table)
+                self.refill_cache(last_table_pk.take(), buffer_table)
                     .await?;
             }
 
             #[for_await]
             for res in self.consume_from_cache(watermark.as_scalar_ref_impl()) {
-                let ((timestamp_val, _), row) = res?;
-                last_timestamp = Some(timestamp_val.into_inner());
+                let row = res?;
+                last_table_pk = Some((&row).project(buffer_table.pk_indices()).into_owned_row());
                 yield row;
             }
 
@@ -169,15 +169,15 @@ impl<S: StateStore> SortBuffer<S> {
         buffer_table.update_watermark(watermark, true);
     }
 
-    #[try_stream(ok = (CacheKey, OwnedRow), error = StreamExecutorError)]
+    #[try_stream(ok = OwnedRow, error = StreamExecutorError)]
     async fn consume_from_cache<'a>(&'a mut self, watermark: ScalarRefImpl<'a>) {
         while self.cache.is_synced() {
             let Some(key) = self.cache.first_key_value().map(|(k, _)| k.clone()) else {
                 break;
             };
             if key.0.as_scalar_ref_impl().default_cmp(&watermark).is_lt() {
                 let row = self.cache.delete(&key).unwrap();
-                yield (key, row);
+                yield row;
             } else {
                 break;
             }
@@ -187,15 +187,14 @@ impl<S: StateStore> SortBuffer<S> {
     /// Clear the cache and refill it with the current content of the buffer table.
     pub async fn refill_cache(
         &mut self,
-        last_timestamp: Option<ScalarImpl>,
+        last_table_pk: Option<OwnedRow>,
         buffer_table: &StateTable<S>,
     ) -> StreamExecutorResult<()> {
         let mut filler = self.cache.begin_syncing();
 
         let pk_range = (
-            last_timestamp
-                .as_ref()
-                .map(|v| Bound::Excluded([Some(v.as_scalar_ref_impl())]))
+            last_table_pk
+                .map(Bound::Excluded)
                 .unwrap_or(Bound::Unbounded),
             Bound::<row::Empty>::Unbounded,
         );

src/stream/src/executor/sort_buffer.rs

@@ -187,7 +187,7 @@ impl TestSuite {
             );
         }
 
-        let timestamptz = DateTime::<Utc>::from_utc(
+        let timestamptz = DateTime::<Utc>::from_naive_utc_and_offset(
             NaiveDate::from_ymd_opt(2022, 1, 1)
                 .unwrap()
                 .and_hms_opt(10, 0, 0)