/*
 * Portions Copyright (C) 2005 NTT DATA Corporation
 * 
 */
package org.postgresforest.vm.core;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Vector;

import org.postgresforest.Driver;
import org.postgresforest.core.Field;
import org.postgresforest.core.PGStream;
import org.postgresforest.core.ParameterList;
import org.postgresforest.core.Query;
import org.postgresforest.core.QueryExecutor;
import org.postgresforest.core.ResultCursor;
import org.postgresforest.core.ResultHandler;
import org.postgresforest.core.ProtocolConnection;
import org.postgresforest.util.GT;
import org.postgresforest.util.PSQLException;
import org.postgresforest.vm.Lock;
import org.postgresforest.vm.LogUtil;
import org.postgresforest.vm.MetaDataReWriter;
import org.postgresforest.vm.NoPartitionParser;
import org.postgresforest.vm.NoPartitionReWriter;
import org.postgresforest.vm.Parser;
import org.postgresforest.vm.Logger;
import org.postgresforest.vm.QueryInfo;
import org.postgresforest.vm.ReWriter;
import org.postgresforest.vm.StatementSub;
import org.postgresforest.vm.StatementThreadMng;
import org.postgresforest.vm.err.ForestPSQLException;
import org.postgresforest.vm.err.ForestSQLState;
import org.postgresforest.vm.gsc.GscData;

/**
 * QueryExecutor implementation for the ForestVM.
 */
/**
 * PostgresForest用Statement基底クラス.
 * 仮想化モジュール用の変更をまとめたクラス。
 * グローバルシステムカタログへのアクセス用のパス(*GD())と、
 * パーティション、分散、並列の実行メソッド(execute())を持つ
 */
public class QueryExecutorImpl implements QueryExecutor{


	/** ログ出力 @since 2.1 */
	protected LogUtil m_logUtil;
	/** グローバルシステムカタログクラス */
	protected GscData m_gsc;
	/** SQL実行スレッド管理クラス */
	protected StatementThreadMng m_threadMng;

	/** ロールバック発生情報 */
	protected RollBackInfo m_rollBackInfo = new RollBackInfo();

    protected List m_metaDataAccess = Collections.synchronizedList(new ArrayList());
    
	protected boolean m_updateSyncMode;

	protected boolean m_updateSyncModeManual = false;

	protected boolean m_autocommit = false;

	
	/**
	 * 
	 * ロールバック対象エラー発生有無と発生エラーを格納
	 *
	 */
	class RollBackInfo{
		/** rollbackフラグ　true ロールバック指示　false  @since 3.1*/
		private boolean m_rollback = false;
		private SQLException  m_rollbackException;

		protected void setRollback(boolean rollback) {
			m_rollback = rollback;
		}

		protected boolean isRollback() {
			return m_rollback;
		}

		protected void setRollbackException(SQLException rollbackException) {
			m_rollbackException = rollbackException;
		}

		protected SQLException getRollbackException() {
			return m_rollbackException;
		}
		
		
	}
	
	/** コネクション */
	protected ProtocolConnectionImpl m_protocolConnectionImpl;
	
	public QueryExecutorImpl(GscData gsc, StatementThreadMng mng, ProtocolConnectionImpl protocolConnectionImpl) {
		super();
		m_gsc = gsc;
		m_logUtil = m_gsc.getLogUtil();
		m_threadMng = mng;
		m_protocolConnectionImpl = protocolConnectionImpl;
	}

	
	public ParameterList createFastpathParameters(int count) {
		// TODO 自動生成されたメソッド・スタブ
		return null;
	}

	public Query createParameterizedQuery(String sql) {


		return createSimpleQuery(sql);

	}

	public Query createSimpleQuery(String sql) {

		
		Parser parser = new Parser(m_logUtil);

		if( m_gsc.hasPartition() == false  &&
			m_gsc.canPartitionMode() == true   ){
		    //非パーティション化モードが可能でパーティションを含んでいない場合
		    //非パーティション化モード
			if (Driver.logInfo)
				m_logUtil.info("NoPartitionParser was selected.");
			parser = new NoPartitionParser(m_logUtil);
		}else{
			if (Driver.logInfo)
				m_logUtil.info("Parser was selected." );
			parser = new Parser(m_logUtil);
		}
		
		SimpleQuery query = new SimpleQuery(parser);
		
		//構文解析実行
		//Ver2.0 ローカルからフィールドへ。結果集約で外部から使用するため　
		try {
			parser.parse(sql);

			if (Driver.logInfo)
				m_logUtil.info("Parser was completed. (" + sql + ")");

		} catch (SQLException e) {
			//SQLException をthrowできないので、Queryに格納
			query.setParserErr(e); 
		}

		
		return query;
	}

	public synchronized void execute(Query query,
									 ParameterList parameters,
									 ResultHandler handler,
									 int maxRows,
									 int fetchSize,
									 int flags)
		throws SQLException {

		SimpleQuery simpleQuery = (SimpleQuery)query;
		
		//パラメータをコピー（非同期モードで上書きされるので）
		SimpleParameterList parameters_wk = null;
		if(parameters != null)
			parameters_wk = (SimpleParameterList)parameters.copy();
		
		/*
		 * autocommitがtrueの場合、テーブルの閉塞状態を検出すると、
		 * 個別のSQLステートメントを実行する前に停止する。
		 */
        if ( getAutocommit()==true && !m_gsc.isTableAvailable() )
		{
				m_gsc.waitTableAvailable();
        }
		/*
		 * autocommitがfalseの場合、テーブルの閉塞状態を検出すると、
		 * トランザクションを開始する前（IDLEの状態の間）に停止する。
		 */
        else if ( getAutocommit()==false && !m_gsc.isTableAvailable() )
		{
			if ( m_protocolConnectionImpl.getTransactionState()==ProtocolConnection.TRANSACTION_IDLE )
				m_gsc.waitTableAvailable();
        }

		Parser parser = simpleQuery.getParser();
		
		//Prepared パラメータをPaserに設定
		parser.setParamList(parameters_wk);
		
		
		//パーティション化モードに応じたRewriter生成
		ReWriter rw;
		if(isMetaDataAccess()){
		    //メタデータアクセスは、すべて分散処理
		    //メタデータアクセス
			if (Driver.logInfo)
				m_logUtil.info("MetaDataReWriter was selected.");
			rw = new MetaDataReWriter(m_gsc);
		
		}else if (parser instanceof NoPartitionParser) {
		    //非パーティション化モードが可能でパーティションを含んでいない場合
		    //非パーティション化モード
			if (Driver.logInfo)
				m_logUtil.info("NoPartitionReWriter was selected.");
			rw = new NoPartitionReWriter(m_gsc);
		}else{
			if (Driver.logInfo)
				m_logUtil.info("ReWriter was selected." );
			rw = new ReWriter(m_gsc);
		}


        try{
		
			//SQL・書き換え
			rw.rewrite(parser);
			if (Driver.logInfo)
				m_logUtil.info("ReWriter was completed. (" + parser.getSrcSql() + ")");

        }catch (SQLException e) {
			//SQLExceptionはそのままthrow
			throw (SQLException) e;

        }catch (Exception e1) {

			throw new PSQLException(GT.tr("SQL analysis error."),ForestSQLState.INTERNAL_ERROR, e1);
		}


		

		ArrayList queryinfoList = rw.getQueryInfo();
		int threadCnt = queryinfoList.size();
		int queryType = rw.getQueryType();
		simpleQuery.setQueryNum(threadCnt);
		simpleQuery.setQueryType(queryType);


        
		boolean flg = true;
		
		
		StatementSub[] th = new StatementSub[threadCnt];
		
		
		//Ver1.1 パフォーマンスチューニング 
		Lock lock = new Lock(threadCnt);

		//スレッド実行
		for( int i = 0; i < threadCnt; i++ )
		{
		    //クエリ情報取得
			QueryInfo qi = (QueryInfo)queryinfoList.get(i);

			//クエリー実行メインスレッド取得
			StatementSub statementSub = m_threadMng.getStatementSub();
			
			//クエリー実行メインスレッド実行
			statementSub.execute(simpleQuery,
									parameters_wk,
									maxRows,
									fetchSize,
									flags,
									this,
									qi,
									rw,
									lock );
			//スレッドリストに格納
			th[i] = statementSub;
		}

		
		Field[] fields = null;
	    Vector tuples = new Vector();
		
		
		try{
	
		//スレッドの終了待ちと結果集約
		StatementSub statementSub;		    
		SimpleResultHandler resHandler; 
		switch (queryType) {
			case ReWriter.TYPE_PARTITION:	//パーティション処理

				//スレッド終了待機

				try {
					
					lock.waitAllEnd();//@@エラーの検出を優先するなら、waitEndにするべき。
				} catch (InterruptedException e1) {}

				
				//エラーチェック　と　結果集約
				for( int i = 0; i < threadCnt; i ++ ){
				    statementSub = th[i];

				    //エラーチェック
					if( statementSub.getErrCode() != StatementSub.STAT_OK )
					{

						throw statementSub.getErrObj();
					}

					//結果集約
					resHandler =  statementSub.getResultHandler();
					if(i == 0){
						fields = resHandler.getFields();
					}
					tuples.addAll(resHandler.getTuples());					
					
				}

				handler.handleResultRows(query,fields,tuples,null);

				break;

			case ReWriter.TYPE_DISPERSION:	//分散処理

				//スレッド終了待機
				try {
					lock.waitEnd();
				} catch (InterruptedException e1) {}
				

			    statementSub = th[0];

				//エラーチェック
				if( statementSub.getErrCode() != StatementSub.STAT_OK )
				{
					throw statementSub.getErrObj();
				}

				//結果格納
				resHandler =  statementSub.getResultHandler();
				handler.handleResultRows(resHandler.getFromQuery(),resHandler.getFields(),resHandler.getTuples(),resHandler.getCursor());
				break;

			case ReWriter.TYPE_PARALLEL:	//並列処理
				
				//スレッド終了待機
				flg = true;
				while(flg)
				{
					
					if(isUpdateSyncMode() == true){
						//更新同期モード Ver3.0

						try {
							lock.waitAllEnd();
						
						} catch (InterruptedException e1) {}
					

						//Ver1.1R8 １つ以上のDBで生じた場合はアプリへ例外を投げる。
						//エラーチェック
						for( int i = 0; i < threadCnt; i ++ ){

						    statementSub = th[i];
							
						    if( statementSub.getErrCode() != StatementSub.STAT_OK )
							{
								throw statementSub.getErrObj();
							}else{
								resHandler =  statementSub.getResultHandler();
								if( resHandler.getStatus() != null && flg == true)
								{
									handler.handleCommandStatus(resHandler.getStatus(),
																resHandler.getUpdateCount(),
																resHandler.getInsertOID());
								}else{
									/*
									 * for updateに限っては最初に返ってきた答え以外は保持しない
									 * （保持してしまうとResultSetが複数になってしまう）
									 */
									if (flg || !parser.isForUpdate()) {
										handler.handleResultRows(resHandler.getFromQuery(),
																 resHandler.getFields(),
																 resHandler.getTuples(),
																 resHandler.getCursor());
									}
								}
								flg = false;
								
							}
						}


					}else{
						//更新非同期モード
	
						try {
							lock.waitEnd();
						} catch (InterruptedException e1) {}
	
						boolean allErr = true; 
						for( int i = 0; i < threadCnt; i ++ )
						{
						    statementSub = th[i];

							if( statementSub.isExecuted() == false )
							{
								allErr = false;
							}else{
								//正常終了したら終了
								if( statementSub.getErrCode() == StatementSub.STAT_OK){
									resHandler =  statementSub.getResultHandler();
									if( resHandler.getStatus()  != null )
									{
										handler.handleCommandStatus(resHandler.getStatus(),resHandler.getUpdateCount(),resHandler.getInsertOID());
									}else{
										handler.handleResultRows(resHandler.getFromQuery(),resHandler.getFields(),resHandler.getTuples(),resHandler.getCursor());
									}
									flg = false;
									break;
								}
							}
						}
	
						//すべてエラーとなった場合
						if(flg == true && allErr == true){
					    
						    //すべてのサーバで処理失敗。各エラーはgetNextException()を呼び出してください。
						    ForestPSQLException err = new ForestPSQLException(GT.tr("It is processing failure at all servers, Each error should call getNextException()."), ForestSQLState.INTERNAL_ERROR);
							for( int i = 0; i < threadCnt; i ++ ) {
								//各サーバーのエラーをメッセージをエラーに追加
								err.setNextException(th[i].getErrObj(), th[i].getURL(), th[i].getSql()); 
							}
							throw err;
	
						}
	
						/*
						 * 並列処理で更新非同期モードの場合、
						 * ロールバック対象エラー以外のエラーはすべて切り離し対象となり、
						 * Connectionのclose()時に切り離し処理が行われる。
						 *
						 * FIXME: なぜ？
						 *
						 * @see StatementSub#close()
						 */
						if(flg == false){
							for( int i = 0; i < threadCnt; i ++ ) {
								th[i].setServerBrokenChk(true); 
							}
							break;
						}

					}

				}

				break;

			default :
				break;
		}

		}catch (Exception e) {
			//SQLException以外のExceptionは内部エラーとしてthrowする
			if (e instanceof SQLException) {
				//SQLExceptionはそのままthrow
				throw (SQLException) e;
			}else{
				//その他のExceptionはSQLExceptionでthrow
				throw new PSQLException(GT.tr("excute internal error.\n{0}",e), ForestSQLState.INTERNAL_ERROR,e);
			}

		}finally{
			//スレッドの開放
			for( int i = 0; i < threadCnt; i ++ ) {
				m_threadMng.closeStatementSub(th[i]); 
			}

		}

		if ( getAutocommit()==false &&
			 m_protocolConnectionImpl.getTransactionState()==ProtocolConnection.TRANSACTION_IDLE )
			m_protocolConnectionImpl.setTransactionState(ProtocolConnection.TRANSACTION_OPEN);
	}

	public synchronized void execute(Query[] queries, ParameterList[] parameterLists, ResultHandler handler, int maxRows, int fetchSize, int flags) throws SQLException {
		// TODO 自動生成されたメソッド・スタブ
		
	}

	public synchronized byte[] fastpathCall(int fnid, ParameterList params, boolean suppressBegin) throws SQLException {
		// TODO 自動生成されたメソッド・スタブ
		return null;
	}

	public synchronized void fetch(ResultCursor cursor, ResultHandler handler, int fetchSize) throws SQLException {
		// TODO 自動生成されたメソッド・スタブ
		
	}




	public LogUtil getLogUtil() {
		return m_logUtil;
	}

	
	/**
     * @return rollback を戻します。
     * @since 3.1
     */
    public boolean isRollback() {
        synchronized (m_rollBackInfo) {
        	return m_rollBackInfo.isRollback();
		}
    	
    }
    /**
     * @param rollback rollback を設定。
     * @since 3.1
     */
    public void setRollback(SQLException ex) {
        if(ex != null){
            synchronized (m_rollBackInfo) {
            	m_rollBackInfo.setRollback(true);
		        //アプリへの返却用Exception生成
	            //ロールバック対象のエラーが発生しました。
            	m_rollBackInfo.setRollbackException(new PSQLException(GT.tr("The error of the rollback object occurred."), ForestSQLState.INTERNAL_ERROR,ex));
            }
        }
    }
    /**
     * @param rollback rollback を設定。
     * @since 3.1
     */
    public void clearRollback() {
        synchronized (m_rollBackInfo) {
        	m_rollBackInfo.setRollback(false);
        }    
    }


	public SQLException getRollbackException() {
       	return m_rollBackInfo.getRollbackException();
	}


	public void waitFree() {
		m_threadMng.waitFree();
	}


	public GscData getGsc() {
		return m_gsc;
	}

	public boolean addMetaDataAccess() {
		return m_metaDataAccess.add(Thread.currentThread());
	}


	public boolean isMetaDataAccess() {
		return m_metaDataAccess.contains(Thread.currentThread());
	}


	public boolean removeMetaDataAccess() {
		return m_metaDataAccess.remove(Thread.currentThread());
	}


	public boolean isUpdateSyncMode() {
		
		if(m_updateSyncModeManual){
			return m_updateSyncMode;
		}else{
			return 	m_gsc.getUpdateSyncMode();
		}
	}


	public void setUpdateSyncMode(boolean updateSyncMode) {
		m_updateSyncMode = updateSyncMode;
		m_updateSyncModeManual = true;
	}


	public boolean getAutocommit() {
		return m_autocommit;
	}


	public void setAutocommit(boolean autocommit) {
		m_autocommit = autocommit;
	}


	public void setTransactionState(int transactionState) {
		m_protocolConnectionImpl.setTransactionState(transactionState);
	}

	public PGStream getPGStream() {
		return null;
	}
	
}
