/*
 * 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.
 */
package org.teiid.odbc;

import static org.teiid.odbc.PGUtil.*;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.charset.Charset;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.ietf.jgss.GSSCredential;
import org.teiid.adminapi.VDB;
import org.teiid.adminapi.impl.SessionMetadata;
import org.teiid.client.RequestMessage.ResultsMode;
import org.teiid.client.security.ILogon;
import org.teiid.client.security.LogonException;
import org.teiid.client.util.ResultsFuture;
import org.teiid.core.TeiidProcessingException;
import org.teiid.core.util.ApplicationInfo;
import org.teiid.core.util.EquivalenceUtil;
import org.teiid.core.util.PropertiesUtils;
import org.teiid.core.util.StringUtil;
import org.teiid.core.util.TimestampWithTimezone;
import org.teiid.deployers.PgCatalogMetadataStore;
import org.teiid.dqp.service.SessionService;
import org.teiid.jdbc.ConnectionImpl;
import org.teiid.jdbc.PreparedStatementImpl;
import org.teiid.jdbc.ResultSetImpl;
import org.teiid.jdbc.StatementImpl;
import org.teiid.jdbc.TeiidDriver;
import org.teiid.jdbc.TeiidSQLException;
import org.teiid.logging.LogConstants;
import org.teiid.logging.LogManager;
import org.teiid.net.TeiidURL;
import org.teiid.net.socket.AuthenticationType;
import org.teiid.net.socket.SocketServerConnection;
import org.teiid.odbc.ODBCClientRemote.CursorDirection;
import org.teiid.odbc.PGUtil.PgColInfo;
import org.teiid.runtime.RuntimePlugin;
import org.teiid.security.GSSResult;
import org.teiid.transport.LocalServerConnection;
import org.teiid.transport.LogonImpl;
import org.teiid.transport.ODBCClientInstance;
import org.teiid.transport.PgBackendProtocol;
import org.teiid.transport.PgFrontendProtocol.NullTerminatedStringDataInputStream;
import org.teiid.transport.pg.TimestampUtils;

/**
 * While executing the multiple prepared statements I see this bug currently
 * http://pgfoundry.org/tracker/?func=detail&atid=538&aid=1007690&group_id=1000125
 */
public class ODBCServerRemoteImpl implements ODBCServerRemote {
	
    private static final boolean HONOR_DECLARE_FETCH_TXN = PropertiesUtils.getBooleanProperty(System.getProperties(), "org.teiid.honorDeclareFetchTxn", false); //$NON-NLS-1$

	public static final String CONNECTION_PROPERTY_PREFIX = "connection."; //$NON-NLS-1$
	private static final String UNNAMED = ""; //$NON-NLS-1$
	private static Pattern setPattern = Pattern.compile("set\\s+(\\w+)\\s+to\\s+((?:'[^']*')+)", Pattern.DOTALL|Pattern.CASE_INSENSITIVE);//$NON-NLS-1$
	
	private static Pattern columnMetadataPattern = Pattern.compile("select n.nspname, c.relname, a.attname, a.atttypid, t.typname, a.attnum, a.attlen, a.atttypmod, a.attnotnull, " //$NON-NLS-1$
			+ "c.relhasrules, c.relkind, c.oid, pg_get_expr\\(d.adbin, d.adrelid\\), case t.typtype when 'd' then t.typbasetype else 0 end, t.typtypmod, c.relhasoids " //$NON-NLS-1$
			+ "from \\(\\(\\(pg_catalog.pg_class c inner join pg_catalog.pg_namespace n on n.oid = c.relnamespace and c.oid = (\\d+)\\) inner join pg_catalog.pg_attribute a " //$NON-NLS-1$
			+ "on \\(not a.attisdropped\\) and a.attnum > 0 and a.attrelid = c.oid\\) inner join pg_catalog.pg_type t on t.oid = a.atttypid\\) left outer join pg_attrdef d " //$NON-NLS-1$
			+ "on a.atthasdef and d.adrelid = a.attrelid and d.adnum = a.attnum order by n.nspname, c.relname, attnum"); //$NON-NLS-1$
	
	private static Pattern pkPattern = Pattern.compile("select ta.attname, ia.attnum, ic.relname, n.nspname, tc.relname " +//$NON-NLS-1$
			"from pg_catalog.pg_attribute ta, pg_catalog.pg_attribute ia, pg_catalog.pg_class tc, pg_catalog.pg_index i, " +//$NON-NLS-1$
			"pg_catalog.pg_namespace n, pg_catalog.pg_class ic where tc.relname = (E?(?:'[^']*')+) AND n.nspname = (E?(?:'[^']*')+).*", Pattern.DOTALL|Pattern.CASE_INSENSITIVE);//$NON-NLS-1$
	
	private static Pattern pkKeyPattern = Pattern.compile("select ta.attname, ia.attnum, ic.relname, n.nspname, NULL from " + //$NON-NLS-1$
			"pg_catalog.pg_attribute ta, pg_catalog.pg_attribute ia, pg_catalog.pg_class ic, pg_catalog.pg_index i, " + //$NON-NLS-1$
			"pg_catalog.pg_namespace n where ic.relname = (E?(?:'[^']*')+) AND n.nspname = (E?(?:'[^']*')+) .*", Pattern.DOTALL|Pattern.CASE_INSENSITIVE); //$NON-NLS-1$
	
	private Pattern fkPattern = Pattern.compile("select\\s+((?:'[^']*')+)::name as PKTABLE_CAT," + //$NON-NLS-1$
			"\\s+n2.nspname as PKTABLE_SCHEM," +  //$NON-NLS-1$
			"\\s+c2.relname as PKTABLE_NAME," +  //$NON-NLS-1$
			"\\s+a2.attname as PKCOLUMN_NAME," +  //$NON-NLS-1$
			"\\s+((?:'[^']*')+)::name as FKTABLE_CAT," +  //$NON-NLS-1$
			"\\s+n1.nspname as FKTABLE_SCHEM," +  //$NON-NLS-1$
			"\\s+c1.relname as FKTABLE_NAME," +  //$NON-NLS-1$
			"\\s+a1.attname as FKCOLUMN_NAME," +  //$NON-NLS-1$
			"\\s+i::int2 as KEY_SEQ," +  //$NON-NLS-1$
			"\\s+case ref.confupdtype" +  //$NON-NLS-1$
			"\\s+when 'c' then (\\d)::int2" +  //$NON-NLS-1$
			"\\s+when 'n' then (\\d)::int2" +  //$NON-NLS-1$
			"\\s+when 'd' then (\\d)::int2" +  //$NON-NLS-1$
			"\\s+when 'r' then (\\d)::int2" +  //$NON-NLS-1$
			"\\s+else 3::int2" +  //$NON-NLS-1$
			"\\s+end as UPDATE_RULE," +  //$NON-NLS-1$
			"\\s+case ref.confdeltype" +  //$NON-NLS-1$
			"\\s+when 'c' then (\\d)::int2" +  //$NON-NLS-1$
			"\\s+when 'n' then (\\d)::int2" +  //$NON-NLS-1$
			"\\s+when 'd' then (\\d)::int2" +  //$NON-NLS-1$
			"\\s+when 'r' then (\\d)::int2" +  //$NON-NLS-1$
			"\\s+else 3::int2" +  //$NON-NLS-1$
			"\\s+end as DELETE_RULE," +							 //$NON-NLS-1$
			"\\s+ref.conname as FK_NAME," +  //$NON-NLS-1$
			"\\s+cn.conname as PK_NAME," +  //$NON-NLS-1$
			"\\s+case" +  //$NON-NLS-1$
			"\\s+when ref.condeferrable then" +  //$NON-NLS-1$
			"\\s+case" +  //$NON-NLS-1$
			"\\s+when ref.condeferred then (\\d)::int2" +  //$NON-NLS-1$
			"\\s+else (\\d)::int2" +  //$NON-NLS-1$
			"\\s+end" +  //$NON-NLS-1$
			"\\s+else (\\d)::int2" +  //$NON-NLS-1$
			"\\s+end as DEFERRABLITY" +  //$NON-NLS-1$
			"\\s+from" +  //$NON-NLS-1$
			"\\s+\\(\\(\\(\\(\\(\\(\\( \\(select cn.oid, conrelid, conkey, confrelid, confkey," +  //$NON-NLS-1$
			"\\s+generate_series\\(array_lower\\(conkey, 1\\), array_upper\\(conkey, 1\\)\\) as i," +  //$NON-NLS-1$
			"\\s+confupdtype, confdeltype, conname," +  //$NON-NLS-1$
			"\\s+condeferrable, condeferred" +  //$NON-NLS-1$
			"\\s+from pg_catalog.pg_constraint cn," +  //$NON-NLS-1$
			"\\s+pg_catalog.pg_class c," +  //$NON-NLS-1$
			"\\s+pg_catalog.pg_namespace n" +  //$NON-NLS-1$
			"\\s+where contype = 'f'" +  //$NON-NLS-1$
			"\\s+and\\s+con(f?)relid = c.oid" +  //$NON-NLS-1$
			"\\s+and\\s+relname = (E?(?:'[^']*')+)" +  //$NON-NLS-1$
			"\\s+and\\s+n.oid = c.relnamespace" +  //$NON-NLS-1$
			"\\s+and\\s+n.nspname = (E?(?:'[^']*')+)" +  //$NON-NLS-1$
			"\\s+\\) ref" +  //$NON-NLS-1$
			"\\s+inner join pg_catalog.pg_class c1" +  //$NON-NLS-1$
			"\\s+on c1.oid = ref.conrelid\\)" +  //$NON-NLS-1$
			"\\s+inner join pg_catalog.pg_namespace n1" +  //$NON-NLS-1$
			"\\s+on\\s+n1.oid = c1.relnamespace\\)" +  //$NON-NLS-1$
			"\\s+inner join pg_catalog.pg_attribute a1" +  //$NON-NLS-1$
			"\\s+on\\s+a1.attrelid = c1.oid" +  //$NON-NLS-1$
			"\\s+and\\s+a1.attnum = conkey\\[i\\]\\)" +  //$NON-NLS-1$
			"\\s+inner join pg_catalog.pg_class c2" +  //$NON-NLS-1$
			"\\s+on\\s+c2.oid = ref.confrelid\\)" +  //$NON-NLS-1$
			"\\s+inner join pg_catalog.pg_namespace n2" +  //$NON-NLS-1$
			"\\s+on\\s+n2.oid = c2.relnamespace\\)" +  //$NON-NLS-1$
			"\\s+inner join pg_catalog.pg_attribute a2" +  //$NON-NLS-1$
			"\\s+on\\s+a2.attrelid = c2.oid" +  //$NON-NLS-1$
			"\\s+and\\s+a2.attnum = confkey\\[i\\]\\)" +  //$NON-NLS-1$
			"\\s+left outer join pg_catalog.pg_constraint cn" +  //$NON-NLS-1$
			"\\s+on cn.conrelid = ref.confrelid" +  //$NON-NLS-1$
			"\\s+and cn.contype = 'p'\\)" +  //$NON-NLS-1$
			"\\s+order by ref.oid, ref.i", Pattern.DOTALL|Pattern.CASE_INSENSITIVE); //$NON-NLS-1$
		
	private static Pattern cursorSelectPattern = Pattern.compile("DECLARE\\s+(\\S+)(\\s+BINARY)?(?:\\s+INSENSITIVE)?(\\s+(NO\\s+)?SCROLL)?\\s+CURSOR\\s+(?:WITH(?:OUT)? HOLD\\s+)?FOR\\s+(.*)", Pattern.CASE_INSENSITIVE|Pattern.DOTALL); //$NON-NLS-1$
	private static Pattern fetchPattern = Pattern.compile("FETCH(?:(?:\\s+(FORWARD|ABSOLUTE|RELATIVE))?\\s+(\\d+)\\s+(?:IN|FROM))?\\s+(\\S+)\\s*", Pattern.DOTALL|Pattern.CASE_INSENSITIVE); //$NON-NLS-1$
	private static Pattern fetchFirstLastPattern = Pattern.compile("FETCH\\s+(FIRST|LAST)\\s+(?:IN|FROM)\\s+(\\S+)\\s*", Pattern.DOTALL|Pattern.CASE_INSENSITIVE); //$NON-NLS-1$
	private static Pattern movePattern = Pattern.compile("MOVE(?:\\s+(FORWARD|BACKWARD))?\\s+(\\d+)\\s+(?:IN|FROM)\\s+(\\S+)\\s*", Pattern.DOTALL|Pattern.CASE_INSENSITIVE); //$NON-NLS-1$
	private static Pattern closePattern = Pattern.compile("CLOSE (\\S+)", Pattern.DOTALL|Pattern.CASE_INSENSITIVE); //$NON-NLS-1$
	
	private static Pattern deallocatePattern = Pattern.compile("DEALLOCATE(?:\\s+PREPARE)?\\s+(.*)", Pattern.DOTALL|Pattern.CASE_INSENSITIVE); //$NON-NLS-1$
	private static Pattern releasePattern = Pattern.compile("RELEASE\\s+(\\w+\\d?_*)", Pattern.DOTALL|Pattern.CASE_INSENSITIVE); //$NON-NLS-1$
	private static Pattern savepointPattern = Pattern.compile("SAVEPOINT\\s+(\\w+\\d?_*)", Pattern.DOTALL|Pattern.CASE_INSENSITIVE); //$NON-NLS-1$
	private static Pattern rollbackPattern = Pattern.compile("ROLLBACK(\\s+to)?\\s+(\\w+\\d+_*)", Pattern.DOTALL|Pattern.CASE_INSENSITIVE); //$NON-NLS-1$
	
	private static Pattern txnPattern = Pattern.compile("(BEGIN(?:\\s+READ\\s+ONLY)?|COMMIT|ROLLBACK)(\\s+(WORK|TRANSACTION))?", Pattern.DOTALL|Pattern.CASE_INSENSITIVE); //$NON-NLS-1$
	
	private TeiidDriver driver;
	private ODBCClientRemote client;
	private Properties props;
	private ConnectionImpl connection;
	private volatile boolean executing;
	private boolean errorOccurred;
	
	private volatile ResultsFuture<Boolean> executionFuture;
	
	// TODO: this is unbounded map; need to define some boundaries as to how many stmts each session can have
	private Map<String, Prepared> preparedMap = Collections.synchronizedMap(new HashMap<String, Prepared>());
	private Map<String, Portal> portalMap = Collections.synchronizedMap(new HashMap<String, Portal>());
	private Map<String, Cursor> cursorMap = Collections.synchronizedMap(new HashMap<String, Cursor>());
	private	LogonImpl logon;
	
	public ODBCServerRemoteImpl(ODBCClientInstance client, TeiidDriver driver, LogonImpl logon) {
		this.driver = driver;
		this.client = client.getClient();
		this.logon = logon;
	}
	
	@Override
	public void initialize(Properties props) {
		this.props = props;		
		this.client.initialized(this.props);

		String user = props.getProperty("user"); //$NON-NLS-1$
		String database = props.getProperty("database"); //$NON-NLS-1$
		
		AuthenticationType authType = null;
		try {
			authType = getAuthenticationType(user, database);
		} catch (LogonException e) {
			errorOccurred(e);
			terminate();
			return;
		}
		
		if (authType.equals(AuthenticationType.USERPASSWORD)) {
			this.client.useClearTextAuthentication();
		}
		else if (authType.equals(AuthenticationType.GSS)) {
			this.client.useAuthenticationGSS();
		} else {
			throw new AssertionError("Unsupported Authentication Type"); //$NON-NLS-1$
		}
	}

	private AuthenticationType getAuthenticationType(String user,
			String database) throws LogonException {
		SessionService ss = this.logon.getSessionService();
		if (ss == null) {
			return AuthenticationType.USERPASSWORD;
		}
		return ss.getAuthenticationType(database, null, user);
	}

	@Override
	public void logon(String databaseName, String user, NullTerminatedStringDataInputStream data, SocketAddress remoteAddress) {
		try {
			java.util.Properties info = new java.util.Properties();
			info.put(TeiidURL.CONNECTION.USER_NAME, user); 
			
			AuthenticationType authType = getAuthenticationType(user, databaseName);
			
			String password = null; 
			if (authType.equals(AuthenticationType.USERPASSWORD)) {
				password = data.readString();
			}
			else if (authType.equals(AuthenticationType.GSS)) {
				byte[] serviceToken = data.readServiceToken();
            	GSSResult result = this.logon.neogitiateGssLogin(serviceToken, databaseName, null, user);
            	serviceToken = result.getServiceToken();
            	if (result.isAuthenticated()) {
                	info.put(ILogon.KRB5TOKEN, serviceToken);
                	if (!result.isNullContinuationToken()) {
                	    this.client.authenticationGSSContinue(serviceToken);
                	}
                	// if delegation is in progress, participate in it.
                	if (result.getDelegationCredential() != null) {
                		info.put(GSSCredential.class.getName(), result.getDelegationCredential());
                	}
            	}
            	else {
	            	this.client.authenticationGSSContinue(serviceToken);
	            	return;            		
            	}
			} else {
				throw new AssertionError("Unsupported Authentication Type"); //$NON-NLS-1$
			}
			
			// this is local connection
			String url = "jdbc:teiid:"+databaseName; //$NON-NLS-1$

			if (password != null) {
				info.put(TeiidURL.CONNECTION.PASSWORD, password); 
			}
			String applicationName = this.props.getProperty(PgBackendProtocol.APPLICATION_NAME);
			if (applicationName == null) {
			    applicationName = PgBackendProtocol.DEFAULT_APPLICATION_NAME;
			    this.props.put(PgBackendProtocol.APPLICATION_NAME, applicationName);
			}
			info.put(TeiidURL.CONNECTION.APP_NAME, applicationName);
			
			if (remoteAddress instanceof InetSocketAddress) {
				SocketServerConnection.updateConnectionProperties(info, ((InetSocketAddress)remoteAddress).getAddress(), false);
			}
			
			this.connection =  driver.connect(url, info);
			//Propagate so that we can use in pg methods
			SessionMetadata sm = ((LocalServerConnection)this.connection.getServerConnection()).getWorkContext().getSession();
			sm.addAttchment(ODBCServerRemoteImpl.class, this);
			setConnectionProperties(this.connection);
			int hash = this.connection.getConnectionId().hashCode();
			Enumeration<?> keys = this.props.propertyNames();
			while (keys.hasMoreElements()) {
				String key = (String)keys.nextElement();
				this.connection.setExecutionProperty(key, this.props.getProperty(key));
			}
			StatementImpl s = this.connection.createStatement();
			try {
				s.execute("select teiid_session_set('resolve_groupby_positional', true)"); //$NON-NLS-1$
			} finally {
				s.close();
			}
			this.client.authenticationSucess(hash, hash);
			ready();
		} catch (SQLException e) {
			errorOccurred(e);
			terminate();
		} catch(LogonException e) {
			errorOccurred(e);
			terminate();
		} catch (IOException e) {
			errorOccurred(e);
			terminate();			
		}
	}

	public static void setConnectionProperties(ConnectionImpl conn)
			throws SQLException {
		SessionMetadata sm = ((LocalServerConnection)conn.getServerConnection()).getWorkContext().getSession();
		VDB vdb = sm.getVdb();
		Properties p = vdb.getProperties();
		setConnectionProperties(conn, p);
	}

	public static void setConnectionProperties(ConnectionImpl conn,
			Properties p) {
		for (Map.Entry<Object, Object> entry : p.entrySet()) {
			String key = (String)entry.getKey();
			
			if (key.startsWith(CONNECTION_PROPERTY_PREFIX)) {
				conn.setExecutionProperty(key.substring(CONNECTION_PROPERTY_PREFIX.length()), (String) entry.getValue());
			}
		}
	}	
	
	private void cursorExecute(String cursorName, final String sql, final ResultsFuture<Integer> completion, boolean scroll, final boolean binary) {
		try {
			// close if the name is already used or the unnamed prepare; otherwise
			// stmt is alive until session ends.
			this.preparedMap.remove(UNNAMED);
			Portal p = this.portalMap.remove(UNNAMED);
			if (p != null) {
				closePortal(p);
			}
			if (cursorName == null || cursorName.length() == 0) {
				cursorName = UNNAMED;
			}
			Cursor cursor = cursorMap.get(cursorName);
			if (cursor != null) {
				errorOccurred(RuntimePlugin.Util.gs(RuntimePlugin.Event.TEIID40111, cursorName));
				return;
			}
			
			final PreparedStatementImpl stmt = this.connection.prepareStatement(sql, scroll?ResultSet.TYPE_SCROLL_INSENSITIVE:ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            this.executionFuture = stmt.submitExecute(ResultsMode.RESULTSET, null);
            final String name = cursorName;
            this.executionFuture.addCompletionListener(new ResultsFuture.CompletionListener<Boolean>() {
        		@Override
        		public void onCompletion(ResultsFuture<Boolean> future) {
        			executionFuture = null;
                    try {
                        if (future.get()) {
		                	List<PgColInfo> cols = getPgColInfo(stmt.getResultSet().getMetaData());
	                        cursorMap.put(name, new Cursor(name, sql, stmt, stmt.getResultSet(), cols, binary));
    						client.sendCommandComplete("DECLARE CURSOR", null); //$NON-NLS-1$		                            
							completion.getResultsReceiver().receiveResults(0);
						}
					} catch (Throwable e) {
						completion.getResultsReceiver().exceptionOccurred(e);
					}
        		}
			});					
		} catch (SQLException e) {
			completion.getResultsReceiver().exceptionOccurred(e);
		} 
	}
	
	private void cursorFetch(String cursorName, CursorDirection direction, int rows, final ResultsFuture<Integer> completion) throws SQLException {
		Cursor cursor = this.cursorMap.get(cursorName);
		if (cursor == null) {
			throw new SQLException(RuntimePlugin.Util.gs(RuntimePlugin.Event.TEIID40078, cursorName));
		}
		if (rows < 1) {
			throw new SQLException(RuntimePlugin.Util.gs(RuntimePlugin.Event.TEIID40112, cursorName, rows));
		}
		this.client.sendResults("FETCH", cursor.rs, cursor.prepared.columnMetadata, completion, direction, rows, true, cursor.resultColumnFormat); //$NON-NLS-1$
	}
	
	private void cursorMove(String prepareName, String direction, final int rows, final ResultsFuture<Integer> completion) throws SQLException {
		
		// win odbc driver sending a move after close; and error is ending up in failure; since the below
		// is not harmful it is ok to send empty move.
		if (rows == 0) {
			client.sendCommandComplete("MOVE", 0); //$NON-NLS-1$
			completion.getResultsReceiver().receiveResults(0);
			return;			
		}
		
		final Cursor cursor = this.cursorMap.get(prepareName);
		if (cursor == null) {
			throw new SQLException(RuntimePlugin.Util.gs(RuntimePlugin.Event.TEIID40078, prepareName));
		}
		
		final boolean forward = direction == null || direction.equalsIgnoreCase("forward"); //$NON-NLS-1$
		Runnable r = new Runnable() {
			public void run() {
				run(null, 0);
			}
			public void run(ResultsFuture<Boolean> next, int i) {
				for (; i < rows; i++) {
					try {
						if (next == null) {
							if (forward) {
								next = cursor.rs.submitNext();
							} else {
								//TODO: we know that we are scrollable in this case, we should just
								//use an absolute positioning
								//as of now previous is non-blocking
								next = StatementImpl.booleanFuture(cursor.rs.previous());
							}
						}
						if (!next.isDone()) {
							final int current = i;
							next.addCompletionListener(new ResultsFuture.CompletionListener<Boolean>() {
								@Override
								public void onCompletion(
										ResultsFuture<Boolean> future) {
									run(future, current); //restart later
								}
							});
							return;
						}
						if (!next.get()) {
							break; //no next row
						}
						next = null;
					} catch (Throwable e) {
						completion.getResultsReceiver().exceptionOccurred(e);
						return;
					}
				} 
				if (!completion.isDone()) {
					client.sendCommandComplete("MOVE", i); //$NON-NLS-1$						
					completion.getResultsReceiver().receiveResults(i);
				}
			}
		};
		r.run();
	}	
	
	private void cursorClose(String prepareName) throws SQLException {
		Cursor cursor = this.cursorMap.remove(prepareName);
		if (cursor != null) {
			closePortal(cursor);
			this.client.sendCommandComplete("CLOSE CURSOR", null); //$NON-NLS-1$
		}
	}	
	
    private void sqlExecute(final String sql, final ResultsFuture<Integer> completion) throws SQLException {
    	String modfiedSQL = fixSQL(sql); 
    	final StatementImpl stmt = connection.createStatement();
        executionFuture = stmt.submitExecute(modfiedSQL, null);
        completion.addCompletionListener(new ResultsFuture.CompletionListener<Integer>() {
        	public void onCompletion(ResultsFuture<Integer> future) {
        		try {
					stmt.close();
				} catch (SQLException e) {
					LogManager.logDetail(LogConstants.CTX_ODBC, e, "Error closing statement"); //$NON-NLS-1$
				}
        	}
        });
        executionFuture.addCompletionListener(new ResultsFuture.CompletionListener<Boolean>() {
    		@Override
    		public void onCompletion(ResultsFuture<Boolean> future) {
    			executionFuture = null;
    			try {
	                if (future.get()) {
                		List<PgColInfo> cols = getPgColInfo(stmt.getResultSet().getMetaData());
                		String tag = PgBackendProtocol.getCompletionTag(sql, null);
                        client.sendResults(sql, stmt.getResultSet(), cols, completion, CursorDirection.FORWARD, -1, tag.equals("SELECT") || tag.equals("SHOW"), null); //$NON-NLS-1$ //$NON-NLS-2$
	                } else {
	                	client.sendUpdateCount(sql, stmt.getUpdateCount());
	                	updateSessionProperties();
	                	completion.getResultsReceiver().receiveResults(1);
	                }
    			} catch (Throwable e) {
    				if (!completion.isDone()) {
    					completion.getResultsReceiver().exceptionOccurred(e);
    				}
    			}
    		}
		});    	
    }	
	
	@Override
	public void prepare(String prepareName, String sql, int[] paramType) {
		if (prepareName == null || prepareName.length() == 0) {
			prepareName  = UNNAMED;
		}
					
		if (sql != null) {
			PreparedStatementImpl stmt = null;
			try {
				// close if the name is already used or the unnamed prepare; otherwise
				// stmt is alive until session ends.
				if (prepareName.equals(UNNAMED)) {
					this.preparedMap.remove(prepareName);
				} else {
					Prepared previous = this.preparedMap.get(prepareName);
					if (previous != null) {
						errorOccurred(RuntimePlugin.Util.gs(RuntimePlugin.Event.TEIID40110, prepareName));
						return;
					}
				}
				//just pull the initial information - leave statement formation until binding 
				String modfiedSQL = fixSQL(sql);
				stmt = this.connection.prepareStatement(modfiedSQL);
				Prepared prepared = new Prepared(prepareName, sql, modfiedSQL, paramType, getPgColInfo(stmt.getMetaData()));
				this.preparedMap.put(prepareName, prepared);
				this.client.prepareCompleted(prepareName);
			} catch (SQLException e) {
				if (e.getCause() instanceof TeiidProcessingException) {
					LogManager.logWarning(LogConstants.CTX_ODBC, e.getCause(), RuntimePlugin.Util.gs(RuntimePlugin.Event.TEIID40020)); 
				}
				errorOccurred(e);
			} finally {
				try {
					if (stmt != null) {
						stmt.close();
					}
				} catch (SQLException e) {
				}
			}
		}
	}	
	
	private long readLong(byte[] bytes, int length) {
		long val = 0;
		for (int k = 0; k < length; k++) {
			val += ((bytes[k] & 255) << ((length - k - 1)*8));
		}
		return val;
	}
	
	@Override
	public void bindParameters(String bindName, String prepareName, Object[] params, int resultCodeCount, short[] resultColumnFormat, Charset encoding) {
		// An unnamed portal is destroyed at the end of the transaction, or as soon as 
		// the next Bind statement specifying the unnamed portal as destination is issued. 
		if (bindName == null || bindName.length() == 0) {
			Portal p = this.portalMap.remove(UNNAMED);
			if (p != null) {
				closePortal(p);
			}
			bindName  = UNNAMED;
		} else if (this.portalMap.get(bindName) != null || this.cursorMap.get(bindName) != null) {
			errorOccurred(RuntimePlugin.Util.gs(RuntimePlugin.Event.TEIID40111, bindName));
			return;
		}
		
		if (prepareName == null || prepareName.length() == 0) {
			prepareName  = UNNAMED;
		}
		
		Prepared prepared = this.preparedMap.get(prepareName);
		if (prepared == null) {
			errorOccurred(RuntimePlugin.Util.gs(RuntimePlugin.Event.TEIID40077, prepareName));
			return;
		}		
		PreparedStatementImpl stmt = null; 
		try {
			stmt = this.connection.prepareStatement(prepared.modifiedSql);
			for (int i = 0; i < params.length; i++) {
				Object param = params[i];
				if (param instanceof byte[] && prepared.paramType.length > i) {
					int oid = prepared.paramType[i];
					switch (oid) {
					case PGUtil.PG_TYPE_UNSPECIFIED:
						//TODO: should infer type from the parameter metadata from the parse message
						break;
					case PGUtil.PG_TYPE_BYTEA:
						break;
					case PGUtil.PG_TYPE_INT2:
						param = (short)readLong((byte[])param, 2);
						break;
					case PGUtil.PG_TYPE_INT4:
						param = (int)readLong((byte[])param, 4);
						break;
					case PGUtil.PG_TYPE_INT8:
						param = readLong((byte[])param, 8);
						break;
					case PGUtil.PG_TYPE_FLOAT4:
						param = Float.intBitsToFloat((int)readLong((byte[])param, 4));
						break;
					case PGUtil.PG_TYPE_FLOAT8:
						param = Double.longBitsToDouble(readLong((byte[])param, 8));
						break;
					case PGUtil.PG_TYPE_DATE:
					    param = TimestampUtils.toDate(TimestampWithTimezone.getCalendar().getTimeZone(), (int)readLong((byte[])param, 4));
						break;
					default:
					    //start with the string conversion
					    param = new String((byte[])param, encoding);
					    break;
					}
				}
				stmt.setObject(i+1, param);
			}
			this.portalMap.put(bindName, new Portal(bindName, prepared, resultColumnFormat, stmt));
			this.client.bindComplete();
			stmt = null;
		} catch (SQLException e) {
			errorOccurred(e);
		} finally {
			if (stmt != null) {
				try {
					stmt.close();
				} catch (SQLException e) {
				}
			}
		}
	}

	@Override
	public void unsupportedOperation(String msg) {
		errorOccurred(msg);
	}

	@Override
	public void execute(String bindName, int maxRows) {
		if (beginExecution()) {
			errorOccurred("Awaiting asynch result"); //$NON-NLS-1$
			return;
		}
		if (bindName == null || bindName.length() == 0) {
			bindName  = UNNAMED;
		}		
		if (maxRows == 0) {
			maxRows = -1;
		}
		// special case cursor execution through portal
		final Cursor cursor = this.cursorMap.get(bindName);
		if (cursor != null) {
			sendCursorResults(cursor, maxRows);
			return;
		}		
		
		final Portal query = this.portalMap.get(bindName);
		if (query == null) {
			errorOccurred(RuntimePlugin.Util.gs(RuntimePlugin.Event.TEIID40078, bindName));
			return;
		}	
		
		if (query.prepared.sql.trim().isEmpty()) {
			this.client.emptyQueryReceived();
			return;
		}
		
        sendPortalResults(maxRows, query);			
	}

	private void sendPortalResults(final int maxRows, final Portal query) {
		if (query.rs != null) {
			//this is a suspended portal
			sendCursorResults(query, maxRows);
			return;
		}
		final PreparedStatementImpl stmt = query.stmt;
        try {
            this.executionFuture = stmt.submitExecute(ResultsMode.EITHER, null);
            executionFuture.addCompletionListener(new ResultsFuture.CompletionListener<Boolean>() {
        		@Override
        		public void onCompletion(ResultsFuture<Boolean> future) {
        			executionFuture = null;
                    try {
		                if (future.get()) {
		                	query.rs = stmt.getResultSet();
		                	sendCursorResults(query, maxRows);
		                } else {
		                	client.sendUpdateCount(query.prepared.sql, stmt.getUpdateCount());
		                	updateSessionProperties();
		                	doneExecuting();
		                }
                    } catch (ExecutionException e) {
                    	if (e.getCause() != null) {
                    		errorOccurred(e.getCause());
                    	} else {
                            errorOccurred(e);
                    	}
                    } catch (Throwable e) {
                        errorOccurred(e);
                    }
        		}
			});
        } catch (SQLException e) {
        	errorOccurred(e);
        }
	}

	private void sendCursorResults(final Portal cursor, final int fetchSize) {
		ResultsFuture<Integer> result = new ResultsFuture<Integer>();
		this.client.sendResults(null, cursor.rs, cursor.prepared.columnMetadata, result, CursorDirection.FORWARD, fetchSize, false, cursor.resultColumnFormat);
		result.addCompletionListener(new ResultsFuture.CompletionListener<Integer>() {
			public void onCompletion(ResultsFuture<Integer> future) {
				try {
					int rowsSent = future.get();
					if (rowsSent < fetchSize || fetchSize <= 0) {
						client.sendCommandComplete(cursor.prepared.sql, rowsSent);
					}
					else {
						client.sendPortalSuspended();
					}
					doneExecuting();
				} catch (InterruptedException e) {
					throw new AssertionError(e);
				} catch (ExecutionException e) {
					errorOccurred(e.getCause());
				}
			};
		});
	}
	
	private String fixSQL(String sql) {
		String modified = modifySQL(sql);
		if (modified != null && !modified.equals(sql)) {
			LogManager.logDetail(LogConstants.CTX_ODBC, "Modified Query:", modified); //$NON-NLS-1$
		}			
		return modified;
	}
	
	private String modifySQL(String sql) {
		String modified = sql;
		if (sql == null) {
			return null;
		}
		Matcher m = null;
		// selects are coming with "select\t" so using a space after "select" does not always work
		if (StringUtil.startsWithIgnoreCase(sql, "select")) { //$NON-NLS-1$
			if ((m = pkPattern.matcher(modified)).matches()) {
				return new StringBuffer("SELECT k.Name AS attname, convert(Position, short) AS attnum, TableName AS relname, SchemaName AS nspname, TableName AS relname") //$NON-NLS-1$
			          .append(" FROM SYS.KeyColumns k") //$NON-NLS-1$ 
			          .append(" WHERE ") //$NON-NLS-1$
			          .append(" UCASE(SchemaName)").append(" LIKE UCASE(").append(m.group(2)).append(")")//$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$
			          .append(" AND UCASE(TableName)") .append(" LIKE UCASE(").append(m.group(1)).append(")")//$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$
			          .append(" AND KeyType LIKE 'Primary'") //$NON-NLS-1$
			          .append(" ORDER BY attnum").toString(); //$NON-NLS-1$					
			}
			else if ((m = pkKeyPattern.matcher(modified)).matches()) {
				String tableName = m.group(1);
				if (tableName.endsWith("_pkey'")) { //$NON-NLS-1$
					tableName = tableName.substring(0, tableName.length()-6) + '\'';
					return "select ia.attname, ia.attnum, ic.relname, n.nspname, NULL "+ //$NON-NLS-1$	
						"from pg_catalog.pg_attribute ia, pg_catalog.pg_class ic, pg_catalog.pg_namespace n, Sys.KeyColumns kc "+ //$NON-NLS-1$	
						"where ic.relname = "+tableName+" AND n.nspname = "+m.group(2)+" AND "+ //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$
						"n.oid = ic.relnamespace AND ia.attrelid = ic.oid AND kc.SchemaName = n.nspname " +//$NON-NLS-1$	
						"AND kc.TableName = ic.relname AND kc.KeyType = 'Primary' AND kc.Name = ia.attname order by ia.attnum";//$NON-NLS-1$	
				}
				return "SELECT NULL, NULL, NULL, NULL, NULL FROM (SELECT 1) as X WHERE 0=1"; //$NON-NLS-1$
			}
			else if ((m = fkPattern.matcher(modified)).matches()){
				String baseQuery = "SELECT PKTABLE_CAT, PKTABLE_SCHEM, PKTABLE_NAME, PKCOLUMN_NAME, FKTABLE_CAT, FKTABLE_SCHEM, "+//$NON-NLS-1$
						"FKTABLE_NAME, FKCOLUMN_NAME, KEY_SEQ, UPDATE_RULE, DELETE_RULE, FK_NAME, PK_NAME, DEFERRABILITY "+//$NON-NLS-1$
						"FROM SYS.ReferenceKeyColumns WHERE  "; //$NON-NLS-1$
				if ("f".equals(m.group(14))) { //$NON-NLS-1$
					//exported keys
					return baseQuery + "PKTABLE_NAME = " + m.group(15)+" and PKTABLE_SCHEM = "+m.group(16);//$NON-NLS-1$ //$NON-NLS-2$
				}
				//imported keys
				return baseQuery + "FKTABLE_NAME = " + m.group(15)+" and FKTABLE_SCHEM = "+m.group(16);//$NON-NLS-1$ //$NON-NLS-2$
			}
			else if (modified.equalsIgnoreCase("select version()")) { //$NON-NLS-1$
				return "SELECT 'Teiid "+ApplicationInfo.getInstance().getReleaseNumber()+"'"; //$NON-NLS-1$ //$NON-NLS-2$
			}
			else if (modified.startsWith("SELECT name FROM master..sysdatabases")) { //$NON-NLS-1$
				return "SELECT 'Teiid'"; //$NON-NLS-1$
			}
			else if (modified.equalsIgnoreCase("select db_name() dbname")) { //$NON-NLS-1$
				return "SELECT current_database()"; //$NON-NLS-1$
			}
			else if (sql.equalsIgnoreCase("select current_schema()")) { //$NON-NLS-1$
				// since teiid can work with multiple schemas at a given time
				// this call resolution is ambiguous
				return "SELECT ''";  //$NON-NLS-1$
			}
			else if (sql.equals("SELECT typinput='array_in'::regproc, typtype FROM pg_catalog.pg_type WHERE typname = $1")) { //$NON-NLS-1$
				return "SELECT substring(typname,1,1) = '_', typtype FROM pg_catalog.pg_type WHERE typname = ?"; //$NON-NLS-1$
			}
			if ((m = columnMetadataPattern.matcher(modified)).matches()) {
				return "select t1.schemaname as nspname, c.relname, t1.name as attname, t.oid as attypid, t.typname, convert(t1.Position, short) as attnum, t.typlen as attlen," //$NON-NLS-1$
				        + PgCatalogMetadataStore.TYPMOD + " as atttypmod, "  //$NON-NLS-1$
						+ "CASE WHEN (t1.NullType = 'No Nulls') THEN true ELSE false END as attnotnull, c.relhasrules, c.relkind, c.oid, pg_get_expr(case when t1.IsAutoIncremented then 'nextval(' else t1.DefaultValue end, c.oid), " //$NON-NLS-1$
						+ " case t.typtype when 'd' then t.typbasetype else 0 end, t.typtypmod, c.relhasoids from sys.columns as t1, pg_catalog.matpg_datatype as t, pg_catalog.pg_class c where c.relnspname=t1.schemaname and c.relname=t1.tablename and t1.DataType = t.Name and c.oid = " //$NON-NLS-1$
						+ m.group(1)
						+ " order by nspname, relname, attnum"; //$NON-NLS-1$
			}
		}
		else if (sql.equalsIgnoreCase("show max_identifier_length")){ //$NON-NLS-1$
			return "select 63"; //$NON-NLS-1$
		}
		else if ((m = setPattern.matcher(sql)).matches()) {
			return "SET " + m.group(1) + " " + m.group(2); //$NON-NLS-1$ //$NON-NLS-2$
		}
		else if ((m = txnPattern.matcher(sql)).matches()) {
			if (StringUtil.startsWithIgnoreCase(m.group(1), "BEGIN")) { //$NON-NLS-1$
				return "START TRANSACTION"; //$NON-NLS-1$
			}
			return m.group(1);
		}
		else if ((m = rollbackPattern.matcher(modified)).matches()) {
			return "set \"dummy-update-pg-odbc\" 0"; //$NON-NLS-1$
		}					
		else if ((m = savepointPattern.matcher(sql)).matches()) {
			return "set \"dummy-update-pg-odbc\" 0"; //$NON-NLS-1$
		}
		else if ((m = releasePattern.matcher(sql)).matches()) {
			return "set \"dummy-update-pg-odbc\" 0"; //$NON-NLS-1$
		} 
		for (int i = 0; i < modified.length(); i++) {
			switch (modified.charAt(i)) {
			case ':':
			case '~':
				ScriptReader reader = new ScriptReader(modified);
				reader.setRewrite(true);
				try {
					return reader.readStatement();
				} catch (IOException e) {
					//can't happen
				}				
			}
		}
		return modified;
	}

	@Override
	public void executeQuery(String query) {
		if (beginExecution()) {
			errorOccurred("Awaiting asynch result"); //$NON-NLS-1$
			ready();
			return;
		}
		//46.2.3 Note that a simple Query message also destroys the unnamed portal.
		Portal p = this.portalMap.remove(UNNAMED);
		if (p != null) {
			closePortal(p);
		}
		this.preparedMap.remove(UNNAMED);
		query = query.trim();
		if (query.length() == 0) {
    		client.emptyQueryReceived();
    		ready();
    	}
		
        QueryWorkItem r = new QueryWorkItem(query);
		r.run();
	}

	private boolean beginExecution() {
		if (this.executionFuture != null) {
			return true;
		}
		this.executing = true;			
		return false;
	}
	
	public boolean isExecuting() {
		return executing;
	}
	
	public boolean isErrorOccurred() {
		return errorOccurred;
	}

	@Override
	public void getParameterDescription(String prepareName) {
		if (prepareName == null || prepareName.length() == 0) {
			prepareName  = UNNAMED;
		}		
		Prepared query = this.preparedMap.get(prepareName);
		if (query == null) {
			errorOccurred(RuntimePlugin.Util.gs(RuntimePlugin.Event.TEIID40079, prepareName));
			return;
		}
		// The response is a ParameterDescription message describing the parameters needed by the statement,
		this.client.sendParameterDescription(query.paramType);
		
		// followed by a RowDescription message describing the rows that will be returned when the statement  
		// is eventually executed (or a NoData message if the statement will not return rows).
		this.client.sendResultSetDescription(query.columnMetadata, null);
	}
	
	private void errorOccurred(String error) {
		this.client.errorOccurred(error);
		synchronized (this) {
			this.errorOccurred = true;
			doneExecuting();
		}
	}
	
	public void errorOccurred(Throwable error) {
		this.client.errorOccurred(error);
		synchronized (this) {
			this.errorOccurred = true;
			doneExecuting();
		}
	}

	@Override
	public void getResultSetMetaDataDescription(String bindName) {
		if (bindName == null || bindName.length() == 0) {
			bindName  = UNNAMED;
		}		
		Portal query = this.portalMap.get(bindName);
		if (query == null) {
			errorOccurred(RuntimePlugin.Util.gs(RuntimePlugin.Event.TEIID40078, bindName));
		}
		else {
			this.client.sendResultSetDescription(query.prepared.columnMetadata, query.resultColumnFormat);
		}
	}

	@Override
	public void sync() {
		ready();
	}
	
	protected void doneExecuting() {
		executing = false;
	}
	
	private void ready() {
		boolean inTxn = false;
		boolean failedTxn = false;
		try {
			if (!this.connection.getAutoCommit()) {
				inTxn = true;
			}
		} catch (SQLException e) {
			failedTxn = true;
		}
		synchronized (this) {
			this.errorOccurred = false;
		}
		this.client.ready(inTxn, failedTxn);
	}
	
	@Override
	public void cancel() {
		// TODO Auto-generated method stub
	}

	@Override
	public void closeBoundStatement(String bindName) {
		if (bindName == null || bindName.length() == 0) {
			bindName  = UNNAMED;
		}		
		Portal query = this.portalMap.remove(bindName);
		if (query != null) {
			closePortal(query);
		}
		this.client.statementClosed();
	}

	private void closePortal(Portal query) {
		ResultSet rs = query.rs;
		if (rs != null) {
			try {
				rs.close();
			} catch (SQLException e) {
				LogManager.logDetail(LogConstants.CTX_ODBC, e, "Did not successfully close portal", query.name); //$NON-NLS-1$
			}
			query.rs = null;
		}
		try {
			query.stmt.close();
		} catch (SQLException e) {
			LogManager.logDetail(LogConstants.CTX_ODBC, e, "Did not successfully close portal", query.name); //$NON-NLS-1$
		}
	}

	@Override
	public void closePreparedStatement(String preparedName) {
		if (preparedName == null || preparedName.length() == 0) {
			preparedName  = UNNAMED;
		}		
		Prepared query = this.preparedMap.remove(preparedName);
		if (query != null) {
			synchronized (this.portalMap) {
				for (Iterator<Portal> iter = this.portalMap.values().iterator(); iter.hasNext();) {
					Portal p = iter.next();
					if (p.prepared.name.equals(preparedName)) {
						iter.remove();
					}
					closePortal(p);
				}
			}
		}	
		this.client.statementClosed();
	}

	@Override
	public void terminate() {
		for (Portal p: this.portalMap.values()) {
			closePortal(p);
		}
		this.portalMap.clear();
		this.preparedMap.clear();
		try {			
			if (this.connection != null) {
				if (!this.connection.getAutoCommit()) {
					this.connection.rollback(false);
				}
				this.connection.close();
			}
		} catch (SQLException e) {
			//ignore
		}
		this.client.terminated();
	}	
	
	@Override
	public void flush() {
		this.client.flush();
	}
	
	@Override
	public void functionCall(int oid) {
		errorOccurred(RuntimePlugin.Util.gs(RuntimePlugin.Event.TEIID40081));
	}
	
	@Override
	public void sslRequest() {
		this.client.sendSslResponse();
	}
	
	private void updateSessionProperties() {
		String encoding = getEncoding();
		if (encoding != null) {
			//this may be unnecessary
			this.client.setEncoding(encoding, false);
		}
		String appName = this.connection.getExecutionProperty(PgBackendProtocol.APPLICATION_NAME);
		if (appName != null) {
    		String existing = props.getProperty(PgBackendProtocol.APPLICATION_NAME);
    		if (!EquivalenceUtil.areEqual(appName, existing)) {
                try {
                    SessionMetadata sm = ((LocalServerConnection)connection.getServerConnection()).getWorkContext().getSession();
                    sm.setApplicationName(appName);
                } catch (SQLException e) {
                    //connection invalid
                }
                this.client.sendParameterStatus(PgBackendProtocol.APPLICATION_NAME, appName);
                this.props.put(PgBackendProtocol.APPLICATION_NAME, appName);
    		}
		}
	}
	
	public String getEncoding() {
		return this.connection.getExecutionProperty(PgBackendProtocol.CLIENT_ENCODING);
	}
	
	static String normalizeName(String name) {
		if (name.length() > 1 && name.startsWith("\"") && name.endsWith("\"")) {
			return StringUtil.replaceAll(name.substring(1, name.length() - 1), "\"", "\"\"");
		}
		//--we are not consistently dealing with identifier naming/casing
		//return name.toUpperCase();
		return name;
	}
	
    private final class QueryWorkItem implements Runnable {
		private final ScriptReader reader;
		String sql;
		private String next;

		private QueryWorkItem(String query) {
			this.reader = new ScriptReader(query);		
		}
		
		private void done(Throwable error) {
			if (error != null) {
				errorOccurred(error);
			} else {
				doneExecuting();
			}
			ready();
		}

		@Override
		public void run() {
			try {
				if (sql == null) {
					sql = reader.readStatement();
				}
		        while (sql != null) {
		            try {
		    			
		            	ResultsFuture<Integer> results = new ResultsFuture<Integer>();
		    			results.addCompletionListener(new ResultsFuture.CompletionListener<Integer>() {
		                	public void onCompletion(ResultsFuture<Integer> future) {
		                		try {		                			
		    						future.get();
		    						if (next != null) {
		    							sql = next;
		    							next = null;
		    						} else {
		    							sql = reader.readStatement();
		    						}
		    					} catch (InterruptedException e) {
		    						throw new AssertionError(e);
		    					} catch (IOException e) {
		    						done(e);
		    						return;
		    					} catch (ExecutionException e) {
		    						Throwable cause = e;
		    						while (cause instanceof ExecutionException && cause.getCause() != null && cause != cause.getCause()) {
		    							cause = cause.getCause();
		    						}
		    						done(cause);
		    						return;
		    					}
		            			QueryWorkItem.this.run(); //continue processing
		                	};
		    			});	
		    			
		    			if (isErrorOccurred()) {
		    				if (!connection.getAutoCommit()) {
		    					connection.rollback(false);
		    				}
		    				break;
		    			}
		    			if (!HONOR_DECLARE_FETCH_TXN && sql.equalsIgnoreCase("BEGIN") && connection.getAutoCommit()) { //$NON-NLS-1$
    						next = reader.readStatement();
    						if (next != null && (cursorSelectPattern.matcher(next)).matches()) { 
    							sql = next;
    							next = null;
    							LogManager.logDetail(LogConstants.CTX_ODBC, "not honoring the transaction for declare/fetch"); //$NON-NLS-1$
    						}
			    		}
		            	Matcher m = null;
		    	        if ((m = cursorSelectPattern.matcher(sql)).matches()){
		    	        	boolean scroll = false;
		    	        	if (m.group(3) != null && m.group(4) == null ) {
	    	        			scroll = true;
		    	        	}
		    				cursorExecute(normalizeName(m.group(1)), fixSQL(m.group(5)), results, scroll, m.group(2) != null);
		    			}
		    			else if ((m = fetchPattern.matcher(sql)).matches()){
		    				int rowCount = 1;
		    				String direction = m.group(1);
		    				CursorDirection cursorDirection = CursorDirection.FORWARD;
		    				if (direction != null) {
		    					cursorDirection = CursorDirection.valueOf(direction.toUpperCase());
		    				}
	    					String rows = m.group(2);
		    				if (rows != null) {
		    					rowCount = Integer.parseInt(rows);
		    				}
	    					cursorFetch(normalizeName(m.group(3)), cursorDirection, rowCount, results);
		    			} 
		    			else if ((m = fetchFirstLastPattern.matcher(sql)).matches()){
		    				int rowCount = 1;
		    				String direction = m.group(1);
		    				CursorDirection cursorDirection = CursorDirection.valueOf(direction.toUpperCase());
	    					cursorFetch(normalizeName(m.group(2)), cursorDirection, rowCount, results);
		    			}
		    			else if ((m = movePattern.matcher(sql)).matches()){
		    				cursorMove(normalizeName(m.group(3)), m.group(1), Integer.parseInt(m.group(2)), results);
		    			}
		    			else if ((m = closePattern.matcher(sql)).matches()){
		    				cursorClose(normalizeName(m.group(1)));
		    				results.getResultsReceiver().receiveResults(1);
		    			}
		    			else if ((m = deallocatePattern.matcher(sql)).matches()) { 
		    				String plan_name = m.group(1);
		    				plan_name = normalizeName(plan_name);
		    				closePreparedStatement(plan_name);
		    				client.sendCommandComplete("DEALLOCATE", null); //$NON-NLS-1$
		    				results.getResultsReceiver().receiveResults(1);
		    			}
		    			else {
		    				sqlExecute(sql, results);
		    			}
		                return; //wait for the execution to finish
		            } catch (SQLException e) {
		            	done(e);
		            	return;
		            } 
		        }
			} catch (NumberFormatException e) {
				//create a sqlexception so that the logic doesn't over log this
				done(TeiidSQLException.create(e, RuntimePlugin.Util.gs(RuntimePlugin.Event.TEIID40147, e.getMessage())));
				return;
			} catch(Exception e) {
				done(e);
				return;
			}
			done(null);
		}
	}

	/**
	 * @see PgCatalogMetadataStore add_pg_attribute for mod calculation
	 */
	private List<PgColInfo> getPgColInfo(ResultSetMetaData meta)
			throws SQLException {
		if (meta == null) {
			return null;
		}
		int columns = meta.getColumnCount();
		final ArrayList<PgColInfo> result = new ArrayList<PgColInfo>(columns);
		for (int i = 1; i <= columns; i++) {
			final PgColInfo info = new PgColInfo();
			info.name = meta.getColumnLabel(i);
			info.type = meta.getColumnType(i);
			info.type = convertType(info.type);
			info.precision = meta.getColumnDisplaySize(i);
			if (info.type == PG_TYPE_NUMERIC) {
				info.mod = 4+ 65536*Math.min(32767, meta.getPrecision(i))+Math.min(32767, meta.getScale(i));
			} else if (info.type == PG_TYPE_BPCHAR || info.type == PG_TYPE_VARCHAR){
				info.mod = (int) Math.min(Integer.MAX_VALUE, 4+(long)meta.getColumnDisplaySize(i));
			} else {
				info.mod = -1;
			}
			String name = meta.getColumnName(i);
			String table = meta.getTableName(i);
			String schema = meta.getSchemaName(i);
			if (schema != null) {
				final PreparedStatementImpl ps = this.connection.prepareStatement("select " //$NON-NLS-1$
						+ "pg_catalog.getOid(SYS.Columns.TableUID), " //$NON-NLS-1$
						+ "cast(SYS.Columns.Position as short), " //$NON-NLS-1$
						+ "cast((select p.value from SYS.Properties p where p.name = 'pg_type:oid' and p.uid = SYS.Columns.uid) as integer) " //$NON-NLS-1$ 
						+ "from SYS.Columns where Name = ? and TableName = ? and SchemaName = ?"); //$NON-NLS-1$
				try {
					ps.setString(1, name);
					ps.setString(2, table);
					ps.setString(3, schema);	
					ResultSet rs = ps.executeQuery();
					if (rs.next()) {
						info.reloid = rs.getInt(1);
						info.attnum = rs.getShort(2);
						int specificType = rs.getInt(3);
						if (!rs.wasNull()) {
							info.type = specificType;
						}
					}	
				} finally {
					ps.close();
				}
			}
			result.add(info);
		}
		return result;
	}  
    
	/**
     * Represents a PostgreSQL Prepared object.  The actual plan preparation is performed lazily.
     */
    static class Prepared {

    	public Prepared (String name, String sql, String modifiedSql, int[] paramType, List<PgColInfo> columnMetadata) {
    		this.name = name;
    		this.sql = sql;
    		this.modifiedSql = modifiedSql;
    		this.paramType = paramType;
    		this.columnMetadata = columnMetadata;
    	}
    	
        /**
         * The object name.
         */
        final String name;

        /**
         * The original SQL statement.
         */
        final String sql;
        
        final String modifiedSql;

        /**
         * The list of pg parameter types (if set).
         */
        final int[] paramType;        
        
        /**
         * calculated column metadata
         */
        final List<PgColInfo> columnMetadata;	
    }

    /**
     * Represents a PostgreSQL Portal object.
     */
    static class Portal {

    	public Portal(String name, Prepared prepared,short[] resultColumnformat, PreparedStatementImpl stmt) {
    		this.name = name;
    		this.prepared = prepared;
    		this.resultColumnFormat = resultColumnformat;
    		this.stmt = stmt;
    	}
        /**
         * The portal name.
         */
        final String name;

        /**
         * The format used in the result set columns (if set).
         */
        final short[] resultColumnFormat;

        final Prepared prepared;
        
        volatile ResultSetImpl rs;
        
        /**
         * The prepared statement.
         */
        final PreparedStatementImpl stmt;
    }
    
    static class Cursor extends Portal {
    	
    	public Cursor (String name, String sql, PreparedStatementImpl stmt, ResultSetImpl rs, List<PgColInfo> colMetadata, boolean binary) {
    		super(name, new Prepared(UNNAMED, sql, sql, null, colMetadata), binary?new short[] {1}:null, stmt);
    		this.rs = rs;
    	}
    }    

}
