This is an automated email from the ASF dual-hosted git repository.
philo-he pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new f04d7dfb02 [GLUTEN-13178][VL] Throw Spark's error for a broadcast
table over maxBroadcastTableSize (#13192)
f04d7dfb02 is described below
commit f04d7dfb0207143fc8e294652b4c08faaa62fb80
Author: rdtr <[email protected]>
AuthorDate: Tue Oct 6 23:04:11 2026 -0700
[GLUTEN-13178][VL] Throw Spark's error for a broadcast table over
maxBroadcastTableSize (#13192)
---
.../gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala | 10 ++++------
.../org/apache/spark/sql/execution/BroadcastUtils.scala | 13 +++++++++++++
.../org/apache/gluten/execution/VeloxHashJoinSuite.scala | 12 +++++++++++-
3 files changed, 28 insertions(+), 7 deletions(-)
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
index 38f2bdbd38..90dfc0b260 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
@@ -30,10 +30,9 @@ import org.apache.gluten.substrait.SubstraitContext
import org.apache.gluten.substrait.expression.{ExpressionBuilder,
ExpressionNode, WindowFunctionNode}
import org.apache.gluten.vectorized.{ColumnarBatchSerializer,
ColumnarBatchSerializeResult}
-import org.apache.spark.{ShuffleDependency, SparkEnv, SparkException}
+import org.apache.spark.{ShuffleDependency, SparkEnv}
import org.apache.spark.api.python.{ColumnarArrowEvalPythonExec,
PullOutArrowEvalPythonPreProjectHelper}
import org.apache.spark.internal.Logging
-import org.apache.spark.memory.SparkMemoryUtil
import org.apache.spark.rdd.RDD
import org.apache.spark.serializer.Serializer
import org.apache.spark.shuffle.{GenShuffleReaderParameters,
GenShuffleWriterParameters, GlutenShuffleReaderWrapper,
GlutenShuffleWriterWrapper, VeloxShuffleUtils}
@@ -887,10 +886,9 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi with
Logging {
val buildSideRowCount = serialized.map(_.numRows).sum
val rawSize = serialized.map(_.sizeInBytes()).sum
if (rawSize >= GlutenConfig.get.maxBroadcastTableSize) {
- throw new SparkException(
- "Cannot broadcast the table that is larger than " +
-
s"${SparkMemoryUtil.bytesToString(GlutenConfig.get.maxBroadcastTableSize)}: " +
- s"${SparkMemoryUtil.bytesToString(rawSize)}")
+ throw BroadcastUtils.cannotBroadcastTableOverMaxTableBytesError(
+ GlutenConfig.get.maxBroadcastTableSize,
+ rawSize)
}
numOutputRows += buildSideRowCount
dataSize += rawSize
diff --git
a/backends-velox/src/main/scala/org/apache/spark/sql/execution/BroadcastUtils.scala
b/backends-velox/src/main/scala/org/apache/spark/sql/execution/BroadcastUtils.scala
index 3746f9529b..4c7ff0d5f6 100644
---
a/backends-velox/src/main/scala/org/apache/spark/sql/execution/BroadcastUtils.scala
+++
b/backends-velox/src/main/scala/org/apache/spark/sql/execution/BroadcastUtils.scala
@@ -28,6 +28,7 @@ import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.UnsafeRow
import org.apache.spark.sql.catalyst.plans.physical.{BroadcastMode,
BroadcastPartitioning, IdentityBroadcastMode, Partitioning}
+import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.joins.{BuildSideRelation,
EmptyHashedRelation, HashedRelation, HashedRelationBroadcastMode,
LongHashedRelation}
import org.apache.spark.sql.execution.unsafe.UnsafeColumnarBuildSideRelation
import org.apache.spark.sql.types.StructType
@@ -140,6 +141,18 @@ object BroadcastUtils {
}
}
+ /**
+ * Returns the error vanilla Spark's BroadcastExchangeExec throws when the
broadcast table is too
+ * large, so that it has the same error condition.
+ */
+ def cannotBroadcastTableOverMaxTableBytesError(
+ maxBroadcastTableBytes: Long,
+ dataSize: Long): Throwable = {
+ QueryExecutionErrors.cannotBroadcastTableOverMaxTableBytesError(
+ maxBroadcastTableBytes,
+ dataSize)
+ }
+
def getBroadcastMode(partitioning: Partitioning): BroadcastMode = {
partitioning match {
case BroadcastPartitioning(mode) =>
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala
index e9a599d8d2..0b86b036b6 100644
---
a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala
@@ -19,13 +19,14 @@ package org.apache.gluten.execution
import org.apache.gluten.config.{GlutenConfig, VeloxConfig}
import org.apache.gluten.sql.shims.SparkShimLoader
-import org.apache.spark.SparkConf
+import org.apache.spark.{SparkConf, SparkException}
import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.expressions.AttributeReference
import org.apache.spark.sql.execution.{ColumnarBroadcastExchangeExec,
ColumnarSubqueryBroadcastExec, InputIteratorTransformer,
SerializedHashTableBroadcastRelation}
import org.apache.spark.sql.execution.exchange.ReusedExchangeExec
import org.apache.spark.sql.execution.joins.BuildSideRelation
import org.apache.spark.sql.execution.joins.HashedRelationBroadcastMode
+import org.apache.spark.sql.functions.broadcast
class VeloxHashJoinSuite extends VeloxWholeStageTransformerSuite {
override protected val resourcePath: String = "/tpch-data-parquet"
@@ -687,4 +688,13 @@ class VeloxHashJoinSuite extends
VeloxWholeStageTransformerSuite {
})
}
+ test("Broadcast over maxBroadcastTableSize fails with Spark's error
condition") {
+ withSQLConf("spark.sql.maxBroadcastTableSize" -> "100") {
+ val df = spark.range(1000).toDF()
+ val ex = intercept[SparkException] {
+ df.join(broadcast(df), "id").collect()
+ }
+ assert(ex.getErrorClass == "_LEGACY_ERROR_TEMP_2249")
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
