From eb2cfce159950859603180fd32593d3a89bd7c73 Mon Sep 17 00:00:00 2001 From: Ke Jia Date: Mon, 6 Jul 2026 03:27:32 +0100 Subject: [PATCH 1/5] Use Velox's HashTableCache to cache the BHJ's HashTable --- .../gluten/vectorized/HashJoinBuilder.java | 4 +-- .../execution/HashJoinExecTransformer.scala | 4 +-- .../VeloxBroadcastBuildSideCache.scala | 2 +- .../execution/ColumnarBuildSideRelation.scala | 7 +++- .../UnsafeColumnarBuildSideRelation.scala | 7 +++- cpp/velox/jni/VeloxJniWrapper.cc | 34 +++++++++++++++++-- cpp/velox/substrait/SubstraitToVeloxPlan.cc | 27 +++------------ .../execution/JoinExecTransformer.scala | 16 +++++++-- .../velox/VeloxAdaptiveQueryExecSuite.scala | 4 ++- 9 files changed, 70 insertions(+), 35 deletions(-) diff --git a/backends-velox/src/main/java/org/apache/gluten/vectorized/HashJoinBuilder.java b/backends-velox/src/main/java/org/apache/gluten/vectorized/HashJoinBuilder.java index dbd8a3a5f70..777636322d1 100644 --- a/backends-velox/src/main/java/org/apache/gluten/vectorized/HashJoinBuilder.java +++ b/backends-velox/src/main/java/org/apache/gluten/vectorized/HashJoinBuilder.java @@ -35,9 +35,9 @@ public long rtHandle() { return runtime.getHandle(); } - public static native void clearHashTable(long hashTableData); + public static native void clearHashTable(String cacheKey, long hashTableData); - public static native long cloneHashTable(long hashTableData); + public static native long cloneHashTable(String cacheKey, long hashTableData); public static native long deserializeHashTableDirect( long address, int size, boolean ignoreNullKeys, boolean joinHasNullKeys); diff --git a/backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala b/backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala index e77b2c61154..95811f0b07d 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala @@ -106,8 +106,8 @@ case class BroadcastHashJoinExecTransformer( right, isNullAwareAntiJoin) { - // Unique ID for the build side. - lazy val buildBroadcastTableId: String = buildPlan.id.toString + // Unique ID for built table + lazy val buildBroadcastTableId: String = canonicalBuildHashTableId(buildPlan) override protected lazy val substraitJoinType: JoinRel.JoinType = joinType match { case _: InnerLike => diff --git a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala index 0d492817b14..0ae3ac9228a 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala @@ -245,7 +245,7 @@ object VeloxBroadcastBuildSideCache } } - HashJoinBuilder.clearHashTable(value.pointer) + HashJoinBuilder.clearHashTable(key, value.pointer) } } } diff --git a/backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarBuildSideRelation.scala b/backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarBuildSideRelation.scala index 35587bde133..844a987847a 100644 --- a/backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarBuildSideRelation.scala +++ b/backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarBuildSideRelation.scala @@ -332,7 +332,12 @@ case class ColumnarBuildSideRelation( (hashTableData(droppedDuplicates), this, droppedDuplicates) } else { - (HashJoinBuilder.cloneHashTable(hashTableData(droppedDuplicates)), null, droppedDuplicates) + ( + HashJoinBuilder.cloneHashTable( + broadcastContext.buildHashTableId, + hashTableData(droppedDuplicates)), + null, + droppedDuplicates) } } diff --git a/backends-velox/src/main/scala/org/apache/spark/sql/execution/unsafe/UnsafeColumnarBuildSideRelation.scala b/backends-velox/src/main/scala/org/apache/spark/sql/execution/unsafe/UnsafeColumnarBuildSideRelation.scala index 2a6963bb727..7ee885e7ae2 100644 --- a/backends-velox/src/main/scala/org/apache/spark/sql/execution/unsafe/UnsafeColumnarBuildSideRelation.scala +++ b/backends-velox/src/main/scala/org/apache/spark/sql/execution/unsafe/UnsafeColumnarBuildSideRelation.scala @@ -305,7 +305,12 @@ class UnsafeColumnarBuildSideRelation( (hashTableData(droppedDuplicates), this, droppedDuplicates) } else { - (HashJoinBuilder.cloneHashTable(hashTableData(droppedDuplicates)), null, droppedDuplicates) + ( + HashJoinBuilder.cloneHashTable( + broadcastContext.buildHashTableId, + hashTableData(droppedDuplicates)), + null, + droppedDuplicates) } } diff --git a/cpp/velox/jni/VeloxJniWrapper.cc b/cpp/velox/jni/VeloxJniWrapper.cc index 36798f5cded..851e5e5b06e 100644 --- a/cpp/velox/jni/VeloxJniWrapper.cc +++ b/cpp/velox/jni/VeloxJniWrapper.cc @@ -46,6 +46,7 @@ #include "velox/common/base/BloomFilter.h" #include "velox/common/file/FileSystems.h" #include "velox/exec/HashTable.h" +#include "velox/exec/HashTableCache.h" #ifdef GLUTEN_ENABLE_GPU #include "cudf/CudfPlanValidator.h" @@ -949,7 +950,7 @@ JNIEXPORT jobject JNICALL Java_org_apache_gluten_execution_IcebergWriteJniWrappe JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_nativeBuild( // NOLINT JNIEnv* env, jobject wrapper, - jstring /*tableId*/, + jstring tableId, jlongArray batchHandles, jobjectArray joinKeys, jobjectArray filterBuildColumns, @@ -974,6 +975,8 @@ JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_native queryConf.get(kAbandonDedupHashMapMinRows, kAbandonDedupHashMapMinRowsDefault); const auto abandonHashBuildDedupMinPct = queryConf.get(kAbandonDedupHashMapMinPct, kAbandonDedupHashMapMinPctDefault); + const auto hashTableId = jStringToCString(env, tableId); + // Convert Java String array to C++ vector std::vector hashJoinKeys; jsize joinKeysCount = env->GetArrayLength(joinKeys); @@ -1052,6 +1055,12 @@ 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->exist(hashTableId)) { + cache->add(hashTableId, builder->hashTable(), builder->joinHasNullKeys(), defaultLeafVeloxMemoryPool()); + } + return gluten::getHashTableObjStore()->save(builder); } @@ -1134,6 +1143,16 @@ 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->exist(hashTableId)) { + cache->add( + hashTableId, + hashTableBuilders[0]->hashTable(), + hashTableBuilders[0]->joinHasNullKeys(), + defaultLeafVeloxMemoryPool()); + } + return gluten::getHashTableObjStore()->save(hashTableBuilders[0]); JNI_METHOD_END(kInvalidObjectHandle) } @@ -1141,9 +1160,17 @@ JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_native 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(tableHandler); + auto* cache = facebook::velox::exec::HashTableCache::instance(); + if (!cache->exist(cacheKeyStr)) { + cache->add( + cacheKeyStr, hashTableHandler->hashTable(), hashTableHandler->joinHasNullKeys(), defaultLeafVeloxMemoryPool()); + } + return gluten::getHashTableObjStore()->save(hashTableHandler); JNI_METHOD_END(kInvalidObjectHandle) } @@ -1151,10 +1178,11 @@ JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_cloneH 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(tableHandler); - hashTableHandler->hashTable()->clear(true); + auto cacheKeyStr = jStringToCString(env, cacheKey); + facebook::velox::exec::HashTableCache::instance()->drop(cacheKeyStr); ObjectStore::release(tableHandler); JNI_METHOD_END() } diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.cc b/cpp/velox/substrait/SubstraitToVeloxPlan.cc index 5bd499582b8..d45409f6655 100644 --- a/cpp/velox/substrait/SubstraitToVeloxPlan.cc +++ b/cpp/velox/substrait/SubstraitToVeloxPlan.cc @@ -458,26 +458,8 @@ core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait:: } else if ( sJoin.has_advanced_extension() && SubstraitParser::configSetInOptimization(sJoin.advanced_extension(), "isBHJ=")) { - std::string hashTableId = sJoin.hashtableid(); - - std::shared_ptr opaqueSharedHashTable = nullptr; - bool joinHasNullKeys = false; - - try { - auto hashTableBuilder = ObjectStore::retrieve(getJoin(hashTableId)); - joinHasNullKeys = hashTableBuilder->joinHasNullKeys(); - auto originalShared = hashTableBuilder->hashTable(); - opaqueSharedHashTable = std::shared_ptr( - originalShared, reinterpret_cast(originalShared.get())); - - LOG(INFO) << "Successfully retrieved and aliased HashTable for reuse. ID: " << hashTableId; - } catch (const std::exception& e) { - LOG(WARNING) - << "Error retrieving HashTable from ObjectStore: " << e.what() - << ". Falling back to building new table. To ensure correct results, please verify that spark.gluten.velox.buildHashTableOncePerExecutor.enabled is set to false."; - opaqueSharedHashTable = nullptr; - } - + const auto& hashTableId = sJoin.hashtableid(); + const auto joinNodeId = hashTableId.empty() ? nextPlanNodeId() : hashTableId; // Create HashJoinNode node return std::make_shared( nextPlanNodeId(), @@ -489,10 +471,11 @@ core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait:: leftNode, rightNode, getJoinOutputType(leftNode, rightNode, joinType), + true, false, false, - joinHasNullKeys, - opaqueSharedHashTable); + nullptr, + sJoin.hashtableid()); } else { // Create HashJoinNode node return std::make_shared( diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala index ba0b55bcbfb..4fd58cbbf96 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala @@ -31,6 +31,8 @@ import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide import org.apache.spark.sql.catalyst.plans._ import org.apache.spark.sql.catalyst.plans.physical._ import org.apache.spark.sql.execution.{ExpandOutputPartitioningShim, ExplainUtils, SparkPlan} +import org.apache.spark.sql.execution.adaptive.BroadcastQueryStageExec +import org.apache.spark.sql.execution.exchange.ReusedExchangeExec import org.apache.spark.sql.execution.joins.{BaseJoinExec, HashedRelationBroadcastMode, HashJoin} import org.apache.spark.sql.execution.metric.SQLMetric import org.apache.spark.sql.types._ @@ -105,6 +107,15 @@ trait HashJoinLikeExecTransformer extends BaseJoinExec with TransformSupport { (right, left) } + protected def canonicalBuildHashTableId(plan: SparkPlan): String = plan match { + case b: BroadcastQueryStageExec => + canonicalBuildHashTableId(b.plan) + case r: ReusedExchangeExec => + canonicalBuildHashTableId(r.child) + case other => + other.id.toString + } + def sameType(from: DataType, to: DataType): Boolean = { (from, to) match { case (ArrayType(fromElement, _), ArrayType(toElement, _)) => @@ -270,7 +281,7 @@ trait HashJoinLikeExecTransformer extends BaseJoinExec with TransformSupport { inputBuildOutput, context, operatorId, - buildPlan.id.toString + canonicalBuildHashTableId(buildPlan) ) context.registerJoinParam(operatorId, joinParams) @@ -397,7 +408,8 @@ abstract class BroadcastHashJoinExecTransformerBase( override def joinBuildSide: BuildSide = buildSide override def hashJoinType: JoinType = joinType - lazy val buildHashTableId: String = buildPlan.id.toString + // Unique ID for builded hash table + lazy val buildHashTableId: String = canonicalBuildHashTableId(buildPlan) override def genJoinParametersInternal(): (Int, Int, String) = { (1, if (isNullAwareAntiJoin) 1 else 0, buildHashTableId) diff --git a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala index 6ccddac0c62..6f1272b7f77 100644 --- a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala +++ b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala @@ -522,7 +522,9 @@ class VeloxAdaptiveQueryExecSuite extends AdaptiveQueryExecSuite with GlutenSQLT withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "20000000", - SQLConf.SUBQUERY_REUSE_ENABLED.key -> "false") { + SQLConf.ADAPTIVE_OPTIMIZER_EXCLUDED_RULES.key -> AQEPropagateEmptyRelation.ruleName, + SQLConf.SUBQUERY_REUSE_ENABLED.key -> "false" + ) { val (plan, adaptivePlan) = runAdaptiveAndVerifyResult( "SELECT a FROM testData join testData2 ON key = a " + "where value >= (" + From db2fac45dd485bb3a3a47882a558803dbd16a80f Mon Sep 17 00:00:00 2001 From: Ke Jia Date: Mon, 6 Jul 2026 04:28:16 +0100 Subject: [PATCH 2/5] fix --- .../gluten/execution/SerializedBroadcastHashTable.scala | 3 ++- .../gluten/execution/VeloxBroadcastBuildSideCache.scala | 1 + .../spark/sql/execution/ColumnarBuildSideRelation.scala | 4 +++- .../execution/unsafe/UnsafeColumnarBuildSideRelation.scala | 4 +++- 4 files changed, 9 insertions(+), 3 deletions(-) diff --git a/backends-velox/src/main/scala/org/apache/gluten/execution/SerializedBroadcastHashTable.scala b/backends-velox/src/main/scala/org/apache/gluten/execution/SerializedBroadcastHashTable.scala index a59326daef8..353c042cae2 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/execution/SerializedBroadcastHashTable.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/execution/SerializedBroadcastHashTable.scala @@ -117,6 +117,7 @@ object SerializedBroadcastHashTable { */ def fromHashTable( hashTableHandle: Long, + cacheKey: String, buildSideRelation: BuildSideRelation, droppedDuplicates: Boolean, numRows: Long): SerializedBroadcastHashTable = { @@ -148,7 +149,7 @@ object SerializedBroadcastHashTable { buildSideRelation) } finally { synchronized { - HashJoinBuilder.clearHashTable(hashTableHandle) + HashJoinBuilder.clearHashTable(cacheKey, hashTableHandle) } } } diff --git a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala index 0ae3ac9228a..e21e61b36f7 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala @@ -161,6 +161,7 @@ object VeloxBroadcastBuildSideCache val result = SerializedBroadcastHashTable.fromHashTable( hashTableHandle, + broadcastId, relation, droppedDuplicates, numRows) diff --git a/backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarBuildSideRelation.scala b/backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarBuildSideRelation.scala index 844a987847a..7374cf0b04d 100644 --- a/backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarBuildSideRelation.scala +++ b/backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarBuildSideRelation.scala @@ -242,7 +242,9 @@ case class ColumnarBuildSideRelation( (hashTableData(droppedDuplicates), this, droppedDuplicates) } else { ( - HashJoinBuilder.cloneHashTable(hashTableData(droppedDuplicates)), + HashJoinBuilder.cloneHashTable( + broadcastContext.buildHashTableId, + hashTableData(droppedDuplicates)), null, droppedDuplicates) } diff --git a/backends-velox/src/main/scala/org/apache/spark/sql/execution/unsafe/UnsafeColumnarBuildSideRelation.scala b/backends-velox/src/main/scala/org/apache/spark/sql/execution/unsafe/UnsafeColumnarBuildSideRelation.scala index 7ee885e7ae2..d0129e887cf 100644 --- a/backends-velox/src/main/scala/org/apache/spark/sql/execution/unsafe/UnsafeColumnarBuildSideRelation.scala +++ b/backends-velox/src/main/scala/org/apache/spark/sql/execution/unsafe/UnsafeColumnarBuildSideRelation.scala @@ -214,7 +214,9 @@ class UnsafeColumnarBuildSideRelation( (hashTableData(droppedDuplicates), this, droppedDuplicates) } else { ( - HashJoinBuilder.cloneHashTable(hashTableData(droppedDuplicates)), + HashJoinBuilder.cloneHashTable( + broadcastContext.buildHashTableId, + hashTableData(droppedDuplicates)), null, droppedDuplicates) } From f90210901ef96d0406b0c0ac08afe6daba821a5f Mon Sep 17 00:00:00 2001 From: Ke Jia Date: Wed, 8 Jul 2026 06:21:50 +0100 Subject: [PATCH 3/5] fix --- .../java/org/apache/gluten/vectorized/HashJoinBuilder.java | 2 +- .../gluten/execution/SerializedBroadcastHashTable.scala | 3 ++- .../gluten/execution/VeloxBroadcastBuildSideCache.scala | 2 +- cpp/velox/jni/VeloxJniWrapper.cc | 7 +++++++ 4 files changed, 11 insertions(+), 3 deletions(-) diff --git a/backends-velox/src/main/java/org/apache/gluten/vectorized/HashJoinBuilder.java b/backends-velox/src/main/java/org/apache/gluten/vectorized/HashJoinBuilder.java index 777636322d1..675db72a844 100644 --- a/backends-velox/src/main/java/org/apache/gluten/vectorized/HashJoinBuilder.java +++ b/backends-velox/src/main/java/org/apache/gluten/vectorized/HashJoinBuilder.java @@ -40,7 +40,7 @@ public long rtHandle() { public static native long cloneHashTable(String cacheKey, long hashTableData); public static native long deserializeHashTableDirect( - long address, int size, boolean ignoreNullKeys, boolean joinHasNullKeys); + String cacheKey, long address, int size, boolean ignoreNullKeys, boolean joinHasNullKeys); public static native boolean getHashTableIgnoreNullKeys(long hashTableHandle); diff --git a/backends-velox/src/main/scala/org/apache/gluten/execution/SerializedBroadcastHashTable.scala b/backends-velox/src/main/scala/org/apache/gluten/execution/SerializedBroadcastHashTable.scala index 353c042cae2..323b88a4203 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/execution/SerializedBroadcastHashTable.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/execution/SerializedBroadcastHashTable.scala @@ -73,8 +73,9 @@ class SerializedBroadcastHashTable( * @return * Hash table builder handle */ - def deserialize(): Long = { + def deserialize(cacheKey: String): Long = { HashJoinBuilder.deserializeHashTableDirect( + cacheKey, serializedData.address(), Math.toIntExact(serializedData.size()), ignoreNullKeys, diff --git a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala index e21e61b36f7..2c68028e9d2 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala @@ -198,7 +198,7 @@ object VeloxBroadcastBuildSideCache (_: String) => { logInfo(s"Deserializing hash table on executor for broadcast ID: $broadcastHashTableId") val startTime = System.currentTimeMillis() - val hashTableHandle = serialized.deserialize() + val hashTableHandle = serialized.deserialize(broadcastHashTableId) val timeMs = System.currentTimeMillis() - startTime deserializeHashTableTimeMetric.foreach(_ += timeMs) BroadcastHashTable( diff --git a/cpp/velox/jni/VeloxJniWrapper.cc b/cpp/velox/jni/VeloxJniWrapper.cc index 851e5e5b06e..eaf1a241302 100644 --- a/cpp/velox/jni/VeloxJniWrapper.cc +++ b/cpp/velox/jni/VeloxJniWrapper.cc @@ -1190,16 +1190,23 @@ JNIEXPORT void JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_clearHa JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_deserializeHashTableDirect( // NOLINT JNIEnv* env, jclass, + jstring cacheKey, jlong address, jint size, jboolean ignoreNullKeys, jboolean joinHasNullKeys) { JNI_METHOD_START + auto cacheKeyStr = jStringToCString(env, cacheKey); auto builder = gluten::deserializeHashTable( reinterpret_cast(address), static_cast(size), static_cast(ignoreNullKeys), static_cast(joinHasNullKeys)); + auto* cache = facebook::velox::exec::HashTableCache::instance(); + if (!cache->exist(cacheKeyStr)) { + cache->add( + cacheKeyStr, builder->hashTable(), builder->joinHasNullKeys(), defaultLeafVeloxMemoryPool()); + } return gluten::getHashTableObjStore()->save(builder); JNI_METHOD_END(kInvalidObjectHandle) } From fe2423b41a453d800fa38a3b1d9867781054935a Mon Sep 17 00:00:00 2001 From: Ke Jia Date: Wed, 8 Jul 2026 06:24:39 +0100 Subject: [PATCH 4/5] format --- cpp/velox/jni/VeloxJniWrapper.cc | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/cpp/velox/jni/VeloxJniWrapper.cc b/cpp/velox/jni/VeloxJniWrapper.cc index eaf1a241302..db189fdd14c 100644 --- a/cpp/velox/jni/VeloxJniWrapper.cc +++ b/cpp/velox/jni/VeloxJniWrapper.cc @@ -1204,8 +1204,7 @@ JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_deseri static_cast(joinHasNullKeys)); auto* cache = facebook::velox::exec::HashTableCache::instance(); if (!cache->exist(cacheKeyStr)) { - cache->add( - cacheKeyStr, builder->hashTable(), builder->joinHasNullKeys(), defaultLeafVeloxMemoryPool()); + cache->add(cacheKeyStr, builder->hashTable(), builder->joinHasNullKeys(), defaultLeafVeloxMemoryPool()); } return gluten::getHashTableObjStore()->save(builder); JNI_METHOD_END(kInvalidObjectHandle) From 48cd516805f2abc195e7abf0d9276250e11d6794 Mon Sep 17 00:00:00 2001 From: Ke Jia Date: Thu, 9 Jul 2026 04:00:24 +0100 Subject: [PATCH 5/5] refine --- .../execution/HashJoinExecTransformer.scala | 4 ++-- cpp/velox/substrait/SubstraitToVeloxPlan.cc | 2 -- .../gluten/execution/JoinExecTransformer.scala | 16 ++-------------- 3 files changed, 4 insertions(+), 18 deletions(-) diff --git a/backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala b/backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala index 95811f0b07d..58120aadc39 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala @@ -106,8 +106,8 @@ case class BroadcastHashJoinExecTransformer( right, isNullAwareAntiJoin) { - // Unique ID for built table - lazy val buildBroadcastTableId: String = canonicalBuildHashTableId(buildPlan) + // Unique ID for the build side + lazy val buildBroadcastTableId: String = buildPlan.id.toString override protected lazy val substraitJoinType: JoinRel.JoinType = joinType match { case _: InnerLike => diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.cc b/cpp/velox/substrait/SubstraitToVeloxPlan.cc index d45409f6655..bdc6d35c29c 100644 --- a/cpp/velox/substrait/SubstraitToVeloxPlan.cc +++ b/cpp/velox/substrait/SubstraitToVeloxPlan.cc @@ -458,8 +458,6 @@ core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait:: } else if ( sJoin.has_advanced_extension() && SubstraitParser::configSetInOptimization(sJoin.advanced_extension(), "isBHJ=")) { - const auto& hashTableId = sJoin.hashtableid(); - const auto joinNodeId = hashTableId.empty() ? nextPlanNodeId() : hashTableId; // Create HashJoinNode node return std::make_shared( nextPlanNodeId(), diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala index 4fd58cbbf96..ba0b55bcbfb 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/execution/JoinExecTransformer.scala @@ -31,8 +31,6 @@ import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide import org.apache.spark.sql.catalyst.plans._ import org.apache.spark.sql.catalyst.plans.physical._ import org.apache.spark.sql.execution.{ExpandOutputPartitioningShim, ExplainUtils, SparkPlan} -import org.apache.spark.sql.execution.adaptive.BroadcastQueryStageExec -import org.apache.spark.sql.execution.exchange.ReusedExchangeExec import org.apache.spark.sql.execution.joins.{BaseJoinExec, HashedRelationBroadcastMode, HashJoin} import org.apache.spark.sql.execution.metric.SQLMetric import org.apache.spark.sql.types._ @@ -107,15 +105,6 @@ trait HashJoinLikeExecTransformer extends BaseJoinExec with TransformSupport { (right, left) } - protected def canonicalBuildHashTableId(plan: SparkPlan): String = plan match { - case b: BroadcastQueryStageExec => - canonicalBuildHashTableId(b.plan) - case r: ReusedExchangeExec => - canonicalBuildHashTableId(r.child) - case other => - other.id.toString - } - def sameType(from: DataType, to: DataType): Boolean = { (from, to) match { case (ArrayType(fromElement, _), ArrayType(toElement, _)) => @@ -281,7 +270,7 @@ trait HashJoinLikeExecTransformer extends BaseJoinExec with TransformSupport { inputBuildOutput, context, operatorId, - canonicalBuildHashTableId(buildPlan) + buildPlan.id.toString ) context.registerJoinParam(operatorId, joinParams) @@ -408,8 +397,7 @@ abstract class BroadcastHashJoinExecTransformerBase( override def joinBuildSide: BuildSide = buildSide override def hashJoinType: JoinType = joinType - // Unique ID for builded hash table - lazy val buildHashTableId: String = canonicalBuildHashTableId(buildPlan) + lazy val buildHashTableId: String = buildPlan.id.toString override def genJoinParametersInternal(): (Int, Int, String) = { (1, if (isNullAwareAntiJoin) 1 else 0, buildHashTableId)