/*
 * 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-2153
// https://issues.jboss.org/browse/TEIID-2171
// https://issues.jboss.org/browse/TEIID-2359
// https://issues.jboss.org/browse/TEIID-2149
// https://issues.jboss.org/browse/TEIID-2358
// https://issues.jboss.org/browse/TEIID-2410
// https://issues.jboss.org/browse/TEIID-2271
// https://issues.jboss.org/browse/TEIID-2467
// https://issues.jboss.org/browse/TEIID-2429
// https://issues.jboss.org/browse/TEIID-2363
// https://issues.jboss.org/browse/TEIID-2160
// https://issues.jboss.org/browse/TEIID-2461
// https://issues.jboss.org/browse/TEIID-2797
// https://issues.jboss.org/browse/TEIID-2326
// https://issues.jboss.org/browse/TEIID-1557
// https://issues.jboss.org/browse/TEIID-3781
// https://issues.jboss.org/browse/TEIID-2352
// https://issues.jboss.org/browse/TEIID-3591
// https://issues.jboss.org/browse/TEIID-2404
// https://issues.jboss.org/browse/TEIID-2408
// https://issues.jboss.org/browse/TEIID-2661
// https://issues.jboss.org/browse/TEIID-1092
// https://issues.jboss.org/browse/TEIID-2507
// https://issues.jboss.org/browse/TEIID-5427
// https://issues.redhat.com/browse/TEIID-4129
package org.teiid.dqp.internal.process;

import com.datavirtuality.dv.core.interfaces.DVCache;
import java.lang.ref.WeakReference;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

import javax.crypto.SecretKey;
import javax.transaction.xa.Xid;

import com.datavirtuality.dv.core.services.DVServices;
import org.teiid.adminapi.AdminException;
import org.teiid.adminapi.Request.ProcessingState;
import org.teiid.adminapi.Request.ThreadState;
import org.teiid.adminapi.VDB.Status;
import org.teiid.adminapi.impl.RequestMetadata;
import org.teiid.adminapi.impl.TransactionMetadata;
import org.teiid.adminapi.impl.WorkerPoolStatisticsMetadata;
import org.teiid.cache.Cache;
import org.teiid.cache.CacheFactory;
import org.teiid.cache.ConcurrentMapCache;
import org.teiid.client.DQP;
import org.teiid.client.RequestMessage;
import org.teiid.client.RequestMessage.StatementType;
import org.teiid.client.ResultsMessage;
import org.teiid.client.lob.LobChunk;
import org.teiid.client.metadata.MetadataResult;
import org.teiid.client.plan.PlanNode;
import org.teiid.client.util.ResultsFuture;
import org.teiid.client.util.ResultsReceiver;
import org.teiid.client.xa.XATransactionException;
import org.teiid.client.xa.XidImpl;
import org.teiid.common.buffer.BufferManager;
import org.teiid.core.TeiidComponentException;
import org.teiid.core.TeiidProcessingException;
import org.teiid.core.TeiidRuntimeException;
import org.teiid.core.types.Streamable;
import org.teiid.core.util.ApplicationInfo;
import org.teiid.core.util.ExecutorUtils;
import org.teiid.core.util.PropertiesUtils;
import org.teiid.dqp.message.AtomicRequestMessage;
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.EnhancedTimer;
import org.teiid.logging.CommandLogMessage;
import org.teiid.logging.CommandLogMessage.Event;
import org.teiid.logging.LogConstants;
import org.teiid.logging.LogManager;
import org.teiid.logging.MessageLevel;
import org.teiid.query.QueryPlugin;
import org.teiid.query.processor.QueryProcessor;
import org.teiid.query.tempdata.TempTableDataManager;
import org.teiid.query.tempdata.TempTableStore;
import org.teiid.query.tempdata.TempTableStore.TransactionMode;
import org.teiid.query.util.DVQueryMonitorIntf;
import org.teiid.query.util.DVUsersHandlerIntf;
import org.teiid.query.util.Options;

import com.datavirtuality.dv.core.interfaces.DVCaller;

/**
 * Implements the core DQP processing.
 */
public class DQPCore implements DQP {
	public static final String USER_QUERIES_LIMIT = "User %s has a limit to %d concurrent queries, but he is already running %d queries. The current query will be canceled";

	public interface CompletionListener<T> {
		void onCompletion(FutureWork<T> future);
	}

	private TeiidExecutor processWorkerPool;
    
    // Resources
    private BufferManager bufferManager;
    private TempTableDataManager dataTierMgr;
    private SessionAwareCache<PreparedPlan> prepPlanCache;
    private SessionAwareCache<CachedResults> rsCache;
	private CacheFactory dvCacheFactory;
    private TransactionService transactionService;
    private EventDistributor eventDistributor;
    
    private DQPConfiguration config = new DQPConfiguration();
    
    private int chunkSize = Streamable.STREAMING_BATCH_SIZE_IN_BYTES;
    
	private final Map<RequestID, RequestWorkItem> requests = new ConcurrentHashMap<>();
	private final Map<String, ClientState> clientState = new ConcurrentHashMap<>();
    
    private int maxActivePlans = DQPConfiguration.DEFAULT_MAX_ACTIVE_PLANS;
    private int currentlyActivePlans;

    private int userRequestSourceConcurrency;
	private final LinkedList<RequestWorkItem> waitingPlans = new LinkedList<>();
	private int maxWaitingPlans = 0;
	private AuthorizationValidator authorizationValidator;

	private EnhancedTimer cancellationTimer;
	private Options options;

	private final DQPRunningCount userActivePlans = new DQPRunningCount();

    /**
     * perform a full shutdown and wait for 10 seconds for all threads to finish
     */
    public void stop() {
    	processWorkerPool.shutdownNow();
    	try {
			processWorkerPool.awaitTermination(10, TimeUnit.SECONDS);
		} catch (InterruptedException ignored) {
		}
    	// TODO: Should we be doing more cleanup here??
		LogManager.logDetail(LogConstants.CTX_DQP, "Stopping the DQP"); //$NON-NLS-1$
    }
    
    /**
     * Return a list of {@link RequestMetadata} for the given session
     */
    public List<RequestMetadata> getRequestsForSession(String sessionId) {
    	ClientState state = getClientState(sessionId, false);
    	if (state == null) {
    		return Collections.emptyList();
    	}
        return buildRequestInfos(state.getRequests(), -1);
    }
    
    public ClientState getClientState(String key, boolean create) {
    	if (key == null) {
    		return null;
    	}
		ClientState state = clientState.get(key);
		if (state == null && create) {
			state = new ClientState(new TempTableStore(key, TransactionMode.ISOLATE_WRITES));
    		clientState.put(key, state);
		}
		return state;
    }

    /**
     * Return a list of all {@link RequestMetadata} 
     */
    public List<RequestMetadata> getRequests() {
		return buildRequestInfos(requests.keySet(), -1);
    } 
    
    public List<RequestMetadata> getLongRunningRequests(){
    	return buildRequestInfos(requests.keySet(), System.currentTimeMillis() - this.config.getQueryThresholdInMilli() );
    }

	public Collection<RequestID> getRequestIds() {
		return requests.keySet();
	}

    private List<RequestMetadata> buildRequestInfos(Collection<RequestID> ids, long longRunningQueryThreshold) {
		List<RequestMetadata> results = new ArrayList<>();
    	
		for (RequestID requestID : ids) {
            RequestWorkItem holder = requests.get(requestID);
            
            if(holder != null && !holder.isCanceled() && (longRunningQueryThreshold == -1 || holder.getProcessingTimestamp() < longRunningQueryThreshold)) {
            	RequestMetadata req = new RequestMetadata();
            	
            	req.setExecutionId(holder.requestID.getExecutionID());
            	req.setSessionId(holder.requestID.getConnectionID());
            	req.setCommand(holder.requestMsg.getCommandString());
            	req.setStartTime(holder.getProcessingTimestamp());
            	req.setState(getWorkItemState(holder));
	            req.setThreadState(getWorkItemThreadState(holder));

            	if (holder.getTransactionContext() != null && holder.getTransactionContext().getTransactionType() != Scope.NONE) {
            		req.setTransactionId(holder.getTransactionContext().getTransactionId());
            	}

                for (DataTierTupleSource conInfo : holder.getConnectorRequests()) {
                    String connectorName = conInfo.getConnectorName();

                    if (connectorName == null) {
                    	continue;
                    }
                    // If the request has not yet completed processing, then
                    // add all the subrequest messages
                	AtomicRequestMessage arm = conInfo.getAtomicRequestMessage();
                	RequestMetadata info = new RequestMetadata();
	                info.setThreadState(getConnectorWorkItemThreadState(conInfo));
                	info.setExecutionId(arm.getRequestID().getExecutionID());
                	info.setSessionId(holder.requestID.getConnectionID());
                	info.setCommand(arm.getCommand().toString());
                	info.setStartTime(arm.getProcessingTimestamp());
                	info.setSourceRequest(true);
                	info.setNodeId(arm.getAtomicRequestID().getNodeID());
                	info.setState(getConnectorWorkItemState(conInfo));
        			results.add(info);
                }
                
            	results.add(req);
            }
        }
    	return results;
	}    

	public ResultsFuture<ResultsMessage> executeRequest(long reqID,RequestMessage requestMsg) throws TeiidProcessingException {
		DQPWorkContext workContext = DQPWorkContext.getWorkContext();
    	checkActive(workContext);
		RequestID requestID = workContext.getRequestID(reqID);
		requestMsg.setFetchSize(Math.min(requestMsg.getFetchSize(), this.config.getMaxRowsFetchSize()));
		Request request;
	    if (requestMsg.isPreparedStatement() || requestMsg.isCallableStatement() || requestMsg.getRequestOptions().isContinuous()) {
	    	request = new PreparedStatementRequest(prepPlanCache);
	    } else {
	    	request = new Request();
	    }
	    ClientState state = this.getClientState(workContext.getSessionId(), true);
		if (state.session == null) {
			state.session = workContext.getSession();
		}
	    request.initialize(requestMsg, bufferManager,
				dataTierMgr, transactionService, state.sessionTables,
				workContext, this.prepPlanCache);
		request.setOptions(options);
	    request.setExecutor(this.processWorkerPool);
		request.setResultSetCacheEnabled(this.rsCache != null);
		request.setAuthorizationValidator(this.authorizationValidator);
		request.setUserRequestConcurrency(this.getUserRequestSourceConcurrency());
		ResultsFuture<ResultsMessage> resultsFuture = new ResultsFuture<>();
		DVCaller.call(32, resultsFuture, requestID);

        @SuppressWarnings("unchecked")
        ResultsFuture<ResultsMessage> resultsFuture2 = (ResultsFuture<ResultsMessage>) DVCaller.callWithThrow(TeiidProcessingException.class, 1, requestMsg, prepPlanCache, requestID, resultsFuture);
        DVCaller.call(113, requestMsg, request, state);
        if (resultsFuture2 != null)
        	return resultsFuture2;
		final RequestWorkItem workItem = new RequestWorkItem(this, requestMsg, request, resultsFuture.getResultsReceiver(), requestID, workContext, resultsFuture);
    	logMMCommand(workItem, Event.NEW, null, resultsFuture, null);
        addRequest(requestID, workItem, state);
     
        long timeout = workContext.getVDB().getQueryTimeout();
        timeout = Math.min(timeout>0?timeout:Long.MAX_VALUE, config.getQueryTimeout()>0?config.getQueryTimeout():Long.MAX_VALUE);
        if (timeout < Long.MAX_VALUE) {
        	final long finalTimeout = timeout;
        	workItem.setCancelTask(this.cancellationTimer.add(new Runnable() {
				final WeakReference<RequestWorkItem> workItemRef = new WeakReference<>(workItem);
				@Override
				public void run() {
					try {
						RequestWorkItem wi = workItemRef.get();
						if (wi != null) {
							LogManager.logInfo(LogConstants.CTX_DQP, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID31096, wi.requestID, finalTimeout));
							wi.requestCancel();
						}
					} catch (TeiidComponentException e) {
						LogManager.logError(LogConstants.CTX_DQP, e, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID30018));
					}
				}
			}, timeout));
        }
        boolean runInThread = requestMsg.isSync();
        synchronized (waitingPlans) {
            checkUserLimits(requestID, workContext.getSession().getUserName()).check(workItem);
			if (runInThread || currentlyActivePlans <= maxActivePlans) {
				startActivePlan(workItem, !runInThread);
			} else {
				if (LogManager.isMessageToBeRecorded(LogConstants.CTX_DQP, MessageLevel.INFO)) {
		            LogManager.logInfo(LogConstants.CTX_DQP, workItem.requestID + " " + "Queuing plan, since max plans has been reached.");  //$NON-NLS-1$
		        }  
				waitingPlans.add(workItem);
				maxWaitingPlans = Math.max(this.maxWaitingPlans, waitingPlans.size());
			}
		}
        
        if (runInThread) {
        	workItem.useCallingThread = true;
        	workItem.run();
        }
        return resultsFuture;
    }

	private LimitChecker checkUserLimits(RequestID rid, String userName) {
		UserLimits ul = DVServices.callne(DVUsersHandlerIntf.class, s -> s.getUserLimits(userName), null);
		if (ul != null) {
			final int limitForUser = ul.getConcurrentQueries();
			if (limitForUser >= 0) {
				final int countForUser = userActivePlans.get(userName);
                if (countForUser >= limitForUser)
					return new LimitChecker(rid, () -> String.format(USER_QUERIES_LIMIT, userName, limitForUser, countForUser));
			}
		}
		return new LimitChecker();
	}

	private static class LimitChecker {
		private final RequestID rid;
		private final boolean hasError;
		private final Supplier<String> message;

		public LimitChecker() {
			rid = null;
			this.hasError = false;
			this.message = () -> "";
		}

		public LimitChecker(RequestID rid, Supplier<String> message) {
            this.rid = rid;
            this.hasError = true;
			this.message = message;
		}

		public boolean ok() {
			return !hasError;
		}

		public void check(RequestWorkItem workItem) throws TeiidProcessingException {
			if (hasError) {
				final String message1 = message.get();
				DVServices.callne(DVQueryMonitorIntf.class, s -> {
					s.markQueryCancelled(rid, message1);
					workItem.close();
					return null;
				}, null);
				throw new TeiidProcessingException(message1);
			}
		}
	}

	public ResultsFuture<ResultsMessage> processCursorRequest(long reqID,
			int batchFirst, int fetchSize) throws TeiidProcessingException {
        if (LogManager.isMessageToBeRecorded(LogConstants.CTX_DQP, MessageLevel.DETAIL)) {
            LogManager.logDetail(LogConstants.CTX_DQP, DQPWorkContext.getWorkContext().getRequestID(reqID) + " DQP process cursor request from " + batchFirst);  //$NON-NLS-1$			
        }
		DQPWorkContext workContext = DQPWorkContext.getWorkContext();
        ResultsFuture<ResultsMessage> resultsFuture = new ResultsFuture<>();
		RequestWorkItem workItem = getRequestWorkItem(workContext.getRequestID(reqID));
		workItem.requestMore(batchFirst, batchFirst + Math.min(fetchSize, this.config.getMaxRowsFetchSize()) - 1, resultsFuture.getResultsReceiver());
		return resultsFuture;
	}

	void addRequest(RequestID requestID, RequestWorkItem workItem, ClientState state) {
		this.requests.put(requestID, workItem);
		state.addRequest(requestID);
	}

	private void startActivePlan(RequestWorkItem workItem, boolean addToQueue) {
		boolean continuous = workItem.requestMsg.getRequestOptions().isContinuous();
		workItem.active = !continuous;
		if (addToQueue) {
			this.addWork(workItem);
		}
		if (!continuous) {
			this.currentlyActivePlans++;
			this.userActivePlans.inc(workItem.getDqpWorkContext().getUserName());
		}
	}
	
    void finishProcessing(final RequestWorkItem workItem) {
    	synchronized (waitingPlans) {
    		if (!workItem.active) {
        		return;
        	}
        	workItem.active = false;
    		currentlyActivePlans--;
			this.userActivePlans.dec(workItem.getDqpWorkContext().getUserName());
			if (!waitingPlans.isEmpty()) {
				Iterator<RequestWorkItem> it = waitingPlans.iterator();
				while (it.hasNext()) {
					RequestWorkItem work = it.next();
					if (checkUserLimits(work.requestID, work.getDqpWorkContext().getUserName()).ok()) {
						it.remove();
				startActivePlan(work, true);
						break;
			}
		}
    }
		}
    }

	public int getActivePlanCount() {
		return this.currentlyActivePlans;
	}
    
    public boolean blockOnOutputBuffer(RequestWorkItem item) {
    	synchronized (waitingPlans) {
    		if (!waitingPlans.isEmpty()) {
    			return false;
    		}
    		if (item.useCallingThread || item.getDqpWorkContext().getSession().isEmbedded()) {
    			return false;
    		}
		}
    	return true;
    }

	public int getWaitingPlanCount() {
		return waitingPlans.size();
	}

	public int getMaxWaitingPlanWatermark() {
		return this.maxWaitingPlans;
	}
    
    void removeRequest(final RequestWorkItem workItem) {
    	finishProcessing(workItem);
        DVCaller.call(94, workItem.requestID);
    	this.requests.remove(workItem.requestID);
    	ClientState state = getClientState(workItem.getDqpWorkContext().getSessionId(), false);
    	if (state != null) {
    		state.removeRequest(workItem.requestID);
    	}
    }
    
    void addWork(Runnable work) {
		this.processWorkerPool.execute(work);
    }
    
    Future<Void> scheduleWork(final Runnable r, long delay) {
    	return this.cancellationTimer.add(r, delay);
    }
    
	public ResultsFuture<?> closeLobChunkStream(int lobRequestId,
			long requestId, String streamId) {
        if (LogManager.isMessageToBeRecorded(LogConstants.CTX_DQP, MessageLevel.DETAIL)) {
            LogManager.logDetail(LogConstants.CTX_DQP, "Request to close the Lob stream with Stream id="+streamId+" instance id="+lobRequestId);  //$NON-NLS-1$//$NON-NLS-2$
        }   
        DQPWorkContext workContext = DQPWorkContext.getWorkContext();
        RequestWorkItem workItem = safeGetWorkItem(workContext.getRequestID(requestId));
        if (workItem != null) {
	        workItem.removeLobStream(lobRequestId);
        }
        return ResultsFuture.NULL_FUTURE;
    }
	    
	public ResultsFuture<LobChunk> requestNextLobChunk(int lobRequestId,
			long requestId, String streamId)
			throws TeiidProcessingException {
        if (LogManager.isMessageToBeRecorded(LogConstants.CTX_DQP, MessageLevel.DETAIL)) {
            LogManager.logDetail(LogConstants.CTX_DQP, "Request for next Lob chunk with Stream id="+streamId+" instance id="+lobRequestId);  //$NON-NLS-1$//$NON-NLS-2$
        }  
        RequestWorkItem workItem = getRequestWorkItem(DQPWorkContext.getWorkContext().getRequestID(requestId));
        ResultsFuture<LobChunk> resultsFuture = new ResultsFuture<>();
        workItem.processLobChunkRequest(streamId, lobRequestId, resultsFuture.getResultsReceiver());
        return resultsFuture;
    }
    
    RequestWorkItem getRequestWorkItem(RequestID reqID) throws TeiidProcessingException {
    	RequestWorkItem result = this.requests.get(reqID);
    	if (result == null) {
    		 throw new TeiidProcessingException(QueryPlugin.Event.TEIID30495, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID30495, reqID));
    	}
    	return result;
    }
    
	public RequestWorkItem safeGetWorkItem(Object processorID) {
        //noinspection SuspiciousMethodCalls
    	return this.requests.get(processorID);
	}
	
    public WorkerPoolStatisticsMetadata getWorkerPoolStatistics() {
    	return this.processWorkerPool.getStats();
    }

	@Override
	public void terminateSession(String sessionId) {
        // sometimes there will not be any atomic requests pending, in that
        // situation we still need to clear the master request from our map
        ClientState state = this.clientState.remove(sessionId);
		if (state != null) {
			for (RequestID reqId : state.getRequests()) {
				try {
					cancelRequest(reqId);
				} catch (TeiidComponentException err) {
					LogManager.logWarning(LogConstants.CTX_DQP, err, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID30026,reqId));
				}
			}
		}
        try {
            transactionService.cancelTransactions(sessionId, false);
        } catch (XATransactionException err) {
            LogManager.logWarning(LogConstants.CTX_DQP, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID30005,sessionId));
        } 
    }

    public boolean cancelRequest(String sessionId, long executionId) throws TeiidComponentException {
    	RequestID requestID = new RequestID(sessionId, executionId);
    	return cancelRequest(requestID);
    }
    
    public PlanNode getPlan(String sessionId, long executionId) {
    	RequestID requestID = new RequestID(sessionId, executionId);
        if (LogManager.isMessageToBeRecorded(LogConstants.CTX_DQP, MessageLevel.DETAIL)) {
            LogManager.logDetail(LogConstants.CTX_DQP, "getPlan for requestID=" + requestID); //$NON-NLS-1$
        }
        RequestWorkItem workItem = safeGetWorkItem(requestID);
        if (workItem == null) {
        	return null;
        }
        QueryProcessor qp = workItem.getProcessor();
        if (qp == null) {
        	return null;
        }
        return qp.getProcessorPlan().getDescriptionProperties();
    }
    
    private boolean cancelRequest(RequestID requestID) throws TeiidComponentException {
        if (LogManager.isMessageToBeRecorded(LogConstants.CTX_DQP, MessageLevel.DETAIL)) {
            LogManager.logDetail(LogConstants.CTX_DQP, "cancelQuery for requestID=" + requestID); //$NON-NLS-1$
        }
        
        boolean markCancelled = false;
        
        RequestWorkItem workItem = safeGetWorkItem(requestID);
        if (workItem != null) {
        	markCancelled = workItem.requestCancel();
        }
    	if (markCancelled) {
		    logMMCommand(workItem, Event.CANCEL, null, null);
    	} else {
    		LogManager.logDetail(LogConstants.CTX_DQP, QueryPlugin.Util.getString("DQPCore.failed_to_cancel", requestID)); //$NON-NLS-1$
    	}
        return markCancelled;
    }
    
	public ResultsFuture<?> closeRequest(long requestId) throws TeiidProcessingException, TeiidComponentException {
        DQPWorkContext workContext = DQPWorkContext.getWorkContext();
        closeRequest(workContext.getRequestID(requestId));
        return ResultsFuture.NULL_FUTURE;
	}
    
    /**
     * Close the request with given ID 
     * @param requestID requestID
     * @throws TeiidComponentException TeiidComponentException
     */
    void closeRequest(RequestID requestID) throws TeiidComponentException {
        if (LogManager.isMessageToBeRecorded(LogConstants.CTX_DQP, MessageLevel.DETAIL)) {
            LogManager.logDetail(LogConstants.CTX_DQP, "closeQuery for requestID=" + requestID); //$NON-NLS-1$
        }
        
        RequestWorkItem workItem = safeGetWorkItem(requestID);
        if (workItem != null) {
        	workItem.requestClose();
        } else {
        	LogManager.logDetail(LogConstants.CTX_DQP, requestID + " close call ignored as the request has already been removed."); //$NON-NLS-1$
        }
    }
    
	public Collection<TransactionMetadata> getTransactions() {
		return this.transactionService.getTransactions();
	}
	
	public void terminateTransaction(String xid) throws AdminException {
		this.transactionService.terminateTransaction(xid);
	}	

	void logMMCommand(RequestWorkItem workItem, Event status, Integer rowCount, Long cpuTime) {
		logMMCommand(workItem, status, rowCount, null, cpuTime);
	}
		
	void logMMCommand(RequestWorkItem workItem, Event status, Integer rowCount, ResultsFuture<ResultsMessage> resultsFuture, Long cpuTime) {
    	RequestMessage msg = workItem.requestMsg;
    	DVCaller.call(109, workItem, msg, status, resultsFuture);
    	if ((status != Event.PLAN && !LogManager.isMessageToBeRecorded(LogConstants.CTX_COMMANDLOGGING, MessageLevel.DETAIL))
			    || (status == Event.PLAN && !LogManager.isMessageToBeRecorded(LogConstants.CTX_COMMANDLOGGING, MessageLevel.TRACE))) {
    		return;
    	}
    	
        DQPWorkContext workContext = DQPWorkContext.getWorkContext();
        RequestID rID = workItem.requestID;
    	String txnID = null;
		TransactionContext tc = workItem.getTransactionContext();
		if (tc != null && tc.getTransactionType() != Scope.NONE) {
			txnID = tc.getTransactionId();
		}
    	String appName = workContext.getAppName();
        // Log to request log
        CommandLogMessage message;
        if (status == Event.NEW) {
            message = new CommandLogMessage(System.currentTimeMillis(), rID.toString(), txnID, workContext.getSessionId(), appName, workContext.getUserName(), workContext.getVdbName(), workContext.getVdbVersion(), msg.getCommandString(), cpuTime);
        } else {
            QueryProcessor qp = workItem.getProcessor();
            PlanNode plan = null;
            if (LogManager.isMessageToBeRecorded(LogConstants.CTX_COMMANDLOGGING, MessageLevel.TRACE) && qp != null) {
            	plan = qp.getProcessorPlan().getDescriptionProperties();
            }
            message = new CommandLogMessage(System.currentTimeMillis(), rID.toString(), txnID, workContext.getSessionId(), workContext.getUserName(), workContext.getVdbName(), workContext.getVdbVersion(), rowCount, status, plan);
        }
		LogManager.log(status == Event.PLAN?MessageLevel.TRACE:MessageLevel.DETAIL, LogConstants.CTX_COMMANDLOGGING, message);
    }
    
    public TempTableDataManager getDataTierManager() {
    	return this.dataTierMgr;
    }
    
	public BufferManager getBufferManager() {
		return bufferManager;
	}

	public TransactionService getTransactionService() {
		return transactionService;
	}

	SessionAwareCache<CachedResults> getRsCache() {
		return rsCache;
	}
	
	int getProcessorTimeSlice() {
		return this.config.getTimeSliceInMilli();
	}	
	
	int getChunkSize() {
		return chunkSize;
	}
	
	public void start(DQPConfiguration theConfig) {
		this.config = theConfig;
        this.authorizationValidator = config.getAuthorizationValidator();
        this.chunkSize = config.getLobChunkSizeInKB() * 1024;

        this.processWorkerPool = config.getTeiidExecutor();
        //we don't want cancellations waiting on normal processing, so they get a small dedicated pool
        //TODO: overflow to the worker pool
        Executor timeoutExecutor = ExecutorUtils.newFixedThreadPool(3, "Server Side Timeout"); //$NON-NLS-1$
        this.cancellationTimer = new EnhancedTimer(timeoutExecutor, timeoutExecutor);
        this.maxActivePlans = config.getMaxActivePlans();
        
        if (this.maxActivePlans > config.getMaxThreads()) {
        	LogManager.logWarning(LogConstants.CTX_DQP, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID30006, this.maxActivePlans, config.getMaxThreads()));
        	this.maxActivePlans = config.getMaxThreads();
        }

		//for now options are scoped to the engine - vdb scoping is a todo
		options = new Options();
		options.setAssumeMatchingCollation(false);
		//options.setProperties(config.getProperties());
		options.setProperties(System.getProperties());
		PropertiesUtils.setBeanProperties(options, options.getProperties(), "org.teiid", true); //$NON-NLS-1$

		this.bufferManager.setOptions(options);
		//hack to set the max active plans
        this.bufferManager.setMaxActivePlans(this.maxActivePlans);
        try {
			this.bufferManager.initialize();
		} catch (TeiidComponentException e) {
			 throw new TeiidRuntimeException(QueryPlugin.Event.TEIID30496, e);
		}
        
        this.userRequestSourceConcurrency = config.getUserRequestSourceConcurrency();
        if (this.userRequestSourceConcurrency < 1) {
        	this.userRequestSourceConcurrency = Math.min(config.getMaxThreads(), 2*config.getMaxThreads()/this.maxActivePlans);
        }
        
        DataTierManagerImpl processorDataManager = new DataTierManagerImpl(this, this.bufferManager, this.config.isDetectingChangeEvents());
        processorDataManager.setEventDistributor(eventDistributor);
		dataTierMgr = new TempTableDataManager(processorDataManager, this.bufferManager, this.rsCache);
		dataTierMgr.setExecutor((command, parameters) -> {
				final String sessionId = DQPWorkContext.getWorkContext().getSessionId();
				RequestMessage request = new RequestMessage(command);
				request.setParameterValues(parameters);
				request.setStatementType(StatementType.PREPARED);
				ResultsFuture<ResultsMessage> result;
				try {
					result = executeRequest(0, request);
				} catch (TeiidProcessingException e) {
					throw new TeiidRuntimeException(e);
				}
            result.addCompletionListener(future -> terminateSession(sessionId));
				});
        dataTierMgr.setEventDistributor(eventDistributor);
		userActivePlans.init(this);

        LogManager.logDetail(LogConstants.CTX_DQP, "DQPCore started maxThreads", this.config.getMaxThreads(), "maxActivePlans", this.maxActivePlans, "source concurrency", this.userRequestSourceConcurrency); //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$
	}
	
	public void setBufferManager(BufferManager mgr) {
		this.bufferManager = mgr;
	}
	
	public void setTransactionService(TransactionService service) {
		this.transactionService = service;
	}
	
	public void setEventDistributor(EventDistributor eventDistributor) {
		this.eventDistributor = eventDistributor;
	}
	
	@Override
	public boolean cancelRequest(long requestID)
			throws TeiidProcessingException, TeiidComponentException {
		DQPWorkContext workContext = DQPWorkContext.getWorkContext();
		return this.cancelRequest(workContext.getRequestID(requestID));
	}
	
	// local txn
	public ResultsFuture<?> begin() throws XATransactionException {
    	String threadId = DQPWorkContext.getWorkContext().getSessionId();
    	this.getTransactionService().begin(threadId);
    	return ResultsFuture.NULL_FUTURE;
	}
	
	// local txn
	public ResultsFuture<?> commit() throws XATransactionException {
		final String threadId = DQPWorkContext.getWorkContext().getSessionId();
		Callable<Void> processor = () -> {
				getTransactionService().commit(threadId);
				return null;
		};
		return addWork(processor, 0);
	}
	
	// local txn
	public ResultsFuture<?> rollback() throws XATransactionException {
		final String threadId = DQPWorkContext.getWorkContext().getSessionId();
		Callable<Void> processor = () -> {
				getTransactionService().rollback(threadId);
				return null;
		};
		return addWork(processor, 0);
	}

	// global txn
	public ResultsFuture<?> commit(final XidImpl xid, final boolean onePhase) throws XATransactionException {
		Callable<Void> processor = () -> {
				DQPWorkContext workContext = DQPWorkContext.getWorkContext();
				getTransactionService().commit(workContext.getSessionId(), xid, onePhase, workContext.getSession().isEmbedded());
				return null;
		};
		return addWork(processor, 0);
	}
	// global txn
	public ResultsFuture<?> end(XidImpl xid, int flags) throws XATransactionException {
		DQPWorkContext workContext = DQPWorkContext.getWorkContext();
		this.getTransactionService().end(workContext.getSessionId(), xid, flags, workContext.getSession().isEmbedded());
		return ResultsFuture.NULL_FUTURE;
	}
	// global txn
	public ResultsFuture<?> forget(XidImpl xid) throws XATransactionException {
		DQPWorkContext workContext = DQPWorkContext.getWorkContext();
		this.getTransactionService().forget(workContext.getSessionId(), xid, workContext.getSession().isEmbedded());
		return ResultsFuture.NULL_FUTURE;
	}
		
	// global txn
	public ResultsFuture<Integer> prepare(final XidImpl xid) throws XATransactionException {
		Callable<Integer> processor = () -> {
				DQPWorkContext workContext = DQPWorkContext.getWorkContext();
				return getTransactionService().prepare(workContext.getSessionId(),xid, workContext.getSession().isEmbedded());
		};
		return addWork(processor, 10);
	}

	private <T> ResultsFuture<T> addWork(final Callable<T> processor, int priority) {
		final ResultsFuture<T> result = new ResultsFuture<>();
		final ResultsReceiver<T> receiver = result.getResultsReceiver();
		Runnable r = () -> {
				try {
					receiver.receiveResults(processor.call());
				} catch (Throwable t) {
					receiver.exceptionOccurred(t);
				}
		};
		FutureWork<T> work = new FutureWork<>(r, null, priority);
		if (DQPWorkContext.getWorkContext().useCallingThread()) {
			work.run();
		} else {
			this.addWork(work);
		}
		return result;
	}
	
	// global txn
	public ResultsFuture<Xid[]> recover(int flag) throws XATransactionException {
		ResultsFuture<Xid[]> result = new ResultsFuture<>();
		DQPWorkContext workContext = DQPWorkContext.getWorkContext();
		result.getResultsReceiver().receiveResults(this.getTransactionService().recover(flag, workContext.getSession().isEmbedded()));
		return result;
	}
	// global txn
	public ResultsFuture<?> rollback(final XidImpl xid) throws XATransactionException {
		Callable<Void> processor = () -> {
				DQPWorkContext workContext = DQPWorkContext.getWorkContext();
				getTransactionService().rollback(workContext.getSessionId(),xid, workContext.getSession().isEmbedded());
				return null;
		};
		return addWork(processor, 0);
	}
	// global txn
	public ResultsFuture<?> start(final XidImpl xid, final int flags, final int timeout)
			throws XATransactionException {
		final DQPWorkContext workContext = DQPWorkContext.getWorkContext();
		if (workContext.getSession().isEmbedded()) {
		    //must be a synch call regardless since the txn should be associated with the thread.
			getTransactionService().start(workContext.getSessionId(), xid, flags, timeout, true);
			return ResultsFuture.NULL_FUTURE;
		}
		Callable<Void> processor = () -> {
				getTransactionService().start(workContext.getSessionId(), xid, flags, timeout, false);
				return null;
		};
		return addWork(processor, 100);
	}

	public MetadataResult getMetadata(long requestID)
			throws TeiidComponentException, TeiidProcessingException {
		DQPWorkContext workContext = DQPWorkContext.getWorkContext();
		MetaDataProcessor processor = new MetaDataProcessor(this, this.prepPlanCache,  workContext.getVdbName(), workContext.getVdbVersion());
		return processor.processMessage(workContext.getRequestID(requestID), workContext, null, true);
	}

	public MetadataResult getMetadata(long requestID, String preparedSql,
			boolean allowDoubleQuotedVariable)
			throws TeiidComponentException, TeiidProcessingException {
		DQPWorkContext workContext = DQPWorkContext.getWorkContext();
		checkActive(workContext);
		MetaDataProcessor processor = new MetaDataProcessor(this, this.prepPlanCache, workContext.getVdbName(), workContext.getVdbVersion());
		return processor.processMessage(workContext.getRequestID(requestID), workContext, preparedSql, allowDoubleQuotedVariable);
	}

	private void checkActive(DQPWorkContext workContext)
			throws TeiidProcessingException {
		if (workContext.getVDB().getStatus() != Status.ACTIVE) {
			throw new TeiidProcessingException(QueryPlugin.Event.TEIID31099, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID31099, workContext.getVDB()));
		}
	}
	
	public boolean isExceptionOnMaxSourceRows() {
		return this.config.isExceptionOnMaxSourceRows();
	}
	
	public int getMaxSourceRows() {
		return this.config.getMaxSourceRows();
	}
	
	public int getMaxRowsFetchSize() {
		return this.config.getMaxRowsFetchSize();
	}
	
	public void setResultsetCache(SessionAwareCache<CachedResults> cache) {
		this.rsCache = cache;
	}
	
	public void setPreparedPlanCache(SessionAwareCache<PreparedPlan> cache) {
		this.prepPlanCache = cache;
	}

	public void setDvCacheFactory(CacheFactory dvCacheFactory) {
		this.dvCacheFactory = dvCacheFactory;
	}

	public SecretKey getSecretKey() {
		return this.config.getSecretKey();
	}

	public <K,V> Cache<K, V> getDvCache(DVCache dvCache) {
		if (Boolean.getBoolean("dv.embedded"))
			return new ConcurrentMapCache<>(dvCache.getCacheName());
		if (dvCacheFactory != null)
			return dvCacheFactory.get(dvCache.getCacheName());
		return null;
	}

	public int getUserRequestSourceConcurrency() {
		return userRequestSourceConcurrency;
	}
	
	void setUserRequestSourceConcurrency(int userRequestSourceConcurrency) {
		this.userRequestSourceConcurrency = userRequestSourceConcurrency;
	}
	
	public int getMaxActivePlans() {
		return maxActivePlans;
	}
	
	public SessionAwareCache<PreparedPlan> getPrepPlanCache() {
		return prepPlanCache;
	}
	
	public String getRuntimeVersion() {
		return ApplicationInfo.getInstance().getBuildNumber();
	}	
	
	public AuthorizationValidator getAuthorizationValidator() {
		return authorizationValidator;
	}

	public ProcessingState getWorkItemState(RequestWorkItem holder) {
		return holder.isCanceled()?ProcessingState.CANCELED:holder.isDoneProcessing()?ProcessingState.DONE:ProcessingState.PROCESSING;
	}

	public ThreadState getWorkItemThreadState(RequestWorkItem holder) {
		switch (holder.getThreadState()) {
			case DONE:
			case IDLE:
				return ThreadState.IDLE;
			default:
				if (holder.isProcessing()) {
					return ThreadState.RUNNING;
				} else {
					return ThreadState.QUEUED;
				}
		}
	}

	public ProcessingState getConnectorWorkItemState(DataTierTupleSource conInfo) {
		return conInfo.isCanceled()?ProcessingState.CANCELED:conInfo.isDone()?ProcessingState.DONE:ProcessingState.PROCESSING;
	}

	public ThreadState getConnectorWorkItemThreadState(DataTierTupleSource conInfo) {
		if (conInfo.isQueued()) {
			return ThreadState.QUEUED;
		} else if (conInfo.isRunning()) {
			return ThreadState.RUNNING;
		} else {
			return ThreadState.IDLE;
		}
	}
}
