You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ap...@apache.org on 2025/04/01 07:00:10 UTC
(ignite-3) branch main updated: IGNITE-24967 Add ability to explicitly specify read timestamp for RO transactions (#5526)
This is an automated email from the ASF dual-hosted git repository.
apolovtsev 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 8076b986934 IGNITE-24967 Add ability to explicitly specify read timestamp for RO transactions (#5526)
8076b986934 is described below
commit 8076b986934e95ebfb5d67dff99abe59c9ef8ca9
Author: Alexander Polovtcev <al...@gmail.com>
AuthorDate: Tue Apr 1 10:00:05 2025 +0300
IGNITE-24967 Add ability to explicitly specify read timestamp for RO transactions (#5526)
---
.../ignite/internal/tx/InternalTxOptions.java | 28 ++++++++++++++++++++--
.../ignite/internal/tx/impl/TxManagerImpl.java | 20 +++++++++-------
.../apache/ignite/internal/tx/TxManagerTest.java | 25 +++++++++++++++++++
3 files changed, 63 insertions(+), 10 deletions(-)
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/InternalTxOptions.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/InternalTxOptions.java
index 36d1ba62068..9586a268419 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/InternalTxOptions.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/InternalTxOptions.java
@@ -17,6 +17,9 @@
package org.apache.ignite.internal.tx;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.jetbrains.annotations.Nullable;
+
/**
* Transaction options for internal use.
*/
@@ -32,9 +35,17 @@ public class InternalTxOptions {
/** Transaction timeout. 0 means 'use default timeout'. */
private final long timeoutMillis;
- private InternalTxOptions(TxPriority priority, long timeoutMillis) {
+ /**
+ * Read timestamp of the transaction. If {@code null} - the most recent available timestamp will be calculated based on the current
+ * node's clock.
+ */
+ @Nullable
+ private final HybridTimestamp readTimestamp;
+
+ private InternalTxOptions(TxPriority priority, long timeoutMillis, @Nullable HybridTimestamp readTimestamp) {
this.priority = priority;
this.timeoutMillis = timeoutMillis;
+ this.readTimestamp = readTimestamp;
}
public static Builder builder() {
@@ -57,11 +68,19 @@ public class InternalTxOptions {
return timeoutMillis;
}
+ public @Nullable HybridTimestamp readTimestamp() {
+ return readTimestamp;
+ }
+
/** Builder for InternalTxOptions. */
public static class Builder {
private TxPriority priority = TxPriority.NORMAL;
+
private long timeoutMillis = 0;
+ @Nullable
+ private HybridTimestamp readTimestamp = null;
+
public Builder priority(TxPriority priority) {
this.priority = priority;
return this;
@@ -72,8 +91,13 @@ public class InternalTxOptions {
return this;
}
+ public Builder readTimestamp(HybridTimestamp readTimestamp) {
+ this.readTimestamp = readTimestamp;
+ return this;
+ }
+
public InternalTxOptions build() {
- return new InternalTxOptions(priority, timeoutMillis);
+ return new InternalTxOptions(priority, timeoutMillis, readTimestamp);
}
}
}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
index cd625927fed..0a34be9ee57 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
@@ -469,11 +469,15 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
) {
UUID txId = transactionIdGenerator.transactionIdFor(beginTimestamp, options.priority());
- HybridTimestamp observableTimestamp = timestampTracker.get();
+ HybridTimestamp readTimestamp = options.readTimestamp();
- HybridTimestamp readTimestamp = observableTimestamp != null
- ? HybridTimestamp.max(observableTimestamp, currentReadTimestamp(beginTimestamp))
- : currentReadTimestamp(beginTimestamp);
+ if (readTimestamp == null) {
+ HybridTimestamp observableTimestamp = timestampTracker.get();
+
+ readTimestamp = observableTimestamp != null
+ ? HybridTimestamp.max(observableTimestamp, currentReadTimestamp(beginTimestamp))
+ : currentReadTimestamp(beginTimestamp);
+ }
boolean lockAcquired = lowWatermark.tryLock(txId, readTimestamp);
if (!lockAcquired) {
@@ -607,7 +611,7 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
Map<ReplicationGroupId, PendingTxPartitionEnlistment> enlistedGroups,
UUID txId
) {
- LOG.debug("Finish [commit={}, txId={}, groups={}].", commitIntent, txId, enlistedGroups);
+ LOG.debug("Finish [commit={}, txId={}, groups={}, commitPartId={}].", commitIntent, txId, enlistedGroups, commitPartition);
if (commitPartition != null) {
assertReplicationGroupType(commitPartition);
@@ -702,7 +706,7 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
private CompletableFuture<Void> prepareFinish(
HybridTimestampTracker observableTimestampTracker,
- ReplicationGroupId commitPartition,
+ @Nullable ReplicationGroupId commitPartition,
boolean commit,
Map<ReplicationGroupId, PendingTxPartitionEnlistment> enlistedGroups,
UUID txId,
@@ -750,7 +754,7 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
) {
return inBusyLockAsync(busyLock, () -> placementDriverHelper.awaitPrimaryReplicaWithExceptionHandling(commitPartition)
.thenCompose(meta ->
- makeFinishRequest(
+ sendFinishRequest(
observableTimestampTracker,
commitPartition,
meta.getLeaseholder(),
@@ -813,7 +817,7 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
.thenCompose(identity()));
}
- private CompletableFuture<Void> makeFinishRequest(
+ private CompletableFuture<Void> sendFinishRequest(
HybridTimestampTracker observableTimestampTracker,
ReplicationGroupId commitPartition,
String primaryConsistentId,
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxManagerTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxManagerTest.java
index a9e72a47008..a0fb8182dbf 100644
--- a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxManagerTest.java
+++ b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxManagerTest.java
@@ -30,6 +30,8 @@ import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFu
import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_COMMIT_ERR;
import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_PRIMARY_REPLICA_EXPIRED_ERR;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -778,6 +780,29 @@ public class TxManagerTest extends IgniteAbstractTest {
doReturn(null).when(localRwTxCounter).inUpdateRwTxCountLock(any());
}
+ @Test
+ void testReadWithExplicitTimestamp() {
+ HybridTimestamp now = clockService.now();
+
+ HybridTimestamp readTimestampInThePast = now.subtractPhysicalTime(100_000);
+
+ InternalTxOptions optionsWithExplicitReadTimestamp = InternalTxOptions.builder()
+ .readTimestamp(readTimestampInThePast)
+ .build();
+
+ InternalTxOptions defaultOptions = InternalTxOptions.defaults();
+
+ InternalTransaction tx1 = txManager.beginExplicitRo(hybridTimestampTracker, defaultOptions);
+ InternalTransaction tx2 = txManager.beginExplicitRo(hybridTimestampTracker, optionsWithExplicitReadTimestamp);
+
+ assertThat(tx1.readTimestamp(), is(greaterThanOrEqualTo(now.subtractPhysicalTime(roReadTimestampSkew()))));
+ assertThat(tx2.readTimestamp(), is(readTimestampInThePast));
+ }
+
+ private long roReadTimestampSkew() {
+ return idleSafeTimePropagationPeriodMsSupplier.getAsLong() + clockService.maxClockSkewMillis();
+ }
+
private InternalTransaction prepareTransaction() {
InternalTransaction tx = txManager.beginExplicitRw(hybridTimestampTracker, InternalTxOptions.defaults());