Method run

engine/src/execution_kernels/BatchJoinProcessing.cpp:512–636

}

kstatus PartwiseJoin::run() {
	CodeTimer timer;

	left_input = this->input_.get_cache("input_a");
	right_input = this->input_.get_cache("input_b");

	bool done = false;

	while (!done) {
		std::unique_ptr<ral::cache::CacheData> left_cache_data, right_cache_data;
		int left_ind, right_ind;
		if (max_left_ind == -1 && max_right_ind == -1){
			// before we load anything, lets make sure each side has data to process
			left_input->wait_for_next();
			right_input->wait_for_next();

			left_cache_data = load_left_set();
			right_cache_data = load_right_set();

			left_ind = this->max_left_ind = 0; // we have loaded just once. This is the highest index for now
			right_ind = this->max_right_ind = 0; // we have loaded just once. This is the highest index for now

			// parsing more of the expression here because we need to have the number of columns of the tables
			std::vector<int> column_indices;
			parseJoinConditionToColumnIndices(this->condition, column_indices);
			for(std::size_t i = 0; i < column_indices.size();i++){
				if(column_indices[i] >= static_cast<int>(left_cache_data->num_columns())){
					this->right_column_indices.push_back(column_indices[i] - left_cache_data->num_columns());
				}else{
					this->left_column_indices.push_back(column_indices[i]);
				}
			}

			std::vector<std::string> left_names = left_cache_data->names();
			std::vector<std::string> right_names = right_cache_data->names();
			this->result_names.reserve(left_names.size() + right_names.size());
			this->result_names.insert(this->result_names.end(), left_names.begin(), left_names.end());
			this->result_names.insert(this->result_names.end(), right_names.begin(), right_names.end());

			computeNormalizationData(left_cache_data->get_schema(), right_cache_data->get_schema());
		} else {
			// Not first load, so we have joined a set pair. Now lets see if there is another set pair we can do, but keeping one of the two sides we already have
			std::tie(left_ind, right_ind) = check_for_another_set_to_do_with_data_we_already_have();
			if (left_ind >= 0 && right_ind >= 0) {
				left_cache_data = this->leftArrayCache->get_or_wait_CacheData(left_ind);
				right_cache_data = this->rightArrayCache->get_or_wait_CacheData(right_ind);
			} else {
				if (this->left_input->wait_for_next()){
					left_cache_data = load_left_set();
					left_ind = this->max_left_ind;
				}
				if (this->right_input->wait_for_next()){
					right_cache_data = load_right_set();
					right_ind = this->max_right_ind;
				}
				if (left_ind >= 0 && right_ind >= 0) {
					// We pulled new data from left and right
				} else if (left_ind >= 0)	{