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: [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]
