improve resharding tests (#5497)

ivanmorozov333 · web-flow · commit 7a49804db36c · 2024-06-13T17:06:15.000+03:00
diff --git a/ydb/core/kqp/ut/olap/blobs_sharing_ut.cpp b/ydb/core/kqp/ut/olap/blobs_sharing_ut.cpp
@@ -269,10 +269,10 @@ Y_UNIT_TEST_SUITE(KqpOlapBlobsSharing) {
     private:
         YDB_ACCESSOR(TString, ShardingType, "HASH_FUNCTION_CONSISTENCY_64");
 
-        void WaitResharding() {
+        void WaitResharding(const TString& hint = "") {
             const TInstant start = TInstant::Now();
             bool clean = false;
-            while (TInstant::Now() - start < TDuration::Seconds(200)) {
+            while (TInstant::Now() - start < TDuration::Seconds(20)) {
                 NYdb::NOperation::TOperationClient operationClient(Kikimr.GetDriver());
                 auto result = operationClient.List<NYdb::NSchemeShard::TBackgroundProcessesResponse>().GetValueSync();
                 UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), NYdb::EStatus::SUCCESS, result.GetIssues().ToString());
@@ -283,7 +283,7 @@ Y_UNIT_TEST_SUITE(KqpOlapBlobsSharing) {
                 }
                 UNIT_ASSERT_VALUES_EQUAL(result.GetList().size(), 1);
                 Sleep(TDuration::Seconds(1));
-                Cerr << "WAIT_FINISHED..." << Endl;
+                Cerr << "RESHARDING_WAIT_FINISHED... (" << hint << ")" << Endl;
             }
             AFL_VERIFY(clean);
         }
@@ -317,7 +317,7 @@ Y_UNIT_TEST_SUITE(KqpOlapBlobsSharing) {
             csController->SetLagForCompactionBeforeTierings(TDuration::Seconds(1));
             csController->SetOverrideReduceMemoryIntervalLimit(1LLU << 30);
 
-            TLocalHelper(Kikimr).SetShardingMethod(ShardingType).CreateTestOlapTable("olapTable", "olapStore", 16, 4);
+            TLocalHelper(Kikimr).SetShardingMethod(ShardingType).CreateTestOlapTable("olapTable", "olapStore", 24, 4);
             auto tableClient = Kikimr.GetTableClient();
 
             Tests::NCommon::TLoggerInit(Kikimr).SetComponents({ NKikimrServices::TX_COLUMNSHARD }, "CS").SetPriority(NActors::NLog::PRI_DEBUG).Initialize();
@@ -354,19 +354,23 @@ Y_UNIT_TEST_SUITE(KqpOlapBlobsSharing) {
             }
 
             CheckCount(230000);
-            {
+            for (ui32 i = 0; i < 2; ++i) {
                 auto alterQuery = TStringBuilder() << R"(ALTER OBJECT `/Root/olapStore/olapTable` (TYPE TABLESTORE) SET (ACTION=ALTER_SHARDING, MODIFICATION=SPLIT);)";
                 auto session = tableClient.CreateSession().GetValueSync().GetSession();
                 auto alterResult = session.ExecuteSchemeQuery(alterQuery).GetValueSync();
                 UNIT_ASSERT_VALUES_EQUAL_C(alterResult.GetStatus(), NYdb::EStatus::SUCCESS, alterResult.GetIssues().ToString());
+                WaitResharding("SPLIT:" + ::ToString(i));
+            }
+            {
+                auto alterQuery = TStringBuilder() << R"(ALTER OBJECT `/Root/olapStore/olapTable` (TYPE TABLESTORE) SET (ACTION=ALTER_SHARDING, MODIFICATION=SPLIT);)";
+                auto session = tableClient.CreateSession().GetValueSync().GetSession();
+                auto alterResult = session.ExecuteSchemeQuery(alterQuery).GetValueSync();
+                UNIT_ASSERT_VALUES_UNEQUAL_C(alterResult.GetStatus(), NYdb::EStatus::SUCCESS, alterResult.GetIssues().ToString());
             }
-
-            WaitResharding();
             AFL_VERIFY(csController->GetShardingFiltersCount().Val() == 0);
             CheckCount(230000);
-
             i64 count = csController->GetShardingFiltersCount().Val();
-            AFL_VERIFY(count == 16)("count", count);
+            AFL_VERIFY(count >= 16)("count", count);
             WriteTestData(Kikimr, "/Root/olapStore/olapTable", 1000000, 300000000, 10000);
             csController->WaitIndexation(TDuration::Seconds(5));
             csController->WaitCompactions(TDuration::Seconds(5));
@@ -376,15 +380,15 @@ Y_UNIT_TEST_SUITE(KqpOlapBlobsSharing) {
             CheckCount(230000);
 
             AFL_VERIFY(count == csController->GetShardingFiltersCount().Val())("count", count)("val", csController->GetShardingFiltersCount().Val());
-            const ui32 portionsCount = 8;
-            for (ui32 i = 0; i < 3; ++i) {
+            const ui32 portionsCount = 16;
+            for (ui32 i = 0; i < 4; ++i) {
                 {
                     auto alterQuery = TStringBuilder() << R"(ALTER OBJECT `/Root/olapStore/olapTable` (TYPE TABLESTORE) SET (ACTION=ALTER_SHARDING, MODIFICATION=MERGE);)";
                     auto session = tableClient.CreateSession().GetValueSync().GetSession();
                     auto alterResult = session.ExecuteSchemeQuery(alterQuery).GetValueSync();
                     UNIT_ASSERT_VALUES_EQUAL_C(alterResult.GetStatus(), NYdb::EStatus::SUCCESS, alterResult.GetIssues().ToString());
                 }
-                WaitResharding();
+                WaitResharding("MERGE:" + ::ToString(i));
                 //            csController->WaitCleaning(TDuration::Seconds(5));
 
                 CheckCount(230000);
diff --git a/ydb/core/protos/flat_scheme_op.proto b/ydb/core/protos/flat_scheme_op.proto
@@ -1649,6 +1649,7 @@ message TShardingTransfer {
     optional uint64 DestinationTabletId = 1;
     repeated uint64 SourceTabletIds = 2;
     optional bool Moving = 3 [default = false];
+    optional string SessionId = 4;
 }
 
 message TShardingTransfers {
diff --git a/ydb/core/tx/columnshard/data_sharing/common/session/common.cpp b/ydb/core/tx/columnshard/data_sharing/common/session/common.cpp
@@ -9,7 +9,7 @@
 namespace NKikimr::NOlap::NDataSharing {
 
 TString TCommonSession::DebugString() const {
-    return TStringBuilder() << "{id=" << SessionId << ";context=" << TransferContext.DebugString() << ";}";
+    return TStringBuilder() << "{id=" << SessionId << ";context=" << TransferContext.DebugString() << ";state=" << State << ";}";
 }
 
 bool TCommonSession::TryStart(const NColumnShard::TColumnShard& shard) {
diff --git a/ydb/core/tx/columnshard/data_sharing/common/session/common.h b/ydb/core/tx/columnshard/data_sharing/common/session/common.h
@@ -21,14 +21,15 @@ class TManager;
 namespace NKikimr::NOlap::NDataSharing {
 
 class TCommonSession {
-private:
+public:
     enum class EState {
         Created,
         Prepared,
         InProgress,
         Finished
     };
 
+private:
     static ui64 GetNextRuntimeId() {
         static TAtomicCounter Counter = 0;
         return (ui64)Counter.Inc();
@@ -56,6 +57,7 @@ class TCommonSession {
         : SessionId(sessionId)
         , Info(info)
         , TransferContext(transferContext) {
+        AFL_VERIFY(!!SessionId);
     }
 
     const TTransferContext& GetTransferContext() const {
@@ -70,6 +72,10 @@ class TCommonSession {
         return State == EState::Prepared;
     }
 
+    bool IsFinished() const {
+        return State == EState::Finished;
+    }
+
     bool IsInProgress() const {
         return State == EState::InProgress;
     }
diff --git a/ydb/core/tx/columnshard/data_sharing/common/session/ya.make b/ydb/core/tx/columnshard/data_sharing/common/session/ya.make
@@ -11,4 +11,6 @@ PEERDIR(
     ydb/core/tablet_flat
 )
 
+GENERATE_ENUM_SERIALIZATION(common.h)
+
 END()
diff --git a/ydb/core/tx/columnshard/transactions/operators/sharing.cpp b/ydb/core/tx/columnshard/transactions/operators/sharing.cpp
@@ -25,6 +25,7 @@ bool TSharingTransactionOperator::DoParse(TColumnShard& owner, const TString& da
     if (currentSession) {
         SessionExistsFlag = true;
         SharingTask = currentSession;
+        AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "session_exists")("session_id", SharingTask->GetSessionId())("info", SharingTask->DebugString());
     } else {
         SharingTask->Confirm();
     }
diff --git a/ydb/core/tx/columnshard/transactions/operators/sharing.h b/ydb/core/tx/columnshard/transactions/operators/sharing.h
@@ -26,7 +26,8 @@ class TSharingTransactionOperator: public IProposeTxOperator {
     virtual void DoFinishProposeOnComplete(TColumnShard& /*owner*/, const TActorContext& /*ctx*/) override {
     }
     virtual bool DoIsAsync() const override {
-        return true;
+        AFL_VERIFY(SharingTask);
+        return !SharingTask->IsFinished();
     }
     virtual bool DoParse(TColumnShard& owner, const TString& data) override;
     virtual TString DoDebugString() const override {
diff --git a/ydb/core/tx/schemeshard/olap/bg_tasks/tx_chain/actor.h b/ydb/core/tx/schemeshard/olap/bg_tasks/tx_chain/actor.h
@@ -17,6 +17,7 @@ class TTxChainActor: public NKikimr::NOlap::NBackground::TSessionActor {
     NActors::TActorId TxAllocatorClient;
 
     void SendCurrentTxToSS() {
+        AFL_NOTICE(NKikimrServices::TX_COLUMNSHARD)("chain_tx", SessionLogic->GetTxData().GetTransactions()[SessionLogic->GetStepForExecute()].DebugString());
         auto evModification = std::make_unique<TEvSchemeShard::TEvModifySchemeTransaction>(SessionLogic->GetCurrentTxIdVerified(), (ui64)TabletId);
         *evModification->Record.AddTransaction() = SessionLogic->GetTxData().GetTransactions()[SessionLogic->GetStepForExecute()];
         NActors::TActivationContext::AsActorContext().Send(TabletActorId, evModification.release());
diff --git a/ydb/core/tx/schemeshard/olap/layout/layout.h b/ydb/core/tx/schemeshard/olap/layout/layout.h
@@ -129,6 +129,13 @@ class TColumnTablesLayout {
 public:
     TColumnTablesLayout(std::vector<TTablesGroup>&& groups);
 
+    void RemoveGroupsWithPathId(const TPathId& pathId) {
+        const auto pred = [&](const TTablesGroup& item) {
+            return item.GetTableIds().GetIds().contains(pathId);
+        };
+        Groups.erase(std::remove_if(Groups.begin(), Groups.end(), pred), Groups.end());
+    }
+
     static std::vector<ui64> ShardIdxToTabletId(const std::vector<TShardIdx>& shards, const TSchemeShard& ss);
 
     static TColumnTablesLayout BuildTrivial(const std::vector<ui64>& tabletIds);
diff --git a/ydb/core/tx/schemeshard/olap/operations/alter/in_store/resharding/update.cpp b/ydb/core/tx/schemeshard/olap/operations/alter/in_store/resharding/update.cpp
@@ -26,6 +26,7 @@ TConclusionStatus TInStoreShardingUpdate::DoInitialize(const TUpdateInitializati
     auto layoutPolicy = storeInfo.GetTablesLayoutPolicy();
     auto currentLayout = context.GetSSOperationContext()->SS->ColumnTables.GetTablesLayout(TColumnTablesLayout::ShardIdxToTabletId(
         storeInfo.GetColumnShards(), *context.GetSSOperationContext()->SS));
+    currentLayout.RemoveGroupsWithPathId(context.GetOriginalEntity().GetPathId());
     auto tablePtr = context.GetSSOperationContext()->SS->ColumnTables.GetVerifiedPtr(context.GetOriginalEntity().GetPathId());
     if (context.GetModification()->GetAlterColumnTable().GetReshardColumnTable().GetIncrease()) {
         const ui32 shardsCount = inStoreTable.GetTableInfoPtrVerified()->GetColumnShards().size();
diff --git a/ydb/core/tx/schemeshard/olap/operations/alter/in_store/transfer/update.cpp b/ydb/core/tx/schemeshard/olap/operations/alter/in_store/transfer/update.cpp
@@ -13,7 +13,7 @@ NKikimr::TConclusionStatus TInStoreShardsTransfer::DoInitializeImpl(const TUpdat
     auto sharding = table.GetTableInfo()->GetShardingVerified(table.GetTableSchemaVerified());
     for (auto&& alter : context.GetModification()->GetAlterColumnTable().GetAlterShards().GetTransfer().GetTransfers()) {
         NKikimrColumnShardDataSharingProto::TDestinationSession destinationSession;
-        destinationSession.SetSessionId("SHARE_TO_SHARD::" + ::ToString(alter.GetDestinationTabletId()));
+        destinationSession.SetSessionId(alter.GetSessionId());
         *destinationSession.MutableInitiatorController() = NKikimr::NOlap::NDataSharing::TInitiatorControllerContainer(
             std::make_shared<NKikimr::NOlap::NDataSharing::TSSInitiatorController>(context.GetSSOperationContext()->SS->TabletID(), 0)).SerializeToProto();
         {
diff --git a/ydb/core/tx/sharding/hash_intervals.cpp b/ydb/core/tx/sharding/hash_intervals.cpp
@@ -28,6 +28,7 @@ NKikimr::TConclusion<std::vector<NKikimrSchemeOp::TAlterShards>> TConsistencySha
                 auto& transfer = *alter.MutableTransfer()->AddTransfers();
                 transfer.SetDestinationTabletId(newTabletIds[idx]);
                 transfer.AddSourceTabletIds(i);
+                transfer.SetSessionId("SPLIT_TO::" + ::ToString(::ToString(newTabletIds[idx])) + "::" + TGUID::CreateTimebased().AsGuidString());
                 result.emplace_back(alter);
             }
             {
@@ -74,6 +75,7 @@ NKikimr::TConclusion<std::vector<NKikimrSchemeOp::TAlterShards>> TConsistencySha
                 transfer.SetDestinationTabletId(newTabletIds[idx]);
                 transfer.AddSourceTabletIds(from1);
                 transfer.AddSourceTabletIds(from2);
+                transfer.SetSessionId("MERGE_TO::" + ::ToString(::ToString(newTabletIds[idx])) + "::" + TGUID::CreateTimebased().AsGuidString());
                 transfer.SetMoving(true);
                 result.emplace_back(alter);
             }
diff --git a/ydb/core/tx/sharding/hash_modulo.cpp b/ydb/core/tx/sharding/hash_modulo.cpp
@@ -64,6 +64,7 @@ NKikimr::TConclusion<std::vector<NKikimrSchemeOp::TAlterShards>> THashShardingMo
                 auto& transfer = *alter.MutableTransfer()->AddTransfers();
                 transfer.SetDestinationTabletId(newTabletIds[idx]);
                 transfer.AddSourceTabletIds(i);
+                transfer.SetSessionId("SPLIT_TO::" + ::ToString(::ToString(newTabletIds[idx])) + "::" + TGUID::CreateTimebased().AsGuidString());
                 result.emplace_back(alter);
             }
             {
@@ -194,6 +195,7 @@ NKikimr::TConclusion<std::vector<NKikimrSchemeOp::TAlterShards>> THashShardingMo
                 transfer.SetDestinationTabletId(newTabletIds[idx]);
                 transfer.AddSourceTabletIds(from1);
                 transfer.AddSourceTabletIds(from2);
+                transfer.SetSessionId("MERGE_TO::" + ::ToString(::ToString(newTabletIds[idx])) + "::" + TGUID::CreateTimebased().AsGuidString());
                 transfer.SetMoving(true);
                 result.emplace_back(alter);
             }

Original file line number	Diff line number	Diff line change
`@@ -1649,6 +1649,7 @@ message TShardingTransfer {`
`1649`	`1649`	`optional uint64 DestinationTabletId = 1;`
`1650`	`1650`	`repeated uint64 SourceTabletIds = 2;`
`1651`	`1651`	`optional bool Moving = 3 [default = false];`
	`1652`	`+ optional string SessionId = 4;`
`1652`	`1653`	`}`
`1653`	`1654`
`1654`	`1655`	`message TShardingTransfers {`
Original file line number	Diff line number	Diff line change
`@@ -9,7 +9,7 @@`
`9`	`9`	`namespace NKikimr::NOlap::NDataSharing {`
`10`	`10`
`11`	`11`	`TString TCommonSession::DebugString() const {`
`12`		`- return TStringBuilder() << "{id=" << SessionId << ";context=" << TransferContext.DebugString() << ";}";`
	`12`	`+ return TStringBuilder() << "{id=" << SessionId << ";context=" << TransferContext.DebugString() << ";state=" << State << ";}";`
`13`	`13`	`}`
`14`	`14`
`15`	`15`	`bool TCommonSession::TryStart(const NColumnShard::TColumnShard& shard) {`
Original file line number	Diff line number	Diff line change
`@@ -11,4 +11,6 @@ PEERDIR(`
`11`	`11`	`ydb/core/tablet_flat`
`12`	`12`	`)`
`13`	`13`
	`14`	`+GENERATE_ENUM_SERIALIZATION(common.h)`
	`15`	`+`
`14`	`16`	`END()`
Original file line number	Diff line number	Diff line change
`@@ -25,6 +25,7 @@ bool TSharingTransactionOperator::DoParse(TColumnShard& owner, const TString& da`
`25`	`25`	`if (currentSession) {`
`26`	`26`	`SessionExistsFlag = true;`
`27`	`27`	`SharingTask = currentSession;`
	`28`	`+ AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "session_exists")("session_id", SharingTask->GetSessionId())("info", SharingTask->DebugString());`
`28`	`29`	`} else {`
`29`	`30`	`SharingTask->Confirm();`
`30`	`31`	`}`
Original file line number	Diff line number	Diff line change
`@@ -26,7 +26,8 @@ class TSharingTransactionOperator: public IProposeTxOperator {`
`26`	`26`	`virtual void DoFinishProposeOnComplete(TColumnShard& /owner/, const TActorContext& /ctx/) override {`
`27`	`27`	`}`
`28`	`28`	`virtual bool DoIsAsync() const override {`
`29`		`- return true;`
	`29`	`+ AFL_VERIFY(SharingTask);`
	`30`	`+ return !SharingTask->IsFinished();`
`30`	`31`	`}`
`31`	`32`	`virtual bool DoParse(TColumnShard& owner, const TString& data) override;`
`32`	`33`	`virtual TString DoDebugString() const override {`
Original file line number	Diff line number	Diff line change
`@@ -13,7 +13,7 @@ NKikimr::TConclusionStatus TInStoreShardsTransfer::DoInitializeImpl(const TUpdat`
`13`	`13`	`auto sharding = table.GetTableInfo()->GetShardingVerified(table.GetTableSchemaVerified());`
`14`	`14`	`for (auto&& alter : context.GetModification()->GetAlterColumnTable().GetAlterShards().GetTransfer().GetTransfers()) {`
`15`	`15`	`NKikimrColumnShardDataSharingProto::TDestinationSession destinationSession;`
`16`		`- destinationSession.SetSessionId("SHARE_TO_SHARD::" + ::ToString(alter.GetDestinationTabletId()));`
	`16`	`+ destinationSession.SetSessionId(alter.GetSessionId());`
`17`	`17`	`*destinationSession.MutableInitiatorController() = NKikimr::NOlap::NDataSharing::TInitiatorControllerContainer(`
`18`	`18`	`std::make_shared<NKikimr::NOlap::NDataSharing::TSSInitiatorController>(context.GetSSOperationContext()->SS->TabletID(), 0)).SerializeToProto();`
`19`	`19`	`{`
Original file line number	Diff line number	Diff line change
`@@ -28,6 +28,7 @@ NKikimr::TConclusion<std::vector<NKikimrSchemeOp::TAlterShards>> TConsistencySha`
`28`	`28`	`auto& transfer = *alter.MutableTransfer()->AddTransfers();`
`29`	`29`	`transfer.SetDestinationTabletId(newTabletIds[idx]);`
`30`	`30`	`transfer.AddSourceTabletIds(i);`
	`31`	`+ transfer.SetSessionId("SPLIT_TO::" + ::ToString(::ToString(newTabletIds[idx])) + "::" + TGUID::CreateTimebased().AsGuidString());`
`31`	`32`	`result.emplace_back(alter);`
`32`	`33`	`}`
`33`	`34`	`{`
`@@ -74,6 +75,7 @@ NKikimr::TConclusion<std::vector<NKikimrSchemeOp::TAlterShards>> TConsistencySha`
`74`	`75`	`transfer.SetDestinationTabletId(newTabletIds[idx]);`
`75`	`76`	`transfer.AddSourceTabletIds(from1);`
`76`	`77`	`transfer.AddSourceTabletIds(from2);`
	`78`	`+ transfer.SetSessionId("MERGE_TO::" + ::ToString(::ToString(newTabletIds[idx])) + "::" + TGUID::CreateTimebased().AsGuidString());`
`77`	`79`	`transfer.SetMoving(true);`
`78`	`80`	`result.emplace_back(alter);`
`79`	`81`	`}`
Original file line number	Diff line number	Diff line change
`@@ -64,6 +64,7 @@ NKikimr::TConclusion<std::vector<NKikimrSchemeOp::TAlterShards>> THashShardingMo`
`64`	`64`	`auto& transfer = *alter.MutableTransfer()->AddTransfers();`
`65`	`65`	`transfer.SetDestinationTabletId(newTabletIds[idx]);`
`66`	`66`	`transfer.AddSourceTabletIds(i);`
	`67`	`+ transfer.SetSessionId("SPLIT_TO::" + ::ToString(::ToString(newTabletIds[idx])) + "::" + TGUID::CreateTimebased().AsGuidString());`
`67`	`68`	`result.emplace_back(alter);`
`68`	`69`	`}`
`69`	`70`	`{`
`@@ -194,6 +195,7 @@ NKikimr::TConclusion<std::vector<NKikimrSchemeOp::TAlterShards>> THashShardingMo`
`194`	`195`	`transfer.SetDestinationTabletId(newTabletIds[idx]);`
`195`	`196`	`transfer.AddSourceTabletIds(from1);`
`196`	`197`	`transfer.AddSourceTabletIds(from2);`
	`198`	`+ transfer.SetSessionId("MERGE_TO::" + ::ToString(::ToString(newTabletIds[idx])) + "::" + TGUID::CreateTimebased().AsGuidString());`
`197`	`199`	`transfer.SetMoving(true);`
`198`	`200`	`result.emplace_back(alter);`
`199`	`201`	`}`