Intra partition parallelism

CMakeLists.txt

@@ -201,6 +201,7 @@ set(SOURCES src/backend/JasmineGraphBackend.cpp
         src/query/processor/cypher/runtime/InstanceHandler.cpp
         src/query/processor/cypher/runtime/OperatorExecutor.cpp
         src/query/processor/cypher/util/SharedBuffer.cpp
+        src/query/processor/executor/IntraPartitionParallelExecutor.cpp
         src/nativestore/MetaPropertyLink.cpp
         src/nativestore/MetaPropertyEdgeLink.cpp
         src/nativestore/MetaPropertyLink.cpp

conf/jasminegraph-server.properties

@@ -124,5 +124,12 @@ org.jasminegraph.vectorstore.dimension=768
 org.jasminegraph.vectorstore.embedding.model=nomic-embed-text
 org.jasminegraph.vectorstore.embedding.ollama.endpoint=http://172.30.5.100:11441
 
+#--------------------------------------------------------------------------------
+# Query execution thread pool settings
+#--------------------------------------------------------------------------------
+#Maximum number of worker threads for intra-partition parallel query execution
+#If not set, defaults to 32.
+org.jasminegraph.query.threadpool.maxworkers=32
+
 
 

src/nativestore/NodeBlock.cpp

@@ -412,14 +412,11 @@ std::list<std::pair<NodeBlock*, RelationBlock*>> NodeBlock::getAllEdgeNodes() {
     return allEdges;
 }
 
-std::map<std::string, char*> NodeBlock::getAllProperties() {
-    std::map<std::string, char*> allProperties;
+std::map<std::string, std::string, std::less<>> NodeBlock::getAllProperties() {
+    std::map<std::string, std::string, std::less<>> allProperties;
     PropertyLink* current = this->getPropertyHead();
     while (current) {
-        // don't forget to free the allocated memory after using this method
-        char* copiedValue = new char[PropertyLink::MAX_VALUE_SIZE];
-        std::strncpy(copiedValue, current->value, PropertyLink::MAX_VALUE_SIZE);
-        allProperties.insert({current->name, copiedValue});
+        allProperties.try_emplace(current->name, current->value);
         PropertyLink* temp = current->next();
         delete current;  // To prevent memory leaks
         current = temp;
@@ -486,8 +483,8 @@ NodeBlock* NodeBlock::get(unsigned int blockAddress) {
         new NodeBlock(id, nodeId, blockAddress, propRef, metaPropRef, edgeRef,
                       centralEdgeRef, edgeRefPID, label, usage);
     if (nodeBlockPointer->id.length() == 0) {  // if label not found in node block look in the properties
-        std::map<std::string, char*> props = nodeBlockPointer->getAllProperties();
-        if (props["label"]) {
+        std::map<std::string, std::string, std::less<>> props = nodeBlockPointer->getAllProperties();
+        if (!props["label"].empty()) {
             nodeBlockPointer->id = props["label"];
         } else {
             node_block_logger.error("Could not find node ID/Label for node with block address = " +

src/nativestore/NodeBlock.h

@@ -77,7 +77,7 @@ class NodeBlock {
     std::map<std::string, char *> getProperty(std::string);
     PropertyLink *getPropertyHead();
     MetaPropertyLink *getMetaPropertyHead();
-    std::map<std::string, char *> getAllProperties();
+    std::map<std::string, std::string, std::less<>> getAllProperties();
 
     bool updateLocalRelation(RelationBlock *, bool relocateHead = true);
     bool updateCentralRelation(RelationBlock *newRelation, bool relocateHead = true);

src/nativestore/NodeManager.cpp

@@ -400,11 +400,16 @@ int NodeManager::dbSize(std::string path) {
  * @Deprecated use NodeBlock.get() instead
  **/
 NodeBlock *NodeManager::get(std::string nodeId) {
-    NodeBlock *nodeBlockPointer = NULL;
+    NodeBlock *nodeBlockPointer = nullptr;
     if (this->nodeIndex.find(nodeId) == this->nodeIndex.end()) {  // Not found
         return nodeBlockPointer;
     }
     unsigned int nodeIndex = this->nodeIndex[nodeId];
+    return get(nodeIndex, nodeId);
+}
+
+NodeBlock *NodeManager::get(unsigned int nodeIndex, const std::string& nodeId) {
+    NodeBlock *nodeBlockPointer = nullptr;
     const unsigned int blockAddress = nodeIndex * NodeBlock::BLOCK_SIZE;
     NodeBlock::nodesDB->seekg(blockAddress);
     unsigned int vertexId;
@@ -455,11 +460,10 @@ NodeBlock *NodeManager::get(std::string nodeId) {
     nodeBlockPointer = new NodeBlock(nodeId, vertexId, blockAddress, propRef, metaPropRef, edgeRef,
                                      centralEdgeRef, edgeRefPID, label, usage);
 
-    node_manager_logger.debug("DEBUG: nodeBlockPointer after creating the object edgeRef " +
-                              std::to_string(nodeBlockPointer->edgeRef));
 
     if (nodeBlockPointer->edgeRef % RelationBlock::BLOCK_SIZE != 0) {
-        node_manager_logger.error("Exception: Invalid edge reference address = " + nodeBlockPointer->edgeRef);
+        node_manager_logger.error("Exception: Invalid edge reference address = " +
+                                  std::to_string(nodeBlockPointer->edgeRef));
     }
     return nodeBlockPointer;
 }

src/nativestore/NodeManager.h

@@ -55,7 +55,10 @@ class NodeManager {
     std::unordered_map<std::string, unsigned int> nodeIndex;
     std::unordered_map<std::string, unsigned int> edgeIndex;
     NodeManager(GraphConfig);
-    ~NodeManager() { delete NodeBlock::nodesDB; };
+
+    NodeBlock *get(std::string);
+    static NodeBlock *get(unsigned int nodeIndex, const std::string& nodeId);
+    void addNode(NodeBlock nodeBlock);
 
     void setIndexKeySize(unsigned long);
     static int dbSize(std::string path);
@@ -72,7 +75,6 @@ class NodeManager {
     RelationBlock* addCentralRelation(NodeBlock source, NodeBlock destination);
 
     NodeBlock* addNode(std::string);  // will return DB block address
-    NodeBlock* get(std::string);
 
     std::list<NodeBlock*> getCentralGraph();
     std::list<NodeBlock> getLimitedGraph(int limit = 10);

src/nativestore/RelationBlock.cpp

@@ -849,14 +849,11 @@ MetaPropertyEdgeLink *RelationBlock::getMetaPropertyHead() {
     return MetaPropertyEdgeLink::get(this->metaPropertyAddress);
 }
 
-std::map<std::string, char*> RelationBlock::getAllProperties() {
-    std::map<std::string, char*> allProperties;
+std::map<std::string, std::string, std::less<>> RelationBlock::getAllProperties() {
+    std::map<std::string, std::string, std::less<>> allProperties;
     PropertyEdgeLink* current = this->getPropertyHead();
     while (current) {
-        // don't forget to free the allocated memory after using this method
-        char* copiedValue = new char[PropertyEdgeLink::MAX_VALUE_SIZE];
-        std::strncpy(copiedValue, current->value, PropertyEdgeLink::MAX_VALUE_SIZE);
-        allProperties.insert({current->name, copiedValue});
+        allProperties.try_emplace(current->name, current->value);
         PropertyEdgeLink* temp = current->next();
         delete current;  // To prevent memory leaks
         current = temp;

src/nativestore/RelationBlock.h

@@ -169,7 +169,7 @@ class RelationBlock {
     std::string getCentralRelationshipType();
     PropertyEdgeLink *getPropertyHead();
     MetaPropertyEdgeLink *getMetaPropertyHead();
-    std::map<std::string, char *> getAllProperties();
+    std::map<std::string, std::string, std::less<>> getAllProperties();
 };
 
 #endif