Optimize skewed intersect lists

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

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} $<TARGET_OBJECTS:lbug_processor_operator_intersect>

src/processor/operator/intersect/intersect.cpp

@@ -3,6 +3,7 @@
 #include <algorithm>
 
 #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);
 }

src/processor/operator/intersect/intersect_kernels.cpp

@@ -0,0 +1,106 @@
+#include "processor/operator/intersect/intersect_kernels.h"
+
+#include <algorithm>
+
+#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

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)

test/processor/CMakeLists.txt

@@ -0,0 +1 @@
+add_lbug_test(processor_test intersect_kernels_test.cpp)

test/processor/intersect_kernels_test.cpp

@@ -0,0 +1,90 @@
+#include <algorithm>
+#include <cstdint>
+#include <random>
+#include <vector>
+
+#include "common/types/types.h"
+#include "gtest/gtest.h"
+#include "processor/operator/intersect/intersect_kernels.h"
+
+using namespace lbug;
+
+namespace {
+
+void verifyMatchesScalar(std::vector<common::nodeID_t> left,
+    const std::vector<common::nodeID_t>& right) {
+    auto scalarLeft = left;
+    std::vector<common::sel_t> scalarLeftPositions(left.size());
+    std::vector<common::sel_t> scalarRightPositions(left.size());
+    std::vector<common::sel_t> fastLeftPositions(left.size());
+    std::vector<common::sel_t> 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<common::nodeID_t> makeSortedIDs(uint64_t count, uint64_t domain,
+    common::table_id_t tableID, uint64_t seed) {
+    std::mt19937_64 random{seed};
+    std::vector<common::nodeID_t> 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<uint64_t>(rightCount * 4, 1);
+            verifyMatchesScalar(makeSortedIDs(leftCount, domain, 7, seed),
+                makeSortedIDs(rightCount, domain, 7, seed + 1000));
+        }
+    }
+}
+
+TEST(IntersectKernelsTest, PreservesDuplicatePairing) {
+    std::vector<common::nodeID_t> left = {{1, 7}, {1, 7}, {2, 7}, {8, 7}, {8, 7}, {8, 7}, {64, 7},
+        {128, 7}};
+    std::vector<common::nodeID_t> 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<common::nodeID_t> left = {{0, 1}, {1, 1}, {0, 2}, {4, 2}};
+    std::vector<common::nodeID_t> 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<common::nodeID_t> left;
+    std::vector<common::nodeID_t> right = {{1, 7}, {2, 7}};
+    verifyMatchesScalar(left, right);
+}