clean

kuzudb · May 2, 2023 · 2a7b455 · 2a7b455
1 parent be1faa8
commit 2a7b455
Show file tree

Hide file tree

Showing 11 changed files with 41 additions and 45 deletions.
diff --git a/src/include/processor/operator/recursive_extend/bfs_state.h b/src/include/processor/operator/recursive_extend/bfs_state.h
@@ -81,7 +81,7 @@ struct BaseBFSMorsel {
 };
 
 struct ShortestPathBFSMorsel : public BaseBFSMorsel {
-    // results
+    // Results
     std::vector<common::offset_t> dstNodeOffsets;
     std::unordered_map<common::offset_t, uint64_t> dstNodeOffset2PathLength;
 
@@ -110,14 +110,13 @@ struct ShortestPathBFSMorsel : public BaseBFSMorsel {
 };
 
 struct VariableLengthBFSMorsel : public BaseBFSMorsel {
-    // results
+    // Results
     std::vector<common::offset_t> dstNodeOffsets;
     std::unordered_map<common::offset_t, uint64_t> dstNodeOffset2NumPath;
 
     explicit VariableLengthBFSMorsel(common::offset_t maxOffset, uint8_t lowerBound,
         uint8_t upperBound, NodeOffsetSemiMask* semiMask)
-        : BaseBFSMorsel{maxOffset, lowerBound, upperBound, semiMask} {
-    }
+        : BaseBFSMorsel{maxOffset, lowerBound, upperBound, semiMask} {}
 
     inline bool isComplete() override { return isCurrentFrontierEmpty() || isUpperBoundReached(); }
     inline void resetState() override {

diff --git a/src/include/processor/operator/recursive_extend/recursive_join.h b/src/include/processor/operator/recursive_extend/recursive_join.h
@@ -122,7 +122,7 @@ class BaseRecursiveJoin : public PhysicalOperator {
     std::shared_ptr<common::ValueVector> tmpDstNodeIDVector; // temporary recursive join result.
 
     std::unique_ptr<BaseBFSMorsel> bfsMorsel;
-    common::offset_t outputCursor;
+    size_t outputCursor;
 };
 
 } // namespace processor

diff --git a/src/include/processor/operator/recursive_extend/shortest_path_recursive_join.h b/src/include/processor/operator/recursive_extend/shortest_path_recursive_join.h
@@ -55,4 +55,4 @@ class ShortestPathRecursiveJoin : public BaseRecursiveJoin {
 };
 
 } // namespace processor
-} // namespace kuzu
+} // namespace kuzu
diff --git a/src/include/processor/operator/recursive_extend/variable_length_recursive_join.h b/src/include/processor/operator/recursive_extend/variable_length_recursive_join.h
@@ -43,10 +43,7 @@ class VariableLengthRecursiveJoin : public BaseRecursiveJoin {
     void computeBFS(kuzu::processor::ExecutionContext* context) override;
 
     void updateVisitedNode(VariableLengthBFSMorsel* morsel, uint64_t multiplicity);
-
-private:
-    uint64_t numScanned = 0;
 };
 
 } // namespace processor
-} // namespace kuzu
+} // namespace kuzu
diff --git a/src/processor/mapper/map_extend.cpp b/src/processor/mapper/map_extend.cpp
@@ -161,10 +161,11 @@ std::unique_ptr<PhysicalOperator> PlanMapper::mapLogicalRecursiveExtendToPhysica
             DataPos{outSchema->getExpressionPos(*rel->getInternalLengthProperty())};
         return std::make_unique<ShortestPathRecursiveJoin>(rel->getLowerBound(),
             rel->getUpperBound(), nodeTable, sharedState, outDataPoses, colIndicesToScan,
-            inNodeIDVectorPos, outNodeIDVectorPos, distanceVectorPos, tmpDstNodePos, std::move(resultCollector),
-            getOperatorID(), extend->getExpressionsForPrinting(), std::move(scanRelTable));
+            inNodeIDVectorPos, outNodeIDVectorPos, distanceVectorPos, tmpDstNodePos,
+            std::move(resultCollector), getOperatorID(), extend->getExpressionsForPrinting(),
+            std::move(scanRelTable));
     }
-    case common::QueryRelType::VARIABLE_LENGTH : {
+    case common::QueryRelType::VARIABLE_LENGTH: {
         return std::make_unique<VariableLengthRecursiveJoin>(rel->getLowerBound(),
             rel->getUpperBound(), nodeTable, sharedState, outDataPoses, colIndicesToScan,
             inNodeIDVectorPos, outNodeIDVectorPos, tmpDstNodePos, std::move(resultCollector),

diff --git a/src/processor/mapper/map_semi_masker.cpp b/src/processor/mapper/map_semi_masker.cpp
@@ -1,8 +1,8 @@
 #include "planner/logical_plan/logical_operator/logical_semi_masker.h"
 #include "processor/mapper/plan_mapper.h"
+#include "processor/operator/recursive_extend/recursive_join.h"
 #include "processor/operator/scan_node_id.h"
 #include "processor/operator/semi_masker.h"
-#include "processor/operator/recursive_extend/recursive_join.h"
 
 using namespace kuzu::planner;
 

diff --git a/src/processor/operator/recursive_extend/recursive_join.cpp b/src/processor/operator/recursive_extend/recursive_join.cpp
@@ -45,26 +45,41 @@ bool BaseRecursiveJoin::getNextTuplesInternal(ExecutionContext* context) {
             inputFTableMorsel->startTupleIdx, inputFTableMorsel->numTuples, colIndicesToScan);
         bfsMorsel->resetState();
         computeBFS(context); // Phase 1
-        outputCursor = 0;
+        outputCursor = 0;    // Reset cursor for result scanning.
     }
 }
 
-std::unique_ptr<ResultSet> BaseRecursiveJoin::getLocalResultSet() {
-    auto numDataChunks = tmpDstNodeIDVectorPos.dataChunkPos + 1;
-    auto resultSet = std::make_unique<ResultSet>(numDataChunks);
-    auto dataChunk0 = std::make_shared<common::DataChunk>(tmpDstNodeIDVectorPos.valueVectorPos + 1);
+// ResultSet for list extend.
+static std::unique_ptr<ResultSet> populateResultWith2DataChunks() {
+    auto resultSet = std::make_unique<ResultSet>(2);
+    auto dataChunk0 = std::make_shared<common::DataChunk>(1);
+    dataChunk0->state = common::DataChunkState::getSingleValueDataChunkState();
+    dataChunk0->insert(0, std::make_shared<common::ValueVector>(common::INTERNAL_ID, nullptr));
+    auto dataChunk1 = std::make_shared<common::DataChunk>(1);
+    dataChunk1->insert(0, std::make_shared<common::ValueVector>(common::INTERNAL_ID, nullptr));
+    resultSet->insert(0, std::move(dataChunk0));
+    resultSet->insert(1, std::move(dataChunk1));
+    return resultSet;
+}
+
+// ResultSet for column extend.
+static std::unique_ptr<ResultSet> populateResultWith1DataChunks() {
+    auto resultSet = std::make_unique<ResultSet>(1);
+    auto dataChunk0 = std::make_shared<common::DataChunk>(2);
     dataChunk0->state = common::DataChunkState::getSingleValueDataChunkState();
     dataChunk0->insert(0, std::make_shared<common::ValueVector>(common::INTERNAL_ID, nullptr));
-    resultSet->insert(0, dataChunk0);
-    if (numDataChunks > 1) {
-        assert(numDataChunks == 2);
-        auto dataChunk1 = std::make_shared<common::DataChunk>(1);
-        dataChunk1->insert(0, std::make_shared<common::ValueVector>(common::INTERNAL_ID, nullptr));
-        resultSet->insert(1, std::move(dataChunk1));
+    dataChunk0->insert(1, std::make_shared<common::ValueVector>(common::INTERNAL_ID, nullptr));
+    resultSet->insert(0, std::move(dataChunk0));
+    return resultSet;
+}
+
+std::unique_ptr<ResultSet> BaseRecursiveJoin::getLocalResultSet() {
+    if (tmpDstNodeIDVectorPos.dataChunkPos == 1) {
+        return populateResultWith2DataChunks();
     } else {
-        dataChunk0->insert(1, std::make_shared<common::ValueVector>(common::INTERNAL_ID, nullptr));
+        assert(tmpDstNodeIDVectorPos.dataChunkPos == 0);
+        return populateResultWith1DataChunks();
     }
-    return resultSet;
 }
 
 void BaseRecursiveJoin::initLocalRecursivePlan(ExecutionContext* context) {

diff --git a/test/test_files/demo_db/demo_db.test b/test/test_files/demo_db/demo_db.test
@@ -1,13 +1,3 @@
--NAME WhereExists1
--QUERY MATCH (a:User) WHERE a.age < 100 AND EXISTS { MATCH (a)-[:Follows*3..3]->(b:User)} RETURN a.name, a.age;
--PARALLELISM 1
----- 1
-Adam|30
-
--NAME WhereExists2
--QUERY MATCH (a:User) WHERE a.age < 100 AND EXISTS { MATCH (a)-[:Follows*3..3]->(b:User) WHERE EXISTS {MATCH (b)-[:Follows]->(c:User)} } RETURN a.name, a.age;
----- 0
-
 -NAME Limit1
 -QUERY MATCH (u:User) RETURN u.name ORDER BY u.age DESC LIMIT 3;
 ---- 3

diff --git a/test/test_files/shortest_path/bfs_sssp.test b/test/test_files/shortest_path/bfs_sssp.test
@@ -1,9 +1,3 @@
--NAME SingleSourceSingleDestination
--QUERY MATCH (a:person)-[r:knows* SHORTEST 1..30]->(b:person) WHERE a.fName = 'Alice' AND b.fName = 'Bob' RETURN a.fName, b.fName, r._length
--PARALLELISM 1
----- 1
-Alice|Bob|1
-
 -NAME SingleSourceAllDestinationsSSP
 -QUERY MATCH (a:person)-[r:knows* SHORTEST 1..30]->(b:person) WHERE a.fName = 'Alice' RETURN a.fName, b.fName, r._length
 ---- 7

diff --git a/test/test_files/tinysnb/var_length_extend/var_length_adj_list_extend.test b/test/test_files/tinysnb/var_length_extend/var_length_adj_list_extend.test
@@ -26,7 +26,7 @@
 
 -NAME KnowsLongPathTest
 -QUERY MATCH (a:person)-[:knows*8..11]->(b:person) RETURN COUNT(*)
--PARALLELISM 1
+-PARALLELISM 8
 ---- 1
 1049760
 

diff --git a/test/test_files/tinysnb/var_length_extend/var_length_column_extend.test b/test/test_files/tinysnb/var_length_extend/var_length_column_extend.test
@@ -9,7 +9,7 @@
 
 -NAME meetsOneToTwoHopTest
 -QUERY MATCH (a:person)-[:meets*1..2]->(b:person) RETURN COUNT(*)
--PARALLELISM 1
+-PARALLELISM 2
 ---- 1
 13