Copilot commented on code in PR #13216: URL: https://github.com/apache/gluten/pull/13216#discussion_r4217163003
########## cpp/velox/operators/hashjoin/HashTableTaskBuilder.cc: ########## @@ -0,0 +1,257 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "operators/hashjoin/HashTableTaskBuilder.h" + +#include <algorithm> +#include <atomic> +#include <thread> + +#include "velox/exec/HashTableCache.h" +#include "velox/exec/RoundRobinPartitionFunction.h" +#include "velox/exec/Task.h" + +using namespace facebook::velox; + +namespace gluten { +namespace { + +// Output type of the fake join. The probe side is empty, so the output is never +// produced; it only has to be valid for the join type. +RowTypePtr fakeJoinOutputType(core::JoinType joinType, const RowTypePtr& probeType, const RowTypePtr& buildType) { + const auto& firstBuildName = buildType->nameOf(0); + const auto& firstBuildType = buildType->childAt(0); + switch (joinType) { + case core::JoinType::kLeftSemiProject: { + auto names = probeType->names(); + auto types = probeType->children(); + names.emplace_back("__fake_match"); + types.emplace_back(BOOLEAN()); + return ROW(std::move(names), std::move(types)); + } + case core::JoinType::kRightSemiFilter: + return ROW({firstBuildName}, {firstBuildType}); + case core::JoinType::kRightSemiProject: + return ROW({firstBuildName, "__fake_match"}, {firstBuildType, BOOLEAN()}); + default: + return probeType; + } +} + +// A table and the pool it is allocated from. 'table' is released first. +struct TableWithPool { + std::shared_ptr<memory::MemoryPool> pool; + std::shared_ptr<exec::BaseHashTable> table; +}; + +// What a task-built table keeps alive: the task's QueryCtx, which owns the +// HashTableCache entry, and the pool of the fake probe input. +struct BuildTaskResources { + std::shared_ptr<core::QueryCtx> queryCtx; + std::shared_ptr<memory::MemoryPool> inputPool; +}; + +// Appends the input columns 'expr' references to 'fields'. +void collectInputFields(const core::TypedExprPtr& expr, std::vector<core::FieldAccessTypedExprPtr>& fields) { + if (auto field = std::dynamic_pointer_cast<const core::FieldAccessTypedExpr>(expr); + field != nullptr && field->isInputColumn()) { + fields.push_back(field); + return; + } + for (const auto& input : expr->inputs()) { + collectInputFields(input, fields); + } +} + +// An empty join table with the join keys of 'spec'. +std::shared_ptr<exec::BaseHashTable> makeEmptyJoinTable(const HashTableTaskBuildSpec& spec, memory::MemoryPool* pool) { + std::vector<std::unique_ptr<exec::VectorHasher>> hashers; + for (auto i = 0; i < spec.joinKeys.size(); ++i) { + hashers.emplace_back(std::make_unique<exec::VectorHasher>(spec.buildType->findChild(spec.joinKeys[i]), i)); + } + auto table = exec::HashTable<false>::createForJoin(std::move(hashers), {}, true, false, false, 1'000, pool); + table->prepareJoinTable({}, exec::BaseHashTable::kNoSpillInputStartPartitionBit, 1'000'000, false, nullptr); + return table; +} Review Comment: `VectorHasher` is being constructed with `i` (the join key ordinal) as the column/channel index. If join keys are not the first N columns of `buildType`, this will hash the wrong build columns and may produce incorrect join behavior in the fallback cached-table path. Use the actual build-side channel index for each key (e.g., look up the child index by name), and keep `i` only as the key ordinal if needed by the API. ########## backends-velox/src/main/scala/org/apache/gluten/execution/BroadcastHashJoinFilter.scala: ########## @@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.gluten.execution + +import org.apache.gluten.expression.ConverterUtils +import org.apache.gluten.substrait.SubstraitContext +import org.apache.gluten.substrait.plan.PlanBuilder +import org.apache.gluten.substrait.rel.RelBuilder +import org.apache.gluten.utils.SubstraitUtil + +import org.apache.spark.sql.catalyst.expressions.Expression + +import scala.collection.JavaConverters._ + +/** The join condition of a broadcast hash join, as needed by the native hash table build. */ +object BroadcastHashJoinFilter { + + /** + * Serializes the join condition as a substrait plan: a filter over an input iterator of the + * attributes the condition references. Returns the plan and the column names of those attributes, + * or empty arrays without a condition. + */ + def serialize(condition: Option[Expression]): (Array[Byte], Array[String]) = condition match { + case Some(cond) => + val inputs = cond.references.toSeq + val context = new SubstraitContext + val operatorId = context.nextOperatorId("BroadcastHashJoinFilter") + val read = RelBuilder.makeReadRelForInputIterator(inputs.asJava, context, operatorId) + val filter = RelBuilder.makeFilterRel( + read, + SubstraitUtil.toSubstraitExpression(cond, inputs, context), + context, + operatorId) Review Comment: `operatorId` is reused for both the read rel and filter rel. If `operatorId` is used as a unique identifier inside `SubstraitContext` (e.g., for bookkeeping, naming, extensions, metrics mapping), reusing it can cause collisions and make debugging/validation harder. Consider allocating distinct operator IDs for each rel node (one for `read`, one for `filter`). ########## backends-velox/src/main/scala/org/apache/gluten/execution/BroadcastHashJoinFilter.scala: ########## @@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.gluten.execution + +import org.apache.gluten.expression.ConverterUtils +import org.apache.gluten.substrait.SubstraitContext +import org.apache.gluten.substrait.plan.PlanBuilder +import org.apache.gluten.substrait.rel.RelBuilder +import org.apache.gluten.utils.SubstraitUtil + +import org.apache.spark.sql.catalyst.expressions.Expression + +import scala.collection.JavaConverters._ + +/** The join condition of a broadcast hash join, as needed by the native hash table build. */ +object BroadcastHashJoinFilter { + + /** + * Serializes the join condition as a substrait plan: a filter over an input iterator of the + * attributes the condition references. Returns the plan and the column names of those attributes, + * or empty arrays without a condition. + */ + def serialize(condition: Option[Expression]): (Array[Byte], Array[String]) = condition match { + case Some(cond) => + val inputs = cond.references.toSeq Review Comment: `cond.references` is a set-like collection; converting it with `.toSeq` can produce nondeterministic ordering across runs. While the plan and returned `filterInputNames` are internally consistent, nondeterminism here can hurt reproducibility (e.g., cache/debug comparisons, test flakiness when plan bytes are logged/compared). Consider applying a stable ordering (e.g., sort by exprId / canonical column name) before generating the iterator schema and the expression. ########## cpp/velox/jni/JniHashTable.cc: ########## @@ -160,95 +172,142 @@ std::shared_ptr<HashTableBuilder> nativeHashTableBuild( return hashTableBuilder; } -long getJoin(const std::string& hashTableId) { - return JniHashTableContext::getInstance().callJavaGet(hashTableId); +facebook::velox::core::TypedExprPtr toVeloxJoinFilter( + const std::string& filterPlan, + const std::vector<std::string>& filterInputNames, + const facebook::velox::config::ConfigBase* veloxCfg, + facebook::velox::memory::MemoryPool* pool) { + if (filterPlan.empty()) { + return nullptr; + } + ::substrait::Plan plan; + VELOX_CHECK(plan.ParseFromString(filterPlan), "Failed to parse the join filter plan"); + SubstraitToVeloxPlanConverter converter( + pool, + veloxCfg, + {}, + VeloxConnectorIds{ + .hive = kHiveConnectorId, + .iceberg = kIcebergConnectorId, + .iterator = kIteratorConnectorId, + .cudfHive = kCudfHiveConnectorId}, + std::nullopt, + std::nullopt, + /*validationMode=*/true); + const auto root = converter.toVeloxPlan(plan); + const auto* filterNode = findFilterNode(root); + VELOX_CHECK_NOT_NULL(filterNode, "The join filter plan has no filter"); + + // The converter names the filter input columns by position. Map them back to + // the column names of the join inputs. + const auto& inputType = filterNode->sources()[0]->outputType(); + VELOX_CHECK_EQ(inputType->size(), filterInputNames.size()); + std::unordered_map<std::string, facebook::velox::core::TypedExprPtr> mapping; + for (auto i = 0; i < inputType->size(); ++i) { + mapping.emplace( + inputType->nameOf(i), + std::make_shared<facebook::velox::core::FieldAccessTypedExpr>(inputType->childAt(i), filterInputNames[i])); + } + return filterNode->filter()->rewriteInputNames(mapping); } -size_t serializedHashTableSize(std::shared_ptr<HashTableBuilder> builder) { - VELOX_CHECK_NOT_NULL(builder, "Hash table builder cannot be null"); - - auto hashTable = builder->hashTable(); - VELOX_CHECK_NOT_NULL(hashTable, "Hash table cannot be null"); - - auto* hashTableFalse = dynamic_cast<facebook::velox::exec::HashTable<false>*>(hashTable.get()); - if (hashTableFalse != nullptr) { - return HashTableSerializer::serializedSize<false>(hashTableFalse); +std::shared_ptr<BroadcastHashTable> nativeHashTableBuildWithTask( + const std::string& hashTableId, + const std::vector<std::string>& joinKeys, + facebook::velox::core::TypedExprPtr filter, + std::vector<std::string> names, + std::vector<facebook::velox::TypePtr> veloxTypeList, + int joinType, + bool isExistenceJoin, + bool isNullAwareAntiJoin, + int64_t bloomFilterPushdownSize, + uint32_t minTableRowsForParallelJoinBuild, + uint32_t joinBuildVectorHasherMaxNumDistinct, + uint32_t abandonHashBuildDedupMinRows, + uint32_t abandonHashBuildDedupMinPct, + std::vector<std::shared_ptr<ColumnarBatch>>& batches, + uint32_t numThreads, + folly::Executor* executor, + std::shared_ptr<facebook::velox::memory::MemoryPool> queryPool, + std::shared_ptr<facebook::velox::memory::MemoryPool> inputPool) { + using facebook::velox::core::QueryConfig; + HashTableTaskBuildSpec spec; + spec.joinType = toVeloxJoinType(joinType, isExistenceJoin); + spec.nullAware = isNullAwareAntiJoin; + spec.filter = std::move(filter); + spec.buildType = std::make_shared<facebook::velox::RowType>(std::move(names), std::move(veloxTypeList)); + spec.joinKeys = joinKeys; + spec.queryConfigs = { + {QueryConfig::kMinTableRowsForParallelJoinBuild, std::to_string(minTableRowsForParallelJoinBuild)}, + {QueryConfig::kJoinBuildVectorHasherMaxNumDistinct, std::to_string(joinBuildVectorHasherMaxNumDistinct)}, + {QueryConfig::kAbandonDedupHashMapMinRows, std::to_string(abandonHashBuildDedupMinRows)}, + {QueryConfig::kAbandonDedupHashMapMinPct, std::to_string(abandonHashBuildDedupMinPct)}, + {QueryConfig::kHashProbeBloomFilterPushdownMaxSize, + std::to_string(std::max<int64_t>(bloomFilterPushdownSize, 0))}, + }; + + std::vector<facebook::velox::RowVectorPtr> buildVectors; + buildVectors.reserve(batches.size()); + for (const auto& batch : batches) { + buildVectors.emplace_back(VeloxColumnarBatch::from(inputPool.get(), batch)->getRowVector()); } - - auto* hashTableTrue = dynamic_cast<facebook::velox::exec::HashTable<true>*>(hashTable.get()); - VELOX_CHECK_NOT_NULL(hashTableTrue, "Hash table must be either HashTable<false> or HashTable<true>"); - return HashTableSerializer::serializedSize<true>(hashTableTrue); + return buildHashTableWithTask(hashTableId, spec, buildVectors, numThreads, executor, std::move(queryPool)); } -int64_t hashTableMemoryUsage(std::shared_ptr<HashTableBuilder> builder) { - VELOX_CHECK_NOT_NULL(builder, "Hash table builder cannot be null"); - return builder->hashTableMemoryUsage(); +std::shared_ptr<BroadcastHashTable> toBroadcastHashTable(std::shared_ptr<HashTableBuilder> builder) { + auto result = std::make_shared<BroadcastHashTable>(); + result->table = builder->hashTable(); + result->joinHasNullKeys = builder->joinHasNullKeys(); + result->memoryUsage = builder->hashTableMemoryUsage(); + result->owner = std::move(builder); + return result; } -void serializeHashTableTo(std::shared_ptr<HashTableBuilder> builder, uint8_t* data, size_t size) { - VELOX_CHECK_NOT_NULL(builder, "Hash table builder cannot be null"); - VELOX_CHECK_NOT_NULL(data, "Serialized buffer cannot be null"); +long getJoin(const std::string& hashTableId) { + return JniHashTableContext::getInstance().callJavaGet(hashTableId); +} - auto hashTable = builder->hashTable(); - VELOX_CHECK_NOT_NULL(hashTable, "Hash table cannot be null"); +size_t serializedHashTableSize(const BroadcastHashTable& hashTable) { + VELOX_CHECK_NOT_NULL(hashTable.table, "Hash table cannot be null"); + if (auto* table = asHashTable<false>(hashTable)) { + return HashTableSerializer::serializedSize<false>(table); + } + auto* table = asHashTable<true>(hashTable); + VELOX_CHECK_NOT_NULL(table, "Hash table must be either HashTable<false> or HashTable<true>"); + return HashTableSerializer::serializedSize<true>(table); +} - auto* hashTableFalse = dynamic_cast<facebook::velox::exec::HashTable<false>*>(hashTable.get()); - if (hashTableFalse != nullptr) { - HashTableSerializer::serializeTo<false>(hashTableFalse, data, size); +void serializeHashTableTo(const BroadcastHashTable& hashTable, uint8_t* data, size_t size) { + VELOX_CHECK_NOT_NULL(hashTable.table, "Hash table cannot be null"); + VELOX_CHECK_NOT_NULL(data, "Serialized buffer cannot be null"); + if (auto* table = asHashTable<false>(hashTable)) { + HashTableSerializer::serializeTo<false>(table, data, size); return; } - - auto* hashTableTrue = dynamic_cast<facebook::velox::exec::HashTable<true>*>(hashTable.get()); - VELOX_CHECK_NOT_NULL(hashTableTrue, "Hash table must be either HashTable<false> or HashTable<true>"); - HashTableSerializer::serializeTo<true>(hashTableTrue, data, size); + auto* table = asHashTable<true>(hashTable); + VELOX_CHECK_NOT_NULL(table, "Hash table must be either HashTable<false> or HashTable<true>"); + HashTableSerializer::serializeTo<true>(table, data, size); } -std::shared_ptr<HashTableBuilder> -deserializeHashTable(const uint8_t* data, size_t size, bool ignoreNullKeys, bool joinHasNullKeys) { +std::shared_ptr<BroadcastHashTable> deserializeHashTable( + const uint8_t* data, + size_t size, + bool ignoreNullKeys, + bool joinHasNullKeys, + std::shared_ptr<facebook::velox::memory::MemoryPool> pool) { VELOX_CHECK_NOT_NULL(data, "Serialized data cannot be null"); VELOX_CHECK_GT(size, 0, "Invalid data size"); - auto pool = defaultLeafVeloxMemoryPool(); - auto* poolPtr = pool.get(); - - std::unique_ptr<facebook::velox::exec::BaseHashTable> hashTable; + auto result = std::make_shared<BroadcastHashTable>(); if (ignoreNullKeys) { - auto derived = HashTableSerializer::deserialize<true>(data, size, poolPtr); - hashTable = std::move(derived); + result->table = HashTableSerializer::deserialize<true>(data, size, pool.get()); } else { - auto derived = HashTableSerializer::deserialize<false>(data, size, poolPtr); - hashTable = std::move(derived); + result->table = HashTableSerializer::deserialize<false>(data, size, pool.get()); } - - std::vector<std::shared_ptr<const facebook::velox::core::FieldAccessTypedExpr>> emptyKeys; - std::vector<uint32_t> emptyChannels; - - auto keyTypes = hashTable->rows()->keyTypes(); - std::vector<std::string> names; - for (size_t i = 0; i < keyTypes.size(); ++i) { - names.push_back("key" + std::to_string(i)); - } - auto rowType = facebook::velox::ROW(std::move(names), std::move(keyTypes)); - - auto builder = std::make_shared<HashTableBuilder>( - facebook::velox::core::JoinType::kInner, - false, - false, - -1, - emptyKeys, - emptyChannels, - false, - rowType, - poolPtr, - 1000, - 1000000, - 100000, - 0); - - builder->setHashTable(std::move(hashTable)); - // Restore the joinHasNullKeys flag - builder->setJoinHasNullKeys(joinHasNullKeys); - return builder; + result->joinHasNullKeys = joinHasNullKeys; + result->memoryUsage = result->table->allocatedBytes(); Review Comment: `memoryUsage` for deserialized tables is computed via `table->allocatedBytes()`, which may not match the memory accounting used elsewhere (e.g., task-built uses `tablePool->usedBytes()`, builder-built uses `hashTableMemoryUsage()`). This can make `hashTableMemorySize` metrics inconsistent/misleading across build paths. Since you already retain the pool in `owner`, consider reporting `pool->usedBytes()` (or another pool-based metric consistent with the other paths) instead of `allocatedBytes()`. ########## cpp/velox/operators/hashjoin/HashTableTaskBuilder.cc: ########## @@ -0,0 +1,257 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "operators/hashjoin/HashTableTaskBuilder.h" + +#include <algorithm> +#include <atomic> +#include <thread> + +#include "velox/exec/HashTableCache.h" +#include "velox/exec/RoundRobinPartitionFunction.h" +#include "velox/exec/Task.h" + +using namespace facebook::velox; + +namespace gluten { +namespace { + +// Output type of the fake join. The probe side is empty, so the output is never +// produced; it only has to be valid for the join type. +RowTypePtr fakeJoinOutputType(core::JoinType joinType, const RowTypePtr& probeType, const RowTypePtr& buildType) { + const auto& firstBuildName = buildType->nameOf(0); + const auto& firstBuildType = buildType->childAt(0); + switch (joinType) { + case core::JoinType::kLeftSemiProject: { + auto names = probeType->names(); + auto types = probeType->children(); + names.emplace_back("__fake_match"); + types.emplace_back(BOOLEAN()); + return ROW(std::move(names), std::move(types)); + } + case core::JoinType::kRightSemiFilter: + return ROW({firstBuildName}, {firstBuildType}); + case core::JoinType::kRightSemiProject: + return ROW({firstBuildName, "__fake_match"}, {firstBuildType, BOOLEAN()}); + default: + return probeType; + } +} + +// A table and the pool it is allocated from. 'table' is released first. +struct TableWithPool { + std::shared_ptr<memory::MemoryPool> pool; + std::shared_ptr<exec::BaseHashTable> table; +}; + +// What a task-built table keeps alive: the task's QueryCtx, which owns the +// HashTableCache entry, and the pool of the fake probe input. +struct BuildTaskResources { + std::shared_ptr<core::QueryCtx> queryCtx; + std::shared_ptr<memory::MemoryPool> inputPool; +}; + +// Appends the input columns 'expr' references to 'fields'. +void collectInputFields(const core::TypedExprPtr& expr, std::vector<core::FieldAccessTypedExprPtr>& fields) { + if (auto field = std::dynamic_pointer_cast<const core::FieldAccessTypedExpr>(expr); + field != nullptr && field->isInputColumn()) { + fields.push_back(field); + return; + } + for (const auto& input : expr->inputs()) { + collectInputFields(input, fields); + } +} + +// An empty join table with the join keys of 'spec'. +std::shared_ptr<exec::BaseHashTable> makeEmptyJoinTable(const HashTableTaskBuildSpec& spec, memory::MemoryPool* pool) { + std::vector<std::unique_ptr<exec::VectorHasher>> hashers; + for (auto i = 0; i < spec.joinKeys.size(); ++i) { + hashers.emplace_back(std::make_unique<exec::VectorHasher>(spec.buildType->findChild(spec.joinKeys[i]), i)); + } + auto table = exec::HashTable<false>::createForJoin(std::move(hashers), {}, true, false, false, 1'000, pool); + table->prepareJoinTable({}, exec::BaseHashTable::kNoSpillInputStartPartitionBit, 1'000'000, false, nullptr); + return table; +} + +std::string nextBuildId() { + static std::atomic<uint64_t> id{0}; + return std::to_string(id++); +} + +void runSerial(const std::shared_ptr<exec::Task>& task) { + while (true) { + ContinueFuture future = ContinueFuture::makeEmpty(); + auto output = task->next(&future); + if (output != nullptr) { + continue; + } + if (!future.valid()) { + break; + } + future.wait(); + } +} + +void runParallel(const std::shared_ptr<exec::Task>& task, uint32_t numDrivers) { + task->start(numDrivers); + task->taskCompletionFuture().wait(); + // Drivers may still hold the task, and so the plan and its input vectors, + // briefly after completion. Wait for them so that the input pool can go. + for (int i = 0; i < 10'000 && task.use_count() > 1; ++i) { + std::this_thread::sleep_for(std::chrono::microseconds(100)); + } Review Comment: This introduces a fixed-duration spin/sleep loop (up to ~1s) on every parallel build, which can add latency and waste CPU under contention. If the goal is to ensure drivers release resources, prefer a more deterministic mechanism (e.g., an explicit “drivers stopped” signal/future if available) or avoid the wait when resources are intentionally kept alive by the returned `BroadcastHashTable` ownership model. ########## cpp/velox/jni/VeloxJniWrapper.cc: ########## @@ -1160,7 +1161,98 @@ JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_native defaultLeafVeloxMemoryPool()); } - return gluten::getHashTableObjStore()->save(hashTableBuilders[0]); + return gluten::getHashTableObjStore()->save(gluten::toBroadcastHashTable(std::move(hashTableBuilders[0]))); + JNI_METHOD_END(kInvalidObjectHandle) +} + +JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_nativeBuildWithTask( // NOLINT + JNIEnv* env, + jobject wrapper, + jstring tableId, + jlongArray batchHandles, + jobjectArray joinKeys, + jbyteArray filterPlan, + jobjectArray filterInputNames, + jint joinType, + jboolean isExistenceJoin, + jbyteArray namedStruct, + jboolean isNullAwareAntiJoin, + jlong bloomFilterPushdownSize, + jint numThreads) { + JNI_METHOD_START + auto ctx = getRuntime(env, wrapper); + auto* runtime = dynamic_cast<VeloxRuntime*>(ctx); + GLUTEN_CHECK(runtime != nullptr, "Not a Velox runtime"); + const auto& queryConf = *(runtime->veloxCfg()); + const auto minTableRowsForParallelJoinBuild = + queryConf.get<uint32_t>(kMinTableRowsForParallelJoinBuild, kMinTableRowsForParallelJoinBuildDefault); + const auto joinBuildVectorHasherMaxNumDistinct = + queryConf.get<uint32_t>(kJoinBuildVectorHasherMaxNumDistinct, kJoinBuildVectorHasherMaxNumDistinctDefault); + const auto abandonHashBuildDedupMinRows = + queryConf.get<uint32_t>(kAbandonDedupHashMapMinRows, kAbandonDedupHashMapMinRowsDefault); + const auto abandonHashBuildDedupMinPct = + queryConf.get<uint32_t>(kAbandonDedupHashMapMinPct, kAbandonDedupHashMapMinPctDefault); + const auto hashTableId = jStringToCString(env, tableId); + + auto toStrings = [env](jobjectArray array) { + std::vector<std::string> strings; + if (array == nullptr) { + return strings; + } + const jsize length = env->GetArrayLength(array); + strings.reserve(length); + for (jsize i = 0; i < length; ++i) { + strings.emplace_back(jStringToCString(env, (jstring)env->GetObjectArrayElement(array, i))); + } + return strings; + }; + auto toBytes = [env](jbyteArray array) { + if (array == nullptr) { + return std::string{}; + } + const auto bytes = gluten::getByteArrayElementsSafe(env, array); + return std::string{ + reinterpret_cast<const char*>(bytes.elems()), static_cast<std::string::size_type>(bytes.length())}; + }; + + substrait::NamedStruct substraitStruct; + substraitStruct.ParseFromString(toBytes(namedStruct)); + auto veloxTypeList = SubstraitParser::parseNamedStruct(substraitStruct); + std::vector<std::string> names{substraitStruct.names().begin(), substraitStruct.names().end()}; + + std::vector<std::shared_ptr<ColumnarBatch>> cb; + int handleCount = env->GetArrayLength(batchHandles); + auto safeArray = getLongArrayElementsSafe(env, batchHandles); + for (int i = 0; i < handleCount; ++i) { + int64_t handle = safeArray.elems()[i]; + cb.push_back(ObjectStore::retrieve<ColumnarBatch>(handle)); + } + + auto filter = gluten::toVeloxJoinFilter( + toBytes(filterPlan), toStrings(filterInputNames), runtime->veloxCfg().get(), defaultLeafVeloxMemoryPool().get()); + + // FIXME: This reuses the io executor which is supposed to only serve async IO tasks. + auto* executor = VeloxBackend::get()->ioExecutor(); + auto hashTable = gluten::nativeHashTableBuildWithTask( Review Comment: Building hash tables (including parallel build) is CPU-intensive work; running it on the IO executor risks starving async IO and can regress overall query stability/latency. Prefer a dedicated CPU executor (or a dedicated “hash build” executor) and/or make the executor selection configurable so deployments can avoid overloading the IO thread pool. -- 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]
