feat: native uuid() implementation compatible with Spark

[andygrove]

Which issue does this PR close?

There is no dedicated tracking issue; uuid was previously listed as planned (🔜) in the expression support guide.

Closes #.

Rationale for this change

Spark's uuid() fell back to Spark because Comet had no native implementation. Spark generates RFC 4122 version 4 UUID strings via org.apache.spark.sql.catalyst.util.RandomUUIDGenerator, which seeds a Commons Math3 MersenneTwister with randomSeed + partitionIndex and, per row, draws two nextLong()s, masks in the version and variant bits, and formats via java.util.UUID.toString. Comet already ports that exact MersenneTwister for native shuffle (SparkMersenneTwister), so uuid() can run natively and produce results that match Spark bit for bit, keeping the enclosing operator on the native pipeline instead of falling back.

What changes are included in this PR?

This change was scaffolded with the project's implement-comet-expression skill (and audited with audit-comet-expression).

• Native UuidExpr (native/spark-expr/src/nondetermenistic_funcs/uuid.rs), modeled on ShuffleExpr: it keeps a per-partition SparkMersenneTwister in a Mutex so state advances across batches, and produces a Utf8 column. Added SparkMersenneTwister::next_long() (a port of Commons Math3 BitsStreamGenerator.nextLong()).
• Proto Uuid message and oneof field, wired through the expression registry with a UuidBuilder that combines the seed with the partition index (matching rand/randn/shuffle).
• CometUuid serde emitting the resolved random seed, registered in QueryPlanSerde.
• Support-status doc flipped from planned to supported, and a misc_funcs audit entry recording the cross-version findings.

The audit confirmed RandomUUIDGenerator is identical across Spark 3.4.3 / 3.5.8 / 4.0.1 / 4.1.1, so no shim is needed. The uuid(seed) SQL form exists only in Spark 4.0+, so the seeded tests are gated with MinSparkVersion: 4.0.

How are these changes tested?

• Rust unit tests in uuid.rs: RFC 4122 v4 format/version/variant, determinism per seed, and per-batch state continuity.
• expressions/misc/uuid.sql (all versions): no-seed uuid() runs natively and satisfies deterministic properties (length, RFC 4122 layout, version/variant nibbles) that hold identically on Comet and Spark.
• expressions/misc/uuid_with_seed.sql (Spark 4.0+): seeded uuid(seed) asserts bit-for-bit equality with Spark across single and multiple rows, negative/zero/boundary seeds, and combined with other expressions. Verified natively on Spark 4.1.2 (default) and 3.5.8 (seeded file correctly skipped).

[comphead]

Thanks @andygrove the PR looks solid.

LLM suggests couple of tests gaps

Gap 1 — Multi-partition bit-for-bit (highest priority)

What's missing. test_uuid_seed has 5 rows in one parquet file, so planner.partition() == 0 and the expr.seed.wrapping_add(partition) offset is never exercised on the SQL path.

Why it matters. The whole reason RandomUUIDGenerator is seeded with randomSeed + partitionIndex is to decorrelate partitions. If planner.partition() ever returned the wrong value (e.g. usize
widening quirks, or a future refactor), the SQL suite would still pass — every partition would silently produce the same UUIDs and neither engine would flag it because both would be wrong in the same way
relative to a single-partition baseline. Only cross-engine comparison across ≥2 partitions catches this.

Add.

  -- forces multiple partitions so partitionIndex ≠ 0 is exercised
  statement
  CREATE TABLE test_uuid_parts(id int) USING parquet

  statement
  INSERT INTO test_uuid_parts SELECT id FROM range(0, 32)

  query
  SELECT uuid(42) FROM test_uuid_parts DISTRIBUTE BY id

Runs in default query mode → bit-for-bit vs Spark across partitions.

---

Gap 2 — Aliasing (stateful semantics)

What's missing. No test covers two uuid projections on the same row.

Why it matters. Spark's Uuid is stateful = true, and freshCopyIfContainsStatefulExpression rewrites aliased instances. For the seeded form uuid(0), uuid(0) both nodes carry the same explicit seed
→ identical sequences. For the unseeded form uuid(), uuid() the analyzer assigns fresh random seeds → different sequences. Comet must not accidentally share a single state_holder across the two nodes
(would produce interleaved draws) or hoist a single instance (would produce identical values for the unseeded case).

Add.

  -- uuid_with_seed.sql: seeded aliases must agree
  query
  SELECT uuid(0) = uuid(0) FROM test_uuid_seed

  -- uuid.sql: unseeded aliases must diverge (spark_answer_only — random)
  query spark_answer_only
  SELECT uuid() = uuid() FROM test_uuid

---

Gap 3 — Empty / zero-row batches

What's missing. evaluate with num_rows == 0 is not tested. The RNG must not advance on an empty batch.

Why it matters. A LIMIT 0 above a uuid projection, or a fully filtered partition, produces a zero-row batch. If the code ever grew a next_uuid() call outside the 0..num_rows loop (bootstrap,
warmup, etc.), the RNG state would drift and downstream bit-for-bit tests would fail intermittently.

Add (Rust unit test — cheaper than a SQL test for this):

  #[test]
  fn test_uuid_empty_batch_does_not_advance_state() {
      let expr = UuidExpr::new(7);
      let _ = collect_uuids(&expr, &empty_batch(0));
      // After an empty eval, the next non-empty eval must equal a fresh generator.
      assert_eq!(collect_uuids(&expr, &empty_batch(3)), eval_uuids(7, 3));
  }

---

Gap 4 — Rust golden fixture

What's missing. test_uuid_version_and_variant_bits only checks the version nibble (bytes[14] == b'4') and variant nibble (bytes[19] ∈ {8,9,a,b}). No test pins the full 128 bits.

Why it matters. next_long() (two next(32) composed high|low) is used only by uuid — the shuffle tests exercise next_int and would not catch a next_long regression (e.g. accidentally swapping
high/low, or masking with 0xffffffff in the wrong place). A single golden fixture makes the Rust suite a first-line defense; today it depends on cross-engine SQL tests running to catch a twister bug.

Add (values captured by running RandomUUIDGenerator(0) in Spark once):

  #[test]
  fn test_uuid_matches_spark_random_uuid_generator() {
      // Seeds and first three UUIDs captured from
      // `new RandomUUIDGenerator(0L).getNextUUID().toString` in spark-shell.
      let expected = vec![
          "…-4…-…", // fill from Spark
          "…-4…-…",
          "…-4…-…",
      ];
      assert_eq!(eval_uuids(0, 3), expected);
  }

Also add a case for a negative seed (e.g. Long.MIN_VALUE) to lock in the sign-extension path through set_seed_long.

---

Gap 5 — No-scan / typeof paths

What's missing. uuid_with_seed.sql has a SELECT uuid(0) line (no FROM), but no assertion around it — just the raw value. And no typeof check.

Why it matters. SELECT uuid(0) with no scan goes through a LocalTableScan/OneRowRelation execution shape rather than the parquet scan path. Different codegen/serde context, and worth pinning that
it doesn't fall back. typeof(uuid()) guards the dataType/collation contract without depending on the random value.

Add.

  -- uuid.sql
  query
  SELECT typeof(uuid())

  -- uuid_with_seed.sql — pin the no-scan path with a deterministic assertion
  query
  SELECT length(uuid(0)), typeof(uuid(0))

---

Priority order: Gap 1 first (real correctness risk), then Gap 4 (cheap Rust guard), then 2 / 3 / 5.