Skip to content

Commit a6e0cde

Browse files
azevaykinazevaykin
authored
UpdateRow & WriteRow (#663)
1 parent 1b81aaf commit a6e0cde

File tree

5 files changed

+64
-63
lines changed

5 files changed

+64
-63
lines changed

‎ydb/core/tx/schemeshard/ut_backup/ut_backup.cpp‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -73,11 +73,11 @@ Y_UNIT_TEST_SUITE(TBackupTests) {
7373

7474
Backup(runtime, ToString(Codec), R"(
7575
Name: "Table"
76-
Columns { Name: "key" Type: "Utf8" }
76+
Columns { Name: "key" Type: "Uint32" }
7777
Columns { Name: "value" Type: "Utf8" }
7878
KeyColumnNames: ["key"]
7979
)", [](TTestBasicRuntime& runtime) {
80-
WriteRow(runtime, "a", "valueA");
80+
UpdateRow(runtime, "Table", 1, "valueA");
8181
});
8282
}
8383

@@ -86,17 +86,17 @@ Y_UNIT_TEST_SUITE(TBackupTests) {
8686

8787
Backup(runtime, ToString(Codec), R"(
8888
Name: "Table"
89-
Columns { Name: "key" Type: "Utf8" }
89+
Columns { Name: "key" Type: "Uint32" }
9090
Columns { Name: "value" Type: "Utf8" }
9191
KeyColumnNames: ["key"]
9292
SplitBoundary {
9393
KeyPrefix {
94-
Tuple { Optional { Text: "b" } }
94+
Tuple { Optional { Uint32: 2 } }
9595
}
9696
}
9797
)", [](TTestBasicRuntime& runtime) {
98-
WriteRow(runtime, "a", "valueA", TTestTxConfig::FakeHiveTablets + 0);
99-
WriteRow(runtime, "b", "valueb", TTestTxConfig::FakeHiveTablets + 1);
98+
UpdateRow(runtime, "Table", 1, "valueA", TTestTxConfig::FakeHiveTablets + 0);
99+
UpdateRow(runtime, "Table", 2, "valueb", TTestTxConfig::FakeHiveTablets + 1);
100100
});
101101
}
102102

@@ -107,12 +107,12 @@ Y_UNIT_TEST_SUITE(TBackupTests) {
107107

108108
const auto actualResult = Backup(runtime, ToString(Codec), R"(
109109
Name: "Table"
110-
Columns { Name: "key" Type: "Utf8" }
110+
Columns { Name: "key" Type: "Uint32" }
111111
Columns { Name: "value" Type: "Utf8" }
112112
KeyColumnNames: ["key"]
113113
)", [](TTestBasicRuntime& runtime) {
114114
for (ui32 i = 0; i < 100 * batchSize; ++i) {
115-
WriteRow(runtime, Sprintf("a%d", i), "valueA");
115+
UpdateRow(runtime, "Table", i, "valueA");
116116
}
117117
}, batchSize, minWriteBatchSize);
118118

‎ydb/core/tx/schemeshard/ut_export/ut_export.cpp‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -687,14 +687,14 @@ partitioning_settings {
687687

688688
TestCreateTable(runtime, ++txId, "/MyRoot", R"(
689689
Name: "Table"
690-
Columns { Name: "key" Type: "Utf8" }
690+
Columns { Name: "key" Type: "Uint32" }
691691
Columns { Name: "value" Type: "Utf8" }
692692
KeyColumnNames: ["key"]
693693
)");
694694
env.TestWaitNotification(runtime, txId);
695695

696-
WriteRow(runtime, "a", "valueA");
697-
WriteRow(runtime, "b", "valueB");
696+
UpdateRow(runtime, "Table", 1, "valueA");
697+
UpdateRow(runtime, "Table", 2, "valueB");
698698

699699
runtime.SetLogPriority(NKikimrServices::S3_WRAPPER, NActors::NLog::PRI_TRACE);
700700
runtime.SetLogPriority(NKikimrServices::DATASHARD_BACKUP, NActors::NLog::PRI_TRACE);
@@ -827,14 +827,14 @@ partitioning_settings {
827827

828828
TestCreateTable(runtime, ++txId, "/MyRoot", R"(
829829
Name: "Table"
830-
Columns { Name: "key" Type: "Utf8" }
830+
Columns { Name: "key" Type: "Uint32" }
831831
Columns { Name: "value" Type: "Utf8" }
832832
KeyColumnNames: ["key"]
833833
)");
834834
env.TestWaitNotification(runtime, txId);
835835

836836
for (int i = 1; i < 500; ++i) {
837-
WriteRow(runtime, Sprintf("a%i", i), "value");
837+
UpdateRow(runtime, "Table", i, "value");
838838
}
839839

840840
// trigger memtable's compaction
@@ -925,13 +925,13 @@ partitioning_settings {
925925

926926
TestCreateTable(runtime, ++txId, "/MyRoot", R"(
927927
Name: "Table"
928-
Columns { Name: "key" Type: "Utf8" }
928+
Columns { Name: "key" Type: "Uint32" }
929929
Columns { Name: "value" Type: "Utf8" }
930930
KeyColumnNames: ["key"]
931931
)");
932932
env.TestWaitNotification(runtime, txId);
933933

934-
WriteRow(runtime, "a", "valueA");
934+
UpdateRow(runtime, "Table", 1, "valueA");
935935

936936
TPortManager portManager;
937937
const ui16 port = portManager.GetPort();
@@ -1277,14 +1277,14 @@ partitioning_settings {
12771277

12781278
TestCreateTable(runtime, ++txId, "/MyRoot", R"(
12791279
Name: "Table"
1280-
Columns { Name: "key" Type: "Utf8" }
1280+
Columns { Name: "key" Type: "Uint32" }
12811281
Columns { Name: "value" Type: "Utf8" }
12821282
KeyColumnNames: ["key"]
12831283
)");
12841284
env.TestWaitNotification(runtime, txId);
12851285

1286-
WriteRow(runtime, "a", "valueA");
1287-
WriteRow(runtime, "b", "valueB");
1286+
UpdateRow(runtime, "Table", 1, "valueA");
1287+
UpdateRow(runtime, "Table", 2, "valueB");
12881288
runtime.SetLogPriority(NKikimrServices::DATASHARD_BACKUP, NActors::NLog::PRI_DEBUG);
12891289

12901290
TPortManager portManager;

‎ydb/core/tx/schemeshard/ut_helpers/helpers.cpp‎

Lines changed: 39 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@
22

33
#include <ydb/core/engine/mkql_proto.h>
44
#include <ydb/core/engine/minikql/flat_local_tx_factory.h>
5+
#include <ydb/core/tx/data_events/events.h>
6+
#include <ydb/core/tx/data_events/payload_helper.h>
57
#include <ydb/core/tx/schemeshard/schemeshard.h>
68
#include <ydb/core/tx/tx_proxy/proxy.h>
79
#include <ydb/core/persqueue/events/global.h>
@@ -2256,31 +2258,31 @@ namespace NSchemeShardUT_Private {
22562258
NKikimr::NPQ::CmdWrite(&runtime, tabletId, edge, partitionId, "sourceid0", msgSeqNo, data, false, {}, true, cookie, 0);
22572259
}
22582260

2259-
void WriteRow(TTestActorRuntime& runtime, const TString& key, const TString& value, ui64 tabletId) {
2261+
void UpdateRow(TTestActorRuntime& runtime, const TString& table, const ui32 key, const TString& value, ui64 tabletId) {
22602262
NKikimrMiniKQL::TResult result;
22612263
TString error;
22622264
NKikimrProto::EReplyStatus status = LocalMiniKQL(runtime, tabletId, Sprintf(R"(
22632265
(
2264-
(let key '( '('key (Utf8 '%s) ) ) )
2266+
(let key '( '('key (Uint32 '%d) ) ) )
22652267
(let row '( '('value (Utf8 '%s) ) ) )
2266-
(return (AsList (UpdateRow '__user__Table key row) ))
2268+
(return (AsList (UpdateRow '__user__%s key row) ))
22672269
)
2268-
)", key.c_str(), value.c_str()), result, error);
2270+
)", key, value.c_str(), table.c_str()), result, error);
22692271

22702272
UNIT_ASSERT_VALUES_EQUAL_C(status, NKikimrProto::EReplyStatus::OK, error);
22712273
UNIT_ASSERT_VALUES_EQUAL(error, "");
22722274
}
22732275

2274-
void WriteRowPg(TTestActorRuntime& runtime, const TString& key, ui32 value, ui64 tabletId) {
2276+
void UpdateRowPg(TTestActorRuntime& runtime, const TString& table, const ui32 key, ui32 value, ui64 tabletId) {
22752277
NKikimrMiniKQL::TResult result;
22762278
TString error;
22772279
NKikimrProto::EReplyStatus status = LocalMiniKQL(runtime, tabletId, Sprintf(R"(
22782280
(
2279-
(let key '( '('key (Utf8 '%s) ) ) )
2281+
(let key '( '('key (Utf8 '%d) ) ) )
22802282
(let row '( '('value (PgConst '%u (PgType 'int4)) ) ) )
2281-
(return (AsList (UpdateRow '__user__Table key row) ))
2283+
(return (AsList (UpdateRow '__user__%s key row) ))
22822284
)
2283-
)", key.c_str(), value), result, error);
2285+
)", key, value, table.c_str()), result, error);
22842286

22852287
UNIT_ASSERT_VALUES_EQUAL_C(status, NKikimrProto::EReplyStatus::OK, error);
22862288
UNIT_ASSERT_VALUES_EQUAL(error, "");
@@ -2291,6 +2293,7 @@ namespace NSchemeShardUT_Private {
22912293
auto tableDesc = DescribePath(runtime, tablePath, true, true);
22922294
const auto& tablePartitions = tableDesc.GetPathDescription().GetTablePartitions();
22932295
UNIT_ASSERT(partitionIdx < tablePartitions.size());
2296+
const ui64 datashardTabletId = tablePartitions[partitionIdx].GetDatashardId();
22942297

22952298
auto ev = MakeHolder<TEvDataShard::TEvUploadRowsRequest>();
22962299
ev->Record.SetTableId(tableDesc.GetPathId());
@@ -2314,7 +2317,34 @@ namespace NSchemeShardUT_Private {
23142317
}
23152318

23162319
const auto& sender = runtime.AllocateEdgeActor();
2317-
ForwardToTablet(runtime, tablePartitions[partitionIdx].GetDatashardId(), sender, ev.Release());
2320+
ForwardToTablet(runtime, datashardTabletId, sender, ev.Release());
23182321
runtime.GrabEdgeEvent<TEvDataShard::TEvUploadRowsResponse>(sender);
23192322
}
2323+
2324+
void WriteRow(TTestActorRuntime& runtime, const ui64 txId, const TString& tablePath, int partitionIdx, const ui32 key, const TString& value, bool successIsExpected) {
2325+
auto tableDesc = DescribePath(runtime, tablePath, true, true);
2326+
const auto& tablePartitions = tableDesc.GetPathDescription().GetTablePartitions();
2327+
UNIT_ASSERT(partitionIdx < tablePartitions.size());
2328+
const ui64 tableId = tableDesc.GetPathId();
2329+
const ui64 datashardTabletId = tablePartitions[partitionIdx].GetDatashardId();
2330+
2331+
const auto& sender = runtime.AllocateEdgeActor();
2332+
2333+
std::vector<ui32> columnIds{1, 2};
2334+
2335+
TVector<TCell> cells{TCell((const char*)&key, sizeof(ui32)), TCell(value.c_str(), value.size())};
2336+
2337+
TSerializedCellMatrix matrix(cells, 1, 2);
2338+
2339+
auto evWrite = std::make_unique<NKikimr::NEvents::TDataEvents::TEvWrite>(txId, NKikimrDataEvents::TEvWrite::MODE_IMMEDIATE);
2340+
ui64 payloadIndex = NKikimr::NEvWrite::TPayloadHelper<NKikimr::NEvents::TDataEvents::TEvWrite>(*evWrite).AddDataToPayload(std::move(matrix.ReleaseBuffer()));
2341+
evWrite->AddOperation(NKikimrDataEvents::TEvWrite::TOperation::OPERATION_UPSERT, tableId, 1, columnIds, payloadIndex, NKikimrDataEvents::FORMAT_CELLVEC);
2342+
2343+
ForwardToTablet(runtime, datashardTabletId, sender, evWrite.release());
2344+
2345+
auto ev = runtime.GrabEdgeEventRethrow<NEvents::TDataEvents::TEvWriteResult>(sender);
2346+
auto status = ev->Get()->Record.GetStatus();
2347+
2348+
UNIT_ASSERT_C(successIsExpected == (status == NKikimrDataEvents::TEvWriteResult::STATUS_COMPLETED), "Status: " << ev->Get()->Record.GetStatus() << " Issues: " << ev->Get()->Record.GetIssues());
2349+
}
23202350
}

‎ydb/core/tx/schemeshard/ut_helpers/helpers.h‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -546,8 +546,9 @@ namespace NSchemeShardUT_Private {
546546

547547
void SendTEvPeriodicTopicStats(TTestActorRuntime& runtime, ui64 topicId, ui64 generation, ui64 round, ui64 dataSize, ui64 usedReserveSize);
548548
void WriteToTopic(TTestActorRuntime& runtime, const TString& path, ui32& msgSeqNo, const TString& message);
549-
void WriteRow(TTestActorRuntime& runtime, const TString& key, const TString& value, ui64 tabletId = TTestTxConfig::FakeHiveTablets);
550-
void WriteRowPg(TTestActorRuntime& runtime, const TString& key, ui32 value, ui64 tabletId = TTestTxConfig::FakeHiveTablets);
549+
void UpdateRow(TTestActorRuntime& runtime, const TString& table, const ui32 key, const TString& value, ui64 tabletId = TTestTxConfig::FakeHiveTablets);
550+
void UpdateRowPg(TTestActorRuntime& runtime, const TString& table, const ui32 key, ui32 value, ui64 tabletId = TTestTxConfig::FakeHiveTablets);
551551
void UploadRows(TTestActorRuntime& runtime, const TString& tablePath, int partitionIdx, const TVector<ui32>& keyTags, const TVector<ui32>& valueTags, const TVector<ui32>& recordIds);
552+
void WriteRow(TTestActorRuntime& runtime, const ui64 txId, const TString& tablePath, int partitionIdx, const ui32 key, const TString& value, bool successIsExpected = true);
552553

553554
} //NSchemeShardUT_Private

‎ydb/core/tx/schemeshard/ut_subdomain/ut_subdomain.cpp‎

Lines changed: 4 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -2816,21 +2816,6 @@ Y_UNIT_TEST_SUITE(TSchemeShardSubDomainTest) {
28162816
TTestEnv env(runtime, opts);
28172817
const auto sender = runtime.AllocateEdgeActor();
28182818

2819-
auto writeRow = [&](ui64 tabletId, ui32 key, const TString& value, const char* table) {
2820-
NKikimrMiniKQL::TResult result;
2821-
TString error;
2822-
NKikimrProto::EReplyStatus status = LocalMiniKQL(runtime, tabletId, Sprintf(R"(
2823-
(
2824-
(let key '( '('key (Uint32 '%u ) ) ) )
2825-
(let row '( '('value (Utf8 '%s) ) ) )
2826-
(return (AsList (UpdateRow '__user__%s key row) ))
2827-
)
2828-
)", key, value.c_str(), table), result, error);
2829-
2830-
UNIT_ASSERT_VALUES_EQUAL_C(status, NKikimrProto::EReplyStatus::OK, error);
2831-
UNIT_ASSERT_VALUES_EQUAL(error, "");
2832-
};
2833-
28342819
auto waitForTableStats = [&](ui32 shards) {
28352820
TDispatchOptions options;
28362821
options.FinalEvents.push_back(TDispatchOptions::TFinalEventCondition(TEvDataShard::EvPeriodicTableStats, shards));
@@ -2865,7 +2850,7 @@ Y_UNIT_TEST_SUITE(TSchemeShardSubDomainTest) {
28652850
)", {NKikimrScheme::StatusAccepted});
28662851
env.TestWaitNotification(runtime, txId);
28672852

2868-
writeRow(tabletId, 1, "value1", "Table1");
2853+
UpdateRow(runtime, "Table1", 1, "value1", tabletId);
28692854
waitForTableStats(1);
28702855

28712856
auto du = getDiskSpaceUsage();
@@ -2888,8 +2873,8 @@ Y_UNIT_TEST_SUITE(TSchemeShardSubDomainTest) {
28882873
)", {NKikimrScheme::StatusAccepted});
28892874
env.TestWaitNotification(runtime, txId);
28902875

2891-
writeRow(tabletId + 0, 1, "value1", "Table2");
2892-
writeRow(tabletId + 1, 2, "value2", "Table2");
2876+
UpdateRow(runtime, "Table2", 1, "value1", tabletId + 0);
2877+
UpdateRow(runtime, "Table2", 2, "value2", tabletId + 1);
28932878
waitForTableStats(1 /* Table1 */ + 2 /* Table2 */);
28942879

28952880
auto du = getDiskSpaceUsage();
@@ -2910,21 +2895,6 @@ Y_UNIT_TEST_SUITE(TSchemeShardSubDomainTest) {
29102895
TTestEnv env(runtime, opts);
29112896
ui64 txId = 100;
29122897

2913-
auto writeRow = [&](ui64 tabletId, ui32 key, const TString& value, const char* table) {
2914-
NKikimrMiniKQL::TResult result;
2915-
TString error;
2916-
NKikimrProto::EReplyStatus status = LocalMiniKQL(runtime, tabletId, Sprintf(R"(
2917-
(
2918-
(let key '( '('key (Uint32 '%u ) ) ) )
2919-
(let row '( '('value (Utf8 '%s) ) ) )
2920-
(return (AsList (UpdateRow '__user__%s key row) ))
2921-
)
2922-
)", key, value.c_str(), table), result, error);
2923-
2924-
UNIT_ASSERT_VALUES_EQUAL_C(status, NKikimrProto::EReplyStatus::OK, error);
2925-
UNIT_ASSERT_VALUES_EQUAL(error, "");
2926-
};
2927-
29282898
auto waitForTableStats = [&](ui32 shards) {
29292899
TDispatchOptions options;
29302900
options.FinalEvents.push_back(TDispatchOptions::TFinalEventCondition(TEvDataShard::EvPeriodicTableStats, shards));
@@ -2972,7 +2942,7 @@ Y_UNIT_TEST_SUITE(TSchemeShardSubDomainTest) {
29722942
)", {NKikimrScheme::StatusAccepted});
29732943
env.TestWaitNotification(runtime, txId);
29742944

2975-
writeRow(tabletId, 1, "value1", "Table1");
2945+
UpdateRow(runtime, "Table1", 1, "value1", tabletId);
29762946
waitForTableStats(1);
29772947

29782948
TestDescribeResult(DescribePath(runtime, "/MyRoot/USER_0"),

0 commit comments

Comments
 (0)