Skip to content

Commit 2e53522

Browse files
alex268alex268
authored
Merge pull request #696 from alex268/master
Added support of StrincSerialiazeble transaction level
2 parents d876581 + e5d21e9 commit 2e53522

10 files changed

Lines changed: 320 additions & 91 deletions

File tree

‎bom/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515

1616
<properties>
1717
<ydb-auth-api.version>1.0.0</ydb-auth-api.version>
18-
<ydb-proto-api.version>1.9.5</ydb-proto-api.version>
18+
<ydb-proto-api.version>1.9.6</ydb-proto-api.version>
1919
<yc-auth.version>2.3.1</yc-auth.version>
2020
</properties>
2121

‎common/src/main/java/tech/ydb/common/transaction/TxMode.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,8 @@ public enum TxMode {
88
NONE,
99

1010
SERIALIZABLE_RW,
11+
STRICT_SERIALIZABLE_RW,
12+
1113
SNAPSHOT_RO,
1214
SNAPSHOT_RW,
1315
READ_COMMITTED_RW,

‎common/src/main/java/tech/ydb/common/transaction/VirtualTimestamp.java‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
package tech.ydb.common.transaction;
2+
3+
/**
4+
*
5+
* @author Aleksandr Gorshenin {@literal <alexandr268@ydb.tech>}
6+
*/
7+
public class VirtualTimestamp {
8+
private final long planStep;
9+
private final long txId;
10+
11+
public VirtualTimestamp(long planStep, long txId) {
12+
this.planStep = planStep;
13+
this.txId = txId;
14+
}
15+
16+
public long getPlanStep() {
17+
return planStep;
18+
}
19+
20+
public long getTxId() {
21+
return txId;
22+
}
23+
24+
@Override
25+
public boolean equals(Object o) {
26+
if (this == o) {
27+
return true;
28+
}
29+
30+
if (!(o instanceof VirtualTimestamp)) {
31+
return false;
32+
}
33+
34+
VirtualTimestamp that = (VirtualTimestamp) o;
35+
return planStep == that.planStep && txId == that.txId;
36+
}
37+
38+
@Override
39+
public int hashCode() {
40+
return Long.hashCode(planStep) * 31 + Long.hashCode(txId);
41+
}
42+
43+
@Override
44+
public String toString() {
45+
return "VirtualTimestamp{"
46+
+ "planStep=" + Long.toUnsignedString(planStep)
47+
+ ", txId=" + Long.toUnsignedString(txId)
48+
+ "}";
49+
}
50+
}

‎common/src/test/java/tech/ydb/common/transaction/VirtualTimestampTest.java‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
package tech.ydb.common.transaction;
2+
3+
import org.junit.Assert;
4+
import org.junit.Test;
5+
6+
/**
7+
*
8+
* @author Aleksandr Gorshenin {@literal <alexandr268@ydb.tech>}
9+
*/
10+
public class VirtualTimestampTest {
11+
12+
@Test
13+
public void getterTest() {
14+
VirtualTimestamp vt = new VirtualTimestamp(0x123456L, 0x9876L);
15+
Assert.assertEquals(0x123456L, vt.getPlanStep());
16+
Assert.assertEquals(0x9876L, vt.getTxId());
17+
18+
Assert.assertEquals("VirtualTimestamp{planStep=1193046, txId=39030}", vt.toString());
19+
}
20+
21+
@Test
22+
public void hashCodeAndEqualsTest() {
23+
VirtualTimestamp vt1 = new VirtualTimestamp(0x123456L, 0x9876L);
24+
VirtualTimestamp vt2 = new VirtualTimestamp(0x123456L, 0x9877L);
25+
VirtualTimestamp vt3 = new VirtualTimestamp(0x123455L, 0x9876L);
26+
VirtualTimestamp vt4 = new VirtualTimestamp(0x123456L, 0x9876L);
27+
28+
Assert.assertNotEquals(vt1, null);
29+
Assert.assertNotEquals(vt1, new Object());
30+
31+
Assert.assertEquals(vt1, vt1);
32+
Assert.assertEquals(vt1, vt4);
33+
Assert.assertNotEquals(vt1, vt2);
34+
Assert.assertNotEquals(vt1, vt3);
35+
Assert.assertNotEquals(vt2, vt3);
36+
37+
Assert.assertEquals(vt1.hashCode(), vt4.hashCode());
38+
Assert.assertNotEquals(vt1.hashCode(), vt2.hashCode());
39+
Assert.assertNotEquals(vt1.hashCode(), vt3.hashCode());
40+
Assert.assertNotEquals(vt2.hashCode(), vt3.hashCode());
41+
}
42+
}

‎query/src/main/java/tech/ydb/query/impl/SessionImpl.java‎

Lines changed: 108 additions & 67 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
import org.slf4j.LoggerFactory;
1818

1919
import tech.ydb.common.transaction.TxMode;
20+
import tech.ydb.common.transaction.VirtualTimestamp;
2021
import tech.ydb.common.transaction.impl.YdbTransactionImpl;
2122
import tech.ydb.core.Issue;
2223
import tech.ydb.core.Result;
@@ -29,8 +30,8 @@
2930
import tech.ydb.core.tracing.Scope;
3031
import tech.ydb.core.tracing.Span;
3132
import tech.ydb.core.utils.URITools;
32-
import tech.ydb.core.utils.UpdatableOptional;
3333
import tech.ydb.proto.ValueProtos;
34+
import tech.ydb.proto.common.CommonProtos;
3435
import tech.ydb.proto.formats.YdbFormats;
3536
import tech.ydb.proto.query.YdbQuery;
3637
import tech.ydb.query.QuerySession;
@@ -74,14 +75,12 @@ abstract class SessionImpl implements QuerySession {
7475
private final QueryServiceRpc rpc;
7576
private final String sessionId;
7677
private final long nodeID;
77-
private final boolean isTraceEnabled;
7878
private final AtomicReference<TransactionImpl> transaction;
7979

8080
SessionImpl(QueryServiceRpc rpc, YdbQuery.CreateSessionResponse response) {
8181
this.rpc = rpc;
8282
this.sessionId = response.getSessionId();
8383
this.nodeID = getNodeBySessionId(response.getSessionId(), response.getNodeId());
84-
this.isTraceEnabled = logger.isTraceEnabled();
8584
this.transaction = new AtomicReference<>(new TransactionImpl(TxMode.SERIALIZABLE_RW, null));
8685
}
8786

@@ -393,68 +392,103 @@ abstract class StreamImpl implements QueryStream {
393392

394393
abstract void handleTxMeta(String txId);
395394

396-
void handleCompletion(Status status, Throwable th) {
397-
}
395+
void handleCompletion(Status status, Throwable th) { }
398396

399397
@Override
400398
public CompletableFuture<Result<QueryInfo>> execute(PartsHandler handler) {
401-
final UpdatableOptional<Status> operationStatus = new UpdatableOptional<>();
402-
final UpdatableOptional<QueryStats> stats = new UpdatableOptional<>();
403-
return Span.endOnResult(span, grpcStream.start(msg -> {
404-
if (isTraceEnabled) {
405-
logger.trace("{} got stream message {}",
406-
SessionImpl.this, TextFormat.shortDebugString(msg));
407-
}
408-
Issue[] issues = Issue.fromPb(msg.getIssuesList());
409-
Status status = Status.of(StatusCode.fromProto(msg.getStatus()), issues);
410-
411-
updateSessionState(status);
412-
413-
if (!status.isSuccess()) {
414-
handleTxMeta(null);
415-
operationStatus.update(status);
416-
return;
417-
}
418-
419-
if (msg.hasTxMeta()) {
420-
handleTxMeta(msg.getTxMeta().getId());
421-
}
422-
if (issues.length > 0) {
423-
if (handler != null) {
424-
handler.onIssues(issues);
425-
} else {
426-
logger.trace("{} lost issues message", SessionImpl.this);
427-
}
428-
}
429-
if (msg.hasExecStats()) {
430-
stats.update(new QueryStats(msg.getExecStats()));
431-
}
432-
433-
if (msg.hasResultSet()) {
434-
long index = msg.getResultSetIndex();
435-
if (handler != null) {
436-
handler.onNextRawPart(index, msg.getResultSet());
437-
} else {
438-
logger.trace("{} lost result set part with index {}", SessionImpl.this, index);
439-
}
440-
}
441-
}).whenComplete(this::handleCompletion).thenApply(streamStatus -> {
399+
Observer observer = new Observer(handler);
400+
CompletableFuture<Result<QueryInfo>> result = grpcStream.start(observer)
401+
.whenComplete(this::handleCompletion)
402+
.thenApply(streamStatus -> {
442403
updateSessionState(streamStatus);
443-
Status status = operationStatus.orElse(streamStatus);
404+
Status status = observer.mergedStatus(streamStatus);
444405
if (status.isSuccess()) {
445-
return Result.success(new QueryInfo(stats.get()), streamStatus);
406+
return Result.success(observer.buildQueryInfo(), status);
446407
} else {
447408
return Result.fail(status);
448409
}
449-
})
450-
);
410+
});
411+
412+
return Span.endOnResult(span, result);
451413
}
452414

453415
@Override
454416
public void cancel() {
455417
updateSessionState(CANCELLED);
456418
grpcStream.cancel();
457419
}
420+
421+
private class Observer implements GrpcReadStream.Observer<YdbQuery.ExecuteQueryResponsePart> {
422+
private final PartsHandler handler;
423+
424+
private volatile Status queryStatus = null;
425+
private volatile QueryStats stats = null;
426+
private volatile VirtualTimestamp commitVt = null;
427+
private volatile VirtualTimestamp snapshotVt = null;
428+
429+
Observer(PartsHandler handler) {
430+
this.handler = handler;
431+
}
432+
433+
public Status mergedStatus(Status streamStatus) {
434+
return (streamStatus.isSuccess() && queryStatus != null) ? queryStatus : streamStatus;
435+
}
436+
437+
public QueryInfo buildQueryInfo() {
438+
return new QueryInfo(stats, commitVt, snapshotVt);
439+
}
440+
441+
@Override
442+
public void onNext(YdbQuery.ExecuteQueryResponsePart msg) {
443+
if (logger.isTraceEnabled()) {
444+
logger.trace("{} got stream message {}", SessionImpl.this, TextFormat.shortDebugString(msg));
445+
}
446+
447+
Issue[] issues = Issue.fromPb(msg.getIssuesList());
448+
Status status = Status.of(StatusCode.fromProto(msg.getStatus()), issues);
449+
450+
updateSessionState(status);
451+
452+
if (!status.isSuccess()) {
453+
handleTxMeta(null);
454+
queryStatus = status;
455+
return;
456+
}
457+
458+
if (msg.hasTxMeta()) {
459+
handleTxMeta(msg.getTxMeta().getId());
460+
}
461+
if (issues.length > 0) {
462+
if (handler != null) {
463+
handler.onIssues(issues);
464+
} else {
465+
logger.trace("{} lost issues message", SessionImpl.this);
466+
}
467+
}
468+
if (msg.hasExecStats()) {
469+
stats = new QueryStats(msg.getExecStats());
470+
}
471+
472+
if (msg.hasCommitTimestamp()) {
473+
CommonProtos.VirtualTimestamp vt = msg.getCommitTimestamp();
474+
commitVt = new VirtualTimestamp(vt.getPlanStep(), vt.getTxId());
475+
}
476+
477+
if (msg.hasSnapshotTimestamp()) {
478+
CommonProtos.VirtualTimestamp vt = msg.getSnapshotTimestamp();
479+
snapshotVt = new VirtualTimestamp(vt.getPlanStep(), vt.getTxId());
480+
}
481+
482+
if (msg.hasResultSet()) {
483+
long index = msg.getResultSetIndex();
484+
if (handler != null) {
485+
handler.onNextRawPart(index, msg.getResultSet());
486+
} else {
487+
logger.trace("{} lost result set part with index {}", SessionImpl.this, index);
488+
}
489+
}
490+
}
491+
}
458492
}
459493

460494
class TransactionImpl extends YdbTransactionImpl implements QueryTransaction {
@@ -544,23 +578,30 @@ public CompletableFuture<Result<QueryInfo>> commit(CommitTransactionSettings set
544578
.build();
545579

546580
try (Scope ignored = span.makeCurrent()) {
547-
return Span.endOnResult(span, rpc.commitTransaction(request, makeOptions(settings, span).build()))
548-
.thenApply(res -> {
549-
Status status = res.getStatus();
550-
currentStatusFuture.complete(status);
551-
updateSessionState(status);
552-
if (!txId.compareAndSet(transactionId, null)) {
553-
logger.warn("{} lost commit response for transaction {}", SessionImpl.this,
554-
transactionId);
555-
}
556-
// TODO: CommitTransactionResponse must contain exec_stats
557-
return res.map(resp -> new QueryInfo(null));
558-
}).whenComplete(((status, th) -> {
559-
if (th != null) {
560-
currentStatusFuture.completeExceptionally(
561-
new RuntimeException("Transaction commit failed with exception", th));
562-
}
563-
}));
581+
GrpcRequestSettings options = makeOptions(settings, span).build();
582+
CompletableFuture<Result<QueryInfo>> result = rpc.commitTransaction(request, options).thenApply(res -> {
583+
Status status = res.getStatus();
584+
currentStatusFuture.complete(status);
585+
updateSessionState(status);
586+
if (!txId.compareAndSet(transactionId, null)) {
587+
logger.warn("{} lost commit response for transaction {}", SessionImpl.this, transactionId);
588+
}
589+
590+
return res.map(resp -> {
591+
VirtualTimestamp commit = null;
592+
if (resp.hasCommitTimestamp()) {
593+
CommonProtos.VirtualTimestamp vt = resp.getCommitTimestamp();
594+
commit = new VirtualTimestamp(vt.getPlanStep(), vt.getTxId());
595+
}
596+
return new QueryInfo(null, commit, null);
597+
});
598+
}).whenComplete(((status, th) -> {
599+
if (th != null) {
600+
currentStatusFuture.completeExceptionally(
601+
new RuntimeException("Transaction commit failed with exception", th));
602+
}
603+
}));
604+
return Span.endOnResult(span, result);
564605
}
565606
}
566607

‎query/src/main/java/tech/ydb/query/impl/TableClientImpl.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,9 @@ private YdbQuery.TransactionControl mapTxControl(YdbTable.TransactionControl tc)
9090
if (tc.getBeginTx().hasSerializableReadWrite()) {
9191
return TxControl.txModeCtrl(TxMode.SERIALIZABLE_RW, tc.getCommitTx());
9292
}
93+
if (tc.getBeginTx().hasStrictSerializableReadWrite()) {
94+
return TxControl.txModeCtrl(TxMode.STRICT_SERIALIZABLE_RW, tc.getCommitTx());
95+
}
9396
if (tc.getBeginTx().hasSnapshotReadOnly()) {
9497
return TxControl.txModeCtrl(TxMode.SNAPSHOT_RO, tc.getCommitTx());
9598
}

‎query/src/main/java/tech/ydb/query/impl/TxControl.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,10 @@ class TxControl {
1212
.setSerializableReadWrite(YdbQuery.SerializableModeSettings.getDefaultInstance())
1313
.build();
1414

15+
private static final YdbQuery.TransactionSettings TS_STRICT_SERIALIZABLE = YdbQuery.TransactionSettings.newBuilder()
16+
.setStrictSerializableReadWrite(YdbQuery.StrictSerializableRWModeSettings.getDefaultInstance())
17+
.build();
18+
1519
private static final YdbQuery.TransactionSettings TS_SNAPSHOT = YdbQuery.TransactionSettings.newBuilder()
1620
.setSnapshotReadOnly(YdbQuery.SnapshotModeSettings.getDefaultInstance())
1721
.build();
@@ -61,6 +65,8 @@ public static YdbQuery.TransactionSettings txSettings(TxMode tx) {
6165
switch (tx) {
6266
case SERIALIZABLE_RW:
6367
return TS_SERIALIZABLE;
68+
case STRICT_SERIALIZABLE_RW:
69+
return TS_STRICT_SERIALIZABLE;
6470
case SNAPSHOT_RO:
6571
return TS_SNAPSHOT;
6672
case SNAPSHOT_RW:

0 commit comments

Comments
 (0)