SQLQuery.java
/*
** Module : SQLQuery.java
** Abstract : Object which performs an SQL query and processes the results.
**
** Copyright (c) 2019-2024, Golden Code Development Corporation.
**
** -#- -I- --Date-- ---------------------------------------Description---------------------------------------
** 001 ECF 20191001 Created initial version with stubbed methods.
** OM 20200202 Added method implementations.
** 002 CA 20200927 Avoid 'log' overhead when computing its string argument (use a lambda for it).
** OM 20201001 Improved DMO manipulation performance by caching slow Property annotation access.
** ECF 20210604 Set configurable JDBC fetch size limit for server-side cursor queries.
** IAS 20210905 Re-working re-writing queries with the values of the mutable SESSION
** attributes
** IAS 20210913 Catch PersistenceException and throw ErrorConditionException in the
** 'list' and 'uniqueResult' methods to handle NO-ERROR option correctly.
** ECF 20211212 Reverted the previous change, as it can result in retrying the enclosing block after a
** fatal database error. This leads in some cases to an infinite loop as the next database
** operation is ignored, throws SQLException, which is converted to ErrorConditionException.
** The cycle repeats indefinitely (at least with PostgreSQL as the back end).
** OM 20211020 Added _DATASOURCE_ROWID property.
** IAS 20211223 Special processing of errors raised by UDFs.
** IAS 20220112 Report SQLWarnings raised by UDFs.
** OM 20220225 Javadoc update.
** OM 20220409 Added support for List/array query parameters.
** OM 20220428 Avoid NPE when running in import mode.
** TJD 20220504 Upgrade do Java 11 minor changes
** OM 20220628 Added parameter permutations storage.
** OM 20220727 FieldId and PropertyId are different for denormalized extent fields.
** AL2 20220828 Added MBean profiling for queries.
** CA 20221006 Added JMX instrumentation for hydrate. Refs #6814
** SBI 20221031 Refactored uniqueResult and hydrateRecordImpl methods to accept the RowStructure parameter
** as a wrapper for two separate parameters, changed hydrateRecordImpl logic to skip fields
** that are not queried.
** TJD 20220504 Upgrade do Java 11 minor changes
** CA 20221109 Implemented runtime for EXCEPT/FIELDS options - only NO-LOCK records are loaded as
** 'partial/incomplete'. Extent fields are always loaded at this time.
** RAA 20221221 Modified the execution of sql's. They now pass through the SQLStatementLogger
** in case logging is intended.
** CA 20230104 If a cached record is incomplete, get the full record from the database when a hydrate is
** is performed.
** IAS 20220913 Re-work processing of UDFs errors/warnings.
** OM 20221103 New class names for FQLPreprocessor, FQLExpression, FQLBundle, and FQLCache.
** OM 20221117 Some properties (datetime-tz) may take a different number of positional parameters when
** used in insert/update or where predicate.
** 003 IAS 20230112 Added fix for H2 bug with bind parameters in recursive CTEs.
** 20230209 Fixed ${TZ} placeholder value
** 004 SR 20230505 Removed readOnly flag from scroll().
** 005 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 006 RAA 20230518 Replaced lambdas with method referencing when using SQLStatementLogger.
** 007 RAA 20230607 SQLs are now executed through SQLExecutor instead of SQLStatementLogger.
** 008 RAA 20230615 Added SQL as a parameter when using SQLExecutor.
** 009 AB 20230821 Modified hydrateRecordImpl() to return null in cases of pk equal to 0.
** AB 20230822 Added a check using ResultSet.wasNull() in hydrateRecordImpl() to ensure that a "0"
** returned by a query is indeed a SQL NULL.
** 010 AL2 20231211 Refactor to use new getters for row structure count and DMO iface.
** 011 RAA 20230724 Replaced FunctionWithException with QueryStatement.
** RAA 20231010 Fixed dependencies.
** 012 AD 20231110 Added getRecordPrimaryKey() method to retrieve primary key of record from ResultSet.
** 013 OM 20231214 The list() method can optionally return a list of plain Java objects instead of attempting
** to hydrate then as Records.
** 014 IAS 20230209 Fixed ${TZ} placeholder value
** IAS 20230208 Dialect-specific error handler support.
** 015 HC 20240222 Enabled JMX on FWD Client.
** 016 CA 20240320 Avoid BDTs within internal FWD runtime. Replaced iterators with Java 'for', where it
** applies.
** 017 TJD 20240123 Java 17 compatibility updates
** 018 SP 20240731 Skip using HYDRATE_RECORD TimeStat when JMX_DEBUG flag is not set.
** CA 20240409 Inlined the logging, to avoid capturing lambdas.
** 019 TJD 20221203 SQLException caused by InterruptedException should be converted to StopCondition
*/
/*
** 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.lang.reflect.Array;
import java.sql.*;
import java.util.*;
import java.util.concurrent.atomic.*;
import java.util.logging.*;
import com.goldencode.p2j.directory.*;
import com.goldencode.p2j.jmx.*;
import com.goldencode.p2j.jmx.QueryProfilerJMX;
import com.goldencode.p2j.persist.*;
import com.goldencode.p2j.persist.FQLPreprocessor.*;
import com.goldencode.p2j.persist.Record;
import com.goldencode.p2j.persist.dialect.*;
import com.goldencode.p2j.persist.orm.SQLExecutor.QueryStatement;
import com.goldencode.p2j.persist.orm.SQLExecutor.UpdateStatement;
import com.goldencode.p2j.persist.orm.types.*;
import com.goldencode.p2j.util.*;
import com.goldencode.p2j.util.ErrorManager;
import com.goldencode.p2j.util.ErrorManager.*;
import com.goldencode.p2j.util.logging.*;
/**
* A SQL query that can be cached and executed multiple times.
* Note: internally, the object is NOT synchronized, so the usage intervals MUST NOT overlap.
*/
public class SQLQuery
{
/** Logger */
private static final CentralLogger LOG = CentralLogger.get(SQLQuery.class);
/** Flag indicating fine logging. */
private static final boolean fine = LOG.isLoggable(Level.FINE);
/** Instrumentation for {@link #hydrateRecord}. */
private static final NanoTimer HYDRATE_RECORD = NanoTimer.getInstance(FwdServerJMX.TimeStat.OrmHydrateRecord);
/** Directory path for fetch size override */
private static final String CFG_FETCH_SIZE = "persistence/jdbc-fetch-size";
/** Default JDBC fetch size (may be overridden by runtime configuration) */
private static final int DEFAULT_FETCH_SIZE = 256;
/** JDBC fetch size (same for all databases) */
private static final int fetchSize;
static
{
Directory dir = DirectoryManager.getInstance();
fetchSize = dir.getInt(Directory.ID_RELATIVE_SERVER, CFG_FETCH_SIZE, DEFAULT_FETCH_SIZE);
LOG.fine("JDBC fetch size initialized to: " + fetchSize);
}
/** Profiler done for queries: cache hits, cache misses, etc. */
private static final QueryProfilerJMX SIMPLE_QUERY_PROFILER =
QueryProfilerJMX.getInstance(FwdServerJMX.QueryProfiler.QueryProfiler);
/** The SQL statement, possibly with placeholders. */
private final String baseSql;
/** The SQL statement to be executed. */
private String sql;
/** The parameter permutations. */
private int[] parPerm = null;
/** Mutable session attributes' values used in the query. */
private final List<SessionAttr> sessionAttrs;
/** Flag indicating that the query contains error handling */
private boolean containsErrorHandling = false;
/** The {@code PreparedStatement}. */
private PreparedStatement stmt;
/** The current list of positional parameters. */
private Object[] parameters = null;
/** The higher index of the parameter set. */
private int maxParamIndex = 0;
/** Flag indicating that the query can raise SQLWarning. */
private boolean canRaiseSqlWarning = false;
/** Flags that query bind parameters should be mapped to session variables */
private boolean bindParam2SessionVar = false;
/**
* Constructs a new object.
*
* @param sql
* The statement to be executed.
* @param sessionAttrs
* Mutable session attributes' values used in the query.
* @param parPerm
* The permutation function, if any available.
*/
public SQLQuery(String sql, List<SessionAttr> sessionAttrs, int[] parPerm)
{
this.baseSql = sql;
this.sql = sql;
this.parPerm = parPerm;
this.sessionAttrs = sessionAttrs;
}
/**
* Constructs a new object.
*
* @param sql
* The statement to be executed.
*/
public SQLQuery(String sql)
{
this(sql, Collections.emptyList(), null);
}
/**
* Set the value of the flag indicating that the query contains error handling.
* @param value
* The new value of the flag;
* @return <code>this</code>
*/
public SQLQuery containsErrorHandling(boolean value)
{
this.containsErrorHandling = value;
return this;
}
/**
* Checks if the query can raise {@code SQLWarning}.
*
* @return {@code true} if the query can raise {@code SQLWarning}.
*/
public boolean canRaiseSqlWarning()
{
return canRaiseSqlWarning;
}
/**
* Set the flag indicating if the query can raise {@code SQLWarning}.
*
* @param canRaiseSqlWarning
* The value of the flag
*
* @return {@code this}
*/
public SQLQuery canRaiseSqlWarning(boolean canRaiseSqlWarning)
{
this.canRaiseSqlWarning = canRaiseSqlWarning;
return this;
}
/**
* Check if the query bind parameters should be mapped to session variables.
*
* @return <code>true</code> if the query parameters should be mapped to session variabales.
*/
public boolean bindParam2SessionVar()
{
return bindParam2SessionVar;
}
/**
* Set the flag indicating if the query can bind parameters should be mapped to session variables.
*
* @param bindParam2SessionVar
* The value of the flag
*
* @return {@code this}
*/
public SQLQuery bindParam2SessionVar(boolean bindParam2SessionVar)
{
this.bindParam2SessionVar = bindParam2SessionVar;
return this;
}
/**
* Expand placeholder.
* @param dialect
* Database dialect
* @return Current SQLQuery instance;
*/
public SQLQuery expandPlaceholders(Dialect dialect)
{
if (sessionAttrs != null && !sessionAttrs.isEmpty())
{
StringBuilder sb = new StringBuilder();
int p = 0;
for (SessionAttr attr: sessionAttrs)
{
String ph = attr.placeHolder;
int pp = baseSql.indexOf(ph, p);
if (pp < 0)
{
throw new IllegalStateException(String.format(
"Cannot find placeholder [%s] in [%s]", ph, baseSql));
}
sb.append(baseSql, p, pp);
sb.append(attr.value.apply(dialect));
p = pp + ph.length();
}
sql = sb.append(baseSql.substring(p)).toString();
}
return this;
}
/**
* Sets the value of a parameter for this query. The parameters are accumulated and the index is preserved
* so a parameter can be overwrite at any moment. Once the query is executed the parameters are set, for
* the eventual missing ones {@code null} value will be used. At that time, the index of parameters will be
* shifted to 1-based JDBC standard.
*
* @param index
* The index of the parameter to be set (0-base).
* @param arg
* The parameter value.
*/
public void setParameter(int index, Object arg)
{
ensureSize(index);
parameters[index] = arg;
}
/**
* Executes the query against the SQL server.
*
* @param <T>
* The type of the rows returned in the list.
*
* @param session
* The {@code Session} to be used.
* @param scrollMode
* The scroll mode.
* @param rowStructure
* The expected structure of a query, as generated by the {@link FqlToSqlConverter}.
*
* @return A scrollable list of rows of type {@code <T>} from the database server, as specified by the
* {@code sql} query.
*
* @throws PersistenceException
* when an error occurred while performing the requested operations.
*/
public <T> ScrollableResults<T> scroll(Session session,
int scrollMode,
ArrayList<RowStructure> rowStructure)
throws PersistenceException
{
Connection conn = session.getConnection();
checkClosed(conn);
try
{
activateErrorHandler(session);
stmt = conn.prepareStatement(sql, scrollMode, ResultSet.CONCUR_READ_ONLY);
if (bindParam2SessionVar)
{
setSessionVars(conn);
}
else
{
setParameters(stmt, false);
}
// hint to the JDBC driver that we want a server-side cursor and a non-default fetch size
stmt.setFetchSize(fetchSize);
if (fine)
{
LOG.log(Level.FINE, "FWD ORM: " + stmt);
}
QueryStatement f = PreparedStatement::executeQuery;
ResultSet resultSet = SQLExecutor.getInstance().execute(session.getDatabase(), sql, f, stmt);
reportSQLWarnings(session);
return new ScrollableResults<T>(stmt, resultSet, rowStructure, session).
onClose(() -> {resetErrorHandler(session); return null;});
}
catch (SQLException exc)
{
close();
Dialect dialect = Query.getDialect(session);
if (dialect.isUdfOriginatedError(exc))
{
throw new UdfException(dialect.errorEntry(exc.getMessage()), exc);
}
throw new PersistenceException("Error scrolling", exc);
}
// finally
// {
// resetErrorHandler(session);
// }
}
/**
* Returns a scrollable list of rows of type {@code <T>} by executing the query against the database
* server.
*
* @param <T>
* The type of the rows returned in the list.
*
* @param session
* The {@code Session} to be used.
* @param rowStructure
* The expected structure of a query, as generated by the {@link FqlToSqlConverter}.
*
* @return A scrollable list of rows of type {@code <T>} from SQL server, as specified by the
* {@code sql} query.
*
* @throws PersistenceException
* when an error occurred while performing the requested operations.
*/
public <T> ScrollableResults<T> scroll(Session session, ArrayList<RowStructure> rowStructure)
throws PersistenceException
{
Connection conn = session.getConnection();
checkClosed(conn);
try
{
activateErrorHandler(session);
stmt = conn.prepareStatement(sql);
if (bindParam2SessionVar)
{
setSessionVars(conn);
}
else
{
setParameters(stmt, false);
}
if (fine)
{
LOG.log(Level.FINE, "FWD ORM: " + stmt);
}
QueryStatement f = PreparedStatement::executeQuery;
ResultSet resultSet = SQLExecutor.getInstance().execute(session.getDatabase(), sql, f, stmt);
return new ScrollableResults<T>(stmt, resultSet, rowStructure, session).
onClose(() -> {resetErrorHandler(session); return null;});
}
catch (SQLException exc)
{
close();
throw new PersistenceException("Error scrolling", exc);
}
// finally
// {
// resetErrorHandler(session);
// }
}
/**
* Returns an unique row of type {@code <T>} by executing the query against the SQL server. The result may
* be either a scalar value, but also can be a full {@code Record} object.
*
* @param <T>
* The type of the row to be returned.
*
* @param session
* The {@code Session} to be used.
*
* @return An object of type {@code <T>} from SQL server, as specified by the {@code sql} query.
*
* @throws PersistenceException
* when an error occurred while performing the requested operations.
*/
public <T> T uniqueResult(Session session)
throws PersistenceException
{
return uniqueResult(session, null);
}
/**
* Returns an unique row of type {@code <T>} by executing the query against the SQL server. The result may
* be either a scalar value, but also can be a full {@code Record} object.
*
* @param <T>
* The type of the row to be returned.
*
* @param session
* The {@code Session} to be used.
* @param rowStructure
* The holder for dmo type and row size of the query. The row size of the query, as generated by
* the {@link FqlToSqlConverter} is usually 1 if a single/plain result is expected, but it can be larger
* if a full {@code Record} object is expected to be returned. The dmo {@code Class} of the expected object
* is used hydration of row.
*
* @return An object of type {@code <T>} from SQL server, as specified by the {@code sql} query.
*
* @throws PersistenceException
* when an error occurred while performing the requested operations.
*/
public <T> T uniqueResult(Session session, RowStructure rowStructure)
throws PersistenceException
{
Connection conn = session.getConnection();
checkClosed(conn);
try
{
activateErrorHandler(session);
stmt = conn.prepareStatement(sql);
if (bindParam2SessionVar)
{
setSessionVars(conn);
}
else
{
setParameters(stmt, false);
}
if (fine)
{
LOG.log(Level.FINE, "FWD ORM: " + stmt);
}
QueryStatement f = PreparedStatement::executeQuery;
ResultSet resultSet = SQLExecutor.getInstance().execute(session.getDatabase(), sql, f, stmt);
reportSQLWarnings(session);
if (!resultSet.next()) // same effect as first()
{
return null; // no record matching predicate
}
T ret = null;
if (rowStructure == null || rowStructure.getCount() == 1)
{
// only the PK or other scalar value is returned by the query
ret = (T) resultSet.getObject(1);
}
else
{
// when multiple values are returned in a row then a fully hydrated Record is expected
ret = (T) hydrateRecord(resultSet, rowStructure, 1, session);
}
if (resultSet.next())
{
close(); // ???
throw new UniqueResultException("Result not unique for " + stmt);
}
SIMPLE_QUERY_PROFILER.updateRowsCount(resultSet, 1);
return ret;
}
catch (SQLException exc)
{
close();
Dialect dialect = Query.getDialect(session);
if (dialect != null && dialect.isUdfOriginatedError(exc))
{
throw new UdfException(dialect.errorEntry(exc.getMessage()), exc);
}
throw new PersistenceException("Error uniqueResult", exc);
}
finally
{
close();
resetErrorHandler(session);
}
}
/**
* Execute the update SQL query. When finished with success returns the number of affected columns.
*
* @param session
* The {@code Session} whose database {@code Connection} to be used.
*
* @return The number of affected rows by the statement.
*
* @throws PersistenceException
* when an error occurred while performing the requested operations.
*/
public int executeUpdate(Session session)
throws PersistenceException
{
Connection conn = session.getConnection();
checkClosed(conn);
try
{
activateErrorHandler(session);
stmt = conn.prepareStatement(sql);
setParameters(stmt, true);
if (fine)
{
LOG.log(Level.FINE, "FWD ORM: " + stmt);
}
UpdateStatement f = PreparedStatement::executeUpdate;
return SQLExecutor.getInstance().execute(session.getDatabase(), sql, f, stmt);
}
catch (SQLException exc)
{
close();
throw new PersistenceException("Error executeUpdate: " + exc.getMessage(), exc);
}
finally
{
close();
resetErrorHandler(session);
}
}
/**
* Returns a list of rows of type {@code <T>} by executing the query against the SQL server. The method
* will always attempt to return a list containing a set of hydrated {@link Record} objects, if possible.
* In the case of projections, the list will contain only the PKs (recid/id) of the records, being the
* responsibility of teh caller to hydrate with subsequent queries only the desired records (aka fetch the
* full {@code Record} using the {@code Loader} of the {@code Persistence}).
*
* @param <T>
* The type of the rows returned in the list.
* @param session
* The {@code Session} to be used.
* @param rowStructures
* The expected structure of a query, as generated by the {@link FqlToSqlConverter}.
*
* @return A list of rows of type {@code <T>} from SQL server, as specified by the {@code sql} query.
*
* @throws PersistenceException
* when an error occurred while performing the requested operations.
*/
public <T> List<T> list(Session session, ArrayList<RowStructure> rowStructures)
throws PersistenceException
{
return list(session, rowStructures, true);
}
/**
* Returns a list of rows of type {@code <T>} by executing the query against the SQL server. Depending on
* the value of teh last parameter, the returned list is composed either from {@code Record} / PKs
* (recid/id) as described in the overloaded method above (when {@code hydrateRecords} is {@code true}), or
* plain Java wrappers (driver specific, which includes: {@code Integer}, {@code String},
* {@code BigDecimal}, etc.), otherwise.
*
* @param <T>
* The type of the rows returned in the list.
* @param session
* The {@code Session} to be used.
* @param rowStructures
* The expected structure of a query, as generated by the {@link FqlToSqlConverter}.
* @param hydrateRecords
* If {@code true} the result is interpreted as legacy records and will be hydrated.
* Otherwise, the returned {@code List} contains an array for each row.
*
* @return A list of rows of type {@code <T>} from SQL server, as specified by the {@code sql} query.
*
* @throws PersistenceException
* when an error occurred while performing the requested operations.
*/
public <T> List<T> list(Session session, ArrayList<RowStructure> rowStructures, boolean hydrateRecords)
throws PersistenceException
{
Connection conn = session.getConnection();
checkClosed(conn);
if (rowStructures == null || rowStructures.isEmpty())
{
throw new IllegalArgumentException("The row structure is null");
}
try
{
activateErrorHandler(session);
int rowSize = rowStructures.size();
List<T> ret = new ArrayList<>();
stmt = conn.prepareStatement(sql);
if (bindParam2SessionVar)
{
setSessionVars(conn);
}
else
{
setParameters(stmt, false);
}
if (fine)
{
LOG.log(Level.FINE, "FWD ORM: " + stmt);
}
QueryStatement f = PreparedStatement::executeQuery;
ResultSet resultSet = SQLExecutor.getInstance().execute(session.getDatabase(), sql, f, stmt);
reportSQLWarnings(session);
int columnCount = resultSet.getMetaData().getColumnCount();
if (!hydrateRecords)
{
while (resultSet.next())
{
if (columnCount == 1)
{
ret.add((T) resultSet.getObject(1));
}
else
{
Object[] row = new Object[columnCount];
for (int i = 0; i < columnCount; i++)
{
row[i] = resultSet.getObject(i + 1);
}
ret.add((T) row);
}
}
return ret;
}
if (columnCount == rowSize)
{
// if the [columnCount] matches the [rowSize] then we have only one column for each DMO to be
// returned in a row. Since for each DMO all queries return at least the PK we deduce that the
// result set of this query is compose only from PKs (recid/id). The caller is responsible for
// handling them (including hydration, if needed).
while (resultSet.next())
{
Object[] tuple = new Object[rowSize];
for (int k = 0; k < rowSize; k++)
{
tuple[k] = resultSet.getObject(k + 1);
}
ret.add((T) tuple);
}
}
else if (rowSize == 1)
{
// when a single DMO is queried and the number of columns is greater than 1 then the full record
// is assumed. Hydrate each record from the current row and return the full list of Records
RowStructure rowStruct = rowStructures.get(0);
while (resultSet.next())
{
ret.add((T) hydrateRecord(resultSet, rowStruct, 1, session));
}
}
else
{
// last case: the returned row is composed of columns from multiple DMOs. We split the row in
// components that match the [rowStructures] map and hydrate all of the Records. This array is
// added as an entry in the list of object returned by this method
while (resultSet.next())
{
Object[] row = new Object[rowSize];
int k = 0;
int resOffset = 1;
for (int i = 0; i < rowStructures.size(); i++)
{
RowStructure rowStruct = rowStructures.get(i);
// populate row[k]'s fields
row[k++] = hydrateRecord(resultSet, rowStruct, resOffset, session);
resOffset += rowStruct.getCount();
}
ret.add((T) row);
}
}
SIMPLE_QUERY_PROFILER.updateRowsCount(resultSet, ret.size());
return ret;
}
catch (SQLException exc)
{
close();
Dialect dialect = Query.getDialect(session);
if (dialect.isUdfOriginatedError(exc))
{
throw new UdfException(dialect.errorEntry(exc.getMessage()), exc);
}
throw new PersistenceException("Error while processing the SQL list", exc);
}
finally
{
resetErrorHandler(session);
close();
}
}
/**
* Report SQLWarnings raised by UDFs.
* @param session
* database session.
* @throws SQLException
* on error in the database layer.
*/
private void reportSQLWarnings(Session session) throws SQLException
{
if (!canRaiseSqlWarning)
{
return;
}
Dialect dialect = Query.getDialect(session);
SQLWarning warning = stmt.getWarnings();
while (warning != null)
{
if (dialect.isUdfOriginatedError(warning))
{
ErrorEntry ee = dialect.errorEntry(warning.getMessage());
ErrorManager.recordOrShowError(
new int[] {ee.num},
new String[] {ee.text},
true,
ee.prefix,
false,
false,
false,
true);
}
warning = warning.getNextWarning();
}
}
/**
* Hydrate a {@code Record} from a {@code ResultSet}, ie a new object that implements the DMO
* interface and having the properties set from the current row of the record set. The method
* takes care of all the DMO properties, record identifier, and including the special
* properties of the temp-tables (_multiplex, before table flags).
*
* @param resultSet
* The data source. Contain the result of a query with the "main" properties. The
* extent properties are queried and hydrated before the record object is returned.
* @param rowStructure
* The holder for the DMO of the object to be returned and the expected number of properties of this object,
* as the select query was constructed. This governs the order and the types of the fields and
* the number of columns used for each one.
* @param rsOffset
* The offset in the {@link ResultSet}'s row where the hydration process begins (in
* case when multiple objects are queried at once).
* @param session
* The {@code Session} to be used.
*
* @return The re-hydrated object.
*
* @throws PersistenceException
* When an issue was encountered in the process.
*/
static BaseRecord hydrateRecord(ResultSet resultSet,
RowStructure rowStructure,
int rsOffset,
Session session)
throws PersistenceException
{
BaseRecord res;
if (FwdServerJMX.JMX_DEBUG)
{
res = HYDRATE_RECORD.timerWithReturn(
() -> hydrateRecordImpl(resultSet, rowStructure, rsOffset, session)
);
}
else
{
res = hydrateRecordImpl(resultSet, rowStructure, rsOffset, session);
}
return res;
}
/**
* Get the primary key of a {@link Record} from a {@link ResultSet}.
*
* @param resultSet
* The data source. Contain the result of a query with the "main" properties.
* @param rowStructure
* The holder for the DMO of the object to be returned and the expected number of properties of this object,
* as the select query was constructed. This governs the order and the types of the fields and
* the number of columns used for each one.
* @param rsOffset
* The offset in the {@link ResultSet}'s row where the primary key should begin (in
* case when multiple objects are queried at once).
* @return The objects primary key.
*
* @throws PersistenceException
* When an issue was encountered in the process.
*/
static Long getRecordPrimaryKey(ResultSet resultSet,
RowStructure rowStructure,
int rsOffset)
throws PersistenceException
{
try
{
Long pk;
pk = resultSet.getLong(rsOffset);
// This can be the case for an outer join. We can't assume that a pk = 0
// represents a SQL NULL without checking that the resultSet was null.
if (pk == 0 && resultSet.wasNull())
{
return null;
}
return pk;
}
catch (SQLException sql)
{
// just log it, but continue to let the generic hydration detect other issues
LOG.log(Level.WARNING, "Failed to locate the primary key of " + rowStructure.getDmoClass().getSimpleName() + ".");
return null;
}
}
/**
* Hydrate a {@code Record} from a {@code ResultSet}, ie a new object that implements the DMO
* interface and having the properties set from the current row of the record set. The method
* takes care of all the DMO properties, record identifier, and including the special
* properties of the temp-tables (_multiplex, before table flags).
*
* @param resultSet
* The data source. Contain the result of a query with the "main" properties. The
* extent properties are queried and hydrated before the record object is returned.
* @param rowStructure
* The holder for the DMO of the object to be returned and the expected number of properties of this object,
* as the select query was constructed. This governs the order and the types of the fields and
* the number of columns used for each one.
* @param rsOffset
* The offset in the {@code ResultSet}'s row where the hydration process begins (in
* case when multiple objects are queried at once).
* @param session
* The {@code Session} to be used.
*
* @return The re-hydrated object.
*
* @throws PersistenceException
* When an issue was encountered in the process.
*/
private static BaseRecord hydrateRecordImpl(ResultSet resultSet,
RowStructure rowStructure,
int rsOffset,
Session session)
throws PersistenceException
{
Connection conn = session.getConnection();
checkClosed(conn);
DmoMeta dmoInfo = rowStructure.getDmoMeta();
Class<? extends Record> recordClass = dmoInfo.getImplementationClass();
int count = rowStructure.getCount();
// Note: we should check whether
// Session.PK.equalsIgnoreCase(resultSet.getMetaData().getColumnName(rsOffset))
// but to avoid the possible extra load of processing the metadata we assume the
// primary key RECID/ID is the FIRST COLUMN in the result set of the query
long pk;
try
{
pk = resultSet.getLong(rsOffset);
// This can be the case for an outer join. We can't assume that a pk = 0
// represents a SQL NULL without checking that the resultSet was null.
if (pk == 0 && resultSet.wasNull())
{
return null;
}
Record cachedRec = session.getCached(recordClass, pk);
if (cachedRec != null && !cachedRec.checkState(DmoState.STALE))
{
// if the record is incomplete, we need to refresh it from the database
if (cachedRec.checkState(DmoState.INCOMPLETE))
{
// this will update the cachedRec reference, so the returned record instance does not change,
// only its data is updated
cachedRec = session.get(recordClass, pk);
}
if (cachedRec != null)
{
SIMPLE_QUERY_PROFILER.updateCacheHits(resultSet, 1);
// we found an unSTALE CACHED record, do not bother reading from result set
return cachedRec;
}
}
if (count == 1)
{
SIMPLE_QUERY_PROFILER.updateCacheHits(resultSet, 1);
cachedRec = session.get(recordClass, pk);
return cachedRec;
}
}
catch (SQLException sql)
{
// just log it, but continue to let the generic hydration detect other issues
LOG.log(Level.WARNING,
"Failed to locate the primary key of " + recordClass.getSimpleName() + ".",
sql);
}
BaseRecord r = null;
try
{
SIMPLE_QUERY_PROFILER.updateCacheMisses(resultSet, 1);
r = recordClass.getDeclaredConstructor().newInstance();
if (rowStructure.isIncomplete())
{
// this should be done before hydrating; otherwise the reading won't mark
// the read properties properly
r.markIncomplete(session);
}
rsOffset = rowStructure.hydrate(session, resultSet, rsOffset, r);
// cache the hydrated record
session.associate(r);
}
catch (SQLException |
ReflectiveOperationException e)
{
// extract Interrupted Exception out of SQLException
if (e instanceof SQLException && e.getCause() instanceof InterruptedException)
{
throw new StopConditionException("Interrupted while populating record for class " + rowStructure.getDmoClass().getName(), e);
}
else
{
throw new PersistenceException("Failed to populate record for class " + rowStructure.getDmoClass().getName(), e);
}
}
return r;
}
/**
* Closes this object by releasing all acquired resources.
*/
private void close()
{
if (stmt != null)
{
try
{
stmt.close();
stmt = null;
}
catch (SQLException e)
{
LOG.severe("", e);
}
}
}
/**
* Makes sure the array that holds the parameters for this query is large enough to accommodate
* the {@code fitIndex}-th parameter. If the index fits the current size nothing is performed.
* Otherwise, the array is reallocated and the existing parameters are copied to their fixed
* positions. To improve performance, more space may be allocated so that this operation will
* not happen at each method call.
*
* @param fitIndex
* The index of the latest added parameter.
*/
private void ensureSize(int fitIndex)
{
if (maxParamIndex < fitIndex)
{
maxParamIndex = fitIndex;
}
if (parameters == null)
{
parameters = new Object[Math.max(5, fitIndex + 1)]; // TODO: optimize this
return;
}
if (parameters.length > fitIndex)
{
// nothing to do
return;
}
// create a new array and copy from previously set [parameters] to new location
Object[] newParam = new Object[1 + fitIndex * 3 / 2]; // TODO: optimize this
System.arraycopy(parameters, 0, newParam, 0, parameters.length);
this.parameters = newParam;
}
/**
* Copy the set of parameters temporarily stored in this object to the {@code PreparedStatement}.
*
* @param stmt
* The statement to be processed.
* @param forUpdate
* {@code true} if the parameter in the statement is used to update a field and {@code false} if
* it is used as part of the where predicate.
*
* @throws SQLException
* If an error occurred in the process.
*/
private void setParameters(PreparedStatement stmt, boolean forUpdate)
throws SQLException
{
if (parameters == null)
{
return; // no parameter have been set
}
AtomicInteger pos = new AtomicInteger(0);
for (int i = 0; i <= maxParamIndex; i++)
{
int pIndex = (parPerm != null && i < parPerm.length) ? parPerm[i] : i;
setParameter(stmt, pos, parameters[pIndex], forUpdate);
}
}
/**
* Copy the set of parameters temporarily stored in this object to the session variables.
*
* @param conn
* The database connection.
*
* @throws SQLException
* If an error occurred in the process.
*/
private void setSessionVars(Connection conn)
throws SQLException
{
if (parameters == null)
{
return; // no parameter have been set
}
AtomicInteger pos = new AtomicInteger(0);
for (int i = 0; i <= maxParamIndex; i++)
{
int pIndex = (parPerm != null && i < parPerm.length) ? parPerm[i] : i;
setSessionVar(conn, pos, parameters[pIndex]);
}
}
/**
* Activate error handler at the database side.
* @param session
* Database session
* @throws SQLException
* on error.
* @throws PersistenceException
* on error.
*/
private void activateErrorHandler(Session session)
throws SQLException, PersistenceException
{
if (!containsErrorHandling)
{
return;
}
Query.getDialect(session).activateErrorHandler(session.getConnection());
}
/**
* Reset error handler at the database side.
* @param session
* Database session
* @throws SQLException
* on error.
* @throws PersistenceException
* on error.
*/
public void resetErrorHandler(Session session)
throws PersistenceException
{
if (!containsErrorHandling)
{
return;
}
try
{
Query.getDialect(session).activateErrorHandler(session.getConnection());
}
catch (SQLException e)
{
throw new PersistenceException("Failed to reset error handler", e);
}
}
/**
* Sets the value for a parameter at a specified position in a statement. The process is done in recursive
* manner for arrays and {@code List}s. The {@code pos} is incremented automatically for each scalar
* parameter configured.
*
* @param stmt
* The statement.
* @param pos
* The start position.
* @param val
* The value
* @param forUpdate
* {@code true} if the parameter in the statement is used to update a field and {@code false} if
* it is used as part of the where predicate.
*
* @throws SQLException
* If an error occurred in the process.
*/
private void setParameter(PreparedStatement stmt, AtomicInteger pos, Object val, boolean forUpdate)
throws SQLException
{
ParameterSetter setter = null;
// validate first: log [null] parameters and their positions
if (val == null)
{
if (fine)
{
LOG.log(Level.FINE, "null value for " + pos + "th parameter.");
}
}
else
{
if (val instanceof BaseDataType && ((BaseDataType) val).isUnknown())
{
if (fine)
{
LOG.log(Level.FINE, "unknown value for " + pos + "th parameter.");
}
}
else if (val.getClass().isArray())
{
int len = Array.getLength(val);
for (int k = 0; k < len; k++)
{
// recursively add all parameters from array
setParameter(stmt, pos, Array.get(val, k), forUpdate);
}
return;
}
else if (val instanceof List)
{
for (Object pp : ((List<?>) val))
{
// recursively add all parameters from list
setParameter(stmt, pos, pp, forUpdate);
}
return;
}
setter = TypeManager.getTypeHandler(val.getClass());
}
int i = pos.incrementAndGet(); // 1-base indexes of parameters
if (setter != null)
{
setter.setParameter(stmt, i, val, forUpdate);
}
else
{
stmt.setObject(i, null);
}
}
/**
* Sets the value for a parameter as a session variable. The process is done in recursive
* manner for arrays and {@code List}s. The {@code pos} is incremented automatically for each scalar
* parameter configured.
*
* @param conn
* Database connection.
* @param pos
* The start position.
* @param val
* The value
*
* @throws SQLException
* If an error occurred in the process.
*/
private void setSessionVar(Connection conn, AtomicInteger pos, Object val)
throws SQLException
{
ParameterSetter setter = null;
// validate first: log [null] parameters and their positions
if (val == null)
{
if (fine)
{
LOG.log(Level.FINE, "null value for " + pos + "th parameter.");
}
}
else
{
if (val instanceof BaseDataType && ((BaseDataType) val).isUnknown())
{
if (fine)
{
LOG.log(Level.FINE, "Unknown value for " + pos + "th parameter.");
}
}
else if (val.getClass().isArray())
{
int len = Array.getLength(val);
for (int k = 0; k < len; k++)
{
// recursively add all parameters from array
setSessionVar(conn, pos, Array.get(val, k));
}
return;
}
else if (val instanceof List)
{
for (Object pp : ((List<?>) val))
{
// recursively add all parameters from list
setSessionVar(conn, pos, pp);
}
return;
}
setter = TypeManager.getTypeHandler(val.getClass());
if (setter == null)
{
throw new SQLException("Unsupported data type " + val.getClass().getName() +
" for " + pos + "th parameter.");
}
}
int i = pos.incrementAndGet(); // 1-base indexes of parameters
PreparedStatement pstmt = conn.prepareStatement("set @_v" + i + " = ?");
if (setter != null) // val != null and has a supported data type
{
setter.setParameter(pstmt, 1, val, false);
}
else
{
pstmt.setObject(1, null);
}
pstmt.executeUpdate();
}
/**
* Test whether the connection is alive (not closed). If the connection is usable the method
* returns without any side-effects. If the session is not usable, {@code PersistenceException}
* is thrown.
*
* @param conn
* The database {@code Connection} to be used.
*
* @throws PersistenceException
* when the connection is not usable because it was closed.
*/
private static void checkClosed(Connection conn)
throws PersistenceException
{
try
{
if (conn == null || conn.isClosed())
{
throw new PersistenceException("Cannot use a closed database session");
}
}
catch (SQLException exc)
{
throw new PersistenceException("Error checking database connection status");
}
}
}