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 875982c4a65b [SPARK-57440][SQL] Gate BIN BY behind
spark.sql.binByRelationOperator.enabled
875982c4a65b is described below
commit 875982c4a65bc703bce88fb17683b423d8f88255
Author: Nikolina Vraneš <[email protected]>
AuthorDate: Sun Jun 14 16:00:38 2026 -0700
[SPARK-57440][SQL] Gate BIN BY behind
spark.sql.binByRelationOperator.enabled
### What changes were proposed in this pull request?
Follows up #56426 (BIN BY parsing and resolution). Tracked under
SPARK-57440.
Adds an internal session config `spark.sql.binByRelationOperator.enabled`
(default `false`) that gates the `BIN BY` relation operator at analysis time.
- New `BIN_BY_ENABLED` conf in `SQLConf` (config key
`spark.sql.binByRelationOperator.enabled`).
- `ResolveBinBy` short-circuits to
`QueryCompilationErrors.binByDisabledError()` when the conf is off, before
resolving the operator.
- The gate reuses the existing `UNSUPPORTED_FEATURE.BIN_BY` condition
(already raised by the execution stub) rather than introducing a new one. Its
message is generalized to "The BIN BY relation operator is not yet supported."
so both the analysis-time gate (flag off) and the execution stub (flag on)
surface the same condition.
### Why are the changes needed?
`BIN BY` resolves today but has no physical execution yet (#56426 stubs
it). A config gate keeps the incomplete operator disabled by default and lets
it be enabled for development and testing, instead of exposing a
half-implemented operator. Defaulting off is honest while execution is
unimplemented.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- `ResolveBinBySuite`: existing cases run with the flag enabled (via a
`test()` override); a new `super.test` case asserts the operator is rejected
with `UNSUPPORTED_FEATURE.BIN_BY` when the flag is off.
- `BinBySuite`: the stub test runs with the flag on; a new case asserts the
flag-off gate fails at analysis.
- `SparkThrowableSuite` passes (error-conditions.json golden).
- `catalyst/Test/compile` + `sql/Test/compile` and scalastyle on the
touched modules are clean.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Fable 5)
Closes #56487 from vranes/bin-by-oss-config.
Authored-by: Nikolina Vraneš <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 2 +-
.../spark/sql/catalyst/analysis/ResolveBinBy.scala | 2 +
.../spark/sql/errors/QueryCompilationErrors.scala | 6 +++
.../org/apache/spark/sql/internal/SQLConf.scala | 10 ++++
.../sql/catalyst/analysis/ResolveBinBySuite.scala | 20 +++++++
.../scala/org/apache/spark/sql/BinBySuite.scala | 61 +++++++++++++++-------
6 files changed, 81 insertions(+), 20 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 3d1ca002719b..e0bea26dc960 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -8094,7 +8094,7 @@
},
"BIN_BY" : {
"message" : [
- "Physical execution of BIN BY is not yet implemented."
+ "The BIN BY relation operator is not yet supported."
]
},
"CATALOG_INTERFACE_METHOD" : {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBy.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBy.scala
index e2d1d41dfcb0..1f7d48787d78 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBy.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBy.scala
@@ -39,6 +39,8 @@ object ResolveBinBy extends Rule[LogicalPlan] {
override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsWithPruning(
_.containsPattern(UNRESOLVED_BIN_BY), ruleId) {
+ case _: UnresolvedBinBy if !SQLConf.get.getConf(SQLConf.BIN_BY_ENABLED) =>
+ throw QueryCompilationErrors.binByDisabledError()
case b: UnresolvedBinBy if !readyToResolve(b) => b
case b: UnresolvedBinBy => resolve(b)
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index ff2fa1684202..7f76f48cc7ef 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -313,6 +313,12 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase with Compilat
messageParameters = Map("columnName" -> toSQLId(columnName)))
}
+ def binByDisabledError(): Throwable = {
+ new AnalysisException(
+ errorClass = "UNSUPPORTED_FEATURE.BIN_BY",
+ messageParameters = Map.empty)
+ }
+
def binByRequiresTopLevelColumnError(columnName: String): Throwable = {
new AnalysisException(
errorClass = "BIN_BY_REQUIRES_TOP_LEVEL_COLUMN",
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 83e28f6a28df..4f76dec323ef 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -5193,6 +5193,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val BIN_BY_ENABLED =
+ buildConf("spark.sql.binByRelationOperator.enabled")
+ .doc("Enable the BIN BY relation operator for aligning range-typed rows
to " +
+ "fixed-width bin boundaries.")
+ .internal()
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(false)
+
object StoreAssignmentPolicy extends Enumeration {
val ANSI, LEGACY, STRICT = Value
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBySuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBySuite.scala
index 5ff103e8288e..5b182cb7c38c 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBySuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBySuite.scala
@@ -17,6 +17,9 @@
package org.apache.spark.sql.catalyst.analysis
+import org.scalactic.source.Position
+import org.scalatest.Tag
+
import org.apache.spark.SparkThrowable
import org.apache.spark.sql.catalyst.QueryPlanningTracker
import org.apache.spark.sql.catalyst.dsl.expressions._
@@ -28,6 +31,17 @@ import org.apache.spark.sql.types._
class ResolveBinBySuite extends AnalysisTest {
+ // BIN BY is gated off by default; run the resolution tests with it enabled.
The dedicated
+ // gate test below uses `super.test` to observe the default-off behavior.
+ override protected def test(testName: String, testTags: Tag*)(testFun: =>
Any)
+ (implicit pos: Position): Unit = {
+ super.test(testName, testTags: _*) {
+ withSQLConf(SQLConf.BIN_BY_ENABLED.key -> "true") {
+ testFun
+ }
+ }
+ }
+
private val tsStart = $"ts_start".timestamp
private val tsEnd = $"ts_end".timestamp
private val tsStartNtz = $"ts_start".timestampNTZ
@@ -304,4 +318,10 @@ class ResolveBinBySuite extends AnalysisTest {
assert(appendedExprIds.distinct.size == appendedExprIds.size,
"appended BinBy attributes must have distinct exprIds across the two
join sides")
}
+
+ // `super.test` escapes the suite-wide flag-on wrapper so this runs with the
default (off).
+ super.test("BIN BY is gated off by default") {
+ assert(!SQLConf.get.getConf(SQLConf.BIN_BY_ENABLED))
+ expectError(unresolved(), "UNSUPPORTED_FEATURE.BIN_BY")
+ }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala
index 36ea5c3ba08f..8ec0a1a44a8c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala
@@ -17,32 +17,55 @@
package org.apache.spark.sql
-import org.apache.spark.SparkUnsupportedOperationException
+import org.apache.spark.{SparkThrowable, SparkUnsupportedOperationException}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
class BinBySuite extends QueryTest with SharedSparkSession {
+ private def createMetricsView(): Unit = {
+ spark.sql(
+ """SELECT TIMESTAMP '2024-01-01 00:00:00' AS ts_start,
+ | TIMESTAMP '2024-01-01 01:00:00' AS ts_end,
+ | CAST(1 AS DOUBLE) AS
value""".stripMargin).createOrReplaceTempView("metrics")
+ }
+
+ private val binByQuery =
+ """SELECT * FROM metrics BIN BY (
+ | RANGE ts_start TO ts_end
+ | BIN WIDTH INTERVAL '5' MINUTE
+ | DISTRIBUTE UNIFORM (value)
+ |)""".stripMargin
+
test("BIN BY analyzes but physical execution is not yet implemented") {
+ withSQLConf(SQLConf.BIN_BY_ENABLED.key -> "true") {
+ withTempView("metrics") {
+ createMetricsView()
+ val df = spark.sql(binByQuery)
+
+ // Analysis is fully functional when the operator is enabled.
+ df.queryExecution.assertAnalyzed()
+
+ // Physical execution is stubbed until the follow-up PR; planning
surfaces a clean
+ // UNSUPPORTED_FEATURE error rather than an internal error.
+ checkError(
+ exception = intercept[SparkUnsupportedOperationException] {
+ df.collect()
+ },
+ condition = "UNSUPPORTED_FEATURE.BIN_BY",
+ parameters = Map.empty[String, String])
+ }
+ }
+ }
+
+ test("BIN BY is gated off by default") {
withTempView("metrics") {
- spark.sql(
- """SELECT TIMESTAMP '2024-01-01 00:00:00' AS ts_start,
- | TIMESTAMP '2024-01-01 01:00:00' AS ts_end,
- | CAST(1 AS DOUBLE) AS
value""".stripMargin).createOrReplaceTempView("metrics")
- val df = spark.sql(
- """SELECT * FROM metrics BIN BY (
- | RANGE ts_start TO ts_end
- | BIN WIDTH INTERVAL '5' MINUTE
- | DISTRIBUTE UNIFORM (value)
- |)""".stripMargin)
-
- // Analysis is fully functional in this PR.
- df.queryExecution.assertAnalyzed()
-
- // Physical execution is stubbed until the follow-up PR; planning
surfaces a clean
- // UNSUPPORTED_FEATURE error rather than an internal error.
+ createMetricsView()
+ // Gated off, the operator is rejected at analysis with the same
UNSUPPORTED_FEATURE.BIN_BY
+ // condition the execution stub raises when enabled.
checkError(
- exception = intercept[SparkUnsupportedOperationException] {
- df.collect()
+ exception = intercept[SparkThrowable] {
+ spark.sql(binByQuery).queryExecution.assertAnalyzed()
},
condition = "UNSUPPORTED_FEATURE.BIN_BY",
parameters = Map.empty[String, String])
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
