diff --git a/modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/jmh/cache/JmhCacheInPlaceUpdateBenchmark.java b/modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/jmh/cache/JmhCacheInPlaceUpdateBenchmark.java new file mode 100644 index 0000000000000..f927f4d9fa916 --- /dev/null +++ b/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 cache0; + + /** Cache without in-place updates. */ + private IgniteCache 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 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("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(); + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageUpdateRecord.java b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageUpdateRecord.java index 6f5d8fd86409b..3a38455e11e14 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageUpdateRecord.java +++ b/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} */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java index 021fdd0408d48..28bc59710c25f 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java +++ b/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; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java index 3247e718ad6cf..168d8b3263384 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java +++ b/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 { * @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,8 +550,10 @@ 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 rows, - IgnitePredicateX initPred) throws IgniteCheckedException; + public void insertRows( + Collection rows, + IgnitePredicateX initPred + ) throws IgniteCheckedException; /** * @param cctx Cache context. @@ -566,16 +561,15 @@ public void insertRows(Collection rows, * @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. diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java index c97aefd68213f..172312c52ca6d 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java @@ -54,7 +54,6 @@ import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow; import org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter; import org.apache.ignite.internal.processors.cache.persistence.CacheSearchRow; -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; @@ -96,6 +95,7 @@ import static org.apache.ignite.internal.pagemem.PageIdAllocator.INDEX_PARTITION; import static org.apache.ignite.internal.processors.cache.GridCacheUtils.TTL_ETERNAL; import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING; +import static org.apache.ignite.internal.processors.rollingupgrade.feature.SupportedFeatureRegistry.MULTI_PAGE_IN_PLACE_ROW_UPDATE_FEATURE; /** * @@ -404,17 +404,16 @@ private Iterator cacheData(boolean primary, boolean backup, Affi /** {@inheritDoc} */ @Override public void update( - GridCacheContext cctx, + GridCacheContext cctx, KeyCacheObject key, CacheObject val, GridCacheVersion ver, long expireTime, - GridDhtLocalPartition part, - @Nullable CacheDataRow oldRow + GridDhtLocalPartition part ) throws IgniteCheckedException { assert expireTime >= 0; - dataStore(part).update(cctx, key, val, ver, expireTime, oldRow); + dataStore(part).update(cctx, key, val, ver, expireTime); } /** {@inheritDoc} */ @@ -988,7 +987,7 @@ private long allocateForTree() throws IgniteCheckedException { IgnitePredicateX initPred) throws IgniteCheckedException { CacheDataStore dataStore = dataStore(part); - List batch = new ArrayList<>(PRELOAD_SIZE_UNDER_CHECKPOINT_LOCK); + List batch = new ArrayList<>(PRELOAD_SIZE_UNDER_CHECKPOINT_LOCK); while (infos.hasNext()) { GridCacheEntryInfo info = infos.next(); @@ -996,7 +995,7 @@ private long allocateForTree() throws IgniteCheckedException { assert info.ttl() == TTL_ETERNAL : info.ttl(); try { - batch.add(new DataRowCacheAware(info.key(), + batch.add(new DataRow(info.key(), info.value(), info.version(), part.id(), @@ -1272,7 +1271,7 @@ public CacheDataStoreImpl( pCntr = grp.shared().logger(PartitionUpdateCounterDebugWrapper.class).isDebugEnabled() ? new PartitionUpdateCounterDebugWrapper(partId, delegate) : new PartitionUpdateCounterErrorWrapper(partId, delegate); - updateValSizeThreshold = grp.shared().database().pageSize() / 2; + updateValSizeThreshold = grp.shared().database().pageSize() * 3 / 4; if (cleaner == null) rowStore.setRowCacheCleaner(() -> rowCacheCleaner); @@ -1437,29 +1436,69 @@ void decrementSize(int cacheId) { } /** + * Checks if in-place row update is possible. + * * @param cctx Cache context. * @param oldRow Old row. * @param dataRow New row. * @return {@code True} if it is possible to update old row data. * @throws IgniteCheckedException If failed. */ - private boolean canUpdateOldRow(GridCacheContext cctx, @Nullable CacheDataRow oldRow, DataRow dataRow) - throws IgniteCheckedException { - if (oldRow == null || cctx.queries().enabled()) + private boolean canUpdateOldRow( + GridCacheContext cctx, + @Nullable CacheDataRow oldRow, + DataRow dataRow + ) throws IgniteCheckedException { + if (oldRow == null) return false; + // In-place update is not possible when queries (indexes) are enabled. + // For indexed entries, if we update entry in-place, after updating entry but before updating index, + // the old index may point to the new entry with fields that do not match the index. For multi-page + // entries, while reading data row by link from index, intermediate pages may be changing, causing + // index to read inconsistent entry. Without in-place update, indexed entries are protected as + // follows: when processing index, leaf page lock prevents removal from index tree; then data page + // read locks are acquired and entry is read. During update, entry is first removed from index, + // then removed from row store. The index page lock guarantees that we do not remove entry from + // row store until index finishes working with this entry. + if (cctx.queries().enabled()) + return false; + + // Pending tree stores entries with their original expire time. During expire, entries for deletion + // are read from pending tree (their links), then entries are initialized (key is read by link) under + // pending tree leaf page lock. During updateб when in-place update is disabled, we first insert new entry + // to row store, then remove old entry link from pending tree (this operation acquires pending tree leaf + // page lock), add new entry link to pending tree, and after that remove old entry from row store. + // The pending tree leaf page lock ensures entry consistency. If in-place update is enabled, during expire + // we may read already updated entry with modified TTL. if (oldRow.expireTime() != dataRow.expireTime()) return false; int oldLen = oldRow.size(); - // Use grp.sharedGroup() flag since it is possible cacheId is not yet set here. - if (!grp.storeCacheIdInDataPage() && grp.sharedGroup() && oldRow.cacheId() != CU.UNDEFINED_CACHE_ID) - oldLen -= 4; + // For multi-page entries with pending tree reference (expireTime != 0), even when old expire time + // equals new expire time, we may fall between page updates during access from pending tree + // (on expiration) and read inconsistent entry, causing unmarshalling failure. + if (oldLen > updateValSizeThreshold && oldRow.expireTime() != 0) + return false; - if (oldLen > updateValSizeThreshold) + // Multi-page in-place row update introduces changes to applying WAL delta records, disable it until + // feature is not activated accross all the cluster. + if (oldLen > updateValSizeThreshold + && !grp.shared().kernalContext().rollingUpgrade().features().isActive(MULTI_PAGE_IN_PLACE_ROW_UPDATE_FEATURE)) return false; + // Entry is read from row store by link only in three places: from index tree, from pending tree, + // and from data tree (key lookup). Row update is executed under write lock on data tree leaf page, + // so KV API operations with in-place update are safe: entry read always happens under data tree + // leaf page lock (even for scan cache - iteration goes through data tree). Unfortunately, fixing + // the other two cases (read from index tree and pending tree) is problematic: under current data + // tree leaf page lock we cannot modify index tree or pending tree, as this may lead to deadlock + // (threads working with pending tree and holding its page lock may request data tree page lock). + // We cannot pre-delete entries from other trees before the data tree lock either, because consistent + // reference to old entry can be obtained only under data tree leaf page lock. Deleting entries + // from other trees after the lock (as done currently) is safe only for non in-place update. + int newLen = dataRow.size(); return oldLen == newLen; @@ -1507,17 +1546,13 @@ private void invoke0(GridCacheContext cctx, CacheSearchRow row, OffheapInvokeClo case PUT: { assert c.newRow() != null : c; - CacheDataRow oldRow = c.oldRow(); - - finishUpdate(cctx, c.newRow(), oldRow, c.oldRowExpiredFlag()); + finishUpdate(cctx, c.newRow(), c.oldRow()); break; } case REMOVE: { - CacheDataRow oldRow = c.oldRow(); - - finishRemove(cctx, row.key(), oldRow); + finishRemove(cctx, row.key(), c.oldRow()); break; } @@ -1532,18 +1567,19 @@ private void invoke0(GridCacheContext cctx, CacheSearchRow row, OffheapInvokeClo } /** {@inheritDoc} */ - @Override public CacheDataRow createRow( - GridCacheContext cctx, + @Override public CacheDataRow updateRow( + GridCacheContext cctx, KeyCacheObject key, CacheObject val, GridCacheVersion ver, long expireTime, - @Nullable CacheDataRow oldRow) throws IgniteCheckedException { + @Nullable CacheDataRow oldRow + ) throws IgniteCheckedException { int cacheId = grp.storeCacheIdInDataPage() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID; DataRow dataRow = makeDataRow(key, val, ver, expireTime, cacheId); - if (canUpdateOldRow(cctx, oldRow, dataRow) && rowStore.updateRow(oldRow.link(), dataRow, grp.statisticsHolderData())) + if (canUpdateOldRow(cctx, oldRow, dataRow) && rowStore.updateRow(oldRow, dataRow, grp.statisticsHolderData())) dataRow.link(oldRow.link()); else { CacheObjectContext coCtx = cctx.cacheObjectContext(); @@ -1563,19 +1599,17 @@ private void invoke0(GridCacheContext cctx, CacheSearchRow row, OffheapInvokeClo } /** {@inheritDoc} */ - @Override public void insertRows(Collection rows, - IgnitePredicateX initPred) throws IgniteCheckedException { + @Override public void insertRows( + Collection rows, + IgnitePredicateX initPred + ) throws IgniteCheckedException { if (!busyLock.enterBusy()) throw operationCancelledException(); try { rowStore.addRows(F.view(rows, row -> row.value() != null), grp.statisticsHolderData()); - boolean cacheIdAwareGrp = grp.sharedGroup() || grp.storeCacheIdInDataPage(); - - for (DataRowCacheAware row : rows) { - row.storeCacheId(cacheIdAwareGrp); - + for (DataRow row : rows) { if (!initPred.applyx(row) && row.value() != null) rowStore.removeRow(row.link(), grp.statisticsHolderData()); } @@ -1598,27 +1632,25 @@ private void invoke0(GridCacheContext cctx, CacheSearchRow row, OffheapInvokeClo if (key.partition() == -1) key.partition(partId); - return new DataRow(key, val, ver, partId, expireTime, cacheId); + assert val != null; + + return new DataRow(key, val, ver, partId, expireTime, cacheId, grp.storeCacheIdInDataPage()); } /** {@inheritDoc} */ - @Override public void update(GridCacheContext cctx, + @Override public void update( + GridCacheContext cctx, KeyCacheObject key, CacheObject val, GridCacheVersion ver, - long expireTime, - @Nullable CacheDataRow oldRow + long expireTime ) throws IgniteCheckedException { - assert oldRow == null || oldRow.link() != 0L : oldRow; - if (!busyLock.enterBusy()) throw operationCancelledException(); try { int cacheId = grp.storeCacheIdInDataPage() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID; - assert oldRow == null || oldRow.cacheId() == cacheId : oldRow; - DataRow dataRow = makeDataRow(key, val, ver, expireTime, cacheId); CacheObjectContext coCtx = cctx.cacheObjectContext(); @@ -1627,31 +1659,16 @@ private void invoke0(GridCacheContext cctx, CacheSearchRow row, OffheapInvokeClo key.valueBytes(coCtx); val.valueBytes(coCtx); - CacheDataRow old; - assert cctx.shared().database().checkpointLockIsHeldByThread(); - if (canUpdateOldRow(cctx, oldRow, dataRow) && rowStore.updateRow(oldRow.link(), dataRow, grp.statisticsHolderData())) { - old = oldRow; - - dataRow.link(oldRow.link()); - } - else { - rowStore.addRow(dataRow, grp.statisticsHolderData()); + rowStore.addRow(dataRow, grp.statisticsHolderData()); - assert dataRow.link() != 0 : dataRow; + assert dataRow.link() != 0 : dataRow; - if (grp.sharedGroup() && dataRow.cacheId() == CU.UNDEFINED_CACHE_ID) - dataRow.cacheId(cctx.cacheId()); + if (grp.sharedGroup() && dataRow.cacheId() == CU.UNDEFINED_CACHE_ID) + dataRow.cacheId(cctx.cacheId()); - if (oldRow != null) { - old = oldRow; - - dataTree.putx(dataRow); - } - else - old = dataTree.put(dataRow); - } + CacheDataRow old = dataTree.put(dataRow); finishUpdate(cctx, dataRow, old); } @@ -1666,24 +1683,15 @@ private void invoke0(GridCacheContext cctx, CacheSearchRow row, OffheapInvokeClo * @param oldRow Old row if available. * @throws IgniteCheckedException If failed. */ - private void finishUpdate(GridCacheContext cctx, CacheDataRow newRow, @Nullable CacheDataRow oldRow) - throws IgniteCheckedException { - finishUpdate(cctx, newRow, oldRow, false); - } - - /** - * @param cctx Cache context. - * @param newRow New row. - * @param oldRow Old row if available. - * @param oldRowExpired Old row expiration flag - * @throws IgniteCheckedException If failed. - */ - private void finishUpdate(GridCacheContext cctx, CacheDataRow newRow, @Nullable CacheDataRow oldRow, boolean oldRowExpired) - throws IgniteCheckedException { - if (oldRow == null && !oldRowExpired) + private void finishUpdate( + GridCacheContext cctx, + CacheDataRow newRow, + @Nullable CacheDataRow oldRow + ) throws IgniteCheckedException { + if (oldRow == null) incrementSize(cctx.cacheId()); - GridCacheQueryManager qryMgr = cctx.queries(); + GridCacheQueryManager qryMgr = cctx.queries(); if (qryMgr.enabled()) qryMgr.store(newRow, oldRow, true); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRow.java index 647e68dbf51b2..970aa992c8241 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRow.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRow.java @@ -58,6 +58,11 @@ public interface CacheDataRow extends CacheSearchRow, Storable { */ public void key(KeyCacheObject key); + /** + * @return {@code True} if store cache ID. + */ + public boolean storeCacheId(); + /** {@inheritDoc} */ @Override public default IOVersions ioVersions() { return DataPageIO.VERSIONS; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java index 810423838a3a8..16092ab9b2cd1 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java @@ -58,6 +58,15 @@ * Cache data row adapter. */ public class CacheDataRowAdapter implements CacheDataRow { + /** Version is ready flag. */ + protected static final byte FLAG_VER_READY = 0x01; + + /** Store cache ID flag. */ + protected static final byte FLAG_STORE_CACHE_ID = 0x02; + + /** Allow null as value. */ + protected static final byte FLAG_ALLOW_NULL_VAL = 0x04; + /** */ @GridToStringExclude protected long link; @@ -68,7 +77,7 @@ public class CacheDataRowAdapter implements CacheDataRow { /** */ @GridToStringInclude - protected CacheObject val; + @Nullable protected CacheObject val; /** */ @GridToStringInclude @@ -78,8 +87,8 @@ public class CacheDataRowAdapter implements CacheDataRow { @GridToStringInclude protected GridCacheVersion ver; - /** Whether version is ready. */ - protected boolean verReady; + /** */ + protected byte flags; /** */ @GridToStringInclude @@ -105,7 +114,7 @@ public CacheDataRowAdapter(KeyCacheObject key, CacheObject val, GridCacheVersion this.ver = ver; this.expireTime = expireTime; - verReady = true; + flags = FLAG_VER_READY; } /** @@ -291,6 +300,9 @@ private void doInitFromLink( assert link != 0 : "link"; assert key == null : "key"; + if (readCacheId) + flags |= FLAG_STORE_CACHE_ID; + long nextLink = link; do { @@ -483,7 +495,7 @@ protected IncompleteObject readFragment( } // Read version. - if (!verReady) { + if ((flags & FLAG_VER_READY) == 0) { incomplete = readIncompleteVersion(buf, incomplete, skipVer); assert skipVer || ver != null || incomplete != null; @@ -562,7 +574,7 @@ protected void readFullRow( verLen = CacheVersionIO.size(ver, false); } - verReady = true; + flags |= FLAG_VER_READY; off += verLen; @@ -737,7 +749,7 @@ protected IncompleteObject readIncompleteVersion( assert ver != null; } - verReady = true; + flags |= FLAG_VER_READY; return null; } @@ -759,7 +771,7 @@ protected IncompleteObject readIncompleteVersion( assert ver != null; } - verReady = true; + flags |= FLAG_VER_READY; } assert !buf.hasRemaining(); @@ -825,7 +837,7 @@ private long[] relatedPageIds( * @return {@code True} if entry is ready. */ public boolean isReady() { - return verReady && val != null && key != null; + return ((flags & FLAG_VER_READY) != 0) && val != null && key != null; } /** {@inheritDoc} */ @@ -844,6 +856,11 @@ public boolean isReady() { this.key = key; } + /** {@inheritDoc} */ + @Override public boolean storeCacheId() { + return (flags & FLAG_STORE_CACHE_ID) != 0; + } + /** {@inheritDoc} */ @Override public int cacheId() { return cacheId; @@ -851,14 +868,14 @@ public boolean isReady() { /** {@inheritDoc} */ @Override public CacheObject value() { - assert val != null : "Value is not ready: " + this; + assert val != null || (flags & FLAG_ALLOW_NULL_VAL) != 0 : "Value is not ready: " + this; return val; } /** {@inheritDoc} */ @Override public GridCacheVersion version() { - assert verReady : "Version is not ready: " + this; + assert (flags & FLAG_VER_READY) != 0 : "Version is not ready: " + this; return ver; } @@ -894,7 +911,7 @@ public boolean isReady() { len += value().valueBytesLength(null) + CacheVersionIO.size(version(), false) + 8; - return len + (cacheId() != 0 ? 4 : 0); + return len + (storeCacheId() ? 4 : 0); } /** diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/DataRowCacheAware.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/DataRowCacheAware.java deleted file mode 100644 index baee401320841..0000000000000 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/DataRowCacheAware.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * 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.processors.cache.persistence; - -import org.apache.ignite.internal.processors.cache.CacheObject; -import org.apache.ignite.internal.processors.cache.KeyCacheObject; -import org.apache.ignite.internal.processors.cache.tree.DataRow; -import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; -import org.apache.ignite.internal.util.typedef.internal.CU; -import org.jetbrains.annotations.Nullable; - -/** - * Data row implementation that can optionally hide the cache identifier and can set {@code null} as value.
It is - * used to simplify storing a data row into page memory, because in some cases the cache identifier is not stored on the - * data pages, but is required to link this data row in {@code BPlusTree}. - */ -public class DataRowCacheAware extends DataRow { - /** Flag indicates that cacheId should be stored in data page. */ - private boolean storeCacheId; - - /** - * @param key Key. - * @param val Value. - * @param ver Version. - * @param part Partition. - * @param expireTime Expire time. - * @param cacheId Cache ID. - * @param storeCacheId Flag indicates that cacheId should be stored in data page. - */ - public DataRowCacheAware(KeyCacheObject key, @Nullable CacheObject val, GridCacheVersion ver, int part, - long expireTime, int cacheId, boolean storeCacheId) { - super(key, val, ver, part, expireTime, cacheId); - - storeCacheId(storeCacheId); - } - - /** - * @param storeCacheId Flag indicates that cacheId should be stored in data page. - */ - public void storeCacheId(boolean storeCacheId) { - this.storeCacheId = storeCacheId; - } - - /** {@inheritDoc} */ - @Override public int cacheId() { - return storeCacheId ? cacheId : CU.UNDEFINED_CACHE_ID; - } - - /** {@inheritDoc} */ - @Override public @Nullable CacheObject value() { - return val; - } -} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java index 102469f82f641..69141f07597bc 100755 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java @@ -2893,8 +2893,7 @@ public boolean applyDataEntry( dataEntry.value(), dataEntry.writeVersion(), dataEntry.expireTime(), - locPart, - null); + locPart); return true; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java index 28911df7c78e1..da86972bb361c 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java @@ -99,6 +99,7 @@ import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer; import org.apache.ignite.internal.processors.cache.tree.CacheDataRowStore; 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.tree.PendingRow; import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; @@ -1638,6 +1639,11 @@ private DataEntryRow(DataEntry entry) { throw new UnsupportedOperationException(); } + /** {@inheritDoc} */ + @Override public boolean storeCacheId() { + throw new UnsupportedOperationException(); + } + /** {@inheritDoc} */ @Override public long link() { return 0; @@ -2466,38 +2472,40 @@ private void checkGapsLinkAndPartMetaStorage(PagePartitionMetaIOV3 io, long page /** {@inheritDoc} */ @Override public void update( - GridCacheContext cctx, + GridCacheContext cctx, KeyCacheObject key, CacheObject val, GridCacheVersion ver, - long expireTime, - @Nullable CacheDataRow oldRow + long expireTime ) throws IgniteCheckedException { assert grp.shared().database().checkpointLockIsHeldByThread(); CacheDataStore delegate = init0(false); - delegate.update(cctx, key, val, ver, expireTime, oldRow); + delegate.update(cctx, key, val, ver, expireTime); } /** {@inheritDoc} */ - @Override public CacheDataRow createRow( - GridCacheContext cctx, + @Override public CacheDataRow updateRow( + GridCacheContext cctx, KeyCacheObject key, CacheObject val, GridCacheVersion ver, long expireTime, - @Nullable CacheDataRow oldRow) throws IgniteCheckedException { + @Nullable CacheDataRow oldRow + ) throws IgniteCheckedException { assert grp.shared().database().checkpointLockIsHeldByThread(); CacheDataStore delegate = init0(false); - return delegate.createRow(cctx, key, val, ver, expireTime, oldRow); + return delegate.updateRow(cctx, key, val, ver, expireTime, oldRow); } /** {@inheritDoc} */ - @Override public void insertRows(Collection rows, - IgnitePredicateX initPred) throws IgniteCheckedException { + @Override public void insertRows( + Collection rows, + IgnitePredicateX initPred + ) throws IgniteCheckedException { CacheDataStore delegate = init0(false); delegate.insertRows(rows, initPred); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/RowStore.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/RowStore.java index cffcf9b1e5be0..4716bc2ac806f 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/RowStore.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/RowStore.java @@ -30,6 +30,8 @@ import org.apache.ignite.internal.processors.query.GridQueryRowCacheCleaner; import org.apache.ignite.internal.util.typedef.internal.U; +import static org.apache.ignite.internal.processors.rollingupgrade.feature.SupportedFeatureRegistry.MULTI_PAGE_IN_PLACE_ROW_UPDATE_FEATURE; + /** * Data store for H2 rows. */ @@ -140,20 +142,23 @@ public void addRows(Collection rows, } /** - * @param link Row link. - * @param row New row data. + * @param oldRow Old row data. + * @param newRow New row data. * @return {@code True} if was able to update row. * @throws IgniteCheckedException If failed. */ - public boolean updateRow(long link, CacheDataRow row, IoStatisticsHolder statHolder) throws IgniteCheckedException { + public boolean updateRow(CacheDataRow oldRow, CacheDataRow newRow, IoStatisticsHolder statHolder) throws IgniteCheckedException { assert !persistenceEnabled || ctx.database().checkpointLockIsHeldByThread(); GridQueryRowCacheCleaner rowCacheCleaner0 = rowCacheCleaner.get(); if (rowCacheCleaner0 != null) - rowCacheCleaner0.remove(link); + rowCacheCleaner0.remove(oldRow.link()); + + boolean allowFragmented = oldRow.expireTime() == 0 + && grp.shared().kernalContext().rollingUpgrade().features().isActive(MULTI_PAGE_IN_PLACE_ROW_UPDATE_FEATURE); - return freeList.updateDataRow(link, row, statHolder); + return freeList.updateDataRow(oldRow, newRow, allowFragmented, statHolder); } /** diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/AbstractFreeList.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/AbstractFreeList.java index 62452095631cf..ad0ed2f74d63c 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/AbstractFreeList.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/AbstractFreeList.java @@ -41,6 +41,7 @@ import org.apache.ignite.internal.processors.cache.persistence.evict.PageEvictionTracker; import org.apache.ignite.internal.processors.cache.persistence.tree.io.AbstractDataPageIO; import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPagePayload; +import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageUpdateResult; import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO; import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.LongListReuseBag; import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseBag; @@ -90,7 +91,10 @@ public abstract class AbstractFreeList extends PagesList imp private final int MIN_SIZE_FOR_DATA_PAGE; /** */ - private final PageHandler updateRow = new UpdateRowHandler(); + private final PageHandler updateSignlePageRow = new UpdateSinglePageRowHandler(); + + /** */ + private final PageHandler updateFragmentedRow = new UpdateFragmentedRowHandler(); /** */ private final DataRegionMetricsImpl memMetrics; @@ -104,7 +108,7 @@ public abstract class AbstractFreeList extends PagesList imp /** * */ - private final class UpdateRowHandler extends PageHandler { + private final class UpdateSinglePageRowHandler extends PageHandler { /** {@inheritDoc} */ @Override public Boolean run( int cacheId, @@ -115,40 +119,122 @@ private final class UpdateRowHandler extends PageHandler { Boolean walPlc, T row, int itemId, - IoStatisticsHolder statHolder) - throws IgniteCheckedException { + IoStatisticsHolder statHolder + ) throws IgniteCheckedException { AbstractDataPageIO io = (AbstractDataPageIO)iox; int rowSize = row.size(); - boolean updated = io.updateRow(pageAddr, itemId, pageSize(), null, row, rowSize); + boolean updated = io.updateRow(pageAddr, itemId, pageSize(), row, rowSize); evictionTracker.touchPage(pageId); - if (updated && needWalDeltaRecord(pageId, page, walPlc)) { - // TODO This record must contain only a reference to a logical WAL record with the actual data. - byte[] payload = new byte[rowSize]; + if (updated) { + statHolder.trackPageRemoveData(rowSize); + statHolder.trackPageInsertData(rowSize); - DataPagePayload data = io.readPayload(pageAddr, itemId, pageSize()); + if (needWalDeltaRecord(pageId, page, walPlc)) { + // TODO IGNITE-5829 This record must contain only a reference to a logical WAL record with the actual data. + byte[] payload = new byte[rowSize]; - assert data.payloadSize() == rowSize; + DataPagePayload data = io.readPayload(pageAddr, itemId, pageSize()); - PageUtils.getBytes(pageAddr, data.offset(), payload, 0, rowSize); + assert data.payloadSize() == rowSize; - statHolder.trackPageRemoveData(rowSize); - statHolder.trackPageInsertData(rowSize); + PageUtils.getBytes(pageAddr, data.offset(), payload, 0, rowSize); - wal.log(new DataPageUpdateRecord( - cacheId, - pageId, - itemId, - payload)); + wal.log(new DataPageUpdateRecord( + cacheId, + pageId, + itemId, + payload)); + } } return updated; } } + /** Current state (for last processed page) of partially written row. */ + private final class PartiallyWritten { + /** */ + private final T row; + + /** */ + private long nextLink; + + /** */ + private int written; + + /** */ + private boolean modified; + + /** */ + public PartiallyWritten(T row) { + this.row = row; + } + } + + /** + * + */ + private final class UpdateFragmentedRowHandler extends PageHandler { + /** {@inheritDoc} */ + @Override public PartiallyWritten run( + int cacheId, + long pageId, + long page, + long pageAddr, + PageIO iox, + Boolean walPlc, + PartiallyWritten fragment, + int itemId, + IoStatisticsHolder statHolder + ) throws IgniteCheckedException { + AbstractDataPageIO io = (AbstractDataPageIO)iox; + + boolean walEnabled = wal != null && !wal.pageRecordsDisabled(grpId, pageId); + + DataPageUpdateResult updateRes = io.updateRowFragment(pageMem, pageAddr, itemId, pageSize(), + fragment.row, fragment.written, walEnabled); + + evictionTracker.touchPage(pageId); + + if (updateRes != null) { + if (updateRes.modifiedPayload() != null && walEnabled && needWalDeltaRecord(pageId, page, walPlc)) { + wal.log(new DataPageUpdateRecord( + cacheId, + pageId, + itemId, + updateRes.modifiedPayload())); + } + + fragment.modified = !walEnabled || updateRes.modifiedPayload() != null; + fragment.nextLink = updateRes.nextLink(); + fragment.written += updateRes.payloadSize(); + } + else { + fragment.modified = true; + fragment.nextLink = 0L; + fragment.written = fragment.row.size(); + } + + return fragment; + } + + /** {@inheritDoc} */ + @Override public boolean markDirtyAfterWrite( + int cacheId, + long pageId, + long page, + long pageAddr, + PartiallyWritten fragment, + int intArg + ) { + return fragment.modified; + } + } + /** Write a single row on a single page. */ private final WriteRowHandler writeRowHnd = new WriteRowHandler(); @@ -788,19 +874,51 @@ private long initReusedPage(T row, long reusedPageId, IoStatisticsHolder statHol } /** {@inheritDoc} */ - @Override public boolean updateDataRow(long link, T row, - IoStatisticsHolder statHolder) throws IgniteCheckedException { + @Override public boolean updateDataRow( + T oldRow, + T newRow, + boolean allowFragmented, + IoStatisticsHolder statHolder + ) throws IgniteCheckedException { + long link = oldRow.link(); + int size = newRow.size(); + assert link != 0; + assert oldRow.size() == size : + "Unexpected row size on update [oldSize=" + oldRow.size() + ", newSize=" + size + ']'; try { long pageId = PageIdUtils.pageId(link); int itemId = PageIdUtils.itemId(link); - Boolean updated = write(pageId, updateRow, row, itemId, null, statHolder); + if (!allowFragmented || size <= pageSize() - AbstractDataPageIO.MIN_DATA_PAGE_OVERHEAD) { + Boolean updated = write(pageId, updateSignlePageRow, newRow, itemId, null, statHolder); - assert updated != null; // Can't fail here. + assert updated != null; // Can't fail here. - return updated; + if (updated || !allowFragmented) + return updated; // If allow fragmented fallback to fragmented row update. + } + + PartiallyWritten updateRes = write(pageId, updateFragmentedRow, new PartiallyWritten(newRow), itemId, + null, statHolder); + + while (updateRes.written < size) { + pageId = PageIdUtils.pageId(updateRes.nextLink); + itemId = PageIdUtils.itemId(updateRes.nextLink); + + updateRes = write(pageId, updateFragmentedRow, updateRes, itemId, null, statHolder); + } + statHolder.trackPageRemoveData(size); + statHolder.trackPageInsertData(size); + + assert updateRes.written == size : + "Unexpected written row size [written=" + updateRes.written + ", rowSize=" + size + ']'; + + assert updateRes.nextLink == 0 : + "Unexpected next page link [nextLink=" + Long.toHexString(updateRes.nextLink) + ']'; + + return true; } catch (AssertionError e) { throw corruptedFreeListException(e); @@ -809,7 +927,7 @@ private long initReusedPage(T row, long reusedPageId, IoStatisticsHolder statHol throw e; } catch (Throwable t) { - throw new CorruptedFreeListException("Failed to update data row", t, grpId); + throw new CorruptedFreeListException("Failed to update data newRow", t, grpId); } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeList.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeList.java index 7d492fd46f8a3..687686060c815 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeList.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeList.java @@ -41,13 +41,19 @@ public interface FreeList { public void insertDataRows(Collection rows, IoStatisticsHolder statHolder) throws IgniteCheckedException; /** - * @param link Row link. - * @param row New row data. + * @param oldRow Old row data. + * @param newRow New row data. + * @param allowFragmented Allow fragmented pages. * @param statHolder Statistics holder to track IO operations. - * @return {@code True} if was able to update row. + * @return {@code True} if was able to update newRow. * @throws IgniteCheckedException If failed. */ - public boolean updateDataRow(long link, T row, IoStatisticsHolder statHolder) throws IgniteCheckedException; + public boolean updateDataRow( + T oldRow, + T newRow, + boolean allowFragmented, + IoStatisticsHolder statHolder + ) throws IgniteCheckedException; /** * @param link Row link. diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/AbstractDataPageIO.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/AbstractDataPageIO.java index 9802a6ece19c0..ca9061cd48311 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/AbstractDataPageIO.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/AbstractDataPageIO.java @@ -776,10 +776,11 @@ private int findIndirectIndexForLastDirect(long pageAddr, int directCnt, int ind } /** + * In-place signle-page row update. + * * @param pageAddr Page address. * @param itemId Item ID. * @param pageSize Page size. - * @param payload Row data. * @param row Row. * @param rowSize Row size. * @return {@code True} if entry is not fragmented. @@ -789,11 +790,10 @@ public boolean updateRow( final long pageAddr, int itemId, int pageSize, - @Nullable byte[] payload, - @Nullable T row, - final int rowSize) throws IgniteCheckedException { + T row, + final int rowSize + ) throws IgniteCheckedException { assert checkIndex(itemId) : itemId; - assert row != null ^ payload != null; assertPageType(pageAddr); final int dataOff = getDataOffset(pageAddr, itemId, pageSize); @@ -801,14 +801,99 @@ public boolean updateRow( if (isFragmented(pageAddr, dataOff)) return false; - if (row != null) - writeRowData(pageAddr, dataOff, rowSize, row, false); - else - writeRowData(pageAddr, dataOff, payload); + writeRowData(pageAddr, dataOff, rowSize, row, false); return true; } + /** + * In-place row update. Modifies only entry payload, keeps size and next fragment link. + * + * @param pageAddr Page address. + * @param itemId Item ID. + * @param pageSize Page size. + * @param payload Row data. + */ + public void updateRow( + final long pageAddr, + int itemId, + int pageSize, + byte[] payload + ) { + assert checkIndex(itemId) : itemId; + assertPageType(pageAddr); + + int dataOff = getDataOffset(pageAddr, itemId, pageSize); + + boolean fragmented = isFragmented(pageAddr, dataOff); + + assert getPageEntrySize(pageAddr, dataOff, 0) == payload.length : "Unexpected payload length [" + + "stored=" + getPageEntrySize(pageAddr, dataOff, 0) + ", updated=" + payload.length + ']'; + + PageUtils.putBytes(pageAddr, dataOff + PAYLOAD_LEN_SIZE + (fragmented ? LINK_SIZE : 0), payload); + } + + /** + * In-place fragmented row update. + * + * @param pageAddr Page address. + * @param itemId Item ID. + * @param pageSize Page size. + * @param row Row. + * @param needPayload If modified payload required (to write WAL records). + * @return Next page and payload if needed, or {@code null} for last page if no payload needed. + * @throws IgniteCheckedException If failed. + */ + public @Nullable DataPageUpdateResult updateRowFragment( + PageMemory pageMem, + long pageAddr, + int itemId, + int pageSize, + T row, + int written, + boolean needPayload + ) throws IgniteCheckedException { + assert checkIndex(itemId) : itemId; + assertPageType(pageAddr); + + int dataOff = getDataOffset(pageAddr, itemId, pageSize); + + // Due to different write formats this method can't be used for non-fragmented rows. + assert isFragmented(pageAddr, dataOff); + + long nextLink = getNextFragmentLink(pageAddr, dataOff); + int payloadSize = getPageEntrySize(pageAddr, dataOff, 0); + + ByteBuffer pageBuf = pageMem.pageBuffer(pageAddr); + pageBuf.position(dataOff + PAYLOAD_LEN_SIZE + LINK_SIZE); + pageBuf.limit(pageBuf.position() + payloadSize); + + byte[] modifiedPayload = null; + + if (needPayload) { + ByteBuffer buf = ByteBuffer.allocate(payloadSize).order(pageBuf.order()); + + writeFragmentData(row, buf, written, payloadSize); + + buf.rewind(); + + boolean modified = buf.compareTo(pageBuf) != 0; + + if (modified) { + pageBuf.put(buf); + modifiedPayload = buf.array(); + } + } + else { + writeFragmentData(row, pageBuf, written, payloadSize); + + if (nextLink == 0L) + return null; + } + + return new DataPageUpdateResult(payloadSize, modifiedPayload, nextLink); + } + /** * @param pageAddr Page address. * @param itemId Fixed item ID (the index used for referencing an entry from the outside). @@ -1157,6 +1242,8 @@ private int addRowFragment( int rowOff = rowSize - written - payloadSize; + assertPageType(buf); + writeFragmentData(row, buf, rowOff, payloadSize); } else { diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java index aaf744969e8d3..0547d059387fb 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java @@ -54,7 +54,7 @@ protected DataPageIO(int ver) { long addr = pageAddr + dataOff; - int cacheIdSize = row.cacheId() != 0 ? 4 : 0; + int cacheIdSize = row.storeCacheId() ? 4 : 0; if (newRow) { PageUtils.putShort(addr, 0, (short)payloadSize); @@ -82,8 +82,6 @@ protected DataPageIO(int ver) { /** {@inheritDoc} */ @Override protected void writeFragmentData(CacheDataRow row, ByteBuffer buf, int rowOff, int payloadSize) throws IgniteCheckedException { - assertPageType(buf); - final int keySize = row.key().valueBytesLength(null); final int valSize = row.value().valueBytesLength(null); @@ -130,7 +128,7 @@ private int writeFragment( final int prevLen; final int curLen; - int cacheIdSize = row.cacheId() == 0 ? 0 : 4; + int cacheIdSize = row.storeCacheId() ? 4 : 0; switch (type) { case CACHE_ID: @@ -174,8 +172,10 @@ private int writeFragment( if (type == EXPIRE_TIME) writeExpireTimeFragment(buf, row.expireTime(), rowOff, len, prevLen); - else if (type == CACHE_ID) - writeCacheIdFragment(buf, row.cacheId(), rowOff, len, prevLen); + else if (type == CACHE_ID) { + if (cacheIdSize != 0) + writeCacheIdFragment(buf, row.cacheId(), rowOff, len, prevLen); + } else if (type != VERSION) { // Write key or value. final CacheObject co = type == KEY ? row.key() : row.value(); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageUpdateResult.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageUpdateResult.java new file mode 100644 index 0000000000000..83376e2f4b0bb --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageUpdateResult.java @@ -0,0 +1,73 @@ +/* + * 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.processors.cache.persistence.tree.io; + +import org.apache.ignite.internal.util.typedef.internal.S; + +/** + * Data page update result. + */ +@SuppressWarnings("AssignmentOrReturnOfFieldWithMutableType") +public class DataPageUpdateResult { + /** */ + private final int payloadSize; + + /** */ + private final byte[] payload; + + /** */ + private final long nextLink; + + /** + * @param payloadSize Payload size. + * @param payload Payload, if it was modified or {@code null}. + * @param nextLink Next link. + */ + DataPageUpdateResult(int payloadSize, byte[] payload, long nextLink) { + this.payloadSize = payloadSize; + this.payload = payload; + this.nextLink = nextLink; + } + + /** + * @return Modified payload. + */ + public byte[] modifiedPayload() { + return payload; + } + + /** + * @return Link to the next fragment or {@code 0} if it is the last fragment or the data row is not fragmented. + */ + public long nextLink() { + return nextLink; + } + + /** + * @return Payload size. + */ + public int payloadSize() { + return payloadSize; + } + + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(DataPageUpdateResult.class, this); + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/util/PageHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/util/PageHandler.java index b62b7d4154caa..aee74765d956e 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/util/PageHandler.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/util/PageHandler.java @@ -93,6 +93,26 @@ public boolean releaseAfterWrite( return true; } + /** + * @param cacheId Cache ID. + * @param pageId Page ID. + * @param page Page pointer. + * @param pageAddr Page address. + * @param arg Argument. + * @param intArg Argument of type {@code int}. + * @return {@code True} if mark page as dirty. + */ + public boolean markDirtyAfterWrite( + int cacheId, + long pageId, + long page, + long pageAddr, + X arg, + int intArg + ) { + return true; + } + /** * @param pageMem Page memory. * @param cacheId Cache ID. @@ -299,7 +319,7 @@ public static R writePage( if (pageAddr == 0L) return lockFailed; - boolean ok = false; + boolean markDirty = false; try { if (init != null) { @@ -312,7 +332,7 @@ public static R writePage( R res = h.run(grpId, pageId, page, pageAddr, init, walPlc, arg, intArg, statHolder); - ok = true; + markDirty = h.markDirtyAfterWrite(grpId, pageId, page, pageAddr, arg, intArg); return res; } @@ -320,7 +340,7 @@ public static R writePage( assert PageIO.getCrc(pageAddr) == 0; //TODO GG-11480 if (releaseAfterWrite = h.releaseAfterWrite(grpId, pageId, page, pageAddr, arg, intArg)) - writeUnlock(pageMem, grpId, pageId, page, pageAddr, lsnr, walPlc, ok); + writeUnlock(pageMem, grpId, pageId, page, pageAddr, lsnr, walPlc, markDirty); } } finally { @@ -367,7 +387,7 @@ public static R writePage( if (pageAddr == 0L) return lockFailed; - boolean ok = false; + boolean markDirty = false; try { if (init != null) { @@ -380,7 +400,7 @@ public static R writePage( R res = h.run(grpId, pageId, page, pageAddr, init, walPlc, arg, intArg, statHolder); - ok = true; + markDirty = h.markDirtyAfterWrite(grpId, pageId, page, pageAddr, arg, intArg); return res; } @@ -388,7 +408,7 @@ public static R writePage( assert PageIO.getCrc(pageAddr) == 0; //TODO GG-11480 if (h.releaseAfterWrite(grpId, pageId, page, pageAddr, arg, intArg)) - writeUnlock(pageMem, grpId, pageId, page, pageAddr, lsnr, walPlc, ok); + writeUnlock(pageMem, grpId, pageId, page, pageAddr, lsnr, walPlc, markDirty); } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/DataRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/DataRow.java index 362a0d9e15aac..b06767ead4492 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/DataRow.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/DataRow.java @@ -24,6 +24,7 @@ import org.apache.ignite.internal.processors.cache.KeyCacheObject; import org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter; import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; +import org.jetbrains.annotations.Nullable; /** * @@ -69,7 +70,15 @@ protected DataRow(CacheGroupContext grp, int hash, long link, int part, RowData * @param expireTime Expire time. * @param cacheId Cache ID. */ - public DataRow(KeyCacheObject key, CacheObject val, GridCacheVersion ver, int part, long expireTime, int cacheId) { + public DataRow( + KeyCacheObject key, + @Nullable CacheObject val, + GridCacheVersion ver, + int part, + long expireTime, + int cacheId, + boolean storeCacheId + ) { super(0); this.hash = key.hashCode(); @@ -80,7 +89,9 @@ public DataRow(KeyCacheObject key, CacheObject val, GridCacheVersion ver, int pa this.expireTime = expireTime; this.cacheId = cacheId; - verReady = true; + flags = (byte)(FLAG_VER_READY + + (storeCacheId ? FLAG_STORE_CACHE_ID : 0) + + (val == null ? FLAG_ALLOW_NULL_VAL : 0)); } /** diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/IgniteNodeFeatureSet.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/IgniteNodeFeatureSet.java index d00439e415602..2445b2c3bd671 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/IgniteNodeFeatureSet.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/IgniteNodeFeatureSet.java @@ -53,6 +53,9 @@ public class IgniteNodeFeatureSet implements Message, Externalizable { /** */ @Nullable private volatile Map featuresByComponent; + /** Core features cache, to avoid hash-map lookup for hot code paths. */ + @Nullable private volatile IgniteComponentFeatureSet coreFeatures; + /** */ public IgniteNodeFeatureSet() { // No-op. @@ -63,7 +66,8 @@ public IgniteNodeFeatureSet(IgniteComponentFeatureSet[] features) { assert features != null; this.features = features; - this.featuresByComponent = indexByComponentName(features); + featuresByComponent = indexByComponentName(features); + coreFeatures = featuresByComponent.get(IgniteCoreFeature.COMPONENT_NAME); } /** */ @@ -96,6 +100,10 @@ public boolean containsAll(IgniteNodeFeatureSet other) { /** */ public boolean contains(IgniteFeature feature) { + //noinspection StringEquality + if (feature.componentName() == IgniteCoreFeature.COMPONENT_NAME && coreFeatures != null) + return coreFeatures.contains(feature.id()); + IgniteComponentFeatureSet cmpFeatures = featuresByComponent().get(feature.componentName()); return cmpFeatures != null && cmpFeatures.contains(feature.id()); @@ -110,6 +118,8 @@ private Map featuresByComponent() { featuresByComponent = indexByComponentName(features); + coreFeatures = featuresByComponent.get(IgniteCoreFeature.COMPONENT_NAME); + this.featuresByComponent = featuresByComponent; return featuresByComponent; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/SupportedFeatureRegistry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/SupportedFeatureRegistry.java index 7b3e55b85d3c4..1c64ab6b66a53 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/SupportedFeatureRegistry.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/SupportedFeatureRegistry.java @@ -93,4 +93,7 @@ public class SupportedFeatureRegistry { /** */ public static final IgniteFeature ROLLING_UPGRADE_FEATURE = new IgniteCoreFeature(0); + + /** */ + public static final IgniteFeature MULTI_PAGE_IN_PLACE_ROW_UPDATE_FEATURE = new IgniteCoreFeature(1); } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/database/CacheFreeListSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/database/CacheFreeListSelfTest.java index 3b317c5db3836..0b0d253edc462 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/database/CacheFreeListSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/database/CacheFreeListSelfTest.java @@ -607,6 +607,11 @@ private TestDataRow(int keySize, int valSize) { return len + (cacheId() != 0 ? 4 : 0); } + /** {@inheritDoc} */ + @Override public boolean storeCacheId() { + return cacheId() != 0; + } + /** {@inheritDoc} */ @Override public long link() { return link; diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/database/MultiPageInPlaceUpdateTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/database/MultiPageInPlaceUpdateTest.java new file mode 100644 index 0000000000000..29f825d254dc8 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/database/MultiPageInPlaceUpdateTest.java @@ -0,0 +1,436 @@ +/* + * 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.processors.database; + +import java.io.File; +import java.io.IOException; +import java.nio.file.OpenOption; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicBoolean; +import javax.cache.expiry.CreatedExpiryPolicy; +import javax.cache.expiry.Duration; +import javax.cache.expiry.ModifiedExpiryPolicy; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.cluster.ClusterState; +import org.apache.ignite.configuration.DataRegionConfiguration; +import org.apache.ignite.configuration.DataStorageConfiguration; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.processors.cache.KeyCacheObject; +import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow; +import org.apache.ignite.internal.processors.cache.persistence.file.FileIO; +import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory; +import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccessFileIOFactory; +import org.apache.ignite.internal.processors.cache.tree.CacheDataTree; +import org.apache.ignite.internal.processors.cache.tree.SearchRow; +import org.apache.ignite.internal.util.typedef.internal.CU; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.spi.metric.LongMetric; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.junit.Test; + +import static org.apache.ignite.configuration.DataStorageConfiguration.DFLT_DATA_REG_DEFAULT_NAME; +import static org.apache.ignite.internal.processors.cache.persistence.DataRegionMetricsImpl.DATAREGION_METRICS_PREFIX; +import static org.apache.ignite.internal.processors.metric.impl.MetricUtils.metricName; + +/** + * Tests for multi-page in-place update. + */ +public class MultiPageInPlaceUpdateTest extends GridCommonAbstractTest { + /** */ + private boolean pds; + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); + + cfg.setDataStorageConfiguration(new DataStorageConfiguration() + .setFileIOFactory(new FailingFileIOFactory()) + .setDefaultDataRegionConfiguration(new DataRegionConfiguration() + .setPersistenceEnabled(pds) + .setMetricsEnabled(true) + ) + ); + + return cfg; + } + + /** {@inheritDoc} */ + @Override protected void beforeTest() throws Exception { + super.beforeTest(); + + cleanPersistenceDir(); + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + super.afterTest(); + + stopAllGrids(); + } + + /** */ + @Test + public void testInPlaceUpdateInMemory() throws Exception { + checkInPlaceUpdate(false); + } + + /** */ + @Test + public void testInPlaceUpdatePersistence() throws Exception { + checkInPlaceUpdate(true); + } + + /** */ + private void checkInPlaceUpdate(boolean pds) throws Exception { + this.pds = pds; + + int entrySize = 100 * 1024; + + IgniteEx ignite = startGrid(0); + + if (pds) + ignite.cluster().state(ClusterState.ACTIVE); + + IgniteCache cache = ignite.getOrCreateCache(DEFAULT_CACHE_NAME); + + byte[] payload = new byte[entrySize]; + + ThreadLocalRandom.current().nextBytes(payload); + + int key = 0; + + cache.put(key, payload); + + long link = link(ignite, key); + + for (int i = 0; i < 100; i++) { + payload[ThreadLocalRandom.current().nextInt(entrySize)] = (byte)i; + + cache.put(key, payload); + + assertEqualsArraysAware(payload, cache.get(0)); + + assertEquals(link, link(ignite, key)); + } + + // Size changed, can't do in-place update. + cache.put(key, new byte[payload.length + 1]); + + assertNotSame(link, link(ignite, key)); + } + + /** */ + @Test + public void testUpdateDifferentSizesInMemory() throws Exception { + checkUpdateDifferentSizes(false); + } + + /** */ + @Test + public void testUpdateDifferentSizesPersistence() throws Exception { + checkUpdateDifferentSizes(true); + } + + /** */ + public void checkUpdateDifferentSizes(boolean pds) throws Exception { + this.pds = pds; + + IgniteEx ignite = startGrid(0); + + if (pds) + ignite.cluster().state(ClusterState.ACTIVE); + + IgniteCache cache = ignite.getOrCreateCache(DEFAULT_CACHE_NAME); + + int pageSize = ignite.context().cache().context().database().pageSize(); + + for (int i = 1000; i < pageSize; i++) + checkLinkChange(ignite, cache, i, true, false); + } + + + /** */ + @Test + public void testDirtyPagesCountAfterUpdate() throws Exception { + pds = true; + int entrySize = 100 * 1024; + int entryCnt = 100; + + IgniteEx ignite = startGrid(0); + + ignite.cluster().state(ClusterState.ACTIVE); + IgniteCache cache = ignite.getOrCreateCache(DEFAULT_CACHE_NAME); + + LongMetric dirtyPages = ignite.context().metric().registry(metricName(DATAREGION_METRICS_PREFIX, + DFLT_DATA_REG_DEFAULT_NAME)).findMetric("DirtyPages"); + + int pageSize = ignite.context().cache().context().database().pageSize(); + + byte[] payload = new byte[entrySize]; + + for (int i = 0; i < entryCnt; i++) + cache.put(i, payload); + + assertTrue(dirtyPages.value() >= entryCnt * entrySize / pageSize); + + forceCheckpoint(); + + // Update only first page. + payload[0] = 1; + + for (int i = 0; i < entryCnt; i++) + cache.put(i, payload); + + assertTrue(dirtyPages.value() < entryCnt * entrySize / pageSize); + + for (int i = 0; i < entryCnt; i++) + assertEqualsArraysAware(payload, cache.get(i)); + + forceCheckpoint(); + + // Update only last page. + payload[entrySize - 1] = 1; + + for (int i = 0; i < entryCnt; i++) + cache.put(i, payload); + + assertTrue(dirtyPages.value() < entryCnt * entrySize / pageSize); + + for (int i = 0; i < entryCnt; i++) + assertEqualsArraysAware(payload, cache.get(i)); + + // Update some intermediate page. + payload[entrySize / 2] = 1; + + for (int i = 0; i < entryCnt; i++) + cache.put(i, payload); + + assertTrue(dirtyPages.value() < entryCnt * entrySize / pageSize); + + for (int i = 0; i < entryCnt; i++) + assertEqualsArraysAware(payload, cache.get(i)); + } + + /** */ + @Test + public void testApplyInPlaceUpdateDeltaRecordsAfterCrash() throws Exception { + pds = true; + int entrySize = 100 * 1024; + int entryCnt = 100; + + IgniteEx ignite = startGrid(0); + + ignite.cluster().state(ClusterState.ACTIVE); + IgniteCache cache = ignite.getOrCreateCache(DEFAULT_CACHE_NAME); + + byte[] payload = new byte[entrySize]; + long[] links = new long[entryCnt]; + + for (int i = 0; i < entryCnt; i++) { + cache.put(i, payload); + links[i] = link(ignite, i); + } + + forceCheckpoint(); + + byte[][] payloads = new byte[entryCnt][entrySize]; + + for (int i = 0; i < entryCnt / 2; i++) { + // First change of the page - produce page snapshot WAL record. + ThreadLocalRandom.current().nextBytes(payloads[i]); + cache.put(i, payloads[i]); + assertEquals(links[i], link(ignite, i)); + // Second change of the page - produce delta WAL record. + ThreadLocalRandom.current().nextBytes(payloads[i]); + cache.put(i, payloads[i]); + assertEquals(links[i], link(ignite, i)); + } + + int pageSize = ignite.context().cache().context().database().pageSize(); + + for (int i = entryCnt / 2; i < entryCnt; i++) { + // Randomly modify pages of entry. + // - Some pages can produce only page snapshot. + // - Some pages can produce both page snapshot and delta pages. + // - Some pages can be untouched. + for (int j = 0; j < entrySize / pageSize; j++) { + payload[ThreadLocalRandom.current().nextInt(entrySize)] = (byte)j; + cache.put(i, payloads[i]); + assertEquals(links[i], link(ignite, i)); + } + } + + FailingFileIOFactory failingFactory = (FailingFileIOFactory)ignite.configuration() + .getDataStorageConfiguration().getFileIOFactory(); + + failingFactory.failFlag.set(true); + + try { + forceCheckpoint(); + + fail("Expected failure on checkpoint"); + } + catch (Exception ignore) { + // Expected. + } + + stopGrid(0); + ignite = startGrid(0); + + cache = ignite.cache(DEFAULT_CACHE_NAME); + + for (int i = 0; i < entryCnt; i++) { + assertEqualsArraysAware(payloads[i], cache.get(i)); + assertEquals(links[i], link(ignite, i)); + } + } + + /** */ + @Test + public void testLogicalRecoveryInPlaceUpdatedEntriesAfterCrash() throws Exception { + pds = true; + int entrySize = 100 * 1024; + int entryCnt = 100; + + IgniteEx ignite = startGrid(0); + + ignite.cluster().state(ClusterState.ACTIVE); + IgniteCache cache = ignite.getOrCreateCache(DEFAULT_CACHE_NAME); + + byte[] payload = new byte[entrySize]; + long[] links = new long[entryCnt]; + + for (int i = 0; i < entryCnt; i++) { + cache.put(i, payload); + links[i] = link(ignite, i); + } + + forceCheckpoint(); + + byte[][] payloads = new byte[entryCnt][entrySize]; + for (int i = 0; i < entryCnt; i++) { + ThreadLocalRandom.current().nextBytes(payloads[i]); + cache.put(i, payloads[i]); + assertEquals(links[i], link(ignite, i)); + } + + stopGrid(0, true); + ignite = startGrid(0); + + cache = ignite.cache(DEFAULT_CACHE_NAME); + + for (int i = 0; i < entryCnt; i++) { + assertEqualsArraysAware(payloads[i], cache.get(i)); + // Entry has the same payload, but link can be changed, since logical recovery applies records never using + // in-place update. + assertNotSame(links[i], link(ignite, i)); + } + } + + /** */ + @Test + public void testInPlaceUpdateWithTtl() throws Exception { + IgniteEx ignite = startGrid(0); + + IgniteCache cache = ignite.getOrCreateCache(DEFAULT_CACHE_NAME); + cache = cache.withExpiryPolicy(new CreatedExpiryPolicy(Duration.ONE_DAY)); + + // In-place update is enabled when TTL is not changed and entry occupies only one page. + checkLinkChange(ignite, cache, 100, true, false); + + // In-place update is disabled when TTL is not changed, but entry occupies more than one page. + checkLinkChange(ignite, cache, ignite.context().cache().context().database().pageSize(), false, false); + + cache = cache.withExpiryPolicy(new ModifiedExpiryPolicy(Duration.ONE_DAY)); + + // In-place update is disabled when TTL is changed. + checkLinkChange(ignite, cache, 100, false, true); + } + + /** */ + private void checkLinkChange( + IgniteEx ignite, + IgniteCache cache, + int payloadSize, + boolean expectInPlaceUpdate, + boolean ensureTtlChanged + ) throws IgniteCheckedException { + int key = 0; + + byte[] payload = new byte[payloadSize]; + ThreadLocalRandom.current().nextBytes(payload); + + cache.put(key, payload); + long link = link(ignite, key); + + long ts = U.currentTimeMillis(); + + if (ensureTtlChanged) { + while (ts == U.currentTimeMillis()) + doSleep(10); + } + + ThreadLocalRandom.current().nextBytes(payload); + cache.put(key, payload); + + assertEquals(expectInPlaceUpdate, link == link(ignite, key)); + assertEqualsArraysAware(payload, cache.get(key)); + } + + /** */ + private long link(IgniteEx ignite, Object key) throws IgniteCheckedException { + KeyCacheObject keyCacheObj = ignite.cachex(DEFAULT_CACHE_NAME).context().toCacheKeyObject(key); + SearchRow searchRow = new SearchRow(CU.cacheId(DEFAULT_CACHE_NAME), keyCacheObj); + + CacheDataTree tree = ignite.cachex(DEFAULT_CACHE_NAME).context().topology() + .localPartition(keyCacheObj.partition()).dataStore().tree(); + + assertNotNull(tree); + + CacheDataRow row = tree.findOne(searchRow); + assertNotNull(row); + + return row.link(); + } + + /** */ + private static final class FailingFileIOFactory implements FileIOFactory { + /** */ + private final FileIOFactory delegateFactory; + + /** */ + private final AtomicBoolean failFlag = new AtomicBoolean(); + + /** */ + FailingFileIOFactory() { + delegateFactory = new RandomAccessFileIOFactory(); + } + + /** {@inheritDoc} */ + @Override public FileIO create(File file, OpenOption... modes) throws IOException { + FileIO delegate = delegateFactory.create(file, modes); + + if (failFlag.get() && file.getName().contains("END.bin")) + throw new IOException("Test exception"); + + return delegate; + } + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite2.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite2.java index 9b5ca62ce8523..835340c55eed1 100644 --- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite2.java +++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite2.java @@ -74,6 +74,7 @@ import org.apache.ignite.internal.processors.database.DataRegionMetricsSelfTest; import org.apache.ignite.internal.processors.database.FreeListCutTailDifferentGcTest; import org.apache.ignite.internal.processors.database.IndexStorageSelfTest; +import org.apache.ignite.internal.processors.database.MultiPageInPlaceUpdateTest; import org.apache.ignite.internal.processors.database.SwapPathConstructionSelfTest; import org.apache.ignite.internal.processors.failure.FailureProcessorLoggingTest; import org.apache.ignite.internal.processors.failure.FailureProcessorThreadDumpThrottlingTest; @@ -230,6 +231,7 @@ MdcCacheReadRequestsRoutingTest.class, IgniteConfigurationTest.class, + MultiPageInPlaceUpdateTest.class, }) public class IgniteBasicTestSuite2 { } diff --git a/modules/core/src/test/java/org/apache/ignite/util/TestStorageUtils.java b/modules/core/src/test/java/org/apache/ignite/util/TestStorageUtils.java index 9c696e81355c0..4ba17f4dd6b2e 100644 --- a/modules/core/src/test/java/org/apache/ignite/util/TestStorageUtils.java +++ b/modules/core/src/test/java/org/apache/ignite/util/TestStorageUtils.java @@ -80,8 +80,7 @@ public static void corruptDataEntry( dataEntry.value(), dataEntry.writeVersion(), dataEntry.expireTime(), - locPart, - null); + locPart); ctx.offheap().dataStore(locPart).updateInitialCounter(dataEntry.partitionCounter() - 1, 1); } diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/H2CacheRow.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/H2CacheRow.java index c9b4db13be711..f1ad9b41ffc9c 100644 --- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/H2CacheRow.java +++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/H2CacheRow.java @@ -263,6 +263,11 @@ private boolean removedRow() { throw new UnsupportedOperationException(); } + /** {@inheritDoc} */ + @Override public boolean storeCacheId() { + throw new UnsupportedOperationException(); + } + /** {@inheritDoc} */ @Override public String toString() { SB sb = new SB("Row@");