DefaultDirtyShareManager.java
/*
** Module : DefaultDirtyShareManager.java
** Abstract : Default implementation of DirtyShareManager
**
** Copyright (c) 2004-2024, Golden Code Development Corporation.
**
** -#- -I- --Date-- --JPRM-- -----------------------------------Description-----------------------------------
** 001 ECF 20080327 @38627 Created initial version. Default
** implementation of DirtyShareManager.
** 002 ECF 20080606 @38639 Fixed NPE. Added safety checks in cleanup
** code.
** 003 ECF 20080612 @38716 Replaced isEntityDirty() with
** getEntityDirtyState(). New method returns
** state information (is entity dirty with
** changes and/or deletes), rather than a simple
** boolean.
** 004 CA 20080626 @39072 When rollbackInsert is called, remove the ID
** from the changes map.
** 005 CA 20080715 @39109 Fixed removeRecords - the session must be
** closed before the method is finished.
** 006 CA 20080716 @39121 Renamed method lockUniqueIndexes to
** lockAllIndexes - this will lock all the
** indexes for a certain entity. Modified the
** delete(...) and rollbackInsert(...) methods
** so the index locking can be disabled.
** 007 ECF 20080815 @39543 Added support for global change events. These
** are events that represent index-changing
** inserts, updates, or deletes, which are
** processed by sessions other than the session
** in which the event originated.
** 008 CA 20080815 @39445 Support API change in DBUtils.
** 009 ECF 20090428 @42130 Store a copy of all inserted DMOs. This ensures a
** DMO from a lookup is in the same state it was
** when first inserted. Modified getDirtyDMO()
** accordingly.
** 010 ECF 20090512 @42157 Added assertion to getDirtyInfo().
** 011 ECF 20090513 @42169 Modified variant of update() method. Added lock
** parameter to make index locking optional.
** 012 ECF 20090601 @42571 Added support for Progress isolation leak quirk.
** Added new methods. Changed implementations of
** getDirtyDMO() and getDirtyInfo().
** 013 ECF 20090609 @42640 Modified cleanupChanges(). The list of entities
** which have had their uncommitted changes leaked
** is now passed in, allowing proper cleanup.
** 014 ECF 20090611 @42662 Optimized getDirtyDMO(). Avoid overhead of index
** locking if requested DMO is not being tracked.
** 015 ECF 20090701 @43226 Removed faulty assert statement. Added debug
** logging.
** 016 ECF 20090720 @43308 Fixed delete(). If the deleted record already is
** being tracked, it must be removed from the dirty
** database.
** 017 ECF 20090723 @43340 Fixed methods which execute queries. Parameters
** to these queries require preprocessing before
** being bound.
** 018 ECF 20100317 @44752 Added protection logic to removeRecords to avoid
** IllegalArgumentException. This is a workaround
** for a situation that causes a null DMO to come
** back from a call to session.get() for a primary
** key that should be valid. Root cause currently is
** unknown, but this change should protect against
** abend at this location.
** 019 SVL 20130331 Upgraded to Hibernate 4.
** 020 ECF 20131001 Added metadata support.
** 021 ECF 20131028 Import change.
** 022 SVL 20140210 Set JPA-style HQL parameters instead of positional.
** 023 ECF 20140825 Track unvalidated inserts. Replaced Apache commons logging with J2SE
** logging.
** 024 ECF 20150923 Return from getDirtyInfo immediately if the target entity is not
** being tracked for changes or deletes.
** 025 ECF 20160225 Changes required by new PropertyHelper implementation.
** 026 OM 20160307 Added support for EXTENTs in update() methods. In case of a
** ROWID/RECID lookup, indexes of transient record don't need to be
** matched/validated.
** 027 ECF 20160608 Removed unnecessary intern calls.
** 028 ECF 20200906 New ORM implementation.
** 029 EVL 20200621 Adding NPE protection for lockSingle when index name is NULL or empty.
** 030 OM 20201012 Force use locally cached meta information instead of map lookup.
** 031 RAA 20221221 Modified the execution of sql's. They now pass through the SQLStatementLogger
** in case logging is intended.
** 032 RAA 20230109 Changed inline statement(s) to prepared statement(s).
** 033 SR 20230510 Removed readOnly parameter.
** 034 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 035 RAA 20230518 Replaced lambdas with method referencing when using SQLStatementLogger.
** 036 SVL 20230516 Support for more granular configuration of dirty sharing functionality.
** 037 RAA 20230607 SQLs are now executed through SQLExecutor instead of SQLStatementLogger.
** 038 DDF 20230627 Replaced static variables of DirtyShareSupport with static method calls.
** 039 AL2 20230822 Removed remaining readOnly parameter from list method.
** 040 OM 20231214 Added a new parameter to Query.list() which allows the caller to select whether
** the result is a set of Records of just an array of Java plain objects.
** 041 TJD 20240123 Java 17 compatibility updates
** 042 OM 20240909 Improved Database API.
** 043 TJD 20230508 Updates needed to support dirty records leaks to other sessions on FIND
** TJD 20240109 Fixes and updates to cross section DirtyShare
** TJD 20240308 DirtyShare optimizations and simplifications
** TJD 20240629 Fix for dirtyshare database search matching few elements
** TJD 20240705 Support for double parameters for dirty share FQLs
** TJD 20240712 Fix for dirtyshare when there are no cross session leaked records
** TJD 20240719 Restore indexUpdateStates as its needed for local dirty check
** TJD 20240722 Dirty share createTable now using table name, and not entity name
** TJD 20240724 Extend indexState parameter meaning for insert method
** TJD 20240912 DirtyShareSupport flags logic dependency update
** TJD 20240912 Handle cross session leak on complete index on first insert
** 044 AL2 20240903 Added dirtyData as a marker on the data that was actually changed.
** AL2 20241025 Null-checks for dirtyData.
** 045 AL2 20250411 Update should also update the earlyInserts if used for intra-session. Otherwise,
** getDirtyInfo will provide old snapshots of data.
*/
/*
** 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.dirty;
import java.lang.reflect.*;
import java.sql.*;
import java.util.*;
import java.util.logging.*;
import com.goldencode.p2j.persist.*;
import com.goldencode.p2j.persist.Record;
import com.goldencode.p2j.persist.dialect.*;
import com.goldencode.p2j.persist.event.*;
import com.goldencode.p2j.persist.lock.*;
import com.goldencode.p2j.persist.orm.*;
import com.goldencode.p2j.util.*;
import com.goldencode.p2j.util.logging.*;
/**
* Default implementation of {@link DirtyShareManager}, to be used for local,
* primary databases (database instances for which the current P2J server is
* authoritative). Stores and shares knowledge of uncommitted modifications
* to database records, specifically, those modifications which affect
* database indexes.
* <p>
* Information about uncommitted inserts and updates to existing records is
* managed at two levels. A modified copy of the inserted or updated record
* is stored in a backing, "dirty" database. In addition, a set of in-memory
* data structures maintains information about the tables (or more accurately,
* the DMO entity names) and primary keys of affected records. The former
* allows the detailed, uncommitted index updates of a record to be queried
* and retrieved if necessary. However, the latter enables a quick, high
* level screening capability to avoid the need for the slower database access
* in many common cases.
* <p>
* Information about uncommitted deletes is managed only in an in-memory data
* structure, because it is unnecessary to store the full record in the case
* of an uncommitted delete in one context; all other contexts still have
* access to the full record. It is simply the knowledge of the delete action
* that must be managed and shared.
* <p>
* Concurrent access to the uncommitted delete information and to the high
* level screening structures for the insert and update information is
* managed via simple monitors by using synchronized blocks on 3 lock objects,
* one for each action type. In some cases, we need to synchronize on both the
* change lock and the delete lock simultaneously. In such cases, we avoid deadlock
* by always synchronizing on the change lock first, then the delete lock.
* <p>
* To maximize concurrent access when it is determined that the dirty database
* must be consulted, this implementation uses a lock manager to serialize
* read (share) and write (exclusive) access to the backing database.
* Granularity of access for record inserts and updates is determined by the
* database indexes in use for a particular update or query. For instance, if
* a query is walking a particular index for its results, only a share lock on
* that single index is acquired when checking for an uncommitted record which
* might override a query's primary result. However, when a record is
* updated, an exclusive lock must be acquired on every index affected by all
* of the updated DMO properties, before the dirty record can be updated in
* the dirty database. For a newly inserted record, every index on the
* associated table must be exclusively locked for the uncommitted insert to
* be recorded in the dirty database.
*
* @see DirtyShareContext
*/
public class DefaultDirtyShareManager
implements DirtyShareManager
{
/** Flag for cross session dirty share support being enabled */
private static final boolean crossSessionEnabled = DirtyShareSupport.isEnabledCrossSession();
/** Bogus record ID for locking at index level */
private static final Long INDEX_ID = 0L;
/** Logger */
private static final CentralLogger LOG = CentralLogger.get(DirtyShareManager.class.getName());
/** Associated dirty database */
private final Database dirtyDatabase;
/** Dialect of the dirty database */
private final Dialect dialect;
/** Helper object to lookup indexes */
private final IndexHelper indexHelper;
/** Set of DMO entity names for tables which are already created */
private final Set<String> createdTables = new HashSet<>();
/** Index lock manager */
// TODO: migrate index references to use faster, long-based RecordIdentifier for index locks
private final InMemoryLockManager<String> lockManager;
/** Index event manager */
private final GlobalEventManager eventManager;
/** Lock for changes map */
private final Object changeLock = new Object();
/** Lock for deletes map */
private final Object deleteLock = new Object();
/** Lock for inserts map */
private final Object insertLock = new Object();
/** Map of record identifiers to inserted records */
private Map<RecordIdentifier<String>, Record> earlyInserts = new HashMap<>();
/** Map of record identifiers to leaked (visible cross sessions) records
*/
private Map<RecordIdentifier<String>, Record> leaks = new HashMap<>();
/** Map of record identifiers for new inserts to set of unique indexes requiring validation */
private Map<RecordIdentifier<String>, IndexState> indexUpdateStates = new HashMap<>();
/** Dirty update ids by entity name */
private Map<String, Set<Long>> changes = new HashMap<>();
/** Dirty deleted ids by entity name */
private Map<String, Set<Long>> deletes = new HashMap<>();
/**
* Constructor. Looks up the dirty database associated with the given
* primary database, and retrieves its <code>SessionFactory</code>.
* Retrieves an <code>IndexHelper</code> instance and creates a headless
* <code>LockManager</code> to coordinate read and write access to the
* dirty database.
*
* @param database
* Primary database for which we are managing a dirty database.
*/
DefaultDirtyShareManager(Database database)
{
Database dirty = database.toType(Database.Type.DIRTY);
dialect = DatabaseManager.getDialect(dirty);
if (dialect == null)
{
throw new IllegalArgumentException("Cannot determine dialect for dirty database " + dirty);
}
this.dirtyDatabase = dirty;
this.indexHelper = IndexHelper.get(database);
this.eventManager = DirtyShareSupport.isEnabledCrossSession() &&
DirtyShareSupport.isGlobalNotificationEnabled() ?
new GlobalEventManager(database) : null;
this.lockManager = new InMemoryLockManager<String>()
{
protected boolean isHeadless() { return true; }
};
lockManager.setDatabase(dirty);
}
/**
* Register with a global event clearinghouse an interest in insert,
* delete, and update events which affect the given DMO entity.
*
* @param entity
* DMO entity for which the caller is interested in tracking
* change.
*
* @return A unique identifier for the registrant, assigned by the event
* clearinghouse, used later to collect events and deregister or
* <code>null</code> if global events are disabled.
*
* @see #getGlobalEvents(long, long)
* @see #deregisterForGlobalEvents(long)
*/
public Long registerForGlobalEvents(String entity)
{
return eventManager != null ? eventManager.register(entity) : null;
}
/**
* Deregister with a global event clearinghouse an interest in insert,
* delete, and update events which affect a particular DMO type.
*
* @param registerID
* A unique identifier for the registrant being deregistered.
*
* @see #getGlobalEvents(long, long)
* @see #registerForGlobalEvents(String)
*/
public void deregisterForGlobalEvents(long registerID)
{
if (eventManager != null)
{
eventManager.deregister(registerID);
}
}
/**
* Retrieve any new events that have been collected by a global event
* clearinghouse, representing inserts, deletes, and updates made to
* record types for which the caller previously has registered interest.
*
* @param registerID
* A unique identifier for an object which has registered interest
* in record-changing events.
* @param lastID
* A unique identifier representing the last event retrieved by
* the caller. If less than 0, it is assumed no events have yet
* been retrieved since the caller registered interest in these
* events.
*
* @return Array of applicable, new events, or <code>null</code> if no
* such events have been collected.
*
* @throws EventRegistrationException
* if <code>registerID</code> is invalid or has expired.
*/
public GlobalChangeEvent[] getGlobalEvents(long registerID, long lastID)
throws EventRegistrationException
{
return eventManager != null ? eventManager.getEvents(registerID, lastID) : null;
}
/**
* Lock or unlock all indexes associated with the given DMO entity.
*
* @param entity
* Name of an entity, which is the name of the DMO implementation
* class associated with a table being tracked for uncommitted
* changes.
* @param lockType
* Type of lock to acquire, or <code>NONE</code> to release
* existing locks. <code>NO_WAIT</code> variants will be treated
* as regular, blocking variants.
*
* @throws PersistenceException
* if there is an error querying JDBC metadata while retrieving
* index information.
*/
public void lockAllIndexes(String entity, LockType lockType)
throws PersistenceException
{
List<String> indexes = indexHelper.getIndexesForEntity(entity, false);
lockMultiple(indexes, lockType);
}
/**
* Begin tracking a newly inserted record in the dirty database.
* <p>
* This record is removed when the adding context's current transaction is
* committed or rolled back.
*
* @param entity
* Name of an entity, which is the name of the DMO implementation class associated
* with a table being tracked for uncommitted changes.
* @param copy
* Copy of the record to be tracked, ready to be inserted into the dirty database.
* @param indexState
* The index update state of the record at the time of insertion.
* If null it means all indexes are complete
* @param lock
* <code>true</code> to acquire write locks to insert the record; <code>false</code>
* to assume write locks are acquired by calling code.
* @param dirtyData
* The data that was changed and caused this update. This is explicitly sent
* because sometimes we can't infer the dirty props from the DMO anymore.
*
* @throws PersistenceException
* if there is any database error.
*/
public void insert(String entity, Record copy, IndexState indexState, boolean lock, DirtyData dirtyData)
throws PersistenceException
{
if (dirtyData != null)
{
applyUpdates(entity, copy, dirtyData);
}
List<String> lockIndexes = lock ? indexHelper.getIndexesForEntity(entity, false) : null;
trackChange(entity, copy, lockIndexes, null, true, indexState);
}
/**
* Rollback the insert of a record previously introduced with the {@link
* #insert(String, Record, IndexState, boolean, DirtyData)} method.
*
* @param ident
* Identifier for the target record, which encapsulates its entity
* name and primary key.
* @param lock
* <code>true</code> to acquire write locks;
* <code>false</code> to assume write locks are acquired by
* calling code.
*
* @throws PersistenceException
* if there is any database error.
*/
public void rollbackInsert(RecordIdentifier<String> ident, boolean lock)
throws PersistenceException
{
String entity = ident.getTable();
Long id = ident.getRecordID();
Session session = null;
boolean inTx = false;
List<String> indexes = (lock ? indexHelper.getIndexesForEntity(entity, false) : null);
try
{
if (indexes != null)
{
lockMultiple(indexes, LockType.EXCLUSIVE);
}
synchronized (insertLock)
{
earlyInserts.remove(ident);
leaks.remove(ident);
indexUpdateStates.remove(ident);
}
synchronized (changeLock)
{
Set<Long> changed = changes.get(entity);
changed.remove(id);
if (changed.isEmpty())
{
changes.remove(entity);
}
}
session = new Session(dirtyDatabase);
inTx = session.beginTransaction();
session.delete(entity, id);
session.commit();
}
catch (PersistenceException exc)
{
if (inTx)
{
session.rollback();
}
DBUtils.handleException(dirtyDatabase, exc);
throw new PersistenceException(exc);
}
finally
{
if (indexes != null)
{
lockMultiple(indexes, LockType.NONE);
}
if (session != null)
{
try
{
session.close();
}
catch (Exception exc)
{
DBUtils.handleException(dirtyDatabase, exc);
}
}
if (LOG.isLoggable(Level.FINE))
{
LOG.log(Level.FINE, "Rolled back insert of " + ident + "; currently tracking " + changes);
}
}
}
/**
* Indicate whether the record specified by the given entity and primary
* key is being tracked for changes within an uncommitted transaction.
* This includes records which have been added to the dirty database due to
* an insert or an update, not due to a delete.
*
* @param entity
* Name of an entity, which is the name of the DMO implementation
* class associated with a table being tracked for uncommitted
* changes.
* @param id
* Primary key of the record to be checked.
*
* @return <code>true</code> to indicate the target record is being
* tracked for changes, else <code>false</code>.
*/
public boolean isTracked(String entity, Long id)
{
synchronized (changeLock)
{
Set<Long> changed = changes.get(entity);
return (changed != null && changed.contains(id));
}
}
/**
* Begin tracking a DMO for changes, based on the updates indicated by the
* {@code properties}, {@code extIndexes} and {@code values} arguments.
* <p>
* Make a copy of the given record, apply the specified updates to that
* copy, and store the copy in the dirty database.
* <p>
* This record is removed when the adding context's current transaction is
* committed or rolled back.
*
* @param entity
* Name of an entity, which is the name of the DMO implementation
* class associated with a table being tracked for uncommitted changes.
* @param copy
* Copy of the record to be tracked.
* @param lock
* <code>true</code> to acquire write locks to update the record;
* <code>false</code> to assume write locks are acquired by calling code.
* @param dirtyData
* The data that was changed and caused this update. This is explicitly sent
* because sometimes we can't infer the dirty props from the DMO anymore.
*
* @throws PersistenceException
* if there is any database error.
*/
public void update(String entity,
Record copy,
boolean lock,
DirtyData dirtyData)
throws PersistenceException
{
if (dirtyData == null)
{
return;
}
applyUpdates(entity, copy, dirtyData);
List<String> indexes = lock ? indexesForProperties(entity, dirtyData.getProperties()) : null;
trackChange(entity, copy, indexes, dirtyData.getProperties(), false, null);
}
/**
* Retrieve a DMO which is being tracked in the dirty database, and apply the specified updates
* to it, as provided by the {@code properties}, {@code extIndexes} and {@code values}
* arguments. It is assumed the DMO already exists in the database, from a previous call to
* {@link #update(String, Record, boolean, DirtyData)}.
*
* @param entity
* Name of an entity, which is the name of the DMO implementation class associated
* with a table being tracked for uncommitted changes.
* @param id
* Primary key id of the record to be retrieved and updated.
* @param validatedTransientIndex
* {@code true} if the update was caused by an index validation of a transient record.
* In this case the record will be updated in {@code inserts} structure so the current
* state will be visible to other contexts.
* @param lock
* {@code true} to acquire write locks to update the record;
* {@code false} to assume write locks are acquired by calling code.
* @param dirtyData
* The data that was changed and caused this update. This is explicitly sent
* because sometimes we can't infer the dirty props from the DMO anymore.
*
* @throws PersistenceException
* if there is any database error.
*/
@Override
public void update(String entity,
Long id,
boolean validatedTransientIndex,
boolean lock,
DirtyData dirtyData)
throws PersistenceException
{
if (dirtyData == null)
{
return;
}
List<String> indexes = lock ? indexesForProperties(entity, dirtyData.getProperties()) : null;
Session session = null;
boolean inTx = false;
try
{
if (lock)
{
lockMultiple(indexes, LockType.EXCLUSIVE);
}
session = new Session(dirtyDatabase);
inTx = session.beginTransaction();
Record copy = session.get(entity, id);
applyUpdates(entity, copy, dirtyData);
session.save(copy, true);
session.commit();
if (!crossSessionEnabled)
{
// intra-session should always "see" the latest version of the DMO
RecordIdentifier<String> ident = new RecordIdentifier<>(entity, id);
synchronized (insertLock)
{
earlyInserts.put(ident, copy);
}
}
if (eventManager != null)
{
eventManager.enqueueUpdate(entity, copy, dirtyData.getProperties());
}
synchronized (changeLock)
{
Set<Long> changed = changes.computeIfAbsent(entity, k -> new HashSet<>());
changed.add(id);
}
}
catch (PersistenceException exc)
{
if (inTx)
{
session.rollback();
}
DBUtils.handleException(dirtyDatabase, exc);
throw new PersistenceException(exc);
}
finally
{
if (lock)
{
lockMultiple(indexes, LockType.NONE);
}
if (session != null)
{
try
{
session.close();
}
catch (Exception exc)
{
DBUtils.handleException(dirtyDatabase, exc);
}
}
}
}
/**
* Indicate whether any uncommitted modification is being tracked for the
* database table associated with the specified entity. A modification is
* the insertion, deletion, or update of a record. Insertions and updates
* considered the same for purposes of this method.
*
* @param entity
* Name of an entity, which is the name of the DMO implementation
* class associated with a table being tracked for uncommitted
* changes.
*
* @return Bitfield indicating dirty status of entity, using constants
* from {@link DirtyInfo}. Zero return indicates entity is not
* dirty. Only <code>CHANGE</code> and <code>DELETE</code> are
* detected, not <code>INSERT</code>.
*/
public int getEntityDirtyState(String entity)
{
int state = DirtyInfo.NONE;
synchronized (changeLock)
{
if (changes.containsKey(entity))
{
state |= DirtyInfo.CHANGE;
}
synchronized (deleteLock)
{
if (deletes.containsKey(entity))
{
state |= DirtyInfo.DELETE;
}
}
}
return state;
}
/**
* Indicate whether the record specified by the given entity and primary
* key has been deleted within an uncommitted transaction in <i>any</i>
* context.
*
* @param entity
* Name of an entity, which is the name of the DMO implementation
* class associated with a table being tracked for uncommitted
* changes.
* @param id
* Primary key of the record to be checked.
*
* @return <code>true</code> if the specified record has been deleted,
* else <code>false</code>.
*/
public boolean isDeleted(String entity, Long id)
{
synchronized (deleteLock)
{
Set<Long> deleted = deletes.get(entity);
return (deleted != null && deleted.contains(id));
}
}
/**
* Mark a record as having been deleted in an uncommitted transaction, or
* unmark a record which previously was so marked.
* <p>
* A record is unmarked when the adding context's current transaction is
* committed or rolled back.
*
* @param entity
* Name of an entity, which is the name of the DMO implementation
* class associated with a table being tracked for uncommitted
* changes.
* @param id
* Primary key of deleted record.
* @param rollback
* If <code>false</code>, the target record is marked as deleted;
* if <code>true</code>, it is unmarked.
* @param lock
* <code>true</code> to acquire write locks to delete the record;
* <code>false</code> to assume write locks are acquired by
* calling code.
*
* @throws PersistenceException
* if there was an error deleting the record from the database.
*/
public void delete(String entity, Long id, boolean rollback, boolean lock)
throws PersistenceException
{
synchronized (deleteLock)
{
Set<Long> deleted = deletes.get(entity);
if (rollback)
{
if (deleted != null)
{
deleted.remove(id);
}
}
else
{
if (deleted == null)
{
deleted = new HashSet<>();
deletes.put(entity, deleted);
}
deleted.add(id);
}
}
if (!rollback)
{
boolean purgeFromDB = false;
synchronized (changeLock)
{
Set<Long> changed = changes.get(entity);
purgeFromDB = (changed != null && changed.remove(id));
}
if (purgeFromDB)
{
RecordIdentifier<String> ident = new RecordIdentifier<>(entity, id);
synchronized (insertLock)
{
earlyInserts.remove(ident);
leaks.remove(ident);
indexUpdateStates.remove(ident);
}
String hql = "delete from " + entity +
" where " + DatabaseManager.PRIMARY_KEY + " = ?";
Session session = null;
boolean inTx = false;
try
{
session = new Session(dirtyDatabase);
inTx = session.beginTransaction();
Query query = Session.createQuery(hql);
query.setParameter(0, id);
query.executeUpdate(session);
session.commit();
}
catch (PersistenceException exc)
{
if (inTx)
{
session.rollback();
}
throw new PersistenceException("Error deleting dirty record", exc);
}
finally
{
if (session != null)
{
try
{
session.close();
}
catch (PersistenceException exc)
{
if (LOG.isLoggable(Level.SEVERE))
{
LOG.log(Level.SEVERE, "Error closing session", exc);
}
}
}
}
}
if (eventManager != null)
{
eventManager.enqueueDelete(entity, id);
}
}
}
/**
* Retrieve from the dirty database the record specified by the given
* entity and primary key, if it exists.
* <p>
* This method temporarily places a share lock on all indexes containing
* the primary key column for the target table.
*
* @param entity
* Name of an entity, which is the name of the DMO implementation
* class associated with a table being tracked for uncommitted
* changes.
* @param id
* Primary key of the record to be retrieved.
*
* @return Target record if it exists in the dirty database, else
* <code>null</code>.
*
* @throws PersistenceException
* never.
*/
public Record getDirtyDMO(String entity, Long id)
throws PersistenceException
{
// Quick out if DMO is not being tracked. Avoids overhead of index locking.
if (!isTracked(entity, id))
{
return null;
}
// shortcut for records that are not yet locked - DirtyDatabase records for unlocked records is mirrored
// in early inserts map, so try to retrieve the DMO as a newly inserted record first. This is
// faster than going to the database.
synchronized(insertLock)
{
RecordIdentifier<String> ident = new RecordIdentifier<>(entity, id);
if (earlyInserts.containsKey(ident))
{
return earlyInserts.get(ident);
}
}
List<String> indexes = indexHelper.getIndexesForProperty(entity, Session.PK);
Record copy = null;
try
{
lockMultiple(indexes, LockType.SHARE);
// Check again, in case status changed before we could get lock(s).
if (!isTracked(entity, id))
{
return null;
}
// Look in the dirty database for the record database
Session session = null;
boolean inTx = false;
try
{
session = new Session(dirtyDatabase);
inTx = session.beginTransaction();
copy = session.get(entity, id);
session.commit();
}
catch (PersistenceException exc)
{
if (inTx)
{
session.rollback();
}
DBUtils.handleException(dirtyDatabase, exc);
throw new PersistenceException(exc);
}
finally
{
if (session != null)
{
try
{
session.close();
}
catch (Exception exc)
{
DBUtils.handleException(dirtyDatabase, exc);
}
}
}
}
finally
{
lockMultiple(indexes, LockType.NONE);
}
return copy;
}
/**
* Given certain search criteria and possibly the primary key of a potential DMO match in the
* primary database, report any significant information currently being tracked for uncommitted
* transactions, which might override data (or the lack thereof) found in the primary database.
* <p>
* If a {@link DirtyInfo} object is returned by this method, it will contain overriding
* information, such as a dirty DMO which matched the search criteria, whether that record was
* newly inserted, or the fact that the candidate record found in the primary search has been
* deleted or modified in an uncommitted transaction. A {@code null} return indicates no
* overriding information was available.
* <p>
* This method acquires a share lock on {@code index}.
*
* @param entity
* Name of an entity, which is the name of the DMO implementation class associated
* with a table being tracked for uncommitted changes.
* @param candidateID
* Primary key ID of DMO, if any, found in the primary database, which matches the
* current query criteria.
* @param index
* UID of index which governs the ordering of data in the current query.
* @param statements
* One or more HQL statements to be executed against the dirty database to find a
* record which matches the current query criteria. The list is in order of most
* specific to least specific search criteria.
* @param params
* Query substitution parameters which match placeholders in the
* most specific HQL statement.
* @param rowidLookup
* The predicate of the query does a lookup for ROWID/RECID of the table. In case of
* transient records they are accessible without the need of having a full matched
* index.
*
* @return An object which summarizes the findings, or {@code null} if no uncommitted change
* information was available for the given arguments.
*
* @throws PersistenceException
* if there was any database error.
*/
@Override
public DirtyInfo getDirtyInfo(String entity,
Long candidateID,
int index,
List<String> statements,
Object[] params,
boolean rowidLookup)
throws PersistenceException
{
// quick out if we're not tracking this entity
synchronized (changeLock)
{
if (!changes.containsKey(entity))
{
synchronized (deleteLock)
{
if (!deletes.containsKey(entity))
{
return null;
}
}
}
}
DirtyInfo info = null;
Session session = null;
boolean inTx = false;
Record dmo = null;
String indexName = entity + index;
try
{
lockSingle(indexName, LockType.SHARE);
session = new Session(dirtyDatabase);
inTx = session.beginTransaction();
int fullCount = params.length;
int callCount = statements.size();
Object[] processedParams = (fullCount == 0 ? params : preprocessQueryParameters(params));
// there are cases where there are many DirtyDMOs that may match query criteria, but only some of them
// are visible to other sessions. Thats why we need to iterate results and find first matching
// dirty DMO from inserts.
List<Object> results = null;
for (int i = 0; dmo == null && i < callCount; i++)
{
// each query requires 2 parameters, so we should take 2 parameters each iteration
int usedCount = fullCount - (i * 2);
Object[] usedParams = new Object[usedCount];
System.arraycopy(processedParams, 0, usedParams, 0, usedCount);
String hql = statements.get(i);
Query query = Session.createQuery(hql);
for (int j = 0; j < usedParams.length; j++)
{
query.setParameter(j, usedParams[j]);
}
results = query.list(session);
if (results != null && results.size() > 0)
{
break;
}
}
session.commit();
if (results != null)
{
Record copy = null;
for(Object result:results)
{
if (result != null)
{
// retrieve a full Record depending on query.list result
if (result instanceof Record)
{
dmo = (Record) result;
}
else if (result instanceof Long)
{
Long id = (Long) result;
dmo = session.get(entity, id);
}
else if (result instanceof Object[])
{
Object [] idArray = (Object[]) result;
if (idArray.length > 0)
{
if (idArray[0] instanceof Long)
{
Long id = (Long) idArray[0];
dmo = session.get(entity, id);
}
}
}
// if DMO is found check cross session leaking conditions
// there are two cases where records are leaking cross sessions:
// - when all fields forming indices for a newly created record are set making all indices
// complete (that also means that unique indices are validated)
// in that case the previous version of a record becomes visible to other sessions
// via leaked table
// - when generic FIND NE no ROWID or FIND EQUAL with ROWID is used on a record buffer - in
// that case record is leaked into leaks table
if (dmo != null)
{
Long id = dmo.primaryKey();
// check if given record is leaked from some other session
// if so - its the right record to send back
// it may be also updated version of a candidate record
RecordIdentifier<String> ident = new RecordIdentifier<>(entity, id);
synchronized (insertLock)
{
IndexState indexState = indexUpdateStates.get(ident);
if (!rowidLookup && indexState != null && indexState.get(index) && !leaks.containsKey(ident))
{
// if this index was not yet updated in the new record, then the record is not visible
// under this index yet, including for the context that manages it
continue;
}
// first check for local session records if we have found anything in master database
// otherwise no further iteration will happen
if (candidateID != null)
{
copy = earlyInserts.get(ident);
if (copy != null)
{
info = new DirtyInfo();
info.setDirtyDMOs(copy, copy);
break;
}
}
// then check for cross session records
copy = leaks.get(ident);
if (copy != null)
{
info = new DirtyInfo();
info.setLeaked();
info.setDirtyDMOs(copy, dmo);
break;
}
}
// If the record is not newly inserted, make a copy of the first
// record; callers may have to retrieve the appropriate version
// of the data from the primary database, as this copy may contain
// newer information than those contexts should be able to "see".
if (copy == null)
{
info = new DirtyInfo();
copy = dmo.snapshot();
info.setDirtyDMOs(copy, copy);
}
}
}
}
}
if (candidateID != null)
{
synchronized (changeLock)
{
Set<Long> changed = changes.get(entity);
if (changed != null && changed.contains(candidateID))
{
if (info == null)
{
info = new DirtyInfo();
}
info.setModified();
}
}
}
}
catch (PersistenceException exc)
{
if (inTx)
{
session.rollback();
}
DBUtils.handleException(dirtyDatabase, exc);
throw new PersistenceException(exc);
}
finally
{
lockSingle(indexName, LockType.NONE);
if (session != null)
{
try
{
session.close();
}
catch (Exception exc)
{
DBUtils.handleException(dirtyDatabase, exc);
}
}
}
return info;
}
/**
* Execute an HQL query and return the results as a list. If the query returns multiple results
* per row, each element in the list will be an array of {@code Object}s.
* <p>
* No locking is attempted.
*
* @param fql
* FQL query statement. This method makes no assumptions as to the types of object(s) returned.
* @param params
* Substitution values for the query. If none, this should be an empty array.
* @param maxResults
* The maximum number of elements to be returned in the list. If this value is
* non-positive, no upper limit is applied.
* @param startOffset
* The 0-based offset of the first record to retrieve. If this value is non-positive,
* an offset of 0 is used by default.
*
* @return A list of results, or {@code null} if no result was found.
*
* @throws PersistenceException
* if there was an error executing the query.
*/
@Override
public <T> List<T> list(String fql, Object[] params, int maxResults, int startOffset)
throws PersistenceException
{
Session session = null;
boolean inTx = false;
try
{
session = new Session(dirtyDatabase);
inTx = session.beginTransaction();
Query query = Session.createQuery(fql);
Object[] processedParams = (params.length == 0 ? params : preprocessQueryParameters(params));
for (int i = 0; i < processedParams.length; i++)
{
query.setParameter(i, processedParams[i]);
}
if (maxResults > 0)
{
query.setMaxResults(maxResults);
}
if (startOffset > 0)
{
query.setFirstResult(startOffset);
}
List<T> results = query.list(session, true);
session.commit();
return (results == null || results.isEmpty() ? null : results);
}
catch (PersistenceException exc)
{
if (inTx)
{
session.rollback();
}
DBUtils.handleException(dirtyDatabase, exc);
throw new PersistenceException(exc);
}
finally
{
if (session != null)
{
try
{
session.close();
}
catch (Exception exc)
{
DBUtils.handleException(dirtyDatabase, exc);
}
}
}
}
/**
* Stop tracking uncommitted changes to the the subset of records
* represented by <code>subChanges</code>, and remove from the master set
* of uncommitted, inserted records the subset of records represented by
* <code>subInserts</code>.
*
* @param subChanges
* Sets of primary keys representing those records to be removed
* from uncommitted change tracking, mapped by entity name.
* @param subInserts
* A set of identifiers of those records to be removed from the
* master set of uncommitted inserts.
*
* @throws PersistenceException
* if there is any database error.
*/
public void cleanupChanges(Map<String, Set<Long>> subChanges, Set<RecordIdentifier<String>> subInserts)
throws PersistenceException
{
if (subChanges == null)
{
return;
}
try
{
Iterator<Map.Entry<String, Set<Long>>> iter = subChanges.entrySet().iterator();
while (iter.hasNext())
{
Map.Entry<String, Set<Long>> entry = iter.next();
String entity = entry.getKey();
Set<Long> ids = entry.getValue();
removeRecords(entity, ids);
}
}
finally
{
if (subInserts != null)
{
// first need to clean leaks
synchronized (insertLock)
{
// clear inserts and reinitialize its HashMap if possible
if (earlyInserts.keySet().removeAll(subInserts) && earlyInserts.isEmpty())
{
earlyInserts = new HashMap<>();
}
// clear leaks
if (leaks.keySet().removeAll(subInserts) && leaks.isEmpty())
{
leaks = new HashMap<>();
}
// clear index updates
if (indexUpdateStates.keySet().removeAll(subInserts) && indexUpdateStates.isEmpty())
{
indexUpdateStates = new HashMap<>();
}
}
}
}
}
/**
* Remove from the master set of uncommitted, deleted records the subset of
* records represented by <code>subDeletes</code>.
*
* @param subDeletes
* Sets of primary keys representing those records to be removed
* from uncommitted delete tracking, mapped by entity name.
*/
public void cleanupDeletes(Map<String, Set<Long>> subDeletes)
{
if (subDeletes == null)
{
return;
}
synchronized (deleteLock)
{
Iterator<Map.Entry<String, Set<Long>>> iter = subDeletes.entrySet().iterator();
while (iter.hasNext())
{
Map.Entry<String, Set<Long>> entry = iter.next();
String entity = entry.getKey();
Set<Long> ids = entry.getValue();
Set<Long> deleted = deletes.get(entity);
deleted.removeAll(ids);
if (deleted.isEmpty())
{
deletes.remove(entity);
}
}
if (deletes.isEmpty())
{
deletes = new HashMap<>();
}
}
}
/**
* Stop tracking uncommitted changes to the the specified set of records
* associated with the given entity name.
*
* @param entity
* Name of an entity, which is the name of the DMO implementation
* class associated with a table being tracked for uncommitted
* changes.
* @param ids
* Set of primary key IDs of the records to be removed from
* tracking.
*
* @throws PersistenceException
* if there is any database error.
*/
private void removeRecords(String entity, Set<Long> ids)
throws PersistenceException
{
List<String> indexes = indexHelper.getIndexesForEntity(entity, false);
Session session = null;
boolean inTx = false;
try
{
// Acquire the necessary write locks.
lockMultiple(indexes, LockType.EXCLUSIVE);
session = new Session(dirtyDatabase);
inTx = session.beginTransaction();
for (Long id : ids)
{
if (!session.delete(entity, id) && LOG.isLoggable(Level.WARNING))
{
LOG.log(Level.WARNING,
"Attempt to remove " + entity + "#" + id + " from dirty database " +
dirtyDatabase + " failed; record not found in session");
}
}
session.commit();
}
catch (PersistenceException exc)
{
if (inTx)
{
session.rollback();
}
DBUtils.handleException(dirtyDatabase, exc);
throw new PersistenceException(exc);
}
finally
{
synchronized (changeLock)
{
Set<Long> changed = changes.get(entity);
if (changed != null)
{
changed.removeAll(ids);
if (changed.isEmpty())
{
changes.remove(entity);
}
}
if (changes.isEmpty())
{
changes = new HashMap<>();
}
}
// Release locks.
lockMultiple(indexes, LockType.NONE);
if (session != null)
{
try
{
session.close();
}
catch (Exception exc)
{
DBUtils.handleException(dirtyDatabase, exc);
}
}
}
}
/**
* Attempt to acquire a lock of the type specified on a single index, or to
* release an existing lock.
* <p>
* If invoked with a <code>lockType</code> that is not <code>NONE</code>,
* this method will block if a target lock is currently unavailable
* (<code>NO_WAIT</code> lock type variants are treated as regular,
* blocking lock types).
*
* @param name
* Name of the index to lock.
* @param lockType
* Type of lock to acquire, or <code>NONE</code> to release an
* existing lock. <code>NO_WAIT</code> variants will be treated
* as regular, blocking variants.
*
* @see com.goldencode.p2j.persist.lock.LockManager
*/
private void lockSingle(String name, LockType lockType)
{
try
{
RecordIdentifier<String> ident = new RecordIdentifier<>(name, INDEX_ID);
lockManager.lock(lockType.toWaitVariant(), ident, true);
}
catch (LockUnavailableException exc)
{
// ok to ignore; never thrown in this situation
}
}
/**
* Attempts to acquire multiple locks of the type specified, on the indexes
* represented by <code>names</code>, or to release existing locks on those
* indexes.
* <p>
* If invoked with a <code>lockType</code> that is not <code>NONE</code>,
* this method will block if a target lock is currently unavailable
* (<code>NO_WAIT</code> lock type variants are treated as regular,
* blocking lock types).
*
* @param names
* Names of the indexes to lock. To prevent deadlocks, this list
* should be consistently ordered for calls made across different
* contexts. An ascending alphabetical sort is recommended.
* @param lockType
* Type of lock to acquire, or <code>NONE</code> to release
* existing locks. <code>NO_WAIT</code> variants will be treated
* as regular, blocking variants.
*/
private void lockMultiple(List<String> names, LockType lockType)
{
try
{
lockType = lockType.toWaitVariant();
for (String name : names)
{
RecordIdentifier<String> ident = new RecordIdentifier<>(name, INDEX_ID);
lockManager.lock(lockType, ident, true);
}
}
catch (LockUnavailableException exc)
{
// ok to ignore; never thrown in this situation
}
}
/**
* Retrieve the names of all indexes in which each of the given DMO
* properties participate and merge them into a single list sorted in
* ascending alphabetical order, discarding duplicates.
*
* @param entity
* Name of an entity, which is the name of the DMO implementation
* class associated with a table being tracked for uncommitted
* changes.
* @param properties
* Array of property names for which index names are to be
* retrieved.
*
* @return Alphabetically ascending list of index names.
*
* @throws PersistenceException
* if an error occurs querying index metadata.
*/
private List<String> indexesForProperties(String entity, String[] properties)
throws PersistenceException
{
List<String> list = null;
int len = properties.length;
if (len == 1)
{
// Only a single list.
list = indexHelper.getIndexesForProperty(entity, properties[0]);
}
else
{
// Multiple lists which must be merged. Duplicates are dropped and
// the list is sorted alphabetically to prevent deadlocks.
SortedSet<String> set = new TreeSet<>();
for (String property : properties)
{
list = indexHelper.getIndexesForProperty(entity, property);
set.addAll(list);
}
list = new ArrayList<>(set);
}
return list;
}
/**
* Begin tracking changes which occur in the calling context's current
* transaction for the given DMO. This entails adding the DMO (a copy of
* the original) into the dirty database and storing its primary key ID in
* the mapping of changed records. If the record was newly inserted into
* its primary database, that fact is recorded as well.
* <p>
* This method acquires exclusive locks on all the indexes in the
* <code>indexes</code> parameter.
*
* @param entity
* Name of an entity, which is the name of the DMO implementation
* class associated with a table being tracked for uncommitted
* changes.
* @param copy
* Copy of the DMO being tracked.
* @param lockIndexes
* List of indexes to lock while recording the uncommitted changes. If
* <code>null</code>, no locking is attempted.
* @param properties
* Array of names of those DMO properties which have been updated.
* @param insert
* <code>true</code> if the record was newly inserted into its
* primary database, <code>false</code> if the record previously
* existed in the primary database and is simply being updated
* now.
* @param indexState
* The index update state of the record at the time of insertion.
*
* @throws PersistenceException
* if there is any database error.
*/
private void trackChange(String entity,
Record copy,
List<String> lockIndexes,
String[] properties,
boolean insert,
IndexState indexState)
throws PersistenceException
{
Long id = copy.primaryKey();
Session session = null;
boolean inTx = false;
boolean updateDirtyDatabase = false;
RecordIdentifier<String> ident = new RecordIdentifier<>(entity, id);
boolean isNewDirtyDMO = false;
try
{
synchronized (changeLock)
{
Set<Long> changed = changes.computeIfAbsent(entity, k -> new HashSet<>());
changed.add(id);
}
if (insert)
{
// is insert table updated so notifications of insert are to be sent out
boolean insertsUpdated = false;
synchronized (insertLock)
{
if (!leaks.containsKey(ident) && !earlyInserts.containsKey(ident))
{
isNewDirtyDMO = true;
}
if (indexState != null)
{
indexUpdateStates.put(ident, indexState);
}
// given insert is locked if insertVisible is set to true
boolean insertLocked = leaks.containsKey(ident);
// if all indices are empty its possible to lock given insert
boolean lockInserts = (!insertLocked && (indexState == null || indexState.isEmpty())) && crossSessionEnabled;
// otherwise - if given insert is not locked it should be updated with the provided value
boolean updateInserts = !lockInserts && !insertLocked || !crossSessionEnabled;
updateDirtyDatabase = updateInserts || lockInserts || insertLocked;
if (lockInserts)
{
Record previousDmo = earlyInserts.remove(ident);
if (previousDmo == null)
{
previousDmo = copy;
}
leaks.put(ident, previousDmo);
insertsUpdated = true;
}
else if (updateInserts)
{
// indices are still not complete, just track another insert as they can be complete anytime soon
updateDirtyDatabase = true;
earlyInserts.put(ident, copy);
leaks.remove(ident);
insertsUpdated = true;
}
}
if (insertsUpdated && eventManager != null)
{
eventManager.enqueueInsert(entity, copy);
}
}
else
{
synchronized (insertLock)
{
earlyInserts.put(ident, copy);
leaks.remove(ident);
updateDirtyDatabase = true;
}
if (eventManager != null)
{
eventManager.enqueueUpdate(entity, copy, properties);
}
}
// update DirtyDatabase if needed
if (updateDirtyDatabase)
{
createTable(entity);
if (lockIndexes != null)
{
lockMultiple(lockIndexes, LockType.EXCLUSIVE);
}
if (isNewDirtyDMO)
{
copy.updateState(null, DmoState.NEW, true);
}
session = new Session(dirtyDatabase);
inTx = session.beginTransaction();
session.save(copy, false);
session.commit();
if (isNewDirtyDMO)
{
copy.updateState(null, DmoState.NEW, false);
}
}
}
catch (PersistenceException exc)
{
if (inTx)
{
session.rollback();
}
DBUtils.handleException(dirtyDatabase, exc);
throw new PersistenceException(exc);
}
finally
{
if (lockIndexes != null)
{
lockMultiple(lockIndexes, LockType.NONE);
}
if (session != null)
{
try
{
session.close();
}
catch (Exception exc)
{
DBUtils.handleException(dirtyDatabase, exc);
}
}
}
}
/**
* Apply the given property updates to the given DMO using reflection.
*
* @param entity
* Name of an entity, which is the name of the DMO implementation class associated
* with a table being tracked for uncommitted changes.
* @param dmo
* DMO being updated.
* @param dirtyData
* The data that was changed and caused this update. This is explicitly sent
* because sometimes we can't infer the dirty props from the DMO anymore.
*
* @throws PersistenceException
* if there is any error applying updates.
*/
private void applyUpdates(String entity, Record dmo, DirtyData dirtyData)
throws PersistenceException
{
String[] properties = dirtyData.getProperties();
int[] extIndexes = dirtyData.getExtIndexes();
BaseDataType[] values = dirtyData.getValues();
Map<String, Method> setters = PropertyHelper.pojoSettersByProperty(entity);
try
{
int len = properties.length;
for (int i = 0; i < len; i++)
{
String property = properties[i];
BaseDataType value = values[i];
Method method = setters.get(property);
if (extIndexes[i] != -1)
{
method.invoke(dmo, extIndexes[i], value);
}
else
{
method.invoke(dmo, value);
}
}
}
catch (Exception exc)
{
throw new PersistenceException("Error storing dirty update(s)", exc);
}
}
/**
* Preprocess an array of query parameters appropriate to the dirty database's dialect.
*
* @param params
* Query parameters. If none, this should be a 0-length array.
*
* @return Preprocessed query parameters.
*/
private Object[] preprocessQueryParameters(Object[] params)
{
Object[] args = params;
int length = params.length;
if (length > 0)
{
args = new Object[length];
for (int i = 0; i < length; i++)
{
args[i] = Persistence.preprocessQueryParameter(params[i], dialect);
}
}
return args;
}
/**
* Given the name of an entity representing a database table and DMO type,
* this method will create the "dirty" table and its associated indexes,
* within the associated "dirty" database. If the table already exists,
* this method does nothing.
*
* @param entity
* Name of an entity, which is the name of the DMO implementation
* class associated with the table being created.
*
* @throws PersistenceException
* if there is any database error.
*/
private void createTable(String entity)
throws PersistenceException
{
DmoMeta dmoInfo = DmoMetadataManager.getDmoInfo(entity, null);
String tableName = dmoInfo.getSqlTableName();
synchronized (createdTables)
{
if (!createdTables.add(tableName))
{
// Already created; nothing to do.
return;
}
Class<? extends DataModelObject> dmoIface = dmoInfo.getAnnotatedInterface();
Session session = null;
Connection conn = null;
Statement stmt = null;
boolean inTx = false;
try
{
session = new Session(dirtyDatabase);
inTx = session.beginTransaction();
conn = session.getConnection();
stmt = conn.createStatement();
// Create table(s).
Iterator<String> tableIter = TempTableHelper.sqlTempTableCreate(dirtyDatabase, dmoIface);
while (tableIter.hasNext())
{
String sql = tableIter.next();
SQLExecutor.getInstance().execute(dirtyDatabase, Statement::execute, stmt, sql);
}
// Create indexes (if any).
Iterator<String> indexIter = TempTableHelper.sqlTempTableIndexCreate(dirtyDatabase, dmoIface);
while (indexIter.hasNext())
{
String sql = indexIter.next();
SQLExecutor.getInstance().execute(dirtyDatabase, Statement::execute, stmt, sql);
}
// Commit the changes.
session.commit();
}
catch (Exception exc)
{
if (inTx)
{
session.rollback();
}
createdTables.remove(tableName);
String msg = "Unable to create table for entity '" +
entity +
"' in database '" +
dirtyDatabase +
"'";
DBUtils.handleException(dirtyDatabase, exc);
throw new PersistenceException(msg, exc);
}
finally
{
if (stmt != null)
{
try
{
stmt.close();
}
catch (SQLException exc)
{
DBUtils.handleException(dirtyDatabase, exc);
if (LOG.isLoggable(Level.WARNING))
{
LOG.log(Level.WARNING, "Error", exc);
}
}
}
if (session != null)
{
try
{
session.close();
}
catch (Exception exc)
{
DBUtils.handleException(dirtyDatabase, exc);
if (LOG.isLoggable(Level.WARNING))
{
LOG.log(Level.WARNING, "Error", exc);
}
}
}
}
}
}
/**
* Update DMO snapshots to be leaked to other sessions
*
* @param entity
* Name of an entity, which is the name of the DMO implementation
* class associated with the table being created.
* @param dmos
* DMOs to be leaked, not NULL
* @throws PersistenceException
* if there is any database error.
*/
@Override
public void updateSnapshots(String entity, Set<Record> dmos)
throws PersistenceException
{
if (dmos == null || !crossSessionEnabled)
{
return;
}
for(Record dmo:dmos)
{
// store DMO snapshot to be leaked to be used later
if (!dmo.checkState(DmoState.COPY))
{
// make a snapshot if DMO is not already a copy
dmo = dmo.snapshot();
}
List<String> lockIndexes = indexHelper.getIndexesForEntity(entity, false);
try {
//trackChange(entity, dmo, indicesToLock, null, true, null);
Long id = dmo.primaryKey();
Session session = null;
boolean inTx = false;
try
{
createTable(entity);
if (lockIndexes != null)
{
lockMultiple(lockIndexes, LockType.EXCLUSIVE);
}
session = new Session(dirtyDatabase);
inTx = session.beginTransaction();
session.save(dmo, false);
session.commit();
synchronized (changeLock)
{
Set<Long> changed = changes.computeIfAbsent(entity, k -> new HashSet<>());
changed.add(id);
RecordIdentifier<String> ident = new RecordIdentifier<>(entity, id);
synchronized (insertLock)
{
leaks.put(ident, dmo);
}
}
}
catch (PersistenceException exc)
{
if (inTx)
{
session.rollback();
}
DBUtils.handleException(dirtyDatabase, exc);
throw new PersistenceException(exc);
}
finally
{
if (lockIndexes != null)
{
lockMultiple(lockIndexes, LockType.NONE);
}
if (session != null)
{
try
{
session.close();
}
catch (Exception exc)
{
DBUtils.handleException(dirtyDatabase, exc);
}
}
}
}
finally
{
if (lockIndexes != null)
{
lockMultiple(lockIndexes, LockType.NONE);
}
}
}
}
}