[SPARK-56573][SQL] Use a full-range non-negative random seed for unseeded sampling

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

@@ -2069,6 +2069,18 @@ object SampleMethod {
 }
 
 object Sample {
+  /**
+   * Resolves the seed of a sample, generating a random one when the user did not specify one.
+   *
+   * Generated seeds are non-negative. A pushed-down sample renders its seed into SQL as
+   * `REPEATABLE (<seed>)`, and the seed in that grammar does not accept a sign. A
+   * user-specified seed is returned unchanged, negative values included.
+   */
+  def resolveSeed(seed: Option[Long]): Long = {
+    // `Utils` in this file is o.a.s.util.collection.Utils, so qualify the one we want here.
+    seed.getOrElse(org.apache.spark.util.Utils.random.nextLong() & Long.MaxValue)
+  }
+
   /**
    * Convenience constructor that wraps a concrete seed in [[Some]].
    * Use the case-class constructor directly with [[None]] when no seed

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/SampleSuite.scala

@@ -0,0 +1,50 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.catalyst.plans.logical
+
+import org.apache.spark.SparkFunSuite
+
+class SampleSuite extends SparkFunSuite {
+
+  test("resolveSeed returns a user-specified seed unchanged") {
+    assert(Sample.resolveSeed(Some(42L)) === 42L)
+    assert(Sample.resolveSeed(Some(0L)) === 0L)
+    assert(Sample.resolveSeed(Some(Long.MaxValue)) === Long.MaxValue)
+    // Only generated seeds are constrained to be non-negative. The Dataset API accepts a
+    // negative seed even though the SQL REPEATABLE grammar does not, so it must pass through.
+    assert(Sample.resolveSeed(Some(-5L)) === -5L)
+    assert(Sample.resolveSeed(Some(Long.MinValue)) === Long.MinValue)
+  }
+
+  test("resolveSeed generates non-negative seeds") {
+    // A pushed-down sample renders its seed into SQL as `REPEATABLE (<seed>)`, and the seed
+    // in that grammar does not accept a sign.
+    for (_ <- 0 until 10000) {
+      assert(Sample.resolveSeed(None) >= 0L)
+    }
+  }
+
+  test("resolveSeed draws from a wide range of values") {
+    // Guards against SPARK-56573, where the generated seed was limited to 1000 distinct
+    // values. Drawing from 2^63 makes 1000 collisions in 10000 draws effectively impossible.
+    val seeds = Seq.fill(10000)(Sample.resolveSeed(None)).toSet
+    assert(seeds.size > 9000, s"expected nearly all seeds to be distinct, got ${seeds.size}")
+    // The old implementation could never exceed 999.
+    assert(seeds.exists(_ > 1000L))
+  }
+}

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/BasicStatsEstimationSuite.scala

@@ -249,14 +249,14 @@ test("range with invalid long value") {
   }
 
   test("sample estimation") {
-    val sample = Sample(0.0, 0.5, withReplacement = false, (math.random() * 1000).toLong, plan)
+    val sample = Sample(0.0, 0.5, withReplacement = false, Sample.resolveSeed(None), plan)
     checkStats(sample, Statistics(sizeInBytes = 60, rowCount = Some(5)))
 
     // Child doesn't have rowCount in stats
     val childStats = Statistics(sizeInBytes = 120)
     val childPlan = DummyLogicalPlan(childStats, childStats)
     val sample2 =
-      Sample(0.0, 0.11, withReplacement = false, (math.random() * 1000).toLong, childPlan)
+      Sample(0.0, 0.11, withReplacement = false, Sample.resolveSeed(None), childPlan)
     checkStats(sample2, Statistics(sizeInBytes = 14))
   }
 

sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala

@@ -32,6 +32,7 @@ import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.BindReferences.bindReferences
 import org.apache.spark.sql.catalyst.expressions.codegen._
 import org.apache.spark.sql.catalyst.optimizer.CollapseProject
+import org.apache.spark.sql.catalyst.plans.logical.Sample
 import org.apache.spark.sql.catalyst.plans.physical._
 import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
 import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
@@ -497,7 +498,7 @@ case class SampleExec(
     seed: Option[Long],
     child: SparkPlan) extends UnaryExecNode with CodegenSupport {
 
-  val resolvedSeed: Long = seed.getOrElse((math.random() * 1000).toLong)
+  val resolvedSeed: Long = Sample.resolveSeed(seed)
 
   override def output: Seq[Attribute] = child.output
 

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala

@@ -1053,10 +1053,7 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
           sample.lowerBound,
           sample.upperBound,
           sample.withReplacement,
-          // TODO(SPARK-56573): The * 1000 limits the seed to only 1000 distinct values.
-          //   Kept here for consistency with SampleExec.resolvedSeed; will be fixed
-          //   across all call sites in SPARK-56573.
-          sample.seed.getOrElse((math.random() * 1000).toLong),
+          Sample.resolveSeed(sample.seed),
           sampleMethod = sample.sampleMethod)
         val pushed = PushDownUtils.pushTableSample(sHolder.builder, tableSample)
         if (pushed) {