# 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");
          }
       }
    }