You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vp...@apache.org on 2022/12/21 09:05:09 UTC
[ignite-3] branch main updated: IGNITE-18050 Fixed possible phantom reads during sorted index scan (#1458)
This is an automated email from the ASF dual-hosted git repository.
vpyatkov 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 7501a2564a IGNITE-18050 Fixed possible phantom reads during sorted index scan (#1458)
7501a2564a is described below
commit 7501a2564a4333a2304b5d8c0af0a6d5ab15236c
Author: Vladislav Pyatkov <vl...@gmail.com>
AuthorDate: Wed Dec 21 13:05:05 2022 +0400
IGNITE-18050 Fixed possible phantom reads during sorted index scan (#1458)
---
.../distributionzones/DistributionZoneManager.java | 4 +-
.../internal/runner/app/ItTablesApiTest.java | 1 +
.../ignite/internal/table/ItTableScanTest.java | 62 ++++++++++++++++--
.../table/distributed/SortedIndexLocker.java | 73 +++++++++++++---------
4 files changed, 103 insertions(+), 37 deletions(-)
diff --git a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java
index 87eb81edae..d8f5117835 100644
--- a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java
+++ b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java
@@ -24,6 +24,7 @@ import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.updateLogicalTopologyAndVersion;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.updateTriggerKey;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zonesLogicalTopologyVersionKey;
+import static org.apache.ignite.internal.metastorage.client.Conditions.notExists;
import static org.apache.ignite.internal.metastorage.client.Conditions.value;
import static org.apache.ignite.internal.metastorage.client.Operations.ops;
import static org.apache.ignite.lang.ErrorGroups.Common.NODE_STOPPING_ERR;
@@ -495,7 +496,8 @@ public class DistributionZoneManager implements IgniteComponent {
if (topVerFromMetastorage == null || ByteUtils.bytesToLong(topVerFromMetastorage) < topologyVersionFromCmg) {
Set<String> topologyFromCmg = snapshot.nodes().stream().map(ClusterNode::name).collect(Collectors.toSet());
- Condition topologyVersionCondition = value(zonesLogicalTopologyVersionKey()).eq(topVerFromMetastorage);
+ Condition topologyVersionCondition = topVerFromMetastorage == null ? notExists(zonesLogicalTopologyVersionKey()) :
+ value(zonesLogicalTopologyVersionKey()).eq(topVerFromMetastorage);
If iff = If.iif(topologyVersionCondition,
updateLogicalTopologyAndVersion(topologyFromCmg, topologyVersionFromCmg),
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItTablesApiTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItTablesApiTest.java
index fbeaf09d8b..13ebf99dad 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItTablesApiTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItTablesApiTest.java
@@ -395,6 +395,7 @@ public class ItTablesApiTest extends IgniteAbstractTest {
* @throws Exception If failed.
*/
@Test
+ @Disabled("https://issues.apache.org/jira/browse/IGNITE-18429")
public void testCreateDropTable() throws Exception {
clusterNodes.forEach(ign -> assertNull(ign.tables().table(TABLE_NAME)));
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItTableScanTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItTableScanTest.java
index f9fcca89bc..ed9d4e7201 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItTableScanTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItTableScanTest.java
@@ -23,6 +23,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.nio.ByteBuffer;
@@ -49,6 +50,7 @@ import org.apache.ignite.internal.sql.engine.AbstractBasicIntegrationTest;
import org.apache.ignite.internal.testframework.IgniteTestUtils;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.lang.IgniteStringFormatter;
+import org.apache.ignite.table.KeyValueView;
import org.apache.ignite.table.Tuple;
import org.apache.ignite.tx.IgniteTransactions;
import org.junit.jupiter.api.AfterEach;
@@ -100,11 +102,11 @@ public class ItTableScanTest extends AbstractBasicIntegrationTest {
InternalTable internalTable = table.internalTable();
- UUID soredIndexId = getSortedIndexId();
+ UUID sortedIndexId = getSortedIndexId();
ArrayList<ByteBuffer> scannedRows = new ArrayList<>();
- Publisher<BinaryRow> publisher = internalTable.scan(0, tx1, soredIndexId, null, null, 0, null);
+ Publisher<BinaryRow> publisher = internalTable.scan(0, tx1, sortedIndexId, null, null, 0, null);
CompletableFuture<Void> scanned = new CompletableFuture<>();
@@ -140,11 +142,11 @@ public class ItTableScanTest extends AbstractBasicIntegrationTest {
InternalTable internalTable = table.internalTable();
- UUID soredIndexId = getSortedIndexId();
+ UUID sortedIndexId = getSortedIndexId();
ArrayList<ByteBuffer> scannedRows = new ArrayList<>();
- Publisher<BinaryRow> publisher = internalTable.scan(0, null, soredIndexId, null, null, 0, null);
+ Publisher<BinaryRow> publisher = internalTable.scan(0, null, sortedIndexId, null, null, 0, null);
CompletableFuture<Void> scanned = new CompletableFuture<>();
@@ -163,8 +165,7 @@ public class ItTableScanTest extends AbstractBasicIntegrationTest {
IgniteTestUtils.await(scanned);
- //TODO: IGNITE-18243 Uncomment this change when the scan on a sorted index will be redesigned.
- //assertEquals(ROW_IDS.size() + 1, scannedRows.size());
+ assertEquals(ROW_IDS.size() + 1, scannedRows.size());
}
@Test
@@ -427,6 +428,55 @@ public class ItTableScanTest extends AbstractBasicIntegrationTest {
assertEquals(ROW_IDS.size() + txOpFut.get(), scanAllRows(pub).size());
}
+ @Test
+ public void testTwiceScanInTransaction() throws Exception {
+ TableImpl table = getOrCreateTable();
+
+ InternalTable internalTable = table.internalTable();
+
+ KeyValueView kvView = table.keyValueView();
+
+ UUID sortedIndexId = getSortedIndexId();
+
+ ArrayList<ByteBuffer> scannedRows = new ArrayList<>();
+
+ InternalTransaction tx = (InternalTransaction) CLUSTER_NODES.get(0).transactions().begin();
+
+ Publisher<BinaryRow> publisher = internalTable.scan(0, tx, sortedIndexId, null, null, 0, null);
+
+ CompletableFuture<Void> scanned = new CompletableFuture<>();
+
+ Subscription subscription = subscribeToPublisher(scannedRows, publisher, scanned);
+
+ subscription.request(3);
+
+ IgniteTestUtils.waitForCondition(() -> scannedRows.size() == 3, 10_000);
+
+ assertEquals(3, scannedRows.size());
+ assertFalse(scanned.isDone());
+
+ assertThrows(Exception.class,
+ () -> kvView.put(null, Tuple.create().set("key", 3), Tuple.create().set("valInt", 3).set("valStr", "New_3")));
+
+ kvView.put(null, Tuple.create().set("key", 8), Tuple.create().set("valInt", 8).set("valStr", "New_8"));
+
+ subscription.request(1_000); // Request so much entries here to close the publisher.
+
+ IgniteTestUtils.waitForCondition(() -> scanned.isDone(), 10_000);
+
+ assertEquals(ROW_IDS.size() + 1, scannedRows.size());
+
+ var pub = internalTable.scan(0, tx, sortedIndexId, null, null, 0, null);
+
+ assertEquals(scanAllRows(pub).size(), scannedRows.size());
+
+ assertTrue(scanned.isDone());
+
+ tx.commit();
+
+ kvView.put(null, Tuple.create().set("key", 3), Tuple.create().set("valInt", 3).set("valStr", "New_3"));
+ }
+
/**
* Scans all rows form given publisher.
*
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/SortedIndexLocker.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/SortedIndexLocker.java
index 01072a3c1d..0e7d6d1450 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/SortedIndexLocker.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/SortedIndexLocker.java
@@ -28,6 +28,7 @@ import org.apache.ignite.internal.schema.BinaryTupleSchema;
import org.apache.ignite.internal.schema.BinaryTupleSchema.Element;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.storage.index.IndexRow;
+import org.apache.ignite.internal.storage.index.PeekCursor;
import org.apache.ignite.internal.storage.index.SortedIndexStorage;
import org.apache.ignite.internal.tx.Lock;
import org.apache.ignite.internal.tx.LockKey;
@@ -92,42 +93,54 @@ public class SortedIndexLocker implements IndexLocker {
*
* @param txId An identifier of the transaction in which the row is read.
* @param cursor Cursor, which next key is to be locked.
- * @return A future representing a result.
+ * @return A future representing a next index row or {@code null} if the cursor has no rows more.
*/
public CompletableFuture<IndexRow> locksForScan(UUID txId, Cursor<IndexRow> cursor) {
- // TODO https://issues.apache.org/jira/browse/IGNITE-18057
- // 1. Do index.peekNext() item and lock it. (Lock INF+ if not found)
- // 2. Do index.peekNext() again and ensure it wasn't changed.
- // 3. If matches then return lock, otherwise release lock on peeked one and repeat from step 1.
-
- // BinaryRow nextRow;
- // BinaryTuple nextKey;
- //
- // if (cursor.hasNext()) {
- // nextKey = cursor.peekNext().indexColumns();
- // } else { // otherwise INF
- // nextKey = POSITIVE_INF;
- //
- //
- // return lockManager.acquire(txId, new LockKey(indexId, nextKey.byteBuffer()), LockMode.S)
- // .thenCompose(lock -> {
- // if (!Objects.equals(nexrRow, indexCursor.peekNext())) { // Concurrent insert into locked range detected. Retry.
- // lockManager.release(lock);
- // return locksForScan(txId, cursor);
- // }
- //
- // return CompletableFuture.completedFuture(lock);
- // });
- //
-
- if (!cursor.hasNext()) { // No upper bound or not found. Lock INF+ and exit loop.
+ assert cursor instanceof PeekCursor : "Cursor has incorrect type [type=" + cursor.getClass().getSimpleName() + ']';
+
+ PeekCursor<IndexRow> peekCursor = (PeekCursor<IndexRow>) cursor;
+
+ return acquireLockNextKey(txId, peekCursor).thenApply(indexRow -> {
+ if (indexRow == null) {
+ return null;
+ }
+
+ return peekCursor.next();
+ });
+ }
+
+ /**
+ * Acquires a lock on the next key without moving a cursor pointer ahead.
+ *
+ * @param txId An identifier of the transaction in which the row is read.
+ * @param peekCursor Cursor, which next key is to be locked.
+ * @return A future representing a locked index row or {@code null} if the cursor has no rows more.
+ */
+ private CompletableFuture<IndexRow> acquireLockNextKey(UUID txId, PeekCursor<IndexRow> peekCursor) {
+ if (!peekCursor.hasNext()) { // No upper bound or not found. Lock INF+ and exit loop.
return lockManager.acquire(txId, new LockKey(indexId, POSITIVE_INF.byteBuffer()), LockMode.S)
- .thenCompose(ignore -> CompletableFuture.completedFuture(null));
+ .thenCompose(ignore -> {
+ if (peekCursor.hasNext()) {
+ lockManager.release(txId, new LockKey(indexId, POSITIVE_INF.byteBuffer()), LockMode.S);
+
+ return acquireLockNextKey(txId, peekCursor);
+ }
+
+ return CompletableFuture.completedFuture(null);
+ });
} else {
- IndexRow nextRow = cursor.next();
+ IndexRow nextRow = peekCursor.peek();
return lockManager.acquire(txId, new LockKey(indexId, nextRow.indexColumns().byteBuffer()), LockMode.S)
- .thenCompose(ignore -> CompletableFuture.completedFuture(nextRow));
+ .thenCompose(ignore -> {
+ if (!peekCursor.hasNext() || !nextRow.rowId().equals(peekCursor.peek().rowId())) {
+ lockManager.release(txId, new LockKey(indexId, nextRow.indexColumns().byteBuffer()), LockMode.S);
+
+ return acquireLockNextKey(txId, peekCursor);
+ }
+
+ return CompletableFuture.completedFuture(nextRow);
+ });
}
}