Consistently change the owner of the table and its child objects

5 files changed, 101 insertions, 45 deletions

diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_modify_acl.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_modify_acl.cpp
index 9fd70b27f81..e8309894077 100644
--- a/ydb/core/tx/schemeshard/schemeshard__operation_modify_acl.cpp
+++ b/ydb/core/tx/schemeshard/schemeshard__operation_modify_acl.cpp
@@ -14,21 +14,21 @@ public:
         const TTabletId ssId = context.SS->SelfTabletId();
 
         const TString& parentPathStr = Transaction.GetWorkingDir();
-        const TString& name = Transaction.GetModifyACL().GetName();
+        const auto& op = Transaction.GetModifyACL();
+        const auto& name = op.GetName();
+        const auto& acl = op.GetDiffACL();
+        const auto& owner = op.GetNewOwner();
 
-        LOG_NOTICE_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD,
-                     "TModifyACL Propose"
-                         << ", path: " << parentPathStr << "/" << name
-                         << ", operationId: " << OperationId
-                         << ", at schemeshard: " << ssId);
+        LOG_NOTICE_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, "TModifyACL Propose"
+            << ", path: " << parentPathStr << "/" << name
+            << ", operationId: " << OperationId
+            << ", at schemeshard: " << ssId);
 
         auto result = MakeHolder<TProposeResponse>(NKikimrScheme::StatusSuccess, ui64(OperationId.GetTxId()), ui64(ssId));
 
-        const TString acl = Transaction.GetModifyACL().GetDiffACL();
-
-        TPath path = TPath::Resolve(parentPathStr, context.SS).Dive(name);
+        const auto path = TPath::Resolve(parentPathStr, context.SS).Dive(name);
         {
-            TPath::TChecker checks = path.Check();
+            const auto checks = path.Check();
             checks
                 .NotEmpty()
                 .NotUnderDomainUpgrade()
@@ -46,62 +46,72 @@ public:
         }
 
         TString errStr;
-
         if (!context.SS->CheckApplyIf(Transaction, errStr)) {
             result->SetError(NKikimrScheme::StatusPreconditionFailed, errStr);
             return result;
         }
 
-        NIceDb::TNiceDb db(context.GetDB());
+        THashSet<TPathId> subTree;
+        if (acl || (owner && path.Base()->IsTable())) {
+            subTree = context.SS->ListSubTree(path.Base()->PathId, context.Ctx);
+        }
 
-        const TString owner = Transaction.GetModifyACL().GetNewOwner();
+        THashSet<TPathId> affectedPaths;
+        NIceDb::TNiceDb db(context.GetDB());
 
-        if (!acl.empty()) {
+        if (acl) {
             ++path.Base()->ACLVersion;
             path.Base()->ApplyACL(acl);
             context.SS->PersistACL(db, path.Base());
 
-            auto subtree = context.SS->ListSubTree(path.Base()->PathId, context.Ctx);
-            for (const TPathId pathId : subtree) {
+            for (const auto& pathId : subTree) {
                 if (context.SS->PathsById.at(pathId)->IsMigrated()) {
                     continue;
                 }
                 context.OnComplete.PublishToSchemeBoard(OperationId, pathId);
             }
 
-            if (!path.Base()->IsPQGroup()) {
-                // YDBOPS-1328
-                // its better we do not republish parent but
-                // it is possible when we do not show ALC for children
-                TPath parent = path.Parent();
-                ++parent.Base()->DirAlterVersion;
-                context.SS->PersistPathDirAlterVersion(db, parent.Base());
-                context.SS->ClearDescribePathCaches(parent.Base());
-                context.OnComplete.PublishToSchemeBoard(OperationId, parent.Base()->PathId);
+            affectedPaths.insert(subTree.begin(), subTree.end());
+        }
+
+        if (owner) {
+            THashSet<TPathId> pathIds = {path.Base()->PathId};
+            if (path.Base()->IsTable()) {
+                pathIds = subTree;
             }
 
-            context.OnComplete.UpdateTenants(std::move(subtree));
-        }
+            for (const auto& pathId : pathIds) {
+                if (!context.SS->PathsById.contains(pathId)) {
+                    Y_VERIFY_DEBUG_S(false, "unreachable");
+                    continue;
+                }
+
+                auto pathEl = context.SS->PathsById.at(pathId);
 
-        if (!owner.empty()) {
-            path.Base()->Owner = owner;
-            context.SS->PersistOwner(db, path.Base());
+                pathEl->Owner = owner;
+                context.SS->PersistOwner(db, pathEl);
 
-            ++path.Base()->DirAlterVersion;
-            context.SS->PersistPathDirAlterVersion(db, path.Base());
+                ++pathEl->DirAlterVersion;
+                context.SS->PersistPathDirAlterVersion(db, pathEl);
+
+                context.SS->ClearDescribePathCaches(pathEl);
+                context.OnComplete.PublishToSchemeBoard(OperationId, pathId);
+            }
 
-            context.OnComplete.PublishToSchemeBoard(OperationId, path.Base()->PathId);
+            affectedPaths.insert(pathIds.begin(), pathIds.end());
+        }
 
-            TPath parent = path.Parent(); // we show owner in children listing, so we have to update it
+        if ((acl && !path.Base()->IsPQGroup()) || owner) {
+            const auto parent = path.Parent();
             ++parent.Base()->DirAlterVersion;
             context.SS->PersistPathDirAlterVersion(db, parent.Base());
             context.SS->ClearDescribePathCaches(parent.Base());
             context.OnComplete.PublishToSchemeBoard(OperationId, parent.Base()->PathId);
-
-            context.OnComplete.UpdateTenants({path.Base()->PathId});
         }
 
+        context.OnComplete.UpdateTenants(std::move(affectedPaths));
         context.OnComplete.DoneOperation(OperationId);
+
         return result;
     }
 
@@ -110,11 +120,11 @@ public:
     }
 
     void ProgressState(TOperationContext&) override {
-        Y_FAIL("no progress state for modify acl");
+        Y_FAIL("no ProgressState for TModifyACL");
     }
 
     void AbortUnsafe(TTxId, TOperationContext&) override {
-        Y_FAIL("no AbortUnsafe for modify acl");
+        Y_FAIL("no AbortUnsafe for TModifyACL");
     }
 };
 
diff --git a/ydb/core/tx/schemeshard/ut_cdc_stream.cpp b/ydb/core/tx/schemeshard/ut_cdc_stream.cpp
index 3622210f2d8..619399646da 100644
--- a/ydb/core/tx/schemeshard/ut_cdc_stream.cpp
+++ b/ydb/core/tx/schemeshard/ut_cdc_stream.cpp
@@ -828,6 +828,45 @@ Y_UNIT_TEST_SUITE(TCdcStreamTests) {
         Metering(false);
     }
 
+    Y_UNIT_TEST(ChangeOwner) {
+        TTestBasicRuntime runtime;
+        TTestEnv env(runtime, TTestEnvOptions().EnableProtoSourceIdInfo(true));
+        ui64 txId = 100;
+
+        TestCreateTable(runtime, ++txId, "/MyRoot", R"(
+            Name: "Table"
+            Columns { Name: "key" Type: "Uint64" }
+            Columns { Name: "value" Type: "Uint64" }
+            KeyColumnNames: ["key"]
+        )");
+        env.TestWaitNotification(runtime, txId);
+
+        TestCreateCdcStream(runtime, ++txId, "/MyRoot", R"(
+            TableName: "Table"
+            StreamDescription {
+              Name: "Stream"
+              Mode: ECdcStreamModeKeysOnly
+              Format: ECdcStreamFormatProto
+            }
+        )");
+        env.TestWaitNotification(runtime, txId);
+
+        for (const auto* path : {"Table", "Table/Stream", "Table/Stream/streamImpl"}) {
+            TestDescribeResult(DescribePrivatePath(runtime, Sprintf("/MyRoot/%s", path)), {
+                NLs::HasOwner("root@builtin"),
+            });
+        }
+
+        TestModifyACL(runtime, ++txId, "/MyRoot", "Table", "", "user@builtin");
+        env.TestWaitNotification(runtime, txId);
+
+        for (const auto* path : {"Table", "Table/Stream", "Table/Stream/streamImpl"}) {
+            TestDescribeResult(DescribePrivatePath(runtime, Sprintf("/MyRoot/%s", path)), {
+                NLs::HasOwner("user@builtin"),
+            });
+        }
+    }
+
 } // TCdcStreamTests
 
 Y_UNIT_TEST_SUITE(TCdcStreamWithInitialScanTests) {
diff --git a/ydb/core/tx/schemeshard/ut_helpers/ls_checks.cpp b/ydb/core/tx/schemeshard/ut_helpers/ls_checks.cpp
index 518649ac343..975b9ae1996 100644
--- a/ydb/core/tx/schemeshard/ut_helpers/ls_checks.cpp
+++ b/ydb/core/tx/schemeshard/ut_helpers/ls_checks.cpp
@@ -997,6 +997,12 @@ TCheckFunc HasColumnTableTtlSettingsTiering(const TString& tieringName) {
     };
 }
 
+TCheckFunc HasOwner(const TString& owner) {
+    return [=](const NKikimrScheme::TEvDescribeSchemeResult& record) {
+        UNIT_ASSERT_VALUES_EQUAL(record.GetPathDescription().GetSelf().GetOwner(), owner);
+    };
+}
+
 void CheckEffectiveRight(const NKikimrScheme::TEvDescribeSchemeResult& record, const TString& right, bool mustHave) {
     const auto& self = record.GetPathDescription().GetSelf();
     TSecurityObject src(self.GetOwner(), self.GetEffectiveACL(), false);
diff --git a/ydb/core/tx/schemeshard/ut_helpers/ls_checks.h b/ydb/core/tx/schemeshard/ut_helpers/ls_checks.h
index f4379b7c0ae..8043bd5fa16 100644
--- a/ydb/core/tx/schemeshard/ut_helpers/ls_checks.h
+++ b/ydb/core/tx/schemeshard/ut_helpers/ls_checks.h
@@ -130,6 +130,7 @@ namespace NLs {
     void NoBackupInFly(const NKikimrScheme::TEvDescribeSchemeResult& record);
     TCheckFunc BackupHistoryCount(ui64 count);
 
+    TCheckFunc HasOwner(const TString& owner);
     TCheckFunc HasEffectiveRight(const TString& right);
     TCheckFunc HasNotEffectiveRight(const TString& right);
 
diff --git a/ydb/core/tx/schemeshard/ut_subdomain.cpp b/ydb/core/tx/schemeshard/ut_subdomain.cpp
index f07c58c2dec..9b8dfae8853 100644
--- a/ydb/core/tx/schemeshard/ut_subdomain.cpp
+++ b/ydb/core/tx/schemeshard/ut_subdomain.cpp
@@ -2254,7 +2254,7 @@ Y_UNIT_TEST_SUITE(TSchemeShardSubDomainTest) {
             env.TestWaitNotification(runtime, txId - 1);
             TestDescribeResult(DescribePath(runtime, "/MyRoot/USER_0"),
                                {NLs::PathExist,
-                                NLs::PathVersionEqual(24),
+                                NLs::PathVersionEqual(23),
                                 NLs::DomainLimitsIs(lowLimits.MaxPaths, lowLimits.MaxShards),
                                 NLs::PathsInsideDomain(0),
                                 NLs::ShardsInsideDomain(2)});
@@ -2301,7 +2301,7 @@ Y_UNIT_TEST_SUITE(TSchemeShardSubDomainTest) {
             env.TestWaitNotification(runtime, txId - 1);
             TestDescribeResult(DescribePath(runtime, "/MyRoot/USER_0"),
                                {NLs::PathExist,
-                                NLs::PathVersionEqual(28),
+                                NLs::PathVersionEqual(27),
                                 NLs::DomainLimitsIs(lowLimits.MaxPaths, lowLimits.MaxShards),
                                 NLs::PathsInsideDomain(0),
                                 NLs::ShardsInsideDomain(2)});
@@ -2322,7 +2322,7 @@ Y_UNIT_TEST_SUITE(TSchemeShardSubDomainTest) {
 
             TestDescribeResult(DescribePath(runtime, "/MyRoot/USER_0"),
                                {NLs::PathExist,
-                                NLs::PathVersionEqual(28),
+                                NLs::PathVersionEqual(27),
                                 NLs::DomainLimitsIs(lowLimits.MaxPaths, lowLimits.MaxShards),
                                 NLs::PathsInsideDomain(0),
                                 NLs::ShardsInsideDomain(2)});
@@ -2339,7 +2339,7 @@ Y_UNIT_TEST_SUITE(TSchemeShardSubDomainTest) {
 
             TestDescribeResult(DescribePath(runtime, "/MyRoot/USER_0"),
                                {NLs::PathExist,
-                                NLs::PathVersionEqual(28),
+                                NLs::PathVersionEqual(27),
                                 NLs::DomainLimitsIs(lowLimits.MaxPaths, lowLimits.MaxShards),
                                 NLs::PathsInsideDomain(0),
                                 NLs::ShardsInsideDomain(2)});
@@ -2357,7 +2357,7 @@ Y_UNIT_TEST_SUITE(TSchemeShardSubDomainTest) {
 
             TestDescribeResult(DescribePath(runtime, "/MyRoot/USER_0"),
                                {NLs::PathExist,
-                                NLs::PathVersionEqual(30),
+                                NLs::PathVersionEqual(29),
                                 NLs::DomainLimitsIs(lowLimits.MaxPaths, lowLimits.MaxShards),
                                 NLs::PathsInsideDomain(1),
                                 NLs::ShardsInsideDomain(4)});
@@ -2374,7 +2374,7 @@ Y_UNIT_TEST_SUITE(TSchemeShardSubDomainTest) {
 
             TestDescribeResult(DescribePath(runtime, "/MyRoot/USER_0"),
                                {NLs::PathExist,
-                                NLs::PathVersionEqual(30),
+                                NLs::PathVersionEqual(29),
                                 NLs::DomainLimitsIs(lowLimits.MaxPaths, lowLimits.MaxShards),
                                 NLs::PathsInsideDomain(1),
                                 NLs::ShardsInsideDomain(5)});
@@ -2391,7 +2391,7 @@ Y_UNIT_TEST_SUITE(TSchemeShardSubDomainTest) {
 
             TestDescribeResult(DescribePath(runtime, "/MyRoot/USER_0"),
                                {NLs::PathExist,
-                                NLs::PathVersionEqual(30),
+                                NLs::PathVersionEqual(29),
                                 NLs::DomainLimitsIs(lowLimits.MaxPaths, lowLimits.MaxShards),
                                 NLs::PathsInsideDomain(1),
                                 NLs::ShardsInsideDomain(5)});