fix

Author: JkSelf

backends-velox/src/main/java/org/apache/gluten/vectorized/HashJoinBuilder.java

@@ -41,6 +41,7 @@ public long rtHandle() {
 
   public native long nativeBuild(
       String buildHashTableId,
+      String executionId,
       long[] batchHandlers,
       String[] joinKeys,
       String[] filterBuildColumns,

backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala

@@ -175,6 +175,7 @@ case class BroadcastHashJoinExecTransformer(
         filterBuildColumns,
         filterPropagatesNulls,
         buildBroadcastTableId,
+        executionId,
         isNullAwareAntiJoin,
         bloomFilterPushdownSize,
         metrics.get("buildHashTableTime")
@@ -195,6 +196,7 @@ case class BroadcastHashJoinContext(
     filterBuildColumns: Array[String],
     filterPropagatesNulls: Boolean,
     buildHashTableId: String,
+    executionId: String,
     isNullAwareAntiJoin: Boolean = false,
     bloomFilterPushdownSize: Long,
     buildHashTableTimeMetric: Option[SQLMetric] = None)

backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarBuildSideRelation.scala

@@ -213,6 +213,7 @@ case class ColumnarBuildSideRelation(
         hashTableData = hashJoinBuilder
           .nativeBuild(
             broadcastContext.buildHashTableId,
+            broadcastContext.executionId,
             batchArray.toArray,
             joinKeys,
             broadcastContext.filterBuildColumns,

backends-velox/src/main/scala/org/apache/spark/sql/execution/unsafe/UnsafeColumnarBuildSideRelation.scala

@@ -183,6 +183,7 @@ class UnsafeColumnarBuildSideRelation(
         hashTableData = hashJoinBuilder
           .nativeBuild(
             broadcastContext.buildHashTableId,
+            broadcastContext.executionId,
             batchArray.toArray,
             joinKeys,
             broadcastContext.filterBuildColumns,

cpp/velox/compute/WholeStageResultIterator.cc

@@ -77,11 +77,13 @@ std::string getVeloxTaskId(const SparkTaskInfo& taskInfo) {
   if (taskInfo.executionId != -1) {
     return fmt::format("Gluten_Execution_{}", std::to_string(taskInfo.executionId));
   }
-  return fmt::format(
-      "Gluten_Stage_{}_TID_{}_VTID_{}",
-      std::to_string(taskInfo.stageId),
-      std::to_string(taskInfo.taskId),
-      std::to_string(taskInfo.vId));
+
+  return "Gluten_Execution_";
+  // return fmt::format(
+  //     "Gluten_Stage_{}_TID_{}_VTID_{}",
+  //     std::to_string(taskInfo.stageId),
+  //     std::to_string(taskInfo.taskId),
+  //     std::to_string(taskInfo.vId));
 }
 } // namespace
 

cpp/velox/jni/VeloxJniWrapper.cc

@@ -962,6 +962,7 @@ JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_native
     JNIEnv* env,
     jobject wrapper,
     jstring tableId,
+    jstring executionId,
     jlongArray batchHandles,
     jobjectArray joinKeys,
     jobjectArray filterBuildColumns,
@@ -987,6 +988,13 @@ JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_native
   const auto abandonHashBuildDedupMinPct =
       queryConf.get<uint32_t>(kAbandonDedupHashMapMinPct, kAbandonDedupHashMapMinPctDefault);
   const auto hashTableId = jStringToCString(env, tableId);
+  const auto sparkExecutionId = jStringToCString(env, executionId);
+
+  auto queryId = fmt::format("Gluten_Execution_{}", sparkExecutionId);
+
+  auto cacheKey = fmt::format("{}:{}", queryId, hashTableId);
+
+  // std::cout << "the cacheKey in VeloxJniWrapper is " << cacheKey << std::endl;
 
   // Convert Java String array to C++ vector<string>
   std::vector<std::string> hashJoinKeys;
@@ -1063,8 +1071,8 @@ JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_native
 
     auto* cache = facebook::velox::exec::HashTableCache::instance();
 
-    if (!cache->hasTable(hashTableId)) {
-      cache->injectTable(hashTableId, builder->hashTable(), builder->joinHasNullKeys(), defaultLeafVeloxMemoryPool());
+    if (!cache->hasTable(cacheKey)) {
+      cache->injectTable(cacheKey, builder->hashTable(), builder->joinHasNullKeys(), defaultLeafVeloxMemoryPool());
     }
 
     return gluten::getHashTableObjStore()->save(builder);
@@ -1147,9 +1155,9 @@ 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)) {
+  if (!cache->hasTable(cacheKey)) {
     cache->injectTable(
-        hashTableId,
+        cacheKey,
         hashTableBuilders[0]->hashTable(),
         hashTableBuilders[0]->joinHasNullKeys(),
         defaultLeafVeloxMemoryPool());