Skip to content

Commit 8bab37e

Browse files
CyberROFLCyberROFL
authored
Merge 7d55970 into 456c93b 
2 parents 456c93b + 7d55970 commit 8bab37e

10 files changed

+227
-57
lines changed

‎ydb/core/protos/counters_datashard.proto‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -490,4 +490,5 @@ enum ETxTypes {
490490
TXTYPE_CLEANUP_VOLATILE = 80 [(TxTypeOpts) = {Name: "TxCleanupVolatile"}];
491491
TXTYPE_PLAN_PREDICTED_TXS = 81 [(TxTypeOpts) = {Name: "TxPlanPredictedTxs"}];
492492
TXTYPE_WRITE = 82 [(TxTypeOpts) = {Name: "TxWrite"}];
493+
TXTYPE_REMOVE_SCHEMA_SNAPSHOTS = 83 [(TxTypeOpts) = {Name: "TxRemoveSchemaSnapshots"}];
493494
}

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

Lines changed: 112 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -855,6 +855,28 @@ void TDataShard::PersistChangeRecord(NIceDb::TNiceDb& db, const TChangeRecord& r
855855
NIceDb::TUpdate<Schema::ChangeRecordDetails::Kind>(record.GetKind()),
856856
NIceDb::TUpdate<Schema::ChangeRecordDetails::Body>(record.GetBody()),
857857
NIceDb::TUpdate<Schema::ChangeRecordDetails::Source>(record.GetSource()));
858+
859+
auto res = ChangesQueue.emplace(record.GetOrder(), record);
860+
Y_VERIFY_S(res.second, "Duplicate change record: " << record.GetOrder());
861+
862+
if (res.first->second.SchemaVersion) {
863+
res.first->second.SchemaSnapshotAcquired = SchemaSnapshotManager.AcquireReference(
864+
TSchemaSnapshotKey(res.first->second.TableId, res.first->second.SchemaVersion));
865+
}
866+
867+
db.GetDatabase().OnRollback([this, order = record.GetOrder()] {
868+
auto it = ChangesQueue.find(order);
869+
Y_VERIFY_S(it != ChangesQueue.end(), "Cannot find change record: " << order);
870+
871+
if (it->second.SchemaSnapshotAcquired) {
872+
const auto snapshotKey = TSchemaSnapshotKey(it->second.TableId, it->second.SchemaVersion);
873+
if (const auto last = SchemaSnapshotManager.ReleaseReference(snapshotKey)) {
874+
ScheduleRemoveSchemaSnapshot(snapshotKey);
875+
}
876+
}
877+
878+
ChangesQueue.erase(it);
879+
});
858880
} else {
859881
auto& state = LockChangeRecords[lockId];
860882
Y_ABORT_UNLESS(state.Changes.empty() || state.Changes.back().LockOffset < record.GetLockOffset(),
@@ -934,6 +956,14 @@ void TDataShard::CommitLockChangeRecords(NIceDb::TNiceDb& db, ui64 lockId, ui64
934956
committed.Step = rowVersion.Step;
935957
committed.TxId = rowVersion.TxId;
936958
collected.push_back(committed);
959+
960+
auto res = ChangesQueue.emplace(committed.Order, committed);
961+
Y_VERIFY_S(res.second, "Duplicate change record: " << committed.Order);
962+
963+
if (res.first->second.SchemaVersion) {
964+
res.first->second.SchemaSnapshotAcquired = SchemaSnapshotManager.AcquireReference(
965+
TSchemaSnapshotKey(res.first->second.TableId, res.first->second.SchemaVersion));
966+
}
937967
}
938968

939969
Y_VERIFY_S(!CommittedLockChangeRecords.contains(lockId), "Cannot commit lock " << lockId << " more than once");
@@ -960,7 +990,26 @@ void TDataShard::CommitLockChangeRecords(NIceDb::TNiceDb& db, ui64 lockId, ui64
960990
LockChangeRecords.erase(it);
961991
});
962992
db.GetDatabase().OnRollback([this, lockId]() {
963-
CommittedLockChangeRecords.erase(lockId);
993+
auto it = CommittedLockChangeRecords.find(lockId);
994+
Y_VERIFY_S(it != CommittedLockChangeRecords.end(), "Unexpected failure to find lockId# " << lockId);
995+
996+
for (size_t i = 0; i < it->second.Count; ++i) {
997+
const ui64 order = it->second.Order + i;
998+
999+
auto cIt = ChangesQueue.find(order);
1000+
Y_VERIFY_S(cIt != ChangesQueue.end(), "Cannot find change record: " << order);
1001+
1002+
if (cIt->second.SchemaSnapshotAcquired) {
1003+
const auto snapshotKey = TSchemaSnapshotKey(cIt->second.TableId, cIt->second.SchemaVersion);
1004+
if (const auto last = SchemaSnapshotManager.ReleaseReference(snapshotKey)) {
1005+
ScheduleRemoveSchemaSnapshot(snapshotKey);
1006+
}
1007+
}
1008+
1009+
ChangesQueue.erase(cIt);
1010+
}
1011+
1012+
CommittedLockChangeRecords.erase(it);
9641013
});
9651014
}
9661015

@@ -1022,23 +1071,9 @@ void TDataShard::RemoveChangeRecord(NIceDb::TNiceDb& db, ui64 order) {
10221071
ChangesQueueBytes -= record.BodySize;
10231072

10241073
if (record.SchemaSnapshotAcquired) {
1025-
Y_ABORT_UNLESS(record.TableId);
1026-
auto tableIt = TableInfos.find(record.TableId.LocalPathId);
1027-
1028-
if (tableIt != TableInfos.end()) {
1029-
const auto snapshotKey = TSchemaSnapshotKey(record.TableId, record.SchemaVersion);
1030-
const bool last = SchemaSnapshotManager.ReleaseReference(snapshotKey);
1031-
1032-
if (last) {
1033-
const auto* snapshot = SchemaSnapshotManager.FindSnapshot(snapshotKey);
1034-
Y_ABORT_UNLESS(snapshot);
1035-
1036-
if (snapshot->Schema->GetTableSchemaVersion() < tableIt->second->GetTableSchemaVersion()) {
1037-
SchemaSnapshotManager.RemoveShapshot(db, snapshotKey);
1038-
}
1039-
}
1040-
} else {
1041-
Y_DEBUG_ABORT_UNLESS(State == TShardState::PreOffline);
1074+
const auto snapshotKey = TSchemaSnapshotKey(record.TableId, record.SchemaVersion);
1075+
if (const bool last = SchemaSnapshotManager.ReleaseReference(snapshotKey)) {
1076+
ScheduleRemoveSchemaSnapshot(snapshotKey);
10421077
}
10431078
}
10441079

@@ -1081,22 +1116,15 @@ void TDataShard::EnqueueChangeRecords(TVector<IDataShardChangeCollector::TChange
10811116
for (const auto& record : records) {
10821117
forward.emplace_back(record.Order, record.PathId, record.BodySize);
10831118

1084-
auto res = ChangesQueue.emplace(
1085-
std::piecewise_construct,
1086-
std::forward_as_tuple(record.Order),
1087-
std::forward_as_tuple(record, now, cookie)
1088-
);
1089-
if (res.second) {
1090-
ChangesList.PushBack(&res.first->second);
1119+
auto it = ChangesQueue.find(record.Order);
1120+
Y_ABORT_UNLESS(it != ChangesQueue.end());
10911121

1092-
Y_ABORT_UNLESS(ChangesQueueBytes <= (Max<ui64>() - record.BodySize));
1093-
ChangesQueueBytes += record.BodySize;
1122+
it->second.EnqueuedAt = now;
1123+
it->second.ReservationCookie = cookie;
1124+
ChangesList.PushBack(&it->second);
10941125

1095-
if (record.SchemaVersion) {
1096-
res.first->second.SchemaSnapshotAcquired = SchemaSnapshotManager.AcquireReference(
1097-
TSchemaSnapshotKey(record.TableId, record.SchemaVersion));
1098-
}
1099-
}
1126+
Y_ABORT_UNLESS(ChangesQueueBytes <= (Max<ui64>() - record.BodySize));
1127+
ChangesQueueBytes += record.BodySize;
11001128
}
11011129

11021130
if (auto it = ChangeQueueReservations.find(cookie); it != ChangeQueueReservations.end()) {
@@ -1265,6 +1293,12 @@ bool TDataShard::LoadChangeRecords(NIceDb::TNiceDb& db, TVector<IDataShardChange
12651293
.SchemaVersion = schemaVersion,
12661294
});
12671295

1296+
auto res = ChangesQueue.emplace(records.back().Order, records.back());
1297+
if (res.second && res.first->second.SchemaVersion) {
1298+
res.first->second.SchemaSnapshotAcquired = SchemaSnapshotManager.AcquireReference(
1299+
TSchemaSnapshotKey(res.first->second.TableId, res.first->second.SchemaVersion));
1300+
}
1301+
12681302
if (!rowset.Next()) {
12691303
return false;
12701304
}
@@ -1363,6 +1397,12 @@ bool TDataShard::LoadChangeRecordCommits(NIceDb::TNiceDb& db, TVector<IDataShard
13631397
});
13641398
entry.Count++;
13651399
needSort = true;
1400+
1401+
auto res = ChangesQueue.emplace(records.back().Order, records.back());
1402+
if (res.second && res.first->second.SchemaVersion) {
1403+
res.first->second.SchemaSnapshotAcquired = SchemaSnapshotManager.AcquireReference(
1404+
TSchemaSnapshotKey(res.first->second.TableId, res.first->second.SchemaVersion));
1405+
}
13661406
}
13671407

13681408
LockChangeRecords.erase(lockId);
@@ -1421,6 +1461,46 @@ void TDataShard::ScheduleRemoveAbandonedLockChanges() {
14211461
}
14221462
}
14231463

1464+
void TDataShard::ScheduleRemoveSchemaSnapshot(const TSchemaSnapshotKey& key) {
1465+
Y_ABORT_UNLESS(!SchemaSnapshotManager.HasReference(key));
1466+
1467+
const auto* snapshot = SchemaSnapshotManager.FindSnapshot(key);
1468+
Y_ABORT_UNLESS(snapshot);
1469+
1470+
auto it = TableInfos.find(key.PathId);
1471+
if (it == TableInfos.end()) {
1472+
Y_DEBUG_ABORT_UNLESS(State == TShardState::PreOffline);
1473+
return;
1474+
}
1475+
1476+
if (snapshot->Schema->GetTableSchemaVersion() < it->second->GetTableSchemaVersion()) {
1477+
bool wasEmpty = PendingSchemaSnapshotsToRemove.empty();
1478+
PendingSchemaSnapshotsToRemove.push_back(key);
1479+
if (wasEmpty) {
1480+
Send(SelfId(), new TEvPrivate::TEvRemoveSchemaSnapshots);
1481+
}
1482+
}
1483+
}
1484+
1485+
void TDataShard::ScheduleRemoveAbandonedSchemaSnapshots() {
1486+
bool wasEmpty = PendingSchemaSnapshotsToRemove.empty();
1487+
1488+
for (const auto& [key, snapshot] : SchemaSnapshotManager.GetSnapshots()) {
1489+
auto it = TableInfos.find(key.PathId);
1490+
if (it == TableInfos.end()) {
1491+
Y_DEBUG_ABORT_UNLESS(State == TShardState::PreOffline);
1492+
break;
1493+
}
1494+
if (snapshot.Schema->GetTableSchemaVersion() < it->second->GetTableSchemaVersion()) {
1495+
PendingSchemaSnapshotsToRemove.push_back(key);
1496+
}
1497+
}
1498+
1499+
if (wasEmpty && !PendingSchemaSnapshotsToRemove.empty()) {
1500+
Send(SelfId(), new TEvPrivate::TEvRemoveSchemaSnapshots);
1501+
}
1502+
}
1503+
14241504
void TDataShard::PersistSchemeTxResult(NIceDb::TNiceDb &db, const TSchemaOperation &op) {
14251505
db.Table<Schema::SchemaOperations>().Key(op.TxId).Update(
14261506
NIceDb::TUpdate<Schema::SchemaOperations::Success>(op.Success),

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

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -425,6 +425,12 @@ bool TDataShard::TTxInit::ReadEverything(TTransactionContext &txc) {
425425
return false;
426426
}
427427

428+
if (Self->State != TShardState::Offline && txc.DB.GetScheme().GetTableInfo(Schema::SchemaSnapshots::TableId)) {
429+
if (!Self->SchemaSnapshotManager.Load(db)) {
430+
return false;
431+
}
432+
}
433+
428434
if (Self->State != TShardState::Offline && txc.DB.GetScheme().GetTableInfo(Schema::ChangeRecords::TableId)) {
429435
if (!Self->LoadChangeRecords(db, ChangeRecords)) {
430436
return false;
@@ -512,12 +518,6 @@ bool TDataShard::TTxInit::ReadEverything(TTransactionContext &txc) {
512518
}
513519
}
514520

515-
if (Self->State != TShardState::Offline && txc.DB.GetScheme().GetTableInfo(Schema::SchemaSnapshots::TableId)) {
516-
if (!Self->SchemaSnapshotManager.Load(db)) {
517-
return false;
518-
}
519-
}
520-
521521
if (Self->State != TShardState::Offline && txc.DB.GetScheme().GetTableInfo(Schema::Locks::TableId)) {
522522
TDataShardLocksDb locksDb(*Self, txc);
523523
if (!Self->SysLocks.Load(locksDb)) {
@@ -547,6 +547,7 @@ bool TDataShard::TTxInit::ReadEverything(TTransactionContext &txc) {
547547
Self->SubscribeNewLocks();
548548

549549
Self->ScheduleRemoveAbandonedLockChanges();
550+
Self->ScheduleRemoveAbandonedSchemaSnapshots();
550551

551552
return true;
552553
}

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

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -340,15 +340,13 @@ class TDataShard::TTxRemoveChangeRecords: public TTransactionBase<TDataShard> {
340340
Self->RemoveChangeRecordsInFly = false;
341341
}
342342

343-
if (!Self->ChangesQueue) { // double check queue
344-
if (ChangeExchangeSplit) {
345-
Self->KillChangeSender(ctx);
346-
Self->ChangeExchangeSplitter.DoSplit(ctx);
347-
}
343+
if (ChangeExchangeSplit) {
344+
Self->KillChangeSender(ctx);
345+
Self->ChangeExchangeSplitter.DoSplit(ctx);
346+
}
348347

349-
for (const auto dstTabletId : ActivationList) {
350-
Self->ChangeSenderActivator.DoSend(dstTabletId, ctx);
351-
}
348+
for (const auto dstTabletId : ActivationList) {
349+
Self->ChangeSenderActivator.DoSend(dstTabletId, ctx);
352350
}
353351

354352
Self->CheckStateChange(ctx);

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

Lines changed: 24 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -241,6 +241,7 @@ class TDataShard
241241
class TTxCdcStreamScanProgress;
242242
class TTxCdcStreamEmitHeartbeats;
243243
class TTxUpdateFollowerReadEdge;
244+
class TTxRemoveSchemaSnapshots;
244245

245246
template <typename T> friend class TTxDirectBase;
246247
class TTxUploadRows;
@@ -374,6 +375,7 @@ class TDataShard
374375
EvPlanPredictedTxs,
375376
EvStatisticsScanFinished,
376377
EvTableStatsError,
378+
EvRemoveSchemaSnapshots,
377379
EvEnd
378380
};
379381

@@ -595,6 +597,8 @@ class TDataShard
595597
struct TEvPlanPredictedTxs : public TEventLocal<TEvPlanPredictedTxs, EvPlanPredictedTxs> {};
596598

597599
struct TEvStatisticsScanFinished : public TEventLocal<TEvStatisticsScanFinished, EvStatisticsScanFinished> {};
600+
601+
struct TEvRemoveSchemaSnapshots : public TEventLocal<TEvRemoveSchemaSnapshots, EvRemoveSchemaSnapshots> {};
598602
};
599603

600604
struct Schema : NIceDb::Schema {
@@ -1383,6 +1387,8 @@ class TDataShard
13831387

13841388
void Handle(TEvPrivate::TEvPlanPredictedTxs::TPtr& ev, const TActorContext& ctx);
13851389

1390+
void Handle(TEvPrivate::TEvRemoveSchemaSnapshots::TPtr& ev, const TActorContext& ctx);
1391+
13861392
void HandleByReplicationSourceOffsetsServer(STATEFN_SIG);
13871393

13881394
void DoPeriodicTasks(const TActorContext &ctx);
@@ -1920,6 +1926,8 @@ class TDataShard
19201926
bool LoadChangeRecordCommits(NIceDb::TNiceDb& db, TVector<IDataShardChangeCollector::TChange>& records);
19211927
void ScheduleRemoveLockChanges(ui64 lockId);
19221928
void ScheduleRemoveAbandonedLockChanges();
1929+
void ScheduleRemoveSchemaSnapshot(const TSchemaSnapshotKey& key);
1930+
void ScheduleRemoveAbandonedSchemaSnapshots();
19231931

19241932
static void PersistCdcStreamScanLastKey(NIceDb::TNiceDb& db, const TSerializedCellVec& value,
19251933
const TPathId& tablePathId, const TPathId& streamPathId);
@@ -2803,24 +2811,29 @@ class TDataShard
28032811
ui64 LockOffset;
28042812
ui64 ReservationCookie;
28052813

2806-
explicit TEnqueuedRecord(ui64 bodySize, const TPathId& tableId,
2807-
ui64 schemaVersion, TInstant created, TInstant enqueued,
2808-
ui64 lockId = 0, ui64 lockOffset = 0, ui64 cookie = 0)
2814+
explicit TEnqueuedRecord(ui64 bodySize, const TPathId& tableId, ui64 schemaVersion,
2815+
TInstant created, ui64 lockId = 0, ui64 lockOffset = 0)
28092816
: BodySize(bodySize)
28102817
, TableId(tableId)
28112818
, SchemaVersion(schemaVersion)
28122819
, SchemaSnapshotAcquired(false)
28132820
, CreatedAt(created)
2814-
, EnqueuedAt(enqueued)
2821+
, EnqueuedAt(TInstant::Zero())
28152822
, LockId(lockId)
28162823
, LockOffset(lockOffset)
2817-
, ReservationCookie(cookie)
2824+
, ReservationCookie(0)
2825+
{
2826+
}
2827+
2828+
explicit TEnqueuedRecord(const IDataShardChangeCollector::TChange& record)
2829+
: TEnqueuedRecord(record.BodySize, record.TableId, record.SchemaVersion,
2830+
record.CreatedAt(), record.LockId, record.LockOffset)
28182831
{
28192832
}
28202833

2821-
explicit TEnqueuedRecord(const IDataShardChangeCollector::TChange& record, TInstant now, ui64 cookie)
2822-
: TEnqueuedRecord(record.BodySize, record.TableId, record.SchemaVersion, record.CreatedAt(), now,
2823-
record.LockId, record.LockOffset, cookie)
2834+
explicit TEnqueuedRecord(const TChangeRecord& record)
2835+
: TEnqueuedRecord(record.GetBody().size(), record.GetTableId(), record.GetSchemaVersion(),
2836+
record.GetApproximateCreationDateTime(), record.GetLockId(), record.GetLockOffset())
28242837
{
28252838
}
28262839
};
@@ -2865,6 +2878,7 @@ class TDataShard
28652878
THashMap<ui64, TUncommittedLockChangeRecords> LockChangeRecords; // ui64 is lock id
28662879
THashMap<ui64, TCommittedLockChangeRecords> CommittedLockChangeRecords; // ui64 is lock id
28672880
TVector<ui64> PendingLockChangeRecordsToRemove;
2881+
TVector<TSchemaSnapshotKey> PendingSchemaSnapshotsToRemove;
28682882

28692883
// in
28702884
THashMap<ui64, TInChangeSender> InChangeSenders; // ui64 is shard id
@@ -2985,6 +2999,7 @@ class TDataShard
29852999
HFuncTraced(TEvMediatorTimecast::TEvNotifyPlanStep, Handle);
29863000
HFuncTraced(TEvPrivate::TEvMediatorRestoreBackup, Handle);
29873001
HFuncTraced(TEvPrivate::TEvRemoveLockChangeRecords, Handle);
3002+
HFuncTraced(TEvPrivate::TEvRemoveSchemaSnapshots, Handle);
29883003
default:
29893004
if (!HandleDefaultEvents(ev, SelfId())) {
29903005
ALOG_WARN(NKikimrServices::TX_DATASHARD, "TDataShard::StateInactive unhandled event type: " << ev->GetTypeRewrite()
@@ -3113,6 +3128,7 @@ class TDataShard
31133128
HFunc(TEvPrivate::TEvPlanPredictedTxs, Handle);
31143129
HFunc(NStat::TEvStatistics::TEvStatisticsRequest, Handle);
31153130
HFunc(TEvPrivate::TEvStatisticsScanFinished, Handle);
3131+
HFuncTraced(TEvPrivate::TEvRemoveSchemaSnapshots, Handle);
31163132
default:
31173133
if (!HandleDefaultEvents(ev, SelfId())) {
31183134
ALOG_WARN(NKikimrServices::TX_DATASHARD, "TDataShard::StateWork unhandled event type: " << ev->GetTypeRewrite() << " event: " << ev->ToString());

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

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,10 @@ void TSchemaSnapshotManager::RenameSnapshots(NTable::TDatabase& db,
119119
}
120120
}
121121

122+
const TSchemaSnapshotManager::TSnapshots& TSchemaSnapshotManager::GetSnapshots() const {
123+
return Snapshots;
124+
}
125+
122126
bool TSchemaSnapshotManager::AcquireReference(const TSchemaSnapshotKey& key) {
123127
auto it = Snapshots.find(key);
124128
if (it == Snapshots.end()) {
@@ -152,6 +156,15 @@ bool TSchemaSnapshotManager::ReleaseReference(const TSchemaSnapshotKey& key) {
152156
return true;
153157
}
154158

159+
bool TSchemaSnapshotManager::HasReference(const TSchemaSnapshotKey& key) const {
160+
auto refIt = References.find(key);
161+
if (refIt == References.end()) {
162+
return refIt->second;
163+
} else {
164+
return false;
165+
}
166+
}
167+
155168
void TSchemaSnapshotManager::PersistAddSnapshot(NIceDb::TNiceDb& db, const TSchemaSnapshotKey& key, const TSchemaSnapshot& snapshot) {
156169
using Schema = TDataShard::Schema;
157170
db.Table<Schema::SchemaSnapshots>()

0 commit comments

Comments
 (0)