# Session.java

    /*
    ** Module   : Session.java
    ** Abstract : Manages a JDBC session with a database
    **
    ** Copyright (c) 2019-2025, Golden Code Development Corporation.
    **
    ** -#- -I- --Date-- ---------------------------------------Description---------------------------------------
    ** 001 ECF 20191001 Created initial version. Manages a JDBC session with a database.
    **     OM  20200202 Added method implementation. Added record caching support based on their
    **                  PropertyState attributes.
    ** 002 ECF 20200910 Verify a record is not stale before saving or caching it.
    **     OM  20200924 Enforced no-stale-in-cache rule.
    **     OM  20201002 Use DmoMeta cached information instead of map lookups.
    **     OM  20201202 Exposed API for removing cached records which were affected by explicit SQL statements
    **                  (these changes invalidate data in cache).
    **     IAS 20201123 createQuery with delayed caching.
    **     ECF 20210519 Reimplemented session cache to use a (lenient) fixed size LRU cache instead of an
    **                  unlimited size hash map.
    **     OM  20210618 Implemented disableCache() method.
    **     OM  20210825 Small changes in error messages.
    **     OM  20210831 Expiring CHANGED records are saved before dropping them.
    **     IAS 20210905 Re-working re-writing queries with the values of the mutable SESSION attributes.
    **     ECF 20210924 Removed unused delete method variant; ensure record state is updated during delete;
    **                  removed RecordBuffer dependency from invalidateRecords.
    **     ECF 20211022 A cache expiration is not tracked as an undoable event.
    **     ECF 20211230 Do not attempt to delete a newly created record that was never flushed to the database.
    **     OM  20220628 Signature change for createSQLQuery() method.
    **     CA  20221109 Implemented runtime for EXCEPT/FIELDS options - only NO-LOCK records are loaded as
    **                  'partial/incomplete'.  Partial/Incomplete records are being cached, but when they try to
    **                  be loaded with no 'partial field information', the cached record is fully refreshed from
    **                  the database, and not just returned.
    **     CA  20230104 If a cached record is incomplete, preserve the dirty properties when loading the full
    **                  record.
    **     OM  20221103 New class names for FQLPreprocessor, FQLExpression, FQLBundle, and FQLCache.
    **     CA  20221109 Implemented runtime for EXCEPT/FIELDS options - only NO-LOCK records are loaded as
    **                  'partial/incomplete'.  Partial/Incomplete records are being cached, but when they try to
    **                  be loaded with no 'partial field information', the cached record is fully refreshed from
    **                  the database, and not just returned.
    **     CA  20230104 If a cached record is incomplete, preserve the dirty properties when loading the full
    **                  record.
    ** 003 RAA 20230208 Removed everything that had to do with redo operations as it is no longer needed.
    ** 004 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
    ** 005 SR  20230519 Changed cacheExpired() so now the expired dmo is also deregistered from DmoVersioning.
    **     SR  20230522 Moved the deregister operation above the state check so it will happen regardless of
    **                  how we exit the method.
    **     AL2 20230525 Added cache extension to support the cache pruning faster.
    ** 006 DDF 20230608 Made the size of the session cache configurable.
    **     DDF 20230613 Use CacheManager to initialize the cache.
    **     DDF 20230620 Replaced static initialization of values from the directory configuration with
    **                  a method called at server bootstrap.
    **     DDF 20230627 Used a standard handling for directory binding.
    ** 007 CA  20231114 In 'getImpl', when refreshing an incomplete record, do not evict and re-cache this DMO.
    ** 008 RAA 20230802 Added support for VALIDATE keyword.
    **     RAA 20230802 Changed name of validateMode to allowDBUniqueCheck.
    ** 009 OM  20240215 A record can be evicted by its DMO class and PK.
    ** 010 CA  20240422 For a NEW temp record which was not flushed and is being deleted or rolled back in the
    **                  transaction, the allocated primary key row (which is an empty template) to reserve this
    **                  primary key, needs to be explicitly removed.
    ** 011 CA  20240501 A NEW and INVALID record has already been removed from the temp-table, so no need to
    **                  clear its primary key.
    ** 012 TJD 20240123 Java 17 compatibility updates
    ** 013 CA  20240701 Always 'clearPrimaryKey' in 'delete', even if the DMO is invalid - this doesn't mean the
    **                  reserved PK has already been deleted.
    ** 014 RAA 20240423 Modified constructors to also include the database configuration and the tenant id.
    **     RAA 20240613 Added constructor that takes both the database configuration and the tenant id
    **                  at the same time.
    **     RAA 20240627 Removed unused constructors.
    ** 015 CA  20240723 Added 'updateCachedRecords', to apply some code to cached temp-table records.
    ** 016 OM  20241128 Multi-tenant runtime support: selected the proper persistence context, eventually
    **                  based on [sharedDb] parameter.
    ** 017 LS  20250211 Added support for executing a hook when a session is established/closed.
    ** 018 OM  20250308 Notified SecurityOps when _user meta table changes.
    ** 019 AP  20250417 Call the onSessionClaim in all necessary constructors. Improved overall logic for this.
    **         20250519 Reverted the logic for session hook in Session(Database, DataSource).
    */

    /*
    ** 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.*;
    import java.sql.*;
    import java.util.*;
    import java.util.function.*;
    import javax.sql.DataSource;
    import com.goldencode.asm.*;
    import com.goldencode.cache.*;
    import com.goldencode.p2j.cfg.*;
    import com.goldencode.p2j.directory.*;
    import com.goldencode.p2j.persist.*;
    import com.goldencode.p2j.persist.FQLPreprocessor.*;
    import com.goldencode.p2j.persist.Record;
    import com.goldencode.p2j.persist.meta.*;
    import com.goldencode.p2j.util.*;
    import com.goldencode.p2j.util.logging.*;

    /**
     * Session which manages a JDBC connection with a database, optionally maintains a cache of record objects,
     * and provides ORM, CRUD, and transaction services.
     * <p>
     * The lifetime of a session object and the JDBC connection it uses are the same. Once the JDBC connection
     * which backs a session instance is closed, the session can no longer be used. If a new connection is
     * needed, a new session must be created.
     */
    public class Session
    implements DmoState,
               CacheExpiryListener<RecordIdentifier<String>, BaseRecord>,
               AutoCloseable
    {
       /**
        * The default name of the primary key. The value is the same as the default value of
        * {@code com.goldencode.p2j.schema.SchemaConfig#primaryKeyName}, used at conversion and
        * import time. The default "recid" is a reserved keyword in 4GL, so it cannot conflict with an
        * existing field/column name.
        */
       public static final String DEFAULT_PK = "recid";

       /** Logger */
       private static final CentralLogger log = CentralLogger.get(Session.class.getName());

       /** Object which tracks the current versions of DMOs to enable stale checking */
       private final DmoVersioning versioning;

       /** Database with which this session is used */
       private final Database database;

       /** Helper used to load records from the database */
       private final Loader loader = new Loader(this);

       /** Helper used to persist records to the database */
       private final Persister persister = new Persister(this);

       /** The tenant that uses this session. This will be {@code null} when multi-tenancy is not enabled. */
       private final String tenantName;

       /**
        * The primary key name in use. It can be configured in static config file ({@code p2j.cfg.xml})
        * for (conversion and) import process and in directory for normal execution.
        */
       public static String PK = DEFAULT_PK;

       /** Cache of database records associated with this session */
       private ExpiryCache<RecordIdentifier<String>, BaseRecord> cache;

       /** A cache extension allowing the look-up per table/multiplex faster */
       private Map<String, Map<Integer, Set<RecordIdentifier<String>>>> cacheExtension;

       /** Active JDBC connection, opened when session is instantiated, closed when it is closed */
       private Connection conn = null;

       /** Flag indicating whether a transaction currently is open */
       private boolean inTx = false;

       /** Active savepoint manager, if any */
       private SavepointManager savepointManager = null;

       /** Lazy flag indicating whether the JDBC data store supports batch operations */
       private Boolean batch = null;

       /** Count of transactions opened during this session's lifetime */
       private int txCount = 0;

       /** Instance used to handle session lifecycle events */
       private SessionHook hook = null;

       /**
        * Construct the session and check out a pooled JDBC connection to use for its services.
        * <p>
        * This variant will construct a session which does not have a built-in record cache.
        *
        * @param   database
        *          Database with which to open a session.
        *
        * @throws  PersistenceException
        *          if there is an error getting a JDBC connection from the data source.
        */
       public Session(Database database)
       throws PersistenceException
       {
          this(database, null, false, null, null);
       }

       /**
        * Construct the session and check out a pooled JDBC connection to use for its services.
        * <p>
        * This variant will construct a session which does not have a built-in record cache.
        *
        * @param   database
        *          Database with which to open a session.
        * @param   cfg
        *          The configuration for the given database.
        *
        * @throws  PersistenceException
        *          if there is an error getting a JDBC connection from the data source.
        */
       public Session(Database database, DatabaseConfig cfg)
       throws PersistenceException
       {
          this(database, cfg, true, null, null);
       }

       /**
        * Construct the session and check out a pooled JDBC connection to use for its services.
        * <p>
        * This variant will construct a session with a built-in record cache.
        *
        * @param   database
        *          Database with which to open a session.
        * @param   config
        *          The configuration for the given database.
        * @param   versioning
        *          Optional object which tracks record versions. May be {@code null} to disable versioning.
        * @param   tenantName
        *          The name of the tenant.
        *
        * @throws  PersistenceException
        *          if there is an error getting a JDBC connection from the data source.
        */
       public Session(Database database, DatabaseConfig config, DmoVersioning versioning, String tenantName)
       throws PersistenceException
       {
          this(database, config, true, versioning, tenantName);
       }

       /**
        * Construct the session and check out a JDBC connection from a specified {@code DataSource} to
        * use for its services.
        *
        * @param   database
        *          Database with which to open a session.
        * @param   ds
        *          The {@code DataSource} that will provide the connection to database.
        *
        * @throws  PersistenceException
        *          if there is an error getting a JDBC connection from the data source.
        */
       public Session(Database database, DataSource ds)
       throws PersistenceException
       {
          this.database = database;
          this.cache = null;
          this.cacheExtension = null;
          this.versioning = null;
          this.tenantName = null;

          try
          {
             this.conn = ds.getConnection();
             DatabaseMetaData dbMeta = conn.getMetaData();
             this.batch = dbMeta.supportsBatchUpdates();
          }
          catch (SQLException sqle)
          {
             throw new PersistenceException(sqle);
          }
       }

       /**
        * Construct the session and check out a pooled JDBC connection to use for its services.
        * <p>
        * Trigger the {@link SessionHook#onSessionClaim(Connection) onSessionOpen()} method if a
        * {@link SessionHook} was provided in the db configuration.
        *
        * @param   database
        *          Database with which to open a session.
        * @param   cfg
        *          The configuration for the given database.
        * @param   useCache
        *          {@code true} to enable record caching; {@code false} to disable it.
        * @param   versioning
        *          Optional object which tracks record versions. May be {@code null} to disable versioning.
        * @param   tenantName
        *          The name of the tenant (optional).
        *
        * @throws  PersistenceException
        *          if there is an error getting a JDBC connection from the data source.
        */
       private Session(Database database,
                       DatabaseConfig cfg,
                       boolean useCache,
                       DmoVersioning versioning,
                       String tenantName)
       throws PersistenceException
       {
          this.database = database;
          this.conn = JdbcDataSource.getConnection(database, cfg, tenantName);
          this.cache = useCache ? createCache() : null;
          this.cacheExtension = null;
          this.versioning = versioning;
          this.tenantName = tenantName;

          DatabaseConfig config = cfg;
          if (config == null)
          {
             config = DatabaseManager.getConfiguration(database);
          }

          this.hook = config.getHook();

          // trigger the onSessionClaim hook
          claim();
       }

       /**
        * Method called when either a {@link Session} is created or when a {@link Session} is reclaimed. The
        * role of this method is to call the associated method from {@link SessionHook}.
        */
       protected void claim()
       {
          if (this.hook != null)
          {
             this.hook.onSessionClaim(this.conn);
          }
       }

       /**
        * Method called at server bootstrap that initializes values from the directory
        * configuration. Until this method is called, default values are used.
        */
       public static void bootstrap()
       {
          String primaryKey = null;

          DirectoryService ds = DirectoryService.getInstance();
          try
          {
             if (ds != null)
             {
                if (!ds.bind())
                {
                   throw new RuntimeException("Directory bind failed");
                }

                try
                {
                   // configure primary key column name from directory, if available
                   String path = "persistence/primary-key-name";
                   primaryKey = Utils.getDirectoryNodeString(ds, path, null, false);
                }
                finally
                {
                   ds.unbind();
                }
             }
          }
          finally
          {
             // Initialize the Persister cache, it can't be done from CacheManager since
             // the classes are not in the same package.
             Persister.initializeCache();
          }

          if (primaryKey == null)
          {
             // if we run in offline mode (convert or import when the directory is not available) or the
             // primary key name is not defined in the directory, attempt to read the value from the
             // p2j.cfg.xml file
             try
             {
                primaryKey = Configuration.getParameter("primary-key-name", DEFAULT_PK);
             }
             catch (Exception exc)
             {
                // if neither directory nor p2j.cfg.xml is available, use the default value
                primaryKey = DEFAULT_PK;
             }
          }

          // finally, set the value into the static final field:
          PK = primaryKey;
       }

       /**
        * Creates and returns a {@code Query} based on a FQL statement. The returned object is not
        * bounded to any session; the {@code Session} upon which it will be executed will be provided
        * at the time of execution.
        *
        * @param   fql
        *          The FQL statement that will be executed by the returned {@code Query}.
        *
        * @return  a {@code SQLQuery} that can be used to execute the specified SQL statement.
        */
       public static Query createQuery(String fql)
       {
          return createQuery(q -> {}, fql);
       }

       /**
        * Creates and returns a {@code Query} based on a FQL statement. The returned object is not
        * bounded to any session; the {@code Session} upon which it will be executed will be provided
        * at the time of execution.
        *
        * @param   cache
        *          caching lambda
        * @param   fql
        *          The FQL statement that will be executed by the returned {@code Query}.
        * @return  a {@code SQLQuery} that can be used to execute the specified SQL statement.
        */
       public static Query createQuery(Consumer<Query> cache, String fql)
       {
          return new Query(cache, fql);
       }

       /**
        * Creates and returns a {@code SQLQuery} based on a SQL statement (not mandatory a SQL query).
        * The returned object is not bounded to any connection; the {@code Connection} upon which it
        * will be executed will be provided at the time of execution.
        *
        * @param   sql
        *          The SQL statement that will be executed by the returned {@code SQLQuery}.
        * @param   sessionAttrs
        *          Mutable session attributes' values used in the query
        * @param   parPerm
        *          The permutation function, if any available.
        *
        * @return  a {@code SQLQuery} that can be used to execute the specified SQL statement.
        */
       public static SQLQuery createSQLQuery(String sql, List<SessionAttr> sessionAttrs, int[] parPerm)
       {
          return new SQLQuery(sql, sessionAttrs, parPerm);
       }

       /**
        * Creates and returns a {@code SQLQuery} based on a SQL statement (not mandatory a SQL query).
        * The returned object is not bounded to any connection; the {@code Connection} upon which it
        * will be executed will be provided at the time of execution.
        *
        * @param   sql
        *          The SQL statement that will be executed by the returned {@code SQLQuery}.
        *
        * @return  a {@code SQLQuery} that can be used to execute the specified SQL statement.
        */
       public static SQLQuery createSQLQuery(String sql)
       {
          return new SQLQuery(sql);
       }

       /**
        * Get the record metadata for the given DMO implementation class.
        *
        * @param   dmoClass
        *          DMO implementation class.
        *
        * @return  Record metadata information.
        *
        * @throws  IllegalArgumentException
        *          if the DMO class does not define a record metadata field or if there is an error
        *          accessing it.
        */
       public static RecordMeta getMetadata(Class<? extends BaseRecord> dmoClass)
       {
          try
          {
             Field metaField = dmoClass.getDeclaredField("_metadata");
             metaField.setAccessible(true);

             return (RecordMeta) metaField.get(null);
          }
          catch (NoSuchFieldException exc)
          {
             throw new IllegalArgumentException(
                "DMO implementation class " + dmoClass.getName() + " does not define a '_metadata' field");
          }
          catch (IllegalArgumentException | IllegalAccessException exc)
          {
             throw new IllegalArgumentException(
                "Error accessing DMO implementation class metadata for " + dmoClass.getName());
          }
       }

       /**
        * Get the cached DMO of the given type with the given primary key.
        * <p>
        * If a record is found in the cache, but it is marked stale, it is removed.
        *
        * @param   dmoClass
        *          DMO implementation type.
        * @param   id
        *          Primary key.
        *
        * @return  The cached DMO instance, or {@code null} if it was not cached under the given
        *          primary key or if it was marked stale.
        *
        * @throws  PersistenceException
        *          if a DMO instance is found in the cache which is not of the given DMO type.
        */
       public <T extends BaseRecord> T getCached(Class<T> dmoClass, Long id)
       throws PersistenceException
       {
          // quick out
          if (id == null || cache == null)
          {
             return null;
          }

          T dmo = getCached(new RecordIdentifier<>(dmoClass.getName(), id));

          // assert that we have the right DMO type for this primary key
          if (dmo != null && !dmoClass.isInstance(dmo))
          {
             // all primary keys should be unique across all tables, so this error suggests an
             // invalid primary key was used to store a record after initial import
             String msg = "Cached record request with identifier " +
                          dmoClass.getName() + "#" + id +
                          " found record of type " + dmo.getClass().getName() + "!";

             throw new PersistenceException(msg);
          }

          return dmo;
       }

       /**
        * Gets a specified record of a table. To lookup the record local cache is analyzed first. If
        * the record is not found in cache, the backing database is queried. If the specified record
        * is not found in database, {@code null} is returned. If the record was fetched from the
        * database at this moment, it is kept in cache for further access.
        *
        * @param   <T>
        *          The type of the record.
        * @param   dmoImplClass
        *          The class of the entity (class that implements the table's DMO interface).
        * @param   id
        *          The id/rowid of the record to be returned.
        *
        * @return  The requested record, if found in cache or database and {@code null} otherwise.
        *
        * @throws  PersistenceException
        *          In case some error occurs in the process.
        */
       public <T extends BaseRecord> T get(Class<T> dmoImplClass, Long id)
       throws PersistenceException
       {
          return getImpl(dmoImplClass, id, new RecordIdentifier<>(dmoImplClass.getName(), id), null);
       }

       /**
        * Gets a specified record of a table. To lookup the record local cache is analyzed first. If
        * the record is not found in cache, the backing database is queried. If the specified record
        * is not found in database, {@code null} is returned. If the record was fetched from the
        * database at this moment, it is kept in cache for further access.
        *
        * @param   <T>
        *          The type of the record.
        * @param   dmoImplClass
        *          The class of the entity (class that implements the table's DMO interface).
        * @param   id
        *          The id/rowid of the record to be returned.
        * @param   partialFields
        *          For an incomplete record, read only these fields.
        *
        * @return  The requested record, if found in cache or database and {@code null} otherwise.
        *
        * @throws  PersistenceException
        *          In case some error occurs in the process.
        */
       public <T extends BaseRecord> T get(Class<T> dmoImplClass, Long id, BitSet partialFields)
       throws PersistenceException
       {
          return getImpl(dmoImplClass, id, new RecordIdentifier<>(dmoImplClass.getName(), id), partialFields);
       }

       /**
        * Gets a specified record of a table. To lookup the record local cache is analyzed first. If
        * the record is not found in cache, the backing database is queried. If the specified record
        * is not found in database, {@code null} is returned. If the record was fetched from the
        * database at this moment, it is kept in cache for further access.
        *
        * @param   <T>
        *          The type of the record.
        * @param   key
        *          The record identifier key with which the DMO is associated in the cache. The {@code table}
        *          portion of the key must be the fully qualified class name of the DMO.
        *
        * @return  The requested record, if found in cache or database and {@code null} otherwise.
        *
        * @throws  PersistenceException
        *          In case some error occurs in the process.
        */
       public <T extends BaseRecord> T get(RecordIdentifier<String> key)
       throws PersistenceException
       {
          return getImpl(null, key.getRecordID(), key, null);
       }

       /**
        * Get a record. If found in cache, it is returned immediately, otherwise the database is
        * queried for the specified record. If found, it is saved to cache and returned. Otherwise
        * {@code null} is returned.
        *
        * @param   entity
        *          The string representation of the {@code Class} of the entity to be returned.
        * @param   id
        *          The {@code recid} of the entity to be returned.
        * @param   <T>
        *          The compile type of the record to be returned.
        *
        * @return  The record with requested recid/id if found in cache/database and {@code null}
        *          otherwise.
        *
        * @throws  PersistenceException
        *          if there is an error getting a JDBC connection from the data source.
        */
       public <T extends BaseRecord> T get(String entity, Long id)
       throws PersistenceException
       {
          return getImpl(null, id, new RecordIdentifier<>(entity, id), null);
       }

       /**
        * Get a record. If found in cache, it is returned immediately, otherwise the database is
        * queried for the specified record. If found, it is saved to cache and returned. Otherwise
        * {@code null} is returned.
        *
        * @param   dmoImplClass
        *          The class of the entity (class that implements the table's DMO interface). Optional if a valid
        *          {@code key} is provided.
        * @param   id
        *          The {@code recid} of the entity to be returned. Optional if a valid {@code key} is provided.
        * @param   key
        *          The cache key of the entity. If {@code null} the returned value is {@code null};
        *          provided.
        * @param   partialFields
        *          For an incomplete record, read only these fields.
        * @param   <T>
        *          The compile type of the record to be returned.
        *
        * @return  The record with requested recid/id if found in cache/database and {@code null}
        *          otherwise.
        *
        * @throws  PersistenceException
        *          if there is an error getting a JDBC connection from the data source.
        */
       private <T extends BaseRecord> T getImpl(Class<T> dmoImplClass,
                                                Long id,
                                                RecordIdentifier<String> key,
                                                BitSet partialFields)
       throws PersistenceException
       {
          // quick out
          if (id == null)
          {
             return null;
          }

          if (key == null)
          {
             if (dmoImplClass == null)
             {
                throw new NullPointerException(
                      "dmoImplClass and key parameters cannot be simultaneously null.");
             }

             key = new RecordIdentifier<>(dmoImplClass.getName(), id);
          }

          T dmo = getCached(key);

          boolean refreshIncomplete = false;
          if (dmo != null)
          {
             // assert that we have the right DMO type for this primary key
             if (!dmo.getClass().getName().equals(key.getTable()))
             {
                // all primary keys should be unique across all tables, so this error suggests an
                // invalid primary key was used to store a record after initial import
                String msg = "Cached record request with identifier " + key.getTable() + "#" + id +
                             " found record of type " + dmo.getClass().getName() + "!";

                throw new PersistenceException(msg);
             }

             if (dmo.checkState(DmoState.INCOMPLETE) && partialFields == null)
             {
                // we have a cached record which is incomplete, and this requires to refresh the incomplete record
                // from the database
                refreshIncomplete = true;
             }
             else
             {
                return dmo;
             }
          }

          if (dmoImplClass == null)
          {
             try
             {
                // normally, [entity] is class dynamically generated by [DmoClass] and loaded by [AsmClassLoader].
                // it will generate [ClassNotFoundException] when queried using the simpler Class.forName() method
                dmoImplClass = (Class<T>) Class.forName(key.getTable(), true, AsmClassLoader.getInstance());
             }
             catch (ClassNotFoundException e)
             {
                throw new PersistenceException(e);
             }
          }

          if (!Record.class.isAssignableFrom(dmoImplClass))
          {
             throw new PersistenceException("Invalid entity name: " + dmoImplClass.getName());
          }

          T loaded = loader.load(dmoImplClass, id, partialFields);
          if (refreshIncomplete)
          {
             if (loaded == null)
             {
                // evict the incomplete record from the cache
                evict(dmo);
                dmo = null;
             }
             else
             {
                BitSet dirtyProps = dmo.getDirtyProps();
                // copy the dirty props to the full record
                for (int p = dirtyProps.nextSetBit(0); p >= 0; p = dirtyProps.nextSetBit(p + 1))
                {
                   // direct access
                   loaded.data[p] = dmo.data[p];
                }
                // copy the full record (preserving the dirty properties) to the cached record, to preserve the
                // instance
                System.arraycopy(loaded.data, 0, dmo.data, 0, dmo.data.length);

                // set the incomplete record as complete
                dmo.updateState(this, DmoState.INCOMPLETE, false);
                dmo.readFields = null;
             }
          }
          else
          {
             dmo = loaded;
          }

          if (dmo != null)
          {
             // cache the result
             // StaleRecordException should not be thrown here, because the record was just read from the
             // database, which means we have a pessimistic lock OR it is a local temp-table record which
             // cannot be changed by another session
             if (!refreshIncomplete)
             {
                // the incomplete record was already found in cache
                associate(key, dmo);
             }
             dmo.updateState(this, READ, true);
          }

          return dmo;
       }

       /**
        * Saves a {@code dmo} to database. If so requested, its internal state information is updated.
        *
        * @param   dmo
        *          The record to be saved.
        * @param   updateDmoState
        *          If {@code true} its internal state information is updated. This is the standard
        *          call. When the method is called for validation, use {@code false} as the database
        *          operation will be rolled back anyway, so the state must not be altered.
        * @param   <T>
        *          The compile type of the record.
        *
        * @throws  PersistenceException
        *          When an exception is encountered. Must be analyzed as the cause might be expected
        *          (ex: when validating a record).
        * @throws  StaleRecordException
        *          if the record being saved is stale.
        */
       public <T extends BaseRecord> void save(T dmo, boolean updateDmoState)
       throws PersistenceException
       {
          checkClosed();

          checkTx();

          Long id = dmo.primaryKey();

          // verify primary key is present
          if (id == null)
          {
             throw new PersistenceException("DMO cannot be persisted without a primary key");
          }

          RecordIdentifier<String> key = new RecordIdentifier<>(dmo.getClass().getName(), id);

          // the record should have been verified by callers before getting here, but just in case...
          try
          {
             checkStale(key, dmo);
          }
          catch (StaleRecordException e)
          {
             // the DMO is stale because the shared version incremented when a copy of the DMO was altered without
             // our [dmo] notice. Normally the unsaved changes should have been copied to that instance and they
             // will be saved when that instance will be flushed
             return;
          }

          // check if we have a different DMO instance in the cache with that ID
          T cached = getCached(key);
          if (cached != null && cached != dmo)
          {
             // cannot have more than one DMO instance associated to the database with the same id
             String msg = "Requested to persist DMO of type " + dmo.getClass().getName() + " with " +
                          Session.PK + " " + id + ". A different DMO instance of type " +
                          cached.getClass().getName() +
                          " is already bound to this session with the same " + Session.PK;

             throw new PersistenceException(msg);
          }

          boolean wasNew = dmo.checkState(NEW);
          boolean wasDirty = wasNew || dmo.isDirty();

          // decide what to do with the record
          if (wasNew)
          {
             persister.insert(dmo);
             wasDirty = true;
          }
          else if (wasDirty)
          {
             persister.update(dmo);
             wasDirty = true;
          }

          if (wasDirty && updateDmoState)
          {
             if (wasNew)
             {
                dmo.updateState(this, NEW, false);
             }

             // clear dirty state, since all changes have been saved to database
             dmo.clearDirtyState(this);
          }

          if (cached == null && updateDmoState)
          {
             associate(key, dmo);
          }

          if (dmo._recordMeta().getDmoMeta().getMetaTable() == MetadataManager.SystemTable._User)
          {
             SecurityOps.invalidate(database.getDefault().getName(), MetadataSecurityOps.META_TABLE_USER);
          }
       }

       /**
        * Persist a list of records, performing inserts using SQL batching, if supported. If not
        * supported, perform the inserts one-by-one.
        * <p>
        * This method is not intended for legacy runtime support. It is meant for bulk inserts and
        * as such, it:
        * <ul>
        * <li>bypasses the session cache (if any), though it should be used with a cache-less
        *     session);</li>
        * <li>does not use savepoints;</li>
        * <li>does not update DMO state.</li>
        * </ul>
        *
        * @param   records
        *          List of records to be saved. Each record must have a unique, positive primary key
        *          assigned.
        *
        * @return  The count of records successfully inserted, which should match the size of the
        *          list provided if there were no errors. If SQL batching is not supported, returns
        *          -1, without saving records.
        *
        * @throws  PersistenceException
        *          if there was an error performing the batch insert. This will wrapper the {@code
        *          SQLException} thrown by the JDBC layer.
        */
       public <T extends BaseRecord> int bulkSave(List<T> records)
       throws PersistenceException
       {
          checkTx();

          return persister.bulkInsert(records);
       }

       /**
        * Delete the DMO's record(s) from the database and remove it from the cache.
        *
        * @param   <T>
        *          DMO type
        * @param   dmo
        *          DMO to be deleted.
        * @return  <code>true</code> on success
        *
        * @throws  PersistenceException
        *          if there is a database error, or if a different DMO instance with the same ID as
        *          the given DMO is found in cache.
        */
       public <T extends BaseRecord> boolean delete(T dmo)
       throws PersistenceException
       {
          checkClosed();

          checkTx();

          Long id = dmo.primaryKey();

          // verify primary key is present
          if (id == null)
          {
             throw new PersistenceException("DMO cannot be deleted without a primary key");
          }

          RecordIdentifier<String> key = new RecordIdentifier<>(dmo.getClass().getName(), id);

          // check if we have a different DMO instance in the cache with that ID
          T cached = getCached(key);
          if (cached != null && cached != dmo)
          {
             // cannot have more than one DMO instance associated to the database with the same id
             String msg = String.format(
                "Requested to delete DMO of type %s with id %d. A different DMO instance of type " +
                "%s is already bound to this session with the same " + Session.PK,
                dmo.getClass().getName(),
                id,
                cached.getClass().getName());

             throw new PersistenceException(msg);
          }

          boolean success = true;

          // attempt to delete the record if it is not newly created (i.e., as of yet unflushed)
          if (!dmo.checkState(NEW))
          {
             success = persister.delete(dmo);
          }
          else if (dmo instanceof TempRecord)
          {
             try
             {
                // remove the slot from MultiplexedScanIndex; invalid records have already been cleared
                DirectAccessHelper.clearPrimaryKey(this, (TempRecord) dmo);
             }
             catch (DirectAccessException e)
             {
                throw new PersistenceException(e);
             }
          }

          if (success)
          {
             dmo.updateState(this, DELETED, true);
             if (cached != null)
             {
                evict(key, cached);
             }

             if (dmo._recordMeta().getDmoMeta().getMetaTable() == MetadataManager.SystemTable._User)
             {
                SecurityOps.invalidate(database.getDefault().getName(), MetadataSecurityOps.META_TABLE_USER);
             }
          }

          return success;
       }

       /**
        * Deletes a record from the database and from internal cache.
        * <p>
        * NOTE: Only used by dirty share manager; currently not integrated with the savepoint manager,
        *       due to inability to discern between undoable and NO-UNDO records
        *
        * @param   entity
        *          The string representation of the {@code Class} of the record to be deleted.
        *          Actually, this represents the table of the record to be deleted.
        * @param   id
        *          The {@code recid} of the record to be deleted.
        *
        * @return  {@code true} if the record wa found and deleted and {@code false} if the record was
        *          not detected in database.
        *
        * @throws  PersistenceException
        *          if there is an error accessing the database.
        * @throws  IllegalArgumentException
        *          if {@code id} is null.
        */
       public boolean delete(String entity, Long id)
       throws PersistenceException
       {
          checkClosed();

          checkTx();

          // verify primary key is present
          if (id == null)
          {
             throw new PersistenceException("DMO cannot be deleted without a primary key");
          }

          Class<?> dmoClass = null;
          try
          {
             // normally, [entity] is class dynamically generated by [DmoClass] and loaded by [AsmClassLoader].
             // it will generate [ClassNotFoundException] when queried using the simpler Class.forName() method
             dmoClass = Class.forName(entity, true, AsmClassLoader.getInstance());
          }
          catch (ClassNotFoundException e)
          {
             throw new PersistenceException(e);
          }

          if (!Record.class.isAssignableFrom(dmoClass))
          {
             throw new PersistenceException("Invalid entity name: " + entity);
          }

          RecordIdentifier<String> key = new RecordIdentifier<>(entity, id);

          // check if we have a different DMO instance in the cache with that ID
          BaseRecord cached = getCached(key);

          boolean success = true;

          // attempt to delete the record if it is not newly created (i.e., as of yet unflushed)
          if (cached == null || !cached.checkState(NEW))
          {
             @SuppressWarnings("unchecked")
             Class<? extends Record> recordClass = (Class<? extends Record>) dmoClass;

             success = persister.delete(recordClass, id);
          }

          if (success && cached != null)
          {
             cached.updateState(this, DELETED, true);
             evict(key, cached);
          }

          return success;
       }

       /**
        * Apply the given update to all the cached records.
        * <p>
        * WARNING: this must be called only for temp-tables, and it must be used with a direct SQL on the same
        * temp-table records, so that the cached and the sql table records have the same state.
        *
        * @param    dmoImplName
        *           The name of the class implementation of DMO.
        * @param    multiplex
        *           The table multiplexer (only for temp tables).
        * @param    update
        *           The code to apply on the cached records.
        */
       public void updateCachedRecords(String dmoImplName, Integer multiplex, Consumer<BaseRecord> update)
       {
          if (cache == null || cacheExtension == null)
          {
             return; // nothing to do
          }

          Map<Integer, Set<RecordIdentifier<String>>> tableMap = cacheExtension.get(dmoImplName);
          if (tableMap == null)
          {
             // there are no records cached for this table
             return;
          }

          Set<RecordIdentifier<String>> multiplexMap = tableMap.get(multiplex);
          if (multiplexMap == null)
          {
             // there are no records cached for this table/multiplex combination
             return;
          }

          for (RecordIdentifier<String> ri : multiplexMap)
          {
             BaseRecord r = cache.get(ri);
             if (r == null)
             {
                log.warning("Inconsistent cache extension: "
                          + "The key which is stored by the extension is no longer in the primary cache.");
                continue;
             }
             update.accept(r);
          }
       }

       /**
        * Invalidates all records from cache belonging to a specific table. If provided a multiplex, only
        * those records are affected.
        *
        * @param   dmoImplName
        *          Name of the DMO implementation class whose associated records are going to be removed from
        *          the cache.
        * @param   multiplex
        *          Optional multiplex used to filter which records are deleted.
        */
       public void invalidateRecords(String dmoImplName, Integer multiplex)
       {
          if (cache == null || cacheExtension == null)
          {
             return; // nothing to do
          }

          Map<Integer, Set<RecordIdentifier<String>>> tableMap = cacheExtension.get(dmoImplName);
          if (tableMap == null)
          {
             // there are no records cached for this table
             return;
          }

          Set<RecordIdentifier<String>> multiplexMap = tableMap.get(multiplex);
          if (multiplexMap == null)
          {
             // there are no records cached for this table/multiplex combination
             return;
          }

          for (RecordIdentifier<String> ri : multiplexMap)
          {
             if (cache.remove(ri) == null)
             {
                log.warning("Inconsistent cache extension: "
                          + "The key which is stored by the extension is no longer in the primary cache.");
             }
          }

          // the multiplex is going to be empty anyway; remove it from the table map
          tableMap.remove(multiplex);

          if (tableMap.isEmpty())
          {
             // the table doesn't have anymore records cached
             cacheExtension.remove(dmoImplName);
          }
       }

       /**
        * Evicts a record of a table specified by its DMO class implementation. The record is specified by its
        * primary key.
        *
        * @param   dmoImplName
        *          Name of the DMO implementation class whose associated records are going to be removed from
        *          the cache.
        * @param   pk
        *          The primary key of the record to be evicted.
        *
        * @return  {@code true} if the record was evicted from cache and {@code false} is the specified record was
        *          not found in cache.
        */
       boolean evictIfExist(String dmoImplName, long pk)
       {
          if (cache == null)
          {
             return false;
          }

          RecordIdentifier<String> key = new RecordIdentifier<>(dmoImplName, pk);

          BaseRecord cached = cache.remove(key);
          if (cached != null)
          {
             // cached should match dmo. if not, cached is safer to use here because it is the element extracted
             // from the primary cache.
             removeFromCacheExtension(key, cached);

             if (versioning != null)
             {
                versioning.deregister(key);
             }
             return true;
          }
          return false;
       }

       /**
        * Refresh the given DMO with data from the database, bypassing the cache. The DMO's primary
        * key is assumed to represent a valid identifier. If no record is found with this primary
        * key, the existing DMO (if any) with that identifier is removed from the cache, and {@code
        * null} is returned.
        *
        * @param   dmo
        *          DMO whose data is to be refreshed from the database.
        *
        * @return  The instance passed into this method as {@code dmo}, containing refreshed data,
        *          or {@code null} if a matching record was not found.
        *
        * @throws  PersistenceException
        *          if a database error occurs.
        */
       public <T extends BaseRecord> T refresh(T dmo)
       throws PersistenceException
       {
          checkClosed();

          // ignore the cache and go directly to underlying database
          // use the variant of load which overwrites the DMOs current field values with data
          // fetched from the database
          T loaded = loader.load(dmo, dmo.id, null);

          RecordIdentifier<String> key = new RecordIdentifier<>(dmo.getClass().getName(), dmo.id);

          if (loaded != null)
          {
             // StaleRecordException should not be thrown here, because the record was just read from the
             // database, which means we have a pessimistic lock OR it is a local temp-table record which
             // cannot be changed by another session
             associate(key, loaded);
             loaded.updateState(this, READ, true);
          }
          else
          {
             evict(key, dmo);
          }

          return loaded;
       }

       // NOTE: Hibernate's Session.merge() saves the record if not done already. FWD ORM does the
       //       same in order to keep the existing code assumptions.
       // TODO: can we get rid of this?
       /**
        * @param   <T>
        *          dmo TYPE
        * @param   dmo
        *          DMO
        *
        * @return  saved record
        *
        * @throws  PersistenceException
        *          on error
        */
       public <T extends BaseRecord> T merge(T dmo)
       throws PersistenceException
       {
          @SuppressWarnings("unchecked")
          T persisted = (T) getCached(dmo.getClass(), dmo.id);

          if (persisted == null)
          {
             persisted = (T) loader.load(dmo.getClass(), dmo.id);
             if (persisted == null)
             {
                return null;
             }
          }

          System.arraycopy(dmo.data, 0, persisted.data, 0, dmo.data.length);

          return persisted;
       }

       /**
        * Associate the given record with the session cache. It is an error if another record instance with
        * the same primary key already is associated with the cache.
        *
        * @param   dmo
        *          The record to associate with the cache.
        *
        * @throws  PersistenceException
        *          if another record instance already is associated with the session under the same key.
        * @throws  StaleRecordException
        *          if the record being associated with the session is stale.
        */
       public <T extends BaseRecord> void associate(T dmo)
       throws PersistenceException,
              StaleRecordException
       {
          if (cache == null)
          {
             return;
          }

          RecordIdentifier<String> key = new RecordIdentifier<>(dmo.getClass().getName(), dmo.id);
          associate(key, dmo);
       }

       /**
        * Associate the given record with the session cache. It is an error if another record instance with
        * the same primary key already is associated with the cache.
        *
        * @param   key
        *          The record identifier key with which to associate the record in the cache. The {@code table}
        *          portion of the key must be the fully qualified class name of the DMO.
        * @param   dmo
        *          The record to associate with the cache.
        *
        * @throws  PersistenceException
        *          if another record instance already is associated with the session under the same key.
        * @throws  StaleRecordException
        *          if the record being associated with the session is stale.
        */
       public <T extends BaseRecord> void associate(RecordIdentifier<String> key, T dmo)
       throws PersistenceException,
              StaleRecordException
       {
          if (cache == null)
          {
             return;
          }

          checkStale(key, dmo);

          BaseRecord cached = cache.putIfAbsent(key, dmo);
          if (cached == null)
          {
             addToCacheExtension(key, dmo);
          }
          else if (cached != dmo)
          {
             // this is weird; we need to do our best to keep the extension consistent
             // an override in the cache means a remove and add in the cache extension
             removeFromCacheExtension(key, cached);
             addToCacheExtension(key, dmo);

             // cannot have more than one DMO instance associated to the database with the same id
             String msg = "Requested to associate DMO of type " + dmo.getClass().getName() + " with " +
                          Session.PK + " " + dmo.id + ". A different DMO instance of type " +
                          cached.getClass().getName() +
                          " is already bound to this session with the same " + Session.PK;
             throw new PersistenceException(msg);
          }

          if (cached == null)
          {
             dmo.updateState(this, CACHED, true);

             if (versioning != null)
             {
                dmo.version(versioning.register(key));
             }
          }
       }

       /**
        * Remove a record from the session cache.
        *
        * @param   dmo
        *          The record to be evicted.
        * @param   <T>
        *          The compile type of the record.
        */
       public <T extends BaseRecord> void evict(T dmo)
       {
          if (cache == null)
          {
             return;
          }

          RecordIdentifier<String> key = new RecordIdentifier<>(dmo.getClass().getName(), dmo.id);
          evict(key, dmo);
       }

       /**
        * Convenience method to access this session database connection.
        * <p>
        * Use with caution and do not directly commit, rollback, set or release savepoints, etc.!
        * Definitely do not close this connection directly! This will render the enclosing session
        * inoperable. Use the {@link Session} API for these services instead.
        *
        * @return  Pooled JDBC connection. There is no guarantee that the connection is in a valid
        *          state.
        *
        * @throws  PersistenceException
        *          if the JDBC connection has been closed or there was an error checking its status.
        */
       public Connection getConnection()
       throws PersistenceException
       {
          checkClosed();

          return conn;
       }

       /**
        * Indicate whether a database transaction currently is in process.
        *
        * @return  {@code true} if a transaction exists, else {@code false}.
        */
       public boolean inTransaction()
       {
          return inTx;
       }

       /**
        * Begin a new database transaction, if one does not already exist for this session.
        * Do not install a savepoint manager.
        *
        * @return  {@code true} if a new transaction was begun; {@code false} if a transaction
        *          already exists for this session.
        *
        * @throws  PersistenceException
        *          if a database error occurs.
        */
       public boolean beginTransaction()
       throws PersistenceException
       {
          return beginTransaction(null);
       }

       /**
        * Begin a new database transaction, if one does not already exist for this session.
        *
        * @param   savepointManager
        *          Optional savepoint manager to install in the session for the duration of the
        *          transaction. This should only be installed when the transaction corresponds with
        *          a legacy, application-level transaction. It will match database savepoints with
        *          application block sub-transactions, and keep in-memory DMO state in sync with
        *          the database through commit and rollback processing. Should be {@code null} when
        *          beginning a database transaction which does not correspond with a legacy
        *          application transactions
        *
        * @return  {@code true} if a new transaction was begun; {@code false} if a transaction
        *          already exists for this session.
        *
        * @throws  PersistenceException
        *          if a database error occurs.
        */
       public boolean beginTransaction(SavepointManager savepointManager)
       throws PersistenceException
       {
          checkClosed();

          if (inTx)
          {
             return false;
          }

          try
          {
             conn.setAutoCommit(false);

             if (savepointManager != null)
             {
                if (this.savepointManager != null)
                {
                   throw new IllegalStateException("Programming error; SavepointManager already exists!");
                }

                this.savepointManager = savepointManager;
                savepointManager.setSession(this);
             }

             inTx = true;

             txCount++;

             return true;
          }
          catch (SQLException exc)
          {
             throw new PersistenceException("Error beginning a new database transaction", exc);
          }
       }

       /**
        * Commit all work done since the beginning of the current transaction.
        * <p>
        * The session's current savepoint manager, if any, is discarded after the transaction ends.
        *
        * @return  {@code true} if the transaction was committed; {@code false} if it there is no
        *          active transaction to commit.
        *
        * @throws  PersistenceException
        *          if there was a database error committing the transaction.
        */
       public boolean commit()
       throws PersistenceException
       {
          checkClosed();

          if (!inTx)
          {
             return false;
          }

          try
          {
             conn.commit();

             return true;
          }
          catch (SQLException exc)
          {
             throw new PersistenceException("Error committing database transaction", exc);
          }
          finally
          {
             inTx = false;
             savepointManager = null;
          }
       }

       /**
        * Roll back all work done since the beginning of the current transaction.
        * <p>
        * The session's current savepoint manager, if any, is discarded after the transaction ends.
        *
        * @return  {@code true} if the transaction was rolled back; {@code false} if it there is no
        *          active transaction to roll back.
        *
        * @throws  PersistenceException
        *          if there was a database error rolling back the transaction.
        */
       public boolean rollback()
       throws PersistenceException
       {
          checkClosed();

          if (!inTx)
          {
             return false;
          }

          try
          {
             conn.rollback();

             return true;
          }
          catch (SQLException exc)
          {
             throw new PersistenceException("Error rolling back database transaction", exc);
          }
          finally
          {
             inTx = false;
             savepointManager = null;
          }
       }

       /**
        * Set the savepoint for the current block.
        */
       public void lazilySetSavepoints()
       throws PersistenceException
       {
          if (inTx && savepointManager != null)
          {
             savepointManager.lazilySetSavepoints(false);
          }
       }

       /**
        * Set a savepoint to which database changes can be rolled back, if we are currently within an active
        * transaction.
        *
        * @return  The savepoint object, if successfully set. {@code null} is returned if we are not within
        *          an active transaction.
        *
        * @throws  PersistenceException
        *          if a database error occurs.
        */
       public Savepoint setSavepoint()
       throws PersistenceException
       {
          checkClosed();

          if (!inTx)
          {
             return null;
          }

          try
          {
             return conn.setSavepoint();
          }
          catch (SQLException exc)
          {
             throw new PersistenceException("Error setting savepoint", exc);
          }
       }

       /**
        * Release the given savepoint, thereby disabling the ability to roll back to the point at
        * which it was created. This represents a commit of the unit of work which began when the
        * savepoint was set.
        *
        * @param   savepoint
        *          The savepoint to be released.
        *
        * @throws  PersistenceException
        *          if a database error occurs.
        */
       public void releaseSavepoint(Savepoint savepoint)
       throws PersistenceException
       {
          checkClosed();

          if (!inTx)
          {
             return;
          }

          try
          {
             conn.releaseSavepoint(savepoint);
          }
          catch (SQLException exc)
          {
             throw new PersistenceException("Error releasing savepoint", exc);
          }
       }

       /**
        * Roll back all database changes since the given savepoint was set.
        *
        * @param   savepoint
        *          Savepoint to which changes should be rolled back.
        *
        * @return  {@code false} if we are not in a transaction when this method is invoked;
        *          {@code true} if we are in a transaction and the rollback was successful.
        *
        * @throws  PersistenceException
        *          if a database error occurs.
        */
       public boolean rollbackSavepoint(Savepoint savepoint)
       throws PersistenceException
       {
          checkClosed();

          if (!inTx)
          {
             return false;
          }

          try
          {
             conn.rollback(savepoint);

             return true;
          }
          catch (SQLException exc)
          {
             throw new PersistenceException("Error rolling back to savepoint", exc);
          }
       }

       /**
        * Test if the connection to database is active / open.
        *
        * @return  {@code true} if the connection to database is usable.
        */
       public boolean isOpen()
       {
          return conn != null;
       }

       /**
        * Clear the session DMO cache.
        * @throws PersistenceException
        *         on error
        */
       public void clear()
       throws PersistenceException
       {
          if (cache == null)
          {
             cacheExtension = null;
             return;
          }

          deregisterVersioned();

          cache = createCache();
          cacheExtension = null;

          // TODO: cancel all pending saves, updates and deletions
       }

       /**
        * Disables cache for this session. This affects performance and should not be done unless there are good
        * reasons like when we are sure the records will never be used twice. This is the case of the import
        * process when a lot of records are just saved to database and never retrieved.
        * @throws PersistenceException
        *         on error
        */
       public void disableCache()
       throws PersistenceException
       {
          if (cache == null)
          {
             cacheExtension = null;
             return;
          }

          deregisterVersioned();

          cache = null;
          cacheExtension = null;
       }

       /**
        * Get the zero-based transaction block nesting level, where 0 indicates the current block is a full
        * transaction block, 1 indicates the current block is the first level of nested subtransaction, and so
        * on. Note that this includes only blocks with transaction properties; no-transaction blocks which may
        * be nested between blocks with transaction properties or ignored for purposes of this count.
        *
        * @return  The current block's transaction nesting level, or -1 if not currently in an active
        *          transaction.
        */
       public Long getTxNestingId()
       {
          long level = savepointManager == null ? -1 : savepointManager.getTxNestingLevel();

          return level < 0 ? null : (level << 32) | txCount;
       }

       /**
        * Begin tracking an undoable record as needing savepoint processing. This method should only be
        * invoked within the context of an active, application-level transaction. Otherwise, it is a no-op.
        *
        * @param   dmo
        *          Record to be tracked.
        *
        * @throws  PersistenceException
        *          if there is an error lazily setting associated savepoints in the savepoint manager.
        *
        * @see     SavepointManager#trackUndoable(BaseRecord)
        */
       public void trackUndoable(BaseRecord dmo)
       throws PersistenceException
       {
          if (savepointManager != null)
          {
             savepointManager.trackUndoable(dmo);
          }
       }

       /**
        * Closes the connection to database and invalidates the local field. The cache is also dropped
        * as the cached values are no more valid without the active connection.
        * <p>
        * Trigger the {@link SessionHook#onSessionClose(Connection) onSessionClose()} method if a
        * {@link SessionHook} was provided in the db configuration.
        * @throws  PersistenceException
        *          in the event of an error.
        */
       @Override
       public void close()
       throws PersistenceException
       {
          if (conn == null)
          {
             // already closed
             return;
          }

          try
          {
             deregisterVersioned();

             if (!conn.isClosed())
             {
                conn.close();

                // trigger the onSessionClose hook
                if (hook != null)
                {
                   hook.onSessionClose(conn);
                   hook = null;
                }

             }
          }
          catch (SQLException exc)
          {
             throw new PersistenceException("Error checking in pooled database connection", exc);
          }
          finally
          {
             conn = null;
             cache = null;
             cacheExtension = null;
          }
       }

       /**
        * Obtain the FWD database this session serves.
        *
        * @return  the {@code Database} this session is attached to.
        */
       public Database getDatabase()
       {
          return database;
       }

       /**
        * This method is invoked when the session cache has expired one or more elements. It unsets the CACHED
        * state from each expired DMO and deregisters from DmoVersioning.
        * <p>
        * <strong>Important:</strong> Delivery of this event is not asynchronous, so any delay in this method's
        * return will impact the performance of the cache! This implementation should be maintained to return as
        * quickly as possible.
        *
        * @param   event
        *          An object containing information about the event.
        */
       @Override
       public void cacheExpired(CacheExpiryEvent<RecordIdentifier<String>, BaseRecord> event)
       {
          for (Map.Entry<RecordIdentifier<String>, BaseRecord> entry : event.getExpiredEntries().entrySet())
          {
             BaseRecord dmo = entry.getValue();

             if (versioning != null)
             {
                versioning.deregister(entry.getKey());
             }

             // keep the cache extension consistent
             removeFromCacheExtension(entry.getKey(), dmo);

             if ((dmo.state.state & CHANGED) != 0)
             {
                // record was not saved since it was added to cache
                try
                {
                   persister.update(dmo);
                }
                catch (PersistenceException e)
                {
                   throw new RuntimeException("Failed to update cached record #" + Long.toHexString(dmo.id), e);
                }
             }
             dmo.updateState(null, CACHED, false);
          }
       }

       /**
        * Indicate whether the JDBC driver used by this session supports batch statement operations.
        *
        * @return  {@code true} if batch operations are supported, else {@code false}.
        *
        * @throws  PersistenceException
        *          if database metadata needs to be queried and there is an error doing so.
        */
       boolean supportsBatch()
       throws PersistenceException
       {
          if (this.batch == null)
          {
             this.batch = JdbcDataSource.supportsBatch(database);
          }

          return this.batch;
       }

       /**
        * Get the cached DMO associated with the given primary key, if any.
        * <p>
        * If a record is found in the cache, but it is stale, it is removed.
        *
        * @param   key
        *          The record identifier key with which the DMO is associated in the cache. The {@code table}
        *          portion of the key must be the fully qualified class name of the DMO.
        *
        * @return  The cached DMO instance, or {@code null} if there is no cache, if the record was not cached
        *          under the given primary key, or if the record was determined to be stale.
        *
        * @throws  PersistenceException
        *          if a DMO instance is found in the cache which is not of the given DMO type.
        */
       private <T extends BaseRecord> T getCached(RecordIdentifier<String> key)
       throws PersistenceException
       {
          if (cache == null)
          {
             return null;
          }

          @SuppressWarnings("unchecked")
          T cached = (T) cache.get(key);

          if (cached != null && versioning != null)
          {
             // if we have a cached record, then it will have been assigned a version number upon being added
             // to the cache; if that version is now out of sync with the canonical version, evict the record
             // from the cache, and return null
             if (cached.isStale())
             {
                // this will deregister the record from version tracking
                evict(key, cached);

                // in case there are other references to this instance, mark it as stale
                cached.updateState(this, STALE, true);

                return null;
             }
          }

          return cached;
       }

       /**
        * Get the persister for this session.
        *
        * @return  Object used to insert and update records.
        */
       Persister getPersister()
       {
          return persister;
       }

       /**
        * Get the savepoint manager registered with this session, if any.
        *
        * @return  Savepoint manager, or {@code null} if none is registered.
        */
       public SavepointManager getSavepointManager()
       {
          return savepointManager;
       }

       /**
        * Getter for the {@code tenantId} field.
        *
        * @return  The id of the tenant that uses this session.
        */
       public String getTenantName()
       {
          return tenantName;
       }

       /**
        * Create a session expiry cache which disallows DMOs which are referenced by buffers or which are being
        * tracked for UNDO purposes from being expired. The cache uses a lenient capacity policy, so it may grow
        * beyond its set size limit, but attempts will be made to shrink its size back down to its limit.
        *
        * @return  A session cache as described above.
        */
       private ExpiryCache<RecordIdentifier<String>, BaseRecord> createCache()
       {
          ExpiryCache<RecordIdentifier<String>, BaseRecord> c =
             CacheManager.createLRUCache(Session.class,
                                         database.toString(),
                                         1024,
                                         (dmo) -> !dmo.isInUse(),
                                         CapacityPolicy.LENIENT);
          c.addCacheExpiryListener(this);

          return c;
       }

       /**
        * Remove a record from the session cache.
        *
        * @param   key
        *          The record identifier key with which the DMO is associated in the cache. The {@code table}
        *          portion of the key must be the fully qualified class name of the DMO.
        * @param   dmo
        *          The record to be evicted.
        */
       private void evict(RecordIdentifier<String> key, BaseRecord dmo)
       {
          if (cache == null)
          {
             return;
          }

          BaseRecord cached = cache.remove(key);
          if (cached != null)
          {
             if (dmo != cached)
             {
                log.warning("Inconsistent session cache. "
                          + "The removed cached record doesn't match the expected record.");
             }

             // cached should match dmo. if not, cached is safer to use here because it is the element extracted
             // from the primary cache.
             removeFromCacheExtension(key, cached);
          }
          dmo.updateState(this, CACHED, false);

          if (versioning != null && cached != null)
          {
             versioning.deregister(key);
          }
       }

       /**
        * Make sure that the cache extension is consistent. Every add in the primary cache should update the cache extension
        * properly. If this is not consistent, the pruning won't cover all targeted cached records.
        *
        * @param   key
        *          The key which was added to the primary cache.
        * @param   dmo
        *          The dmo which was added. This is used to retrieve the multiplex.
        */
       private void addToCacheExtension(RecordIdentifier<String> key, BaseRecord dmo)
       {
          if (cacheExtension == null)
          {
             cacheExtension = new HashMap<>();
          }

          String tableName = key.getTable();
          Map<Integer, Set<RecordIdentifier<String>>> tableMap = cacheExtension.get(tableName);
          if (tableMap == null)
          {
             tableMap = new HashMap<>();
             cacheExtension.put(tableName, tableMap);
          }

          Integer multiplex = dmo instanceof TempRecord ? ((TempRecord) dmo)._multiplex() : null;
          Set<RecordIdentifier<String>> multiplexMap = tableMap.get(multiplex);
          if (multiplexMap == null)
          {
             multiplexMap = new HashSet<>();
             tableMap.put(multiplex, multiplexMap);
          }

          multiplexMap.add(key);
       }

       /**
        * Make sure that the cache extension is consistent. Every removal from the primary cache (due to eviction
        * or expiration) should update the cache extension properly. If this is not consistent, the extension
        * will leak.
        *
        * @param   key
        *          The key which is removed from the primary cache.
        * @param   dmo
        *          The dmo which was removed. This is used to retrieve the multiplex.
        */
       private void removeFromCacheExtension(RecordIdentifier<String> key, BaseRecord dmo)
       {
          if (cacheExtension == null)
          {
             log.warning("Inconsistent cache extension: "
                       + "The key which is removed from the primary cache is not in the extension.");
             return;
          }

          String tableName = key.getTable();
          Map<Integer, Set<RecordIdentifier<String>>> tableMap = cacheExtension.get(tableName);
          if (tableMap == null)
          {
             log.warning("Inconsistent cache extension: "
                       + "The key which is removed from the primary cache is not in the extension.");
             return;
          }

          Integer multiplex = dmo instanceof TempRecord ? ((TempRecord) dmo)._multiplex() : null;
          Set<RecordIdentifier<String>> multiplexMap = tableMap.get(multiplex);
          if (multiplexMap == null)
          {
             log.warning("Inconsistent cache extension: "
                      + "The key which is removed from the primary cache is not in the extension.");
             return;
          }

          if (!multiplexMap.remove(key))
          {
             log.warning("Inconsistent cache extension: "
                       + "The key which is removed from the primary cache is not in the extension.");
          }

          if (multiplexMap.isEmpty())
          {
             tableMap.remove(multiplex);
             if (tableMap.isEmpty())
             {
                cacheExtension.remove(tableName);
             }
          }
       }

       /**
        * Deregister any cached records from DMO version tracking.
        */
       private void deregisterVersioned()
       {
          if (versioning == null || cache == null)
          {
             return;
          }

          for (RecordIdentifier<String> key : cache.keys())
          {
             versioning.deregister(key);
          }
       }

       /**
        * 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.
        *
        * @throws  PersistenceException
        *          when the connection is not usable because it was closed.
        */
       private void checkClosed()
       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");
          }
       }

       /**
        * A sanity check to ensure any DML being executed is performed within an active transaction or on an
        * auto-commit connection. This method should only be invoked from within methods which perform DML and
        * after it is confirmed that an open connection exists.
        * <p>
        * TODO: the conditions for which this method checks represent programming errors and as such, we may
        * want to remove this method once the persistence subsystem is well tested. On the other hand, the
        * overhead is quite low: in the common case, there is just the overhead of the method call itself and
        * of an instance field check.
        *
        * @throws  PersistenceException
        *          if we are not in an active transaction and the connection is not set to auto-commit.
        */
       private void checkTx()
       throws PersistenceException
       {
          if (inTx)
          {
             // expected state, quick out
             return;
          }

          try
          {
             if (!conn.getAutoCommit())
             {
                throw new PersistenceException("Attempting DML which cannot be committed!");
             }
          }
          catch (SQLException exc)
          {
             throw new PersistenceException("Error checking database connection auto-commit status");
          }
       }

       /**
        * If DMO versioning is enabled for this session, check whether the given record is stale. If versioning
        * is disabled, or the record is not stale, return normally.
        *
        * @param   dmo
        *          Record to be checked.
        *
        * @throws  StaleRecordException
        *          if the record is stale.
        */
       private <T extends BaseRecord> void checkStale(RecordIdentifier<String> key, T dmo)
       throws StaleRecordException
       {
          if (versioning == null)
          {
             return;
          }

          boolean stale = false;

          if (dmo.checkState(STALE))
          {
             stale = true;
          }
          else if (dmo.isStale())
          {
             stale = true;
             dmo.updateState(this, STALE, true);
             evict(key, dmo); // no STALE record in Session's cache
          }

          if (stale)
          {
             throw new StaleRecordException("Cannot use stale record with database session: " + key);
          }
       }
    }