Skip to content

Commit 0d36b47

Browse files
kungakunga
authored
Merge ebbc776 into 53f75b6 
2 parents 53f75b6 + ebbc776 commit 0d36b47

16 files changed

+254
-88
lines changed

‎ydb/core/tx/datashard/build_index.cpp

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -328,7 +328,7 @@ class TBuildIndexScan : public TActor<TBuildIndexScan>, public NTable::IScan {
328328

329329
EScan Seek(TLead& lead, ui64 seq) noexcept override {
330330
auto ctx = TActivationContext::AsActorContext().MakeFor(SelfId());
331-
LOG_DEBUG_S(ctx, NKikimrServices::TX_DATASHARD,
331+
LOG_TRACE_S(ctx, NKikimrServices::TX_DATASHARD,
332332
"Seek no " << seq << " " << Debug());
333333
if (seq) {
334334
if (!WriteBuf.IsEmpty()) {
@@ -367,7 +367,7 @@ class TBuildIndexScan : public TActor<TBuildIndexScan>, public NTable::IScan {
367367

368368
EScan Feed(TArrayRef<const TCell> key, const TRow& row) noexcept override {
369369
auto ctx = TActivationContext::AsActorContext().MakeFor(SelfId());
370-
LOG_DEBUG_S(ctx, NKikimrServices::TX_DATASHARD,
370+
LOG_TRACE_S(ctx, NKikimrServices::TX_DATASHARD,
371371
"Feed key " << DebugPrintPoint(KeyTypes, key, *AppData()->TypeRegistry)
372372
<< " " << Debug());
373373

‎ydb/core/tx/datashard/datashard__compact_borrowed.cpp

Lines changed: 39 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -21,45 +21,59 @@ class TDataShard::TTxCompactBorrowed : public NTabletFlatExecutor::TTransactionB
2121
<< " for table " << pathId
2222
<< " at tablet " << Self->TabletID());
2323

24-
auto response = MakeHolder<TEvDataShard::TEvCompactBorrowedResult>(Self->TabletID(), pathId);
24+
auto nothingToCompactResult = MakeHolder<TEvDataShard::TEvCompactBorrowedResult>(Self->TabletID(), pathId);
2525

26-
if (pathId.OwnerId != Self->GetPathOwnerId()) {
27-
// Ignore unexpected owner
28-
ctx.Send(Ev->Sender, std::move(response));
26+
if (pathId.OwnerId != Self->GetPathOwnerId()) { // ignore unexpected owner
27+
ctx.Send(Ev->Sender, std::move(nothingToCompactResult));
2928
return true;
3029
}
31-
3230
auto it = Self->TableInfos.find(pathId.LocalPathId);
33-
if (it == Self->TableInfos.end()) {
34-
// Ignore unexpected table (may normally happen with races)
35-
ctx.Send(Ev->Sender, std::move(response));
31+
if (it == Self->TableInfos.end()) { // ignore unexpected table (may normally happen with races)
32+
ctx.Send(Ev->Sender, std::move(nothingToCompactResult));
3633
return true;
3734
}
38-
3935
const TUserTable& tableInfo = *it->second;
36+
37+
THashSet<ui32> tablesToCompact;
38+
if (txc.DB.HasBorrowed(tableInfo.LocalTid, Self->TabletID())) {
39+
tablesToCompact.insert(tableInfo.LocalTid);
40+
}
41+
if (tableInfo.ShadowTid && txc.DB.HasBorrowed(tableInfo.ShadowTid, Self->TabletID())) {
42+
tablesToCompact.insert(tableInfo.ShadowTid);
43+
}
44+
45+
auto waiter = MakeIntrusive<TCompactBorrowedWaiter>(Ev->Sender, pathId.LocalPathId);
4046

41-
bool hasBorrowed = txc.DB.HasBorrowed(tableInfo.LocalTid, Self->TabletID());
42-
if (!hasBorrowed) {
47+
for (auto tableToCompact : tablesToCompact) {
4348
LOG_DEBUG_S(ctx, NKikimrServices::TX_DATASHARD,
4449
"TEvCompactBorrowed request from " << Ev->Sender
4550
<< " for table " << pathId
46-
<< " has no borrowed parts"
51+
<< " starting compaction for local table " << tableToCompact
4752
<< " at tablet " << Self->TabletID());
48-
ctx.Send(Ev->Sender, std::move(response));
49-
return true;
50-
}
5153

52-
LOG_DEBUG_S(ctx, NKikimrServices::TX_DATASHARD,
53-
"TEvCompactBorrowed request from " << Ev->Sender
54-
<< " for table " << pathId
55-
<< " starting compaction for local table " << tableInfo.LocalTid
56-
<< " at tablet " << Self->TabletID());
57-
58-
Self->Executor()->CompactBorrowed(tableInfo.LocalTid);
59-
Self->IncCounter(COUNTER_TX_COMPACT_BORROWED);
60-
++tableInfo.Stats.CompactBorrowedCount;
54+
if (Self->Executor()->CompactBorrowed(tableToCompact)) {
55+
Self->IncCounter(COUNTER_TX_COMPACT_BORROWED);
56+
++tableInfo.Stats.CompactBorrowedCount;
57+
58+
waiter->CompactingTables.insert(tableToCompact);
59+
Self->CompactBorrowedWaiters[tableToCompact].push_back(waiter);
60+
} else {
61+
LOG_DEBUG_S(ctx, NKikimrServices::TX_DATASHARD,
62+
"TEvCompactBorrowed request from " << Ev->Sender
63+
<< " for table " << pathId
64+
<< " can not be compacted"
65+
<< " at tablet " << Self->TabletID());
66+
}
67+
}
6168

62-
Self->CompactBorrowedWaiters[tableInfo.LocalTid].emplace_back(Ev->Sender);
69+
if (waiter->CompactingTables.empty()) { // none has been triggered
70+
LOG_DEBUG_S(ctx, NKikimrServices::TX_DATASHARD,
71+
"TEvCompactBorrowed request from " << Ev->Sender
72+
<< " for table " << pathId
73+
<< " has no parts for borrowed compaction"
74+
<< " at tablet " << Self->TabletID());
75+
ctx.Send(Ev->Sender, std::move(nothingToCompactResult));
76+
}
6377

6478
return true;
6579
}

‎ydb/core/tx/datashard/datashard__compaction.cpp

Lines changed: 31 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -196,6 +196,12 @@ void TDataShard::Handle(TEvDataShard::TEvCompactTable::TPtr& ev, const TActorCon
196196
void TDataShard::CompactionComplete(ui32 tableId, const TActorContext &ctx) {
197197
auto finishedInfo = Executor()->GetFinishedCompactionInfo(tableId);
198198

199+
LOG_DEBUG_S(ctx, NKikimrServices::TX_DATASHARD,
200+
"CompactionComplete of tablet# "<< TabletID()
201+
<< ", table# " << tableId
202+
<< ", finished edge# " << finishedInfo.Edge
203+
<< ", ts " << finishedInfo.FullCompactionTs);
204+
199205
TLocalPathId localPathId = InvalidLocalPathId;
200206
if (tableId >= Schema::MinLocalTid) {
201207
for (auto& ti : TableInfos) {
@@ -234,11 +240,12 @@ void TDataShard::ReplyCompactionWaiters(
234240
<< ", finished edge# " << compactionInfo.Edge
235241
<< ", front# " << (CompactionWaiters[tableId].empty() ? 0UL : std::get<0>(CompactionWaiters[tableId].front())));
236242

237-
auto& fullCompactionQueue = CompactionWaiters[tableId];
238-
while (!fullCompactionQueue.empty()) {
239-
const auto& waiter = CompactionWaiters[tableId].front();
240-
if (std::get<0>(waiter) > compactionInfo.Edge)
243+
auto fullCompactionQueue = CompactionWaiters.FindPtr(tableId);
244+
while (fullCompactionQueue && !fullCompactionQueue->empty()) {
245+
const auto& waiter = fullCompactionQueue->front();
246+
if (std::get<0>(waiter) > compactionInfo.Edge) {
241247
break;
248+
}
242249

243250
const auto& sender = std::get<1>(waiter);
244251
auto response = MakeHolder<TEvDataShard::TEvCompactTableResult>(
@@ -252,27 +259,30 @@ void TDataShard::ReplyCompactionWaiters(
252259
"Sending TEvCompactTableResult to# " << sender
253260
<< "pathId# " << TPathId(GetPathOwnerId(), localPathId));
254261

255-
fullCompactionQueue.pop_front();
262+
fullCompactionQueue->pop_front();
256263
}
257264

258-
auto& compactBorrowedQueue = CompactBorrowedWaiters[tableId];
259-
if (!compactBorrowedQueue.empty()) {
265+
auto compactBorrowedQueue = CompactBorrowedWaiters.FindPtr(tableId);
266+
if (compactBorrowedQueue && !compactBorrowedQueue->empty()) {
260267
const bool hasBorrowed = Executor()->HasBorrowed(tableId, TabletID());
261268
if (!hasBorrowed) {
262-
while (!compactBorrowedQueue.empty()) {
263-
const auto& waiter = compactBorrowedQueue.front();
264-
265-
auto response = MakeHolder<TEvDataShard::TEvCompactBorrowedResult>(
266-
TabletID(),
267-
GetPathOwnerId(),
268-
localPathId);
269-
ctx.Send(waiter, std::move(response));
270-
271-
LOG_DEBUG_S(ctx, NKikimrServices::TX_DATASHARD,
272-
"Sending TEvCompactBorrowedResult to# " << waiter
273-
<< "pathId# " << TPathId(GetPathOwnerId(), localPathId));
274-
275-
compactBorrowedQueue.pop_front();
269+
while (!compactBorrowedQueue->empty()) {
270+
const auto& waiter = compactBorrowedQueue->front();
271+
waiter->CompactingTables.erase(tableId);
272+
273+
if (waiter->CompactingTables.empty()) { // all requested tables have been compacted
274+
auto response = MakeHolder<TEvDataShard::TEvCompactBorrowedResult>(
275+
TabletID(),
276+
GetPathOwnerId(),
277+
waiter->RequestedTable);
278+
ctx.Send(waiter->ActorId, std::move(response));
279+
280+
LOG_DEBUG_S(ctx, NKikimrServices::TX_DATASHARD,
281+
"Sending TEvCompactBorrowedResult to# " << waiter->ActorId
282+
<< "pathId# " << TPathId(GetPathOwnerId(), waiter->RequestedTable));
283+
}
284+
285+
compactBorrowedQueue->pop_front();
276286
}
277287
}
278288
}

‎ydb/core/tx/datashard/datashard__stats.cpp

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -361,7 +361,9 @@ void TDataShard::Handle(TEvPrivate::TEvAsyncTableStats::TPtr& ev, const TActorCo
361361
Actors.erase(ev->Sender);
362362

363363
ui64 tableId = ev->Get()->TableId;
364-
LOG_DEBUG_S(ctx, NKikimrServices::TX_DATASHARD, "Stats rebuilt at datashard " << TabletID() << ", for tableId " << tableId);
364+
LOG_DEBUG_S(ctx, NKikimrServices::TX_DATASHARD, "Stats rebuilt at datashard " << TabletID() << ", for tableId " << tableId
365+
<< ": RowCount " << ev->Get()->Stats.RowCount << ", DataSize " << ev->Get()->Stats.DataSize.Size
366+
<< (ev->Get()->PartOwners.size() > 1 || ev->Get()->PartOwners.size() == 1 && *ev->Get()->PartOwners.begin() != TabletID() ? ", with borrowed parts" : ""));
365367

366368
i64 dataSize = 0;
367369
if (TableInfos.contains(tableId)) {

‎ydb/core/tx/datashard/datashard_impl.h

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2822,10 +2822,19 @@ class TDataShard
28222822
// from the front
28232823
THashMap<ui32, TCompactionWaiterList> CompactionWaiters;
28242824

2825-
using TCompactBorrowedWaiterList = TList<TActorId>;
2825+
struct TCompactBorrowedWaiter : public TThrRefBase {
2826+
TCompactBorrowedWaiter(TActorId actorId, TLocalPathId requestedTable)
2827+
: ActorId(actorId)
2828+
, RequestedTable(requestedTable)
2829+
{ }
2830+
2831+
TActorId ActorId;
2832+
TLocalPathId RequestedTable;
2833+
THashSet<ui32> CompactingTables;
2834+
};
28262835

28272836
// tableLocalTid -> waiters, similar to CompactionWaiters
2828-
THashMap<ui32, TCompactBorrowedWaiterList> CompactBorrowedWaiters;
2837+
THashMap<ui32, TList<TIntrusivePtr<TCompactBorrowedWaiter>>> CompactBorrowedWaiters;
28292838

28302839
struct TReplicationSourceOffsetsReceiveState {
28312840
// A set of tables for which we already received offsets

‎ydb/core/tx/datashard/datashard_loans.cpp

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@ NTabletFlatExecutor::ITransaction* TDataShard::CreateTxInitiateBorrowedPartsRetu
5050
}
5151

5252
void TDataShard::CompletedLoansChanged(const TActorContext &ctx) {
53+
LOG_INFO_S(ctx, NKikimrServices::TX_DATASHARD, TabletID() << " CompletedLoansChanged");
5354
Y_ABORT_UNLESS(Executor()->GetStats().CompactedPartLoans);
5455

5556
CheckInitiateBorrowedPartsReturn(ctx);

‎ydb/core/tx/datashard/datashard_ut_build_index.cpp

Lines changed: 113 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -95,7 +95,7 @@ Y_UNIT_TEST_SUITE(TTxDataShardBuildIndexScan) {
9595
CreateShardedTable(server, sender, root, name, opts);
9696
}
9797

98-
Y_UNIT_TEST(TestRunScan) {
98+
Y_UNIT_TEST(RunScan) {
9999
TPortManager pm;
100100
TServerSettings serverSettings(pm.GetPort(2134));
101101
serverSettings.SetDomainName("Root")
@@ -130,8 +130,7 @@ Y_UNIT_TEST_SUITE(TTxDataShardBuildIndexScan) {
130130
// Alter table: disable shadow data and change compaction policy
131131
auto policy = NLocalDb::CreateDefaultUserTablePolicy();
132132
policy->KeepEraseMarkers = false;
133-
WaitTxNotification(server,
134-
AsyncAlterAndDisableShadow(server, "/Root", "table-2", policy.Get()));
133+
WaitTxNotification(server, AsyncAlterAndDisableShadow(server, "/Root", "table-2", policy.Get()));
135134

136135
// Shadow data must be visible now
137136
auto data2 = ReadShardedTable(server, "/Root/table-2");
@@ -140,6 +139,117 @@ Y_UNIT_TEST_SUITE(TTxDataShardBuildIndexScan) {
140139
"value = 300, key = 3\n"
141140
"value = 500, key = 5\n");
142141
}
142+
143+
Y_UNIT_TEST(ShadowBorrowCompaction) {
144+
TPortManager pm;
145+
TServerSettings serverSettings(pm.GetPort(2134));
146+
serverSettings
147+
.SetDomainName("Root")
148+
.SetUseRealThreads(false);
149+
150+
Tests::TServer::TPtr server = new TServer(serverSettings);
151+
auto &runtime = *server->GetRuntime();
152+
auto sender = runtime.AllocateEdgeActor();
153+
154+
runtime.SetLogPriority(NKikimrServices::TX_DATASHARD, NLog::PRI_DEBUG);
155+
runtime.SetLogPriority(NKikimrServices::FLAT_TX_SCHEMESHARD, NLog::PRI_TRACE);
156+
// runtime.SetLogPriority(NKikimrServices::TABLET_EXECUTOR, NLog::PRI_DEBUG);
157+
158+
// Allow manipulating shadow data using normal schemeshard operations
159+
runtime.GetAppData().AllowShadowDataInSchemeShardForTests = true;
160+
161+
InitRoot(server, sender);
162+
163+
CreateShardedTable(server, sender, "/Root", "table-1", 1, false);
164+
165+
// Upsert some initial values
166+
ExecSQL(server, sender, "UPSERT INTO `/Root/table-1` (key, value) VALUES (1, 100), (2, 200), (3, 300), (4, 400), (5, 500);");
167+
168+
CreateShardedTableForIndex(server, sender, "/Root", "table-2", 1, false);
169+
170+
auto observer = runtime.AddObserver<TEvDataShard::TEvCompactBorrowed>([&](TEvDataShard::TEvCompactBorrowed::TPtr& event) {
171+
IActor *actor = runtime.FindActor(event->Sender);
172+
if (actor && actor->GetActivityType() == 186) {
173+
Cerr << "Ignore SchemeShard TEvCompactBorrowed from " << event->Sender << "(" << actor->GetActivityType() << ")" << " to " << event->Recipient << Endl;
174+
Y_UNUSED(event.Release());
175+
}
176+
});
177+
178+
auto snapshot = CreateVolatileSnapshot(server, { "/Root/table-1" });
179+
180+
DoBuildIndex(server, sender, "/Root/table-1", "/Root/table-2", snapshot, NKikimrTxDataShard::TEvBuildIndexProgressResponse::DONE);
181+
182+
// Writes to shadow data should not be visible yet
183+
auto data = ReadShardedTable(server, "/Root/table-2");
184+
UNIT_ASSERT_VALUES_EQUAL(data, "");
185+
186+
// Split index
187+
auto shards1 = GetTableShards(server, sender, "/Root/table-2");
188+
UNIT_ASSERT_VALUES_EQUAL(shards1.size(), 1u);
189+
190+
// Split would fail otherwise :(
191+
SetSplitMergePartCountLimit(server->GetRuntime(), -1);
192+
193+
auto senderSplit = runtime.AllocateEdgeActor();
194+
ui64 txId = AsyncSplitTable(server, senderSplit, "/Root/table-2", shards1.at(0), 300);
195+
WaitTxNotification(server, senderSplit, txId);
196+
197+
auto shards2 = GetTableShards(server, sender, "/Root/table-2");
198+
UNIT_ASSERT_VALUES_EQUAL(shards2.size(), 2u);
199+
200+
for (auto shardIndex : xrange(2u)) {
201+
auto stats = WaitTableStats(runtime, shards2.at(shardIndex));
202+
// Cerr << "Received shard stats:" << Endl << stats.DebugString() << Endl;
203+
204+
UNIT_ASSERT_VALUES_EQUAL(stats.GetTableStats().GetRowCount(), shardIndex == 0 ? 2 : 3);
205+
206+
const auto& ownersProto = stats.GetUserTablePartOwners();
207+
THashSet<ui64> owners(ownersProto.begin(), ownersProto.end());
208+
// Note: datashard always adds current shard to part owners, even if there are no parts
209+
UNIT_ASSERT_VALUES_EQUAL(owners.size(), 2u);
210+
UNIT_ASSERT(owners.contains(shards1.at(0)));
211+
UNIT_ASSERT(owners.contains(shards2.at(shardIndex)));
212+
213+
auto tableId = ResolveTableId(server, sender, "/Root/table-2");
214+
auto result = CompactBorrowed(runtime, shards2.at(shardIndex), tableId);
215+
// Cerr << "Compact result " << result.DebugString() << Endl;
216+
UNIT_ASSERT_VALUES_EQUAL(result.GetTabletId(), shards2.at(shardIndex));
217+
UNIT_ASSERT_VALUES_EQUAL(result.GetPathId().GetOwnerId(), tableId.PathId.OwnerId);
218+
UNIT_ASSERT_VALUES_EQUAL(result.GetPathId().GetLocalId(), tableId.PathId.LocalPathId);
219+
220+
for (int i = 0; i < 5; ++i) {
221+
auto stats = WaitTableStats(runtime, shards2.at(shardIndex));
222+
// Cerr << "Received shard stats:" << Endl << stats.DebugString() << Endl;
223+
const auto& ownersProto = stats.GetUserTablePartOwners();
224+
THashSet<ui64> owners(ownersProto.begin(), ownersProto.end());
225+
if (i < 4) {
226+
if (owners.size() > 1) {
227+
continue;
228+
}
229+
if (owners.contains(shards1.at(0))) {
230+
continue;
231+
}
232+
}
233+
UNIT_ASSERT_VALUES_EQUAL(owners.size(), 1u);
234+
UNIT_ASSERT(owners.contains(shards2.at(shardIndex)));
235+
Cerr << "OK " << shards2.at(shardIndex) << Endl;
236+
}
237+
}
238+
239+
// Alter table: disable shadow data and change compaction policy
240+
auto policy = NLocalDb::CreateDefaultUserTablePolicy();
241+
policy->KeepEraseMarkers = false;
242+
WaitTxNotification(server, AsyncAlterAndDisableShadow(server, "/Root", "table-2", policy.Get()));
243+
244+
// Shadow data must be visible now
245+
auto data2 = ReadShardedTable(server, "/Root/table-2");
246+
UNIT_ASSERT_VALUES_EQUAL(data2,
247+
"value = 100, key = 1\n"
248+
"value = 200, key = 2\n"
249+
"value = 300, key = 3\n"
250+
"value = 400, key = 4\n"
251+
"value = 500, key = 5\n");
252+
}
143253
}
144254

145255
} // namespace NKikimr

‎ydb/core/tx/datashard/datashard_ut_compaction.cpp

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,7 @@ Y_UNIT_TEST_SUITE(DataShardCompaction) {
8787

8888
{
8989
auto stats = WaitTableStats(runtime, shards2.at(0));
90-
// Cerr << "Received shard stats:" << Endl << stats.DebugString();
90+
Cerr << "Received shard stats:" << Endl << stats.DebugString();
9191
const auto& ownersProto = stats.GetUserTablePartOwners();
9292
THashSet<ui64> owners(ownersProto.begin(), ownersProto.end());
9393
// NOTE: datashard always adds current shard to part owners, even if there are no parts
@@ -96,12 +96,18 @@ Y_UNIT_TEST_SUITE(DataShardCompaction) {
9696
UNIT_ASSERT(owners.contains(shards2.at(0)));
9797
}
9898

99-
auto tableId = ResolveTableId(server, sender, "/Root/table-1");
100-
CompactBorrowed(runtime, shards2.at(0), tableId);
99+
{
100+
auto tableId = ResolveTableId(server, sender, "/Root/table-1");
101+
auto result = CompactBorrowed(runtime, shards2.at(0), tableId);
102+
Cerr << "Compact result " << result.DebugString() << Endl;
103+
UNIT_ASSERT_VALUES_EQUAL(result.GetTabletId(), shards2.at(0));
104+
UNIT_ASSERT_VALUES_EQUAL(result.GetPathId().GetOwnerId(), tableId.PathId.OwnerId);
105+
UNIT_ASSERT_VALUES_EQUAL(result.GetPathId().GetLocalId(), tableId.PathId.LocalPathId);
106+
}
101107

102108
for (int i = 0; i < 5; ++i) {
103109
auto stats = WaitTableStats(runtime, shards2.at(0));
104-
// Cerr << "Received shard stats:" << Endl << stats.DebugString();
110+
// Cerr << "Received shard stats:" << Endl << stats.DebugString() << Endl;
105111
const auto& ownersProto = stats.GetUserTablePartOwners();
106112
THashSet<ui64> owners(ownersProto.begin(), ownersProto.end());
107113
if (i < 4) {

0 commit comments

Comments
 (0)