mbutrovich commented on code in PR #5034: URL: https://github.com/apache/datafusion-comet/pull/5034#discussion_r3659786837
########## spark/src/test/resources/sql-tests/expressions/misc/uuid_with_seed.sql: ########## @@ -0,0 +1,79 @@ +-- 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. + +-- The one argument uuid(seed) form only exists in Spark 4.0+. With a fixed seed the output is +-- deterministic, so Comet must reproduce Spark's output exactly. Comet drives the same Commons +-- Math3 MersenneTwister as org.apache.spark.sql.catalyst.util.RandomUUIDGenerator, combining the +-- seed with the partition index, so these queries assert bit-for-bit equality with Spark in the +-- default query mode. + +-- MinSparkVersion: 4.0 + +statement +CREATE TABLE test_uuid_seed(id int) USING parquet + +statement +INSERT INTO test_uuid_seed VALUES (1), (2), (3), (4), (5) + +-- Multi-partition table used by the DISTRIBUTE BY query below to exercise partitionIndex != 0. +statement +CREATE TABLE test_uuid_parts(id int) USING parquet + +statement +INSERT INTO test_uuid_parts SELECT id FROM range(0, 32) + +-- fixed seed, single row -- also exercises the no-scan (OneRowRelation) planning path +query +SELECT uuid(0) + +-- pin the no-scan path with a deterministic assertion on top of the raw value above +query +SELECT length(uuid(0)) + +-- fixed seed, multiple rows: the generator advances per row within the partition +query +SELECT uuid(42) FROM test_uuid_seed + +-- Forces a hash exchange so uuid runs post-shuffle across several partitions. Locks the +-- `seed + partitionIndex` offset used by RandomUUIDGenerator; a single-partition test would +-- silently agree with Spark even if Comet ignored partitionIndex. +query +SELECT uuid(42) FROM test_uuid_parts DISTRIBUTE BY id Review Comment: This query does not exercise `partitionIndex != 0`, so the multi-partition coverage it was added for is still missing. The update says it makes "uuid run post-shuffle". The plan is the other way round. `DISTRIBUTE BY` is applied by `withQueryResultClauses`, which calls `withRepartitionByExpression(ctx, expressionList(distributeBy), query)` where `query` is the already-constructed plan including the `Project` (Spark `AstBuilder.scala:1272-1274`). So the shape is `RepartitionByExpression [id]` on top of `Project [uuid(42)]`, and uuid is evaluated on the scan's partitions, before the exchange. The exchange's partition count never reaches this expression. That leaves the scan's partition count, and it is very likely 1. `CometTestBase` runs `local[5]` (`CometTestBase.scala:545`), so `range(0, 32)` produces 5 partitions and the insert writes 5 tiny files. Under the default `spark.sql.files.maxPartitionBytes` of 128MB and `openCostInBytes` of 4MB, 5 tiny files pack into a single `FilePartition`. So `planner.partition()` is 0 and `expr.seed.wrapping_add(partition)` in `UuidBuilder::build` is never exercised on the SQL path. This is exactly the silent-agreement failure that motivated adding this query: with one partition, Comet and Spark agree even if the partition offset were dropped entirely, so the test passes while covering nothing. Please move the projection above the exchange so it runs on post-shuffle partitions: ```sql SELECT uuid(42) FROM (SELECT id FROM test_uuid_parts DISTRIBUTE BY id) ``` Since the value of this test rests entirely on an assumption about partition count, make that assumption checkable rather than implicit. A companion query such as `SELECT count(DISTINCT spark_partition_id()) FROM (SELECT id FROM test_uuid_parts DISTRIBUTE BY id)` fails loudly if the plan ever collapses back to one partition, instead of quietly returning to vacuous. The local run passing is consistent with either reading, since a one-partition plan agrees with Spark too, so the partition count is the thing to check rather than the pass. ########## spark/src/test/resources/sql-tests/expressions/misc/uuid_with_seed.sql: ########## @@ -0,0 +1,79 @@ +-- 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. + +-- The one argument uuid(seed) form only exists in Spark 4.0+. With a fixed seed the output is +-- deterministic, so Comet must reproduce Spark's output exactly. Comet drives the same Commons +-- Math3 MersenneTwister as org.apache.spark.sql.catalyst.util.RandomUUIDGenerator, combining the +-- seed with the partition index, so these queries assert bit-for-bit equality with Spark in the +-- default query mode. + +-- MinSparkVersion: 4.0 + +statement +CREATE TABLE test_uuid_seed(id int) USING parquet + +statement +INSERT INTO test_uuid_seed VALUES (1), (2), (3), (4), (5) + +-- Multi-partition table used by the DISTRIBUTE BY query below to exercise partitionIndex != 0. +statement +CREATE TABLE test_uuid_parts(id int) USING parquet + +statement +INSERT INTO test_uuid_parts SELECT id FROM range(0, 32) + +-- fixed seed, single row -- also exercises the no-scan (OneRowRelation) planning path +query +SELECT uuid(0) + +-- pin the no-scan path with a deterministic assertion on top of the raw value above +query +SELECT length(uuid(0)) Review Comment: This line replaced the `typeof(uuid(0))` variant, and the reason recorded for dropping it is not the actual cause. It is not a Spark 4.1 codegen quirk. It is a latent bug in Spark's `TypeOf.doGenCode`, which interpolates the catalog string unquoted into generated Java: ```scala defineCodeGen(ctx, ev, _ => s"""UTF8String.fromString(${child.dataType.catalogString})""") ``` For `typeof(uuid(0))` that emits `UTF8String.fromString(string)`, which is precisely the Janino error observed. It needs to be a Java string literal. This code is byte-identical on 3.5 (`misc.scala:280-282`), 4.0 (`:330-332`) and master (`:370-372`), so it reproduces on every supported version and has nothing to do with uuid. It is normally invisible because `TypeOf.foldable = true` and ConstantFolding eliminates the node before codegen runs. The SQL fixture suite excludes ConstantFolding, which is what exposes it. Two things worth capturing rather than losing. This deserves an upstream Spark report. And it is a harness-level constraint: no Comet SQL fixture can use `typeof` on a non-foldable input, which the next person will otherwise rediscover the same way. Please record the real cause in the description instead of the version-specific framing. ########## spark/src/test/resources/sql-tests/expressions/misc/uuid.sql: ########## @@ -0,0 +1,43 @@ +-- 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. + +-- uuid() runs natively. The raw random value cannot be compared against Spark because Spark assigns +-- a fresh random seed on each planning pass, so these queries assert deterministic properties that +-- hold identically on both engines. The seeded uuid(seed) form (Spark 4.0+) asserts bit-for-bit +-- equality with Spark and lives in uuid_with_seed.sql. + +statement +CREATE TABLE test_uuid(id int) USING parquet + +statement +INSERT INTO test_uuid VALUES (1), (2), (3), (4), (5) + +-- canonical form is 36 characters +query +SELECT length(uuid()) FROM test_uuid + +-- matches the RFC 4122 version 4 layout (version nibble 4, variant nibble 8/9/a/b), lowercase hex. +-- This regex subsumes the length, alphabet, version, and variant checks. +query +SELECT uuid() RLIKE '^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$' FROM test_uuid + +-- Two unseeded uuid() nodes get distinct fresh random seeds during analysis (Uuid is stateful, +-- freshCopyIfContainsStatefulExpression rewrites aliases). Every row must therefore compare +-- unequal -- if Comet ever hoisted a single instance, this would flip to true. +-- spark_answer_only: value is random, so we only cross-check that Spark and Comet agree. +query spark_answer_only +SELECT uuid() = uuid() FROM test_uuid Review Comment: `spark_answer_only` gives up the operator assertion here for no reason. `SparkAnswerOnly` maps to `checkSparkAnswer` (`CometSqlFileTestSuite.scala:153`), which compares values but does not assert the expression ran in Comet. The comment above says this query guards against Comet hoisting a single shared instance, but that guard only holds while the projection is actually native, which this mode does not check. It also does not need the relaxed mode. `uuid() = uuid()` is deterministic: the two nodes get distinct fresh seeds, so the result is `false` on every row on both engines. Dropping `spark_answer_only` and using default `query` mode keeps the value comparison and adds the nativeness assertion. ########## native/spark-expr/src/nondetermenistic_funcs/uuid.rs: ########## @@ -0,0 +1,281 @@ +// 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. + +use crate::nondetermenistic_funcs::internal::mersenne::SparkMersenneTwister; +use arrow::array::{RecordBatch, StringBuilder}; +use arrow::datatypes::{DataType, Schema}; +use datafusion::common::Result; +use datafusion::logical_expr::ColumnarValue; +use datafusion::physical_expr::PhysicalExpr; +use std::fmt::{Display, Formatter}; +use std::hash::{Hash, Hasher}; +use std::sync::{Arc, Mutex}; +use uuid::Uuid; + +/// Draw one RFC 4122 version 4 UUID from the generator, matching +/// `org.apache.spark.sql.catalyst.util.RandomUUIDGenerator.getNextUUID`: two +/// `nextLong()` draws with the version (4) and variant (10) bits masked in. +/// `Uuid`'s canonical lowercase hyphenated form matches `java.util.UUID.toString()`. +fn next_uuid(rng: &mut SparkMersenneTwister) -> Uuid { + let most = (rng.next_long() as u64 & 0xFFFF_FFFF_FFFF_0FFF) | 0x0000_0000_0000_4000; + let least = (rng.next_long() as u64 | 0x8000_0000_0000_0000) & 0xBFFF_FFFF_FFFF_FFFF; + Uuid::from_u64_pair(most, least) +} + +/// Physical expression for Spark's `uuid()`. Like `ShuffleExpr`, the generator +/// state is kept in a `Mutex` so that it advances continuously across every +/// batch in a partition, matching Spark's stateful per-partition evaluation. +/// Spark seeds a fresh `RandomUUIDGenerator` (a Commons Math3 `MersenneTwister`) +/// per partition with `randomSeed + partitionIndex`. +#[derive(Debug)] +pub struct UuidExpr { + /// Random seed already combined with the partition index by the planner. + seed: i64, + state_holder: Arc<Mutex<Option<SparkMersenneTwister>>>, +} + +impl UuidExpr { + pub fn new(seed: i64) -> Self { + Self { + seed, + state_holder: Arc::new(Mutex::new(None)), + } + } +} + +impl Display for UuidExpr { + fn fmt(&self, f: &mut Formatter) -> std::fmt::Result { + write!(f, "Uuid({})", self.seed) + } +} + +impl PartialEq for UuidExpr { + fn eq(&self, other: &Self) -> bool { + self.seed.eq(&other.seed) + } +} + +impl Eq for UuidExpr {} + +impl Hash for UuidExpr { + fn hash<H: Hasher>(&self, state: &mut H) { + self.seed.hash(state); + } +} + +impl PhysicalExpr for UuidExpr { + fn data_type(&self, _input_schema: &Schema) -> Result<DataType> { + Ok(DataType::Utf8) + } + + fn nullable(&self, _input_schema: &Schema) -> Result<bool> { + Ok(false) + } + + fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> { + let num_rows = batch.num_rows(); + + let mut state = self.state_holder.lock().unwrap(); + let rng = state.get_or_insert_with(|| SparkMersenneTwister::new(self.seed)); + + // Each canonical UUID is exactly 36 bytes, so pre-size both builder buffers and encode + // into a reused stack buffer to avoid a per-row heap allocation. + const LEN: usize = uuid::fmt::Hyphenated::LENGTH; + let mut builder = StringBuilder::with_capacity(num_rows, num_rows * LEN); + let mut buf = [0u8; LEN]; + for _ in 0..num_rows { + builder.append_value(next_uuid(rng).hyphenated().encode_lower(&mut buf)); + } + Ok(ColumnarValue::Array(Arc::new(builder.finish()))) + } + + fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> { + vec![] + } + + fn with_new_children( + self: Arc<Self>, + _children: Vec<Arc<dyn PhysicalExpr>>, + ) -> Result<Arc<dyn PhysicalExpr>> { + Ok(Arc::new(UuidExpr::new(self.seed))) + } + + fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + Display::fmt(self, f) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use arrow::array::{Array, RecordBatchOptions, StringArray}; + + fn empty_batch(num_rows: usize) -> RecordBatch { + RecordBatch::try_new_with_options( + Arc::new(Schema::empty()), + vec![], + &RecordBatchOptions::new().with_row_count(Some(num_rows)), + ) + .unwrap() + } + + fn collect_uuids(expr: &UuidExpr, batch: &RecordBatch) -> Vec<String> { + let arr = expr + .evaluate(batch) + .unwrap() + .into_array(batch.num_rows()) + .unwrap(); + let arr = arr.as_any().downcast_ref::<StringArray>().unwrap(); + (0..arr.len()).map(|i| arr.value(i).to_string()).collect() + } + + fn eval_uuids(seed: i64, num_rows: usize) -> Vec<String> { + collect_uuids(&UuidExpr::new(seed), &empty_batch(num_rows)) + } + + #[test] + fn test_uuid_version_and_variant_bits() { + // The RNG and the version/variant masking are ours; the canonical string layout is + // guaranteed by the `uuid` crate. Assert the RFC 4122 v4 bits our masking sets, at their + // fixed positions in `xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx`. + for uuid in eval_uuids(42, 20) { + let bytes = uuid.as_bytes(); + assert_eq!(bytes[14], b'4'); // version nibble + assert!(matches!(bytes[19], b'8' | b'9' | b'a' | b'b')); // variant nibble + } + } + + #[test] + fn test_uuid_deterministic_for_seed() { + // Same seed -> identical sequence. + assert_eq!(eval_uuids(42, 5), eval_uuids(42, 5)); + // Different seed -> different sequence. + assert_ne!(eval_uuids(42, 5), eval_uuids(0, 5)); + } + + #[test] + fn test_uuid_state_advances_across_batches() { + // A single expression evaluated over two batches yields the same UUIDs as + // one batch of the combined size (state persists across batches). + let expr = UuidExpr::new(7); + let mut streamed = Vec::new(); + for n in [3usize, 4usize] { + streamed.extend(collect_uuids(&expr, &empty_batch(n))); + } + assert_eq!(streamed, eval_uuids(7, 7)); + // All distinct. + let mut sorted = streamed.clone(); + sorted.sort(); + sorted.dedup(); + assert_eq!(sorted.len(), streamed.len()); + } + + #[test] + fn test_uuid_empty_batch_does_not_advance_state() { + // A zero-row batch (e.g. `LIMIT 0` over a `uuid` projection, or a fully filtered + // partition) must not consume any RNG draws. If the code ever grew a `next_uuid()` + // call outside the `0..num_rows` loop, the state would drift and downstream + // bit-for-bit tests would fail intermittently. + let expr = UuidExpr::new(7); + let _ = collect_uuids(&expr, &empty_batch(0)); + let after_empty = collect_uuids(&expr, &empty_batch(3)); + assert_eq!(after_empty, eval_uuids(7, 3)); + } + + /// Golden fixtures captured from Commons Math3's `MersenneTwister` (the exact RNG that + /// backs Spark's `org.apache.spark.sql.catalyst.util.RandomUUIDGenerator`). Regenerate + /// with a tiny Java program: + /// + /// ```java + /// import org.apache.commons.math3.random.MersenneTwister; + /// import java.util.UUID; + /// static UUID next(MersenneTwister r) { + /// long m = (r.nextLong() & 0xFFFFFFFFFFFF0FFFL) | 0x0000000000004000L; + /// long l = (r.nextLong() | 0x8000000000000000L) & 0xBFFFFFFFFFFFFFFFL; + /// return new UUID(m, l); + /// } + /// ``` + /// + /// `next_long()` is used *only* by `uuid`, so the shuffle tests (which exercise + /// `next_int`) do not guard against a `next_long` regression such as swapping the + /// high/low words or masking with the wrong constant. This test locks all 128 bits + /// per row, catches sign-extension bugs in `set_seed_long` via the negative seeds, + /// and runs entirely in Rust so it fires without a JVM roundtrip. + #[test] + fn test_uuid_matches_commons_math3_random_uuid_generator() { Review Comment: On 3.4 and 3.5 these golden values are the only line of defense, and their provenance cannot be checked from the repo. The doc comment above documents how to regenerate them with a Java snippet, which is the right instinct, but nothing here lets a reviewer confirm the committed values came from Commons Math3 rather than from this same Rust code. If they are ever regenerated from the Rust side the test becomes circular and silently stops testing anything. On 4.0+ that risk is contained, because `uuid_with_seed.sql` compares against Spark bit for bit and would catch a `next_long` port bug regardless. On 3.4 and 3.5 there is no such backstop: the seeded fixture is skipped by `MinSparkVersion: 4.0`, and unseeded `uuid()` cannot be compared across engines, so a twister regression on those profiles would be caught only by these constants. `Uuid(Some(seed))` is constructible from Scala on 3.4 and 3.5 even though the SQL form is not. A small Scala test building the expression directly and comparing Comet against Spark would give every profile a bit-for-bit assertion and remove the dependence on hand-captured constants. That closes the one place where this PR can regress without any test noticing. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
