Copilot commented on code in PR #12163:
URL: https://github.com/apache/gluten/pull/12163#discussion_r3386149611
##########
cpp/velox/substrait/SubstraitToVeloxPlan.cc:
##########
@@ -448,29 +448,22 @@ 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<core::OpaqueHashTable> opaqueSharedHashTable = nullptr;
- bool joinHasNullKeys = false;
-
+ const auto& hashTableId = sJoin.hashtableid();
+ auto useHashTableCache = true;
try {
auto hashTableBuilder =
ObjectStore::retrieve<gluten::HashTableBuilder>(getJoin(hashTableId));
- joinHasNullKeys = hashTableBuilder->joinHasNullKeys();
- auto originalShared = hashTableBuilder->hashTable();
- opaqueSharedHashTable = std::shared_ptr<core::OpaqueHashTable>(
- originalShared,
reinterpret_cast<core::OpaqueHashTable*>(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;
+ useHashTableCache = false;
}
+ size_t pos = hashTableId.find(':');
+ auto planNodeId = hashTableId.substr(pos + 1);
+ std::cout << "the cacheKey is " << hashTableId << " the planNode id is "
<< planNodeId << std::endl;
Review Comment:
Using `std::cout` for debug logging will spam executor logs and bypasses the
project's logging configuration/levels. Please remove this line (or switch to
an existing logging macro that can be gated by verbosity).
##########
cpp/velox/compute/WholeStageResultIterator.cc:
##########
@@ -72,6 +72,15 @@ const std::string kHiveDefaultPartition =
"__HIVE_DEFAULT_PARTITION__";
} // namespace
+namespace {
+std::string getVeloxTaskId(const SparkTaskInfo& taskInfo) {
+ if (taskInfo.executionId != -1) {
+ return fmt::format("Gluten_Execution_{}",
std::to_string(taskInfo.executionId));
+ }
+ return fmt::format("Gluten_Execution_{}", "");
+}
Review Comment:
`getVeloxTaskId` now returns the same ID for all Spark tasks that share an
SQL executionId. Velox `Task::create` expects task IDs to be unique within a
process; reusing the same ID across concurrent Spark tasks can lead to task
registry collisions, incorrect stats attribution, or runtime failures. Keep the
executionId in the string if needed for cache scoping, but also include
stageId/taskId/vId (or otherwise guarantee uniqueness).
##########
cpp/velox/substrait/SubstraitToVeloxPlan.cc:
##########
@@ -448,29 +448,22 @@ 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<core::OpaqueHashTable> opaqueSharedHashTable = nullptr;
- bool joinHasNullKeys = false;
-
+ const auto& hashTableId = sJoin.hashtableid();
+ auto useHashTableCache = true;
try {
auto hashTableBuilder =
ObjectStore::retrieve<gluten::HashTableBuilder>(getJoin(hashTableId));
- joinHasNullKeys = hashTableBuilder->joinHasNullKeys();
- auto originalShared = hashTableBuilder->hashTable();
- opaqueSharedHashTable = std::shared_ptr<core::OpaqueHashTable>(
- originalShared,
reinterpret_cast<core::OpaqueHashTable*>(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;
+ useHashTableCache = false;
}
+ size_t pos = hashTableId.find(':');
+ auto planNodeId = hashTableId.substr(pos + 1);
+ std::cout << "the cacheKey is " << hashTableId << " the planNode id is "
<< planNodeId << std::endl;
+ const auto joinNodeId = hashTableId.empty() ? nextPlanNodeId() :
planNodeId;
+
Review Comment:
`joinNodeId` is being set to `planNodeId` parsed from the Spark-side
hashTableId (e.g. a numeric Spark plan id).
`SubstraitToVeloxPlanConverter::nextPlanNodeId()` also generates sequential
numeric IDs starting from 0, so this can easily collide with already-assigned
Velox plan node IDs and produce an invalid plan (duplicate PlanNodeId). The
join node ID must be guaranteed unique within the converted Velox plan while
still allowing HashTableCache lookup.
##########
cpp/velox/jni/VeloxJniWrapper.cc:
##########
@@ -986,6 +987,7 @@ 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);
+ std::cout << "the cacheKey in nativeBuild is " << hashTableId << std::endl;
Review Comment:
`std::cout` debug output in JNI code will produce noisy executor logs and is
not controllable via log levels. Please remove this line (or replace with an
existing logging facility that can be configured/disabled).
##########
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_05
-VELOX_ENHANCED_BRANCH=ibm-2026_06_05
+VELOX_REPO=https://github.com/JkSelf/velox.git
+VELOX_BRANCH=dft-2026_06_05-hashtable-cache
+VELOX_ENHANCED_BRANCH=ibm-2026_06_05-hashtable-cache
Review Comment:
Defaulting the build to a personal Velox fork/branch
(`github.com/JkSelf/velox.git`) is a supply-chain risk and makes builds
non-reproducible for CI and other contributors. The default repo/branch should
remain an official upstream (or the existing IBM fork used by this project),
with developer overrides handled via `--velox_repo/--velox_branch` or
`UPSTREAM_VELOX_PR_ID`.
##########
gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala:
##########
@@ -602,6 +602,7 @@ class VeloxTestSettings extends BackendTestSettings {
enableSuite[GlutenResolvedDataSourceSuite]
enableSuite[GlutenSaveLoadSuite]
enableSuite[GlutenTableScanSuite]
+ .includeByPrefix("Caching")
Review Comment:
`.includeByPrefix("Caching")` limits `GlutenTableScanSuite` to only tests
whose names start with "Caching", which substantially reduces coverage for
table-scan behavior in the Velox CI test matrix. If the goal is to add/enable
specific caching-related cases, consider including only the new tests (without
filtering out the rest of the suite) or moving the filtering behind a dedicated
CI profile.
--
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]
