IGNITE-29028 Implement multi-page in-place row update

modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/jmh/cache/JmhCacheInPlaceUpdateBenchmark.java

@@ -0,0 +1,160 @@
+/*
+ * 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.benchmarks.jmh.cache;
+
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import javax.cache.expiry.CreatedExpiryPolicy;
+import javax.cache.expiry.Duration;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.cluster.ClusterState;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.DataRegionConfiguration;
+import org.apache.ignite.configuration.DataStorageConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.benchmarks.jmh.runner.JmhIdeBenchmarkRunner;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+
+/**
+ * Compare put with in-place update and without in-place update.
+ */
+@State(Scope.Benchmark)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Warmup(iterations = 3, time = 3)
+@Measurement(iterations = 3, time = 10)
+public class JmhCacheInPlaceUpdateBenchmark {
+    /** Items count. */
+    private static final int CNT = 100;
+
+    /** Entry size. */
+    private static final int ENTRY_SIZE = 100 * 1024;
+
+    /** Ignite. */
+    private Ignite ignite;
+
+    /** Cache with in-place updates. */
+    private IgniteCache<Integer, byte[]> cache0;
+
+    /** Cache without in-place updates. */
+    private IgniteCache<Integer, byte[]> cache1;
+
+    /** Entry payloads for cache 0. */
+    private final byte[][] payloads0 = new byte[CNT][ENTRY_SIZE];
+
+    /** Entry payloads for cache 1. */
+    private final byte[][] payloads1 = new byte[CNT][ENTRY_SIZE];
+
+    /** Persistence enabled. */
+    @Param({"FALSE", "TRUE"})
+    private String persistence;
+
+    /** */
+    @Benchmark
+    public void putWithInPlaceUpdate() {
+        int key = ThreadLocalRandom.current().nextInt(CNT);
+
+        changeAndPutPayload(cache0, key, payloads0[key]);
+    }
+
+    /** */
+    @Benchmark
+    public void putWithoutInPlaceUpdate() {
+        int key = ThreadLocalRandom.current().nextInt(CNT);
+
+        changeAndPutPayload(cache1, key, payloads1[key]);
+    }
+
+    /** */
+    private void changeAndPutPayload(IgniteCache<Integer, byte[]> cache, int key, byte[] payload) {
+        // Change 1 byte.
+        payload[ThreadLocalRandom.current().nextInt(payload.length)] = (byte)ThreadLocalRandom.current().nextInt(256);
+
+        cache.put(key, payload);
+    }
+
+    /**
+     * Initiate Ignite and caches.
+     */
+    @Setup(Level.Trial)
+    public void setup() {
+        ignite = Ignition.start(new IgniteConfiguration().setIgniteInstanceName("test")
+            .setDataStorageConfiguration(new DataStorageConfiguration().setDefaultDataRegionConfiguration(
+                new DataRegionConfiguration().setPersistenceEnabled(Boolean.parseBoolean(persistence))
+            ))
+        );
+
+        ignite.cluster().state(ClusterState.ACTIVE);
+
+        cache0 = ignite.getOrCreateCache(new CacheConfiguration<>("CACHE0"));
+
+        // Enable expiration for second cache, but set eager ttl to false, this will disable in-place updates,
+        // but without performance overhead to maintain expiration.
+        cache1 = ignite.getOrCreateCache(
+            new CacheConfiguration<Integer, byte[]>("CACHE1")
+                .setEagerTtl(false)
+                .setExpiryPolicyFactory(CreatedExpiryPolicy.factoryOf(Duration.ONE_DAY))
+        );
+    }
+
+    /**
+     * Clear caches.
+     */
+    @Setup(Level.Iteration)
+    public void setupIteration() {
+        for (int i = 0; i < CNT; i++) {
+            ThreadLocalRandom.current().nextBytes(payloads0[i]);
+            ThreadLocalRandom.current().nextBytes(payloads1[i]);
+            cache0.put(i, payloads0[i]);
+            cache1.put(i, payloads1[i]);
+        }
+    }
+
+    /**
+     * Stop Ignite instance.
+     */
+    @TearDown
+    public void tearDown() {
+        ignite.close();
+    }
+
+    /**
+     * Run benchmarks.
+     *
+     * @param args Args.
+     * @throws Exception Exception.
+     */
+    public static void main(String[] args) throws Exception {
+        JmhIdeBenchmarkRunner.create()
+            .benchmarks(JmhCacheInPlaceUpdateBenchmark.class.getSimpleName())
+            .run();
+    }
+}

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

@@ -71,7 +71,7 @@ public byte[] payload() {
 
         AbstractDataPageIO io = PageIO.getPageIO(pageAddr);
 
-        io.updateRow(pageAddr, itemId, pageMem.realPageSize(groupId()), payload, null, 0);
+        io.updateRow(pageAddr, itemId, pageMem.realPageSize(groupId()), payload);
     }
 
     /** {@inheritDoc} */

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

@@ -4354,9 +4354,6 @@ private static class UpdateClosure implements IgniteCacheOffheapManager.OffheapI
         /** */
         private CacheDataRow oldRow;
 
-        /** */
-        private boolean oldRowExpiredFlag;
-
         /** */
         private IgniteTree.OperationType treeOp = IgniteTree.OperationType.PUT;
 
@@ -4379,14 +4376,14 @@ private static class UpdateClosure implements IgniteCacheOffheapManager.OffheapI
 
         /** {@inheritDoc} */
         @Override public void call(@Nullable CacheDataRow oldRow) throws IgniteCheckedException {
+            this.oldRow = oldRow;
+
             if (oldRow != null) {
                 oldRow.key(entry.key);
 
                 oldRow = checkRowExpired(oldRow);
             }
 
-            this.oldRow = oldRow;
-
             if (predicate != null && !predicate.apply(oldRow)) {
                 treeOp = IgniteTree.OperationType.NOOP;
 
@@ -4395,7 +4392,7 @@ private static class UpdateClosure implements IgniteCacheOffheapManager.OffheapI
 
             if (val != null) {
                 if (newRow == null) {
-                    newRow = entry.cctx.offheap().dataStore(entry.localPartition()).createRow(
+                    newRow = entry.cctx.offheap().dataStore(entry.localPartition()).updateRow(
                         entry.cctx,
                         entry.key,
                         val,
@@ -4426,11 +4423,6 @@ private static class UpdateClosure implements IgniteCacheOffheapManager.OffheapI
             return oldRow;
         }
 
-        /** {@inheritDoc} */
-        @Override public boolean oldRowExpiredFlag() {
-            return oldRowExpiredFlag;
-        }
-
         /**
          * Checks row for expiration and fire expire events if needed.
          *
@@ -4476,8 +4468,6 @@ private CacheDataRow checkRowExpired(CacheDataRow row) throws IgniteCheckedExcep
 
             entry.updatePlatformCache(null, null);
 
-            oldRowExpiredFlag = true;
-
             return null;
         }
     }
@@ -4637,11 +4627,6 @@ private static class AtomicCacheUpdateClosure implements IgniteCacheOffheapManag
             return oldRow;
         }
 
-        /** {@inheritDoc} */
-        @Override public boolean oldRowExpiredFlag() {
-            return oldRowExpiredFlag;
-        }
-
         /** {@inheritDoc} */
         @Override public CacheDataRow newRow() {
             return newRow;
@@ -4917,7 +4902,7 @@ else if (updateExpireTime && expiryPlc != null && entry.val != null) {
             }
 
             if (needUpdate) {
-                newRow = entry.localPartition().dataStore().createRow(
+                newRow = entry.localPartition().dataStore().updateRow(
                     entry.cctx,
                     entry.key,
                     storeLoadedVal,
@@ -5081,7 +5066,7 @@ else if (interceptorVal != updated0) {
             entry.logUpdate(op, updated, newVer, newExpireTime, updateCntr0, primary);
 
             if (!entry.isNear()) {
-                newRow = entry.localPartition().dataStore().createRow(
+                newRow = entry.localPartition().dataStore().updateRow(
                     entry.cctx,
                     entry.key,
                     updated,
@@ -5090,7 +5075,7 @@ else if (interceptorVal != updated0) {
                     oldRow);
 
                 treeOp = oldRow != null && oldRow.link() == newRow.link() ?
-                    IgniteTree.OperationType.NOOP : IgniteTree.OperationType.PUT;
+                    IgniteTree.OperationType.IN_PLACE : IgniteTree.OperationType.PUT;
             }
             else
                 treeOp = IgniteTree.OperationType.PUT;

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

@@ -26,13 +26,13 @@
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtDemandedPartitionsMap;
 import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
 import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
-import org.apache.ignite.internal.processors.cache.persistence.DataRowCacheAware;
 import org.apache.ignite.internal.processors.cache.persistence.RootPage;
 import org.apache.ignite.internal.processors.cache.persistence.RowStore;
 import org.apache.ignite.internal.processors.cache.persistence.freelist.SimpleDataRow;
 import org.apache.ignite.internal.processors.cache.persistence.partstorage.PartitionMetaStorage;
 import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
 import org.apache.ignite.internal.processors.cache.tree.CacheDataTree;
+import org.apache.ignite.internal.processors.cache.tree.DataRow;
 import org.apache.ignite.internal.processors.cache.tree.PendingEntriesTree;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
 import org.apache.ignite.internal.processors.query.GridQueryRowCacheCleaner;
@@ -194,18 +194,16 @@ public void invoke(GridCacheContext cctx, KeyCacheObject key, GridDhtLocalPartit
      * @param val Value.
      * @param ver Version.
      * @param expireTime Expire time.
-     * @param oldRow Old row if available.
      * @param part Partition.
      * @throws IgniteCheckedException If failed.
      */
     public void update(
-        GridCacheContext cctx,
+        GridCacheContext<?, ?> cctx,
         KeyCacheObject key,
         CacheObject val,
         GridCacheVersion ver,
         long expireTime,
-        GridDhtLocalPartition part,
-        @Nullable CacheDataRow oldRow
+        GridDhtLocalPartition part
     ) throws IgniteCheckedException;
 
     /**
@@ -418,12 +416,6 @@ interface OffheapInvokeClosure extends IgniteTree.InvokeClosure<CacheDataRow> {
          * @return Old row.
          */
         @Nullable public CacheDataRow oldRow();
-
-        /**
-         * Flag that indicates if oldRow was expired during invoke.
-         * @return {@code true} if old row was expired, {@code false} otherwise.
-         */
-        public boolean oldRowExpiredFlag();
     }
 
     /**
@@ -542,13 +534,14 @@ interface CacheDataStore {
          * @return New row.
          * @throws IgniteCheckedException If failed.
          */
-        CacheDataRow createRow(
-            GridCacheContext cctx,
+        CacheDataRow updateRow(
+            GridCacheContext<?, ?> cctx,
             KeyCacheObject key,
             CacheObject val,
             GridCacheVersion ver,
             long expireTime,
-            @Nullable CacheDataRow oldRow) throws IgniteCheckedException;
+            @Nullable CacheDataRow oldRow
+        ) throws IgniteCheckedException;
 
         /**
          * Insert rows into page memory.
@@ -557,25 +550,26 @@ CacheDataRow createRow(
          * @param initPred Applied to all rows. Each row that not matches the predicate is removed.
          * @throws IgniteCheckedException If failed.
          */
-        public void insertRows(Collection<DataRowCacheAware> rows,
-            IgnitePredicateX<CacheDataRow> initPred) throws IgniteCheckedException;
+        public void insertRows(
+            Collection<DataRow> rows,
+            IgnitePredicateX<CacheDataRow> initPred
+        ) throws IgniteCheckedException;
 
         /**
          * @param cctx Cache context.
          * @param key Key.
          * @param val Value.
          * @param ver Version.
          * @param expireTime Expire time.
-         * @param oldRow Old row if available.
          * @throws IgniteCheckedException If failed.
          */
         void update(
-            GridCacheContext cctx,
+            GridCacheContext<?, ?> cctx,
             KeyCacheObject key,
             CacheObject val,
             GridCacheVersion ver,
-            long expireTime,
-            @Nullable CacheDataRow oldRow) throws IgniteCheckedException;
+            long expireTime
+        ) throws IgniteCheckedException;
 
         /**
          * @param cctx Cache context.