Copilot commented on code in PR #12765:
URL: https://github.com/apache/gluten/pull/12765#discussion_r3925848320
##########
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:
`releaseSerializedData()` frees the underlying off-heap buffer, but
`sizeInBytes` will still report the pre-release size because
`UnsafeByteArray.release()` nulls the buffer without changing the `size` field.
This can lead to incorrect memory accounting/metrics (and can be misleading in
logs/diagnostics). Consider returning 0 when `serializedData.isReleased` is
true, or updating `UnsafeByteArray.release()` to also reset `size` when
releasing.
##########
cpp/velox/jni/VeloxJniWrapper.cc:
##########
@@ -1208,7 +1220,11 @@ JNIEXPORT jlong JNICALL
Java_org_apache_gluten_vectorized_HashJoinBuilder_deseri
reinterpret_cast<const uint8_t*>(address),
static_cast<size_t>(size),
static_cast<bool>(ignoreNullKeys),
- static_cast<bool>(joinHasNullKeys));
+ static_cast<bool>(joinHasNullKeys),
+ // Deserializing on one thread costs about as much as building the table
from the raw build
+ // side does on all of them, which would leave the driver-side build
with no upside at all.
+ // FIXME: This reuses the io executor which is supposed to only serve
async IO tasks.
+ VeloxBackend::get()->ioExecutor());
Review Comment:
This explicitly schedules CPU-heavy deserialization work onto the IO
executor (and notes it as a FIXME). That can starve genuine async IO and cause
query-wide latency regressions under load. Consider introducing (or using, if
already available) a dedicated CPU executor/thread-pool for hash table
build/deserialize work, or plumb an appropriate executor from the caller/config
instead of reusing `ioExecutor()`.
##########
backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala:
##########
@@ -199,11 +234,14 @@ object VeloxBroadcastBuildSideCache
buildSideRelationCache.get(
broadcastHashTableId,
(_: String) => {
- logInfo(s"Deserializing hash table on executor for broadcast ID:
$broadcastHashTableId")
- val startTime = System.currentTimeMillis()
- val hashTableHandle = serialized.deserialize(broadcastHashTableId)
- val timeMs = System.currentTimeMillis() - startTime
- deserializeHashTableTimeMetric.foreach(_ += timeMs)
+ val shared = getOrDeserializeShared(
+ serialized,
+ broadcastHashTableId,
+ deserializeHashTableTimeMetric)
+ // Register the shared table under this join's id as well, so that the
native probe side
+ // can resolve it. The clone holds its own reference to the same
native table.
+ val hashTableHandle =
+ HashJoinBuilder.cloneHashTable(broadcastHashTableId, shared.pointer)
Review Comment:
The new shared-deserialization + per-join clone behavior is
performance-critical and subtly interacts with lifecycle (eviction + native
ref-counting + `releaseSerializedData()`). Consider adding an integration test
that forces a reused broadcast exchange with multiple BHJs on the same
executor, and asserts that deserialization happens once per broadcast (e.g.,
via metrics/log counters) and that subsequent joins use clones successfully
even after the serialized bytes are released.
##########
backends-velox/src/main/scala/org/apache/spark/sql/execution/SerializedHashTableBroadcastRelation.scala:
##########
@@ -83,9 +93,15 @@ case class SerializedHashTableBroadcastRelation(
/**
* Transform is used for DPP (Dynamic Partition Pruning) to extract keys. We
delegate to the
* underlying buildSideRelation in the serialized hash table.
+ *
+ * Driver-only, for the same reason as [[deserialized]].
*/
override def transform(key: Expression): Array[InternalRow] = {
- serializedHashTable.buildSideRelation.transform(key)
+ val relation = serializedHashTable.buildSideRelation
+ if (relation == null) {
+ throw new IllegalStateException(driverOnlyMessage("transform"))
+ }
+ relation.transform(key)
Review Comment:
After making `buildSideRelation` transient/not serialized, this relation
will have `buildSideRelation == null` on executors. Throwing from `transform`
(and similarly from `deserialized`) is a behavioral break for any Spark paths
that legitimately call these methods on executors (e.g., DPP key extraction in
some runtimes). If executors may need these, consider preserving the minimal
data required for `transform`/`deserialized` in the broadcast payload (instead
of the full raw relation), or implement these methods using the serialized
bytes (or explicitly ensure via call sites that they are never invoked
off-driver).
##########
cpp/velox/jni/VeloxJniWrapper.cc:
##########
@@ -1297,9 +1313,10 @@ JNIEXPORT void JNICALL
Java_org_apache_gluten_vectorized_HashJoinBuilder_seriali
auto builder =
ObjectStore::retrieve<gluten::HashTableBuilder>(hashTableHandle);
VELOX_CHECK_GT(address, 0, "Serialized hash table buffer address must be
positive");
VELOX_CHECK_GE(size, 0, "Serialized hash table buffer size must be
non-negative");
- const auto serializedSize = gluten::serializedHashTableSize(builder);
- VELOX_CHECK_EQ(static_cast<size_t>(size), serializedSize, "Hash table buffer
size mismatch");
- gluten::serializeHashTableTo(builder, reinterpret_cast<uint8_t*>(address),
serializedSize);
+ // Do not recompute serializedHashTableSize() here just to validate 'size':
that walks every
+ // build row again to re-measure the variable-width columns. serializeTo()
bounds-checks each
+ // write and asserts that it filled the buffer exactly, which covers the
same mistake.
+ gluten::serializeHashTableTo(builder, reinterpret_cast<uint8_t*>(address),
static_cast<size_t>(size));
Review Comment:
Removing the explicit size equality check changes error reporting and relies
on internal checks that may be asserts (often compiled out in release builds),
potentially turning a clear size-mismatch into a harder-to-diagnose failure. If
recomputing the size is too expensive, consider caching the computed serialized
size at allocation time (and passing it through), or ensure `serializeTo()`
uses non-assert runtime checks for both overflow and exact-size agreement so
mismatches fail deterministically with a clear message.
--
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]
