Add support for volatile scalar functions

This PR adds the ability to mark scalar functions as volatile, which prevents
DuckDB from optimizing them as constants. This is essential for functions that
generate random or unique values per row, such as:
- Random number generators
- UUID generators
- Fake data generators
- Current timestamp functions

Added a new method to `ScalarFunction` that calls the existing FFI binding
`duckdb_scalar_function_set_volatile()`. This method follows the builder
pattern used by other methods in the API.

Added two new convenience methods to `Connection`:
- `register_volatile_scalar_function<S: VScalar>(name)` - Register with default state
- `register_volatile_scalar_function_with_state<S: VScalar>(name, state)` - Register with custom state

These mirror the existing `register_scalar_function` methods but automatically
mark the functions as volatile.

Added comprehensive tests demonstrating:
- `test_volatile_scalar` - Verifies volatile functions are evaluated per row
- `test_non_volatile_scalar` - Verifies non-volatile functions are optimized as constants

```rust
use duckdb::Connection;
use duckdb::vscalar::VScalar;

// Assume RandomUUID implements VScalar
let conn = Connection::open_in_memory()?;
conn.register_volatile_scalar_function::<RandomUUID>("random_uuid")?;

// Each row gets a unique UUID
let mut stmt = conn.prepare("SELECT random_uuid() FROM generate_series(1, 10)")?;
```

By default, DuckDB optimizes zero-argument scalar functions as constants,
evaluating them only once and reusing the result. For deterministic functions
this is correct, but for non-deterministic functions (random generators, UUIDs,
fake data), this produces incorrect results where all rows get the same value.

The VOLATILE flag tells DuckDB's optimizer to re-evaluate the function for
each row, which is the correct behavior for non-deterministic functions.

Fixes functionality needed by DuckDB extensions that generate unique data per row.

Author: tobilg

crates/duckdb/src/vscalar/function.rs

@@ -53,7 +53,8 @@ use libduckdb_sys::{
     duckdb_create_scalar_function_set, duckdb_data_chunk, duckdb_delete_callback_t, duckdb_destroy_scalar_function,
     duckdb_function_info, duckdb_scalar_function, duckdb_scalar_function_add_parameter, duckdb_scalar_function_set,
     duckdb_scalar_function_set_extra_info, duckdb_scalar_function_set_function, duckdb_scalar_function_set_name,
-    duckdb_scalar_function_set_return_type, duckdb_scalar_function_set_varargs, duckdb_vector, DuckDBSuccess,
+    duckdb_scalar_function_set_return_type, duckdb_scalar_function_set_varargs, duckdb_scalar_function_set_volatile,
+    duckdb_vector, DuckDBSuccess,
 };
 
 use crate::{core::LogicalTypeHandle, Error};
@@ -112,6 +113,33 @@ impl ScalarFunction {
         self
     }
 
+    /// Marks the scalar function as volatile.
+    ///
+    /// Volatile functions are re-evaluated for each row, even if they have no parameters.
+    /// This is useful for functions that generate random or unique values, such as random
+    /// number generators, UUID generators, or fake data generators.
+    ///
+    /// By default, DuckDB optimizes zero-argument scalar functions as constants, evaluating
+    /// them only once. Setting a function as volatile prevents this optimization.
+    ///
+    /// # Example
+    /// ```no_run
+    /// use duckdb::vscalar::ScalarFunction;
+    /// use duckdb::core::LogicalTypeHandle;
+    /// use libduckdb_sys::LogicalTypeId;
+    ///
+    /// let func = ScalarFunction::new("my_random_func").unwrap();
+    /// func.set_return_type(&LogicalTypeHandle::from(LogicalTypeId::Varchar))
+    ///     .set_volatile()  // Mark as volatile so it's evaluated per row
+    ///     .set_function(Some(my_random_impl));
+    /// ```
+    pub fn set_volatile(&self) -> &Self {
+        unsafe {
+            duckdb_scalar_function_set_volatile(self.ptr);
+        }
+        self
+    }
+
     /// Assigns extra information to the scalar function using raw pointers.
     ///
     /// For most use cases, prefer [`set_extra_info`](Self::set_extra_info) which handles memory management automatically.

crates/duckdb/src/vscalar/mod.rs

@@ -168,6 +168,73 @@ impl Connection {
         }
         self.db.borrow_mut().register_scalar_function_set(set)
     }
+
+    /// Register the given ScalarFunction with default state, marked as volatile.
+    ///
+    /// Volatile functions are re-evaluated for each row, even if they have no parameters.
+    /// This is useful for functions that generate random or unique values per row, such as:
+    /// - Random number generators
+    /// - UUID generators
+    /// - Fake data generators
+    /// - Current timestamp functions
+    ///
+    /// By default, DuckDB optimizes zero-argument scalar functions as constants.
+    /// Use this method when you need the function to be evaluated independently for each row.
+    ///
+    /// # Example
+    /// ```no_run
+    /// use duckdb::Connection;
+    /// // Assume RandomUUID implements VScalar
+    /// let conn = Connection::open_in_memory()?;
+    /// conn.register_volatile_scalar_function::<RandomUUID>("random_uuid")?;
+    ///
+    /// // Each row gets a unique UUID
+    /// let mut stmt = conn.prepare("SELECT random_uuid() FROM generate_series(1, 10)")?;
+    /// # Ok::<(), duckdb::Error>(())
+    /// ```
+    #[inline]
+    pub fn register_volatile_scalar_function<S: VScalar>(&self, name: &str) -> crate::Result<()>
+    where
+        S::State: Default,
+    {
+        let set = ScalarFunctionSet::new(name);
+        for signature in S::signatures() {
+            let scalar_function = ScalarFunction::new(name)?;
+            signature.register_with_scalar(&scalar_function);
+            scalar_function.set_function(Some(scalar_func::<S>));
+            scalar_function.set_volatile(); // Mark as volatile
+            scalar_function.set_extra_info(S::State::default());
+            set.add_function(scalar_function)?;
+        }
+        self.db.borrow_mut().register_scalar_function_set(set)
+    }
+
+    /// Register the given ScalarFunction with custom state, marked as volatile.
+    ///
+    /// Volatile functions are re-evaluated for each row, even if they have no parameters.
+    /// This is the volatile variant of `register_scalar_function_with_state`.
+    ///
+    /// See [`register_volatile_scalar_function`](Self::register_volatile_scalar_function) for more details on volatile functions.
+    #[inline]
+    pub fn register_volatile_scalar_function_with_state<S: VScalar>(
+        &self,
+        name: &str,
+        state: &S::State,
+    ) -> crate::Result<()>
+    where
+        S::State: Clone,
+    {
+        let set = ScalarFunctionSet::new(name);
+        for signature in S::signatures() {
+            let scalar_function = ScalarFunction::new(name)?;
+            signature.register_with_scalar(&scalar_function);
+            scalar_function.set_function(Some(scalar_func::<S>));
+            scalar_function.set_volatile(); // Mark as volatile
+            scalar_function.set_extra_info(state.clone());
+            set.add_function(scalar_function)?;
+        }
+        self.db.borrow_mut().register_scalar_function_set(set)
+    }
 }
 
 impl InnerConnection {
@@ -374,4 +441,96 @@ mod test {
 
         Ok(())
     }
+
+    // Counter for testing volatile functions
+    use std::sync::atomic::{AtomicU64, Ordering};
+    static COUNTER: AtomicU64 = AtomicU64::new(0);
+
+    struct CounterScalar {}
+
+    impl VScalar for CounterScalar {
+        type State = ();
+
+        unsafe fn invoke(
+            _: &Self::State,
+            input: &mut DataChunkHandle,
+            output: &mut dyn WritableVector,
+        ) -> Result<(), Box<dyn std::error::Error>> {
+            let len = input.len();
+            let mut output_vec = output.flat_vector();
+            let data = output_vec.as_mut_slice::<i64>();
+
+            for i in 0..len {
+                let count = COUNTER.fetch_add(1, Ordering::SeqCst);
+                data[i] = count as i64;
+            }
+            Ok(())
+        }
+
+        fn signatures() -> Vec<ScalarFunctionSignature> {
+            vec![ScalarFunctionSignature::exact(
+                vec![],
+                LogicalTypeHandle::from(LogicalTypeId::Bigint),
+            )]
+        }
+    }
+
+    #[test]
+    fn test_volatile_scalar() -> Result<(), Box<dyn Error>> {
+        let conn = Connection::open_in_memory()?;
+
+        // Reset counter
+        COUNTER.store(0, Ordering::SeqCst);
+
+        // Register as volatile
+        conn.register_volatile_scalar_function::<CounterScalar>("volatile_counter")?;
+
+        // Query should get different values for each row
+        let mut stmt = conn.prepare("SELECT volatile_counter() FROM generate_series(1, 5)")?;
+        let mut rows = stmt.query([])?;
+
+        let mut values = Vec::new();
+        while let Some(row) = rows.next()? {
+            let val: i64 = row.get(0)?;
+            values.push(val);
+        }
+
+        // Each value should be unique (counter increments)
+        assert_eq!(values.len(), 5);
+        for (i, val) in values.iter().enumerate() {
+            assert_eq!(*val, i as i64, "Row {} should have counter value {}", i, i);
+        }
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_non_volatile_scalar() -> Result<(), Box<dyn Error>> {
+        let conn = Connection::open_in_memory()?;
+
+        // Reset counter
+        COUNTER.store(100, Ordering::SeqCst);
+
+        // Register WITHOUT volatile flag
+        conn.register_scalar_function::<CounterScalar>("non_volatile_counter")?;
+
+        // Query should get the SAME value for all rows (optimized as constant)
+        let mut stmt = conn.prepare("SELECT non_volatile_counter() FROM generate_series(1, 5)")?;
+        let mut rows = stmt.query([])?;
+
+        let mut values = Vec::new();
+        while let Some(row) = rows.next()? {
+            let val: i64 = row.get(0)?;
+            values.push(val);
+        }
+
+        // All values should be the same (function was only called once)
+        assert_eq!(values.len(), 5);
+        let first_val = values[0];
+        for val in values.iter() {
+            assert_eq!(*val, first_val, "All rows should have the same value when not volatile");
+        }
+
+        Ok(())
+    }
 }