UserTableStatUpdater.java
/*
** Module : UserTableStatUpdater.java
** Abstract : Manages _UserTableStat VST
**
** Copyright (c) 2020-2025, Golden Code Development Corporation.
**
** -#- -I- --Date-- --------------------------------------Description----------------------------------------
** 001 IAS 20200819 Created initial version.
** 002 IAS 20210401 Fixed bugs and added new helper methods.
** ECF 20210910 Refactored to remove upstream synchronization when called from ConnectTableUpdater, and
** to avoid using context-local functionality when invoked during context cleanup.
** ECF 20220630 Fixed MinimalUsertablestat getter/setter names to match conversion of historical names in
** _UserTableStat table. Fixed hard-coded SQL to reflect corresponding changes in column
** names.
** TJD 20220504 Upgrade do Java 11 minor changes
** 003 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 004 GBB 20230825 SecurityManager session methods calls updated.
** 005 TJD 20240123 Java 17 compatibility updates.
** 006 OM 20240909 Improved Database API.
** 007 OM 20241128 Multi-tenant runtime support: selected the proper persistence context, eventually
** based on [sharedDb] parameter.
** 008 ICP 20250123 Used integer constants to leverage caches instances.
*/
/*
** 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.meta;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.logging.*;
import com.goldencode.p2j.persist.*;
import com.goldencode.p2j.persist.Record;
import com.goldencode.p2j.persist.meta.TableStats.*;
import com.goldencode.p2j.persist.orm.*;
import com.goldencode.p2j.security.SecurityManager;
import com.goldencode.p2j.util.*;
import com.goldencode.p2j.util.logging.*;
/**
* Maintains the CRUD operations' statistics in the {@code _usertablestat} VST.
*
* For performance reasons we do not update the table immediately when the value of one of
* its counters is updated. This will be too expensive since the counters are updated very often
* but their values are retrieved rarely (if retrieved at all).
*
* We keep the up-to-date values of the counters in memory (see {@code tableStats} map) and flush
* them to the underlying table only when the application tries to retrieve the table data.
*
* The flushes are "incremental", so we maintain the the 'version' of the statistics
* in-memory record and the version of the most recently persisted state (see {@code TableState}.
*
* An additional optimization is achieved my maintaining the versions of the in-memory and
* persisted statistics. This allows us in some situation to completely avoid iteration over
* in-memory data and {@code persist} just returns.
*/
public class UserTableStatUpdater
{
/** Logger */
private static final CentralLogger log = CentralLogger.get(UserTableStatUpdater.class);
/** Max batch size for commit */
private static final int MAX_BATCH = 100;
/** Usertablestat meta-table name */
private static final String USERTABLESTAT_TABLE = "meta_usertablestat";
/** Is user table stat metadata in use? */
private static final boolean isUserTableStats = MetadataManager.inUse(MetadataManager._USERTABLESTAT);
/** Instances of this class, by primary database */
private static final Map<Database, UserTableStatUpdater> instances = new ConcurrentHashMap<>();
/** MetaUsertablestat DMO wrapper */
// TODO: this assumes a particular DMO interface name conversion, which can change with configuration;
// we should only allow a single, deterministic name conversion for the metaschema tables
private final MetaTableWrapper<MinimalUsertablestat> usertablestatWrapper =
new MetaTableWrapper<>(MinimalUsertablestat.class, "MetaUsertablestat");
/** MetaUsertablestatImpl DMO proxy */
private final MinimalUsertablestat usertablestat = usertablestatWrapper.getMetaTableProxy();
/** meta database */
private final Database metaDatabase;
/** meta database Persistence */
private final Persistence persistence;
/** Security manager */
private final SecurityManager securityManager;
/** Tables' CRUD operations statistics per connection */
private final ConcurrentMap<integer, ConcurrentMap<integer, TableStats>> tableStats =
new ConcurrentSkipListMap<>();
/** Persisted statistics version */
private final AtomicLong databaseVersion = new AtomicLong(-1);
/** In-memory statistics version */
private final AtomicLong statisticsVersion = new AtomicLong(0);
/** Next available primary key for a new _UserTableStat metadata record */
private final AtomicLong nextPrimaryKey = new AtomicLong(1);
/**
* Get the instance of this class associated with the given metadata database, creating and mapping it
* to the database first, if necessary.
*
* @param database
* Primary database whose stats are tracked by the associated instance.
*
* @return The instance of this class associated with the given database.
*/
public static UserTableStatUpdater get(Database database)
{
if (!isUserTableStats)
{
return null;
}
return instances.computeIfAbsent(database, d -> new UserTableStatUpdater(d.toType(Database.Type.META)));
}
/**
* Persist statistics in the meta database
*
* @param database
* Primary database.
*/
public static void persist(Database database)
{
if (isUserTableStats)
{
try
{
UserTableStatUpdater.get(database).persist();
}
catch (PersistenceException e)
{
if (log.isLoggable(Level.WARNING))
{
log.log(Level.WARNING, "Failed to flush _UserTableStat", e);
}
}
}
}
/**
* Register create operation.
*
* @param db
* Primary database.
* @param buffer
* target buffer
*/
public static void create(Database db, RecordBuffer buffer)
{
stat(db, buffer).ifPresent(TableStats::create);
}
/**
* Register retrieve operation.
*
* @param db
* Primary database.
* @param buffer
* target buffer
*/
public static void retrieve(Database db, RecordBuffer buffer)
{
stat(db, buffer).ifPresent(TableStats::retrieve);
}
/**
* Register update operation
*
* @param db
* Primary database.
* @param buffer
* target buffer
*/
public static void update(Database db, RecordBuffer buffer)
{
stat(db, buffer).ifPresent(TableStats::update);
}
/**
* Register delete operation.
*
* @param db
* Primary database.
* @param buffer
* target buffer
*/
public static void delete(Database db, RecordBuffer buffer)
{
stat(db, buffer).ifPresent(TableStats::delete);
}
/**
* Retrieve the table statistics record.
*
* @param db
* Primary database
* @param buffer
* target buffer
*
* @return table statistics record
*/
private static Optional<TableStats> stat(Database db, RecordBuffer buffer)
{
if (!isUserTableStats)
{
return Optional.empty();
}
DmoMeta dmoMeta = buffer.getDmoInfo();
if (dmoMeta.isMeta())
{
return Optional.empty();
}
Class<? extends Record> dmoClass = dmoMeta.getImplementationClass();
integer filenum = MetadataManager.getFileNum(db, dmoClass);
if (filenum == null)
{
return Optional.empty();
}
UserTableStatUpdater utsu = get(db);
return Optional.ofNullable(
utsu.getTableStats(integer.of(utsu.getSecurityManager().sessionSm.getSessionId()), filenum)
);
}
/**
* Execute action uninterruptibly.
*
* @param action
* action to be executed
*
* @throws Exception
*/
private static void uninterruptibly(Action action)
throws Exception
{
Thread worker = Thread.currentThread();
log.fine("Executing on " + worker.getId() + "/" + worker.getName());
boolean interrupted = false;
while (true)
{
interrupted = Thread.interrupted();
try
{
action.exec();
break;
}
catch (Throwable t)
{
Throwable e = t;
while(e != null && !(e instanceof InterruptedException))
{
e = e.getCause();
}
if (e == null)
{
throw t;
}
log.fine("InterruptedException caught, retrying");
continue;
}
}
if (interrupted)
{
worker.interrupt();
}
}
/**
* Get the SecurityManager instance.
*
* @return SecurityManager instance
*/
public SecurityManager getSecurityManager()
{
return securityManager;
}
/**
* Constructor.
*
* @param metaDatabase
* meta database
*/
public UserTableStatUpdater(Database metaDatabase)
{
this.metaDatabase = metaDatabase;
this.persistence = PersistenceFactory.getInstance(metaDatabase);
this.securityManager = SecurityManager.getInstance();
}
/**
* Get table statistics holder for a give connection and table
*
* @param connectId
* connection id
* @param tableNo
* table number
*
* @return Table statistics holder.
*/
public TableStats getTableStats(integer connectId, integer tableNo)
{
if (connectId == null)
{
return null;
}
return tableStats.computeIfAbsent(
connectId,
k -> new ConcurrentSkipListMap<>()).computeIfAbsent(tableNo,
k -> new TableStats(statisticsVersion));
}
/**
* Persist statistics in the meta database
*
* @return {@code true} if success.
*
* @throws PersistenceException
* If a persistence error is encountered during execution of the method.
*/
public boolean persist()
throws PersistenceException
{
long statVersion = statisticsVersion.get();
if (statVersion <= databaseVersion.get())
{
return true;
}
try
{
boolean commit = persistence.beginTransaction(Persistence.META_CTX);
Map<integer, Map<integer, Long>> updates = new HashMap<>();
new Worker().run(updates);
persistence.flush(Persistence.META_CTX);
if (commit)
{
persistence.commit(Persistence.META_CTX);
}
databaseVersion.set(statVersion);
// update "persisted" value
for (Map.Entry<integer, Map<integer, Long>> ce : updates.entrySet())
{
ConcurrentMap<integer, TableStats> cs = tableStats.get(ce.getKey());
if (cs == null)
{
continue;
}
for (Map.Entry<integer, Long> te : ce.getValue().entrySet())
{
TableStats ts = cs.get(te.getKey());
if (ts != null)
{
ts.persisted(te.getValue());
}
}
}
return true;
}
catch (PersistenceException e)
{
persistence.rollback(Persistence.META_CTX);
return false;
}
}
/**
* Drop data for the closed connection.
*
* @param connectId
* Connection ID.
*/
public void disconnected(Session session, Integer connectId)
{
try
{
uninterruptibly(() -> disconnected(session, new integer(connectId)));
if (log.isLoggable(Level.FINE))
{
log.fine("disconnected for " + session.getDatabase().getId() + "/" + connectId);
}
}
catch (Exception exc)
{
if (log.isLoggable(Level.WARNING))
{
log.log(Level.WARNING,
"Failed to cleanup statistics for the closed connection: " +
session.getDatabase().getId() + "/" + connectId.intValue(),
exc);
}
}
}
/**
* Drop data for the closed connection.
*
* @param session
* Database session.
* @param connectId
* Connection ID.
*
* @throws PersistenceException
* if an error occurs executing the database update.
*/
public void disconnected(Session session, int64 connectId)
throws PersistenceException
{
tableStats.remove(connectId);
String sql = "DELETE FROM " + USERTABLESTAT_TABLE + " WHERE user_table_stat_conn = ?1";
SQLQuery query = Session.createSQLQuery(sql);
query.setParameter(0, connectId.intValue());
query.executeUpdate(session);
}
/**
* Implements incremental update of the _UserTableStat table
*/
private class Worker
{
/**
* Perform incremental database update
*
* @param updates
* version of the updated/inserted records
*
* @throws PersistenceException
*/
public void run(Map<integer, Map<integer, Long>> updates)
throws PersistenceException
{
for (Map.Entry<integer, ConcurrentMap<integer, TableStats>> ce : tableStats.entrySet())
{
integer connId = ce.getKey();
for (Map.Entry<integer, TableStats> te : ce.getValue().entrySet())
{
Snapshot data = te.getValue().getStat();
if (data.version <= data.persisted)
{
continue;
}
integer tblId = te.getKey();
Record dmo = data.persisted < 0
? createRecord(connId, tblId, data)
: updateRecord(connId, tblId, data);
updates.computeIfAbsent(connId, k -> new HashMap<>()).put(tblId, data.version);
persistence.save(dmo, dmo.primaryKey());
}
}
}
/**
* Update _usertablestat record
*
* @param connId
* connection id
* @param tblId
* table id
* @param data
* data snapshot
* @return updated
* Record object
*
* @throws PersistenceException
*/
private Record updateRecord(integer connId, integer tblId, Snapshot data)
throws PersistenceException
{
String sql = "SELECT USER_TABLE_STAT_ID FROM "
+ USERTABLESTAT_TABLE + " WHERE "
+ "USER_TABLE_STAT_CONN = ?1 AND USER_TABLE_STAT_NUM = ?2";
Long id = persistence.getSingleSQLResult(sql,
Persistence.META_CTX,
new Object[] { connId.longValue(), tblId.longValue() });
Record dmo = null;
if (id != null)
{
RecordIdentifier<String> metaIdent =
new RecordIdentifier<>(usertablestatWrapper.getDmoClass().getName(), id);
dmo = persistence.quickLoad(metaIdent, Persistence.META_CTX);
}
if (dmo == null)
{
throw new PersistenceException(String.format(
"No _usertablestat record found for connId = %s, tblId = %s",
connId.longValue(),
tblId.longValue()));
}
usertablestatWrapper.setDelegate(dmo);
MinimalUsertablestat proxy = usertablestatWrapper.getMetaTableProxy();
populateStat(data, proxy);
return dmo;
}
/**
* Creates an {@code _usertablestat} record based on passed parameters.
*
* @param connId
* The connection id.
* @param tblId
* The table id.
* @param data
* The data snapshot.
*
* @return created
* The newly created {@code Record} object.
*
* @throws PersistenceException
* if a persistence error occurs in the process.
*/
private Record createRecord(integer connId, integer tblId, Snapshot data)
throws PersistenceException
{
Record dmo = null;
try
{
dmo = usertablestatWrapper.getDmoClass().getDeclaredConstructor().newInstance();
dmo.initialize(null, true);
long id = nextPrimaryKey.getAndIncrement();
dmo.primaryKey(id);
usertablestatWrapper.setDelegate(dmo);
MinimalUsertablestat proxy = usertablestatWrapper.getMetaTableProxy();
proxy.setUserTableStatId(new integer(id));
proxy.setUserTableStatConn(connId);
proxy.setUserTableStatNum(tblId);
populateStat(data, proxy);
}
catch (ReflectiveOperationException e)
{
throw new PersistenceException("Error creating meta usertablestat record", e);
}
return dmo;
}
/**
* Populate a statistics DMO.
*
* @param data
* Table statistics snapshot.
* @param proxy
* Proxy used to populate the DMO.
*/
private void populateStat(Snapshot data, MinimalUsertablestat proxy)
{
proxy.setUserTableStatCreate(new integer(data.create));
proxy.setUserTableStatRead(new integer(data.retrieve));
proxy.setUserTableStatUpdate(new integer(data.update));
proxy.setUserTableStatDelete(new integer(data.delete));
}
}
/**
* Interface which defines the minimally required methods of the
* {@code MinimalUsertablestat} interface. This interface defines the proxy
* used to create and manipulate <codeMinimalUsertablestatImpl</code DMO
* instances. The DMO's methods cannot be invoked directly from code here, since
* the {@code MinimalUsertablestat} interface resides in an
* application-specific package, which is not known at compile time.
*/
public static interface MinimalUsertablestat
extends Persistable
{
/**
* Getter: UserTableStat-Id
*
* @return UserTableStat-Id
*/
public int64 getUserTableStatId();
/**
* Setter: UserTableStat-Id
*
* @param userTableStatId
* UserTableStat-Id
*/
public void setUserTableStatId(NumberType userTableStatId);
/**
* Getter: Connection
*
* @return Connection
*/
public integer getUserTableStatConn();
/**
* Setter: Connection
*
* @param userTableStatConn
* Connection
*/
public void setUserTableStatConn(NumberType userTableStatConn);
/**
* Getter: Table #
*
* @return Table #
*/
public integer getUserTableStatNum();
/**
* Setter: Table #
*
* @param userTableStatNum
* Table #
*/
public void setUserTableStatNum(NumberType userTableStatNum);
/**
* Getter: read
*
* @return read
*/
public int64 getUserTableStatRead();
/**
* Setter: read
*
* @param userTableStatRead
* read
*/
public void setUserTableStatRead(NumberType userTableStatRead);
/**
* Getter: update
*
* @return update
*/
public int64 getUserTableStatUpdate();
/**
* Setter: update
*
* @param userTableStatUpdate
* update
*/
public void setUserTableStatUpdate(NumberType userTableStatUpdate);
/**
* Getter: create
*
* @return create
*/
public int64 getUserTableStatCreate();
/**
* Setter: create
*
* @param userTableStatCreate
* create
*/
public void setUserTableStatCreate(NumberType userTableStatCreate);
/**
* Getter: delete
*
* @return delete
*/
public int64 getUserTableStatDelete();
/**
* Setter: delete
*
* @param userTableStatDelete
* delete
*/
public void setUserTableStatDelete(NumberType userTableStatDelete);
/**
* Getter: Partition Id
*
* @return Partition Id
*/
public integer getUserTableStatPartitionId();
/**
* Setter: Partition Id
*
* @param userTableStatPartitionId
* Partition Id
*/
public void setUserTableStatPartitionId(NumberType userTableStatPartitionId);
/**
* Getter: Tenant Id
*
* @return Tenant Id
*/
public integer getUserTableStatTenantId();
/**
* Setter: Tenant Id
*
* @param userTableStatTenantId
* Tenant Id
*/
public void setUserTableStatTenantId(NumberType userTableStatTenantId);
}
/**
* Runnable which can throw an Exception.
*/
private static interface Action
{
/**
* Do the job.
*
* @throws Exception
*/
public void exec() throws Exception;
}
}