Transaction savepoint poc

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridNearUnlockRequest.java

@@ -36,6 +36,10 @@
     @Order(0)
     public List<KeyCacheObject> keys;
 
+    /** Savepoint rollback flag. */
+    @Order(1)
+    public boolean forSavepoint;
+
     /**
      * Empty constructor.
      */
@@ -60,6 +64,20 @@
         return keys;
     }
 
+    /**
+     * @return {@code True} if unlock request is sent during rollback to savepoint.
+     */
+    public boolean forSavepoint() {
+        return forSavepoint;
+    }
+
+    /**
+     * @param forSavepoint Savepoint rollback flag.
+     */
+    public void forSavepoint(boolean forSavepoint) {
+        this.forSavepoint = forSavepoint;
+    }
+
     /**
      * @param key Key.
      */

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java

@@ -59,6 +59,7 @@
 import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockResponse;
 import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
 import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTransactionalCache;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalEx;
@@ -580,6 +581,9 @@
     private void processDhtUnlockRequest(UUID nodeId, GridDhtUnlockRequest req) {
         clearLocks(nodeId, req);
 
+        if (req.forSavepoint())
+            clearTxEntries(req.version(), req.keys());
+
         if (isNearEnabled(cacheCfg))
             near().clearLocks(nodeId, req);
     }
@@ -1488,7 +1492,14 @@
         assert ctx.affinityNode();
         assert nodeId != null;
 
-        removeLocks(nodeId, req.version(), req.keys(), true);
+        removeLocks(
+            nodeId,
+            req.version(),
+            req.keys(),
+            //TODO: PVD:: Need to investigate this parameter.
+            true,
+            req.forSavepoint()
+        );
     }
 
     /**
@@ -1565,6 +1576,23 @@
      * @param unmap Flag for un-mapping version.
      */
     public void removeLocks(UUID nodeId, GridCacheVersion ver, Iterable<KeyCacheObject> keys, boolean unmap) {
+        removeLocks(nodeId, ver, keys, unmap, false);
+    }
+
+    /**
+     * @param nodeId Node ID.
+     * @param ver Version.
+     * @param keys Keys.
+     * @param unmap Flag for un-mapping version.
+     * @param forSavepoint Savepoint rollback flag.
+     */
+    public void removeLocks(
+        UUID nodeId,
+        GridCacheVersion ver,
+        Iterable<KeyCacheObject> keys,
+        boolean unmap,
+        boolean forSavepoint
+    ) {
         assert nodeId != null;
         assert ver != null;
 
@@ -1574,7 +1602,8 @@
         // Remove mapped versions.
         GridCacheVersion dhtVer = unmap ? ctx.mvcc().unmapVersion(ver) : ver;
 
-        ctx.mvcc().addRemoved(ctx, ver);
+        if (!forSavepoint)
+            ctx.mvcc().addRemoved(ctx, ver);
 
         Map<ClusterNode, List<KeyCacheObject>> dhtMap = new HashMap<>();
         Map<ClusterNode, List<KeyCacheObject>> nearMap = new HashMap<>();
@@ -1636,6 +1665,9 @@
                     // as there is no point to reorder relative to the version
                     // we are about to remove.
                     if (entry.removeLock(dhtVer)) {
+                        if (forSavepoint)
+                            clearTxEntry(dhtVer, key);
+
                         // Map to backups and near readers.
                         map(nodeId, topVer, entry, readers, dhtMap, nearMap);
 
@@ -1674,10 +1706,11 @@
             GridDhtUnlockRequest req = new GridDhtUnlockRequest(ctx.cacheId(), keyBytes.size());
 
             req.version(dhtVer);
+            req.forSavepoint(forSavepoint);
 
             try {
-                for (KeyCacheObject key : keyBytes)
-                    req.addKey(key);
+                for (int i = 0; i < keyBytes.size(); i++)
+                    req.addKey(keyBytes.get(i));
 
                 keyBytes = nearMap.get(n);
 
@@ -1708,6 +1741,7 @@
                 GridDhtUnlockRequest req = new GridDhtUnlockRequest(ctx.cacheId(), keyBytes.size());
 
                 req.version(dhtVer);
+                req.forSavepoint(forSavepoint);
 
                 try {
                     for (KeyCacheObject key : keyBytes)
@@ -1728,6 +1762,47 @@
         }
     }
 
+    /**
+     * @param ver Tx version.
+     * @param keys Keys to clear from remote tx.
+     */
+    private void clearTxEntries(GridCacheVersion ver, List<KeyCacheObject> keys) {
+        if (F.isEmpty(keys))
+            return;
+
+        for (KeyCacheObject key : keys)
+            clearTxEntry(ver, key);
+    }
+
+    /**
+     * @param ver Tx version.
+     * @param key Key.
+     */
+    private void clearTxEntry(GridCacheVersion ver, KeyCacheObject key) {
+        IgniteInternalTx tx = ctx.tm().tx(ver);
+
+        if (tx instanceof GridDhtTxLocal) {
+            ((GridDhtTxLocal)tx).clearEntry(ctx.txKey(key));
+
+            try {
+                if (tx.empty()) {
+                    ((GridDhtTxLocal)tx).rollbackDhtLocal();
+                }
+            }
+            catch (IgniteCheckedException e) {
+                U.error(log, "Failed to remove transaction container during rollback to savepoint: " + tx, e);
+            }
+        }
+        else if (configuration().getNearConfiguration() != null) {
+            try {
+                invalidateNearEntry(key, ver);
+            }
+            catch (IgniteCheckedException e) {
+                U.error(log, "Failed to invalidate near entry during rollback to savepoint: " + key, e);
+            }
+        }
+    }
+
     /**
      * @param key Key
      * @param ver Version.

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java

@@ -41,6 +41,7 @@
 import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareResponse;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalAdapter;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
 import org.apache.ignite.internal.util.F0;
@@ -388,6 +389,23 @@ boolean removeNearMapping(UUID nodeId, GridCacheEntryEx entry) {
         return removeMapping(nodeId, entry, nearMap);
     }
 
+    /**
+     * Removes tx entry from local DHT transaction state and all DHT/near mappings.
+     *
+     * @param key Tx key.
+     */
+    public void clearEntry(IgniteTxKey key) {
+        IgniteTxEntry txEntry = entry(key);
+
+        if (txEntry == null)
+            return;
+
+        removeEntryFromMappings(txEntry, dhtMap);
+        removeEntryFromMappings(txEntry, nearMap);
+
+        txState().removeEntry(key);
+    }
+
     /**
      * @param nodeId Node ID.
      * @param entry Entry to remove.
@@ -418,6 +436,19 @@ private boolean removeMapping(UUID nodeId, @Nullable GridCacheEntryEx entry,
             return map.remove(nodeId) != null;
     }
 
+    /**
+     * @param txEntry Entry.
+     * @param map Mappings.
+     */
+    private void removeEntryFromMappings(IgniteTxEntry txEntry, Map<UUID, GridDistributedTxMapping> map) {
+        for (Map.Entry<UUID, GridDistributedTxMapping> e : map.entrySet()) {
+            GridDistributedTxMapping mapping = e.getValue();
+
+            if (mapping.removeEntry(txEntry) && mapping.empty())
+                map.remove(e.getKey(), mapping);
+        }
+    }
+
     /**
      * @param mappings Entry mappings.
      * @param dst Transaction mappings.

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java

@@ -476,6 +476,30 @@
      * @param keys Keys.
      */
     public void removeLocks(GridCacheVersion ver, Collection<KeyCacheObject> keys) {
+        removeLocks(ver, keys, false);
+    }
+
+    /**
+     * Removes locks regardless of whether they are owned or not for given
+     * version and keys. In savepoint mode lock version is not marked as globally cancelled.
+     *
+     * @param ver Lock version.
+     * @param keys Keys.
+     */
+    public void removeLocksForSavepoint(GridCacheVersion ver, Collection<KeyCacheObject> keys) {
+        removeLocks(ver, keys, true);
+    }
+
+    /**
+     * @param ver Lock version.
+     * @param keys Keys.
+     * @param forSavepoint Savepoint rollback flag.
+     */
+    private void removeLocks(
+        GridCacheVersion ver,
+        Collection<KeyCacheObject> keys,
+        boolean forSavepoint
+    ) {
         if (keys.isEmpty())
             return;
 
@@ -523,13 +547,20 @@
                                         map.put(primary, req = new GridNearUnlockRequest(ctx.cacheId(), keyCnt));
 
                                         req.version(ver);
+                                        req.forSavepoint(forSavepoint);
                                     }
                                 }
 
                                 // Remove candidate from local node first.
                                 if (entry.removeLock(cand.version())) {
                                     if (primary.isLocal()) {
-                                        dht.removeLocks(primary.id(), ver, F.asList(key), true);
+                                        dht.removeLocks(
+                                            primary.id(),
+                                            ver,
+                                            F.asList(key),
+                                            true,
+                                            forSavepoint
+                                        );
 
                                         assert req == null;