/*
 * $Source: /opt/cvsroot/erserver/src/com/postgres/replic/util/MasterProcessor.java,v $
 * $Author: ronz $ $Revision: 1.4 $ $Date: 2003/12/29 22:43:47 $
 *
 */
package com.postgres.replic.util;

import java.sql.*;
import java.util.*;

import com.postgres.replic.server.props.*;
import com.postgres.replic.util.struct.*;

/**
 */
public class MasterProcessor extends DBProcessor {

	private final static String ACTIVE_SEP = ",";
	// Modes of activeLogTable:
	private final static int ALT_1 = 1;// $CTable = "_RSERV_LOG_1_"; $OTable
	// = "_RSERV_LOG_2_";
	private final static int ALT_2 = 2;
	private final static int EXCLUSIVE = 1;

	// Number of rows to fetch in cursors:
	//private final static int FETCH_MAX = 1000;

	// TAble id-s:
	String LOG_1 = "_RSERV_LOG_1_";
	String LOG_2 = "_RSERV_LOG_2_";
	private final static int MARK_CLEAN = 1;

	// Lock modes:
	private final static int SHARED = 0;

	private boolean resetTableMap = false;

	// Sql satatements:
	String SQL_TABLE_MAP = "" +
			"select pgc.oid, pgc.relname, pgc.attname, pgt.typname " +
			"from _RSERV_TABLES_ rt, _ers_class_attr pgc, " +
			"pg_type pgt " +
			"where pgc.oid = rt.reloid  " +
			"and pgt.oid = pgc.atttypid and pgc.attnum = rt.key";

	private final static int TRUNCATE = 0;
	// Modes defining useOldLogTable:
	private final static int USE = 2;

	// Statuses for clean log:
	private final static int USE_OLD_NEW_LOG = 2;

	private int activeLogTable = 0;
	private LastSyncData lastSync;// of TableMap objects
	private int server = 0;
	private String sinfo = null;
	private SnapshotData snapData = new SnapshotData();
	//private TableColumnCache  tableColumnCache;
	private boolean snapshotRemembered = false;
	private long syncid = T_NO;

	private TableMap tableMap;
	ArrayList usTable;
	//private USTable usTable;// $CTable = "_RSERV_LOG_2_"; $OTable
	// = "_RSERV_LOG_1_";

	private boolean useOldLogTable = true;
	private int useolt = 0;

	/**
	 *  Gets the snapshotData attribute
	 *
	 * @return                     The snapshotData value
	 * @exception  RservException  Description of the Exception
	 */
	public SnapshotData getSnapshotData() throws RservException {
		if (!snapshotRemembered) {
			throw new RservException("::getSnapshotData: snapshot is not remembered");
		}
		return snapData;
	}


	/**
	 *  Description of the Method
	 *
	 * @param  conn                Parameter
	 * @param  server              Parameter
	 * @param  serverProps         Parameter
	 * @return                     Return Value
	 * @exception  RservException  Description of the Exception
	 * @exception  Exception       Description of the Exception
	 */
	public MasterProcessor reset(Connection conn, int server, ServerProps serverProps)
			throws RservException, Exception {
		super.init(conn, serverProps);
		SQL_TABLE_MAP =
			"select pgc.oid, pgc.relname, pgc.attname, pgt.typname " +
			"from "+serverProps.createRelName("_RSERV_TABLES_")+" rt, "+serverProps.createRelName("_ers_class_attr")+" pgc, " +
			"pg_type pgt " +
			"where pgc.oid = rt.reloid  " +
			"and pgt.oid = pgc.atttypid and pgc.attnum = rt.key";

		LOG_1 = serverProps.createRelName("_RSERV_LOG_1_");
		LOG_2 = serverProps.createRelName("_RSERV_LOG_2_");

		lastSync = null;
		snapData.clear();
		this.server = server;
		snapshotRemembered = false;
		usTable = null;
		sinfo = null;
		syncid = T_NO;
		getLogger().debug("***** **** *** MasterProcessor::reset: server=" + server + "; getPstmtUseCount()=" + getPstmtUseCount() + "; conn=" + conn + "; tableColumnCache=" + tableColumnCache);
		setTableColumnCache();
		getLogger().debug("MasterProcessor::reset: After setTableColumnCache");
		if (resetTableMap) {
			getLogger().info("MasterProcessor::reset: Resetting table map");
			tableMap = null;
			tableMap = getTableMap();
			resetTableMap = false;
		}

		setInitialized(true);
		return this;
	}

	/* Updates _RSERV_SYNC_ on Master with Slave SyncID
    */
	/**
	 *  Description of the Method
	 *
	 * @param  syncid              Parameter
	 * @exception  RservException  Description of the Exception
	 */
	protected void SyncSyncID(long syncid) throws RservException {
		validateInit();
		ResultSet rs = null;
		Statement stmt = null;
		String sql = "select synctime, status from "+serverProps.createRelName("_RSERV_SYNC_") +
				" where server = " + server + " and syncid = " +
				syncid +
				" for update";
		getLogger().debug("MasterProcessor::SyncSyncID: sql(1)=" + sql);
		long rc = T_NO;

		try {
			stmt = getConnection().createStatement();
			rs = stmt.executeQuery(sql);
			int row = 0;
			while (rs.next()) {
				row++;
				rc = rs.getLong(2);
			}

			if (row == 0) {
				throw new RservException(
						"MasterProcessor::SyncSyncID: No SyncID " + syncid
						+ " found for server " + server);
			} else if (rc > 0) {
				if (getDebug()) {
					getLogger().debug("MasterProcessor::SyncSyncID: "
							+ syncid + " for server " + server + " already updated");
				}
				return;
			}

			sql = "update "+serverProps.createRelName("_RSERV_SYNC_") +
					" set synctime = now(), status = 1" +
					" where server = " + server + " and syncid = " + syncid;
			getLogger().debug("MasterProcessor::SyncSyncID: sql(2)=" + sql);

			rc = stmt.executeUpdate(sql);

			sql = "delete from "+serverProps.createRelName("_RSERV_SYNC_") +
					" where server = " + server + " and syncid < " + syncid;
			getLogger().debug("MasterProcessor::SyncSyncID: sql(3)=" + sql);

			rc = stmt.executeUpdate(sql);
			rs.close();
		} catch (Exception e) {
			throw new RservException(
					"MasterProcessor::SyncSyncID: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}
		}
	}

	/* this thing is multi-transationary
    * @param - howold time in sec how long keep the log record
    */
	/**
	 *  Description of the Method
	 *
	 * @param  tout                Parameter
	 * @return                     Return Value
	 * @exception  RservException  Description of the Exception
	 * @exception  Exception       Description of the Exception
	 */
	protected int cleanLog(int tout) throws RservException, Exception {
		getLogger().info("MasterProcessor::cleanLog BEGIN tout=" + tout);
		int olt = 0;
		try {
			// Optimizer hints:
			setOptimizerHints();
			olt = prepareCleanLog();
			commit();

			//
			// If old log has status 1 (MARK_CLEAN) then just truncate it and return.
			//
			if (useolt == MARK_CLEAN) {
				truncateTable(olt);
				postTruncateTable(olt);
				lock(EXCLUSIVE);
				setLogStatus(TRUNCATE);
				unlock(EXCLUSIVE);
				commit();
				return olt;
			}

			getLogger().info("MasterProcessor::cleanLog: WAITING (" + tout + " millisec cycle); Log" + olt + " is not clean yet");
			waitUnreplicated(olt, tout);

			//
			// Ok - all records from _RSERV_LOG_${olt}_ are replicated.
			// Now advise "Prepare" to ignore old log table in queries
			// but mark old log as clean only after truncate.
			//
			lock(EXCLUSIVE);
			setLogStatus(MARK_CLEAN);
			unlock(EXCLUSIVE);
			commit();

			//
			// Now we can just truncate old log table. Truncate will
			// grab AccessExclusive lock and so will wait for "Prepare"
			// if "Prepare" decided to read from old log table before
			// we changed old log status. This is because of "Prepare"
			// grab SHARE ROW lock on old log table while holding
			// "altid" lock and so *we wait for "Prepare"*, which is not
			// bad and *do not block "Prepare"*, which is very good.
			//
			truncateTable(olt);
			lock(EXCLUSIVE);
			setLogStatus(TRUNCATE);
			unlock(EXCLUSIVE);
			commit();

			getLogger().info("MasterProcessor::cleanLog: END ");
		} catch (Throwable e) {
			try {
				rollback();
			} catch (Exception ex) {
			}
			getLogger().error("MasterProcessor::cleanLog: ", e);
			throw new RservException("MasterProcessor::cleanLog: " + e.toString());
		} finally {

		}
		return olt;
	}


	/*
    * @param - howold time in sec how long keep the log record
    */
	/**
	 *  Description of the Method
	 *
	 * @param  howold              Parameter
	 * @exception  RservException  Description of the Exception
	 */
	protected void cleanLogOld(int howold) throws RservException {
		validateInit();
		ResultSet rs = null;
		Statement stmt = null;
		String sql = "select rs.maxid, rs.active from "+serverProps.createRelName("_RSERV_SYNC_")+" rs" +
				" where rs.syncid = (select max(rs2.syncid) from "+serverProps.createRelName("_RSERV_SYNC_")+" rs2" +
				" where rs2.server = rs.server and rs2.status > 0) order by rs.maxid";

		getLogger().debug("MasterProcessor::cleanLog: sql(1)=" + sql);
		long rc = T_NO;
		Hashtable active = new Hashtable();
		try {
			stmt = getConnection().createStatement();
			rs = stmt.executeQuery(sql);
			long maxid = 0;
			int row = 0;
			String pattern = ",";
			while (rs.next()) {
				row++;
				rc = rs.getLong(1);
				maxid = maxid == 0 ? rc : maxid;
				if (maxid > rc) {// what the hell does this mean?
					continue;
				}
				String activeValue = rs.getString(2);
				StringTokenizer st = new StringTokenizer(activeValue, pattern);
				while (st.hasMoreTokens()) {
					String token = st.nextToken().trim();
					if (!active.containsKey(token)) {
						active.put(token, token);
					}
				}
				if (maxid == 0) {
					if (getDebug()) {
						getLogger().debug("MasterProcessor::cleanLog: No maxid");
					}
					return;
				}

				String sinfo = "logid < '" + maxid+"'::xxid";
				Iterator iterator = active.keySet().iterator();
				String alist = "";
				int count = 0;
				while (iterator.hasNext()) {
					count++;
					String key = (String) iterator.next();
					alist = (count == 1) ? key : alist + ", " + key;
				}
				if (alist != null) {
					sinfo += " and logid not in (" + alist + ")";
				}

				sql = "delete from "+serverProps.createRelName("_RSERV_LOG_")+" where "
						+ "logtime < now() - '" + howold
						+ " second'::interval and " + sinfo;
				getLogger().debug("MasterProcessor::cleanLog: sql(2)=" + sql);
				stmt.executeUpdate(sql);
			}
			rs.close();
		} catch (Exception e) {
			getLogger().error("MasterProcessor::cleanLog: ", e);
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}
		}
	}

	/**
	 *  Gets the activeLog attribute
	 *
	 * @return                     The activeLog value
	 * @exception  RservException  Description of the Exception
	 */
	protected int getActiveLog() throws RservException {
		return getActiveLog(false);
	}

	/**
	 *  Gets the errorPrefix attribute
	 *
	 * @return    The errorPrefix value
	 */
	protected String getErrorPrefix() {
		return "MasterProcessor:: ";
	}


	/* MAP oid --> tabname, keyname
     * retrieve it once since table struct does not change
    */
	/**
	 *  Gets the tableMap attribute
	 *
	 * @return                     The tableMap value
	 * @exception  RservException  Description of the Exception
	 */
	protected TableMap getTableMap() throws RservException {
		if (getDebug()) {
			getLogger().debug("MasterProcessor::getTableMap: BEGIN");
		}
		if (tableMap != null) {
			if (getDebug()) {
				getLogger().debug("MasterProcessor::getTableMap:  tableMap=" +
						tableMap);
			}
			return tableMap;
		}

		if (getDebug()) {
			getLogger().debug("MasterProcessor::getTableMap: !tableMapSet - let's set it");
		}
		ResultSet rs = null;
		Statement stmt = null;

		/* SQL_TABLE_MAP =
        sql = "select pgc.oid, pgc.relname, pga.attname" +
        " from _RSERV_TABLES_ rt, pg_class pgc, pg_attribute pga" +
        " where pgc.oid = rt.reloid and pga.attrelid = rt.reloid" +
        " and pga.attnum = rt.key";
        */
		try {
			tableMap = new TableMap();
			stmt = getConnection().createStatement();
			rs = stmt.executeQuery(SQL_TABLE_MAP);
			getLogger().debug("MasterProcessor::getTableMap: executed sql=" + SQL_TABLE_MAP);
			int row = 0;
			while (rs.next()) {
				row++;
				String oid = rs.getString(1);
				String relName = rs.getString(2);
				String attName = rs.getString(3);
				String keyType = rs.getString(4);

				// Adding element with row key == oid
				tableMap.addElement(oid).
						setOid(oid).
						setRelName(relName).
						setAttName(attName).
						setKeyType(keyType);
			}

			if (getDebug()) {
				getLogger().debug("MasterProcessor::getTableMap: retrieved rows=" + row);
			}
			rs.close();
		} catch (Exception e) {
			throw new RservException(
					"MasterProcessor::getTableMap: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}
		}

		return tableMap;
	}

	/*
    */
	/**
	 *  Description of the Method
	 *
	 * @exception  RservException  Description of the Exception
	 */
	protected void initPrepare() throws RservException {
		markLog();
		getTableMap();
		getUSTable();
	}

	/*
    * @param - howold time in sec how long keep the log record
    */
	/**
	 *  Description of the Method
	 *
	 * @return                     Return Value
	 * @exception  RservException  Description of the Exception
	 */
	protected int prepareCleanLog() throws RservException {
		int olt = 0;
		getLogger().debug("MasterProcessor::prepareCleanLog BEGIN");
		try {
			lock(EXCLUSIVE);
			int alt = getActiveLog(true);
			olt = notAlt(alt);
			useolt = useOldLogState();

			if (useolt == TRUNCATE) {// Old log is clean - switch logs
				setLogStatus(USE_OLD_NEW_LOG);
				changeActiveLog(olt);
				alt = olt;
				olt = notAlt(alt);
			}

			//
			// Else - do not switch logs and cleanup old log
			//
			unlock(EXCLUSIVE);
			//
			// Make sure that no triggers use old log table.
			// Note that triggers always place ROW EXCLUSIVE lock
			// on active table while holding share "altid" lock,
			// so *we wait for triggers* here and do not block
			// any triggers while waiting. "Prepare" uses ROW SHARE
			// lock on old log - so no conflict at this point.
			//
			lockTable(olt);

			//
			// Got SHARED lock ==> all ROW EXCLUSIVE LOCKs from triggers
			// released, ie triggers now use new active table only
			//

		} catch (Exception e) {
			getLogger().error("MasterProcessor::prepareCleanLog: ", e);
			throw new RservException(
					"MasterProcessor::prepareCleanLog: " + e.toString());
		} finally {
		}

		return olt;
	}

	/*
    */
	/**
	 *  Description of the Method
	 *
	 * @return                     Return Value
	 * @exception  RservException  Description of the Exception
	 */
	protected boolean prepareTables() throws RservException {

		boolean gotdata = false;

		try {
			TableMap tableMap = getTableMap();
			tableMap.setIterator();

			while (tableMap.hasNext()) {
				String taboid = tableMap.nextRowKey();
				tableMap.setRow(taboid);
				String tabname = tableMap.getRelName();
				String tabkey = tableMap.getAttName();
				String keytype = tableMap.getKeyType();
				String oidkey = (tabkey.equals("oid")) ? "_" + tabname + ".oid," : "";
				if (getDebug()) {
					getLogger().debug("MasterProcessor::prepareTables inside iterator while: taboid="
							+ taboid + "; tabname=" + tabname + "; tabkey=" + tabkey +
							"; oidkey=" + oidkey);
				}
				boolean gotdataPart = PrepareTable(taboid, tabname, tabkey, keytype);
				gotdata = !gotdata ? gotdataPart : gotdata;

			}
		} catch (Exception e) {
			throw new RservException(
					"MasterProcessor::prepareTables: " + e.toString());
		}

		return gotdata;
	}

	/* Remember this snapshot info
    */
	/**
	 *  Description of the Method
	 *
	 * @exception  RservException  Description of the Exception
	 */
	protected void rememberSnapshot() throws RservException {
		validateInit();
		String sql_2 = "select "+serverProps.createRelName("_rserv_sync_")+"(" + server + ")";
		ResultSet rs = null;
		Statement stmt = null;

		try {
			stmt = getConnection().createStatement();
			rs = stmt.executeQuery(sql_2);
			snapshotRemembered = true;
			rs.close();
		} catch (Exception e) {
			throw new RservException("MasterProcessor::rememberSnapshot: "
					+ e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}

		}
	}


	// update pg_class set relpages=1000,reltuples=100000 where relname = [your_table_name];

	/**
	 *  Description of the Method
	 *
	 * @return                     Return Value
	 * @exception  RservException  Description of the Exception
	 */
	protected boolean useOldLogTable() throws RservException {
		return useOldLogTable(false);
	}

	/* Get syncid for new snapshot
    */
	/**
	 *  Description of the Method
	 *
	 * @return                     Return Value
	 * @exception  RservException  Description of the Exception
	 */
	private long GetSYNCID() throws RservException {
		if (syncid != T_NO) {
			return syncid;
		}
		ResultSet rs = null;
		Statement stmt = null;
		String sql = "select nextval('"+serverProps.createRelName("_rserv_sync_seq_")+"')";
		getLogger().debug("MasterProcessor::GetSYNCID: sql=" + sql);
		long rc = T_NO;

		try {
			stmt = getConnection().createStatement();
			rs = stmt.executeQuery(sql);
			while (rs.next()) {
				rc = rs.getLong(1);
			}
			if (rc <= 0) {
				throw new RservException(
						"Cannot get nextval('_rserv_sync_seq_')");
			}

			CommandType comType = new SnapshotCommand(CommandType.SYNCID);
			SnapshotDataBlock snapDataBlock = new SnapshotDataBlock(comType,
					(new Long(rc)).toString());
			snapData.addBlock(snapDataBlock);
		} catch (Exception e) {
			throw new RservException(
					"MasterProcessor::GetSYNCID: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}
		}

		return rc;
	}


	/*
    */
	/**
	 *  Description of the Method
	 *
	 * @param  taboid              Parameter
	 * @param  tabname             Parameter
	 * @param  tabkey              Parameter
	 * @param  keytype             Parameter
	 * @return                     Return Value
	 * @exception  RservException  Description of the Exception
	 * @exception  Exception       Description of the Exception
	 */
	private boolean PrepareTable(String taboid, String tabname, String tabkey, String keytype)
			throws RservException, Exception {

		String cTable = null;
		String oTable = null;

		if (getActiveLog() == ALT_1) {
			cTable = LOG_1;
			oTable = LOG_2;
		} else if (getActiveLog() == ALT_2) {
			cTable = LOG_2;
			oTable = LOG_1;
		} else {
			throw new RservException("MasterProcessor::PrepareTable: unsexpected value for getActiveLog=" +
					getActiveLog());
		}

		// Declare cursor to read from current log:
		//declareLogCursor(taboid, tabname, tabkey, keytype, cTable, oTable, false);
		// boolean gotdata = ReadData(taboid, tabname, tabkey, keytype, cTable, oTable, false);

		boolean gotdata = false;
		if (useOldLogTable()) {
			getLogger().debug("MasterProcessor::PrepareTable -- I AM GOING TO USE OLD LOG --");
			//declareLogCursor(taboid, tabname, tabkey, keytype, cTable, oTable, true);
			gotdata = ReadData(taboid, tabname, tabkey, keytype, oTable, cTable, true);
		}
		gotdata = gotdata | ReadData(taboid, tabname, tabkey, keytype, cTable, oTable, false);

		if (getDebug()) {
			getLogger().debug("MasterProcessor::PrepareTable gotdata=" + gotdata);
		}

		return gotdata;
	}

	/*  Main method to build the Snapshot - reads data in snapData
    */
	/**
	 *  Description of the Method
	 *
	 * @param  taboid              Parameter
	 * @param  tabname             Parameter
	 * @param  tabkey              Parameter
	 * @param  keytype             Parameter
	 * @param  cTable              Parameter
	 * @param  oTable              Parameter
	 * @param  useOldLogTable      Parameter
	 * @return                     Return Value
	 * @exception  RservException  Description of the Exception
	 */
	private boolean ReadData(
			String taboid,
			String tabname,
			String tabkey,
			String keytype,
			String cTable,
			String oTable,
			boolean useOldLogTable) throws RservException {

		ResultSet rs = null;
		boolean gotdata = false;
		SnapshotDataBlock deletedBlock = null;
		SnapshotDataBlock updatedBlock = null;
		SnapshotDataBlock insertedBlock = null;
		int columnCount = 0;

//		sql = "select ";
		long syncid = 0;
		ArrayList usTable = getUSTable();
		int row = 0;

		getLogger().debug("MasterProcessor::ReadData BEGIN");
		try {
			//for ( ; ; ) {
			//rs = fetchCursor(FETCH_MAX); // this is table specific

			rs = executeLogCursor(taboid, tabname, tabkey, keytype, cTable, oTable, useOldLogTable);
			columnCount = rs.getMetaData().getColumnCount();
			// columnCount = getTableColumnCache().size(tabname) + 3;

			while (rs.next()) {
				//
				// Find to what SyncID this change belongs:
				//
				long logid = rs.getLong(1);// $row[0]
				int deleted = rs.getInt(2);// $row[1]
				boolean useoid = (tabkey == "oid");
				String key = rs.getString(3);// $row[2]
				getLogger().debug("MasterProcessor::ReadData logid=" + logid + "; deleted=" + deleted + "; key=" + key + "; tabkey=" + tabkey + "; keytype=" + keytype);

				syncid = getSnapshotSyncid(logid, usTable);
				//syncid = syncid == null ? SnapshotDataRow.THIS : syncid; // this snapshot

				if (deleted > 0) {// deleted
					getLogger().debug("MasterProcessor::ReadData ADDING DELETED TO SNAP DATA");
					deletedBlock = getDataBlock(deletedBlock,
							columnCount, CommandType.DELETE, tabname, key, useoid, syncid, rs);
					row++;
				} else if (deleted == 0) {// update
					getLogger().debug("MasterProcessor::ReadData ADDING UPDATED TO SNAP DATA");
					updatedBlock = getDataBlock(updatedBlock,
							columnCount, CommandType.UPDATE, tabname, key, useoid, syncid, rs);
					row++;
				} else {// insert
					getLogger().debug("MasterProcessor::ReadData ADDING INSERTED TO SNAP DATA");
					insertedBlock = getDataBlock(insertedBlock,
							columnCount, CommandType.INSERT, tabname, key, useoid, syncid, rs);
					row++;
				}
			}// end of while
			//

			//}
			//if (getDebug())
			getLogger().debug("MasterProcessor::ReadData row=" + row);
			if (row > 0) {
				GetSYNCID();
				if (deletedBlock != null) {
					snapData.addBlock(deletedBlock);
				}
				if (insertedBlock != null) {
					snapData.addBlock(insertedBlock);
				}
				if (updatedBlock != null) {
					snapData.addBlock(updatedBlock);
				}
				gotdata = true;
			}

		} catch (Exception e) {
			e.printStackTrace();
			throw new RservException(
					"MasterProcessor::ReadData: " + e.toString());
		} finally {
			try {
				Statement s = rs.getStatement();
				if (s != null) {
					s.close();
				}
			} catch (Exception ex) {
			}
		}

		return gotdata;
	}

	/*
    */
	/* change active log
    */
	/**
	 *  Description of the Method
	 *
	 * @param  logId               Parameter
	 * @exception  RservException  Description of the Exception
	 */
	private void changeActiveLog(int logId) throws RservException {
		PreparedStatement stmt = null;

		String sql = "select setval('"+serverProps.createRelName("_rserv_active_log_id_'")+", " + logId + ")";
		getLogger().debug("MasterProcessor::changeActiveLog: sql=" + sql);

		try {
			stmt = getConnection().prepareStatement(sql);
			stmt.execute();
			getLogger().info("MasterProcessor::changeActiveLog: LOG " + logId + " IS ACTIVE NOW");

		} catch (Exception e) {
			getLogger().error("MasterProcessor::changeActiveLog: ", e);
			throw new RservException("MasterProcessor::changeActiveLog: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}
		}
	}

	// Template method for query processing:

	/*
    */
	/**
	 *  Description of the Method
	 *
	 * @exception  RservException  Description of the Exception
	 */
	private void execTemplate() throws RservException {
		ResultSet rs = null;
		PreparedStatement stmt = null;
		String sql = null;

		sql = "select ";

		getLogger().debug("MasterProcessor:: sql=" + sql);
		try {
			stmt = getConnection().prepareStatement(sql);
			stmt.setInt(1, server);
			stmt.setInt(2, server);
			rs = stmt.executeQuery();
			while (rs.next()) {
				//String syncid = rs.getString(1);
				// String lastSynctime = rs.getString(2);

				getLogger().debug("");
			}

		} catch (Exception e) {
			throw new RservException(
					"MasterProcessor::: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}

		}
	}


	/* Declare 2 log cursor cursor (current & old):
    */
	/**
	 *  Description of the Method
	 *
	 * @param  taboid              Parameter
	 * @param  tabname             Parameter
	 * @param  tabkey              Parameter
	 * @param  keytype             Parameter
	 * @param  cTable              Parameter
	 * @param  oTable              Parameter
	 * @param  useOldLogTable      Parameter
	 * @return                     Return Value
	 * @exception  RservException  Description of the Exception
	 */
	private ResultSet executeLogCursor(
			String taboid,
			String tabname,
			String tabkey,
			String keytype,
			String cTable,
			String oTable,
			boolean useOldLogTable) throws RservException {
		ResultSet rs = null;
		Statement stmt = null;

		StringBuffer sqlBuff = new StringBuffer();
		//sqlBuff.append("DECLARE c cursor FOR ")
		sqlBuff.append(" SELECT l.logid, l.deleted, l.key, _")
				.append(tabname.replace('.', '_'))
				.append(".* ")
				.append(" FROM ")
				.append(cTable)
				.append(" l LEFT JOIN ")
				.append(tabname)
				.append(" _")
				.append(tabname.replace('.', '_'))
				.append(" ON l.key::")
				.append(keytype)
				.append("= _")
				.append(tabname.replace('.', '_'))
				.append(".")
				.append(tabkey)
				.append(" WHERE l.reloid = ")
				.append(taboid)
				.append(" ")
				.append(getSinfo());

		if (useOldLogTable) {
			sqlBuff.append(" ")
					.append("and not exists (")
					.append("select lc.logid from ")
					.append(oTable)
					.append(" lc where lc.reloid = ")
					.append(taboid)
					.append(" and lc.key = l.key)");
		}

		String sql = sqlBuff.toString();
		getLogger().debug("MasterProcessor::executeLogCursor: sql=" + sql);

		try {
			stmt = getConnection().createStatement();
			rs = stmt.executeQuery(sql);

		} catch (Exception e) {
			System.err.println(sql);
			throw new RservException(
					"MasterProcessor::executeLogCursor: " + e.toString());
		} finally {

		}

		return rs;
	}

	/*
    */
	/**
	 *  Description of the Method
	 *
	 * @param  numRows             Parameter
	 * @return                     Return Value
	 * @exception  RservException  Description of the Exception
	 */
	private ResultSet fetchCursor(int numRows) throws RservException {
		if (numRows < 0) {
			throw new RservException("MasterProcessor::fetchCursor: numRows < 0");
		}

		ResultSet rs = null;
		PreparedStatement stmt = null;
		String sql = "fetch " + numRows + " in c";
		getLogger().debug("MasterProcessor::fetchCursor: sql=" + sql);

		try {
			stmt = getConnection().prepareStatement(sql);
			rs = stmt.executeQuery();
		} catch (Exception e) {
			getLogger().error("MasterProcessor::fetchCursor: ", e);
			rs = null;
		} finally {

		}

		return rs;
	}

	//private TableColumnCache getTableColumnCache() {
	//    return tableColumnCache;
	//}

	/*
    */
	/**
	 *  Gets the activeLog attribute
	 *
	 * @param  reset               Parameter
	 * @return                     The activeLog value
	 * @exception  RservException  Description of the Exception
	 */
	private int getActiveLog(boolean reset) throws RservException {
		if (!reset) {
			return activeLogTable;
		}

		ResultSet rs = null;
		PreparedStatement stmt = null;
		String sql = null;

		sql = "select last_value::int4 FROM "+serverProps.createRelName("_rserv_active_log_id_");

		getLogger().debug("MasterProcessor::getActiveLog: sql=" + sql);
		try {
			stmt = getConnection().prepareStatement(sql);
			rs = stmt.executeQuery();
			while (rs.next()) {
				activeLogTable = rs.getInt(1);

				if (getDebug()) {
					getLogger().debug("MasterProcessor::getActiveLog: activeLogTable=" +
							activeLogTable);
				}
			}
			rs.close();
		} catch (Exception e) {
			throw new RservException(
					"MasterProcessor::getActiveLog: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception e) {
			}
		}

		return activeLogTable;
	}


	/*
    */
	/**
	 *  Gets the dataBlock attribute
	 *
	 * @param  dataBlock           Parameter
	 * @param  columnCount         Parameter
	 * @param  command             Parameter
	 * @param  tabname             Parameter
	 * @param  key                 Parameter
	 * @param  useoid              Parameter
	 * @param  syncid              Parameter
	 * @param  rs                  Parameter
	 * @return                     The dataBlock value
	 * @exception  RservException  Description of the Exception
	 * @exception  Exception       Description of the Exception
	 */
	private SnapshotDataBlock getDataBlock(
			SnapshotDataBlock dataBlock,
			int columnCount,
			int command,
			String tabname,
			String key,
			boolean useoid,
			long syncid,
			ResultSet rs) throws RservException, Exception {

		//if (getDebug())
		getLogger().debug("\n MasterProcessor:: getDataBlock BEGIN; tabname=" + tabname +
				"; dataBlock=" + dataBlock + "; command=" + command +
				"\n");
		if (dataBlock == null) {
			dataBlock =
					new SnapshotDataBlock(
							new SnapshotCommand(command), tabname);
			getLogger().debug("\n MasterProcessor:: getDataBlock BEGIN; dataBlock == null - create new one for table " + tabname +
					"; command=" + command);
		}

		getLogger().debug(" MasterProcessor:: getDataBlock param=" + dataBlock.getParam() + "; key=" + key + "; syncid=" + syncid);

		SnapshotDataRow dataRow = new SnapshotDataRow(key, syncid, useoid);
		if (command == CommandType.UPDATE || command == CommandType.INSERT) {
			int i = (useoid) ? 3 : 4;
			try {
				for (; i <= columnCount; i++) {
					String data = rs.getString(i);
					dataRow.add(data);
				}
			} catch (Exception e) {
				getLogger().error("MasterProcessor:: getDataBlock: ", e);
				throw new RservException("MasterProcessor:: getDataBlock: illegal command=" + e.toString());
			}

		} else if (command != CommandType.DELETE) {
			throw new RservException("MasterProcessor:: getDataBlock: illegal command=" + command);
		}

		dataBlock.addRow(dataRow);
		getLogger().debug("MasterProcessor:: getDataBlock: dataBlock.size()=" + dataBlock.size() + "; key=" + dataBlock.getRow(dataBlock.size() - 1).getKeyInt());

		return dataBlock;
	}


	/*
     * Read last succeeded sync id that was synchronized with slave
    */
	/**
	 *  Gets the lastSync attribute
	 *
	 * @return                     The lastSync value
	 * @exception  RservException  Description of the Exception
	 */
	private LastSyncData getLastSync() throws RservException {
		if (lastSync != null) {
			return lastSync;
		}
		validateInit();
		lastSync = new LastSyncData();

		ResultSet rs = null;
		PreparedStatement stmt = null;
		String sql = null;

		sql = "select syncid, synctime, minid, maxid, active from "+serverProps.createRelName("_RSERV_SYNC_") +
				" where server = ? and syncid = (select max(syncid) from " +
				serverProps.createRelName("_RSERV_SYNC_")+" where server = ? and status > 0)";
		getLogger().debug("MasterProcessor::getLastSync sql=" + sql);

		try {
			stmt = getConnection().prepareStatement(sql);
			stmt.setInt(1, server);
			stmt.setInt(2, server);
			rs = stmt.executeQuery();
			while (rs.next()) {
				String lastSyncid = rs.getString(1);
				String lastSynctime = rs.getString(2);
				String lastMinid = rs.getString(3);
				String lastMaxid = rs.getString(4);
				String lastActive = rs.getString(5);
				if (lastActive == null) {
					getLogger().warn("MasterProcessor::getLastSync lastActive==null");
				}
				lastSync.setLastSyncid(lastSyncid);
				lastSync.setLastSynctime(lastSynctime);
				lastSync.setLastMinid(lastMinid);
				lastSync.setLastMaxid(lastMaxid);
				lastSync.setLastActive(lastActive);
				if (getDebug()) {
					getLogger().debug("MasterProcessor::getLastSync lastSyncid=" + lastSyncid
							+ "; lastSynctime=" + lastSynctime
							+ "; lastMinid=" + lastMinid
							+ "; lastMaxid=" + lastMaxid + "; getLastMaxid=" +
							lastSync.getLastMaxid()
							+ "; lastActive=" + lastActive);
				}
			}
		} catch (Exception e) {
			throw new RservException(
					"MasterProcessor::setLastSync: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}
		}

		return lastSync;
	}


	/**
	 *  Gets the sinfo attribute
	 *
	 * @return                     The sinfo value
	 * @exception  RservException  Description of the Exception
	 */
	private String getSinfo() throws RservException {
		if (sinfo != null) {
			if (getDebug()) {
				getLogger().debug("MasterProcessor::getSinfo sinfo(1)=" + sinfo);
			}
			return sinfo;
		}
		validateInit();
		LastSyncData lastSync = getLastSync();
		StringBuffer sinfoBuff = null;
		// sync info
		if (getDebug()) {
			getLogger().debug("MasterProcessor::getSinfo lastSync.getLastMaxid()=" +
					lastSync.getLastMaxid());
		}
		if (lastSync.getLastMaxid() != null &&
				lastSync.getLastMinid() != null) {
			sinfoBuff = new StringBuffer();
			sinfoBuff.append(" and l.logid >='")
					.append(lastSync.getLastMinid())
					.append("'::xxid and (l.logid >= '")
					.append(lastSync.getLastMaxid()).append("'::xxid");

			if (lastSync.getLastActive() != null &&
					!lastSync.getLastActive().trim().equals("")) {
				sinfoBuff.append(" or l.logid in (")
						.append(lastSync.getLastActive())
						.append(")");
			}
			sinfoBuff.append(")");
			sinfo = sinfoBuff.toString();
		}
		if (getDebug()) {
			getLogger().debug("MasterProcessor::getSinfo sinfo(2)=" + sinfo);
		}

		return (sinfo == null) ? "" : sinfo;
	}


	/*
       Gets last syncid that belongs to this snapshot
    */
	/**
	 *  Gets the snapshotSyncid attribute
	 *
	 * @param  logid    Parameter
	 * @param  usTable  Parameter
	 * @return          The snapshotSyncid value
	 */
	private long getSnapshotSyncid(long logid, ArrayList usTable) {
		long syncid = SnapshotDataRow.THIS;
		USTableRow row;
		for (int i = 0; i < usTable.size(); i++) {
			row = (USTableRow) usTable.get(i);
			 if (transactionIdPrecedes(logid, row.getMinid())) {
			//if (logid < row.getMinid()) {// LT than minid?
				syncid = row.getSyncid();// belongs to this
				break;
			}

			if (transactionIdPrecedes(row.getMaxid(), logid)) {
			//if (logid > row.getMaxid()) {// GE than maxid?
				continue;// belongs to the next
			}

			if (row.containsKey(
					new Long(logid).toString())) {// was active?
				continue;// belongs to the next
			}

			// (xid > minid) AND (xid < maxid) AND was committed
			syncid = row.getSyncid();
			break;
		}
		return syncid;
	}

	/*
     * Read last succeeded sync ids NEW - 2LOG
    */
	/**
	 *  Gets the uSTable attribute
	 *
	 * @return                     The uSTable value
	 * @exception  RservException  Description of the Exception
	 */
	private ArrayList getUSTable() throws RservException {
		if (usTable != null) {
			return usTable;
		}
		validateInit();
		usTable = new ArrayList();
		String lastSyncid = getLastSync().getLastSyncid();

		ResultSet rs = null;
		PreparedStatement stmt = null;
		String sql = null;

		sql = "select syncid, synctime, minid, maxid, active " +
				"from "+serverProps.createRelName("_RSERV_SYNC_")+" where server =? ";
		if (lastSyncid != null) {
			sql += " and syncid > " + lastSyncid;
		}
		sql += " order by minid";

		getLogger().debug("MasterProcessor::getUSTable sql=" + sql);
		try {
			stmt = getConnection().prepareStatement(sql);
			stmt.setInt(1, server);
			//stmt.setInt(2, server);
			rs = stmt.executeQuery();
			while (rs.next()) {
				USTableRow usRow = new USTableRow();
				int syncid = rs.getInt(1);// just syncid - may not be confirmed by slave
				long minid = rs.getLong(3);
				long maxid = rs.getLong(4);
				String active = rs.getString(5);

				usRow.setSyncid(syncid);
				usRow.setMinid(minid);
				usRow.setMaxid(maxid);

				if (active != null) {
					StringTokenizer st = new StringTokenizer(active, ACTIVE_SEP);
					while (st.hasMoreTokens()) {
						String token = st.nextToken().trim();
						if (!usRow.containsKey(token)) {
							usRow.putActive(token);
						}
					}
				} else {
					getLogger().warn("MasterProcessor::getUSTable active==null");
				}

				// Add row to Us table:
				usTable.add(usRow);

				if (getDebug()) {
					getLogger().debug("MasterProcessor::getUSTable lastSyncid=" + lastSyncid
							+ "; minid=" + minid + "; maxid=" + maxid + "; getLastMaxid="
							+ lastSync.getLastMaxid() + "; active=" + active);
				}
			}
		} catch (Exception e) {
			throw new RservException(
					"MasterProcessor::getLastMaxid: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}
		}

		return usTable;
	}

	/* modes are  SHARED & EXCLUSIVE
    */
	/**
	 *  Description of the Method
	 *
	 * @param  mode                Parameter
	 * @exception  RservException  Description of the Exception
	 */
	private void lock(int mode) throws RservException {
		ResultSet rs = null;
		PreparedStatement stmt = null;
		String sql = null;
		sql = "select "+serverProps.createRelName("_rserv_lock_altid_")+"(" + mode + ")";

		getLogger().debug("MasterProcessor::lock: sql=" + sql);
		try {
			stmt = getConnection().prepareStatement(sql);
			rs = stmt.executeQuery();
			rs.close();
		} catch (Exception e) {
			throw new RservException(
					"MasterProcessor::lock: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception e) {
			}
		}
	}


	/*
    */
	/**
	 *  Description of the Method
	 *
	 * @exception  RservException  Description of the Exception
	 */
	private void lockTable()
			throws RservException {
		PreparedStatement stmt = null;

		try {
			if (useOldLogTable()) {
				String sql = "LOCK TABLE "
						+ ((getActiveLog() == ALT_1) ? LOG_2 : LOG_1)
						+ " IN ROW SHARE MODE";
				getLogger().debug("MasterProcessor::lockTable: sql=" + sql);
				stmt = getConnection().prepareStatement(sql);
				stmt.executeUpdate();
			}
		} catch (Exception e) {
			throw new RservException(
					"MasterProcessor::lockTable: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}
		}
	}

	/*
    */
	/**
	 *  Description of the Method
	 *
	 * @param  logId               Parameter
	 * @exception  RservException  Description of the Exception
	 */
	private void lockTable(int logId)
			throws RservException {
		PreparedStatement stmt = null;
		String table = null;

		if (logId == ALT_1) {
			table = LOG_1;
		} else if (logId == ALT_2) {
			table = LOG_2;
		} else {
			throw new RservException(
					"MasterProcessor::lockTable: wrong logId = " + logId);
		}

		try {
			String sql = "LOCK TABLE " + table + " IN SHARE MODE";
			getLogger().debug("MasterProcessor::lockTable: sql=" + sql);
			stmt = getConnection().prepareStatement(sql);
			stmt.executeUpdate();
		} catch (Exception e) {
			throw new RservException(
					"MasterProcessor::lockTable: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}
		}
	}


	/*
    */
	/**
	 *  Description of the Method
	 *
	 * @exception  RservException  Description of the Exception
	 */
	private void markLog() throws RservException {
		lock(SHARED);
		activeLogTable = getActiveLog(true);
		useOldLogTable = useOldLogTable(true);
		lockTable();
		unlock(SHARED);
	}

	/* Vadim calls it $olt
    */
	/**
	 *  Description of the Method
	 *
	 * @param  alt  Parameter
	 * @return      Return Value
	 */
	private int notAlt(int alt) {
		return (alt == ALT_1) ? ALT_2 : ALT_1;
	}

	/**
	 *  Description of the Method
	 *
	 * @param  logId               Parameter
	 * @exception  RservException  Description of the Exception
	 * @exception  SQLException    Description of the Exception
	 */
	private void postTruncateTable(int logId)
			throws RservException, SQLException {

		ResultSet rs = null;
		PreparedStatement stmt = null;
		String table = null;

		if (logId == ALT_1) {
			table = LOG_1;
		} else if (logId == ALT_2) {
			table = LOG_2;
		} else {
			throw new RservException(
					"MasterProcessor::postTruncateTable: wrong logId = " + logId);
		}

		try {
			// We have to explicitely set autocommit to true
			// To pass truncate since it cannot be run inside of transaction:
			getConnection().setAutoCommit(true);

			String sql = "update pg_class set relpages=1000,reltuples=100000 where relname = '" + table + "'";
			getLogger().debug("MasterProcessor::postTruncateTable: sql=" + sql);
			stmt = getConnection().prepareStatement(sql);
			stmt.executeUpdate();
		} catch (Exception e) {
			getLogger().error("MasterProcessor::postTruncateTable: " + e.toString());
		} finally {
			if (stmt != null) {
				stmt.close();
			}
			getConnection().setAutoCommit(false);
		}
	}

	/**
	 *  Sets the logStatus attribute
	 *
	 * @param  status              The new logStatus value
	 * @exception  RservException  Description of the Exception
	 */
	private void setLogStatus(int status) throws RservException {
		PreparedStatement stmt = null;

		String sql = "select setval('"+serverProps.createRelName("_rserv_old_log_status_")+"', " + status + ")";
		getLogger().debug("MasterProcessor::setLogStatus: sql=" + sql);

		try {
			stmt = getConnection().prepareStatement(sql);
			stmt.execute();

		} catch (Exception e) {
			getLogger().error("MasterProcessor::setLogStatus: ", e);
			throw new RservException("MasterProcessor::setLogStatus: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}
		}

	}

	/**
	 *  Description of the Method
	 *
	 * @param  logId               Parameter
	 * @exception  RservException  Description of the Exception
	 * @exception  SQLException    Description of the Exception
	 */
	private void truncateTable(int logId)
			throws RservException, SQLException {

		PreparedStatement stmt = null;
		String table = null;

		if (logId == ALT_1) {
			table = LOG_1;
		} else if (logId == ALT_2) {
			table = LOG_2;
		} else {
			throw new RservException(
					"MasterProcessor::truncateTable: wrong logId = " + logId);
		}

		try {
			// We have to explicitely set autocommit to true
			// To pass truncate since it cannot be run inside of transaction:
			getConnection().setAutoCommit(true);

			String sql = "TRUNCATE TABLE  " + table;
			getLogger().debug("MasterProcessor::truncateTable: sql=" + sql);
			stmt = getConnection().prepareStatement(sql);
			stmt.executeUpdate();
		} catch (Exception e) {
			throw new RservException(
					"MasterProcessor::truncateTable: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}

			getConnection().setAutoCommit(false);
		}
	}

	/* modes are  SHARED=0 & EXCLUSIVE=1
    */
	/**
	 *  Description of the Method
	 *
	 * @param  mode                Parameter
	 * @exception  RservException  Description of the Exception
	 */
	private void unlock(int mode) throws RservException {
		ResultSet rs = null;
		PreparedStatement stmt = null;
		String sql = "select "+serverProps.createRelName("_rserv_unlock_altid_")+"(" + mode + ")";

		getLogger().debug("MasterProcessor::unlock sql=" + sql);
		try {
			stmt = getConnection().prepareStatement(sql);
			rs = stmt.executeQuery();
			rs.close();
		} catch (Exception e) {
			throw new RservException(
					"MasterProcessor::unlock: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception e) {
			}
		}
	}


	/*
    */
	/**
	 *  Description of the Method
	 *
	 * @return                     Return Value
	 * @exception  RservException  Description of the Exception
	 */
	private int useOldLogState() throws RservException {

		ResultSet rs = null;
		PreparedStatement stmt = null;
		String sql = null;
		int use = 0;

		sql = "select last_value::int4 FROM "+serverProps.createRelName("_rserv_old_log_status_");

		getLogger().debug("MasterProcessor::useOldLogState: sql=" + sql);
		try {
			stmt = getConnection().prepareStatement(sql);
			rs = stmt.executeQuery();
			int row = 0;
			while (rs.next()) {
				row++;
				use = rs.getInt(1);
				if (getDebug()) {
					getLogger().debug("MasterProcessor::useOldLogState (2) USE OLD LOG ? : " + useOldLogTable);
				}
			}
			if (row == 0 || use < 0 || use > 2) {
				throw new RservException(" row == 0 || use < 0 || use >2");
			}
		} catch (Exception e) {
			throw new RservException("MasterProcessor::useOldLogState: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}
		}

		return use;
	}

	/*
    */
	/**
	 *  Description of the Method
	 *
	 * @param  reset               Parameter
	 * @return                     Return Value
	 * @exception  RservException  Description of the Exception
	 */
	private boolean useOldLogTable(boolean reset) throws RservException {
		if (!reset) {
			if (getDebug()) {
				getLogger().debug("MasterProcessor::useOldLogTable (1) USE OLD LOG ? : " + useOldLogTable);
			}
			return useOldLogTable;
		}
		ResultSet rs = null;
		PreparedStatement stmt = null;
		String sql = null;

		sql = "select last_value::int4 FROM "+serverProps.createRelName("_rserv_old_log_status_");

		getLogger().debug("MasterProcessor::useOldLogTable: sql=" + sql);
		try {
			stmt = getConnection().prepareStatement(sql);
			rs = stmt.executeQuery();
			while (rs.next()) {
				int use = rs.getInt(1);
				if (use < 0 || use > 2) {
					throw new RservException(" use < 0 || use >2");
				}

				useOldLogTable = (use == USE) ? true : false;
				if (getDebug()) {
					getLogger().debug("MasterProcessor::useOldLogTable (2) USE OLD LOG ? : " + useOldLogTable);
				}
			}
		} catch (Exception e) {
			throw new RservException("MasterProcessor::useOldLogTable: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}
		}

		return useOldLogTable;
	}

	/* Now wait as long as there are unreplicated records in old log
    */
	/**
	 *  Description of the Method
	 *
	 * @param  logId               Parameter
	 * @param  tout                Parameter
	 * @exception  RservException  Description of the Exception
	 */
	private void waitUnreplicated(int logId, int tout) throws RservException {
		ResultSet rs = null;
		ResultSet rsLogid = null;
		Statement stmt = null;
		Statement stmtLogid = null;
		String sql = null;
		Hashtable sinfoHash = new Hashtable();
		sql = "";

		try {
			for (; ;) {
				String server = "";
				setTransactionSerializable();
				sql = "select s.server, s.minid, s.maxid, s.active " +
						"from "+serverProps.createRelName("_RSERV_SYNC_")+" s " +
						"where s.syncid = (select max(s2.syncid) from " +
						serverProps.createRelName("_RSERV_SYNC_")+" s2 where s2.server = s.server and s2.status > 0)";
				getLogger().debug("MasterProcessor::waitUnreplicated: sql(1)=" + sql);
				stmt = getConnection().createStatement();
				rs = stmt.executeQuery(sql);

				sinfoHash.clear();
				String sqlAnd = "";
				while (rs.next()) {
					server = rs.getString(1);
					long minid = rs.getLong(2);
					long maxid = rs.getLong(3);
					String active = rs.getString(4);

					sqlAnd = "and l.logid >= '" + minid + "'::xxid and (l.logid >= '" + maxid+"'::xxid";
					sqlAnd += (active != null && !active.equals("")) ?
							" or l.logid in (" + active + ")" : "";
					sqlAnd += ")";
					if (sinfoHash.contains(server)) {
						sinfoHash.remove(server);
					}

					sinfoHash.put(server, sqlAnd);
				}

				boolean found = false;
				//
				// For each server and for each table
				//
				Iterator sinfoIterator = sinfoHash.keySet().iterator();
				while (sinfoIterator.hasNext()) {
					server = (String) sinfoIterator.next();
					getTableMap().setIterator();
					while (getTableMap().hasNext()) {
						String taboid = tableMap.nextRowKey();
						sql = "select l.logid from "+serverProps.createRelName("_RSERV_LOG_" + logId + "_")+" l "
								+ "WHERE l.reloid = " + taboid +" ";
						sql += (String) sinfoHash.get(server);
						sql += " LIMIT 1";
						getLogger().debug("MasterProcessor::waitUnreplicated: sql(2)=" + sql);
						stmtLogid = getConnection().createStatement();
						rsLogid = stmt.executeQuery(sql);
						while (rsLogid.next()) {
							found = true;
							break;
						}
						if (found) {
							break;
						}
					}
					if (found) {
						break;
					}
				}

				rollback();

				if (!found) {
					break;
				}
				Thread.sleep(tout);
			}// end of for (;;;) loop

		} catch (Exception e) {
			e.printStackTrace();
			throw new RservException(
					"MasterProcessor::waitUnreplicated: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}
			try {
				if (stmtLogid != null) {
					stmtLogid.close();
				}
			} catch (Exception ex) {
			}

		}
	}

	public void resetTableMap() {
		resetTableMap = true;
	}


}

