This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 85bc81df8434 [SPARK-56573][SQL] Use a full-range non-negative random
seed for unseeded sampling
85bc81df8434 is described below
commit 85bc81df84341129ef99a812e95d114ce0e60638
Author: Stanley Yao <[email protected]>
AuthorDate: Wed Aug 5 17:17:24 2026 +0800
[SPARK-56573][SQL] Use a full-range non-negative random seed for unseeded
sampling
### What changes were proposed in this pull request?
When a `Sample` has no user-specified seed, Spark generates one with
`(math.random() * 1000).toLong`. This PR replaces both call sites with a shared
`Sample.resolveSeed` helper that returns a non-negative 63-bit seed via
`Utils.random.nextLong() & Long.MaxValue`.
The seed is generated in two places -- `SampleExec.resolvedSeed` and
`V2ScanRelationPushDown.pushDownSample` -- which SPARK-56392 duplicated when it
moved the expression out of `AstBuilder`. Both now delegate to one helper so
they cannot drift apart.
Generated seeds must be non-negative: a pushed-down sample renders its seed
into SQL as `REPEATABLE (<seed>)`, and the seed in that grammar accepts no
sign. A user-specified seed passes through unchanged, negative values included.
### Why are the changes needed?
A 1000-value seed space means Spark can produce at most 1000 distinct
samples of a table regardless of its size. Two unseeded samples are identical
0.1% of the time; among ~37 samples the odds of some pair colliding exceed 50%;
1000 bootstrap resamples yield only ~632 distinct samples.
This is not a regression -- the expression dates to SPARK-1251 (2014), and
SPARK-56392 relocated it verbatim. But `TABLESAMPLE SYSTEM` rejects the
`REPEATABLE` clause, so block-sampling users cannot pin a seed and always take
this path, making the generated seed their only source of variation.
`TABLESAMPLE` was the last sampling path in Spark still using a narrow
seed. PySpark (`random.randint(0, sys.maxsize)`) and the RDD APIs
(`Utils.random.nextLong`) already draw from the full range, so this removes an
outlier rather than introducing a new convention.
Separately considered and intentionally left out of scope:
`SampleExec.resolvedSeed` is a non-constructor `val`, so structurally identical
`SampleExec` nodes canonicalize equal while holding different seeds. Whether
plan reuse can collapse two independent samples into one is an independence
question that seed width does not address, and it warrants its own JIRA.
### Does this PR introduce _any_ user-facing change?
No behavior change users can depend on. Unseeded sampling was already
nondeterministic; it now draws from a much larger seed space. Explicitly seeded
sampling (`REPEATABLE(n)`, `sample(fraction, seed)`) is unaffected.
`TABLESAMPLE SYSTEM` is unreleased, so no released behavior changes there.
### How was this patch tested?
New `SampleSuite` covering seed passthrough (including negative
user-specified seeds), non-negativity of generated seeds, and distinctness
across 10000 draws.
Existing suites were run across catalyst, sql/core and connect:
`PlanParserSuite`, `DataSourceV2TableSampleSuite`, `JDBCV2Suite`, `JDBCSuite`,
`BasicStatsEstimationSuite`, `SparkConnectProtoSuite`, `DataFrameSuite`,
`DatasetSuite`, `SQLQuerySuite`, `ColumnPruningSuite`, `CollapseProjectSuite`,
`NestedColumnAliasingSuite`, `UnsupportedOperationsSuite`, `AnalysisErrorSuite`
and others -- 2302 tests passing.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored-by: Claude Code (Claude Opus 5) with thorough human review and
iterations
Closes #57732 from stanyao/spark-56573-tablesample-seed.
Lead-authored-by: Stanley Yao <[email protected]>
Co-authored-by: stanleyy <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../plans/logical/basicLogicalOperators.scala | 12 ++++++
.../sql/catalyst/plans/logical/SampleSuite.scala | 50 ++++++++++++++++++++++
.../BasicStatsEstimationSuite.scala | 4 +-
.../sql/execution/basicPhysicalOperators.scala | 3 +-
.../datasources/v2/V2ScanRelationPushDown.scala | 5 +--
5 files changed, 67 insertions(+), 7 deletions(-)
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 f441b7df9263..2798e2a82d24 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 (<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
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 000000000000..0f3efb0c7698
--- /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
(<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))
+ }
+}
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 f07c19120438..afb4618fea79 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 e94b774a37d1..a45ea80c2ee5 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 1bd43e07afb2..d8817b9aa758 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) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
