--- /dev/null
+.idea
+target
+mdbc.iml
//TODO: This is a place holder for taking snapshots and moving data from redo record into actual tables
/**
- * This method is called whenever there is a DELETE on the transaction digest and should be called when ownership changes, if required
- * It updates the MUSIC/Cassandra tables (both dirty bits and actual data) corresponding to the SQL DELETE.
- * Music propagates it to the other replicas.
+ * This method is called whenever there is a DELETE on a local SQL table, and should be called by the underlying databases
+ * triggering mechanism. It updates the MUSIC/Cassandra tables (both dirty bits and actual data) corresponding to the SQL DELETE.
+ * Music propagates it to the other replicas. If the local database is in the middle of a transaction, the DELETEs to MUSIC are
+ * delayed until the transaction is either committed or rolled back.
* @param tableName This is the table on which the select is being performed
* @param oldRow This is information about the row that is being deleted
*/
}
/**
- * This method is called whenever there is an INSERT or UPDATE to a the transaction digest, and should be called by an
- * ownership chance. It updates the MUSIC/Cassandra tables (both dirty bits and actual data) corresponding to the SQL write.
+ * This method is called whenever there is an INSERT or UPDATE to a local SQL table, and should be called by the underlying databases
+ * triggering mechanism. It updates the MUSIC/Cassandra tables (both dirty bits and actual data) corresponding to the SQL write.
* Music propagates it to the other replicas. If the local database is in the middle of a transaction, the updates to MUSIC are
* delayed until the transaction is either committed or rolled back.
*
return id;
}
+ /**
+ * This function creates the TransactionInformation table. It contain information related
+ * to the transactions happening in a given partition.
+ * * The schema of the table is
+ * * Id, uiid.
+ * * Partition, uuid id of the partition
+ * * LatestApplied, int indicates which values from the redologtable wast the last to be applied to the data tables
+ * * Applied: boolean, indicates if all the values in this redo log table where already applied to data tables
+ * * Redo: list of uiids associated to the Redo Records Table
+ *
+ */
+ public static void CreateTransactionInformationTable( String musicNamespace, String transactionInformationTableName) throws MDBCServiceException {
+ String tableName = transactionInformationTableName;
+ String priKey = "id";
+ StringBuilder fields = new StringBuilder();
+ fields.append("id uuid, ");
+ fields.append("partition uuid, ");
+ fields.append("latestapplied int, ");
+ fields.append("applied boolean, ");
+ //TODO: Frozen is only needed for old versions of cassandra, please update correspondingly
+ fields.append("redo list<frozen<tuple<text,tuple<text,varint>>>> ");
+ String cql = String.format("CREATE TABLE IF NOT EXISTS %s.%s (%s, PRIMARY KEY (%s));", musicNamespace, tableName, fields, priKey);
+ try {
+ executeMusicWriteQuery(musicNamespace,tableName,cql);
+ } catch (MDBCServiceException e) {
+ logger.error("Initialization error: Failure to create transaction information table");
+ throw(e);
+ }
+ }
-
-
+ /**
+ * This function creates the RedoRecords table. It contain information related to each transaction committed
+ * * LeaseId: id associated with the lease, text
+ * * LeaseCounter: transaction number under this lease, bigint \TODO this may need to be a varint later
+ * * TransactionDigest: text that contains all the changes in the transaction
+ */
+ public static void CreateRedoRecordsTable(int redoTableNumber, String musicNamespace, String redoRecordTableName) throws MDBCServiceException {
+ String tableName = redoRecordTableName;
+ if(redoTableNumber >= 0) {
+ StringBuilder table = new StringBuilder();
+ table.append(tableName);
+ table.append("-");
+ table.append(Integer.toString(redoTableNumber));
+ tableName=table.toString();
+ }
+ String priKey = "leaseid,leasecounter";
+ StringBuilder fields = new StringBuilder();
+ fields.append("leaseid text, ");
+ fields.append("leasecounter varint, ");
+ fields.append("transactiondigest text ");//notice lack of ','
+ String cql = String.format("CREATE TABLE IF NOT EXISTS %s.%s (%s, PRIMARY KEY (%s));", musicNamespace, tableName, fields, priKey);
+ try {
+ executeMusicWriteQuery(musicNamespace,tableName,cql);
+ } catch (MDBCServiceException e) {
+ logger.error("Initialization error: Failure to create redo records table");
+ throw(e);
+ }
+ }
/**
* This function creates the Table To Partition table. It contain information related to
}
}
}
-
- /**
- * This function creates the MusicTxDigest table. It contain information related to each transaction committed
- * * LeaseId: id associated with the lease, text
- * * LeaseCounter: transaction number under this lease, bigint \TODO this may need to be a varint later
- * * TransactionDigest: text that contains all the changes in the transaction
- */
- public static void CreateMusicTxDigest(int musicTxDigestTableNumber, String musicNamespace, String musicTxDigestTableName) throws MDBCServiceException {
- String tableName = musicTxDigestTableName;
- if(musicTxDigestTableNumber >= 0) {
- StringBuilder table = new StringBuilder();
- table.append(tableName);
- table.append("-");
- table.append(Integer.toString(musicTxDigestTableNumber));
- tableName=table.toString();
- }
- String priKey = "leaseid,leasecounter";
- StringBuilder fields = new StringBuilder();
- fields.append("leaseid text, ");
- fields.append("leasecounter varint, ");
- fields.append("transactiondigest text ");//notice lack of ','
- String cql = String.format("CREATE TABLE IF NOT EXISTS %s.%s (%s, PRIMARY KEY (%s));", musicNamespace, tableName, fields, priKey);
- try {
- executeMusicWriteQuery(musicNamespace,tableName,cql);
- } catch (MDBCServiceException e) {
- logger.error("Initialization error: Failure to create redo records table");
- throw(e);
- }
- }
-
- /**
- * This function creates the TransactionInformation table. It contain information related
- * to the transactions happening in a given partition.
- * * The schema of the table is
- * * Id, uiid.
- * * Partition, uuid id of the partition
- * * LatestApplied, int indicates which values from the redologtable wast the last to be applied to the data tables
- * * Applied: boolean, indicates if all the values in this redo log table where already applied to data tables
- * * Redo: list of uiids associated to the Redo Records Table
- *
- */
- public static void CreateMusicRangeInformationTable(String musicNamespace, String musicRangeInformationTableName) throws MDBCServiceException {
- String tableName = musicRangeInformationTableName;
- String priKey = "id";
- StringBuilder fields = new StringBuilder();
- fields.append("id uuid, ");
- fields.append("partition uuid, ");
- fields.append("latestapplied int, ");
- fields.append("applied boolean, ");
- //TODO: Frozen is only needed for old versions of cassandra, please update correspondingly
- fields.append("redo list<frozen<tuple<text,tuple<text,varint>>>> ");
- String cql = String.format("CREATE TABLE IF NOT EXISTS %s.%s (%s, PRIMARY KEY (%s));", musicNamespace, tableName, fields, priKey);
- try {
- executeMusicWriteQuery(musicNamespace,tableName,cql);
- } catch (MDBCServiceException e) {
- logger.error("Initialization error: Failure to create transaction information table");
- throw(e);
- }
- }
-
-
}
import java.util.Set;
import com.att.research.logging.EELFLoggerDelegate;
+import com.att.research.mdbc.mixins.CassandraMixin;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
public class DatabasePartition {
private transient static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(DatabasePartition.class);
- private String musicRangeInformationTable;//Table that currently contains the REDO log for this partition
- private String musicRangeInformationIndex;//Index that can be obtained either from
- private String musicTxDigestTable;
+ private String transactionInformationTable;//Table that currently contains the REDO log for this partition
+ private String transactionInformationIndex;//Index that can be obtained either from
+ private String redoRecordsTable;
private String partitionId;
private String lockId;
protected Set<Range> ranges;
ranges = new HashSet<>();
}
- public DatabasePartition(Set<Range> knownRanges, String mriIndex, String mriTable, String partitionId, String lockId, String musicTxDigestTable) {
+ public DatabasePartition(Set<Range> knownRanges, String titIndex, String titTable, String partitionId, String lockId, String redoRecordsTable) {
if(knownRanges != null) {
ranges = knownRanges;
}
ranges = new HashSet<>();
}
- if(musicTxDigestTable != null) {
- this.setMusicTxDigestTable(musicTxDigestTable);
+ if(redoRecordsTable != null) {
+ this.setRedoRecordsTable(redoRecordsTable);
}
else{
- this.setMusicTxDigestTable("");
+ this.setRedoRecordsTable("");
}
- if(mriIndex != null) {
- this.setMusicRangeInformationIndex(mriIndex);
+ if(titIndex != null) {
+ this.setTransactionInformationIndex(titIndex);
}
else {
- this.setMusicRangeInformationIndex("");
+ this.setTransactionInformationIndex("");
}
- if(mriTable != null) {
- this.setMusicRangeInformationTable(mriTable);
+ if(titTable != null) {
+ this.setTransactionInformationTable(titTable);
}
else {
- this.setMusicRangeInformationTable("");
+ this.setTransactionInformationTable("");
}
if(partitionId != null) {
}
}
- public String getMusicRangeInformationTable() {
- return musicRangeInformationTable;
+ public String getTransactionInformationTable() {
+ return transactionInformationTable;
}
- public void setMusicRangeInformationTable(String musicRangeInformationTable) {
- this.musicRangeInformationTable = musicRangeInformationTable;
+ public void setTransactionInformationTable(String transactionInformationTable) {
+ this.transactionInformationTable = transactionInformationTable;
}
- public String getMusicRangeInformationIndex() {
- return musicRangeInformationIndex;
+ public String getTransactionInformationIndex() {
+ return transactionInformationIndex;
}
- public void setMusicRangeInformationIndex(String musicRangeInformationIndex) {
- this.musicRangeInformationIndex = musicRangeInformationIndex;
+ public void setTransactionInformationIndex(String transactionInformationIndex) {
+ this.transactionInformationIndex = transactionInformationIndex;
}
/**
this.lockId = lockId;
}
- public String getMusicTxDigestTable() {
- return musicTxDigestTable;
+ public String getRedoRecordsTable() {
+ return redoRecordsTable;
}
- public void setMusicTxDigestTable(String musicTxDigestTable) {
- this.musicTxDigestTable = musicTxDigestTable;
+ public void setRedoRecordsTable(String redoRecordsTable) {
+ this.redoRecordsTable = redoRecordsTable;
}
}
package com.att.research.mdbc;
import java.sql.Connection;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Properties;
+import java.util.Set;
import org.json.JSONObject;
*/
public class MusicSqlManager {
- private static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(MusicSqlManager.class);
+ private static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(MusicSqlManager.class);
+
+ private final DBInterface dbi;
+ private final MusicInterface mi;
+ private final Set<String> table_set;
+ private final HashMap<Range,StagingTable> transactionDigest;
+ private boolean autocommit; // a copy of the autocommit flag from the JDBC Connection
- private final DBInterface dbi;
- private final MusicInterface mi;
- private final Set<String> table_set;
- private final HashMap<Range, StagingTable> transactionDigest;
- private boolean autocommit; // a copy of the autocommit flag from the JDBC Connection
-
- /**
- * Build a MusicSqlManager for a DB connection. This construct may only be called by getMusicSqlManager(),
- * which will ensure that only one MusicSqlManager is created per URL.
- * This is the location where the appropriate mixins to use for the MusicSqlManager should be determined.
- * They should be picked based upon the URL and the properties passed to this constructor.
- * <p>
- * At the present time, we only support the use of the H2Mixin (for access to a local H2 database),
- * with the CassandraMixin (for direct access to a Cassandra noSQL DB as the persistence layer).
- * </p>
- *
- * @param url the JDBC URL which was used to connection to the database
- * @param conn the actual connection to the database
- * @param info properties passed from the initial JDBC connect() call
- * @throws MDBCServiceException
- */
- public MusicSqlManager(String url, Connection conn, Properties info, MusicInterface mi) throws MDBCServiceException {
- try {
- info.putAll(Utils.getMdbcProperties());
- String mixinDb = info.getProperty(Configuration.KEY_DB_MIXIN_NAME, Configuration.DB_MIXIN_DEFAULT);
- this.dbi = MixinFactory.createDBInterface(mixinDb, this, url, conn, info);
- this.mi = mi;
- this.table_set = Collections.synchronizedSet(new HashSet<String>());
- this.autocommit = true;
- this.transactionDigest = new HashMap<Range, StagingTable>();
-
- } catch (Exception e) {
- throw new MDBCServiceException(e.getMessage());
- }
- }
-
- public void setAutoCommit(boolean b, String txId, TxCommitProgress progressKeeper, DatabasePartition partition) throws MDBCServiceException {
- if (b != autocommit) {
- autocommit = b;
- logger.debug(EELFLoggerDelegate.applicationLogger, "autocommit changed to " + b);
- if (b) {
- // My reading is that turning autoCOmmit ON should automatically commit any outstanding transaction
- if (txId == null || txId.isEmpty()) {
- logger.error(EELFLoggerDelegate.errorLogger, "Connection ID is null", AppMessages.UNKNOWNERROR, ErrorSeverity.CRITICAL, ErrorTypes.QUERYERROR);
- throw new MDBCServiceException("tx id is null");
- }
- commit(txId, progressKeeper, partition);
- }
- }
- }
+ /**
+ * Build a MusicSqlManager for a DB connection. This construct may only be called by getMusicSqlManager(),
+ * which will ensure that only one MusicSqlManager is created per URL.
+ * This is the location where the appropriate mixins to use for the MusicSqlManager should be determined.
+ * They should be picked based upon the URL and the properties passed to this constructor.
+ * <p>
+ * At the present time, we only support the use of the H2Mixin (for access to a local H2 database),
+ * with the CassandraMixin (for direct access to a Cassandra noSQL DB as the persistence layer).
+ * </p>
+ *
+ * @param url the JDBC URL which was used to connection to the database
+ * @param conn the actual connection to the database
+ * @param info properties passed from the initial JDBC connect() call
+ * @throws MDBCServiceException
+ */
+ public MusicSqlManager(String url, Connection conn, Properties info, MusicInterface mi) throws MDBCServiceException {
+ try {
+ info.putAll(Utils.getMdbcProperties());
+ String mixinDb = info.getProperty(Configuration.KEY_DB_MIXIN_NAME, Configuration.DB_MIXIN_DEFAULT);
+ this.dbi = MixinFactory.createDBInterface(mixinDb, this, url, conn, info);
+ this.mi = mi;
+ this.table_set = Collections.synchronizedSet(new HashSet<String>());
+ this.autocommit = true;
+ this.transactionDigest = new HashMap<Range,StagingTable>();
- /**
- * Close this MusicSqlManager.
- */
- public void close() {
- if (dbi != null) {
- dbi.close();
- }
- }
+ }catch(Exception e) {
+ throw new MDBCServiceException(e.getMessage());
+ }
+ }
- /**
- * Code to be run within the DB driver before a SQL statement is executed. This is where tables
- * can be synchronized before a SELECT, for those databases that do not support SELECT triggers.
- *
- * @param sql the SQL statement that is about to be executed
- */
- public void preStatementHook(final String sql) {
- dbi.preStatementHook(sql);
- }
+ public void setAutoCommit(boolean b,String txId, TxCommitProgress progressKeeper, DatabasePartition partition) throws MDBCServiceException {
+ if (b != autocommit) {
+ autocommit = b;
+ logger.debug(EELFLoggerDelegate.applicationLogger,"autocommit changed to "+b);
+ if (b) {
+ // My reading is that turning autoCOmmit ON should automatically commit any outstanding transaction
+ if(txId == null || txId.isEmpty()) {
+ logger.error(EELFLoggerDelegate.errorLogger, "Connection ID is null",AppMessages.UNKNOWNERROR, ErrorSeverity.CRITICAL, ErrorTypes.QUERYERROR);
+ throw new MDBCServiceException("tx id is null");
+ }
+ commit(txId,progressKeeper,partition);
+ }
+ }
+ }
- /**
- * Code to be run within the DB driver after a SQL statement has been executed. This is where remote
- * statement actions can be copied back to Cassandra/MUSIC.
- *
- * @param sql the SQL statement that was executed
- */
- public void postStatementHook(final String sql) {
- dbi.postStatementHook(sql, transactionDigest);
- }
+ /**
+ * Close this MusicSqlManager.
+ */
+ public void close() {
+ if (dbi != null) {
+ dbi.close();
+ }
+ }
- /**
- * Synchronize the list of tables in SQL with the list in MUSIC. This function should be called when the
- * proxy first starts, and whenever there is the possibility that tables were created or dropped. It is synchronized
- * in order to prevent multiple threads from running this code in parallel.
- */
- public synchronized void synchronizeTables() throws QueryException {
- Set<String> set1 = dbi.getSQLTableSet(); // set of tables in the database
- logger.debug(EELFLoggerDelegate.applicationLogger, "synchronizing tables:" + set1);
- for (String tableName : set1) {
- // This map will be filled in if this table was previously discovered
- if (!table_set.contains(tableName) && !dbi.getReservedTblNames().contains(tableName)) {
- logger.info(EELFLoggerDelegate.applicationLogger, "New table discovered: " + tableName);
- try {
- TableInfo ti = dbi.getTableInfo(tableName);
- mi.initializeMusicForTable(ti, tableName);
- //\TODO Verify if table info can be modify in the previous step, if not this step can be deleted
- ti = dbi.getTableInfo(tableName);
- mi.createDirtyRowTable(ti, tableName);
- dbi.createSQLTriggers(tableName);
- table_set.add(tableName);
- synchronizeTableData(tableName);
- logger.debug(EELFLoggerDelegate.applicationLogger, "synchronized tables:" +
- table_set.size() + "/" + set1.size() + "tables uploaded");
- } catch (Exception e) {
- logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), AppMessages.UNKNOWNERROR, ErrorSeverity.CRITICAL, ErrorTypes.QUERYERROR);
- //logger.error(EELFLoggerDelegate.errorLogger, "Exception synchronizeTables: "+e);
- throw new QueryException();
- }
- }
- }
+ /**
+ * Code to be run within the DB driver before a SQL statement is executed. This is where tables
+ * can be synchronized before a SELECT, for those databases that do not support SELECT triggers.
+ * @param sql the SQL statement that is about to be executed
+ */
+ public void preStatementHook(final String sql) {
+ dbi.preStatementHook(sql);
+ }
+ /**
+ * Code to be run within the DB driver after a SQL statement has been executed. This is where remote
+ * statement actions can be copied back to Cassandra/MUSIC.
+ * @param sql the SQL statement that was executed
+ */
+ public void postStatementHook(final String sql) {
+ dbi.postStatementHook(sql,transactionDigest);
+ }
+ /**
+ * Synchronize the list of tables in SQL with the list in MUSIC. This function should be called when the
+ * proxy first starts, and whenever there is the possibility that tables were created or dropped. It is synchronized
+ * in order to prevent multiple threads from running this code in parallel.
+ */
+ public synchronized void synchronizeTables() throws QueryException {
+ Set<String> set1 = dbi.getSQLTableSet(); // set of tables in the database
+ logger.debug(EELFLoggerDelegate.applicationLogger, "synchronizing tables:" + set1);
+ for (String tableName : set1) {
+ // This map will be filled in if this table was previously discovered
+ if (!table_set.contains(tableName) && !dbi.getReservedTblNames().contains(tableName)) {
+ logger.info(EELFLoggerDelegate.applicationLogger, "New table discovered: "+tableName);
+ try {
+ TableInfo ti = dbi.getTableInfo(tableName);
+ mi.initializeMusicForTable(ti,tableName);
+ //\TODO Verify if table info can be modify in the previous step, if not this step can be deleted
+ ti = dbi.getTableInfo(tableName);
+ mi.createDirtyRowTable(ti,tableName);
+ dbi.createSQLTriggers(tableName);
+ table_set.add(tableName);
+ synchronizeTableData(tableName);
+ logger.debug(EELFLoggerDelegate.applicationLogger, "synchronized tables:" +
+ table_set.size() + "/" + set1.size() + "tables uploaded");
+ } catch (Exception e) {
+ logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.UNKNOWNERROR, ErrorSeverity.CRITICAL, ErrorTypes.QUERYERROR);
+ //logger.error(EELFLoggerDelegate.errorLogger, "Exception synchronizeTables: "+e);
+ throw new QueryException();
+ }
+ }
+ }
// Set<String> set2 = getMusicTableSet(music_ns);
- // not working - fix later
+ // not working - fix later
// for (String tbl : set2) {
// if (!set1.contains(tbl)) {
// logger.debug("Old table dropped: "+tbl);
// // ZZTODO drop camunda table ?
// }
// }
- }
-
- /**
- * On startup, copy dirty data from Cassandra to H2. May not be needed.
- *
- * @param tableName
- */
- public void synchronizeTableData(String tableName) {
- // TODO - copy MUSIC -> H2
- dbi.synchronizeData(tableName);
- }
+ }
- /**
- * This method is called whenever there is a SELECT on a local SQL table, and should be called by the underlying databases
- * triggering mechanism. It first checks the local dirty bits table to see if there are any keys in Cassandra whose value
- * has not yet been sent to SQL. If there are, the appropriate values are copied from Cassandra to the local database.
- * Under normal execution, this function behaves as a NOP operation.
- *
- * @param tableName This is the table on which the SELECT is being performed
- */
- public void readDirtyRowsAndUpdateDb(String tableName) {
- mi.readDirtyRowsAndUpdateDb(dbi, tableName);
- }
-
-
- /**
- * This method gets the primary key that the music interfaces uses by default.
- * If the front end uses a primary key, this will not match what is used in the MUSIC interface
- *
- * @return
- */
- public String getMusicDefaultPrimaryKeyName() {
- return mi.getMusicDefaultPrimaryKeyName();
- }
-
- /**
- * Asks music interface to provide the function to create a primary key
- * e.g. uuid(), 1, "unique_aksd419fjc"
- *
- * @return
- */
- public String generateUniqueKey() {
- //
- return mi.generateUniqueKey();
- }
-
-
- /**
- * Perform a commit, as requested by the JDBC driver. If any row updates have been delayed,
- * they are performed now and copied into MUSIC.
- *
- * @throws MDBCServiceException
- */
- public synchronized void commit(String txId, TxCommitProgress progressKeeper, DatabasePartition partition) throws MDBCServiceException {
- logger.debug(EELFLoggerDelegate.applicationLogger, " commit ");
- // transaction was committed -- add all the updates into the REDO-Log in MUSIC
- try {
- mi.commitLog(dbi, partition, transactionDigest, txId, progressKeeper);
- } catch (MDBCServiceException e) {
- logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), AppMessages.QUERYERROR, ErrorTypes.QUERYERROR, ErrorSeverity.CRITICAL);
- throw e;
- }
- }
-
- /**
- * Perform a rollback, as requested by the JDBC driver. If any row updates have been delayed,
- * they are discarded.
- */
- public synchronized void rollback() {
- // transaction was rolled back - discard the updates
- logger.debug(EELFLoggerDelegate.applicationLogger, "Rollback");
- ;
- transactionDigest.clear();
- }
-
- /**
- * Get all
- *
- * @param table
- * @param dbRow
- * @return
- */
- public String getMusicKeyFromRowWithoutPrimaryIndexes(String table, JSONObject dbRow) {
- TableInfo ti = dbi.getTableInfo(table);
- return mi.getMusicKeyFromRowWithoutPrimaryIndexes(ti, table, dbRow);
- }
+ /**
+ * On startup, copy dirty data from Cassandra to H2. May not be needed.
+ * @param tableName
+ */
+ public void synchronizeTableData(String tableName) {
+ // TODO - copy MUSIC -> H2
+ dbi.synchronizeData(tableName);
+ }
+ /**
+ * This method is called whenever there is a SELECT on a local SQL table, and should be called by the underlying databases
+ * triggering mechanism. It first checks the local dirty bits table to see if there are any keys in Cassandra whose value
+ * has not yet been sent to SQL. If there are, the appropriate values are copied from Cassandra to the local database.
+ * Under normal execution, this function behaves as a NOP operation.
+ * @param tableName This is the table on which the SELECT is being performed
+ */
+ public void readDirtyRowsAndUpdateDb(String tableName) {
+ mi.readDirtyRowsAndUpdateDb(dbi,tableName);
+ }
+
+
+
+
+ /**
+ * This method gets the primary key that the music interfaces uses by default.
+ * If the front end uses a primary key, this will not match what is used in the MUSIC interface
+ * @return
+ */
+ public String getMusicDefaultPrimaryKeyName() {
+ return mi.getMusicDefaultPrimaryKeyName();
+ }
+
+ /**
+ * Asks music interface to provide the function to create a primary key
+ * e.g. uuid(), 1, "unique_aksd419fjc"
+ * @return
+ */
+ public String generateUniqueKey() {
+ //
+ return mi.generateUniqueKey();
+ }
+
+
+ /**
+ * Perform a commit, as requested by the JDBC driver. If any row updates have been delayed,
+ * they are performed now and copied into MUSIC.
+ * @throws MDBCServiceException
+ */
+ public synchronized void commit(String txId, TxCommitProgress progressKeeper, DatabasePartition partition) throws MDBCServiceException {
+ logger.debug(EELFLoggerDelegate.applicationLogger, " commit ");
+ // transaction was committed -- add all the updates into the REDO-Log in MUSIC
+ try {
+ mi.commitLog(dbi, partition, transactionDigest, txId, progressKeeper);
+ }catch(MDBCServiceException e) {
+ logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), AppMessages.QUERYERROR, ErrorTypes.QUERYERROR, ErrorSeverity.CRITICAL);
+ throw e;
+ }
+ }
- public String getMusicKeyFromRow(String table, JSONObject dbRow) {
- TableInfo ti = dbi.getTableInfo(table);
- return mi.getMusicKeyFromRow(ti, table, dbRow);
- }
+ /**
+ * Perform a rollback, as requested by the JDBC driver. If any row updates have been delayed,
+ * they are discarded.
+ */
+ public synchronized void rollback() {
+ // transaction was rolled back - discard the updates
+ logger.debug(EELFLoggerDelegate.applicationLogger, "Rollback");;
+ transactionDigest.clear();
+ }
- /**
- * Returns all keys that matches the current sql statement, and not in already updated keys.
- *
- * @param sql the query that we are getting keys for
- * @deprecated
- */
- public ArrayList<String> getMusicKeys(String sql) {
- ArrayList<String> musicKeys = new ArrayList<String>();
- //\TODO See if this is required
+ /**
+ * Get all
+ * @param table
+ * @param dbRow
+ * @return
+ */
+ public String getMusicKeyFromRowWithoutPrimaryIndexes(String table, JSONObject dbRow) {
+ TableInfo ti = dbi.getTableInfo(table);
+ return mi.getMusicKeyFromRowWithoutPrimaryIndexes(ti,table, dbRow);
+ }
+
+ public String getMusicKeyFromRow(String table, JSONObject dbRow) {
+ TableInfo ti = dbi.getTableInfo(table);
+ return mi.getMusicKeyFromRow(ti,table, dbRow);
+ }
+
+ /**
+ * Returns all keys that matches the current sql statement, and not in already updated keys.
+ *
+ * @param sql the query that we are getting keys for
+ * @deprecated
+ */
+ public ArrayList<String> getMusicKeys(String sql) {
+ ArrayList<String> musicKeys = new ArrayList<String>();
+ //\TODO See if this is required
/*
try {
net.sf.jsqlparser.statement.Statement stmt = CCJSqlParserUtil.parse(sql);
System.out.print(musicKey + ",");
}
*/
- return musicKeys;
- }
-
- public void own(List<Range> ranges) {
- throw new java.lang.UnsupportedOperationException("function not implemented yet");
- }
-
- public void appendRange(String rangeId, List<Range> ranges) {
- throw new java.lang.UnsupportedOperationException("function not implemented yet");
- }
-
- public void relinquish(String ownerId, String rangeId) {
- throw new java.lang.UnsupportedOperationException("function not implemented yet");
- }
+ return musicKeys;
+ }
}
//\fixme this is not really used, delete!
String cassandraUrl = info.getProperty(Configuration.KEY_CASSANDRA_URL, Configuration.CASSANDRA_URL_DEFAULT);
String mixin = info.getProperty(Configuration.KEY_MUSIC_MIXIN_NAME, Configuration.MUSIC_MIXIN_DEFAULT);
- init(mixin,cassandraUrl);
- }
-
- protected void init(String mixin, String cassandraUrl) throws MDBCServiceException {
this.musicManager = MixinFactory.createMusicInterface(mixin, cassandraUrl, info,ranges);
this.musicManager.createKeyspace();
try {
- this.musicManager.initializeMetricDataStructures();
+ this.musicManager.initializeMdbcDataStructures();
} catch (MDBCServiceException e) {
logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.UNKNOWNERROR, ErrorSeverity.CRITICAL, ErrorTypes.GENERALSERVICEERROR);
throw(e);
public DatabasePartition partition;
public String nodeName;
- public NodeConfiguration(String tables, String mriIndex, String mriTableName, String partitionId, String sqlDatabaseName, String node, String redoRecordsTable){
- partition = new DatabasePartition(toRanges(tables), mriIndex, mriTableName, partitionId, null, redoRecordsTable) ;
+ public NodeConfiguration(String tables, String titIndex, String titTableName, String partitionId, String sqlDatabaseName, String node, String redoRecordsTable){
+ partition = new DatabasePartition(toRanges(tables), titIndex, titTableName, partitionId, null, redoRecordsTable) ;
this.sqlDatabaseName = sqlDatabaseName;
this.nodeName = node;
}
public class TablesConfiguration {
private final String TIT_TABLE_NAME = "transactioninformation";
- private final String MUSIC_TX_DIGEST_TABLE_NAME = "musictxdigest";
+ private final String REDO_RECORDS_NAME = "redorecords";
private transient static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(TablesConfiguration.class);
private List<PartitionInformation> partitions;
throw new MDBCServiceException("Partition was not correctly initialized");
}
for(PartitionInformation partitionInfo : partitions){
- String titTableName = partitionInfo.mriTableName;
+ String titTableName = partitionInfo.titTableName;
titTableName = (titTableName==null || titTableName.isEmpty())?TIT_TABLE_NAME:titTableName;
//0) Create the corresponding TIT table
- DatabaseOperations.CreateMusicRangeInformationTable(musicNamespace,titTableName);
- String musicTxDigestTableName = partitionInfo.mtxdTableName;
- musicTxDigestTableName = (musicTxDigestTableName==null || musicTxDigestTableName.isEmpty())? MUSIC_TX_DIGEST_TABLE_NAME :musicTxDigestTableName;
- DatabaseOperations.CreateMusicTxDigest(-1,musicNamespace,musicTxDigestTableName);
+ DatabaseOperations.CreateTransactionInformationTable(musicNamespace,titTableName);
+ String redoRecordsName = partitionInfo.rrtTableName;
+ redoRecordsName = (redoRecordsName==null || redoRecordsName.isEmpty())?REDO_RECORDS_NAME:redoRecordsName;
+ DatabaseOperations.CreateRedoRecordsTable(-1,musicNamespace,redoRecordsName);
//0) Create the corresponding TIT table
String partitionId;
if(partitionInfo.partitionId==null || partitionInfo.partitionId.isEmpty()){
//5) Add it to the redo history table
DatabaseOperations.createRedoHistoryBeginRow(musicNamespace,rhName,newRedoRow,partitionId,null);
//6) Create config for this node
- nodeConfigs.add(new NodeConfiguration(String.join(",",partitionInfo.tables),titIndex,titTableName,partitionId,sqlDatabaseName,partitionInfo.owner,musicTxDigestTableName));
+ nodeConfigs.add(new NodeConfiguration(String.join(",",partitionInfo.tables),titIndex,titTableName,partitionId,sqlDatabaseName,partitionInfo.owner,redoRecordsName));
}
return nodeConfigs;
}
public class PartitionInformation{
private List<String> tables;
private String owner;
- private String mriTableName;
- private String mtxdTableName;
+ private String titTableName;
+ private String rrtTableName;
private String partitionId;
private int replicationFactor;
this.owner = owner;
}
- public String getMriTableName() {
- return mriTableName;
+ public String getTitTableName() {
+ return titTableName;
}
- public void setMriTableName(String mriTableName) {
- this.mriTableName = mriTableName;
+ public void setTitTableName(String titTableName) {
+ this.titTableName = titTableName;
}
public String getPartitionId() {
this.replicationFactor = replicationFactor;
}
- public String getMtxdTableName(){
- return mtxdTableName;
+ public String getRrtTableName(){
+ return rrtTableName;
}
- public void setMtxdTableName(String mtxdTableName) {
- this.mtxdTableName = mtxdTableName;
+ public void setRrtTableName(String rrtTableName) {
+ this.rrtTableName = rrtTableName;
}
}
}
{
- "musicRangeInformationTable": "transactioninformation",
- "musicRangeInformationIndex": "d0e8ef2e-aeca-4261-8d9d-1679f560b85b",
+ "transactionInformationTable": "transactioninformation",
+ "transactionInformationIndex": "d0e8ef2e-aeca-4261-8d9d-1679f560b85b",
"partitionId": "798110cf-9c61-4db2-9446-cb2dbab5a143",
"lockId": "",
"ranges": [
{
"tables":["table11"],
"owner":"",
- "mriTableName":"musicrangeinformation",
- "mtxdTableName":"musictxdigest",
+ "titTableName":"transactioninformation",
+ "rrtTableName":"redorecords",
"partitionId":"",
"replicationFactor":1
}
public static final String PARTITION_INFORMATION_TABLE_NAME = "partitioninfo";
public static final String REDO_HISTORY_TABLE_NAME= "redohistory";
//\TODO Add logic to change the names when required and create the tables when necessary
- private String musicTxDigestTableName = "musictxdigest";
- private String musicRangeInformationTableName = "musicrangeinformation";
+ private String redoRecordTableName = "redorecords";
+ private String transactionInformationTableName = "transactioninformation";
private EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(CassandraMixin.class);
this.music_ns = info.getProperty(KEY_MUSIC_NAMESPACE,DEFAULT_MUSIC_NAMESPACE);
logger.info(EELFLoggerDelegate.applicationLogger,"MusicSqlManager: music_ns="+music_ns);
- musicRangeInformationTableName = "transactioninformation";
+ transactionInformationTableName = "transactioninformation";
createMusicKeyspace();
}
}
}
@Override
- public void initializeMetricDataStructures() throws MDBCServiceException {
+ public void initializeMdbcDataStructures() throws MDBCServiceException {
try {
- DatabaseOperations.CreateMusicTxDigest(-1, music_ns, musicTxDigestTableName);//\TODO If we start partitioning the data base, we would need to use the redotable number
- DatabaseOperations.CreateMusicRangeInformationTable(music_ns, musicRangeInformationTableName);
+ DatabaseOperations.CreateRedoRecordsTable(-1, music_ns, redoRecordTableName);//\TODO If we start partitioning the data base, we would need to use the redotable number
+ DatabaseOperations.CreateTransactionInformationTable(music_ns, transactionInformationTableName);
DatabaseOperations.CreateTableToPartitionTable(music_ns, TABLE_TO_PARTITION_TABLE_NAME);
DatabaseOperations.CreatePartitionInfoTable(music_ns, PARTITION_INFORMATION_TABLE_NAME);
DatabaseOperations.CreateRedoHistoryTable(music_ns, REDO_HISTORY_TABLE_NAME);
}
- private PreparedQueryObject createAppendMtxdIndexToMriQuery(String titTable, String uuid, String table, String redoUuid){
+ private PreparedQueryObject createAppendRRTIndexToTitQuery(String titTable, String uuid, String table, String redoUuid){
PreparedQueryObject query = new PreparedQueryObject();
StringBuilder appendBuilder = new StringBuilder();
appendBuilder.append("UPDATE ")
return lockId;
}
- protected void pushRowToMtxd(String lockId, String commitId, HashMap<Range,StagingTable> transactionDigest) throws MDBCServiceException{
+ protected void pushRowToRRT(String lockId, String commitId, HashMap<Range,StagingTable> transactionDigest) throws MDBCServiceException{
PreparedQueryObject query = new PreparedQueryObject();
StringBuilder cqlQuery = new StringBuilder("INSERT INTO ")
.append(music_ns)
.append('.')
- .append(musicTxDigestTableName)
+ .append(redoRecordTableName)
.append(" (leaseid,leasecounter,transactiondigest) ")
.append("VALUES ('")
.append( lockId ).append("',")
}
}
- protected void appendIndexToMri(String lockId, String commitId, String TITIndex) throws MDBCServiceException{
+ protected void appendIndexToTit(String lockId, String commitId, String TITIndex) throws MDBCServiceException{
StringBuilder redoUuidBuilder = new StringBuilder();
redoUuidBuilder.append("('")
.append(lockId)
.append("',")
.append(commitId)
.append(")");
- PreparedQueryObject appendQuery = createAppendMtxdIndexToMriQuery(musicRangeInformationTableName, TITIndex, musicTxDigestTableName, redoUuidBuilder.toString());
- ReturnType returnType = MusicPureCassaCore.criticalPut(music_ns, musicRangeInformationTableName, TITIndex, appendQuery, lockId, null);
+ PreparedQueryObject appendQuery = createAppendRRTIndexToTitQuery(transactionInformationTableName, TITIndex, redoRecordTableName, redoUuidBuilder.toString());
+ ReturnType returnType = MusicPureCassaCore.criticalPut(music_ns, transactionInformationTableName, TITIndex, appendQuery, lockId, null);
if(returnType.getResult().compareTo(ResultType.SUCCESS) != 0 ){
logger.error(EELFLoggerDelegate.errorLogger, "Error when executing append operation with return type: "+returnType.getMessage());
throw new MDBCServiceException("Error when executing append operation with return type: "+returnType.getMessage());
@Override
public void commitLog(DBInterface dbi, DatabasePartition partition, HashMap<Range,StagingTable> transactionDigest, String txId ,TxCommitProgress progressKeeper) throws MDBCServiceException{
- String TITIndex = partition.getMusicRangeInformationIndex();
+ String TITIndex = partition.getTransactionInformationIndex();
if(TITIndex.isEmpty()) {
//\TODO Fetch TITIndex from the Range Information Table
throw new MDBCServiceException("TIT Index retrieval not yet implemented");
}
- String fullyQualifiedTitKey = music_ns+"."+ musicRangeInformationTableName +"."+TITIndex;
+ String fullyQualifiedTitKey = music_ns+"."+ transactionInformationTableName +"."+TITIndex;
//0. See if reference to lock was already created
String lockId = partition.getLockId();
if(lockId == null || lockId.isEmpty()) {
- lockId = createAndAssignLock(fullyQualifiedTitKey,partition,music_ns, musicRangeInformationTableName,TITIndex);
+ lockId = createAndAssignLock(fullyQualifiedTitKey,partition,music_ns,transactionInformationTableName,TITIndex);
}
String commitId;
//Add creation type of transaction digest
//1. Push new row to RRT and obtain its index
- pushRowToMtxd(lockId, commitId, transactionDigest);
+ pushRowToRRT(lockId, commitId, transactionDigest);
//2. Save RRT index to RQ
if(progressKeeper!= null) {
progressKeeper.setRecordId(txId,new RedoRecordId(lockId, commitId));
}
//3. Append RRT index into the corresponding TIT row array
- appendIndexToMri(lockId,commitId,TITIndex);
+ appendIndexToTit(lockId,commitId,TITIndex);
}
/**
* This function is used to created all the required data structures, both local
* \TODO Check if this function is required in the MUSIC interface or could be just created on the constructor
*/
- void initializeMetricDataStructures() throws MDBCServiceException;
+ void initializeMdbcDataStructures() throws MDBCServiceException;
/**
* Get the name of this MusicInterface mixin object.
* @return the name
}
@Override
- public void initializeMetricDataStructures() {
+ public void initializeMdbcDataStructures() {
//
}
+++ /dev/null
-package com.att.research.mdbc.mixins;
-
-import java.io.Serializable;
-
-import org.json.JSONObject;
-import org.json.JSONTokener;
-
-public final class Operation implements Serializable{
-
- private static final long serialVersionUID = -1215301985078183104L;
-
- final OperationType TYPE;
- final String OLD_VAL;
- final String NEW_VAL;
-
- public Operation(OperationType type, String newVal, String oldVal) {
- TYPE = type;
- NEW_VAL = newVal;
- OLD_VAL = oldVal;
- }
-
- public JSONObject getNewVal(){
- JSONObject newRow = new JSONObject(new JSONTokener(NEW_VAL));
- return newRow;
- }
-
- public JSONObject getOldVal(){
- JSONObject keydata = new JSONObject(new JSONTokener(OLD_VAL));
- return keydata;
- }
-}
+++ /dev/null
-package com.att.research.mdbc.mixins;
-
-public enum OperationType{
- DELETE, UPDATE, INSERT, SELECT
-}
+++ /dev/null
-package com.att.research.mdbc.mixins;
-
-import java.util.List;
-
-public class PartitionInformation {
- public final String partition;
- public final TitReference tit;
- public final List<String> tables;
- public final int replicationFactor;
- public final String currentOwner;
-
- public PartitionInformation(String partition, TitReference tit, List<String> tables, int replicationFactor, String currentOwner) {
- this.partition=partition;
- this.tit=tit;
- this.tables=tables;
- this.replicationFactor=replicationFactor;
- this.currentOwner=currentOwner;
- }
-}
+++ /dev/null
-package com.att.research.mdbc.mixins;
-
-import java.util.List;
-
-public final class RedoHistoryElement {
- public final String partition;
- public final TitReference current;
- public final List<TitReference> previous;
-
- public RedoHistoryElement(String partition, TitReference current, List<TitReference> previous) {
- this.partition = partition;
- this.current = current;
- this.previous = previous;
- }
-}
+++ /dev/null
-package com.att.research.mdbc.mixins;
-
-public final class RedoRecordId {
- public final String leaseId;
- public final String commitId;
-
- public RedoRecordId(String leaseId, String commitId) {
- this.leaseId = leaseId;
- this.commitId = commitId;
- }
-
- public boolean isEmpty() {
- return (this.leaseId==null || this.leaseId.isEmpty())&&(this.commitId==null||this.commitId.isEmpty());
- }
-}
+++ /dev/null
-package com.att.research.mdbc.mixins;
-
-import java.io.Serializable;
-import java.util.Deque;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.Set;
-import org.apache.commons.lang3.tuple.Pair;
-import org.json.JSONObject;
-
-import com.att.research.logging.EELFLoggerDelegate;
-
-public class StagingTable implements Serializable{
- /**
- *
- */
- private static final long serialVersionUID = 7583182634761771943L;
- private transient static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(StagingTable.class);
- private HashMap<String,Deque<Operation>> operations;
-
- public StagingTable() {
- operations = new HashMap<>();
- }
-
- synchronized public void addOperation(String key, OperationType type, String oldVal, String newVal) {
- if(!operations.containsKey(key)) {
- operations.put(key, new LinkedList<>());
- }
- operations.get(key).add(new Operation(type,newVal,oldVal));
- }
-
- synchronized public Deque<Pair<String,Operation>> getIterableSnapshot() throws NoSuchFieldException{
- Deque<Pair<String,Operation>> response=new LinkedList<Pair<String,Operation>>();
- //\TODO: check if we can just return the last change to a given key
- Set<String> keys = operations.keySet();
- for(String key : keys) {
- Deque<Operation> ops = operations.get(key);
- if(ops.isEmpty()) {
- logger.error(EELFLoggerDelegate.errorLogger, "Invalid state of the Operation data structure when creating snapshot");
- throw new NoSuchFieldException("Invalid state of the operation data structure");
- }
- response.add(Pair.of(key,ops.getLast()));
- }
- return response;
- }
-
- synchronized public void clean() {
- operations.clear();
- }
-}
+++ /dev/null
-package com.att.research.mdbc.mixins;
-
-import java.util.List;
-
-public final class TablePartitionInformation {
- public final String table;
- public final String partition;
- public final List<String> oldPartitions;
-
- public TablePartitionInformation(String table, String partition, List<String> oldPartitions) {
- this.table = table;
- this.partition = partition;
- this.oldPartitions = oldPartitions;
- }
-}
+++ /dev/null
-package com.att.research.mdbc.mixins;
-
-public final class TitReference {
- public final String table;
- public final String index;
-
- public TitReference(String table, String index) {
- this.table = table;
- this.index= index;
- }
-
-}
+++ /dev/null
-package com.att.research.mdbc.mixins;
-
-import java.util.List;
-
-public final class TransactionInformationElement {
- public final String index;
- public final List<RedoRecordId> redoLog;
- public final String partition;
- public final int latestApplied;
- public final boolean applied;
-
- public TransactionInformationElement(String index, List<RedoRecordId> redoLog, String partition, int latestApplied, boolean applied) {
- this.index = index;
- this.redoLog = redoLog;
- this.partition = partition;
- this.latestApplied = latestApplied;
- this.applied = applied;
- }
-}
+++ /dev/null
-package com.att.research.mdbc.mixins;
-
-import java.math.BigInteger;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicLong;
-import com.att.research.logging.EELFLoggerDelegate;
-
-import java.sql.Connection;
-import java.util.concurrent.atomic.AtomicReference;
-
-
-public class TxCommitProgress{
- private EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(TxCommitProgress.class);
-
- private AtomicReference<BigInteger> nextCommitId;
- private Map<String, CommitProgress> transactionInfo;
-
- public TxCommitProgress(){
- nextCommitId=new AtomicReference<>(BigInteger.ZERO);
- transactionInfo = new ConcurrentHashMap<>();
- }
-
- public boolean containsTx(String txId) {
- return transactionInfo.containsKey(txId);
- }
-
- public BigInteger getCommitId(String txId) {
- CommitProgress prog = transactionInfo.get(txId);
- if(prog.isCommitIdAssigned()) {
- return prog.getCommitId();
- }
- BigInteger commitId = nextCommitId.getAndUpdate((a)-> a.add(BigInteger.ONE));
- prog.setCommitId(commitId);
- return commitId;
- }
-
- public void createNewTransactionTracker(String id, Connection conn) {
- transactionInfo.put(id, new CommitProgress(id,conn));
- }
-
- public void commitRequested(String txId) {
- CommitProgress prog = transactionInfo.get(txId);
- if(prog == null){
- logger.error(EELFLoggerDelegate.errorLogger, "Transaction doesn't exist: [%l], failure when storing commit request",txId);
- }
- prog.setCommitRequested();
- }
-
- public void setSQLDone(String txId) {
- CommitProgress prog = transactionInfo.get(txId);
- if(prog == null){
- logger.error(EELFLoggerDelegate.errorLogger, "Transaction doesn't exist: [%l], failure when storing saving completion of SQL",txId);
- }
- prog.setSQLCompleted();
- }
-
- public void setMusicDone(String txId) {
- CommitProgress prog = transactionInfo.get(txId);
- if(prog == null){
- logger.error(EELFLoggerDelegate.errorLogger, "Transaction doesn't exist: [%l], failure when storing saving completion of Music",txId);
- }
- prog.setMusicCompleted();
- }
-
- public Connection getConnection(String txId){
- CommitProgress prog = transactionInfo.get(txId);
- if(prog == null){
- logger.error(EELFLoggerDelegate.errorLogger, "Transaction doesn't exist: [%l], failure when retrieving statement",txId);
- }
- return prog.getConnection();
- }
-
- public void setRecordId(String txId, RedoRecordId recordId){
- CommitProgress prog = transactionInfo.get(txId);
- if(prog == null){
- logger.error(EELFLoggerDelegate.errorLogger, "Transaction doesn't exist: [%l], failure when setting record Id",txId);
- }
- prog.setRecordId(recordId);
- }
-
- public RedoRecordId getRecordId(String txId) {
- CommitProgress prog = transactionInfo.get(txId);
- if(prog == null){
- logger.error(EELFLoggerDelegate.errorLogger, "Transaction doesn't exist: [%l], failure when getting record Id",txId);
- }
- return prog.getRecordId();
- }
-
- public boolean isRecordIdAssigned(String txId) {
- CommitProgress prog = transactionInfo.get(txId);
- if(prog == null){
- logger.error(EELFLoggerDelegate.errorLogger, "Transaction doesn't exist: [%l], failure when checking record",txId);
- }
- return prog.isRedoRecordAssigned();
- }
-
- public boolean isComplete(String txId) {
- CommitProgress prog = transactionInfo.get(txId);
- if(prog == null){
- logger.error(EELFLoggerDelegate.errorLogger, "Transaction doesn't exist: [%l], failure when checking completion",txId);
- }
- return prog.isComplete();
- }
-
- public void reinitializeTxProgress(String txId) {
- CommitProgress prog = transactionInfo.get(txId);
- if(prog == null){
- logger.error(EELFLoggerDelegate.errorLogger, "Transaction doesn't exist: [%l], failure when reinitializing tx progress",txId);
- }
- prog.reinitialize();
- }
-
- public void deleteTxProgress(String txId){
- transactionInfo.remove(txId);
- }
-}
-
-final class CommitProgress{
- private String lTxId; // local transaction id
- private BigInteger commitId; // commit id
- private boolean commitRequested; //indicates if the user tried to commit the request already.
- private boolean SQLDone; // indicates if SQL was already committed
- private boolean MusicDone; // indicates if music commit was already performed, atomic bool
- private Connection connection;// reference to a connection object. This is used to complete a commit if it failed in the original thread.
- private Long timestamp; // last time this data structure was updated
- private RedoRecordId redoRecordId;// record id for each partition
-
- public CommitProgress(String id,Connection conn){
- redoRecordId=null;
- lTxId = id;
- commitRequested = false;
- SQLDone = false;
- MusicDone = false;
- connection = conn;
- commitId = null;
- timestamp = System.currentTimeMillis();
- }
-
- public synchronized boolean isComplete() {
- return commitRequested && SQLDone && MusicDone;
- }
-
- public synchronized void setCommitId(BigInteger commitId) {
- this.commitId = commitId;
- timestamp = System.currentTimeMillis();
- }
-
- public synchronized void reinitialize() {
- commitId = null;
- redoRecordId=null;
- commitRequested = false;
- SQLDone = false;
- MusicDone = false;
- timestamp = System.currentTimeMillis();
- }
-
- public synchronized void setCommitRequested() {
- commitRequested = true;
- timestamp = System.currentTimeMillis();
- }
-
- public synchronized void setSQLCompleted() {
- SQLDone = true;
- timestamp = System.currentTimeMillis();
- }
-
- public synchronized void setMusicCompleted() {
- MusicDone = true;
- timestamp = System.currentTimeMillis();
- }
-
- public Connection getConnection() {
- timestamp = System.currentTimeMillis();
- return connection;
- }
-
- public long getTimestamInMillis() {
- return timestamp;
- }
-
- public synchronized void setRecordId(RedoRecordId id) {
- redoRecordId = id;
- timestamp = System.currentTimeMillis();
- }
-
- public synchronized boolean isRedoRecordAssigned() {
- return this.redoRecordId!=null;
- }
-
- public synchronized RedoRecordId getRecordId() {
- return redoRecordId;
- }
-
- public synchronized BigInteger getCommitId() {
- return commitId;
- }
-
- public synchronized String getId() {
- return this.lTxId;
- }
-
- public synchronized boolean isCommitIdAssigned() {
- return this.commitId!= null;
- }
-}
\ No newline at end of file
package com.att.research.mdbc.tools;
import com.att.research.logging.EELFLoggerDelegate;
+import com.att.research.mdbc.DatabasePartition;
+import com.att.research.mdbc.MDBCUtils;
+import com.att.research.mdbc.Range;
import com.att.research.mdbc.configurations.NodeConfiguration;
import com.beust.jcommander.JCommander;
import com.beust.jcommander.Parameter;
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+
public class CreatePartition {
public static final EELFLoggerDelegate LOG = EELFLoggerDelegate.getLogger(CreatePartition.class);
@Parameter(names = { "-n", "--tit-table-name" }, required = true,
description = "Tit Table name")
private String titTable;
- @Parameter(names = { "-r", "--music-tx-digest-table-name" }, required = true,
- description = "Music Transaction Digest Table name")
- private String mTxDTable;
+ @Parameter(names = { "-r", "--redorecords-table-name" }, required = true,
+ description = "Redo Records Table name")
+ private String rrTable;
@Parameter(names = { "-p", "--partition-id" }, required = true,
description = "Partition Id")
private String partitionId;
}
public void convert(){
- config = new NodeConfiguration(tables,titIndex,titTable,partitionId,"test","", mTxDTable);
+ config = new NodeConfiguration(tables,titIndex,titTable,partitionId,"test","",rrTable);
}
public void saveToFile(){
Code Review / music / mdbc.git / commitdiff