You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tk...@apache.org on 2025/05/25 06:58:23 UTC
(ignite-3) 01/01: IGNITE-25477 wip
This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a commit to branch ignite-25477
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 5bc10277dfb543a13a356e2975f3ad2d1aa0dbc4
Author: Kirill Tkalenko <tk...@yandex.ru>
AuthorDate: Sun May 25 09:57:52 2025 +0300
IGNITE-25477 wip
---
.../raft/snapshot/PartitionDataStorage.java | 10 +++++
.../ignite/internal/storage/CommitResult.java | 45 ++++++++++++++++++++
.../internal/storage/CommitResultStatus.java | 31 ++++++++++++++
.../internal/storage/MvPartitionStorage.java | 8 ++++
.../internal/storage/TxIdMismatchException.java | 8 ++--
.../mv/AbstractPageMemoryMvPartitionStorage.java | 33 +++++++++++++++
.../pagememory/mv/CommitWriteInvokeClosure.java | 25 +++++++++++
.../table/distributed/StorageUpdateHandler.java | 49 ++++++++--------------
.../SnapshotAwarePartitionDataStorage.java | 6 +++
9 files changed, 178 insertions(+), 37 deletions(-)
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/snapshot/PartitionDataStorage.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/snapshot/PartitionDataStorage.java
index 15a73274fbd..3dec8052395 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/snapshot/PartitionDataStorage.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/snapshot/PartitionDataStorage.java
@@ -23,6 +23,8 @@ import org.apache.ignite.internal.close.ManuallyCloseable;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.raft.RaftGroupConfiguration;
import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.storage.CommitResult;
+import org.apache.ignite.internal.storage.CommitResultStatus;
import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.MvPartitionStorage.WriteClosure;
import org.apache.ignite.internal.storage.PartitionTimestampCursor;
@@ -187,6 +189,14 @@ public interface PartitionDataStorage extends ManuallyCloseable {
*/
void commitWrite(RowId rowId, HybridTimestamp timestamp) throws StorageException;
+ /** Will be removed soon. */
+ // TODO: IGNITE-25477 Get rid of it
+ default CommitResult commitWrite(RowId rowId, HybridTimestamp timestamp, UUID txId) throws StorageException {
+ commitWrite(rowId, timestamp);
+
+ return new CommitResult(CommitResultStatus.SUCCESS, null);
+ }
+
/**
* Scans all versions of a single row.
*
diff --git a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/CommitResult.java b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/CommitResult.java
new file mode 100644
index 00000000000..92c721e2484
--- /dev/null
+++ b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/CommitResult.java
@@ -0,0 +1,45 @@
+/*
+ * 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.storage;
+
+import java.util.UUID;
+import org.jetbrains.annotations.Nullable;
+
+/** Will be removed soon. */
+// TODO: IGNITE-25477 Get rid of it
+public class CommitResult {
+ private final CommitResultStatus status;
+
+ private final @Nullable UUID expectedTxId;
+
+ /** Constructor. */
+ public CommitResult(CommitResultStatus status, @Nullable UUID expectedTxId) {
+ this.status = status;
+ this.expectedTxId = expectedTxId;
+ }
+
+ /** Will be removed soon. */
+ public @Nullable UUID expectedTxId() {
+ return expectedTxId;
+ }
+
+ /** Will be removed soon. */
+ public CommitResultStatus status() {
+ return status;
+ }
+}
diff --git a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/CommitResultStatus.java b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/CommitResultStatus.java
new file mode 100644
index 00000000000..3569d2d78a3
--- /dev/null
+++ b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/CommitResultStatus.java
@@ -0,0 +1,31 @@
+/*
+ * 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.storage;
+
+/** Will be removed soon. */
+// TODO: IGNITE-25477 Get rid of it
+public enum CommitResultStatus {
+ /** Will be removed soon. */
+ SUCCESS,
+
+ /** Will be removed soon. */
+ NO_WRITE_INTENT,
+
+ /** Will be removed soon. */
+ MISMATCH_TX;
+}
diff --git a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/MvPartitionStorage.java b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/MvPartitionStorage.java
index fbaf53274e5..d32b0d6d8c1 100644
--- a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/MvPartitionStorage.java
+++ b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/MvPartitionStorage.java
@@ -210,6 +210,14 @@ public interface MvPartitionStorage extends ManuallyCloseable {
*/
void commitWrite(RowId rowId, HybridTimestamp timestamp) throws StorageException;
+ /** Will be removed soon. */
+ // TODO: IGNITE-25477 Get rid of it
+ default CommitResult commitWrite(RowId rowId, HybridTimestamp timestamp, UUID txId) throws StorageException {
+ commitWrite(rowId, timestamp);
+
+ return new CommitResult(CommitResultStatus.SUCCESS, null);
+ }
+
/**
* Creates a committed version.
* In details:
diff --git a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/TxIdMismatchException.java b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/TxIdMismatchException.java
index edc2316f825..70c3571b8b9 100644
--- a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/TxIdMismatchException.java
+++ b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/TxIdMismatchException.java
@@ -23,9 +23,7 @@ import java.util.UUID;
import org.apache.ignite.internal.tostring.S;
import org.apache.ignite.lang.IgniteException;
-/**
- * Exception class that describes the situation when two independent transactions attempt to write values for the same key.
- */
+/** Exception describing the situation where transaction IDs do not match when performing an operation. */
public class TxIdMismatchException extends IgniteException {
/** Expected transaction id. */
private final UUID expectedTransactionId;
@@ -36,8 +34,8 @@ public class TxIdMismatchException extends IgniteException {
/**
* Constructor.
*
- * @param expectedTxId Expected transaction id.
- * @param conflictingTxId Conflicting transaction id.
+ * @param expectedTxId Expected transaction ID.
+ * @param conflictingTxId Conflicting transaction ID.
*/
public TxIdMismatchException(UUID expectedTxId, UUID conflictingTxId) {
super(INTERNAL_ERR, S.toString(
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java
index cfeeea8a978..d7b3c57c8aa 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java
@@ -44,6 +44,7 @@ import org.apache.ignite.internal.pagememory.tree.BplusTree.TreeRowMapClosure;
import org.apache.ignite.internal.pagememory.tree.IgniteTree.InvokeClosure;
import org.apache.ignite.internal.pagememory.util.GradualTaskExecutor;
import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.storage.CommitResult;
import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.PartitionTimestampCursor;
import org.apache.ignite.internal.storage.ReadResult;
@@ -487,6 +488,38 @@ public abstract class AbstractPageMemoryMvPartitionStorage implements MvPartitio
CommitWriteInvokeClosure commitWrite = new CommitWriteInvokeClosure(
rowId,
timestamp,
+ null,
+ updateTimestampHandler,
+ this
+ );
+
+ renewableState.versionChainTree().invoke(new VersionChainKey(rowId), null, commitWrite);
+
+ commitWrite.afterCompletion();
+
+ return commitWrite.commitResult();
+ } catch (IgniteInternalCheckedException e) {
+ throwStorageExceptionIfItCause(e);
+
+ throw new StorageException("Error while executing commitWrite: [rowId={}, {}]", e, rowId, createStorageInfo());
+ }
+ });
+ }
+
+ @Override
+ public CommitResult commitWrite(RowId rowId, HybridTimestamp timestamp, UUID txId) throws StorageException {
+ assert rowId.partitionId() == partitionId : rowId;
+
+ return busy(() -> {
+ throwExceptionIfStorageNotInRunnableOrRebalanceState(state.get(), this::createStorageInfo);
+
+ assert rowIsLocked(rowId);
+
+ try {
+ CommitWriteInvokeClosure commitWrite = new CommitWriteInvokeClosure(
+ rowId,
+ timestamp,
+ txId,
updateTimestampHandler,
this
);
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/CommitWriteInvokeClosure.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/CommitWriteInvokeClosure.java
index 39fa14feb6b..0d144502ac6 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/CommitWriteInvokeClosure.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/CommitWriteInvokeClosure.java
@@ -21,6 +21,7 @@ import static org.apache.ignite.internal.pagememory.util.PageIdUtils.NULL_LINK;
import static org.apache.ignite.internal.storage.pagememory.mv.AbstractPageMemoryMvPartitionStorage.DONT_LOAD_VALUE;
import static org.apache.ignite.internal.util.GridUnsafe.pageSize;
+import java.util.UUID;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.lang.IgniteInternalCheckedException;
import org.apache.ignite.internal.pagememory.freelist.FreeList;
@@ -31,6 +32,8 @@ import org.apache.ignite.internal.pagememory.tree.IgniteTree.InvokeClosure;
import org.apache.ignite.internal.pagememory.tree.IgniteTree.OperationType;
import org.apache.ignite.internal.pagememory.util.PageHandler;
import org.apache.ignite.internal.pagememory.util.PageIdUtils;
+import org.apache.ignite.internal.storage.CommitResult;
+import org.apache.ignite.internal.storage.CommitResultStatus;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.storage.StorageException;
import org.apache.ignite.internal.storage.pagememory.mv.gc.GcQueue;
@@ -48,6 +51,8 @@ class CommitWriteInvokeClosure implements InvokeClosure<VersionChain> {
private final HybridTimestamp timestamp;
+ private final @Nullable UUID txId;
+
private final AbstractPageMemoryMvPartitionStorage storage;
private final FreeList freeList;
@@ -78,14 +83,18 @@ class CommitWriteInvokeClosure implements InvokeClosure<VersionChain> {
@Nullable
private RowVersion prevRowVersion;
+ private CommitResult commitResult;
+
CommitWriteInvokeClosure(
RowId rowId,
HybridTimestamp timestamp,
+ @Nullable UUID txId,
UpdateTimestampHandler updateTimestampHandler,
AbstractPageMemoryMvPartitionStorage storage
) {
this.rowId = rowId;
this.timestamp = timestamp;
+ this.txId = txId;
this.storage = storage;
this.updateTimestampHandler = updateTimestampHandler;
@@ -123,9 +132,21 @@ class CommitWriteInvokeClosure implements InvokeClosure<VersionChain> {
// Row doesn't exist or the chain doesn't contain an uncommitted write intent.
operationType = OperationType.NOOP;
+ commitResult = new CommitResult(CommitResultStatus.NO_WRITE_INTENT, null);
+
+ return;
+ }
+
+ if (txId != null && !txId.equals(oldRow.transactionId())) {
+ operationType = OperationType.NOOP;
+
+ commitResult = new CommitResult(CommitResultStatus.MISMATCH_TX, oldRow.transactionId());
+
return;
}
+ commitResult = new CommitResult(CommitResultStatus.SUCCESS, null);
+
operationType = OperationType.PUT;
currentRowVersion = storage.readRowVersion(oldRow.headLink(), DONT_LOAD_VALUE);
@@ -222,4 +243,8 @@ class CommitWriteInvokeClosure implements InvokeClosure<VersionChain> {
}
}
}
+
+ CommitResult commitResult() {
+ return commitResult;
+ }
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandler.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandler.java
index d05df5adc07..c58c4cb5bc9 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandler.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandler.java
@@ -33,9 +33,12 @@ import org.apache.ignite.internal.replicator.PartitionGroupId;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.configuration.ReplicationConfiguration;
import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.storage.CommitResult;
+import org.apache.ignite.internal.storage.CommitResultStatus;
import org.apache.ignite.internal.storage.MvPartitionStorage.Locker;
import org.apache.ignite.internal.storage.ReadResult;
import org.apache.ignite.internal.storage.RowId;
+import org.apache.ignite.internal.storage.TxIdMismatchException;
import org.apache.ignite.internal.table.distributed.index.IndexUpdateHandler;
import org.apache.ignite.internal.table.distributed.replicator.PendingRows;
import org.apache.ignite.internal.util.Cursor;
@@ -301,7 +304,7 @@ public class StorageUpdateHandler {
// No more data => the write intent we have is actually the first version of this row
// and lastCommitTs is the commit timestamp of it.
// Action: commit this write intent.
- performCommitWrite(item.transactionId(), Set.of(rowId), lastCommitTs);
+ performCommitWriteWithCheckMatchingTxs(item.transactionId(), Set.of(rowId), lastCommitTs);
return;
}
// Otherwise there are other versions in the chain.
@@ -317,7 +320,7 @@ public class StorageUpdateHandler {
if (lastCommitTs.compareTo(committedItem.commitTimestamp()) > 0) {
// We see that lastCommitTs is later than the timestamp of the committed value => we need to commit the write intent.
// Action: commit this write intent.
- performCommitWrite(item.transactionId(), Set.of(rowId), lastCommitTs);
+ performCommitWriteWithCheckMatchingTxs(item.transactionId(), Set.of(rowId), lastCommitTs);
} else {
// lastCommitTs == committedItem.commitTimestamp()
// So we see a write intent from a different transaction, which was not committed on primary.
@@ -406,7 +409,7 @@ public class StorageUpdateHandler {
pendingRowIds.forEach(locker::lock);
if (commit) {
- performCommitWrite(txId, pendingRowIds, commitTimestamp);
+ performCommitWriteWithCheckMatchingTxs(txId, pendingRowIds, commitTimestamp);
} else {
performAbortWrite(txId, pendingRowIds, indexIds);
}
@@ -421,42 +424,24 @@ public class StorageUpdateHandler {
}
/**
- * Commit write intents created by the provided transaction.
+ * Commits write intents created by the provided transaction. Transaction that created write intent is expected to commit it.
*
- * @param txId Transaction id
- * @param pendingRowIds Row ids of write-intents to be committed.
+ * @param txId Transaction ID.
+ * @param pendingRowIds Row IDs of write-intents to be committed.
* @param commitTimestamp Commit timestamp.
+ * @throws TxIdMismatchException If commit of write intent is performed by a transaction that did not create it.
*/
- private void performCommitWrite(UUID txId, Set<RowId> pendingRowIds, HybridTimestamp commitTimestamp) {
- assert commitTimestamp != null : "Commit timestamp is null";
+ // TODO: IGNITE-25477 Wait for the linked ticket and wait for the tests
+ private void performCommitWriteWithCheckMatchingTxs(UUID txId, Set<RowId> pendingRowIds, HybridTimestamp commitTimestamp) {
+ assert commitTimestamp != null : "Commit timestamp is null: " + txId;
- // Please note: `pendingRowIds` might not contain the complete set of rows that were changed by this transaction:
- // Pending rows are stored in memory and will be lost in case a node restarts.
- // This method might be called by a write intent resolving transaction that will find only those rows that it needs itself.
- List<RowId> rowIds = new ArrayList<>();
+ for (RowId rowId : pendingRowIds) {
+ CommitResult commitResult = storage.commitWrite(rowId, commitTimestamp, txId);
- for (RowId pendingRowId : pendingRowIds) {
-
- // Here we check that the write intent we are going to commit still belongs to the provided transaction.
- //
- // This check is required to cover the following case caused by asynchronous cleanup of write intents:
- // 1. RO Transaction A sees a write intent for a row1, resolves it and schedules a cleanup for it.
- // 2. RW Transaction B sees the same write intent for a row1, resolves it and schedules a cleanup for it.
- // This cleanup action finishes first. Then Transaction B adds its own write intent for the row1.
- // 3. Transaction A starts executing the cleanup action.
- // Without this check it would commit the write intent from a different transaction.
- //
- // This is just a workaround. The proper fix is to check the transaction id for the row in the storage.
- // TODO: https://issues.apache.org/jira/browse/IGNITE-20347 to check transaction id in the storage
- ReadResult result = storage.getStorage().read(pendingRowId, HybridTimestamp.MAX_VALUE);
- if (result.isWriteIntent() && txId.equals(result.transactionId())) {
- // In case of an asynchronous cleanup of write intents, we might get into a situation when some of the
- // write intents were already cleaned up. In this case, we just ignore them.
- rowIds.add(pendingRowId);
+ if (commitResult.status() == CommitResultStatus.MISMATCH_TX) {
+ throw new TxIdMismatchException(commitResult.expectedTxId(), txId);
}
}
-
- rowIds.forEach(rowId -> storage.commitWrite(rowId, commitTimestamp));
}
/**
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/SnapshotAwarePartitionDataStorage.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/SnapshotAwarePartitionDataStorage.java
index b82380bc8bb..958e654e2a1 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/SnapshotAwarePartitionDataStorage.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/SnapshotAwarePartitionDataStorage.java
@@ -29,6 +29,7 @@ import org.apache.ignite.internal.partition.replicator.raft.snapshot.outgoing.Pa
import org.apache.ignite.internal.raft.RaftGroupConfiguration;
import org.apache.ignite.internal.raft.RaftGroupConfigurationConverter;
import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.storage.CommitResult;
import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.MvPartitionStorage.WriteClosure;
import org.apache.ignite.internal.storage.PartitionTimestampCursor;
@@ -162,6 +163,11 @@ public class SnapshotAwarePartitionDataStorage implements PartitionDataStorage {
partitionStorage.commitWrite(rowId, timestamp);
}
+ @Override
+ public CommitResult commitWrite(RowId rowId, HybridTimestamp timestamp, UUID txId) throws StorageException {
+ return partitionStorage.commitWrite(rowId, timestamp, txId);
+ }
+
@Override
public Cursor<ReadResult> scanVersions(RowId rowId) throws StorageException {
return partitionStorage.scanVersions(rowId);