Skip to content

Copy locks in observer & Single observer #973

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Jan 12, 2024
Merged

Copy locks in observer & Single observer #973

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion ydb/core/tx/datashard/datashard_ut_order.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4058,7 +4058,7 @@ Y_UNIT_TEST(TestShardSnapshotReadNoEarlyReply) {
}
}

Y_UNIT_TEST(TestSnapshotReadAfterBrokenLock) {
Y_UNIT_TEST_TWIN(TestSnapshotReadAfterBrokenLock, EvWrite) {
TPortManager pm;
TServerSettings serverSettings(pm.GetPort(2134));
serverSettings.SetDomainName("Root")
Expand All @@ -4073,6 +4073,9 @@ Y_UNIT_TEST(TestSnapshotReadAfterBrokenLock) {
CreateShardedTable(server, sender, "/Root", "table-1", 1);
CreateShardedTable(server, sender, "/Root", "table-2", 1);

auto rows = EvWrite ? TEvWriteRows{{{1, 1}}, {{2, 2}}, {{3, 3}}, {{5, 5}}} : TEvWriteRows{};
auto evWriteObservers = ReplaceEvProposeTransactionWithEvWrite(runtime, rows);

ExecSQL(server, sender, Q_("UPSERT INTO `/Root/table-1` (key, value) VALUES (1, 1)"));
ExecSQL(server, sender, Q_("UPSERT INTO `/Root/table-2` (key, value) VALUES (2, 2)"));

Expand Down
3 changes: 1 addition & 2 deletions ydb/core/tx/datashard/datashard_ut_write.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,8 +36,7 @@ Y_UNIT_TEST_SUITE(DataShardWrite) {
auto [shards, tableId] = CreateShardedTable(server, sender, "/Root", "table-1", opts);

auto rows = EvWrite ? TEvWriteRows{{{0, 1}}, {{2, 3}}, {{4, 5}}} : TEvWriteRows{};
auto upsertObserver = ReplaceEvProposeTransactionWithEvWrite(runtime, rows);
auto upsertResultObserver = ReplaceEvProposeTransactionResultWithEvWrite(runtime, rows);
auto evWriteObservers = ReplaceEvProposeTransactionWithEvWrite(runtime, rows);

ExecSQL(server, sender, Q_("UPSERT INTO `/Root/table-1` (key, value) VALUES (0, 1);"));
ExecSQL(server, sender, Q_("UPSERT INTO `/Root/table-1` (key, value) VALUES (2, 3);"));
Expand Down
21 changes: 13 additions & 8 deletions ydb/core/tx/datashard/ut_common/datashard_ut_common.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1877,11 +1877,11 @@ NKikimrDataEvents::TEvWriteResult Write(TTestActorRuntime& runtime, TActorId sen



TTestActorRuntimeBase::TEventObserverHolder ReplaceEvProposeTransactionWithEvWrite(TTestActorRuntime& runtime, TEvWriteRows& rows) {
TTestActorRuntimeBase::TEventObserverHolderPair ReplaceEvProposeTransactionWithEvWrite(TTestActorRuntime& runtime, TEvWriteRows& rows) {
if (rows.empty())
return {};

return runtime.AddObserver([&rows](TAutoPtr<IEventHandle>& event) {
auto requestObserver = runtime.AddObserver([&rows](TAutoPtr<IEventHandle>& event) {
if (event->GetTypeRewrite() != TEvDataShard::EvProposeTransaction)
return;

Expand Down Expand Up @@ -1939,18 +1939,21 @@ TTestActorRuntimeBase::TEventObserverHolder ReplaceEvProposeTransactionWithEvWri
ui64 payloadIndex = NKikimr::NEvWrite::TPayloadHelper<NKikimr::NEvents::TDataEvents::TEvWrite>(*evWrite).AddDataToPayload(std::move(blobData));
evWrite->AddOperation(NKikimrDataEvents::TEvWrite::TOperation::OPERATION_UPSERT, tableId, columnIds, payloadIndex, NKikimrDataEvents::FORMAT_CELLVEC);

// Copy locks
if (tx.HasLockTxId())
evWrite->Record.SetLockTxId(tx.GetLockTxId());
if (tx.HasLockNodeId())
evWrite->Record.SetLockNodeId(tx.GetLockNodeId());
if (tx.GetKqpTransaction().HasLocks())
evWrite->Record.MutableLocks()->CopyFrom(tx.GetKqpTransaction().GetLocks());

// Replace event
auto handle = new IEventHandle(event->Recipient, event->Sender, evWrite.release(), 0, event->Cookie);
handle->Rewrite(handle->GetTypeRewrite(), event->GetRecipientRewrite());
event.Reset(handle);
});
}

TTestActorRuntimeBase::TEventObserverHolder ReplaceEvProposeTransactionResultWithEvWrite(TTestActorRuntime& runtime, TEvWriteRows& rows) {
if (rows.empty())
return {};

return runtime.AddObserver([&rows](TAutoPtr<IEventHandle>& event) {
auto responseObserver = runtime.AddObserver([&rows](TAutoPtr<IEventHandle>& event) {
if (event->GetTypeRewrite() != NEvents::TDataEvents::EvWriteResult)
return;

Expand All @@ -1971,6 +1974,8 @@ TTestActorRuntimeBase::TEventObserverHolder ReplaceEvProposeTransactionResultWit
handle->Rewrite(handle->GetTypeRewrite(), event->GetRecipientRewrite());
event.Reset(handle);
});

return {std::move(requestObserver), std::move(responseObserver)};
}

void UploadRows(TTestActorRuntime& runtime, const TString& tablePath, const TVector<std::pair<TString, Ydb::Type_PrimitiveTypeId>>& types, const TVector<TCell>& keys, const TVector<TCell>& values)
Expand Down
3 changes: 1 addition & 2 deletions ydb/core/tx/datashard/ut_common/datashard_ut_common.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -749,8 +749,7 @@ class TEvWriteRows : public std::vector<TEvWriteRow> {
}
};

TTestActorRuntimeBase::TEventObserverHolder ReplaceEvProposeTransactionWithEvWrite(TTestActorRuntime& runtime, TEvWriteRows& rows);
TTestActorRuntimeBase::TEventObserverHolder ReplaceEvProposeTransactionResultWithEvWrite(TTestActorRuntime& runtime, TEvWriteRows& rows);
TTestActorRuntimeBase::TEventObserverHolderPair ReplaceEvProposeTransactionWithEvWrite(TTestActorRuntime& runtime, TEvWriteRows& rows);

void UploadRows(TTestActorRuntime& runtime, const TString& tablePath, const TVector<std::pair<TString, Ydb::Type_PrimitiveTypeId>>& types, const TVector<TCell>& keys, const TVector<TCell>& values);

Expand Down
1 change: 1 addition & 0 deletions ydb/library/actors/testlib/test_runtime.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -349,6 +349,7 @@ namespace NActors {
TEventObserverCollection* List;
TEventObserverCollection::iterator Iter;
};
using TEventObserverHolderPair = std::pair<TEventObserverHolder,TEventObserverHolder>;

// An example of using AddObserver in unit tests
/*
Expand Down