refactor to move common methods of transactions and micro-tx to inter…

…nal interface

client/src/main/java/com/orientechnologies/orient/client/remote/OStorageRemote.java

@@ -61,8 +61,8 @@
 import com.orientechnologies.orient.core.storage.impl.local.paginated.ORecordSerializationContext;
 import com.orientechnologies.orient.core.storage.ridbag.sbtree.OBonsaiCollectionPointer;
 import com.orientechnologies.orient.core.storage.ridbag.sbtree.OSBTreeCollectionManager;
-import com.orientechnologies.orient.core.tx.OTransaction;
 import com.orientechnologies.orient.core.tx.OTransactionAbstract;
+import com.orientechnologies.orient.core.tx.OTransactionInternal;
 import com.orientechnologies.orient.core.tx.OTransactionOptimistic;
 import com.orientechnologies.orient.enterprise.channel.binary.OChannelBinary;
 import com.orientechnologies.orient.enterprise.channel.binary.OChannelBinaryProtocol;
@@ -901,11 +901,11 @@ public void fetchNextPage(ODatabaseDocumentRemote database, ORemoteResultSet rs)
     rs.fetched(response.getResult(), response.isHasNextPage(), response.getExecutionPlan(), response.getQueryStats());
   }
 
-  public List<ORecordOperation> commit(final OTransaction iTx, final Runnable callback) {
+  public List<ORecordOperation> commit(final OTransactionInternal iTx, final Runnable callback) {
     OCommit37Request request;
     if (((OTransactionOptimistic) iTx).isChanged()) {
-      request = new OCommit37Request(iTx.getId(), true, iTx.isUsingLog(), (Iterable<ORecordOperation>) iTx.getAllRecordEntries(),
-          ((OTransactionOptimistic) iTx).getIndexEntries());
+      request = new OCommit37Request(iTx.getId(), true, iTx.isUsingLog(), (Iterable<ORecordOperation>) iTx.getRecordOperations(),
+          ((OTransactionOptimistic) iTx).getIndexOperations());
     } else {
       request = new OCommit37Request(iTx.getId(), false, iTx.isUsingLog(), null, null);
     }
@@ -933,15 +933,15 @@ public List<ORecordOperation> commit(final OTransaction iTx, final Runnable call
     updateCollectionsFromChanges(((OTransactionOptimistic) iTx).getDatabase().getSbTreeCollectionManager(),
         response.getCollectionChanges());
     // SET ALL THE RECORDS AS UNDIRTY
-    for (ORecordOperation txEntry : iTx.getAllRecordEntries())
+    for (ORecordOperation txEntry : iTx.getRecordOperations())
       ORecordInternal.unsetDirty(txEntry.getRecord());
 
     // UPDATE THE CACHE ONLY IF THE ITERATOR ALLOWS IT. 
-    OTransactionAbstract.updateCacheFromEntries(iTx, iTx.getAllRecordEntries(), true);
+    OTransactionAbstract.updateCacheFromEntries(iTx.getDatabase(), iTx.getRecordOperations(), true);
     return null;
   }
 
-  public void rollback(OTransaction iTx) {
+  public void rollback(OTransactionInternal iTx) {
     if (((OTransactionOptimistic) iTx).isAlreadyCleared()) {
       ORollbackTransactionRequest request = new ORollbackTransactionRequest(iTx.getId());
       ORollbackTransactionResponse response = networkOperation(request, "Error on fetching next page for statment: " + request);
@@ -1823,7 +1823,7 @@ public void addNewClusterToConfiguration(int clusterId, String iClusterName) {
 
   public void beginTransaction(ODatabaseDocumentRemote database, OTransactionOptimistic transaction) {
     OBeginTransactionRequest request = new OBeginTransactionRequest(transaction.getId(), true, transaction.isUsingLog(),
-        transaction.getAllRecordEntries(), transaction.getIndexEntries());
+        transaction.getRecordOperations(), transaction.getIndexOperations());
     OBeginTransactionResponse response = networkOperationNoRetry(request, "Error on remote treansaction begin");
     for (Map.Entry<ORID, ORID> entry : response.getUpdatedIds().entrySet()) {
       transaction.updateIdentityAfterCommit(entry.getKey(), entry.getValue());
@@ -1832,7 +1832,7 @@ public void beginTransaction(ODatabaseDocumentRemote database, OTransactionOptim
 
   public void reBeginTransaction(ODatabaseDocumentRemote database, OTransactionOptimistic transaction) {
     ORebeginTransactionRequest request = new ORebeginTransactionRequest(transaction.getId(), transaction.isUsingLog(),
-        transaction.getAllRecordEntries(), transaction.getIndexEntries());
+        transaction.getRecordOperations(), transaction.getIndexOperations());
     OBeginTransactionResponse response = networkOperationNoRetry(request, "Error on remote treansaction begin");
     for (Map.Entry<ORID, ORID> entry : response.getUpdatedIds().entrySet()) {
       transaction.updateIdentityAfterCommit(entry.getKey(), entry.getValue());

core/src/main/java/com/orientechnologies/orient/core/db/document/ODatabaseDocumentAbstract.java

@@ -2492,7 +2492,7 @@ public long countClass(final String iClassName, final boolean iPolymorphic) {
     long deletedInTx = 0;
     long addedInTx = 0;
     if (getTransaction().isActive())
-      for (ORecordOperation op : getTransaction().getAllRecordEntries()) {
+      for (ORecordOperation op : getTransaction().getRecordOperations()) {
         if (op.type == ORecordOperation.DELETED) {
           final ORecord rec = op.getRecord();
           if (rec != null && rec instanceof ODocument) {

core/src/main/java/com/orientechnologies/orient/core/index/OIndexTxAwareMultiValue.java

@@ -306,7 +306,7 @@ public Set<OIdentifiable> get(Object key) {
     if (indexChanges == null) {
       Set<OIdentifiable> res = super.get(key);
       //In case of active transaction we use to return null instead of empty list, make check to be backward compatible
-      if (database.getTransaction().isActive() && ((OTransactionOptimistic) database.getTransaction()).getIndexEntries().size() != 0
+      if (database.getTransaction().isActive() && ((OTransactionOptimistic) database.getTransaction()).getIndexOperations().size() != 0
           && res.isEmpty())
         return null;
       return res;

core/src/main/java/com/orientechnologies/orient/core/sql/executor/FetchTemporaryFromTxStep.java

@@ -98,7 +98,7 @@ private void init() {
     long begin = profilingEnabled ? System.nanoTime() : 0;
     try {
       if (this.txEntries == null) {
-        Iterable<? extends ORecordOperation> iterable = ctx.getDatabase().getTransaction().getAllRecordEntries();
+        Iterable<? extends ORecordOperation> iterable = ctx.getDatabase().getTransaction().getRecordOperations();
 
         List<ORecord> records = new ArrayList<>();
         if (iterable != null) {

core/src/main/java/com/orientechnologies/orient/core/storage/OBasicTransaction.java

@@ -136,4 +136,13 @@ void addIndexEntry(OIndex<?> index, String indexName, OTransactionIndexChanges.O
    */
   void setCustomData(String name, Object value);
 
+  /**
+   * @return this transaction ID as seen by the client of this transaction.
+   */
+  default int getClientTransactionId() {
+    return getId();
+  }
+
+  int getId();
+
 }

core/src/main/java/com/orientechnologies/orient/core/storage/OStorage.java

@@ -30,7 +30,7 @@
 import com.orientechnologies.orient.core.id.ORID;
 import com.orientechnologies.orient.core.id.ORecordId;
 import com.orientechnologies.orient.core.storage.ridbag.sbtree.OSBTreeCollectionManager;
-import com.orientechnologies.orient.core.tx.OTransaction;
+import com.orientechnologies.orient.core.tx.OTransactionInternal;
 import com.orientechnologies.orient.core.util.OBackupable;
 
 import java.io.IOException;
@@ -102,10 +102,10 @@ OStorageOperationResult<Integer> updateRecord(ORecordId iRecordId, boolean updat
   boolean cleanOutRecord(ORecordId recordId, int recordVersion, int iMode, ORecordCallback<Boolean> callback);
 
   // TX OPERATIONS
-  List<ORecordOperation> commit(OTransaction iTx, Runnable callback);
+  List<ORecordOperation> commit(OTransactionInternal iTx, Runnable callback);
 
   // TX OPERATIONS
-  void rollback(OTransaction iTx);
+  void rollback(OTransactionInternal iTx);
 
   // MISC
   OStorageConfiguration getConfiguration();

core/src/main/java/com/orientechnologies/orient/core/storage/impl/local/OAbstractPaginatedStorage.java

@@ -88,7 +88,7 @@
 import com.orientechnologies.orient.core.tx.OTransaction;
 import com.orientechnologies.orient.core.tx.OTransactionAbstract;
 import com.orientechnologies.orient.core.tx.OTransactionIndexChanges;
-import com.orientechnologies.orient.core.tx.OTransactionOptimistic;
+import com.orientechnologies.orient.core.tx.OTransactionInternal;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 
 import java.io.*;
@@ -1487,7 +1487,7 @@ public void preallocateRids(final OTransaction clientTx) {
       checkLowDiskSpaceRequestsAndBackgroundDataFlushExceptionsAndBrokenPages();
 
       @SuppressWarnings("unchecked")
-      final Iterable<ORecordOperation> entries = (Iterable<ORecordOperation>) clientTx.getAllRecordEntries();
+      final Iterable<ORecordOperation> entries = (Iterable<ORecordOperation>) clientTx.getRecordOperations();
       final TreeMap<Integer, OCluster> clustersToLock = new TreeMap<>();
 
       final Set<ORecordOperation> newRecords = new TreeSet<>(COMMIT_RECORD_OPERATION_COMPARATOR);
@@ -1562,7 +1562,7 @@ public void preallocateRids(final OTransaction clientTx) {
    * @return The list of operations applied by the transaction
    */
   @Override
-  public List<ORecordOperation> commit(final OTransaction clientTx, Runnable callback) {
+  public List<ORecordOperation> commit(final OTransactionInternal clientTx, Runnable callback) {
     return commit(clientTx, false);
   }
 
@@ -1574,7 +1574,7 @@ public List<ORecordOperation> commit(final OTransaction clientTx, Runnable callb
    * @return The list of operations applied by the transaction
    */
   @SuppressWarnings("UnusedReturnValue")
-  public List<ORecordOperation> commitPreAllocated(final OTransaction clientTx) {
+  public List<ORecordOperation> commitPreAllocated(final OTransactionInternal clientTx) {
     return commit(clientTx, true);
   }
 
@@ -1595,7 +1595,7 @@ public List<ORecordOperation> commitPreAllocated(final OTransaction clientTx) {
    *
    * @return The list of operations applied by the transaction
    */
-  private List<ORecordOperation> commit(final OTransaction clientTx, boolean allocated) {
+  private List<ORecordOperation> commit(final OTransactionInternal clientTx, boolean allocated) {
     // XXX: At this moment, there are two implementations of the commit method. One for regular client transactions and one for
     // implicit micro-transactions. The implementations are quite identical, but operate on slightly different data. If you change
     // this method don't forget to change its counterpart:
@@ -1615,7 +1615,7 @@ private List<ORecordOperation> commit(final OTransaction clientTx, boolean alloc
       databaseRecord.getMetadata().makeThreadLocalSchemaSnapshot();
 
       @SuppressWarnings("unchecked")
-      final Iterable<ORecordOperation> entries = (Iterable<ORecordOperation>) clientTx.getAllRecordEntries();
+      final Iterable<ORecordOperation> entries = (Iterable<ORecordOperation>) clientTx.getRecordOperations();
       final TreeMap<Integer, OCluster> clustersToLock = new TreeMap<>();
       final Map<ORecordOperation, Integer> clusterOverrides = new IdentityHashMap<>();
 
@@ -1737,7 +1737,7 @@ private List<ORecordOperation> commit(final OTransaction clientTx, boolean alloc
                 });
             }
 
-            OTransactionAbstract.updateCacheFromEntries(clientTx, entries, true);
+            OTransactionAbstract.updateCacheFromEntries(clientTx.getDatabase(), entries, true);
 
             txCommit.incrementAndGet();
 
@@ -1872,7 +1872,7 @@ public void commit(OMicroTransaction microTransaction) {
 
                 rid.setClusterPosition(physicalPosition.clusterPosition);
 
-                microTransaction.updateIdentityAfterRecordCommit(oldRID, rid);
+                microTransaction.updateIdentityAfterCommit(oldRID, rid);
               }
             }
 
@@ -1887,7 +1887,7 @@ public void commit(OMicroTransaction microTransaction) {
 
             endStorageTx();
 
-            microTransaction.updateRecordCacheAfterCommit();
+            OTransactionAbstract.updateCacheFromEntries(microTransaction.getDatabase(), microTransaction.getRecordOperations(), true);
 
             txCommit.incrementAndGet();
 
@@ -1946,13 +1946,8 @@ private void commitIndexes(final Map<String, OTransactionIndexChanges> indexesTo
     }
   }
 
-  private TreeMap<String, OTransactionIndexChanges> getSortedIndexOperations(OTransaction clientTx) {
-    assert clientTx instanceof OTransactionOptimistic;
-    return new TreeMap<>(((OTransactionOptimistic) clientTx).getIndexEntries());
-  }
-
-  private TreeMap<String, OTransactionIndexChanges> getSortedIndexOperations(OMicroTransaction microTransaction) {
-    return new TreeMap<>(microTransaction.getIndexOperations());
+  private TreeMap<String, OTransactionIndexChanges> getSortedIndexOperations(OTransactionInternal clientTx) {
+    return new TreeMap<>(clientTx.getIndexOperations());
   }
 
   public int loadIndexEngine(String name) {
@@ -2914,7 +2909,7 @@ private boolean doHasRangeQuerySupport(int indexId) throws OInvalidIndexEngineId
     return engine.hasRangeQuerySupport();
   }
 
-  private void makeRollback(OTransaction clientTx, Exception e) {
+  private void makeRollback(OTransactionInternal clientTx, Exception e) {
     // WE NEED TO CALL ROLLBACK HERE, IN THE LOCK
     OLogManager.instance()
         .debug(this, "Error during transaction commit, transaction will be rolled back (tx-id=%d)", e, clientTx.getId());
@@ -2945,7 +2940,7 @@ private void makeRollback(OMicroTransaction microTransaction, Exception e) {
   }
 
   @Override
-  public void rollback(final OTransaction clientTx) {
+  public void rollback(final OTransactionInternal clientTx) {
     try {
       checkOpenness();
       stateLock.acquireReadLock();
@@ -2963,7 +2958,7 @@ public void rollback(final OTransaction clientTx) {
           makeStorageDirty();
           rollbackStorageTx();
 
-          OTransactionAbstract.updateCacheFromEntries(clientTx, clientTx.getAllRecordEntries(), false);
+          OTransactionAbstract.updateCacheFromEntries(clientTx.getDatabase(), clientTx.getRecordOperations(), false);
 
           txRollback.incrementAndGet();
 
@@ -3911,7 +3906,7 @@ private OLogSequenceNumber endStorageTx() throws IOException {
     return lsn;
   }
 
-  private void startStorageTx(OTransaction clientTx) throws IOException {
+  private void startStorageTx(OTransactionInternal clientTx) throws IOException {
     final OStorageTransaction storageTx = transaction.get();
     if (storageTx != null && storageTx.getClientTx().getId() != clientTx.getId())
       rollback(clientTx);

core/src/main/java/com/orientechnologies/orient/core/storage/impl/local/OMicroTransaction.java

@@ -46,9 +46,7 @@
 import com.orientechnologies.orient.core.record.impl.ODocumentInternal;
 import com.orientechnologies.orient.core.storage.OBasicTransaction;
 import com.orientechnologies.orient.core.storage.ORecordCallback;
-import com.orientechnologies.orient.core.tx.OTransactionIndexChanges;
-import com.orientechnologies.orient.core.tx.OTransactionIndexChangesPerKey;
-import com.orientechnologies.orient.core.tx.OTransactionRecordIndexOperation;
+import com.orientechnologies.orient.core.tx.*;
 
 import java.util.*;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -59,7 +57,7 @@
  *
  * @author Sergey Sitnikov
  */
-public final class OMicroTransaction implements OBasicTransaction {
+public final class OMicroTransaction implements OBasicTransaction, OTransactionInternal {
 
   private static final AtomicInteger transactionSerial = new AtomicInteger(0);
 
@@ -113,6 +111,16 @@ public int getId() {
     return id;
   }
 
+  @Override
+  public int getClientTransactionId() {
+    return -1;
+  }
+
+  @Override
+  public void updateIdentityAfterCommit(ORID oldRID, ORID rid) {
+    updateIdentityAfterRecordCommit(oldRID, rid);
+  }
+
   /**
    * @return the micro-transaction's database.
    */
@@ -123,7 +131,7 @@ public ODatabaseDocumentInternal getDatabase() {
   /**
    * @return the record operations done in the context of this micro-transaction.
    */
-  public Iterable<ORecordOperation> getRecordOperations() { // ordered by operation time
+  public Collection<ORecordOperation> getRecordOperations() { // ordered by operation time
     return recordOperations.values();
   }
 
@@ -269,20 +277,6 @@ public void updateIdentityAfterRecordCommit(final ORID oldRid, final ORID newRid
     }
   }
 
-  /**
-   * Updates the record cache after successful micro-transaction commit.
-   */
-  public void updateRecordCacheAfterCommit() {
-    final OLocalRecordCache databaseLocalCache = database.getLocalCache();
-
-    for (ORecordOperation recordOperation : recordOperations.values()) {
-      if (recordOperation.type == ORecordOperation.DELETED)
-        databaseLocalCache.deleteRecord(recordOperation.getRecord().getIdentity());
-      else if (recordOperation.type == ORecordOperation.UPDATED || recordOperation.type == ORecordOperation.CREATED)
-        databaseLocalCache.updateRecord(recordOperation.getRecord());
-    }
-  }
-
   /**
    * Updates the record cache after unsuccessful micro-transaction commit.
    */
@@ -711,6 +705,11 @@ private static boolean isIndexKeyMayDependOnRid(Object key, ORID rid, Dependency
     return dependency == Dependency.Unknown || dependency == null;
   }
 
+  @Override
+  public void setStatus(OTransaction.TXSTATUS iStatus) {
+    //IGNORE
+  }
+
   private enum Dependency {
     Unknown, Yes, No
   }
@@ -725,4 +724,13 @@ public KeyChangesUpdateRecord(OTransactionIndexChangesPerKey keyChanges, OTransa
     }
   }
 
+  @Override
+  public boolean isUsingLog() {
+    return true;
+  }
+
+  @Override
+  public ORecordOperation getRecordEntry(ORID currentRid) {
+    return recordOperations.get(currentRid);
+  }
 }

core/src/main/java/com/orientechnologies/orient/core/storage/impl/local/paginated/OStorageTransaction.java

@@ -16,17 +16,17 @@
 package com.orientechnologies.orient.core.storage.impl.local.paginated;
 
 import com.orientechnologies.orient.core.storage.impl.local.OMicroTransaction;
-import com.orientechnologies.orient.core.tx.OTransaction;
+import com.orientechnologies.orient.core.tx.OTransactionInternal;
 
 /**
  * @author Andrey Lomakin (a.lomakin-at-orientdb.com)
  * @since 12.06.13
  */
 public class OStorageTransaction {
-  private final OTransaction      clientTx;
-  private final OMicroTransaction microTransaction;
+  private final OTransactionInternal clientTx;
+  private final OMicroTransaction    microTransaction;
 
-  public OStorageTransaction(OTransaction clientTx) {
+  public OStorageTransaction(OTransactionInternal clientTx) {
     this.clientTx = clientTx;
     this.microTransaction = null;
   }
@@ -41,7 +41,7 @@ public OStorageTransaction(OMicroTransaction microTransaction) {
     this.clientTx = null;
   }
 
-  public OTransaction getClientTx() {
+  public OTransactionInternal getClientTx() {
     return clientTx;
   }
 

core/src/main/java/com/orientechnologies/orient/core/storage/impl/local/paginated/atomicoperations/OAtomicOperationsManager.java

@@ -39,7 +39,7 @@
 import com.orientechnologies.orient.core.storage.impl.local.paginated.wal.OOperationUnitId;
 import com.orientechnologies.orient.core.storage.impl.local.paginated.wal.OWriteAheadLog;
 import com.orientechnologies.orient.core.storage.impl.local.statistic.OPerformanceStatisticManager;
-import com.orientechnologies.orient.core.tx.OTransaction;
+import com.orientechnologies.orient.core.tx.OTransactionInternal;
 
 import javax.management.*;
 import java.io.IOException;
@@ -554,7 +554,7 @@ private boolean useWal() {
     if (storageTransaction == null)
       return true;
 
-    final OTransaction clientTx = storageTransaction.getClientTx();
+    final OTransactionInternal clientTx = storageTransaction.getClientTx();
     return clientTx == null || clientTx.isUsingLog();
 
   }