loop filter apply

Author: kungasc

ydb/core/sys_view/auth/auth_scan_base.h

@@ -57,11 +57,11 @@ class TAuthScanBase : public TScanActorBase<TDerived> {
         , RequireUserAdministratorAccess(requireUserAdministratorAccess)
     {
         if (applyPathTableRange) {
-            if (auto cell = TBase::GetCellFrom(0)) {
-                PathFrom = cell->AsBuf();
+            if (auto cellsFrom = TBase::TableRange.From.GetCells(); cellsFrom.size() > 0 && !cellsFrom[0].IsNull()) {
+                PathFrom = cellsFrom[0].AsBuf();
             }
-            if (auto cell = TBase::GetCellTo(0)) {
-                PathTo = cell->AsBuf();
+            if (auto cellsTo = TBase::TableRange.To.GetCells(); cellsTo.size() > 0 && !cellsTo[0].IsNull()) {
+                PathTo = cellsTo[0].AsBuf();
             }
         }
     }

ydb/core/sys_view/auth/group_members.cpp

@@ -34,7 +34,9 @@ class TGroupMembersScan : public TAuthScanBase<TGroupMembersScan> {
         TVector<std::pair<const TDomainInfo::TGroup*, const TString*>> memberships;
         for (const auto& group : entry.DomainInfo->Groups) {
             for (const auto& member : group.Members) {
-                memberships.emplace_back(&group, &member);
+                if (StringKeyIsInTableRange({group.Sid, member})) {
+                    memberships.emplace_back(&group, &member);
+                }
             }
         }
         SortBatch(memberships, [](const auto& left, const auto& right) {

ydb/core/sys_view/auth/groups.cpp

@@ -33,10 +33,9 @@ class TGroupsScan : public TAuthScanBase<TGroupsScan> {
 
         TVector<const TDomainInfo::TGroup*> groups(::Reserve(entry.DomainInfo->Groups.size()));
         for (const auto& group : entry.DomainInfo->Groups) {
-            if (!OneCellStringKeyIsInTableRange(group.Sid)) {
-                continue;
+            if (StringKeyIsInTableRange({group.Sid})) {
+                groups.push_back(&group);
             }
-            groups.push_back(&group);
         }
         SortBatch(groups, [](const auto* left, const auto* right) {
             return left->Sid < right->Sid;

ydb/core/sys_view/auth/owners.cpp

@@ -34,7 +34,7 @@ class TOwnersScan : public TAuthScanBase<TOwnersScan> {
 
         auto entryPath = CanonizePath(entry.Path);
 
-        if (OneCellStringKeyIsInTableRange(entryPath)) {
+        if (StringKeyIsInTableRange({entryPath})) {
             for (auto& column : Columns) {
                 switch (column.Tag) {
                 case Schema::AuthOwners::Path::ColumnId:

ydb/core/sys_view/auth/permissions.cpp

@@ -36,6 +36,8 @@ class TPermissionsScan : public TAuthScanBase<TPermissionsScan> {
             batch.Finished = false;
             return;
         }
+
+        auto entryPath = CanonizePath(entry.Path);
 
         TVector<std::pair<TString, TString>> permissions;
         for (const NACLibProto::TACE& ace : entry.SecurityObject->GetACL().GetACE()) {
@@ -45,10 +47,15 @@ class TPermissionsScan : public TAuthScanBase<TPermissionsScan> {
             if (!Effective && ace.GetInherited()) {
                 continue;
             }
+            if (!ace.HasSID()) {
+                continue;
+            }
 
             auto acePermissions = ConvertACLMaskToYdbPermissionNames(ace.GetAccessRight());
             for (const auto& permission : acePermissions) {
-                permissions.emplace_back(ace.HasSID() ? ace.GetSID() : TString{}, std::move(permission));
+                if (StringKeyIsInTableRange({entryPath, ace.GetSID(), permission})) {
+                    permissions.emplace_back(ace.GetSID(), std::move(permission));
+                }
             }
         }
         // Note: due to rights inheritance permissions may be duplicated
@@ -59,8 +66,6 @@ class TPermissionsScan : public TAuthScanBase<TPermissionsScan> {
 
         TVector<TCell> cells(::Reserve(Columns.size()));
 
-        auto entryPath = CanonizePath(entry.Path);
-
         for (const auto& [sid, permission] : permissions) {
             for (auto& column : Columns) {
                 switch (column.Tag) {

ydb/core/sys_view/auth/users.cpp

@@ -89,7 +89,7 @@ class TUsersScan : public TScanActorBase<TUsersScan> {
             if (!user.HasName() || !CanAccessUser(user.GetName())) {
                 continue;
             }
-            if (!OneCellStringKeyIsInTableRange(user.GetName())) {
+            if (!StringKeyIsInTableRange({user.GetName()})) {
                 continue;
             }
             users.push_back(&user);

ydb/core/sys_view/common/scan_actor_base_impl.h

@@ -162,25 +162,52 @@ class TScanActorBase : public TActorBootstrapped<TDerived> {
         SendBatch(std::move(batch));
     }
 
-    std::optional<TCell> GetCellFrom(size_t index) const {
-        return GetCell(TableRange.From.GetCells(), index);
-    }
-
-    std::optional<TCell> GetCellTo(size_t index) const {
-        return GetCell(TableRange.To.GetCells(), index);
-    }
-
-    bool OneCellStringKeyIsInTableRange(const TString value) const {
-        if (auto pathFrom = GetCellFrom(0); pathFrom) {
-            if (int cmp = pathFrom->AsBuf().compare(value); cmp > 0 || cmp == 0 && !TableRange.FromInclusive) {
+    bool StringKeyIsInTableRange(const TVector<TString>& key) const {
+        {
+            bool equalPrefixes = true;
+            for (size_t index : xrange(Min(TableRange.From.GetCells().size(), key.size()))) {
+                if (auto cellFrom = TableRange.From.GetCells()[index]; !cellFrom.IsNull()) {
+                    int cmp = cellFrom.AsBuf().compare(key[index]);
+                    if (cmp < 0) {
+                        equalPrefixes = false;
+                        break;
+                    }
+                    if (cmp > 0) {
+                        return false;
+                    }
+                    // cmp == 0, prefixes are equal, go further
+                } else {
+                    equalPrefixes = false;
+                    break;
+                }
+            }
+            if (equalPrefixes && !TableRange.FromInclusive) {
                 return false;
             }
         }
-        if (auto pathTo = GetCellTo(0); pathTo) {
-            if (int cmp = pathTo->AsBuf().compare(value); cmp < 0 || cmp == 0 && !TableRange.ToInclusive) {
+
+        if (TableRange.To.GetCells().size()) {
+            bool equalPrefixes = true;
+            for (size_t index : xrange(Min(TableRange.To.GetCells().size(), key.size()))) {
+                if (auto cellTo = TableRange.To.GetCells()[index]; !cellTo.IsNull()) {
+                    int cmp = cellTo.AsBuf().compare(key[index]);
+                    if (cmp > 0) {
+                        equalPrefixes = false;
+                        break;
+                    }
+                    if (cmp < 0) {
+                        return false;
+                    }
+                    // cmp == 0, prefixes are equal, go further
+                } else {
+                    break;
+                }
+            }
+            if (equalPrefixes && !TableRange.ToInclusive) {
                 return false;
             }
         }
+
         return true;
     }
 
@@ -331,13 +358,6 @@ class TScanActorBase : public TActorBootstrapped<TDerived> {
         }
     }
 
-    std::optional<TCell> GetCell(TConstArrayRef<TCell> cells, size_t index) const {
-        if (index < cells.size() && !cells[index].IsNull()) {
-            return cells[index];
-        }
-        return {};
-    }
-
 protected:
     static constexpr TDuration Timeout = TDuration::Seconds(60);