Copilot commented on code in PR #12163: URL: https://github.com/apache/gluten/pull/12163#discussion_r3401313383
########## ep/build-velox/src/get-velox.sh: ########## @@ -17,9 +17,9 @@ set -exu CURRENT_DIR=$(cd "$(dirname "$BASH_SOURCE")"; pwd) -VELOX_REPO=https://github.com/IBM/velox.git -VELOX_BRANCH=dft-2026_06_06 -VELOX_ENHANCED_BRANCH=ibm-2026_06_06 +VELOX_REPO=https://github.com/JkSelf/velox.git +VELOX_BRANCH=dft-2026_06_06-hashtable-cache +VELOX_ENHANCED_BRANCH=ibm-2026_06_06-hashtable-cache Review Comment: The default Velox source is switched to a personal fork/branch (JkSelf/velox). This makes builds non-reproducible and is risky for CI and downstream users; please keep the default pointing at the project-approved upstream and use the script flags when testing forks. ########## cpp/velox/jni/VeloxJniWrapper.cc: ########## @@ -1138,30 +1146,52 @@ JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_native } hashTableBuilders[0]->setHashTable(std::move(mainTable)); + + auto* cache = facebook::velox::exec::HashTableCache::instance(); + if (!cache->hasTable(hashTableId)) { + std::cout << "VeloxJniWrapper inject hash table id is " << hashTableId << "\n"; + cache->injectTable( + hashTableId, + hashTableBuilders[0]->hashTable(), + hashTableBuilders[0]->joinHasNullKeys(), + defaultLeafVeloxMemoryPool()); + } + return gluten::getHashTableObjStore()->save(hashTableBuilders[0]); JNI_METHOD_END(kInvalidObjectHandle) } JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_cloneHashTable( // NOLINT JNIEnv* env, jclass, + jstring cacheKey, jlong tableHandler) { JNI_METHOD_START + auto cacheKeyStr = jStringToCString(env, cacheKey); auto hashTableHandler = ObjectStore::retrieve<gluten::HashTableBuilder>(tableHandler); + auto* cache = facebook::velox::exec::HashTableCache::instance(); + if (!cache->hasTable(cacheKeyStr)) { + cache->injectTable( + cacheKeyStr, hashTableHandler->hashTable(), hashTableHandler->joinHasNullKeys(), defaultLeafVeloxMemoryPool()); + } + return gluten::getHashTableObjStore()->save(hashTableHandler); JNI_METHOD_END(kInvalidObjectHandle) } JNIEXPORT void JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_clearHashTable( // NOLINT JNIEnv* env, jclass, + jstring cacheKey, jlong tableHandler) { JNI_METHOD_START - auto hashTableHandler = ObjectStore::retrieve<gluten::HashTableBuilder>(tableHandler); - hashTableHandler->hashTable()->clear(true); + auto cacheKeyStr = jStringToCString(env, cacheKey); + std::cout << "VeloxJniWrapper clear hash table id is " << cacheKeyStr << "\n"; + facebook::velox::exec::HashTableCache::instance()->drop(cacheKeyStr); Review Comment: `std::cout` in `clearHashTable` will emit on every eviction/cleanup and can be very noisy. Use `LOG(INFO)`/`VLOG` instead. ########## gluten-substrait/src/main/scala/org/apache/spark/sql/execution/ColumnarBroadcastExchangeExec.scala: ########## @@ -76,8 +80,14 @@ case class ColumnarBroadcastExchangeExec(mode: BroadcastMode, child: SparkPlan) child, longMetric("numOutputRows"), longMetric("dataSize"), - metrics.getOrElse("buildThreads", null)) + metrics.getOrElse("buildThreads", null), + queryId) } + logWarning( Review Comment: Same issue as above: this is routine tracing (`driver-broadcast-built`) and should not be logged at warning level. ########## cpp/velox/jni/VeloxJniWrapper.cc: ########## @@ -1060,6 +1061,13 @@ JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_native nullptr); builder->setHashTable(std::move(mainTable)); + auto* cache = facebook::velox::exec::HashTableCache::instance(); + + if (!cache->hasTable(hashTableId)) { + std::cout << "VeloxJniWrapper inject hash table id is " << hashTableId << "\n"; + cache->injectTable(hashTableId, builder->hashTable(), builder->joinHasNullKeys(), defaultLeafVeloxMemoryPool()); + } Review Comment: `std::cout` in JNI code will spam executor logs and bypass the existing logging framework (glog). Prefer `LOG(INFO)` (or `VLOG`) so output can be controlled via log configuration. ########## backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala: ########## @@ -183,6 +190,17 @@ case class BroadcastHashJoinExecTransformer( // FIXME: Do we have to make build side a RDD? streamedRDD :+ broadcastRDD } + + private def extractQueryId(relation: BuildSideRelation): String = relation match { + case columnar: ColumnarBuildSideRelation if columnar.queryId.nonEmpty => + columnar.queryId + case unsafe: UnsafeColumnarBuildSideRelation if unsafe.queryIdValue.nonEmpty => + unsafe.queryIdValue + case _ => + throw new IllegalStateException( + s"Missing query id in build-side relation for broadcast hash join " + + s"$buildBroadcastTableId, relationType=${relation.getClass.getName}") + } Review Comment: Throwing here will fail the query if a build-side relation is created without a non-empty `queryId` (e.g., default argument path or unexpected BuildSideRelation implementation). Consider falling back to the driver-scoped `GlutenQueryContext` query id (and warn) rather than hard-failing. ########## gluten-substrait/src/main/scala/org/apache/spark/sql/execution/ColumnarBroadcastExchangeExec.scala: ########## @@ -57,11 +57,15 @@ case class ColumnarBroadcastExchangeExec(mode: BroadcastMode, child: SparkPlan) @transient override lazy val relationFuture: java.util.concurrent.Future[broadcast.Broadcast[Any]] = { + val queryId = org.apache.gluten.execution.GlutenQueryContext.getOrCreateQueryId(sparkContext) SQLExecution.withThreadLocalCaptured[broadcast.Broadcast[Any]]( session, BroadcastExchangeExec.executionContext) { try { SparkShimLoader.getSparkShims.setJobDescriptionOrTagForBroadcastExchange(sparkContext, this) + logWarning( Review Comment: `logWarning` is used for normal broadcast lifecycle tracing. This is expected to run frequently and will pollute warning logs; please switch these trace logs to `logInfo`/`logDebug`. ########## gluten-substrait/src/main/scala/org/apache/spark/sql/execution/ColumnarBroadcastExchangeExec.scala: ########## @@ -86,6 +96,9 @@ case class ColumnarBroadcastExchangeExec(mode: BroadcastMode, child: SparkPlan) sparkContext, relation.asInstanceOf[Any]) } + logWarning( Review Comment: Same issue as above: `driver-broadcast-published` is normal operation and should not be warning-level. ########## gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala: ########## @@ -267,7 +278,7 @@ trait HashJoinLikeExecTransformer extends BaseJoinExec with TransformSupport { inputBuildOutput, context, operatorId, - buildPlan.id.toString + canonicalBuildHashTableId(buildPlan) Review Comment: Potential key mismatch for BHJ hash table caching: the build side injects into Velox `HashTableCache` using `BroadcastHashJoinContext.buildHashTableId` (now `cacheKey = s"$queryId:$buildHashTableId"`), but the probe side uses `JoinRel.hashtableid` (set via the `hashTableId` argument passed to `JoinUtils.createJoinRel`). With the current code, `hashTableId` is only `canonicalBuildHashTableId(buildPlan)` (no queryId prefix), so the HashJoinNode key will not match the cache key used during build, and cache lookup will fail. ########## backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala: ########## @@ -135,16 +137,21 @@ case class BroadcastHashJoinExecTransformer( override def columnarInputRDDs: Seq[RDD[ColumnarBatch]] = { val streamedRDD = getColumnarInputRDDs(streamedPlan) val executionId = sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY) + val broadcast = buildPlan.executeBroadcast[BuildSideRelation]() + val queryId = extractQueryId(broadcast.value) + val cacheKey = s"$queryId:$buildHashTableId" + logWarning( Review Comment: `logWarning` is used for routine tracing (cache key creation). This will generate warning-level noise for every broadcast hash join even when nothing is wrong; please use `logInfo`/`logDebug` instead. ########## cpp/velox/jni/VeloxJniWrapper.cc: ########## @@ -1138,30 +1146,52 @@ JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_native } hashTableBuilders[0]->setHashTable(std::move(mainTable)); + + auto* cache = facebook::velox::exec::HashTableCache::instance(); + if (!cache->hasTable(hashTableId)) { + std::cout << "VeloxJniWrapper inject hash table id is " << hashTableId << "\n"; + cache->injectTable( + hashTableId, + hashTableBuilders[0]->hashTable(), + hashTableBuilders[0]->joinHasNullKeys(), + defaultLeafVeloxMemoryPool()); + } Review Comment: Same as above: please avoid `std::cout` for routine tracing in JNI and use the repo's logger so it can be filtered. ########## cpp/velox/compute/WholeStageResultIterator.cc: ########## @@ -72,6 +72,18 @@ const std::string kHiveDefaultPartition = "__HIVE_DEFAULT_PARTITION__"; } // namespace +namespace { +std::string getVeloxTaskId(const SparkTaskInfo& taskInfo) { + if (!taskInfo.queryId.empty()) { + return taskInfo.queryId; + } + if (taskInfo.executionId != -1) { + return fmt::format("Gluten_Execution_{}", std::to_string(taskInfo.executionId)); + } + return fmt::format("Gluten_Execution_{}", ""); +} Review Comment: When neither `queryId` nor `executionId` is available, `getVeloxTaskId` currently returns `Gluten_Execution_` with an empty suffix, which is not unique and not very helpful for debugging/Velox context naming. Use the existing stage/task/vId fields as a deterministic fallback. ########## backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala: ########## @@ -867,25 +868,36 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi with Logging { val buildThreadsValue = if (rawThreads < 1) 1 else rawThreads buildThreads += buildThreadsValue - if (useOffheapBroadcastBuildRelation) { - TaskResources.runUnsafe { - UnsafeColumnarBuildSideRelation( + val relation = + if (useOffheapBroadcastBuildRelation) { + TaskResources.runUnsafe { + UnsafeColumnarBuildSideRelation( + newOutput, + serialized.flatMap(_.offHeapData().asScala), + mode, + newBuildKeys, + offload, + buildThreadsValue, + queryId) + } + } else { + ColumnarBuildSideRelation( newOutput, - serialized.flatMap(_.offHeapData().asScala), + serialized.flatMap(_.onHeapData().asScala).toArray, mode, newBuildKeys, offload, - buildThreadsValue) + buildThreadsValue, + queryId) } - } else { - ColumnarBuildSideRelation( - newOutput, - serialized.flatMap(_.onHeapData().asScala).toArray, - mode, - newBuildKeys, - offload, - buildThreadsValue) - } + logWarning( Review Comment: `logWarning` is used for normal relation creation tracing. This is not an exceptional condition and may be extremely noisy in production; prefer `logInfo`/`logDebug`. -- 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]
