Used Mem changes and Db changes + removed micro framework

Author: GrigoriyPA

ydb/core/tx/schemeshard/schemeshard__operation_alter_streaming_query.cpp

@@ -1,7 +1,10 @@
 #include "schemeshard__operation_common.h"
-#include "schemeshard__operation_common_streaming_query.h"
 #include "schemeshard_impl.h"
 
+#define LOG_I(stream) LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, "[" << context.SS->TabletID() << "] " << stream)
+#define LOG_N(stream) LOG_NOTICE_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, "[" << context.SS->TabletID() << "] " << stream)
+#define RETURN_RESULT_UNLESS(x) if (!(x)) return result;
+
 namespace NKikimr::NSchemeShard {
 
 namespace NStreamingQuery {
@@ -24,13 +27,9 @@ class TPropose : public TSubOperationState {
 
         const TPathId& pathId = txState->TargetPathId;
         const TPath& path = TPath::Init(pathId, context.SS);
-        const TPathElement::TPtr pathPtr = context.SS->PathsById.at(pathId);
-
         NIceDb::TNiceDb db(context.GetDB());
 
         IncParentDirAlterVersionWithRepublish(OperationId, path, context);
-        context.SS->ClearDescribePathCaches(pathPtr);
-        context.OnComplete.PublishToSchemeBoard(OperationId, pathId);
 
         context.SS->ChangeTxState(db, OperationId, TTxState::Done);
         return true;
@@ -57,6 +56,8 @@ class TPropose : public TSubOperationState {
 };
 
 class TAlterStreamingQuery : public TSubOperation {
+    static constexpr ui64 MAX_PROTOBUF_SIZE = 2_MB;
+
     static TTxState::ETxState NextState() {
         return TTxState::Propose;
     }
@@ -83,9 +84,25 @@ class TAlterStreamingQuery : public TSubOperation {
         }
     }
 
+    static bool IsParentPathValid(const THolder<TProposeResponse>& result, const TPath& parentPath) {
+        const auto checks = parentPath.Check();
+        checks.NotUnderDomainUpgrade()
+            .IsAtLocalSchemeShard()
+            .IsResolved()
+            .NotDeleted()
+            .NotUnderDeleting()
+            .IsCommonSensePath()
+            .IsLikeDirectory();
+
+        if (!checks) {
+            result->SetError(checks.GetStatus(), checks.GetError());
+        }
+
+        return static_cast<bool>(checks);
+    }
+
     static bool IsDestinationPathValid(const THolder<TProposeResponse>& result, const TPath& dstPath) {
         const auto checks = dstPath.Check();
-
         checks.IsAtLocalSchemeShard()
             .IsResolved()
             .NotUnderDeleting()
@@ -103,13 +120,78 @@ class TAlterStreamingQuery : public TSubOperation {
         return static_cast<bool>(checks);
     }
 
-    TPathElement::TPtr ReplaceStreamingQueryPathElement(const TPath& dstPath) const {
+    bool IsApplyIfChecksPassed(const THolder<TProposeResponse>& result, const TOperationContext& context) const {
+        if (TString errorStr; !context.SS->CheckApplyIf(Transaction, errorStr)) {
+            result->SetError(NKikimrScheme::StatusPreconditionFailed, errorStr);
+            return false;
+        }
+
+        return true;
+    }
+
+    TStreamingQueryInfo::TPtr GetAlteredQueryInfo(const TPath& dstPath, const TOperationContext& context) const {
+        const auto& oldStreamingQueryInfo = context.SS->StreamingQueries.Value(dstPath->PathId, nullptr);
+        Y_ABORT_UNLESS(oldStreamingQueryInfo);
+        auto streamingQueryInfo = MakeIntrusive<TStreamingQueryInfo>(TStreamingQueryInfo{
+            .AlterVersion = oldStreamingQueryInfo->AlterVersion + 1,
+            .Properties = Transaction.GetCreateStreamingQuery().GetProperties(),
+        });
+
+        auto& properties = *streamingQueryInfo->Properties.MutableProperties();
+        for (const auto& [property, value] : oldStreamingQueryInfo->Properties.GetProperties()) {
+            properties.emplace(property, value);
+        }
+
+        return streamingQueryInfo;
+    }
+
+    bool IsDescriptionValid(const THolder<TProposeResponse>& result, TStreamingQueryInfo::TPtr queryInfo) const {
+        if (const ui64 propertiesSize = queryInfo->Properties.ByteSizeLong(); propertiesSize > MAX_PROTOBUF_SIZE) {
+            result->SetError(NKikimrScheme::StatusSchemeError, TStringBuilder() << "Maximum size of properties must be less or equal equal to " << MAX_PROTOBUF_SIZE << " but got " << propertiesSize << " after alter");
+            return false;
+        }
+
+        return true;
+    }
+
+    void PersistAlterStreamingQuery(const TPath& dstPath, const TOperationContext& context) const {
+        const TPathId& pathId = dstPath.Base()->PathId;
+
+        context.MemChanges.GrabPath(context.SS, dstPath->ParentPathId);
+        context.MemChanges.GrabStreamingQuery(context.SS, pathId);
+        context.MemChanges.GrabNewTxState(context.SS, OperationId);
+
+        context.DbChanges.PersistPath(pathId);
+        context.DbChanges.PersistStreamingQuery(pathId);
+        context.DbChanges.PersistTxState(OperationId);
+    }
+
+    void CreateTransaction(const TPath& dstPath, const TOperationContext& context) const {
+        Y_ABORT_UNLESS(!context.SS->FindTx(OperationId));
+
+        TTxState& txState = context.SS->CreateTx(OperationId, TTxState::TxAlterStreamingQuery, dstPath.Base()->PathId);
+        txState.Shards.clear();
+        txState.State = TTxState::Propose;
+        txState.MinStep = TStepId(1);
+        context.OnComplete.ActivateTx(OperationId);
+
+        if (const auto parent = dstPath.Parent().Base(); parent->HasActiveChanges()) {
+            const TTxId parentTxId = parent->PlannedToCreate() ? parent->CreateTxId : parent->LastTxId;
+            context.OnComplete.Dependence(parentTxId, OperationId.GetTxId());
+        }
+    }
+
+    void AlterStreamingQueryPathElement(const TPath& dstPath, TStreamingQueryInfo::TPtr queryInfo, const TOperationContext& context) const {
         TPathElement::TPtr streamingQuery = dstPath.Base();
 
         streamingQuery->PathState = TPathElement::EPathState::EPathStateAlter;
         streamingQuery->LastTxId  = OperationId.GetTxId();
 
-        return streamingQuery;
+        if (const auto& acl = Transaction.GetModifyACL().GetDiffACL()) {
+            streamingQuery->ApplyACL(acl);
+        }
+
+        context.SS->StreamingQueries[dstPath.Base()->PathId] = queryInfo;
     }
 
 public:
@@ -128,36 +210,27 @@ class TAlterStreamingQuery : public TSubOperation {
                                                    static_cast<ui64>(context.SS->SelfTabletId()));
 
         const TPath& parentPath = TPath::Resolve(parentPathStr, context.SS);
-        RETURN_RESULT_UNLESS(IsParentPathValid(result, parentPath, /* isCreate */ false));
+        RETURN_RESULT_UNLESS(IsParentPathValid(result, parentPath));
 
         const TPath& dstPath = parentPath.Child(name);
         RETURN_RESULT_UNLESS(IsDestinationPathValid(result, dstPath));
-        RETURN_RESULT_UNLESS(IsApplyIfChecksPassed(result, Transaction, context));
-        RETURN_RESULT_UNLESS(IsDescriptionValid(result, streamingQueryDescription));
-
-        const auto& oldStreamingQueryInfo = context.SS->StreamingQueries.Value(dstPath->PathId, nullptr);
-        Y_ABORT_UNLESS(oldStreamingQueryInfo);
-        const auto streamingQueryInfo = CreateModifyStreamingQuery(streamingQueryDescription, oldStreamingQueryInfo);
-        Y_ABORT_UNLESS(streamingQueryInfo);
+        RETURN_RESULT_UNLESS(IsApplyIfChecksPassed(result, context));
+        const auto queryInfo = GetAlteredQueryInfo(dstPath, context);
+        RETURN_RESULT_UNLESS(IsDescriptionValid(result, queryInfo));
 
         result->SetPathId(dstPath.Base()->PathId.LocalPathId);
-        const TPathElement::TPtr streamingQuery = ReplaceStreamingQueryPathElement(dstPath);
-        CreateTransaction(OperationId, context, streamingQuery->PathId, TTxState::TxAlterStreamingQuery);
-        RegisterParentPathDependencies(OperationId, context, parentPath);
-
-        NIceDb::TNiceDb db(context.GetDB());
-        AdvanceTransactionStateToPropose(OperationId, context, db);
-        PersistStreamingQuery(OperationId, context, db, streamingQuery, streamingQueryInfo, /* acl */ TString());
 
-        IncParentDirAlterVersionWithRepublishSafeWithUndo(OperationId, dstPath, context.SS, context.OnComplete);
+        const auto guard = context.DbGuard();
+        PersistAlterStreamingQuery(dstPath, context);
+        CreateTransaction(dstPath, context);
+        AlterStreamingQueryPathElement(dstPath, queryInfo, context);
 
         SetState(NextState());
         return result;
     }
 
     void AbortPropose(TOperationContext& context) override {
         LOG_N("TAlterStreamingQuery AbortPropose: opId# " << OperationId);
-        Y_ABORT("no AbortPropose for TAlterStreamingQuery");
     }
 
     void AbortUnsafe(TTxId forceDropTxId, TOperationContext& context) override {