You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by "JkSelf (via GitHub)" <gi...@apache.org> on 2026/10/07 15:25:45 UTC
[PR] [GLUTEN][VL][TEST] Use fake task to build hash table [gluten]
JkSelf opened a new pull request, #13216:
URL: https://github.com/apache/gluten/pull/13216
<!--
Thank you for submitting a pull request! Here are some tips:
1. For first-time contributors, please read our contributing guide:
https://github.com/apache/gluten/blob/main/CONTRIBUTING.md
2. If necessary, create a GitHub issue for discussion beforehand to avoid duplicate work.
3. If the PR is specific to a single backend, include [VL] or [CH] in the PR title to indicate the
Velox or ClickHouse backend, respectively.
4. If the PR is not ready for review, please mark it as a draft.
-->
## What changes are proposed in this pull request?
<!--
Provide a clear and concise description of the changes introduced in this PR.
Ensure the PR description aligns with the code changes, especially after updates.
If applicable, include "Fixes #<GitHub_Issue_ID>" to automatically close the corresponding issue
when the PR is merged.
-->
## How was this patch tested?
<!--
Describe how the changes were tested, if applicable.
Include new tests to validate the functionality, if necessary.
For UI-related changes, attach screenshots to demonstrate the updates.
-->
## Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN][VL][TEST] Use fake task to build hash table [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #13216:
URL: https://github.com/apache/gluten/pull/13216#discussion_r4312757956
##########
cpp/velox/jni/JniHashTable.cc:
##########
@@ -116,139 +120,87 @@ std::shared_ptr<HashTableBuilder> nativeHashTableBuild(
VELOX_NYI("Unsupported Join type: {}", std::to_string(sJoin));
}
- std::vector<std::shared_ptr<const facebook::velox::core::FieldAccessTypedExpr>> joinKeyTypes;
- joinKeyTypes.reserve(joinKeys.size());
- for (const auto& name : joinKeys) {
- joinKeyTypes.emplace_back(
- std::make_shared<facebook::velox::core::FieldAccessTypedExpr>(rowType->findChild(name), name));
- }
-
- std::vector<uint32_t> filterInputChannels;
- filterInputChannels.reserve(filterBuildColumns.size());
+ using facebook::velox::core::QueryConfig;
+ HashTableTaskBuildSpec spec;
+ spec.joinType = vJoin;
+ spec.nullAware = isNullAwareAntiJoin;
+ spec.withFilter = hasMixedJoinCondition;
for (const auto& name : filterBuildColumns) {
- if (const auto idx = rowType->getChildIdxIfExists(name)) {
- filterInputChannels.push_back(*idx);
+ if (rowType->containsChild(name) &&
+ std::find(spec.filterBuildColumns.begin(), spec.filterBuildColumns.end(), name) ==
+ spec.filterBuildColumns.end()) {
+ spec.filterBuildColumns.push_back(name);
}
}
- std::sort(filterInputChannels.begin(), filterInputChannels.end());
- filterInputChannels.erase(
- std::unique(filterInputChannels.begin(), filterInputChannels.end()), filterInputChannels.end());
-
- auto hashTableBuilder = std::make_shared<HashTableBuilder>(
- vJoin,
- isNullAwareAntiJoin,
- hasMixedJoinCondition,
- bloomFilterPushdownSize,
- joinKeyTypes,
- filterInputChannels,
- filterPropagatesNulls,
- rowType,
- memoryPool.get(),
- minTableRowsForParallelJoinBuild,
- joinBuildVectorHasherMaxNumDistinct,
- abandonHashBuildDedupMinRows,
- abandonHashBuildDedupMinPct);
-
- for (auto i = 0; i < batches.size(); i++) {
- auto rowVector = VeloxColumnarBatch::from(memoryPool.get(), batches[i])->getRowVector();
- hashTableBuilder->addInput(rowVector);
- if (hashTableBuilder->noMoreInput()) {
- break;
- }
+ spec.filterPropagatesNulls = filterPropagatesNulls;
+ spec.buildType = rowType;
+ 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(memoryPool.get(), batch)->getRowVector());
}
-
- return hashTableBuilder;
+ return buildHashTableWithTask(
+ hashTableId, spec, buildVectors, numThreads, executor, std::move(queryPool), memoryPool.get());
}
long getJoin(const std::string& hashTableId) {
return JniHashTableContext::getInstance().callJavaGet(hashTableId);
}
-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");
+size_t serializedHashTableSize(const BroadcastHashTable& hashTable) {
+ const auto& table = hashTable.hashTable();
+ VELOX_CHECK_NOT_NULL(table, "Hash table cannot be null");
- auto* hashTableFalse = dynamic_cast<facebook::velox::exec::HashTable<false>*>(hashTable.get());
+ auto* hashTableFalse = dynamic_cast<facebook::velox::exec::HashTable<false>*>(table.get());
if (hashTableFalse != nullptr) {
return HashTableSerializer::serializedSize<false>(hashTableFalse);
}
- auto* hashTableTrue = dynamic_cast<facebook::velox::exec::HashTable<true>*>(hashTable.get());
+ auto* hashTableTrue = dynamic_cast<facebook::velox::exec::HashTable<true>*>(table.get());
VELOX_CHECK_NOT_NULL(hashTableTrue, "Hash table must be either HashTable<false> or HashTable<true>");
return HashTableSerializer::serializedSize<true>(hashTableTrue);
}
-int64_t hashTableMemoryUsage(std::shared_ptr<HashTableBuilder> builder) {
- VELOX_CHECK_NOT_NULL(builder, "Hash table builder cannot be null");
- return builder->hashTableMemoryUsage();
-}
-
-void serializeHashTableTo(std::shared_ptr<HashTableBuilder> builder, uint8_t* data, size_t size) {
- VELOX_CHECK_NOT_NULL(builder, "Hash table builder cannot be null");
+void serializeHashTableTo(const BroadcastHashTable& hashTable, uint8_t* data, size_t size) {
VELOX_CHECK_NOT_NULL(data, "Serialized buffer cannot be null");
- auto hashTable = builder->hashTable();
- VELOX_CHECK_NOT_NULL(hashTable, "Hash table cannot be null");
+ const auto& table = hashTable.hashTable();
+ VELOX_CHECK_NOT_NULL(table, "Hash table cannot be null");
- auto* hashTableFalse = dynamic_cast<facebook::velox::exec::HashTable<false>*>(hashTable.get());
+ auto* hashTableFalse = dynamic_cast<facebook::velox::exec::HashTable<false>*>(table.get());
if (hashTableFalse != nullptr) {
HashTableSerializer::serializeTo<false>(hashTableFalse, data, size);
return;
}
- auto* hashTableTrue = dynamic_cast<facebook::velox::exec::HashTable<true>*>(hashTable.get());
+ auto* hashTableTrue = dynamic_cast<facebook::velox::exec::HashTable<true>*>(table.get());
VELOX_CHECK_NOT_NULL(hashTableTrue, "Hash table must be either HashTable<false> or HashTable<true>");
HashTableSerializer::serializeTo<true>(hashTableTrue, data, size);
}
-std::shared_ptr<HashTableBuilder>
+std::shared_ptr<BroadcastHashTable>
deserializeHashTable(const uint8_t* data, size_t size, bool ignoreNullKeys, bool joinHasNullKeys) {
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;
+ std::shared_ptr<facebook::velox::exec::BaseHashTable> hashTable;
if (ignoreNullKeys) {
- auto derived = HashTableSerializer::deserialize<true>(data, size, poolPtr);
- hashTable = std::move(derived);
+ hashTable = HashTableSerializer::deserialize<true>(data, size, pool.get());
} else {
- auto derived = HashTableSerializer::deserialize<false>(data, size, poolPtr);
- hashTable = std::move(derived);
- }
-
- 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));
+ hashTable = HashTableSerializer::deserialize<false>(data, size, pool.get());
}
Review Comment:
The deserialized `BaseHashTable` is allocated from `pool.get()`, but the returned/cached `shared_ptr<BaseHashTable>` doesn’t share ownership of that pool. If the `BroadcastHashTable` handle is released while tasks still hold table pointers (e.g., via `HashTableCache`), the underlying `MemoryPool` can be destroyed before the table destructor runs, leading to frees against a dead pool. Fix by ensuring the table `shared_ptr` itself keeps the allocation pool alive (e.g., via an aliasing `shared_ptr` pattern similar to `TableWithPool` in `HashTableTaskBuilder.cc`), and/or ensure the cache stores/retains the exact pool that backs the table.
##########
cpp/velox/operators/hashjoin/HashTableTaskBuilder.cc:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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 <atomic>
+
+#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 join outputs
+// nothing and needs no columns, except the match column of a semi project join.
+RowTypePtr fakeJoinOutputType(core::JoinType joinType) {
+ if (core::isLeftSemiProjectJoin(joinType) || core::isRightSemiProjectJoin(joinType)) {
+ return ROW({"__fake_match"}, {BOOLEAN()});
+ }
+ return ROW({}, {});
+}
+
+// A stand-in for the join filter with the properties HashBuild looks at, see
+// HashTableTaskBuildSpec. It is compiled by HashProbe but never evaluated.
+core::TypedExprPtr makePlaceholderFilter(const HashTableTaskBuildSpec& spec) {
+ if (!spec.withFilter) {
+ return nullptr;
+ }
+ if (!spec.filterPropagatesNulls || spec.filterBuildColumns.empty()) {
+ // References no build column, so HashBuild drops no build rows for it.
+ return std::make_shared<core::ConstantTypedExpr>(BOOLEAN(), Variant(true));
+ }
+ // Chains the build columns, cast to VARCHAR to fit any type, with equalto.
+ // Both have default null behavior, so the result is null-intolerant in each
+ // of them.
+ auto toVarchar = [&](const core::TypedExprPtr& expr) -> core::TypedExprPtr {
+ return std::make_shared<core::CastTypedExpr>(VARCHAR(), expr, false);
+ };
+ auto column = [&](const std::string& name) -> core::TypedExprPtr {
+ return toVarchar(std::make_shared<core::FieldAccessTypedExpr>(spec.buildType->findChild(name), name));
+ };
+ auto equalTo = [](const core::TypedExprPtr& left, const core::TypedExprPtr& right) -> core::TypedExprPtr {
+ return std::make_shared<core::CallTypedExpr>(BOOLEAN(), std::vector<core::TypedExprPtr>{left, right}, "equalto");
+ };
+ auto filter = equalTo(column(spec.filterBuildColumns[0]), column(spec.filterBuildColumns[0]));
+ for (auto i = 1; i < spec.filterBuildColumns.size(); ++i) {
+ filter = equalTo(toVarchar(filter), column(spec.filterBuildColumns[i]));
Review Comment:
The placeholder filter assumes all referenced build columns can be cast to VARCHAR and compared via `equalto`. If a build column is a complex/un-castable type (or if `equalto` isn’t supported for that type), filter compilation can fail even though the filter is “never evaluated”, causing hash table build to fail unexpectedly. Consider building a null-intolerant placeholder predicate using a function that is known to accept all Velox types (e.g., hashing/encoding-based) and default-null-propagates, or use a `try(...)`-style construct (if available) so compilation doesn’t throw for unsupported casts.
##########
cpp/velox/operators/hashjoin/HashTableTaskBuilder.cc:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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 <atomic>
+
+#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 join outputs
+// nothing and needs no columns, except the match column of a semi project join.
+RowTypePtr fakeJoinOutputType(core::JoinType joinType) {
+ if (core::isLeftSemiProjectJoin(joinType) || core::isRightSemiProjectJoin(joinType)) {
+ return ROW({"__fake_match"}, {BOOLEAN()});
+ }
+ return ROW({}, {});
+}
+
+// A stand-in for the join filter with the properties HashBuild looks at, see
+// HashTableTaskBuildSpec. It is compiled by HashProbe but never evaluated.
+core::TypedExprPtr makePlaceholderFilter(const HashTableTaskBuildSpec& spec) {
+ if (!spec.withFilter) {
+ return nullptr;
+ }
+ if (!spec.filterPropagatesNulls || spec.filterBuildColumns.empty()) {
+ // References no build column, so HashBuild drops no build rows for it.
+ return std::make_shared<core::ConstantTypedExpr>(BOOLEAN(), Variant(true));
+ }
+ // Chains the build columns, cast to VARCHAR to fit any type, with equalto.
+ // Both have default null behavior, so the result is null-intolerant in each
+ // of them.
+ auto toVarchar = [&](const core::TypedExprPtr& expr) -> core::TypedExprPtr {
+ return std::make_shared<core::CastTypedExpr>(VARCHAR(), expr, false);
+ };
+ auto column = [&](const std::string& name) -> core::TypedExprPtr {
+ return toVarchar(std::make_shared<core::FieldAccessTypedExpr>(spec.buildType->findChild(name), name));
+ };
+ auto equalTo = [](const core::TypedExprPtr& left, const core::TypedExprPtr& right) -> core::TypedExprPtr {
+ return std::make_shared<core::CallTypedExpr>(BOOLEAN(), std::vector<core::TypedExprPtr>{left, right}, "equalto");
+ };
+ auto filter = equalTo(column(spec.filterBuildColumns[0]), column(spec.filterBuildColumns[0]));
+ for (auto i = 1; i < spec.filterBuildColumns.size(); ++i) {
+ filter = equalTo(toVarchar(filter), column(spec.filterBuildColumns[i]));
+ }
Review Comment:
`auto i = 1` deduces `int`, which is then compared with `size_t` from `.size()`, potentially triggering `-Wsign-compare` (often promoted to errors in stricter builds). Prefer `size_t i = 1;` (and similarly for other index loops over `.size()` in this file).
##########
cpp/velox/operators/hashjoin/HashTableTaskBuilder.h:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <folly/Executor.h>
+
+#include "operators/hashjoin/BroadcastHashTable.h"
+#include "velox/core/PlanNode.h"
Review Comment:
This header introduces several STL and Velox types in its public API (`std::string`, `std::vector`, `std::unordered_map`, `uint32_t`, `facebook::velox::RowVectorPtr`, `facebook::velox::memory::MemoryPool`) but doesn’t include the corresponding standard/Velox headers directly. This can cause fragile builds that depend on transitive includes. Add the needed direct includes (e.g., `<string>`, `<vector>`, `<unordered_map>`, `<cstdint>`, plus the Velox headers that define `RowVectorPtr` and `MemoryPool`).
##########
cpp/velox/operators/hashjoin/HashTableTaskBuilder.cc:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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 <atomic>
+
+#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 join outputs
+// nothing and needs no columns, except the match column of a semi project join.
+RowTypePtr fakeJoinOutputType(core::JoinType joinType) {
+ if (core::isLeftSemiProjectJoin(joinType) || core::isRightSemiProjectJoin(joinType)) {
+ return ROW({"__fake_match"}, {BOOLEAN()});
+ }
+ return ROW({}, {});
+}
+
+// A stand-in for the join filter with the properties HashBuild looks at, see
+// HashTableTaskBuildSpec. It is compiled by HashProbe but never evaluated.
+core::TypedExprPtr makePlaceholderFilter(const HashTableTaskBuildSpec& spec) {
+ if (!spec.withFilter) {
+ return nullptr;
+ }
+ if (!spec.filterPropagatesNulls || spec.filterBuildColumns.empty()) {
+ // References no build column, so HashBuild drops no build rows for it.
+ return std::make_shared<core::ConstantTypedExpr>(BOOLEAN(), Variant(true));
+ }
+ // Chains the build columns, cast to VARCHAR to fit any type, with equalto.
+ // Both have default null behavior, so the result is null-intolerant in each
+ // of them.
+ auto toVarchar = [&](const core::TypedExprPtr& expr) -> core::TypedExprPtr {
+ return std::make_shared<core::CastTypedExpr>(VARCHAR(), expr, false);
+ };
+ auto column = [&](const std::string& name) -> core::TypedExprPtr {
+ return toVarchar(std::make_shared<core::FieldAccessTypedExpr>(spec.buildType->findChild(name), name));
+ };
+ auto equalTo = [](const core::TypedExprPtr& left, const core::TypedExprPtr& right) -> core::TypedExprPtr {
+ return std::make_shared<core::CallTypedExpr>(BOOLEAN(), std::vector<core::TypedExprPtr>{left, right}, "equalto");
+ };
+ auto filter = equalTo(column(spec.filterBuildColumns[0]), column(spec.filterBuildColumns[0]));
+ for (auto i = 1; i < spec.filterBuildColumns.size(); ++i) {
+ filter = equalTo(toVarchar(filter), column(spec.filterBuildColumns[i]));
+ }
+ return filter;
+}
+
+// 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 constructed with `channel = i`, but `i` is the join-key ordinal, not necessarily the channel index of `spec.joinKeys[i]` in `spec.buildType`. This can make the “empty cached table” inconsistent with how HashBuild would have created hashers (which typically use the build-side input channel indices for the join keys). To avoid mismatches, use the actual child index from `buildType` for the hasher channel (or use the same factory/utility HashBuild uses to create these hashers).
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN][VL][TEST] Use fake task to build hash table [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #13216:
URL: https://github.com/apache/gluten/pull/13216#discussion_r4269518745
##########
cpp/velox/operators/hashjoin/HashTableTaskBuilder.cc:
##########
@@ -0,0 +1,262 @@
+/*
+ * 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 <atomic>
+
+#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 stand-in for the join filter with the properties HashBuild looks at, see
+// HashTableTaskBuildSpec. It is compiled by HashProbe but never evaluated.
+core::TypedExprPtr makePlaceholderFilter(const HashTableTaskBuildSpec& spec) {
+ if (!spec.withFilter) {
+ return nullptr;
+ }
+ if (!spec.filterPropagatesNulls || spec.filterBuildColumns.empty()) {
+ // References no build column, so HashBuild drops no build rows for it.
+ return std::make_shared<core::ConstantTypedExpr>(BOOLEAN(), Variant(true));
+ }
+ // Chains the build columns, cast to VARCHAR to fit any type, with equalto.
+ // Both have default null behavior, so the result is null-intolerant in each
Review Comment:
The placeholder filter relies on casting arbitrary build columns (and intermediate boolean expressions) to VARCHAR and comparing via `equalto`. This can fail during expression compilation if a build column type (or BOOLEAN->VARCHAR cast) isn’t supported, even though the probe side is empty and the filter is never evaluated. Consider constructing a placeholder that only uses universally-supported null checks (e.g., a boolean expression that returns NULL when any referenced build column is NULL) without requiring casts/equality on the build column types.
##########
cpp/velox/operators/hashjoin/HashTableTaskBuilder.cc:
##########
@@ -0,0 +1,262 @@
+/*
+ * 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 <atomic>
+
+#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 stand-in for the join filter with the properties HashBuild looks at, see
+// HashTableTaskBuildSpec. It is compiled by HashProbe but never evaluated.
+core::TypedExprPtr makePlaceholderFilter(const HashTableTaskBuildSpec& spec) {
+ if (!spec.withFilter) {
+ return nullptr;
+ }
+ if (!spec.filterPropagatesNulls || spec.filterBuildColumns.empty()) {
+ // References no build column, so HashBuild drops no build rows for it.
+ return std::make_shared<core::ConstantTypedExpr>(BOOLEAN(), Variant(true));
+ }
+ // Chains the build columns, cast to VARCHAR to fit any type, with equalto.
+ // Both have default null behavior, so the result is null-intolerant in each
+ // of them.
+ auto toVarchar = [&](const core::TypedExprPtr& expr) -> core::TypedExprPtr {
+ return std::make_shared<core::CastTypedExpr>(VARCHAR(), expr, false);
+ };
+ auto column = [&](const std::string& name) -> core::TypedExprPtr {
+ return toVarchar(std::make_shared<core::FieldAccessTypedExpr>(spec.buildType->findChild(name), name));
+ };
+ auto equalTo = [](const core::TypedExprPtr& left, const core::TypedExprPtr& right) -> core::TypedExprPtr {
+ return std::make_shared<core::CallTypedExpr>(BOOLEAN(), std::vector<core::TypedExprPtr>{left, right}, "equalto");
+ };
+ auto filter = equalTo(column(spec.filterBuildColumns[0]), column(spec.filterBuildColumns[0]));
+ for (auto i = 1; i < spec.filterBuildColumns.size(); ++i) {
+ filter = equalTo(toVarchar(filter), column(spec.filterBuildColumns[i]));
+ }
+ return filter;
+}
+
+// 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++);
+}
+
+// A table and the pool it is allocated from. Members are destroyed in reverse
+// order, so 'table' returns its memory before 'pool' goes.
+struct TableWithPool {
+ std::shared_ptr<memory::MemoryPool> pool;
+ std::shared_ptr<exec::BaseHashTable> table;
+};
+
+} // namespace
+
+std::shared_ptr<HashTableBuilder> buildHashTableWithTask(
+ const std::string& cacheKey,
+ const HashTableTaskBuildSpec& spec,
+ const std::vector<RowVectorPtr>& buildVectors,
+ uint32_t numDrivers,
+ folly::Executor* executor,
+ std::shared_ptr<memory::MemoryPool> queryPool,
+ memory::MemoryPool* vectorPool) {
+ VELOX_CHECK_GT(numDrivers, 0);
+ VELOX_CHECK(numDrivers == 1 || executor != nullptr, "Parallel hash table build requires an executor");
+ const auto buildId = nextBuildId();
+ auto queryCtx = core::QueryCtx::create(
+ numDrivers > 1 ? executor : nullptr,
+ core::QueryConfig(spec.queryConfigs),
+ {},
+ nullptr,
+ std::move(queryPool),
+ nullptr,
+ fmt::format("Gluten_HashTableBuild_{}_{}", cacheKey, buildId));
+
+ // Probe side: an empty input with one column per join key.
+ std::vector<std::string> probeNames;
+ std::vector<TypePtr> probeTypes;
+ std::vector<core::FieldAccessTypedExprPtr> leftKeys;
+ std::vector<core::FieldAccessTypedExprPtr> rightKeys;
+ for (auto i = 0; i < spec.joinKeys.size(); ++i) {
+ const auto& type = spec.buildType->findChild(spec.joinKeys[i]);
+ probeNames.emplace_back(fmt::format("__fake_probe_k{}", i));
+ probeTypes.emplace_back(type);
+ leftKeys.emplace_back(std::make_shared<core::FieldAccessTypedExpr>(type, probeNames.back()));
+ rightKeys.emplace_back(std::make_shared<core::FieldAccessTypedExpr>(type, spec.joinKeys[i]));
+ }
+ auto probeType = ROW(std::move(probeNames), std::move(probeTypes));
+
+ int planNodeId = 0;
+ auto nextPlanNodeId = [&]() { return std::to_string(planNodeId++); };
+ core::PlanNodePtr probe = std::make_shared<core::ValuesNode>(
+ nextPlanNodeId(), std::vector<RowVectorPtr>{BaseVector::create<RowVector>(probeType, 0, vectorPool)});
+
+ // Build side: the broadcast vectors, renamed to 'buildType'.
+ std::vector<RowVectorPtr> values;
+ values.reserve(buildVectors.size());
+ for (const auto& vector : buildVectors) {
+ values.emplace_back(std::make_shared<RowVector>(
+ vector->pool(), spec.buildType, vector->nulls(), vector->size(), vector->children()));
+ }
Review Comment:
The comment says the build input is ‘fanned out’ to all drivers, but `LocalPartitionNode::Type::kRepartition` with `RoundRobinPartitionFunctionSpec` partitions the rows across drivers (it doesn’t duplicate/broadcast them). Please adjust the comment to reflect the actual behavior (partitioning/distribution) to avoid misleading future maintainers.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN][VL][TEST] Use fake task to build hash table [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN][VL][TEST] Use fake task to build hash table [gluten]
Posted by "JkSelf (via GitHub)" <gi...@apache.org>.
JkSelf closed pull request #13216: [GLUTEN][VL][TEST] Use fake task to build hash table
URL: https://github.com/apache/gluten/pull/13216
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN][VL][TEST] Use fake task to build hash table [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #13216:
URL: https://github.com/apache/gluten/pull/13216#discussion_r4312928794
##########
cpp/velox/operators/hashjoin/HashTableTaskBuilder.cc:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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 <atomic>
+
+#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 join outputs
+// nothing and needs no columns, except the match column of a semi project join.
+RowTypePtr fakeJoinOutputType(core::JoinType joinType) {
+ if (core::isLeftSemiProjectJoin(joinType) || core::isRightSemiProjectJoin(joinType)) {
+ return ROW({"__fake_match"}, {BOOLEAN()});
+ }
+ return ROW({}, {});
+}
+
+// A stand-in for the join filter with the properties HashBuild looks at, see
+// HashTableTaskBuildSpec. It is compiled by HashProbe but never evaluated.
+core::TypedExprPtr makePlaceholderFilter(const HashTableTaskBuildSpec& spec) {
+ if (!spec.withFilter) {
+ return nullptr;
+ }
+ if (!spec.filterPropagatesNulls || spec.filterBuildColumns.empty()) {
+ // References no build column, so HashBuild drops no build rows for it.
+ return std::make_shared<core::ConstantTypedExpr>(BOOLEAN(), Variant(true));
+ }
+ // Chains the build columns, cast to VARCHAR to fit any type, with equalto.
+ // Both have default null behavior, so the result is null-intolerant in each
+ // of them.
+ auto toVarchar = [&](const core::TypedExprPtr& expr) -> core::TypedExprPtr {
+ return std::make_shared<core::CastTypedExpr>(VARCHAR(), expr, false);
+ };
+ auto column = [&](const std::string& name) -> core::TypedExprPtr {
+ return toVarchar(std::make_shared<core::FieldAccessTypedExpr>(spec.buildType->findChild(name), name));
+ };
+ auto equalTo = [](const core::TypedExprPtr& left, const core::TypedExprPtr& right) -> core::TypedExprPtr {
+ return std::make_shared<core::CallTypedExpr>(BOOLEAN(), std::vector<core::TypedExprPtr>{left, right}, "equalto");
+ };
+ auto filter = equalTo(column(spec.filterBuildColumns[0]), column(spec.filterBuildColumns[0]));
+ for (auto i = 1; i < spec.filterBuildColumns.size(); ++i) {
+ filter = equalTo(toVarchar(filter), column(spec.filterBuildColumns[i]));
+ }
+ return filter;
+}
+
+// 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:
`makeEmptyJoinTable()` hard-codes both the hasher channel indices (`i`) and the hash table template (`HashTable<false>`). The hasher channel should correspond to the actual build-side column index (e.g., `buildType->getChildIdxIfExists(joinKey)`), not the ordinal position in `joinKeys`. Also, using `HashTable<false>` can change the “ignore null keys” behavior observed by downstream code (e.g., JNI `getIgnoreNullKeys()` uses a `dynamic_cast<HashTable<true>>`). Even if this table is only used as a fallback, it can still be serialized/broadcast or consulted for flags, so the shape/type should match what Velox `HashBuild` would have produced for this join/spec. Update the fallback to (1) build hashers with correct input channels, and (2) pick `HashTable<true>` vs `HashTable<false>` consistently with the join’s null-key handling (and ideally align any other flags that affect row layout/serialization like allowDuplicates / nextOffset).
##########
cpp/velox/jni/JniHashTable.cc:
##########
@@ -116,139 +120,87 @@ std::shared_ptr<HashTableBuilder> nativeHashTableBuild(
VELOX_NYI("Unsupported Join type: {}", std::to_string(sJoin));
}
- std::vector<std::shared_ptr<const facebook::velox::core::FieldAccessTypedExpr>> joinKeyTypes;
- joinKeyTypes.reserve(joinKeys.size());
- for (const auto& name : joinKeys) {
- joinKeyTypes.emplace_back(
- std::make_shared<facebook::velox::core::FieldAccessTypedExpr>(rowType->findChild(name), name));
- }
-
- std::vector<uint32_t> filterInputChannels;
- filterInputChannels.reserve(filterBuildColumns.size());
+ using facebook::velox::core::QueryConfig;
+ HashTableTaskBuildSpec spec;
+ spec.joinType = vJoin;
+ spec.nullAware = isNullAwareAntiJoin;
+ spec.withFilter = hasMixedJoinCondition;
for (const auto& name : filterBuildColumns) {
- if (const auto idx = rowType->getChildIdxIfExists(name)) {
- filterInputChannels.push_back(*idx);
+ if (rowType->containsChild(name) &&
+ std::find(spec.filterBuildColumns.begin(), spec.filterBuildColumns.end(), name) ==
+ spec.filterBuildColumns.end()) {
+ spec.filterBuildColumns.push_back(name);
}
}
Review Comment:
This de-duplicates `filterBuildColumns` using a linear `std::find` on a growing vector, making it O(n²). If `filterBuildColumns` can be large (or this is called frequently), consider using a `folly::F14FastSet`/`std::unordered_set` for membership checks while preserving insertion order (e.g., keep the vector for order but guard pushes with a set).
##########
cpp/velox/jni/JniHashTable.cc:
##########
@@ -116,139 +120,87 @@ std::shared_ptr<HashTableBuilder> nativeHashTableBuild(
VELOX_NYI("Unsupported Join type: {}", std::to_string(sJoin));
}
- std::vector<std::shared_ptr<const facebook::velox::core::FieldAccessTypedExpr>> joinKeyTypes;
- joinKeyTypes.reserve(joinKeys.size());
- for (const auto& name : joinKeys) {
- joinKeyTypes.emplace_back(
- std::make_shared<facebook::velox::core::FieldAccessTypedExpr>(rowType->findChild(name), name));
- }
-
- std::vector<uint32_t> filterInputChannels;
- filterInputChannels.reserve(filterBuildColumns.size());
+ using facebook::velox::core::QueryConfig;
+ HashTableTaskBuildSpec spec;
+ spec.joinType = vJoin;
+ spec.nullAware = isNullAwareAntiJoin;
+ spec.withFilter = hasMixedJoinCondition;
for (const auto& name : filterBuildColumns) {
- if (const auto idx = rowType->getChildIdxIfExists(name)) {
- filterInputChannels.push_back(*idx);
+ if (rowType->containsChild(name) &&
+ std::find(spec.filterBuildColumns.begin(), spec.filterBuildColumns.end(), name) ==
+ spec.filterBuildColumns.end()) {
+ spec.filterBuildColumns.push_back(name);
}
}
- std::sort(filterInputChannels.begin(), filterInputChannels.end());
- filterInputChannels.erase(
- std::unique(filterInputChannels.begin(), filterInputChannels.end()), filterInputChannels.end());
-
- auto hashTableBuilder = std::make_shared<HashTableBuilder>(
- vJoin,
- isNullAwareAntiJoin,
- hasMixedJoinCondition,
- bloomFilterPushdownSize,
- joinKeyTypes,
- filterInputChannels,
- filterPropagatesNulls,
- rowType,
- memoryPool.get(),
- minTableRowsForParallelJoinBuild,
- joinBuildVectorHasherMaxNumDistinct,
- abandonHashBuildDedupMinRows,
- abandonHashBuildDedupMinPct);
-
- for (auto i = 0; i < batches.size(); i++) {
- auto rowVector = VeloxColumnarBatch::from(memoryPool.get(), batches[i])->getRowVector();
- hashTableBuilder->addInput(rowVector);
- if (hashTableBuilder->noMoreInput()) {
- break;
- }
+ spec.filterPropagatesNulls = filterPropagatesNulls;
+ spec.buildType = rowType;
+ 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))},
Review Comment:
The new path clamps `bloomFilterPushdownSize` to `>= 0` before passing it via `QueryConfig`. If callers previously used a negative value (e.g., `-1`) with a special meaning (commonly “disabled”), this changes semantics to `0`, which might not be interpreted the same way by Velox config parsing. To preserve behavior, pass the value through unchanged (or map it using the same rule Velox expects for this specific config).
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN][VL][TEST] Use fake task to build hash table [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #13216:
URL: https://github.com/apache/gluten/pull/13216#discussion_r4269853975
##########
cpp/velox/jni/JniHashTable.cc:
##########
@@ -116,139 +120,87 @@ std::shared_ptr<HashTableBuilder> nativeHashTableBuild(
VELOX_NYI("Unsupported Join type: {}", std::to_string(sJoin));
}
- std::vector<std::shared_ptr<const facebook::velox::core::FieldAccessTypedExpr>> joinKeyTypes;
- joinKeyTypes.reserve(joinKeys.size());
- for (const auto& name : joinKeys) {
- joinKeyTypes.emplace_back(
- std::make_shared<facebook::velox::core::FieldAccessTypedExpr>(rowType->findChild(name), name));
- }
-
- std::vector<uint32_t> filterInputChannels;
- filterInputChannels.reserve(filterBuildColumns.size());
+ using facebook::velox::core::QueryConfig;
+ HashTableTaskBuildSpec spec;
+ spec.joinType = vJoin;
+ spec.nullAware = isNullAwareAntiJoin;
+ spec.withFilter = hasMixedJoinCondition;
for (const auto& name : filterBuildColumns) {
- if (const auto idx = rowType->getChildIdxIfExists(name)) {
- filterInputChannels.push_back(*idx);
+ if (rowType->containsChild(name) &&
+ std::find(spec.filterBuildColumns.begin(), spec.filterBuildColumns.end(), name) ==
+ spec.filterBuildColumns.end()) {
Review Comment:
When `hasMixedJoinCondition` is true, missing/incorrect `filterBuildColumns` are silently ignored. For anti-joins with null-intolerant filters, this can change semantics because HashBuild’s “drop rows with null build filter columns” behavior depends on these columns. It would be safer to validate that each requested filter build column exists (fail fast), or at least error/LOG when a specified column name is not found instead of silently proceeding with an incomplete spec.
##########
cpp/velox/operators/hashjoin/HashTableTaskBuilder.cc:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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 <atomic>
+
+#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 join outputs
+// nothing and needs no columns, except the match column of a semi project join.
+RowTypePtr fakeJoinOutputType(core::JoinType joinType) {
+ if (core::isLeftSemiProjectJoin(joinType) || core::isRightSemiProjectJoin(joinType)) {
+ return ROW({"__fake_match"}, {BOOLEAN()});
+ }
+ return ROW({}, {});
+}
+
+// A stand-in for the join filter with the properties HashBuild looks at, see
+// HashTableTaskBuildSpec. It is compiled by HashProbe but never evaluated.
+core::TypedExprPtr makePlaceholderFilter(const HashTableTaskBuildSpec& spec) {
+ if (!spec.withFilter) {
+ return nullptr;
+ }
+ if (!spec.filterPropagatesNulls || spec.filterBuildColumns.empty()) {
+ // References no build column, so HashBuild drops no build rows for it.
+ return std::make_shared<core::ConstantTypedExpr>(BOOLEAN(), Variant(true));
+ }
+ // Chains the build columns, cast to VARCHAR to fit any type, with equalto.
+ // Both have default null behavior, so the result is null-intolerant in each
+ // of them.
+ auto toVarchar = [&](const core::TypedExprPtr& expr) -> core::TypedExprPtr {
+ return std::make_shared<core::CastTypedExpr>(VARCHAR(), expr, false);
+ };
+ auto column = [&](const std::string& name) -> core::TypedExprPtr {
+ return toVarchar(std::make_shared<core::FieldAccessTypedExpr>(spec.buildType->findChild(name), name));
+ };
+ auto equalTo = [](const core::TypedExprPtr& left, const core::TypedExprPtr& right) -> core::TypedExprPtr {
+ return std::make_shared<core::CallTypedExpr>(BOOLEAN(), std::vector<core::TypedExprPtr>{left, right}, "equalto");
+ };
+ auto filter = equalTo(column(spec.filterBuildColumns[0]), column(spec.filterBuildColumns[0]));
+ for (auto i = 1; i < spec.filterBuildColumns.size(); ++i) {
+ filter = equalTo(toVarchar(filter), column(spec.filterBuildColumns[i]));
+ }
+ return filter;
+}
+
+// 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 constructed with `channel = i`, but `i` is just the position within `spec.joinKeys`, not the actual child index in `spec.buildType`. If join keys are not the first N columns (or not in the same order), the empty table will have hashers pointing at the wrong input channels, which can break probing or trigger invariants. Use the actual column index from `buildType` (e.g., `buildType->getChildIdx(...)` / equivalent) when constructing hashers.
##########
cpp/velox/tests/HashTableTaskBuilderTest.cc:
##########
@@ -0,0 +1,328 @@
+/*
+ * 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 <folly/init/Init.h>
+#include <gtest/gtest.h>
+
+#include "jni/JniHashTable.h"
+#include "operators/hashjoin/HashTableSerializer.h"
+#include "operators/hashjoin/HashTableTaskBuilder.h"
+#include "velox/exec/HashTableCache.h"
+#include "velox/exec/tests/utils/AssertQueryBuilder.h"
+#include "velox/exec/tests/utils/OperatorTestBase.h"
+#include "velox/exec/tests/utils/QueryAssertions.h"
+#include "velox/functions/sparksql/registration/Register.h"
+
+using namespace facebook::velox;
+using namespace facebook::velox::exec;
+
+namespace gluten {
+namespace {
+
+enum class Filter {
+ kNone,
+ // t_v > u_v: null-intolerant, so an anti join drops build rows with a null u_v.
+ kNullIntolerant,
+ // u_v IS NULL OR t_v > u_v: an anti join keeps build rows with a null u_v.
+ kNullTolerant,
+};
+
+struct JoinCase {
+ std::string name;
+ core::JoinType joinType;
+ bool nullAware{false};
+ Filter filter{Filter::kNone};
+ bool nullBuildKeys{false};
+ uint32_t numDrivers{1};
+};
+
+std::string caseName(const testing::TestParamInfo<JoinCase>& info) {
+ return info.param.name;
+}
+
+class HashTableTaskBuilderTest : public exec::test::OperatorTestBase, public testing::WithParamInterface<JoinCase> {
+ protected:
+ static void SetUpTestCase() {
+ OperatorTestBase::SetUpTestCase();
+ functions::sparksql::registerFunctions("");
+ }
+
+ // Build side: 'numBatches' x 'batchSize' rows with duplicate keys.
+ std::vector<RowVectorPtr> makeBuildVectors(bool nullKeys, int numBatches = 8, int batchSize = 500) {
+ std::vector<RowVectorPtr> vectors;
+ for (int batch = 0; batch < numBatches; ++batch) {
+ vectors.push_back(makeRowVector(
+ {"u_k", "u_v"},
+ {makeFlatVector<int32_t>(
+ batchSize,
+ [batch](auto row) { return (row * 7 + batch) % 997; },
+ [nullKeys, batch](auto row) { return nullKeys && batch == 3 && row == 17; }),
+ makeFlatVector<int64_t>(
+ batchSize, [](auto row) { return row % 101; }, [](auto row) { return row % 37 == 0; })}));
+ }
+ return vectors;
+ }
+
+ std::vector<RowVectorPtr> makeProbeVectors() {
+ std::vector<RowVectorPtr> vectors;
+ for (int batch = 0; batch < 4; ++batch) {
+ vectors.push_back(makeRowVector(
+ {"t_k", "t_v"},
+ {makeFlatVector<int32_t>(
+ 1'000, [batch](auto row) { return (row + batch * 1'000) % 1'500; }, [](auto row) { return row == 5; }),
+ makeFlatVector<int64_t>(1'000, [](auto row) { return row % 103; })}));
+ }
+ return vectors;
+ }
+
+ // The real join filter, see Filter.
+ static core::TypedExprPtr makeFilter(Filter filter) {
+ if (filter == Filter::kNone) {
+ return nullptr;
+ }
+ auto uv = std::make_shared<core::FieldAccessTypedExpr>(BIGINT(), "u_v");
+ core::TypedExprPtr greaterThan = std::make_shared<core::CallTypedExpr>(
+ BOOLEAN(),
+ std::vector<core::TypedExprPtr>{std::make_shared<core::FieldAccessTypedExpr>(BIGINT(), "t_v"), uv},
+ "greaterthan");
+ if (filter == Filter::kNullIntolerant) {
+ return greaterThan;
+ }
+ return std::make_shared<core::CallTypedExpr>(
+ BOOLEAN(),
+ std::vector<core::TypedExprPtr>{
+ std::make_shared<core::CallTypedExpr>(BOOLEAN(), std::vector<core::TypedExprPtr>{uv}, "isnull"),
+ greaterThan},
+ "or");
+ }
+
+ // Builds with the filter properties the Scala side derives from makeFilter().
+ std::shared_ptr<BroadcastHashTable>
+ build(const JoinCase& c, const std::string& key, const std::vector<RowVectorPtr>& vectors) {
+ HashTableTaskBuildSpec spec;
+ spec.joinType = c.joinType;
+ spec.nullAware = c.nullAware;
+ spec.withFilter = c.filter != Filter::kNone;
+ if (spec.withFilter) {
+ spec.filterBuildColumns = {"u_v"};
+ }
+ spec.filterPropagatesNulls = c.filter == Filter::kNullIntolerant;
+ spec.buildType = ROW({"u_k", "u_v"}, {INTEGER(), BIGINT()});
+ spec.joinKeys = {"u_k"};
+ // Build in parallel whatever the table size, so that parallel cases merge per-driver tables.
+ spec.queryConfigs = {{core::QueryConfig::kMinTableRowsForParallelJoinBuild, "1"}};
+ return buildHashTableWithTask(key, spec, vectors, c.numDrivers, driverExecutor_.get(), rootPool_, pool());
+ }
+
+ // Joins the probe vectors with 'buildVectors'. With 'cacheKey', the probe
+ // uses the table cached under that key and 'buildVectors' must be empty.
+ RowVectorPtr join(
+ const JoinCase& c,
+ const std::vector<RowVectorPtr>& buildVectors,
+ const std::optional<std::string>& cacheKey = std::nullopt) {
+ auto probeNode = std::make_shared<core::ValuesNode>("0", makeProbeVectors());
+ auto buildNode = std::make_shared<core::ValuesNode>(
+ "1",
+ buildVectors.empty()
+ ? std::vector<RowVectorPtr>{makeRowVector(
+ {"u_k", "u_v"},
+ {makeFlatVector<int32_t>(std::vector<int32_t>{}), makeFlatVector<int64_t>(std::vector<int64_t>{})})}
+ : buildVectors);
+ RowTypePtr outputType;
+ switch (c.joinType) {
+ case core::JoinType::kInner:
+ case core::JoinType::kLeft:
+ case core::JoinType::kRight:
+ case core::JoinType::kFull:
+ outputType = ROW({"t_k", "t_v", "u_k", "u_v"}, {INTEGER(), BIGINT(), INTEGER(), BIGINT()});
+ break;
+ case core::JoinType::kLeftSemiProject:
+ outputType = ROW({"t_k", "t_v", "match"}, {INTEGER(), BIGINT(), BOOLEAN()});
+ break;
+ case core::JoinType::kRightSemiFilter:
+ outputType = ROW({"u_k", "u_v"}, {INTEGER(), BIGINT()});
+ break;
+ case core::JoinType::kRightSemiProject:
+ outputType = ROW({"u_k", "u_v", "match"}, {INTEGER(), BIGINT(), BOOLEAN()});
+ break;
+ default:
+ outputType = ROW({"t_k", "t_v"}, {INTEGER(), BIGINT()});
+ }
+ auto joinNode = core::HashJoinNode::Builder()
+ .id("2")
+ .joinType(c.joinType)
+ .leftKeys({std::make_shared<core::FieldAccessTypedExpr>(INTEGER(), "t_k")})
+ .rightKeys({std::make_shared<core::FieldAccessTypedExpr>(INTEGER(), "u_k")})
+ .filter(makeFilter(c.filter))
+ .left(probeNode)
+ .right(buildNode)
+ .outputType(outputType)
+ .nullAware(c.nullAware)
+ .useHashTableCache(cacheKey.has_value())
+ .cacheKey(cacheKey)
+ .build();
+ return exec::test::AssertQueryBuilder(joinNode).copyResults(pool());
+ }
+};
+
+// Probing the table built by the fake join gives the same result as a regular
+// Velox hash join over the same build rows.
+TEST_P(HashTableTaskBuilderTest, sameResultAsHashJoin) {
+ const auto& c = GetParam();
+ auto* cache = HashTableCache::instance();
+ const auto key = "task_" + c.name;
+ auto vectors = makeBuildVectors(c.nullBuildKeys);
+
+ auto hashTable = build(c, key, vectors);
+ ASSERT_NE(hashTable, nullptr);
+ ASSERT_TRUE(cache->exist(key));
+ EXPECT_GT(hashTable->memoryUsage(), 0);
+ if (c.numDrivers > 1 && !hashTable->joinHasNullKeys()) {
+ // HashBuild built one table per driver and merged them.
+ EXPECT_EQ(hashTable->hashTable()->allRows().size(), c.numDrivers);
+ }
+
+ exec::test::assertEqualResults({join(c, vectors)}, {join(c, {}, key)});
+
+ // The cache entry, which owns the table's pool, can go first. The table
+ // must keep its memory pool alive in between.
+ cache->drop(key);
+ auto table = hashTable->hashTable();
+ hashTable.reset();
+ EXPECT_NE(table->rows(), nullptr);
+ table.reset();
+}
+
+// Releasing the last reference to the table drops the cache entry.
+TEST_P(HashTableTaskBuilderTest, releasingTableDropsCacheEntry) {
+ const auto& c = GetParam();
+ auto* cache = HashTableCache::instance();
+ const auto key = "release_" + c.name;
+ auto hashTable = build(c, key, makeBuildVectors(c.nullBuildKeys));
+ ASSERT_TRUE(cache->exist(key));
+ hashTable.reset();
+ EXPECT_FALSE(cache->exist(key));
+}
+
+// A probe task takes the table from the cache and can outlive the cache entry
+// and the returned builder, e.g. when the task fails while the broadcast table
+// is released. The table it holds must keep its memory pool alive.
+TEST_P(HashTableTaskBuilderTest, cachedTableOutlivesBuilder) {
+ const auto& c = GetParam();
+ auto* cache = HashTableCache::instance();
+ const auto key = "outlive_" + c.name;
+ auto hashTable = build(c, key, makeBuildVectors(c.nullBuildKeys));
+
+ ContinueFuture future = ContinueFuture::makeEmpty();
+ auto probeQueryCtx = core::QueryCtx::create();
+ auto entry = cache->get(key, "probe_task", probeQueryCtx.get(), &future);
+ ASSERT_FALSE(future.valid());
+ auto probeTable = entry->table;
+ std::weak_ptr<memory::MemoryPool> tablePool = entry->tablePool;
+ entry.reset();
+
+ hashTable.reset();
+ EXPECT_FALSE(cache->exist(key));
+ EXPECT_FALSE(tablePool.expired());
+ EXPECT_NE(probeTable->rows(), nullptr);
+ probeTable.reset();
+ EXPECT_TRUE(tablePool.expired());
+}
+
+// The driver-side build serializes the table and executors deserialize it.
+TEST_P(HashTableTaskBuilderTest, serializeRoundTrip) {
+ const auto& c = GetParam();
+ auto* cache = HashTableCache::instance();
+ const auto key = "serialize_" + c.name;
+ auto vectors = makeBuildVectors(c.nullBuildKeys);
+ auto hashTable = build(c, key, vectors);
+
+ std::vector<uint8_t> data(serializedHashTableSize(*hashTable));
+ serializeHashTableTo(*hashTable, data.data(), data.size());
+ // As deserializeHashTable(), with the test's pool.
+ std::shared_ptr<BaseHashTable> deserialized;
+ if (dynamic_cast<HashTable<true>*>(hashTable->hashTable().get()) != nullptr) {
+ deserialized = HashTableSerializer::deserialize<true>(data.data(), data.size(), pool());
+ } else {
+ deserialized = HashTableSerializer::deserialize<false>(data.data(), data.size(), pool());
+ }
+ const auto deserializedKey = "deserialized_" + c.name;
+ cache->add(deserializedKey, deserialized, hashTable->joinHasNullKeys(), pool_);
+
+ exec::test::assertEqualResults({join(c, vectors)}, {join(c, {}, deserializedKey)});
+ cache->drop(deserializedKey);
+}
+
+// For tests that do not take a JoinCase parameter.
+class HashTableTaskBuilderSingleTest : public HashTableTaskBuilderTest {};
+
+// A build without input rows still caches an empty table.
+TEST_F(HashTableTaskBuilderSingleTest, emptyBuildSide) {
+ const JoinCase c{"empty", core::JoinType::kInner};
+ auto* cache = HashTableCache::instance();
+ auto hashTable = build(c, "empty", {});
+ ASSERT_TRUE(cache->exist("empty"));
+ EXPECT_EQ(hashTable->hashTable()->numDistinct(), 0);
+ hashTable.reset();
+ EXPECT_FALSE(cache->exist("empty"));
+}
+
+INSTANTIATE_TEST_SUITE_P(
+ HashTableTaskBuilderTest,
+ HashTableTaskBuilderTest,
+ testing::Values(
+ JoinCase{"inner", core::JoinType::kInner},
+ JoinCase{"innerParallel", core::JoinType::kInner, false, Filter::kNone, false, 4},
+ JoinCase{"innerFilter", core::JoinType::kInner, false, Filter::kNullIntolerant},
+ JoinCase{"leftParallel", core::JoinType::kLeft, false, Filter::kNone, true, 4},
+ JoinCase{"leftFilter", core::JoinType::kLeft, false, Filter::kNullIntolerant, true},
+ JoinCase{"right", core::JoinType::kRight, false, Filter::kNone, true},
+ JoinCase{"rightFilterParallel", core::JoinType::kRight, false, Filter::kNullIntolerant, true, 4},
+ JoinCase{"full", core::JoinType::kFull, false, Filter::kNone, true},
+ JoinCase{"fullParallel", core::JoinType::kFull, false, Filter::kNone, true, 4},
+ JoinCase{"rightSemi", core::JoinType::kRightSemiFilter, false, Filter::kNone, true},
+ JoinCase{"rightSemiFilterParallel", core::JoinType::kRightSemiFilter, false, Filter::kNullIntolerant, true, 4},
+ JoinCase{"rightSemiProject", core::JoinType::kRightSemiProject, false, Filter::kNone, true},
+ JoinCase{"rightSemiProjectParallel", core::JoinType::kRightSemiProject, false, Filter::kNone, true, 4},
+ JoinCase{"leftSemi", core::JoinType::kLeftSemiFilter},
+ JoinCase{"leftSemiParallel", core::JoinType::kLeftSemiFilter, false, Filter::kNone, true, 4},
+ JoinCase{"leftSemiFilterParallel", core::JoinType::kLeftSemiFilter, false, Filter::kNullIntolerant, false, 4},
+ JoinCase{"existence", core::JoinType::kLeftSemiProject},
+ JoinCase{"existenceFilter", core::JoinType::kLeftSemiProject, false, Filter::kNullIntolerant, true},
+ JoinCase{"anti", core::JoinType::kAnti, false, Filter::kNone, true},
+ JoinCase{"antiParallel", core::JoinType::kAnti, false, Filter::kNone, true, 4},
+ JoinCase{"antiFilter", core::JoinType::kAnti, false, Filter::kNullIntolerant, true},
+ JoinCase{"antiFilterParallel", core::JoinType::kAnti, false, Filter::kNullIntolerant, true, 4},
+ JoinCase{"antiNullTolerantFilter", core::JoinType::kAnti, false, Filter::kNullTolerant, true},
+ JoinCase{"antiNullTolerantFilterParallel", core::JoinType::kAnti, false, Filter::kNullTolerant, true, 4},
+ JoinCase{"nullAwareAnti", core::JoinType::kAnti, true},
+ JoinCase{"nullAwareAntiParallel", core::JoinType::kAnti, true, Filter::kNone, false, 4},
+ JoinCase{"nullAwareAntiNullKey", core::JoinType::kAnti, true, Filter::kNone, true},
+ JoinCase{"nullAwareAntiNullKeyParallel", core::JoinType::kAnti, true, Filter::kNone, true, 4},
+ JoinCase{"nullAwareAntiFilterNullKey", core::JoinType::kAnti, true, Filter::kNullIntolerant, true}),
+ caseName);
+
+} // namespace
+} // namespace gluten
+
+// folly::init() registers folly singletons, e.g. the Timekeeper that
+// AssertQueryBuilder waits with. Without it Linux builds abort.
+int main(int argc, char** argv) {
+ testing::InitGoogleTest(&argc, argv);
+ folly::init(&argc, &argv, false);
+ return RUN_ALL_TESTS();
+}
Review Comment:
Defining a custom `main()` in a test source can cause duplicate-symbol link failures (if the test target already links a shared gtest/Velox test main), or can bypass any common test initialization your `add_velox_test(...)` infrastructure expects. Prefer wiring `folly::init()` into the existing shared test main (or the test framework’s init path) rather than providing a per-test `main()`. If `add_velox_test` intentionally builds each test without a default main, make that explicit in CMake to prevent future collisions.
##########
cpp/velox/tests/HashTableTaskBuilderTest.cc:
##########
@@ -0,0 +1,328 @@
+/*
+ * 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 <folly/init/Init.h>
+#include <gtest/gtest.h>
+
+#include "jni/JniHashTable.h"
Review Comment:
This unit test pulls in `jni/JniHashTable.h`, which includes `<jni.h>` and JNI-related dependencies. That can make the C++ test target depend on a JVM/JNI dev environment even though the test is exercising hash-table serialization/build logic. Consider moving `serializedHashTableSize(...)` / `serializeHashTableTo(...)` declarations to a non-JNI header (or using `HashTableSerializer` directly in the test) so pure C++ tests don’t require JNI headers/tooling.
##########
cpp/velox/jni/VeloxJniWrapper.cc:
##########
@@ -1022,145 +1022,32 @@ JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_HashJoinBuilder_native
cb.push_back(ObjectStore::retrieve<ColumnarBatch>(handle));
}
- if (numThreads == 1) {
- // Use default global pool for driver-side build
- // The hash table will be serialized and broadcast, so it doesn't need runtime's pool
- // Using runtime pool causes lifecycle management issues
- auto memoryPool = defaultLeafVeloxMemoryPool();
-
- auto builder = nativeHashTableBuild(
- hashJoinKeys,
- filterColumns,
- filterPropagatesNulls,
- names,
- veloxTypeList,
- joinType,
- hasMixedJoinCondition,
- isExistenceJoin,
- isNullAwareAntiJoin,
- bloomFilterPushdownSize,
- minTableRowsForParallelJoinBuild,
- joinBuildVectorHasherMaxNumDistinct,
- abandonHashBuildDedupMinRows,
- abandonHashBuildDedupMinPct,
- cb,
- memoryPool);
-
- auto mainTable = builder->uniqueTable();
- mainTable->prepareJoinTable(
- {},
- facebook::velox::exec::BaseHashTable::kNoSpillInputStartPartitionBit,
- builder->joinBuildVectorHasherMaxNumDistinct(),
- builder->dropDuplicates(),
- nullptr);
- builder->setHashTable(std::move(mainTable));
- builder->setHashTableMemoryUsage(builder->hashTable()->allocatedBytes());
-
- auto* cache = facebook::velox::exec::HashTableCache::instance();
-
- if (!cache->exist(hashTableId)) {
- cache->add(hashTableId, builder->hashTable(), builder->joinHasNullKeys(), defaultLeafVeloxMemoryPool());
- }
-
- return gluten::getHashTableObjStore()->save(builder);
- }
-
- // Use thread pool (executor) instead of creating threads directly.
// FIXME: This reuses the io executor which is supposed to only serve async IO tasks.
- auto executor = VeloxBackend::get()->ioExecutor();
-
- std::vector<std::shared_ptr<gluten::HashTableBuilder>> hashTableBuilders(numThreads);
- std::vector<std::unique_ptr<facebook::velox::exec::BaseHashTable>> otherTables(numThreads);
- std::vector<folly::Future<folly::Unit>> futures;
- futures.reserve(numThreads);
-
- for (size_t t = 0; t < numThreads; ++t) {
- size_t start = (handleCount * t) / numThreads;
- size_t end = (handleCount * (t + 1)) / numThreads;
-
- // Submit task to thread pool
- auto future = folly::via(executor, [&, t, start, end]() {
- std::vector<std::shared_ptr<gluten::ColumnarBatch>> threadBatches;
- threadBatches.reserve(end - start);
- for (size_t i = start; i < end; ++i) {
- threadBatches.push_back(cb[i]);
- }
-
- // Use default global pool for driver-side build
- // The hash table will be serialized and broadcast, so it doesn't need runtime's pool
- auto threadMemoryPool = defaultLeafVeloxMemoryPool();
-
- auto builder = nativeHashTableBuild(
- hashJoinKeys,
- filterColumns,
- filterPropagatesNulls,
- names,
- veloxTypeList,
- joinType,
- hasMixedJoinCondition,
- isExistenceJoin,
- isNullAwareAntiJoin,
- bloomFilterPushdownSize,
- minTableRowsForParallelJoinBuild,
- joinBuildVectorHasherMaxNumDistinct,
- abandonHashBuildDedupMinRows,
- abandonHashBuildDedupMinPct,
- threadBatches,
- threadMemoryPool);
-
- hashTableBuilders[t] = std::move(builder);
- otherTables[t] = hashTableBuilders[t]->uniqueTable();
- });
-
- futures.push_back(std::move(future));
- }
-
- // Wait for all tasks to complete
- folly::collectAll(futures).wait();
-
- auto mainTable = std::move(otherTables[0]);
- std::vector<std::unique_ptr<facebook::velox::exec::BaseHashTable>> tables;
- for (int i = 1; i < numThreads; ++i) {
- tables.push_back(std::move(otherTables[i]));
- }
- int64_t otherTablesMemoryUsage = 0;
- for (const auto& table : tables) {
- otherTablesMemoryUsage += table->allocatedBytes();
- }
-
- // TODO: Get accurate signal if parallel join build is going to be applied
- // from hash table. Currently there is still a chance inside hash table that
- // it might decide it is not going to trigger parallel join build.
- const bool allowParallelJoinBuild = !tables.empty();
-
- mainTable->prepareJoinTable(
- std::move(tables),
- facebook::velox::exec::BaseHashTable::kNoSpillInputStartPartitionBit,
- hashTableBuilders[0]->joinBuildVectorHasherMaxNumDistinct(),
- hashTableBuilders[0]->dropDuplicates(),
- allowParallelJoinBuild ? VeloxBackend::get()->ioExecutor() : nullptr);
-
- for (int i = 1; i < numThreads; ++i) {
- if (hashTableBuilders[i]->joinHasNullKeys()) {
- hashTableBuilders[0]->setJoinHasNullKeys(true);
- break;
- }
- }
-
- hashTableBuilders[0]->setHashTable(std::move(mainTable));
- hashTableBuilders[0]->setHashTableMemoryUsage(
- hashTableBuilders[0]->hashTable()->allocatedBytes() + otherTablesMemoryUsage);
-
- 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]);
+ auto* executor = VeloxBackend::get()->ioExecutor();
+ // The build vectors and the table use global pools: the table outlives the
+ // runtime when it is cached or serialized and broadcast.
+ auto hashTable = nativeHashTableBuild(
+ hashTableId,
+ hashJoinKeys,
+ filterColumns,
+ filterPropagatesNulls,
+ names,
+ veloxTypeList,
+ joinType,
+ hasMixedJoinCondition,
+ isExistenceJoin,
+ isNullAwareAntiJoin,
+ bloomFilterPushdownSize,
+ minTableRowsForParallelJoinBuild,
+ joinBuildVectorHasherMaxNumDistinct,
+ abandonHashBuildDedupMinRows,
+ abandonHashBuildDedupMinPct,
+ cb,
+ executor == nullptr ? 1 : std::max(numThreads, 1),
+ executor,
+ getDefaultMemoryManager()->getAggregateMemoryPool(),
+ defaultLeafVeloxMemoryPool());
Review Comment:
The new build path still schedules CPU-heavy hash table build work on the IO executor (called out by the existing FIXME). This can starve IO tasks and create latency spikes under load. Prefer a dedicated CPU executor (or a dedicated executor owned by the Velox backend for compute tasks) for parallel join build, and reserve the IO executor for IO-bound work.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN][VL][TEST] Use fake task to build hash table [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #13216:
URL: https://github.com/apache/gluten/pull/13216#discussion_r4269693571
##########
cpp/velox/jni/VeloxJniWrapper.cc:
##########
Review Comment:
This `cloneHashTable` path adds a cache entry under a *new key* but (1) always uses `defaultLeafVeloxMemoryPool()` as `tablePool`, which may not be the pool the table was actually allocated from, and (2) does not tie the new cache entry’s lifetime to the returned handle (the original task-build code drops only the *original* cache key via a QueryCtx release callback). This can lead to incorrect memory accounting and/or leaked cache entries under the cloned key. Consider extending `BroadcastHashTable` to carry the correct `tablePool` (or enough info to retrieve it) and adding a release hook that drops `cacheKeyStr`, or provide a dedicated “clone/alias cache entry” API that preserves the original pool + lifetime semantics.
##########
cpp/velox/operators/hashjoin/HashTableTaskBuilder.cc:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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 <atomic>
+
+#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 join outputs
+// nothing and needs no columns, except the match column of a semi project join.
+RowTypePtr fakeJoinOutputType(core::JoinType joinType) {
+ if (core::isLeftSemiProjectJoin(joinType) || core::isRightSemiProjectJoin(joinType)) {
+ return ROW({"__fake_match"}, {BOOLEAN()});
+ }
+ return ROW({}, {});
+}
+
+// A stand-in for the join filter with the properties HashBuild looks at, see
+// HashTableTaskBuildSpec. It is compiled by HashProbe but never evaluated.
+core::TypedExprPtr makePlaceholderFilter(const HashTableTaskBuildSpec& spec) {
+ if (!spec.withFilter) {
+ return nullptr;
+ }
+ if (!spec.filterPropagatesNulls || spec.filterBuildColumns.empty()) {
+ // References no build column, so HashBuild drops no build rows for it.
+ return std::make_shared<core::ConstantTypedExpr>(BOOLEAN(), Variant(true));
+ }
+ // Chains the build columns, cast to VARCHAR to fit any type, with equalto.
+ // Both have default null behavior, so the result is null-intolerant in each
+ // of them.
+ auto toVarchar = [&](const core::TypedExprPtr& expr) -> core::TypedExprPtr {
+ return std::make_shared<core::CastTypedExpr>(VARCHAR(), expr, false);
+ };
+ auto column = [&](const std::string& name) -> core::TypedExprPtr {
+ return toVarchar(std::make_shared<core::FieldAccessTypedExpr>(spec.buildType->findChild(name), name));
+ };
+ auto equalTo = [](const core::TypedExprPtr& left, const core::TypedExprPtr& right) -> core::TypedExprPtr {
+ return std::make_shared<core::CallTypedExpr>(BOOLEAN(), std::vector<core::TypedExprPtr>{left, right}, "equalto");
+ };
+ auto filter = equalTo(column(spec.filterBuildColumns[0]), column(spec.filterBuildColumns[0]));
+ for (auto i = 1; i < spec.filterBuildColumns.size(); ++i) {
+ filter = equalTo(toVarchar(filter), column(spec.filterBuildColumns[i]));
+ }
+ return filter;
+}
+
+// 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));
Review Comment:
The `VectorHasher` channel is set to `i`, but `i` is just the position in `spec.joinKeys`, not necessarily the actual column index in `spec.buildType` (and it breaks outright if join keys repeat the same build column, e.g. `t.k1 = u.k AND t.k2 = u.k`). Use the real channel index from `buildType` (e.g. `getChildIdx(...)` / equivalent) and prefer `VectorHasher::create(type, channel)` to match Velox’s expected construction path.
##########
cpp/velox/tests/HashTableTaskBuilderTest.cc:
##########
@@ -0,0 +1,328 @@
+/*
+ * 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 <folly/init/Init.h>
+#include <gtest/gtest.h>
+
+#include "jni/JniHashTable.h"
+#include "operators/hashjoin/HashTableSerializer.h"
+#include "operators/hashjoin/HashTableTaskBuilder.h"
+#include "velox/exec/HashTableCache.h"
+#include "velox/exec/tests/utils/AssertQueryBuilder.h"
+#include "velox/exec/tests/utils/OperatorTestBase.h"
+#include "velox/exec/tests/utils/QueryAssertions.h"
+#include "velox/functions/sparksql/registration/Register.h"
+
+using namespace facebook::velox;
+using namespace facebook::velox::exec;
+
+namespace gluten {
+namespace {
+
+enum class Filter {
+ kNone,
+ // t_v > u_v: null-intolerant, so an anti join drops build rows with a null u_v.
+ kNullIntolerant,
+ // u_v IS NULL OR t_v > u_v: an anti join keeps build rows with a null u_v.
+ kNullTolerant,
+};
+
+struct JoinCase {
+ std::string name;
+ core::JoinType joinType;
+ bool nullAware{false};
+ Filter filter{Filter::kNone};
+ bool nullBuildKeys{false};
+ uint32_t numDrivers{1};
+};
+
+std::string caseName(const testing::TestParamInfo<JoinCase>& info) {
+ return info.param.name;
+}
+
+class HashTableTaskBuilderTest : public exec::test::OperatorTestBase, public testing::WithParamInterface<JoinCase> {
+ protected:
+ static void SetUpTestCase() {
+ OperatorTestBase::SetUpTestCase();
+ functions::sparksql::registerFunctions("");
+ }
+
+ // Build side: 'numBatches' x 'batchSize' rows with duplicate keys.
+ std::vector<RowVectorPtr> makeBuildVectors(bool nullKeys, int numBatches = 8, int batchSize = 500) {
+ std::vector<RowVectorPtr> vectors;
+ for (int batch = 0; batch < numBatches; ++batch) {
+ vectors.push_back(makeRowVector(
+ {"u_k", "u_v"},
+ {makeFlatVector<int32_t>(
+ batchSize,
+ [batch](auto row) { return (row * 7 + batch) % 997; },
+ [nullKeys, batch](auto row) { return nullKeys && batch == 3 && row == 17; }),
+ makeFlatVector<int64_t>(
+ batchSize, [](auto row) { return row % 101; }, [](auto row) { return row % 37 == 0; })}));
+ }
+ return vectors;
+ }
+
+ std::vector<RowVectorPtr> makeProbeVectors() {
+ std::vector<RowVectorPtr> vectors;
+ for (int batch = 0; batch < 4; ++batch) {
+ vectors.push_back(makeRowVector(
+ {"t_k", "t_v"},
+ {makeFlatVector<int32_t>(
+ 1'000, [batch](auto row) { return (row + batch * 1'000) % 1'500; }, [](auto row) { return row == 5; }),
+ makeFlatVector<int64_t>(1'000, [](auto row) { return row % 103; })}));
+ }
+ return vectors;
+ }
+
+ // The real join filter, see Filter.
+ static core::TypedExprPtr makeFilter(Filter filter) {
+ if (filter == Filter::kNone) {
+ return nullptr;
+ }
+ auto uv = std::make_shared<core::FieldAccessTypedExpr>(BIGINT(), "u_v");
+ core::TypedExprPtr greaterThan = std::make_shared<core::CallTypedExpr>(
+ BOOLEAN(),
+ std::vector<core::TypedExprPtr>{std::make_shared<core::FieldAccessTypedExpr>(BIGINT(), "t_v"), uv},
+ "greaterthan");
+ if (filter == Filter::kNullIntolerant) {
+ return greaterThan;
+ }
+ return std::make_shared<core::CallTypedExpr>(
+ BOOLEAN(),
+ std::vector<core::TypedExprPtr>{
+ std::make_shared<core::CallTypedExpr>(BOOLEAN(), std::vector<core::TypedExprPtr>{uv}, "isnull"),
+ greaterThan},
+ "or");
+ }
+
+ // Builds with the filter properties the Scala side derives from makeFilter().
+ std::shared_ptr<HashTableBuilder>
+ build(const JoinCase& c, const std::string& key, const std::vector<RowVectorPtr>& vectors) {
+ HashTableTaskBuildSpec spec;
+ spec.joinType = c.joinType;
+ spec.nullAware = c.nullAware;
+ spec.withFilter = c.filter != Filter::kNone;
+ if (spec.withFilter) {
+ spec.filterBuildColumns = {"u_v"};
+ }
+ spec.filterPropagatesNulls = c.filter == Filter::kNullIntolerant;
+ spec.buildType = ROW({"u_k", "u_v"}, {INTEGER(), BIGINT()});
+ spec.joinKeys = {"u_k"};
+ // Build in parallel whatever the table size, so that parallel cases merge per-driver tables.
+ spec.queryConfigs = {{core::QueryConfig::kMinTableRowsForParallelJoinBuild, "1"}};
+ return buildHashTableWithTask(key, spec, vectors, c.numDrivers, driverExecutor_.get(), rootPool_, pool());
+ }
+
+ // Joins the probe vectors with 'buildVectors'. With 'cacheKey', the probe
+ // uses the table cached under that key and 'buildVectors' must be empty.
+ RowVectorPtr join(
+ const JoinCase& c,
+ const std::vector<RowVectorPtr>& buildVectors,
+ const std::optional<std::string>& cacheKey = std::nullopt) {
+ auto probeNode = std::make_shared<core::ValuesNode>("0", makeProbeVectors());
+ auto buildNode = std::make_shared<core::ValuesNode>(
+ "1",
+ buildVectors.empty()
+ ? std::vector<RowVectorPtr>{makeRowVector(
+ {"u_k", "u_v"},
+ {makeFlatVector<int32_t>(std::vector<int32_t>{}), makeFlatVector<int64_t>(std::vector<int64_t>{})})}
+ : buildVectors);
+ RowTypePtr outputType;
+ switch (c.joinType) {
+ case core::JoinType::kInner:
+ case core::JoinType::kLeft:
+ case core::JoinType::kRight:
+ case core::JoinType::kFull:
+ outputType = ROW({"t_k", "t_v", "u_k", "u_v"}, {INTEGER(), BIGINT(), INTEGER(), BIGINT()});
+ break;
+ case core::JoinType::kLeftSemiProject:
+ outputType = ROW({"t_k", "t_v", "match"}, {INTEGER(), BIGINT(), BOOLEAN()});
+ break;
+ case core::JoinType::kRightSemiFilter:
+ outputType = ROW({"u_k", "u_v"}, {INTEGER(), BIGINT()});
+ break;
+ case core::JoinType::kRightSemiProject:
+ outputType = ROW({"u_k", "u_v", "match"}, {INTEGER(), BIGINT(), BOOLEAN()});
+ break;
+ default:
+ outputType = ROW({"t_k", "t_v"}, {INTEGER(), BIGINT()});
+ }
+ auto joinNode = core::HashJoinNode::Builder()
+ .id("2")
+ .joinType(c.joinType)
+ .leftKeys({std::make_shared<core::FieldAccessTypedExpr>(INTEGER(), "t_k")})
+ .rightKeys({std::make_shared<core::FieldAccessTypedExpr>(INTEGER(), "u_k")})
+ .filter(makeFilter(c.filter))
+ .left(probeNode)
+ .right(buildNode)
+ .outputType(outputType)
+ .nullAware(c.nullAware)
+ .useHashTableCache(cacheKey.has_value())
+ .cacheKey(cacheKey)
+ .build();
+ return exec::test::AssertQueryBuilder(joinNode).copyResults(pool());
+ }
+};
+
+// Probing the table built by the fake join gives the same result as a regular
+// Velox hash join over the same build rows.
+TEST_P(HashTableTaskBuilderTest, sameResultAsHashJoin) {
+ const auto& c = GetParam();
+ auto* cache = HashTableCache::instance();
+ const auto key = "task_" + c.name;
+ auto vectors = makeBuildVectors(c.nullBuildKeys);
+
+ auto hashTable = build(c, key, vectors);
+ ASSERT_NE(hashTable, nullptr);
+ ASSERT_TRUE(cache->exist(key));
+ EXPECT_GT(hashTable->hashTableMemoryUsage(), 0);
+ if (c.numDrivers > 1 && !hashTable->joinHasNullKeys()) {
+ // HashBuild built one table per driver and merged them.
+ EXPECT_EQ(hashTable->hashTable()->allRows().size(), c.numDrivers);
+ }
+
+ exec::test::assertEqualResults({join(c, vectors)}, {join(c, {}, key)});
+
+ // The cache entry, which owns the table's pool, can go first. The table
+ // must keep its memory pool alive in between.
+ cache->drop(key);
+ auto table = hashTable->hashTable();
+ hashTable.reset();
+ EXPECT_NE(table->rows(), nullptr);
+ table.reset();
+}
+
+// Releasing the last reference to the table drops the cache entry.
+TEST_P(HashTableTaskBuilderTest, releasingTableDropsCacheEntry) {
+ const auto& c = GetParam();
+ auto* cache = HashTableCache::instance();
+ const auto key = "release_" + c.name;
+ auto hashTable = build(c, key, makeBuildVectors(c.nullBuildKeys));
+ ASSERT_TRUE(cache->exist(key));
+ hashTable.reset();
+ EXPECT_FALSE(cache->exist(key));
+}
+
+// A probe task takes the table from the cache and can outlive the cache entry
+// and the returned builder, e.g. when the task fails while the broadcast table
+// is released. The table it holds must keep its memory pool alive.
+TEST_P(HashTableTaskBuilderTest, cachedTableOutlivesBuilder) {
+ const auto& c = GetParam();
+ auto* cache = HashTableCache::instance();
+ const auto key = "outlive_" + c.name;
+ auto hashTable = build(c, key, makeBuildVectors(c.nullBuildKeys));
+
+ ContinueFuture future = ContinueFuture::makeEmpty();
+ auto probeQueryCtx = core::QueryCtx::create();
+ auto entry = cache->get(key, "probe_task", probeQueryCtx.get(), &future);
+ ASSERT_FALSE(future.valid());
+ auto probeTable = entry->table;
+ std::weak_ptr<memory::MemoryPool> tablePool = entry->tablePool;
+ entry.reset();
+
+ hashTable.reset();
+ EXPECT_FALSE(cache->exist(key));
+ EXPECT_FALSE(tablePool.expired());
+ EXPECT_NE(probeTable->rows(), nullptr);
+ probeTable.reset();
+ EXPECT_TRUE(tablePool.expired());
+}
+
+// The driver-side build serializes the table and executors deserialize it.
+TEST_P(HashTableTaskBuilderTest, serializeRoundTrip) {
+ const auto& c = GetParam();
+ auto* cache = HashTableCache::instance();
+ const auto key = "serialize_" + c.name;
+ auto vectors = makeBuildVectors(c.nullBuildKeys);
+ auto hashTable = build(c, key, vectors);
+
+ std::vector<uint8_t> data(serializedHashTableSize(hashTable));
+ serializeHashTableTo(hashTable, data.data(), data.size());
+ // As deserializeHashTable(), with the test's pool.
+ std::shared_ptr<BaseHashTable> deserialized;
+ if (dynamic_cast<HashTable<true>*>(hashTable->hashTable().get()) != nullptr) {
+ deserialized = HashTableSerializer::deserialize<true>(data.data(), data.size(), pool());
+ } else {
+ deserialized = HashTableSerializer::deserialize<false>(data.data(), data.size(), pool());
+ }
+ const auto deserializedKey = "deserialized_" + c.name;
+ cache->add(deserializedKey, deserialized, hashTable->joinHasNullKeys(), pool_);
+
+ exec::test::assertEqualResults({join(c, vectors)}, {join(c, {}, deserializedKey)});
+ cache->drop(deserializedKey);
+}
+
+// For tests that do not take a JoinCase parameter.
+class HashTableTaskBuilderSingleTest : public HashTableTaskBuilderTest {};
+
+// A build without input rows still caches an empty table.
+TEST_F(HashTableTaskBuilderSingleTest, emptyBuildSide) {
+ const JoinCase c{"empty", core::JoinType::kInner};
+ auto* cache = HashTableCache::instance();
+ auto hashTable = build(c, "empty", {});
+ ASSERT_TRUE(cache->exist("empty"));
+ EXPECT_EQ(hashTable->hashTable()->numDistinct(), 0);
+ hashTable.reset();
+ EXPECT_FALSE(cache->exist("empty"));
+}
+
+INSTANTIATE_TEST_SUITE_P(
+ HashTableTaskBuilderTest,
+ HashTableTaskBuilderTest,
+ testing::Values(
+ JoinCase{"inner", core::JoinType::kInner},
+ JoinCase{"innerParallel", core::JoinType::kInner, false, Filter::kNone, false, 4},
+ JoinCase{"innerFilter", core::JoinType::kInner, false, Filter::kNullIntolerant},
+ JoinCase{"leftParallel", core::JoinType::kLeft, false, Filter::kNone, true, 4},
+ JoinCase{"leftFilter", core::JoinType::kLeft, false, Filter::kNullIntolerant, true},
+ JoinCase{"right", core::JoinType::kRight, false, Filter::kNone, true},
+ JoinCase{"rightFilterParallel", core::JoinType::kRight, false, Filter::kNullIntolerant, true, 4},
+ JoinCase{"full", core::JoinType::kFull, false, Filter::kNone, true},
+ JoinCase{"fullParallel", core::JoinType::kFull, false, Filter::kNone, true, 4},
+ JoinCase{"rightSemi", core::JoinType::kRightSemiFilter, false, Filter::kNone, true},
+ JoinCase{"rightSemiFilterParallel", core::JoinType::kRightSemiFilter, false, Filter::kNullIntolerant, true, 4},
+ JoinCase{"rightSemiProject", core::JoinType::kRightSemiProject, false, Filter::kNone, true},
+ JoinCase{"rightSemiProjectParallel", core::JoinType::kRightSemiProject, false, Filter::kNone, true, 4},
+ JoinCase{"leftSemi", core::JoinType::kLeftSemiFilter},
+ JoinCase{"leftSemiParallel", core::JoinType::kLeftSemiFilter, false, Filter::kNone, true, 4},
+ JoinCase{"leftSemiFilterParallel", core::JoinType::kLeftSemiFilter, false, Filter::kNullIntolerant, false, 4},
+ JoinCase{"existence", core::JoinType::kLeftSemiProject},
+ JoinCase{"existenceFilter", core::JoinType::kLeftSemiProject, false, Filter::kNullIntolerant, true},
+ JoinCase{"anti", core::JoinType::kAnti, false, Filter::kNone, true},
+ JoinCase{"antiParallel", core::JoinType::kAnti, false, Filter::kNone, true, 4},
+ JoinCase{"antiFilter", core::JoinType::kAnti, false, Filter::kNullIntolerant, true},
+ JoinCase{"antiFilterParallel", core::JoinType::kAnti, false, Filter::kNullIntolerant, true, 4},
+ JoinCase{"antiNullTolerantFilter", core::JoinType::kAnti, false, Filter::kNullTolerant, true},
+ JoinCase{"antiNullTolerantFilterParallel", core::JoinType::kAnti, false, Filter::kNullTolerant, true, 4},
+ JoinCase{"nullAwareAnti", core::JoinType::kAnti, true},
+ JoinCase{"nullAwareAntiParallel", core::JoinType::kAnti, true, Filter::kNone, false, 4},
+ JoinCase{"nullAwareAntiNullKey", core::JoinType::kAnti, true, Filter::kNone, true},
+ JoinCase{"nullAwareAntiNullKeyParallel", core::JoinType::kAnti, true, Filter::kNone, true, 4},
+ JoinCase{"nullAwareAntiFilterNullKey", core::JoinType::kAnti, true, Filter::kNullIntolerant, true}),
+ caseName);
+
+} // namespace
+} // namespace gluten
+
+// folly::init() registers folly singletons, e.g. the Timekeeper that
+// AssertQueryBuilder waits with. Without it Linux builds abort.
+int main(int argc, char** argv) {
+ testing::InitGoogleTest(&argc, argv);
+ folly::init(&argc, &argv, false);
+ return RUN_ALL_TESTS();
+}
Review Comment:
Defining `main()` inside an `add_velox_test(...)` unit test source is very likely to cause a link error (multiple definitions of `main`) because the test target typically links a shared gtest/Velox test main already. Instead, rely on the repo’s existing test main (and ensure *that* main calls `folly::init()`), or adjust the test target to not link the default main and provide a single dedicated test-main source for this binary.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN][VL][TEST] Use fake task to build hash table [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #13216:
URL: https://github.com/apache/gluten/pull/13216#discussion_r4268243252
##########
cpp/velox/operators/hashjoin/HashTableTaskBuilder.cc:
##########
@@ -0,0 +1,255 @@
+/*
+ * 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 <atomic>
+
+#include "velox/exec/HashTableCache.h"
+#include "velox/exec/RoundRobinPartitionFunction.h"
+#include "velox/exec/Task.h"
Review Comment:
`buildHashTableWithTask()` uses `fmt::format(...)` later in this file, but there is no direct include for the fmt format header in the added includes. If this isn’t guaranteed by transitive includes, it will fail to compile under stricter include hygiene. Add the needed fmt include (e.g., the project-standard fmt header) in this file to make the dependency explicit.
##########
cpp/velox/tests/HashTableTaskBuilderTest.cc:
##########
@@ -0,0 +1,283 @@
+/*
+ * 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 <folly/init/Init.h>
+#include <gtest/gtest.h>
+
+#include "jni/JniHashTable.h"
+#include "operators/hashjoin/HashTableSerializer.h"
+#include "operators/hashjoin/HashTableTaskBuilder.h"
+#include "velox/exec/HashTableCache.h"
+#include "velox/exec/tests/utils/AssertQueryBuilder.h"
+#include "velox/exec/tests/utils/OperatorTestBase.h"
+#include "velox/exec/tests/utils/QueryAssertions.h"
+#include "velox/functions/sparksql/registration/Register.h"
+
+using namespace facebook::velox;
+using namespace facebook::velox::exec;
+
+namespace gluten {
+namespace {
+
+enum class Filter {
+ kNone,
+ // t_v > u_v: null-intolerant, so an anti join drops build rows with a null u_v.
+ kNullIntolerant,
+ // u_v IS NULL OR t_v > u_v: an anti join keeps build rows with a null u_v.
+ kNullTolerant,
+};
+
+struct JoinCase {
+ std::string name;
+ core::JoinType joinType;
+ bool nullAware{false};
+ Filter filter{Filter::kNone};
+ bool nullBuildKeys{false};
+ uint32_t numDrivers{1};
+};
+
+std::string caseName(const testing::TestParamInfo<JoinCase>& info) {
+ return info.param.name;
+}
+
+class HashTableTaskBuilderTest : public exec::test::OperatorTestBase, public testing::WithParamInterface<JoinCase> {
+ protected:
+ static void SetUpTestCase() {
+ OperatorTestBase::SetUpTestCase();
+ functions::sparksql::registerFunctions("");
+ }
+
+ // Build side: 'numBatches' x 'batchSize' rows with duplicate keys.
+ std::vector<RowVectorPtr> makeBuildVectors(bool nullKeys, int numBatches = 8, int batchSize = 500) {
+ std::vector<RowVectorPtr> vectors;
+ for (int batch = 0; batch < numBatches; ++batch) {
+ vectors.push_back(makeRowVector(
+ {"u_k", "u_v"},
+ {makeFlatVector<int32_t>(
+ batchSize,
+ [batch](auto row) { return (row * 7 + batch) % 997; },
+ [nullKeys, batch](auto row) { return nullKeys && batch == 3 && row == 17; }),
+ makeFlatVector<int64_t>(
+ batchSize, [](auto row) { return row % 101; }, [](auto row) { return row % 37 == 0; })}));
+ }
+ return vectors;
+ }
+
+ std::vector<RowVectorPtr> makeProbeVectors() {
+ std::vector<RowVectorPtr> vectors;
+ for (int batch = 0; batch < 4; ++batch) {
+ vectors.push_back(makeRowVector(
+ {"t_k", "t_v"},
+ {makeFlatVector<int32_t>(
+ 1'000, [batch](auto row) { return (row + batch * 1'000) % 1'500; }, [](auto row) { return row == 5; }),
+ makeFlatVector<int64_t>(1'000, [](auto row) { return row % 103; })}));
+ }
+ return vectors;
+ }
+
+ // The real join filter, see Filter.
+ static core::TypedExprPtr makeFilter(Filter filter) {
+ if (filter == Filter::kNone) {
+ return nullptr;
+ }
+ auto uv = std::make_shared<core::FieldAccessTypedExpr>(BIGINT(), "u_v");
+ core::TypedExprPtr greaterThan = std::make_shared<core::CallTypedExpr>(
+ BOOLEAN(),
+ std::vector<core::TypedExprPtr>{std::make_shared<core::FieldAccessTypedExpr>(BIGINT(), "t_v"), uv},
+ "greaterthan");
+ if (filter == Filter::kNullIntolerant) {
+ return greaterThan;
+ }
+ return std::make_shared<core::CallTypedExpr>(
+ BOOLEAN(),
+ std::vector<core::TypedExprPtr>{
+ std::make_shared<core::CallTypedExpr>(BOOLEAN(), std::vector<core::TypedExprPtr>{uv}, "isnull"),
+ greaterThan},
+ "or");
+ }
+
+ // Builds with the filter properties the Scala side derives from makeFilter().
+ std::shared_ptr<HashTableBuilder>
+ build(const JoinCase& c, const std::string& key, const std::vector<RowVectorPtr>& vectors) {
+ HashTableTaskBuildSpec spec;
+ spec.joinType = c.joinType;
+ spec.nullAware = c.nullAware;
+ spec.withFilter = c.filter != Filter::kNone;
+ if (spec.withFilter) {
+ spec.filterBuildColumns = {"u_v"};
+ }
+ spec.filterPropagatesNulls = c.filter == Filter::kNullIntolerant;
+ spec.buildType = ROW({"u_k", "u_v"}, {INTEGER(), BIGINT()});
+ spec.joinKeys = {"u_k"};
+ // Build in parallel whatever the table size, so that parallel cases merge per-driver tables.
+ spec.queryConfigs = {{core::QueryConfig::kMinTableRowsForParallelJoinBuild, "1"}};
+ return buildHashTableWithTask(key, spec, vectors, c.numDrivers, driverExecutor_.get(), rootPool_, pool());
+ }
+
+ // Joins the probe vectors with 'buildVectors'. With 'cacheKey', the probe
+ // uses the table cached under that key and 'buildVectors' must be empty.
+ RowVectorPtr join(
+ const JoinCase& c,
+ const std::vector<RowVectorPtr>& buildVectors,
+ const std::optional<std::string>& cacheKey = std::nullopt) {
+ auto probeNode = std::make_shared<core::ValuesNode>("0", makeProbeVectors());
+ auto buildNode = std::make_shared<core::ValuesNode>(
+ "1",
+ buildVectors.empty()
+ ? std::vector<RowVectorPtr>{makeRowVector(
+ {"u_k", "u_v"},
+ {makeFlatVector<int32_t>(std::vector<int32_t>{}), makeFlatVector<int64_t>(std::vector<int64_t>{})})}
+ : buildVectors);
+ RowTypePtr outputType;
+ if (c.joinType == core::JoinType::kInner || c.joinType == core::JoinType::kLeft) {
+ outputType = ROW({"t_k", "t_v", "u_k", "u_v"}, {INTEGER(), BIGINT(), INTEGER(), BIGINT()});
+ } else if (c.joinType == core::JoinType::kLeftSemiProject) {
+ outputType = ROW({"t_k", "t_v", "match"}, {INTEGER(), BIGINT(), BOOLEAN()});
+ } else {
+ outputType = ROW({"t_k", "t_v"}, {INTEGER(), BIGINT()});
+ }
+ auto joinNode = core::HashJoinNode::Builder()
+ .id("2")
+ .joinType(c.joinType)
+ .leftKeys({std::make_shared<core::FieldAccessTypedExpr>(INTEGER(), "t_k")})
+ .rightKeys({std::make_shared<core::FieldAccessTypedExpr>(INTEGER(), "u_k")})
+ .filter(makeFilter(c.filter))
+ .left(probeNode)
+ .right(buildNode)
+ .outputType(outputType)
+ .nullAware(c.nullAware)
+ .useHashTableCache(cacheKey.has_value())
+ .cacheKey(cacheKey)
+ .build();
+ return exec::test::AssertQueryBuilder(joinNode).copyResults(pool());
+ }
+};
+
+// Probing the table built by the fake join gives the same result as a regular
+// Velox hash join over the same build rows.
+TEST_P(HashTableTaskBuilderTest, sameResultAsHashJoin) {
+ const auto& c = GetParam();
+ auto* cache = HashTableCache::instance();
+ const auto key = "task_" + c.name;
+ auto vectors = makeBuildVectors(c.nullBuildKeys);
+
+ auto hashTable = build(c, key, vectors);
+ ASSERT_NE(hashTable, nullptr);
+ ASSERT_TRUE(cache->exist(key));
+ EXPECT_GT(hashTable->hashTableMemoryUsage(), 0);
+ if (c.numDrivers > 1 && !hashTable->joinHasNullKeys()) {
+ // HashBuild built one table per driver and merged them.
+ EXPECT_EQ(hashTable->hashTable()->allRows().size(), c.numDrivers);
+ }
+
+ exec::test::assertEqualResults({join(c, vectors)}, {join(c, {}, key)});
+
+ // The cache entry, which owns the table's pool, can go first. The table
+ // must keep its memory pool alive in between.
+ cache->drop(key);
+ auto table = hashTable->hashTable();
+ hashTable.reset();
+ EXPECT_NE(table->rows(), nullptr);
+ table.reset();
+}
+
+// Releasing the last reference to the table drops the cache entry.
+TEST_P(HashTableTaskBuilderTest, releasingTableDropsCacheEntry) {
+ const auto& c = GetParam();
+ auto* cache = HashTableCache::instance();
+ const auto key = "release_" + c.name;
+ auto hashTable = build(c, key, makeBuildVectors(c.nullBuildKeys));
+ ASSERT_TRUE(cache->exist(key));
+ hashTable.reset();
+ EXPECT_FALSE(cache->exist(key));
+}
+
+// The driver-side build serializes the table and executors deserialize it.
+TEST_P(HashTableTaskBuilderTest, serializeRoundTrip) {
+ const auto& c = GetParam();
+ auto* cache = HashTableCache::instance();
+ const auto key = "serialize_" + c.name;
+ auto vectors = makeBuildVectors(c.nullBuildKeys);
+ auto hashTable = build(c, key, vectors);
+
+ std::vector<uint8_t> data(serializedHashTableSize(hashTable));
+ serializeHashTableTo(hashTable, data.data(), data.size());
+ // As deserializeHashTable(), with the test's pool.
+ std::shared_ptr<BaseHashTable> deserialized;
+ if (dynamic_cast<HashTable<true>*>(hashTable->hashTable().get()) != nullptr) {
+ deserialized = HashTableSerializer::deserialize<true>(data.data(), data.size(), pool());
+ } else {
+ deserialized = HashTableSerializer::deserialize<false>(data.data(), data.size(), pool());
+ }
+ const auto deserializedKey = "deserialized_" + c.name;
+ cache->add(deserializedKey, deserialized, hashTable->joinHasNullKeys(), pool_);
+
+ exec::test::assertEqualResults({join(c, vectors)}, {join(c, {}, deserializedKey)});
+ cache->drop(deserializedKey);
+}
+
+// For tests that do not take a JoinCase parameter.
+class HashTableTaskBuilderSingleTest : public HashTableTaskBuilderTest {};
+
+// A build without input rows still caches an empty table.
+TEST_F(HashTableTaskBuilderSingleTest, emptyBuildSide) {
+ const JoinCase c{"empty", core::JoinType::kInner};
+ auto* cache = HashTableCache::instance();
+ auto hashTable = build(c, "empty", {});
+ ASSERT_TRUE(cache->exist("empty"));
+ EXPECT_EQ(hashTable->hashTable()->numDistinct(), 0);
+ hashTable.reset();
+ EXPECT_FALSE(cache->exist("empty"));
+}
+
+INSTANTIATE_TEST_SUITE_P(
+ HashTableTaskBuilderTest,
+ HashTableTaskBuilderTest,
+ testing::Values(
+ JoinCase{"inner", core::JoinType::kInner},
+ JoinCase{"innerParallel", core::JoinType::kInner, false, Filter::kNone, false, 4},
+ JoinCase{"innerFilter", core::JoinType::kInner, false, Filter::kNullIntolerant},
+ JoinCase{"leftParallel", core::JoinType::kLeft, false, Filter::kNone, true, 4},
+ JoinCase{"leftFilter", core::JoinType::kLeft, false, Filter::kNullIntolerant, true},
+ JoinCase{"leftSemi", core::JoinType::kLeftSemiFilter},
+ JoinCase{"leftSemiParallel", core::JoinType::kLeftSemiFilter, false, Filter::kNone, true, 4},
+ JoinCase{"leftSemiFilterParallel", core::JoinType::kLeftSemiFilter, false, Filter::kNullIntolerant, false, 4},
+ JoinCase{"existence", core::JoinType::kLeftSemiProject},
+ JoinCase{"existenceFilter", core::JoinType::kLeftSemiProject, false, Filter::kNullIntolerant, true},
+ JoinCase{"anti", core::JoinType::kAnti, false, Filter::kNone, true},
+ JoinCase{"antiParallel", core::JoinType::kAnti, false, Filter::kNone, true, 4},
+ JoinCase{"antiFilter", core::JoinType::kAnti, false, Filter::kNullIntolerant, true},
+ JoinCase{"antiFilterParallel", core::JoinType::kAnti, false, Filter::kNullIntolerant, true, 4},
+ JoinCase{"antiNullTolerantFilter", core::JoinType::kAnti, false, Filter::kNullTolerant, true},
+ JoinCase{"antiNullTolerantFilterParallel", core::JoinType::kAnti, false, Filter::kNullTolerant, true, 4},
+ JoinCase{"nullAwareAnti", core::JoinType::kAnti, true},
+ JoinCase{"nullAwareAntiParallel", core::JoinType::kAnti, true, Filter::kNone, false, 4},
+ JoinCase{"nullAwareAntiNullKey", core::JoinType::kAnti, true, Filter::kNone, true},
+ JoinCase{"nullAwareAntiNullKeyParallel", core::JoinType::kAnti, true, Filter::kNone, true, 4},
+ JoinCase{"nullAwareAntiFilterNullKey", core::JoinType::kAnti, true, Filter::kNullIntolerant, true}),
+ caseName);
+
+} // namespace
+} // namespace gluten
+
+// folly::init() registers folly singletons, e.g. the Timekeeper that
+// AssertQueryBuilder waits with. Without it Linux builds abort.
+int main(int argc, char** argv) {
+ testing::InitGoogleTest(&argc, argv);
+ folly::init(&argc, &argv, false);
+ return RUN_ALL_TESTS();
+}
Review Comment:
Defining `main()` inside a test source is very likely to cause duplicate-symbol link failures if `add_velox_test(...)` already links a common gtest main (which is the typical setup when other tests in the same directory don’t define `main`). To avoid breaking the build, remove this `main()` and instead ensure `folly::init()` is invoked from the shared test main used by `add_velox_test` (preferred), or from the test framework initialization path that’s guaranteed to run once per test binary.
##########
cpp/velox/operators/hashjoin/HashTableBuilder.cc:
##########
@@ -25,6 +25,40 @@
namespace gluten {
+std::shared_ptr<HashTableBuilder> HashTableBuilder::wrap(
+ std::shared_ptr<facebook::velox::exec::BaseHashTable> table,
+ bool joinHasNullKeys,
+ facebook::velox::memory::MemoryPool* pool,
+ std::shared_ptr<void> owner) {
+ auto keyTypes = table->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));
+
+ const std::vector<facebook::velox::core::FieldAccessTypedExprPtr> noKeys;
+ const std::vector<column_index_t> noChannels;
+ auto builder = std::make_shared<HashTableBuilder>(
+ facebook::velox::core::JoinType::kInner,
+ false,
+ false,
+ -1,
+ noKeys,
+ noChannels,
+ false,
+ rowType,
+ pool,
+ 1000,
+ 1000000,
+ 100000,
+ 0);
Review Comment:
`HashTableBuilder::wrap()` currently constructs a `HashTableBuilder` using several hard-coded tuning parameters and a `-1` bloom filter size sentinel. Even if these values are ‘unused’ for wrapped tables, they create hidden coupling to constructor behavior and make future refactors riskier (e.g., if the constructor starts validating ranges). Consider introducing a dedicated/private constructor or a small internal factory specifically for wrapped tables that sets clearly safe defaults (e.g., `0` instead of `-1`) and/or documents which fields are meaningful in the wrapped-table scenario.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org