You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by "felixloesing (via GitHub)" <gi...@apache.org> on 2025/08/26 22:47:13 UTC
[PR] [GLUTEN-9671][VL] Fix broadcast exchange stackoverflow due to Kryo serialization [incubator-gluten]
felixloesing opened a new pull request, #10541:
URL: https://github.com/apache/incubator-gluten/pull/10541
This pull request introduces a safer and more robust approach for handling Spark's `BroadcastMode` during serialization. The main improvement is the introduction of a new `SafeBroadcastMode` abstraction and related utilities, which help avoid serialization issues that caused a Stackoverflow exception during broadcast exchanges. BroadcastMode was introduced in this [PR](https://github.com/apache/incubator-gluten/pull/8116) that caused the issue we observed. HashedRelationBroadcastMode embeds Catalyst expression trees, which are not safe to Kryo-serialize when running with `spark.kryo.referenceTracking=false` (default internally).
With this change, the broadcast payload now contains only primitives and byte arrays (no Catalyst trees). For bound keys, we serialize just column ordinals (+ null-aware flag) and for computed keys (e.g., upper(col)), we serialize the key expressions once as Java bytes and deserialize only where needed to build projections.
#### Test Plan
Ran internal test set (50 queries) and ran other query specifically checking if
`spark.gluten.velox.offHeapBroadcastBuildRelation.enabled=true;` works.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9671][VL] Fix broadcast exchange stackoverflow due to Kryo serialization [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #10541:
URL: https://github.com/apache/incubator-gluten/pull/10541#issuecomment-3225964190
https://github.com/apache/incubator-gluten/issues/9671
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9671][VL] Fix broadcast exchange stackoverflow due to Kryo serialization [incubator-gluten]
Posted by "JkSelf (via GitHub)" <gi...@apache.org>.
JkSelf commented on code in PR #10541:
URL: https://github.com/apache/incubator-gluten/pull/10541#discussion_r2306600659
##########
backends-velox/src/main/scala/org/apache/spark/sql/execution/BroadcastModeUtils.scala:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.execution
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BoundReference, Expression}
+import org.apache.spark.sql.catalyst.plans.physical.{BroadcastMode, IdentityBroadcastMode}
+import org.apache.spark.sql.execution.joins.HashedRelationBroadcastMode
+
+import java.io.{ByteArrayInputStream, ByteArrayOutputStream, ObjectInputStream, ObjectOutputStream}
+
+/**
+ * Provides serialization-safe representations of BroadcastMode to avoid issues with circular
+ * references in complex expression trees during Kryo serialization.
+ */
+sealed trait SafeBroadcastMode extends Serializable
+
+/** Safe representation of IdentityBroadcastMode */
+case object IdentitySafeBroadcastMode extends SafeBroadcastMode
+
+/**
+ * Safe wrapper for HashedRelationBroadcastMode. Stores only column ordinals instead of full
+ * BoundReference expressions.
+ */
+final case class HashSafeBroadcastMode(ordinals: Array[Int], isNullAware: Boolean)
+ extends SafeBroadcastMode
+
+/**
+ * Safe wrapper for HashedRelationBroadcastMode when keys are not simple BoundReferences. Stores key
+ * expressions as serialized Java bytes.
+ */
+final case class HashExprSafeBroadcastMode(exprBytes: Array[Byte], isNullAware: Boolean)
+ extends SafeBroadcastMode
+
+object BroadcastModeUtils {
+
+ /**
+ * Converts a BroadcastMode to its SafeBroadcastMode equivalent. Uses ordinals for simple
+ * BoundReferences, otherwise serializes the expressions.
+ */
+ private[execution] def toSafe(mode: BroadcastMode): SafeBroadcastMode = mode match {
+ case IdentityBroadcastMode =>
+ IdentitySafeBroadcastMode
+ case HashedRelationBroadcastMode(keys, isNullAware) =>
+ // Fast path: all keys are already BoundReference(i, ..,..).
+ val ords = keys.collect { case BoundReference(ord, _, _) => ord }
+ if (ords.size == keys.size) {
+ HashSafeBroadcastMode(ords.toArray, isNullAware)
+ } else {
+ // Fallback: store the key expressions as Java-serialized bytes.
+ HashExprSafeBroadcastMode(serializeExpressions(keys), isNullAware)
+ }
+
+ case other =>
+ throw new IllegalArgumentException(s"Unsupported BroadcastMode: $other")
+ }
+
+ /** Converts a SafeBroadcastMode to its BroadcastMode equivalent. */
+ private[execution] def fromSafe(safe: SafeBroadcastMode, output: Seq[Attribute]): BroadcastMode =
+ safe match {
+ case IdentitySafeBroadcastMode =>
+ IdentityBroadcastMode
+
+ case HashSafeBroadcastMode(ords, isNullAware) =>
+ val bound = ords.map(i => BoundReference(i, output(i).dataType, output(i).nullable)).toSeq
+ HashedRelationBroadcastMode(bound, isNullAware)
+
+ case HashExprSafeBroadcastMode(bytes, isNullAware) =>
+ HashedRelationBroadcastMode(deserializeExpressions(bytes), isNullAware)
+ }
+
+ // Helpers for expression serialization (used in HashExprSafeBroadcastMode)
+ private[execution] def serializeExpressions(keys: Seq[Expression]): Array[Byte] = {
+ val bos = new ByteArrayOutputStream()
+ var oos: ObjectOutputStream = null
+ try {
+ oos = new ObjectOutputStream(bos)
+ oos.writeObject(keys)
+ oos.flush()
+ bos.toByteArray
+ } finally {
Review Comment:
nit: Can we add a catch clause to log the exception and make debugging easier?
##########
backends-velox/src/main/scala/org/apache/spark/sql/execution/BroadcastModeUtils.scala:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.execution
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BoundReference, Expression}
+import org.apache.spark.sql.catalyst.plans.physical.{BroadcastMode, IdentityBroadcastMode}
+import org.apache.spark.sql.execution.joins.HashedRelationBroadcastMode
+
+import java.io.{ByteArrayInputStream, ByteArrayOutputStream, ObjectInputStream, ObjectOutputStream}
+
+/**
+ * Provides serialization-safe representations of BroadcastMode to avoid issues with circular
+ * references in complex expression trees during Kryo serialization.
+ */
+sealed trait SafeBroadcastMode extends Serializable
+
+/** Safe representation of IdentityBroadcastMode */
+case object IdentitySafeBroadcastMode extends SafeBroadcastMode
+
+/**
+ * Safe wrapper for HashedRelationBroadcastMode. Stores only column ordinals instead of full
+ * BoundReference expressions.
+ */
+final case class HashSafeBroadcastMode(ordinals: Array[Int], isNullAware: Boolean)
+ extends SafeBroadcastMode
+
+/**
+ * Safe wrapper for HashedRelationBroadcastMode when keys are not simple BoundReferences. Stores key
+ * expressions as serialized Java bytes.
+ */
+final case class HashExprSafeBroadcastMode(exprBytes: Array[Byte], isNullAware: Boolean)
+ extends SafeBroadcastMode
+
+object BroadcastModeUtils {
+
+ /**
+ * Converts a BroadcastMode to its SafeBroadcastMode equivalent. Uses ordinals for simple
+ * BoundReferences, otherwise serializes the expressions.
+ */
+ private[execution] def toSafe(mode: BroadcastMode): SafeBroadcastMode = mode match {
+ case IdentityBroadcastMode =>
+ IdentitySafeBroadcastMode
+ case HashedRelationBroadcastMode(keys, isNullAware) =>
+ // Fast path: all keys are already BoundReference(i, ..,..).
+ val ords = keys.collect { case BoundReference(ord, _, _) => ord }
+ if (ords.size == keys.size) {
+ HashSafeBroadcastMode(ords.toArray, isNullAware)
+ } else {
+ // Fallback: store the key expressions as Java-serialized bytes.
+ HashExprSafeBroadcastMode(serializeExpressions(keys), isNullAware)
+ }
+
+ case other =>
+ throw new IllegalArgumentException(s"Unsupported BroadcastMode: $other")
+ }
+
+ /** Converts a SafeBroadcastMode to its BroadcastMode equivalent. */
+ private[execution] def fromSafe(safe: SafeBroadcastMode, output: Seq[Attribute]): BroadcastMode =
+ safe match {
+ case IdentitySafeBroadcastMode =>
+ IdentityBroadcastMode
+
+ case HashSafeBroadcastMode(ords, isNullAware) =>
+ val bound = ords.map(i => BoundReference(i, output(i).dataType, output(i).nullable)).toSeq
+ HashedRelationBroadcastMode(bound, isNullAware)
+
+ case HashExprSafeBroadcastMode(bytes, isNullAware) =>
+ HashedRelationBroadcastMode(deserializeExpressions(bytes), isNullAware)
+ }
+
+ // Helpers for expression serialization (used in HashExprSafeBroadcastMode)
+ private[execution] def serializeExpressions(keys: Seq[Expression]): Array[Byte] = {
+ val bos = new ByteArrayOutputStream()
+ var oos: ObjectOutputStream = null
+ try {
+ oos = new ObjectOutputStream(bos)
+ oos.writeObject(keys)
+ oos.flush()
+ bos.toByteArray
+ } finally {
+ if (oos != null) oos.close()
+ bos.close()
+ }
+ }
+
+ private[execution] def deserializeExpressions(bytes: Array[Byte]): Seq[Expression] = {
+ val bis = new ByteArrayInputStream(bytes)
+ var ois: ObjectInputStream = null
+ try {
+ ois = new ObjectInputStream(bis)
+ ois.readObject().asInstanceOf[Seq[Expression]]
+ } finally {
Review Comment:
ditto.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9671][VL] Fix broadcast exchange stackoverflow due to Kryo serialization [incubator-gluten]
Posted by "FelixYBW (via GitHub)" <gi...@apache.org>.
FelixYBW commented on PR #10541:
URL: https://github.com/apache/incubator-gluten/pull/10541#issuecomment-3226002372
@JkSelf It fixed the odd issue #9671. Can you take a look?
Spark set `spark.kryo.referenceTracking=true`. The fix can set it to false which boosts performance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9671][VL] Fix broadcast exchange stackoverflow due to Kryo serialization [incubator-gluten]
Posted by "felixloesing (via GitHub)" <gi...@apache.org>.
felixloesing commented on PR #10541:
URL: https://github.com/apache/incubator-gluten/pull/10541#issuecomment-3228988171
> @felixloesing Thanks for your fix. Just curious — why does Gluten require this change, while vanilla Spark doesn't?
@JkSelf This is also described in the [PR](https://github.com/apache/incubator-gluten/pull/8116) that introduced the `BroadcastMode` to `BuildSideRelation` but let me summarize. Vanilla Spark doesn’t have this because it computes the transform inside BroadcastExchangeExec, so there’s no post‑reuse mismatch to reconcile. In Gluten, if exchange re-use, the buildKeys of ColumnarSubqueryBroadcastExec and the buildKeys of BroadcastExchangeExec may not be the same, so it is difficult to construct the projection of the transform.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9671][VL] Fix broadcast exchange stackoverflow due to Kryo serialization [incubator-gluten]
Posted by "felixloesing (via GitHub)" <gi...@apache.org>.
felixloesing commented on PR #10541:
URL: https://github.com/apache/incubator-gluten/pull/10541#issuecomment-3235077127
@JkSelf Thanks for the review! Added a catch clause and logged the exception.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9671][VL] Fix broadcast exchange stackoverflow due to Kryo serialization [incubator-gluten]
Posted by "JkSelf (via GitHub)" <gi...@apache.org>.
JkSelf commented on PR #10541:
URL: https://github.com/apache/incubator-gluten/pull/10541#issuecomment-3226497068
@felixloesing Thanks for your fix. Just curious — why does Gluten require this change, while vanilla Spark doesn't?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9671][VL] Fix broadcast exchange stackoverflow due to Kryo serialization [incubator-gluten]
Posted by "FelixYBW (via GitHub)" <gi...@apache.org>.
FelixYBW merged PR #10541:
URL: https://github.com/apache/incubator-gluten/pull/10541
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org