This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new a51777f72c IGNITE-20958
PersistentPageMemoryMvPartitionStorageTest#groupConfigShorteningWorksCorrectly
threw an AssertionError (#2888)
a51777f72c is described below
commit a51777f72cd53ae1a43d72e41d14d73cdf0b971e
Author: Kirill Tkalenko <[email protected]>
AuthorDate: Tue Nov 28 15:13:56 2023 +0300
IGNITE-20958
PersistentPageMemoryMvPartitionStorageTest#groupConfigShorteningWorksCorrectly
threw an AssertionError (#2888)
---
.../persistence/PartitionMetaManager.java | 50 ++++-----
.../persistence/store/FilePageStoreManager.java | 95 +++++++---------
.../persistence/store/GroupPageStoresMap.java | 12 ++
.../persistence/PartitionMetaManagerTest.java | 22 +++-
.../PersistentPageMemoryNoLoadTest.java | 29 +++--
.../store/FilePageStoreManagerTest.java | 118 ++++++++++++++-----
.../PersistentPageMemoryTableStorage.java | 125 +++++++++++----------
7 files changed, 269 insertions(+), 182 deletions(-)
diff --git
a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/PartitionMetaManager.java
b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/PartitionMetaManager.java
index bfda58770f..67f4833de3 100644
---
a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/PartitionMetaManager.java
+++
b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/PartitionMetaManager.java
@@ -18,9 +18,7 @@
package org.apache.ignite.internal.pagememory.persistence;
import static
org.apache.ignite.internal.pagememory.persistence.PartitionMeta.partitionMetaPageId;
-import static org.apache.ignite.internal.util.GridUnsafe.allocateBuffer;
import static org.apache.ignite.internal.util.GridUnsafe.bufferAddress;
-import static org.apache.ignite.internal.util.GridUnsafe.freeBuffer;
import java.nio.ByteBuffer;
import java.util.Map;
@@ -80,55 +78,51 @@ public class PartitionMetaManager {
/**
* Reads the partition {@link PartitionMeta meta} from the partition file
or creates a new one.
*
- * <p>If it creates a new one, it writes the meta to the file.
+ * <p>If it creates a new one, it writes the meta to the file.</p>
*
* @param checkpointId Checkpoint ID.
* @param groupPartitionId Partition of the group.
* @param filePageStore Partition file page store.
+ * @param buffer Buffer for reading and writing pages.
*/
public PartitionMeta readOrCreateMeta(
@Nullable UUID checkpointId,
GroupPartitionId groupPartitionId,
- FilePageStore filePageStore
+ FilePageStore filePageStore,
+ ByteBuffer buffer
) throws IgniteInternalCheckedException {
- ByteBuffer buffer = allocateBuffer(pageSize);
-
long bufferAddr = bufferAddress(buffer);
long partitionMetaPageId =
partitionMetaPageId(groupPartitionId.getPartitionId());
- try {
- if (containsPartitionMeta(filePageStore)) {
- // Reads the partition meta.
- try {
- filePageStore.readWithoutPageIdCheck(partitionMetaPageId,
buffer, false);
+ if (containsPartitionMeta(filePageStore)) {
+ // Reads the partition meta.
+ try {
+ filePageStore.readWithoutPageIdCheck(partitionMetaPageId,
buffer, false);
- return new PartitionMeta(checkpointId,
ioRegistry.resolve(bufferAddr), bufferAddr);
- } catch (IgniteInternalDataIntegrityViolationException e) {
- LOG.info(() -> "Error reading partition meta page, will be
recreated: " + groupPartitionId, e);
- }
+ return new PartitionMeta(checkpointId,
ioRegistry.resolve(bufferAddr), bufferAddr);
+ } catch (IgniteInternalDataIntegrityViolationException e) {
+ LOG.info(() -> "Error reading partition meta page, will be
recreated: " + groupPartitionId, e);
}
+ }
- // Creates and writes a partition meta.
- PartitionMetaIo io = PartitionMetaIo.VERSIONS.latest();
+ // Creates and writes a partition meta.
+ PartitionMetaIo io = PartitionMetaIo.VERSIONS.latest();
- io.initNewPage(bufferAddr, partitionMetaPageId, pageSize);
+ io.initNewPage(bufferAddr, partitionMetaPageId, pageSize);
- // Because we will now write this page.
- io.setPageCount(bufferAddr, 1);
+ // Because we will now write this page.
+ io.setPageCount(bufferAddr, 1);
- int pageIdx = filePageStore.allocatePage();
+ int pageIdx = filePageStore.allocatePage();
- assert pageIdx == 0 : pageIdx;
+ assert pageIdx == 0 : pageIdx;
- filePageStore.write(partitionMetaPageId, buffer.rewind(), true);
+ filePageStore.write(partitionMetaPageId, buffer.rewind(), true);
- filePageStore.sync();
+ filePageStore.sync();
- return new PartitionMeta(checkpointId, io, bufferAddr);
- } finally {
- freeBuffer(buffer);
- }
+ return new PartitionMeta(checkpointId, io, bufferAddr);
}
/**
diff --git
a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStoreManager.java
b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStoreManager.java
index 502b139c5c..831d10ced8 100644
---
a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStoreManager.java
+++
b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStoreManager.java
@@ -24,8 +24,6 @@ import static java.util.stream.Collectors.toList;
import static
org.apache.ignite.internal.pagememory.PageIdAllocator.MAX_PARTITION_ID;
import static org.apache.ignite.internal.pagememory.util.PageIdUtils.pageId;
import static
org.apache.ignite.internal.pagememory.util.PageIdUtils.partitionId;
-import static org.apache.ignite.internal.util.GridUnsafe.allocateBuffer;
-import static org.apache.ignite.internal.util.GridUnsafe.freeBuffer;
import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
import java.io.File;
@@ -50,7 +48,6 @@ import
org.apache.ignite.internal.pagememory.persistence.GroupPartitionId;
import org.apache.ignite.internal.pagememory.persistence.PageReadWriteManager;
import
org.apache.ignite.internal.pagememory.persistence.store.GroupPageStoresMap.GroupPartitionPageStore;
import
org.apache.ignite.internal.pagememory.persistence.store.LongOperationAsyncExecutor.RunnableX;
-import org.apache.ignite.internal.util.IgniteStripedLock;
import org.apache.ignite.internal.util.IgniteUtils;
import org.jetbrains.annotations.Nullable;
@@ -97,20 +94,12 @@ public class FilePageStoreManager implements
PageReadWriteManager {
/** Starting directory for all file page stores, for example:
'db/group-123/index.bin'. */
private final Path dbDir;
- /** Page size in bytes. */
- private final int pageSize;
-
/** Executor to disallow running code that modifies data in {@link
#groupPageStores} concurrently with cleanup of file page store. */
private final LongOperationAsyncExecutor cleanupAsyncExecutor;
/** Mapping: group ID -> group page stores. */
private final GroupPageStoresMap<FilePageStore> groupPageStores;
- /** Striped lock for partition initialization. */
- private final IgniteStripedLock initPartitionStripedLock = new
IgniteStripedLock(
- Math.max(Runtime.getRuntime().availableProcessors(), 8)
- );
-
/** {@link FilePageStore} factory. */
private final FilePageStoreFactory filePageStoreFactory;
@@ -131,7 +120,6 @@ public class FilePageStoreManager implements
PageReadWriteManager {
int pageSize
) throws IgniteInternalCheckedException {
this.dbDir = storagePath.resolve("db");
- this.pageSize = pageSize;
cleanupAsyncExecutor = new
LongOperationAsyncExecutor(igniteInstanceName, LOG);
@@ -271,48 +259,6 @@ public class FilePageStoreManager implements
PageReadWriteManager {
}
}
- /**
- * Initialization of the page storage for the group partition.
- *
- * @param groupPartitionId Pair of group ID with partition ID.
- * @throws IgniteInternalCheckedException If failed.
- */
- public void initialize(GroupPartitionId groupPartitionId) throws
IgniteInternalCheckedException {
- initPartitionStripedLock.lock(groupPartitionId.hashCode());
-
- try {
- if (!groupPageStores.contains(groupPartitionId)) {
- Path tableWorkDir =
ensureGroupWorkDir(groupPartitionId.getGroupId());
-
- ByteBuffer buffer = allocateBuffer(pageSize);
-
- try {
- Path partFilePath =
tableWorkDir.resolve(String.format(PART_FILE_TEMPLATE,
groupPartitionId.getPartitionId()));
-
- Path[] partDeltaFiles =
findPartitionDeltaFiles(tableWorkDir, groupPartitionId.getPartitionId());
-
- FilePageStore filePageStore =
filePageStoreFactory.createPageStore(buffer.rewind(), partFilePath,
partDeltaFiles);
-
- FilePageStore previous =
groupPageStores.put(groupPartitionId, filePageStore);
-
- assert previous == null : IgniteStringFormatter.format(
- "Parallel creation is not allowed: [tableId={},
partitionId={}]",
- groupPartitionId.getGroupId(),
- groupPartitionId.getPartitionId()
- );
- } finally {
- freeBuffer(buffer);
- }
- }
- } catch (IgniteInternalCheckedException e) {
- // TODO: IGNITE-16899 By analogy with 2.0, fail a node
-
- throw e;
- } finally {
- initPartitionStripedLock.unlock(groupPartitionId.hashCode());
- }
- }
-
/**
* Returns view for all page stores of all groups.
*/
@@ -324,7 +270,7 @@ public class FilePageStoreManager implements
PageReadWriteManager {
* Returns partition file page store for the corresponding parameters.
*
* @param groupPartitionId Pair of group ID with partition ID.
- * @return Partition file page, {@code null} if not initialized or has
been removed.
+ * @return Partition file page store, {@code null} if not initialized or
has been removed.
*/
public @Nullable FilePageStore getStore(GroupPartitionId groupPartitionId)
{
return groupPageStores.get(groupPartitionId);
@@ -488,4 +434,43 @@ public class FilePageStoreManager implements
PageReadWriteManager {
delete(partitionDeleteFilePath);
}, "destroy-group-" + groupPartitionId.getGroupId() + "-partition-" +
groupPartitionId.getPartitionId());
}
+
+ /**
+ * Reads a partition file page store from the file system with its delta
files if it exists, otherwise creates a new one but without
+ * saving it to the file system.
+ *
+ * <p>Also does not initialize the storage, i.e. does not call {@link
FilePageStore#ensure()}.</p>
+ *
+ * @param groupPartitionId Pair of group ID with partition ID.
+ * @param readBuffer Buffer for reading file headers and other supporting
information from files.
+ */
+ public FilePageStore readOrCreateStore(
+ GroupPartitionId groupPartitionId,
+ ByteBuffer readBuffer
+ ) throws IgniteInternalCheckedException {
+ Path tableWorkDir = ensureGroupWorkDir(groupPartitionId.getGroupId());
+
+ Path partFilePath =
tableWorkDir.resolve(String.format(PART_FILE_TEMPLATE,
groupPartitionId.getPartitionId()));
+
+ Path[] partDeltaFiles = findPartitionDeltaFiles(tableWorkDir,
groupPartitionId.getPartitionId());
+
+ return filePageStoreFactory.createPageStore(readBuffer.rewind(),
partFilePath, partDeltaFiles);
+ }
+
+ /**
+ * Adds a partition file page storage.
+ *
+ * <p>It is expected that the storage has not been added previously and is
also ready to be used by other components such as checkpoint
+ * or delta file compactor.</p>
+ *
+ * @param groupPartitionId Pair of group ID with partition ID.
+ * @param filePageStore Partition file page store.
+ */
+ public void addStore(GroupPartitionId groupPartitionId, FilePageStore
filePageStore) {
+ groupPageStores.compute(groupPartitionId, oldFilePageStore -> {
+ assert oldFilePageStore == null : groupPartitionId;
+
+ return filePageStore;
+ });
+ }
}
diff --git
a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/store/GroupPageStoresMap.java
b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/store/GroupPageStoresMap.java
index 17a7fa2de1..0554d47378 100644
---
a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/store/GroupPageStoresMap.java
+++
b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/store/GroupPageStoresMap.java
@@ -19,6 +19,7 @@ package
org.apache.ignite.internal.pagememory.persistence.store;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
import java.util.stream.Stream;
import org.apache.ignite.internal.pagememory.persistence.GroupPartitionId;
import org.jetbrains.annotations.Nullable;
@@ -99,6 +100,17 @@ public class GroupPageStoresMap<T extends PageStore> {
groupPartitionIdPageStore.clear();
}
+ /**
+ * Attempts to compute a mapping for the group partition and its current
mapped value (or {@code null} if there is no current mapping).
+ * The entire method invocation is performed atomically.
+ *
+ * @param groupPartitionId Pair of group ID with partition ID.
+ * @param remappingFunction Function to compute a value.
+ */
+ public T compute(GroupPartitionId groupPartitionId, Function<? super T, ?
extends T> remappingFunction) {
+ return groupPartitionIdPageStore.compute(groupPartitionId,
(groupPartitionId1, t) -> remappingFunction.apply(t));
+ }
+
/**
* Group partition page storage.
*/
diff --git
a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/PartitionMetaManagerTest.java
b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/PartitionMetaManagerTest.java
index 9ebcc03853..d511d47737 100644
---
a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/PartitionMetaManagerTest.java
+++
b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/PartitionMetaManagerTest.java
@@ -98,7 +98,7 @@ public class PartitionMetaManagerTest extends
BaseIgniteAbstractTest {
try {
// Check for an empty file.
try (FilePageStore filePageStore =
createFilePageStore(testFilePath)) {
- PartitionMeta meta = manager.readOrCreateMeta(null, partId,
filePageStore);
+ PartitionMeta meta = readOrCreateMeta(manager, partId,
filePageStore);
assertEquals(0, meta.lastAppliedIndex());
assertEquals(0, meta.lastAppliedTerm());
@@ -125,7 +125,7 @@ public class PartitionMetaManagerTest extends
BaseIgniteAbstractTest {
// Check not empty file.
try (FilePageStore filePageStore =
createFilePageStore(testFilePath)) {
- PartitionMeta meta = manager.readOrCreateMeta(null, partId,
filePageStore);
+ PartitionMeta meta = readOrCreateMeta(manager, partId,
filePageStore);
assertEquals(50, meta.lastAppliedIndex());
assertEquals(10, meta.lastAppliedTerm());
@@ -152,7 +152,7 @@ public class PartitionMetaManagerTest extends
BaseIgniteAbstractTest {
deltaFilePageStoreIo.sync();
- PartitionMeta meta = manager.readOrCreateMeta(null, partId,
filePageStore);
+ PartitionMeta meta = readOrCreateMeta(manager, partId,
filePageStore);
assertEquals(100, meta.lastAppliedIndex());
assertEquals(10, meta.lastAppliedTerm());
@@ -174,7 +174,7 @@ public class PartitionMetaManagerTest extends
BaseIgniteAbstractTest {
fileIo.writeFully(buffer.rewind(), filePageStore.headerSize());
- PartitionMeta meta = manager.readOrCreateMeta(null, partId,
filePageStore);
+ PartitionMeta meta = readOrCreateMeta(manager, partId,
filePageStore);
assertEquals(0, meta.lastAppliedIndex());
assertEquals(0, meta.lastAppliedTerm());
@@ -216,4 +216,18 @@ public class PartitionMetaManagerTest extends
BaseIgniteAbstractTest {
return filePageStore;
}
+
+ private static PartitionMeta readOrCreateMeta(
+ PartitionMetaManager manager,
+ GroupPartitionId partId,
+ FilePageStore filePageStore
+ ) throws Exception {
+ ByteBuffer buffer = allocateBuffer(PAGE_SIZE);
+
+ try {
+ return manager.readOrCreateMeta(null, partId, filePageStore,
buffer);
+ } finally {
+ freeBuffer(buffer);
+ }
+ }
}
diff --git
a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/PersistentPageMemoryNoLoadTest.java
b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/PersistentPageMemoryNoLoadTest.java
index 29162f0fd5..0ee440c756 100644
---
a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/PersistentPageMemoryNoLoadTest.java
+++
b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/PersistentPageMemoryNoLoadTest.java
@@ -22,8 +22,10 @@ import static
org.apache.ignite.internal.pagememory.persistence.PersistentPageMe
import static
org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointState.FINISHED;
import static
org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointState.PAGES_SORTED;
import static
org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointTestUtils.mockCheckpointTimeoutLock;
-import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
+import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.util.Constants.MiB;
+import static org.apache.ignite.internal.util.GridUnsafe.allocateBuffer;
+import static org.apache.ignite.internal.util.GridUnsafe.freeBuffer;
import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.empty;
@@ -365,7 +367,7 @@ public class PersistentPageMemoryNoLoadTest extends
AbstractPageMemoryNoLoadSelf
doAnswer(answer -> {
startWriteMetaToBufferFuture.complete(null);
- await(finishWaitWriteMetaToBufferFuture, 1, SECONDS);
+ assertThat(finishWaitWriteMetaToBufferFuture,
willCompleteSuccessfully());
return answer.callRealMethod();
})
@@ -497,13 +499,15 @@ public class PersistentPageMemoryNoLoadTest extends
AbstractPageMemoryNoLoadSelf
checkpointManager.checkpointTimeoutLock().checkpointReadLock();
+ ByteBuffer buffer = null;
+
try {
+ buffer = allocateBuffer(PAGE_SIZE);
+
for (int partition = 0; partition < partitions; partition++) {
GroupPartitionId groupPartitionId = new
GroupPartitionId(GRP_ID, partition);
- filePageStoreManager.initialize(groupPartitionId);
-
- FilePageStore filePageStore =
filePageStoreManager.getStore(groupPartitionId);
+ FilePageStore filePageStore =
filePageStoreManager.readOrCreateStore(groupPartitionId, buffer.rewind());
filePageStore.ensure();
@@ -512,11 +516,10 @@ public class PersistentPageMemoryNoLoadTest extends
AbstractPageMemoryNoLoadSelf
PartitionMeta partitionMeta =
partitionMetaManager.readOrCreateMeta(
lastCheckpointProgress == null ? null :
lastCheckpointProgress.id(),
groupPartitionId,
- filePageStore
+ filePageStore,
+ buffer.rewind()
);
- partitionMetaManager.addMeta(groupPartitionId, partitionMeta);
-
filePageStore.setPageAllocationListener(pageIdx -> {
assert
checkpointManager.checkpointTimeoutLock().checkpointLockIsHeldByThread();
@@ -526,9 +529,17 @@ public class PersistentPageMemoryNoLoadTest extends
AbstractPageMemoryNoLoadSelf
});
filePageStore.pages(partitionMeta.pageCount());
+
+ filePageStoreManager.addStore(groupPartitionId, filePageStore);
+ partitionMetaManager.addMeta(groupPartitionId, partitionMeta);
}
} finally {
- checkpointManager.checkpointTimeoutLock().checkpointReadUnlock();
+ ByteBuffer bufferToClose = buffer;
+
+ closeAll(
+ bufferToClose == null ? null : () ->
freeBuffer(bufferToClose),
+ () ->
checkpointManager.checkpointTimeoutLock().checkpointReadUnlock()
+ );
}
}
}
diff --git
a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStoreManagerTest.java
b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStoreManagerTest.java
index 0ee0eb377d..fa16fda64e 100644
---
a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStoreManagerTest.java
+++
b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStoreManagerTest.java
@@ -28,8 +28,12 @@ import static
org.apache.ignite.internal.pagememory.persistence.store.FilePageSt
import static
org.apache.ignite.internal.pagememory.persistence.store.FilePageStoreManager.TMP_PART_DELTA_FILE_TEMPLATE;
import static
org.apache.ignite.internal.pagememory.persistence.store.FilePageStoreManager.findPartitionDeltaFiles;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.runRace;
+import static org.apache.ignite.internal.util.GridUnsafe.allocateBuffer;
+import static org.apache.ignite.internal.util.GridUnsafe.freeBuffer;
+import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.arrayContainingInAnyOrder;
+import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.empty;
@@ -39,16 +43,22 @@ import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
-import java.util.Iterator;
+import java.util.Collection;
import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import org.apache.ignite.internal.fileio.RandomAccessFileIoFactory;
@@ -56,10 +66,12 @@ import
org.apache.ignite.internal.lang.IgniteInternalCheckedException;
import org.apache.ignite.internal.pagememory.persistence.GroupPartitionId;
import
org.apache.ignite.internal.pagememory.persistence.store.GroupPageStoresMap.GroupPartitionPageStore;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
import org.apache.ignite.internal.testframework.WorkDirectory;
import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
import org.apache.ignite.internal.util.ArrayUtils;
import org.apache.ignite.internal.util.IgniteUtils;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
@@ -72,12 +84,21 @@ import org.junit.jupiter.params.provider.ValueSource;
*/
@ExtendWith(WorkDirectoryExtension.class)
public class FilePageStoreManagerTest extends BaseIgniteAbstractTest {
+ private static final int PAGE_SIZE = 1024;
+
/** To be used in a loop. {@link RepeatedTest} cannot be combined with
{@link ParameterizedTest}. */
private static final int REPEATS = 100;
@WorkDirectory
private Path workDir;
+ private final Collection<FilePageStoreManager> managers = new
ConcurrentLinkedQueue<>();
+
+ @AfterEach
+ void tearDown() throws Exception {
+ closeAll(managers.stream().filter(Objects::nonNull).map(manager ->
manager::stop));
+ }
+
@Test
void testCreateManager() {
// Checks if the manager was successfully created.
@@ -109,13 +130,14 @@ public class FilePageStoreManagerTest extends
BaseIgniteAbstractTest {
IgniteInternalCheckedException exception =
assertThrows(IgniteInternalCheckedException.class, manager::start);
assertThat(exception.getMessage(), containsString("Could not create
work directory for page stores"));
-
}
@Test
- void testInitialize() throws Exception {
+ void testReadOrCreateStore() throws Exception {
FilePageStoreManager manager = createManager();
+ ByteBuffer buffer = allocateBuffer(PAGE_SIZE);
+
try {
Files.createDirectories(workDir.resolve("db"));
@@ -125,19 +147,21 @@ public class FilePageStoreManagerTest extends
BaseIgniteAbstractTest {
GroupPartitionId groupPartitionId00 = new GroupPartitionId(0, 0);
- IgniteInternalCheckedException exception = assertThrows(
+ IgniteTestUtils.assertThrows(
IgniteInternalCheckedException.class,
- () -> manager.initialize(groupPartitionId00)
+ () -> manager.readOrCreateStore(groupPartitionId00,
buffer.rewind()),
+ "Failed to initialize group working directory"
);
- assertThat(exception.getMessage(), containsString("Failed to
initialize group working directory"));
-
Files.delete(testGroupDir);
GroupPartitionId groupPartitionId01 = new GroupPartitionId(0, 1);
- assertDoesNotThrow(() -> manager.initialize(groupPartitionId00));
- assertDoesNotThrow(() -> manager.initialize(groupPartitionId01));
+ FilePageStore filePageStore0 =
manager.readOrCreateStore(groupPartitionId00, buffer.rewind());
+ FilePageStore filePageStore1 =
manager.readOrCreateStore(groupPartitionId01, buffer.rewind());
+
+ assertNull(manager.getStore(groupPartitionId00));
+ assertNull(manager.getStore(groupPartitionId01));
assertTrue(Files.isDirectory(testGroupDir));
@@ -145,10 +169,8 @@ public class FilePageStoreManagerTest extends
BaseIgniteAbstractTest {
assertThat(files.count(), is(0L));
}
- Iterator<GroupPartitionPageStore<FilePageStore>> iterator =
manager.allPageStores().iterator();
- while (iterator.hasNext()) {
- iterator.next().pageStore().ensure();
- }
+ filePageStore0.ensure();
+ filePageStore1.ensure();
try (Stream<Path> files = Files.list(testGroupDir)) {
assertThat(
@@ -157,7 +179,7 @@ public class FilePageStoreManagerTest extends
BaseIgniteAbstractTest {
);
}
} finally {
- manager.stop();
+ freeBuffer(buffer);
}
}
@@ -170,7 +192,7 @@ public class FilePageStoreManagerTest extends
BaseIgniteAbstractTest {
try {
GroupPartitionId groupPartitionId00 = new GroupPartitionId(0, 0);
- manager0.initialize(groupPartitionId00);
+ createAndAddFilePageStore(manager0, groupPartitionId00);
manager0.getStore(groupPartitionId00).ensure();
@@ -194,7 +216,7 @@ public class FilePageStoreManagerTest extends
BaseIgniteAbstractTest {
try {
GroupPartitionId groupPartitionId10 = new GroupPartitionId(1, 0);
- manager1.initialize(groupPartitionId10);
+ createAndAddFilePageStore(manager1, groupPartitionId10);
manager1.getStore(groupPartitionId10).ensure();
@@ -216,8 +238,8 @@ public class FilePageStoreManagerTest extends
BaseIgniteAbstractTest {
GroupPartitionId groupPartitionId10 = new GroupPartitionId(1, 0);
GroupPartitionId groupPartitionId20 = new GroupPartitionId(2, 0);
- manager.initialize(groupPartitionId10);
- manager.initialize(groupPartitionId20);
+ createAndAddFilePageStore(manager, groupPartitionId10);
+ createAndAddFilePageStore(manager, groupPartitionId20);
Path grpDir0 = workDir.resolve("db/table-1");
Path grpDir1 = workDir.resolve("db/table-2");
@@ -247,8 +269,8 @@ public class FilePageStoreManagerTest extends
BaseIgniteAbstractTest {
GroupPartitionId groupPartitionId10 = new GroupPartitionId(1, 0);
GroupPartitionId groupPartitionId20 = new GroupPartitionId(2, 0);
- manager.initialize(groupPartitionId10);
- manager.initialize(groupPartitionId20);
+ createAndAddFilePageStore(manager, groupPartitionId10);
+ createAndAddFilePageStore(manager, groupPartitionId20);
Path grpDir0 = workDir.resolve("db/table-1");
Path grpDir1 = workDir.resolve("db/table-2");
@@ -328,8 +350,8 @@ public class FilePageStoreManagerTest extends
BaseIgniteAbstractTest {
manager.start();
- manager.initialize(new GroupPartitionId(1, 0));
- manager.initialize(new GroupPartitionId(2, 0));
+ createAndAddFilePageStore(manager, new GroupPartitionId(1, 0));
+ createAndAddFilePageStore(manager, new GroupPartitionId(2, 0));
List<Path> allPageStoreFiles = manager.allPageStores()
.map(GroupPartitionPageStore::pageStore)
@@ -354,8 +376,8 @@ public class FilePageStoreManagerTest extends
BaseIgniteAbstractTest {
GroupPartitionId groupPartitionId00 = new GroupPartitionId(0, 0);
GroupPartitionId groupPartitionId10 = new GroupPartitionId(1, 0);
- manager.initialize(groupPartitionId00);
- manager.initialize(groupPartitionId10);
+ createAndAddFilePageStore(manager, groupPartitionId00);
+ createAndAddFilePageStore(manager, groupPartitionId10);
FilePageStore filePageStore0 = manager.getStore(groupPartitionId00);
FilePageStore filePageStore1 = manager.getStore(groupPartitionId10);
@@ -399,10 +421,10 @@ public class FilePageStoreManagerTest extends
BaseIgniteAbstractTest {
GroupPartitionId groupPartitionId10 = new GroupPartitionId(1, 0);
GroupPartitionId groupPartitionId11 = new GroupPartitionId(1, 1);
- manager.initialize(groupPartitionId00);
- manager.initialize(groupPartitionId01);
- manager.initialize(groupPartitionId10);
- manager.initialize(groupPartitionId11);
+ createAndAddFilePageStore(manager, groupPartitionId00);
+ createAndAddFilePageStore(manager, groupPartitionId01);
+ createAndAddFilePageStore(manager, groupPartitionId10);
+ createAndAddFilePageStore(manager, groupPartitionId11);
FilePageStore filePageStore00 = manager.getStore(groupPartitionId00);
FilePageStore filePageStore01 = manager.getStore(groupPartitionId01);
@@ -480,8 +502,34 @@ public class FilePageStoreManagerTest extends
BaseIgniteAbstractTest {
}
}
+ @Test
+ void testAddStore() throws Exception {
+ FilePageStoreManager manager = createManager();
+
+ GroupPartitionId groupPartitionId0 = new GroupPartitionId(0, 0);
+ GroupPartitionId groupPartitionId1 = new GroupPartitionId(0, 1);
+ GroupPartitionId groupPartitionId2 = new GroupPartitionId(1, 0);
+
+ FilePageStore filePageStore = mock(FilePageStore.class);
+
+ manager.addStore(groupPartitionId0, filePageStore);
+
+ assertSame(filePageStore, manager.getStore(groupPartitionId0));
+ assertNull(manager.getStore(groupPartitionId1));
+ assertNull(manager.getStore(groupPartitionId2));
+
+ assertThat(
+
manager.allPageStores().map(GroupPartitionPageStore::pageStore).collect(toList()),
+ contains(filePageStore)
+ );
+ }
+
private FilePageStoreManager createManager() throws Exception {
- return new FilePageStoreManager("test", workDir, new
RandomAccessFileIoFactory(), 1024);
+ FilePageStoreManager manager = new FilePageStoreManager("test",
workDir, new RandomAccessFileIoFactory(), PAGE_SIZE);
+
+ managers.add(manager);
+
+ return manager;
}
private static List<Path> collectFilesOnly(Path start) throws Exception {
@@ -519,4 +567,16 @@ public class FilePageStoreManagerTest extends
BaseIgniteAbstractTest {
assertTrue(IgniteUtils.deleteIfExists(tmpDeltaFile),
tmpDeltaFile.toString());
}
}
+
+ private static void createAndAddFilePageStore(FilePageStoreManager
manager, GroupPartitionId groupPartitionId) throws Exception {
+ ByteBuffer buffer = allocateBuffer(PAGE_SIZE);
+
+ try {
+ FilePageStore filePageStore =
manager.readOrCreateStore(groupPartitionId, buffer);
+
+ manager.addStore(groupPartitionId, filePageStore);
+ } finally {
+ freeBuffer(buffer);
+ }
+ }
}
diff --git
a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java
b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java
index c39a114062..9114c245ea 100644
---
a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java
+++
b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java
@@ -18,7 +18,10 @@
package org.apache.ignite.internal.storage.pagememory;
import static org.apache.ignite.internal.pagememory.PageIdAllocator.FLAG_AUX;
+import static org.apache.ignite.internal.util.GridUnsafe.allocateBuffer;
+import static org.apache.ignite.internal.util.GridUnsafe.freeBuffer;
+import java.nio.ByteBuffer;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
@@ -132,41 +135,6 @@ public class PersistentPageMemoryTableStorage extends
AbstractPageMemoryTableSto
});
}
- /**
- * Initializes the partition file page store if it hasn't already.
- *
- * @param groupPartitionId Pair of group ID with partition ID.
- * @return Partition file page store.
- * @throws StorageException If failed.
- */
- private FilePageStore ensurePartitionFilePageStoreExists(GroupPartitionId
groupPartitionId) throws StorageException {
- try {
- dataRegion.filePageStoreManager().initialize(groupPartitionId);
-
- FilePageStore filePageStore =
dataRegion.filePageStoreManager().getStore(groupPartitionId);
-
- assert !filePageStore.isMarkedToDestroy() :
IgniteStringFormatter.format(
- "Should not be marked for deletion: [tableId={},
partitionId={}]",
- groupPartitionId.getGroupId(),
- groupPartitionId.getPartitionId()
- );
-
- filePageStore.ensure();
-
- if (filePageStore.deltaFileCount() > 0) {
- dataRegion.checkpointManager().triggerCompaction();
- }
-
- return filePageStore;
- } catch (IgniteInternalCheckedException e) {
- throw new StorageException(
- "Error initializing file page store: [tableId={},
partitionId={}]",
- e,
- groupPartitionId.getGroupId(),
groupPartitionId.getPartitionId()
- );
- }
- }
-
/**
* Returns id of the last started checkpoint, or {@code null} if no
checkpoints were started yet.
*/
@@ -458,47 +426,90 @@ public class PersistentPageMemoryTableStorage extends
AbstractPageMemoryTableSto
}
/**
- * Creates or gets partition meta from a file.
+ * Creates or receives partition meta from a file.
*
- * <p>Safe to use without a checkpointReadLock as we read the meta
directly without using {@link PageMemory}.
+ * <p>Safe to use without a checkpointReadLock as we read the meta
directly without using {@link PageMemory}.</p>
*
* @param groupPartitionId Partition of the group.
- * @param filePageStore Partition file page store.
*/
- private PartitionMeta getOrCreatePartitionMeta(GroupPartitionId
groupPartitionId, FilePageStore filePageStore) {
+ private PartitionMeta
getOrCreatePartitionMetaOnCreatePartition(GroupPartitionId groupPartitionId) {
+ ByteBuffer buffer = allocateBuffer(dataRegion.pageMemory().pageSize());
+
try {
- PartitionMeta meta =
dataRegion.partitionMetaManager().readOrCreateMeta(lastCheckpointId(),
groupPartitionId, filePageStore);
+ FilePageStore filePageStore =
dataRegion.filePageStoreManager().getStore(groupPartitionId);
+
+ // TODO: IGNITE-20983 This shouldn't happen, we should read the
page store and its meta again
+ if (filePageStore != null) {
+ PartitionMeta partitionMeta =
dataRegion.partitionMetaManager().getMeta(groupPartitionId);
+
+ assert partitionMeta != null : groupPartitionId;
- dataRegion.partitionMetaManager().addMeta(groupPartitionId, meta);
+ return partitionMeta;
+ }
+
+ filePageStore = readOrCreateAndInitFilePageStore(groupPartitionId,
buffer);
+
+ PartitionMeta partitionMeta =
readOrCreatePartitionMeta(groupPartitionId, filePageStore, buffer.rewind());
- filePageStore.pages(meta.pageCount());
+ filePageStore.pages(partitionMeta.pageCount());
filePageStore.setPageAllocationListener(pageIdx -> {
assert
dataRegion.checkpointManager().checkpointTimeoutLock().checkpointLockIsHeldByThread();
- meta.incrementPageCount(lastCheckpointId());
+ partitionMeta.incrementPageCount(lastCheckpointId());
});
- return meta;
+ dataRegion.filePageStoreManager().addStore(groupPartitionId,
filePageStore);
+ dataRegion.partitionMetaManager().addMeta(groupPartitionId,
partitionMeta);
+
+ if (filePageStore.deltaFileCount() > 0) {
+ dataRegion.checkpointManager().triggerCompaction();
+ }
+
+ return partitionMeta;
+ } finally {
+ freeBuffer(buffer);
+ }
+ }
+
+ private FilePageStore readOrCreateAndInitFilePageStore(
+ GroupPartitionId groupPartitionId,
+ ByteBuffer buffer
+ ) throws StorageException {
+ try {
+ FilePageStore filePageStore =
dataRegion.filePageStoreManager().readOrCreateStore(groupPartitionId, buffer);
+
+ assert !filePageStore.isMarkedToDestroy() :
IgniteStringFormatter.format(
+ "Should not be marked for deletion: [tableId={},
partitionId={}]",
+ groupPartitionId.getGroupId(),
+ groupPartitionId.getPartitionId()
+ );
+
+ filePageStore.ensure();
+
+ return filePageStore;
} catch (IgniteInternalCheckedException e) {
throw new StorageException(
- "Error reading or creating partition meta information:
[tableId={}, partitionId={}]",
+ "Error read and initializing file page store: [tableId={},
partitionId={}]",
e,
- getTableId(), groupPartitionId.getPartitionId()
+ groupPartitionId.getGroupId(),
groupPartitionId.getPartitionId()
);
}
}
- /**
- * Creates or receives partition meta from a file.
- *
- * <p>Safe to use without a checkpointReadLock as we read the meta
directly without using {@link PageMemory}.
- *
- * @param groupPartitionId Partition of the group.
- */
- private PartitionMeta
getOrCreatePartitionMetaOnCreatePartition(GroupPartitionId groupPartitionId) {
- FilePageStore filePageStore =
ensurePartitionFilePageStoreExists(groupPartitionId);
-
- return getOrCreatePartitionMeta(groupPartitionId, filePageStore);
+ private PartitionMeta readOrCreatePartitionMeta(
+ GroupPartitionId groupPartitionId,
+ FilePageStore filePageStore,
+ ByteBuffer buffer
+ ) throws StorageException {
+ try {
+ return
dataRegion.partitionMetaManager().readOrCreateMeta(lastCheckpointId(),
groupPartitionId, filePageStore, buffer);
+ } catch (IgniteInternalCheckedException e) {
+ throw new StorageException(
+ "Error reading or creating partition meta information:
[tableId={}, partitionId={}]",
+ e,
+ getTableId(), groupPartitionId.getPartitionId()
+ );
+ }
}
}
