Repository: ArroyoSystems/arroyo (language: rust)

Issue:
Implement DataFusion Hashing Functions
DataFusion [supports](https://arrow.apache.org/datafusion/user-guide/sql/scalar_functions.html#hashing-functions) md5, sha224, sha256, sha384, sha512.


Discussion / hints:
I have been working on this.

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index 465e70155..eb8b08a32 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -627,6 +627,7 @@ dependencies = [
  "arroyo-sql-macro",
  "arroyo-types",
  "chrono",
+ "hex",
  "quote",
  "syn 1.0.109",
 ]
@@ -665,8 +666,12 @@ dependencies = [
 name = "arroyo-types"
 version = "0.1.0"
 dependencies = [
+ "base64 0.21.0",
  "bincode 2.0.0-rc.3",
+ "hex",
+ "md-5 0.10.5",
  "serde",
+ "sha2 0.10.6",
 ]
 
 [[package]]
diff --git a/arroyo-sql-macro/src/lib.rs b/arroyo-sql-macro/src/lib.rs
index 7c0706f95..e20ceb3ec 100644
--- a/arroyo-sql-macro/src/lib.rs
+++ b/arroyo-sql-macro/src/lib.rs
@@ -4,6 +4,32 @@ use quote::quote;
 use syn::parse::{Parse, ParseStream};
 use syn::{parse_macro_input, Expr, LitStr, Token};
 
+/// This macro is used to generate a test function for a single test case.
+/// Used in the `arroyo-sql-testing` crate.
+/// 
+/// # Arguments
+/// 
+/// * `test_name` - The name of the test.
+/// * `calculation_string` - The calculation string to be tested.
+/// * `input_value` - The input value to be used in the calculation.
+/// * `expected_result` - The expected result of the calculation.
+/// 
+/// # Returns
+/// 
+/// A test function that can be used to test a single test case.
+/// 
+/// # Example
+/// 
+/// ```
+/// use arroyo_sql_testing::single_test_codegen;
+/// 
+/// single_test_codegen!(
+///    "test_name",
+///   "calculation_string",
+///   1,
+///   2
+/// );
+/// ```
 #[proc_macro]
 pub fn single_test_codegen(input: TokenStream) -> TokenStream {
     let test_case = parse_macro_input!(input as TestCase);
diff --git a/arroyo-sql/src/expressions.rs b/arroyo-sql/src/expressions.rs
index d836b770f..5b988e385 100644
--- a/arroyo-sql/src/expressions.rs
+++ b/arroyo-sql/src/expressions.rs
@@ -40,10 +40,11 @@ pub enum Expression {
     Cast(CastExpression),
     Numeric(NumericExpression),
     String(StringFunction),
+    Hash(HashExpression),
 }
 
-impl Expression {
-    pub fn to_syn_expression(&self) -> syn::Expr {
+impl ExpressionGenerator for Expression {
+    fn to_syn_expression(&self) -> syn::Expr {
         match self {
             Expression::Column(column_expression) => column_expression.to_syn_expression(),
             Expression::UnaryBoolean(unary_boolean_expression) => {
@@ -63,9 +64,11 @@ impl Expression {
             Expression::Cast(cast_expression) => cast_expression.to_syn_expression(),
             Expression::Numeric(numeric_expression) => numeric_expression.to_syn_expression(),
             Expression::String(string_function) => string_function.to_syn_expression(),
+            Expression::Hash(hash_expression) => hash_expression.to_syn_expression(),
         }
     }
-    pub fn return_type(&self) -> TypeDef {
+
+    fn return_type(&self) -> TypeDef {
         match self {
             Expression::Column(column_expression) => column_expression.return_type(),
             Expression::UnaryBoolean(unary_boolean_expression) => {
@@ -83,12 +86,12 @@ impl Expression {
             Expression::Cast(cast_expression) => cast_expression.return_type(),
             Expression::Numeric(numeric_expression) => numeric_expression.return_type(),
             Expression::String(string_function) => string_function.return_type(),
+            Expression::Hash(hash_expression) => hash_expression.return_type(),
         }
     }
-    pub fn nullable(&self) -> bool {
-        self.return_type().is_optional()
-    }
+}
 
+impl Expression {
     pub(crate) fn has_max_value(&self, field: &StructField) -> Option<i64> {
         match self {
             Expression::BinaryComparison(BinaryComparisonExpression { left, op, right }) => {
@@ -364,9 +367,10 @@ pub fn to_expression_generator(expression: &Expr, input_struct: &StructDef) -> R
                 | BuiltinScalarFunction::SHA224
                 | BuiltinScalarFunction::SHA256
                 | BuiltinScalarFunction::SHA384
-                | BuiltinScalarFunction::SHA512 => {
-                    bail!("hashing function {:?} not implemented", fun)
-                }
+                | BuiltinScalarFunction::SHA512 => Ok(HashExpression::new(
+                    fun.clone(),
+                    Box::new(arg_expressions.remove(0)),
+                )?),
                 BuiltinScalarFunction::ToHex => bail!("hex not implemented"),
                 BuiltinScalarFunction::Uuid => bail!("UUID unimplemented"),
                 BuiltinScalarFunction::Cbrt => bail!("cube root unimplemented"),
@@ -1123,6 +1127,7 @@ impl SortExpression {
             nulls_first,
         })
     }
+
     pub fn tuple_type(&self) -> syn::Type {
         let value_type = self.value.return_type().return_type();
         match (self.value.nullable(), &self.direction, self.nulls_first) {
@@ -1138,6 +1143,7 @@ impl SortExpression {
             }
         }
     }
+
     pub fn to_syn_expr(&self) -> syn::Expr {
         let value_expr = self.value.to_syn_expression();
         match (self.value.nullable(), &self.direction, self.nulls_first) {
@@ -1158,6 +1164,7 @@ impl SortExpression {
         }
     }
 }
+
 #[derive(Debug)]
 pub enum StringFunction {
     Ascii(Box<Expression>),
@@ -1196,6 +1203,86 @@ pub enum StringFunction {
     Rtrim(Box<Expression>, Option<Box<Expression>>),
 }
 
+#[derive(Debug)]
+pub enum HashFunction {
+    MD5,
+    SHA224,
+    SHA256,
+    SHA384,
+    SHA512,
+}
+
+impl ToString for HashFunction {
+    fn to_string(&self) -> String {
+        match self {
+            Self::MD5 => "md5".to_string(),
+            Self::SHA224 => "sha224".to_string(),
+            Self::SHA256 => "sha256".to_string(),
+            Self::SHA384 => "sha384".to_string(),
+            Self::SHA512 => "sha512".to_string(),
+        }
+    }
+}
+
+impl TryFrom<BuiltinScalarFunction> for HashFunction {
+    type Error = anyhow::Error;
+
+    fn try_from(value: BuiltinScalarFunction) -> Result<Self> {
+        match value {
+            BuiltinScalarFunction::MD5 => Ok(Self::MD5),
+            BuiltinScalarFunction::SHA224 => Ok(Self::SHA224),
+            BuiltinScalarFunction::SHA256 => Ok(Self::SHA256),
+            BuiltinScalarFunction::SHA384 => Ok(Self::SHA384),
+            BuiltinScalarFunction::SHA512 => Ok(Self::SHA512),
+            _ => bail!("function {} is not a hash function", value),
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct HashExpression {
+    function: HashFunction,
+    input: Box<Expression>,
+}
+
+impl HashExpression {
+    pub fn new(function: BuiltinScalarFunction, input: Box<Expression>) -> Result<Expression> {
+        Ok(Expression::Hash(HashExpression {
+            function: function.try_into()?,
+            input,
+        }))
+    }
+}
+
+impl ExpressionGenerator for HashExpression {
+    fn to_syn_expression(&self) -> syn::Expr {
+        let input = self.input.to_syn_expression();
+        let hash_fn = format_ident!("{}", self.function.to_string());
+
+        let coerce = match self.input.return_type() {
+            TypeDef::StructDef(_, _) => unreachable!(),
+            TypeDef::DataType(DataType::Utf8, _) => quote!(.as_bytes().to_vec()),
+            TypeDef::DataType(DataType::Binary, _) => quote!(),
+            TypeDef::DataType(_, _) => unreachable!(),
+        };
+
+        match self.input.nullable() {
+            true => parse_quote!({
+                match #input {
+                    Some(unwrapped) => Some(arroyo_types::functions::hash::#hash_fn(unwrapped #coerce)),
+                    None => None,
+                }
+            }),
+            false => parse_quote!(arroyo_types::functions::hash::#hash_fn(#input #coerce)),
+        }
+    }
+
+    fn return_type(&self) -> TypeDef {
+        // this *can* be null because in SQL - MD5(NULL) = NULL
+        TypeDef::DataType(DataType::Utf8, self.input.nullable())
+    }
+}
+
 impl TryFrom<(BuiltinScalarFunction, Vec<Expression>)> for StringFunction {
     type Error = anyhow::Error;
 
diff --git a/arroyo-sql/src/lib.rs b/arroyo-sql/src/lib.rs
index 29b6edd94..8a002a5da 100644
--- a/arroyo-sql/src/lib.rs
+++ b/arroyo-sql/src/lib.rs
@@ -37,6 +37,8 @@ use types::{StructDef, StructField, TypeDef};
 use std::time::SystemTime;
 use std::{collections::HashMap, sync::Arc};
 
+use crate::expressions::ExpressionGenerator;
+
 #[cfg(test)]
 mod test;
 
@@ -266,6 +268,8 @@ pub struct TestStruct {
     pub nullable_string: Option<String>,
     pub non_nullable_timestamp: SystemTime,
     pub nullable_timestamp: Option<SystemTime>,
+    pub non_nullable_bytes: Vec<u8>,
+    pub nullable_bytes: Option<Vec<u8>>,
 }
 
 impl Default for TestStruct {
@@ -283,6 +287,8 @@ impl Default for TestStruct {
             nullable_string: Default::default(),
             non_nullable_timestamp: SystemTime::UNIX_EPOCH,
             nullable_timestamp: None,
+            non_nullable_bytes: Default::default(),
+            nullable_bytes: Default::default(),
         }
     }
 }
@@ -357,6 +363,16 @@ fn test_struct_def() -> StructDef {
                     true,
                 ),
             },
+            StructField {
+                name: "non_nullable_bytes".to_string(),
+                alias: None,
+                data_type: TypeDef::DataType(DataType::Binary, false),
+            },
+            StructField {
+                name: "nullable_bytes".to_string(),
+                alias: None,
+                data_type: TypeDef::DataType(DataType::Binary, true),
+            },
         ],
     }
 }
diff --git a/arroyo-types/Cargo.toml b/arroyo-types/Cargo.toml
index 19b2ddd1b..2d36c34e7 100644
--- a/arroyo-types/Cargo.toml
+++ b/arroyo-types/Cargo.toml
@@ -5,4 +5,8 @@ edition = "2021"
 
 [dependencies]
 bincode = "2.0.0-rc.3"
+base64 = "0.21"
+md-5 = "0.10"
+sha2 = "0.10"
+hex = "0.4"
 serde = { version = "1.0", features = ["derive"] }
diff --git a/arroyo-types/src/functions.rs b/arroyo-types/src/functions.rs
deleted file mode 100644
index f57966caf..000000000
--- a/arroyo-types/src/functions.rs
+++ /dev/null
@@ -1,189 +0,0 @@
-pub mod strings {
-    pub fn ascii(argument: String) -> i32 {
-        argument
-            .chars()
-            .next()
-            .map(|char| char as i32)
-            .unwrap_or_default()
-    }
-
-    pub fn bit_length(argument: String) -> i32 {
-        argument.len() as i32 * 8
-    }
-
-    pub fn trim(string: String, trimming_characters: String) -> String {
-        let chars: Vec<char> = trimming_characters.chars().collect();
-        let char_slice: &[char] = &chars;
-        string.trim_matches(char_slice).to_string()
-    }
-
-    pub fn char_length(argument: String) -> i32 {
-        length(argument)
-    }
-
-    pub fn chr(argument: i64) -> String {
-        std::char::from_u32(argument as u32)
-            .unwrap_or_default()
-            .to_string()
-    }
-
-    pub fn initcap(argument: String) -> String {
-        let mut result = String::with_capacity(argument.len());
-        let mut new_word = true;
-
-        for c in argument.chars() {
-            if c.is_alphanumeric() {
-                if new_word {
-                    result.extend(c.to_uppercase());
-                    new_word = false;
-                } else {
-                    result.extend(c.to_lowercase());
-                }
-            } else {
-                new_word = true;
-                result.push(c);
-            }
-        }
-        result
-    }
-
-    pub fn left(s: String, n: i64) -> String {
-        if n < 0 {
-            s.chars()
-                .take(s.chars().count().saturating_sub(n.abs() as usize))
-                .collect()
-        } else {
-            s.chars().take(n as usize).collect()
-        }
-    }
-
-    pub fn length(s: String) -> i32 {
-        s.chars().count() as i32
-    }
-
-    pub fn lpad(s: String, length: usize, fill: String) -> String {
-        let s_len = s.chars().count();
-        if s_len == length {
-            return s;
-        }
-        if s_len > length {
-            return s.chars().take(length).collect();
-        }
-        let fill_chars: Vec<char> = fill.chars().collect();
-        let mut result = String::new();
-
-        if length > s_len {
-            for i in 0..(length - s_len) {
-                result.push(fill_chars[i % fill_chars.len()]);
-            }
-        }
-
-        result.push_str(&s);
-        result
-    }
-
-    pub fn ltrim(string: String, trimming_characters: String) -> String {
-        let chars: Vec<char> = trimming_characters.chars().collect();
-        let char_slice: &[char] = &chars;
-        string.trim_start_matches(char_slice).to_string()
-    }
-
-    pub fn split_part(s: String, delimiter: String, n: isize) -> String {
-        let parts: Vec<&str> = s.split(&delimiter).collect();
-
-        if n > 0 {
-            parts
-                .get((n - 1) as usize)
-                .cloned()
-                .unwrap_or("")
-                .to_string()
-        } else {
-            parts
-                .get(parts.len().saturating_sub((-n) as usize))
-                .cloned()
-                .unwrap_or("")
-                .to_string()
-        }
-    }
-
-    pub fn starts_with(s: String, prefix: String) -> bool {
-        s.starts_with(&prefix)
-    }
-
-    pub fn strpos(s: String, substring: String) -> usize {
-        s.find(&substring).map_or(0, |index| index + 1)
-    }
-
-    pub fn substr(s: String, start: usize, count: Option<usize>) -> String {
-        let start_index = start.saturating_sub(1);
-        match count {
-            Some(count) => s.chars().skip(start_index).take(count).collect(),
-            None => s.chars().skip(start_index).collect(),
-        }
-    }
-
-    pub fn translate(s: String, from: String, to: String) -> String {
-        let from_chars: Vec<char> = from.chars().collect();
-        let to_chars: Vec<char> = to.chars().collect();
-        let mut translation_map = std::collections::HashMap::new();
-
-        for (i, from_char) in from_chars.iter().enumerate() {
-            let to_char = to_chars.get(i).copied();
-            translation_map.insert(from_char, to_char);
-        }
-
-        s.chars()
-            .filter_map(|c| {
-                if let Some(mapped_char) = translation_map.get(&c) {
-                    if mapped_char.is_none() {
-                        None // Delete the character
-                    } else {
-                        *mapped_char
-                    }
-                } else {
-                    Some(c)
-                }
-            })
-            .collect()
-    }
-
-    pub fn octet_length(s: String) -> usize {
-        s.as_bytes().len()
-    }
-
-    pub fn right(s: String, n: isize) -> String {
-        if n < 0 {
-            s.chars().skip(n.abs() as usize).collect()
-        } else {
-            s.chars()
-                .rev()
-                .take(n as usize)
-                .collect::<String>()
-                .chars()
-                .rev()
-                .collect()
-        }
-    }
-
-    pub fn rpad(s: String, length: usize, fill: String) -> String {
-        let s_len = s.chars().count();
-        if s_len == length {
-            return s;
-        }
-        if s_len > length {
-            return s.chars().take(length).collect();
-        }
-        let fill_chars: Vec<char> = fill.chars().collect();
-        let mut result = s;
-
-        for i in 0..(length.saturating_sub(s_len)) {
-            result.push(fill_chars[i % fill_chars.len()]);
-        }
-        result
-    }
-    pub fn rtrim(string: String, trimming_characters: String) -> String {
-        let chars: Vec<char> = trimming_characters.chars().collect();
-        let char_slice: &[char] = &chars;
-        string.trim_end_matches(char_slice).to_string()
-    }
-}
diff --git a/arroyo-types/src/functions/hash.rs b/arroyo-types/src/functions/hash.rs
new file mode 100644
index 000000000..a836e9dfe
--- /dev/null
+++ b/arroyo-types/src/functions/hash.rs
@@ -0,0 +1,33 @@
+use hex;
+use md5::{Digest, Md5};
+
+pub fn md5(argument: Vec<u8>) -> String {
+    let mut hasher = Md5::new();
+    hasher.update(argument);
+    let res = hasher.finalize();
+    hex::encode(&res)
+}
+
+pub fn sha224(argument: Vec<u8>) -> Vec<u8> {
+    let mut hasher = sha2::Sha224::new();
+    hasher.update(argument);
+    hasher.finalize().to_vec()
+}
+
+pub fn sha256(argument: Vec<u8>) -> Vec<u8> {
+    let mut hasher = sha2::Sha256::new();
+    hasher.update(argument);
+    hasher.finalize().to_vec()
+}
+
+pub fn sha384(argument: Vec<u8>) -> Vec<u8> {
+    let mut hasher = sha2::Sha384::new();
+    hasher.update(argument);
+    hasher.finalize().to_vec()
+}
+
+pub fn sha512(argument: Vec<u8>) -> Vec<u8> {
+    let mut hasher = sha2::Sha512::new();
+    hasher.update(argument);
+    hasher.finalize().to_vec()
+}
diff --git a/arroyo-types/src/functions/mod.rs b/arroyo-types/src/functions/mod.rs
new file mode 100644
index 000000000..472bcebe2
--- /dev/null
+++ b/arroyo-types/src/functions/mod.rs
@@ -0,0 +1,2 @@
+pub mod hash;
+pub mod strings;
diff --git a/arroyo-types/src/functions/strings.rs b/arroyo-types/src/functions/strings.rs
new file mode 100644
index 000000000..4b85f0484
--- /dev/null
+++ b/arroyo-types/src/functions/strings.rs
@@ -0,0 +1,187 @@
+pub fn ascii(argument: String) -> i32 {
+    argument
+        .chars()
+        .next()
+        .map(|char| char as i32)
+        .unwrap_or_default()
+}
+
+pub fn bit_length(argument: String) -> i32 {
+    argument.len() as i32 * 8
+}
+
+pub fn trim(string: String, trimming_characters: String) -> String {
+    let chars: Vec<char> = trimming_characters.chars().collect();
+    let char_slice: &[char] = &chars;
+    string.trim_matches(char_slice).to_string()
+}
+
+pub fn char_length(argument: String) -> i32 {
+    length(argument)
+}
+
+pub fn chr(argument: i64) -> String {
+    std::char::from_u32(argument as u32)
+        .unwrap_or_default()
+        .to_string()
+}
+
+pub fn initcap(argument: String) -> String {
+    let mut result = String::with_capacity(argument.len());
+    let mut new_word = true;
+
+    for c in argument.chars() {
+        if c.is_alphanumeric() {
+            if new_word {
+                result.extend(c.to_uppercase());
+                new_word = false;
+            } else {
+                result.extend(c.to_lowercase());
+            }
+        } else {
+            new_word = true;
+            result.push(c);
+        }
+    }
+    result
+}
+
+pub fn left(s: String, n: i64) -> String {
+    if n < 0 {
+        s.chars()
+            .take(s.chars().count().saturating_sub(n.abs() as usize))
+            .collect()
+    } else {
+        s.chars().take(n as usize).collect()
+    }
+}
+
+pub fn length(s: String) -> i32 {
+    s.chars().count() as i32
+}
+
+pub fn lpad(s: String, length: usize, fill: String) -> String {
+    let s_len = s.chars().count();
+    if s_len == length {
+        return s;
+    }
+    if s_len > length {
+        return s.chars().take(length).collect();
+    }
+    let fill_chars: Vec<char> = fill.chars().collect();
+    let mut result = String::new();
+
+    if length > s_len {
+        for i in 0..(length - s_len) {
+            result.push(fill_chars[i % fill_chars.len()]);
+        }
+    }
+
+    result.push_str(&s);
+    result
+}
+
+pub fn ltrim(string: String, trimming_characters: String) -> String {
+    let chars: Vec<char> = trimming_characters.chars().collect();
+    let char_slice: &[char] = &chars;
+    string.trim_start_matches(char_slice).to_string()
+}
+
+pub fn split_part(s: String, delimiter: String, n: isize) -> String {
+    let parts: Vec<&str> = s.split(&delimiter).collect();
+
+    if n > 0 {
+        parts
+            .get((n - 1) as usize)
+            .cloned()
+            .unwrap_or("")
+            .to_string()
+    } else {
+        parts
+            .get(parts.len().saturating_sub((-n) as usize))
+            .cloned()
+            .unwrap_or("")
+            .to_string()
+    }
+}
+
+pub fn starts_with(s: String, prefix: String) -> bool {
+    s.starts_with(&prefix)
+}
+
+pub fn strpos(s: String, substring: String) -> usize {
+    s.find(&substring).map_or(0, |index| index + 1)
+}
+
+pub fn substr(s: String, start: usize, count: Option<usize>) -> String {
+    let start_index = start.saturating_sub(1);
+    match count {
+        Some(count) => s.chars().skip(start_index).take(count).collect(),
+        None => s.chars().skip(start_index).collect(),
+    }
+}
+
+pub fn translate(s: String, from: String, to: String) -> String {
+    let from_chars: Vec<char> = from.chars().collect();
+    let to_chars: Vec<char> = to.chars().collect();
+    let mut translation_map = std::collections::HashMap::new();
+
+    for (i, from_char) in from_chars.iter().enumerate() {
+        let to_char = to_chars.get(i).copied();
+        translation_map.insert(from_char, to_char);
+    }
+
+    s.chars()
+        .filter_map(|c| {
+            if let Some(mapped_char) = translation_map.get(&c) {
+                if mapped_char.is_none() {
+                    None // Delete the character
+                } else {
+                    *mapped_char
+                }
+            } else {
+                Some(c)
+            }
+        })
+        .collect()
+}
+
+pub fn octet_length(s: String) -> usize {
+    s.as_bytes().len()
+}
+
+pub fn right(s: String, n: isize) -> String {
+    if n < 0 {
+        s.chars().skip(n.abs() as usize).collect()
+    } else {
+        s.chars()
+            .rev()
+            .take(n as usize)
+            .collect::<String>()
+            .chars()
+            .rev()
+            .collect()
+    }
+}
+
+pub fn rpad(s: String, length: usize, fill: String) -> String {
+    let s_len = s.chars().count();
+    if s_len == length {
+        return s;
+    }
+    if s_len > length {
+        return s.chars().take(length).collect();
+    }
+    let fill_chars: Vec<char> = fill.chars().collect();
+    let mut result = s;
+
+    for i in 0..(length.saturating_sub(s_len)) {
+        result.push(fill_chars[i % fill_chars.len()]);
+    }
+    result
+}
+pub fn rtrim(string: String, trimming_characters: String) -> String {
+    let chars: Vec<char> = trimming_characters.chars().collect();
+    let char_slice: &[char] = &chars;
+    string.trim_end_matches(char_slice).to_string()
+}


Test patch:
diff --git a/arroyo-sql-testing/Cargo.toml b/arroyo-sql-testing/Cargo.toml
index ef3c16394..18b6c2e80 100644
--- a/arroyo-sql-testing/Cargo.toml
+++ b/arroyo-sql-testing/Cargo.toml
@@ -8,6 +8,7 @@ edition = "2021"
 syn = "1.0"
 quote = "1.0"
 chrono = "0.4"
+hex = "0.4"
 arroyo-sql = { path = "../arroyo-sql" }
 arroyo-types = { path = "../arroyo-types" }
 arroyo-sql-macro = { path = "../arroyo-sql-macro" }
\ No newline at end of file
diff --git a/arroyo-sql-testing/src/lib.rs b/arroyo-sql-testing/src/lib.rs
index b4642068a..58260bb3c 100644
--- a/arroyo-sql-testing/src/lib.rs
+++ b/arroyo-sql-testing/src/lib.rs
@@ -611,4 +611,105 @@ mod tests {
         },
         None
     );
+
+    // Hash
+    single_test_codegen!(
+        "md5_non_null",
+        "md5(non_nullable_string)",
+        arroyo_sql::TestStruct {
+            non_nullable_string: "Hello, World".into(),
+            ..Default::default()
+        },
+        "82bb413746aee42f89dea2b59614f9ef".to_string()
+    );
+
+    single_test_codegen!(
+        "md5_null",
+        "md5(nullable_string)",
+        arroyo_sql::TestStruct {
+            nullable_string: None,
+            ..Default::default()
+        },
+        None
+    );
+
+    single_test_codegen!(
+        "sha224_non_null",
+        "sha224(non_nullable_bytes)",
+        arroyo_sql::TestStruct {
+            non_nullable_bytes: "asdf".as_bytes().to_vec(),
+            ..Default::default()
+        },
+        hex::decode("7872a74bcbf298a1e77d507cd95d4f8d96131cbbd4cdfc571e776c8a").unwrap()
+    );
+
+    single_test_codegen!(
+        "sha224_null",
+        "sha224(nullable_bytes)",
+        arroyo_sql::TestStruct {
+            nullable_bytes: None,
+            ..Default::default()
+        },
+        None
+    );
+
+    single_test_codegen!(
+        "sha256_non_null",
+        "sha256(non_nullable_bytes)",
+        arroyo_sql::TestStruct {
+            non_nullable_bytes: "asdf".as_bytes().to_vec(),
+            ..Default::default()
+        },
+        hex::decode("f0e4c2f76c58916ec258f246851bea091d14d4247a2fc3e18694461b1816e13b").unwrap()
+    );
+
+    single_test_codegen!(
+        "sha256_null",
+        "sha256(nullable_bytes)",
+        arroyo_sql::TestStruct {
+            nullable_bytes: None,
+            ..Default::default()
+        },
+        None
+    );
+
+    single_test_codegen!(
+        "sha384_non_null",
+        "sha384(non_nullable_bytes)",
+        arroyo_sql::TestStruct {
+            non_nullable_bytes: "asdf".as_bytes().to_vec(),
+            ..Default::default()
+        },
+        hex::decode("a69e7df30b24c042ec540ccbbdbfb1562c85787038c885749c1e408e2d62fa36642cd0075fa351e822e2b8a59139cd9d").unwrap()
+    );
+
+    single_test_codegen!(
+        "sha384_null",
+        "sha384(nullable_bytes)",
+        arroyo_sql::TestStruct {
+            nullable_bytes: None,
+            ..Default::default()
+        },
+        None
+    );
+
+    single_test_codegen!(
+        "sha512_non_null",
+        "sha512(non_nullable_bytes)",
+        arroyo_sql::TestStruct {
+            non_nullable_bytes: "asdf".as_bytes().to_vec(),
+            ..Default::default()
+        },
+        hex::decode("401b09eab3c013d4ca54922bb802bec8fd5318192b0a75f201d8b3727429080fb337591abd3e44453b954555b7a0812e1081c39b740293f765eae731f5a65ed1").unwrap()
+    );
+
+    single_test_codegen!(
+        "sha512_null",
+        "sha512(nullable_bytes)",
+        arroyo_sql::TestStruct {
+            nullable_bytes: None,
+            ..Default::default()
+        },
+        None
+    );
 }