﻿/*
	Copyright (c) 2017 TOSHIBA Digital Solutions Corporation

	This program is free software: you can redistribute it and/or modify
	it under the terms of the GNU Affero General Public License as
	published by the Free Software Foundation, either version 3 of the
	License, or (at your option) any later version.

	This program is distributed in the hope that it will be useful,
	but WITHOUT ANY WARRANTY; without even the implied warranty of
	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
	GNU Affero General Public License for more details.

	You should have received a copy of the GNU Affero General Public License
	along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

#include "sql_operator_join.h"
#include "sql_operator_utils.h"


const SQLOps::OpRegistrar
SQLJoinOps::Registrar::REGISTRAR_INSTANCE((Registrar()));

void SQLJoinOps::Registrar::operator()() const {
	add<SQLOpTypes::OP_JOIN, Join>();
	add<SQLOpTypes::OP_JOIN_OUTER, JoinOuter>();
	add<SQLOpTypes::OP_JOIN_SORTED, JoinSorted>();
	add<SQLOpTypes::OP_JOIN_NESTED, JoinNested>();
	add<SQLOpTypes::OP_JOIN_COMP_MERGE, JoinCompMerge>();
}


void SQLJoinOps::Join::compile(OpContext &cxt) const {
	OpPlan &plan = cxt.getPlan();
	const OpCode &code = getCode();

	const uint32_t inCount = getCode().getInputCount();

	if (code.getJoinType() != SQLType::JOIN_INNER) {
		OpNode &node = plan.createNode(SQLOpTypes::OP_JOIN_OUTER);

		for (uint32_t i = 0; i < inCount; i++) {
			node.addInput(plan.getParentInput(i));
		}
		plan.getParentOutput(0).addInput(node);

		node.setCode(code);
		return;
	}

	const SQLValues::CompColumnList *keyColumnList = code.findKeyColumnList();
	const bool sorted = (keyColumnList != NULL);

	OpNode &joinNode = plan.createNode(
			sorted ? SQLOpTypes::OP_JOIN_SORTED : SQLOpTypes::OP_JOIN_NESTED);

	for (uint32_t i = 0; i < inCount; i++) {
		joinNode.addInput(plan.getParentInput(i));
	}

	{
		OpCodeBuilder builder(OpCodeBuilder::ofContext(cxt));
		OpCode joinCode = code;

		std::pair<Projection*, Projection*> projections =
				builder.arrangeProjections(joinCode, true, true);

		if (sorted) {
			projections.first = &builder.createFilteredProjection(
					*projections.first,
					builder.getExprRewriter().compColumnListToPredicate(
							builder.getExprFactoryContext(),
							*keyColumnList));
		}

		joinCode.setPipeProjection(projections.first);
		joinCode.setFinishProjection(projections.second);

		joinNode.setCode(joinCode);
	}
	plan.getParentOutput(0).addInput(joinNode);
}


void SQLJoinOps::JoinOuter::compile(OpContext &cxt) const {
	OpPlan &plan = cxt.getPlan();

	OpCodeBuilder builder(OpCodeBuilder::ofContext(cxt));

	SQLExprs::ExprRewriter &rewriter = builder.getExprRewriter();
	SQLExprs::ExprFactoryContext &factoryCxt = builder.getExprFactoryContext();
	rewriter.activateColumnMapping(factoryCxt);

	const uint32_t inCount = 2;

	const SQLType::JoinType joinType = getCode().getJoinType();
	const bool outerList[inCount] = {
			(joinType == SQLType::JOIN_LEFT_OUTER ||
			joinType == SQLType::JOIN_FULL_OUTER),
			(joinType == SQLType::JOIN_RIGHT_OUTER ||
			joinType == SQLType::JOIN_FULL_OUTER)
	};

	for (uint32_t i = 0; i < inCount; i++) {
		if (outerList[i]) {
			rewriter.setIdOfInput(i, true);
		}
	}

	const OpNode *inNodeList[inCount];
	for (uint32_t i = 0; i < inCount; i++) {
		const OpNode *&inNode = inNodeList[i];

		inNode = &plan.getParentInput(i);

		SQLExprs::ExprRewriter::Scope scope(rewriter);
		rewriter.setMappedInput(i, 0);
		rewriter.setInputNull((inCount - i - 1), true);

		OpNode &node = plan.createNode(SQLOpTypes::OP_SELECT);
		node.addInput(*inNode);

		OpCode code;
		code.setPipeProjection(&builder.createProjectionByUsage(false));
		node.setCode(code);

		inNode = &node;
	}

	rewriter.setIdProjected(true);

	OpNode &joinNode = plan.createNode(SQLOpTypes::OP_JOIN);
	for (uint32_t i = 0; i < inCount; i++) {
		joinNode.addInput(*inNodeList[i]);
	}
	{
		SQLExprs::ExprFactoryContext::Scope scope(factoryCxt);
		factoryCxt.setAggregationPhase(true, SQLType::AGG_PHASE_ALL_PIPE);

		OpCode code;
		code.setJoinType(SQLType::JOIN_INNER);
		if (getCode().findKeyColumnList() != NULL) {
			code.setKeyColumnList(&rewriter.rewriteCompColumnList(
					factoryCxt, getCode().getKeyColumnList(), false));
		}
		if (getCode().getJoinPredicate() != NULL) {
			code.setFilterPredicate(&rewriter.rewritePredicate(
					factoryCxt, getCode().getJoinPredicate()));
		}
		code.setPipeProjection(&builder.createProjectionByUsage(false));
		joinNode.setCode(code);
	}

	rewriter.setInputProjected(true);
	builder.applyJoinType(joinType);

	OpNode *compensatingNode = &joinNode;
	for (uint32_t i = 0; i < inCount; i++) {
		if (!outerList[i]) {
			continue;
		}

		OpNode &node = plan.createNode(SQLOpTypes::OP_UNION);
		node.addInput(*compensatingNode);
		node.addInput(*inNodeList[i]);

		SQLValues::CompColumnList &keyColumnList =
				builder.createCompColumnList();
		{
			SQLValues::CompColumn keyColumn;
			keyColumn.setColumnPos(rewriter.getMappedIdColumn(i), true);
			keyColumnList.push_back(keyColumn);
		}

		OpCode code;
		code.setUnionType(SQLType::END_UNION);
		code.setKeyColumnList(&keyColumnList);
		code.setPipeProjection(&builder.createProjectionByUsage(true));
		node.setCode(code);

		compensatingNode = &node;
	}

	{
		OpNode &node = plan.createNode(SQLOpTypes::OP_SELECT);
		node.addInput(*compensatingNode);

		OpCode code = getCode();

		code.setJoinType(SQLType::END_JOIN);
		code.setJoinPredicate(NULL);
		code.setKeyColumnList(NULL);

		std::pair<Projection*, Projection*> projections =
				builder.arrangeProjections(code, true, true);
		code.setPipeProjection(projections.first);
		code.setFinishProjection(projections.second);
		node.setCode(code);

		plan.getParentOutput(0).addInput(node);
	}
}


void SQLJoinOps::JoinSorted::compile(OpContext &cxt) const {
	OpPlan &plan = cxt.getPlan();
	const OpCode &code = getCode();

	OpNode &joinNode = plan.createNode(SQLOpTypes::OP_JOIN_COMP_MERGE);

	OpCodeBuilder builder(OpCodeBuilder::ofContext(cxt));

	const uint32_t inCount = getCode().getInputCount();
	for (uint32_t i = 0; i < inCount; i++) {
		OpNode &subNode = plan.createNode(SQLOpTypes::OP_SORT);
		OpCode subCode;

		SQLExprs::ExprRewriter &rewriter = builder.getExprRewriter();
		SQLExprs::ExprRewriter::Scope scope(rewriter);
		rewriter.activateColumnMapping(builder.getExprFactoryContext());
		rewriter.setMappedInput(i, 0);

		SQLValues::CompColumnList &list = builder.createCompColumnList();

		const SQLValues::CompColumnList srcList = code.getKeyColumnList();
		for (SQLValues::CompColumnList::const_iterator it = srcList.begin();
				it != srcList.end(); ++it) {
			SQLValues::CompColumn key;
			key.setColumnPos(it->getColumnPos((i == 0)), true);
			list.push_back(key);
		}

		subCode.setKeyColumnList(&list);
		subCode.setPipeProjection(
				&builder.createIdenticalProjection(false, i));

		subNode.setCode(subCode);
		subNode.addInput(plan.getParentInput(i));

		joinNode.addInput(subNode);
	}

	joinNode.setCode(code);
	plan.getParentOutput(0).addInput(joinNode);
}


void SQLJoinOps::JoinNested::execute(OpContext &cxt) const {
	uint32_t inList[] = { IN_LEFT, IN_RIGHT }; 
	cxt.setReaderRandomAccess(inList[1]);

	TupleListReader &reader1 = cxt.getReader(inList[0]); 
	TupleListReader &reader2 = cxt.getReader(inList[1]); 
	TupleListReader &reader2Top = cxt.getReader(inList[1], 1); 

	if (!cxt.isInputCompleted(inList[1]) || !reader2.exists()) {
		return;
	}
	reader2Top.exists(); 

	for (; reader1.exists(); reader1.next()) {
		if (cxt.checkSuspended()) { 
			return;
		}
		for (; reader2.exists(); reader2.next()) {
			if (cxt.checkSuspended()) { 
				return;
			}
			getCode().getPipeProjection()->project(cxt);
		}

		reader2.assign(reader2Top);
	}
}


void SQLJoinOps::JoinCompMerge::execute(OpContext &cxt) const {
	uint32_t inList[] = { IN_LEFT, IN_RIGHT }; 
	cxt.setReaderRandomAccess(inList[1]);

	TupleListReader &reader1 = cxt.getReader(inList[0]); 
	TupleListReader &reader2 = cxt.getReader(inList[1], IN_INSIDE_MAIN); 
	TupleListReader &reader2Lower = cxt.getReader(inList[1], IN_INSIDE_SUB); 
	const SQLValues::TupleRangeComparator &comp =
			getCode().getJoinRangePredicate();

	if (!reader2.exists()) {
		return;
	}
	reader2Lower.exists(); 

	bool lowerAssignable = false;
	for (; reader1.exists(); reader1.next()) {
		if (cxt.checkSuspended()) { 
			return;
		}

		bool lowerReached = false;
		for (; reader2.exists(); reader2.next()) {
			if (cxt.checkSuspended()) { 
				return;
			}

			const int32_t ret = comp(reader1.get(), reader2.get());

			if (!lowerReached) {
				if (ret > 0) {
					continue; 
				}
				if (lowerAssignable) {
					reader2Lower.assign(reader2);
				}
				lowerReached = true;
			}

			if (ret < 0) {
				break; 
			}

			getCode().getPipeProjection()->project(cxt);
		}

		reader2.assign(reader2Lower);
		lowerAssignable = true;
	}
}
