You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by se...@apache.org on 2020/11/12 13:42:41 UTC
[ignite] branch master updated: IGNITE-13684 Page memory
refactoring and improvements needed for defragmentation (IEP-47) - Fixes
#8435.
This is an automated email from the ASF dual-hosted git repository.
sergeychugunov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 431f7e76 IGNITE-13684 Page memory refactoring and improvements needed for defragmentation (IEP-47) - Fixes #8435.
431f7e76 is described below
commit 431f7e76acf4b4a13f18b6f59167c68fc01e31a6
Author: ibessonov <be...@gmail.com>
AuthorDate: Thu Nov 12 16:42:06 2020 +0300
IGNITE-13684 Page memory refactoring and improvements needed for defragmentation (IEP-47) - Fixes #8435.
Signed-off-by: Sergey Chugunov <se...@gmail.com>
---
.../ignite/internal/pagemem/PageIdAllocator.java | 5 +
.../pagemem/store/IgnitePageStoreManager.java | 29 +---
.../store/PageStoreCollection.java} | 38 +++---
.../processors/cache/CacheGroupContext.java | 2 +-
.../cache/IgniteCacheOffheapManager.java | 7 +
.../cache/IgniteCacheOffheapManagerImpl.java | 145 ++++++++++++++------
.../cache/PartitionUpdateCounterMvccImpl.java | 15 +++
.../cache/PartitionUpdateCounterTrackingImpl.java | 8 +-
.../processors/cache/mvcc/MvccProcessorImpl.java | 3 +-
.../processors/cache/mvcc/txlog/TxLog.java | 2 +-
.../cache/persistence/DataStructure.java | 22 +++-
.../GridCacheDatabaseSharedManager.java | 32 +++--
.../cache/persistence/GridCacheOffheapManager.java | 129 ++++++++++++------
.../IgniteCacheDatabaseSharedManager.java | 28 +++-
.../persistence/checkpoint/CheckpointManager.java | 2 +-
.../checkpoint/LightweightCheckpointManager.java | 2 +-
.../cache/persistence/file/FileIOFactory.java | 1 +
.../cache/persistence/file/FilePageStore.java | 16 +--
.../persistence/file/FilePageStoreFactory.java | 6 +-
.../persistence/file/FilePageStoreManager.java | 144 ++++----------------
.../cache/persistence/file/FilePageStoreV2.java | 4 +-
.../file/FileVersionCheckingFactory.java | 6 +-
.../cache/persistence/freelist/PagesList.java | 3 +-
.../cache/persistence/pagemem/PageMemoryEx.java | 17 +--
.../cache/persistence/pagemem/PageMemoryImpl.java | 45 ++++---
.../persistence/pagemem/PageReadWriteManager.java | 58 ++++++++
.../pagemem/PageReadWriteManagerImpl.java | 146 +++++++++++++++++++++
.../snapshot/IgniteSnapshotManager.java | 3 +-
.../cache/persistence/tree/BPlusTree.java | 16 ++-
.../cache/persistence/tree/io/PageIoResolver.java} | 23 ++--
.../cache/persistence/tree/util/InsertLast.java} | 14 +-
.../cache/persistence/tree/util/PageHandler.java | 24 ++--
.../processors/cache/tree/CacheDataRowStore.java | 19 +--
.../cache/WalModeChangeAdvancedSelfTest.java | 1 -
.../IgnitePdsRecoveryAfterFileCorruptionTest.java | 2 +-
.../persistence/IgnitePdsTaskCancelingTest.java | 4 +-
...CheckpointSimulationWithRealCpDisabledTest.java | 2 +-
.../pagemem/BPlusTreePageMemoryImplTest.java | 1 +
.../BPlusTreeReuseListPageMemoryImplTest.java | 1 +
.../IgnitePageMemReplaceDelayedWriteUnitTest.java | 2 +-
.../pagemem/IndexStoragePageMemoryImplTest.java | 1 +
.../persistence/pagemem/NoOpPageStoreManager.java | 17 +--
.../pagemem/PageMemoryImplNoLoadTest.java | 1 +
.../persistence/pagemem/PageMemoryImplTest.java | 11 +-
.../h2/DurableBackgroundCleanupIndexTreeTask.java | 4 +-
.../processors/query/h2/IgniteH2Indexing.java | 1 -
.../processors/query/h2/database/H2Tree.java | 74 ++++++++---
.../processors/query/h2/database/H2TreeIndex.java | 57 +++++---
.../inlinecolumn/AbstractInlineIndexColumn.java | 2 +-
.../inlinecolumn/ObjectHashInlineIndexColumn.java | 94 ++++++++++++-
.../h2/database/io/AbstractH2ExtrasInnerIO.java | 5 +
.../h2/database/io/AbstractH2ExtrasLeafIO.java | 8 +-
.../query/h2/database/io/AbstractH2InnerIO.java | 5 +
.../query/h2/database/io/AbstractH2LeafIO.java | 7 +-
.../query/h2/database/io/H2ExtrasLeafIO.java | 2 +-
.../processors/query/h2/database/io/H2LeafIO.java | 2 +-
.../query/h2/database/io/H2MvccExtrasLeafIO.java | 4 +-
.../query/h2/database/io/H2MvccLeafIO.java | 2 +-
.../query/h2/database/io/H2RowLinkIO.java | 5 +
.../processors/query/h2/opt/H2CacheRow.java | 14 ++
.../db/LongDestroyDurableBackgroundTaskTest.java | 7 +-
.../MultipleParallelCacheDeleteDeadlockTest.java | 7 +-
62 files changed, 914 insertions(+), 443 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/PageIdAllocator.java b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/PageIdAllocator.java
index d91d31d..7395695 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/PageIdAllocator.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/PageIdAllocator.java
@@ -19,6 +19,8 @@ package org.apache.ignite.internal.pagemem;
import org.apache.ignite.IgniteCheckedException;
+import static org.apache.ignite.internal.pagemem.PageIdUtils.pageId;
+
/**
* Allocates page ID's.
*/
@@ -41,6 +43,9 @@ public interface PageIdAllocator {
/** Special partition reserved for metastore space. */
public static final int METASTORE_PARTITION = 0x1;
+ /** Cache group meta page id. */
+ public static final long META_PAGE_ID = pageId(INDEX_PARTITION, FLAG_IDX, 0);
+
/**
* Allocates a page from the space for the given partition ID and the given flags.
*
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/IgnitePageStoreManager.java b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/IgnitePageStoreManager.java
index 8216b96..50af2a4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/IgnitePageStoreManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/IgnitePageStoreManager.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.pagemem.store;
import java.nio.ByteBuffer;
import java.util.Map;
+import java.util.function.LongConsumer;
import java.util.function.Predicate;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.configuration.CacheConfiguration;
@@ -27,13 +28,13 @@ import org.apache.ignite.internal.processors.cache.CacheGroupContext;
import org.apache.ignite.internal.processors.cache.CacheGroupDescriptor;
import org.apache.ignite.internal.processors.cache.GridCacheSharedManager;
import org.apache.ignite.internal.processors.cache.StoredCacheData;
+import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageReadWriteManager;
import org.apache.ignite.internal.processors.cluster.IgniteChangeGlobalStateSupport;
-import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric;
/**
*
*/
-public interface IgnitePageStoreManager extends GridCacheSharedManager, IgniteChangeGlobalStateSupport {
+public interface IgnitePageStoreManager extends GridCacheSharedManager, IgniteChangeGlobalStateSupport, PageReadWriteManager {
/**
* Invoked before starting checkpoint recover.
*/
@@ -53,7 +54,7 @@ public interface IgnitePageStoreManager extends GridCacheSharedManager, IgniteCh
* @param tracker Allocation tracker.
* @throws IgniteCheckedException If failed.
*/
- void initialize(int cacheId, int partitions, String workingDir, LongAdderMetric tracker)
+ void initialize(int cacheId, int partitions, String workingDir, LongConsumer tracker)
throws IgniteCheckedException;
/**
@@ -102,16 +103,6 @@ public interface IgnitePageStoreManager extends GridCacheSharedManager, IgniteCh
public void onPartitionDestroyed(int grpId, int partId, int tag) throws IgniteCheckedException;
/**
- * Reads a page for the given cache ID. Cache ID may be {@code 0} if the page is a meta page.
- *
- * @param grpId Cache group ID.
- * @param pageId PageID to read.
- * @param pageBuf Page buffer to write to.
- * @throws IgniteCheckedException If failed to read the page.
- */
- public void read(int grpId, long pageId, ByteBuffer pageBuf) throws IgniteCheckedException;
-
- /**
* Checks if partition store exists.
*
* @param grpId Cache group ID.
@@ -139,7 +130,7 @@ public interface IgnitePageStoreManager extends GridCacheSharedManager, IgniteCh
* @param pageBuf Page buffer to write.
* @throws IgniteCheckedException If failed to write page.
*/
- public void write(int grpId, long pageId, ByteBuffer pageBuf, int tag) throws IgniteCheckedException;
+ @Override public PageStore write(int grpId, long pageId, ByteBuffer pageBuf, int tag, boolean calculateCrc) throws IgniteCheckedException;
/**
* Gets page offset within the page store file.
@@ -176,7 +167,7 @@ public interface IgnitePageStoreManager extends GridCacheSharedManager, IgniteCh
* @return Allocated page ID.
* @throws IgniteCheckedException If IO exception occurred while allocating a page ID.
*/
- public long allocatePage(int grpId, int partId, byte flags) throws IgniteCheckedException;
+ @Override public long allocatePage(int grpId, int partId, byte flags) throws IgniteCheckedException;
/**
* Gets total number of allocated pages for the given space.
@@ -189,14 +180,6 @@ public interface IgnitePageStoreManager extends GridCacheSharedManager, IgniteCh
public int pages(int grpId, int partId) throws IgniteCheckedException;
/**
- * Gets meta page ID for specified cache.
- *
- * @param grpId Cache group ID.
- * @return Meta page ID.
- */
- public long metaPageId(int grpId);
-
- /**
* @return Saved cache configurations.
* @throws IgniteCheckedException If failed.
*/
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionUpdateCounterMvccImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/PageStoreCollection.java
similarity index 51%
copy from modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionUpdateCounterMvccImpl.java
copy to modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/PageStoreCollection.java
index cbb5896..8caf040 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionUpdateCounterMvccImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/PageStoreCollection.java
@@ -15,31 +15,27 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.cache;
+package org.apache.ignite.internal.pagemem.store;
+
+import java.util.Collection;
+import org.apache.ignite.IgniteCheckedException;
/**
- * Update counter implementation for MVCC mode.
+ * A collection that contains {@link PageStore} elements.
*/
-public class PartitionUpdateCounterMvccImpl extends PartitionUpdateCounterTrackingImpl {
+public interface PageStoreCollection {
/**
- * @param grp Group.
+ * @param grpId Cache group ID.
+ * @param partId Partition ID.
+ * @return Page store for the corresponding parameters.
+ * @throws IgniteCheckedException If cache or partition with the given ID was not created.
*/
- public PartitionUpdateCounterMvccImpl(CacheGroupContext grp) {
- super(grp);
- }
-
- /** {@inheritDoc} */
- @Override public long reserve(long delta) {
- return next(delta);
- }
-
- /** {@inheritDoc} */
- @Override public long reserved() {
- return get();
- }
+ public PageStore getStore(int grpId, int partId) throws IgniteCheckedException;
- /** {@inheritDoc} */
- @Override protected PartitionUpdateCounterTrackingImpl createInstance() {
- return new PartitionUpdateCounterMvccImpl(grp);
- }
+ /**
+ * @param grpId Cache group ID.
+ * @return Collection of related page stores.
+ * @throws IgniteCheckedException If failed.
+ */
+ public Collection<PageStore> getStores(int grpId) throws IgniteCheckedException;
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
index 3621d0b..1994457 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
@@ -200,7 +200,7 @@ public class CacheGroupContext {
* @param persistenceEnabled Persistence enabled flag.
* @param walEnabled Wal enabled flag.
*/
- CacheGroupContext(
+ public CacheGroupContext(
GridCacheSharedContext ctx,
int grpId,
UUID rcvdFrom,
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 d0fd64c..ceebc72 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
@@ -38,6 +38,7 @@ import org.apache.ignite.internal.processors.cache.persistence.freelist.SimpleDa
import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId;
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.PendingEntriesTree;
import org.apache.ignite.internal.processors.cache.tree.mvcc.data.MvccUpdateResult;
import org.apache.ignite.internal.processors.cache.tree.mvcc.search.MvccLinkAwareSearchRow;
@@ -606,6 +607,12 @@ public interface IgniteCacheOffheapManager {
*
*/
interface CacheDataStore {
+
+ /**
+ * @return Cache data tree object.
+ */
+ public CacheDataTree tree();
+
/**
* Initialize data store if it exists.
*
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 048179a..4743d89 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
@@ -32,6 +32,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
import javax.cache.Cache;
import javax.cache.processor.EntryProcessor;
import org.apache.ignite.IgniteCheckedException;
@@ -43,6 +44,8 @@ import org.apache.ignite.failure.FailureType;
import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.metric.IoStatisticsHolder;
import org.apache.ignite.internal.pagemem.FullPageId;
+import org.apache.ignite.internal.pagemem.PageMemory;
+import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
import org.apache.ignite.internal.pagemem.wal.record.delta.DataPageMvccMarkUpdatedRecord;
import org.apache.ignite.internal.pagemem.wal.record.delta.DataPageMvccUpdateNewTxStateHintRecord;
import org.apache.ignite.internal.pagemem.wal.record.delta.DataPageMvccUpdateTxStateHintRecord;
@@ -152,10 +155,6 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
/** The maximum number of entries that can be preloaded under checkpoint read lock. */
public static final int PRELOAD_SIZE_UNDER_CHECKPOINT_LOCK = 100;
- /** */
- private final boolean failNodeOnPartitionInconsistency = Boolean.getBoolean(
- IgniteSystemProperties.IGNITE_FAIL_NODE_ON_UNRECOVERABLE_PARTITION_INCONSISTENCY);
-
/** Batch size for cache removals during destroy. */
private static final int BATCH_SIZE = 1000;
@@ -184,9 +183,6 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
protected final GridSpinBusyLock busyLock = new GridSpinBusyLock();
/** */
- private int updateValSizeThreshold;
-
- /** */
protected GridStripedLock partStoreLock = new GridStripedLock(Runtime.getRuntime().availableProcessors());
/** {@inheritDoc} */
@@ -200,8 +196,6 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
this.grp = grp;
this.log = ctx.logger(getClass());
- updateValSizeThreshold = ctx.database().pageSize() / 2;
-
if (grp.affinityNode()) {
ctx.database().checkpointReadLock();
@@ -1300,7 +1294,7 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
lsnr
);
- return new CacheDataStoreImpl(p, rowStore, dataTree);
+ return new CacheDataStoreImpl(p, rowStore, dataTree, () -> pendingEntries, grp, busyLock, log);
}
/** {@inheritDoc} */
@@ -1447,7 +1441,7 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
/**
*
*/
- protected class CacheDataStoreImpl implements CacheDataStore {
+ public static class CacheDataStoreImpl implements CacheDataStore {
/** */
private final int partId;
@@ -1457,6 +1451,15 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
/** */
private final CacheDataTree dataTree;
+ /** */
+ private final Supplier<PendingEntriesTree> pendingEntries;
+
+ /** */
+ private final CacheGroupContext grp;
+
+ /** */
+ private final GridSpinBusyLock busyLock;
+
/** Update counter. */
protected final PartitionUpdateCounter pCntr;
@@ -1467,13 +1470,24 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
private final IntMap<AtomicLong> cacheSizes = new IntRWHashMap();
/** Mvcc remove handler. */
- private final PageHandler<MvccUpdateDataRow, Boolean> mvccUpdateMarker = new MvccMarkUpdatedHandler();
+ private final PageHandler<MvccUpdateDataRow, Boolean> mvccUpdateMarker;
/** Mvcc update tx state hint handler. */
- private final PageHandler<Void, Boolean> mvccUpdateTxStateHint = new MvccUpdateTxStateHintHandler();
+ private final PageHandler<Void, Boolean> mvccUpdateTxStateHint;
/** */
- private final PageHandler<MvccDataRow, Boolean> mvccApplyChanges = new MvccApplyChangesHandler();
+ private final PageHandler<MvccDataRow, Boolean> mvccApplyChanges;
+
+ /** */
+ private final IgniteLogger log;
+
+ /** */
+ private final Boolean failNodeOnPartitionInconsistency = Boolean.getBoolean(
+ IgniteSystemProperties.IGNITE_FAIL_NODE_ON_UNRECOVERABLE_PARTITION_INCONSISTENCY
+ );
+
+ /** */
+ private final int updateValSizeThreshold;
/**
* @param partId Partition number.
@@ -1483,18 +1497,37 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
public CacheDataStoreImpl(
int partId,
CacheDataRowStore rowStore,
- CacheDataTree dataTree
+ CacheDataTree dataTree,
+ Supplier<PendingEntriesTree> pendingEntries,
+ CacheGroupContext grp,
+ GridSpinBusyLock busyLock,
+ IgniteLogger log
) {
this.partId = partId;
this.rowStore = rowStore;
this.dataTree = dataTree;
+ this.pendingEntries = pendingEntries;
+ this.grp = grp;
+ this.busyLock = busyLock;
+ this.log = log;
PartitionUpdateCounter delegate = grp.mvccEnabled() ? new PartitionUpdateCounterMvccImpl(grp) :
!grp.persistenceEnabled() || grp.hasAtomicCaches() ? new PartitionUpdateCounterVolatileImpl(grp) :
new PartitionUpdateCounterTrackingImpl(grp);
- pCntr = ctx.logger(PartitionUpdateCounterDebugWrapper.class).isDebugEnabled() ?
+ pCntr = grp.shared().logger(PartitionUpdateCounterDebugWrapper.class).isDebugEnabled() ?
new PartitionUpdateCounterDebugWrapper(partId, delegate) : delegate;
+
+ updateValSizeThreshold = grp.shared().database().pageSize() / 2;
+
+ mvccUpdateMarker = new MvccMarkUpdatedHandler(grp);
+ mvccUpdateTxStateHint = new MvccUpdateTxStateHintHandler(grp);
+ mvccApplyChanges = new MvccApplyChangesHandler(grp);
+ }
+
+ /** {@inheritDoc} */
+ @Override public CacheDataTree tree() {
+ return dataTree;
}
/**
@@ -1640,7 +1673,7 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
grp.cacheOrGroupName() + ", partId=" + partId + ']', e);
if (failNodeOnPartitionInconsistency)
- ctx.kernalContext().failure().process(new FailureContext(FailureType.CRITICAL_ERROR, e));
+ grp.shared().kernalContext().failure().process(new FailureContext(FailureType.CRITICAL_ERROR, e));
}
}
@@ -2985,11 +3018,11 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
while (cur.next()) {
if (++rmv == BATCH_SIZE) {
- ctx.database().checkpointReadUnlock();
+ grp.shared().database().checkpointReadUnlock();
rmv = 0;
- ctx.database().checkpointReadLock();
+ grp.shared().database().checkpointReadLock();
}
CacheDataRow row = cur.get();
@@ -3019,9 +3052,9 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
throw new IgniteCheckedException("Fail destroy store", ex);
// Allow checkpointer to progress if a partition contains less than BATCH_SIZE keys.
- ctx.database().checkpointReadUnlock();
+ grp.shared().database().checkpointReadUnlock();
- ctx.database().checkpointReadLock();
+ grp.shared().database().checkpointReadLock();
}
/** {@inheritDoc} */
@@ -3053,7 +3086,7 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
/** {@inheritDoc} */
@Override public PendingEntriesTree pendingTree() {
- return pendingEntries;
+ return pendingEntries.get();
}
/** {@inheritDoc} */
@@ -3251,16 +3284,26 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
/**
* Mvcc remove handler.
*/
- private final class MvccMarkUpdatedHandler extends PageHandler<MvccUpdateDataRow, Boolean> {
+ private static final class MvccMarkUpdatedHandler extends PageHandler<MvccUpdateDataRow, Boolean> {
+ /** */
+ private final CacheGroupContext grp;
+
+ private MvccMarkUpdatedHandler(CacheGroupContext grp) {
+ this.grp = grp;
+ }
+
/** {@inheritDoc} */
@Override public Boolean run(int cacheId, long pageId, long page, long pageAddr, PageIO io, Boolean walPlc,
MvccUpdateDataRow updateDataRow, int itemId, IoStatisticsHolder statHolder) throws IgniteCheckedException {
assert grp.mvccEnabled();
+ PageMemory pageMem = grp.dataRegion().pageMemory();
+ IgniteWriteAheadLogManager wal = grp.shared().wal();
+
DataPageIO iox = (DataPageIO)io;
int off = iox.getPayloadOffset(pageAddr, itemId,
- grp.dataRegion().pageMemory().realPageSize(grp.groupId()), MVCC_INFO_SIZE);
+ pageMem.realPageSize(grp.groupId()), MVCC_INFO_SIZE);
long newCrd = iox.newMvccCoordinator(pageAddr, off);
long newCntr = iox.newMvccCounter(pageAddr, off);
@@ -3273,8 +3316,8 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
iox.updateNewVersion(pageAddr, off, updateDataRow.mvccCoordinatorVersion(), updateDataRow.mvccCounter(),
updateDataRow.mvccOperationCounter() | keyAbsentBeforeFlag, TxState.NA);
- if (isWalDeltaRecordNeeded(grp.dataRegion().pageMemory(), cacheId, pageId, page, ctx.wal(), walPlc))
- ctx.wal().log(new DataPageMvccMarkUpdatedRecord(cacheId, pageId, itemId,
+ if (isWalDeltaRecordNeeded(pageMem, cacheId, pageId, page, wal, walPlc))
+ wal.log(new DataPageMvccMarkUpdatedRecord(cacheId, pageId, itemId,
updateDataRow.mvccCoordinatorVersion(), updateDataRow.mvccCounter(), updateDataRow.mvccOperationCounter()));
return TRUE;
@@ -3284,15 +3327,26 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
/**
* Mvcc update operation counter hints handler.
*/
- private final class MvccUpdateTxStateHintHandler extends PageHandler<Void, Boolean> {
+ private static final class MvccUpdateTxStateHintHandler extends PageHandler<Void, Boolean> {
+ /** */
+ private final CacheGroupContext grp;
+
+ /** */
+ private MvccUpdateTxStateHintHandler(CacheGroupContext grp) {
+ this.grp = grp;
+ }
+
/** {@inheritDoc} */
@Override public Boolean run(int cacheId, long pageId, long page, long pageAddr, PageIO io,
Boolean walPlc, Void ignore, int itemId, IoStatisticsHolder statHolder) throws IgniteCheckedException {
DataPageIO iox = (DataPageIO)io;
+ PageMemory pageMem = grp.dataRegion().pageMemory();
+ IgniteWriteAheadLogManager wal = grp.shared().wal();
+
int off = iox.getPayloadOffset(pageAddr, itemId,
- grp.dataRegion().pageMemory().realPageSize(grp.groupId()), MVCC_INFO_SIZE);
+ pageMem.realPageSize(grp.groupId()), MVCC_INFO_SIZE);
long crd = iox.mvccCoordinator(pageAddr, off);
long cntr = iox.mvccCounter(pageAddr, off);
@@ -3305,8 +3359,8 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
if (state == TxState.COMMITTED || state == TxState.ABORTED) {
iox.rawMvccOperationCounter(pageAddr, off, opCntr | (state << MVCC_HINTS_BIT_OFF));
- if (isWalDeltaRecordNeeded(grp.dataRegion().pageMemory(), cacheId, pageId, page, ctx.wal(), walPlc))
- ctx.wal().log(new DataPageMvccUpdateTxStateHintRecord(cacheId, pageId, itemId, state));
+ if (isWalDeltaRecordNeeded(pageMem, cacheId, pageId, page, wal, walPlc))
+ wal.log(new DataPageMvccUpdateTxStateHintRecord(cacheId, pageId, itemId, state));
}
else
throw unexpectedStateException(grp, state, crd, cntr, opCntr);
@@ -3323,8 +3377,8 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
if (state == TxState.COMMITTED || state == TxState.ABORTED) {
iox.rawNewMvccOperationCounter(pageAddr, off, newOpCntr | (state << MVCC_HINTS_BIT_OFF));
- if (isWalDeltaRecordNeeded(grp.dataRegion().pageMemory(), cacheId, pageId, page, ctx.wal(), walPlc))
- ctx.wal().log(new DataPageMvccUpdateNewTxStateHintRecord(cacheId, pageId, itemId, state));
+ if (isWalDeltaRecordNeeded(pageMem, cacheId, pageId, page, wal, walPlc))
+ wal.log(new DataPageMvccUpdateNewTxStateHintRecord(cacheId, pageId, itemId, state));
}
// We do not throw an exception here because new version may be updated by active Tx at this moment.
@@ -3337,7 +3391,15 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
/**
* Applies changes to the row.
*/
- private final class MvccApplyChangesHandler extends PageHandler<MvccDataRow, Boolean> {
+ private static final class MvccApplyChangesHandler extends PageHandler<MvccDataRow, Boolean> {
+ /** */
+ private final CacheGroupContext grp;
+
+ /** */
+ private MvccApplyChangesHandler(CacheGroupContext grp) {
+ this.grp = grp;
+ }
+
/** {@inheritDoc} */
@Override public Boolean run(int cacheId, long pageId, long page, long pageAddr, PageIO io, Boolean walPlc,
MvccDataRow newRow, int itemId, IoStatisticsHolder statHolder) throws IgniteCheckedException {
@@ -3345,8 +3407,11 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
DataPageIO iox = (DataPageIO)io;
+ PageMemory pageMem = grp.dataRegion().pageMemory();
+ IgniteWriteAheadLogManager wal = grp.shared().wal();
+
int off = iox.getPayloadOffset(pageAddr, itemId,
- grp.dataRegion().pageMemory().realPageSize(grp.groupId()), MVCC_INFO_SIZE);
+ pageMem.realPageSize(grp.groupId()), MVCC_INFO_SIZE);
long crd = iox.mvccCoordinator(pageAddr, off);
long cntr = iox.mvccCounter(pageAddr, off);
@@ -3371,8 +3436,8 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
iox.rawMvccOperationCounter(pageAddr, off, opCntr | (newRow.mvccTxState() << MVCC_HINTS_BIT_OFF));
- if (isWalDeltaRecordNeeded(grp.dataRegion().pageMemory(), cacheId, pageId, page, ctx.wal(), walPlc))
- ctx.wal().log(new DataPageMvccUpdateTxStateHintRecord(cacheId, pageId, itemId, newRow.mvccTxState()));
+ if (isWalDeltaRecordNeeded(pageMem, cacheId, pageId, page, wal, walPlc))
+ wal.log(new DataPageMvccUpdateTxStateHintRecord(cacheId, pageId, itemId, newRow.mvccTxState()));
}
if (compare(newCrd,
@@ -3387,8 +3452,8 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
iox.updateNewVersion(pageAddr, off, newRow.newMvccCoordinatorVersion(), newRow.newMvccCounter(),
newRow.newMvccOperationCounter(), newRow.newMvccTxState());
- if (isWalDeltaRecordNeeded(grp.dataRegion().pageMemory(), cacheId, pageId, page, ctx.wal(), walPlc))
- ctx.wal().log(new DataPageMvccMarkUpdatedRecord(cacheId, pageId, itemId,
+ if (isWalDeltaRecordNeeded(pageMem, cacheId, pageId, page, wal, walPlc))
+ wal.log(new DataPageMvccMarkUpdatedRecord(cacheId, pageId, itemId,
newRow.newMvccCoordinatorVersion(), newRow.newMvccCounter(),
newRow.newMvccOperationCounter() | (newRow.newMvccTxState() << MVCC_HINTS_BIT_OFF)));
}
@@ -3397,8 +3462,8 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
iox.rawNewMvccOperationCounter(pageAddr, off, newOpCntr | (newRow.newMvccTxState() << MVCC_HINTS_BIT_OFF));
- if (isWalDeltaRecordNeeded(grp.dataRegion().pageMemory(), cacheId, pageId, page, ctx.wal(), walPlc))
- ctx.wal().log(new DataPageMvccUpdateNewTxStateHintRecord(cacheId, pageId, itemId, newRow.newMvccTxState()));
+ if (isWalDeltaRecordNeeded(pageMem, cacheId, pageId, page, wal, walPlc))
+ wal.log(new DataPageMvccUpdateNewTxStateHintRecord(cacheId, pageId, itemId, newRow.newMvccTxState()));
}
return TRUE;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionUpdateCounterMvccImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionUpdateCounterMvccImpl.java
index cbb5896..2e3066a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionUpdateCounterMvccImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionUpdateCounterMvccImpl.java
@@ -17,6 +17,8 @@
package org.apache.ignite.internal.processors.cache;
+import java.util.TreeMap;
+
/**
* Update counter implementation for MVCC mode.
*/
@@ -42,4 +44,17 @@ public class PartitionUpdateCounterMvccImpl extends PartitionUpdateCounterTracki
@Override protected PartitionUpdateCounterTrackingImpl createInstance() {
return new PartitionUpdateCounterMvccImpl(grp);
}
+
+ /** {@inheritDoc} */
+ @Override public PartitionUpdateCounter copy() {
+ PartitionUpdateCounterMvccImpl copy = new PartitionUpdateCounterMvccImpl(grp);
+
+ copy.cntr.set(cntr.get());
+ copy.first = first;
+ copy.queue = new TreeMap<>(queue);
+ copy.initCntr = initCntr;
+ copy.reserveCntr.set(reserveCntr.get());
+
+ return copy;
+ }
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionUpdateCounterTrackingImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionUpdateCounterTrackingImpl.java
index 7671396..0f9d72c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionUpdateCounterTrackingImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/PartitionUpdateCounterTrackingImpl.java
@@ -68,16 +68,16 @@ public class PartitionUpdateCounterTrackingImpl implements PartitionUpdateCounte
private static final byte VERSION = 1;
/** Queue of applied out of order counter updates. */
- private NavigableMap<Long, Item> queue = new TreeMap<>();
+ protected NavigableMap<Long, Item> queue = new TreeMap<>();
/** LWM. */
- private final AtomicLong cntr = new AtomicLong();
+ protected final AtomicLong cntr = new AtomicLong();
/** HWM. */
protected final AtomicLong reserveCntr = new AtomicLong();
/** */
- private boolean first = true;
+ protected boolean first = true;
/** */
protected final CacheGroupContext grp;
@@ -86,7 +86,7 @@ public class PartitionUpdateCounterTrackingImpl implements PartitionUpdateCounte
* Initial counter points to last sequential update after WAL recovery.
* @deprecated TODO FIXME https://issues.apache.org/jira/browse/IGNITE-11794
*/
- @Deprecated private volatile long initCntr;
+ @Deprecated protected volatile long initCntr;
/**
* @param grp Group.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java
index 472c681..9fbf83c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java
@@ -34,7 +34,6 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
@@ -405,7 +404,7 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
//noinspection ConstantConditions
ctx.cache().context().pageStore().initialize(TX_LOG_CACHE_ID, 0,
- TX_LOG_CACHE_NAME, mgr.dataRegion(TX_LOG_CACHE_NAME).memoryMetrics().totalAllocatedPages());
+ TX_LOG_CACHE_NAME, mgr.dataRegion(TX_LOG_CACHE_NAME).memoryMetrics().totalAllocatedPages()::add);
}
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/txlog/TxLog.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/txlog/TxLog.java
index 9b8c73b..0e98b8d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/txlog/TxLog.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/txlog/TxLog.java
@@ -115,7 +115,7 @@ public class TxLog implements CheckpointListener {
IgniteWriteAheadLogManager wal = ctx.cache().context().wal();
PageMemoryEx pageMemory = (PageMemoryEx)txLogDataRegion.pageMemory();
- long metaId = pageMemory.metaPageId(TX_LOG_CACHE_ID);
+ long metaId = PageMemory.META_PAGE_ID;
long metaPage = pageMemory.acquirePage(TX_LOG_CACHE_ID, metaId);
long treeRoot, reuseListRoot;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/DataStructure.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/DataStructure.java
index 300e98a..68fd48e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/DataStructure.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/DataStructure.java
@@ -29,6 +29,7 @@ import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
import org.apache.ignite.internal.pagemem.wal.record.delta.RecycleRecord;
import org.apache.ignite.internal.pagemem.wal.record.delta.RotatedIdPartRecord;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIoResolver;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseBag;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
import org.apache.ignite.internal.processors.cache.persistence.tree.util.PageHandler;
@@ -66,18 +67,24 @@ public abstract class DataStructure {
/** */
protected ReuseList reuseList;
+ /** */
+ protected final PageIoResolver pageIoRslvr;
+
/**
* @param cacheGrpId Cache group ID.
* @param grpName Cache group name.
* @param pageMem Page memory.
* @param wal Write ahead log manager.
+ * @param lockLsnr Page lock listener.
+ * @param pageIoRslvr Page IO resolver.
*/
public DataStructure(
int cacheGrpId,
String grpName,
PageMemory pageMem,
IgniteWriteAheadLogManager wal,
- PageLockListener lockLsnr
+ PageLockListener lockLsnr,
+ PageIoResolver pageIoRslvr
) {
assert pageMem != null;
@@ -86,6 +93,7 @@ public abstract class DataStructure {
this.pageMem = pageMem;
this.wal = wal;
this.lockLsnr = lockLsnr == null ? NOOP_LSNR : lockLsnr;
+ this.pageIoRslvr = pageIoRslvr;
}
/**
@@ -251,7 +259,7 @@ public abstract class DataStructure {
R lockFailed,
IoStatisticsHolder statHolder) throws IgniteCheckedException {
return PageHandler.writePage(pageMem, grpId, pageId, lockLsnr, h,
- null, null, null, null, intArg, lockFailed, statHolder);
+ null, null, null, null, intArg, lockFailed, statHolder, pageIoRslvr);
}
/**
@@ -272,7 +280,7 @@ public abstract class DataStructure {
R lockFailed,
IoStatisticsHolder statHolder) throws IgniteCheckedException {
return PageHandler.writePage(pageMem, grpId, pageId, lockLsnr, h,
- null, null, null, arg, intArg, lockFailed, statHolder);
+ null, null, null, arg, intArg, lockFailed, statHolder, pageIoRslvr);
}
/**
@@ -295,7 +303,7 @@ public abstract class DataStructure {
R lockFailed,
IoStatisticsHolder statHolder) throws IgniteCheckedException {
return PageHandler.writePage(pageMem, grpId, pageId, page, lockLsnr, h,
- null, null, null, arg, intArg, lockFailed, statHolder);
+ null, null, null, arg, intArg, lockFailed, statHolder, pageIoRslvr);
}
/**
@@ -318,7 +326,7 @@ public abstract class DataStructure {
R lockFailed,
IoStatisticsHolder statHolder) throws IgniteCheckedException {
return PageHandler.writePage(pageMem, grpId, pageId, lockLsnr, h,
- init, wal, null, arg, intArg, lockFailed, statHolder);
+ init, wal, null, arg, intArg, lockFailed, statHolder, pageIoRslvr);
}
/**
@@ -339,7 +347,7 @@ public abstract class DataStructure {
R lockFailed,
IoStatisticsHolder statHolder) throws IgniteCheckedException {
return PageHandler.readPage(pageMem, grpId, pageId, lockLsnr,
- h, arg, intArg, lockFailed, statHolder);
+ h, arg, intArg, lockFailed, statHolder, pageIoRslvr);
}
/**
@@ -362,7 +370,7 @@ public abstract class DataStructure {
R lockFailed,
IoStatisticsHolder statHolder) throws IgniteCheckedException {
return PageHandler.readPage(pageMem, grpId, pageId, page, lockLsnr, h,
- arg, intArg, lockFailed, statHolder);
+ arg, intArg, lockFailed, statHolder, pageIoRslvr);
}
/**
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 6a0ecb2..8f6f683 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
@@ -126,6 +126,7 @@ import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaS
import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetastorageLifecycleListener;
import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryEx;
import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryImpl;
+import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageReadWriteManager;
import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteCacheSnapshotManager;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
@@ -790,7 +791,7 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
walTail = CheckpointStatus.NULL_PTR.equals(status.endPtr) ? null : status.endPtr;
}
- cctx.wal().resumeLogging(walTail);
+ resumeWalLogging();
walTail = null;
@@ -947,10 +948,11 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
DataStorageConfiguration memCfg,
DataRegionConfiguration plcCfg,
DataRegionMetricsImpl memMetrics,
- final boolean trackable
+ final boolean trackable,
+ PageReadWriteManager pmPageMgr
) {
if (!plcCfg.isPersistenceEnabled())
- return super.createPageMemory(memProvider, memCfg, plcCfg, memMetrics, trackable);
+ return super.createPageMemory(memProvider, memCfg, plcCfg, memMetrics, trackable, pmPageMgr);
memMetrics.persistenceEnabled(true);
@@ -991,6 +993,7 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
chpBufSize
),
cctx,
+ pmPageMgr,
memCfg.getPageSize(),
(fullId, pageBuf, tag) -> {
memMetrics.onPageWritten();
@@ -999,7 +1002,7 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
snapshotMgr.beforePageWrite(fullId);
// Write page to disk.
- storeMgr.write(fullId.groupId(), fullId.pageId(), pageBuf, tag);
+ pmPageMgr.write(fullId.groupId(), fullId.pageId(), pageBuf, tag, true);
getCheckpointer().currentProgress().updateEvictedPages(1);
},
@@ -1320,6 +1323,7 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
/**
* Gets the checkpoint read lock. While this lock is held, checkpoint thread will not acquireSnapshotWorker memory
* state.
+ *
* @throws IgniteException If failed.
*/
@Override public void checkpointReadLock() {
@@ -1630,6 +1634,8 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
checkpointReadLock();
+ RestoreLogicalState logicalState;
+
try {
// Preform early regions startup before restoring state.
initAndStartRegions(kctx.config().getDataStorageConfiguration());
@@ -1652,7 +1658,7 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
CheckpointStatus status = readCheckpointStatus();
- RestoreLogicalState logicalState = applyLogicalUpdates(
+ logicalState = applyLogicalUpdates(
status,
groupsWithEnabledWal(),
logicalRecords(),
@@ -1666,10 +1672,6 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
}
startTimer.finishGlobalStage("Restore logical state");
-
- walTail = tailPointer(logicalState);
-
- cctx.wal().onDeActivate(kctx);
}
catch (IgniteCheckedException e) {
releaseFileLock();
@@ -1679,6 +1681,15 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
finally {
checkpointReadUnlock();
}
+
+ walTail = tailPointer(logicalState);
+
+ cctx.wal().onDeActivate(kctx);
+ }
+
+ /** */
+ public void resumeWalLogging() throws IgniteCheckedException {
+ cctx.wal().resumeLogging(walTail);
}
/**
@@ -3041,7 +3052,8 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
* @param log Logger.
* @throws IgniteCheckedException If failed.
*/
- private static void dumpPartitionsInfo(GridCacheSharedContext cctx, IgniteLogger log) throws IgniteCheckedException {
+ private static void dumpPartitionsInfo(GridCacheSharedContext cctx,
+ IgniteLogger log) throws IgniteCheckedException {
for (CacheGroupContext grp : cctx.cache().cacheGroups()) {
if (grp.isLocal() || !grp.persistenceEnabled())
continue;
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 4bfda53..cb7b6dc 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
@@ -68,6 +68,7 @@ import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheMvccEntryInfo;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.GridCacheTtlManager;
import org.apache.ignite.internal.processors.cache.IgniteCacheOffheapManagerImpl;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
@@ -91,6 +92,7 @@ import org.apache.ignite.internal.processors.cache.persistence.partstate.PagesAl
import org.apache.ignite.internal.processors.cache.persistence.partstate.PartitionAllocationMap;
import org.apache.ignite.internal.processors.cache.persistence.partstorage.PartitionMetaStorage;
import org.apache.ignite.internal.processors.cache.persistence.partstorage.PartitionMetaStorageImpl;
+import org.apache.ignite.internal.processors.cache.persistence.tree.BPlusTree;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageMetaIO;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageMetaIOV2;
@@ -110,6 +112,7 @@ import org.apache.ignite.internal.processors.cache.tree.mvcc.search.MvccLinkAwar
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.query.GridQueryRowCacheCleaner;
import org.apache.ignite.internal.util.GridLongList;
+import org.apache.ignite.internal.util.GridSpinBusyLock;
import org.apache.ignite.internal.util.lang.GridCursor;
import org.apache.ignite.internal.util.lang.IgniteInClosure2X;
import org.apache.ignite.internal.util.lang.IgnitePredicateX;
@@ -243,7 +246,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
boolean exists = ctx.pageStore() != null && ctx.pageStore().exists(grp.groupId(), p);
- return new GridCacheDataStore(p, exists);
+ return new GridCacheDataStore(grp, p, exists, busyLock, log);
}
/** {@inheritDoc} */
@@ -766,7 +769,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
* return null if counter page does not exist.
* @throws IgniteCheckedException If page memory operation failed.
*/
- @Nullable private static Map<Integer, Long> readSharedGroupCacheSizes(PageSupport pageMem, int grpId,
+ @Nullable public static Map<Integer, Long> readSharedGroupCacheSizes(PageSupport pageMem, int grpId,
long cntrsPageId) throws IgniteCheckedException {
if (cntrsPageId == 0L)
@@ -817,7 +820,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
* @return new counter page Id. Same as {@code cntrsPageId} or new value if cache size pages were initialized.
* @throws IgniteCheckedException if page memory operation failed.
*/
- private static long writeSharedGroupCacheSizes(PageMemory pageMem, int grpId,
+ public static long writeSharedGroupCacheSizes(PageMemory pageMem, int grpId,
long cntrsPageId, int partId, Map<Integer, Long> sizes) throws IgniteCheckedException {
byte[] data = PagePartitionCountersIO.VERSIONS.latest().serializeCacheSizes(sizes);
@@ -880,7 +883,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
int grpId = grp.groupId();
PageMemoryEx pageMem = (PageMemoryEx)grp.dataRegion().pageMemory();
- long metaPageId = pageMem.metaPageId(grpId);
+ long metaPageId = PageMemory.META_PAGE_ID;
long metaPage = pageMem.acquirePage(grpId, metaPageId);
try {
@@ -1048,7 +1051,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
IgniteWriteAheadLogManager wal = ctx.wal();
int grpId = grp.groupId();
- long metaId = pageMem.metaPageId(grpId);
+ long metaId = PageMemory.META_PAGE_ID;
long metaPage = pageMem.acquirePage(grpId, metaId);
try {
@@ -1191,7 +1194,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
int cleared = 0;
for (CacheDataStore store : cacheDataStores()) {
- cleared += ((GridCacheDataStore)store).purgeExpired(cctx, c, amount - cleared);
+ cleared += ((GridCacheDataStore)store).purgeExpired(cctx, c, unwindThrottlingTimeout, amount - cleared);
if (amount != -1 && cleared >= amount)
return true;
@@ -1240,7 +1243,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
for (CacheDataStore store : partDataStores.values()) {
assert store instanceof GridCacheDataStore;
- AbstractFreeList freeList = ((GridCacheDataStore)store).freeList;
+ AbstractFreeList freeList = ((GridCacheDataStore)store).getCacheStoreFreeList();
if (freeList == null)
continue;
@@ -1262,7 +1265,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
for (CacheDataStore store : partDataStores.values()) {
assert store instanceof GridCacheDataStore;
- AbstractFreeList freeList = ((GridCacheDataStore)store).freeList;
+ AbstractFreeList freeList = ((GridCacheDataStore)store).getCacheStoreFreeList();
if (freeList == null)
continue;
@@ -1315,7 +1318,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
PageMemoryEx pageMem = (PageMemoryEx)grp.dataRegion().pageMemory();
- long metaPageId = pageMem.metaPageId(grpId);
+ long metaPageId = PageIdAllocator.META_PAGE_ID;
long metaPage = pageMem.acquirePage(grpId, metaPageId);
try {
@@ -1352,6 +1355,11 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
}
}
+ /** */
+ public GridSpinBusyLock busyLock() {
+ return busyLock;
+ }
+
/**
*
*/
@@ -1763,22 +1771,22 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
/**
*
*/
- private static class Metas {
+ static class Metas {
/** */
@GridToStringInclude
- private final RootPage reuseListRoot;
+ public final RootPage reuseListRoot;
/** */
@GridToStringInclude
- private final RootPage treeRoot;
+ public final RootPage treeRoot;
/** */
@GridToStringInclude
- private final RootPage pendingTreeRoot;
+ public final RootPage pendingTreeRoot;
/** */
@GridToStringInclude
- private final RootPage partMetastoreReuseListRoot;
+ public final RootPage partMetastoreReuseListRoot;
/**
* @param treeRoot Metadata storage root.
@@ -1800,11 +1808,14 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
/**
*
*/
- public class GridCacheDataStore implements CacheDataStore {
+ public static class GridCacheDataStore implements CacheDataStore {
/** */
private final int partId;
/** */
+ private final CacheGroupContext grp;
+
+ /** */
private volatile AbstractFreeList<CacheDataRow> freeList;
/** */
@@ -1832,18 +1843,37 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
private final boolean exists;
/** */
+ private final GridSpinBusyLock busyLock;
+
+ /** */
+ private final IgniteLogger log;
+
+ /** */
private final AtomicBoolean init = new AtomicBoolean();
/** */
private final CountDownLatch latch = new CountDownLatch(1);
+ /** */
+ private CacheDataTree dataTree;
+
/**
* @param partId Partition.
* @param exists {@code True} if store exists.
*/
- private GridCacheDataStore(int partId, boolean exists) {
+ public GridCacheDataStore(CacheGroupContext grp, int partId, boolean exists,
+ GridSpinBusyLock busyLock,
+ IgniteLogger log) {
+ this.grp = grp;
this.partId = partId;
this.exists = exists;
+ this.busyLock = busyLock;
+ this.log = log;
+ }
+
+ /** */
+ public AbstractFreeList<CacheDataRow> getCacheStoreFreeList() {
+ return freeList;
}
/**
@@ -1864,7 +1894,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
* @return Name of data tree.
*/
private String dataTreeName() {
- return grp.cacheOrGroupName() + "-" + treeName(partId);
+ return grp.cacheOrGroupName() + "-" + BPlusTree.treeName("p-" + partId, "CacheData");
}
/**
@@ -1892,6 +1922,10 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
return null;
}
+ final GridCacheSharedContext ctx = grp.shared();
+
+ AtomicLong pageListCacheLimit = ((GridCacheDatabaseSharedManager) ctx.database()).pageListCacheLimitHolder(grp.dataRegion());
+
if (init.compareAndSet(false, true)) {
IgniteCacheDatabaseSharedManager dbMgr = ctx.database();
@@ -1904,7 +1938,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
PageIdUtils.partId(metas.treeRoot.pageId().pageId()) != partId ||
PageIdUtils.partId(metas.pendingTreeRoot.pageId().pageId()) != partId ||
PageIdUtils.partId(metas.partMetastoreReuseListRoot.pageId().pageId()) != partId
- ) {
+ ) {
throw new IgniteCheckedException("Invalid meta root allocated [" +
"cacheOrGroupName=" + grp.cacheOrGroupName() +
", partId=" + partId +
@@ -1929,7 +1963,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
) {
/** {@inheritDoc} */
@Override protected long allocatePageNoReuse() throws IgniteCheckedException {
- assert grp.shared().database().checkpointLockIsHeldByThread();
+ assert ctx.database().checkpointLockIsHeldByThread();
return pageMem.allocatePage(grpId, partId, PageIdAllocator.FLAG_DATA);
}
@@ -1954,7 +1988,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
) {
/** {@inheritDoc} */
@Override protected long allocatePageNoReuse() throws IgniteCheckedException {
- assert grp.shared().database().checkpointLockIsHeldByThread();
+ assert ctx.database().checkpointLockIsHeldByThread();
return pageMem.allocatePage(grpId, partId, PageIdAllocator.FLAG_DATA);
}
@@ -1966,7 +2000,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
RootPage treeRoot = metas.treeRoot;
- CacheDataTree dataTree = new CacheDataTree(
+ dataTree = new CacheDataTree(
grp,
dataTreeName,
freeList,
@@ -1977,7 +2011,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
) {
/** {@inheritDoc} */
@Override protected long allocatePageNoReuse() throws IgniteCheckedException {
- assert grp.shared().database().checkpointLockIsHeldByThread();
+ assert ctx.database().checkpointLockIsHeldByThread();
return pageMem.allocatePage(grpId, partId, PageIdAllocator.FLAG_DATA);
}
@@ -1998,15 +2032,24 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
) {
/** {@inheritDoc} */
@Override protected long allocatePageNoReuse() throws IgniteCheckedException {
- assert grp.shared().database().checkpointLockIsHeldByThread();
+ assert ctx.database().checkpointLockIsHeldByThread();
return pageMem.allocatePage(grpId, partId, PageIdAllocator.FLAG_DATA);
}
};
- PageMemoryEx pageMem = (PageMemoryEx)grp.dataRegion().pageMemory();
+ PageMemoryEx pageMem = (PageMemoryEx) grp.dataRegion().pageMemory();
+
+ int grpId = grp.groupId();
- delegate0 = new CacheDataStoreImpl(partId, rowStore, dataTree) {
+ delegate0 = new CacheDataStoreImpl(partId,
+ rowStore,
+ dataTree,
+ () -> pendingTree0,
+ grp,
+ busyLock,
+ log
+ ) {
/** {@inheritDoc} */
@Override public PendingEntriesTree pendingTree() {
return pendingTree0;
@@ -2019,20 +2062,20 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
if (pageStoreMgr == null)
return;
- final int pages = pageStoreMgr.pages(grp.groupId(), partId);
+ final int pages = pageStoreMgr.pages(grpId, partId);
- long pageId = pageMem.partitionMetaPageId(grp.groupId(), partId);
+ long pageId = pageMem.partitionMetaPageId(grpId, partId);
// For each page sequentially pin/unpin.
for (int pageNo = 0; pageNo < pages; pageId++, pageNo++) {
long pagePointer = -1;
try {
- pagePointer = pageMem.acquirePage(grp.groupId(), pageId);
+ pagePointer = pageMem.acquirePage(grpId, pageId);
}
finally {
if (pagePointer != -1)
- pageMem.releasePage(grp.groupId(), pageId, pagePointer);
+ pageMem.releasePage(grpId, pageId, pagePointer);
}
}
}
@@ -2043,7 +2086,6 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
if (!pendingTree0.isEmpty())
grp.caches().forEach(cctx -> cctx.ttl().hasPendingEntries(true));
- int grpId = grp.groupId();
long partMetaId = pageMem.partitionMetaPageId(grpId, partId);
long partMetaPage = pageMem.acquirePage(grpId, partMetaId);
@@ -2072,7 +2114,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
grp, partId, io.getEncryptedPageIndex(pageAddr), encrPageCnt);
}
- globalRemoveId().setIfGreater(io.getGlobalRemoveId(pageAddr));
+ grp.offheap().globalRemoveId().setIfGreater(io.getGlobalRemoveId(pageAddr));
}
}
finally {
@@ -2116,7 +2158,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
*/
private Metas getOrAllocatePartitionMetas() throws IgniteCheckedException {
PageMemoryEx pageMem = (PageMemoryEx)grp.dataRegion().pageMemory();
- IgniteWriteAheadLogManager wal = ctx.wal();
+ IgniteWriteAheadLogManager wal = grp.shared().wal();
int grpId = grp.groupId();
long partMetaId = pageMem.partitionMetaPageId(grpId, partId);
@@ -2177,7 +2219,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
assert pageVer == 1 || pageVer == 2;
if (log.isDebugEnabled())
- log.info("Upgrade partition meta page version: [part=" + partId +
+ log.debug("Upgrade partition meta page version: [part=" + partId +
", grpId=" + grpId + ", oldVer=" + pageVer +
", newVer=" + io.getVersion()
);
@@ -2251,6 +2293,11 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
}
/** {@inheritDoc} */
+ @Override public CacheDataTree tree() {
+ return dataTree;
+ }
+
+ /** {@inheritDoc} */
@Override public boolean init() {
try {
return init0(true) != null;
@@ -2495,7 +2542,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
long expireTime,
@Nullable CacheDataRow oldRow
) throws IgniteCheckedException {
- assert ctx.database().checkpointLockIsHeldByThread();
+ assert grp.shared().database().checkpointLockIsHeldByThread();
CacheDataStore delegate = init0(false);
@@ -2625,7 +2672,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
GridCacheVersion ver,
long expireTime,
@Nullable CacheDataRow oldRow) throws IgniteCheckedException {
- assert ctx.database().checkpointLockIsHeldByThread();
+ assert grp.shared().database().checkpointLockIsHeldByThread();
CacheDataStore delegate = init0(false);
@@ -2658,7 +2705,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
/** {@inheritDoc} */
@Override public void invoke(GridCacheContext cctx, KeyCacheObject key, OffheapInvokeClosure c)
throws IgniteCheckedException {
- assert ctx.database().checkpointLockIsHeldByThread();
+ assert grp.shared().database().checkpointLockIsHeldByThread();
CacheDataStore delegate = init0(false);
@@ -2668,7 +2715,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
/** {@inheritDoc} */
@Override public void remove(GridCacheContext cctx, KeyCacheObject key, int partId)
throws IgniteCheckedException {
- assert ctx.database().checkpointLockIsHeldByThread();
+ assert grp.shared().database().checkpointLockIsHeldByThread();
CacheDataStore delegate = init0(false);
@@ -2828,7 +2875,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
/** {@inheritDoc} */
@Override public void clear(int cacheId) throws IgniteCheckedException {
- assert ctx.database().checkpointLockIsHeldByThread();
+ assert grp.shared().database().checkpointLockIsHeldByThread();
CacheDataStore delegate0 = init0(true);
@@ -2879,6 +2926,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
public int purgeExpired(
GridCacheContext cctx,
IgniteInClosure2X<GridCacheEntryEx, GridCacheVersion> c,
+ long throttlingTimeout,
int amount
) throws IgniteCheckedException {
CacheDataStore delegate0 = init0(true);
@@ -2896,7 +2944,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
if (cleared < amount) {
lastThrottledCacheId = cctx.cacheId();
- nextStoreCleanTimeNanos = nowNanos + U.millisToNanos(unwindThrottlingTimeout);
+ nextStoreCleanTimeNanos = nowNanos + U.millisToNanos(throttlingTimeout);
}
return cleared;
@@ -3038,7 +3086,8 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
}
}
- @Override public PartitionMetaStorage partStorage() {
+ /** */
+ @Override public PartitionMetaStorage<SimpleDataRow> partStorage() {
return partStorage;
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java
index 42ad2a9..9eb02fb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java
@@ -75,6 +75,7 @@ import org.apache.ignite.internal.processors.cache.persistence.freelist.CacheFre
import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeList;
import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage;
import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetastorageLifecycleListener;
+import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageReadWriteManager;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
import org.apache.ignite.internal.processors.cache.persistence.tree.util.PageLockListener;
import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
@@ -418,6 +419,21 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
DataRegionConfiguration dataRegionCfg,
boolean trackable
) throws IgniteCheckedException {
+ return addDataRegion(dataStorageCfg, dataRegionCfg, trackable, cctx.pageStore());
+ }
+
+ /**
+ * @param dataStorageCfg Database config.
+ * @param dataRegionCfg Data region config.
+ * @param pmPageMgr Page manager.
+ * @throws IgniteCheckedException If failed to initialize swap path.
+ */
+ protected DataRegion addDataRegion(
+ DataStorageConfiguration dataStorageCfg,
+ DataRegionConfiguration dataRegionCfg,
+ boolean trackable,
+ PageReadWriteManager pmPageMgr
+ ) throws IgniteCheckedException {
String dataRegionName = dataRegionCfg.getName();
String dfltMemPlcName = dataStorageCfg.getDefaultDataRegionConfiguration().getName();
@@ -430,7 +446,7 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
cctx.kernalContext().metric(),
dataRegionMetricsProvider(dataRegionCfg));
- DataRegion region = initMemory(dataStorageCfg, dataRegionCfg, memMetrics, trackable);
+ DataRegion region = initMemory(dataStorageCfg, dataRegionCfg, memMetrics, trackable, pmPageMgr);
dataRegionMap.put(dataRegionName, region);
@@ -1205,6 +1221,7 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
* @param memCfg memory configuration with common parameters.
* @param plcCfg data region with PageMemory specific parameters.
* @param memMetrics {@link DataRegionMetrics} object to collect memory usage metrics.
+ * @param pmPageMgr Page manager.
* @return data region instance.
*
* @throws IgniteCheckedException If failed to initialize swap path.
@@ -1213,9 +1230,10 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
DataStorageConfiguration memCfg,
DataRegionConfiguration plcCfg,
DataRegionMetricsImpl memMetrics,
- boolean trackable
+ boolean trackable,
+ PageReadWriteManager pmPageMgr
) throws IgniteCheckedException {
- PageMemory pageMem = createPageMemory(createOrReuseMemoryProvider(plcCfg), memCfg, plcCfg, memMetrics, trackable);
+ PageMemory pageMem = createPageMemory(createOrReuseMemoryProvider(plcCfg), memCfg, plcCfg, memMetrics, trackable, pmPageMgr);
return new DataRegion(pageMem, plcCfg, memMetrics, createPageEvictionTracker(plcCfg, pageMem));
}
@@ -1314,6 +1332,7 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
* @param memCfg Memory configuartion.
* @param memPlcCfg data region configuration.
* @param memMetrics DataRegionMetrics to collect memory usage metrics.
+ * @param pmPageMgr Page manager.
* @return PageMemory instance.
*/
protected PageMemory createPageMemory(
@@ -1321,7 +1340,8 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
DataStorageConfiguration memCfg,
DataRegionConfiguration memPlcCfg,
DataRegionMetricsImpl memMetrics,
- boolean trackable
+ boolean trackable,
+ PageReadWriteManager pmPageMgr
) {
memMetrics.persistenceEnabled(false);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/CheckpointManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/CheckpointManager.java
index 2beac7b..7fe7002 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/CheckpointManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/CheckpointManager.java
@@ -170,7 +170,7 @@ public class CheckpointManager {
checkpointPagesWriterFactory = new CheckpointPagesWriterFactory(
logger, snapshotMgr,
- (pageMemEx, fullPage, buf, tag) -> pageStoreManager.writeInternal(fullPage.groupId(), fullPage.pageId(), buf, tag, true),
+ (pageMemEx, fullPage, buf, tag) -> pageStoreManager.write(fullPage.groupId(), fullPage.pageId(), buf, tag, true),
persStoreMetrics,
throttlingPolicy, threadBuf,
pageMemoryGroupResolver
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/LightweightCheckpointManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/LightweightCheckpointManager.java
index 7a329f7..9e7b3dc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/LightweightCheckpointManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/LightweightCheckpointManager.java
@@ -139,7 +139,7 @@ public class LightweightCheckpointManager {
logger,
snapshotMgr,
(pageMemEx, fullPage, buf, tag) ->
- pageStoreManager.writeInternal(fullPage.groupId(), fullPage.pageId(), buf, tag, true),
+ pageStoreManager.write(fullPage.groupId(), fullPage.pageId(), buf, tag, true),
persStoreMetrics,
throttlingPolicy,
threadBuf,
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FileIOFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FileIOFactory.java
index 29f7e6f..4deb047 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FileIOFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FileIOFactory.java
@@ -29,6 +29,7 @@ import static java.nio.file.StandardOpenOption.WRITE;
/**
* {@link FileIO} factory definition.
*/
+@FunctionalInterface
public interface FileIOFactory extends Serializable {
/**
* Creates I/O interface for file with default I/O mode.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStore.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStore.java
index adb49a3..6b63d31 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStore.java
@@ -30,6 +30,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.LongConsumer;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteSystemProperties;
@@ -42,7 +43,6 @@ import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSn
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
import org.apache.ignite.internal.processors.cache.persistence.wal.crc.FastCrc;
import org.apache.ignite.internal.processors.cache.persistence.wal.crc.IgniteDataIntegrityViolationException;
-import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteOutClosure;
@@ -101,7 +101,7 @@ public class FilePageStore implements PageStore {
private final AtomicLong allocated;
/** Region metrics updater. */
- private final LongAdderMetric allocatedTracker;
+ private final LongConsumer allocatedTracker;
/** List of listeners for current page store to handle. */
private final List<PageWriteListener> lsnrs = new CopyOnWriteArrayList<>();
@@ -130,7 +130,7 @@ public class FilePageStore implements PageStore {
IgniteOutClosure<Path> pathProvider,
FileIOFactory factory,
DataStorageConfiguration cfg,
- LongAdderMetric allocatedTracker
+ LongConsumer allocatedTracker
) {
this.type = type;
this.pathProvider = pathProvider;
@@ -357,7 +357,7 @@ public class FilePageStore implements PageStore {
}
}
finally {
- allocatedTracker.add(-1L * allocated.getAndSet(0) / pageSize);
+ allocatedTracker.accept(-1L * allocated.getAndSet(0) / pageSize);
inited = false;
@@ -406,7 +406,7 @@ public class FilePageStore implements PageStore {
throw new StorageException("Failed to truncate partition file [file=" + filePath.toAbsolutePath() + "]", e);
}
finally {
- allocatedTracker.add(-1L * allocated.getAndSet(0) / pageSize);
+ allocatedTracker.accept(-1L * allocated.getAndSet(0) / pageSize);
inited = false;
@@ -443,7 +443,7 @@ public class FilePageStore implements PageStore {
assert delta % pageSize == 0 : delta;
- allocatedTracker.add(delta / pageSize);
+ allocatedTracker.accept(delta / pageSize);
}
recover = false;
@@ -592,7 +592,7 @@ public class FilePageStore implements PageStore {
// Order is important, update of total allocated pages must be called after allocated update
// and setting inited to true, because it affects pages() returned value.
- allocatedTracker.add(pages());
+ allocatedTracker.accept(pages());
}
catch (IOException e) {
err = new StorageException(
@@ -839,7 +839,7 @@ public class FilePageStore implements PageStore {
off = allocated.get();
if (allocated.compareAndSet(off, off + pageSize)) {
- allocatedTracker.increment();
+ allocatedTracker.accept(1);
break;
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreFactory.java
index 6607ebe..ae92359 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreFactory.java
@@ -19,10 +19,10 @@ package org.apache.ignite.internal.processors.cache.persistence.file;
import java.io.File;
import java.nio.file.Path;
+import java.util.function.LongConsumer;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.pagemem.PageIdAllocator;
import org.apache.ignite.internal.pagemem.store.PageStore;
-import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric;
import org.apache.ignite.lang.IgniteOutClosure;
/**
@@ -38,7 +38,7 @@ public interface FilePageStoreFactory {
* @return page store
* @throws IgniteCheckedException if failed.
*/
- default PageStore createPageStore(byte type, File file, LongAdderMetric allocatedTracker)
+ default PageStore createPageStore(byte type, File file, LongConsumer allocatedTracker)
throws IgniteCheckedException {
return createPageStore(type, file::toPath, allocatedTracker);
}
@@ -52,6 +52,6 @@ public interface FilePageStoreFactory {
* @return page store
* @throws IgniteCheckedException if failed
*/
- PageStore createPageStore(byte type, IgniteOutClosure<Path> pathProvider, LongAdderMetric allocatedTracker)
+ PageStore createPageStore(byte type, IgniteOutClosure<Path> pathProvider, LongConsumer allocatedTracker)
throws IgniteCheckedException;
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java
index 024f52c..7152b1e 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java
@@ -49,9 +49,9 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;
+import java.util.function.LongConsumer;
import java.util.function.Predicate;
import java.util.stream.Collectors;
-
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
@@ -67,9 +67,9 @@ import org.apache.ignite.internal.pagemem.PageIdUtils;
import org.apache.ignite.internal.pagemem.PageMemory;
import org.apache.ignite.internal.pagemem.store.IgnitePageStoreManager;
import org.apache.ignite.internal.pagemem.store.PageStore;
+import org.apache.ignite.internal.pagemem.store.PageStoreCollection;
import org.apache.ignite.internal.processors.cache.CacheGroupContext;
import org.apache.ignite.internal.processors.cache.CacheGroupDescriptor;
-import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheSharedManagerAdapter;
import org.apache.ignite.internal.processors.cache.StoredCacheData;
import org.apache.ignite.internal.processors.cache.persistence.DataRegion;
@@ -78,8 +78,9 @@ import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabase
import org.apache.ignite.internal.processors.cache.persistence.StorageException;
import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolderSettings;
import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage;
+import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageReadWriteManager;
+import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageReadWriteManagerImpl;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteCacheSnapshotManager;
-import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric;
import org.apache.ignite.internal.util.GridStripedReadWriteLock;
import org.apache.ignite.internal.util.typedef.X;
@@ -104,7 +105,8 @@ import static org.apache.ignite.internal.pagemem.PageIdAllocator.MAX_PARTITION_I
/**
* File page store manager.
*/
-public class FilePageStoreManager extends GridCacheSharedManagerAdapter implements IgnitePageStoreManager {
+public class FilePageStoreManager extends GridCacheSharedManagerAdapter implements IgnitePageStoreManager,
+ PageStoreCollection {
/** File suffix. */
public static final String FILE_SUFFIX = ".bin";
@@ -118,7 +120,10 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
public static final String PART_FILE_PREFIX = "part-";
/** */
- public static final String INDEX_FILE_NAME = "index" + FILE_SUFFIX;
+ public static final String INDEX_FILE_PREFIX = "index";
+
+ /** */
+ public static final String INDEX_FILE_NAME = INDEX_FILE_PREFIX + FILE_SUFFIX;
/** */
public static final String PART_FILE_TEMPLATE = PART_FILE_PREFIX + "%d" + FILE_SUFFIX;
@@ -157,6 +162,9 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
/** Marshaller. */
private final Marshaller marshaller;
+ /** Page manager. */
+ private final PageReadWriteManager pmPageMgr;
+
/**
* Executor to disallow running code that modifies data in idxCacheStores concurrently with cleanup of file page
* store.
@@ -188,9 +196,6 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
private File storeWorkDir;
/** */
- private final long metaPageId = PageIdUtils.pageId(-1, PageMemory.FLAG_IDX, 0);
-
- /** */
private final Set<Integer> grpsWithoutIdx = Collections.newSetFromMap(new ConcurrentHashMap<Integer, Boolean>());
/** */
@@ -217,6 +222,8 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
pageStoreV1FileIoFactory = pageStoreFileIoFactory = dsCfg.getFileIOFactory();
marshaller = MarshallerUtils.jdkMarshaller(ctx.igniteInstanceName());
+
+ pmPageMgr = new PageReadWriteManagerImpl(ctx, this, FilePageStoreManager.class.getSimpleName());
}
/** {@inheritDoc} */
@@ -446,7 +453,7 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
}
/** {@inheritDoc} */
- @Override public void initialize(int cacheId, int partitions, String workingDir, LongAdderMetric tracker)
+ @Override public void initialize(int cacheId, int partitions, String workingDir, LongConsumer tracker)
throws IgniteCheckedException {
assert storeWorkDir != null;
@@ -493,7 +500,7 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
new File(storeWorkDir, META_STORAGE_NAME),
grpId,
PageIdAllocator.METASTORE_PARTITION + 1,
- dataRegion.memoryMetrics().totalAllocatedPages(),
+ dataRegion.memoryMetrics().totalAllocatedPages()::add,
false);
CacheStoreHolder old = idxCacheStores.put(grpId, holder);
@@ -596,34 +603,8 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
}
/** {@inheritDoc} */
- @Override public void read(int grpId, long pageId, ByteBuffer pageBuf) throws IgniteCheckedException {
- read(grpId, pageId, pageBuf, false);
- }
-
- /**
- * Will preserve crc in buffer if keepCrc is true.
- *
- * @param grpId Group ID.
- * @param pageId Page ID.
- * @param pageBuf Page buffer.
- * @param keepCrc Keep CRC flag.
- * @throws IgniteCheckedException If failed.
- */
- public void read(int grpId, long pageId, ByteBuffer pageBuf, boolean keepCrc) throws IgniteCheckedException {
- PageStore store = getStore(grpId, PageIdUtils.partId(pageId));
-
- try {
- store.read(pageId, pageBuf, keepCrc);
-
- assert keepCrc || PageIO.getCrc(pageBuf) == 0 : store.size() - store.pageOffset(pageId);
-
- cctx.kernalContext().compress().decompressPage(pageBuf, store.getPageSize());
- }
- catch (StorageException e) {
- cctx.kernalContext().failure().process(new FailureContext(FailureType.CRITICAL_ERROR, e));
-
- throw e;
- }
+ @Override public void read(int grpId, long pageId, ByteBuffer pageBuf, boolean keepCrc) throws IgniteCheckedException {
+ pmPageMgr.read(grpId, pageId, pageBuf, keepCrc);
}
/** {@inheritDoc} */
@@ -648,8 +629,8 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
}
/** {@inheritDoc} */
- @Override public void write(int grpId, long pageId, ByteBuffer pageBuf, int tag) throws IgniteCheckedException {
- writeInternal(grpId, pageId, pageBuf, tag, true);
+ @Override public PageStore write(int grpId, long pageId, ByteBuffer pageBuf, int tag, boolean calculateCrc) throws IgniteCheckedException {
+ return pmPageMgr.write(grpId, pageId, pageBuf, tag, calculateCrc);
}
/** {@inheritDoc} */
@@ -660,59 +641,6 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
}
/**
- * @param cacheId Cache ID to write.
- * @param pageId Page ID.
- * @param pageBuf Page buffer.
- * @param tag Partition tag (growing 1-based partition file version). Used to validate page is not outdated
- * @param calculateCrc if {@code False} crc calculation will be forcibly skipped.
- * @return PageStore to which the page has been written.
- * @throws IgniteCheckedException If IO error occurred.
- */
- public PageStore writeInternal(int cacheId, long pageId, ByteBuffer pageBuf, int tag, boolean calculateCrc)
- throws IgniteCheckedException {
- int partId = PageIdUtils.partId(pageId);
-
- PageStore store = getStore(cacheId, partId);
-
- try {
- int pageSize = store.getPageSize();
- int compressedPageSize = pageSize;
-
- GridCacheContext cctx0 = cctx.cacheContext(cacheId);
-
- if (cctx0 != null) {
- assert pageBuf.position() == 0 && pageBuf.limit() == pageSize : pageBuf;
-
- ByteBuffer compressedPageBuf = cctx0.compress().compressPage(pageBuf, store);
-
- if (compressedPageBuf != pageBuf) {
- compressedPageSize = PageIO.getCompressedSize(compressedPageBuf);
-
- if (!calculateCrc) {
- calculateCrc = true;
- PageIO.setCrc(compressedPageBuf, 0); // It will be recalculated over compressed data further.
- }
-
- PageIO.setCrc(pageBuf, 0); // It is expected to be reset to 0 after each write.
- pageBuf = compressedPageBuf;
- }
- }
-
- store.write(pageId, pageBuf, tag, calculateCrc);
-
- if (pageSize > compressedPageSize)
- store.punchHole(pageId, compressedPageSize); // TODO maybe add async punch mode?
- }
- catch (StorageException e) {
- cctx.kernalContext().failure().process(new FailureContext(FailureType.CRITICAL_ERROR, e));
-
- throw e;
- }
-
- return store;
- }
-
- /**
*
*/
public Path getPath(boolean isSharedGroup, String cacheOrGroupName, int partId) {
@@ -741,7 +669,7 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
cacheWorkDir,
grpDesc.groupId(),
grpDesc.config().getAffinity().partitions(),
- allocatedTracker,
+ allocatedTracker::add,
ccfg.isEncryptionEnabled()
);
}
@@ -799,7 +727,7 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
private CacheStoreHolder initDir(File cacheWorkDir,
int grpId,
int partitions,
- LongAdderMetric allocatedTracker,
+ LongConsumer allocatedTracker,
boolean encrypted) throws IgniteCheckedException {
try {
boolean dirExisted = checkAndInitCacheWorkDir(cacheWorkDir);
@@ -829,7 +757,7 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
allocatedTracker);
partStores[partId] = partStore;
- }
+ }
return new CacheStoreHolder(idxStore, partStores);
}
@@ -983,25 +911,7 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
/** {@inheritDoc} */
@Override public long allocatePage(int grpId, int partId, byte flags) throws IgniteCheckedException {
- assert partId <= MAX_PARTITION_ID || partId == INDEX_PARTITION;
-
- PageStore store = getStore(grpId, partId);
-
- try {
- long pageIdx = store.allocatePage();
-
- return PageIdUtils.pageId(partId, flags, (int)pageIdx);
- }
- catch (StorageException e) {
- cctx.kernalContext().failure().process(new FailureContext(FailureType.CRITICAL_ERROR, e));
-
- throw e;
- }
- }
-
- /** {@inheritDoc} */
- @Override public long metaPageId(final int grpId) {
- return metaPageId;
+ return pmPageMgr.allocatePage(grpId, partId, flags);
}
/** {@inheritDoc} */
@@ -1341,7 +1251,7 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
* @return Collection of related page stores.
* @throws IgniteCheckedException If failed.
*/
- public Collection<PageStore> getStores(int grpId) throws IgniteCheckedException {
+ @Override public Collection<PageStore> getStores(int grpId) throws IgniteCheckedException {
return getHolder(grpId);
}
@@ -1353,7 +1263,7 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
*
* Note: visible for testing.
*/
- public PageStore getStore(int grpId, int partId) throws IgniteCheckedException {
+ @Override public PageStore getStore(int grpId, int partId) throws IgniteCheckedException {
CacheStoreHolder holder = getHolder(grpId);
if (holder == null)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreV2.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreV2.java
index 8109dbb..84609c9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreV2.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreV2.java
@@ -17,8 +17,8 @@
package org.apache.ignite.internal.processors.cache.persistence.file;
import java.nio.file.Path;
+import java.util.function.LongConsumer;
import org.apache.ignite.configuration.DataStorageConfiguration;
-import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric;
import org.apache.ignite.lang.IgniteOutClosure;
/**
@@ -45,7 +45,7 @@ public class FilePageStoreV2 extends FilePageStore {
IgniteOutClosure<Path> pathProvider,
FileIOFactory factory,
DataStorageConfiguration cfg,
- LongAdderMetric allocatedTracker) {
+ LongConsumer allocatedTracker) {
super(type, pathProvider, factory, cfg, allocatedTracker);
hdrSize = cfg.getPageSize();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FileVersionCheckingFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FileVersionCheckingFactory.java
index e547122..e650242 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FileVersionCheckingFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FileVersionCheckingFactory.java
@@ -22,10 +22,10 @@ import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.function.LongConsumer;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.internal.pagemem.store.PageStore;
-import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric;
import org.apache.ignite.lang.IgniteOutClosure;
/**
@@ -69,7 +69,7 @@ public class FileVersionCheckingFactory implements FilePageStoreFactory {
@Override public PageStore createPageStore(
byte type,
IgniteOutClosure<Path> pathProvider,
- LongAdderMetric allocatedTracker) throws IgniteCheckedException {
+ LongConsumer allocatedTracker) throws IgniteCheckedException {
Path filePath = pathProvider.apply();
if (!Files.exists(filePath))
@@ -124,7 +124,7 @@ public class FileVersionCheckingFactory implements FilePageStoreFactory {
byte type,
IgniteOutClosure<Path> pathProvider,
int ver,
- LongAdderMetric allocatedTracker) {
+ LongConsumer allocatedTracker) {
switch (ver) {
case FilePageStore.VERSION:
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/PagesList.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/PagesList.java
index e944b1e..530690e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/PagesList.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/PagesList.java
@@ -67,6 +67,7 @@ import static org.apache.ignite.internal.pagemem.PageIdAllocator.FLAG_DATA;
import static org.apache.ignite.internal.pagemem.PageIdAllocator.FLAG_IDX;
import static org.apache.ignite.internal.pagemem.PageIdUtils.MAX_ITEMID_NUM;
import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO.getPageId;
+import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIoResolver.DEFAULT_PAGE_IO_RESOLVER;
/**
* Striped doubly-linked list of page IDs optionally organized in buckets.
@@ -211,7 +212,7 @@ public abstract class PagesList extends DataStructure {
PageLockListener lockLsnr,
GridKernalContext ctx
) {
- super(cacheId, null, pageMem, wal, lockLsnr);
+ super(cacheId, null, pageMem, wal, lockLsnr, DEFAULT_PAGE_IO_RESOLVER);
this.name = name;
this.buckets = buckets;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryEx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryEx.java
index 8bba4e1..4ddcb94 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryEx.java
@@ -66,23 +66,13 @@ public interface PageMemoryEx extends PageMemory {
boolean dirtyFlag, boolean restore);
/**
- * Gets or allocates metadata page for specified grpId.
- *
- * @param grpId Group ID.
- * @return Meta page for grpId.
- * @throws IgniteCheckedException If failed.
- */
- public long metaPageId(int grpId) throws IgniteCheckedException;
-
- /**
- * Gets or allocates partition metadata page for specified grpId and partId.
+ * Gets partition metadata page for specified grpId and partId.
*
* @param grpId Group ID.
* @param partId Partition ID.
* @return Meta page for grpId and partId.
- * @throws IgniteCheckedException If failed.
*/
- public long partitionMetaPageId(int grpId, int partId) throws IgniteCheckedException;
+ public long partitionMetaPageId(int grpId, int partId);
/**
* @see #acquirePage(int, long)
@@ -154,6 +144,9 @@ public interface PageMemoryEx extends PageMemory {
CheckpointMetricsTracker tracker
) throws IgniteCheckedException;
+ /** */
+ public PageReadWriteManager pageManager();
+
/**
* Marks partition as invalid / outdated.
*
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImpl.java
index f349cad..4872c6d 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImpl.java
@@ -57,7 +57,6 @@ import org.apache.ignite.internal.pagemem.FullPageId;
import org.apache.ignite.internal.pagemem.PageIdAllocator;
import org.apache.ignite.internal.pagemem.PageIdUtils;
import org.apache.ignite.internal.pagemem.PageUtils;
-import org.apache.ignite.internal.pagemem.store.IgnitePageStoreManager;
import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
import org.apache.ignite.internal.pagemem.wal.WALIterator;
import org.apache.ignite.internal.pagemem.wal.record.CheckpointRecord;
@@ -196,8 +195,8 @@ public class PageMemoryImpl implements PageMemoryEx {
/** */
private final ExecutorService asyncRunner;
- /** Page store manager. */
- private IgnitePageStoreManager storeMgr;
+ /** Page manager. */
+ private final PageReadWriteManager pmPageMgr;
/** */
private IgniteWriteAheadLogManager walMgr;
@@ -276,6 +275,7 @@ public class PageMemoryImpl implements PageMemoryEx {
* @param directMemoryProvider Memory allocator to use.
* @param sizes segments sizes, last is checkpoint pool size.
* @param ctx Cache shared context.
+ * @param pmPageMgr Page store manager.
* @param pageSize Page size.
* @param flushDirtyPage write callback invoked when a dirty page is removed for replacement.
* @param changeTracker Callback invoked to track changes in pages.
@@ -288,6 +288,7 @@ public class PageMemoryImpl implements PageMemoryEx {
DirectMemoryProvider directMemoryProvider,
long[] sizes,
GridCacheSharedContext<?, ?> ctx,
+ PageReadWriteManager pmPageMgr,
int pageSize,
PageStoreWriter flushDirtyPage,
@Nullable GridInClosure3X<Long, FullPageId, PageMemoryEx> changeTracker,
@@ -315,12 +316,12 @@ public class PageMemoryImpl implements PageMemoryEx {
this.throttlingPlc = throttlingPlc != null ? throttlingPlc : ThrottlingPolicy.CHECKPOINT_BUFFER_ONLY;
this.cpProgressProvider = cpProgressProvider;
- storeMgr = ctx.pageStore();
+ this.pmPageMgr = pmPageMgr;
walMgr = ctx.wal();
encMgr = ctx.kernalContext().encryption();
encryptionDisabled = ctx.gridConfig().getEncryptionSpi() instanceof NoopEncryptionSpi;
- assert storeMgr != null;
+ assert pmPageMgr != null;
assert walMgr != null;
assert encMgr != null;
@@ -523,7 +524,7 @@ public class PageMemoryImpl implements PageMemoryEx {
if (isThrottlingEnabled())
writeThrottle.onMarkDirty(false);
- long pageId = storeMgr.allocatePage(grpId, partId, flags);
+ long pageId = pmPageMgr.allocatePage(grpId, partId, flags);
assert PageIdUtils.pageIndex(pageId) > 0; //it's crucial for tracking pages (zero page is super one)
@@ -673,13 +674,6 @@ public class PageMemoryImpl implements PageMemoryEx {
}
/** {@inheritDoc} */
- @Override public long metaPageId(int grpId) {
- assert started;
-
- return storeMgr.metaPageId(grpId);
- }
-
- /** {@inheritDoc} */
@Override public long partitionMetaPageId(int grpId, int partId) {
assert started;
@@ -881,7 +875,7 @@ public class PageMemoryImpl implements PageMemoryEx {
long actualPageId = 0;
try {
- storeMgr.read(grpId, pageId, buf);
+ pmPageMgr.read(grpId, pageId, buf, false);
statHolder.trackPhysicalAndLogicalRead(pageAddr);
@@ -1188,6 +1182,11 @@ public class PageMemoryImpl implements PageMemoryEx {
}
/** {@inheritDoc} */
+ @Override public PageReadWriteManager pageManager() {
+ return pmPageMgr;
+ }
+
+ /** {@inheritDoc} */
@Override public void checkpointWritePage(
FullPageId fullId,
ByteBuffer buf,
@@ -2148,7 +2147,7 @@ public class PageMemoryImpl implements PageMemoryEx {
assert writeLock().isHeldByCurrentThread();
// Do not evict cache meta pages.
- if (fullPageId.pageId() == storeMgr.metaPageId(fullPageId.groupId()))
+ if (fullPageId.pageId() == META_PAGE_ID)
return false;
if (PageHeader.isAcquired(absPtr))
@@ -2161,7 +2160,7 @@ public class PageMemoryImpl implements PageMemoryEx {
// Can evict a dirty page only if should be written by a checkpoint.
// These pages does not have tmp buffer.
if (checkpointPages != null && checkpointPages.allowToSave(fullPageId)) {
- assert storeMgr != null;
+ assert pmPageMgr != null;
memMetrics.updatePageReplaceRate(U.currentTimeMillis() - PageHeader.readTimestamp(absPtr));
@@ -2334,7 +2333,7 @@ public class PageMemoryImpl implements PageMemoryEx {
CheckpointPages checkpointPages = this.checkpointPages;
if (relRmvAddr == rndAddr || pinned || skip ||
- fullId.pageId() == storeMgr.metaPageId(fullId.groupId()) ||
+ fullId.pageId() == META_PAGE_ID ||
(dirty && (checkpointPages == null || !checkpointPages.contains(fullId)))
) {
i--;
@@ -2487,10 +2486,14 @@ public class PageMemoryImpl implements PageMemoryEx {
", pinnedInSegment=" + pinnedCnt +
", failedToPrepare=" + failToPrepare +
']' + U.nl() + "Out of memory in data region [" +
- "name=" + dataRegionCfg.getName() +
- ", initSize=" + U.readableSize(dataRegionCfg.getInitialSize(), false) +
- ", maxSize=" + U.readableSize(dataRegionCfg.getMaxSize(), false) +
- ", persistenceEnabled=" + dataRegionCfg.isPersistenceEnabled() + "] Try the following:" + U.nl() +
+ (dataRegionCfg == null ? "NULL" : (
+ "name=" + dataRegionCfg.getName() +
+ ", initSize=" + U.readableSize(dataRegionCfg.getInitialSize(), false) +
+ ", maxSize=" + U.readableSize(dataRegionCfg.getMaxSize(), false) +
+ ", persistenceEnabled=" + dataRegionCfg.isPersistenceEnabled()
+ )) +
+ "]" +
+ " Try the following:" + U.nl() +
" ^-- Increase maximum off-heap memory size (DataRegionConfiguration.maxSize)" + U.nl() +
" ^-- Enable Ignite persistence (DataRegionConfiguration.persistenceEnabled)" + U.nl() +
" ^-- Enable eviction or expiration policies"
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageReadWriteManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageReadWriteManager.java
new file mode 100644
index 0000000..531da8b
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageReadWriteManager.java
@@ -0,0 +1,58 @@
+/*
+ * 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.pagemem;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.pagemem.PageMemory;
+import org.apache.ignite.internal.pagemem.store.PageStore;
+
+/** */
+public interface PageReadWriteManager {
+ /**
+ * Reads a page for the given cache ID. Cache ID may be {@code 0} if the page is a meta page.
+ *
+ * @param grpId Cache group ID.
+ * @param pageId PageID to read.
+ * @param pageBuf Page buffer to write to.
+ * @param keepCrc Keep CRC flag.
+ * @throws IgniteCheckedException If failed to read the page.
+ */
+ public void read(int grpId, long pageId, ByteBuffer pageBuf, boolean keepCrc) throws IgniteCheckedException;
+
+ /**
+ * Writes the page for the given cache ID. Cache ID may be {@code 0} if the page is a meta page.
+ *
+ * @param grpId Cache group ID.
+ * @param pageId Page ID.
+ * @param pageBuf Page buffer to write.
+ * @throws IgniteCheckedException If failed to write page.
+ */
+ public PageStore write(int grpId, long pageId, ByteBuffer pageBuf, int tag, boolean calculateCrc) throws IgniteCheckedException;
+
+ /**
+ * Allocates a page for the given page space.
+ *
+ * @param grpId Cache group ID.
+ * @param partId Partition ID. Used only if {@code flags} is equal to {@link PageMemory#FLAG_DATA}.
+ * @param flags Page allocation flags.
+ * @return Allocated page ID.
+ * @throws IgniteCheckedException If IO exception occurred while allocating a page ID.
+ */
+ public long allocatePage(int grpId, int partId, byte flags) throws IgniteCheckedException;
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageReadWriteManagerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageReadWriteManagerImpl.java
new file mode 100644
index 0000000..5fbd7c1
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageReadWriteManagerImpl.java
@@ -0,0 +1,146 @@
+/*
+ * 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.pagemem;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.failure.FailureContext;
+import org.apache.ignite.failure.FailureType;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.pagemem.PageIdUtils;
+import org.apache.ignite.internal.pagemem.store.PageStore;
+import org.apache.ignite.internal.pagemem.store.PageStoreCollection;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.persistence.StorageException;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
+import org.apache.ignite.internal.util.tostring.GridToStringExclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+import static org.apache.ignite.internal.pagemem.PageIdAllocator.INDEX_PARTITION;
+import static org.apache.ignite.internal.pagemem.PageIdAllocator.MAX_PARTITION_ID;
+
+/** */
+public class PageReadWriteManagerImpl implements PageReadWriteManager {
+ /** */
+ private final GridKernalContext ctx;
+
+ /** */
+ @GridToStringExclude
+ protected final PageStoreCollection pageStores;
+
+ /** */
+ @SuppressWarnings("unused")
+ private final String name;
+
+ /**
+ * @param ctx Kernal context.
+ * @param pageStores Page stores.
+ */
+ public PageReadWriteManagerImpl(
+ GridKernalContext ctx,
+ PageStoreCollection pageStores,
+ String name
+ ) {
+ this.ctx = ctx;
+ this.pageStores = pageStores;
+ this.name = name;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void read(int grpId, long pageId, ByteBuffer pageBuf, boolean keepCrc) throws IgniteCheckedException {
+ PageStore store = pageStores.getStore(grpId, PageIdUtils.partId(pageId));
+
+ try {
+ store.read(pageId, pageBuf, keepCrc);
+
+ ctx.compress().decompressPage(pageBuf, store.getPageSize());
+ }
+ catch (StorageException e) {
+ ctx.failure().process(new FailureContext(FailureType.CRITICAL_ERROR, e));
+
+ throw e;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public PageStore write(int grpId, long pageId, ByteBuffer pageBuf, int tag, boolean calculateCrc) throws IgniteCheckedException {
+ int partId = PageIdUtils.partId(pageId);
+
+ PageStore store = pageStores.getStore(grpId, partId);
+
+ try {
+ int pageSize = store.getPageSize();
+ int compressedPageSize = pageSize;
+
+ GridCacheContext<?, ?> cctx0 = ctx.cache().context().cacheContext(grpId);
+
+ if (cctx0 != null) {
+ assert pageBuf.position() == 0 && pageBuf.limit() == pageSize : pageBuf;
+
+ ByteBuffer compressedPageBuf = cctx0.compress().compressPage(pageBuf, store);
+
+ if (compressedPageBuf != pageBuf) {
+ compressedPageSize = PageIO.getCompressedSize(compressedPageBuf);
+
+ if (!calculateCrc) {
+ calculateCrc = true;
+ PageIO.setCrc(compressedPageBuf, 0); // It will be recalculated over compressed data further.
+ }
+
+ PageIO.setCrc(pageBuf, 0); // It is expected to be reset to 0 after each write.
+ pageBuf = compressedPageBuf;
+ }
+ }
+
+ store.write(pageId, pageBuf, tag, calculateCrc);
+
+ if (pageSize > compressedPageSize)
+ store.punchHole(pageId, compressedPageSize); // TODO maybe add async punch mode?
+ }
+ catch (StorageException e) {
+ ctx.failure().process(new FailureContext(FailureType.CRITICAL_ERROR, e));
+
+ throw e;
+ }
+
+ return store;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long allocatePage(int grpId, int partId, byte flags) throws IgniteCheckedException {
+ assert partId <= MAX_PARTITION_ID || partId == INDEX_PARTITION;
+
+ PageStore store = pageStores.getStore(grpId, partId);
+
+ try {
+ long pageIdx = store.allocatePage();
+
+ return PageIdUtils.pageId(partId, flags, (int)pageIdx);
+ }
+ catch (StorageException e) {
+ ctx.failure().process(new FailureContext(FailureType.CRITICAL_ERROR, e));
+
+ throw e;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(PageReadWriteManagerImpl.class, this);
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
index 03e7e1c..681485d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
@@ -87,7 +87,6 @@ import org.apache.ignite.internal.processors.cache.persistence.wal.crc.FastCrc;
import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState;
import org.apache.ignite.internal.processors.marshaller.MappedName;
import org.apache.ignite.internal.processors.metric.MetricRegistry;
-import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric;
import org.apache.ignite.internal.processors.task.GridInternal;
import org.apache.ignite.internal.util.GridBusyLock;
import org.apache.ignite.internal.util.distributed.DistributedProcess;
@@ -1209,7 +1208,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
.apply(pair.getGroupId(), false)
.createPageStore(getFlagByPartId(pair.getPartitionId()),
snpPart::toPath,
- new LongAdderMetric("NO_OP", null))
+ val -> {})
) {
ByteBuffer pageBuf = ByteBuffer.allocate(pageSize)
.order(ByteOrder.nativeOrder());
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/BPlusTree.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/BPlusTree.java
index a46b3a8..5e67b7c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/BPlusTree.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/BPlusTree.java
@@ -59,9 +59,11 @@ import org.apache.ignite.internal.processors.cache.persistence.tree.io.BPlusLeaf
import org.apache.ignite.internal.processors.cache.persistence.tree.io.BPlusMetaIO;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.IOVersions;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIoResolver;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.LongListReuseBag;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseBag;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
+import org.apache.ignite.internal.processors.cache.persistence.tree.util.InsertLast;
import org.apache.ignite.internal.processors.cache.persistence.tree.util.PageHandler;
import org.apache.ignite.internal.processors.cache.persistence.tree.util.PageHandlerWrapper;
import org.apache.ignite.internal.processors.cache.persistence.tree.util.PageLockListener;
@@ -90,6 +92,7 @@ import static org.apache.ignite.internal.processors.cache.persistence.tree.BPlus
import static org.apache.ignite.internal.processors.cache.persistence.tree.BPlusTree.Result.NOT_FOUND;
import static org.apache.ignite.internal.processors.cache.persistence.tree.BPlusTree.Result.RETRY;
import static org.apache.ignite.internal.processors.cache.persistence.tree.BPlusTree.Result.RETRY_ROOT;
+import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIoResolver.DEFAULT_PAGE_IO_RESOLVER;
/**
* Abstract B+Tree.
@@ -778,7 +781,8 @@ public abstract class BPlusTree<L, T extends L> extends DataStructure implements
metaPageId,
reuseList,
failureProcessor,
- lockLsnr
+ lockLsnr,
+ DEFAULT_PAGE_IO_RESOLVER
);
setIos(innerIos, leafIos);
@@ -806,9 +810,10 @@ public abstract class BPlusTree<L, T extends L> extends DataStructure implements
long metaPageId,
ReuseList reuseList,
@Nullable FailureProcessor failureProcessor,
- @Nullable PageLockListener lsnr
- ) throws IgniteCheckedException {
- super(cacheGrpId, grpName, pageMem, wal, lsnr);
+ @Nullable PageLockListener lsnr,
+ PageIoResolver pageIoRslvr
+ ) {
+ super(cacheGrpId, grpName, pageMem, wal, lsnr, pageIoRslvr);
assert !F.isEmpty(name);
@@ -5289,6 +5294,9 @@ public abstract class BPlusTree<L, T extends L> extends DataStructure implements
throws IgniteCheckedException {
assert row != null;
+ if (row instanceof InsertLast)
+ return -cnt - 1;
+
int high = cnt - 1;
while (low <= high) {
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2ExtrasLeafIO.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageIoResolver.java
similarity index 66%
copy from modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2ExtrasLeafIO.java
copy to modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageIoResolver.java
index 085f98b..2354c09 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2ExtrasLeafIO.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageIoResolver.java
@@ -15,18 +15,15 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.query.h2.database.io;
+package org.apache.ignite.internal.processors.cache.persistence.tree.io;
-/**
- * Leaf page for H2 row references.
- */
-public class H2ExtrasLeafIO extends AbstractH2ExtrasLeafIO {
- /**
- * @param type Page type.
- * @param ver Page format version.
- * @param payloadSize Payload size.
- */
- H2ExtrasLeafIO(short type, int ver, int payloadSize) {
- super(type, ver, 8, payloadSize);
- }
+import org.apache.ignite.IgniteCheckedException;
+
+/** */
+public interface PageIoResolver {
+ /** */
+ public static final PageIoResolver DEFAULT_PAGE_IO_RESOLVER = PageIO::getPageIO;
+
+ /** */
+ PageIO resolve(long pageAddr) throws IgniteCheckedException;
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2ExtrasLeafIO.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/util/InsertLast.java
similarity index 67%
copy from modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2ExtrasLeafIO.java
copy to modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/util/InsertLast.java
index 085f98b..e7cfd41 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2ExtrasLeafIO.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/util/InsertLast.java
@@ -15,18 +15,10 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.query.h2.database.io;
+package org.apache.ignite.internal.processors.cache.persistence.tree.util;
/**
- * Leaf page for H2 row references.
+ * Rows with this marker interface will always be inserted in the very end of the tree.
*/
-public class H2ExtrasLeafIO extends AbstractH2ExtrasLeafIO {
- /**
- * @param type Page type.
- * @param ver Page format version.
- * @param payloadSize Payload size.
- */
- H2ExtrasLeafIO(short type, int ver, int payloadSize) {
- super(type, ver, 8, payloadSize);
- }
+public interface InsertLast {
}
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 4d05e5e..9461cca 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
@@ -25,11 +25,13 @@ import org.apache.ignite.internal.pagemem.PageSupport;
import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
import org.apache.ignite.internal.pagemem.wal.record.delta.InitNewPageRecord;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIoResolver;
import org.apache.ignite.internal.util.GridUnsafe;
import org.jetbrains.annotations.Nullable;
import static java.lang.Boolean.FALSE;
import static java.lang.Boolean.TRUE;
+import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIoResolver.DEFAULT_PAGE_IO_RESOLVER;
/**
* Page handler.
@@ -112,7 +114,8 @@ public abstract class PageHandler<X, R> {
X arg,
int intArg,
R lockFailed,
- IoStatisticsHolder statHolder
+ IoStatisticsHolder statHolder,
+ PageIoResolver pageIoRslvr
) throws IgniteCheckedException {
long page = pageMem.acquirePage(cacheId, pageId, statHolder);
try {
@@ -121,7 +124,7 @@ public abstract class PageHandler<X, R> {
if (pageAddr == 0L)
return lockFailed;
try {
- PageIO io = PageIO.getPageIO(pageAddr);
+ PageIO io = pageIoRslvr.resolve(pageAddr);
return h.run(cacheId, pageId, page, pageAddr, io, null, arg, intArg, statHolder);
}
finally {
@@ -157,7 +160,8 @@ public abstract class PageHandler<X, R> {
X arg,
int intArg,
R lockFailed,
- IoStatisticsHolder statHolder
+ IoStatisticsHolder statHolder,
+ PageIoResolver pageIoRslvr
) throws IgniteCheckedException {
long pageAddr = 0L;
@@ -165,7 +169,7 @@ public abstract class PageHandler<X, R> {
if ((pageAddr = readLock(pageMem, cacheId, pageId, page, lsnr)) == 0L)
return lockFailed;
- PageIO io = PageIO.getPageIO(pageAddr);
+ PageIO io = pageIoRslvr.resolve(pageAddr);
return h.run(cacheId, pageId, page, pageAddr, io, null, arg, intArg, statHolder);
}
finally {
@@ -236,7 +240,7 @@ public abstract class PageHandler<X, R> {
PageLockListener lsnr,
IoStatisticsHolder statHolder
) throws IgniteCheckedException {
- Boolean res = writePage(pageMem, grpId, pageId, lsnr, PageHandler.NO_OP, init, wal, null, null, 0, FALSE, statHolder);
+ Boolean res = writePage(pageMem, grpId, pageId, lsnr, PageHandler.NO_OP, init, wal, null, null, 0, FALSE, statHolder, DEFAULT_PAGE_IO_RESOLVER);
assert res != FALSE;
}
@@ -269,7 +273,8 @@ public abstract class PageHandler<X, R> {
X arg,
int intArg,
R lockFailed,
- IoStatisticsHolder statHolder
+ IoStatisticsHolder statHolder,
+ PageIoResolver pageIoRslvr
) throws IgniteCheckedException {
boolean releaseAfterWrite = true;
long page = pageMem.acquirePage(grpId, pageId, statHolder);
@@ -288,7 +293,7 @@ public abstract class PageHandler<X, R> {
walPlc = FALSE;
}
else
- init = PageIO.getPageIO(pageAddr);
+ init = pageIoRslvr.resolve(pageAddr);
R res = h.run(grpId, pageId, page, pageAddr, init, walPlc, arg, intArg, statHolder);
@@ -339,7 +344,8 @@ public abstract class PageHandler<X, R> {
X arg,
int intArg,
R lockFailed,
- IoStatisticsHolder statHolder
+ IoStatisticsHolder statHolder,
+ PageIoResolver pageIoRslvr
) throws IgniteCheckedException {
long pageAddr = writeLock(pageMem, grpId, pageId, page, lsnr, false);
@@ -355,7 +361,7 @@ public abstract class PageHandler<X, R> {
walPlc = FALSE;
}
else
- init = PageIO.getPageIO(pageAddr);
+ init = pageIoRslvr.resolve(pageAddr);
R res = h.run(grpId, pageId, page, pageAddr, init, walPlc, arg, intArg, statHolder);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataRowStore.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataRowStore.java
index 1032801..f066771 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataRowStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/CacheDataRowStore.java
@@ -31,7 +31,7 @@ import org.apache.ignite.internal.util.typedef.internal.CU;
*/
public class CacheDataRowStore extends RowStore {
/** Whether version should be skipped. */
- private static ThreadLocal<Boolean> SKIP_VER = ThreadLocal.withInitial(() -> false);
+ private static final ThreadLocal<Boolean> SKIP_VER = ThreadLocal.withInitial(() -> false);
/**
* @return Skip version flag.
@@ -74,17 +74,8 @@ public class CacheDataRowStore extends RowStore {
* @param link Link.
* @return Search row.
*/
- CacheSearchRow keySearchRow(int cacheId, int hash, long link) {
- DataRow dataRow = new DataRow(
- grp,
- hash,
- link,
- partId,
- CacheDataRowAdapter.RowData.KEY_ONLY,
- SKIP_VER.get()
- );
-
- return initDataRow(dataRow, cacheId);
+ protected CacheSearchRow keySearchRow(int cacheId, int hash, long link) {
+ return dataRow(cacheId, hash, link, CacheDataRowAdapter.RowData.KEY_ONLY);
}
/**
@@ -97,7 +88,7 @@ public class CacheDataRowStore extends RowStore {
* @param opCntr Mvcc operation counter.
* @return Search row.
*/
- MvccDataRow mvccRow(int cacheId, int hash, long link, CacheDataRowAdapter.RowData rowData, long crdVer, long mvccCntr, int opCntr) {
+ protected MvccDataRow mvccRow(int cacheId, int hash, long link, CacheDataRowAdapter.RowData rowData, long crdVer, long mvccCntr, int opCntr) {
MvccDataRow row = new MvccDataRow(
grp,
hash,
@@ -120,7 +111,7 @@ public class CacheDataRowStore extends RowStore {
* @param rowData Required row data.
* @return Data row.
*/
- CacheDataRow dataRow(int cacheId, int hash, long link, CacheDataRowAdapter.RowData rowData) {
+ protected CacheDataRow dataRow(int cacheId, int hash, long link, CacheDataRowAdapter.RowData rowData) {
DataRow dataRow = new DataRow(
grp,
hash,
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/WalModeChangeAdvancedSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/WalModeChangeAdvancedSelfTest.java
index 9eb3b7c..10e1db1 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/WalModeChangeAdvancedSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/WalModeChangeAdvancedSelfTest.java
@@ -24,7 +24,6 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
-
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsRecoveryAfterFileCorruptionTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsRecoveryAfterFileCorruptionTest.java
index 12774b4..82dfb71 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsRecoveryAfterFileCorruptionTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsRecoveryAfterFileCorruptionTest.java
@@ -339,7 +339,7 @@ public class IgnitePdsRecoveryAfterFileCorruptionTest extends GridCommonAbstract
long writeStart = System.nanoTime();
- storeMgr.write(cacheId, pageId, buf, tag);
+ storeMgr.write(cacheId, pageId, buf, tag, true);
long writeEnd = System.nanoTime();
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsTaskCancelingTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsTaskCancelingTest.java
index 95e24cc..300ce4e 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsTaskCancelingTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsTaskCancelingTest.java
@@ -44,7 +44,6 @@ import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactor
import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStore;
import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccessFileIOFactory;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
-import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric;
import org.apache.ignite.internal.util.GridUnsafe;
import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -199,8 +198,7 @@ public class IgnitePdsTaskCancelingTest extends GridCommonAbstractTest {
DataStorageConfiguration dbCfg = getDataStorageConfiguration();
- FilePageStore pageStore = new FilePageStore(PageMemory.FLAG_DATA, () -> file.toPath(), factory, dbCfg,
- new LongAdderMetric("NO_OP", null));
+ FilePageStore pageStore = new FilePageStore(PageMemory.FLAG_DATA, file::toPath, factory, dbCfg, val -> {});
int pageSize = dbCfg.getPageSize();
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/file/IgnitePdsCheckpointSimulationWithRealCpDisabledTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/file/IgnitePdsCheckpointSimulationWithRealCpDisabledTest.java
index 04a0cf1..f829b90 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/file/IgnitePdsCheckpointSimulationWithRealCpDisabledTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/file/IgnitePdsCheckpointSimulationWithRealCpDisabledTest.java
@@ -1016,7 +1016,7 @@ public class IgnitePdsCheckpointSimulationWithRealCpDisabledTest extends GridCom
long writeStart = System.nanoTime();
- storeMgr.write(cacheId, fullId.pageId(), tmpBuf, tag);
+ storeMgr.write(cacheId, fullId.pageId(), tmpBuf, tag, true);
long writeEnd = System.nanoTime();
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreePageMemoryImplTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreePageMemoryImplTest.java
index 8f985cd..3e73281 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreePageMemoryImplTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreePageMemoryImplTest.java
@@ -108,6 +108,7 @@ public class BPlusTreePageMemoryImplTest extends BPlusTreeSelfTest {
PageMemory mem = new PageMemoryImpl(
provider, sizes,
sharedCtx,
+ sharedCtx.pageStore(),
PAGE_SIZE,
(fullPageId, byteBuf, tag) -> {
assert false : "No page replacement should happen during the test";
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreeReuseListPageMemoryImplTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreeReuseListPageMemoryImplTest.java
index 1e67469..456c9d6 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreeReuseListPageMemoryImplTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreeReuseListPageMemoryImplTest.java
@@ -107,6 +107,7 @@ public class BPlusTreeReuseListPageMemoryImplTest extends BPlusTreeReuseSelfTest
PageMemory mem = new PageMemoryImpl(
provider, sizes,
sharedCtx,
+ sharedCtx.pageStore(),
PAGE_SIZE,
(fullPageId, byteBuf, tag) -> {
assert false : "No page replacement (rotation with disk) should happen during the test";
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/IgnitePageMemReplaceDelayedWriteUnitTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/IgnitePageMemReplaceDelayedWriteUnitTest.java
index a8f9ee2..1f6d5ce 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/IgnitePageMemReplaceDelayedWriteUnitTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/IgnitePageMemReplaceDelayedWriteUnitTest.java
@@ -275,7 +275,7 @@ public class IgnitePageMemReplaceDelayedWriteUnitTest {
}
};
- PageMemoryImpl memory = new PageMemoryImpl(provider, sizes, sctx, pageSize,
+ PageMemoryImpl memory = new PageMemoryImpl(provider, sizes, sctx, sctx.pageStore(), pageSize,
pageWriter, null, () -> true, memMetrics, PageMemoryImpl.ThrottlingPolicy.DISABLED,
clo);
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/IndexStoragePageMemoryImplTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/IndexStoragePageMemoryImplTest.java
index 67aa12b..19e67b4 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/IndexStoragePageMemoryImplTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/IndexStoragePageMemoryImplTest.java
@@ -123,6 +123,7 @@ public class IndexStoragePageMemoryImplTest extends IndexStorageSelfTest {
return new PageMemoryImpl(
provider, sizes,
sharedCtx,
+ sharedCtx.pageStore(),
PAGE_SIZE,
(fullPageId, byteBuf, tag) -> {
assert false : "No page replacement (rotation with disk) should happen during the test";
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/NoOpPageStoreManager.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/NoOpPageStoreManager.java
index 3c7a2ed..c5cc372 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/NoOpPageStoreManager.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/NoOpPageStoreManager.java
@@ -23,6 +23,7 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.LongConsumer;
import java.util.function.Predicate;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.configuration.CacheConfiguration;
@@ -30,11 +31,11 @@ import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.pagemem.FullPageId;
import org.apache.ignite.internal.pagemem.PageIdUtils;
import org.apache.ignite.internal.pagemem.store.IgnitePageStoreManager;
+import org.apache.ignite.internal.pagemem.store.PageStore;
import org.apache.ignite.internal.processors.cache.CacheGroupContext;
import org.apache.ignite.internal.processors.cache.CacheGroupDescriptor;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.StoredCacheData;
-import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.lang.IgniteFuture;
@@ -57,7 +58,7 @@ public class NoOpPageStoreManager implements IgnitePageStoreManager {
/** {@inheritDoc} */
@Override public void initialize(int cacheId, int partitions, String workingDir,
- LongAdderMetric tracker) throws IgniteCheckedException {
+ LongConsumer tracker) throws IgniteCheckedException {
// No-op.
}
@@ -88,8 +89,8 @@ public class NoOpPageStoreManager implements IgnitePageStoreManager {
}
/** {@inheritDoc} */
- @Override public void read(int grpId, long pageId, ByteBuffer pageBuf) throws IgniteCheckedException {
-
+ @Override public void read(int grpId, long pageId, ByteBuffer pageBuf, boolean keepCrc) throws IgniteCheckedException {
+ // No-op.
}
/** {@inheritDoc} */
@@ -103,8 +104,9 @@ public class NoOpPageStoreManager implements IgnitePageStoreManager {
}
/** {@inheritDoc} */
- @Override public void write(int grpId, long pageId, ByteBuffer pageBuf, int tag) throws IgniteCheckedException {
+ @Override public PageStore write(int grpId, long pageId, ByteBuffer pageBuf, int tag, boolean calculateCrc) throws IgniteCheckedException {
// No-op.
+ return null;
}
/** {@inheritDoc} */
@@ -151,11 +153,6 @@ public class NoOpPageStoreManager implements IgnitePageStoreManager {
}
/** {@inheritDoc} */
- @Override public long metaPageId(int grpId) {
- return 1;
- }
-
- /** {@inheritDoc} */
@Override public void start(GridCacheSharedContext cctx) throws IgniteCheckedException {
// No-op.
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplNoLoadTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplNoLoadTest.java
index 9f74ec4..51e29ff 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplNoLoadTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplNoLoadTest.java
@@ -114,6 +114,7 @@ public class PageMemoryImplNoLoadTest extends PageMemoryNoLoadSelfTest {
provider,
sizes,
sharedCtx,
+ sharedCtx.pageStore(),
PAGE_SIZE,
(fullPageId, byteBuf, tag) -> {
assert false : "No page replacement (rotation with disk) should happen during the test";
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplTest.java
index 9ea27f9..1632e73 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplTest.java
@@ -43,6 +43,7 @@ import org.apache.ignite.internal.mem.unsafe.UnsafeMemoryProvider;
import org.apache.ignite.internal.pagemem.FullPageId;
import org.apache.ignite.internal.pagemem.PageUtils;
import org.apache.ignite.internal.pagemem.store.IgnitePageStoreManager;
+import org.apache.ignite.internal.pagemem.store.PageStore;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.persistence.CheckpointLockStateChecker;
import org.apache.ignite.internal.processors.cache.persistence.DataRegionMetricsImpl;
@@ -327,7 +328,7 @@ public class PageMemoryImplTest extends GridCommonAbstractTest {
PageStoreWriter pageStoreWriter = (fullPageId, buf, tag) -> {
assertNotNull(tag);
- pageStoreMgr.write(fullPageId.groupId(), fullPageId.pageId(), buf, 1);
+ pageStoreMgr.write(fullPageId.groupId(), fullPageId.pageId(), buf, 1, false);
};
for (FullPageId cpPage : cpPages) {
@@ -642,6 +643,7 @@ public class PageMemoryImplTest extends GridCommonAbstractTest {
provider,
sizes,
sharedCtx,
+ sharedCtx.pageStore(),
PAGE_SIZE,
replaceWriter,
new GridInClosure3X<Long, FullPageId, PageMemoryEx>() {
@@ -661,6 +663,7 @@ public class PageMemoryImplTest extends GridCommonAbstractTest {
provider,
sizes,
sharedCtx,
+ sharedCtx.pageStore(),
PAGE_SIZE,
replaceWriter,
new GridInClosure3X<Long, FullPageId, PageMemoryEx>() {
@@ -701,7 +704,7 @@ public class PageMemoryImplTest extends GridCommonAbstractTest {
private Map<FullPageId, byte[]> storedPages = new HashMap<>();
/** {@inheritDoc} */
- @Override public void read(int grpId, long pageId, ByteBuffer pageBuf) throws IgniteCheckedException {
+ @Override public void read(int grpId, long pageId, ByteBuffer pageBuf, boolean keepCrc) throws IgniteCheckedException {
FullPageId fullPageId = new FullPageId(pageId, grpId);
byte[] bytes = storedPages.get(fullPageId);
@@ -713,12 +716,14 @@ public class PageMemoryImplTest extends GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override public void write(int grpId, long pageId, ByteBuffer pageBuf, int tag) throws IgniteCheckedException {
+ @Override public PageStore write(int grpId, long pageId, ByteBuffer pageBuf, int tag, boolean calculateCrc) throws IgniteCheckedException {
byte[] data = new byte[PAGE_SIZE];
pageBuf.get(data);
storedPages.put(new FullPageId(pageId, grpId), data);
+
+ return null;
}
/** {@inheritDoc} */
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/DurableBackgroundCleanupIndexTreeTask.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/DurableBackgroundCleanupIndexTreeTask.java
index 23e9af7..4d91660 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/DurableBackgroundCleanupIndexTreeTask.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/DurableBackgroundCleanupIndexTreeTask.java
@@ -28,6 +28,7 @@ import org.apache.ignite.internal.metric.IoStatisticsHolderIndex;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.persistence.metastorage.pendingtask.DurableBackgroundTask;
import org.apache.ignite.internal.processors.cache.persistence.tree.BPlusTree;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIoResolver;
import org.apache.ignite.internal.processors.query.h2.database.H2Tree;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -141,7 +142,8 @@ public class DurableBackgroundCleanupIndexTreeTask implements DurableBackgroundT
null,
stats,
null,
- 0
+ 0,
+ PageIoResolver.DEFAULT_PAGE_IO_RESOLVER
);
trees0.add(tree);
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
index de64068..7a31059 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
@@ -479,7 +479,6 @@ public class IgniteH2Indexing implements GridQueryIndexing {
wrappedCols,
inlineSize,
segments,
- qryCtxRegistry,
log
);
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2Tree.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2Tree.java
index 66badc0..9c6ca13 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2Tree.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2Tree.java
@@ -41,6 +41,7 @@ import org.apache.ignite.internal.processors.cache.persistence.tree.BPlusTree;
import org.apache.ignite.internal.processors.cache.persistence.tree.CorruptedTreeException;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.BPlusIO;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.BPlusMetaIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIoResolver;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
import org.apache.ignite.internal.processors.cache.tree.mvcc.data.MvccDataRow;
import org.apache.ignite.internal.processors.failure.FailureProcessor;
@@ -208,7 +209,8 @@ public class H2Tree extends BPlusTree<H2Row, H2Row> {
IgniteLogger log,
IoStatisticsHolder stats,
InlineIndexColumnFactory factory,
- int configuredInlineSize
+ int configuredInlineSize,
+ PageIoResolver pageIoRslvr
) throws IgniteCheckedException {
super(
name,
@@ -220,7 +222,8 @@ public class H2Tree extends BPlusTree<H2Row, H2Row> {
metaPageId,
reuseList,
failureProcessor,
- null
+ null,
+ pageIoRslvr
);
this.cctx = cctx;
@@ -296,7 +299,7 @@ public class H2Tree extends BPlusTree<H2Row, H2Row> {
*
* @return Indexed columns.
*/
- IndexColumn[] cols() {
+ public IndexColumn[] cols() {
return cols;
}
@@ -308,11 +311,15 @@ public class H2Tree extends BPlusTree<H2Row, H2Row> {
* @throws IgniteCheckedException if failed.
*/
public H2Row createRow(long link) throws IgniteCheckedException {
+ return createRow(link, true);
+ }
+
+ public H2Row createRow(long link, boolean follow) throws IgniteCheckedException {
if (rowCache != null) {
H2CacheRow row = rowCache.get(link);
if (row == null) {
- row = createRow0(link);
+ row = createRow0(link, follow);
rowCache.put(row);
}
@@ -320,7 +327,7 @@ public class H2Tree extends BPlusTree<H2Row, H2Row> {
return row;
}
else
- return createRow0(link);
+ return createRow0(link, follow);
}
/**
@@ -332,14 +339,16 @@ public class H2Tree extends BPlusTree<H2Row, H2Row> {
* @return Row.
* @throws IgniteCheckedException If failed.
*/
- private H2CacheRow createRow0(long link) throws IgniteCheckedException {
+ private H2CacheRow createRow0(long link, boolean follow) throws IgniteCheckedException {
CacheDataRowAdapter row = new CacheDataRowAdapter(link);
- row.initFromLink(
- cctx.group(),
- CacheDataRowAdapter.RowData.FULL,
- true
- );
+ if (follow) {
+ row.initFromLink(
+ cctx.group(),
+ CacheDataRowAdapter.RowData.FULL,
+ true
+ );
+ }
return table.rowDescriptor().createRow(row);
}
@@ -352,12 +361,35 @@ public class H2Tree extends BPlusTree<H2Row, H2Row> {
* @return Row.
* @throws IgniteCheckedException if failed.
*/
- public H2Row createMvccRow(long link, long mvccCrdVer, long mvccCntr, int mvccOpCntr) throws IgniteCheckedException {
+ public H2Row createMvccRow(
+ long link,
+ long mvccCrdVer,
+ long mvccCntr,
+ int mvccOpCntr
+ ) throws IgniteCheckedException {
+ return createMvccRow(link, mvccCrdVer, mvccCntr, mvccOpCntr, null);
+ }
+
+ /**
+ * Create row from link.
+ *
+ * @param link Link.
+ * @param mvccOpCntr MVCC operation counter.
+ * @return Row.
+ * @throws IgniteCheckedException if failed.
+ */
+ public H2Row createMvccRow(
+ long link,
+ long mvccCrdVer,
+ long mvccCntr,
+ int mvccOpCntr,
+ CacheDataRowAdapter.RowData rowData
+ ) throws IgniteCheckedException {
if (rowCache != null) {
H2CacheRow row = rowCache.get(link);
if (row == null) {
- row = createMvccRow0(link, mvccCrdVer, mvccCntr, mvccOpCntr);
+ row = createMvccRow0(link, mvccCrdVer, mvccCntr, mvccOpCntr, rowData);
rowCache.put(row);
}
@@ -365,7 +397,15 @@ public class H2Tree extends BPlusTree<H2Row, H2Row> {
return row;
}
else
- return createMvccRow0(link, mvccCrdVer, mvccCntr, mvccOpCntr);
+ return createMvccRow0(link, mvccCrdVer, mvccCntr, mvccOpCntr, rowData);
+ }
+
+ public boolean getPk() {
+ return pk;
+ }
+
+ public boolean getAffinityKey() {
+ return affinityKey;
}
/**
@@ -375,8 +415,8 @@ public class H2Tree extends BPlusTree<H2Row, H2Row> {
* @param mvccOpCntr Mvcc operation counter.
* @return Row.
*/
- private H2CacheRow createMvccRow0(long link, long mvccCrdVer, long mvccCntr, int mvccOpCntr)
- throws IgniteCheckedException {
+ private H2CacheRow createMvccRow0(long link, long mvccCrdVer, long mvccCntr, int mvccOpCntr, CacheDataRowAdapter.RowData rowData)
+ throws IgniteCheckedException {
int partId = PageIdUtils.partId(PageIdUtils.pageId(link));
MvccDataRow row = new MvccDataRow(
@@ -384,7 +424,7 @@ public class H2Tree extends BPlusTree<H2Row, H2Row> {
0,
link,
partId,
- null,
+ rowData,
mvccCrdVer,
mvccCntr,
mvccOpCntr,
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2TreeIndex.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2TreeIndex.java
index 339ee70..1b7ffe7 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2TreeIndex.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2TreeIndex.java
@@ -41,11 +41,13 @@ import org.apache.ignite.internal.metric.IoStatisticsHolderIndex;
import org.apache.ignite.internal.pagemem.PageMemory;
import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.IgniteCacheOffheapManager;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.persistence.IgniteCacheDatabaseSharedManager;
import org.apache.ignite.internal.processors.cache.persistence.RootPage;
import org.apache.ignite.internal.processors.cache.persistence.metastorage.pendingtask.DurableBackgroundTask;
import org.apache.ignite.internal.processors.cache.persistence.tree.BPlusTree;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIoResolver;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
import org.apache.ignite.internal.processors.failure.FailureProcessor;
import org.apache.ignite.internal.processors.query.GridQueryTypeDescriptor;
@@ -240,7 +242,6 @@ public class H2TreeIndex extends H2TreeIndexBase {
* @param wrappedCols Index columns as is.
* @param inlineSize Inline size.
* @param segmentsCnt Count of tree segments.
- * @param qryCtxRegistry Query context registry.
* @throws IgniteCheckedException If failed.
*/
public static H2TreeIndex createIndex(
@@ -254,7 +255,31 @@ public class H2TreeIndex extends H2TreeIndexBase {
List<IndexColumn> wrappedCols,
int inlineSize,
int segmentsCnt,
- QueryContextRegistry qryCtxRegistry,
+ IgniteLogger log
+ ) throws IgniteCheckedException {
+ return createIndex(cctx, rowCache, tbl, idxName, pk, affinityKey, unwrappedCols, wrappedCols, inlineSize,
+ segmentsCnt, cctx.dataRegion().pageMemory(),
+ cctx.offheap(),
+ PageIoResolver.DEFAULT_PAGE_IO_RESOLVER,
+ log
+ );
+ }
+
+ /** */
+ public static H2TreeIndex createIndex(
+ GridCacheContext<?, ?> cctx,
+ @Nullable H2RowCache rowCache,
+ GridH2Table tbl,
+ String idxName,
+ boolean pk,
+ boolean affinityKey,
+ List<IndexColumn> unwrappedCols,
+ List<IndexColumn> wrappedCols,
+ int inlineSize,
+ int segmentsCnt,
+ PageMemory pageMemory,
+ IgniteCacheOffheapManager offheap,
+ PageIoResolver pageIoRslvr,
IgniteLogger log
) throws IgniteCheckedException {
assert segmentsCnt > 0 : segmentsCnt;
@@ -274,10 +299,10 @@ public class H2TreeIndex extends H2TreeIndexBase {
AtomicInteger maxCalculatedInlineSize = new AtomicInteger();
IoStatisticsHolderIndex stats = new IoStatisticsHolderIndex(
- SORTED_INDEX,
- cctx.name(),
- idxName,
- cctx.kernalContext().metric()
+ SORTED_INDEX,
+ cctx.name(),
+ idxName,
+ cctx.kernalContext().metric()
);
InlineIndexColumnFactory idxHelperFactory = new InlineIndexColumnFactory(tbl.getCompareMode());
@@ -286,7 +311,7 @@ public class H2TreeIndex extends H2TreeIndexBase {
db.checkpointReadLock();
try {
- RootPage page = getMetaPage(cctx, treeName, i);
+ RootPage page = getMetaPage(offheap, cctx, treeName, i);
segments[i] = h2TreeFactory.create(
cctx,
@@ -295,12 +320,12 @@ public class H2TreeIndex extends H2TreeIndexBase {
idxName,
tbl.getName(),
tbl.cacheName(),
- cctx.offheap().reuseListForIndex(treeName),
+ offheap.reuseListForIndex(treeName),
cctx.groupId(),
cctx.group().name(),
- cctx.dataRegion().pageMemory(),
+ pageMemory,
cctx.shared().wal(),
- cctx.offheap().globalRemoveId(),
+ offheap.globalRemoveId(),
page.pageId().pageId(),
page.isAllocated(),
unwrappedCols,
@@ -314,7 +339,8 @@ public class H2TreeIndex extends H2TreeIndexBase {
log,
stats,
idxHelperFactory,
- inlineSize
+ inlineSize,
+ pageIoRslvr
);
}
finally {
@@ -593,7 +619,7 @@ public class H2TreeIndex extends H2TreeIndexBase {
* @param segment Segment Id.
* @return Snapshot for requested segment if there is one.
*/
- private H2Tree treeForRead(int segment) {
+ public H2Tree treeForRead(int segment) {
return segments[segment];
}
@@ -627,9 +653,9 @@ public class H2TreeIndex extends H2TreeIndexBase {
* @return RootPage for meta page.
* @throws IgniteCheckedException If failed.
*/
- private static RootPage getMetaPage(GridCacheContext<?, ?> cctx, String treeName, int segIdx)
+ private static RootPage getMetaPage(IgniteCacheOffheapManager offheap, GridCacheContext<?, ?> cctx, String treeName, int segIdx)
throws IgniteCheckedException {
- return cctx.offheap().rootPageForIndex(cctx.cacheId(), treeName, segIdx);
+ return offheap.rootPageForIndex(cctx.cacheId(), treeName, segIdx);
}
/**
@@ -1018,7 +1044,8 @@ public class H2TreeIndex extends H2TreeIndexBase {
IgniteLogger log,
IoStatisticsHolder stats,
InlineIndexColumnFactory factory,
- int configuredInlineSize
+ int configuredInlineSize,
+ PageIoResolver pageIoRslvr
) throws IgniteCheckedException;
}
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/inlinecolumn/AbstractInlineIndexColumn.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/inlinecolumn/AbstractInlineIndexColumn.java
index 313c542..7380c53 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/inlinecolumn/AbstractInlineIndexColumn.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/inlinecolumn/AbstractInlineIndexColumn.java
@@ -100,7 +100,7 @@ public abstract class AbstractInlineIndexColumn implements InlineIndexColumn {
*
* @return Restored value or {@code null} if value can't be restored.
*/
- @Nullable Value get(long pageAddr, int off, int maxSize) {
+ @Nullable public Value get(long pageAddr, int off, int maxSize) {
if (size > 0 && size + 1 > maxSize)
return null;
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/inlinecolumn/ObjectHashInlineIndexColumn.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/inlinecolumn/ObjectHashInlineIndexColumn.java
index 7226a5a..b146436 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/inlinecolumn/ObjectHashInlineIndexColumn.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/inlinecolumn/ObjectHashInlineIndexColumn.java
@@ -17,8 +17,11 @@
package org.apache.ignite.internal.processors.query.h2.database.inlinecolumn;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
import org.apache.ignite.internal.pagemem.PageUtils;
import org.h2.table.Column;
+import org.h2.value.CompareMode;
import org.h2.value.Value;
import org.h2.value.ValueInt;
@@ -60,7 +63,9 @@ public class ObjectHashInlineIndexColumn extends AbstractInlineIndexColumn {
/** {@inheritDoc} */
@Override protected Value get0(long pageAddr, int off) {
- return null;
+ int hashCode = PageUtils.getInt(pageAddr, off + 1);
+
+ return new ValueObjectHashCode(hashCode);
}
/**
@@ -80,4 +85,91 @@ public class ObjectHashInlineIndexColumn extends AbstractInlineIndexColumn {
return size() + 1;
}
+
+ /**
+ * Value for object with hashcode.
+ */
+ private static class ValueObjectHashCode extends Value {
+ /**
+ * The precision in digits.
+ */
+ public static final int PRECISION = 10;
+
+ /**
+ * The maximum display size of an int.
+ * Example: -2147483648
+ */
+ public static final int DISPLAY_SIZE = 11;
+
+ /**
+ * Hashcode of object.
+ */
+ private final int value;
+
+ public ValueObjectHashCode(int value) {
+ this.value = value;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String getSQL() {
+ return getString();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getType() {
+ return Value.JAVA_OBJECT;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getInt() {
+ return value;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getLong() {
+ return value;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected int compareSecure(Value o, CompareMode mode) {
+ ValueObjectHashCode v = (ValueObjectHashCode) o;
+ return Integer.compare(value, v.value);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String getString() {
+ return String.valueOf(value);
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getPrecision() {
+ return PRECISION;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return value;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object getObject() {
+ return value;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void set(PreparedStatement prep, int parameterIndex)
+ throws SQLException {
+ prep.setInt(parameterIndex, value);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getDisplaySize() {
+ return DISPLAY_SIZE;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object other) {
+ return other instanceof ValueObjectHashCode && value == ((ValueObjectHashCode) other).value;
+ }
+ }
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/AbstractH2ExtrasInnerIO.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/AbstractH2ExtrasInnerIO.java
index f62952f..efb942c 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/AbstractH2ExtrasInnerIO.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/AbstractH2ExtrasInnerIO.java
@@ -161,4 +161,9 @@ public abstract class AbstractH2ExtrasInnerIO extends BPlusInnerIO<H2Row> implem
@Override public final long getLink(long pageAddr, int idx) {
return PageUtils.getLong(pageAddr, offset(idx) + payloadSize);
}
+
+ /** {@inheritDoc} */
+ @Override public int getPayloadSize() {
+ return payloadSize;
+ }
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/AbstractH2ExtrasLeafIO.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/AbstractH2ExtrasLeafIO.java
index 4bddaee..6e4236c 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/AbstractH2ExtrasLeafIO.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/AbstractH2ExtrasLeafIO.java
@@ -64,7 +64,6 @@ public abstract class AbstractH2ExtrasLeafIO extends BPlusLeafIO<H2Row> implemen
* @param mvccEnabled Mvcc flag.
* @return IOVersions for given payload.
*/
- @SuppressWarnings("unchecked")
public static IOVersions<? extends BPlusLeafIO<H2Row>> getVersions(int payload, boolean mvccEnabled) {
assert payload >= 0 && payload <= PageIO.MAX_PAYLOAD_SIZE;
@@ -140,7 +139,7 @@ public abstract class AbstractH2ExtrasLeafIO extends BPlusLeafIO<H2Row> implemen
}
/** {@inheritDoc} */
- @Override public final H2Row getLookupRow(BPlusTree<H2Row, ?> tree, long pageAddr, int idx)
+ @Override public H2Row getLookupRow(BPlusTree<H2Row, ?> tree, long pageAddr, int idx)
throws IgniteCheckedException {
long link = getLink(pageAddr, idx);
@@ -159,4 +158,9 @@ public abstract class AbstractH2ExtrasLeafIO extends BPlusLeafIO<H2Row> implemen
@Override public final long getLink(long pageAddr, int idx) {
return PageUtils.getLong(pageAddr, offset(idx) + payloadSize);
}
+
+ /** {@inheritDoc} */
+ @Override public int getPayloadSize() {
+ return payloadSize;
+ }
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/AbstractH2InnerIO.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/AbstractH2InnerIO.java
index a782ffb..76cccf3 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/AbstractH2InnerIO.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/AbstractH2InnerIO.java
@@ -71,4 +71,9 @@ public abstract class AbstractH2InnerIO extends BPlusInnerIO<H2Row> implements H
@Override public long getLink(long pageAddr, int idx) {
return PageUtils.getLong(pageAddr, offset(idx));
}
+
+ /** {@inheritDoc} */
+ @Override public int getPayloadSize() {
+ return 0;
+ }
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/AbstractH2LeafIO.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/AbstractH2LeafIO.java
index ccacb4e..058b5bc 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/AbstractH2LeafIO.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/AbstractH2LeafIO.java
@@ -54,7 +54,7 @@ public abstract class AbstractH2LeafIO extends BPlusLeafIO<H2Row> implements H2R
}
/** {@inheritDoc} */
- @Override public final H2Row getLookupRow(BPlusTree<H2Row,?> tree, long pageAddr, int idx)
+ @Override public H2Row getLookupRow(BPlusTree<H2Row,?> tree, long pageAddr, int idx)
throws IgniteCheckedException {
long link = getLink(pageAddr, idx);
@@ -73,4 +73,9 @@ public abstract class AbstractH2LeafIO extends BPlusLeafIO<H2Row> implements H2R
@Override public long getLink(long pageAddr, int idx) {
return PageUtils.getLong(pageAddr, offset(idx));
}
+
+ /** {@inheritDoc} */
+ @Override public int getPayloadSize() {
+ return 0;
+ }
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2ExtrasLeafIO.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2ExtrasLeafIO.java
index 085f98b..3c79df7 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2ExtrasLeafIO.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2ExtrasLeafIO.java
@@ -26,7 +26,7 @@ public class H2ExtrasLeafIO extends AbstractH2ExtrasLeafIO {
* @param ver Page format version.
* @param payloadSize Payload size.
*/
- H2ExtrasLeafIO(short type, int ver, int payloadSize) {
+ public H2ExtrasLeafIO(short type, int ver, int payloadSize) {
super(type, ver, 8, payloadSize);
}
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2LeafIO.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2LeafIO.java
index 8954de0..466cd1c 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2LeafIO.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2LeafIO.java
@@ -31,7 +31,7 @@ public class H2LeafIO extends AbstractH2LeafIO {
/**
* @param ver Page format version.
*/
- private H2LeafIO(int ver) {
+ public H2LeafIO(int ver) {
super(T_H2_REF_LEAF, ver, 8);
}
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2MvccExtrasLeafIO.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2MvccExtrasLeafIO.java
index 60a1598..7585471 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2MvccExtrasLeafIO.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2MvccExtrasLeafIO.java
@@ -22,13 +22,13 @@ import org.apache.ignite.internal.pagemem.PageUtils;
/**
* Leaf page for H2 row references.
*/
-class H2MvccExtrasLeafIO extends AbstractH2ExtrasLeafIO {
+public class H2MvccExtrasLeafIO extends AbstractH2ExtrasLeafIO {
/**
* @param type Page type.
* @param ver Page format version.
* @param payloadSize Payload size.
*/
- H2MvccExtrasLeafIO(short type, int ver, int payloadSize) {
+ protected H2MvccExtrasLeafIO(short type, int ver, int payloadSize) {
super(type, ver, 28, payloadSize);
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2MvccLeafIO.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2MvccLeafIO.java
index c7cd998..5575806 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2MvccLeafIO.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2MvccLeafIO.java
@@ -32,7 +32,7 @@ public class H2MvccLeafIO extends AbstractH2LeafIO {
/**
* @param ver Page format version.
*/
- private H2MvccLeafIO(int ver) {
+ protected H2MvccLeafIO(int ver) {
super(T_H2_MVCC_REF_LEAF, ver, 28);
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2RowLinkIO.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2RowLinkIO.java
index 1942069..55a36d6 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2RowLinkIO.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/io/H2RowLinkIO.java
@@ -61,4 +61,9 @@ public interface H2RowLinkIO {
public default boolean storeMvccInfo() {
return false;
}
+
+ /**
+ * @return Size of reserved data array for data inlining.
+ */
+ public int getPayloadSize();
}
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 527a33e..86df146 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
@@ -183,6 +183,13 @@ public class H2CacheRow extends H2Row implements CacheDataRow {
}
/**
+ * @return Cache data row.
+ */
+ public CacheDataRow getRow() {
+ return row;
+ }
+
+ /**
* @return {@code True} if this is removed row (doesn't have value).
*/
private boolean removedRow() {
@@ -345,4 +352,11 @@ public class H2CacheRow extends H2Row implements CacheDataRow {
return sb.toString();
}
+
+ /**
+ * @return H2 row descriptor.
+ */
+ public GridH2RowDescriptor getDesc() {
+ return desc;
+ }
}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/LongDestroyDurableBackgroundTaskTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/LongDestroyDurableBackgroundTaskTest.java
index ca96684..1f03ef4 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/LongDestroyDurableBackgroundTaskTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/LongDestroyDurableBackgroundTaskTest.java
@@ -60,6 +60,7 @@ import org.apache.ignite.internal.processors.cache.persistence.checkpoint.Checkp
import org.apache.ignite.internal.processors.cache.persistence.metastorage.ReadOnlyMetastorage;
import org.apache.ignite.internal.processors.cache.persistence.metastorage.ReadWriteMetastorage;
import org.apache.ignite.internal.processors.cache.persistence.metastorage.pendingtask.DurableBackgroundTask;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIoResolver;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.LongListReuseBag;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
import org.apache.ignite.internal.processors.failure.FailureProcessor;
@@ -758,7 +759,8 @@ public class LongDestroyDurableBackgroundTaskTest extends GridCommonAbstractTest
IgniteLogger log,
IoStatisticsHolder stats,
InlineIndexColumnFactory factory,
- int configuredInlineSize
+ int configuredInlineSize,
+ PageIoResolver pageIoRslvr
) throws IgniteCheckedException {
super(
cctx,
@@ -786,7 +788,8 @@ public class LongDestroyDurableBackgroundTaskTest extends GridCommonAbstractTest
log,
stats,
factory,
- configuredInlineSize
+ configuredInlineSize,
+ pageIoRslvr
);
}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/MultipleParallelCacheDeleteDeadlockTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/MultipleParallelCacheDeleteDeadlockTest.java
index 9e0f93a..ed90b66 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/MultipleParallelCacheDeleteDeadlockTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/MultipleParallelCacheDeleteDeadlockTest.java
@@ -36,6 +36,7 @@ import org.apache.ignite.internal.metric.IoStatisticsHolder;
import org.apache.ignite.internal.pagemem.PageMemory;
import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIoResolver;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.LongListReuseBag;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
import org.apache.ignite.internal.processors.failure.FailureProcessor;
@@ -266,7 +267,8 @@ public class MultipleParallelCacheDeleteDeadlockTest extends GridCommonAbstractT
IgniteLogger log,
IoStatisticsHolder stats,
InlineIndexColumnFactory factory,
- int configuredInlineSize
+ int configuredInlineSize,
+ PageIoResolver pageIoRslvr
) throws IgniteCheckedException {
super(
cctx,
@@ -294,7 +296,8 @@ public class MultipleParallelCacheDeleteDeadlockTest extends GridCommonAbstractT
log,
stats,
factory,
- configuredInlineSize
+ configuredInlineSize,
+ pageIoRslvr
);
}