package org.teiid.query.function.aggregate;

import java.util.List;

import org.teiid.api.exception.query.ExpressionEvaluationException;
import org.teiid.api.exception.query.FunctionExecutionException;
import org.teiid.common.buffer.IndexedTupleSource;
import org.teiid.core.TeiidComponentException;
import org.teiid.core.TeiidProcessingException;
import org.teiid.query.sql.symbol.AggregateSymbol.Type;
import org.teiid.query.util.CommandContext;

public class LeadLagFunction extends AggregateFunction implements FullTableFunction {
	private Type type;
	private Object result;
	private int offset = 1;
	private Object def = null;

	public LeadLagFunction(Type type) {
		this.type = type;
	}

	@Override
	public void addInputDirect(List<?> tuple, CommandContext commandContext) throws TeiidComponentException, TeiidProcessingException { }

	@Override
	public void reset() { }

	@Override
	public void addInputTable(IndexedTupleSource ts, int currentRow, int startRow, int endRow, CommandContext commandContext) throws TeiidProcessingException, TeiidComponentException {
		ts.setPosition(currentRow);
		List<?> tuple = ts.nextTuple();
		if (argIndexes.length > 1) {
			offset = (Integer) tuple.get(argIndexes[1]);
		}
		if (argIndexes.length > 2) {
			def = tuple.get(argIndexes[2]);
		}
		int actualRow = 0;
		if (type == Type.LEAD)
			actualRow = currentRow + offset;
		else if (type == Type.LAG)
			actualRow = currentRow - offset;
		if (actualRow >= startRow && actualRow <= endRow) {
			ts.setPosition(actualRow);
			result = ts.nextTuple().get(argIndexes[0]);
		} else
			result = def;
	}

	@Override
	public Object getResult(CommandContext commandContext) throws FunctionExecutionException, ExpressionEvaluationException, TeiidComponentException, TeiidProcessingException {
		return result;
	}
}
