feat: support sql func encrypt/decrypt

e2e_test/streaming/encrypt.slt

@@ -0,0 +1,26 @@
+query T
+SELECT encrypt('\x00112233445566778899aabbccddeeff','\x000102030405060708090a0b0c0d0e0f','aes-ecb/pad:none')
+----
+\x69c4e0d86a7b0430d8cdb78070b4c55a
+
+query T
+SELECT encrypt('\x00112233445566778899aabbccddeeff','\x000102030405060708090a0b0c0d0e0f1011121314151617','aes-ecb/pad:none');
+----
+\xdda97ca4864cdfe06eaf70a0ec0d7191
+
+query T
+SELECT encrypt('\x00112233445566778899aabbccddeeff','\x000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f','aes-ecb/pad:none')
+----
+\x8ea2b7ca516745bfeafc49904b496089
+
+query T
+SELECT encrypt('\x00112233445566778899aabbccddeeff','\x000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f','aes-cbc/pad:none')
+----
+\x8ea2b7ca516745bfeafc49904b496089
+
+# input not multiple of block size when not allowing padding
+statement error
+SELECT encrypt('\x00112233445566778899aabbccddeeff00','\x000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f','aes-cbc/pad:none')
+
+
+

proto/expr.proto

@@ -259,6 +259,10 @@ message ExprNode {
     JSONB_PATH_QUERY_ARRAY = 622;
     JSONB_PATH_QUERY_FIRST = 623;
 
+    // encrypt and decrypt functions
+    DECRYPT = 700;
+    ENCRYPT = 701;
+
     // Non-pure functions below (> 1000)
     // ------------------------
     // Internal functions

src/expr/core/Cargo.toml

@@ -39,6 +39,7 @@ futures-util = "0.3"
 itertools = "0.12"
 moka = { version = "0.12", features = ["future"] }
 num-traits = "0.2"
+openssl = { version = "0.10", features = ["vendored"] }
 parse-display = "0.8"
 paste = "1"
 risingwave_common = { workspace = true }

src/expr/core/src/error.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::fmt::Display;
+use std::fmt::{Debug, Display};
 
 use risingwave_common::array::{ArrayError, ArrayRef};
 use risingwave_common::error::{ErrorCode, RwError};
@@ -117,6 +117,23 @@ pub enum ExprError {
 
     #[error("invalid state: {0}")]
     InvalidState(String),
+
+    #[error("error in cryptography: {0}")]
+    Cryptography(Box<CryptographyError>),
+}
+
+#[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,
 }
 
 static_assertions::const_assert_eq!(std::mem::size_of::<ExprError>(), 40);

src/expr/core/src/lib.rs

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

src/expr/impl/Cargo.toml

@@ -34,6 +34,7 @@ itertools = "0.12"
 jsonbb = "0.1.2"
 md5 = "0.7"
 num-traits = "0.2"
+openssl = { version = "0.10", features = ["vendored"] }
 regex = "1"
 risingwave_common = { workspace = true }
 risingwave_expr = { workspace = true }