IGNITE-13596 Flag to distinguish DataRecord on primary and backup add…

…ed (#8904)

modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerClusterByClassTest.java

@@ -1523,7 +1523,8 @@ private void corruptDataEntry(
                 new GridCacheVersion(),
                 0L,
                 partId,
-                updateCntr
+                updateCntr,
+                false
             );
 
             GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)ctx.shared().database();

modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java

@@ -2630,7 +2630,8 @@ private void corruptDataEntry(
                 new GridCacheVersion(),
                 0L,
                 partId,
-                updateCntr
+                updateCntr,
+                false
             );
 
             GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)ctx.shared().database();

modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/DataEntry.java

@@ -61,6 +61,10 @@ public class DataEntry {
     @GridToStringInclude
     protected long partCnt;
 
+    /** If {@code true} then change made on primary node. */
+    @GridToStringInclude
+    protected boolean primary;
+
     /** Constructor. */
     private DataEntry() {
         // No-op, used from factory methods.
@@ -76,6 +80,7 @@ private DataEntry() {
      * @param expireTime Expire time.
      * @param partId Partition ID.
      * @param partCnt Partition counter.
+     * @param primary {@code True} if node is primary for partition in the moment of logging.
      */
     public DataEntry(
         int cacheId,
@@ -86,7 +91,8 @@ public DataEntry(
         GridCacheVersion writeVer,
         long expireTime,
         int partId,
-        long partCnt
+        long partCnt,
+        boolean primary
     ) {
         this.cacheId = cacheId;
         this.key = key;
@@ -97,6 +103,7 @@ public DataEntry(
         this.expireTime = expireTime;
         this.partId = partId;
         this.partCnt = partCnt;
+        this.primary = primary;
 
         // Only READ, CREATE, UPDATE and DELETE operations should be stored in WAL.
         assert op == GridCacheOperation.READ || op == GridCacheOperation.CREATE || op == GridCacheOperation.UPDATE || op == GridCacheOperation.DELETE : op;
@@ -177,6 +184,13 @@ public long expireTime() {
         return expireTime;
     }
 
+    /**
+     * @return {@code True} if node is primary for partition in the moment of logging.
+     */
+    public boolean primary() {
+        return primary;
+    }
+
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(DataEntry.class, this);

modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/DataRecord.java

@@ -35,7 +35,7 @@ public class DataRecord extends TimeStampRecord {
 
     /** {@inheritDoc} */
     @Override public RecordType type() {
-        return RecordType.DATA_RECORD;
+        return RecordType.DATA_RECORD_V2;
     }
 
     /**

modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/LazyDataEntry.java

@@ -60,6 +60,7 @@ public class LazyDataEntry extends DataEntry implements MarshalledDataEntry {
      * @param expireTime Expire time.
      * @param partId Partition ID.
      * @param partCnt Partition counter.
+     * @param primary {@code True} if node is primary for partition in the moment of logging.
      */
     public LazyDataEntry(
         GridCacheSharedContext cctx,
@@ -73,9 +74,10 @@ public LazyDataEntry(
         GridCacheVersion writeVer,
         long expireTime,
         int partId,
-        long partCnt
+        long partCnt,
+        boolean primary
     ) {
-        super(cacheId, null, null, op, nearXidVer, writeVer, expireTime, partId, partCnt);
+        super(cacheId, null, null, op, nearXidVer, writeVer, expireTime, partId, partCnt, primary);
 
         this.cctx = cctx;
         this.keyType = keyType;

modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/MvccDataEntry.java

@@ -56,7 +56,7 @@ public MvccDataEntry(
         long partCnt,
         MvccVersion mvccVer
     ) {
-        super(cacheId, key, val, op, nearXidVer, writeVer, expireTime, partId, partCnt);
+        super(cacheId, key, val, op, nearXidVer, writeVer, expireTime, partId, partCnt, false);
 
         this.mvccVer = mvccVer;
     }

modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/UnwrapDataEntry.java

@@ -48,6 +48,7 @@ public class UnwrapDataEntry extends DataEntry implements UnwrappedDataEntry {
      * @param partCnt Partition counter.
      * @param cacheObjValCtx cache object value context for unwrapping objects.
      * @param keepBinary disable unwrapping for non primitive objects, Binary Objects would be returned instead.
+     * @param primary {@code True} if node is primary for partition in the moment of logging.
      */
     public UnwrapDataEntry(
         final int cacheId,
@@ -60,8 +61,9 @@ public UnwrapDataEntry(
         final int partId,
         final long partCnt,
         final CacheObjectValueContext cacheObjValCtx,
-        final boolean keepBinary) {
-        super(cacheId, key, val, op, nearXidVer, writeVer, expireTime, partId, partCnt);
+        final boolean keepBinary,
+        final boolean primary) {
+        super(cacheId, key, val, op, nearXidVer, writeVer, expireTime, partId, partCnt, primary);
         this.cacheObjValCtx = cacheObjValCtx;
         this.keepBinary = keepBinary;
     }

modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/WALRecord.java

@@ -45,7 +45,8 @@ public enum RecordType {
         /** */
         PAGE_RECORD(1, PHYSICAL),
 
-        /** */
+        /** @deprecated Use {@link #DATA_RECORD_V2} instead. */
+        @Deprecated
         DATA_RECORD(2, LOGICAL),
 
         /** Checkpoint (begin) record */
@@ -206,7 +207,11 @@ public enum RecordType {
         /** Encrypted WAL-record. */
         ENCRYPTED_RECORD(52, PHYSICAL),
 
-        /** Ecnrypted data record. */
+        /**
+         * Ecnrypted data record.
+         * @deprecated Use {@link #ENCRYPTED_DATA_RECORD_V3} instead.
+         */
+        @Deprecated
         ENCRYPTED_DATA_RECORD(53, LOGICAL),
 
         /** Mvcc data record. */
@@ -239,7 +244,11 @@ public enum RecordType {
         /** Encrypted WAL-record. */
         ENCRYPTED_RECORD_V2(63, PHYSICAL),
 
-        /** Ecnrypted data record. */
+        /**
+         * Ecnrypted data record.
+         * @deprecated Use {@link #ENCRYPTED_DATA_RECORD_V3} instead.
+         */
+        @Deprecated
         ENCRYPTED_DATA_RECORD_V2(64, LOGICAL),
 
         /** Master key change record containing multiple keys for single cache group. */
@@ -252,7 +261,13 @@ public enum RecordType {
         PARTITION_META_PAGE_DELTA_RECORD_V3(67, PHYSICAL),
 
         /** Index meta page delta record includes encryption status data. */
-        INDEX_META_PAGE_DELTA_RECORD(68, PHYSICAL);
+        INDEX_META_PAGE_DELTA_RECORD(68, PHYSICAL),
+
+        /** Data record V2. */
+        DATA_RECORD_V2(69, LOGICAL),
+
+        /** Ecnrypted data record. */
+        ENCRYPTED_DATA_RECORD_V3(70, LOGICAL);
 
         /** Index for serialization. Should be consistent throughout all versions. */
         private final int idx;

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

@@ -1569,7 +1569,7 @@ else if (interceptorVal != val0)
             updateCntr0 = nextPartitionCounter(tx, updateCntr);
 
             if (tx != null && cctx.group().persistenceEnabled() && cctx.group().walEnabled())
-                logPtr = logTxUpdate(tx, val, expireTime, updateCntr0);
+                logPtr = logTxUpdate(tx, val, expireTime, updateCntr0, topVer);
 
             update(val, expireTime, ttl, newVer, true);
 
@@ -1791,7 +1791,7 @@ protected Object keyValue(boolean cpy) {
             updateCntr0 = nextPartitionCounter(tx, updateCntr);
 
             if (tx != null && cctx.group().persistenceEnabled() && cctx.group().walEnabled())
-                logPtr = logTxUpdate(tx, null, 0, updateCntr0);
+                logPtr = logTxUpdate(tx, null, 0, updateCntr0, topVer);
 
             drReplicate(drType, null, newVer, topVer);
 
@@ -2148,7 +2148,7 @@ else if (ttl != CU.TTL_ZERO)
 
                 update(updated, expireTime, ttl, ver, true);
 
-                logUpdate(op, updated, ver, expireTime, 0);
+                logUpdate(op, updated, ver, expireTime, 0, cctx.affinity().affinityTopologyVersion());
 
                 if (evt) {
                     CacheObject evtOld = null;
@@ -2180,7 +2180,7 @@ else if (ttl != CU.TTL_ZERO)
 
                 update(null, CU.TTL_ETERNAL, CU.EXPIRE_TIME_ETERNAL, ver, true);
 
-                logUpdate(op, null, ver, CU.EXPIRE_TIME_ETERNAL, 0);
+                logUpdate(op, null, ver, CU.EXPIRE_TIME_ETERNAL, 0, cctx.affinity().affinityTopologyVersion());
 
                 if (evt) {
                     CacheObject evtOld = null;
@@ -3495,7 +3495,8 @@ else if (deletedUnlocked())
                             ver,
                             expireTime,
                             partition(),
-                            updateCntr
+                            updateCntr,
+                            cctx.affinity().primaryByPartition(cctx.localNode(), partition(), topVer)
                         )));
                     }
                 }
@@ -4323,9 +4324,16 @@ protected boolean storeValue(
      * @param writeVer Write version.
      * @param expireTime Expire time.
      * @param updCntr Update counter.
+     * @param topVer Topology version.
      */
-    protected void logUpdate(GridCacheOperation op, CacheObject val, GridCacheVersion writeVer, long expireTime, long updCntr)
-        throws IgniteCheckedException {
+    protected void logUpdate(
+        GridCacheOperation op,
+        CacheObject val,
+        GridCacheVersion writeVer,
+        long expireTime,
+        long updCntr,
+        AffinityTopologyVersion topVer
+    ) throws IgniteCheckedException {
         // We log individual updates only in ATOMIC cache.
         assert cctx.atomic();
 
@@ -4340,7 +4348,8 @@ protected void logUpdate(GridCacheOperation op, CacheObject val, GridCacheVersio
                     writeVer,
                     expireTime,
                     partition(),
-                    updCntr)));
+                    updCntr,
+                    cctx.affinity().primaryByPartition(cctx.localNode(), partition(), topVer))));
         }
         catch (StorageException e) {
             throw new IgniteCheckedException("Failed to log ATOMIC cache update [key=" + key + ", op=" + op +
@@ -4353,10 +4362,16 @@ protected void logUpdate(GridCacheOperation op, CacheObject val, GridCacheVersio
      * @param val Value.
      * @param expireTime Expire time (or 0 if not applicable).
      * @param updCntr Update counter.
+     * @param topVer Topology version.
      * @throws IgniteCheckedException In case of log failure.
      */
-    protected WALPointer logTxUpdate(IgniteInternalTx tx, CacheObject val, long expireTime, long updCntr)
-        throws IgniteCheckedException {
+    protected WALPointer logTxUpdate(
+        IgniteInternalTx tx,
+        CacheObject val,
+        long expireTime,
+        long updCntr,
+        AffinityTopologyVersion topVer
+    ) throws IgniteCheckedException {
         assert cctx.transactional() && !cctx.transactionalSnapshot();
 
         if (tx.local()) { // For remote tx we log all updates in batch: GridDistributedTxRemoteAdapter.commitIfLocked()
@@ -4375,7 +4390,8 @@ protected WALPointer logTxUpdate(IgniteInternalTx tx, CacheObject val, long expi
                 tx.writeVersion(),
                 expireTime,
                 key.partition(),
-                updCntr)));
+                updCntr,
+                cctx.affinity().primaryByPartition(cctx.localNode(), partition(), topVer))));
         }
         else
             return null;
@@ -6484,7 +6500,7 @@ else if (interceptorVal != updated0) {
 
             long updateCntr0 = entry.nextPartitionCounter(topVer, primary, false, updateCntr);
 
-            entry.logUpdate(op, updated, newVer, newExpireTime, updateCntr0);
+            entry.logUpdate(op, updated, newVer, newExpireTime, updateCntr0, topVer);
 
             if (!entry.isNear()) {
                 newRow = entry.localPartition().dataStore().createRow(
@@ -6571,7 +6587,7 @@ private void remove(@Nullable GridCacheVersionConflictContext<?, ?> conflictCtx,
 
             long updateCntr0 = entry.nextPartitionCounter(topVer, primary, false, updateCntr);
 
-            entry.logUpdate(op, null, newVer, 0, updateCntr0);
+            entry.logUpdate(op, null, newVer, 0, updateCntr0, topVer);
 
             if (oldVal != null) {
                 assert !entry.deletedUnlocked();

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

@@ -631,7 +631,8 @@ else if (conflictCtx.isMerge()) {
                                                     writeVersion(),
                                                     0,
                                                     txEntry.key().partition(),
-                                                    txEntry.updateCounter()
+                                                    txEntry.updateCounter(),
+                                                    cacheCtx.affinity().primaryByPartition(cctx.localNode(), txEntry.key().partition(), topVer)
                                                 ),
                                                 txEntry
                                             )

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

@@ -18,6 +18,7 @@
 package org.apache.ignite.internal.processors.cache.distributed.dht.colocated;
 
 import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.processors.cache.CacheObject;
 import org.apache.ignite.internal.processors.cache.GridCacheContext;
 import org.apache.ignite.internal.processors.cache.GridCacheOperation;
@@ -73,13 +74,14 @@ public void resetFromPrimary(CacheObject val, GridCacheVersion ver) {
     }
 
     /** {@inheritDoc} */
-    @Override protected void logUpdate(GridCacheOperation op, CacheObject val, GridCacheVersion writeVer, long expireTime, long updCntr) throws IgniteCheckedException {
+    @Override protected void logUpdate(GridCacheOperation op, CacheObject val, GridCacheVersion writeVer,
+        long expireTime, long updCntr, AffinityTopologyVersion topVer) {
         // No-op for detached entries, index is updated on primary or backup nodes.
     }
 
     /** {@inheritDoc} */
-    @Override protected WALPointer logTxUpdate(IgniteInternalTx tx, CacheObject val, long expireTime, long updCntr)
-        throws IgniteCheckedException {
+    @Override protected WALPointer logTxUpdate(IgniteInternalTx tx, CacheObject val, long expireTime, long updCntr,
+        AffinityTopologyVersion topVer) {
         return null;
     }
 

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

@@ -471,14 +471,14 @@ public boolean loadedValue(@Nullable IgniteInternalTx tx,
     }
 
     /** {@inheritDoc} */
-    @Override protected void logUpdate(GridCacheOperation op, CacheObject val, GridCacheVersion ver, long expireTime, long updCntr)
-        throws IgniteCheckedException {
+    @Override protected void logUpdate(GridCacheOperation op, CacheObject val, GridCacheVersion ver, long expireTime,
+        long updCntr, AffinityTopologyVersion topVer) {
         // No-op: queries are disabled for near cache.
     }
 
     /** {@inheritDoc} */
-    @Override protected WALPointer logTxUpdate(IgniteInternalTx tx, CacheObject val, long expireTime, long updCntr)
-        throws IgniteCheckedException {
+    @Override protected WALPointer logTxUpdate(IgniteInternalTx tx, CacheObject val, long expireTime, long updCntr,
+        AffinityTopologyVersion topVer) {
         return null;
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java

@@ -2496,6 +2496,7 @@ public void applyUpdatesOnRecovery(
                 switch (rec.type()) {
                     case MVCC_DATA_RECORD:
                     case DATA_RECORD:
+                    case DATA_RECORD_V2:
                         checkpointReadLock();
 
                         try {
@@ -2635,8 +2636,10 @@ private RestoreLogicalState applyLogicalUpdates(
 
                     case MVCC_DATA_RECORD:
                     case DATA_RECORD:
+                    case DATA_RECORD_V2:
                     case ENCRYPTED_DATA_RECORD:
                     case ENCRYPTED_DATA_RECORD_V2:
+                    case ENCRYPTED_DATA_RECORD_V3:
                         DataRecord dataRec = (DataRecord)rec;
 
                         for (DataEntry dataEntry : dataRec.writeEntries()) {

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/StandaloneWalRecordsIterator.java

@@ -354,7 +354,9 @@ private boolean checkBounds(long idx) {
         GridKernalContext kernalCtx = sharedCtx.kernalContext();
         IgniteCacheObjectProcessor processor = kernalCtx.cacheObjects();
 
-        if (processor != null && (rec.type() == RecordType.DATA_RECORD || rec.type() == RecordType.MVCC_DATA_RECORD)) {
+        if (processor != null && (rec.type() == RecordType.DATA_RECORD
+            || rec.type() == RecordType.DATA_RECORD_V2
+            || rec.type() == RecordType.MVCC_DATA_RECORD)) {
             try {
                 return postProcessDataRecord((DataRecord)rec, kernalCtx, processor);
             }
@@ -498,7 +500,8 @@ private DataEntry unwrapDataEntry(CacheObjectContext coCtx, DataEntry dataEntry,
                 dataEntry.partitionId(),
                 dataEntry.partitionCounter(),
                 coCtx,
-                keepBinary);
+                keepBinary,
+                dataEntry.primary());
     }
 
     /** {@inheritDoc} */