From 45895f6d53c10c735cb127a32749216a36f5d63c Mon Sep 17 00:00:00 2001 From: rahul-iyer Date: Thu, 20 Aug 2026 18:36:50 -0700 Subject: [PATCH 1/2] Optimize skewed intersect lists --- .../operator/intersect/intersect_kernels.h | 22 ++++ .../operator/intersect/CMakeLists.txt | 3 +- .../operator/intersect/intersect.cpp | 22 +--- .../operator/intersect/intersect_kernels.cpp | 106 ++++++++++++++++++ test/CMakeLists.txt | 1 + test/processor/CMakeLists.txt | 1 + test/processor/intersect_kernels_test.cpp | 90 +++++++++++++++ 7 files changed, 226 insertions(+), 19 deletions(-) create mode 100644 src/include/processor/operator/intersect/intersect_kernels.h create mode 100644 src/processor/operator/intersect/intersect_kernels.cpp create mode 100644 test/processor/CMakeLists.txt create mode 100644 test/processor/intersect_kernels_test.cpp diff --git a/src/include/processor/operator/intersect/intersect_kernels.h b/src/include/processor/operator/intersect/intersect_kernels.h new file mode 100644 index 0000000000..0b5bab0b71 --- /dev/null +++ b/src/include/processor/operator/intersect/intersect_kernels.h @@ -0,0 +1,22 @@ +#pragma once + +#include "common/api.h" +#include "common/types/types.h" + +namespace lbug { +namespace processor { + +// Reference merge used for mixed-table lists, balanced lists, and differential tests. +LBUG_API common::sel_t intersectNodeIDsScalar(common::nodeID_t* left, common::sel_t leftCount, + const common::nodeID_t* right, common::sel_t rightCount, common::sel_t* leftPositions, + common::sel_t* rightPositions); + +// Selects the skew-aware same-table fast path when it is profitable and otherwise uses the +// reference merge. Both inputs must be sorted by nodeID_t's lexicographic ordering, and leftCount +// must be less than or equal to rightCount. +LBUG_API common::sel_t intersectNodeIDs(common::nodeID_t* left, common::sel_t leftCount, + const common::nodeID_t* right, common::sel_t rightCount, common::sel_t* leftPositions, + common::sel_t* rightPositions); + +} // namespace processor +} // namespace lbug diff --git a/src/processor/operator/intersect/CMakeLists.txt b/src/processor/operator/intersect/CMakeLists.txt index 8df0db7892..64b12e7fa7 100644 --- a/src/processor/operator/intersect/CMakeLists.txt +++ b/src/processor/operator/intersect/CMakeLists.txt @@ -1,6 +1,7 @@ add_library(lbug_processor_operator_intersect OBJECT - intersect.cpp) + intersect.cpp + intersect_kernels.cpp) set(ALL_OBJECT_FILES ${ALL_OBJECT_FILES} $ diff --git a/src/processor/operator/intersect/intersect.cpp b/src/processor/operator/intersect/intersect.cpp index f711d73fe4..b5d1915d88 100644 --- a/src/processor/operator/intersect/intersect.cpp +++ b/src/processor/operator/intersect/intersect.cpp @@ -3,6 +3,7 @@ #include #include "function/hash/hash_functions.h" +#include "processor/operator/intersect/intersect_kernels.h" #include "processor/result/factorized_table.h" using namespace lbug::common; @@ -67,24 +68,9 @@ void Intersect::twoWayIntersect(nodeID_t* leftNodeIDs, SelectionVector& lSelVect DASSERT(lSelVector.getSelSize() <= rSelVector.getSelSize()); auto leftPositionBuffer = lSelVector.getMutableBuffer(); auto rightPositionBuffer = rSelVector.getMutableBuffer(); - sel_t leftPosition = 0, rightPosition = 0; - uint64_t outputValuePosition = 0; - while (leftPosition < lSelVector.getSelSize() && rightPosition < rSelVector.getSelSize()) { - auto leftNodeID = leftNodeIDs[leftPosition]; - auto rightNodeID = rightNodeIDs[rightPosition]; - if (leftNodeID < rightNodeID) { - leftPosition++; - } else if (leftNodeID > rightNodeID) { - rightPosition++; - } else { - leftPositionBuffer[outputValuePosition] = leftPosition; - rightPositionBuffer[outputValuePosition] = rightPosition; - leftNodeIDs[outputValuePosition] = leftNodeID; - leftPosition++; - rightPosition++; - outputValuePosition++; - } - } + const auto outputValuePosition = + intersectNodeIDs(leftNodeIDs, lSelVector.getSelSize(), rightNodeIDs, + rSelVector.getSelSize(), leftPositionBuffer.data(), rightPositionBuffer.data()); lSelVector.setToFiltered(outputValuePosition); rSelVector.setToFiltered(outputValuePosition); } diff --git a/src/processor/operator/intersect/intersect_kernels.cpp b/src/processor/operator/intersect/intersect_kernels.cpp new file mode 100644 index 0000000000..e552e85c18 --- /dev/null +++ b/src/processor/operator/intersect/intersect_kernels.cpp @@ -0,0 +1,106 @@ +#include "processor/operator/intersect/intersect_kernels.h" + +#include + +#include "common/assert.h" + +namespace lbug { +namespace processor { + +namespace { + +constexpr common::sel_t MIN_GALLOPING_RIGHT_COUNT = 64; +constexpr common::sel_t MIN_GALLOPING_SIZE_RATIO = 8; + +bool isHomogeneous(const common::nodeID_t* values, common::sel_t count) { + return count != 0 && values[0].tableID == values[count - 1].tableID; +} + +common::sel_t lowerBoundOffset(const common::nodeID_t* values, common::sel_t begin, + common::sel_t end, common::offset_t target) { + while (begin < end) { + const auto middle = begin + (end - begin) / 2; + if (values[middle].offset < target) { + begin = middle + 1; + } else { + end = middle; + } + } + return begin; +} + +common::sel_t intersectSameTableGalloping(common::nodeID_t* left, common::sel_t leftCount, + const common::nodeID_t* right, common::sel_t rightCount, common::sel_t* leftPositions, + common::sel_t* rightPositions) { + common::sel_t leftPosition = 0; + common::sel_t rightPosition = 0; + common::sel_t outputPosition = 0; + while (leftPosition < leftCount && rightPosition < rightCount) { + const auto leftNodeID = left[leftPosition]; + const auto target = leftNodeID.offset; + if (right[rightPosition].offset < target) { + common::sel_t step = 1; + const auto remaining = rightCount - rightPosition; + while (step < remaining && right[rightPosition + step].offset < target) { + step = step > remaining - step ? remaining : step + step; + } + const auto begin = std::min(rightPosition + (step >> 1) + 1, rightCount); + const auto end = std::min(rightPosition + step + 1, rightCount); + rightPosition = lowerBoundOffset(right, begin, end, target); + if (rightPosition == rightCount) { + break; + } + } + if (right[rightPosition].offset == target) { + leftPositions[outputPosition] = leftPosition; + rightPositions[outputPosition] = rightPosition; + left[outputPosition++] = leftNodeID; + ++rightPosition; + } + ++leftPosition; + } + return outputPosition; +} + +} // namespace + +common::sel_t intersectNodeIDsScalar(common::nodeID_t* left, common::sel_t leftCount, + const common::nodeID_t* right, common::sel_t rightCount, common::sel_t* leftPositions, + common::sel_t* rightPositions) { + common::sel_t leftPosition = 0; + common::sel_t rightPosition = 0; + common::sel_t outputPosition = 0; + while (leftPosition < leftCount && rightPosition < rightCount) { + const auto leftNodeID = left[leftPosition]; + const auto rightNodeID = right[rightPosition]; + if (leftNodeID < rightNodeID) { + ++leftPosition; + } else if (leftNodeID > rightNodeID) { + ++rightPosition; + } else { + leftPositions[outputPosition] = leftPosition; + rightPositions[outputPosition] = rightPosition; + left[outputPosition++] = leftNodeID; + ++leftPosition; + ++rightPosition; + } + } + return outputPosition; +} + +common::sel_t intersectNodeIDs(common::nodeID_t* left, common::sel_t leftCount, + const common::nodeID_t* right, common::sel_t rightCount, common::sel_t* leftPositions, + common::sel_t* rightPositions) { + DASSERT(leftCount <= rightCount); + if (leftCount == 0 || rightCount < MIN_GALLOPING_RIGHT_COUNT || + rightCount / leftCount < MIN_GALLOPING_SIZE_RATIO || !isHomogeneous(left, leftCount) || + !isHomogeneous(right, rightCount) || left[0].tableID != right[0].tableID) { + return intersectNodeIDsScalar(left, leftCount, right, rightCount, leftPositions, + rightPositions); + } + return intersectSameTableGalloping(left, leftCount, right, rightCount, leftPositions, + rightPositions); +} + +} // namespace processor +} // namespace lbug diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 37163d5cf9..3ec1b8cf91 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -10,6 +10,7 @@ add_subdirectory(common) add_subdirectory(graph_test) add_subdirectory(optimizer) add_subdirectory(planner) +add_subdirectory(processor) add_subdirectory(runner) add_subdirectory(storage) add_subdirectory(test_helper) diff --git a/test/processor/CMakeLists.txt b/test/processor/CMakeLists.txt new file mode 100644 index 0000000000..57a6659b2d --- /dev/null +++ b/test/processor/CMakeLists.txt @@ -0,0 +1 @@ +add_lbug_test(processor_test intersect_kernels_test.cpp) diff --git a/test/processor/intersect_kernels_test.cpp b/test/processor/intersect_kernels_test.cpp new file mode 100644 index 0000000000..d7005c15fd --- /dev/null +++ b/test/processor/intersect_kernels_test.cpp @@ -0,0 +1,90 @@ +#include +#include +#include +#include + +#include "common/types/types.h" +#include "gtest/gtest.h" +#include "processor/operator/intersect/intersect_kernels.h" + +using namespace lbug; + +namespace { + +void verifyMatchesScalar(std::vector left, + const std::vector& right) { + auto scalarLeft = left; + std::vector scalarLeftPositions(left.size()); + std::vector scalarRightPositions(left.size()); + std::vector fastLeftPositions(left.size()); + std::vector fastRightPositions(left.size()); + + const auto scalarCount = processor::intersectNodeIDsScalar(scalarLeft.data(), scalarLeft.size(), + right.data(), right.size(), scalarLeftPositions.data(), scalarRightPositions.data()); + const auto fastCount = processor::intersectNodeIDs(left.data(), left.size(), right.data(), + right.size(), fastLeftPositions.data(), fastRightPositions.data()); + + ASSERT_EQ(fastCount, scalarCount); + EXPECT_TRUE(std::equal(left.begin(), left.begin() + fastCount, scalarLeft.begin())); + EXPECT_TRUE(std::equal(fastLeftPositions.begin(), fastLeftPositions.begin() + fastCount, + scalarLeftPositions.begin())); + EXPECT_TRUE(std::equal(fastRightPositions.begin(), fastRightPositions.begin() + fastCount, + scalarRightPositions.begin())); +} + +std::vector makeSortedIDs(uint64_t count, uint64_t domain, + common::table_id_t tableID, uint64_t seed) { + std::mt19937_64 random{seed}; + std::vector result; + result.reserve(count); + for (auto i = 0u; i < count; ++i) { + result.emplace_back(random() % domain, tableID); + } + std::sort(result.begin(), result.end()); + return result; +} + +} // namespace + +TEST(IntersectKernelsTest, MatchesScalarAcrossSizesAndSkews) { + for (const auto [leftCount, rightCount] : + {std::pair{1u, 64u}, std::pair{8u, 8u}, std::pair{16u, 128u}, std::pair{32u, 2048u}, + std::pair{128u, 128u}, std::pair{256u, 2048u}}) { + for (auto seed = 0u; seed < 20; ++seed) { + const auto domain = std::max(rightCount * 4, 1); + verifyMatchesScalar(makeSortedIDs(leftCount, domain, 7, seed), + makeSortedIDs(rightCount, domain, 7, seed + 1000)); + } + } +} + +TEST(IntersectKernelsTest, PreservesDuplicatePairing) { + std::vector left = {{1, 7}, {1, 7}, {2, 7}, {8, 7}, {8, 7}, {8, 7}, {64, 7}, + {128, 7}}; + std::vector right(128, common::nodeID_t{3, 7}); + right[0] = {1, 7}; + right[1] = {1, 7}; + right[2] = {1, 7}; + right[60] = {8, 7}; + right[61] = {8, 7}; + right[126] = {128, 7}; + right[127] = {128, 7}; + std::sort(right.begin(), right.end()); + verifyMatchesScalar(left, right); +} + +TEST(IntersectKernelsTest, FallsBackForMixedTableIDs) { + std::vector left = {{0, 1}, {1, 1}, {0, 2}, {4, 2}}; + std::vector right; + for (auto i = 0u; i < 128; ++i) { + right.emplace_back(i, i < 64 ? 1 : 2); + } + std::sort(right.begin(), right.end()); + verifyMatchesScalar(left, right); +} + +TEST(IntersectKernelsTest, HandlesEmptyLeftInput) { + std::vector left; + std::vector right = {{1, 7}, {2, 7}}; + verifyMatchesScalar(left, right); +} From fcff7ae8b9fa9698628c8ca08efd5147855fca2e Mon Sep 17 00:00:00 2001 From: rahul-iyer Date: Thu, 20 Aug 2026 21:12:52 -0700 Subject: [PATCH 2/2] Fix GCC warning in intersect test --- test/processor/intersect_kernels_test.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/processor/intersect_kernels_test.cpp b/test/processor/intersect_kernels_test.cpp index d7005c15fd..b7df08f3e3 100644 --- a/test/processor/intersect_kernels_test.cpp +++ b/test/processor/intersect_kernels_test.cpp @@ -47,7 +47,7 @@ std::vector makeSortedIDs(uint64_t count, uint64_t domain, } // namespace TEST(IntersectKernelsTest, MatchesScalarAcrossSizesAndSkews) { - for (const auto [leftCount, rightCount] : + for (const auto& [leftCount, rightCount] : {std::pair{1u, 64u}, std::pair{8u, 8u}, std::pair{16u, 128u}, std::pair{32u, 2048u}, std::pair{128u, 128u}, std::pair{256u, 2048u}}) { for (auto seed = 0u; seed < 20; ++seed) {