Merge branch 'main' of https://github.com/risingwavelabs/risingwave i…

…nto li0k/test_builder_oom

src/expr/core/Cargo.toml

@@ -39,7 +39,6 @@ futures-util = "0.3"
 itertools = { workspace = true }
 linkme = { version = "0.3", features = ["used_linker"] }
 num-traits = "0.2"
-openssl = { version = "0.10", features = ["vendored"] }
 parse-display = "0.9"
 paste = "1"
 prometheus = "0.13"

src/expr/core/src/error.rs

@@ -110,26 +110,13 @@ pub enum ExprError {
     #[error("invalid state: {0}")]
     InvalidState(String),
 
-    #[error("error in cryptography: {0}")]
-    Cryptography(Box<CryptographyError>),
-
     /// Function error message returned by UDF.
     #[error("{0}")]
     Custom(String),
-}
 
-#[derive(Debug)]
-pub enum CryptographyStage {
-    Encrypt,
-    Decrypt,
-}
-
-#[derive(Debug, Error)]
-#[error("{stage:?} stage, reason: {reason}")]
-pub struct CryptographyError {
-    pub stage: CryptographyStage,
-    #[source]
-    pub reason: openssl::error::ErrorStack,
+    /// Error from a function call.
+    #[error("{0}")]
+    Function(#[source] Box<dyn std::error::Error + Send + Sync>),
 }
 
 static_assertions::const_assert_eq!(std::mem::size_of::<ExprError>(), 40);

src/expr/core/src/lib.rs

@@ -35,6 +35,6 @@ pub mod sig;
 pub mod table_function;
 pub mod window_function;
 
-pub use error::{ContextUnavailable, CryptographyError, CryptographyStage, ExprError, Result};
+pub use error::{ContextUnavailable, ExprError, Result};
 pub use risingwave_common::{bail, ensure};
 pub use risingwave_expr_macro::*;

src/expr/impl/src/scalar/encrypt.rs

@@ -18,7 +18,7 @@ use std::sync::LazyLock;
 use openssl::error::ErrorStack;
 use openssl::symm::{Cipher, Crypter, Mode as CipherMode};
 use regex::Regex;
-use risingwave_expr::{function, CryptographyError, CryptographyStage, ExprError, Result};
+use risingwave_expr::{function, ExprError, Result};
 
 #[derive(Debug, Clone, PartialEq)]
 enum Algorithm {
@@ -129,14 +129,13 @@ impl CipherConfig {
         })
     }
 
-    fn eval(&self, input: &[u8], stage: CryptographyStage) -> Result<Box<[u8]>> {
+    fn eval(&self, input: &[u8], stage: CryptographyStage) -> Result<Box<[u8]>, CryptographyError> {
         let operation = match stage {
             CryptographyStage::Encrypt => CipherMode::Encrypt,
             CryptographyStage::Decrypt => CipherMode::Decrypt,
         };
-        self.eval_inner(input, operation).map_err(|reason| {
-            ExprError::Cryptography(Box::new(CryptographyError { stage, reason }))
-        })
+        self.eval_inner(input, operation)
+            .map_err(|reason| CryptographyError { stage, reason })
     }
 
     fn eval_inner(
@@ -163,18 +162,32 @@ impl CipherConfig {
     "decrypt(bytea, bytea, varchar) -> bytea",
     prebuild = "CipherConfig::parse_cipher_config($1, $2)?"
 )]
-pub fn decrypt(data: &[u8], config: &CipherConfig) -> Result<Box<[u8]>> {
+fn decrypt(data: &[u8], config: &CipherConfig) -> Result<Box<[u8]>, CryptographyError> {
     config.eval(data, CryptographyStage::Decrypt)
 }
 
 #[function(
     "encrypt(bytea, bytea, varchar) -> bytea",
     prebuild = "CipherConfig::parse_cipher_config($1, $2)?"
 )]
-pub fn encrypt(data: &[u8], config: &CipherConfig) -> Result<Box<[u8]>> {
+fn encrypt(data: &[u8], config: &CipherConfig) -> Result<Box<[u8]>, CryptographyError> {
     config.eval(data, CryptographyStage::Encrypt)
 }
 
+#[derive(Debug)]
+enum CryptographyStage {
+    Encrypt,
+    Decrypt,
+}
+
+#[derive(Debug, thiserror::Error)]
+#[error("{stage:?} stage, reason: {reason}")]
+struct CryptographyError {
+    pub stage: CryptographyStage,
+    #[source]
+    pub reason: openssl::error::ErrorStack,
+}
+
 #[cfg(test)]
 mod test {
     use super::*;
@@ -197,24 +210,23 @@ mod test {
 
     #[test]
     fn encrypt_testcase() {
-        let encrypt_wrapper = |data: &[u8], key: &[u8], mode: &str| -> Result<Box<[u8]>> {
-            let config = CipherConfig::parse_cipher_config(key, mode)?;
-            encrypt(data, &config)
+        let encrypt_wrapper = |data: &[u8], key: &[u8], mode: &str| -> Box<[u8]> {
+            let config = CipherConfig::parse_cipher_config(key, mode).unwrap();
+            encrypt(data, &config).unwrap()
         };
-        let decrypt_wrapper = |data: &[u8], key: &[u8], mode: &str| -> Result<Box<[u8]>> {
-            let config = CipherConfig::parse_cipher_config(key, mode)?;
-            decrypt(data, &config)
+        let decrypt_wrapper = |data: &[u8], key: &[u8], mode: &str| -> Box<[u8]> {
+            let config = CipherConfig::parse_cipher_config(key, mode).unwrap();
+            decrypt(data, &config).unwrap()
         };
         let key = b"\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f";
 
         let encrypted = encrypt_wrapper(
             b"\x00\x11\x22\x33\x44\x55\x66\x77\x88\x99\xaa\xbb\xcc\xdd\xee\xff",
             key,
             "aes-ecb/pad:none",
-        )
-        .unwrap();
+        );
 
-        let decrypted = decrypt_wrapper(&encrypted, key, "aes-ecb/pad:none").unwrap();
+        let decrypted = decrypt_wrapper(&encrypted, key, "aes-ecb/pad:none");
         assert_eq!(
             decrypted,
             (*b"\x00\x11\x22\x33\x44\x55\x66\x77\x88\x99\xaa\xbb\xcc\xdd\xee\xff").into()

src/expr/macro/src/gen.rs

@@ -365,13 +365,13 @@ impl FunctionAttr {
             ReturnTypeKind::Result => quote! {
                 match #output {
                     Ok(x) => Some(x),
-                    Err(e) => { errors.push(e); None }
+                    Err(e) => { errors.push(ExprError::Function(Box::new(e))); None }
                 }
             },
             ReturnTypeKind::ResultOption => quote! {
                 match #output {
                     Ok(x) => x,
-                    Err(e) => { errors.push(e); None }
+                    Err(e) => { errors.push(ExprError::Function(Box::new(e))); None }
                 }
             },
         };