(spark) branch master updated: [SPARK-57440][SQL] Gate BIN BY behind spark.sql.binByRelationOperator.enabled

[we...@apache.org]

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š <ni...@databricks.com>
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š <ni...@databricks.com>
    Signed-off-by: Wenchen Fan <we...@databricks.com>
---
 .../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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org