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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
