Copilot commented on code in PR #12765:
URL: https://github.com/apache/gluten/pull/12765#discussion_r3902900869
##########
backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala:
##########
@@ -212,6 +250,50 @@ object VeloxBroadcastBuildSideCache
)
}
+ /**
+ * Returns a handle to the native hash table for `serialized`, deserializing
it at most once per
+ * driver-side broadcast id. The returned handle is owned by
[[sharedDeserializedCache]]; callers
+ * must clone it rather than releasing it.
+ */
+ private def getOrDeserializeShared(
+ serialized: SerializedBroadcastHashTable,
+ broadcastHashTableId: String,
+ deserializeHashTableTimeMetric:
Option[org.apache.spark.sql.execution.metric.SQLMetric])
+ : SharedDeserializedHashTable = {
+ // Older payloads, and any path that did not go through the driver-side
build, carry no
+ // broadcast id. Fall back to keying on the join id, which is what the
previous behavior was.
+ val sharedKey =
+ if (serialized.broadcastId != null) serialized.broadcastId else
broadcastHashTableId
+
+ sharedDeserializedCache.get(
+ sharedKey,
+ (key: String) => {
+ logInfo(s"Deserializing hash table on executor for broadcast ID: $key")
+ val startTime = System.currentTimeMillis()
+ val hashTableHandle = serialized.deserialize(key)
+ val timeMs = System.currentTimeMillis() - startTime
+ deserializeHashTableTimeMetric.foreach(_ += timeMs)
+
+ // The serialized bytes have been fully consumed into the native
table. On an executor
+ // nothing else refers to them, so free the off-heap copy instead of
waiting for the
+ // broadcast object to be collected. On the driver the same object is
still owned by the
+ // broadcast variable and by driverSerializedCache, so it must be left
alone.
+ if (!isDriver) {
+ serialized.releaseSerializedData()
+ }
Review Comment:
Releasing `serialized`’s off-heap bytes mutates the broadcast value on
executors and can make later re-materialization impossible if the shared native
table is evicted (e.g., `expireAfterAccess`) or if Spark reuses the same cached
broadcast value for a later stage without forcing a re-deserialize from blocks.
This can surface as runtime failures in
`SerializedBroadcastHashTable.deserialize` (“already been released”). Prefer
keeping `serializedData` intact (remove `releaseSerializedData()`), or only
releasing bytes when you can guarantee the native table will remain available
for the broadcast’s full usable lifetime (e.g., tie it to broadcast
lifecycle/unpersist rather than cache eviction).
##########
gluten-arrow/src/main/java/org/apache/spark/sql/execution/unsafe/UnsafeByteArray.java:
##########
@@ -102,12 +120,12 @@ public void read(Kryo kryo, Input input) {
this.buffer = ArrowBufferAllocators.globalInstance().buffer((int) size);
Review Comment:
Casting `size` (a `long`) to `int` can truncate for payloads > 2GB, leading
to an under-allocated `ArrowBuf` and subsequent out-of-bounds reads/writes
during streaming. Since the serialized format persists `size` as a `long`, add
an explicit bound check (e.g., fail fast with an exception) before allocating
and before using the `int index` loop variable. The same issue applies to the
identical allocation in `readExternal`.
##########
gluten-arrow/src/main/java/org/apache/spark/sql/execution/unsafe/UnsafeByteArray.java:
##########
@@ -45,9 +49,14 @@ public class UnsafeByteArray implements Externalizable,
KryoSerializable {
public UnsafeByteArray() {}
- private byte[] chunkBuf() {
- if (chunkBuf == null) {
- chunkBuf = new byte[CHUNK_SIZE];
+ /**
+ * Returns a scratch buffer of at most {@link #CHUNK_SIZE} bytes, never
larger than the payload
+ * itself so that relations holding many small batches do not each allocate
a full-sized chunk.
+ */
+ private byte[] chunkBuf(long dataSize) {
+ final int wanted = (int) Math.max(1, Math.min(CHUNK_SIZE, dataSize));
+ if (chunkBuf == null || chunkBuf.length < wanted) {
+ chunkBuf = new byte[wanted];
Review Comment:
The doc comment says the scratch buffer is “never larger than the payload
itself”, but the implementation only grows the buffer and never shrinks it;
after handling a large payload once, later smaller payloads can still reuse a
larger-than-payload `chunkBuf`. Either adjust the wording to reflect the
non-shrinking behavior (recommended to avoid realloc churn), or implement
shrinking with care (tradeoff: more allocations/GC).
##########
backends-velox/src/main/scala/org/apache/gluten/execution/SerializedBroadcastHashTable.scala:
##########
@@ -82,8 +105,20 @@ class SerializedBroadcastHashTable(
joinHasNullKeys)
}
+ /**
+ * Frees the off-heap buffer holding the serialized bytes. Only safe once
the native hash table
+ * has been materialized from it, and only on an executor: on the driver the
very same object is
+ * still owned by the broadcast variable and by
+ *
[[VeloxBroadcastBuildSideCache.buildAndSerializeOnDriverInBroadcastExchange]]'s
cache.
+ */
+ def releaseSerializedData(): Unit = {
+ if (serializedData != null) {
+ serializedData.release()
+ }
+ }
+
/** Get the size of serialized data in bytes. */
- def sizeInBytes: Long = serializedData.size()
+ def sizeInBytes: Long = if (serializedData == null) 0L else
serializedData.size()
Review Comment:
After `releaseSerializedData()` runs, `serializedData` can be non-null but
released; `sizeInBytes` will still report the original size, which can be
misleading for memory accounting/debugging. Consider returning 0 when
`serializedData.isReleased` is true (or documenting that `sizeInBytes` reflects
the original payload size, not current resident bytes).
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
