IGNITE-11704 Tombstones

modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java

@@ -53,6 +53,7 @@
 import org.apache.ignite.internal.processors.cache.GridCacheReturn;
 import org.apache.ignite.internal.processors.cache.GridChangeGlobalStateMessageResponse;
 import org.apache.ignite.internal.processors.cache.KeyCacheObjectImpl;
+import org.apache.ignite.internal.processors.cache.TombstoneCacheObject;
 import org.apache.ignite.internal.processors.cache.WalStateAckMessage;
 import org.apache.ignite.internal.processors.cache.binary.MetadataRequestMessage;
 import org.apache.ignite.internal.processors.cache.binary.MetadataResponseMessage;
@@ -1166,6 +1167,11 @@ public GridIoMessageFactory(MessageFactory[] ext) {
 
                 break;
 
+            case 176:
+                msg = TombstoneCacheObject.INSTANCE;
+
+                break;
+
             // [-3..119] [124..129] [-23..-28] [-36..-55] - this
             // [120..123] - DR
             // [-4..-22, -30..-35] - SQL

modules/core/src/main/java/org/apache/ignite/internal/pagemem/PageUtils.java

@@ -56,8 +56,8 @@ public static int getUnsignedByte(long addr, int off) {
      */
     public static byte[] getBytes(long addr, int off, int len) {
         assert addr > 0 : addr;
-        assert off >= 0;
-        assert len >= 0;
+        assert off >= 0 : off;
+        assert len >= 0 : len;
 
         byte[] bytes = new byte[len];
 

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

@@ -220,8 +220,11 @@ public enum RecordType {
         /** Rollback tx record. */
         ROLLBACK_TX_RECORD (57, LOGICAL),
 
-        /** */
-        PARTITION_META_PAGE_UPDATE_COUNTERS_V2 (58, PHYSICAL);
+        /** Partition meta page containing update counter gaps. */
+        PARTITION_META_PAGE_UPDATE_COUNTERS_V2 (58, PHYSICAL),
+
+        /** Partition meta page containing tombstone presence flag. */
+        PARTITION_META_PAGE_UPDATE_COUNTERS_V3 (59, PHYSICAL);
 
         /** Index for serialization. Should be consistent throughout all versions. */
         private final int idx;

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

@@ -46,7 +46,7 @@ public class MetaPageUpdatePartitionDataRecord extends PageDeltaRecord {
     private int allocatedIdxCandidate;
 
     /** */
-    private long cntrsPageId;
+    private long cacheSizesPageId;
 
     /**
      * @param grpId Cache group ID.
@@ -59,17 +59,18 @@ public MetaPageUpdatePartitionDataRecord(
         long updateCntr,
         long globalRmvId,
         int partSize,
-        long cntrsPageId,
+        long cacheSizesPageId,
         byte state,
-        int allocatedIdxCandidate) {
+        int allocatedIdxCandidate
+    ) {
         super(grpId, pageId);
 
         this.updateCntr = updateCntr;
         this.globalRmvId = globalRmvId;
         this.partSize = partSize;
         this.state = state;
         this.allocatedIdxCandidate = allocatedIdxCandidate;
-        this.cntrsPageId = cntrsPageId;
+        this.cacheSizesPageId = cacheSizesPageId;
     }
 
     /**
@@ -81,7 +82,7 @@ public MetaPageUpdatePartitionDataRecord(DataInput in) throws IOException{
         this.updateCntr = in.readLong();
         this.globalRmvId = in.readLong();
         this.partSize = in.readInt();
-        this.cntrsPageId = in.readLong();
+        this.cacheSizesPageId = in.readLong();
         this.state = in.readByte();
         this.allocatedIdxCandidate = in.readInt();
     }
@@ -110,8 +111,8 @@ public int partitionSize() {
     /**
      * @return Partition size.
      */
-    public long countersPageId() {
-        return cntrsPageId;
+    public long cacheSizesPageId() {
+        return cacheSizesPageId;
     }
 
     /**
@@ -128,7 +129,7 @@ public byte state() {
         io.setUpdateCounter(pageAddr, updateCntr);
         io.setGlobalRemoveId(pageAddr, globalRmvId);
         io.setSize(pageAddr, partSize);
-        io.setCountersPageId(pageAddr, cntrsPageId);
+        io.setSizesPageId(pageAddr, cacheSizesPageId);
         io.setPartitionState(pageAddr, state);
         io.setCandidatePageCount(pageAddr, allocatedIdxCandidate);
     }
@@ -150,7 +151,7 @@ public void toBytes(ByteBuffer buf) {
         buf.putLong(updateCounter());
         buf.putLong(globalRemoveId());
         buf.putInt(partitionSize());
-        buf.putLong(countersPageId());
+        buf.putLong(cacheSizesPageId());
         buf.put(state());
         buf.putInt(allocatedIndexCandidate());
     }

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

@@ -28,11 +28,12 @@
 import org.apache.ignite.internal.util.typedef.internal.S;
 
 /**
- *
+ * Partition meta page delta record.
+ * Contains reference to update counters gaps.
  */
 public class MetaPageUpdatePartitionDataRecordV2 extends MetaPageUpdatePartitionDataRecord {
     /** */
-    private long link;
+    private long gapsLink;
 
     /**
      * @param grpId Group id.
@@ -43,7 +44,7 @@ public class MetaPageUpdatePartitionDataRecordV2 extends MetaPageUpdatePartition
      * @param cntrsPageId Cntrs page id.
      * @param state State.
      * @param allocatedIdxCandidate Allocated index candidate.
-     * @param link Link.
+     * @param gapsLink Link.
      */
     public MetaPageUpdatePartitionDataRecordV2(
         int grpId,
@@ -54,9 +55,10 @@ public MetaPageUpdatePartitionDataRecordV2(
         long cntrsPageId,
         byte state,
         int allocatedIdxCandidate,
-        long link) {
+        long gapsLink
+    ) {
         super(grpId, pageId, updateCntr, globalRmvId, partSize, cntrsPageId, state, allocatedIdxCandidate);
-        this.link = link;
+        this.gapsLink = gapsLink;
     }
 
     /**
@@ -65,7 +67,7 @@ public MetaPageUpdatePartitionDataRecordV2(
     public MetaPageUpdatePartitionDataRecordV2(DataInput in) throws IOException {
         super(in);
 
-        this.link = in.readLong();
+        this.gapsLink = in.readLong();
     }
 
     /** {@inheritDoc} */
@@ -74,21 +76,21 @@ public MetaPageUpdatePartitionDataRecordV2(DataInput in) throws IOException {
 
         PagePartitionMetaIOV2 io = (PagePartitionMetaIOV2)PagePartitionMetaIO.VERSIONS.forPage(pageAddr);
 
-        io.setGapsLink(pageAddr, link);
+        io.setGapsLink(pageAddr, gapsLink);
     }
 
     /**
      *
      */
-    public long link() {
-        return link;
+    public long gapsLink() {
+        return gapsLink;
     }
 
     /** {@inheritDoc} */
     @Override public void toBytes(ByteBuffer buf) {
         super.toBytes(buf);
 
-        buf.putLong(link());
+        buf.putLong(gapsLink());
     }
 
     /** {@inheritDoc} */

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

@@ -0,0 +1,108 @@
+/*
+ * 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.delta;
+
+import java.io.DataInput;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.pagemem.PageIdUtils;
+import org.apache.ignite.internal.pagemem.PageMemory;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PagePartitionMetaIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PagePartitionMetaIOV2;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+/**
+ * Partition meta page delta record.
+ * Contains information about tombstones count.
+ */
+public class MetaPageUpdatePartitionDataRecordV3 extends MetaPageUpdatePartitionDataRecordV2 {
+    /** Tombstones count. */
+    private long tombstonesCnt;
+
+    /**
+     * @param grpId Group id.
+     * @param pageId Page id.
+     * @param updateCntr Update counter.
+     * @param globalRmvId Global remove id.
+     * @param partSize Partition size.
+     * @param cacheSizesPageId Cache sizes page id.
+     * @param state State.
+     * @param allocatedIdxCandidate Allocated index candidate.
+     * @param gapsLink Gaps link.
+     * @param tombstonesCnt Tombstones count.
+     */
+    public MetaPageUpdatePartitionDataRecordV3(
+        int grpId,
+        long pageId,
+        long updateCntr,
+        long globalRmvId,
+        int partSize,
+        long cacheSizesPageId,
+        byte state,
+        int allocatedIdxCandidate,
+        long gapsLink,
+        long tombstonesCnt
+    ) {
+        super(grpId, pageId, updateCntr, globalRmvId, partSize, cacheSizesPageId, state, allocatedIdxCandidate, gapsLink);
+        this.tombstonesCnt = tombstonesCnt;
+    }
+
+    /**
+     * @param in In.
+     */
+    public MetaPageUpdatePartitionDataRecordV3(DataInput in) throws IOException {
+        super(in);
+
+        this.tombstonesCnt = in.readLong();
+    }
+
+    /**
+     * @return Tombstones count.
+     */
+    public long tombstonesCount() {
+        return tombstonesCnt;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void applyDelta(PageMemory pageMem, long pageAddr) throws IgniteCheckedException {
+        super.applyDelta(pageMem, pageAddr);
+
+        PagePartitionMetaIOV2 io = (PagePartitionMetaIOV2) PagePartitionMetaIO.VERSIONS.forPage(pageAddr);
+
+        io.setTombstonesCount(pageAddr, tombstonesCnt);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void toBytes(ByteBuffer buf) {
+        super.toBytes(buf);
+
+        buf.putLong(tombstonesCnt);
+    }
+
+    /** {@inheritDoc} */
+    @Override public RecordType type() {
+        return RecordType.PARTITION_META_PAGE_UPDATE_COUNTERS_V3;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(MetaPageUpdatePartitionDataRecordV2.class, this, "partId", PageIdUtils.partId(pageId()),
+            "super", super.toString());
+    }
+}

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

@@ -46,6 +46,8 @@
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtAffinityAssignmentRequest;
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtAffinityAssignmentResponse;
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPreloader;
+import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
+import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
 import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology;
 import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopologyImpl;
 import org.apache.ignite.internal.processors.cache.persistence.DataRegion;
@@ -261,6 +263,8 @@ public class CacheGroupContext {
             statHolderIdx = new IoStatisticsHolderIndex(HASH_INDEX, cacheOrGroupName(), HASH_PK_IDX_NAME, mmgr);
             statHolderData = new IoStatisticsHolderCache(cacheOrGroupName(), grpId, mmgr);
         }
+
+        hasAtomicCaches = ccfg.getAtomicityMode() == ATOMIC;
     }
 
     /**
@@ -1298,6 +1302,21 @@ public boolean hasAtomicCaches() {
         return hasAtomicCaches;
     }
 
+    /**
+     * @return {@code True} if need create temporary tombstones entries for removed data.
+     */
+    public boolean supportsTombstone() {
+        return !mvccEnabled && !isLocal();
+    }
+
+    /**
+     * @param part Partition.
+     * @return {@code True} if need create tombstone for remove in given partition.
+     */
+    public boolean shouldCreateTombstone(@Nullable GridDhtLocalPartition part) {
+        return part != null && supportsTombstone() && part.state() == GridDhtPartitionState.MOVING;
+    }
+
     /**
      * @return Metrics.
      */

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

@@ -70,6 +70,7 @@ public class CacheGroupMetricsImpl {
     private final LongMetric sparseStorageSize;
 
     /** Interface describing a predicate of two integers. */
+    @FunctionalInterface
     private interface IntBiPredicate {
         /**
          * Predicate body.
@@ -169,6 +170,10 @@ public void onTopologyInitialized() {
         mreg.register("TotalAllocatedSize",
             this::getTotalAllocatedSize,
             "Total size of memory allocated for group, in bytes.");
+
+        mreg.register("Tombstones",
+            this::getTombstones,
+            "Number of tombstone entries.");
     }
 
     /** */
@@ -253,20 +258,12 @@ private int numberOfPartitionCopies(IntBiPredicate pred) {
 
     /** */
     public int getMinimumNumberOfPartitionCopies() {
-        return numberOfPartitionCopies(new IntBiPredicate() {
-            @Override public boolean apply(int targetVal, int nextVal) {
-                return nextVal < targetVal;
-            }
-        });
+        return numberOfPartitionCopies((targetVal, nextVal) -> nextVal < targetVal);
     }
 
     /** */
     public int getMaximumNumberOfPartitionCopies() {
-        return numberOfPartitionCopies(new IntBiPredicate() {
-            @Override public boolean apply(int targetVal, int nextVal) {
-                return nextVal > targetVal;
-            }
-        });
+        return numberOfPartitionCopies((targetVal, nextVal) -> nextVal > targetVal);
     }
 
     /**
@@ -462,6 +459,12 @@ public long getSparseStorageSize() {
         return sparseStorageSize == null ? 0 : sparseStorageSize.value();
     }
 
+    /** */
+    public long getTombstones() {
+        return ctx.topology().localPartitions().stream()
+            .map(part -> part.dataStore().tombstonesCount()).reduce(Long::sum).orElse(0L);
+    }
+
     /** Removes all metric for cache group. */
     public void remove() {
         ctx.shared().kernalContext().metric().remove(metricGroupName());

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

@@ -38,6 +38,9 @@ public interface CacheObject extends Message {
     /** */
     public static final byte TYPE_BINARY_ENUM = 101;
 
+    /** */
+    public static final byte TOMBSTONE = -1;
+
     /**
      * @param ctx Context.
      * @param cpy If {@code true} need to copy value.