diff options
| author | ijon <[email protected]> | 2024-10-21 18:14:36 +0300 |
|---|---|---|
| committer | GitHub <[email protected]> | 2024-10-21 18:14:36 +0300 |
| commit | ff01ea7d10c8495ae6ce28097ff63d13760ae785 (patch) | |
| tree | 348220cb2b1a175865c40db2dd75003775923150 | |
| parent | 989a4fa04b3fdab90e58f07f7f995cc936cc6763 (diff) | |
schemeshard: split schemeshard__operation_common.cpp, move code from .h (#10631)
Move operations specific content of schemeshard__operation_common.cpp (bsv, cdc_stream, pq, subdomain) into separate files.
Move dependency rich methods definitions out of schemeshard__operation_common.h.
This is part of "improve schemeshard operations build time" effort (#10633).
8 files changed, 2117 insertions, 1990 deletions
diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_common.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_common.cpp index 19fe974c4b6..c6260ddbd40 100644 --- a/ydb/core/tx/schemeshard/schemeshard__operation_common.cpp +++ b/ydb/core/tx/schemeshard/schemeshard__operation_common.cpp @@ -82,9 +82,393 @@ THolder<TEvHive::TEvCreateTablet> CreateEvCreateTablet(TPathElement::TPtr target return ev; } +// TCreateParts +// +TCreateParts::TCreateParts(const TOperationId& id) + : OperationId(id) +{ + IgnoreMessages(DebugHint(), {}); +} + +bool TCreateParts::HandleReply(TEvHive::TEvAdoptTabletReply::TPtr& ev, TOperationContext& context) { + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " HandleReply TEvAdoptTablet" + << ", at tabletId: " << context.SS->SelfTabletId()); + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " HandleReply TEvAdoptTablet" + << ", message% " << DebugReply(ev)); + + NIceDb::TNiceDb db(context.GetDB()); + + const TString& explain = ev->Get()->Record.GetExplain(); + TTabletId tabletId = TTabletId(ev->Get()->Record.GetTabletID()); // global id from hive + auto shardIdx = context.SS->MakeLocalId(TLocalShardIdx(ev->Get()->Record.GetOwnerIdx())); // global id from hive + TTabletId hive = TTabletId(ev->Get()->Record.GetOrigin()); + + Y_ABORT_UNLESS(ui64(context.SS->SelfTabletId()) == ev->Get()->Record.GetOwner()); + + NKikimrProto::EReplyStatus status = ev->Get()->Record.GetStatus(); + Y_VERIFY_S(status == NKikimrProto::OK || status == NKikimrProto::ALREADY, + "Unexpected status " << NKikimrProto::EReplyStatus_Name(status) + << " in AdoptTabletReply for tabletId " << tabletId + << " with explain " << explain); + + // Note that HIVE might send duplicate TTxAdoptTabletReply in case of restarts + // So we just ignore the event if we cannot find the Tx or if it is in a different + // state + + Y_ABORT_UNLESS(context.SS->ShardInfos.contains(shardIdx)); + + if (!context.SS->AdoptedShards.contains(shardIdx)) { + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " HandleReply TEvAdoptTablet" + << " Got TTxAdoptTabletReply for shard but it is not present in AdoptedShards" + << ", shardIdx: " << shardIdx + << ", tabletId:" << tabletId); + return false; + } + + TShardInfo& shardInfo = context.SS->ShardInfos[shardIdx]; + Y_ABORT_UNLESS(shardInfo.TabletID == InvalidTabletId || shardInfo.TabletID == tabletId); + + Y_ABORT_UNLESS(tabletId != InvalidTabletId); + shardInfo.TabletID = tabletId; + context.SS->TabletIdToShardIdx[tabletId] = shardIdx; + + TTxState* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + Y_ABORT_UNLESS(txState->State == TTxState::CreateParts); + + txState->ShardsInProgress.erase(shardIdx); + + context.SS->AdoptedShards.erase(shardIdx); + context.SS->PersistDeleteAdopted(db, shardIdx); + context.SS->PersistShardMapping(db, shardIdx, tabletId, shardInfo.PathId, OperationId.GetTxId(), shardInfo.TabletType); + + context.OnComplete.UnbindMsgFromPipe(OperationId, hive, shardIdx); + context.OnComplete.ActivateShardCreated(shardIdx, OperationId.GetTxId()); + + // If all datashards have been created + if (txState->ShardsInProgress.empty()) { + context.SS->ChangeTxState(db, OperationId, TTxState::ConfigureParts); + return true; + } + + return false; +} + +bool TCreateParts::HandleReply(TEvHive::TEvCreateTabletReply::TPtr& ev, TOperationContext& context) { + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " HandleReply TEvCreateTabletReply" + << ", at tabletId: " << context.SS->SelfTabletId()); + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " HandleReply TEvCreateTabletReply" + << ", message: " << DebugReply(ev)); + + NIceDb::TNiceDb db(context.GetDB()); + + auto shardIdx = TShardIdx(ev->Get()->Record.GetOwner(), + TLocalShardIdx(ev->Get()->Record.GetOwnerIdx())); + + auto tabletId = TTabletId(ev->Get()->Record.GetTabletID()); // global id from hive + auto hive = TTabletId(ev->Get()->Record.GetOrigin()); + + NKikimrProto::EReplyStatus status = ev->Get()->Record.GetStatus(); + + Y_VERIFY_S(status == NKikimrProto::OK + || status == NKikimrProto::ALREADY + || status == NKikimrProto::INVALID_OWNER + || status == NKikimrProto::BLOCKED, + "Unexpected status " << NKikimrProto::EReplyStatus_Name(status) + << " in CreateTabletReply shard idx " << shardIdx << " tabletId " << tabletId); + + if (status == NKikimrProto::BLOCKED) { + Y_ABORT_UNLESS(!context.SS->IsDomainSchemeShard); + + LOG_NOTICE_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " CreateRequest BLOCKED " + << " at Hive: " << hive + << " msg: " << DebugReply(ev)); + + // do not unsubscribe message + // context.OnComplete.UnbindMsgFromPipe(OperationId, hive, shardIdx); + + // just stay calm and hung until tenant schemeshard is deleted + return false; + } + + TTxState& txState = *context.SS->FindTx(OperationId); + + TShardInfo& shardInfo = context.SS->ShardInfos.at(shardIdx); + Y_ABORT_UNLESS(shardInfo.TabletID == InvalidTabletId || shardInfo.TabletID == tabletId); + + if (status == NKikimrProto::INVALID_OWNER) { + auto redirectTo = TTabletId(ev->Get()->Record.GetForwardRequest().GetHiveTabletId()); + Y_ABORT_UNLESS(redirectTo); + Y_ABORT_UNLESS(tabletId); + + context.OnComplete.UnbindMsgFromPipe(OperationId, hive, shardIdx); + + auto path = context.SS->PathsById.at(txState.TargetPathId); + auto request = CreateEvCreateTablet(path, shardIdx, context); + + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " CreateRequest" + << " Redirect from Hive: " << hive + << " to Hive: " << redirectTo + << " msg: " << request->Record.DebugString()); + + context.OnComplete.BindMsgToPipe(OperationId, redirectTo, shardIdx, request.Release()); + return false; + } + + if (shardInfo.TabletID == InvalidTabletId) { + switch (shardInfo.TabletType) { + case ETabletType::DataShard: + context.SS->TabletCounters->Simple()[COUNTER_TABLE_SHARD_ACTIVE_COUNT].Add(1); + break; + case ETabletType::Coordinator: + context.SS->TabletCounters->Simple()[COUNTER_SUB_DOMAIN_COORDINATOR_COUNT].Add(1); + break; + case ETabletType::Mediator: + context.SS->TabletCounters->Simple()[COUNTER_SUB_DOMAIN_MEDIATOR_COUNT].Add(1); + break; + case ETabletType::Hive: + context.SS->TabletCounters->Simple()[COUNTER_SUB_DOMAIN_HIVE_COUNT].Add(1); + break; + case ETabletType::SysViewProcessor: + context.SS->TabletCounters->Simple()[COUNTER_SYS_VIEW_PROCESSOR_COUNT].Add(1); + break; + case ETabletType::StatisticsAggregator: + context.SS->TabletCounters->Simple()[COUNTER_STATISTICS_AGGREGATOR_COUNT].Add(1); + break; + case ETabletType::BackupController: + context.SS->TabletCounters->Simple()[COUNTER_BACKUP_CONTROLLER_TABLET_COUNT].Add(1); + break; + default: + break; + } + } + + Y_ABORT_UNLESS(tabletId != InvalidTabletId); + shardInfo.TabletID = tabletId; + context.SS->TabletIdToShardIdx[tabletId] = shardIdx; + + Y_ABORT_UNLESS(OperationId.GetTxId() == shardInfo.CurrentTxId); + + txState.ShardsInProgress.erase(shardIdx); + + context.SS->PersistShardMapping(db, shardIdx, tabletId, shardInfo.PathId, OperationId.GetTxId(), shardInfo.TabletType); + context.OnComplete.UnbindMsgFromPipe(OperationId, hive, shardIdx); + context.OnComplete.ActivateShardCreated(shardIdx, OperationId.GetTxId()); + + // If all datashards have been created + if (txState.ShardsInProgress.empty()) { + context.SS->ChangeTxState(db, OperationId, TTxState::ConfigureParts); + return true; + } + + return false; +} + +THolder<TEvHive::TEvAdoptTablet> TCreateParts::AdoptRequest(TShardIdx shardIdx, TOperationContext& context) { + Y_ABORT_UNLESS(context.SS->AdoptedShards.contains(shardIdx)); + auto& adoptedShard = context.SS->AdoptedShards[shardIdx]; + auto& shard = context.SS->ShardInfos[shardIdx]; + + THolder<TEvHive::TEvAdoptTablet> ev = MakeHolder<TEvHive::TEvAdoptTablet>( + ui64(shard.TabletID), + adoptedShard.PrevOwner, ui64(adoptedShard.PrevShardIdx), + shard.TabletType, + ui64(context.SS->SelfTabletId()), ui64(shardIdx.GetLocalId())); + + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " AdoptRequest" + << " Event to Hive: " << ev->Record.DebugString().c_str()); + + return ev; +} + +bool TCreateParts::ProgressState(TOperationContext& context) { + TTabletId ssId = context.SS->SelfTabletId(); + + TTxState* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " ProgressState" + << ", operation type: " << TTxState::TypeName(txState->TxType) + << ", at tablet" << ssId); + + if (txState->TxType == TTxState::TxDropTable + || txState->TxType == TTxState::TxAlterTable + || txState->TxType == TTxState::TxBackup + || txState->TxType == TTxState::TxRestore) { + if (NTableState::CheckPartitioningChangedForTableModification(*txState, context)) { + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " ProgressState" + << " SourceTablePartitioningChangedForModification" + << ", tx type: " << TTxState::TypeName(txState->TxType)); + NTableState::UpdatePartitioningForTableModification(OperationId, *txState, context); + } + } else if (txState->TxType == TTxState::TxCopyTable) { + if (NTableState::SourceTablePartitioningChangedForCopyTable(*txState, context)) { + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " ProgressState" + << " SourceTablePartitioningChangedForCopyTable" + << ", tx type: " << TTxState::TypeName(txState->TxType)); + NTableState::UpdatePartitioningForCopyTable(OperationId, *txState, context); + } + } + + txState->ClearShardsInProgress(); + + bool nothingToDo = true; + for (const auto& shard : txState->Shards) { + if (shard.Operation != TTxState::CreateParts) { + continue; + } + nothingToDo = false; + + if (context.SS->AdoptedShards.contains(shard.Idx)) { + auto ev = AdoptRequest(shard.Idx, context); + context.OnComplete.BindMsgToPipe(OperationId, context.SS->GetGlobalHive(context.Ctx), shard.Idx, ev.Release()); + } else { + auto path = context.SS->PathsById.at(txState->TargetPathId); + auto ev = CreateEvCreateTablet(path, shard.Idx, context); + + auto hiveToRequest = context.SS->ResolveHive(shard.Idx, context.Ctx); + + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " CreateRequest" + << " Event to Hive: " << hiveToRequest + << " msg: "<< ev->Record.DebugString().c_str()); + + context.OnComplete.BindMsgToPipe(OperationId, hiveToRequest, shard.Idx, ev.Release()); + } + context.OnComplete.RouteByShardIdx(OperationId, shard.Idx); + } + + if (nothingToDo) { + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " ProgressState" + << " no shards to create, do next state"); + + NIceDb::TNiceDb db(context.GetDB()); + context.SS->ChangeTxState(db, OperationId, TTxState::ConfigureParts); + return true; + } + + txState->UpdateShardsInProgress(TTxState::CreateParts); + return false; +} + +// TDeleteParts +// +TDeleteParts::TDeleteParts(const TOperationId& id, TTxState::ETxState nextState) + : OperationId(id) + , NextState(nextState) +{ + IgnoreMessages(DebugHint(), {}); +} + +void TDeleteParts::DeleteShards(TOperationContext& context) { + const auto* txState = context.SS->FindTx(OperationId); + + // Initiate asynchronous deletion of all shards + for (const auto& shard : txState->Shards) { + context.OnComplete.DeleteShard(shard.Idx); + } +} -namespace +bool TDeleteParts::ProgressState(TOperationContext& context) { + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "[" << context.SS->SelfTabletId() << "] " << DebugHint() << " ProgressState"); + DeleteShards(context); + + NIceDb::TNiceDb db(context.GetDB()); + context.SS->ChangeTxState(db, OperationId, NextState); + return true; +} + +// TDeletePartsAndDone +// +TDeletePartsAndDone::TDeletePartsAndDone(const TOperationId& id) + : TDeleteParts(id) { + Y_UNUSED(NextState); +} + +bool TDeletePartsAndDone::ProgressState(TOperationContext& context) { + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "[" << context.SS->SelfTabletId() << "] " << DebugHint() << " ProgressState"); + DeleteShards(context); + + context.OnComplete.DoneOperation(OperationId); + return true; +} + +// TDone +// +TDone::TDone(const TOperationId& id) + : OperationId(id) +{ + IgnoreMessages(DebugHint(), AllIncomingEvents()); +} + +bool TDone::ProgressState(TOperationContext& context) { + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "[" << context.SS->SelfTabletId() << "] " << DebugHint() << " ProgressState"); + + const auto* txState = context.SS->FindTx(OperationId); + + const auto& pathId = txState->TargetPathId; + Y_ABORT_UNLESS(context.SS->PathsById.contains(pathId)); + TPathElement::TPtr path = context.SS->PathsById.at(pathId); + Y_VERIFY_S(path->PathState != TPathElement::EPathState::EPathStateNoChanges, "with context" + << ", PathState: " << NKikimrSchemeOp::EPathState_Name(path->PathState) + << ", PathId: " << path->PathId + << ", PathName: " << path->Name); + + if (path->IsPQGroup() && txState->IsCreate()) { + TPathElement::TPtr parentDir = context.SS->PathsById.at(path->ParentPathId); + // can't be removed until KIKIMR-8861 + // at least lets wrap it into condition + // it helps to actualize PATHSTATE inside children listing + context.SS->ClearDescribePathCaches(parentDir); + } + + if (txState->IsDrop()) { + context.OnComplete.ReleasePathState(OperationId, path->PathId, TPathElement::EPathState::EPathStateNotExist); + } else { + context.OnComplete.ReleasePathState(OperationId, path->PathId, TPathElement::EPathState::EPathStateNoChanges); + } + + if (txState->SourcePathId != InvalidPathId) { + Y_ABORT_UNLESS(context.SS->PathsById.contains(txState->SourcePathId)); + TPathElement::TPtr srcPath = context.SS->PathsById.at(txState->SourcePathId); + if (srcPath->PathState == TPathElement::EPathState::EPathStateCopying) { + context.OnComplete.ReleasePathState(OperationId, srcPath->PathId, TPathElement::EPathState::EPathStateNoChanges); + } + } + + // OlapStore tracks all tables that are under operation, make sure to unlink + if (context.SS->ColumnTables.contains(pathId)) { + auto tableInfo = context.SS->ColumnTables.at(pathId); + if (!tableInfo->IsStandalone()) { + const auto storePathId = tableInfo->GetOlapStorePathIdVerified(); + if (context.SS->OlapStores.contains(storePathId)) { + auto storeInfo = context.SS->OlapStores.at(storePathId); + storeInfo->ColumnTablesUnderOperation.erase(pathId); + } + } + } + + context.OnComplete.DoneOperation(OperationId); + return true; +} + +namespace { template <typename T, typename TFuncCheck, typename TFuncToString> bool CollectProposeTxResults( @@ -158,7 +542,9 @@ bool CollectProposeTxResults( } // anonymous namespace -bool NTableState::CollectProposeTransactionResults( +namespace NTableState { + +bool CollectProposeTransactionResults( const NKikimr::NSchemeShard::TOperationId &operationId, const TEvDataShard::TEvProposeTransactionResult::TPtr &ev, NKikimr::NSchemeShard::TOperationContext &context) @@ -174,7 +560,7 @@ bool NTableState::CollectProposeTransactionResults( return CollectProposeTxResults(ev, operationId, context, prepared, toString); } -bool NTableState::CollectProposeTransactionResults( +bool CollectProposeTransactionResults( const NKikimr::NSchemeShard::TOperationId& operationId, const TEvColumnShard::TEvProposeTransactionResult::TPtr& ev, NKikimr::NSchemeShard::TOperationContext& context) @@ -190,7 +576,7 @@ bool NTableState::CollectProposeTransactionResults( return CollectProposeTxResults(ev, operationId, context, prepared, toString); } -bool NTableState::CollectSchemaChanged( +bool CollectSchemaChanged( const TOperationId& operationId, const TEvDataShard::TEvSchemaChanged::TPtr& ev, TOperationContext& context) @@ -263,7 +649,7 @@ bool NTableState::CollectSchemaChanged( return false; } -void NTableState::AckAllSchemaChanges(const TOperationId &operationId, TTxState &txState, TOperationContext &context) { +void AckAllSchemaChanges(const TOperationId &operationId, TTxState &txState, TOperationContext &context) { TTabletId ssId = context.SS->SelfTabletId(); LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, @@ -290,7 +676,7 @@ void NTableState::AckAllSchemaChanges(const TOperationId &operationId, TTxState } } -bool NTableState::CheckPartitioningChangedForTableModification(TTxState &txState, TOperationContext &context) { +bool CheckPartitioningChangedForTableModification(TTxState &txState, TOperationContext &context) { Y_ABORT_UNLESS(context.SS->Tables.contains(txState.TargetPathId)); TTableInfo::TPtr table = context.SS->Tables.at(txState.TargetPathId); @@ -309,7 +695,7 @@ bool NTableState::CheckPartitioningChangedForTableModification(TTxState &txState return !shardIdxsLeft.empty(); } -void NTableState::UpdatePartitioningForTableModification(TOperationId operationId, TTxState &txState, TOperationContext &context) { +void UpdatePartitioningForTableModification(TOperationId operationId, TTxState &txState, TOperationContext &context) { Y_ABORT_UNLESS(!txState.TxShardsListFinalized, "Rebuilding the list of shards must not happen twice"); NIceDb::TNiceDb db(context.GetDB()); @@ -428,7 +814,7 @@ void NTableState::UpdatePartitioningForTableModification(TOperationId operationI txState.TxShardsListFinalized = true; } -bool NTableState::SourceTablePartitioningChangedForCopyTable(const TTxState &txState, TOperationContext &context) { +bool SourceTablePartitioningChangedForCopyTable(const TTxState &txState, TOperationContext &context) { Y_ABORT_UNLESS(txState.SourcePathId != InvalidPathId); Y_ABORT_UNLESS(txState.TargetPathId != InvalidPathId); const TTableInfo::TPtr srcTableInfo = *context.SS->Tables.FindPtr(txState.SourcePathId); @@ -453,7 +839,7 @@ bool NTableState::SourceTablePartitioningChangedForCopyTable(const TTxState &txS return !srcShardIdxsLeft.empty(); } -void NTableState::UpdatePartitioningForCopyTable(TOperationId operationId, TTxState &txState, TOperationContext &context) { +void UpdatePartitioningForCopyTable(TOperationId operationId, TTxState &txState, TOperationContext &context) { Y_ABORT_UNLESS(!txState.TxShardsListFinalized, "CopyTable can adjust partitioning only once"); // Source table must not be altered or drop while we are performing copying. So we put it into a special state. @@ -555,7 +941,7 @@ void NTableState::UpdatePartitioningForCopyTable(TOperationId operationId, TTxSt txState.TxShardsListFinalized = true; } -TVector<TTableShardInfo> NTableState::ApplyPartitioningCopyTable(const TShardInfo &templateDatashardInfo, TTableInfo::TPtr srcTableInfo, TTxState &txState, TSchemeShard *ss) { +TVector<TTableShardInfo> ApplyPartitioningCopyTable(const TShardInfo &templateDatashardInfo, TTableInfo::TPtr srcTableInfo, TTxState &txState, TSchemeShard *ss) { TVector<TTableShardInfo> dstPartitions = srcTableInfo->GetPartitions(); // Source table must not be altered or drop while we are performing copying. So we put it into a special state. @@ -580,336 +966,109 @@ TVector<TTableShardInfo> NTableState::ApplyPartitioningCopyTable(const TShardInf return dstPartitions; } -namespace NPQState { - -bool CollectProposeTransactionResults(const TOperationId& operationId, - const TEvPersQueue::TEvProposeTransactionResult::TPtr& ev, - TOperationContext& context) -{ - auto prepared = [](NKikimrPQ::TEvProposeTransactionResult::EStatus status) -> bool { - return status == NKikimrPQ::TEvProposeTransactionResult::PREPARED; - }; - - auto toString = [](NKikimrPQ::TEvProposeTransactionResult::EStatus status) -> TString { - return NKikimrPQ::TEvProposeTransactionResult_EStatus_Name(status); - }; - - return CollectProposeTxResults(ev, operationId, context, prepared, toString); -} - -bool CollectPQConfigChanged(const TOperationId& operationId, - const TEvPersQueue::TEvProposeTransactionResult::TPtr& ev, - TOperationContext& context) -{ - Y_ABORT_UNLESS(context.SS->FindTx(operationId)); - TTxState& txState = *context.SS->FindTx(operationId); - - const auto& evRecord = ev->Get()->Record; - if (evRecord.GetStatus() == NKikimrPQ::TEvProposeTransactionResult::COMPLETE) { - const auto ssId = context.SS->SelfTabletId(); - const TTabletId shardId(evRecord.GetOrigin()); - - const auto shardIdx = context.SS->MustGetShardIdx(shardId); - Y_ABORT_UNLESS(context.SS->ShardInfos.contains(shardIdx)); - - Y_ABORT_UNLESS(txState.State == TTxState::Propose); - - txState.ShardsInProgress.erase(shardIdx); - - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "CollectPQConfigChanged accept TEvPersQueue::TEvProposeTransactionResult" - << ", operationId: " << operationId - << ", shardIdx: " << shardIdx - << ", shard: " << shardId - << ", left await: " << txState.ShardsInProgress.size() - << ", txState.State: " << TTxState::StateName(txState.State) - << ", txState.ReadyForNotifications: " << txState.ReadyForNotifications - << ", at schemeshard: " << ssId); - } - - return txState.ShardsInProgress.empty(); -} - -bool CollectPQConfigChanged(const TOperationId& operationId, - const TEvPersQueue::TEvProposeTransactionAttachResult::TPtr& ev, - TOperationContext& context) +// NTableState::TProposedWaitParts +// +TProposedWaitParts::TProposedWaitParts(TOperationId id, TTxState::ETxState nextState) + : OperationId(id) + , NextState(nextState) { - Y_ABORT_UNLESS(context.SS->FindTx(operationId)); - TTxState& txState = *context.SS->FindTx(operationId); - - // - // The PQ tablet can perform a transaction and send a TEvProposeTransactionResult(COMPLETE) response. - // The SchemeShard tablet can restart at this point. After restarting at the TPropose step, it will - // send the TEvProposeTransactionAttach message to the PQ tablets. If the NODATA status is specified in - // the response TEvProposeTransactionAttachResult, then the PQ tablet has already completed the transaction. - // Otherwise, she continues to execute the transaction - // - - const auto& evRecord = ev->Get()->Record; - if (evRecord.GetStatus() != NKikimrProto::NODATA) { - // - // If the PQ tablet returned something other than NODATA, then it continues to execute the transaction - // - return txState.ShardsInProgress.empty(); - } - - // - // Otherwise, she has already completed the transaction and has forgotten about it. Then we can - // remove PQ tablet from the list of shards - // - - const auto ssId = context.SS->SelfTabletId(); - const TTabletId shardId(evRecord.GetTabletId()); - - const auto shardIdx = context.SS->MustGetShardIdx(shardId); - Y_ABORT_UNLESS(context.SS->ShardInfos.contains(shardIdx)); - - Y_ABORT_UNLESS(txState.State == TTxState::Propose); - - txState.ShardsInProgress.erase(shardIdx); - - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "CollectPQConfigChanged accept TEvPersQueue::TEvProposeTransactionAttachResult" - << ", operationId: " << operationId - << ", shardIdx: " << shardIdx - << ", shard: " << shardId - << ", left await: " << txState.ShardsInProgress.size() - << ", txState.State: " << TTxState::StateName(txState.State) - << ", txState.ReadyForNotifications: " << txState.ReadyForNotifications - << ", at schemeshard: " << ssId); - - return txState.ShardsInProgress.empty(); -} - -bool TConfigureParts::HandleReply(TEvPersQueue::TEvProposeTransactionResult::TPtr& ev, TOperationContext& context) -{ - const TTabletId ssId = context.SS->SelfTabletId(); - - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvProposeTransactionResult" - << ", at schemeshard: " << ssId); - - return CollectProposeTransactionResults(OperationId, ev, context); + IgnoreMessages(DebugHint(), + { TEvHive::TEvCreateTabletReply::EventType + , TEvDataShard::TEvProposeTransactionResult::EventType + , TEvPrivate::TEvOperationPlan::EventType } + ); } -THolder<TEvPersQueue::TEvProposeTransaction> TConfigureParts::MakeEvProposeTransaction(TTxId txId, - const TTopicInfo& pqGroup, - const TTopicTabletInfo& pqShard, - const TString& topicName, - const TString& topicPath, - const std::optional<TBootstrapConfigWrapper>& bootstrapConfig, - const TString& cloudId, - const TString& folderId, - const TString& databaseId, - const TString& databasePath, - TTxState::ETxType txType, - const TOperationContext& context) -{ - auto event = MakeHolder<TEvPersQueue::TEvProposeTransactionBuilder>(); - event->Record.SetTxId(ui64(txId)); - ActorIdToProto(context.SS->SelfId(), event->Record.MutableSourceActor()); - - MakePQTabletConfig(context, - *event->Record.MutableConfig()->MutableTabletConfig(), - pqGroup, - pqShard, - topicName, - topicPath, - cloudId, - folderId, - databaseId, - databasePath); - if (bootstrapConfig) { - Y_ABORT_UNLESS(txType == TTxState::TxCreatePQGroup); - event->PreSerializedData += bootstrapConfig->GetPreSerializedProposeTransaction(); - } - - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "Propose configure PersQueue" << - ", message: " << event->Record.ShortUtf8DebugString()); - - return event; -} - -THolder<TEvPersQueue::TEvUpdateConfig> TConfigureParts::MakeEvUpdateConfig(TTxId txId, - const TTopicInfo& pqGroup, - const TTopicTabletInfo& pqShard, - const TString& topicName, - const TString& topicPath, - const std::optional<TBootstrapConfigWrapper>& bootstrapConfig, - const TString& cloudId, - const TString& folderId, - const TString& databaseId, - const TString& databasePath, - TTxState::ETxType txType, - const TOperationContext& context) -{ - auto event = MakeHolder<TEvPersQueue::TEvUpdateConfigBuilder>(); - event->Record.SetTxId(ui64(txId)); - - MakePQTabletConfig(context, - *event->Record.MutableTabletConfig(), - pqGroup, - pqShard, - topicName, - topicPath, - cloudId, - folderId, - databaseId, - databasePath); - if (bootstrapConfig) { - Y_ABORT_UNLESS(txType == TTxState::TxCreatePQGroup); - event->PreSerializedData += bootstrapConfig->GetPreSerializedUpdateConfig(); - } - - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "Propose configure PersQueue" << - ", message: " << event->Record.ShortUtf8DebugString()); - - return event; -} - -bool TPropose::HandleReply(TEvPersQueue::TEvProposeTransactionResult::TPtr& ev, TOperationContext& context) -{ - const TTabletId ssId = context.SS->SelfTabletId(); - const auto& evRecord = ev->Get()->Record; - - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvProposeTransactionResult" - << " triggers early" - << ", at schemeshard: " << ssId - << " message# " << evRecord.ShortDebugString()); - - const bool collected = CollectPQConfigChanged(OperationId, ev, context); - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvProposeTransactionResult" - << " CollectPQConfigChanged: " << (collected ? "true" : "false")); - - return TryPersistState(context); -} - -bool TPropose::HandleReply(TEvPersQueue::TEvProposeTransactionAttachResult::TPtr& ev, TOperationContext& context) -{ - const auto ssId = context.SS->SelfTabletId(); +bool TProposedWaitParts::HandleReply(TEvDataShard::TEvSchemaChanged::TPtr& ev, TOperationContext& context) { + TTabletId ssId = context.SS->SelfTabletId(); const auto& evRecord = ev->Get()->Record; LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvProposeTransactionAttachResult" - << " triggers early" - << ", at schemeshard: " << ssId - << " message# " << evRecord.ShortDebugString()); - - const bool collected = CollectPQConfigChanged(OperationId, ev, context); + DebugHint() << " HandleReply TEvSchemaChanged" + << " at tablet: " << ssId); LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvProposeTransactionAttachResult" - << " CollectPQConfigChanged: " << (collected ? "true" : "false")); - - return TryPersistState(context); -} - -void TPropose::PrepareShards(TTxState& txState, TSet<TTabletId>& shardSet, TOperationContext& context) -{ - txState.UpdateShardsInProgress(); - - for (const auto& shard : txState.Shards) { - const TShardIdx idx = shard.Idx; - // - // The operation involves the ETabletType::PersQueue and Tabletype::PersQueueReadBalancer shards. - // The program receives responses from PersQueueReadBalancer at the previous stage. At this stage, - // it only expects TEvProposeTransactionResult from PersQueue - // - if (shard.TabletType == ETabletType::PersQueue) { - const TTabletId tablet = context.SS->ShardInfos.at(idx).TabletID; - - shardSet.insert(tablet); + DebugHint() << " HandleReply TEvSchemaChanged" + << " at tablet: " << ssId + << " message: " << evRecord.ShortDebugString()); - // - // By this point, the SchemeShard tablet could restart and the actor ID changed. Therefore, we send - // the TEvProposeTransactionAttach message to the PQ tablets so that they recognize the new recipient - // - SendEvProposeTransactionAttach(idx, tablet, context); - } else { - txState.ShardsInProgress.erase(idx); - } - } -} - -void TPropose::SendEvProposeTransactionAttach(TShardIdx shard, TTabletId tablet, - TOperationContext& context) -{ - auto event = - MakeHolder<TEvPersQueue::TEvProposeTransactionAttach>(ui64(tablet), - ui64(OperationId.GetTxId())); - context.OnComplete.BindMsgToPipe(OperationId, tablet, shard, event.Release()); -} - -bool TPropose::CanPersistState(const TTxState& txState, - TOperationContext& context) -{ - if (!txState.ShardsInProgress.empty()) { + if (!CollectSchemaChanged(OperationId, ev, context)) { LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " can't persist state: " << - "ShardsInProgress is not empty, remain: " << txState.ShardsInProgress.size()); + DebugHint() << " HandleReply TEvSchemaChanged" + << " CollectSchemaChanged: false"); return false; } - PathId = txState.TargetPathId; - Path = context.SS->PathsById.at(PathId); + Y_ABORT_UNLESS(context.SS->FindTx(OperationId)); + TTxState& txState = *context.SS->FindTx(OperationId); - if (Path->StepCreated == InvalidStepId) { + if (!txState.ReadyForNotifications) { LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " can't persist state: " << - "StepCreated is invalid"); + DebugHint() << " HandleReply TEvSchemaChanged" + << " ReadyForNotifications: false"); return false; } return true; } -void TPropose::PersistState(const TTxState& txState, - TOperationContext& context) const -{ - NIceDb::TNiceDb db(context.GetDB()); +bool TProposedWaitParts::ProgressState(TOperationContext& context) { + TTabletId ssId = context.SS->SelfTabletId(); - if (txState.TxType == TTxState::TxCreatePQGroup) { - auto parentDir = context.SS->PathsById.at(Path->ParentPathId); - ++parentDir->DirAlterVersion; - context.SS->PersistPathDirAlterVersion(db, parentDir); - context.SS->ClearDescribePathCaches(parentDir); - context.OnComplete.PublishToSchemeBoard(OperationId, parentDir->PathId); - } + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " ProgressState" + << " at tablet: " << ssId); - context.SS->ClearDescribePathCaches(Path); - context.OnComplete.PublishToSchemeBoard(OperationId, PathId); + TTxState* txState = context.SS->FindTx(OperationId); - TTopicInfo::TPtr pqGroup = context.SS->Topics[PathId]; + NIceDb::TNiceDb db(context.GetDB()); - NKikimrPQ::TPQTabletConfig tabletConfig = pqGroup->GetTabletConfig(); - NKikimrPQ::TPQTabletConfig newTabletConfig = pqGroup->AlterData->GetTabletConfig(); + txState->ClearShardsInProgress(); + for (TTxState::TShardOperation& shard : txState->Shards) { + if (shard.Operation < TTxState::ProposedWaitParts) { + shard.Operation = TTxState::ProposedWaitParts; + context.SS->PersistUpdateTxShard(db, OperationId, shard.Idx, shard.Operation); + } + Y_ABORT_UNLESS(context.SS->ShardInfos.contains(shard.Idx)); + context.OnComplete.RouteByTablet(OperationId, context.SS->ShardInfos.at(shard.Idx).TabletID); + } + txState->UpdateShardsInProgress(TTxState::ProposedWaitParts); - pqGroup->FinishAlter(); + // Move all notifications that were already received + // NOTE: SchemeChangeNotification is sent form DS after it has got PlanStep from coordinator and the schema tx has completed + // At that moment the SS might not have received PlanStep from coordinator yet (this message might be still on its way to SS) + // So we are going to accumulate SchemeChangeNotification that are received before this Tx switches to WaitParts state + txState->AcceptPendingSchemeNotification(); - context.SS->PersistPersQueueGroup(db, PathId, pqGroup); - context.SS->PersistRemovePersQueueGroupAlter(db, PathId); + // Got notifications from all datashards? + if (txState->ShardsInProgress.empty()) { + AckAllSchemaChanges(OperationId, *txState, context); + context.SS->ChangeTxState(db, OperationId, NextState); + return true; + } - context.SS->ChangeTxState(db, OperationId, TTxState::Done); + return false; } -bool TPropose::TryPersistState(TOperationContext& context) -{ - TTxState* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); +} // namespace NTableState - if (!CanPersistState(*txState, context)) { - return false; - } +namespace NPQState { + +bool CollectProposeTransactionResults(const TOperationId& operationId, + const TEvPersQueue::TEvProposeTransactionResult::TPtr& ev, + TOperationContext& context) +{ + auto prepared = [](NKikimrPQ::TEvProposeTransactionResult::EStatus status) -> bool { + return status == NKikimrPQ::TEvProposeTransactionResult::PREPARED; + }; - PersistState(*txState, context); + auto toString = [](NKikimrPQ::TEvProposeTransactionResult::EStatus status) -> TString { + return NKikimrPQ::TEvProposeTransactionResult_EStatus_Name(status); + }; - return true; + return CollectProposeTxResults(ev, operationId, context, prepared, toString); } -} +} // namespace NPQState TSet<ui32> AllIncomingEvents() { TSet<ui32> result; @@ -923,7 +1082,9 @@ TSet<ui32> AllIncomingEvents() { return result; } -void NForceDrop::CollectShards(const THashSet<TPathId>& paths, TOperationId operationId, TTxState *txState, TOperationContext &context) { +namespace NForceDrop { + +void CollectShards(const THashSet<TPathId>& paths, TOperationId operationId, TTxState *txState, TOperationContext &context) { NIceDb::TNiceDb db(context.GetDB()); auto shards = context.SS->CollectAllShards(paths); @@ -950,7 +1111,7 @@ void NForceDrop::CollectShards(const THashSet<TPathId>& paths, TOperationId oper context.SS->PersistTxState(db, operationId); } -void NForceDrop::ValidateNoTransactionOnPaths(TOperationId operationId, const THashSet<TPathId>& paths, TOperationContext &context) { +void ValidateNoTransactionOnPaths(TOperationId operationId, const THashSet<TPathId>& paths, TOperationContext &context) { // No transaction should materialize in a subdomain that is being deleted -- // -- all operations should be checking parent dir status at Propose stage. // However, it is better to verify that, just in case. @@ -963,6 +1124,8 @@ void NForceDrop::ValidateNoTransactionOnPaths(TOperationId operationId, const TH } } +} // namespace NForceDrop + void IncParentDirAlterVersionWithRepublishSafeWithUndo(const TOperationId& opId, const TPath& path, TSchemeShard* ss, TSideEffects& onComplete) { auto parent = path.Parent(); if (parent.Base()->IsDirectory() || parent.Base()->IsDomainRoot()) { diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_common.h b/ydb/core/tx/schemeshard/schemeshard__operation_common.h index 701a0f00726..7c8372a8782 100644 --- a/ydb/core/tx/schemeshard/schemeshard__operation_common.h +++ b/ydb/core/tx/schemeshard/schemeshard__operation_common.h @@ -8,8 +8,9 @@ #include <ydb/core/tx/columnshard/columnshard.h> #include <ydb/core/tx/tx_processing.h> -namespace NKikimr { -namespace NSchemeShard { +namespace NKikimr::NSchemeShard { + +class TSchemeShard; TSet<ui32> AllIncomingEvents(); @@ -52,86 +53,10 @@ private: } public: - TProposedWaitParts(TOperationId id, TTxState::ETxState nextState = TTxState::Done) - : OperationId(id) - , NextState(nextState) - { - IgnoreMessages(DebugHint(), - { TEvHive::TEvCreateTabletReply::EventType - , TEvDataShard::TEvProposeTransactionResult::EventType - , TEvPrivate::TEvOperationPlan::EventType } - ); - } - - bool HandleReply(TEvDataShard::TEvSchemaChanged::TPtr& ev, TOperationContext& context) override { - TTabletId ssId = context.SS->SelfTabletId(); - const auto& evRecord = ev->Get()->Record; - - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvSchemaChanged" - << " at tablet: " << ssId); - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvSchemaChanged" - << " at tablet: " << ssId - << " message: " << evRecord.ShortDebugString()); - - if (!NTableState::CollectSchemaChanged(OperationId, ev, context)) { - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvSchemaChanged" - << " CollectSchemaChanged: false"); - return false; - } - - Y_ABORT_UNLESS(context.SS->FindTx(OperationId)); - TTxState& txState = *context.SS->FindTx(OperationId); - - if (!txState.ReadyForNotifications) { - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvSchemaChanged" - << " ReadyForNotifications: false"); - return false; - } - - return true; - } - - bool ProgressState(TOperationContext& context) override { - TTabletId ssId = context.SS->SelfTabletId(); + TProposedWaitParts(TOperationId id, TTxState::ETxState nextState = TTxState::Done); - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " ProgressState" - << " at tablet: " << ssId); - - TTxState* txState = context.SS->FindTx(OperationId); - - NIceDb::TNiceDb db(context.GetDB()); - - txState->ClearShardsInProgress(); - for (TTxState::TShardOperation& shard : txState->Shards) { - if (shard.Operation < TTxState::ProposedWaitParts) { - shard.Operation = TTxState::ProposedWaitParts; - context.SS->PersistUpdateTxShard(db, OperationId, shard.Idx, shard.Operation); - } - Y_ABORT_UNLESS(context.SS->ShardInfos.contains(shard.Idx)); - context.OnComplete.RouteByTablet(OperationId, context.SS->ShardInfos.at(shard.Idx).TabletID); - } - txState->UpdateShardsInProgress(TTxState::ProposedWaitParts); - - // Move all notifications that were already received - // NOTE: SchemeChangeNotification is sent form DS after it has got PlanStep from coordinator and the schema tx has completed - // At that moment the SS might not have received PlanStep from coordinator yet (this message might be still on its way to SS) - // So we are going to accumulate SchemeChangeNotification that are received before this Tx switches to WaitParts state - txState->AcceptPendingSchemeNotification(); - - // Got notifications from all datashards? - if (txState->ShardsInProgress.empty()) { - NTableState::AckAllSchemaChanges(OperationId, *txState, context); - context.SS->ChangeTxState(db, OperationId, NextState); - return true; - } - - return false; - } + bool ProgressState(TOperationContext& context) override; + bool HandleReply(TEvDataShard::TEvSchemaChanged::TPtr& ev, TOperationContext& context) override; }; } // namespace NTableState @@ -144,285 +69,14 @@ class TCreateParts: public TSubOperationState { << " opId# " << OperationId; } -public: - explicit TCreateParts(const TOperationId& id) - : OperationId(id) - { - IgnoreMessages(DebugHint(), {}); - } - - bool HandleReply(TEvHive::TEvAdoptTabletReply::TPtr& ev, TOperationContext& context) override { - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvAdoptTablet" - << ", at tabletId: " << context.SS->TabletID()); - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvAdoptTablet" - << ", message% " << DebugReply(ev)); - - NIceDb::TNiceDb db(context.GetDB()); - - const TString& explain = ev->Get()->Record.GetExplain(); - TTabletId tabletId = TTabletId(ev->Get()->Record.GetTabletID()); // global id from hive - auto shardIdx = context.SS->MakeLocalId(TLocalShardIdx(ev->Get()->Record.GetOwnerIdx())); // global id from hive - TTabletId hive = TTabletId(ev->Get()->Record.GetOrigin()); - - Y_ABORT_UNLESS(context.SS->TabletID() == ev->Get()->Record.GetOwner()); - - NKikimrProto::EReplyStatus status = ev->Get()->Record.GetStatus(); - Y_VERIFY_S(status == NKikimrProto::OK || status == NKikimrProto::ALREADY, - "Unexpected status " << NKikimrProto::EReplyStatus_Name(status) - << " in AdoptTabletReply for tabletId " << tabletId - << " with explain " << explain); - - // Note that HIVE might send duplicate TTxAdoptTabletReply in case of restarts - // So we just ignore the event if we cannot find the Tx or if it is in a different - // state - - Y_ABORT_UNLESS(context.SS->ShardInfos.contains(shardIdx)); - - if (!context.SS->AdoptedShards.contains(shardIdx)) { - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvAdoptTablet" - << " Got TTxAdoptTabletReply for shard but it is not present in AdoptedShards" - << ", shardIdx: " << shardIdx - << ", tabletId:" << tabletId); - return false; - } - - TShardInfo& shardInfo = context.SS->ShardInfos[shardIdx]; - Y_ABORT_UNLESS(shardInfo.TabletID == InvalidTabletId || shardInfo.TabletID == tabletId); - - Y_ABORT_UNLESS(tabletId != InvalidTabletId); - shardInfo.TabletID = tabletId; - context.SS->TabletIdToShardIdx[tabletId] = shardIdx; - - TTxState* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - Y_ABORT_UNLESS(txState->State == TTxState::CreateParts); - - txState->ShardsInProgress.erase(shardIdx); - - context.SS->AdoptedShards.erase(shardIdx); - context.SS->PersistDeleteAdopted(db, shardIdx); - context.SS->PersistShardMapping(db, shardIdx, tabletId, shardInfo.PathId, OperationId.GetTxId(), shardInfo.TabletType); - - context.OnComplete.UnbindMsgFromPipe(OperationId, hive, shardIdx); - context.OnComplete.ActivateShardCreated(shardIdx, OperationId.GetTxId()); - - // If all datashards have been created - if (txState->ShardsInProgress.empty()) { - context.SS->ChangeTxState(db, OperationId, TTxState::ConfigureParts); - return true; - } - - return false; - } - - bool HandleReply(TEvHive::TEvCreateTabletReply::TPtr& ev, TOperationContext& context) override { - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvCreateTabletReply" - << ", at tabletId: " << context.SS->TabletID()); - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvCreateTabletReply" - << ", message: " << DebugReply(ev)); - - NIceDb::TNiceDb db(context.GetDB()); - - auto shardIdx = TShardIdx(ev->Get()->Record.GetOwner(), - TLocalShardIdx(ev->Get()->Record.GetOwnerIdx())); - - auto tabletId = TTabletId(ev->Get()->Record.GetTabletID()); // global id from hive - auto hive = TTabletId(ev->Get()->Record.GetOrigin()); - - NKikimrProto::EReplyStatus status = ev->Get()->Record.GetStatus(); - - Y_VERIFY_S(status == NKikimrProto::OK - || status == NKikimrProto::ALREADY - || status == NKikimrProto::INVALID_OWNER - || status == NKikimrProto::BLOCKED, - "Unexpected status " << NKikimrProto::EReplyStatus_Name(status) - << " in CreateTabletReply shard idx " << shardIdx << " tabletId " << tabletId); - - if (status == NKikimrProto::BLOCKED) { - Y_ABORT_UNLESS(!context.SS->IsDomainSchemeShard); - - LOG_NOTICE_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " CreateRequest BLOCKED " - << " at Hive: " << hive - << " msg: " << DebugReply(ev)); - - // do not unsubscribe message - // context.OnComplete.UnbindMsgFromPipe(OperationId, hive, shardIdx); - - // just stay calm and hung until tenant schemeshard is deleted - return false; - } - - TTxState& txState = *context.SS->FindTx(OperationId); - - TShardInfo& shardInfo = context.SS->ShardInfos.at(shardIdx); - Y_ABORT_UNLESS(shardInfo.TabletID == InvalidTabletId || shardInfo.TabletID == tabletId); - - if (status == NKikimrProto::INVALID_OWNER) { - auto redirectTo = TTabletId(ev->Get()->Record.GetForwardRequest().GetHiveTabletId()); - Y_ABORT_UNLESS(redirectTo); - Y_ABORT_UNLESS(tabletId); - - context.OnComplete.UnbindMsgFromPipe(OperationId, hive, shardIdx); - - auto path = context.SS->PathsById.at(txState.TargetPathId); - auto request = CreateEvCreateTablet(path, shardIdx, context); - - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " CreateRequest" - << " Redirect from Hive: " << hive - << " to Hive: " << redirectTo - << " msg: " << request->Record.DebugString()); - - context.OnComplete.BindMsgToPipe(OperationId, redirectTo, shardIdx, request.Release()); - return false; - } - - if (shardInfo.TabletID == InvalidTabletId) { - switch (shardInfo.TabletType) { - case ETabletType::DataShard: - context.SS->TabletCounters->Simple()[COUNTER_TABLE_SHARD_ACTIVE_COUNT].Add(1); - break; - case ETabletType::Coordinator: - context.SS->TabletCounters->Simple()[COUNTER_SUB_DOMAIN_COORDINATOR_COUNT].Add(1); - break; - case ETabletType::Mediator: - context.SS->TabletCounters->Simple()[COUNTER_SUB_DOMAIN_MEDIATOR_COUNT].Add(1); - break; - case ETabletType::Hive: - context.SS->TabletCounters->Simple()[COUNTER_SUB_DOMAIN_HIVE_COUNT].Add(1); - break; - case ETabletType::SysViewProcessor: - context.SS->TabletCounters->Simple()[COUNTER_SYS_VIEW_PROCESSOR_COUNT].Add(1); - break; - case ETabletType::StatisticsAggregator: - context.SS->TabletCounters->Simple()[COUNTER_STATISTICS_AGGREGATOR_COUNT].Add(1); - break; - case ETabletType::BackupController: - context.SS->TabletCounters->Simple()[COUNTER_BACKUP_CONTROLLER_TABLET_COUNT].Add(1); - break; - default: - break; - } - } - - Y_ABORT_UNLESS(tabletId != InvalidTabletId); - shardInfo.TabletID = tabletId; - context.SS->TabletIdToShardIdx[tabletId] = shardIdx; - - Y_ABORT_UNLESS(OperationId.GetTxId() == shardInfo.CurrentTxId); - - txState.ShardsInProgress.erase(shardIdx); - - context.SS->PersistShardMapping(db, shardIdx, tabletId, shardInfo.PathId, OperationId.GetTxId(), shardInfo.TabletType); - context.OnComplete.UnbindMsgFromPipe(OperationId, hive, shardIdx); - context.OnComplete.ActivateShardCreated(shardIdx, OperationId.GetTxId()); - - // If all datashards have been created - if (txState.ShardsInProgress.empty()) { - context.SS->ChangeTxState(db, OperationId, TTxState::ConfigureParts); - return true; - } - - return false; - } - - THolder<TEvHive::TEvAdoptTablet> AdoptRequest(TShardIdx shardIdx, TOperationContext& context) { - Y_ABORT_UNLESS(context.SS->AdoptedShards.contains(shardIdx)); - auto& adoptedShard = context.SS->AdoptedShards[shardIdx]; - auto& shard = context.SS->ShardInfos[shardIdx]; - - THolder<TEvHive::TEvAdoptTablet> ev = MakeHolder<TEvHive::TEvAdoptTablet>( - ui64(shard.TabletID), - adoptedShard.PrevOwner, ui64(adoptedShard.PrevShardIdx), - shard.TabletType, - context.SS->TabletID(), ui64(shardIdx.GetLocalId())); - - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " AdoptRequest" - << " Event to Hive: " << ev->Record.DebugString().c_str()); - - return ev; - } - - bool ProgressState(TOperationContext& context) override { - TTabletId ssId = context.SS->SelfTabletId(); - - TTxState* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " ProgressState" - << ", operation type: " << TTxState::TypeName(txState->TxType) - << ", at tablet" << ssId); - - if (txState->TxType == TTxState::TxDropTable - || txState->TxType == TTxState::TxAlterTable - || txState->TxType == TTxState::TxBackup - || txState->TxType == TTxState::TxRestore) { - if (NTableState::CheckPartitioningChangedForTableModification(*txState, context)) { - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " ProgressState" - << " SourceTablePartitioningChangedForModification" - << ", tx type: " << TTxState::TypeName(txState->TxType)); - NTableState::UpdatePartitioningForTableModification(OperationId, *txState, context); - } - } else if (txState->TxType == TTxState::TxCopyTable) { - if (NTableState::SourceTablePartitioningChangedForCopyTable(*txState, context)) { - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " ProgressState" - << " SourceTablePartitioningChangedForCopyTable" - << ", tx type: " << TTxState::TypeName(txState->TxType)); - NTableState::UpdatePartitioningForCopyTable(OperationId, *txState, context); - } - } - - txState->ClearShardsInProgress(); - - bool nothingToDo = true; - for (const auto& shard : txState->Shards) { - if (shard.Operation != TTxState::CreateParts) { - continue; - } - nothingToDo = false; - - if (context.SS->AdoptedShards.contains(shard.Idx)) { - auto ev = AdoptRequest(shard.Idx, context); - context.OnComplete.BindMsgToPipe(OperationId, context.SS->GetGlobalHive(context.Ctx), shard.Idx, ev.Release()); - } else { - auto path = context.SS->PathsById.at(txState->TargetPathId); - auto ev = CreateEvCreateTablet(path, shard.Idx, context); + THolder<TEvHive::TEvAdoptTablet> AdoptRequest(TShardIdx shardIdx, TOperationContext& context); - auto hiveToRequest = context.SS->ResolveHive(shard.Idx, context.Ctx); - - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " CreateRequest" - << " Event to Hive: " << hiveToRequest - << " msg: "<< ev->Record.DebugString().c_str()); - - context.OnComplete.BindMsgToPipe(OperationId, hiveToRequest, shard.Idx, ev.Release()); - } - context.OnComplete.RouteByShardIdx(OperationId, shard.Idx); - } - - if (nothingToDo) { - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " ProgressState" - << " no shards to create, do next state"); - - NIceDb::TNiceDb db(context.GetDB()); - context.SS->ChangeTxState(db, OperationId, TTxState::ConfigureParts); - return true; - } +public: + explicit TCreateParts(const TOperationId& id); - txState->UpdateShardsInProgress(TTxState::CreateParts); - return false; - } + bool ProgressState(TOperationContext& context) override; + bool HandleReply(TEvHive::TEvCreateTabletReply::TPtr& ev, TOperationContext& context) override; + bool HandleReply(TEvHive::TEvAdoptTabletReply::TPtr& ev, TOperationContext& context) override; }; class TDeleteParts: public TSubOperationState { @@ -435,50 +89,19 @@ protected: << " opId# " << OperationId << " "; } - void DeleteShards(TOperationContext& context) { - const auto* txState = context.SS->FindTx(OperationId); - - // Initiate asynchronous deletion of all shards - for (const auto& shard : txState->Shards) { - context.OnComplete.DeleteShard(shard.Idx); - } - } + void DeleteShards(TOperationContext& context); public: - explicit TDeleteParts(const TOperationId& id, TTxState::ETxState nextState = TTxState::Propose) - : OperationId(id) - , NextState(nextState) - { - IgnoreMessages(DebugHint(), {}); - } + explicit TDeleteParts(const TOperationId& id, TTxState::ETxState nextState = TTxState::Propose); - bool ProgressState(TOperationContext& context) override { - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "[" << context.SS->TabletID() << "] " << DebugHint() << " ProgressState"); - DeleteShards(context); - - NIceDb::TNiceDb db(context.GetDB()); - context.SS->ChangeTxState(db, OperationId, NextState); - return true; - } + bool ProgressState(TOperationContext& context) override; }; class TDeletePartsAndDone: public TDeleteParts { public: - explicit TDeletePartsAndDone(const TOperationId& id) - : TDeleteParts(id) - { - Y_UNUSED(NextState); - } + explicit TDeletePartsAndDone(const TOperationId& id); - bool ProgressState(TOperationContext& context) override { - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "[" << context.SS->TabletID() << "] " << DebugHint() << " ProgressState"); - DeleteShards(context); - - context.OnComplete.DoneOperation(OperationId); - return true; - } + bool ProgressState(TOperationContext& context) override; }; class TDone: public TSubOperationState { @@ -491,114 +114,14 @@ protected: } public: - explicit TDone(const TOperationId& id) - : OperationId(id) - { - IgnoreMessages(DebugHint(), AllIncomingEvents()); - } - - bool ProgressState(TOperationContext& context) override { - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "[" << context.SS->TabletID() << "] " << DebugHint() << " ProgressState"); - - const auto* txState = context.SS->FindTx(OperationId); - - const auto& pathId = txState->TargetPathId; - Y_ABORT_UNLESS(context.SS->PathsById.contains(pathId)); - TPathElement::TPtr path = context.SS->PathsById.at(pathId); - Y_VERIFY_S(path->PathState != TPathElement::EPathState::EPathStateNoChanges, "with context" - << ", PathState: " << NKikimrSchemeOp::EPathState_Name(path->PathState) - << ", PathId: " << path->PathId - << ", PathName: " << path->Name); + explicit TDone(const TOperationId& id); - if (path->IsPQGroup() && txState->IsCreate()) { - TPathElement::TPtr parentDir = context.SS->PathsById.at(path->ParentPathId); - // can't be removed until KIKIMR-8861 - // at least lets wrap it into condition - // it helps to actualize PATHSTATE inside children listing - context.SS->ClearDescribePathCaches(parentDir); - } - - if (txState->IsDrop()) { - context.OnComplete.ReleasePathState(OperationId, path->PathId, TPathElement::EPathState::EPathStateNotExist); - } else { - context.OnComplete.ReleasePathState(OperationId, path->PathId, TPathElement::EPathState::EPathStateNoChanges); - } - - if (txState->SourcePathId != InvalidPathId) { - Y_ABORT_UNLESS(context.SS->PathsById.contains(txState->SourcePathId)); - TPathElement::TPtr srcPath = context.SS->PathsById.at(txState->SourcePathId); - if (srcPath->PathState == TPathElement::EPathState::EPathStateCopying) { - context.OnComplete.ReleasePathState(OperationId, srcPath->PathId, TPathElement::EPathState::EPathStateNoChanges); - } - } - - // OlapStore tracks all tables that are under operation, make sure to unlink - if (context.SS->ColumnTables.contains(pathId)) { - auto tableInfo = context.SS->ColumnTables.at(pathId); - if (!tableInfo->IsStandalone()) { - const auto storePathId = tableInfo->GetOlapStorePathIdVerified(); - if (context.SS->OlapStores.contains(storePathId)) { - auto storeInfo = context.SS->OlapStores.at(storePathId); - storeInfo->ColumnTablesUnderOperation.erase(pathId); - } - } - } - - context.OnComplete.DoneOperation(OperationId); - return true; - } + bool ProgressState(TOperationContext& context) override; }; namespace NPQState { -class TBootstrapConfigWrapper: public NKikimrPQ::TBootstrapConfig { - struct TSerializedProposeTransaction { - TString Value; - - static TSerializedProposeTransaction Serialize(const NKikimrPQ::TBootstrapConfig& value) { - NKikimrPQ::TEvProposeTransaction record; - record.MutableConfig()->MutableBootstrapConfig()->CopyFrom(value); - return {record.SerializeAsString()}; - } - }; - - struct TSerializedUpdateConfig { - TString Value; - - static TSerializedUpdateConfig Serialize(const NKikimrPQ::TBootstrapConfig& value) { - NKikimrPQ::TUpdateConfig record; - record.MutableBootstrapConfig()->CopyFrom(value); - return {record.SerializeAsString()}; - } - }; - - mutable std::optional<std::variant< - TSerializedProposeTransaction, - TSerializedUpdateConfig - >> PreSerialized; - - template <typename T> - const TString& Get() const { - if (!PreSerialized) { - PreSerialized.emplace(T::Serialize(*this)); - } - - const auto* value = std::get_if<T>(&PreSerialized.value()); - Y_ABORT_UNLESS(value); - - return value->Value; - } - -public: - const TString& GetPreSerializedProposeTransaction() const { - return Get<TSerializedProposeTransaction>(); - } - - const TString& GetPreSerializedUpdateConfig() const { - return Get<TSerializedUpdateConfig>(); - } -}; +bool CollectProposeTransactionResults(const TOperationId& operationId, const TEvPersQueue::TEvProposeTransactionResult::TPtr& ev, TOperationContext& context); class TConfigureParts: public TSubOperationState { private: @@ -611,397 +134,11 @@ private: } public: - TConfigureParts(TOperationId id) - : OperationId(id) - { - IgnoreMessages(DebugHint(), {TEvHive::TEvCreateTabletReply::EventType}); - } + TConfigureParts(TOperationId id); + bool ProgressState(TOperationContext& context) override; bool HandleReply(TEvPersQueue::TEvProposeTransactionResult::TPtr& ev, TOperationContext& context) override; - - bool HandleReply(TEvPersQueue::TEvUpdateConfigResponse::TPtr& ev, TOperationContext& context) override { - TTabletId ssId = context.SS->SelfTabletId(); - - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvUpdateConfigResponse" - << " at tablet" << ssId); - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvUpdateConfigResponse" - << " message: " << ev->Get()->Record.ShortUtf8DebugString() - << " at tablet" << ssId); - - TTxState* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreatePQGroup || txState->TxType == TTxState::TxAlterPQGroup); - - TTabletId tabletId = TTabletId(ev->Get()->Record.GetOrigin()); - NKikimrPQ::EStatus status = ev->Get()->Record.GetStatus(); - - // Schemeshard always sends a valid config to the PQ tablet and PQ tablet is not supposed to reject it - // Also Schemeshard never send multiple different config updates simultaneously (same config - // can be sent more than once in case of retries due to restarts or disconnects) - // If PQ tablet is not able to save a valid config it should kill itself and restart without - // sending error response - Y_VERIFY_S(status == NKikimrPQ::OK || status == NKikimrPQ::ERROR_UPDATE_IN_PROGRESS, - "Unexpected error in UpdateConfigResponse," - << " status: " << NKikimrPQ::EStatus_Name(status) - << " Tx " << OperationId - << " tablet "<< tabletId - << " at schemeshard: " << ssId); - - if (status == NKikimrPQ::ERROR_UPDATE_IN_PROGRESS) { - LOG_ERROR_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "PQ reconfiguration is in progress. We'll try to finish it later." - << " Tx " << OperationId - << " tablet " << tabletId - << " at schemeshard: " << ssId); - return false; - } - - Y_ABORT_UNLESS(txState->State == TTxState::ConfigureParts); - - TShardIdx idx = context.SS->MustGetShardIdx(tabletId); - txState->ShardsInProgress.erase(idx); - - // Detach datashard pipe - context.OnComplete.UnbindMsgFromPipe(OperationId, tabletId, idx); - - if (txState->ShardsInProgress.empty()) { - NIceDb::TNiceDb db(context.GetDB()); - context.SS->ChangeTxState(db, OperationId, TTxState::Propose); - return true; - } - - return false; - } - - bool ProgressState(TOperationContext& context) override { - TTabletId ssId = context.SS->SelfTabletId(); - - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() - << " HandleReply ProgressState" - << ", at schemeshard: " << ssId); - - TTxState* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreatePQGroup || txState->TxType == TTxState::TxAlterPQGroup); - - txState->ClearShardsInProgress(); - - TString topicName = context.SS->PathsById.at(txState->TargetPathId)->Name; - Y_VERIFY_S(topicName.size(), - "topicName is empty" - <<", pathId: " << txState->TargetPathId); - - TTopicInfo::TPtr pqGroup = context.SS->Topics[txState->TargetPathId]; - Y_VERIFY_S(pqGroup, - "pqGroup is null" - << ", pathId " << txState->TargetPathId); - - const TPathElement::TPtr dbRootEl = context.SS->PathsById.at(context.SS->RootPathId()); - TString cloudId; - if (dbRootEl->UserAttrs->Attrs.contains("cloud_id")) { - cloudId = dbRootEl->UserAttrs->Attrs.at("cloud_id"); - } - TString folderId; - if (dbRootEl->UserAttrs->Attrs.contains("folder_id")) { - folderId = dbRootEl->UserAttrs->Attrs.at("folder_id"); - } - TString databaseId; - if (dbRootEl->UserAttrs->Attrs.contains("database_id")) { - databaseId = dbRootEl->UserAttrs->Attrs.at("database_id"); - } - - TString databasePath = TPath::Init(context.SS->RootPathId(), context.SS).PathString(); - auto topicPath = TPath::Init(txState->TargetPathId, context.SS); - - std::optional<TBootstrapConfigWrapper> bootstrapConfig; - if (txState->TxType == TTxState::TxCreatePQGroup && topicPath.Parent().IsCdcStream()) { - bootstrapConfig.emplace(); - - auto tablePath = topicPath.Parent().Parent(); // table/cdc_stream/topic - Y_ABORT_UNLESS(tablePath.IsResolved()); - - Y_ABORT_UNLESS(context.SS->Tables.contains(tablePath.Base()->PathId)); - auto table = context.SS->Tables.at(tablePath.Base()->PathId); - - const auto& partitions = table->GetPartitions(); - - for (ui32 i = 0; i < partitions.size(); ++i) { - const auto& cur = partitions.at(i); - - Y_ABORT_UNLESS(context.SS->ShardInfos.contains(cur.ShardIdx)); - const auto& shard = context.SS->ShardInfos.at(cur.ShardIdx); - - auto& mg = *bootstrapConfig->AddExplicitMessageGroups(); - mg.SetId(NPQ::NSourceIdEncoding::EncodeSimple(ToString(shard.TabletID))); - - if (i != partitions.size() - 1) { - mg.MutableKeyRange()->SetToBound(cur.EndOfRange); - } - - if (i) { - const auto& prev = partitions.at(i - 1); - mg.MutableKeyRange()->SetFromBound(prev.EndOfRange); - } - } - } - - for (auto shard : txState->Shards) { - TShardIdx idx = shard.Idx; - TTabletId tabletId = context.SS->ShardInfos.at(idx).TabletID; - - if (shard.TabletType == ETabletType::PersQueue) { - TTopicTabletInfo::TPtr pqShard = pqGroup->Shards.at(idx); - Y_VERIFY_S(pqShard, "pqShard is null, idx is " << idx << " has was "<< THash<TShardIdx>()(idx)); - - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "Propose configure PersQueue" - << ", opId: " << OperationId - << ", tabletId: " << tabletId - << ", Partitions size: " << pqShard->Partitions.size() - << ", at schemeshard: " << ssId); - - THolder<NActors::IEventBase> event; - if (context.SS->EnablePQConfigTransactionsAtSchemeShard) { - event = MakeEvProposeTransaction(OperationId.GetTxId(), - *pqGroup, - *pqShard, - topicName, - topicPath.PathString(), - bootstrapConfig, - cloudId, - folderId, - databaseId, - databasePath, - txState->TxType, - context); - } else { - event = MakeEvUpdateConfig(OperationId.GetTxId(), - *pqGroup, - *pqShard, - topicName, - topicPath.PathString(), - bootstrapConfig, - cloudId, - folderId, - databaseId, - databasePath, - txState->TxType, - context); - } - - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "Propose configure PersQueue" - << ", opId: " << OperationId - << ", tabletId: " << tabletId - << ", at schemeshard: " << ssId); - - context.OnComplete.BindMsgToPipe(OperationId, tabletId, idx, event.Release()); - } else { - Y_ABORT_UNLESS(shard.TabletType == ETabletType::PersQueueReadBalancer); - - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "Propose configure PersQueueReadBalancer" - << ", opId: " << OperationId - << ", tabletId: " << tabletId - << ", at schemeshard: " << ssId); - - pqGroup->BalancerTabletID = tabletId; - if (pqGroup->AlterData) { - pqGroup->AlterData->BalancerTabletID = tabletId; - } - - if (pqGroup->AlterData) { - pqGroup->AlterData->BalancerShardIdx = idx; - } - - TAutoPtr<TEvPersQueue::TEvUpdateBalancerConfig> event(new TEvPersQueue::TEvUpdateBalancerConfig()); - event->Record.SetTxId(ui64(OperationId.GetTxId())); - - ParsePQTabletConfig(*event->Record.MutableTabletConfig(), *pqGroup); - - Y_ABORT_UNLESS(pqGroup->AlterData); - - event->Record.SetTopicName(topicName); - event->Record.SetPathId(txState->TargetPathId.LocalPathId); - event->Record.SetPath(TPath::Init(txState->TargetPathId, context.SS).PathString()); - event->Record.SetPartitionPerTablet(pqGroup->AlterData ? pqGroup->AlterData->MaxPartsPerTablet : pqGroup->MaxPartsPerTablet); - event->Record.SetSchemeShardId(context.SS->TabletID()); - - event->Record.SetTotalGroupCount(pqGroup->AlterData ? pqGroup->AlterData->TotalGroupCount : pqGroup->TotalGroupCount); - event->Record.SetNextPartitionId(pqGroup->AlterData ? pqGroup->AlterData->NextPartitionId : pqGroup->NextPartitionId); - - event->Record.SetVersion(pqGroup->AlterData->AlterVersion); - - for (const auto& p : pqGroup->Shards) { - const auto& pqShard = p.second; - const auto& tabletId = context.SS->ShardInfos[p.first].TabletID; - auto tablet = event->Record.AddTablets(); - tablet->SetTabletId(ui64(tabletId)); - tablet->SetOwner(context.SS->TabletID()); - tablet->SetIdx(ui64(p.first.GetLocalId())); - for (const auto& pq : pqShard->Partitions) { - auto info = event->Record.AddPartitions(); - info->SetPartition(pq->PqId); - info->SetTabletId(ui64(tabletId)); - info->SetGroup(pq->GroupId); - info->SetCreateVersion(pq->CreateVersion); - - if (pq->KeyRange) { - pq->KeyRange->SerializeToProto(*info->MutableKeyRange()); - } - info->SetStatus(pq->Status); - info->MutableParentPartitionIds()->Reserve(pq->ParentPartitionIds.size()); - for (const auto parent : pq->ParentPartitionIds) { - info->MutableParentPartitionIds()->AddAlreadyReserved(parent); - } - info->MutableChildPartitionIds()->Reserve(pq->ChildPartitionIds.size()); - for (const auto children : pq->ChildPartitionIds) { - info->MutableChildPartitionIds()->AddAlreadyReserved(children); - } - } - } - - if (const ui64 subDomainPathId = context.SS->ResolvePathIdForDomain(txState->TargetPathId).LocalPathId) { - event->Record.SetSubDomainPathId(subDomainPathId); - } - - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "Propose configure PersQueueReadBalancer" - << ", opId: " << OperationId - << ", tabletId: " << tabletId - << ", message: " << event->Record.ShortUtf8DebugString() - << ", at schemeshard: " << ssId); - - context.OnComplete.BindMsgToPipe(OperationId, tabletId, idx, event.Release()); - } - } - - txState->UpdateShardsInProgress(); - return false; - } - -private: - static void FillPartition(NKikimrPQ::TPQTabletConfig::TPartition& partition, const TTopicTabletInfo::TTopicPartitionInfo* pq, ui64 tabletId) { - partition.SetPartitionId(pq->PqId); - partition.SetCreateVersion(pq->CreateVersion); - if (pq->KeyRange) { - pq->KeyRange->SerializeToProto(*partition.MutableKeyRange()); - } - partition.SetStatus(pq->Status); - partition.MutableParentPartitionIds()->Reserve(pq->ParentPartitionIds.size()); - for (const auto parent : pq->ParentPartitionIds) { - partition.MutableParentPartitionIds()->AddAlreadyReserved(parent); - } - partition.MutableChildPartitionIds()->Reserve(pq->ChildPartitionIds.size()); - for (const auto children : pq->ChildPartitionIds) { - partition.MutableChildPartitionIds()->AddAlreadyReserved(children); - } - partition.SetTabletId(tabletId); - } - - static void MakePQTabletConfig(const TOperationContext& context, - NKikimrPQ::TPQTabletConfig& config, - const TTopicInfo& pqGroup, - const TTopicTabletInfo& pqShard, - const TString& topicName, - const TString& topicPath, - const TString& cloudId, - const TString& folderId, - const TString& databaseId, - const TString& databasePath) - { - ParsePQTabletConfig(config, pqGroup); - - config.SetTopicName(topicName); - config.SetTopicPath(topicPath); - config.MutablePartitionConfig()->SetTotalPartitions(pqGroup.AlterData ? pqGroup.AlterData->TotalGroupCount : pqGroup.TotalGroupCount); - - config.SetYcCloudId(cloudId); - config.SetYcFolderId(folderId); - config.SetYdbDatabaseId(databaseId); - config.SetYdbDatabasePath(databasePath); - - if (pqGroup.AlterData) { - config.SetVersion(pqGroup.AlterData->AlterVersion); - } - - THashSet<ui32> linkedPartitions; - - for(const auto& pq : pqShard.Partitions) { - config.AddPartitionIds(pq->PqId); - - auto& partition = *config.AddPartitions(); - FillPartition(partition, pq.Get(), 0); - - linkedPartitions.insert(pq->PqId); - linkedPartitions.insert(pq->ParentPartitionIds.begin(), pq->ParentPartitionIds.end()); - linkedPartitions.insert(pq->ChildPartitionIds.begin(), pq->ChildPartitionIds.end()); - for (auto c : pq->ChildPartitionIds) { - auto it = pqGroup.Partitions.find(c); - if (it == pqGroup.Partitions.end()) { - continue; - } - linkedPartitions.insert(it->second->ParentPartitionIds.begin(), it->second->ParentPartitionIds.end()); - } - } - - for(auto lp : linkedPartitions) { - auto it = pqGroup.Partitions.find(lp); - if (it == pqGroup.Partitions.end()) { - continue; - } - - auto* partitionInfo = it->second; - const auto& tabletId = context.SS->ShardInfos[partitionInfo->ShardIdx].TabletID; - - auto& partition = *config.AddAllPartitions(); - FillPartition(partition, partitionInfo, ui64(tabletId)); - } - } - - static void ParsePQTabletConfig(NKikimrPQ::TPQTabletConfig& config, - const TTopicInfo& pqGroup) - { - const TString* source = &pqGroup.TabletConfig; - if (pqGroup.AlterData) { - if (!pqGroup.AlterData->TabletConfig.empty()) - source = &pqGroup.AlterData->TabletConfig; - } - - if (!source->empty()) { - Y_ABORT_UNLESS(ParseFromStringNoSizeLimit(config, *source)); - } - } - - static THolder<TEvPersQueue::TEvProposeTransaction> - MakeEvProposeTransaction(TTxId txId, - const TTopicInfo& pqGroup, - const TTopicTabletInfo& pqShard, - const TString& topicName, - const TString& topicPath, - const std::optional<TBootstrapConfigWrapper>& bootstrapConfig, - const TString& cloudId, - const TString& folderId, - const TString& databaseId, - const TString& databasePath, - TTxState::ETxType txType, - const TOperationContext& context); - static THolder<TEvPersQueue::TEvUpdateConfig> - MakeEvUpdateConfig(TTxId txId, - const TTopicInfo& pqGroup, - const TTopicTabletInfo& pqShard, - const TString& topicName, - const TString& topicPath, - const std::optional<TBootstrapConfigWrapper>& bootstrapConfig, - const TString& cloudId, - const TString& folderId, - const TString& databaseId, - const TString& databasePath, - TTxState::ETxType txType, - const TOperationContext& context); + bool HandleReply(TEvPersQueue::TEvUpdateConfigResponse::TPtr& ev, TOperationContext& context) override; }; class TPropose: public TSubOperationState { @@ -1015,71 +152,12 @@ private: } public: - TPropose(TOperationId id) - : OperationId(id) - { - IgnoreMessages(DebugHint(), {TEvHive::TEvCreateTabletReply::EventType, TEvPersQueue::TEvUpdateConfigResponse::EventType}); - } + TPropose(TOperationId id); + bool ProgressState(TOperationContext& context) override; bool HandleReply(TEvPersQueue::TEvProposeTransactionResult::TPtr& ev, TOperationContext& context) override; bool HandleReply(TEvPersQueue::TEvProposeTransactionAttachResult::TPtr& ev, TOperationContext& context) override; - - bool HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOperationContext& context) override { - TStepId step = TStepId(ev->Get()->StepId); - TTabletId ssId = context.SS->SelfTabletId(); - - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() - << " HandleReply TEvOperationPlan" - << ", step: " << step - << ", at tablet: " << ssId); - - TTxState* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreatePQGroup || txState->TxType == TTxState::TxAlterPQGroup); - - TPathId pathId = txState->TargetPathId; - TPathElement::TPtr path = context.SS->PathsById.at(pathId); - - NIceDb::TNiceDb db(context.GetDB()); - - if (path->StepCreated == InvalidStepId) { - path->StepCreated = step; - context.SS->PersistCreateStep(db, pathId, step); - } - - return TryPersistState(context); - } - - bool ProgressState(TOperationContext& context) override { - TTabletId ssId = context.SS->SelfTabletId(); - - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "NPQState::TPropose ProgressState" - << ", operationId: " << OperationId - << ", at schemeshard: " << ssId); - - TTxState* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreatePQGroup || txState->TxType == TTxState::TxAlterPQGroup); - - // - // If the program works according to the new scheme, then we must add PQ tablets to the list for - // the Coordinator. At this stage, we cannot rely on the value of - // the EnablePQConfigTransactionsAtSchemeShard flag. Because the operation could have started on one tablet - // and moved to another by that time. - // - // Therefore, here we check the value of the minStep field, which is filled in in - // the TEvProposeTransactionResult handler - // - TSet<TTabletId> shardSet; - if (ui64(txState->MinStep) > 0) { - PrepareShards(*txState, shardSet, context); - } - context.OnComplete.ProposeToCoordinator(OperationId, txState->TargetPathId, txState->MinStep, shardSet); - - return false; - } + bool HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOperationContext& context) override; private: bool CanPersistState(const TTxState& txState, @@ -1111,122 +189,12 @@ private: } public: - TConfigureParts(TOperationId id) - : OperationId(id) - { - IgnoreMessages(DebugHint(), {TEvHive::TEvCreateTabletReply::EventType}); - } - - bool HandleReply(TEvBlockStore::TEvUpdateVolumeConfigResponse::TPtr& ev, TOperationContext& context) override { - TTabletId ssId = context.SS->SelfTabletId(); - - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvSetConfigResult" - << ", at schemeshard: " << ssId); - - TTxState* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateBlockStoreVolume || txState->TxType == TTxState::TxAlterBlockStoreVolume); - Y_ABORT_UNLESS(txState->State == TTxState::ConfigureParts); - - TTabletId tabletId = TTabletId(ev->Get()->Record.GetOrigin()); - NKikimrBlockStore::EStatus status = ev->Get()->Record.GetStatus(); - - // Schemeshard never sends invalid or outdated configs - Y_VERIFY_S(status == NKikimrBlockStore::OK || status == NKikimrBlockStore::ERROR_UPDATE_IN_PROGRESS, - "Unexpected error in UpdateVolumeConfigResponse," - << " status " << NKikimrBlockStore::EStatus_Name(status) - << " Tx " << OperationId - << " tablet " << tabletId); - - if (status == NKikimrBlockStore::ERROR_UPDATE_IN_PROGRESS) { - LOG_ERROR_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "BlockStore reconfiguration is in progress. We'll try to finish it later." - << " Tx " << OperationId - << " tablet " << tabletId); - return false; - } - - TShardIdx idx = context.SS->MustGetShardIdx(tabletId); - txState->ShardsInProgress.erase(idx); - - context.OnComplete.UnbindMsgFromPipe(OperationId, tabletId, idx); - - if (txState->ShardsInProgress.empty()) { - NIceDb::TNiceDb db(context.GetDB()); - context.SS->ChangeTxState(db, OperationId, TTxState::Propose); - context.OnComplete.ActivateTx(OperationId); - return true; - } - - return false; - } - - bool ProgressState(TOperationContext& context) override { - TTabletId ssId = context.SS->SelfTabletId(); - - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " ProgressState" - << ", at schemeshard" << ssId); - - TTxState* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateBlockStoreVolume || txState->TxType == TTxState::TxAlterBlockStoreVolume); - Y_ABORT_UNLESS(!txState->Shards.empty()); - - txState->ClearShardsInProgress(); - - TBlockStoreVolumeInfo::TPtr volume = context.SS->BlockStoreVolumes[txState->TargetPathId]; - Y_VERIFY_S(volume, "volume is null. PathId: " << txState->TargetPathId); - - ui64 version = volume->AlterVersion; - const auto* volumeConfig = &volume->VolumeConfig; - if (volume->AlterData) { - version = volume->AlterData->AlterVersion; - volumeConfig = &volume->AlterData->VolumeConfig; - } - - for (auto shard : txState->Shards) { - if (shard.TabletType == ETabletType::BlockStorePartition || - shard.TabletType == ETabletType::BlockStorePartition2) { - continue; - } - - Y_ABORT_UNLESS(shard.TabletType == ETabletType::BlockStoreVolume); - TShardIdx shardIdx = shard.Idx; - TTabletId tabletId = context.SS->ShardInfos[shardIdx].TabletID; - - volume->VolumeTabletId = tabletId; - if (volume->AlterData) { - volume->AlterData->VolumeTabletId = tabletId; - volume->AlterData->VolumeShardIdx = shardIdx; - } + TConfigureParts(TOperationId id); - TAutoPtr<TEvBlockStore::TEvUpdateVolumeConfig> event(new TEvBlockStore::TEvUpdateVolumeConfig()); - event->Record.SetTxId(ui64(OperationId.GetTxId())); - - event->Record.MutableVolumeConfig()->CopyFrom(*volumeConfig); - event->Record.MutableVolumeConfig()->SetVersion(version); - - for (const auto& p : volume->Shards) { - const auto& part = p.second; - const auto& partTabletId = context.SS->ShardInfos[p.first].TabletID; - auto info = event->Record.AddPartitions(); - info->SetPartitionId(part->PartitionId); - info->SetTabletId(ui64(partTabletId)); - } - - context.OnComplete.BindMsgToPipe(OperationId, tabletId, shardIdx, event.Release()); - - // Wait for results from this shard - txState->ShardsInProgress.insert(shardIdx); - } - - return false; - } + bool ProgressState(TOperationContext& context) override; + bool HandleReply(TEvBlockStore::TEvUpdateVolumeConfigResponse::TPtr& ev, TOperationContext& context) override; }; - class TPropose: public TSubOperationState { private: TOperationId OperationId; @@ -1238,79 +206,10 @@ private: } public: - TPropose(TOperationId id) - : OperationId(id) - { - IgnoreMessages(DebugHint(), {TEvHive::TEvCreateTabletReply::EventType, TEvBlockStore::TEvUpdateVolumeConfigResponse::EventType}); - } - - bool HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOperationContext& context) override { - TStepId step = TStepId(ev->Get()->StepId); - TTabletId ssId = context.SS->SelfTabletId(); - - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvOperationPlan" - << ", at schemeshard: " << ssId); - - TTxState* txState = context.SS->FindTx(OperationId); - if (!txState) { - return false; - } - Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateBlockStoreVolume || txState->TxType == TTxState::TxAlterBlockStoreVolume); - - TPathId pathId = txState->TargetPathId; - TPathElement::TPtr path = context.SS->PathsById.at(pathId); - - NIceDb::TNiceDb db(context.GetDB()); - - if (path->StepCreated == InvalidStepId) { - path->StepCreated = step; - context.SS->PersistCreateStep(db, pathId, step); - } - - TBlockStoreVolumeInfo::TPtr volume = context.SS->BlockStoreVolumes.at(pathId); - - auto oldVolumeSpace = volume->GetVolumeSpace(); - volume->FinishAlter(); - auto newVolumeSpace = volume->GetVolumeSpace(); - // Decrease in occupied space is applied on tx finish - auto domainDir = context.SS->PathsById.at(context.SS->ResolvePathIdForDomain(path)); - Y_ABORT_UNLESS(domainDir); - domainDir->ChangeVolumeSpaceCommit(newVolumeSpace, oldVolumeSpace); - - context.SS->PersistBlockStoreVolume(db, pathId, volume); - context.SS->PersistRemoveBlockStoreVolumeAlter(db, pathId); - - if (txState->TxType == TTxState::TxCreateBlockStoreVolume) { - auto parentDir = context.SS->PathsById.at(path->ParentPathId); - ++parentDir->DirAlterVersion; - context.SS->PersistPathDirAlterVersion(db, parentDir); - context.SS->ClearDescribePathCaches(parentDir); - context.OnComplete.PublishToSchemeBoard(OperationId, parentDir->PathId); - } + TPropose(TOperationId id); - context.SS->ClearDescribePathCaches(path); - context.OnComplete.PublishToSchemeBoard(OperationId, pathId); - - context.SS->ChangeTxState(db, OperationId, TTxState::Done); - return true; - } - - bool ProgressState(TOperationContext& context) override { - TTabletId ssId = context.SS->SelfTabletId(); - - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " ProgressState" - << ", at schemeshard: " << ssId); - - TTxState* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateBlockStoreVolume || txState->TxType == TTxState::TxAlterBlockStoreVolume); - - - context.OnComplete.ProposeToCoordinator(OperationId, txState->TargetPathId, TStepId(0)); - return false; - } + bool ProgressState(TOperationContext& context) override; + bool HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOperationContext& context) override; }; } // NBSVState @@ -1324,77 +223,17 @@ class TConfigurePartsAtTable: public TSubOperationState { << " operationId: " << OperationId; } - static bool IsExpectedTxType(TTxState::ETxType txType) { - switch (txType) { - case TTxState::TxCreateCdcStreamAtTable: - case TTxState::TxCreateCdcStreamAtTableWithInitialScan: - case TTxState::TxAlterCdcStreamAtTable: - case TTxState::TxAlterCdcStreamAtTableDropSnapshot: - case TTxState::TxDropCdcStreamAtTable: - case TTxState::TxDropCdcStreamAtTableDropSnapshot: - return true; - default: - return false; - } - } - protected: virtual void FillNotice(const TPathId& pathId, NKikimrTxDataShard::TFlatSchemeTransaction& tx, TOperationContext& context) const = 0; public: - explicit TConfigurePartsAtTable(TOperationId id) - : OperationId(id) - { - IgnoreMessages(DebugHint(), {}); - } - - bool ProgressState(TOperationContext& context) override { - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " ProgressState" - << ", at schemeshard: " << context.SS->TabletID()); - - auto* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - Y_ABORT_UNLESS(IsExpectedTxType(txState->TxType)); - const auto& pathId = txState->TargetPathId; - - if (NTableState::CheckPartitioningChangedForTableModification(*txState, context)) { - NTableState::UpdatePartitioningForTableModification(OperationId, *txState, context); - } - - NKikimrTxDataShard::TFlatSchemeTransaction tx; - context.SS->FillSeqNo(tx, context.SS->StartRound(*txState)); - FillNotice(pathId, tx, context); - - txState->ClearShardsInProgress(); - Y_ABORT_UNLESS(txState->Shards.size()); - - for (ui32 i = 0; i < txState->Shards.size(); ++i) { - const auto& idx = txState->Shards[i].Idx; - const auto datashardId = context.SS->ShardInfos[idx].TabletID; - auto ev = context.SS->MakeDataShardProposal(pathId, OperationId, tx.SerializeAsString(), context.Ctx); - context.OnComplete.BindMsgToPipe(OperationId, datashardId, idx, ev.Release()); - } - - txState->UpdateShardsInProgress(TTxState::ConfigureParts); - return false; - } - - bool HandleReply(TEvDataShard::TEvProposeTransactionResult::TPtr& ev, TOperationContext& context) override { - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply " << ev->Get()->ToString() - << ", at schemeshard: " << context.SS->TabletID()); + explicit TConfigurePartsAtTable(TOperationId id); - if (!NTableState::CollectProposeTransactionResults(OperationId, ev, context)) { - return false; - } - - return true; - } + bool ProgressState(TOperationContext& context) override; + bool HandleReply(TEvDataShard::TEvProposeTransactionResult::TPtr& ev, TOperationContext& context) override; private: const TOperationId OperationId; - }; // TConfigurePartsAtTable class TProposeAtTable: public TSubOperationState { @@ -1404,131 +243,31 @@ class TProposeAtTable: public TSubOperationState { << " operationId: " << OperationId; } - static bool IsExpectedTxType(TTxState::ETxType txType) { - switch (txType) { - case TTxState::TxCreateCdcStreamAtTable: - case TTxState::TxCreateCdcStreamAtTableWithInitialScan: - case TTxState::TxAlterCdcStreamAtTable: - case TTxState::TxAlterCdcStreamAtTableDropSnapshot: - case TTxState::TxDropCdcStreamAtTable: - case TTxState::TxDropCdcStreamAtTableDropSnapshot: - return true; - default: - return false; - } - } - public: - explicit TProposeAtTable(TOperationId id) - : OperationId(id) - { - IgnoreMessages(DebugHint(), {TEvDataShard::TEvProposeTransactionResult::EventType}); - } - - bool ProgressState(TOperationContext& context) override { - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " ProgressState" - << ", at schemeshard: " << context.SS->TabletID()); - - const auto* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - Y_ABORT_UNLESS(IsExpectedTxType(txState->TxType)); - - TSet<TTabletId> shardSet; - for (const auto& shard : txState->Shards) { - Y_ABORT_UNLESS(context.SS->ShardInfos.contains(shard.Idx)); - shardSet.insert(context.SS->ShardInfos.at(shard.Idx).TabletID); - } - - context.OnComplete.ProposeToCoordinator(OperationId, txState->TargetPathId, txState->MinStep, shardSet); - return false; - } - - bool HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOperationContext& context) override { - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " HandleReply TEvOperationPlan" - << ", step: " << ev->Get()->StepId - << ", at schemeshard: " << context.SS->TabletID()); - - const auto* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - Y_ABORT_UNLESS(IsExpectedTxType(txState->TxType)); - const auto& pathId = txState->TargetPathId; - - Y_ABORT_UNLESS(context.SS->PathsById.contains(pathId)); - auto path = context.SS->PathsById.at(pathId); - - Y_ABORT_UNLESS(context.SS->Tables.contains(pathId)); - auto table = context.SS->Tables.at(pathId); - - table->AlterVersion += 1; + explicit TProposeAtTable(TOperationId id); - NIceDb::TNiceDb db(context.GetDB()); - context.SS->PersistTableAlterVersion(db, pathId, table); - - context.SS->ClearDescribePathCaches(path); - context.OnComplete.PublishToSchemeBoard(OperationId, pathId); - - context.SS->ChangeTxState(db, OperationId, TTxState::ProposedWaitParts); - return true; - } - - bool HandleReply(TEvDataShard::TEvSchemaChanged::TPtr& ev, TOperationContext& context) override { - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << " TEvDataShard::TEvSchemaChanged" - << " triggers early, save it" - << ", at schemeshard: " << context.SS->TabletID()); - - NTableState::CollectSchemaChanged(OperationId, ev, context); - return false; - } + bool ProgressState(TOperationContext& context) override; + bool HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOperationContext& context) override; + bool HandleReply(TEvDataShard::TEvSchemaChanged::TPtr& ev, TOperationContext& context) override; protected: const TOperationId OperationId; - }; // TProposeAtTable class TProposeAtTableDropSnapshot: public TProposeAtTable { public: using TProposeAtTable::TProposeAtTable; - bool HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOperationContext& context) override { - TProposeAtTable::HandleReply(ev, context); - - const auto* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - const auto& pathId = txState->TargetPathId; - - Y_ABORT_UNLESS(context.SS->TablesWithSnapshots.contains(pathId)); - const auto snapshotTxId = context.SS->TablesWithSnapshots.at(pathId); - - auto it = context.SS->SnapshotTables.find(snapshotTxId); - if (it != context.SS->SnapshotTables.end()) { - it->second.erase(pathId); - if (it->second.empty()) { - context.SS->SnapshotTables.erase(it); - } - } - - context.SS->SnapshotsStepIds.erase(snapshotTxId); - context.SS->TablesWithSnapshots.erase(pathId); - - NIceDb::TNiceDb db(context.GetDB()); - context.SS->PersistDropSnapshot(db, snapshotTxId, pathId); - - context.SS->TabletCounters->Simple()[COUNTER_SNAPSHOTS_COUNT].Sub(1); - return true; - } - + bool HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOperationContext& context) override; }; // TProposeAtTableDropSnapshot } // NCdcStreamState namespace NForceDrop { -void ValidateNoTransactionOnPaths(TOperationId operationId, const THashSet<TPathId>& paths, TOperationContext& context); +void ValidateNoTransactionOnPaths(TOperationId operationId, const THashSet<TPathId>& paths, TOperationContext& context); void CollectShards(const THashSet<TPathId>& paths, TOperationId operationId, TTxState* txState, TOperationContext& context); + } // namespace NForceDrop -} // namespace NSchemeShard -} // namespace NKikimr +} // namespace NKikimr::NSchemeShard diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_common_bsv.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_common_bsv.cpp new file mode 100644 index 00000000000..f58c437f2c3 --- /dev/null +++ b/ydb/core/tx/schemeshard/schemeshard__operation_common_bsv.cpp @@ -0,0 +1,203 @@ +#include "schemeshard__operation_common.h" + +#include "schemeshard_private.h" +#include <ydb/core/base/hive.h> +#include <ydb/core/blockstore/core/blockstore.h> + + +namespace NKikimr::NSchemeShard::NBSVState { + +// NBSVState::TConfigureParts +// +TConfigureParts::TConfigureParts(TOperationId id) + : OperationId(id) +{ + IgnoreMessages(DebugHint(), {TEvHive::TEvCreateTabletReply::EventType}); +} + +bool TConfigureParts::HandleReply(TEvBlockStore::TEvUpdateVolumeConfigResponse::TPtr& ev, TOperationContext& context) { + TTabletId ssId = context.SS->SelfTabletId(); + + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " HandleReply TEvSetConfigResult" + << ", at schemeshard: " << ssId); + + TTxState* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateBlockStoreVolume || txState->TxType == TTxState::TxAlterBlockStoreVolume); + Y_ABORT_UNLESS(txState->State == TTxState::ConfigureParts); + + TTabletId tabletId = TTabletId(ev->Get()->Record.GetOrigin()); + NKikimrBlockStore::EStatus status = ev->Get()->Record.GetStatus(); + + // Schemeshard never sends invalid or outdated configs + Y_VERIFY_S(status == NKikimrBlockStore::OK || status == NKikimrBlockStore::ERROR_UPDATE_IN_PROGRESS, + "Unexpected error in UpdateVolumeConfigResponse," + << " status " << NKikimrBlockStore::EStatus_Name(status) + << " Tx " << OperationId + << " tablet " << tabletId); + + if (status == NKikimrBlockStore::ERROR_UPDATE_IN_PROGRESS) { + LOG_ERROR_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "BlockStore reconfiguration is in progress. We'll try to finish it later." + << " Tx " << OperationId + << " tablet " << tabletId); + return false; + } + + TShardIdx idx = context.SS->MustGetShardIdx(tabletId); + txState->ShardsInProgress.erase(idx); + + context.OnComplete.UnbindMsgFromPipe(OperationId, tabletId, idx); + + if (txState->ShardsInProgress.empty()) { + NIceDb::TNiceDb db(context.GetDB()); + context.SS->ChangeTxState(db, OperationId, TTxState::Propose); + context.OnComplete.ActivateTx(OperationId); + return true; + } + + return false; +} + +bool TConfigureParts::ProgressState(TOperationContext& context) { + TTabletId ssId = context.SS->SelfTabletId(); + + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " ProgressState" + << ", at schemeshard" << ssId); + + TTxState* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateBlockStoreVolume || txState->TxType == TTxState::TxAlterBlockStoreVolume); + Y_ABORT_UNLESS(!txState->Shards.empty()); + + txState->ClearShardsInProgress(); + + TBlockStoreVolumeInfo::TPtr volume = context.SS->BlockStoreVolumes[txState->TargetPathId]; + Y_VERIFY_S(volume, "volume is null. PathId: " << txState->TargetPathId); + + ui64 version = volume->AlterVersion; + const auto* volumeConfig = &volume->VolumeConfig; + if (volume->AlterData) { + version = volume->AlterData->AlterVersion; + volumeConfig = &volume->AlterData->VolumeConfig; + } + + for (auto shard : txState->Shards) { + if (shard.TabletType == ETabletType::BlockStorePartition || + shard.TabletType == ETabletType::BlockStorePartition2) { + continue; + } + + Y_ABORT_UNLESS(shard.TabletType == ETabletType::BlockStoreVolume); + TShardIdx shardIdx = shard.Idx; + TTabletId tabletId = context.SS->ShardInfos[shardIdx].TabletID; + + volume->VolumeTabletId = tabletId; + if (volume->AlterData) { + volume->AlterData->VolumeTabletId = tabletId; + volume->AlterData->VolumeShardIdx = shardIdx; + } + + TAutoPtr<TEvBlockStore::TEvUpdateVolumeConfig> event(new TEvBlockStore::TEvUpdateVolumeConfig()); + event->Record.SetTxId(ui64(OperationId.GetTxId())); + + event->Record.MutableVolumeConfig()->CopyFrom(*volumeConfig); + event->Record.MutableVolumeConfig()->SetVersion(version); + + for (const auto& p : volume->Shards) { + const auto& part = p.second; + const auto& partTabletId = context.SS->ShardInfos[p.first].TabletID; + auto info = event->Record.AddPartitions(); + info->SetPartitionId(part->PartitionId); + info->SetTabletId(ui64(partTabletId)); + } + + context.OnComplete.BindMsgToPipe(OperationId, tabletId, shardIdx, event.Release()); + + // Wait for results from this shard + txState->ShardsInProgress.insert(shardIdx); + } + + return false; +} + + +// NBSVState::TPropose +// +TPropose::TPropose(TOperationId id) + : OperationId(id) +{ + IgnoreMessages(DebugHint(), {TEvHive::TEvCreateTabletReply::EventType, TEvBlockStore::TEvUpdateVolumeConfigResponse::EventType}); +} + +bool TPropose::HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOperationContext& context) { + TStepId step = TStepId(ev->Get()->StepId); + TTabletId ssId = context.SS->SelfTabletId(); + + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " HandleReply TEvOperationPlan" + << ", at schemeshard: " << ssId); + + TTxState* txState = context.SS->FindTx(OperationId); + if (!txState) { + return false; + } + Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateBlockStoreVolume || txState->TxType == TTxState::TxAlterBlockStoreVolume); + + TPathId pathId = txState->TargetPathId; + TPathElement::TPtr path = context.SS->PathsById.at(pathId); + + NIceDb::TNiceDb db(context.GetDB()); + + if (path->StepCreated == InvalidStepId) { + path->StepCreated = step; + context.SS->PersistCreateStep(db, pathId, step); + } + + TBlockStoreVolumeInfo::TPtr volume = context.SS->BlockStoreVolumes.at(pathId); + + auto oldVolumeSpace = volume->GetVolumeSpace(); + volume->FinishAlter(); + auto newVolumeSpace = volume->GetVolumeSpace(); + // Decrease in occupied space is applied on tx finish + auto domainDir = context.SS->PathsById.at(context.SS->ResolvePathIdForDomain(path)); + Y_ABORT_UNLESS(domainDir); + domainDir->ChangeVolumeSpaceCommit(newVolumeSpace, oldVolumeSpace); + + context.SS->PersistBlockStoreVolume(db, pathId, volume); + context.SS->PersistRemoveBlockStoreVolumeAlter(db, pathId); + + if (txState->TxType == TTxState::TxCreateBlockStoreVolume) { + auto parentDir = context.SS->PathsById.at(path->ParentPathId); + ++parentDir->DirAlterVersion; + context.SS->PersistPathDirAlterVersion(db, parentDir); + context.SS->ClearDescribePathCaches(parentDir); + context.OnComplete.PublishToSchemeBoard(OperationId, parentDir->PathId); + } + + context.SS->ClearDescribePathCaches(path); + context.OnComplete.PublishToSchemeBoard(OperationId, pathId); + + context.SS->ChangeTxState(db, OperationId, TTxState::Done); + return true; +} + +bool TPropose::ProgressState(TOperationContext& context) { + TTabletId ssId = context.SS->SelfTabletId(); + + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " ProgressState" + << ", at schemeshard: " << ssId); + + TTxState* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateBlockStoreVolume || txState->TxType == TTxState::TxAlterBlockStoreVolume); + + + context.OnComplete.ProposeToCoordinator(OperationId, txState->TargetPathId, TStepId(0)); + return false; +} + +} // NKikimr::NSchemeShard::NBSVState diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_common_cdc_stream.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_common_cdc_stream.cpp new file mode 100644 index 00000000000..6c9d3b9058e --- /dev/null +++ b/ydb/core/tx/schemeshard/schemeshard__operation_common_cdc_stream.cpp @@ -0,0 +1,179 @@ +#include "schemeshard__operation_common.h" + +#include "schemeshard_private.h" +#include <ydb/core/base/hive.h> +#include <ydb/core/tx/datashard/datashard.h> + + +namespace NKikimr::NSchemeShard::NCdcStreamState { + +namespace { + +bool IsExpectedTxType(TTxState::ETxType txType) { + switch (txType) { + case TTxState::TxCreateCdcStreamAtTable: + case TTxState::TxCreateCdcStreamAtTableWithInitialScan: + case TTxState::TxAlterCdcStreamAtTable: + case TTxState::TxAlterCdcStreamAtTableDropSnapshot: + case TTxState::TxDropCdcStreamAtTable: + case TTxState::TxDropCdcStreamAtTableDropSnapshot: + return true; + default: + return false; + } +} + +} // namespace anonymous + + +// NCdcStreamState::TConfigurePartsAtTable +// +TConfigurePartsAtTable::TConfigurePartsAtTable(TOperationId id) + : OperationId(id) +{ + IgnoreMessages(DebugHint(), {}); +} + +bool TConfigurePartsAtTable::ProgressState(TOperationContext& context) { + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " ProgressState" + << ", at schemeshard: " << context.SS->SelfTabletId()); + + auto* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + Y_ABORT_UNLESS(IsExpectedTxType(txState->TxType)); + const auto& pathId = txState->TargetPathId; + + if (NTableState::CheckPartitioningChangedForTableModification(*txState, context)) { + NTableState::UpdatePartitioningForTableModification(OperationId, *txState, context); + } + + NKikimrTxDataShard::TFlatSchemeTransaction tx; + context.SS->FillSeqNo(tx, context.SS->StartRound(*txState)); + FillNotice(pathId, tx, context); + + txState->ClearShardsInProgress(); + Y_ABORT_UNLESS(txState->Shards.size()); + + for (ui32 i = 0; i < txState->Shards.size(); ++i) { + const auto& idx = txState->Shards[i].Idx; + const auto datashardId = context.SS->ShardInfos[idx].TabletID; + auto ev = context.SS->MakeDataShardProposal(pathId, OperationId, tx.SerializeAsString(), context.Ctx); + context.OnComplete.BindMsgToPipe(OperationId, datashardId, idx, ev.Release()); + } + + txState->UpdateShardsInProgress(TTxState::ConfigureParts); + return false; +} + +bool TConfigurePartsAtTable::HandleReply(TEvDataShard::TEvProposeTransactionResult::TPtr& ev, TOperationContext& context) { + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " HandleReply " << ev->Get()->ToString() + << ", at schemeshard: " << context.SS->SelfTabletId()); + + if (!NTableState::CollectProposeTransactionResults(OperationId, ev, context)) { + return false; + } + + return true; +} + + +// NCdcStreamState::TProposeAtTable +// +TProposeAtTable::TProposeAtTable(TOperationId id) + : OperationId(id) +{ + IgnoreMessages(DebugHint(), {TEvDataShard::TEvProposeTransactionResult::EventType}); +} + +bool TProposeAtTable::ProgressState(TOperationContext& context) { + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " ProgressState" + << ", at schemeshard: " << context.SS->SelfTabletId()); + + const auto* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + Y_ABORT_UNLESS(IsExpectedTxType(txState->TxType)); + + TSet<TTabletId> shardSet; + for (const auto& shard : txState->Shards) { + Y_ABORT_UNLESS(context.SS->ShardInfos.contains(shard.Idx)); + shardSet.insert(context.SS->ShardInfos.at(shard.Idx).TabletID); + } + + context.OnComplete.ProposeToCoordinator(OperationId, txState->TargetPathId, txState->MinStep, shardSet); + return false; +} + +bool TProposeAtTable::HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOperationContext& context) { + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " HandleReply TEvOperationPlan" + << ", step: " << ev->Get()->StepId + << ", at schemeshard: " << context.SS->SelfTabletId()); + + const auto* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + Y_ABORT_UNLESS(IsExpectedTxType(txState->TxType)); + const auto& pathId = txState->TargetPathId; + + Y_ABORT_UNLESS(context.SS->PathsById.contains(pathId)); + auto path = context.SS->PathsById.at(pathId); + + Y_ABORT_UNLESS(context.SS->Tables.contains(pathId)); + auto table = context.SS->Tables.at(pathId); + + table->AlterVersion += 1; + + NIceDb::TNiceDb db(context.GetDB()); + context.SS->PersistTableAlterVersion(db, pathId, table); + + context.SS->ClearDescribePathCaches(path); + context.OnComplete.PublishToSchemeBoard(OperationId, pathId); + + context.SS->ChangeTxState(db, OperationId, TTxState::ProposedWaitParts); + return true; +} + +bool TProposeAtTable::HandleReply(TEvDataShard::TEvSchemaChanged::TPtr& ev, TOperationContext& context) { + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " TEvDataShard::TEvSchemaChanged" + << " triggers early, save it" + << ", at schemeshard: " << context.SS->SelfTabletId()); + + NTableState::CollectSchemaChanged(OperationId, ev, context); + return false; +} + + +// NCdcStreamState::TProposeAtTableDropSnapshot +// +bool TProposeAtTableDropSnapshot::HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOperationContext& context) { + TProposeAtTable::HandleReply(ev, context); + + const auto* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + const auto& pathId = txState->TargetPathId; + + Y_ABORT_UNLESS(context.SS->TablesWithSnapshots.contains(pathId)); + const auto snapshotTxId = context.SS->TablesWithSnapshots.at(pathId); + + auto it = context.SS->SnapshotTables.find(snapshotTxId); + if (it != context.SS->SnapshotTables.end()) { + it->second.erase(pathId); + if (it->second.empty()) { + context.SS->SnapshotTables.erase(it); + } + } + + context.SS->SnapshotsStepIds.erase(snapshotTxId); + context.SS->TablesWithSnapshots.erase(pathId); + + NIceDb::TNiceDb db(context.GetDB()); + context.SS->PersistDropSnapshot(db, snapshotTxId, pathId); + + context.SS->TabletCounters->Simple()[COUNTER_SNAPSHOTS_COUNT].Sub(1); + return true; +} + +} // NKikimr::NSchemeShard::NCdcStreamState diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_common_pq.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_common_pq.cpp new file mode 100644 index 00000000000..004a9af3e15 --- /dev/null +++ b/ydb/core/tx/schemeshard/schemeshard__operation_common_pq.cpp @@ -0,0 +1,814 @@ +#include "schemeshard__operation_common.h" + +#include <ydb/core/persqueue/writer/source_id_encoding.h> + +#include "schemeshard_private.h" +#include <ydb/core/base/hive.h> +#include <ydb/core/persqueue/events/global.h> + + +namespace NKikimr::NSchemeShard::NPQState { + +namespace { + +void ParsePQTabletConfig(NKikimrPQ::TPQTabletConfig& config, + const TTopicInfo& pqGroup) +{ + const TString* source = &pqGroup.TabletConfig; + if (pqGroup.AlterData) { + if (!pqGroup.AlterData->TabletConfig.empty()) + source = &pqGroup.AlterData->TabletConfig; + } + + if (!source->empty()) { + Y_ABORT_UNLESS(ParseFromStringNoSizeLimit(config, *source)); + } +} + +void FillPartition(NKikimrPQ::TPQTabletConfig::TPartition& partition, const TTopicTabletInfo::TTopicPartitionInfo* pq, ui64 tabletId) { + partition.SetPartitionId(pq->PqId); + partition.SetCreateVersion(pq->CreateVersion); + if (pq->KeyRange) { + pq->KeyRange->SerializeToProto(*partition.MutableKeyRange()); + } + partition.SetStatus(pq->Status); + partition.MutableParentPartitionIds()->Reserve(pq->ParentPartitionIds.size()); + for (const auto parent : pq->ParentPartitionIds) { + partition.MutableParentPartitionIds()->AddAlreadyReserved(parent); + } + partition.MutableChildPartitionIds()->Reserve(pq->ChildPartitionIds.size()); + for (const auto children : pq->ChildPartitionIds) { + partition.MutableChildPartitionIds()->AddAlreadyReserved(children); + } + partition.SetTabletId(tabletId); +} + +void MakePQTabletConfig(const TOperationContext& context, + NKikimrPQ::TPQTabletConfig& config, + const TTopicInfo& pqGroup, + const TTopicTabletInfo& pqShard, + const TString& topicName, + const TString& topicPath, + const TString& cloudId, + const TString& folderId, + const TString& databaseId, + const TString& databasePath) +{ + ParsePQTabletConfig(config, pqGroup); + + config.SetTopicName(topicName); + config.SetTopicPath(topicPath); + config.MutablePartitionConfig()->SetTotalPartitions(pqGroup.AlterData ? pqGroup.AlterData->TotalGroupCount : pqGroup.TotalGroupCount); + + config.SetYcCloudId(cloudId); + config.SetYcFolderId(folderId); + config.SetYdbDatabaseId(databaseId); + config.SetYdbDatabasePath(databasePath); + + if (pqGroup.AlterData) { + config.SetVersion(pqGroup.AlterData->AlterVersion); + } + + THashSet<ui32> linkedPartitions; + + for(const auto& pq : pqShard.Partitions) { + config.AddPartitionIds(pq->PqId); + + auto& partition = *config.AddPartitions(); + FillPartition(partition, pq.Get(), 0); + + linkedPartitions.insert(pq->PqId); + linkedPartitions.insert(pq->ParentPartitionIds.begin(), pq->ParentPartitionIds.end()); + linkedPartitions.insert(pq->ChildPartitionIds.begin(), pq->ChildPartitionIds.end()); + for (auto c : pq->ChildPartitionIds) { + auto it = pqGroup.Partitions.find(c); + if (it == pqGroup.Partitions.end()) { + continue; + } + linkedPartitions.insert(it->second->ParentPartitionIds.begin(), it->second->ParentPartitionIds.end()); + } + } + + for(auto lp : linkedPartitions) { + auto it = pqGroup.Partitions.find(lp); + if (it == pqGroup.Partitions.end()) { + continue; + } + + auto* partitionInfo = it->second; + const auto& tabletId = context.SS->ShardInfos[partitionInfo->ShardIdx].TabletID; + + auto& partition = *config.AddAllPartitions(); + FillPartition(partition, partitionInfo, ui64(tabletId)); + } +} + +class TBootstrapConfigWrapper: public NKikimrPQ::TBootstrapConfig { + struct TSerializedProposeTransaction { + TString Value; + + static TSerializedProposeTransaction Serialize(const NKikimrPQ::TBootstrapConfig& value) { + NKikimrPQ::TEvProposeTransaction record; + record.MutableConfig()->MutableBootstrapConfig()->CopyFrom(value); + return {record.SerializeAsString()}; + } + }; + + struct TSerializedUpdateConfig { + TString Value; + + static TSerializedUpdateConfig Serialize(const NKikimrPQ::TBootstrapConfig& value) { + NKikimrPQ::TUpdateConfig record; + record.MutableBootstrapConfig()->CopyFrom(value); + return {record.SerializeAsString()}; + } + }; + + mutable std::optional<std::variant< + TSerializedProposeTransaction, + TSerializedUpdateConfig + >> PreSerialized; + + template <typename T> + const TString& Get() const { + if (!PreSerialized) { + PreSerialized.emplace(T::Serialize(*this)); + } + + const auto* value = std::get_if<T>(&PreSerialized.value()); + Y_ABORT_UNLESS(value); + + return value->Value; + } + +public: + const TString& GetPreSerializedProposeTransaction() const { + return Get<TSerializedProposeTransaction>(); + } + + const TString& GetPreSerializedUpdateConfig() const { + return Get<TSerializedUpdateConfig>(); + } +}; + +THolder<TEvPersQueue::TEvProposeTransaction> MakeEvProposeTransaction( + TTxId txId, + const TTopicInfo& pqGroup, + const TTopicTabletInfo& pqShard, + const TString& topicName, + const TString& topicPath, + const std::optional<TBootstrapConfigWrapper>& bootstrapConfig, + const TString& cloudId, + const TString& folderId, + const TString& databaseId, + const TString& databasePath, + TTxState::ETxType txType, + const TOperationContext& context + ) +{ + auto event = MakeHolder<TEvPersQueue::TEvProposeTransactionBuilder>(); + event->Record.SetTxId(ui64(txId)); + ActorIdToProto(context.SS->SelfId(), event->Record.MutableSourceActor()); + + MakePQTabletConfig(context, + *event->Record.MutableConfig()->MutableTabletConfig(), + pqGroup, + pqShard, + topicName, + topicPath, + cloudId, + folderId, + databaseId, + databasePath); + if (bootstrapConfig) { + Y_ABORT_UNLESS(txType == TTxState::TxCreatePQGroup); + event->PreSerializedData += bootstrapConfig->GetPreSerializedProposeTransaction(); + } + + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "Propose configure PersQueue" << + ", message: " << event->Record.ShortUtf8DebugString()); + + return event; +} + +THolder<TEvPersQueue::TEvUpdateConfig> MakeEvUpdateConfig( + TTxId txId, + const TTopicInfo& pqGroup, + const TTopicTabletInfo& pqShard, + const TString& topicName, + const TString& topicPath, + const std::optional<TBootstrapConfigWrapper>& bootstrapConfig, + const TString& cloudId, + const TString& folderId, + const TString& databaseId, + const TString& databasePath, + TTxState::ETxType txType, + const TOperationContext& context + ) +{ + auto event = MakeHolder<TEvPersQueue::TEvUpdateConfigBuilder>(); + event->Record.SetTxId(ui64(txId)); + + MakePQTabletConfig(context, + *event->Record.MutableTabletConfig(), + pqGroup, + pqShard, + topicName, + topicPath, + cloudId, + folderId, + databaseId, + databasePath); + if (bootstrapConfig) { + Y_ABORT_UNLESS(txType == TTxState::TxCreatePQGroup); + event->PreSerializedData += bootstrapConfig->GetPreSerializedUpdateConfig(); + } + + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "Propose configure PersQueue" << + ", message: " << event->Record.ShortUtf8DebugString()); + + return event; +} + +} // anonymous namespace + + +bool CollectPQConfigChanged(const TOperationId& operationId, + const TEvPersQueue::TEvProposeTransactionResult::TPtr& ev, + TOperationContext& context) +{ + Y_ABORT_UNLESS(context.SS->FindTx(operationId)); + TTxState& txState = *context.SS->FindTx(operationId); + + const auto& evRecord = ev->Get()->Record; + if (evRecord.GetStatus() == NKikimrPQ::TEvProposeTransactionResult::COMPLETE) { + const auto ssId = context.SS->SelfTabletId(); + const TTabletId shardId(evRecord.GetOrigin()); + + const auto shardIdx = context.SS->MustGetShardIdx(shardId); + Y_ABORT_UNLESS(context.SS->ShardInfos.contains(shardIdx)); + + Y_ABORT_UNLESS(txState.State == TTxState::Propose); + + txState.ShardsInProgress.erase(shardIdx); + + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "CollectPQConfigChanged accept TEvPersQueue::TEvProposeTransactionResult" + << ", operationId: " << operationId + << ", shardIdx: " << shardIdx + << ", shard: " << shardId + << ", left await: " << txState.ShardsInProgress.size() + << ", txState.State: " << TTxState::StateName(txState.State) + << ", txState.ReadyForNotifications: " << txState.ReadyForNotifications + << ", at schemeshard: " << ssId); + } + + return txState.ShardsInProgress.empty(); +} + +bool CollectPQConfigChanged(const TOperationId& operationId, + const TEvPersQueue::TEvProposeTransactionAttachResult::TPtr& ev, + TOperationContext& context) +{ + Y_ABORT_UNLESS(context.SS->FindTx(operationId)); + TTxState& txState = *context.SS->FindTx(operationId); + + // + // The PQ tablet can perform a transaction and send a TEvProposeTransactionResult(COMPLETE) response. + // The SchemeShard tablet can restart at this point. After restarting at the TPropose step, it will + // send the TEvProposeTransactionAttach message to the PQ tablets. If the NODATA status is specified in + // the response TEvProposeTransactionAttachResult, then the PQ tablet has already completed the transaction. + // Otherwise, she continues to execute the transaction + // + + const auto& evRecord = ev->Get()->Record; + if (evRecord.GetStatus() != NKikimrProto::NODATA) { + // + // If the PQ tablet returned something other than NODATA, then it continues to execute the transaction + // + return txState.ShardsInProgress.empty(); + } + + // + // Otherwise, she has already completed the transaction and has forgotten about it. Then we can + // remove PQ tablet from the list of shards + // + + const auto ssId = context.SS->SelfTabletId(); + const TTabletId shardId(evRecord.GetTabletId()); + + const auto shardIdx = context.SS->MustGetShardIdx(shardId); + Y_ABORT_UNLESS(context.SS->ShardInfos.contains(shardIdx)); + + Y_ABORT_UNLESS(txState.State == TTxState::Propose); + + txState.ShardsInProgress.erase(shardIdx); + + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "CollectPQConfigChanged accept TEvPersQueue::TEvProposeTransactionAttachResult" + << ", operationId: " << operationId + << ", shardIdx: " << shardIdx + << ", shard: " << shardId + << ", left await: " << txState.ShardsInProgress.size() + << ", txState.State: " << TTxState::StateName(txState.State) + << ", txState.ReadyForNotifications: " << txState.ReadyForNotifications + << ", at schemeshard: " << ssId); + + return txState.ShardsInProgress.empty(); +} + + +// NPQState::TConfigureParts +// +TConfigureParts::TConfigureParts(TOperationId id) + : OperationId(id) +{ + IgnoreMessages(DebugHint(), {TEvHive::TEvCreateTabletReply::EventType}); +} + +bool TConfigureParts::HandleReply(TEvPersQueue::TEvProposeTransactionResult::TPtr& ev, TOperationContext& context) +{ + const TTabletId ssId = context.SS->SelfTabletId(); + + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " HandleReply TEvProposeTransactionResult" + << ", at schemeshard: " << ssId); + + return NPQState::CollectProposeTransactionResults(OperationId, ev, context); +} + +bool TConfigureParts::HandleReply(TEvPersQueue::TEvUpdateConfigResponse::TPtr& ev, TOperationContext& context) { + TTabletId ssId = context.SS->SelfTabletId(); + + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " HandleReply TEvUpdateConfigResponse" + << " at tablet" << ssId); + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " HandleReply TEvUpdateConfigResponse" + << " message: " << ev->Get()->Record.ShortUtf8DebugString() + << " at tablet" << ssId); + + TTxState* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreatePQGroup || txState->TxType == TTxState::TxAlterPQGroup); + + TTabletId tabletId = TTabletId(ev->Get()->Record.GetOrigin()); + NKikimrPQ::EStatus status = ev->Get()->Record.GetStatus(); + + // Schemeshard always sends a valid config to the PQ tablet and PQ tablet is not supposed to reject it + // Also Schemeshard never send multiple different config updates simultaneously (same config + // can be sent more than once in case of retries due to restarts or disconnects) + // If PQ tablet is not able to save a valid config it should kill itself and restart without + // sending error response + Y_VERIFY_S(status == NKikimrPQ::OK || status == NKikimrPQ::ERROR_UPDATE_IN_PROGRESS, + "Unexpected error in UpdateConfigResponse," + << " status: " << NKikimrPQ::EStatus_Name(status) + << " Tx " << OperationId + << " tablet "<< tabletId + << " at schemeshard: " << ssId); + + if (status == NKikimrPQ::ERROR_UPDATE_IN_PROGRESS) { + LOG_ERROR_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "PQ reconfiguration is in progress. We'll try to finish it later." + << " Tx " << OperationId + << " tablet " << tabletId + << " at schemeshard: " << ssId); + return false; + } + + Y_ABORT_UNLESS(txState->State == TTxState::ConfigureParts); + + TShardIdx idx = context.SS->MustGetShardIdx(tabletId); + txState->ShardsInProgress.erase(idx); + + // Detach datashard pipe + context.OnComplete.UnbindMsgFromPipe(OperationId, tabletId, idx); + + if (txState->ShardsInProgress.empty()) { + NIceDb::TNiceDb db(context.GetDB()); + context.SS->ChangeTxState(db, OperationId, TTxState::Propose); + return true; + } + + return false; +} + +bool TConfigureParts::ProgressState(TOperationContext& context) { + TTabletId ssId = context.SS->SelfTabletId(); + + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() + << " HandleReply ProgressState" + << ", at schemeshard: " << ssId); + + TTxState* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreatePQGroup || txState->TxType == TTxState::TxAlterPQGroup); + + txState->ClearShardsInProgress(); + + TString topicName = context.SS->PathsById.at(txState->TargetPathId)->Name; + Y_VERIFY_S(topicName.size(), + "topicName is empty" + <<", pathId: " << txState->TargetPathId); + + TTopicInfo::TPtr pqGroup = context.SS->Topics[txState->TargetPathId]; + Y_VERIFY_S(pqGroup, + "pqGroup is null" + << ", pathId " << txState->TargetPathId); + + const TPathElement::TPtr dbRootEl = context.SS->PathsById.at(context.SS->RootPathId()); + TString cloudId; + if (dbRootEl->UserAttrs->Attrs.contains("cloud_id")) { + cloudId = dbRootEl->UserAttrs->Attrs.at("cloud_id"); + } + TString folderId; + if (dbRootEl->UserAttrs->Attrs.contains("folder_id")) { + folderId = dbRootEl->UserAttrs->Attrs.at("folder_id"); + } + TString databaseId; + if (dbRootEl->UserAttrs->Attrs.contains("database_id")) { + databaseId = dbRootEl->UserAttrs->Attrs.at("database_id"); + } + + TString databasePath = TPath::Init(context.SS->RootPathId(), context.SS).PathString(); + auto topicPath = TPath::Init(txState->TargetPathId, context.SS); + + std::optional<TBootstrapConfigWrapper> bootstrapConfig; + if (txState->TxType == TTxState::TxCreatePQGroup && topicPath.Parent().IsCdcStream()) { + bootstrapConfig.emplace(); + + auto tablePath = topicPath.Parent().Parent(); // table/cdc_stream/topic + Y_ABORT_UNLESS(tablePath.IsResolved()); + + Y_ABORT_UNLESS(context.SS->Tables.contains(tablePath.Base()->PathId)); + auto table = context.SS->Tables.at(tablePath.Base()->PathId); + + const auto& partitions = table->GetPartitions(); + + for (ui32 i = 0; i < partitions.size(); ++i) { + const auto& cur = partitions.at(i); + + Y_ABORT_UNLESS(context.SS->ShardInfos.contains(cur.ShardIdx)); + const auto& shard = context.SS->ShardInfos.at(cur.ShardIdx); + + auto& mg = *bootstrapConfig->AddExplicitMessageGroups(); + mg.SetId(NPQ::NSourceIdEncoding::EncodeSimple(ToString(shard.TabletID))); + + if (i != partitions.size() - 1) { + mg.MutableKeyRange()->SetToBound(cur.EndOfRange); + } + + if (i) { + const auto& prev = partitions.at(i - 1); + mg.MutableKeyRange()->SetFromBound(prev.EndOfRange); + } + } + } + + for (auto shard : txState->Shards) { + TShardIdx idx = shard.Idx; + TTabletId tabletId = context.SS->ShardInfos.at(idx).TabletID; + + if (shard.TabletType == ETabletType::PersQueue) { + TTopicTabletInfo::TPtr pqShard = pqGroup->Shards.at(idx); + Y_VERIFY_S(pqShard, "pqShard is null, idx is " << idx << " has was "<< THash<TShardIdx>()(idx)); + + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "Propose configure PersQueue" + << ", opId: " << OperationId + << ", tabletId: " << tabletId + << ", Partitions size: " << pqShard->Partitions.size() + << ", at schemeshard: " << ssId); + + THolder<NActors::IEventBase> event; + if (context.SS->EnablePQConfigTransactionsAtSchemeShard) { + event = MakeEvProposeTransaction(OperationId.GetTxId(), + *pqGroup, + *pqShard, + topicName, + topicPath.PathString(), + bootstrapConfig, + cloudId, + folderId, + databaseId, + databasePath, + txState->TxType, + context); + } else { + event = MakeEvUpdateConfig(OperationId.GetTxId(), + *pqGroup, + *pqShard, + topicName, + topicPath.PathString(), + bootstrapConfig, + cloudId, + folderId, + databaseId, + databasePath, + txState->TxType, + context); + } + + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "Propose configure PersQueue" + << ", opId: " << OperationId + << ", tabletId: " << tabletId + << ", at schemeshard: " << ssId); + + context.OnComplete.BindMsgToPipe(OperationId, tabletId, idx, event.Release()); + } else { + Y_ABORT_UNLESS(shard.TabletType == ETabletType::PersQueueReadBalancer); + + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "Propose configure PersQueueReadBalancer" + << ", opId: " << OperationId + << ", tabletId: " << tabletId + << ", at schemeshard: " << ssId); + + pqGroup->BalancerTabletID = tabletId; + if (pqGroup->AlterData) { + pqGroup->AlterData->BalancerTabletID = tabletId; + } + + if (pqGroup->AlterData) { + pqGroup->AlterData->BalancerShardIdx = idx; + } + + TAutoPtr<TEvPersQueue::TEvUpdateBalancerConfig> event(new TEvPersQueue::TEvUpdateBalancerConfig()); + event->Record.SetTxId(ui64(OperationId.GetTxId())); + + ParsePQTabletConfig(*event->Record.MutableTabletConfig(), *pqGroup); + + Y_ABORT_UNLESS(pqGroup->AlterData); + + event->Record.SetTopicName(topicName); + event->Record.SetPathId(txState->TargetPathId.LocalPathId); + event->Record.SetPath(TPath::Init(txState->TargetPathId, context.SS).PathString()); + event->Record.SetPartitionPerTablet(pqGroup->AlterData ? pqGroup->AlterData->MaxPartsPerTablet : pqGroup->MaxPartsPerTablet); + event->Record.SetSchemeShardId(ui64(context.SS->SelfTabletId())); + + event->Record.SetTotalGroupCount(pqGroup->AlterData ? pqGroup->AlterData->TotalGroupCount : pqGroup->TotalGroupCount); + event->Record.SetNextPartitionId(pqGroup->AlterData ? pqGroup->AlterData->NextPartitionId : pqGroup->NextPartitionId); + + event->Record.SetVersion(pqGroup->AlterData->AlterVersion); + + for (const auto& p : pqGroup->Shards) { + const auto& pqShard = p.second; + const auto& tabletId = context.SS->ShardInfos[p.first].TabletID; + auto tablet = event->Record.AddTablets(); + tablet->SetTabletId(ui64(tabletId)); + tablet->SetOwner(ui64(context.SS->SelfTabletId())); + tablet->SetIdx(ui64(p.first.GetLocalId())); + for (const auto& pq : pqShard->Partitions) { + auto info = event->Record.AddPartitions(); + info->SetPartition(pq->PqId); + info->SetTabletId(ui64(tabletId)); + info->SetGroup(pq->GroupId); + info->SetCreateVersion(pq->CreateVersion); + + if (pq->KeyRange) { + pq->KeyRange->SerializeToProto(*info->MutableKeyRange()); + } + info->SetStatus(pq->Status); + info->MutableParentPartitionIds()->Reserve(pq->ParentPartitionIds.size()); + for (const auto parent : pq->ParentPartitionIds) { + info->MutableParentPartitionIds()->AddAlreadyReserved(parent); + } + info->MutableChildPartitionIds()->Reserve(pq->ChildPartitionIds.size()); + for (const auto children : pq->ChildPartitionIds) { + info->MutableChildPartitionIds()->AddAlreadyReserved(children); + } + } + } + + if (const ui64 subDomainPathId = context.SS->ResolvePathIdForDomain(txState->TargetPathId).LocalPathId) { + event->Record.SetSubDomainPathId(subDomainPathId); + } + + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "Propose configure PersQueueReadBalancer" + << ", opId: " << OperationId + << ", tabletId: " << tabletId + << ", message: " << event->Record.ShortUtf8DebugString() + << ", at schemeshard: " << ssId); + + context.OnComplete.BindMsgToPipe(OperationId, tabletId, idx, event.Release()); + } + } + + txState->UpdateShardsInProgress(); + return false; +} + + +// NPQState::TPropose +// +TPropose::TPropose(TOperationId id) + : OperationId(id) +{ + IgnoreMessages(DebugHint(), {TEvHive::TEvCreateTabletReply::EventType, TEvPersQueue::TEvUpdateConfigResponse::EventType}); +} + +bool TPropose::HandleReply(TEvPersQueue::TEvProposeTransactionResult::TPtr& ev, TOperationContext& context) +{ + const TTabletId ssId = context.SS->SelfTabletId(); + const auto& evRecord = ev->Get()->Record; + + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " HandleReply TEvProposeTransactionResult" + << " triggers early" + << ", at schemeshard: " << ssId + << " message# " << evRecord.ShortDebugString()); + + const bool collected = CollectPQConfigChanged(OperationId, ev, context); + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " HandleReply TEvProposeTransactionResult" + << " CollectPQConfigChanged: " << (collected ? "true" : "false")); + + return TryPersistState(context); +} + +bool TPropose::HandleReply(TEvPersQueue::TEvProposeTransactionAttachResult::TPtr& ev, TOperationContext& context) +{ + const auto ssId = context.SS->SelfTabletId(); + const auto& evRecord = ev->Get()->Record; + + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " HandleReply TEvProposeTransactionAttachResult" + << " triggers early" + << ", at schemeshard: " << ssId + << " message# " << evRecord.ShortDebugString()); + + const bool collected = CollectPQConfigChanged(OperationId, ev, context); + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " HandleReply TEvProposeTransactionAttachResult" + << " CollectPQConfigChanged: " << (collected ? "true" : "false")); + + return TryPersistState(context); +} + +bool TPropose::HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOperationContext& context) +{ + TStepId step = TStepId(ev->Get()->StepId); + TTabletId ssId = context.SS->SelfTabletId(); + + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() + << " HandleReply TEvOperationPlan" + << ", step: " << step + << ", at tablet: " << ssId); + + TTxState* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreatePQGroup || txState->TxType == TTxState::TxAlterPQGroup); + + TPathId pathId = txState->TargetPathId; + TPathElement::TPtr path = context.SS->PathsById.at(pathId); + + NIceDb::TNiceDb db(context.GetDB()); + + if (path->StepCreated == InvalidStepId) { + path->StepCreated = step; + context.SS->PersistCreateStep(db, pathId, step); + } + + return TryPersistState(context); +} + +bool TPropose::ProgressState(TOperationContext& context) +{ + TTabletId ssId = context.SS->SelfTabletId(); + + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "NPQState::TPropose ProgressState" + << ", operationId: " << OperationId + << ", at schemeshard: " << ssId); + + TTxState* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreatePQGroup || txState->TxType == TTxState::TxAlterPQGroup); + + // + // If the program works according to the new scheme, then we must add PQ tablets to the list for + // the Coordinator. At this stage, we cannot rely on the value of + // the EnablePQConfigTransactionsAtSchemeShard flag. Because the operation could have started on one tablet + // and moved to another by that time. + // + // Therefore, here we check the value of the minStep field, which is filled in in + // the TEvProposeTransactionResult handler + // + TSet<TTabletId> shardSet; + if (ui64(txState->MinStep) > 0) { + PrepareShards(*txState, shardSet, context); + } + context.OnComplete.ProposeToCoordinator(OperationId, txState->TargetPathId, txState->MinStep, shardSet); + + return false; +} + +void TPropose::PrepareShards(TTxState& txState, TSet<TTabletId>& shardSet, TOperationContext& context) +{ + txState.UpdateShardsInProgress(); + + for (const auto& shard : txState.Shards) { + const TShardIdx idx = shard.Idx; + // + // The operation involves the ETabletType::PersQueue and Tabletype::PersQueueReadBalancer shards. + // The program receives responses from PersQueueReadBalancer at the previous stage. At this stage, + // it only expects TEvProposeTransactionResult from PersQueue + // + if (shard.TabletType == ETabletType::PersQueue) { + const TTabletId tablet = context.SS->ShardInfos.at(idx).TabletID; + + shardSet.insert(tablet); + + // + // By this point, the SchemeShard tablet could restart and the actor ID changed. Therefore, we send + // the TEvProposeTransactionAttach message to the PQ tablets so that they recognize the new recipient + // + SendEvProposeTransactionAttach(idx, tablet, context); + } else { + txState.ShardsInProgress.erase(idx); + } + } +} + +void TPropose::SendEvProposeTransactionAttach(TShardIdx shard, TTabletId tablet, + TOperationContext& context) +{ + auto event = + MakeHolder<TEvPersQueue::TEvProposeTransactionAttach>(ui64(tablet), + ui64(OperationId.GetTxId())); + context.OnComplete.BindMsgToPipe(OperationId, tablet, shard, event.Release()); +} + +bool TPropose::CanPersistState(const TTxState& txState, + TOperationContext& context) +{ + if (!txState.ShardsInProgress.empty()) { + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " can't persist state: " << + "ShardsInProgress is not empty, remain: " << txState.ShardsInProgress.size()); + return false; + } + + PathId = txState.TargetPathId; + Path = context.SS->PathsById.at(PathId); + + if (Path->StepCreated == InvalidStepId) { + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << " can't persist state: " << + "StepCreated is invalid"); + return false; + } + + return true; +} + +void TPropose::PersistState(const TTxState& txState, + TOperationContext& context) const +{ + NIceDb::TNiceDb db(context.GetDB()); + + if (txState.TxType == TTxState::TxCreatePQGroup) { + auto parentDir = context.SS->PathsById.at(Path->ParentPathId); + ++parentDir->DirAlterVersion; + context.SS->PersistPathDirAlterVersion(db, parentDir); + context.SS->ClearDescribePathCaches(parentDir); + context.OnComplete.PublishToSchemeBoard(OperationId, parentDir->PathId); + } + + context.SS->ClearDescribePathCaches(Path); + context.OnComplete.PublishToSchemeBoard(OperationId, PathId); + + TTopicInfo::TPtr pqGroup = context.SS->Topics[PathId]; + + NKikimrPQ::TPQTabletConfig tabletConfig = pqGroup->GetTabletConfig(); + NKikimrPQ::TPQTabletConfig newTabletConfig = pqGroup->AlterData->GetTabletConfig(); + + pqGroup->FinishAlter(); + + context.SS->PersistPersQueueGroup(db, PathId, pqGroup); + context.SS->PersistRemovePersQueueGroupAlter(db, PathId); + + context.SS->ChangeTxState(db, OperationId, TTxState::Done); +} + +bool TPropose::TryPersistState(TOperationContext& context) +{ + TTxState* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + + if (!CanPersistState(*txState, context)) { + return false; + } + + PersistState(*txState, context); + + return true; +} + +} // namespace NKikimr::NSchemeShard::NPQState diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_common_subdomain.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_common_subdomain.cpp new file mode 100644 index 00000000000..7166bb69c2b --- /dev/null +++ b/ydb/core/tx/schemeshard/schemeshard__operation_common_subdomain.cpp @@ -0,0 +1,393 @@ +#include "schemeshard__operation_part.h" +#include "schemeshard__operation_common_subdomain.h" + +#include <ydb/core/base/hive.h> +#include "schemeshard_private.h" + +#include <ydb/core/statistics/events.h> + +namespace NKikimr::NSchemeShard::NSubDomainState { + +// NSubDomainState::TConfigureParts +// +TConfigureParts::TConfigureParts(TOperationId id) + : OperationId(id) +{ + IgnoreMessages(DebugHint(), {TEvHive::TEvCreateTabletReply::EventType}); +} + +bool TConfigureParts::HandleReply(TEvSchemeShard::TEvInitTenantSchemeShardResult::TPtr& ev, TOperationContext& context) { + TTabletId ssId = context.SS->SelfTabletId(); + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() + << " HandleReply TEvInitTenantSchemeShardResult" + << " operationId: " << OperationId + << " at schemeshard: " << ssId); + + TTxState* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateSubDomain + || txState->TxType == TTxState::TxAlterSubDomain + || txState->TxType == TTxState::TxAlterExtSubDomain + || txState->TxType == TTxState::TxAlterExtSubDomainCreateHive + ); + + const auto& record = ev->Get()->Record; + + NIceDb::TNiceDb db(context.GetDB()); + + TTabletId tabletId = TTabletId(record.GetTenantSchemeShard()); + auto status = record.GetStatus(); + + auto shardIdx = context.SS->MustGetShardIdx(tabletId); + Y_ABORT_UNLESS(context.SS->ShardInfos.contains(shardIdx)); + + if (status != NKikimrScheme::EStatus::StatusSuccess && status != NKikimrScheme::EStatus::StatusAlreadyExists) { + LOG_CRIT_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() + << " Got error status on SubDomain Configure" + << "from tenant schemeshard tablet: " << tabletId + << " shard: " << shardIdx + << " status: " << NKikimrScheme::EStatus_Name(status) + << " opId: " << OperationId + << " schemeshard: " << ssId); + return false; + } + + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() + << " Got OK TEvInitTenantSchemeShardResult from schemeshard" + << " tablet: " << tabletId + << " shardIdx: " << shardIdx + << " at schemeshard: " << ssId); + + txState->ShardsInProgress.erase(shardIdx); + context.OnComplete.UnbindMsgFromPipe(OperationId, tabletId, shardIdx); + + if (txState->ShardsInProgress.empty()) { + // All tablets have replied so we can done this transaction + context.SS->ChangeTxState(db, OperationId, TTxState::Propose); + context.OnComplete.ActivateTx(OperationId); + return true; + } + + return false; +} + +bool TConfigureParts::HandleReply(TEvSubDomain::TEvConfigureStatus::TPtr& ev, TOperationContext& context) { + TTabletId ssId = context.SS->SelfTabletId(); + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() + << " HandleReply TEvConfigureStatus" + << " operationId:" << OperationId + << " at schemeshard:" << ssId); + + TTxState* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateSubDomain + || txState->TxType == TTxState::TxAlterSubDomain + || txState->TxType == TTxState::TxAlterExtSubDomain + || txState->TxType == TTxState::TxAlterExtSubDomainCreateHive + ); + + const auto& record = ev->Get()->Record; + + NIceDb::TNiceDb db(context.GetDB()); + + TTabletId tabletId = TTabletId(record.GetOnTabletId()); + auto status = record.GetStatus(); + + auto shardIdx = context.SS->MustGetShardIdx(tabletId); + Y_ABORT_UNLESS(context.SS->ShardInfos.contains(shardIdx)); + + if (status == NKikimrTx::TEvSubDomainConfigurationAck::REJECT) { + LOG_CRIT_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() + << " Got REJECT on SubDomain Configure" + << "from tablet: " << tabletId + << " shard: " << shardIdx + << " opId: " << OperationId + << " schemeshard: " << ssId); + return false; + } + + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() << + " Got OK TEvConfigureStatus from " + << " tablet# " << tabletId + << " shardIdx# " << shardIdx + << " at schemeshard# " << ssId); + + txState->ShardsInProgress.erase(shardIdx); + context.OnComplete.UnbindMsgFromPipe(OperationId, tabletId, shardIdx); + + if (txState->ShardsInProgress.empty()) { + // All tablets have replied so we can done this transaction + context.SS->ChangeTxState(db, OperationId, TTxState::Propose); + context.OnComplete.ActivateTx(OperationId); + return true; + } + + return false; +} + + +bool TConfigureParts::ProgressState(TOperationContext& context) { + TTabletId ssId = context.SS->SelfTabletId(); + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + DebugHint() + << " ProgressState" + << ", at schemeshard: " << ssId); + + TTxState* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateSubDomain + || txState->TxType == TTxState::TxAlterSubDomain + || txState->TxType == TTxState::TxAlterExtSubDomain + || txState->TxType == TTxState::TxAlterExtSubDomainCreateHive + ); + + txState->ClearShardsInProgress(); + + if (txState->Shards.empty()) { + NIceDb::TNiceDb db(context.GetDB()); + context.SS->ChangeTxState(db, OperationId, TTxState::Propose); + context.OnComplete.ActivateTx(OperationId); + return true; + } + + auto pathId = txState->TargetPathId; + Y_ABORT_UNLESS(context.SS->PathsById.contains(pathId)); + TPath path = TPath::Init(pathId, context.SS); + + Y_ABORT_UNLESS(context.SS->SubDomains.contains(pathId)); + auto subDomain = context.SS->SubDomains.at(pathId); + auto alterData = subDomain->GetAlter(); + Y_ABORT_UNLESS(alterData); + alterData->Initialize(context.SS->ShardInfos); + auto processing = alterData->GetProcessingParams(); + auto storagePools = alterData->GetStoragePools(); + auto& schemeLimits = subDomain->GetSchemeLimits(); + + for (auto& shard : txState->Shards) { + if (shard.Operation != TTxState::CreateParts) { + continue; + } + TShardIdx idx = shard.Idx; + Y_ABORT_UNLESS(context.SS->ShardInfos.contains(idx)); + TTabletId tabletID = context.SS->ShardInfos[idx].TabletID; + auto type = context.SS->ShardInfos[idx].TabletType; + + switch (type) { + case ETabletType::Coordinator: + case ETabletType::Mediator: { + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "Send configure request to coordinator/mediator: " << tabletID << + " opId: " << OperationId << + " schemeshard: " << ssId); + shard.Operation = TTxState::ConfigureParts; + auto event = new TEvSubDomain::TEvConfigure(processing); + context.OnComplete.BindMsgToPipe(OperationId, tabletID, idx, event); + break; + } + case ETabletType::Hive: { + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "Send configure request to hive: " << tabletID << + " opId: " << OperationId << + " schemeshard: " << ssId); + shard.Operation = TTxState::ConfigureParts; + auto event = new TEvHive::TEvConfigureHive(TSubDomainKey(pathId.OwnerId, pathId.LocalPathId)); + context.OnComplete.BindMsgToPipe(OperationId, tabletID, idx, event); + break; + } + case ETabletType::SysViewProcessor: { + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "Send configure request to sys view processor: " << tabletID << + " opId: " << OperationId << + " schemeshard: " << ssId); + auto event = new NSysView::TEvSysView::TEvConfigureProcessor(path.PathString()); + shard.Operation = TTxState::ConfigureParts; + context.OnComplete.BindMsgToPipe(OperationId, tabletID, idx, event); + break; + } + case ETabletType::StatisticsAggregator: { + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "Send configure request to statistics aggregator: " << tabletID << + " opId: " << OperationId << + " schemeshard: " << ssId); + auto event = new NStat::TEvStatistics::TEvConfigureAggregator(path.PathString()); + shard.Operation = TTxState::ConfigureParts; + context.OnComplete.BindMsgToPipe(OperationId, tabletID, idx, event); + break; + } + case ETabletType::SchemeShard: { + auto event = new TEvSchemeShard::TEvInitTenantSchemeShard(ui64(ssId), + pathId.LocalPathId, path.PathString(), + path.Base()->Owner, path.GetEffectiveACL(), path.GetEffectiveACLVersion(), + processing, storagePools, + path.Base()->UserAttrs->Attrs, path.Base()->UserAttrs->AlterVersion, + schemeLimits, ui64(alterData->GetSharedHive()), alterData->GetResourcesDomainId() + ); + if (alterData->GetDeclaredSchemeQuotas()) { + event->Record.MutableDeclaredSchemeQuotas()->CopyFrom(*alterData->GetDeclaredSchemeQuotas()); + } + if (alterData->GetDatabaseQuotas()) { + event->Record.MutableDatabaseQuotas()->CopyFrom(*alterData->GetDatabaseQuotas()); + } + if (alterData->GetAuditSettings()) { + event->Record.MutableAuditSettings()->CopyFrom(*alterData->GetAuditSettings()); + } + if (alterData->GetServerlessComputeResourcesMode()) { + event->Record.SetServerlessComputeResourcesMode(*alterData->GetServerlessComputeResourcesMode()); + } + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "Send configure request to schemeshard: " << tabletID << + " opId: " << OperationId << + " schemeshard: " << ssId << + " msg: " << event->Record.ShortDebugString()); + + shard.Operation = TTxState::ConfigureParts; + context.OnComplete.BindMsgToPipe(OperationId, tabletID, idx, event); + break; + } + case ETabletType::GraphShard: { + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "Send configure request to graph shard: " << tabletID << + " opId: " << OperationId << + " schemeshard: " << ssId); + shard.Operation = TTxState::ConfigureParts; + auto event = new TEvSubDomain::TEvConfigure(processing); + context.OnComplete.BindMsgToPipe(OperationId, tabletID, idx, event); + break; + } + case ETabletType::BackupController: { + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "Send configure request to backup controller tablet: " << tabletID << + " opId: " << OperationId << + " schemeshard: " << ssId); + shard.Operation = TTxState::ConfigureParts; + auto event = new TEvSubDomain::TEvConfigure(processing); + context.OnComplete.BindMsgToPipe(OperationId, tabletID, idx, event); + break; + } + default: + Y_FAIL_S("Unexpected type, we don't create tablets with type " << ETabletType::TypeToStr(type)); + } + } + + txState->UpdateShardsInProgress(TTxState::ConfigureParts); + return false; +} + + +// NSubDomainState::TPropose +// +TPropose::TPropose(TOperationId id) + : OperationId(id) +{ + IgnoreMessages(DebugHint(), { + TEvHive::TEvCreateTabletReply::EventType, + TEvSubDomain::TEvConfigureStatus::EventType, + TEvPrivate::TEvCompleteBarrier::EventType, + }); +} + +bool TPropose::HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOperationContext& context) { + TStepId step = TStepId(ev->Get()->StepId); + TTabletId ssId = context.SS->SelfTabletId(); + + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "NSubDomainState::TPropose HandleReply TEvOperationPlan" + << ", operationId " << OperationId + << ", at tablet " << ssId); + + TTxState* txState = context.SS->FindTx(OperationId); + if (!txState) { + return false; + } + Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateSubDomain + || txState->TxType == TTxState::TxAlterSubDomain + || txState->TxType == TTxState::TxCreateExtSubDomain + || txState->TxType == TTxState::TxAlterExtSubDomain + || txState->TxType == TTxState::TxAlterExtSubDomainCreateHive + ); + + TPathId pathId = txState->TargetPathId; + Y_ABORT_UNLESS(context.SS->PathsById.contains(pathId)); + TPathElement::TPtr path = context.SS->PathsById.at(pathId); + + NIceDb::TNiceDb db(context.GetDB()); + + if (path->StepCreated == InvalidStepId) { + path->StepCreated = step; + context.SS->PersistCreateStep(db, pathId, step); + } + + Y_ABORT_UNLESS(context.SS->SubDomains.contains(pathId)); + auto subDomain = context.SS->SubDomains.at(pathId); + auto alter = subDomain->GetAlter(); + Y_ABORT_UNLESS(alter); + Y_VERIFY_S(subDomain->GetVersion() < alter->GetVersion(), "" << subDomain->GetVersion() << " and " << alter->GetVersion()); + + subDomain->ActualizeAlterData(context.SS->ShardInfos, context.Ctx.Now(), + /* isExternal */ path->PathType == TPathElement::EPathType::EPathTypeExtSubDomain, + context.SS); + + context.SS->SubDomains[pathId] = alter; + context.SS->PersistSubDomain(db, pathId, *alter); + context.SS->PersistSubDomainSchemeQuotas(db, pathId, *alter); + + if (txState->TxType == TTxState::TxCreateSubDomain || txState->TxType == TTxState::TxCreateExtSubDomain) { + auto parentDir = context.SS->PathsById.at(path->ParentPathId); + ++parentDir->DirAlterVersion; + context.SS->PersistPathDirAlterVersion(db, parentDir); + context.SS->ClearDescribePathCaches(parentDir); + context.OnComplete.PublishToSchemeBoard(OperationId, parentDir->PathId); + } + + context.OnComplete.UpdateTenant(pathId); + context.SS->ClearDescribePathCaches(path); + context.OnComplete.PublishToSchemeBoard(OperationId, pathId); + + if (txState->NeedSyncHive) { + context.SS->ChangeTxState(db, OperationId, TTxState::SyncHive); + } else { + context.SS->ChangeTxState(db, OperationId, TTxState::Done); + } + + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "NSubDomainState::TPropose HandleReply TEvOperationPlan" + << ", operationId " << OperationId + << ", at tablet " << ssId); + + return true; +} + +bool TPropose::ProgressState(TOperationContext& context) { + TTabletId ssId = context.SS->SelfTabletId(); + + LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "NSubDomainState::TPropose ProgressState" + << ", operationId: " << OperationId + << ", at schemeshard: " << ssId); + + TTxState* txState = context.SS->FindTx(OperationId); + Y_ABORT_UNLESS(txState); + Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateSubDomain + || txState->TxType == TTxState::TxAlterSubDomain + || txState->TxType == TTxState::TxCreateExtSubDomain + || txState->TxType == TTxState::TxAlterExtSubDomain + || txState->TxType == TTxState::TxAlterExtSubDomainCreateHive + ); + + context.OnComplete.ProposeToCoordinator(OperationId, txState->TargetPathId, TStepId(0)); + + LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, + "NSubDomainState::TPropose ProgressState leave" + << ", operationId " << OperationId + << ", at tablet " << ssId); + + return false; +} + +} // namespace NKikimr::NSchemeShard::NSubDomainState diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_common_subdomain.h b/ydb/core/tx/schemeshard/schemeshard__operation_common_subdomain.h index a2d422c954b..6afcde46dca 100644 --- a/ydb/core/tx/schemeshard/schemeshard__operation_common_subdomain.h +++ b/ydb/core/tx/schemeshard/schemeshard__operation_common_subdomain.h @@ -3,8 +3,7 @@ #include "schemeshard_impl.h" #include "schemeshard__operation_part.h" -namespace NKikimr { -namespace NSchemeShard { +namespace NKikimr::NSchemeShard { inline bool CheckStoragePoolsInQuotas( const Ydb::Cms::DatabaseQuotas& quotas, @@ -60,274 +59,11 @@ private: << " operationId#" << OperationId; } public: - TConfigureParts(TOperationId id) - : OperationId(id) - { - IgnoreMessages(DebugHint(), {TEvHive::TEvCreateTabletReply::EventType}); - } - - bool HandleReply(TEvSchemeShard::TEvInitTenantSchemeShardResult::TPtr& ev, TOperationContext& context) override { - TTabletId ssId = context.SS->SelfTabletId(); - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() - << " HandleReply TEvInitTenantSchemeShardResult" - << " operationId: " << OperationId - << " at schemeshard: " << ssId); - - TTxState* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateSubDomain - || txState->TxType == TTxState::TxAlterSubDomain - || txState->TxType == TTxState::TxAlterExtSubDomain - || txState->TxType == TTxState::TxAlterExtSubDomainCreateHive - ); - - const auto& record = ev->Get()->Record; - - NIceDb::TNiceDb db(context.GetDB()); - - TTabletId tabletId = TTabletId(record.GetTenantSchemeShard()); - auto status = record.GetStatus(); - - auto shardIdx = context.SS->MustGetShardIdx(tabletId); - Y_ABORT_UNLESS(context.SS->ShardInfos.contains(shardIdx)); - - if (status != NKikimrScheme::EStatus::StatusSuccess && status != NKikimrScheme::EStatus::StatusAlreadyExists) { - LOG_CRIT_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() - << " Got error status on SubDomain Configure" - << "from tenant schemeshard tablet: " << tabletId - << " shard: " << shardIdx - << " status: " << NKikimrScheme::EStatus_Name(status) - << " opId: " << OperationId - << " schemeshard: " << ssId); - return false; - } - - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() - << " Got OK TEvInitTenantSchemeShardResult from schemeshard" - << " tablet: " << tabletId - << " shardIdx: " << shardIdx - << " at schemeshard: " << ssId); - - txState->ShardsInProgress.erase(shardIdx); - context.OnComplete.UnbindMsgFromPipe(OperationId, tabletId, shardIdx); - - if (txState->ShardsInProgress.empty()) { - // All tablets have replied so we can done this transaction - context.SS->ChangeTxState(db, OperationId, TTxState::Propose); - context.OnComplete.ActivateTx(OperationId); - return true; - } - - return false; - } - - bool HandleReply(TEvSubDomain::TEvConfigureStatus::TPtr& ev, TOperationContext& context) override { - TTabletId ssId = context.SS->SelfTabletId(); - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() - << " HandleReply TEvConfigureStatus" - << " operationId:" << OperationId - << " at schemeshard:" << ssId); - - TTxState* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateSubDomain - || txState->TxType == TTxState::TxAlterSubDomain - || txState->TxType == TTxState::TxAlterExtSubDomain - || txState->TxType == TTxState::TxAlterExtSubDomainCreateHive - ); - - const auto& record = ev->Get()->Record; - - NIceDb::TNiceDb db(context.GetDB()); - - TTabletId tabletId = TTabletId(record.GetOnTabletId()); - auto status = record.GetStatus(); - - auto shardIdx = context.SS->MustGetShardIdx(tabletId); - Y_ABORT_UNLESS(context.SS->ShardInfos.contains(shardIdx)); - - if (status == NKikimrTx::TEvSubDomainConfigurationAck::REJECT) { - LOG_CRIT_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() - << " Got REJECT on SubDomain Configure" - << "from tablet: " << tabletId - << " shard: " << shardIdx - << " opId: " << OperationId - << " schemeshard: " << ssId); - return false; - } - - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() << - " Got OK TEvConfigureStatus from " - << " tablet# " << tabletId - << " shardIdx# " << shardIdx - << " at schemeshard# " << ssId); - - txState->ShardsInProgress.erase(shardIdx); - context.OnComplete.UnbindMsgFromPipe(OperationId, tabletId, shardIdx); - - if (txState->ShardsInProgress.empty()) { - // All tablets have replied so we can done this transaction - context.SS->ChangeTxState(db, OperationId, TTxState::Propose); - context.OnComplete.ActivateTx(OperationId); - return true; - } - - return false; - } - - - bool ProgressState(TOperationContext& context) override { - TTabletId ssId = context.SS->SelfTabletId(); - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - DebugHint() - << " ProgressState" - << ", at schemeshard: " << ssId); - - TTxState* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateSubDomain - || txState->TxType == TTxState::TxAlterSubDomain - || txState->TxType == TTxState::TxAlterExtSubDomain - || txState->TxType == TTxState::TxAlterExtSubDomainCreateHive - ); - - txState->ClearShardsInProgress(); + TConfigureParts(TOperationId id); - if (txState->Shards.empty()) { - NIceDb::TNiceDb db(context.GetDB()); - context.SS->ChangeTxState(db, OperationId, TTxState::Propose); - context.OnComplete.ActivateTx(OperationId); - return true; - } - - auto pathId = txState->TargetPathId; - Y_ABORT_UNLESS(context.SS->PathsById.contains(pathId)); - TPath path = TPath::Init(pathId, context.SS); - - Y_ABORT_UNLESS(context.SS->SubDomains.contains(pathId)); - auto subDomain = context.SS->SubDomains.at(pathId); - auto alterData = subDomain->GetAlter(); - Y_ABORT_UNLESS(alterData); - alterData->Initialize(context.SS->ShardInfos); - auto processing = alterData->GetProcessingParams(); - auto storagePools = alterData->GetStoragePools(); - auto& schemeLimits = subDomain->GetSchemeLimits(); - - for (auto& shard : txState->Shards) { - if (shard.Operation != TTxState::CreateParts) { - continue; - } - TShardIdx idx = shard.Idx; - Y_ABORT_UNLESS(context.SS->ShardInfos.contains(idx)); - TTabletId tabletID = context.SS->ShardInfos[idx].TabletID; - auto type = context.SS->ShardInfos[idx].TabletType; - - switch (type) { - case ETabletType::Coordinator: - case ETabletType::Mediator: { - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "Send configure request to coordinator/mediator: " << tabletID << - " opId: " << OperationId << - " schemeshard: " << ssId); - shard.Operation = TTxState::ConfigureParts; - auto event = new TEvSubDomain::TEvConfigure(processing); - context.OnComplete.BindMsgToPipe(OperationId, tabletID, idx, event); - break; - } - case ETabletType::Hive: { - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "Send configure request to hive: " << tabletID << - " opId: " << OperationId << - " schemeshard: " << ssId); - shard.Operation = TTxState::ConfigureParts; - auto event = new TEvHive::TEvConfigureHive(TSubDomainKey(pathId.OwnerId, pathId.LocalPathId)); - context.OnComplete.BindMsgToPipe(OperationId, tabletID, idx, event); - break; - } - case ETabletType::SysViewProcessor: { - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "Send configure request to sys view processor: " << tabletID << - " opId: " << OperationId << - " schemeshard: " << ssId); - auto event = new NSysView::TEvSysView::TEvConfigureProcessor(path.PathString()); - shard.Operation = TTxState::ConfigureParts; - context.OnComplete.BindMsgToPipe(OperationId, tabletID, idx, event); - break; - } - case ETabletType::StatisticsAggregator: { - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "Send configure request to statistics aggregator: " << tabletID << - " opId: " << OperationId << - " schemeshard: " << ssId); - auto event = new NStat::TEvStatistics::TEvConfigureAggregator(path.PathString()); - shard.Operation = TTxState::ConfigureParts; - context.OnComplete.BindMsgToPipe(OperationId, tabletID, idx, event); - break; - } - case ETabletType::SchemeShard: { - auto event = new TEvSchemeShard::TEvInitTenantSchemeShard(ui64(ssId), - pathId.LocalPathId, path.PathString(), - path.Base()->Owner, path.GetEffectiveACL(), path.GetEffectiveACLVersion(), - processing, storagePools, - path.Base()->UserAttrs->Attrs, path.Base()->UserAttrs->AlterVersion, - schemeLimits, ui64(alterData->GetSharedHive()), alterData->GetResourcesDomainId() - ); - if (alterData->GetDeclaredSchemeQuotas()) { - event->Record.MutableDeclaredSchemeQuotas()->CopyFrom(*alterData->GetDeclaredSchemeQuotas()); - } - if (alterData->GetDatabaseQuotas()) { - event->Record.MutableDatabaseQuotas()->CopyFrom(*alterData->GetDatabaseQuotas()); - } - if (alterData->GetAuditSettings()) { - event->Record.MutableAuditSettings()->CopyFrom(*alterData->GetAuditSettings()); - } - if (alterData->GetServerlessComputeResourcesMode()) { - event->Record.SetServerlessComputeResourcesMode(*alterData->GetServerlessComputeResourcesMode()); - } - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "Send configure request to schemeshard: " << tabletID << - " opId: " << OperationId << - " schemeshard: " << ssId << - " msg: " << event->Record.ShortDebugString()); - - shard.Operation = TTxState::ConfigureParts; - context.OnComplete.BindMsgToPipe(OperationId, tabletID, idx, event); - break; - } - case ETabletType::GraphShard: { - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "Send configure request to graph shard: " << tabletID << - " opId: " << OperationId << - " schemeshard: " << ssId); - shard.Operation = TTxState::ConfigureParts; - auto event = new TEvSubDomain::TEvConfigure(processing); - context.OnComplete.BindMsgToPipe(OperationId, tabletID, idx, event); - break; - } - case ETabletType::BackupController: { - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "Send configure request to backup controller tablet: " << tabletID << - " opId: " << OperationId << - " schemeshard: " << ssId); - shard.Operation = TTxState::ConfigureParts; - auto event = new TEvSubDomain::TEvConfigure(processing); - context.OnComplete.BindMsgToPipe(OperationId, tabletID, idx, event); - break; - } - default: - Y_FAIL_S("Unexpected type, we don't create tablets with type " << ETabletType::TypeToStr(type)); - } - } - - txState->UpdateShardsInProgress(TTxState::ConfigureParts); - return false; - } + bool ProgressState(TOperationContext& context) override; + bool HandleReply(TEvSchemeShard::TEvInitTenantSchemeShardResult::TPtr& ev, TOperationContext& context) override; + bool HandleReply(TEvSubDomain::TEvConfigureStatus::TPtr& ev, TOperationContext& context) override; }; class TPropose: public TSubOperationState { @@ -341,116 +77,12 @@ private: } public: - TPropose(TOperationId id) - : OperationId(id) - { - IgnoreMessages(DebugHint(), { - TEvHive::TEvCreateTabletReply::EventType, - TEvSubDomain::TEvConfigureStatus::EventType, - TEvPrivate::TEvCompleteBarrier::EventType, - }); - } - - bool HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOperationContext& context) override { - TStepId step = TStepId(ev->Get()->StepId); - TTabletId ssId = context.SS->SelfTabletId(); - - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "NSubDomainState::TPropose HandleReply TEvOperationPlan" - << ", operationId " << OperationId - << ", at tablet " << ssId); - - TTxState* txState = context.SS->FindTx(OperationId); - if (!txState) { - return false; - } - Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateSubDomain - || txState->TxType == TTxState::TxAlterSubDomain - || txState->TxType == TTxState::TxCreateExtSubDomain - || txState->TxType == TTxState::TxAlterExtSubDomain - || txState->TxType == TTxState::TxAlterExtSubDomainCreateHive - ); - - TPathId pathId = txState->TargetPathId; - Y_ABORT_UNLESS(context.SS->PathsById.contains(pathId)); - TPathElement::TPtr path = context.SS->PathsById.at(pathId); - - NIceDb::TNiceDb db(context.GetDB()); - - if (path->StepCreated == InvalidStepId) { - path->StepCreated = step; - context.SS->PersistCreateStep(db, pathId, step); - } + TPropose(TOperationId id); - Y_ABORT_UNLESS(context.SS->SubDomains.contains(pathId)); - auto subDomain = context.SS->SubDomains.at(pathId); - auto alter = subDomain->GetAlter(); - Y_ABORT_UNLESS(alter); - Y_VERIFY_S(subDomain->GetVersion() < alter->GetVersion(), "" << subDomain->GetVersion() << " and " << alter->GetVersion()); - - subDomain->ActualizeAlterData(context.SS->ShardInfos, context.Ctx.Now(), - /* isExternal */ path->PathType == TPathElement::EPathType::EPathTypeExtSubDomain, - context.SS); - - context.SS->SubDomains[pathId] = alter; - context.SS->PersistSubDomain(db, pathId, *alter); - context.SS->PersistSubDomainSchemeQuotas(db, pathId, *alter); - - if (txState->TxType == TTxState::TxCreateSubDomain || txState->TxType == TTxState::TxCreateExtSubDomain) { - auto parentDir = context.SS->PathsById.at(path->ParentPathId); - ++parentDir->DirAlterVersion; - context.SS->PersistPathDirAlterVersion(db, parentDir); - context.SS->ClearDescribePathCaches(parentDir); - context.OnComplete.PublishToSchemeBoard(OperationId, parentDir->PathId); - } - - context.OnComplete.UpdateTenant(pathId); - context.SS->ClearDescribePathCaches(path); - context.OnComplete.PublishToSchemeBoard(OperationId, pathId); - - if (txState->NeedSyncHive) { - context.SS->ChangeTxState(db, OperationId, TTxState::SyncHive); - } else { - context.SS->ChangeTxState(db, OperationId, TTxState::Done); - } - - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "NSubDomainState::TPropose HandleReply TEvOperationPlan" - << ", operationId " << OperationId - << ", at tablet " << ssId); - - return true; - } - - bool ProgressState(TOperationContext& context) override { - TTabletId ssId = context.SS->SelfTabletId(); - - LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "NSubDomainState::TPropose ProgressState" - << ", operationId: " << OperationId - << ", at schemeshard: " << ssId); - - TTxState* txState = context.SS->FindTx(OperationId); - Y_ABORT_UNLESS(txState); - Y_ABORT_UNLESS(txState->TxType == TTxState::TxCreateSubDomain - || txState->TxType == TTxState::TxAlterSubDomain - || txState->TxType == TTxState::TxCreateExtSubDomain - || txState->TxType == TTxState::TxAlterExtSubDomain - || txState->TxType == TTxState::TxAlterExtSubDomainCreateHive - ); - - context.OnComplete.ProposeToCoordinator(OperationId, txState->TargetPathId, TStepId(0)); - - LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, - "NSubDomainState::TPropose ProgressState leave" - << ", operationId " << OperationId - << ", at tablet " << ssId); - - return false; - } + bool ProgressState(TOperationContext& context) override; + bool HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOperationContext& context) override; }; } // namespace NSubDomainState -} // namespace NSchemeShard -} // namespace NKikimr +} // namespace NKikimr::NSchemeShard diff --git a/ydb/core/tx/schemeshard/ya.make b/ydb/core/tx/schemeshard/ya.make index 7e0a802b9f2..7c71e695124 100644 --- a/ydb/core/tx/schemeshard/ya.make +++ b/ydb/core/tx/schemeshard/ya.make @@ -103,12 +103,16 @@ SRCS( schemeshard__operation_blob_depot.cpp schemeshard__operation_cancel_tx.cpp schemeshard__operation_cansel_build_index.cpp - schemeshard__operation_common.cpp schemeshard__operation_common.h + schemeshard__operation_common.cpp + schemeshard__operation_common_pq.cpp + schemeshard__operation_common_bsv.cpp + schemeshard__operation_common_cdc_stream.cpp schemeshard__operation_common_external_data_source.cpp schemeshard__operation_common_external_table.cpp schemeshard__operation_common_resource_pool.cpp schemeshard__operation_common_subdomain.h + schemeshard__operation_common_subdomain.cpp schemeshard__operation_consistent_copy_tables.cpp schemeshard__operation_copy_sequence.cpp schemeshard__operation_copy_table.cpp |