refactor(expr): better error handling and reporting for expressions

Cargo.toml

@@ -74,6 +74,7 @@ etcd-client = { package = "madsim-etcd-client", version = "0.3" }
 futures-async-stream = "0.2"
 hytra = "0.1"
 rdkafka = { package = "madsim-rdkafka", version = "=0.2.14-alpha", features = ["cmake-build"] }
+snafu = { version = "0.7", features = ["unstable-provider-api", "rust_1_61", "backtraces-impl-std"] }
 hashbrown = { version = "0.14.0", features = ["ahash", "inline-more", "nightly"] }
 criterion = { version = "0.5", features = ["async_futures"] }
 tonic = { package = "madsim-tonic", version = "0.3.1" }

src/batch/src/executor/aggregation/orderby.rs

@@ -14,7 +14,6 @@
 
 use std::ops::Range;
 
-use anyhow::anyhow;
 use futures_util::FutureExt;
 use risingwave_common::array::{Op, RowRef, StreamChunk};
 use risingwave_common::estimate_size::EstimateSize;
@@ -24,7 +23,7 @@ use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
 use risingwave_common::util::memcmp_encoding;
 use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
 use risingwave_expr::agg::{Aggregator, BoxedAggState};
-use risingwave_expr::{ExprError, Result};
+use risingwave_expr::Result;
 
 /// `ProjectionOrderBy` is a wrapper of `Aggregator` that sorts rows by given columns and then
 /// projects columns.
@@ -65,8 +64,7 @@ impl ProjectionOrderBy {
 
     fn push_row(&mut self, row: RowRef<'_>) -> Result<()> {
         let key =
-            memcmp_encoding::encode_row(row.project(&self.order_col_indices), &self.order_types)
-                .map_err(|e| ExprError::Internal(anyhow!("failed to encode row, error: {}", e)))?;
+            memcmp_encoding::encode_row(row.project(&self.order_col_indices), &self.order_types)?;
         let projected_row = row.project(&self.arg_indices).to_owned_row();
 
         self.unordered_values_estimated_heap_size +=

src/common/Cargo.toml

@@ -74,6 +74,7 @@ serde_default = "0.1"
 serde_json = "1"
 serde_with = "3"
 smallbitset = "0.7.1"
+snafu = { workspace = true }
 speedate = "0.11.0"
 static_assertions = "1"
 strum = "0.25"

src/common/src/cast/bytea.rs

@@ -0,0 +1,185 @@
+// Copyright 2023 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use itertools::Itertools;
+use snafu::{OptionExt, Snafu};
+
+type Result<T> = std::result::Result<T, ByteaCastError>;
+
+/// Error type for bytea cast.
+#[derive(Snafu, Debug)]
+#[snafu(visibility(pub))]
+pub enum ByteaCastError {
+    #[snafu(display("invalid hex digit `{from}`"))]
+    InvalidHexDigit { from: char },
+
+    #[snafu(display("invalid base64 symbol: `{from}`"))]
+    InvalidBase64 { from: char },
+
+    #[snafu(display("invalid format (`{from}`) for bytea: {message}"))]
+    InvalidBytea { from: Box<str>, message: Box<str> },
+}
+
+impl From<ByteaCastError> for crate::error::RwError {
+    fn from(value: ByteaCastError) -> Self {
+        crate::error::ErrorCode::ExprError(value.into()).into()
+    }
+}
+
+/// Refer to PostgreSQL's implementation <https://github.com/postgres/postgres/blob/5cb54fc310fb84287cbdc74533f3420490a2f63a/src/backend/utils/adt/varlena.c#L276-L288>
+pub fn str_to_bytea(elem: &str) -> Result<Box<[u8]>> {
+    if let Some(remainder) = elem.strip_prefix(r"\x") {
+        Ok(parse_bytes_hex(remainder)?.into())
+    } else {
+        Ok(parse_bytes_traditional(elem)?.into())
+    }
+}
+
+/// Ref: <https://docs.rs/hex/0.4.3/src/hex/lib.rs.html#175-185>
+#[inline(always)]
+fn get_hex(c: u8) -> Result<u8> {
+    match c {
+        b'A'..=b'F' => Ok(c - b'A' + 10),
+        b'a'..=b'f' => Ok(c - b'a' + 10),
+        b'0'..=b'9' => Ok(c - b'0'),
+        _ => InvalidHexDigitSnafu { from: c }.fail(),
+    }
+}
+
+/// Refer to <https://www.postgresql.org/docs/current/datatype-binary.html#id-1.5.7.12.10> for specification.
+pub fn parse_bytes_hex(s: &str) -> Result<Vec<u8>> {
+    let mut res = Vec::with_capacity(s.len() / 2);
+
+    let mut bytes = s.bytes();
+    while let Some(c) = bytes.next() {
+        // white spaces are tolerated
+        if c == b' ' || c == b'\n' || c == b'\t' || c == b'\r' {
+            continue;
+        }
+        let v1 = get_hex(c)?;
+
+        let c = bytes.next().context(InvalidByteaSnafu {
+            from: s,
+            message: "odd number of digits in hex format",
+        })?;
+
+        let v2 = get_hex(c)?;
+        res.push((v1 << 4) | v2);
+    }
+
+    Ok(res)
+}
+
+/// Refer to <https://www.postgresql.org/docs/current/datatype-binary.html#id-1.5.7.12.10> for specification.
+pub fn parse_bytes_traditional(s: &str) -> Result<Vec<u8>> {
+    let context = InvalidByteaSnafu {
+        from: s,
+        message: "invalid input syntax for escape format",
+    };
+
+    let mut bytes = s.bytes();
+
+    let mut res = Vec::new();
+    while let Some(b) = bytes.next() {
+        if b != b'\\' {
+            res.push(b);
+        } else {
+            match bytes.next() {
+                Some(b'\\') => {
+                    res.push(b'\\');
+                }
+                Some(b1 @ b'0'..=b'3') => match bytes.next_tuple() {
+                    Some((b2 @ b'0'..=b'7', b3 @ b'0'..=b'7')) => {
+                        res.push(((b1 - b'0') << 6) + ((b2 - b'0') << 3) + (b3 - b'0'));
+                    }
+                    _ => {
+                        // one backslash, not followed by another or ### valid octal
+                        return context.fail();
+                    }
+                },
+                _ => {
+                    // one backslash, not followed by another or ### valid octal
+                    return context.fail();
+                }
+            }
+        }
+    }
+
+    Ok(res)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_bytea() {
+        use crate::types::ToText;
+        assert_eq!(str_to_bytea("fgo").unwrap().as_ref().to_text(), r"\x66676f");
+        assert_eq!(
+            str_to_bytea(r"\xDeadBeef").unwrap().as_ref().to_text(),
+            r"\xdeadbeef"
+        );
+        assert_eq!(
+            str_to_bytea("12CD").unwrap().as_ref().to_text(),
+            r"\x31324344"
+        );
+        assert_eq!(
+            str_to_bytea("1234").unwrap().as_ref().to_text(),
+            r"\x31323334"
+        );
+        assert_eq!(
+            str_to_bytea(r"\x12CD").unwrap().as_ref().to_text(),
+            r"\x12cd"
+        );
+        assert_eq!(
+            str_to_bytea(r"\x De Ad Be Ef ").unwrap().as_ref().to_text(),
+            r"\xdeadbeef"
+        );
+        assert_eq!(
+            str_to_bytea("x De Ad Be Ef ").unwrap().as_ref().to_text(),
+            r"\x7820446520416420426520456620"
+        );
+        assert_eq!(
+            str_to_bytea(r"De\\123dBeEf").unwrap().as_ref().to_text(),
+            r"\x44655c3132336442654566"
+        );
+        assert_eq!(
+            str_to_bytea(r"De\123dBeEf").unwrap().as_ref().to_text(),
+            r"\x4465536442654566"
+        );
+        assert_eq!(
+            str_to_bytea(r"De\\000dBeEf").unwrap().as_ref().to_text(),
+            r"\x44655c3030306442654566"
+        );
+
+        assert_eq!(str_to_bytea(r"\123").unwrap().as_ref().to_text(), r"\x53");
+        assert_eq!(str_to_bytea(r"\\").unwrap().as_ref().to_text(), r"\x5c");
+        assert_eq!(
+            str_to_bytea(r"123").unwrap().as_ref().to_text(),
+            r"\x313233"
+        );
+        assert_eq!(
+            str_to_bytea(r"\\123").unwrap().as_ref().to_text(),
+            r"\x5c313233"
+        );
+
+        assert!(str_to_bytea(r"\1").is_err());
+        assert!(str_to_bytea(r"\12").is_err());
+        assert!(str_to_bytea(r"\400").is_err());
+        assert!(str_to_bytea(r"\378").is_err());
+        assert!(str_to_bytea(r"\387").is_err());
+        assert!(str_to_bytea(r"\377").is_ok());
+    }
+}