/*
 * 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-2496
// https://issues.jboss.org/browse/TEIID-2499
// https://issues.jboss.org/browse/TEIID-2582
// https://issues.jboss.org/browse/TEIID-2672
// https://issues.jboss.org/browse/TEIID-3046
// https://issues.jboss.org/browse/TEIID-4076
// https://issues.jboss.org/browse/TEIID-4135
// https://issues.jboss.org/browse/TEIID-2381
// https://issues.jboss.org/browse/TEIID-5004
package org.teiid.query.processor.relational;

import java.util.HashMap;
import java.util.List;

import org.teiid.client.plan.PlanNode;
import org.teiid.common.buffer.BlockedException;
import org.teiid.common.buffer.BufferManager;
import org.teiid.common.buffer.TupleBatch;
import org.teiid.common.buffer.TupleBuffer;
import org.teiid.core.TeiidComponentException;
import org.teiid.core.TeiidProcessingException;
import org.teiid.language.SQLConstants;
import org.teiid.query.analysis.AnalysisRecord;
import org.teiid.query.processor.ProcessorDataManager;
import org.teiid.query.processor.ProcessorPlan;
import org.teiid.query.processor.QueryProcessor;
import org.teiid.query.sql.LanguageObject;
import org.teiid.query.sql.lang.SetQuery;
import org.teiid.query.sql.lang.WithQueryCommand;
import org.teiid.query.sql.symbol.Expression;
import org.teiid.query.tempdata.TempTableStore;
import org.teiid.query.tempdata.TempTableStore.RecursiveTableProcessor;
import org.teiid.query.tempdata.TempTableStore.TableProcessor;
import org.teiid.query.tempdata.TempTableStore.TransactionMode;
import org.teiid.query.util.CommandContext;

/**
 */
public class RelationalPlan extends ProcessorPlan {

	// Initialize state - don't reset
	private RelationalNode root;
	private List<? extends Expression> outputCols;
	private List<WithQueryCommand> with;
	
	private TempTableStore tempTableStore;
	private boolean multisourceUpdate;

    /**
     * Constructor for RelationalPlan.
     */
    public RelationalPlan(RelationalNode node) {
        this.root = node;
    }

    public RelationalNode getRootNode() {
        return this.root;
    }
    
    public void setRootNode(RelationalNode root) {
        this.root = root;
    }
    
    public void setWith(List<WithQueryCommand> with) {
		this.with = with;
	}
    
    /**
     * @see ProcessorPlan#connectDataManager(ProcessorDataManager)
     */
    public void initialize(CommandContext context, ProcessorDataManager dataMgr, BufferManager bufferMgr) {
        if (this.with != null) {
    		context = context.clone();
    		tempTableStore = new TempTableStore(context.getConnectionId(), TransactionMode.NONE, true);
            tempTableStore.setParentTempTableStore(context.getTempTableStore());
            tempTableStore.setRemoveParentTables(false);
            context.setTempTableStore(tempTableStore);
        } 
    	setContext(context);
        connectExternal(this.root, context, dataMgr, bufferMgr);	
    }

	private void connectExternal(RelationalNode node, CommandContext context, ProcessorDataManager dataMgr, BufferManager bufferMgr) {		

        node.initialize(context, bufferMgr, dataMgr);

        RelationalNode[] children = node.getChildren();  
        for(int i=0; i<children.length; i++) {
            if(children[i] != null) {
                connectExternal(children[i], context, dataMgr, bufferMgr);                
            } else {
                break;
            }
        }
    }
    
    /**
     * Get list of resolved elements describing output columns for this plan.
     * @return List of SingleElementSymbol
     */
    public List<? extends Expression> getOutputElements() {
        return this.outputCols;
    }

    // EXTENSION
    // ISSUE: DVCORE-1078
    // FIXED BY: Roman Pedchenko
    // DATE: 2014.03.26
    // ADOPTED FROM: https://issues.jboss.org/browse/TEIID-2141
    @Override
    // END EXTENSION
    public void open()
        throws TeiidComponentException, TeiidProcessingException {
    	// EXTENSION
		// ISSUE: DV-69
		// FIXED BY: Roman Pedchenko
		// DATE: 2013.04.30
    	// the fix adopted from https://issues.jboss.org/browse/TEIID-2442
    if (with != null && tempTableStore.getProcessors() == null) {
//  if (with != null) {
	    	HashMap<String, TableProcessor> processors = new HashMap<String, TableProcessor>();
	        tempTableStore.setProcessors(processors);
			for (WithQueryCommand withCommand : this.with) {
				if (withCommand.isRecursive()) {
					SetQuery setQuery = (SetQuery)withCommand.getCommand();
					ProcessorPlan initial = setQuery.getLeftQuery().getProcessorPlan();
					QueryProcessor withProcessor = new QueryProcessor(initial, getContext().clone(), root.getBufferManager(), root.getDataManager());
					processors.put(withCommand.getGroupSymbol().getName(), new RecursiveTableProcessor(withProcessor, withCommand.getColumns(), setQuery.getRightQuery().getProcessorPlan(), setQuery.isAll()));
					continue;
				}
				ProcessorPlan plan = withCommand.getCommand().getProcessorPlan();
				QueryProcessor withProcessor = new QueryProcessor(plan, getContext().clone(), root.getBufferManager(), root.getDataManager()); 
				processors.put(withCommand.getGroupSymbol().getName(), new TableProcessor(withProcessor, withCommand.getColumns()));
			}
    	}
    	// END EXTENSION
        this.root.open();
    }

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

		final TupleBatch tupleBatch = this.root.nextBatch();
		if (tupleBatch.getTerminationFlag())
			finalizeRun();
		return tupleBatch;
    }

    public void close()
        throws TeiidComponentException {
		if (this.tempTableStore != null) {
			this.tempTableStore.removeTempTables();
			if (this.tempTableStore.getProcessors() != null) {
    			for (TableProcessor proc : this.tempTableStore.getProcessors().values()) {
				    proc.close();
    			}
    			this.tempTableStore.setProcessors(null);
			}
		}
        this.root.close();
    }

    /**
     * @see org.teiid.query.processor.ProcessorPlan#reset()
     */
    public void reset() {
        super.reset();
        
        this.root.reset();
        if (this.with != null) {
        	for (WithQueryCommand withCommand : this.with) {
		        if (withCommand.isRecursive()) {
			        SetQuery setQuery = (SetQuery)withCommand.getCommand();
			        setQuery.getLeftQuery().getProcessorPlan().reset();
			        setQuery.getLeftQuery().getProcessorPlan().reset();
			    } else {
			        withCommand.getCommand().getProcessorPlan().reset();
		        }
			}
        }
    }

	public String toString() {
		StringBuilder sb = new StringBuilder();
		if (this.with != null) {
			sb.append(SQLConstants.Reserved.WITH);
			for (WithQueryCommand withCommand : this.with) {
				sb.append("\n"); //$NON-NLS-1$
				sb.append(withCommand.getGroupSymbol());
				if (withCommand.isRecursive()) {
					sb.append(" anchor\n").append(((SetQuery)withCommand.getCommand()).getLeftQuery().getProcessorPlan()); //$NON-NLS-1$
					sb.append("recursive\n").append(((SetQuery)withCommand.getCommand()).getRightQuery().getProcessorPlan()); //$NON-NLS-1$
				} else {
					sb.append("\n"); //$NON-NLS-1$
				sb.append(withCommand.getCommand().getProcessorPlan());
			}
		}
			sb.append("body\n"); //$NON-NLS-1$
		}
		sb.append(this.root.toString());
		return sb.toString();
	}
    
	public RelationalPlan clone(){
		RelationalPlan plan = new RelationalPlan((RelationalNode)root.clone());
		plan.setOutputElements(outputCols);
		if (with != null) {
			List<WithQueryCommand> newWith = LanguageObject.Util.deepClone(this.with, WithQueryCommand.class);
			for (WithQueryCommand withQueryCommand : newWith) {
				if (withQueryCommand.isRecursive()) {
					SetQuery setQuery = (SetQuery)withQueryCommand.getCommand();
					setQuery.getLeftQuery().setProcessorPlan(setQuery.getLeftQuery().getProcessorPlan().clone());
					setQuery.getRightQuery().setProcessorPlan(setQuery.getRightQuery().getProcessorPlan().clone());
				} else {
					withQueryCommand.getCommand().setProcessorPlan(withQueryCommand.getCommand().getProcessorPlan().clone());
				}
			}
			plan.setWith(newWith);
		}
		return plan;
	}
	
    public PlanNode getDescriptionProperties() {
    	PlanNode node = this.root.getDescriptionProperties();
    	if (this.with != null) {
    		AnalysisRecord.addLanaguageObjects(node, AnalysisRecord.PROP_WITH, this.with);
        }
    	return node;
    }
    
    /** 
     * @param outputCols The outputCols to set.
     */
    public void setOutputElements(List<? extends Expression> outputCols) {
        this.outputCols = outputCols;
    }
    
    public void setMultisourceUpdate(boolean multisourceUpdate) {
		this.multisourceUpdate = multisourceUpdate;
	}
    
    @Override
    public Boolean requiresTransaction(boolean transactionalReads) {
    	boolean txnWtih = false;
    	if (multisourceUpdate) {
    		return true;
    	}
    	if (this.with != null) {
    		for (WithQueryCommand withCommand : this.with) {
			    if (withCommand.isRecursive()) {
				    SetQuery setQuery = (SetQuery)withCommand.getCommand();
				    Boolean leftRequires = setQuery.getLeftQuery().getProcessorPlan().requiresTransaction(transactionalReads);
				 	Boolean rightRequires = setQuery.getLeftQuery().getProcessorPlan().requiresTransaction(transactionalReads); 
				 	if (Boolean.TRUE.equals(leftRequires) || Boolean.TRUE.equals(rightRequires)) {
					    return true;
				    }
				 	if (leftRequires == null || rightRequires == null) {
				 	 	if (txnWtih) {
				 	 		return true;
				 	 	}
				 	 	txnWtih = true;
				 	}
			    } else {
					Boolean requires = withCommand.getCommand().getProcessorPlan().requiresTransaction(transactionalReads);
					if (Boolean.TRUE.equals(requires)) {
						return true;
					}
					if (requires == null) {
						if (txnWtih) {
							return true;
						}
						txnWtih = true;
					}
				}
			}
    	}
		Boolean requires = requiresTransaction(transactionalReads, root);
		if (Boolean.TRUE.equals(requires)) {
			return true;
		}
		if (requires == null && txnWtih) {
			return true;
		}
		if (txnWtih || requires == null) {
			return null;
		}
		return false;
	}
    
    static Boolean requiresTransaction(boolean transactionalReads, RelationalNode node) {
    	Boolean requiresTxn = node.requiresTransaction(transactionalReads);
    	if (Boolean.TRUE.equals(requiresTxn)) {
    		return true;
    	}
		for (RelationalNode child : node.getChildren()) {
			if (child == null) {
				continue;
			}
			Boolean childRequires = requiresTransaction(transactionalReads, child);
			if (Boolean.TRUE.equals(childRequires)) {
				return true;
			}
			if (childRequires == null) {
				if (requiresTxn == null) {
					return true;
				}
				requiresTxn = null;
			}
		}
		return requiresTxn;
    }
    
    @Override
    public TupleBuffer getBuffer(int maxRows) throws BlockedException, TeiidComponentException, TeiidProcessingException {
    	return root.getBuffer(maxRows);
    }
    
    @Override
    public boolean hasBuffer(boolean requireFinal) {
    	return root.hasBuffer(requireFinal);
    }

}
