Merge pull request #1085 from williamBlazing/fix/join_leak

[REVIEW] clearing array caches after PartwiseJoin is done

CHANGELOG.md

@@ -37,6 +37,7 @@
 - #1073 Fixed parseSchemaPython can throw exceptions
 - #1074: Remove lock inside grow() method from PinnedBufferProvider
 - #1071 Fix crash when loading an empty folder
+- #1085 Fixed intra-query memory leak in joins. Fixed by clearing array caches after PartwiseJoin is done 
 
 
 # BlazingSQL 0.15.0 (August 31, 2020)

engine/src/execution_graph/logic_controllers/BatchJoinProcessing.h

@@ -412,6 +412,10 @@ class PartwiseJoin : public kernel {
 									"duration"_a=timer.elapsed_time(),
 									"kernel_id"_a=this->get_id());
 
+		// these are intra kernel caches. We want to make sure they are empty before we finish.
+		this->leftArrayCache->clear();
+		this->rightArrayCache->clear();
+
 		return kstatus::proceed;
 	}
 

engine/src/execution_graph/logic_controllers/CacheMachine.cpp

@@ -200,10 +200,7 @@ void CacheMachine::put(size_t message_id, std::unique_ptr<ral::frame::BlazingTab
 }
 
 void CacheMachine::clear() {
-	std::unique_ptr<message> message_data;
-	while(message_data = waitingCache->pop_or_wait()) {
-		printf("...cleaning cache\n");
-	}
+	auto messages = this->waitingCache->get_all();
 	this->waitingCache->finish();
 }
 

engine/src/execution_graph/logic_controllers/CacheMachine.h

@@ -766,6 +766,14 @@ class WaitingQueue {
 		return std::move(data);
 	}
 
+	/**
+	 * gets all the messages
+	 */
+	std::vector<message_ptr> get_all(){
+		std::unique_lock<std::mutex> lock(mutex_);
+		return get_all_unsafe();
+	}
+
 	/**
 	* Waits until all messages are ready then returns all of them.
 	* You should never call this function more than once on a WaitingQueue else