Merge 69c4eac into b76c6f8

aavdonkin · web-flow · commit ac73a0555240 · 2024-11-21T12:37:02.000Z
diff --git a/ydb/core/protos/counters_columnshard.proto b/ydb/core/protos/counters_columnshard.proto
@@ -203,4 +203,6 @@ enum ETxTypes {
     TXTYPE_START_INTERNAL_SCAN = 36                         [(TxTypeOpts) = {Name: "TxStartInternalScan"}];
     TXTYPE_DATA_SHARING_START_SOURCE_CURSOR = 37            [(TxTypeOpts) = {Name: "TxDataSharingStartSourceCursor"}];
     TXTYPE_ASK_PORTION_METADATA = 38                        [(TxTypeOpts) = {Name: "TxAskPortionMetadata"}];
+    TXTYPE_PERSIST_SUBDOMAIN_OUT_OF_SPACE = 39              [(TxTypeOpts) = {Name: "TxPersistSubDomainOutOfSpace"}];
+    TXTYPE_PERSIST_SUBDOMAIN_PATH_ID = 40                   [(TxTypeOpts) = {Name: "TxPersistSubDomainPathId"}];
 }
diff --git a/ydb/core/protos/tx_columnshard.proto b/ydb/core/protos/tx_columnshard.proto
@@ -162,6 +162,7 @@ message TEvProposeTransaction {
     optional NKikimrSubDomains.TProcessingParams ProcessingParams = 6;
     optional uint64 Flags = 7;
     optional NKikimrTx.TMessageSeqNo SeqNo = 8;
+    optional uint64 SubDomainPathId = 9;
 }
 
 message TEvCheckPlannedTransaction {
diff --git a/ydb/core/tx/columnshard/columnshard__init.cpp b/ydb/core/tx/columnshard/columnshard__init.cpp
@@ -107,6 +107,11 @@ void TTxInit::Complete(const TActorContext& ctx) {
     Self->ProgressTxController->OnTabletInit();
     Self->SwitchToWork(ctx);
     NYDBTest::TControllers::GetColumnShardController()->OnTabletInitCompleted(*Self);
+    if (Self->SubDomainPathId) {
+        Self->StartWatchingSubDomainPathId();
+    } else {
+        Self->StartFindSubDomainPathId();
+    }
 }
 
 class TTxUpdateSchema: public TTransactionBase<TColumnShard> {
diff --git a/ydb/core/tx/columnshard/columnshard__propose_transaction.cpp b/ydb/core/tx/columnshard/columnshard__propose_transaction.cpp
@@ -55,6 +55,14 @@ class TTxProposeTransaction: public NTabletFlatExecutor::TTransactionBase<TColum
                 AFL_VERIFY(Self->CurrentSchemeShardId == record.GetSchemeShardId());
             }
         }
+        if (txKind == NKikimrTxColumnShard::TX_KIND_SCHEMA) {
+            ui64 subDomainPathId = record.GetSubDomainPathId();
+            AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "propose")("subdomain_id", subDomainPathId);
+            if (subDomainPathId) {
+                Self->PersistSubDomainPathId(record.GetSchemeShardId(), subDomainPathId, txc);
+                Self->StartWatchingSubDomainPathId();
+            }
+        }
         std::optional<TMessageSeqNo> msgSeqNo;
         if (Ev->Get()->Record.HasSeqNo()) {
             TMessageSeqNo seqNo;
diff --git a/ydb/core/tx/columnshard/columnshard__quota.cpp b/ydb/core/tx/columnshard/columnshard__quota.cpp
@@ -0,0 +1,131 @@
+#include "columnshard_impl.h"
+
+namespace NKikimr::NColumnShard {
+
+class TTxPersistSubDomainOutOfSpace : public NTabletFlatExecutor::TTransactionBase<TColumnShard> {
+public:
+    TTxPersistSubDomainOutOfSpace(TColumnShard* self, bool outOfSpace)
+        : TTransactionBase(self)
+        , OutOfSpace(outOfSpace)
+    { }
+
+    TTxType GetTxType() const override { return TXTYPE_PERSIST_SUBDOMAIN_OUT_OF_SPACE; }
+
+    bool Execute(TTransactionContext& txc, const TActorContext&) override {
+        NIceDb::TNiceDb db(txc.DB);
+
+        if (Self->SubDomainOutOfSpace != OutOfSpace) {
+            Schema::SaveSpecialValue(db, Schema::EValueIds::SubDomainOutOfSpace, ui64(OutOfSpace ? 1 : 0));
+            Self->SubDomainOutOfSpace = OutOfSpace;
+        }
+
+        return true;
+    }
+
+    void Complete(const TActorContext&) override {
+        // nothing
+    }
+
+private:
+    const bool OutOfSpace = false;
+};
+
+class TTxPersistSubDomainPathId : public NTabletFlatExecutor::TTransactionBase<TColumnShard> {
+public:
+    TTxPersistSubDomainPathId(TColumnShard* self, ui64 schemeShardId, ui64 localPathId)
+        : TTransactionBase(self)
+        , SchemeShardId(schemeShardId)
+        , LocalPathId(localPathId)
+    { }
+
+    TTxType GetTxType() const override { return TXTYPE_PERSIST_SUBDOMAIN_PATH_ID; }
+
+    bool Execute(TTransactionContext& txc, const TActorContext&) override {
+        if (Self->CurrentSchemeShardId == SchemeShardId &&
+            !Self->SubDomainPathId || Self->SubDomainPathId->OwnerId != SchemeShardId)
+        {
+            Self->PersistSubDomainPathId(SchemeShardId, LocalPathId, txc);
+            Self->StartWatchingSubDomainPathId();
+        }
+        return true;
+    }
+
+    void Complete(const TActorContext&) override {
+        // nothing
+    }
+
+private:
+    const ui64 SchemeShardId;
+    const ui64 LocalPathId;
+};
+
+void TColumnShard::PersistSubDomainPathId(ui64 ownerId, ui64 localPathId,
+                                               NTabletFlatExecutor::TTransactionContext &txc) {
+    SubDomainPathId.emplace(ownerId, localPathId);
+    NIceDb::TNiceDb db(txc.DB);
+    Schema::SaveSpecialValue(db, Schema::EValueIds::SubDomainOwnerId, ownerId);
+    Schema::SaveSpecialValue(db, Schema::EValueIds::SubDomainLocalPathId, localPathId);
+}
+
+void TColumnShard::StopWatchingSubDomainPathId() {
+    if (WatchingSubDomainPathId) {
+        Send(MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvWatchRemove());
+        WatchingSubDomainPathId.reset();
+    }
+}
+
+void TColumnShard::StartWatchingSubDomainPathId() {
+    if (!SubDomainPathId || SubDomainPathId->OwnerId != CurrentSchemeShardId) {
+        return;
+    }
+
+    if (WatchingSubDomainPathId && *WatchingSubDomainPathId != *SubDomainPathId) {
+        StopWatchingSubDomainPathId();
+    }
+
+    if (!WatchingSubDomainPathId) {
+        AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("started_watching_subdomain", *SubDomainPathId);
+        Send(MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvWatchPathId(*SubDomainPathId));
+        WatchingSubDomainPathId = *SubDomainPathId;
+    }
+}
+
+void TColumnShard::Handle(TEvTxProxySchemeCache::TEvWatchNotifyUpdated::TPtr& ev, const TActorContext& ctx) {
+    const auto* msg = ev->Get();
+    AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("notify_subdomain", msg->PathId);
+    if (SubDomainPathId && msg->PathId == *SubDomainPathId) {
+        const bool outOfSpace = msg->Result->GetPathDescription()
+            .GetDomainDescription()
+            .GetDomainState()
+            .GetDiskQuotaExceeded();
+
+        LOG_DEBUG_S(ctx, NKikimrServices::TX_COLUMNSHARD,
+            "Discovered subdomain " << msg->PathId << " state, outOfSpace = " << outOfSpace
+            << " at columnshard " << TabletID());
+
+        Execute(new TTxPersistSubDomainOutOfSpace(this, outOfSpace), ctx);
+    }
+}
+
+static constexpr TDuration MaxFindSubDomainPathIdDelay = TDuration::Minutes(10);
+
+void TColumnShard::StartFindSubDomainPathId(bool delayFirstRequest) {
+    if (!FindSubDomainPathIdActor &&
+        CurrentSchemeShardId != 0 &&
+        (!SubDomainPathId || SubDomainPathId->OwnerId != CurrentSchemeShardId))
+    {
+        FindSubDomainPathIdActor = Register(CreateFindSubDomainPathIdActor(SelfId(), TabletID(), CurrentSchemeShardId, delayFirstRequest, MaxFindSubDomainPathIdDelay));
+    }
+}
+
+
+void TColumnShard::Handle(NSchemeShard::TEvSchemeShard::TEvSubDomainPathIdFound::TPtr& ev, const TActorContext& ctx) {
+    const auto* msg = ev->Get();
+    if (FindSubDomainPathIdActor == ev->Sender) {
+        FindSubDomainPathIdActor = { };
+    }
+    AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "subdomain_found")("scheme_shard_id", msg->SchemeShardId)("local_path_id", msg->LocalPathId);
+    Execute(new TTxPersistSubDomainPathId(this, msg->SchemeShardId, msg->LocalPathId), ctx);
+}
+
+}
diff --git a/ydb/core/tx/columnshard/columnshard__write.cpp b/ydb/core/tx/columnshard/columnshard__write.cpp
@@ -203,14 +203,20 @@ void TColumnShard::Handle(TEvColumnShard::TEvWrite::TPtr& ev, const TActorContex
     writeMeta.SetLongTxId(NLongTxService::TLongTxId::FromProto(record.GetLongTxId()));
     writeMeta.SetWritePartId(record.GetWritePartId());
 
-    const auto returnFail = [&](const NColumnShard::ECumulativeCounters signalIndex, const EWriteFailReason reason) {
+    const auto returnFail = [&](const NColumnShard::ECumulativeCounters signalIndex, const EWriteFailReason reason, NKikimrTxColumnShard::EResultStatus resultStatus = NKikimrTxColumnShard::EResultStatus::ERROR) {
         Counters.GetTabletCounters()->IncCounter(signalIndex);
 
-        ctx.Send(source, std::make_unique<TEvColumnShard::TEvWriteResult>(TabletID(), writeMeta, NKikimrTxColumnShard::EResultStatus::ERROR));
+        ctx.Send(source, std::make_unique<TEvColumnShard::TEvWriteResult>(TabletID(), writeMeta, resultStatus));
         Counters.GetCSCounters().OnFailedWriteResponse(reason);
         return;
     };
 
+    if (SubDomainOutOfSpace && record.HasModificationType() && (record.GetModificationType() != NKikimrTxColumnShard::TEvWrite::OPERATION_DELETE)) {
+        AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "skip_writing")("reason", "quota_exceeded");
+        Counters.GetTabletCounters()->IncCounter(COUNTER_OUT_OF_SPACE);
+        return returnFail(COUNTER_WRITE_FAIL, EWriteFailReason::Overload, NKikimrTxColumnShard::EResultStatus::OVERLOADED);
+    }
+
     if (!AppDataVerified().ColumnShardConfig.GetWritingEnabled()) {
         AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "skip_writing")("reason", "disabled");
         return returnFail(COUNTER_WRITE_FAIL, EWriteFailReason::Disabled);
diff --git a/ydb/core/tx/columnshard/columnshard_impl.h b/ydb/core/tx/columnshard/columnshard_impl.h
@@ -174,6 +174,8 @@ class TColumnShard: public TActor<TColumnShard>, public NTabletFlatExecutor::TTa
     friend class TTxRemoveSharedBlobs;
     friend class TTxFinishAsyncTransaction;
     friend class TWaitEraseTablesTxSubscriber;
+    friend class TTxPersistSubDomainOutOfSpace;
+    friend class TTxPersistSubDomainPathId;
 
     friend class NOlap::TCleanupPortionsColumnEngineChanges;
     friend class NOlap::TCleanupTablesColumnEngineChanges;
@@ -282,8 +284,9 @@ class TColumnShard: public TActor<TColumnShard>, public NTabletFlatExecutor::TTa
     void Handle(NOlap::NDataSharing::NEvents::TEvFinishedFromSource::TPtr& ev, const TActorContext& ctx);
     void Handle(NOlap::NDataSharing::NEvents::TEvAckFinishToSource::TPtr& ev, const TActorContext& ctx);
     void Handle(NOlap::NDataSharing::NEvents::TEvAckFinishFromInitiator::TPtr& ev, const TActorContext& ctx);
-
     void Handle(NOlap::NDataAccessorControl::TEvAskTabletDataAccessors::TPtr& ev, const TActorContext& ctx);
+    void Handle(TEvTxProxySchemeCache::TEvWatchNotifyUpdated::TPtr& ev, const TActorContext& ctx);
+    void Handle(NSchemeShard::TEvSchemeShard::TEvSubDomainPathIdFound::TPtr& ev, const TActorContext&);
 
     ITransaction* CreateTxInitSchema();
 
@@ -362,6 +365,11 @@ class TColumnShard: public TActor<TColumnShard>, public NTabletFlatExecutor::TTa
         return true;
     }
 
+    void PersistSubDomainPathId(ui64 ownerId, ui64 localPathId, NTabletFlatExecutor::TTransactionContext &txc);
+    void StopWatchingSubDomainPathId();
+    void StartWatchingSubDomainPathId();
+    void StartFindSubDomainPathId(bool delayFirstRequest = true);
+
 private:
     void OverloadWriteFail(const EOverloadStatus overloadReason, const NEvWrite::TWriteMeta& writeMeta, const ui64 writeSize, const ui64 cookie,
         std::unique_ptr<NActors::IEventBase>&& event, const TActorContext& ctx);
@@ -445,6 +453,8 @@ class TColumnShard: public TActor<TColumnShard>, public NTabletFlatExecutor::TTa
             HFunc(NOlap::NDataSharing::NEvents::TEvAckFinishToSource, Handle);
             HFunc(NOlap::NDataSharing::NEvents::TEvAckFinishFromInitiator, Handle);
             HFunc(NOlap::NDataAccessorControl::TEvAskTabletDataAccessors, Handle);
+            HFunc(TEvTxProxySchemeCache::TEvWatchNotifyUpdated, Handle);
+            HFunc(NSchemeShard::TEvSchemeShard::TEvSubDomainPathIdFound, Handle);
 
             default:
                 if (!HandleDefaultEvents(ev, SelfId())) {
@@ -531,6 +541,11 @@ class TColumnShard: public TActor<TColumnShard>, public NTabletFlatExecutor::TTa
     TLimits Limits;
     NOlap::TNormalizationController NormalizerController;
     NDataShard::TSysLocks SysLocks;
+    TActorId FindSubDomainPathIdActor;
+    std::optional<TPathId> SubDomainPathId;
+    std::optional<TPathId> WatchingSubDomainPathId;
+    bool SubDomainOutOfSpace = false;
+
     static TDuration GetMaxReadStaleness();
 
     void TryRegisterMediatorTimeCast();
diff --git a/ydb/core/tx/columnshard/columnshard_schema.h b/ydb/core/tx/columnshard/columnshard_schema.h
@@ -84,7 +84,10 @@ struct Schema : NIceDb::Schema {
         LastCompletedTxId = 14,
         LastNormalizerSequentialId = 15,
         GCBarrierPreparationGen = 16,
-        GCBarrierPreparationStep = 17
+        GCBarrierPreparationStep = 17,
+        SubDomainOwnerId = 18,
+        SubDomainLocalPathId = 19,
+        SubDomainOutOfSpace = 20
     };
 
     enum class EInsertTableIds : ui8 {
diff --git a/ydb/core/tx/columnshard/loading/stages.cpp b/ydb/core/tx/columnshard/loading/stages.cpp
@@ -172,6 +172,25 @@ bool TSpecialValuesInitializer::DoExecute(NTabletFlatExecutor::TTransactionConte
         return false;
     }
 
+    std::optional<ui64> subDomainOwnerId;
+    std::optional<ui64> subDomainLocalPathId;
+    if (!Schema::GetSpecialValueOpt(db, Schema::EValueIds::SubDomainOwnerId, subDomainOwnerId)) {
+        return false;
+    }
+    if (!Schema::GetSpecialValueOpt(db, Schema::EValueIds::SubDomainLocalPathId, subDomainLocalPathId)) {
+        return false;
+    }
+
+    if (subDomainOwnerId.has_value() && subDomainLocalPathId.has_value()) {
+        Self->SubDomainPathId.emplace(*subDomainOwnerId, *subDomainLocalPathId);
+    }
+
+    ui64 outOfSpace = 0;
+    if (!Schema::GetSpecialValueOpt(db, Schema::EValueIds::SubDomainOutOfSpace, outOfSpace)) {
+        return false;
+    }
+    Self->SubDomainOutOfSpace = outOfSpace;
+
     {
         ui64 lastCompletedStep = 0;
         ui64 lastCompletedTx = 0;
diff --git a/ydb/core/tx/columnshard/ya.make b/ydb/core/tx/columnshard/ya.make
@@ -10,6 +10,7 @@ SRCS(
     columnshard__progress_tx.cpp
     columnshard__propose_cancel.cpp
     columnshard__propose_transaction.cpp
+    columnshard__quota.cpp
     columnshard__scan.cpp
     columnshard__statistics.cpp
     columnshard__write.cpp
diff --git a/ydb/core/tx/datashard/datashard__init.cpp b/ydb/core/tx/datashard/datashard__init.cpp
@@ -104,8 +104,10 @@ void TDataShard::TTxInit::Complete(const TActorContext &ctx) {
     // Find subdomain path id if needed
     if (Self->State == TShardState::Ready) {
         if (Self->SubDomainPathId) {
+            AFL_DEBUG(NKikimrServices::TX_DATASHARD)("event", "start_watching_subdomain");
             Self->StartWatchingSubDomainPathId();
         } else {
+            AFL_DEBUG(NKikimrServices::TX_DATASHARD)("event", "start_find_subdomain");
             Self->StartFindSubDomainPathId();
         }
     }
diff --git a/ydb/core/tx/datashard/datashard_subdomain_path_id.cpp b/ydb/core/tx/datashard/datashard_subdomain_path_id.cpp
@@ -56,6 +56,7 @@ class TDataShard::TTxPersistSubDomainPathId : public NTabletFlatExecutor::TTrans
 };
 
 void TDataShard::Handle(NSchemeShard::TEvSchemeShard::TEvSubDomainPathIdFound::TPtr& ev, const TActorContext& ctx) {
+    AFL_DEBUG(NKikimrServices::TX_DATASHARD)("event", "subdomain_found");
     const auto* msg = ev->Get();
 
     if (FindSubDomainPathIdActor == ev->Sender) {
diff --git a/ydb/core/tx/schemeshard/olap/operations/create_store.cpp b/ydb/core/tx/schemeshard/olap/operations/create_store.cpp
@@ -100,6 +100,9 @@ class TConfigureParts: public TSubOperationState {
                     ui64(OperationId.GetTxId()),
                     columnShardTxBody, seqNo,
                     context.SS->SelectProcessingParams(txState->TargetPathId));
+                if (const ui64 subDomainPathId = context.SS->ResolvePathIdForDomain(txState->TargetPathId).LocalPathId) {
+                     event->Record.SetSubDomainPathId(subDomainPathId);
+                }
 
                 context.OnComplete.BindMsgToPipe(OperationId, tabletId, shard.Idx, event.release());
             } else {
diff --git a/ydb/core/tx/schemeshard/olap/operations/create_table.cpp b/ydb/core/tx/schemeshard/olap/operations/create_table.cpp
@@ -317,7 +317,9 @@ class TConfigureParts: public TSubOperationState {
                     ui64(OperationId.GetTxId()),
                     columnShardTxBody, seqNo,
                     context.SS->SelectProcessingParams(txState->TargetPathId));
-
+                if (const ui64 subDomainPathId = context.SS->ResolvePathIdForDomain(txState->TargetPathId).LocalPathId) {
+                     event->Record.SetSubDomainPathId(subDomainPathId);
+                }
                 context.OnComplete.BindMsgToPipe(OperationId, tabletId, shard.Idx, event.release());
             } else {
                 LOG_ERROR_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, DebugHint() << " unexpected tablet type");
diff --git a/ydb/core/tx/schemeshard/schemeshard_info_types.cpp b/ydb/core/tx/schemeshard/schemeshard_info_types.cpp
@@ -158,6 +158,7 @@ bool TSubDomainInfo::CheckDiskSpaceQuotas(IQuotaCounters* counters) {
         // they are not accounted for when we make a decision to change the state of the subdomain.
         ui64 totalUsage = TotalDiskSpaceUsage();
 
+        LOG_DEBUG_S(TlsActivationContext->AsActorContext(), NKikimrServices::FLAT_TX_SCHEMESHARD, "Total usage: " << totalUsage << " soft quota: " << quotas.SoftQuota << " hard quota: " << quotas.HardQuota);
         // If a quota is equal to zero, then it sets no limit on the disk space usage.
         const bool isHardQuotaExceeded = quotas.HardQuota && totalUsage > quotas.HardQuota;
         const bool isTotalUsageBelowSoftQuota = !quotas.SoftQuota || totalUsage < quotas.SoftQuota;

Original file line number	Diff line number	Diff line change
`@@ -203,4 +203,6 @@ enum ETxTypes {`
`203`	`203`	`TXTYPE_START_INTERNAL_SCAN = 36 [(TxTypeOpts) = {Name: "TxStartInternalScan"}];`
`204`	`204`	`TXTYPE_DATA_SHARING_START_SOURCE_CURSOR = 37 [(TxTypeOpts) = {Name: "TxDataSharingStartSourceCursor"}];`
`205`	`205`	`TXTYPE_ASK_PORTION_METADATA = 38 [(TxTypeOpts) = {Name: "TxAskPortionMetadata"}];`
	`206`	`+ TXTYPE_PERSIST_SUBDOMAIN_OUT_OF_SPACE = 39 [(TxTypeOpts) = {Name: "TxPersistSubDomainOutOfSpace"}];`
	`207`	`+ TXTYPE_PERSIST_SUBDOMAIN_PATH_ID = 40 [(TxTypeOpts) = {Name: "TxPersistSubDomainPathId"}];`
`206`	`208`	`}`
Original file line number	Diff line number	Diff line change
`@@ -162,6 +162,7 @@ message TEvProposeTransaction {`
`162`	`162`	`optional NKikimrSubDomains.TProcessingParams ProcessingParams = 6;`
`163`	`163`	`optional uint64 Flags = 7;`
`164`	`164`	`optional NKikimrTx.TMessageSeqNo SeqNo = 8;`
	`165`	`+ optional uint64 SubDomainPathId = 9;`
`165`	`166`	`}`
`166`	`167`
`167`	`168`	`message TEvCheckPlannedTransaction {`