IGNITE-10052: MVCC: fixed local node recovery. This closes #5245. This …

…closes #5345.

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

@@ -143,5 +143,4 @@ public byte getValType() {
     public byte[] getValBytes() {
         return valBytes;
     }
-
 }

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

@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.pagemem.wal.record;
+
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheOperation;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccVersion;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.processors.cacheobject.IgniteCacheObjectProcessor;
+
+/**
+ * Represents Data Entry ({@link #key}, {@link #val value}) pair update {@link #op operation}. <br>
+ * This Data entry was not converted to key, value pair during record deserialization.
+ */
+public class LazyMvccDataEntry extends MvccDataEntry {
+    /** */
+    private GridCacheSharedContext cctx;
+
+    /** Data Entry key type code. See {@link CacheObject} for built-in value type codes */
+    private byte keyType;
+
+    /** Key value bytes. */
+    private byte[] keyBytes;
+
+    /** Data Entry Value type code. See {@link CacheObject} for built-in value type codes */
+    private byte valType;
+
+    /** Value value bytes. */
+    private byte[] valBytes;
+
+    /**
+     * @param cctx Shared context.
+     * @param cacheId Cache ID.
+     * @param keyType Object type code for Key.
+     * @param keyBytes Data Entry Key value bytes.
+     * @param valType Object type code for Value.
+     * @param valBytes Data Entry Value value bytes.
+     * @param op Operation.
+     * @param nearXidVer Near transaction version.
+     * @param writeVer Write version.
+     * @param expireTime Expire time.
+     * @param partId Partition ID.
+     * @param partCnt Partition counter.
+     * @param mvccVer Mvcc version.
+     */
+    public LazyMvccDataEntry(
+        GridCacheSharedContext cctx,
+        int cacheId,
+        byte keyType,
+        byte[] keyBytes,
+        byte valType,
+        byte[] valBytes,
+        GridCacheOperation op,
+        GridCacheVersion nearXidVer,
+        GridCacheVersion writeVer,
+        long expireTime,
+        int partId,
+        long partCnt,
+        MvccVersion mvccVer
+    ) {
+        super(cacheId, null, null, op, nearXidVer, writeVer, expireTime, partId, partCnt, mvccVer);
+
+        this.cctx = cctx;
+        this.keyType = keyType;
+        this.keyBytes = keyBytes;
+        this.valType = valType;
+        this.valBytes = valBytes;
+    }
+
+    /** {@inheritDoc} */
+    @Override public KeyCacheObject key() {
+        try {
+            if (key == null) {
+                GridCacheContext cacheCtx = cctx.cacheContext(cacheId);
+
+                if (cacheCtx == null)
+                    throw new IgniteException("Failed to find cache context for the given cache ID: " + cacheId);
+
+                IgniteCacheObjectProcessor co = cctx.kernalContext().cacheObjects();
+
+                key = co.toKeyCacheObject(cacheCtx.cacheObjectContext(), keyType, keyBytes);
+
+                if (key.partition() == -1)
+                    key.partition(partId);
+            }
+
+            return key;
+        }
+        catch (IgniteCheckedException e) {
+            throw new IgniteException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public CacheObject value() {
+        if (val == null && valBytes != null) {
+            GridCacheContext cacheCtx = cctx.cacheContext(cacheId);
+
+            if (cacheCtx == null)
+                throw new IgniteException("Failed to find cache context for the given cache ID: " + cacheId);
+
+            IgniteCacheObjectProcessor co = cctx.kernalContext().cacheObjects();
+
+            val = co.toCacheObject(cacheCtx.cacheObjectContext(), valType, valBytes);
+        }
+
+        return val;
+    }
+
+    /** @return Data Entry Key type code. See {@link CacheObject} for built-in value type codes */
+    public byte getKeyType() {
+        return keyType;
+    }
+
+    /** @return Key value bytes. */
+    public byte[] getKeyBytes() {
+        return keyBytes;
+    }
+
+    /** @return Data Entry Value type code. See {@link CacheObject} for built-in value type codes */
+    public byte getValType() {
+        return valType;
+    }
+
+    /** @return Value value bytes. */
+    public byte[] getValBytes() {
+        return valBytes;
+    }
+}

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

@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.pagemem.wal.record;
+
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.GridCacheOperation;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccVersion;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Represents Data Entry ({@link #key}, {@link #val value}) pair for mvcc update {@link #op operation} in WAL log.
+ */
+public class MvccDataEntry extends DataEntry {
+    /** Entry version. */
+    private MvccVersion mvccVer;
+
+    /**
+     * @param cacheId Cache ID.
+     * @param key Key.
+     * @param val Value or null for delete operation.
+     * @param op Operation.
+     * @param nearXidVer Near transaction version.
+     * @param writeVer Write version.
+     * @param expireTime Expire time.
+     * @param partId Partition ID.
+     * @param partCnt Partition counter.
+     * @param mvccVer Mvcc version.
+     */
+    public MvccDataEntry(
+        int cacheId,
+        KeyCacheObject key,
+        @Nullable CacheObject val,
+        GridCacheOperation op,
+        GridCacheVersion nearXidVer,
+        GridCacheVersion writeVer,
+        long expireTime,
+        int partId,
+        long partCnt,
+        MvccVersion mvccVer
+    ) {
+        super(cacheId, key, val, op, nearXidVer, writeVer, expireTime, partId, partCnt);
+
+        this.mvccVer = mvccVer;
+    }
+
+    /**
+     * @return Mvcc version.
+     */
+    public MvccVersion mvccVer() {
+        return mvccVer;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(MvccDataEntry.class, this);
+    }
+}

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

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.pagemem.wal.record;
+
+import java.util.Collections;
+import java.util.List;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+
+/**
+ * Logical data record with cache operation description.
+ * This record contains information about operation we want to do.
+ * Contains operation type (put, remove) and (Key, Value, Version) for each {@link MvccDataEntry}
+ */
+public class MvccDataRecord extends DataRecord {
+    /** {@inheritDoc} */
+    @Override public RecordType type() {
+        return RecordType.MVCC_DATA_RECORD;
+    }
+
+    /**
+     * @param writeEntry Write entry.
+     */
+    public MvccDataRecord(MvccDataEntry writeEntry) {
+        this(writeEntry, U.currentTimeMillis());
+    }
+
+    /**
+     * @param writeEntries Write entries.
+     */
+    public MvccDataRecord(List<DataEntry> writeEntries) {
+        this(writeEntries, U.currentTimeMillis());
+    }
+
+    /**
+     * @param writeEntry Write entry.
+     */
+    public MvccDataRecord(MvccDataEntry writeEntry, long timestamp) {
+        this(Collections.singletonList(writeEntry), timestamp);
+    }
+
+    /**
+     * @param writeEntries Write entries.
+     * @param timestamp TimeStamp.
+     */
+    public MvccDataRecord(List<DataEntry> writeEntries, long timestamp) {
+        super(writeEntries, timestamp);
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(MvccDataRecord.class, this, "super", super.toString());
+    }
+}

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

@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.pagemem.wal.record;
+
+import java.util.Collection;
+import java.util.Map;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccVersion;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.transactions.TransactionState;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Logical data record indented for MVCC transaction related actions.<br>
+ * This record is marker of prepare, commit, and rollback transactions.
+ */
+public class MvccTxRecord extends TxRecord {
+    /** Transaction mvcc snapshot version. */
+    private final MvccVersion mvccVer;
+
+    /**
+     * @param state Transaction state.
+     * @param nearXidVer Transaction id.
+     * @param writeVer Transaction entries write topology version.
+     * @param participatingNodes Primary -> Backup nodes compact IDs participating in transaction.
+     * @param mvccVer Transaction snapshot version.
+     */
+    public MvccTxRecord(
+        TransactionState state,
+        GridCacheVersion nearXidVer,
+        GridCacheVersion writeVer,
+        @Nullable Map<Short, Collection<Short>> participatingNodes,
+        MvccVersion mvccVer
+    ) {
+        super(state, nearXidVer, writeVer, participatingNodes);
+
+        this.mvccVer = mvccVer;
+    }
+
+    /**
+     * @param state Transaction state.
+     * @param nearXidVer Transaction id.
+     * @param writeVer Transaction entries write topology version.
+     * @param mvccVer Transaction snapshot version.
+     * @param participatingNodes Primary -> Backup nodes participating in transaction.
+     * @param ts TimeStamp.
+     */
+    public MvccTxRecord(
+        TransactionState state,
+        GridCacheVersion nearXidVer,
+        GridCacheVersion writeVer,
+        @Nullable Map<Short, Collection<Short>> participatingNodes,
+        MvccVersion mvccVer,
+        long ts
+    ) {
+        super(state, nearXidVer, writeVer, participatingNodes, ts);
+
+        this.mvccVer = mvccVer;
+    }
+
+    /** {@inheritDoc} */
+    @Override public RecordType type() {
+        return RecordType.MVCC_TX_RECORD;
+    }
+
+    /**
+     * @return Mvcc version.
+     */
+    public MvccVersion mvccVersion() {
+        return mvccVer;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(MvccTxRecord.class, this, "super", super.toString());
+    }
+}