Skip to content

Commit 53df48f

Browse files
ivanmorozov333ivanmorozov333
authored
use ss tablet id for schemeshard and sender for tx_proxy (#8684)
1 parent 65e10f5 commit 53df48f

File tree

6 files changed

+43
-22
lines changed

6 files changed

+43
-22
lines changed

‎ydb/core/tx/columnshard/transactions/operators/long_tx_write.cpp

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,4 +48,9 @@ bool TLongTxTransactionOperator::DoParse(TColumnShard& /*owner*/, const TString&
4848
return true;
4949
}
5050

51+
void TLongTxTransactionOperator::DoSendReply(TColumnShard& owner, const TActorContext& ctx) {
52+
const auto& txInfo = GetTxInfo();
53+
ctx.Send(txInfo.Source, BuildProposeResultEvent(owner).release());
54+
}
55+
5156
}

‎ydb/core/tx/columnshard/transactions/operators/long_tx_write.h

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@ namespace NKikimr::NColumnShard {
2424
virtual void DoStartProposeOnComplete(TColumnShard& /*owner*/, const TActorContext& /*ctx*/) override {
2525

2626
}
27+
virtual void DoSendReply(TColumnShard& owner, const TActorContext& ctx) override;
28+
2729
virtual void DoFinishProposeOnExecute(TColumnShard& /*owner*/, NTabletFlatExecutor::TTransactionContext& /*txc*/) override {
2830
}
2931
virtual void DoFinishProposeOnComplete(TColumnShard& /*owner*/, const TActorContext& /*ctx*/) override {

‎ydb/core/tx/columnshard/transactions/operators/propose_tx.cpp

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -3,21 +3,36 @@
33
namespace NKikimr::NColumnShard {
44

55
void IProposeTxOperator::DoSendReply(TColumnShard& owner, const TActorContext& ctx) {
6+
if (owner.CurrentSchemeShardId) {
7+
AFL_VERIFY(owner.CurrentSchemeShardId);
8+
ctx.Send(MakePipePerNodeCacheID(false),
9+
new TEvPipeCache::TEvForward(BuildProposeResultEvent(owner).release(), (ui64)owner.CurrentSchemeShardId, true));
10+
} else {
11+
AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "scheme_shard_tablet_not_initialized")("source", GetTxInfo().Source);
12+
ctx.Send(GetTxInfo().Source, BuildProposeResultEvent(owner).release());
13+
}
14+
}
15+
16+
std::unique_ptr<NKikimr::TEvColumnShard::TEvProposeTransactionResult> IProposeTxOperator::BuildProposeResultEvent(const TColumnShard& owner) const {
617
const auto& txInfo = GetTxInfo();
7-
std::unique_ptr<TEvColumnShard::TEvProposeTransactionResult> evResult = std::make_unique<TEvColumnShard::TEvProposeTransactionResult>(
8-
owner.TabletID(), txInfo.TxKind, txInfo.TxId, GetProposeStartInfoVerified().GetStatus(), GetProposeStartInfoVerified().GetStatusMessage());
18+
std::unique_ptr<TEvColumnShard::TEvProposeTransactionResult> evResult =
19+
std::make_unique<TEvColumnShard::TEvProposeTransactionResult>(owner.TabletID(), txInfo.TxKind, txInfo.TxId,
20+
GetProposeStartInfoVerified().GetStatus(), GetProposeStartInfoVerified().GetStatusMessage());
921
if (IsFail()) {
1022
owner.Counters.GetTabletCounters()->IncCounter(COUNTER_PREPARE_ERROR);
11-
AFL_ERROR(NKikimrServices::TX_COLUMNSHARD)("message", GetProposeStartInfoVerified().GetStatusMessage())("tablet_id", owner.TabletID())("tx_id", txInfo.TxId);
23+
AFL_ERROR(NKikimrServices::TX_COLUMNSHARD)("message", GetProposeStartInfoVerified().GetStatusMessage())("tablet_id", owner.TabletID())(
24+
"tx_id", txInfo.TxId);
1225
} else {
1326
evResult->Record.SetMinStep(txInfo.MinStep);
1427
evResult->Record.SetMaxStep(txInfo.MaxStep);
1528
if (owner.ProcessingParams) {
1629
evResult->Record.MutableDomainCoordinators()->CopyFrom(owner.ProcessingParams->GetCoordinators());
1730
}
1831
owner.Counters.GetTabletCounters()->IncCounter(COUNTER_PREPARE_SUCCESS);
32+
AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("message", GetProposeStartInfoVerified().GetStatusMessage())("tablet_id", owner.TabletID())(
33+
"tx_id", txInfo.TxId);
1934
}
20-
ctx.Send(txInfo.Source, evResult.release());
35+
return evResult;
2136
}
2237

2338
}

‎ydb/core/tx/columnshard/transactions/operators/propose_tx.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ class IProposeTxOperator: public TTxController::ITransactionOperator {
1212
virtual bool DoCheckTxInfoForReply(const TFullTxInfo& originalTxInfo) const override {
1313
return GetTxInfo() == originalTxInfo;
1414
}
15+
std::unique_ptr<TEvColumnShard::TEvProposeTransactionResult> BuildProposeResultEvent(const TColumnShard& owner) const;
1516
virtual void DoSendReply(TColumnShard& owner, const TActorContext& ctx) override;
1617
virtual bool DoCheckAllowUpdate(const TFullTxInfo& currentTxInfo) const override {
1718
if (!currentTxInfo.SeqNo || !GetTxInfo().SeqNo) {

‎ydb/core/tx/columnshard/transactions/operators/schema.cpp

Lines changed: 11 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,17 @@ class TWaitEraseTablesTxSubscriber: public NSubscriber::ISubscriber {
4040
}
4141
};
4242

43-
NKikimr::NColumnShard::TTxController::TProposeResult TSchemaTransactionOperator::DoStartProposeOnExecute(TColumnShard& owner, NTabletFlatExecutor::TTransactionContext& txc) {
43+
TTxController::TProposeResult TSchemaTransactionOperator::DoStartProposeOnExecute(TColumnShard& owner, NTabletFlatExecutor::TTransactionContext& txc) {
44+
auto seqNo = SeqNoFromProto(SchemaTxBody.GetSeqNo());
45+
auto lastSeqNo = owner.LastSchemaSeqNo;
46+
47+
// Check if proposal is outdated
48+
if (seqNo < lastSeqNo) {
49+
auto errorMessage = TStringBuilder() << "Ignoring outdated schema tx proposal at tablet " << owner.TabletID() << " txId " << GetTxId()
50+
<< " ssId " << owner.CurrentSchemeShardId << " seqNo " << seqNo << " lastSeqNo " << lastSeqNo;
51+
return TProposeResult(NKikimrTxColumnShard::EResultStatus::SCHEMA_CHANGED, errorMessage);
52+
}
53+
4454
switch (SchemaTxBody.TxBody_case()) {
4555
case NKikimrTxColumnShard::TSchemaTxBody::kInitShard:
4656
{
@@ -67,21 +77,6 @@ NKikimr::NColumnShard::TTxController::TProposeResult TSchemaTransactionOperator:
6777
break;
6878
}
6979

70-
auto seqNo = SeqNoFromProto(SchemaTxBody.GetSeqNo());
71-
auto lastSeqNo = owner.LastSchemaSeqNo;
72-
73-
// Check if proposal is outdated
74-
if (seqNo < lastSeqNo) {
75-
auto errorMessage = TStringBuilder()
76-
<< "Ignoring outdated schema tx proposal at tablet "
77-
<< owner.TabletID()
78-
<< " txId " << GetTxId()
79-
<< " ssId " << owner.CurrentSchemeShardId
80-
<< " seqNo " << seqNo
81-
<< " lastSeqNo " << lastSeqNo;
82-
return TProposeResult(NKikimrTxColumnShard::EResultStatus::SCHEMA_CHANGED, errorMessage);
83-
}
84-
8580
owner.UpdateSchemaSeqNo(seqNo, txc);
8681
return TProposeResult();
8782
}

‎ydb/core/tx/columnshard/transactions/tx_controller.cpp

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -278,7 +278,10 @@ TDuration TTxController::GetTxCompleteLag(ui64 timecastStep) const {
278278
TTxController::EPlanResult TTxController::PlanTx(const ui64 planStep, const ui64 txId, NTabletFlatExecutor::TTransactionContext& txc) {
279279
auto it = Operators.find(txId);
280280
if (it == Operators.end()) {
281+
AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "skip_plan_tx")("tx_id", txId);
281282
return EPlanResult::Skipped;
283+
} else {
284+
AFL_TRACE(NKikimrServices::TX_COLUMNSHARD)("event", "plan_tx")("tx_id", txId)("plan_step", it->second->MutableTxInfo().PlanStep);
282285
}
283286
auto& txInfo = it->second->MutableTxInfo();
284287
if (txInfo.PlanStep == 0) {
@@ -304,8 +307,8 @@ void TTxController::OnTabletInit() {
304307

305308
std::shared_ptr<TTxController::ITransactionOperator> TTxController::StartProposeOnExecute(
306309
const TTxController::TTxInfo& txInfo, const TString& txBody, NTabletFlatExecutor::TTransactionContext& txc) {
307-
NActors::TLogContextGuard lGuard = NActors::TLogContextBuilder::Build()("method", "TTxController::StartProposeOnExecute")(
308-
"tx_info", txInfo.DebugString())("tx_info", txInfo.DebugString());
310+
NActors::TLogContextGuard lGuard =
311+
NActors::TLogContextBuilder::Build()("method", "TTxController::StartProposeOnExecute")("tx_info", txInfo.DebugString());
309312
AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "start");
310313
std::shared_ptr<TTxController::ITransactionOperator> txOperator(
311314
TTxController::ITransactionOperator::TFactory::Construct(txInfo.TxKind, txInfo));

0 commit comments

Comments
 (0)