/*
 * JBoss, Home of Professional Open Source.
 * See the COPYRIGHT.txt file distributed with this work for information
 * regarding copyright ownership.  Some portions may be licensed
 * to Red Hat, Inc. under one or more contributor license agreements.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library 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
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301 USA.
 * Modified by
 * @author  Data Virtuality GmbH
 * @date    $Date$
 *
 */
// https://issues.jboss.org/browse/TEIID-3552
// https://issues.jboss.org/browse/TEIID-247
// https://issues.jboss.org/browse/TEIID-2166
// https://issues.jboss.org/browse/TEIID-2151
// https://issues.jboss.org/browse/TEIID-2234
// https://issues.jboss.org/browse/TEIID-2581
// https://issues.jboss.org/browse/TEIID-2571
// https://issues.jboss.org/browse/TEIID-2369
// https://issues.jboss.org/browse/TEIID-1386
// https://issues.jboss.org/browse/TEIID-2226
// https://issues.jboss.org/browse/TEIID-2233
// https://issues.jboss.org/browse/TEIID-2224
// https://issues.jboss.org/browse/TEIID-2225
// https://issues.jboss.org/browse/TEIID-2257
// https://issues.jboss.org/browse/TEIID-2235
// https://issues.jboss.org/browse/TEIID-2661
// https://issues.jboss.org/browse/TEIID-2178
// https://issues.jboss.org/browse/TEIID-2297
// https://issues.jboss.org/browse/TEIID-3127
// https://issues.jboss.org/browse/TEIID-2111
// https://issues.jboss.org/browse/TEIID-2635
// https://issues.jboss.org/browse/TEIID-4017
// https://issues.jboss.org/browse/TEIID-2205
// https://issues.jboss.org/browse/TEIID-2641
// https://issues.jboss.org/browse/TEIID-3053
// https://issues.jboss.org/browse/TEIID-4076
// https://issues.jboss.org/browse/TEIID-4508
// https://issues.jboss.org/browse/TEIID-4509
// https://issues.jboss.org/browse/TEIID-2152
// https://issues.jboss.org/browse/TEIID-2639
// https://issues.jboss.org/browse/TEIID-5412
// https://issues.jboss.org/browse/TEIID-4504
// https://issues.jboss.org/browse/TEIID-4993 (only changes for this class)
// https://issues.jboss.org/browse/TEIID-5445
// https://issues.jboss.org/browse/TEIID-3995
// https://issues.jboss.org/browse/TEIID-5686
// https://issues.redhat.com/browse/TEIID-2227
package org.teiid.query.processor.proc;

import com.datavirtuality.dv.core.interfaces.DVCaller;
import org.teiid.metadata.ExecuteAs;
import com.datavirtuality.dv.core.services.DVServices;
import org.teiid.adminapi.impl.SessionMetadata;
import org.teiid.api.exception.query.QueryMetadataException;
import org.teiid.api.exception.query.QueryValidatorException;
import org.teiid.client.ProcedureErrorInstructionException;
import org.teiid.client.plan.PlanNode;
import org.teiid.client.xa.XATransactionException;
import org.teiid.common.buffer.*;
import org.teiid.common.buffer.BufferManager.TupleSourceType;
import org.teiid.core.TeiidComponentException;
import org.teiid.core.TeiidProcessingException;
import org.teiid.core.TeiidRuntimeException;
import org.teiid.core.types.ArrayImpl;
import org.teiid.core.types.DataTypeManager;
import org.teiid.core.util.Assertion;
import org.teiid.dqp.internal.process.DQPWorkContext;
import org.teiid.dqp.internal.process.DVProcedureContext;
import org.teiid.dqp.internal.process.DataTierTupleSource;
import org.teiid.dqp.message.RequestID;
import org.teiid.dqp.service.TransactionContext;
import org.teiid.dqp.service.TransactionContext.Scope;
import org.teiid.dqp.service.TransactionService;
import org.teiid.events.EventDistributor;
import org.teiid.jdbc.TeiidSQLException;
import org.teiid.logging.LogConstants;
import org.teiid.logging.LogManager;
import org.teiid.metadata.Procedure;
import org.teiid.query.QueryPlugin;
import org.teiid.query.analysis.AnalysisRecord;
import org.teiid.query.metadata.QueryMetadataInterface;
import org.teiid.query.metadata.SupportConstants;
import org.teiid.query.metadata.TempMetadataID;
import org.teiid.query.processor.*;
import org.teiid.query.processor.proc.CreateCursorResultSetInstruction.Mode;
import org.teiid.query.processor.relational.SubqueryAwareEvaluator;
import org.teiid.query.processor.relational.SubqueryAwareRelationalNode;
import org.teiid.query.resolver.command.UpdateProcedureResolver;
import org.teiid.query.sql.ProcedureReservedWords;
import org.teiid.query.sql.lang.Command;
import org.teiid.query.sql.lang.Criteria;
import org.teiid.query.sql.proc.CreateProcedureCommand;
import org.teiid.query.sql.symbol.ElementSymbol;
import org.teiid.query.sql.symbol.Expression;
import org.teiid.query.sql.symbol.GroupSymbol;
import org.teiid.query.sql.util.VariableContext;
import org.teiid.query.sql.visitor.ValueIteratorProviderCollectorVisitor;
import org.teiid.query.tempdata.TempTableStore;
import org.teiid.query.util.CommandContext;
import org.teiid.query.util.DVContextGetter;
import org.teiid.query.util.DVUsersHandlerIntf;

import javax.transaction.SystemException;
import java.lang.ref.WeakReference;
import java.util.*;

import static org.teiid.query.analysis.AnalysisRecord.PROP_OUTPUT_COLS;

/**
 */
public class ProcedurePlan extends ProcessorPlan {
	private static class CursorState {
		QueryProcessor processor;
		IndexedTupleSource ts;
		List<?> currentRow;
		TupleBuffer resultsBuffer;
		public boolean returning;
		public boolean usesLocalTemp;
	}

	public static ElementSymbol ROWCOUNT =
		new ElementSymbol(ProcedureReservedWords.VARIABLES+"."+ProcedureReservedWords.ROWCOUNT); //$NON-NLS-1$

	static {
		ROWCOUNT.setType(DataTypeManager.DefaultDataClasses.INTEGER);
	}

	public static ElementSymbol DVPROCCONTEXT =
			new ElementSymbol(ProcedureReservedWords.VARIABLES + "." + ProcedureReservedWords.DVPROCCONTEXT); //$NON-NLS-1$

	static {
		DVPROCCONTEXT.setType(DataTypeManager.DefaultDataClasses.OBJECT);
	}

	private Program originalProgram;
	private final Command command;

	// State initialized by processor
	private ProcessorDataManager dataMgr;
	private ProcessorDataManager parentDataMrg;
    private BufferManager bufferMgr;
    private int batchSize;

    private boolean done = false;
    private CursorState currentState;

    // Temp state for final results
    private TupleSource finalTupleSource;
    private int beginBatch = 1;
    private List<List<?>> batchRows;
    private boolean lastBatch = false;
    private LinkedHashMap<ElementSymbol, Expression> params;
	private boolean runInContext = true;
    private List<ElementSymbol> outParams;
    private QueryMetadataInterface metadata;

    private VariableContext cursorStates;

	private VariableContext currentVarContext;
	private VariableContext parentContext;
	private volatile SessionMetadata parentSession;

    private List outputElements;

	private SubqueryAwareEvaluator evaluator;

    // Stack of programs, with current program on top
    private Stack<Program> programs = new Stack<Program>();

    private boolean evaluatedParams;

    private int updateCount = Procedure.AUTO_UPDATECOUNT;

    private TransactionContext blockContext;
    /**
     * Resources cannot be held open across the txn boundary.  This list is a hack at ensuring the resources are closed.
     */
    private LinkedList<WeakReference<DataTierTupleSource>> txnTupleSources = new LinkedList<WeakReference<DataTierTupleSource>>();

	private boolean validateAccess = true;

	private GroupSymbol virtualGroup;

	private boolean procContextPushed;

    /**
     * Constructor for ProcedurePlan.
     */
    public ProcedurePlan(Program originalProgram, Command command) {
    	this.originalProgram = originalProgram;
		this.command = command;
		createVariableContext();
    }

    public Program getOriginalProgram() {
		return originalProgram;
	}

    /**
     * @see ProcessorPlan#initialize(ProcessorDataManager, Object)
     */
    public void initialize(CommandContext context, ProcessorDataManager dataMgr, BufferManager bufferMgr) {
        this.bufferMgr = bufferMgr;
        this.batchSize = bufferMgr.getProcessorBatchSize(getOutputElements());
	    this.parentContext = context.getVariableContext();
		context.setInherited(true);
		CommandContext clonedContext = null;
		if (command instanceof CreateProcedureCommand) {
			final CreateProcedureCommand c1 = (CreateProcedureCommand) command;
			if (c1.getVirtualGroup() != null) {
				Object mid = c1.getVirtualGroup().getMetadataID();
				if (mid instanceof TempMetadataID)
					mid = ((TempMetadataID) mid).getOriginalMetadataID();
				if (mid instanceof Procedure) {
					final Procedure p = (Procedure) mid;
					final SessionMetadata session = context.getSession();
					if (p.getExecuteAs() == ExecuteAs.OWNER && session != null) {
						final String userName = p.getOwner() + "@" + session.getSecurityDomain();
						boolean hasUser = DVServices.callne(DVUsersHandlerIntf.class,
								s -> s.hasUser(p.getOwner()), false);
						if (!hasUser)
							throw new TeiidRuntimeException(QueryPlugin.Util.gs(QueryPlugin.Event.DVCORE94189, userName, p.getFullName()));
						if (!Objects.equals(userName, session.getUserName())) {
							clonedContext = DVServices.callne(DVContextGetter.class,
									s -> s.getContextForUser(context, p.getOwner()), null);
							this.parentSession = session;
						}
					}
				}
			}
		}
		if (clonedContext == null)
			clonedContext = CommandContext.newChild(context);
		setContext(clonedContext);
        this.dataMgr = new ProcessorDataManager() {

			@Override
			public TupleSource registerRequest(CommandContext context, Command command,
					String modelName, RegisterRequestParameter parameterObject)
					throws TeiidComponentException, TeiidProcessingException {
// DVCORE-3832
//				if (command instanceof StoredProcedure) {
//					getContext().setSchema(((StoredProcedure) command).getGroup().getSchema());
//				}
				TupleSource ts = parentDataMrg.registerRequest(context, command, modelName, parameterObject);
				if (blockContext != null && ts instanceof DataTierTupleSource) {
					txnTupleSources.add(new WeakReference<DataTierTupleSource>((DataTierTupleSource)ts));
				}
				return ts;
			}

			@Override
			public Object lookupCodeValue(CommandContext context, String codeTableName,
					String returnElementName, String keyElementName, Object keyValue)
					throws BlockedException, TeiidComponentException,
					TeiidProcessingException {
				return parentDataMrg.lookupCodeValue(context, codeTableName, returnElementName, keyElementName, keyValue);
			}

			@Override
			public EventDistributor getEventDistributor() {
				return parentDataMrg.getEventDistributor();
			}
		};
        this.parentDataMrg = dataMgr;
        if (evaluator == null) {
        	this.evaluator = new SubqueryAwareEvaluator(Collections.emptyMap(), getDataManager(), getContext(), this.bufferMgr);
        }
    }

	public void reset() {
        super.reset();
        if (evaluator != null) {
        	evaluator.reset();
        }
        evaluatedParams = false;
	    if (parentContext != null) {
		    super.getContext().setVariableContext(parentContext);
		}
        createVariableContext();

        done = false;
        currentState = null;
        finalTupleSource = null;
        beginBatch = 1;
        batchRows = null;
        lastBatch = false;
        //reset program stack
        programs.clear();
		LogManager.logTrace(org.teiid.logging.LogConstants.CTX_DQP, "ProcedurePlan reset"); //$NON-NLS-1$
    }

    public ProcessorDataManager getDataManager() {
        return this.dataMgr;
    }

    public void open() throws TeiidProcessingException, TeiidComponentException {
    	if (!this.evaluatedParams) {
    		if (this.outParams != null) {
    			for (ElementSymbol param : this.outParams) {
					setParameterValue(param, getCurrentVariableContext(), null);
				}
    		}
    		if (this.params != null) {
		        for (Map.Entry<ElementSymbol, Expression> entry : this.params.entrySet()) {
		            ElementSymbol param = entry.getKey();
		            Expression expr = entry.getValue();

		            VariableContext context = getCurrentVariableContext();
		            if (context.containsVariable(param)) {
		            	continue;
		            }
		            Object value = this.evaluateExpression(expr);

		            //check constraint
		            checkNotNull(param, value);
		            setParameterValue(param, context, value);
		        }
		        this.evaluator.close();
		    } else if (runInContext) {
			    //if there are no params, this needs to run in the current variable context
			    this.currentVarContext.setParentContext(parentContext);
    		}
    		this.push(originalProgram);
    	}
    	this.evaluatedParams = true;
		final DVProcedureContextImpl ctx = new DVProcedureContextImpl(virtualGroup);
		DVProcedureContext.Store.push(ctx);
		procContextPushed = true;
		this.currentVarContext.setValue(DVPROCCONTEXT, ctx);
		if (parentSession != null)
			DQPWorkContext.getWorkContext().setSession(getContext().getSession());
    }

	private void checkNotNull(ElementSymbol param, Object value)
			throws TeiidComponentException, QueryMetadataException,
			QueryValidatorException {
		if (metadata.elementSupports(param.getMetadataID(), SupportConstants.Element.NULL)) {
			return;
		}
		if (metadata.isVariadic(param.getMetadataID())) {
		 	if (value instanceof ArrayImpl) {
				ArrayImpl av = (ArrayImpl)value;
				for (Object o : av.getValues()) {
					if (o == null) {
					     throw new QueryValidatorException(QueryPlugin.Event.TEIID30164, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID30164, param));
					}
				}
			}
		} else if (value == null) {
		 	throw new QueryValidatorException(QueryPlugin.Event.TEIID30164, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID30164, param));
		}
	}

	protected void setParameterValue(ElementSymbol param,
			VariableContext context, Object value) {
		context.setValue(param, value);
	}

	@Override
	public TupleBatch nextBatch() throws BlockedException,
			TeiidComponentException, TeiidProcessingException {
		if (blockContext != null) {
			this.getContext().getTransactionServer().resume(blockContext);
		}
		try {
			return nextBatchDirect();
		} finally {
			//ensure that the generatedkeys don't escape
			getContext().clearGeneratedKeys();
			if (blockContext != null) {
				this.getContext().getTransactionServer().suspend(blockContext);
			}
		}
	}

    /**
     * @see ProcessorPlan#nextBatch()
     */
    public TupleBatch nextBatchDirect()
        throws TeiidComponentException, TeiidProcessingException, BlockedException {

        // Already returned results?
        if(done) {
            // Already returned all results
            TupleBatch emptyTerminationBatch = new TupleBatch(beginBatch, new List[0]);
            emptyTerminationBatch.setTerminationFlag(true);
			finalizeRun();
            return emptyTerminationBatch;

        }
        // First attempt to process
        if(this.finalTupleSource == null) {
            // Still need to process - this should either
            // throw a BlockedException or return a finalTupleSource
            this.finalTupleSource = processProcedure();
        }

        // Next, attempt to return batches if processing completed
        while(! isBatchFull()) {
            // May throw BlockedException and exit here
            List<?> tuple = this.finalTupleSource.nextTuple();
            if(tuple == null) {
            	if (outParams != null) {
            		VariableContext vc = getCurrentVariableContext();
            		List<Object> paramTuple = Arrays.asList(new Object[this.getOutputElements().size()]);
            		int i = this.getOutputElements().size() - this.outParams.size();
            		for (ElementSymbol param : outParams) {
            			Object value = vc.getValue(param);
            			checkNotNull(param, value);
						paramTuple.set(i++, value);
					}
            		addBatchRow(paramTuple, true);
            	}
                terminateBatches();
                done = true;
                break;
            }
            addBatchRow(tuple, false);
        }

        return pullBatch();
    }

    /**
     * <p>Process the procedure, using the stack of Programs supplied by the
     * ProcessorEnvironment.  With each pass through the loop, the
     * current Program is gotten off the top of the stack, and the
     * current instruction is gotten from that program; each call
     * to an instruction's process method may alter the Program
     * Stack and/or the current instruction pointer of a Program,
     * so it's important that this method's loop refer to the
     * call stack of the ProcessorEnvironment each time, and not
     * cache things in local variables.  If the current Program's
     * current instruction is null, then it's time to pop that
     * Program off the stack.</p>
     *
     * @return List a single tuple containing one Integer: the update
     * count resulting from the procedure execution.
     */
    private TupleSource processProcedure()
        throws TeiidComponentException, TeiidProcessingException, BlockedException {

        // execute plan
	    ProgramInstruction inst = null;

	    while (!this.programs.empty()){
			clearExceptionTrackingData();
            Program program = peek();
            if (virtualGroup != null && virtualGroup.getSchema() != null) {
	            CommandContext commandContext = CommandContext.getThreadLocalContext();
	            if (commandContext != null)
	                commandContext.setSchema(virtualGroup.getSchema());
            }
            inst = program.getCurrentInstruction();
	        if (inst == null){
	        	LogManager.logTrace(org.teiid.logging.LogConstants.CTX_DQP, "Finished program", program); //$NON-NLS-1$
	        	//look ahead to see if we need to process in place
        		VariableContext vc = this.cursorStates.getParentContext();
        		CursorState last = (CursorState) this.cursorStates.getValue(null);
                if(last != null){
	                if (last.resultsBuffer == null) {
		                if (last.usesLocalTemp || !txnTupleSources.isEmpty()) {
			                last.returning = true;
		                } else {
			                //check to see if we are returning from a loop - the processorplan can be reused, so we need to save this result
			                //
			                //the other strategy here would be to clone the plan on invocation and use different tracking rather
			                //than the plan itself
			                for (int i = 0; i < this.programs.size() - 1; i++) {
				                if (this.programs.get(i).getCurrentInstruction() instanceof RepeatedInstruction) {
					                last.returning = true;
					                break;
				                }
			                }
		                }
		                if (last.returning) {
			                last.resultsBuffer = bufferMgr.createTupleBuffer(last.processor.getOutputElements(), getContext().getConnectionId(), TupleSourceType.PROCESSOR);
		                }
                	}
                	if (last.returning) {
	                	while (last.ts.hasNext()) {
	                		List<?> tuple = last.ts.nextTuple();
	                		last.resultsBuffer.addTuple(tuple);
	                	}
	                	last.resultsBuffer.close();
	                	last.ts = last.resultsBuffer.createIndexedTupleSource(true);
	                	last.returning = false;
                	}
                }
                this.pop(true);
                continue;
            }
	        try {
	            if (inst instanceof RepeatedInstruction) {
	    	        LogManager.logTrace(org.teiid.logging.LogConstants.CTX_DQP, "Executing repeated instruction", inst); //$NON-NLS-1$
	                RepeatedInstruction loop = (RepeatedInstruction)inst;
	                if (loop.testCondition(this)) {
	                    LogManager.logTrace(org.teiid.logging.LogConstants.CTX_DQP, "Passed condition, executing program " + loop.getNestedProgram()); //$NON-NLS-1$
	                    inst.process(this);
	                    this.push(loop.getNestedProgram());
	                    continue;
	                }
	                LogManager.logTrace(org.teiid.logging.LogConstants.CTX_DQP, "Exiting repeated instruction", inst); //$NON-NLS-1$
	                loop.postInstruction(this);
	            } else {
	            	LogManager.logTrace(org.teiid.logging.LogConstants.CTX_DQP, "Executing instruction", inst); //$NON-NLS-1$
	                inst.process(this);
		            this.evaluator.close();
	            }
	        } catch (RuntimeException e) {
	        	throw e;
	        } catch (TeiidComponentException e) {
	        	throw e;
	        } catch (Exception e) {
	        	boolean atomic = program.isAtomic();
	        	addExceptionTrackingData();
	        	//processing or teiidsqlexception
	        	try{
	        		while (program.getExceptionGroup() == null) {
		    			this.pop(false);
		        		if (this.programs.empty()) {
		        			//reached the top without a handler, so throw
			        		if (e instanceof TeiidProcessingException) {
			        			throw (TeiidProcessingException)e;
			        		}
							ProcedureErrorInstructionException procedureErrorInstructionException = new ProcedureErrorInstructionException(QueryPlugin.Event.TEIID30167, e);
			        		if (this.getVirtualGroup() != null) {
			        			procedureErrorInstructionException.setProcedureName(this.getVirtualGroup().getName());
							}
							throw procedureErrorInstructionException;
		        		}
		    			program = peek();
		    			atomic |= program.isAtomic();
		        	}
	        	} finally {
	        		DVCaller.call(128, null, null, getContext(), null, atomic, program.getExceptionGroup() != null);
	        	}
				this.pop(false); // allow the current program to go out of scope
				if (atomic) {
					TransactionContext tc = this.getContext().getTransactionContext();
					if (tc != null && tc.getTransactionType() != Scope.NONE) {
						// a non-completing atomic block under a higher level transaction

						// this will not work correctly until we support
						// checkpoints/subtransactions
						try {
							tc.getTransaction().setRollbackOnly();
						} catch (IllegalStateException e1) {
							throw new TeiidComponentException(e1);
						} catch (SystemException e1) {
							throw new TeiidComponentException(e1);
						}
					}
				}
        		if (program.getExceptionProgram() == null) {
        			continue;
        		}
	        	Program exceptionProgram = program.getExceptionProgram();
				this.push(exceptionProgram);
		        LogManager.logDetail(LogConstants.CTX_DQP, "Caught exception in exception hanlding block", e); //$NON-NLS-1$
				TeiidSQLException tse = TeiidSQLException.create(e);
				GroupSymbol gs = new GroupSymbol(program.getExceptionGroup());
				this.currentVarContext.setValue(exceptionSymbol(gs, 0), tse.getSQLState());
				this.currentVarContext.setValue(exceptionSymbol(gs, 1), tse.getErrorCode());
				this.currentVarContext.setValue(exceptionSymbol(gs, 2), tse.getTeiidCode());
				this.currentVarContext.setValue(exceptionSymbol(gs, 3), tse);
				this.currentVarContext.setValue(exceptionSymbol(gs, 4), tse.getCause());
				continue;
	        }
            program.incrementProgramCounter();
	    }

	    CursorState last = (CursorState) this.cursorStates.getValue(null);
        if(last == null){
            return CollectionTupleSource.createNullTupleSource();
        }
        return last.ts;
    }

	private void addExceptionTrackingData() {
		CommandContext context = this.getContext();
		GroupSymbol virtualGroup = this.getVirtualGroup();

		if (virtualGroup != null && context != null) {
			RequestID requestID = context.getRequestID();
			String symbolName = virtualGroup.getName();

			if (symbolName != null && requestID != null) {
				DVCaller.call(140, requestID + ":" + ProcedureErrorInstructionException.class.getName(), symbolName);
			}
		}
	}

	private void clearExceptionTrackingData() {
		CommandContext context = this.getContext();
		if (context != null) {
			RequestID requestID = context.getRequestID();
			if (requestID != null) {
				DVCaller.call(140, requestID + ":" + ProcedureErrorInstructionException.class.getName(), null);
			}
		}
	}

	private ElementSymbol exceptionSymbol(GroupSymbol gs, int pos) {
		ElementSymbol es = UpdateProcedureResolver.exceptionGroup.get(pos).clone();
		es.setGroupSymbol(gs);
		return es;
	}

    public void close()
        throws TeiidComponentException {
		if (parentSession != null) {
			DQPWorkContext.getWorkContext().setSession(parentSession);
			parentSession = null;
		}
		if (procContextPushed) {
			procContextPushed = false;
			DVProcedureContext.Store.pop();
		}
        while (!programs.isEmpty()) {
        	try {
        		pop(false);
        	} catch (TeiidComponentException e) {
        		LogManager.logDetail(LogConstants.CTX_DQP, e, "Error closing program"); //$NON-NLS-1$
        	}
        }
        if (this.evaluator != null) {
        	this.evaluator.close();
        }
		if (this.cursorStates != null) {
			removeAllCursors(this.cursorStates);
			this.cursorStates = null;
		}
        this.dataMgr = parentDataMrg;
        this.txnTupleSources.clear();
        this.blockContext = null;
        this.currentVarContext = null;
    }

    public String toString() {
        return "ProcedurePlan:\n" + this.originalProgram; //$NON-NLS-1$
    }

	public ProcessorPlan clone(){
        ProcedurePlan plan = new ProcedurePlan(originalProgram.clone(), command);
        plan.setOutputElements(this.getOutputElements());
        plan.setParams(params);
        plan.setOutParams(outParams);
        plan.setMetadata(metadata);
        plan.updateCount = updateCount;
		plan.runInContext = runInContext;
        plan.virtualGroup = this.virtualGroup;
        plan.parentSession = this.parentSession;
		plan.replace = this.replace;
        return plan;
    }

	private void addBatchRow(List<?> row, boolean last) {
        if(this.batchRows == null) {
            this.batchRows = new ArrayList<List<?>>(this.batchSize/4);
        }
        if (!last && this.outParams != null) {
        	List<Object> newRow = Arrays.asList(new Object[row.size() + this.outParams.size()]);
        	for (int i = 0; i < row.size(); i++) {
				newRow.set(i, row.get(i));
			}
        	row = newRow;
        }
        this.batchRows.add(row);
    }

    protected void terminateBatches() {
        this.lastBatch = true;
    }

    protected boolean isBatchFull() {
        return (this.batchRows != null) && (this.batchRows.size() == this.batchSize);
    }

    protected TupleBatch pullBatch() {
        TupleBatch batch = null;
        if(this.batchRows != null) {
            batch = new TupleBatch(this.beginBatch, this.batchRows);
            beginBatch += this.batchRows.size();
        } else {
            batch = new TupleBatch(this.beginBatch, Collections.EMPTY_LIST);
        }

        batch.setTerminationFlag(this.lastBatch);

        // Reset batch state
        this.batchRows = null;
        this.lastBatch = false;

        // Return batch
        return batch;
    }

    public PlanNode getDescriptionProperties() {
    	PlanNode node = this.originalProgram.getDescriptionProperties();
    	node.addProperty(PROP_OUTPUT_COLS, AnalysisRecord.getOutputColumnProperties(getOutputElements()));
    	return node;
    }

    public void setMetadata( QueryMetadataInterface metadata ) {
        this.metadata = metadata;
    }

    public void setOutParams(List<ElementSymbol> outParams) {
		this.outParams = outParams;
	}

    public void setParams( LinkedHashMap<ElementSymbol, Expression> params ) {
        this.params = params;
    }

	private void createVariableContext() {
		this.currentVarContext = new VariableContext(false);
        this.currentVarContext.setValue(ROWCOUNT, 0);
        this.cursorStates = new VariableContext(false);
     	this.cursorStates.setValue(null, null);
	}

    /**
     * <p> Get the current <code>VariavleContext</code> on this environment.
     * The VariableContext is updated with variables and their values by
     * {@link ProgramInstruction}s that are part of the ProcedurePlan that use
     * this environment.</p>
     * @return The current <code>VariariableContext</code>.
     */
    public VariableContext getCurrentVariableContext() {
		return this.currentVarContext;
    }

    /**
 	* 
 	* @param command
 	* @param rsName
 	* @param procAssignments
 	* @param mode
 	* @param usesLocalTemp - only matters in HOLD mode
 	* @throws TeiidComponentException
 	* @throws TeiidProcessingException
 	*/
 	public void executePlan(ProcessorPlan command, String rsName, Map<ElementSymbol, ElementSymbol> procAssignments, CreateCursorResultSetInstruction.Mode mode, boolean usesLocalTemp)
        throws TeiidComponentException, TeiidProcessingException {

    	CursorState state = (CursorState) this.cursorStates.getValue(rsName);
        if (state == null || rsName == null) {
        	if (this.currentState != null && this.currentState.processor.getProcessorPlan() != command) {
        		//sanity check for non-deterministic paths
        		removeState(this.currentState);
        		this.currentState = null;
        	}
        	if (this.currentState == null) {
		        //this may not be the first time the plan is being run
		        command.reset();

		        CommandContext subContext = CommandContext.newChild(getContext());
		        subContext.setVariableContext(this.currentVarContext);
		        state = new CursorState();
		        state.usesLocalTemp = usesLocalTemp;
		        state.processor = new QueryProcessor(command, subContext, this.bufferMgr, this.dataMgr);
		        state.ts = new BatchIterator(state.processor);
		        if (mode == Mode.HOLD && procAssignments != null && state.processor.getOutputElements().size() - procAssignments.size() > 0) {
		        	state.resultsBuffer = bufferMgr.createTupleBuffer(state.processor.getOutputElements().subList(0, state.processor.getOutputElements().size() - procAssignments.size()), getContext().getConnectionId(), TupleSourceType.PROCESSOR);
		        } else if ((this.blockContext != null || this.programs.peek().isTrappingExceptions()) && (mode == Mode.HOLD || rsName != null)) {
		        	state.resultsBuffer = bufferMgr.createTupleBuffer(state.processor.getOutputElements(), getContext().getConnectionId(), TupleSourceType.PROCESSOR);
		        }
	            this.currentState = state;
        	}

			//Iterate until ts.hasNext() doesn't throw BlockedException
			boolean noBlockedException;
			do {
				try {
					this.currentState.ts.hasNext();
					noBlockedException = true;
				}
				catch (QueryProcessor.ExpiredTimeSliceException e) {
					throw e;
				}
				catch (BlockedException e) {
					noBlockedException = false;
				}
			} while (!noBlockedException);

        	if (procAssignments != null) {
            	while (this.currentState.ts.hasNext()) {
            		if (this.currentState.currentRow != null && this.currentState.resultsBuffer != null) {
            			this.currentState.resultsBuffer.addTuple(this.currentState.currentRow.subList(0, this.currentState.resultsBuffer.getSchema().size()));
            			this.currentState.currentRow = null;
            		}
            		this.currentState.currentRow = this.currentState.ts.nextTuple();
            	}
            	//process assignments
            	Assertion.assertTrue(this.currentState.currentRow != null);
            	for (Map.Entry<ElementSymbol, ElementSymbol> entry : procAssignments.entrySet()) {
            		if (entry.getValue() == null || !metadata.elementSupports(entry.getValue().getMetadataID(), SupportConstants.Element.UPDATE)) {
	            		continue;
	            	}
            		int index = this.currentState.processor.getOutputElements().indexOf(entry.getKey());
            		getCurrentVariableContext().setValue(entry.getValue(), DataTypeManager.transformValue(this.currentState.currentRow.get(index), entry.getValue().getType()));
				}
            } else if (this.currentState.resultsBuffer != null) {
            	//result should be saved, typically to respect txn semantics
            	while (this.currentState.ts.hasNext()) {
            		List<?> tuple = this.currentState.ts.nextTuple();
        			this.currentState.resultsBuffer.addTuple(tuple);
            	}
            } else if (mode == Mode.UPDATE) {
        		List<?> t = this.currentState.ts.nextTuple();
        		if (this.currentState.ts.hasNext()) {
        			throw new AssertionError(QueryPlugin.Util.gs(QueryPlugin.Event.DVCORE301261));
        		}
        		removeState(this.currentState);
        		this.currentState = null;
        		int rowCount = 0;
        		if (t != null) {
        			rowCount = (Integer)t.get(0);
        		}
        		getCurrentVariableContext().setValue(ProcedurePlan.ROWCOUNT, rowCount);
        		return;
        	}
        	if (rsName == null && mode == Mode.NOHOLD) {
        		//unnamed without hold
        		//process fully, but don't save
            	//TODO: could set the rowcount in this case
            	while (this.currentState.ts.hasNext()) {
            		this.currentState.ts.nextTuple();
            	}
            	this.currentState = null;
            	return;
        	}
        	if (this.currentState.resultsBuffer != null) {
            	//close the results buffer and use a buffer backed tuplesource
            	this.currentState.resultsBuffer.close();
            	this.currentState.ts = this.currentState.resultsBuffer.createIndexedTupleSource(true);
        	}
        	CursorState old = (CursorState) this.cursorStates.setValue(rsName, this.currentState);
	        if (old != null) {
	        	removeState(old);
	        }
	        this.currentState = null;
        }
    }

    /**
     * @param success
     * @throws TeiidComponentException
     * @throws XATransactionException
     */
	public void pop(boolean success) throws TeiidComponentException {
		this.evaluator.close();
		Program program = this.programs.pop();

		VariableContext vc = this.currentVarContext;
		VariableContext cs = this.cursorStates;
		try {
			this.currentVarContext = this.currentVarContext.getParentContext();
			this.cursorStates = this.cursorStates.getParentContext();
			program.getTempTableStore().removeTempTables();
			this.getContext().setTempTableStore(getTempTableStore());
			if (program.startedTxn()) {
				TransactionService ts = this.getContext().getTransactionServer();
				TransactionContext tc = this.blockContext;
				this.blockContext = null;
				try {
					try {
						ts.resume(tc);
					} catch (IllegalStateException ise) {
					}
					for (WeakReference<DataTierTupleSource> ref : txnTupleSources) {
						DataTierTupleSource dtts = ref.get();
						if (dtts != null) {
							dtts.fullyCloseSource();
						}
					}
					this.txnTupleSources.clear();
					if (success) {
						ts.commit(tc);
					} else {
						ts.rollback(tc);
					}
				} catch (XATransactionException e) {
					throw new TeiidComponentException(QueryPlugin.Event.TEIID30165, e);
				}
			}
		} finally {
			removeAllCursors(cs);
		}
	}
	
	private void removeAllCursors(VariableContext cs) {
		if (this.currentState != null) {
			//in error situations the current state will still need cleared
			removeState(currentState);
			this.currentState = null;
		}
		for (Map.Entry<Object, Object> entry : cs.getVariableMap().entrySet()) {
			removeState((CursorState) entry.getValue());
		}
	}

    public void push(Program program) throws XATransactionException {
    	this.evaluator.close();
    	program.reset(this.getContext().getConnectionId());
		program.setTrappingExceptions(program.getExceptionGroup() != null || (!this.programs.isEmpty() && this.programs.peek().isTrappingExceptions()));
    	TempTableStore tts = getTempTableStore();
		getContext().setTempTableStore(program.getTempTableStore());
		program.getTempTableStore().setParentTempTableStore(tts);
        this.programs.push(program);
        VariableContext context = new VariableContext(true);
        context.setParentContext(this.currentVarContext);
        this.currentVarContext = context;
        VariableContext cc = new VariableContext(true);
     	cc.setParentContext(this.cursorStates);
     	this.cursorStates = cc;

        if (program.isAtomic() && this.blockContext == null) {
        	TransactionContext tc = this.getContext().getTransactionContext();
	        if (tc != null && tc.getTransactionType() == Scope.NONE) {
		        //need to toggle the atomic block flag prior to asking if txn required
		        CommandContext tlc = CommandContext.getThreadLocalContext();
		        boolean saved = false;
		        if (tlc != null) {
			        saved = tlc.isAtomicBlock();
			        tlc.setAtomicBlock(true);
		        }
		        try {
			        if (Boolean.TRUE.equals(program.instructionsRequireTransaction(false))) {
				        //start a transaction
				        this.getContext().getTransactionServer().begin(tc);
				        this.blockContext = tc;
				        program.setStartedTxn(true);
			        }
		        } finally {
			        if (tlc != null) {
				        tlc.setAtomicBlock(saved);
			        }
		        }
	        }
        }
    }

    public void incrementProgramCounter() throws TeiidComponentException {
    	if (this.programs.isEmpty()) {
    		return;
    	}
        Program program = peek();
        ProgramInstruction instr = program.getCurrentInstruction();
        if (instr instanceof RepeatedInstruction) {
            RepeatedInstruction repeated = (RepeatedInstruction)instr;
            repeated.postInstruction(this);
        }
        peek().incrementProgramCounter();
    }

    public List<?> getCurrentRow(String rsName) throws TeiidComponentException {
        return getCursorState(rsName).currentRow;
    }

    public boolean iterateCursor(String rsName)
        throws TeiidComponentException, TeiidProcessingException {

        CursorState state = getCursorState(rsName);

        state.currentRow = state.ts.nextTuple();
        return (state.currentRow != null);
    }

	private CursorState getCursorState(String rsKey) throws TeiidComponentException {
		CursorState state = (CursorState) this.cursorStates.getValue(rsKey);
		if (state == null) {
			 throw new TeiidComponentException(QueryPlugin.Event.TEIID30166, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID30166, rsKey));
		}
		return state;
	}

    public void removeResults(String rsName) {
    	CursorState state = (CursorState) this.cursorStates.remove(rsName);
        removeState(state);
    }

	private void removeState(CursorState state) {
		if (state != null) {
        	state.processor.closeProcessing();
        	if (state.resultsBuffer != null) {
        		state.resultsBuffer.remove();
        	}
        }
	}

    /**
     * Get the schema from the tuple source that
     * represents the columns in a result set
     * @param rsName the ResultSet name (not a temp group)
     * @return List of elements
     * @throws QueryProcessorException if the list of elements is null
     */
    public List getSchema(String rsName) throws TeiidComponentException {
        CursorState cursorState = getCursorState(rsName);
        // get the schema from the tuple source
        List schema = cursorState.processor.getOutputElements();
        return schema;
    }

    public boolean resultSetExists(String rsName) {
    	return this.cursorStates.containsVariable(rsName);
    }

    public CommandContext getContext() {
    	CommandContext context = super.getContext();
    	if (evaluatedParams) {
    		context.setVariableContext(currentVarContext);
    	}
    	return context;
    }

    public List getOutputElements() {
		return outputElements;
	}

	public void setOutputElements(List outputElements) {
		this.outputElements = outputElements;
	}

    /**
     * @return Returns the tempTableStore.
     * @since 5.5
     */
    public TempTableStore getTempTableStore() {
    	if (this.programs.isEmpty()) {
			if (runInContext && params == null) {
				return getContext().getTempTableStore();
			}
    		return null;
    	}
        return this.peek().getTempTableStore();
    }

    boolean evaluateCriteria(Criteria condition) throws BlockedException, TeiidProcessingException, TeiidComponentException {
    	evaluator.initialize(getContext(), getDataManager());
		return evaluator.evaluate(condition, Collections.emptyList());
    }

    Object evaluateExpression(Expression expression) throws BlockedException, TeiidProcessingException, TeiidComponentException {
    	evaluator.initialize(getContext(), getDataManager());
    	return evaluator.evaluate(expression, Collections.emptyList());
    }

    public Program peek() {
	    if (this.programs.isEmpty()) {
		    return null;
	    }
        return programs.peek();
    }

    public void setUpdateCount(int updateCount) {
     	this.updateCount = updateCount;
    }

	@Override
	public Boolean requiresTransaction(boolean transactionalReads) {
		Boolean paramRequires = false;
		if (params != null) {
			paramRequires = SubqueryAwareRelationalNode.requiresTransaction(transactionalReads, ValueIteratorProviderCollectorVisitor.getValueIteratorProviders(params.values()));
			if (paramRequires != null && paramRequires) {
				return true;
			}
		}
		if ((transactionalReads && updateCount < 2) || updateCount == Procedure.AUTO_UPDATECOUNT) {
			Boolean programRequires = this.originalProgram.requiresTransaction(transactionalReads);
			if (programRequires == null) {
				return paramRequires == null ? true : null;
			}
			if (programRequires) {
				return true;
			}
			return paramRequires;
		}
		if (updateCount == 0) {
			return paramRequires;
		}
		if (updateCount == 1) {
			return paramRequires == null ? true : null;
		}
		return true;
	}

	/**
	 * For procedures without explicit parameters, sets whether the
	 * procedure should run in the parent variable context.
	 * @param runInContext
	 */
	public void setRunInContext(boolean runInContext) {
		this.runInContext = runInContext;
	}

	public void setValidateAccess(boolean b) {
		this.validateAccess = b;
	}

	public boolean isValidateAccess() {
		return validateAccess;
	}

	public GroupSymbol getVirtualGroup() {
		return virtualGroup;
	}

	public void setVirtualGroup(GroupSymbol virtualGroup) {
		this.virtualGroup = virtualGroup;
	}
}
