TempTableDataSourceProvider.java
/*
** Module : TempTableDataSourceProvider.java
** Abstract : Provides the context-local data source for legacy temp-table use.
**
** Copyright (c) 2019-2024, Golden Code Development Corporation.
**
** -#- -I- --Date-- --------------------------------Description----------------------------------
** 001 ECF 20191001 Created initial version with basic runtime support.
** OM 20191202 Implemented getConnection() methods. Using delegation instead of proxy because
** the proxied object lacks the default c'tor.
** 002 AIL 20201020 Added prepared statement cache at connection level.
** ECF 20201022 Javadoc cleanup.
** CA 20220908 Increased the prepared statement cache.
** CA 20221006 Increased prepared statement cache to 65536.
** CA 20221011 Decreased prepared statement cache to 2048. The PreparedStatements are not cleaned when
** a temp-table is dropped, and this will pin in memory the entire temp-table (with rows,
** indexes, etc) at H2 level.
** 003 AL2 20230210 Major refactor. Proxied connection is no longer an annonymous class. Make it named
** to identify if the connection is proxied or not.
** 004 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 005 AL2 20230512 Made psCache size configurable. Set default to 8196.
** 006 RAA 20230613 Added SQL as a parameter when creating an UnclosablePreparedStatement.
** 007 DDF 20230627 Changed the configuration path for the psCache size.
** DDF 20230627 Removed the static block that reads the configuration sizes of the cache.
** CacheManager will handle the finding of the configuration size and the
** cache creation.
** DDF 20230706 Replaced createLRUCache call with another one that will use a null discriminator
** by default.
** 008 CA 20240307 Added a secondary pool of prepared statements, to be used when the instance from the main
** cache is checked out.
** 009 RAA 20240423 getDataSource now includes the database configuration.
*/
/*
** This program is free software: you can redistribute it and/or modify
** it under the terms of the GNU Affero General Public License as
** published by the Free Software Foundation, either version 3 of the
** License, or (at your option) any later version.
**
** This program 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 Affero General Public License for more details.
**
** You may find a copy of the GNU Affero GPL version 3 at the following
** location: https://www.gnu.org/licenses/agpl-3.0.en.html
**
** Additional terms under GNU Affero GPL version 3 section 7:
**
** Under Section 7 of the GNU Affero GPL version 3, the following additional
** terms apply to the works covered under the License. These additional terms
** are non-permissive additional terms allowed under Section 7 of the GNU
** Affero GPL version 3 and may not be removed by you.
**
** 0. Attribution Requirement.
**
** You must preserve all legal notices or author attributions in the covered
** work or Appropriate Legal Notices displayed by works containing the covered
** work. You may not remove from the covered work any author or developer
** credit already included within the covered work.
**
** 1. No License To Use Trademarks.
**
** This license does not grant any license or rights to use the trademarks
** Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks
** of Golden Code Development Corporation. You are not authorized to use the
** name Golden Code, FWD, or the names of any author or contributor, for
** publicity purposes without written authorization.
**
** 2. No Misrepresentation of Affiliation.
**
** You may not represent yourself as Golden Code Development Corporation or FWD.
**
** You may not represent yourself for publicity purposes as associated with
** Golden Code Development Corporation, FWD, or any author or contributor to
** the covered work, without written authorization.
**
** 3. No Misrepresentation of Source or Origin.
**
** You may not represent the covered work as solely your work. All modified
** versions of the covered work must be marked in a reasonable way to make it
** clear that the modified work is not originating from Golden Code Development
** Corporation or FWD. All modified versions must contain the notices of
** attribution required in this license.
*/
package com.goldencode.p2j.persist.orm;
import java.io.*;
import java.sql.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.logging.*;
import javax.sql.DataSource;
import com.goldencode.cache.*;
import com.goldencode.p2j.persist.*;
import com.goldencode.p2j.persist.dialect.*;
import com.goldencode.p2j.persist.orm.types.*;
import com.goldencode.p2j.security.*;
import com.goldencode.p2j.util.logging.CentralLogger;
/**
* A data source provider which supplies a JDBC connection for legacy temp-table use. The connection is not
* closed until all temp-tables have been dropped explicitly. The connection provided is a hard-coded proxy
* which delegates most service requests to a backing, physical JDBC connection, but intercepts certain
* requests to implement the desired behavior.
* <p>
* Implements prepared statement caching in a manner compatible with H2's prepared statement implementation.
*/
class TempTableDataSourceProvider
implements DataSourceProvider
{
/**
* Check if the connection is a proxy delivered by this provider.
*
* @param conn
* The connection to be checked.
*
* @return {@code true} if the provided connection is proxied by this provider.
*/
public static boolean isProxy(Connection conn)
{
return conn instanceof DataSourceImpl.TempProxyConnection;
}
/**
* Retrieve the original connection without proxy. Use this with care as it bypassed
* some critical business logic. Avoid executing SQL directly on the physical connection,
* unless you are sure that the wrapper is not affected.
*
* @param conn
* The connection to unwrap.
*
* @return The physical connection. The unwrapping works only if this is a proxy of this
* provider.
*/
public static Connection unwrapProxy(Connection conn)
{
if (conn == null || !isProxy(conn))
{
return null;
}
DataSourceImpl.TempProxyConnection proxy = (DataSourceImpl.TempProxyConnection) conn;
return proxy.ctx.physicalConn;
}
/**
* Obtain the {@code DataSource} for a specific {@code Database}, within the current context.
*
* @param database
* The FWD {@code Database} that needs a connection to backing database.
* @param cfg
* The configuration for the given database.
*
* @return the the {@code DataSource} for the specified {@code Database}.
*/
@Override
public DataSource getDataSource(Database database, DatabaseConfig cfg)
{
return new DataSourceImpl();
}
/**
* Implementation of the {@code DataSource} interface, which maintains context-local data for each
* private temp-table connection.
*/
private static class DataSourceImpl
implements DataSource
{
/** Logger */
private static final CentralLogger LOG = CentralLogger.get(DataSourceImpl.class.getName());
/**
* Attempts to establish a connection with the data source of the {@code _temp} database.
* Uses the credentials from the {@code DatabaseManager} configuration.
*
* @return a connection to the data source.
*
* @throws SQLException
* if a database access error occurs.
*/
@Override
public Connection getConnection()
throws SQLException
{
return getConnection(null, null);
}
/**
* Attempts to establish a connection with the data source of the {@code _temp} database.
* If credentials are {@code null}, the {@code DatabaseManager} configuration is inspected
* for their values.
*
* @param username
* The database user on whose behalf the connection is made. If {@code null}, the
* value is read form @code DatabaseManager}'s configuration.
* @param password
* The password of the database user on whose behalf the connection is made. If
* {@code null}, the value is read form @code DatabaseManager}'s configuration.
*
* @return a connection to the data source.
*
* @throws SQLException
* if a database access error occurs.
*/
@Override
public Connection getConnection(String username, String password)
throws SQLException
{
Context ctx = context.get();
if (ctx.physicalConn == null)
{
DatabaseConfig cfg;
try
{
cfg = DatabaseManager.getConfiguration(DatabaseManager.TEMP_TABLE_DB);
}
catch (PersistenceException pe)
{
LOG.warning("", pe);
throw new SQLException("Failed to get configuration for TEMP database", pe);
}
Settings settings = cfg.getOrmSettings();
String url = settings.getString(Settings.URL, null);
if (username == null)
{
username = settings.getString(Settings.USER, null);
}
if (password == null)
{
password = settings.getString(Settings.PASSWORD, null);
}
ctx.physicalConn = DriverManager.getConnection(url, username, password);
// setup the blob and clob creators
Dialect dialect = Dialect.getDialect(settings);
TypeManager.addBlobCreator(url, dialect::blobCreator);
TypeManager.addClobCreator(url, dialect::clobCreator);
ctx.psCache = CacheManager.createLRUCache(TempTableDataSourceProvider.class, 8192);
ctx.psCache.addCacheExpiryListener(ctx);
}
if (ctx.proxyConn == null)
{
ctx.proxyConn = new TempProxyConnection(ctx);
}
return ctx.proxyConn;
}
@Override
public boolean isWrapperFor(Class<?> iface)
throws SQLException
{
if (Wrapper.class.equals(iface) || DataSourceImpl.class.isAssignableFrom(iface))
{
return true;
}
throw new SQLException("Data source neither implements nor is a wrapper for " + iface);
}
@Override
public <T> T unwrap(Class<T> iface)
throws SQLException
{
// will not actually return null, because isWrapperFor will throw an exception if it
// cannot respond with true, per the Wrapper contract
return isWrapperFor(iface) ? (T) this : null;
}
@Override public PrintWriter getLogWriter() { return null; }
@Override public void setLogWriter(PrintWriter out) { }
@Override public void setLoginTimeout(int seconds) { }
@Override public int getLoginTimeout() { return 0; }
@Override
public Logger getParentLogger()
throws SQLFeatureNotSupportedException
{
throw new SQLFeatureNotSupportedException();
}
/**
* Intercept close requests to close the physical database connection only when there is no temp-table
* which still needs it.
*
* @throws SQLException
* if there is an error closing the connection.
*/
public void close()
throws SQLException
{
Context ctx = context.get();
if (ctx.physicalConn != null && TemporaryBuffer.getTableCount() == 0)
{
ctx.physicalConn.close();
ctx.physicalConn = null;
ctx.psCache = null;
}
}
/** Context local variable to hold context-specific data */
private static final ContextLocal<Context> context = new ContextLocal<Context>()
{
/**
* Get the weight of this context-local var.
*
* @return Always {@link WeightFactor#LEVEL_2}.
*/
@Override
public WeightFactor getWeight()
{
return WeightFactor.LEVEL_2;
};
/**
* Instantiate a new {@code Context} instance for each context.
*
* @return Context object.
*/
protected Context initialValue()
{
// System.out.println("+++ New TT DS Context " + this);
return new Context();
}
/**
* Clean up the context before it is removed.
*
* @param context
* Context object stored in this variable.
*/
protected void cleanup(Context context)
{
if (context != null)
{
if (context.physicalConn != null)
{
try
{
context.physicalConn.close();
}
catch (SQLException throwable)
{
LOG.severe("", throwable);
}
context.physicalConn = null;
}
context.proxyConn = null;
}
// System.out.println("+++ CleanedUp TT DS Context " + this);
}
};
/**
* Context-specific data.
*/
private static class Context
implements CacheExpiryListener<PSKey, UnclosablePreparedStatement>
{
/** Physical JDBC connection */
private Connection physicalConn = null;
/** Proxied JDBC connection which overrides close behavior */
private Connection proxyConn = null;
/** Prepared statement cache which closes statements as they expire from the cache */
private ExpiryCache<PSKey, UnclosablePreparedStatement> psCache = null;
/**
* Manage prepared statements which expire from the cache, by closing them.
*
* @param event
* Event which contains the prepared statement(s) which have expired from the cache.
*/
@Override
public void cacheExpired(CacheExpiryEvent<PSKey, UnclosablePreparedStatement> event)
{
for (UnclosablePreparedStatement ps : event.getExpiredEntries().values())
{
try
{
if (!ps.isClosed())
{
if (ps.isCheckedOut())
{
ps.setCloseOnCheckIn(true);
}
else
{
ps.forceClose();
}
}
}
catch (SQLException exc)
{
// eat it
}
}
}
}
/**
* A proxy connection for the temporary database. We do such proxy to allow us to cache prepared
* statements, even if we close them from the FWD code.
*/
public class TempProxyConnection
implements Connection
{
/** */
private Context ctx;
/**
* @param ctx
*/
TempProxyConnection(Context ctx)
{
this.ctx = ctx;
}
public void close() throws SQLException
{
DataSourceImpl.this.close();
}
public PreparedStatement prepareStatement(String sql)
throws SQLException
{
// prepare using default result set type and concurrency
return prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
}
public PreparedStatement prepareStatement(String sql,
int resultSetType,
int resultSetConcurrency)
throws SQLException
{
PSKey key = new PSKey(sql, resultSetType, resultSetConcurrency);
UnclosablePreparedStatement ps = ctx.psCache.get(key);
if (ps == null || ps.isClosed())
{
if (ps != null)
{
if (LOG.isLoggable(Level.SEVERE))
{
LOG.log(Level.SEVERE, "Closed statements shouldn't be cached.");
}
}
PreparedStatement raw = ctx.physicalConn.prepareStatement(sql, resultSetType, resultSetConcurrency);
ps = new UnclosablePreparedStatement(raw, sql, false);
ctx.psCache.put(key, ps);
}
else if (ps.isCheckedOut())
{
UnclosablePreparedStatement pooledPs = ps.getFromPool();
if (pooledPs == null)
{
// the prepared statement is already in use
// create another one and add it to the secondary pool.
PreparedStatement raw = ctx.physicalConn.prepareStatement(sql, resultSetType, resultSetConcurrency);
pooledPs = new UnclosablePreparedStatement(raw, sql, false);
ps.addToPool(pooledPs);
pooledPs = ps.getFromPool();
}
ps = pooledPs;
if (LOG.isLoggable(Level.FINE))
{
LOG.log(Level.FINE, "Multiply-cached PreparedStatement: " + sql);
}
}
ps.checkOut();
return ps;
}
public <T> T unwrap(Class<T> iface) throws SQLException { return ctx.physicalConn.unwrap(iface); }
public boolean isWrapperFor(Class<?> iface) throws SQLException { return ctx.physicalConn.isWrapperFor(iface); }
public Statement createStatement() throws SQLException { return ctx.physicalConn.createStatement(); }
public CallableStatement prepareCall(String sql) throws SQLException { return ctx.physicalConn.prepareCall(sql); }
public String nativeSQL(String sql) throws SQLException { return ctx.physicalConn.nativeSQL(sql); }
public void setAutoCommit(boolean autoCommit) throws SQLException { ctx.physicalConn.setAutoCommit(autoCommit); }
public boolean getAutoCommit() throws SQLException { return ctx.physicalConn.getAutoCommit(); }
public void commit() throws SQLException { ctx.physicalConn.commit(); }
public void rollback() throws SQLException { ctx.physicalConn.rollback(); }
public boolean isClosed() throws SQLException { return ctx.physicalConn == null || ctx.physicalConn.isClosed(); }
public DatabaseMetaData getMetaData() throws SQLException { return ctx.physicalConn.getMetaData(); }
public void setReadOnly(boolean readOnly) throws SQLException { ctx.physicalConn.setReadOnly(readOnly); }
public boolean isReadOnly() throws SQLException { return ctx.physicalConn.isReadOnly(); }
public void setCatalog(String catalog) throws SQLException { ctx.physicalConn.setCatalog(catalog); }
public String getCatalog() throws SQLException { return ctx.physicalConn.getCatalog(); }
public void setTransactionIsolation(int level) throws SQLException { ctx.physicalConn.setTransactionIsolation(level); }
public int getTransactionIsolation() throws SQLException { return ctx.physicalConn.getTransactionIsolation(); }
public SQLWarning getWarnings() throws SQLException { return ctx.physicalConn.getWarnings(); }
public void clearWarnings() throws SQLException { ctx.physicalConn.clearWarnings(); }
public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException {
return ctx.physicalConn.createStatement(resultSetType, resultSetConcurrency); }
public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
return ctx.physicalConn.prepareCall(sql, resultSetType, resultSetConcurrency); }
public Map<String, Class<?>> getTypeMap() throws SQLException {
return ctx.physicalConn.getTypeMap(); }
public void setTypeMap(Map<String, Class<?>> map) throws SQLException {
ctx.physicalConn.setTypeMap(map); }
public void setHoldability(int holdability) throws SQLException { ctx.physicalConn.setHoldability(holdability); }
public int getHoldability() throws SQLException { return ctx.physicalConn.getHoldability(); }
public Savepoint setSavepoint() throws SQLException { return ctx.physicalConn.setSavepoint(); }
public Savepoint setSavepoint(String name) throws SQLException { return ctx.physicalConn.setSavepoint(name); }
public void rollback(Savepoint savepoint) throws SQLException { ctx.physicalConn.rollback(savepoint); }
public void releaseSavepoint(Savepoint savepoint) throws SQLException { ctx.physicalConn.releaseSavepoint(savepoint); }
public Statement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
return ctx.physicalConn.createStatement(resultSetType, resultSetConcurrency, resultSetHoldability); }
public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
return ctx.physicalConn.prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability); }
public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
return ctx.physicalConn.prepareCall(sql, resultSetType, resultSetConcurrency, resultSetHoldability); }
public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException {
return ctx.physicalConn.prepareStatement(sql, autoGeneratedKeys); }
public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException {
return ctx.physicalConn.prepareStatement(sql, columnIndexes); }
public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException {
return ctx.physicalConn.prepareStatement(sql, columnNames); }
public Clob createClob() throws SQLException { return ctx.physicalConn.createClob(); }
public Blob createBlob() throws SQLException { return ctx.physicalConn.createBlob(); }
public NClob createNClob() throws SQLException { return ctx.physicalConn.createNClob(); }
public SQLXML createSQLXML() throws SQLException { return ctx.physicalConn.createSQLXML(); }
public boolean isValid(int timeout) throws SQLException { return ctx.physicalConn.isValid(timeout); }
public void setClientInfo(String name, String value) throws SQLClientInfoException { ctx.physicalConn.setClientInfo(name, value); }
public void setClientInfo(Properties properties) throws SQLClientInfoException { ctx.physicalConn.setClientInfo(properties); }
public String getClientInfo(String name) throws SQLException { return ctx.physicalConn.getClientInfo(name); }
public Properties getClientInfo() throws SQLException { return ctx.physicalConn.getClientInfo(); }
public Array createArrayOf(String typeName, Object[] elements) throws SQLException {
return ctx.physicalConn.createArrayOf(typeName, elements); }
public Struct createStruct(String typeName, Object[] attributes) throws SQLException {
return ctx.physicalConn.createStruct(typeName, attributes); }
public void setSchema(String schema) throws SQLException { ctx.physicalConn.setSchema(schema); }
public String getSchema() throws SQLException { return ctx.physicalConn.getSchema(); }
public void abort(Executor executor) throws SQLException { ctx.physicalConn.abort(executor); }
public void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException {
ctx.physicalConn.setNetworkTimeout(executor, milliseconds); }
public int getNetworkTimeout() throws SQLException { return ctx.physicalConn.getNetworkTimeout(); }
}
}
/**
* Prepared statement cache key.
*/
private static class PSKey
{
/** SQL string, including placeholders */
private final String sql;
/** Statement result set type */
private final int resultSetType;
/** Statement result set concurrency */
private final int resultSetConcurrency;
/** Cached hash code */
private final int hashCode;
/**
* Constructor.
*
* @param sql
* SQL string, including placeholders
* @param resultSetType
* Statement result set type
* @param resultSetConcurrency
* Statement result set concurrency
*/
private PSKey(String sql, int resultSetType, int resultSetConcurrency)
{
this.sql = sql;
this.resultSetType = resultSetType;
this.resultSetConcurrency = resultSetConcurrency;
this.hashCode = computeHash();
}
/**
* Compute and return a well distributed hash code.
*
* @return Computed hash code.
*/
private int computeHash()
{
int result = 17;
result = 37 * result + sql.hashCode();
result = 37 * result + resultSetType;
result = 37 * result + resultSetConcurrency;
return result;
}
/**
* Return a hash code for this object.
*
* @return Hash code.
*/
@Override
public int hashCode()
{
return hashCode;
}
/**
* Test this object for equality (equivalence) with the given object.
*
* @param o
* Object to test.
*
* @return {@code true} if this object is the same instance as the given object, or if it is
* considered equivalent to the given object, else {@code false}.
*/
@Override
public boolean equals(Object o)
{
if (this == o)
{
// same instance
return true;
}
if (!(o instanceof PSKey))
{
// impossible to be equivalent
return false;
}
PSKey that = (PSKey) o;
return this.resultSetType == that.resultSetType &&
this.resultSetConcurrency == that.resultSetConcurrency &&
this.sql.equals(that.sql);
}
}
}