JDBCConnection.java

package tap.db;

/*
 * This file is part of TAPLibrary.
 * 
 * TAPLibrary 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 3 of the License, or
 * (at your option) any later version.
 * 
 * TAPLibrary 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 TAPLibrary.  If not, see <http://www.gnu.org/licenses/>.
 * 
 * Copyright 2012-2016 - UDS/Centre de Données astronomiques de Strasbourg (CDS),
 *                       Astronomisches Rechen Institut (ARI)
 */

import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.Driver;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLFeatureNotSupportedException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import adql.db.DBColumn;
import adql.db.DBType;
import adql.db.DBType.DBDatatype;
import adql.db.STCS;
import adql.db.STCS.Region;
import adql.query.ADQLQuery;
import adql.query.IdentifierField;
import adql.translator.ADQLTranslator;
import adql.translator.JDBCTranslator;
import adql.translator.TranslationException;
import tap.data.DataReadException;
import tap.data.ResultSetTableIterator;
import tap.data.TableIterator;
import tap.log.TAPLog;
import tap.metadata.TAPColumn;
import tap.metadata.TAPForeignKey;
import tap.metadata.TAPMetadata;
import tap.metadata.TAPMetadata.STDSchema;
import tap.metadata.TAPMetadata.STDTable;
import tap.metadata.TAPSchema;
import tap.metadata.TAPTable;
import tap.metadata.TAPTable.TableType;
import uws.ISO8601Format;
import uws.service.log.UWSLog.LogLevel;

/**
 * <p>This {@link DBConnection} implementation is theoretically able to deal with any DBMS JDBC connection.</p>
 * 
 * <p><i>Note:
 * 	"Theoretically", because its design has been done using information about Postgres, SQLite, Oracle, MySQL and Java DB (Derby).
 * 	Then it has been really tested successfully with Postgres and SQLite.
 * </i></p>
 * 
 * 
 * <h3>Only one query executed at a time!</h3>
 * 
 * <p>
 * 	With a single instance of {@link JDBCConnection} it is possible to execute only one query (whatever the type: SELECT, UPDATE, DELETE, ...)
 * 	at a time. This is indeed the simple way chosen with this implementation in order to allow the cancellation of any query by managing only
 * 	one {@link Statement}. Indeed, only a {@link Statement} has a cancel function able to stop any query execution on the database.
 * 	So all queries are executed with the same {@link Statement}. Thus, allowing the execution of one query at a time lets
 * 	abort only one query rather than several in once (though just one should have been stopped).
 * </p>
 * 
 * <p>
 * 	All the following functions are synchronized in order to prevent parallel execution of them by several threads:
 * 	{@link #addUploadedTable(TAPTable, TableIterator)}, {@link #dropUploadedTable(TAPTable)}, {@link #executeQuery(ADQLQuery)},
 * 	{@link #getTAPSchema()} and {@link #setTAPSchema(TAPMetadata)}.
 * </p>
 * 
 * <p>
 * 	To cancel a query execution the function {@link #cancel(boolean)} must be called. No error is returned by this function in case
 * 	no query is currently executing.
 * </p>
 * 
 * 
 * <h3>Deal with different DBMS features</h3>
 * 
 * <p>Update queries are taking into account whether the following features are supported by the DBMS:</p>
 * <ul>
 * 	<li><b>data definition</b>: when not supported, no update operation will be possible.
 * 	                            All corresponding functions will then throw a {@link DBException} ;
 * 	                            only {@link #executeQuery(ADQLQuery)} will be possibly called.</li>
 * 
 * 	<li><b>transactions</b>: when not supported, no transaction is started or merely used.
 * 	                         It means that in case of update failure, no rollback will be possible
 * 	                         and that already done modification will remain in the database.</li>
 * 
 * 	<li><b>schemas</b>: when the DBMS does not have the notion of schema (like SQLite), no schema creation or dropping will be obviously processed.
 * 	                    Besides, if not already done, database name of all tables will be prefixed by the schema name.</li>
 * 
 * 	<li><b>batch updates</b>: when not supported, updates will just be done, "normally, one by one.
 * 	                          In one word, there will be merely no optimization.
 * 	                          Anyway, this feature concerns only the insertions into tables.</li>
 * 
 * 	<li><b>case sensitivity of identifiers</b>: the case sensitivity of quoted identifier varies from the used DBMS. This {@link DBConnection}
 * 	                                            implementation is able to adapt itself in function of the way identifiers are stored and
 * 	                                            researched in the database. How the case sensitivity is managed by the DBMS is the problem
 * 	                                            of only one function (which can be overwritten if needed): {@link #equals(String, String, boolean)}.</li>
 * </ul>
 * 
 * <p><i><b>Warning</b>:
 * 	All these features have no impact at all on ADQL query executions ({@link #executeQuery(ADQLQuery)}).
 * </i></p>
 * 
 * 
 * <h3>Datatypes</h3>
 * 
 * <p>
 * 	All datatype conversions done while fetching a query result (via a {@link ResultSet})
 * 	are done exclusively by the returned {@link TableIterator} (so, here {@link ResultSetTableIterator}).
 * </p>
 * 
 * <p>
 * 	However, datatype conversions done while uploading a table are done here by the function
 * 	{@link #convertTypeToDB(DBType)}. This function uses first the conversion function of the translator
 * 	({@link JDBCTranslator#convertTypeToDB(DBType)}), and then {@link #defaultTypeConversion(DBType)}
 * 	if it fails.
 * </p>
 * 
 * <p>
 * 	In this default conversion, all typical DBMS datatypes are taken into account, <b>EXCEPT the geometrical types</b>
 * 	(POINT and REGION). That's why it is recommended to use a translator in which the geometrical types are supported
 * 	and managed.
 * </p>
 * 
 * 
 * <h3>Fetch size</h3>
 * 
 * <p>
 * 	The possibility to specify a "fetch size" to the JDBC driver (and more exactly to a {@link Statement}) may reveal
 * 	very helpful when dealing with large datasets. Thus, it is possible to fetch rows by block of a size represented
 * 	by this "fetch size". This is also possible with this {@link DBConnection} thanks to the function {@link #setFetchSize(int)}.
 * </p>
 * 
 * <p>
 * 	However, some JDBC driver or DBMS may not support this feature. In such case, it is then automatically disabled by
 * 	{@link JDBCConnection} so that any subsequent queries do not attempt to use it again. The {@link #supportsFetchSize}
 * 	is however reset to <code>true</code> when {@link #setFetchSize(int)} is called.
 * </p>
 * 
 * <p><i>Note 1:
 * 	The "fetch size" feature is used only for SELECT queries executed by {@link #executeQuery(ADQLQuery)}. In all other functions,
 * 	results of SELECT queries are fetched with the default parameter of the JDBC driver and its {@link Statement} implementation.
 * </i></p>
 * 
 * <p><i>Note 2:
 * 	By default, this feature is disabled. So the default value of the JDBC driver is used.
 * 	To enable it, a simple call to {@link #setFetchSize(int)} is enough, whatever is the given value.
 * </i></p>
 * 
 * @author Gr&eacute;gory Mantelet (CDS;ARI)
 * @version 2.1 (04/2016)
 * @since 2.0
 */
public class JDBCConnection implements DBConnection {

	/** DBMS name of PostgreSQL used in the database URL. */
	protected final static String DBMS_POSTGRES = "postgresql";

	/** DBMS name of SQLite used in the database URL. */
	protected final static String DBMS_SQLITE = "sqlite";

	/** DBMS name of MySQL used in the database URL. */
	protected final static String DBMS_MYSQL = "mysql";

	/** DBMS name of Oracle used in the database URL. */
	protected final static String DBMS_ORACLE = "oracle";

	/** Name of the database column giving the database name of a TAP column, table or schema. */
	protected final static String DB_NAME_COLUMN = "dbname";

	/** Connection ID (typically, the job ID). It lets identify the DB errors linked to the Job execution in the logs. */
	protected final String ID;

	/** JDBC connection (created and initialized at the creation of this {@link JDBCConnection} instance). */
	protected final Connection connection;

	/** <p>The only {@link Statement} instance that should be used in this {@link JDBCConnection}.
	 * Having the same {@link Statement} for all the interactions with the database lets cancel any when needed (e.g. when the execution is too long).</p>
	 * <p>This statement is by default NULL ; it must be initialized by the function {@link #getStatement()}.</p>
	 * @since 2.1 */
	protected Statement stmt = null;

	/**
	 * <p>It <code>true</code>, this flag indicates that the function {@link #cancel(boolean)} has been called successfully.</p>
	 * 
	 * <p>{@link #cancel(boolean)} sets this flag to <code>true</code>.</p>
	 * <p>
	 * 	All functions executing any kind of query on the database MUST set this flag to <code>false</code> before doing anything
	 * 	by calling the function {@link #resetCancel()}.
	 * </p>
	 * <p>
	 * 	This flag is particularly useful for debugging: when an exception is detected inside a function executing a query,
	 * 	this flag is used to know whether the exception should be ignored for logging (if <code>true</code>) or not.
	 * </p>
	 * <p>
	 * 	Any access (write AND read) to this flag MUST be synchronized on it using one of the following functions:
	 * 	{@link #cancel(boolean)}, {@link #resetCancel()} and {@link #isCancelled()}.
	 * </p>
	 * @since 2.1 */
	private Boolean cancelled = false;

	/** The translator this connection must use to translate ADQL into SQL. It is also used to get information about the case sensitivity of all types of identifier (schema, table, column). */
	protected final JDBCTranslator translator;

	/** Object to use if any message needs to be logged. <i>note: this logger may be NULL. If NULL, messages will never be printed.</i> */
	protected final TAPLog logger;

	/* JDBC URL MANAGEMENT */

	/** JDBC prefix of any database URL (for instance: jdbc:postgresql://127.0.0.1/myDB or jdbc:postgresql:myDB). */
	public final static String JDBC_PREFIX = "jdbc";

	/** Name (in lower-case) of the DBMS with which the connection is linked. */
	protected final String dbms;

	/* DBMS SUPPORTED FEATURES */

	/** Indicate whether the DBMS supports transactions (start, commit, rollback and end). <i>note: If no transaction is possible, no transaction will be used, but then, it will never possible to cancel modifications in case of error.</i> */
	protected boolean supportsTransaction;

	/** Indicate whether the DBMS supports the definition of data (create, update, drop, insert into schemas and tables). <i>note: If not supported, it will never possible to create TAP_SCHEMA from given metadata (see {@link #setTAPSchema(TAPMetadata)}) and to upload/drop tables (see {@link #addUploadedTable(TAPTable, TableIterator)} and {@link #dropUploadedTable(TAPTable)}).</i> */
	protected boolean supportsDataDefinition;

	/** Indicate whether the DBMS supports several updates in once (using {@link Statement#addBatch(String)} and {@link Statement#executeBatch()}). <i>note: If not supported, every updates will be done one by one. So it is not really a problem, but just a loss of optimization.</i> */
	protected boolean supportsBatchUpdates;

	/** Indicate whether the DBMS has the notion of SCHEMA. Most of the DBMS has it, but not SQLite for instance. <i>note: If not supported, the DB table name will be prefixed by the DB schema name followed by the character "_". Nevertheless, if the DB schema name is NULL, the DB table name will never be prefixed.</i> */
	protected boolean supportsSchema;

	/** <p>Indicate whether a DBMS statement is able to cancel a query execution.</p>
	 * <p> Since this information is not provided by {@link DatabaseMetaData} a first attempt is always performed.
	 * In case a {@link SQLFeatureNotSupportedException} is caught, this flag is set to false preventing any further
	 * attempt of canceling a query.</p>
	 * @since 2.1 */
	protected boolean supportsCancel = true;

	/* CASE SENSITIVITY SUPPORT */

	/** Indicate whether UNquoted identifiers will be considered as case INsensitive and stored in mixed case by the DBMS. <i>note: If FALSE, unquoted identifiers will still be considered as case insensitive for the researches, but will be stored in lower or upper case (in function of {@link #lowerCaseUnquoted} and {@link #upperCaseUnquoted}). If none of these two flags is TRUE, the storage case will be though considered as mixed.</i> */
	protected boolean supportsMixedCaseUnquotedIdentifier;
	/** Indicate whether the unquoted identifiers are stored in lower case in the DBMS. */
	protected boolean lowerCaseUnquoted;
	/** Indicate whether the unquoted identifiers are stored in upper case in the DBMS. */
	protected boolean upperCaseUnquoted;

	/** Indicate whether quoted identifiers will be considered as case INsensitive and stored in mixed case by the DBMS. <i>note: If FALSE, quoted identifiers will be considered as case sensitive and will be stored either in lower, upper or in mixed case (in function of {@link #lowerCaseQuoted}, {@link #upperCaseQuoted} and {@link #mixedCaseQuoted}). If none of these three flags is TRUE, the storage case will be mixed case.</i> */
	protected boolean supportsMixedCaseQuotedIdentifier;
	/** Indicate whether the quoted identifiers are stored in lower case in the DBMS. */
	protected boolean lowerCaseQuoted;
	/** Indicate whether the quoted identifiers are stored in mixed case in the DBMS. */
	protected boolean mixedCaseQuoted;
	/** Indicate whether the quoted identifiers are stored in upper case in the DBMS. */
	protected boolean upperCaseQuoted;

	/* FETCH SIZE */

	/** Special fetch size meaning that the JDBC driver is free to set its own guess for this value. */
	public final static int IGNORE_FETCH_SIZE = 0;
	/** Default fetch size.
	 * <i>Note 1: this value may be however ignored if the JDBC driver does not support this feature.</i>
	 * <i>Note 2: by default set to {@link #IGNORE_FETCH_SIZE}.</i> */
	public final static int DEFAULT_FETCH_SIZE = IGNORE_FETCH_SIZE;

	/** <p>Indicate whether the last fetch size operation works.</p>
	 * <p>By default, this attribute is set to <code>false</code>, meaning that the "fetch size" feature is
	 * disabled. To enable it, a simple call to {@link #setFetchSize(int)} is enough, whatever is the given value.</p>
	 * <p>If just once this operation fails, the fetch size feature will be always considered as unsupported in this {@link JDBCConnection}
	 * until the next call of {@link #setFetchSize(int)}.</p> */
	protected boolean supportsFetchSize = false;

	/** <p>Fetch size to set in the {@link Statement} in charge of executing a SELECT query.</p>
	 * <p><i>Note 1: this value must always be positive. If negative or null, it will be ignored and the {@link Statement} will keep its default behavior.</i></p>
	 * <p><i>Note 2: if this feature is enabled (i.e. has a value &gt; 0), the AutoCommit will be disabled.</i></p> */
	protected int fetchSize = DEFAULT_FETCH_SIZE;

	/**
	 * <p>Creates a JDBC connection to the specified database and with the specified JDBC driver.
	 * This connection is established using the given user name and password.<p>
	 * 
	 * <p><i><u>note:</u> the JDBC driver is loaded using <pre>Class.forName(driverPath)</pre> and the connection is created with <pre>DriverManager.getConnection(dbUrl, dbUser, dbPassword)</pre>.</i></p>
	 * 
	 * <p><i><b>Warning:</b>
	 * 	This constructor really creates a new SQL connection. Creating a SQL connection is time consuming!
	 * 	That's why it is recommended to use a pool of connections. When doing so, you should use the other constructor of this class
	 * 	({@link #JDBCConnection(Connection, JDBCTranslator, String, TAPLog)}).
	 * </i></p>
	 * 
	 * @param driverPath	Full class name of the JDBC driver.
	 * @param dbUrl			URL to the database. <i><u>note</u> This URL may not be prefixed by "jdbc:". If not, the prefix will be automatically added.</i>
	 * @param dbUser		Name of the database user.
	 * @param dbPassword	Password of the given database user.
	 * @param translator	{@link ADQLTranslator} to use in order to get SQL from an ADQL query and to get qualified DB table names.
	 * @param connID		ID of this connection. <i>note: may be NULL ; but in this case, logs concerning this connection will be more difficult to localize.</i>
	 * @param logger		Logger to use in case of need. <i>note: may be NULL ; in this case, error will never be logged, but sometimes DBException may be raised.</i>
	 * 
	 * @throws DBException	If the driver can not be found or if the connection can not merely be created (usually because DB parameters are wrong).
	 */
	public JDBCConnection(final String driverPath, final String dbUrl, final String dbUser, final String dbPassword, final JDBCTranslator translator, final String connID, final TAPLog logger) throws DBException{
		this(createConnection(driverPath, dbUrl, dbUser, dbPassword), translator, connID, logger);
	}

	/**
	 * Create a JDBC connection by wrapping the given connection.
	 * 
	 * @param conn			Connection to wrap.
	 * @param translator	{@link ADQLTranslator} to use in order to get SQL from an ADQL query and to get qualified DB table names.
	 * @param connID		ID of this connection. <i>note: may be NULL ; but in this case, logs concerning this connection will be more difficult to localize.</i>
	 * @param logger		Logger to use in case of need. <i>note: may be NULL ; in this case, error will never be logged, but sometimes DBException may be raised.</i>
	 */
	public JDBCConnection(final Connection conn, final JDBCTranslator translator, final String connID, final TAPLog logger) throws DBException{
		if (conn == null)
			throw new NullPointerException("Missing SQL connection! => can not create a JDBCConnection object.");
		if (translator == null)
			throw new NullPointerException("Missing ADQL translator! => can not create a JDBCConnection object.");

		this.connection = conn;
		this.translator = translator;
		this.ID = connID;
		this.logger = logger;

		// Set the supporting features' flags + DBMS type:
		try{
			DatabaseMetaData dbMeta = connection.getMetaData();
			dbms = getDBMSName(dbMeta.getURL());
			supportsTransaction = dbMeta.supportsTransactions();
			supportsBatchUpdates = dbMeta.supportsBatchUpdates();
			supportsDataDefinition = dbMeta.supportsDataDefinitionAndDataManipulationTransactions();
			supportsSchema = dbMeta.supportsSchemasInTableDefinitions();
			lowerCaseUnquoted = dbMeta.storesLowerCaseIdentifiers();
			upperCaseUnquoted = dbMeta.storesUpperCaseIdentifiers();
			supportsMixedCaseUnquotedIdentifier = dbMeta.supportsMixedCaseIdentifiers();
			lowerCaseQuoted = dbMeta.storesLowerCaseQuotedIdentifiers();
			mixedCaseQuoted = dbMeta.storesMixedCaseQuotedIdentifiers();
			upperCaseQuoted = dbMeta.storesUpperCaseQuotedIdentifiers();
			supportsMixedCaseQuotedIdentifier = dbMeta.supportsMixedCaseQuotedIdentifiers();
		}catch(SQLException se){
			throw new DBException("Unable to access to one or several DB metadata (url, supportsTransaction, supportsBatchUpdates, supportsDataDefinitionAndDataManipulationTransactions, supportsSchemasInTableDefinitions, storesLowerCaseIdentifiers, storesUpperCaseIdentifiers, supportsMixedCaseIdentifiers, storesLowerCaseQuotedIdentifiers, storesMixedCaseQuotedIdentifiers, storesUpperCaseQuotedIdentifiers and supportsMixedCaseQuotedIdentifiers) from the given Connection!");
		}
	}

	/**
	 * Extract the DBMS name from the given database URL.
	 * 
	 * @param dbUrl	JDBC URL to access the database. <b>This URL must start with "jdbc:" ; otherwise an exception will be thrown.</b>
	 * 
	 * @return	The DBMS name as found in the given URL.
	 * 
	 * @throws DBException	If NULL has been given, if the URL is not a JDBC one (starting with "jdbc:") or if the DBMS name is missing.
	 */
	protected static final String getDBMSName(String dbUrl) throws DBException{
		if (dbUrl == null)
			throw new DBException("Missing database URL!");

		if (!dbUrl.startsWith(JDBC_PREFIX + ":"))
			throw new DBException("This DBConnection implementation is only able to deal with JDBC connection! (the DB URL must start with \"" + JDBC_PREFIX + ":\" ; given url: " + dbUrl + ")");

		dbUrl = dbUrl.substring(5);
		int indSep = dbUrl.indexOf(':');
		if (indSep <= 0)
			throw new DBException("Incorrect database URL: " + dbUrl);

		return dbUrl.substring(0, indSep).toLowerCase();
	}

	/**
	 * Create a {@link Connection} instance using the given database parameters.
	 * The path of the JDBC driver will be used to load the adequate driver if none is found by default.
	 * 
	 * @param driverPath	Path to the JDBC driver.
	 * @param dbUrl			JDBC URL to connect to the database. <i><u>note</u> This URL may not be prefixed by "jdbc:". If not, the prefix will be automatically added.</i>
	 * @param dbUser		Name of the user to use to connect to the database.
	 * @param dbPassword	Password of the user to use to connect to the database.
	 * 
	 * @return	A new DB connection.
	 * 
	 * @throws DBException	If the driver can not be found or if the connection can not merely be created (usually because DB parameters are wrong).
	 * 
	 * @see DriverManager#getDriver(String)
	 * @see Driver#connect(String, Properties)
	 */
	private final static Connection createConnection(final String driverPath, final String dbUrl, final String dbUser, final String dbPassword) throws DBException{
		// Normalize the DB URL:
		String url = dbUrl.startsWith(JDBC_PREFIX) ? dbUrl : (JDBC_PREFIX + dbUrl);

		// Select the JDBDC driver:
		Driver d;
		try{
			d = DriverManager.getDriver(dbUrl);
		}catch(SQLException e){
			try{
				// ...load it, if necessary:
				if (driverPath == null)
					throw new DBException("Missing JDBC driver path! Since the required JDBC driver is not yet loaded, this path is needed to load it.");
				Class.forName(driverPath);
				// ...and try again:
				d = DriverManager.getDriver(dbUrl);
			}catch(ClassNotFoundException cnfe){
				throw new DBException("Impossible to find the JDBC driver \"" + driverPath + "\" !", cnfe);
			}catch(SQLException se){
				throw new DBException("No suitable JDBC driver found for the database URL \"" + dbUrl + "\" and the driver path \"" + driverPath + "\"!", se);
			}
		}

		// Build a connection to the specified database:
		try{
			Properties p = new Properties();
			if (dbUser != null)
				p.setProperty("user", dbUser);
			if (dbPassword != null)
				p.setProperty("password", dbPassword);
			Connection con = d.connect(url, p);
			return con;
		}catch(SQLException se){
			throw new DBException("Impossible to establish a connection to the database \"" + url + "\"!", se);
		}
	}

	@Override
	public final String getID(){
		return ID;
	}

	/**
	 * <p>Get the JDBC connection wrapped by this {@link JDBCConnection} object.</p>
	 * 
	 * <p><i>Note:
	 * 	This is the best way to get the JDBC connection in order to properly close it.
	 * </i></p>
	 * 
	 * @return	The wrapped JDBC connection.
	 */
	public final Connection getInnerConnection(){
		return connection;
	}

	/**
	 * <p>Tell whether this {@link JDBCConnection} is already associated with a {@link Statement}.</p>
	 * 
	 * @return	<code>true</code> if a {@link Statement} instance is already associated with this {@link JDBCConnection}
	 *        	<code>false</code> otherwise.
	 * 
	 * @throws SQLException	In case the open/close status of the current {@link Statement} instance can not be checked.
	 * 
	 * @since 2.1
	 */
	protected boolean hasStatement() throws SQLException{
		return (stmt != null && !stmt.isClosed());
	}

	/**
	 * <p>Get the only statement associated with this {@link JDBCConnection}.</p>
	 * 
	 * <p>
	 * 	If no {@link Statement} is yet existing, one is created, stored in this {@link JDBCConnection} (for further uses)
	 * 	and then returned.
	 * </p>
	 * 
	 * @return	The {@link Statement} instance associated with this {@link JDBCConnection}. <i>Never NULL</i>
	 * 
	 * @throws SQLException	In case a {@link Statement} can not be created.
	 * 
	 * @since 2.1
	 */
	protected Statement getStatement() throws SQLException{
		if (hasStatement())
			return stmt;
		else
			return (stmt = connection.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY));
	}

	/**
	 * Close the only statement associated with this {@link JDBCConnection}.
	 * 
	 * @since 2.1
	 */
	protected void closeStatement(){
		close(stmt);
		stmt = null;
	}

	/**
	 * <p>Cancel (and rollback when possible) the currently running query of this {@link JDBCConnection} instance.</p>
	 * 
	 * <p><b>Important note:</b>
	 * 	This function is effective only if the JDBC driver and DBMS both support
	 * 	this operation.
	 * </p>
	 * <p>
	 * 	If a call of this function fails the flag {@link #supportsCancel} is set to false
	 * 	so that any subsequent call of this function for this instance of {@link JDBCConnection}
	 * 	does not try any other cancellation attempt. <b>HOWEVER</b> the rollback will still continue
	 * 	to be performed if the parameter <code>rollback</code> is set to <code>true</code>.
	 * </p>
	 * 
	 * <p><i>Note 1:
	 * 	A failure of a rollback is not considered as a not supported cancellation feature by the JDBC driver or the DBMS.
	 * 	So if the cancellation succeeds but a rollback fails, a next call of this function will still try cancelling the given statement.
	 * 	In case of a rollback failure, only a WARNING is written in the log file ; no exception is thrown.
	 * </i></p>
	 * 
	 * <p><i>Note 2:
	 * 	In case of cancellation success, the flag {@link #cancelled} is set to <code>true</code>.
	 * 	Thus, the function executing a query can know that if any SQL exception is thrown, it will be due to the cancellation and
	 * 	should not be then considered as a real error (=> exception not logged but anyway propagated in order to stop any processing).
	 * </i></p></p>
	 * 
	 * <p><i>Note 3:
	 * 	This function is synchronized on the {@link #cancelled} flag.
	 * 	Thus, it may block until another synchronized block on this same flag is finished.
	 * </i></p>
	 * 
	 * @param rollback	<code>true</code> to cancel the statement AND rollback the current connection transaction,
	 *                	<code>false</code> to just cancel the statement.
	 * 
	 * @see DBConnection#cancel(boolean)
	 * @see #cancel(Statement, boolean)
	 * 
	 * @since 2.1
	 */
	@Override
	public final void cancel(final boolean rollback){
		if (supportsCancel && stmt != null){
			synchronized(cancelled){
				cancelled = cancel(stmt, rollback);
				// Log the success of the cancellation:
				if (cancelled && logger != null)
					logger.logDB(LogLevel.INFO, this, "CANCEL", "Query execution successfully stopped!", null);
			}
		}
	}

	/**
	 * <p>Cancel (and rollback when asked and if possible) the given statement.</p>
	 * 
	 * <p><b>Important note:</b>
	 * 	This function is effective only if the JDBC driver and DBMS both support
	 * 	this operation.
	 * </p>
	 * <p>
	 * 	If a call of this function fails the flag {@link #supportsCancel} is set to false
	 * 	so that any subsequent call of this function for this instance of {@link JDBCConnection}
	 * 	does not try any other cancellation attempt. <b>HOWEVER</b> the rollback will still continue
	 * 	to be performed if the parameter <code>rollback</code> is set to <code>true</code>.
	 * </p>
	 * 
	 * <p><i>Note:
	 * 	A failure of a rollback is not considered as a not supported cancellation feature by the JDBC driver or the DBMS.
	 * 	So if the cancellation succeeds but a rollback fails, a next call of this function will still try canceling the given statement.
	 * 	In case of a rollback failure, only a WARNING is written in the log file ; no exception is thrown.
	 * </i></p>
	 * 
	 * @param stmt		The statement to cancel. <i>Note: if closed or NULL, no exception will be thrown and only a rollback will be attempted if asked in parameter.</i>
	 * @param rollback	<code>true</code> to cancel the statement AND rollback the current connection transaction,
	 *                	<code>false</code> to just cancel the statement.
	 * 
	 * @return	<code>true</code> if the cancellation succeeded (or none was running),
	 *        	<code>false</code> otherwise (and especially if the "cancel" operation is not supported).
	 * 
	 * @since 2.1
	 */
	protected boolean cancel(final Statement stmt, final boolean rollback){
		try{
			// If the statement is not already closed, cancel its current query execution:
			if (supportsCancel && stmt != null && !stmt.isClosed()){
				stmt.cancel();
				return true;
			}else
				return false;

		}catch(SQLFeatureNotSupportedException sfnse){
			// prevent further cancel attempts:
			supportsCancel = false;
			// log a warning:
			if (logger != null)
				logger.logDB(LogLevel.WARNING, this, "CANCEL", "This JDBC driver does not support Statement.cancel(). No further cancel attempt will be performed with this JDBCConnection instance.", sfnse);
			return false;

		}catch(SQLException se){
			if (logger != null)
				logger.logDB(LogLevel.ERROR, this, "CANCEL", "Abortion of the current query apparently fails! The query may still run on the database server.", se);
			return false;
		}
		// Whatever happens, rollback all executed operations (only if rollback=true and if in a transaction ; that's to say if AutoCommit = false):
		finally{
			if (rollback && supportsTransaction){
				try{
					if (!connection.getAutoCommit()){
						connection.rollback();
						connection.setAutoCommit(true);
					}
				}catch(SQLException se){
					if (logger != null)
						logger.logDB(LogLevel.ERROR, this, "CANCEL", "Query execution successfully stopped BUT the rollback fails!", se);
				}
			}
		}
	}

	/**
	 * <p>Tell whether the last query execution has been canceled.</p>
	 * 
	 * <p><i>Note:
	 * 	This function is synchronized on the {@link #cancelled} flag.
	 * 	Thus, it may block until another synchronized block on this same flag is finished.
	 * </i></p>
	 * 
	 * @return	<code>true</code> if the last query execution has been cancelled,
	 *        	<code>false</code> otherwise.
	 * 
	 * @since 2.1
	 */
	protected final boolean isCancelled(){
		synchronized(cancelled){
			return cancelled;
		}
	}

	/**
	 * <p>Reset the {@link #cancelled} flag to <code>false</code>.</p>
	 * 
	 * <p><i>Note:
	 * 	This function is synchronized on the {@link #cancelled} flag.
	 * 	Thus, it may block until another synchronized block on this same flag is finished.
	 * </i></p>
	 * 
	 * @since 2.1
	 */
	protected final void resetCancel(){
		synchronized(cancelled){
			cancelled = false;
		}
	}

	/* ********************* */
	/* INTERROGATION METHODS */
	/* ********************* */
	@Override
	public synchronized TableIterator executeQuery(final ADQLQuery adqlQuery) throws DBException{
		// Starting of new query execution => disable the cancel flag:
		resetCancel();

		String sql = null;
		ResultSet result = null;
		try{
			// 1. Translate the ADQL query into SQL:
			if (logger != null)
				logger.logDB(LogLevel.INFO, this, "TRANSLATE", "Translating ADQL: " + adqlQuery.toADQL().replaceAll("(\t|\r?\n)+", " "), null);
			sql = translator.translate(adqlQuery);

			// 2. Create the statement and if needed, configure it for the given fetch size:
			if (supportsTransaction && supportsFetchSize && fetchSize > 0){
				try{
					connection.setAutoCommit(false);
				}catch(SQLException se){
					if (!isCancelled()){
						supportsFetchSize = false;
						if (logger != null)
							logger.logDB(LogLevel.WARNING, this, "RESULT", "Fetch size unsupported!", null);
					}
				}
			}

			getStatement();

			if (supportsFetchSize){
				try{
					stmt.setFetchSize(fetchSize);
				}catch(SQLException se){
					if (!isCancelled()){
						supportsFetchSize = false;
						if (logger != null)
							logger.logDB(LogLevel.WARNING, this, "RESULT", "Fetch size unsupported!", null);
					}
				}
			}

			// 3. Execute the SQL query:
			if (logger != null)
				logger.logDB(LogLevel.INFO, this, "EXECUTE", "SQL query: " + sql.replaceAll("(\t|\r?\n)+", " "), null);
			result = stmt.executeQuery(sql);

			// 4. Return the result through a TableIterator object:
			if (logger != null)
				logger.logDB(LogLevel.INFO, this, "RESULT", "Returning result (" + (supportsFetchSize ? "fetch size = " + fetchSize : "all in once") + ").", null);
			return createTableIterator(result, adqlQuery.getResultingColumns());

		}catch(Exception ex){
			// Close the ResultSet, if one was open:
			close(result);
			// Ensure the query is not running anymore and rollback the transaction:
			cancel(true);
			// If still not canceled, log this fact as an error:
			if (!isCancelled() && logger != null)
				logger.logDB(LogLevel.ERROR, this, "EXECUTE", "Unexpected error while EXECUTING SQL query!", null);
			// Close the open statement, if one was open:
			closeStatement();
			// Propagate the exception with an appropriate error message:
			if (ex instanceof SQLException)
				throw new DBException("Unexpected error while executing a SQL query: " + ex.getMessage(), ex);
			else if (ex instanceof TranslationException)
				throw new DBException("Unexpected error while translating ADQL into SQL: " + ex.getMessage(), ex);
			else if (ex instanceof DataReadException)
				throw new DBException("Impossible to read the query result, because: " + ex.getMessage(), ex);
			else
				throw new DBException("Unexpected error while executing an ADQL query: " + ex.getMessage(), ex);
		}
	}

	/**
	 * <p>Create a {@link TableIterator} instance which lets reading the given result table.</p>
	 * 
	 * <p><b>Important note:</b>
	 * 	This function also set to NULL the statement of this {@link JDBCConnection} instance: {@link #stmt}.
	 * 	However, the statement is not closed ; it is just given to a {@link ResultSetTableIterator} iterator
	 * 	which will close it in the same time as the given {@link ResultSet}, when its function
	 * 	{@link ResultSetTableIterator#close()} is called.
	 * </p>
	 * 
	 * @param rs				Result of an SQL query.
	 * @param resultingColumns	Metadata corresponding to each columns of the result.
	 * 
	 * @return	A {@link TableIterator} instance.
	 * 
	 * @throws DataReadException	If the metadata (columns count and types) can not be fetched
	 *                          	or if any other error occurs.
	 */
	protected TableIterator createTableIterator(final ResultSet rs, final DBColumn[] resultingColumns) throws DataReadException{
		// Dis-associate the current Statement from this JDBCConnection instance:
		Statement itStmt = stmt;
		stmt = null;
		// Return a TableIterator wrapping the given ResultSet:
		try{
			return new ResultSetTableIterator(itStmt, rs, translator, dbms, resultingColumns);
		}catch(Throwable t){
			throw (t instanceof DataReadException) ? (DataReadException)t : new DataReadException(t);
		}
	}

	/* *********************** */
	/* TAP_SCHEMA MANIPULATION */
	/* *********************** */

	/**
	 * Tell when, compared to the other TAP standard tables, a given standard TAP table should be created.
	 * 
	 * @param table	Standard TAP table.
	 * 
	 * @return	An index between 0 and 4 (included) - 0 meaning the first table to create whereas 4 is the last one.
	 *          -1 is returned if NULL is given in parameter of if the standard table is not taken into account here.
	 */
	protected int getCreationOrder(final STDTable table){
		if (table == null)
			return -1;

		switch(table){
			case SCHEMAS:
				return 0;
			case TABLES:
				return 1;
			case COLUMNS:
				return 2;
			case KEYS:
				return 3;
			case KEY_COLUMNS:
				return 4;
			default:
				return -1;
		}
	}

	/* ************************************ */
	/* GETTING TAP_SCHEMA FROM THE DATABASE */
	/* ************************************ */

	/**
	 * <p>In this implementation, this function is first creating a virgin {@link TAPMetadata} object
	 * that will be filled progressively by calling the following functions:</p>
	 * <ol>
	 * 	<li>{@link #loadSchemas(TAPTable, TAPMetadata, Statement)}</li>
	 * 	<li>{@link #loadTables(TAPTable, TAPMetadata, Statement)}</li>
	 * 	<li>{@link #loadColumns(TAPTable, List, Statement)}</li>
	 * 	<li>{@link #loadKeys(TAPTable, TAPTable, List, Statement)}</li>
	 * </ol>
	 * 
	 * <p><i>Note:
	 * 	If schemas are not supported by this DBMS connection, the DB name of all tables will be set to NULL
	 * 	and the DB name of all tables will be prefixed by the ADQL name of their respective schema.
	 * </i></p>
	 * 
	 * @see tap.db.DBConnection#getTAPSchema()
	 */
	@Override
	public synchronized TAPMetadata getTAPSchema() throws DBException{
		// Starting of new query execution => disable the cancel flag:
		resetCancel();

		// Build a virgin TAP metadata:
		TAPMetadata metadata = new TAPMetadata();

		// Get the definition of the standard TAP_SCHEMA tables:
		TAPSchema tap_schema = TAPMetadata.getStdSchema(supportsSchema);

		// LOAD ALL METADATA FROM THE STANDARD TAP TABLES:
		try{
			// create a common statement for all loading functions:
			getStatement();

			// load all schemas from TAP_SCHEMA.schemas:
			if (logger != null)
				logger.logDB(LogLevel.INFO, this, "LOAD_TAP_SCHEMA", "Loading TAP_SCHEMA.schemas.", null);
			loadSchemas(tap_schema.getTable(STDTable.SCHEMAS.label), metadata, stmt);

			// load all tables from TAP_SCHEMA.tables:
			if (logger != null)
				logger.logDB(LogLevel.INFO, this, "LOAD_TAP_SCHEMA", "Loading TAP_SCHEMA.tables.", null);
			List<TAPTable> lstTables = loadTables(tap_schema.getTable(STDTable.TABLES.label), metadata, stmt);

			// load all columns from TAP_SCHEMA.columns:
			if (logger != null)
				logger.logDB(LogLevel.INFO, this, "LOAD_TAP_SCHEMA", "Loading TAP_SCHEMA.columns.", null);
			loadColumns(tap_schema.getTable(STDTable.COLUMNS.label), lstTables, stmt);

			// load all foreign keys from TAP_SCHEMA.keys and TAP_SCHEMA.key_columns:
			if (logger != null)
				logger.logDB(LogLevel.INFO, this, "LOAD_TAP_SCHEMA", "Loading TAP_SCHEMA.keys and TAP_SCHEMA.key_columns.", null);
			loadKeys(tap_schema.getTable(STDTable.KEYS.label), tap_schema.getTable(STDTable.KEY_COLUMNS.label), lstTables, stmt);

		}catch(SQLException se){
			if (!isCancelled() && logger != null)
				logger.logDB(LogLevel.ERROR, this, "LOAD_TAP_SCHEMA", "Impossible to create a Statement!", se);
			throw new DBException("Can not create a Statement!", se);
		}finally{
			cancel(true);
			closeStatement();
		}

		return metadata;
	}

	/**
	 * <p>Load into the given metadata all schemas listed in TAP_SCHEMA.schemas.</p>
	 * 
	 * <p><i>Note:
	 * 	If schemas are not supported by this DBMS connection, the DB name of the loaded schemas is set to NULL.
	 * </i></p>
	 * 
	 * @param tableDef		Definition of the table TAP_SCHEMA.schemas.
	 * @param metadata		Metadata to fill with all found schemas.
	 * @param stmt			Statement to use in order to interact with the database.
	 * 
	 * @throws DBException	If any error occurs while interacting with the database.
	 */
	protected void loadSchemas(final TAPTable tableDef, final TAPMetadata metadata, final Statement stmt) throws DBException{
		ResultSet rs = null;
		try{
			// Determine whether the dbName column exists:
			/* note: if the schema notion is not supported by this DBMS, the column "dbname" is ignored. */
			boolean hasDBName = supportsSchema && isColumnExisting(tableDef.getDBSchemaName(), tableDef.getDBName(), DB_NAME_COLUMN, connection.getMetaData());

			// Build the SQL query:
			StringBuffer sqlBuf = new StringBuffer("SELECT ");
			sqlBuf.append(translator.getColumnName(tableDef.getColumn("schema_name")));
			sqlBuf.append(", ").append(translator.getColumnName(tableDef.getColumn("description")));
			sqlBuf.append(", ").append(translator.getColumnName(tableDef.getColumn("utype")));
			if (hasDBName)
				sqlBuf.append(", ").append(DB_NAME_COLUMN);
			sqlBuf.append(" FROM ").append(translator.getTableName(tableDef, supportsSchema)).append(';');

			// Execute the query:
			rs = stmt.executeQuery(sqlBuf.toString());

			// Create all schemas:
			while(rs.next()){
				String schemaName = rs.getString(1),
						description = rs.getString(2), utype = rs.getString(3),
						dbName = (hasDBName ? rs.getString(4) : null);

				// create the new schema:
				TAPSchema newSchema = new TAPSchema(schemaName, nullifyIfNeeded(description), nullifyIfNeeded(utype));
				if (dbName != null && dbName.trim().length() > 0)
					newSchema.setDBName(dbName);

				// add the new schema inside the given metadata:
				metadata.addSchema(newSchema);
			}
		}catch(SQLException se){
			if (!isCancelled() && logger != null)
				logger.logDB(LogLevel.ERROR, this, "LOAD_TAP_SCHEMA", "Impossible to load schemas from TAP_SCHEMA.schemas!", se);
			throw new DBException("Impossible to load schemas from TAP_SCHEMA.schemas!", se);
		}finally{
			close(rs);
		}
	}

	/**
	 * <p>Load into the corresponding metadata all tables listed in TAP_SCHEMA.tables.</p>
	 * 
	 * <p><i>Note:
	 * 	Schemas are searched in the given metadata by their ADQL name and case sensitively.
	 * 	If they can not be found a {@link DBException} is thrown.
	 * </i></p>
	 * 
	 * <p><i>Note:
	 * 	If schemas are not supported by this DBMS connection, the DB name of the loaded
	 * 	{@link TAPTable}s is prefixed by the ADQL name of their respective schema.
	 * </i></p>
	 * 
	 * @param tableDef		Definition of the table TAP_SCHEMA.tables.
	 * @param metadata		Metadata (containing already all schemas listed in TAP_SCHEMA.schemas).
	 * @param stmt			Statement to use in order to interact with the database.
	 * 
	 * @return	The complete list of all loaded tables. <i>note: this list is required by {@link #loadColumns(TAPTable, List, Statement)}.</i>
	 * 
	 * @throws DBException	If a schema can not be found, or if any other error occurs while interacting with the database.
	 */
	protected List<TAPTable> loadTables(final TAPTable tableDef, final TAPMetadata metadata, final Statement stmt) throws DBException{
		ResultSet rs = null;
		try{
			// Determine whether the dbName column exists:
			boolean hasDBName = isColumnExisting(tableDef.getDBSchemaName(), tableDef.getDBName(), DB_NAME_COLUMN, connection.getMetaData());

			// Build the SQL query:
			StringBuffer sqlBuf = new StringBuffer("SELECT ");
			sqlBuf.append(translator.getColumnName(tableDef.getColumn("schema_name")));
			sqlBuf.append(", ").append(translator.getColumnName(tableDef.getColumn("table_name")));
			sqlBuf.append(", ").append(translator.getColumnName(tableDef.getColumn("table_type")));
			sqlBuf.append(", ").append(translator.getColumnName(tableDef.getColumn("description")));
			sqlBuf.append(", ").append(translator.getColumnName(tableDef.getColumn("utype")));
			if (hasDBName)
				sqlBuf.append(", ").append(DB_NAME_COLUMN);
			sqlBuf.append(" FROM ").append(translator.getTableName(tableDef, supportsSchema)).append(';');

			// Execute the query:
			rs = stmt.executeQuery(sqlBuf.toString());

			// Create all tables:
			ArrayList<TAPTable> lstTables = new ArrayList<TAPTable>();
			while(rs.next()){
				String schemaName = rs.getString(1),
						tableName = rs.getString(2), typeStr = rs.getString(3),
						description = rs.getString(4), utype = rs.getString(5),
						dbName = (hasDBName ? rs.getString(6) : null);

				// get the schema:
				TAPSchema schema = metadata.getSchema(schemaName);
				if (schema == null){
					if (logger != null)
						logger.logDB(LogLevel.ERROR, this, "LOAD_TAP_SCHEMA", "Impossible to find the schema of the table \"" + tableName + "\": \"" + schemaName + "\"!", null);
					throw new DBException("Impossible to find the schema of the table \"" + tableName + "\": \"" + schemaName + "\"!");
				}

				// If the table name is qualified, check its prefix (it must match to the schema name):
				int endPrefix = tableName.indexOf('.');
				if (endPrefix >= 0){
					if (endPrefix == 0)
						throw new DBException("Incorrect table name syntax: \"" + tableName + "\"! Missing schema name (before '.').");
					else if (endPrefix == tableName.length() - 1)
						throw new DBException("Incorrect table name syntax: \"" + tableName + "\"! Missing table name (after '.').");
					else if (schemaName == null)
						throw new DBException("Incorrect schema prefix for the table \"" + tableName.substring(endPrefix + 1) + "\": this table is not in a schema, according to the column \"schema_name\" of TAP_SCHEMA.tables!");
					else if (!tableName.substring(0, endPrefix).trim().equalsIgnoreCase(schemaName))
						throw new DBException("Incorrect schema prefix for the table \"" + schemaName + "." + tableName.substring(tableName.indexOf('.') + 1) + "\": " + tableName + "! Mismatch between the schema specified in prefix of the column \"table_name\" and in the column \"schema_name\".");
				}

				// resolve the table type (if any) ; by default, it will be "table":
				TableType type = TableType.table;
				if (typeStr != null){
					try{
						type = TableType.valueOf(typeStr.toLowerCase());
					}catch(IllegalArgumentException iae){}
				}

				// create the new table:
				TAPTable newTable = new TAPTable(tableName, type, nullifyIfNeeded(description), nullifyIfNeeded(utype));
				newTable.setDBName(dbName);

				// add the new table inside its corresponding schema:
				schema.addTable(newTable);