Skip to content

Fix TExportToS3WithRebootsTests CancelShouldSucceedOnManyTables test after auto dropping #17381

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 9 commits into from
Apr 21, 2025
Merged

Fix TExportToS3WithRebootsTests CancelShouldSucceedOnManyTables test after auto dropping #17381

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
73799e2
fix
stanislav-shchetinin Apr 17, 2025
369f5e7
fix blockPartiotion01
stanislav-shchetinin Apr 17, 2025
82e22ea
fix
stanislav-shchetinin Apr 17, 2025
c775b27
Apply suggestions from code review
stanislav-shchetinin Apr 17, 2025
edb7ae0
extra diff
stanislav-shchetinin Apr 17, 2025
994ac08
removed test
stanislav-shchetinin Apr 17, 2025
3241249
removed
stanislav-shchetinin Apr 17, 2025
c7cd25f
extra
stanislav-shchetinin Apr 17, 2025
a565453
fix view
stanislav-shchetinin Apr 18, 2025
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
1 change: 0 additions & 1 deletion .github/config/muted_ya.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,6 @@ ydb/core/quoter/ut QuoterWithKesusTest.KesusRecreation
ydb/core/quoter/ut QuoterWithKesusTest.PrefetchCoefficient
ydb/core/statistics/aggregator/ut AnalyzeColumnshard.AnalyzeRebootColumnShard
ydb/core/tx/datashard/ut_incremental_backup IncrementalBackup.ComplexRestoreBackupCollection+WithIncremental
ydb/core/tx/schemeshard/ut_export_reboots_s3 TExportToS3WithRebootsTests.CancelShouldSucceedOnManyTables
ydb/core/tx/schemeshard/ut_export_reboots_s3 TExportToS3WithRebootsTests.ShouldSucceedOnManyTables
ydb/core/tx/schemeshard/ut_login_large TSchemeShardLoginLargeTest.RemoveLogin_Many
ydb/core/tx/schemeshard/ut_move_reboots TSchemeShardMoveRebootsTest.WithData
Expand Down
20 changes: 15 additions & 5 deletions ydb/core/tx/schemeshard/schemeshard_export__create.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -554,6 +554,18 @@ struct TSchemeShard::TExport::TTxProgress: public TSchemeShard::TXxport::TTxBase
Send(Self->TxAllocatorClient, new TEvTxAllocatorClient::TEvAllocate(), 0, exportInfo->Id);
}

void PrepareAutoDropping(TSchemeShard* ss, TExportInfo::TPtr exportInfo, NIceDb::TNiceDb& db) {
bool isContinued = false;
PrepareDropping(ss, exportInfo, db, TExportInfo::EState::AutoDropping, [&](ui64 itemIdx) {
exportInfo->PendingDropItems.push_back(itemIdx);
isContinued = true;
AllocateTxId(exportInfo);
});
if (!isContinued) {
AllocateTxId(exportInfo);
}
}

void SubscribeTx(TTxId txId) {
Send(Self->SelfId(), new TEvSchemeShard::TEvNotifyTxCompletion(ui64(txId)));
}
Expand Down Expand Up @@ -1146,8 +1158,7 @@ struct TSchemeShard::TExport::TTxProgress: public TSchemeShard::TXxport::TTxBase
Self->PersistExportItemState(db, exportInfo, itemIdx);

if (AllOf(exportInfo->Items, &TExportInfo::TItem::IsDone)) {
PrepareDropping(Self, exportInfo, db, true);
AllocateTxId(exportInfo);
PrepareAutoDropping(Self, exportInfo, db);
}
} else if (exportInfo->State == EState::Cancellation) {
item.State = EState::Cancelled;
Expand Down Expand Up @@ -1347,8 +1358,7 @@ struct TSchemeShard::TExport::TTxProgress: public TSchemeShard::TXxport::TTxBase
}
}
if (!itemHasIssues && AllOf(exportInfo->Items, &TExportInfo::TItem::IsDone)) {
PrepareDropping(Self, exportInfo, db, true);
AllocateTxId(exportInfo);
PrepareAutoDropping(Self, exportInfo, db);
}

Self->PersistExportItemState(db, exportInfo, itemIdx);
Expand All @@ -1365,7 +1375,7 @@ struct TSchemeShard::TExport::TTxProgress: public TSchemeShard::TXxport::TTxBase
item.WaitTxId = InvalidTxId;
Self->PersistExportItemState(db, exportInfo, itemIdx);

if (exportInfo->AllItemsAreDropped() || exportInfo->State == EState::AutoDropping) {
if (exportInfo->AllItemsAreDropped()) {
AllocateTxId(exportInfo);
}
} else {
Expand Down
20 changes: 16 additions & 4 deletions ydb/core/tx/schemeshard/schemeshard_export_flow_proposals.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -328,9 +328,17 @@ TString ExportItemPathName(const TString& exportPathName, ui32 itemIdx) {
return TStringBuilder() << exportPathName << "/" << itemIdx;
}

void PrepareDropping(TSchemeShard* ss, TExportInfo::TPtr exportInfo, NIceDb::TNiceDb& db, bool isAutoDropping) {
void PrepareDropping(
TSchemeShard* ss,
TExportInfo::TPtr exportInfo,
NIceDb::TNiceDb& db,
TExportInfo::EState droppingState,
std::function<void(ui64)> func)
{
Y_ABORT_UNLESS(IsIn({TExportInfo::EState::AutoDropping, TExportInfo::EState::Dropping}, droppingState));

exportInfo->WaitTxId = InvalidTxId;
exportInfo->State = isAutoDropping ? TExportInfo::EState::AutoDropping : TExportInfo::EState::Dropping;
exportInfo->State = droppingState;
ss->PersistExportState(db, exportInfo);

for (ui32 itemIdx : xrange(exportInfo->Items.size())) {
Expand All @@ -341,14 +349,18 @@ void PrepareDropping(TSchemeShard* ss, TExportInfo::TPtr exportInfo, NIceDb::TNi
const TPath itemPath = TPath::Resolve(ExportItemPathName(ss, exportInfo, itemIdx), ss);
if (itemPath.IsResolved() && !itemPath.IsDeleted()) {
item.State = TExportInfo::EState::Dropping;
if (isAutoDropping) {
exportInfo->PendingDropItems.push_back(itemIdx);
if (exportInfo->State == TExportInfo::EState::AutoDropping) {
func(itemIdx);
}
}

ss->PersistExportItemState(db, exportInfo, itemIdx);
}
}

void PrepareDropping(TSchemeShard* ss, TExportInfo::TPtr exportInfo, NIceDb::TNiceDb& db) {
PrepareDropping(ss, exportInfo, db, TExportInfo::EState::Dropping, [](ui64){});
}

} // NSchemeShard
} // NKikimr
4 changes: 3 additions & 1 deletion ydb/core/tx/schemeshard/schemeshard_export_flow_proposals.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,9 @@ THolder<TEvSchemeShard::TEvCancelTx> CancelPropose(
TString ExportItemPathName(TSchemeShard* ss, const TExportInfo::TPtr exportInfo, ui32 itemIdx);
TString ExportItemPathName(const TString& exportPathName, ui32 itemIdx);

void PrepareDropping(TSchemeShard* ss, TExportInfo::TPtr exportInfo, NIceDb::TNiceDb& db, bool isAutoDropping = false);
void PrepareDropping(TSchemeShard* ss, TExportInfo::TPtr exportInfo, NIceDb::TNiceDb& db,
TExportInfo::EState droppingState, std::function<void(ui64)> func);
void PrepareDropping(TSchemeShard* ss, TExportInfo::TPtr exportInfo, NIceDb::TNiceDb& db);

} // NSchemeShard
} // NKikimr
8 changes: 4 additions & 4 deletions ydb/core/tx/schemeshard/ut_export/ut_export.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1264,7 +1264,7 @@ partitioning_settings {
TTestEnv env(runtime);
ui64 txId = 100;

TBlockEvents<NKikimr::NWrappers::NExternalStorage::TEvPutObjectRequest> blockPartition0(runtime, [](auto&& ev) {
TBlockEvents<NKikimr::NWrappers::NExternalStorage::TEvPutObjectRequest> blockPartition01(runtime, [](auto&& ev) {
return ev->Get()->Request.GetKey() == "/data_01.csv";
});

Expand Down Expand Up @@ -1300,7 +1300,7 @@ partitioning_settings {
)", port));


runtime.WaitFor("put object request from 01 partition", [&]{ return blockPartition0.size() >= 1; });
runtime.WaitFor("put object request from 01 partition", [&]{ return blockPartition01.size() >= 1; });
bool isCompleted = false;

while (!isCompleted) {
Expand All @@ -1318,8 +1318,8 @@ partitioning_settings {
}
}

blockPartition0.Stop();
blockPartition0.Unblock();
blockPartition01.Stop();
blockPartition01.Unblock();

env.TestWaitNotification(runtime, txId);

Expand Down
Loading