feat: native uuid() implementation compatible with Spark

.github/workflows/pr_build_linux.yml

@@ -401,6 +401,7 @@ jobs:
               org.apache.comet.CometCodegenFuzzSuite
               org.apache.comet.CometStringDecodeSuite
               org.apache.comet.CometWidthBucketSuite
+              org.apache.comet.CometUuidExpressionSuite
       fail-fast: false
     name: ${{ matrix.profile.name }} [${{ matrix.suite.name }}]
     runs-on: ubuntu-24.04

.github/workflows/pr_build_macos.yml

@@ -217,6 +217,7 @@ jobs:
               org.apache.comet.CometCodegenFuzzSuite
               org.apache.comet.CometStringDecodeSuite
               org.apache.comet.CometWidthBucketSuite
+              org.apache.comet.CometUuidExpressionSuite
 
       fail-fast: false
     name: ${{ matrix.os }}/${{ matrix.profile.name }} [${{ matrix.suite.name }}]

docs/source/contributor-guide/expression-audits/misc_funcs.md

@@ -89,4 +89,11 @@
 - Spark 4.0.1 (audited 2026-05-27): identical to 3.4.3 except the resulting literal carries the default string collation.
 - Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1.
 
+## uuid
+
+- Spark 3.4.3 (audited 2026-07-24): `Uuid(randomSeed: Option[Long]) extends LeafExpression with Nondeterministic with ExpressionWithRandomSeed`. The analyzer's `ResolveRandomSeed` fills `randomSeed` with a random `Long`, so it is always defined before Comet sees the plan. Per partition it seeds `RandomUUIDGenerator(randomSeed + partitionIndex)`, a Commons Math3 `MersenneTwister`, and per row draws two `nextLong()`s, masks in the RFC 4122 version 4 and variant bits, and formats via `java.util.UUID.toString`. Only the no-argument `uuid()` form exists (no seed constructor). Comet emits a `Uuid` proto with the resolved seed and reproduces the generator bit for bit via `SparkMersenneTwister`.
+- Spark 3.5.8 (audited 2026-07-24): identical to 3.4.3.
+- Spark 4.0.1 (audited 2026-07-24): adds `def this(seed: Expression)`, exposing the `uuid(seed)` SQL form (the seed must be an integer or long literal, validated at analysis time). `RandomUUIDGenerator` and the per-row algorithm are unchanged, so results are identical to 3.4.3 for a given seed.
+- Spark 4.1.1 (audited 2026-07-24): identical to 4.0.1, plus `withShiftedSeed`. No runtime change.
+
 [Spark Expression Support]: ../../user-guide/latest/expressions.md

docs/source/user-guide/latest/expressions.md

@@ -511,7 +511,7 @@ expression-level). The `outer` variants are wired but marked `Incompatible`; the
 | `try_variant_get` | 🔜 | — | Requires `VariantType` support |
 | `typeof` | ✅ | — | Foldable; resolved to a literal before Comet sees the plan |
 | `user` | ✅ | — | Resolved to a literal by the Spark analyzer before reaching Comet |
-| `uuid` | 🔜 | — | Nondeterministic random UUID |
+| `uuid` | ✅ | Native |  |
 | `variant_get` | 🔜 | — | Requires `VariantType` support |
 
 ---

native/core/src/execution/expressions/random.rs

@@ -22,7 +22,7 @@ use crate::extract_expr;
 use arrow::datatypes::SchemaRef;
 use datafusion::physical_expr::PhysicalExpr;
 use datafusion_comet_proto::spark_expression::Expr;
-use datafusion_comet_spark_expr::{RandExpr, RandStrExpr, RandnExpr, ShuffleExpr};
+use datafusion_comet_spark_expr::{RandExpr, RandStrExpr, RandnExpr, ShuffleExpr, UuidExpr};
 use std::sync::Arc;
 
 pub struct RandBuilder;
@@ -57,6 +57,22 @@ impl ExpressionBuilder for ShuffleBuilder {
     }
 }
 
+pub struct UuidBuilder;
+
+impl ExpressionBuilder for UuidBuilder {
+    fn build(
+        &self,
+        spark_expr: &Expr,
+        _input_schema: SchemaRef,
+        planner: &PhysicalPlanner,
+    ) -> Result<Arc<dyn PhysicalExpr>, ExecutionError> {
+        let expr = extract_expr!(spark_expr, Uuid);
+        // Spark seeds a fresh generator per partition with `randomSeed + partitionIndex`.
+        let seed = expr.seed.wrapping_add(planner.partition().into());
+        Ok(Arc::new(UuidExpr::new(seed)))
+    }
+}
+
 pub struct RandStrBuilder;
 
 impl ExpressionBuilder for RandStrBuilder {

native/core/src/execution/planner/expression_registry.rs

@@ -102,6 +102,7 @@ pub enum ExpressionType {
     Randn,
     RandStr,
     Shuffle,
+    Uuid,
     SparkPartitionId,
     MonotonicallyIncreasingId,
     ArraysZip,
@@ -378,6 +379,7 @@ impl ExpressionRegistry {
             Some(ExprStruct::Randn(_)) => Ok(ExpressionType::Randn),
             Some(ExprStruct::RandStr(_)) => Ok(ExpressionType::RandStr),
             Some(ExprStruct::Shuffle(_)) => Ok(ExpressionType::Shuffle),
+            Some(ExprStruct::Uuid(_)) => Ok(ExpressionType::Uuid),
             Some(ExprStruct::SparkPartitionId(_)) => Ok(ExpressionType::SparkPartitionId),
             Some(ExprStruct::MonotonicallyIncreasingId(_)) => {
                 Ok(ExpressionType::MonotonicallyIncreasingId)
@@ -411,6 +413,8 @@ impl ExpressionRegistry {
             .insert(ExpressionType::Randn, Box::new(RandnBuilder));
         self.builders
             .insert(ExpressionType::Shuffle, Box::new(ShuffleBuilder));
+        self.builders
+            .insert(ExpressionType::Uuid, Box::new(UuidBuilder));
         self.builders
             .insert(ExpressionType::RandStr, Box::new(RandStrBuilder));
     }

native/proto/src/proto/expr.proto

@@ -93,6 +93,7 @@ message Expr {
     PreciseTimestampConversion precise_timestamp_conversion = 71;
     Shuffle shuffle = 72;
     RandStr rand_str = 73;
+    Uuid uuid = 74;
   }
 
   reserved 20;
@@ -570,6 +571,14 @@ message Shuffle {
   int64 seed = 2;
 }
 
+// Spark's Uuid returns a random RFC 4122 version 4 UUID string. It is
+// non-deterministic: the resolved random seed is combined with the partition
+// index and seeds a Commons Math3 MersenneTwister, matching
+// org.apache.spark.sql.catalyst.util.RandomUUIDGenerator.
+message Uuid {
+  int64 seed = 1;
+}
+
 // Spark's RandStr (Spark 4.0+) returns a random alphanumeric string of the given
 // length. It is non-deterministic: the resolved random seed is combined with the
 // partition index and seeds an XORShiftRandom, matching

native/spark-expr/Cargo.toml

@@ -43,6 +43,7 @@ futures = { workspace = true }
 twox-hash = "2.1.2"
 rand = { workspace = true }
 base64 = "0.23.0"
+uuid = "1.23.3"
 
 [dev-dependencies]
 arrow = {workspace = true}

native/spark-expr/src/nondetermenistic_funcs/internal/mersenne.rs

@@ -0,0 +1,172 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+//! A bit-for-bit port of Apache Commons Math3's `MersenneTwister`, the PRNG Spark
+//! seeds per partition with `randomSeed + partitionIndex`. It backs both `shuffle`
+//! (via `RandomIndicesGenerator`) and `uuid` (via `RandomUUIDGenerator`).
+//!
+//! See:
+//! - `org/apache/commons/math3/random/MersenneTwister.java`
+//! - `org/apache/commons/math3/random/BitsStreamGenerator.java`
+
+#[derive(Debug, Clone)]
+pub(crate) struct SparkMersenneTwister {
+    /// Bytes pool.
+    mt: [i32; Self::N],
+    /// Current index in the bytes pool.
+    mti: usize,
+}
+
+impl SparkMersenneTwister {
+    /// Size of the bytes pool.
+    const N: usize = 624;
+    /// Period second parameter.
+    const M: usize = 397;
+    /// X * MATRIX_A for X = {0, 1}.
+    const MAG01: [i32; 2] = [0x0, 0x9908b0dfu32 as i32];
+
+    pub(crate) fn new(seed: i64) -> Self {
+        // `mti` is set by seeding before it is ever read; 0 is just a placeholder.
+        let mut twister = SparkMersenneTwister {
+            mt: [0i32; Self::N],
+            mti: 0,
+        };
+        twister.set_seed_long(seed);
+        twister
+    }
+
+    fn set_seed_int(&mut self, seed: i32) {
+        // We use a long masked by 0xffffffff as a poor man's unsigned int.
+        let mut long_mt = seed as i64;
+        self.mt[0] = long_mt as i32;
+        let mut mti = 1usize;
+        while mti < Self::N {
+            long_mt = (1812433253i64.wrapping_mul(long_mt ^ (long_mt >> 30)) + mti as i64)
+                & 0xffffffffi64;
+            self.mt[mti] = long_mt as i32;
+            mti += 1;
+        }
+        self.mti = mti;
+    }
+
+    fn set_seed_int_array(&mut self, seed: &[i32]) {
+        self.set_seed_int(19650218);
+        let mut i = 1usize;
+        let mut j = 0usize;
+
+        for _ in 0..Self::N.max(seed.len()) {
+            let mt_i = self.mt[i] as i64 & 0xffffffffi64;
+            let mt_im1 = self.mt[i - 1] as i64 & 0xffffffffi64;
+            let l = (mt_i ^ ((mt_im1 ^ (mt_im1 >> 30)).wrapping_mul(1664525)))
+                .wrapping_add(seed[j] as i64)
+                .wrapping_add(j as i64);
+            self.mt[i] = (l & 0xffffffffi64) as i32;
+            i += 1;
+            j += 1;
+            if i >= Self::N {
+                self.mt[0] = self.mt[Self::N - 1];
+                i = 1;
+            }
+            if j >= seed.len() {
+                j = 0;
+            }
+        }
+
+        for _ in 0..(Self::N - 1) {
+            let mt_i = self.mt[i] as i64 & 0xffffffffi64;
+            let mt_im1 = self.mt[i - 1] as i64 & 0xffffffffi64;
+            let l = (mt_i ^ ((mt_im1 ^ (mt_im1 >> 30)).wrapping_mul(1566083941)))
+                .wrapping_sub(i as i64);
+            self.mt[i] = (l & 0xffffffffi64) as i32;
+            i += 1;
+            if i >= Self::N {
+                self.mt[0] = self.mt[Self::N - 1];
+                i = 1;
+            }
+        }
+
+        // MSB is 1, assuring a non-zero initial array.
+        self.mt[0] = 0x80000000u32 as i32;
+    }
+
+    fn set_seed_long(&mut self, seed: i64) {
+        self.set_seed_int_array(&[(seed >> 32) as i32, seed as i32]);
+    }
+
+    fn next(&mut self, bits: u32) -> i32 {
+        let mut y: i32;
+        if self.mti >= Self::N {
+            // Generate N words at one time.
+            let mut mt_next = self.mt[0];
+            for k in 0..(Self::N - Self::M) {
+                let mt_curr = mt_next;
+                mt_next = self.mt[k + 1];
+                y = (mt_curr & (0x80000000u32 as i32)) | (mt_next & 0x7fffffff);
+                self.mt[k] = self.mt[k + Self::M]
+                    ^ (((y as u32) >> 1) as i32)
+                    ^ Self::MAG01[(y & 0x1) as usize];
+            }
+            for k in (Self::N - Self::M)..(Self::N - 1) {
+                let mt_curr = mt_next;
+                mt_next = self.mt[k + 1];
+                y = (mt_curr & (0x80000000u32 as i32)) | (mt_next & 0x7fffffff);
+                self.mt[k] = self.mt[k + Self::M - Self::N]
+                    ^ (((y as u32) >> 1) as i32)
+                    ^ Self::MAG01[(y & 0x1) as usize];
+            }
+            y = (mt_next & (0x80000000u32 as i32)) | (self.mt[0] & 0x7fffffff);
+            self.mt[Self::N - 1] =
+                self.mt[Self::M - 1] ^ (((y as u32) >> 1) as i32) ^ Self::MAG01[(y & 0x1) as usize];
+            self.mti = 0;
+        }
+
+        y = self.mt[self.mti];
+        self.mti += 1;
+
+        // Tempering.
+        y ^= ((y as u32) >> 11) as i32;
+        y ^= (y << 7) & (0x9d2c5680u32 as i32);
+        y ^= (y << 15) & (0xefc60000u32 as i32);
+        y ^= ((y as u32) >> 18) as i32;
+
+        ((y as u32) >> (32 - bits)) as i32
+    }
+
+    /// Port of `BitsStreamGenerator.nextInt(int n)`. The caller always passes a
+    /// strictly positive `n`, matching Spark's `random.nextInt(i + 1)`.
+    pub(crate) fn next_int(&mut self, n: i32) -> i32 {
+        if (n & n.wrapping_neg()) == n {
+            // n is a power of two.
+            return ((n as i64 * self.next(31) as i64) >> 31) as i32;
+        }
+        loop {
+            let bits = self.next(31);
+            let val = bits % n;
+            if bits.wrapping_sub(val).wrapping_add(n.wrapping_sub(1)) >= 0 {
+                return val;
+            }
+        }
+    }
+
+    /// Port of `BitsStreamGenerator.nextLong()`: two 32-bit draws combined into a
+    /// signed 64-bit value. Used by `RandomUUIDGenerator`.
+    pub(crate) fn next_long(&mut self) -> i64 {
+        let high = (self.next(32) as i64) << 32;
+        let low = (self.next(32) as i64) & 0xffffffffi64;
+        high | low
+    }
+}

native/spark-expr/src/nondetermenistic_funcs/internal/mod.rs

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+pub(crate) mod mersenne;
 mod rand_utils;
 
 pub use rand_utils::evaluate_batch_for_rand;

native/spark-expr/src/nondetermenistic_funcs/mod.rs

@@ -22,9 +22,11 @@ pub mod rand;
 pub mod randn;
 pub mod randstr;
 pub mod shuffle;
+pub mod uuid;
 
 pub use bernoulli_cell_sampler::BernoulliCellSampler;
 pub use rand::RandExpr;
 pub use randn::RandnExpr;
 pub use randstr::RandStrExpr;
 pub use shuffle::ShuffleExpr;
+pub use uuid::UuidExpr;