diff options
17 files changed, 194 insertions, 313 deletions
diff --git a/src/main/java/com/att/research/mdbc/mixins/CassandraMixin.java b/src/main/java/com/att/research/mdbc/mixins/CassandraMixin.java index 5293682..63c95c2 100755 --- a/src/main/java/com/att/research/mdbc/mixins/CassandraMixin.java +++ b/src/main/java/com/att/research/mdbc/mixins/CassandraMixin.java @@ -17,12 +17,10 @@ import java.util.UUID; import com.att.research.mdbc.*; import com.att.research.mdbc.tables.PartitionInformation; -import com.att.research.mdbc.tables.RedoHistoryElement; -import com.att.research.mdbc.tables.RedoRecordId; +import com.att.research.mdbc.tables.MusixTxDigestId; import com.att.research.mdbc.tables.StagingTable; -import com.att.research.mdbc.tables.TablePartitionInformation; -import com.att.research.mdbc.tables.TitReference; -import com.att.research.mdbc.tables.TransactionInformationElement; +import com.att.research.mdbc.tables.MriReference; +import com.att.research.mdbc.tables.MusicRangeInformationRow; import com.att.research.mdbc.tables.TxCommitProgress; import org.json.JSONObject; @@ -1025,13 +1023,13 @@ public class CassandraMixin implements MusicInterface { } - private PreparedQueryObject createAppendMtxdIndexToMriQuery(String titTable, String uuid, String table, String redoUuid){ + private PreparedQueryObject createAppendMtxdIndexToMriQuery(String mriTable, String uuid, String table, UUID redoUuid){ PreparedQueryObject query = new PreparedQueryObject(); StringBuilder appendBuilder = new StringBuilder(); appendBuilder.append("UPDATE ") .append(music_ns) .append(".") - .append(titTable) + .append(mriTable) .append(" SET redo = redo +[('") .append(table) .append("',") @@ -1095,15 +1093,14 @@ public class CassandraMixin implements MusicInterface { return lockId; } - protected void pushRowToMtxd(String lockId, String commitId, HashMap<Range,StagingTable> transactionDigest) throws MDBCServiceException{ + protected void pushRowToMtxd(UUID commitId, HashMap<Range,StagingTable> transactionDigest) throws MDBCServiceException{ PreparedQueryObject query = new PreparedQueryObject(); StringBuilder cqlQuery = new StringBuilder("INSERT INTO ") .append(music_ns) .append('.') .append(musicTxDigestTableName) - .append(" (leaseid,leasecounter,transactiondigest) ") + .append(" (txid,transactiondigest) ") .append("VALUES ('") - .append( lockId ).append("',") .append( commitId ).append(",'"); try { cqlQuery.append( MDBCUtils.toString(transactionDigest) ); @@ -1122,14 +1119,8 @@ public class CassandraMixin implements MusicInterface { } } - protected void appendIndexToMri(String lockId, String commitId, String MriIndex) throws MDBCServiceException{ - StringBuilder redoUuidBuilder = new StringBuilder(); - redoUuidBuilder.append("('") - .append(lockId) - .append("',") - .append(commitId) - .append(")"); - PreparedQueryObject appendQuery = createAppendMtxdIndexToMriQuery(musicRangeInformationTableName, MriIndex, musicTxDigestTableName, redoUuidBuilder.toString()); + protected void appendIndexToMri(String lockId, UUID commitId, String MriIndex) throws MDBCServiceException{ + PreparedQueryObject appendQuery = createAppendMtxdIndexToMriQuery(musicRangeInformationTableName, MriIndex, musicTxDigestTableName, commitId); ReturnType returnType = MusicCore.criticalPut(music_ns, musicRangeInformationTableName, MriIndex, appendQuery, lockId, null); if(returnType.getResult().compareTo(ResultType.SUCCESS) != 0 ){ logger.error(EELFLoggerDelegate.errorLogger, "Error when executing append operation with return type: "+returnType.getMessage()); @@ -1151,10 +1142,10 @@ public class CassandraMixin implements MusicInterface { lockId = createAndAssignLock(fullyQualifiedTitKey,partition,music_ns, musicRangeInformationTableName,MriIndex); } - String commitId; + UUID commitId; //Generate a local commit id if(progressKeeper.containsTx(txId)) { - commitId = progressKeeper.getCommitId(txId).toString(); + commitId = progressKeeper.getCommitId(txId); } else{ logger.error(EELFLoggerDelegate.errorLogger, "Tx with id "+txId+" was not created in the TxCommitProgress "); @@ -1163,11 +1154,11 @@ public class CassandraMixin implements MusicInterface { //Add creation type of transaction digest //1. Push new row to RRT and obtain its index - pushRowToMtxd(lockId, commitId, transactionDigest); + pushRowToMtxd(commitId, transactionDigest); //2. Save RRT index to RQ if(progressKeeper!= null) { - progressKeeper.setRecordId(txId,new RedoRecordId(lockId, commitId)); + progressKeeper.setRecordId(txId,new MusixTxDigestId(commitId)); } //3. Append RRT index into the corresponding TIT row array appendIndexToMri(lockId,commitId,MriIndex); @@ -1223,73 +1214,48 @@ public class CassandraMixin implements MusicInterface { } @Override - public TransactionInformationElement getTransactionInformation(String id){ - throw new UnsupportedOperationException(); - } - - @Override - public TitReference createTransactionInformationRow(TransactionInformationElement info){ - throw new UnsupportedOperationException(); - } - - @Override - public void appendToRedoLog(TitReference titRow, DatabasePartition partition, RedoRecordId newRecord){ - throw new UnsupportedOperationException(); - } - - @Override - public void appendRedoRecord(String redoRecordTable, RedoRecordId newRecord, String transactionDigest){ - throw new UnsupportedOperationException(); - } - - @Override - public void updateTablePartition(String table, DatabasePartition partition){ - throw new UnsupportedOperationException(); - } - @Override - public TitReference createPartition(List<String> tables, int replicationFactor, String currentOwner){ + public MusicRangeInformationRow getMusicRangeInformation(UUID id){ throw new UnsupportedOperationException(); } @Override - public void updatePartitionOwner(String partition, String owner){ + public MriReference createMusicRangeInformation(MusicRangeInformationRow info){ throw new UnsupportedOperationException(); } @Override - public void updateTitReference(String partition, TitReference tit){ + public void appendToRedoLog(MriReference mriRowId, DatabasePartition partition, MusixTxDigestId newRecord){ throw new UnsupportedOperationException(); } @Override - public void updatePartitionReplicationFactor(String partition, int replicationFactor){ + public void addTxDigest(String musicTxDigestTable, MusixTxDigestId newId, String transactionDigest){ throw new UnsupportedOperationException(); } @Override - public void addRedoHistory(DatabasePartition partition, TitReference newTit, List<TitReference> old){ + public List<PartitionInformation> getPartitionInformation(DatabasePartition partition){ throw new UnsupportedOperationException(); } @Override - public List<RedoHistoryElement> getHistory(DatabasePartition partition){ + public HashMap<Range,StagingTable> getTransactionDigest(MusixTxDigestId id){ throw new UnsupportedOperationException(); } @Override - public List<PartitionInformation> getPartitionInformation(DatabasePartition partition){ + public void own(List<Range> ranges){ throw new UnsupportedOperationException(); } @Override - public TablePartitionInformation getTablePartitionInformation(String table){ + public void appendRange(String rangeId, List<Range> ranges){ throw new UnsupportedOperationException(); } @Override - public HashMap<Range,StagingTable> getTransactionDigest(RedoRecordId id){ + public void relinquish(String ownerId, String rangeId){ throw new UnsupportedOperationException(); } - - } +} diff --git a/src/main/java/com/att/research/mdbc/mixins/MusicInterface.java b/src/main/java/com/att/research/mdbc/mixins/MusicInterface.java index 6e2e0ca..7fb6d18 100755 --- a/src/main/java/com/att/research/mdbc/mixins/MusicInterface.java +++ b/src/main/java/com/att/research/mdbc/mixins/MusicInterface.java @@ -3,6 +3,7 @@ package com.att.research.mdbc.mixins; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.UUID; import org.json.JSONObject; @@ -11,16 +12,12 @@ import com.att.research.mdbc.DatabasePartition; import com.att.research.mdbc.Range; import com.att.research.mdbc.TableInfo; import com.att.research.mdbc.tables.PartitionInformation; -import com.att.research.mdbc.tables.RedoHistoryElement; -import com.att.research.mdbc.tables.RedoRecordId; +import com.att.research.mdbc.tables.MusixTxDigestId; import com.att.research.mdbc.tables.StagingTable; -import com.att.research.mdbc.tables.TablePartitionInformation; -import com.att.research.mdbc.tables.TitReference; -import com.att.research.mdbc.tables.TransactionInformationElement; +import com.att.research.mdbc.tables.MriReference; +import com.att.research.mdbc.tables.MusicRangeInformationRow; import com.att.research.mdbc.tables.TxCommitProgress; -import org.onap.music.exceptions.MusicLockingException; - /** * This Interface defines the methods that MDBC needs for a class to provide access to the persistence layer of MUSIC. * @@ -154,34 +151,23 @@ public interface MusicInterface { */ void commitLog(DBInterface dbi, DatabasePartition partition, HashMap<Range,StagingTable> transactionDigest, String txId,TxCommitProgress progressKeeper) throws MDBCServiceException; - TransactionInformationElement getTransactionInformation(String id); + MusicRangeInformationRow getMusicRangeInformation(UUID id); - TitReference createTransactionInformationRow(TransactionInformationElement info); + MriReference createMusicRangeInformation(MusicRangeInformationRow info); - void appendToRedoLog(TitReference titRow, DatabasePartition partition, RedoRecordId newRecord); + void appendToRedoLog(MriReference mriRowId, DatabasePartition partition, MusixTxDigestId newRecord); - void appendRedoRecord(String redoRecordTable, RedoRecordId newRecord, String transactionDigest); + void addTxDigest(String musicTxDigestTable, MusixTxDigestId newId, String transactionDigest); - void updateTablePartition(String table, DatabasePartition partition); - - TitReference createPartition(List<String> tables, int replicationFactor, String currentOwner); + List<PartitionInformation> getPartitionInformation(DatabasePartition partition); - void updatePartitionOwner(String partition, String owner); + HashMap<Range,StagingTable> getTransactionDigest(MusixTxDigestId id); - void updateTitReference(String partition, TitReference tit); - - void updatePartitionReplicationFactor(String partition, int replicationFactor); - - void addRedoHistory(DatabasePartition partition, TitReference newTit, List<TitReference> old); - - List<RedoHistoryElement> getHistory(DatabasePartition partition); + void own(List<Range> ranges); - List<PartitionInformation> getPartitionInformation(DatabasePartition partition); - - TablePartitionInformation getTablePartitionInformation(String table); - - HashMap<Range,StagingTable> getTransactionDigest(RedoRecordId id); + void appendRange(String rangeId, List<Range> ranges); + void relinquish(String ownerId, String rangeId); } diff --git a/src/main/java/com/att/research/mdbc/mixins/MusicMixin.java b/src/main/java/com/att/research/mdbc/mixins/MusicMixin.java index 43f36e3..630b75e 100644 --- a/src/main/java/com/att/research/mdbc/mixins/MusicMixin.java +++ b/src/main/java/com/att/research/mdbc/mixins/MusicMixin.java @@ -2,13 +2,7 @@ package com.att.research.mdbc.mixins; import java.io.IOException; import java.io.InputStream; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Properties; -import java.util.Set; +import java.util.*; import com.att.research.mdbc.LockId; import org.json.JSONObject; @@ -19,12 +13,10 @@ import com.att.research.mdbc.DatabasePartition; import com.att.research.mdbc.Range; import com.att.research.mdbc.TableInfo; import com.att.research.mdbc.tables.PartitionInformation; -import com.att.research.mdbc.tables.RedoHistoryElement; -import com.att.research.mdbc.tables.RedoRecordId; +import com.att.research.mdbc.tables.MusixTxDigestId; import com.att.research.mdbc.tables.StagingTable; -import com.att.research.mdbc.tables.TablePartitionInformation; -import com.att.research.mdbc.tables.TitReference; -import com.att.research.mdbc.tables.TransactionInformationElement; +import com.att.research.mdbc.tables.MriReference; +import com.att.research.mdbc.tables.MusicRangeInformationRow; import com.att.research.mdbc.tables.TxCommitProgress; import org.onap.music.main.MusicCore; @@ -36,7 +28,7 @@ import org.onap.music.main.MusicCore; public class MusicMixin implements MusicInterface { public static Map<Integer, Set<String>> currentLockMap = new HashMap<>(); - public static List<String> criticalTables = new ArrayList<>(); + public static List<String> criticalTables = new ArrayList<>(); @Override public String getMixinName() { @@ -65,49 +57,49 @@ public class MusicMixin implements MusicInterface { @Override public void close() { // - + } @Override public void createKeyspace() { // - + } @Override public void initializeMusicForTable(TableInfo ti, String tableName) { // - + } @Override public void createDirtyRowTable(TableInfo ti, String tableName) { // - + } @Override public void dropDirtyRowTable(String tableName) { // - + } @Override public void clearMusicForTable(String tableName) { // - + } @Override - public void markDirtyRow(TableInfo ti, String tableName, JSONObject keys) { + public void markDirtyRow(TableInfo ti, String tableName, JSONObject keys) { // - + } @Override public void cleanDirtyRow(TableInfo ti, String tableName, JSONObject keys) { // - + } @Override @@ -119,55 +111,54 @@ public class MusicMixin implements MusicInterface { @Override public void deleteFromEntityTableInMusic(TableInfo ti, String tableName, JSONObject oldRow) { // - + } @Override public void readDirtyRowsAndUpdateDb(DBInterface dbi, String tableName) { // - + } @Override public void updateDirtyRowAndEntityTableInMusic(TableInfo ti, String tableName, JSONObject changedRow) { updateDirtyRowAndEntityTableInMusic(tableName, changedRow, false); - + } - - public void updateDirtyRowAndEntityTableInMusic(String tableName, JSONObject changedRow, boolean isCritical) { } - + + public void updateDirtyRowAndEntityTableInMusic(String tableName, JSONObject changedRow, boolean isCritical) { + } + public static void loadProperties() { - Properties prop = new Properties(); - InputStream input = null; - try { - input = MusicMixin.class.getClassLoader().getResourceAsStream("mdbc.properties"); - prop.load(input); - String crTable = prop.getProperty("critical.tables"); - String[] tableArr = crTable.split(","); - criticalTables = Arrays.asList(tableArr); - - } - catch (Exception ex) { - ex.printStackTrace(); - } - finally { - if (input != null) { - try { - input.close(); - } catch (IOException e) { - e.printStackTrace(); - } - } - } - } - + Properties prop = new Properties(); + InputStream input = null; + try { + input = MusicMixin.class.getClassLoader().getResourceAsStream("mdbc.properties"); + prop.load(input); + String crTable = prop.getProperty("critical.tables"); + String[] tableArr = crTable.split(","); + criticalTables = Arrays.asList(tableArr); + + } catch (Exception ex) { + ex.printStackTrace(); + } finally { + if (input != null) { + try { + input.close(); + } catch (IOException e) { + e.printStackTrace(); + } + } + } + } + public static void releaseZKLocks(Set<LockId> lockIds) { - for(LockId lockId: lockIds) { - System.out.println("Releasing lock: "+lockId); + for (LockId lockId : lockIds) { + System.out.println("Releasing lock: " + lockId); try { - MusicCore.voluntaryReleaseLock(lockId.getFullyQualifiedLockKey(),lockId.getLockReference()); - MusicCore.destroyLockRef(lockId.getFullyQualifiedLockKey(),lockId.getLockReference()); + MusicCore.voluntaryReleaseLock(lockId.getFullyQualifiedLockKey(), lockId.getLockReference()); + MusicCore.destroyLockRef(lockId.getFullyQualifiedLockKey(), lockId.getLockReference()); } catch (MusicLockingException e) { e.printStackTrace(); } @@ -183,7 +174,7 @@ public class MusicMixin implements MusicInterface { @Override public void initializeMetricDataStructures() { // - + } @Override @@ -192,67 +183,51 @@ public class MusicMixin implements MusicInterface { } @Override - public void commitLog(DBInterface dbi, DatabasePartition partition, HashMap<Range,StagingTable> transactionDigest, String txId,TxCommitProgress progressKeeper) - throws MDBCServiceException{ + public void commitLog(DBInterface dbi, DatabasePartition partition, HashMap<Range, StagingTable> transactionDigest, String txId, TxCommitProgress progressKeeper) + throws MDBCServiceException { // TODO Auto-generated method stub } @Override - public TablePartitionInformation getTablePartitionInformation(String table){ - return null; - } - - @Override - public HashMap<Range,StagingTable> getTransactionDigest(RedoRecordId id){ - return null; - } - - @Override - public TransactionInformationElement getTransactionInformation(String id){ - return null; - } - - @Override - public void updateTitReference(String partition, TitReference tit){} - - @Override - public List<RedoHistoryElement> getHistory(DatabasePartition partition){ - return null; - } - - @Override - public void addRedoHistory(DatabasePartition partition, TitReference newTit, List<TitReference> old){ - } - - @Override - public TitReference createPartition(List<String> tables, int replicationFactor, String currentOwner){ - return null; - } - - @Override - public List<PartitionInformation> getPartitionInformation(DatabasePartition partition){ - return null; - } - - @Override - public TitReference createTransactionInformationRow(TransactionInformationElement info){ - return null; - } - - @Override - public void appendToRedoLog(TitReference titRow, DatabasePartition partition, RedoRecordId newRecord){ - } - - @Override - public void appendRedoRecord(String redoRecordTable, RedoRecordId newRecord, String transactionDigest){ - } - - @Override - public void updateTablePartition(String table, DatabasePartition partition){} + public HashMap<Range, StagingTable> getTransactionDigest(MusixTxDigestId id) { + return null; + } + + @Override + public List<PartitionInformation> getPartitionInformation(DatabasePartition partition) { + return null; + } + + @Override + public MriReference createMusicRangeInformation(MusicRangeInformationRow info) { + return null; + } + + @Override + public void appendToRedoLog(MriReference mriRowId, DatabasePartition partition, MusixTxDigestId newRecord) { + } + + @Override + public void addTxDigest(String musicTxDigestTable, MusixTxDigestId newId, String transactionDigest) { + } + + @Override + public void own(List<Range> ranges) { + throw new java.lang.UnsupportedOperationException("function not implemented yet"); + } @Override - public void updatePartitionOwner(String partition, String owner){} + public void appendRange(String rangeId, List<Range> ranges) { + throw new java.lang.UnsupportedOperationException("function not implemented yet"); + } @Override - public void updatePartitionReplicationFactor(String partition, int replicationFactor){} + public void relinquish(String ownerId, String rangeId) { + throw new java.lang.UnsupportedOperationException("function not implemented yet"); + } + + @Override + public MusicRangeInformationRow getMusicRangeInformation(UUID id){ + return null; + } } diff --git a/src/main/java/com/att/research/mdbc/mixins/MySQLMixin.java b/src/main/java/com/att/research/mdbc/mixins/MySQLMixin.java index 4f70147..5615ffb 100755 --- a/src/main/java/com/att/research/mdbc/mixins/MySQLMixin.java +++ b/src/main/java/com/att/research/mdbc/mixins/MySQLMixin.java @@ -586,7 +586,7 @@ NEW.field refers to the new value if(!transactionDigests.containsKey(range)) { transactionDigests.put(range, new StagingTable()); } - transactionDigests.get(range).addOperation(musicKey, opType, keydataStr, newRow.toString()); + transactionDigests.get(range).addOperation(musicKey, opType, newRow.toString()); rows.add(ix); } rs.getStatement().close(); diff --git a/src/main/java/com/att/research/mdbc/tables/MriReference.java b/src/main/java/com/att/research/mdbc/tables/MriReference.java new file mode 100644 index 0000000..a1d0b61 --- /dev/null +++ b/src/main/java/com/att/research/mdbc/tables/MriReference.java @@ -0,0 +1,14 @@ +package com.att.research.mdbc.tables; + +import java.util.UUID; + +public final class MriReference { + public final String table; + public final UUID index; + + public MriReference(String table, UUID index) { + this.table = table; + this.index= index; + } + +} diff --git a/src/main/java/com/att/research/mdbc/tables/MusicRangeInformationRow.java b/src/main/java/com/att/research/mdbc/tables/MusicRangeInformationRow.java new file mode 100644 index 0000000..e069666 --- /dev/null +++ b/src/main/java/com/att/research/mdbc/tables/MusicRangeInformationRow.java @@ -0,0 +1,16 @@ +package com.att.research.mdbc.tables; + +import java.util.List; +import java.util.UUID; + +public final class MusicRangeInformationRow { + public final UUID index; + public final PartitionInformation partition; + public final List<MusixTxDigestId> redoLog; + + public MusicRangeInformationRow(UUID index, List<MusixTxDigestId> redoLog, PartitionInformation partition) { + this.index = index; + this.redoLog = redoLog; + this.partition = partition; + } +} diff --git a/src/main/java/com/att/research/mdbc/tables/MusixTxDigestId.java b/src/main/java/com/att/research/mdbc/tables/MusixTxDigestId.java new file mode 100644 index 0000000..71e715a --- /dev/null +++ b/src/main/java/com/att/research/mdbc/tables/MusixTxDigestId.java @@ -0,0 +1,15 @@ +package com.att.research.mdbc.tables; + +import java.util.UUID; + +public final class MusixTxDigestId { + public final UUID tablePrimaryKey; + + public MusixTxDigestId(UUID primaryKey) { + this.tablePrimaryKey= primaryKey; + } + + public boolean isEmpty() { + return (this.tablePrimaryKey==null); + } +} diff --git a/src/main/java/com/att/research/mdbc/tables/Operation.java b/src/main/java/com/att/research/mdbc/tables/Operation.java index 026fa40..abb21c8 100644 --- a/src/main/java/com/att/research/mdbc/tables/Operation.java +++ b/src/main/java/com/att/research/mdbc/tables/Operation.java @@ -10,13 +10,11 @@ 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) { + public Operation(OperationType type, String newVal) { TYPE = type; NEW_VAL = newVal; - OLD_VAL = oldVal; } public JSONObject getNewVal(){ @@ -24,11 +22,6 @@ public final class Operation implements Serializable{ return newRow; } - public JSONObject getOldVal(){ - JSONObject keydata = new JSONObject(new JSONTokener(OLD_VAL)); - return keydata; - } - public OperationType getOperationType() { return this.TYPE; } diff --git a/src/main/java/com/att/research/mdbc/tables/PartitionInformation.java b/src/main/java/com/att/research/mdbc/tables/PartitionInformation.java index 9249844..0e4b80d 100644 --- a/src/main/java/com/att/research/mdbc/tables/PartitionInformation.java +++ b/src/main/java/com/att/research/mdbc/tables/PartitionInformation.java @@ -3,17 +3,9 @@ package com.att.research.mdbc.tables; 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; + + public PartitionInformation(List<String> tables) { this.tables=tables; - this.replicationFactor=replicationFactor; - this.currentOwner=currentOwner; } } diff --git a/src/main/java/com/att/research/mdbc/tables/RedoHistoryElement.java b/src/main/java/com/att/research/mdbc/tables/RedoHistoryElement.java deleted file mode 100644 index 8d92216..0000000 --- a/src/main/java/com/att/research/mdbc/tables/RedoHistoryElement.java +++ /dev/null @@ -1,15 +0,0 @@ -package com.att.research.mdbc.tables; - -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; - } -} diff --git a/src/main/java/com/att/research/mdbc/tables/RedoRecordId.java b/src/main/java/com/att/research/mdbc/tables/RedoRecordId.java deleted file mode 100644 index 225c89e..0000000 --- a/src/main/java/com/att/research/mdbc/tables/RedoRecordId.java +++ /dev/null @@ -1,15 +0,0 @@ -package com.att.research.mdbc.tables; - -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()); - } -} diff --git a/src/main/java/com/att/research/mdbc/tables/StagingTable.java b/src/main/java/com/att/research/mdbc/tables/StagingTable.java index c16f11c..f5e5101 100644 --- a/src/main/java/com/att/research/mdbc/tables/StagingTable.java +++ b/src/main/java/com/att/research/mdbc/tables/StagingTable.java @@ -23,11 +23,11 @@ public class StagingTable implements Serializable{ operations = new HashMap<>(); } - synchronized public void addOperation(String key, OperationType type, String oldVal, String newVal) { + synchronized public void addOperation(String key, OperationType type, String newVal) { if(!operations.containsKey(key)) { operations.put(key, new LinkedList<>()); } - operations.get(key).add(new Operation(type,newVal,oldVal)); + operations.get(key).add(new Operation(type,newVal)); } synchronized public Deque<Pair<String,Operation>> getIterableSnapshot() throws NoSuchFieldException{ diff --git a/src/main/java/com/att/research/mdbc/tables/TablePartitionInformation.java b/src/main/java/com/att/research/mdbc/tables/TablePartitionInformation.java deleted file mode 100644 index 9201de5..0000000 --- a/src/main/java/com/att/research/mdbc/tables/TablePartitionInformation.java +++ /dev/null @@ -1,15 +0,0 @@ -package com.att.research.mdbc.tables; - -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; - } -} diff --git a/src/main/java/com/att/research/mdbc/tables/TitReference.java b/src/main/java/com/att/research/mdbc/tables/TitReference.java deleted file mode 100644 index 2abb989..0000000 --- a/src/main/java/com/att/research/mdbc/tables/TitReference.java +++ /dev/null @@ -1,12 +0,0 @@ -package com.att.research.mdbc.tables; - -public final class TitReference { - public final String table; - public final String index; - - public TitReference(String table, String index) { - this.table = table; - this.index= index; - } - -} diff --git a/src/main/java/com/att/research/mdbc/tables/TransactionInformationElement.java b/src/main/java/com/att/research/mdbc/tables/TransactionInformationElement.java deleted file mode 100644 index a80cab1..0000000 --- a/src/main/java/com/att/research/mdbc/tables/TransactionInformationElement.java +++ /dev/null @@ -1,19 +0,0 @@ -package com.att.research.mdbc.tables; - -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; - } -} diff --git a/src/main/java/com/att/research/mdbc/tables/TxCommitProgress.java b/src/main/java/com/att/research/mdbc/tables/TxCommitProgress.java index cecdb08..d87e33d 100644 --- a/src/main/java/com/att/research/mdbc/tables/TxCommitProgress.java +++ b/src/main/java/com/att/research/mdbc/tables/TxCommitProgress.java @@ -2,8 +2,10 @@ package com.att.research.mdbc.tables; import java.math.BigInteger; import java.util.Map; +import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.atomic.AtomicLong; +import com.datastax.driver.core.utils.UUIDs; + import com.att.research.logging.EELFLoggerDelegate; import java.sql.Connection; @@ -13,11 +15,9 @@ 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<>(); } @@ -25,12 +25,12 @@ public class TxCommitProgress{ return transactionInfo.containsKey(txId); } - public BigInteger getCommitId(String txId) { + public UUID getCommitId(String txId) { CommitProgress prog = transactionInfo.get(txId); if(prog.isCommitIdAssigned()) { return prog.getCommitId(); } - BigInteger commitId = nextCommitId.getAndUpdate((a)-> a.add(BigInteger.ONE)); + UUID commitId = UUIDs.random(); prog.setCommitId(commitId); return commitId; } @@ -71,7 +71,7 @@ public class TxCommitProgress{ return prog.getConnection(); } - public void setRecordId(String txId, RedoRecordId recordId){ + public void setRecordId(String txId, MusixTxDigestId recordId){ CommitProgress prog = transactionInfo.get(txId); if(prog == null){ logger.error(EELFLoggerDelegate.errorLogger, "Transaction doesn't exist: [%l], failure when setting record Id",txId); @@ -79,7 +79,7 @@ public class TxCommitProgress{ prog.setRecordId(recordId); } - public RedoRecordId getRecordId(String txId) { + public MusixTxDigestId 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); @@ -118,16 +118,16 @@ public class TxCommitProgress{ final class CommitProgress{ private String lTxId; // local transaction id - private BigInteger commitId; // commit id + private UUID 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 + private MusixTxDigestId musixTxDigestId;// record id for each partition public CommitProgress(String id,Connection conn){ - redoRecordId=null; + musixTxDigestId =null; lTxId = id; commitRequested = false; SQLDone = false; @@ -141,14 +141,14 @@ final class CommitProgress{ return commitRequested && SQLDone && MusicDone; } - public synchronized void setCommitId(BigInteger commitId) { + public synchronized void setCommitId(UUID commitId) { this.commitId = commitId; timestamp = System.currentTimeMillis(); } public synchronized void reinitialize() { commitId = null; - redoRecordId=null; + musixTxDigestId =null; commitRequested = false; SQLDone = false; MusicDone = false; @@ -179,20 +179,20 @@ final class CommitProgress{ return timestamp; } - public synchronized void setRecordId(RedoRecordId id) { - redoRecordId = id; + public synchronized void setRecordId(MusixTxDigestId id) { + musixTxDigestId = id; timestamp = System.currentTimeMillis(); } public synchronized boolean isRedoRecordAssigned() { - return this.redoRecordId!=null; + return this.musixTxDigestId !=null; } - public synchronized RedoRecordId getRecordId() { - return redoRecordId; + public synchronized MusixTxDigestId getRecordId() { + return musixTxDigestId; } - public synchronized BigInteger getCommitId() { + public synchronized UUID getCommitId() { return commitId; } diff --git a/src/test/java/com/att/research/mdbc/MDBCUtilsTest.java b/src/test/java/com/att/research/mdbc/MDBCUtilsTest.java index 28af754..7e25fe0 100644 --- a/src/test/java/com/att/research/mdbc/MDBCUtilsTest.java +++ b/src/test/java/com/att/research/mdbc/MDBCUtilsTest.java @@ -17,7 +17,7 @@ public class MDBCUtilsTest { @Test public void toStringTest1() { StagingTable table = new StagingTable(); - table.addOperation("test",OperationType.INSERT,(new JSONObject(new String[]{"test1", "test2"})).toString(),(new JSONObject(new String[]{"test3", "Test4"})).toString()); + table.addOperation("test",OperationType.INSERT,(new JSONObject(new String[]{"test3", "Test4"})).toString()); String output=null; try { output = MDBCUtils.toString(table); @@ -33,7 +33,7 @@ public class MDBCUtilsTest { public void toStringTest2() { HashMap<String,StagingTable> mapToSerialize = new HashMap<>(); StagingTable table = new StagingTable(); - table.addOperation("test",OperationType.INSERT,(new JSONObject(new String[]{"test1", "test2"})).toString(),(new JSONObject(new String[]{"test3", "Test4"})).toString()); + table.addOperation("test",OperationType.INSERT,(new JSONObject(new String[]{"test3", "Test4"})).toString()); mapToSerialize.put("table",table); String output=null; try { |