You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ib...@apache.org on 2022/07/07 10:57:59 UTC
[ignite-3] branch main updated: IGNITE-17008 Add support for NextLink field in VersionChain (#886)
This is an automated email from the ASF dual-hosted git repository.
ibessonov 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 173c799bf IGNITE-17008 Add support for NextLink field in VersionChain (#886)
173c799bf is described below
commit 173c799bf48556899bd962c376edb54401d22866
Author: Roman Puchkovskiy <ro...@gmail.com>
AuthorDate: Thu Jul 7 14:57:55 2022 +0400
IGNITE-17008 Add support for NextLink field in VersionChain (#886)
---
.../mv/PageMemoryMvPartitionStorage.java | 65 +++++++++++++++++-----
...adLatestRowVersion.java => ReadRowVersion.java} | 4 +-
.../storage/pagememory/mv/VersionChain.java | 38 +++++++++++--
.../pagememory/mv/VersionChainDataPageReader.java | 5 +-
.../pagememory/mv/io/VersionChainDataIo.java | 4 +-
.../mv/PageMemoryMvPartitionStorageTest.java | 51 ++++++++++++++++-
6 files changed, 140 insertions(+), 27 deletions(-)
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PageMemoryMvPartitionStorage.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PageMemoryMvPartitionStorage.java
index 5f4163b6c..a59217468 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PageMemoryMvPartitionStorage.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PageMemoryMvPartitionStorage.java
@@ -54,6 +54,7 @@ public class PageMemoryMvPartitionStorage implements MvPartitionStorage {
private static final Predicate<BinaryRow> MATCH_ALL = row -> true;
private static final Predicate<Timestamp> ALWAYS_LOAD_VALUE = timestamp -> true;
+ private static final Predicate<Timestamp> NEVER_LOAD_VALUE = timestamp -> false;
private static final Predicate<Timestamp> LOAD_VALUE_WHEN_UNCOMMITTED = RowVersion::isUncommitted;
private final int partId;
@@ -65,7 +66,7 @@ public class PageMemoryMvPartitionStorage implements MvPartitionStorage {
private final RowVersionFreeList rowVersionFreeList;
private final DataPageReader rowVersionDataPageReader;
- private final ThreadLocal<ReadLatestRowVersion> readLatestRowVersionCache = ThreadLocal.withInitial(ReadLatestRowVersion::new);
+ private final ThreadLocal<ReadRowVersion> readRowVersionCache = ThreadLocal.withInitial(ReadRowVersion::new);
private final ThreadLocal<ScanVersionChainByTimestamp> scanVersionChainByTimestampCache = ThreadLocal.withInitial(
ScanVersionChainByTimestamp::new
);
@@ -179,7 +180,11 @@ public class PageMemoryMvPartitionStorage implements MvPartitionStorage {
private RowVersion findLatestRowVersion(VersionChain versionChain, Predicate<Timestamp> loadValue) {
long nextLink = PartitionlessLinks.addPartitionIdToPartititionlessLink(versionChain.headLink(), partId);
- ReadLatestRowVersion read = freshReadLatestRowVersion();
+ return readRowVersion(nextLink, loadValue);
+ }
+
+ private RowVersion readRowVersion(long nextLink, Predicate<Timestamp> loadValue) {
+ ReadRowVersion read = freshReadRowVersion();
try {
rowVersionDataPageReader.traverse(nextLink, read, loadValue);
@@ -190,8 +195,8 @@ public class PageMemoryMvPartitionStorage implements MvPartitionStorage {
return read.result();
}
- private ReadLatestRowVersion freshReadLatestRowVersion() {
- ReadLatestRowVersion traversal = readLatestRowVersionCache.get();
+ private ReadRowVersion freshReadRowVersion() {
+ ReadRowVersion traversal = readRowVersionCache.get();
traversal.reset();
return traversal;
}
@@ -227,13 +232,17 @@ public class PageMemoryMvPartitionStorage implements MvPartitionStorage {
}
private @Nullable ByteBufferRow findRowVersionByTimestamp(VersionChain versionChain, Timestamp timestamp) {
- long nextRowPartitionlessLink = versionChain.headLink();
- long nextLink = PartitionlessLinks.addPartitionIdToPartititionlessLink(nextRowPartitionlessLink, partId);
+ if (!versionChain.hasCommittedVersions()) {
+ return null;
+ }
+
+ long newestCommittedRowPartitionlessLink = versionChain.newestCommittedPartitionlessLink();
+ long newestCommittedLink = PartitionlessLinks.addPartitionIdToPartititionlessLink(newestCommittedRowPartitionlessLink, partId);
ScanVersionChainByTimestamp scanByTimestamp = freshScanByTimestamp();
try {
- rowVersionDataPageReader.traverse(nextLink, scanByTimestamp, timestamp);
+ rowVersionDataPageReader.traverse(newestCommittedLink, scanByTimestamp, timestamp);
} catch (IgniteInternalCheckedException e) {
throw new StorageException("Cannot search for a row version", e);
}
@@ -252,7 +261,12 @@ public class PageMemoryMvPartitionStorage implements MvPartitionStorage {
public LinkRowId insert(BinaryRow row, UUID txId) throws StorageException {
RowVersion rowVersion = insertRowVersion(Objects.requireNonNull(row), RowVersion.NULL_LINK);
- VersionChain versionChain = new VersionChain(partId, txId, PartitionlessLinks.removePartitionIdFromLink(rowVersion.link()));
+ VersionChain versionChain = new VersionChain(
+ partId,
+ txId,
+ PartitionlessLinks.removePartitionIdFromLink(rowVersion.link()),
+ RowVersion.NULL_LINK
+ );
try {
versionChainFreeList.insertDataRow(versionChain);
@@ -306,7 +320,8 @@ public class PageMemoryMvPartitionStorage implements MvPartitionStorage {
VersionChain chainReplacement = new VersionChain(
partId,
txId,
- PartitionlessLinks.removePartitionIdFromLink(newVersion.link())
+ PartitionlessLinks.removePartitionIdFromLink(newVersion.link()),
+ currentChain.headLink()
);
updateVersionChain(currentChain, chainReplacement);
@@ -336,16 +351,24 @@ public class PageMemoryMvPartitionStorage implements MvPartitionStorage {
return null;
}
- RowVersion currentVersion = findLatestRowVersion(currentVersionChain, ALWAYS_LOAD_VALUE);
- assert currentVersion.isUncommitted();
+ RowVersion latestVersion = findLatestRowVersion(currentVersionChain, ALWAYS_LOAD_VALUE);
+ assert latestVersion.isUncommitted();
+
+ removeRowVersion(latestVersion);
- removeRowVersion(currentVersion);
+ if (latestVersion.hasNextLink()) {
+ // This load can be avoided, see the comment below.
+ RowVersion latestCommittedVersion = readNextInChainOrderHeaderOnly(latestVersion);
- if (currentVersion.hasNextLink()) {
VersionChain versionChainReplacement = VersionChain.withoutTxId(
partId,
currentVersionChain.link(),
- currentVersion.nextLink()
+ latestVersion.nextLink(),
+ // Next can be safely replaced with any value (like -1), because this field is only used when there
+ // is some uncommitted value, but when we add an uncommitted value, we 'fix' such placeholder value
+ // (like -1) by replacing it with a valid value. But it seems that this optimization is not critical
+ // as aborts are pretty rare; let's strive for internal consistency for now and write the correct value.
+ latestCommittedVersion.nextLink()
);
updateVersionChain(currentVersionChain, versionChainReplacement);
} else {
@@ -353,7 +376,19 @@ public class PageMemoryMvPartitionStorage implements MvPartitionStorage {
removeVersionChain(currentVersionChain);
}
- return rowVersionToBinaryRow(currentVersion);
+ return rowVersionToBinaryRow(latestVersion);
+ }
+
+ /**
+ * Reads next row version in chain order (that is, the predecessor of the given version in creation order); payload is not loaded.
+ *
+ * @param rowVersion Version from which to start.
+ * @return Next row version in chain order (that is, the predecessor of the given version in creation order).
+ */
+ private RowVersion readNextInChainOrderHeaderOnly(RowVersion rowVersion) {
+ long preLatestVersionLink = PartitionlessLinks.addPartitionIdToPartititionlessLink(rowVersion.nextLink(), partId);
+
+ return readRowVersion(preLatestVersionLink, NEVER_LOAD_VALUE);
}
private void removeVersionChain(VersionChain currentVersionChain) {
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/ReadLatestRowVersion.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/ReadRowVersion.java
similarity index 94%
rename from modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/ReadLatestRowVersion.java
rename to modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/ReadRowVersion.java
index 363867483..ed410237d 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/ReadLatestRowVersion.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/ReadRowVersion.java
@@ -27,9 +27,9 @@ import org.apache.ignite.internal.tx.Timestamp;
import org.jetbrains.annotations.Nullable;
/**
- * Traversal for reading the latest row version. If the version is uncommitted, returns its value; otherwise, does NOT return it.
+ * Traversal for reading a row version by its link. Loads the version value conditionally.
*/
-class ReadLatestRowVersion implements PageMemoryTraversal<Predicate<Timestamp>> {
+class ReadRowVersion implements PageMemoryTraversal<Predicate<Timestamp>> {
private RowVersion result;
private boolean readingFirstSlot = true;
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VersionChain.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VersionChain.java
index 15c702a9d..c845d5935 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VersionChain.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VersionChain.java
@@ -35,40 +35,50 @@ public class VersionChain extends VersionChainLink implements Storable {
private static final int TRANSACTION_ID_STORE_SIZE_BYTES = 2 * Long.BYTES;
private static final int HEAD_LINK_STORE_SIZE_BYTES = PartitionlessLinks.PARTITIONLESS_LINK_SIZE_BYTES;
+ private static final int NEXT_LINK_STORE_SIZE_BYTES = PartitionlessLinks.PARTITIONLESS_LINK_SIZE_BYTES;
public static final int TRANSACTION_ID_OFFSET = 0;
private final int partitionId;
@Nullable
private final UUID transactionId;
+
+ /**
+ * Link to the latest version.
+ */
private final long headLink;
- // TODO: IGNITE-17008 - add nextLink
+ /**
+ * Link to the pre-latest version ({@link RowVersion#NULL_LINK} if there is just one version).
+ */
+ private final long nextLink;
/**
* Constructs a VersionChain without a transaction ID.
*/
- public static VersionChain withoutTxId(int partitionId, long link, long headLink) {
- return new VersionChain(partitionId, link, null, headLink);
+ public static VersionChain withoutTxId(int partitionId, long link, long headLink, long nextLink) {
+ return new VersionChain(partitionId, link, null, headLink, nextLink);
}
/**
* Constructor.
*/
- public VersionChain(int partitionId, @Nullable UUID transactionId, long headLink) {
+ public VersionChain(int partitionId, @Nullable UUID transactionId, long headLink, long nextLink) {
this.partitionId = partitionId;
this.transactionId = transactionId;
this.headLink = headLink;
+ this.nextLink = nextLink;
}
/**
* Constructor.
*/
- public VersionChain(int partitionId, long link, @Nullable UUID transactionId, long headLink) {
+ public VersionChain(int partitionId, long link, @Nullable UUID transactionId, long headLink, long nextLink) {
super(link);
this.partitionId = partitionId;
this.transactionId = transactionId;
this.headLink = headLink;
+ this.nextLink = nextLink;
}
@Nullable
@@ -80,6 +90,22 @@ public class VersionChain extends VersionChainLink implements Storable {
return headLink;
}
+ public long nextLink() {
+ return nextLink;
+ }
+
+ public long newestCommittedPartitionlessLink() {
+ return isUncommitted() ? nextLink : headLink;
+ }
+
+ private boolean isUncommitted() {
+ return transactionId != null;
+ }
+
+ public boolean hasCommittedVersions() {
+ return newestCommittedPartitionlessLink() != RowVersion.NULL_LINK;
+ }
+
@Override
public final int partition() {
return partitionId;
@@ -87,7 +113,7 @@ public class VersionChain extends VersionChainLink implements Storable {
@Override
public int size() {
- return TRANSACTION_ID_STORE_SIZE_BYTES + HEAD_LINK_STORE_SIZE_BYTES;
+ return TRANSACTION_ID_STORE_SIZE_BYTES + HEAD_LINK_STORE_SIZE_BYTES + NEXT_LINK_STORE_SIZE_BYTES;
}
@Override
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VersionChainDataPageReader.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VersionChainDataPageReader.java
index 2b8a8221f..fabf03d8d 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VersionChainDataPageReader.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VersionChainDataPageReader.java
@@ -65,7 +65,10 @@ class VersionChainDataPageReader extends NonFragmentableDataPageReader<VersionCh
long headLink = PartitionlessLinks.readFromMemory(pageAddr, offset);
offset += PartitionlessLinks.PARTITIONLESS_LINK_SIZE_BYTES;
- return new VersionChain(partitionOfLink(link), link, txId, headLink);
+ long nextLink = PartitionlessLinks.readFromMemory(pageAddr, offset);
+ offset += PartitionlessLinks.PARTITIONLESS_LINK_SIZE_BYTES;
+
+ return new VersionChain(partitionOfLink(link), link, txId, headLink, nextLink);
}
private int partitionOfLink(long link) {
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/io/VersionChainDataIo.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/io/VersionChainDataIo.java
index 7d7618792..90a9bb2a6 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/io/VersionChainDataIo.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/io/VersionChainDataIo.java
@@ -68,6 +68,7 @@ public class VersionChainDataIo extends AbstractDataPageIo<VersionChain> {
addr += TransactionIds.writeTransactionId(addr, 0, row.transactionId());
addr += PartitionlessLinks.writeToMemory(addr, row.headLink());
+ addr += PartitionlessLinks.writeToMemory(addr, row.nextLink());
}
/** {@inheritDoc} */
@@ -75,7 +76,8 @@ public class VersionChainDataIo extends AbstractDataPageIo<VersionChain> {
protected void writeFragmentData(VersionChain row, ByteBuffer buf, int rowOff, int payloadSize) {
assertPageType(buf);
- throw new UnsupportedOperationException("Splitting version chain rows to fragments is ridiculous!");
+ throw new UnsupportedOperationException("Version chains must never be split to fragments, this should be guaranteed "
+ + "by VersionChain#headerSize()");
}
/**
diff --git a/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/mv/PageMemoryMvPartitionStorageTest.java b/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/mv/PageMemoryMvPartitionStorageTest.java
index 4e5ccbe31..11cb5e156 100644
--- a/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/mv/PageMemoryMvPartitionStorageTest.java
+++ b/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/mv/PageMemoryMvPartitionStorageTest.java
@@ -18,7 +18,11 @@
package org.apache.ignite.internal.storage.pagememory.mv;
import static java.util.stream.Collectors.joining;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import java.nio.file.Path;
@@ -56,6 +60,8 @@ import org.junit.jupiter.api.extension.ExtendWith;
class PageMemoryMvPartitionStorageTest extends AbstractMvPartitionStorageTest<PageMemoryMvPartitionStorage> {
private final PageIoRegistry ioRegistry = new PageIoRegistry();
+ private final BinaryRow binaryRow3 = binaryRow(key, new TestValue(22, "bar3"));
+
{
ioRegistry.loadFromServiceLoader();
}
@@ -77,8 +83,6 @@ class PageMemoryMvPartitionStorageTest extends AbstractMvPartitionStorageTest<Pa
private AbstractPageMemoryTableStorage table;
- private int nextPageIndex = 100;
-
@WorkDirectory
private Path workDir;
@@ -212,4 +216,47 @@ class PageMemoryMvPartitionStorageTest extends AbstractMvPartitionStorageTest<Pa
assertRowMatches(foundRow, longRow);
}
}
+
+ @Test
+ void readByTimestampWorksCorrectlyAfterCommitAndAbortFollowedByUncommittedWrite() {
+ RowId rowId = commitAbortAndAddUncommitted();
+
+ BinaryRow foundRow = storage.read(rowId, Timestamp.nextVersion());
+
+ assertRowMatches(foundRow, binaryRow);
+ }
+
+ private RowId commitAbortAndAddUncommitted() {
+ RowId rowId = storage.insert(binaryRow, txId);
+ storage.commitWrite(rowId, Timestamp.nextVersion());
+
+ storage.addWrite(rowId, binaryRow2, newTransactionId());
+ storage.abortWrite(rowId);
+
+ storage.addWrite(rowId, binaryRow3, newTransactionId());
+
+ return rowId;
+ }
+
+ @Test
+ void scanByTimestampWorksCorrectlyAfterCommitAndAbortFollowedByUncommittedWrite() throws Exception {
+ commitAbortAndAddUncommitted();
+
+ try (Cursor<BinaryRow> cursor = storage.scan(k -> true, Timestamp.nextVersion())) {
+ BinaryRow foundRow = cursor.next();
+
+ assertRowMatches(foundRow, binaryRow);
+
+ assertFalse(cursor.hasNext());
+ }
+ }
+
+ @Test
+ void readByTimestampWorksCorrectlyIfNoUncommittedValueExists() {
+ LinkRowId rowId = storage.insert(binaryRow, txId);
+
+ BinaryRow foundRow = storage.read(rowId, Timestamp.nextVersion());
+
+ assertThat(foundRow, is(nullValue()));
+ }
}