diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala index f441b7df9263e..2798e2a82d24f 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala +++ b/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 ()`, 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 diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/SampleSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/SampleSuite.scala new file mode 100644 index 0000000000000..0f3efb0c76983 --- /dev/null +++ b/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 ()`, 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)) + } +} diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/BasicStatsEstimationSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/BasicStatsEstimationSuite.scala index f07c19120438d..afb4618fea79c 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/BasicStatsEstimationSuite.scala +++ b/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)) } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala index e94b774a37d19..a45ea80c2ee52 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala +++ b/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 diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala index 1bd43e07afb2c..d8817b9aa7582 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala +++ b/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) {