From 87bdd624027d43a4da1d2d00de5e63a959b5f17c Mon Sep 17 00:00:00 2001 From: aoiasd <45024769+aoiasd@users.noreply.github.com> Date: Sat, 10 Oct 2026 13:45:39 +0800 Subject: [PATCH] feat: [RLS8] decouple RLS metadata from enforcement (#54059) relate: #50263 design doc: docs/design-docs/design_docs/20250610-rls_design.md design doc PR: #53173 ## Summary Allow changing `rls.enabled` independently of policy/tag management. Preserve metadata across toggles and permit configuration before enabling enforcement; keep schema dependency checks and explicit drop cleanup. Recover all saved policies into resident Coord metadata before readiness, including disabled collections, and maintain the map after successful persistence. Principal tags remain persisted in etcd and point-read by Coord without startup preload; Proxy execution caches are unchanged. Reuse existing collection-cache invalidation, continue invalidation after an ignored missing-resource-group result, and simplify redundant request/enum adapters without removing supported protocol modes. ## Validation Two fresh independent static reviewers completed the full eligible production diff at `96688e8484`; primary reconciliation found no new actionable finding. Static failure-path checks cover recovery failure, persisted mutation failure, metadata preservation across toggles and cache invalidation retries. Regression coverage was updated but has not been executed for this revision. No new local build or test run has been performed for this revision. Earlier build and test results apply to earlier commits only; current runtime validation remains with PR CI. Static review excludes tests/generated contents and does not verify distributed execution. Distributed cutover is not claimed to be atomic. --------- Signed-off-by: aoiasd --- .../design_docs/20250610-rls_design.md | 47 ++-- internal/metastore/model/collection.go | 5 +- internal/metastore/model/rls.go | 59 +---- internal/metastore/model/rls_test.go | 35 +-- .../proxy/dml/task_rls_enforcement_test.go | 9 +- internal/proxy/impl.go | 14 +- internal/proxy/impl_rls_snapshot_test.go | 42 ++++ internal/proxy/rls/manager_test.go | 4 +- internal/proxy/task.go | 3 - internal/proxy/task_test.go | 15 +- .../ddl_callbacks_alter_collection_field.go | 9 +- ...l_callbacks_alter_collection_field_test.go | 22 ++ ...l_callbacks_alter_collection_properties.go | 11 +- ...lbacks_alter_collection_properties_test.go | 34 ++- .../ddl_callbacks_collection_function.go | 12 - .../ddl_callbacks_collection_function_test.go | 1 + internal/rootcoord/ddl_callbacks_rls.go | 26 +-- internal/rootcoord/ddl_callbacks_rls_test.go | 138 +++++++++++ internal/rootcoord/meta_table.go | 124 ++-------- internal/rootcoord/meta_table_test.go | 219 ++++++++++++++++-- internal/rootcoord/rls_api.go | 58 +---- internal/rootcoord/rls_api_test.go | 22 +- internal/rootcoord/root_coord.go | 5 +- internal/rootcoord/root_coord_test.go | 1 + internal/util/rlsutil/metadata.go | 9 +- internal/util/rlsutil/types.go | 195 ++-------------- internal/util/rlsutil/validation.go | 6 +- internal/util/rlsutil/validation_test.go | 22 ++ pkg/common/common.go | 22 -- pkg/common/common_test.go | 29 --- 30 files changed, 614 insertions(+), 584 deletions(-) diff --git a/docs/design-docs/design_docs/20250610-rls_design.md b/docs/design-docs/design_docs/20250610-rls_design.md index f1373cbca5..90a287d134 100644 --- a/docs/design-docs/design_docs/20250610-rls_design.md +++ b/docs/design-docs/design_docs/20250610-rls_design.md @@ -24,14 +24,20 @@ are outside the initial scope. ## Collection Switch -`rls.enabled` can be changed through collection properties. Disabled -collections bypass RLS. Enabling RLS makes the collection deny-by-default and -must not become observable until the RLS metadata state is ready and every -serving Proxy will refresh the collection state before handling later -requests. Disabling RLS removes all policies and tag bindings for the -collection and invalidates Proxy RLS state. These synchronization steps are -performed by Milvus as part of the property transition; users do not manage -Proxy caches directly. +`rls.enabled` controls enforcement, not metadata ownership, and can be changed +through collection properties. Policy and tag management is available whether +RLS is enabled or disabled, subject to the same management privileges and +validation. Applications can configure rules before enabling enforcement. + +Disabled collections bypass RLS without deleting policies or tag bindings. +Enabling or re-enabling applies the current configuration and remains +deny-by-default when no applicable permissive policy exists. Deleting the +property restores the disabled default. The switch uses the existing +AlterCollection cache-invalidation path; policy and tag mutations continue to +invalidate their caches even while enforcement is disabled. Requests load +missing or stale RLS state on demand and fail closed if that fails. Already +captured request snapshots are unchanged. Metadata is removed by explicit +policy/tag deletion or collection drop, not by toggling enforcement. Every row-bearing request on an enabled collection must provide request-level RLS context. Most APIs use top-level `rls_principal` and `skip_rls`; bulk import @@ -39,8 +45,9 @@ carries the same values in its existing `options` map. Sub-searches inherit the top-level decision. A skip is allowed only when authorization is disabled or the authenticated Milvus user has `SkipRLS` on the collection. -`rls.force=true` rejects `skip_rls=true` and is meaningful only while RLS is -enabled. +`rls.force=true` rejects `skip_rls=true` and requires `rls.enabled=true`. +To disable a forced collection, set both properties to false together, or +delete both properties. ## Principal Tag Bindings @@ -208,9 +215,18 @@ functions such as `now()`. RootCoord owns policies and principal tag bindings. Records use globally unique collection IDs as identity; database and collection names are descriptive. -RootCoord keeps complete policies in a name-keyed collection map, including -their internal IDs. Principal tag bindings remain in the catalog and are read -by `(collectionID, principalName)` instead of being loaded during recovery. +Like collection schemas, policies are recovered before RootCoord becomes ready +and maintained in a name-keyed map including their internal IDs. Recovery +includes stored policies on disabled collections; a recovery failure fails +metadata initialization. Reads use the resident metadata, and successful +persisted mutations update it. Enforcement changes neither load nor discard +this metadata, so management and schema-dependency checks remain available +while RLS is disabled. + +Principal tag bindings remain in the catalog and are read by +`(collectionID, principalName)` instead of being loaded during recovery. +Disabling enforcement stops execution-driven tag reads, not metadata +management. Tag caches expire naturally without deleting stored bindings. The initial design assumes policy and tag mutations are low-frequency control-plane operations. Each mutation uses a CChannel broadcast with the same @@ -297,5 +313,6 @@ RLS uses an explicit application principal rather than the Milvus username, lazy principal-tag loading rather than collection-wide Proxy snapshots, and the existing broadcast/ACK path rather than direct catalog mutation. -Follow-ups include an atomic tag patch API, high-frequency tag mutation, -pagination for bulk principal APIs, and dedicated CDC validation. +Follow-ups include a bounded, demand-loaded RootCoord tag cache, an atomic tag +patch API, high-frequency tag mutation, pagination for bulk principal APIs, +and dedicated CDC validation. diff --git a/internal/metastore/model/collection.go b/internal/metastore/model/collection.go index 1534a784c0..81f0943721 100644 --- a/internal/metastore/model/collection.go +++ b/internal/metastore/model/collection.go @@ -42,8 +42,9 @@ type Collection struct { Fields []*Field StructArrayFields []*StructArrayField Functions []*Function - // RLS policies are cached by RootCoord and persisted in their own KV namespace. - RLSPolicies map[string]*RLSPolicy + // RLS policies are owned by RootCoord and persisted in their own KV namespace. + RLSPolicies map[string]*RLSPolicy + VirtualChannelNames []string PhysicalChannelNames []string ShardsNum int32 diff --git a/internal/metastore/model/rls.go b/internal/metastore/model/rls.go index 73b4ddbd41..bca1cce941 100644 --- a/internal/metastore/model/rls.go +++ b/internal/metastore/model/rls.go @@ -17,9 +17,10 @@ package model import ( + "maps" + "slices" "sort" - "github.com/milvus-io/milvus-proto/go-api/v3/milvuspb" "github.com/milvus-io/milvus/internal/util/rlsutil" "github.com/milvus-io/milvus/pkg/v3/proto/rootcoordpb" ) @@ -45,8 +46,8 @@ func MarshalRLSPolicyModel(policy *RLSPolicy) *rootcoordpb.RLSPolicyInfo { CollectionId: policy.CollectionID, PolicyId: policy.PolicyID, PolicyName: policy.PolicyName, - PolicyType: milvuspb.RowPolicyType(policy.PolicyType), - Actions: policyActionsToProto(policy.Actions), + PolicyType: policy.PolicyType, + Actions: slices.Clone(policy.Actions), UsingExpr: policy.UsingExpr, CheckExpr: policy.CheckExpr, Description: policy.Description, @@ -62,8 +63,8 @@ func UnmarshalRLSPolicyModel(policy *rootcoordpb.RLSPolicyInfo) *RLSPolicy { CollectionID: policy.GetCollectionId(), PolicyID: policy.GetPolicyId(), PolicyName: policy.GetPolicyName(), - PolicyType: rlsutil.PolicyType(policy.GetPolicyType()), - Actions: policyActionsFromProto(policy.GetActions()), + PolicyType: policy.GetPolicyType(), + Actions: slices.Clone(policy.GetActions()), UsingExpr: policy.GetUsingExpr(), CheckExpr: policy.GetCheckExpr(), Description: policy.GetDescription(), @@ -77,7 +78,7 @@ func (policy *RLSPolicy) ToRowPolicy() *rlsutil.RowPolicy { return &rlsutil.RowPolicy{ PolicyName: policy.PolicyName, PolicyType: policy.PolicyType, - Actions: cloneRowPolicyActions(policy.Actions), + Actions: slices.Clone(policy.Actions), UsingExpr: policy.UsingExpr, CheckExpr: policy.CheckExpr, Description: policy.Description, @@ -95,7 +96,7 @@ func CloneRLSPolicy(policy *RLSPolicy) *RLSPolicy { PolicyID: policy.PolicyID, PolicyName: policy.PolicyName, PolicyType: policy.PolicyType, - Actions: cloneRowPolicyActions(policy.Actions), + Actions: slices.Clone(policy.Actions), UsingExpr: policy.UsingExpr, CheckExpr: policy.CheckExpr, Description: policy.Description, @@ -187,48 +188,6 @@ func CloneRLSPrincipal(principal *RLSPrincipal) *RLSPrincipal { DBID: principal.DBID, CollectionID: principal.CollectionID, PrincipalName: principal.PrincipalName, - Tags: cloneRLSTags(principal.Tags), + Tags: maps.Clone(principal.Tags), } } - -func cloneRowPolicyActions(actions []rlsutil.PolicyAction) []rlsutil.PolicyAction { - if actions == nil { - return nil - } - cloned := make([]rlsutil.PolicyAction, len(actions)) - copy(cloned, actions) - return cloned -} - -func policyActionsToProto(actions []rlsutil.PolicyAction) []milvuspb.RowPolicyAction { - if actions == nil { - return nil - } - converted := make([]milvuspb.RowPolicyAction, len(actions)) - for i, action := range actions { - converted[i] = milvuspb.RowPolicyAction(action) - } - return converted -} - -func policyActionsFromProto(actions []milvuspb.RowPolicyAction) []rlsutil.PolicyAction { - if actions == nil { - return nil - } - converted := make([]rlsutil.PolicyAction, len(actions)) - for i, action := range actions { - converted[i] = rlsutil.PolicyAction(action) - } - return converted -} - -func cloneRLSTags(in map[string]rlsutil.TagValue) map[string]rlsutil.TagValue { - if in == nil { - return nil - } - out := make(map[string]rlsutil.TagValue, len(in)) - for k, v := range in { - out[k] = v - } - return out -} diff --git a/internal/metastore/model/rls_test.go b/internal/metastore/model/rls_test.go index 2b908fad7a..fd44dd1504 100644 --- a/internal/metastore/model/rls_test.go +++ b/internal/metastore/model/rls_test.go @@ -21,27 +21,30 @@ import ( "github.com/stretchr/testify/require" - "github.com/milvus-io/milvus-proto/go-api/v3/milvuspb" "github.com/milvus-io/milvus/internal/util/rlsutil" ) -func TestRLSPolicyActionValuesMatchPublicProto(t *testing.T) { - tests := []struct { - internal rlsutil.PolicyAction - proto milvuspb.RowPolicyAction - }{ - {rlsutil.PolicyActionQuery, milvuspb.RowPolicyAction_Query}, - {rlsutil.PolicyActionSearch, milvuspb.RowPolicyAction_Search}, - {rlsutil.PolicyActionInsert, milvuspb.RowPolicyAction_Insert}, - {rlsutil.PolicyActionDelete, milvuspb.RowPolicyAction_Delete}, - {rlsutil.PolicyActionUpsert, milvuspb.RowPolicyAction_Upsert}, - {rlsutil.PolicyActionQueryIterator, milvuspb.RowPolicyAction_QueryIterator}, - {rlsutil.PolicyActionSearchIterator, milvuspb.RowPolicyAction_SearchIterator}, - {rlsutil.PolicyActionHybridSearch, milvuspb.RowPolicyAction_HybridSearch}, +func TestRLSPolicyModelCopiesActions(t *testing.T) { + policy := &RLSPolicy{ + DBID: 10, CollectionID: 20, PolicyID: 100, PolicyName: "tenant", + PolicyType: rlsutil.PolicyTypeRestrictive, + Actions: []rlsutil.PolicyAction{rlsutil.PolicyActionQuery, rlsutil.PolicyActionUpsert}, + UsingExpr: "tenant == $current_principal", CheckExpr: "tenant == $current_principal", } + encoded := MarshalRLSPolicyModel(policy) + decoded := UnmarshalRLSPolicyModel(encoded) + require.Equal(t, policy, decoded) + encoded.Actions[0] = rlsutil.PolicyActionDelete + require.Equal(t, rlsutil.PolicyActionQuery, policy.Actions[0]) + require.Equal(t, rlsutil.PolicyActionQuery, decoded.Actions[0]) + decoded.Actions[1] = rlsutil.PolicyActionInsert + require.Equal(t, rlsutil.PolicyActionUpsert, policy.Actions[1]) - for _, test := range tests { - require.Equal(t, int32(test.proto), int32(test.internal)) + require.Nil(t, MarshalRLSPolicyModel(nil)) + require.Nil(t, UnmarshalRLSPolicyModel(nil)) + for _, actions := range [][]rlsutil.PolicyAction{nil, {}} { + policy.Actions = actions + require.Equal(t, actions, UnmarshalRLSPolicyModel(MarshalRLSPolicyModel(policy)).Actions) } } diff --git a/internal/proxy/dml/task_rls_enforcement_test.go b/internal/proxy/dml/task_rls_enforcement_test.go index 81b82e2430..6bcdbb4f35 100644 --- a/internal/proxy/dml/task_rls_enforcement_test.go +++ b/internal/proxy/dml/task_rls_enforcement_test.go @@ -18,6 +18,7 @@ package dml import ( "context" + "slices" "testing" "github.com/bytedance/mockey" @@ -118,16 +119,12 @@ func rlsPolicyInfos(collectionID int64, policies []*rlsutil.RowPolicy) []*rootco if policy == nil { continue } - actions := make([]milvuspb.RowPolicyAction, len(policy.Actions)) - for i, action := range policy.Actions { - actions[i] = milvuspb.RowPolicyAction(action) - } converted = append(converted, &rootcoordpb.RLSPolicyInfo{ CollectionId: collectionID, PolicyId: int64(len(converted) + 1), PolicyName: policy.PolicyName, - PolicyType: milvuspb.RowPolicyType(policy.PolicyType), - Actions: actions, + PolicyType: policy.PolicyType, + Actions: slices.Clone(policy.Actions), UsingExpr: policy.UsingExpr, CheckExpr: policy.CheckExpr, Description: policy.Description, diff --git a/internal/proxy/impl.go b/internal/proxy/impl.go index 9ac52521b3..33ea78cd64 100644 --- a/internal/proxy/impl.go +++ b/internal/proxy/impl.go @@ -7025,20 +7025,12 @@ func (node *Proxy) resolveRLSRequestTarget(ctx context.Context, req rlsManagemen return dbName, collectionName, nil } -func rlsPolicyActionsFromProto(actions []milvuspb.RowPolicyAction) []rlsutil.PolicyAction { - converted := make([]rlsutil.PolicyAction, len(actions)) - for i, action := range actions { - converted[i] = rlsutil.PolicyAction(action) - } - return converted -} - func normalizeCreateRowPolicyType(req *milvuspb.CreateRowPolicyRequest) rlsutil.PolicyType { if req.PolicyType == nil { policyType := milvuspb.RowPolicyType_RowPolicyTypePermissive req.PolicyType = &policyType } - return rlsutil.PolicyType(req.GetPolicyType()) + return req.GetPolicyType() } func (node *Proxy) CreateRowPolicy(ctx context.Context, req *milvuspb.CreateRowPolicyRequest) (*commonpb.Status, error) { @@ -7060,7 +7052,7 @@ func (node *Proxy) CreateRowPolicy(ctx context.Context, req *milvuspb.CreateRowP if err := rlsutil.ValidatePolicyActionCount(len(req.GetActions())); err != nil { return merr.Status(err), nil } - if err := rlsutil.ValidatePolicy(req.GetPolicyName(), normalizeCreateRowPolicyType(req), rlsPolicyActionsFromProto(req.GetActions()), req.GetUsingExpr(), req.GetCheckExpr()); err != nil { + if err := rlsutil.ValidatePolicy(req.GetPolicyName(), normalizeCreateRowPolicyType(req), req.GetActions(), req.GetUsingExpr(), req.GetCheckExpr()); err != nil { return merr.Status(err), nil } if err := rlsutil.ValidatePolicyDescription(req.GetDescription()); err != nil { @@ -7100,7 +7092,7 @@ func (node *Proxy) UpdateRowPolicy(ctx context.Context, req *milvuspb.UpdateRowP if err := rlsutil.ValidatePolicyActionCount(len(req.GetActions())); err != nil { return merr.Status(err), nil } - if err := rlsutil.ValidatePolicyForUpdate(req.GetPolicyName(), rlsutil.PolicyType(req.GetPolicyType()), rlsPolicyActionsFromProto(req.GetActions()), req.GetUsingExpr(), req.GetCheckExpr()); err != nil { + if err := rlsutil.ValidatePolicyForUpdate(req.GetPolicyName(), req.GetPolicyType(), req.GetActions(), req.GetUsingExpr(), req.GetCheckExpr()); err != nil { return merr.Status(err), nil } if err := rlsutil.ValidatePolicyDescription(req.GetDescription()); err != nil { diff --git a/internal/proxy/impl_rls_snapshot_test.go b/internal/proxy/impl_rls_snapshot_test.go index 17104c3c07..5fdf83bceb 100644 --- a/internal/proxy/impl_rls_snapshot_test.go +++ b/internal/proxy/impl_rls_snapshot_test.go @@ -20,13 +20,55 @@ import ( "context" "testing" + "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" "github.com/milvus-io/milvus-proto/go-api/v3/commonpb" + "github.com/milvus-io/milvus-proto/go-api/v3/milvuspb" + "github.com/milvus-io/milvus-proto/go-api/v3/schemapb" + "github.com/milvus-io/milvus/internal/mocks" + "github.com/milvus-io/milvus/internal/proxy/shardclient" "github.com/milvus-io/milvus/pkg/v3/common" "github.com/milvus-io/milvus/pkg/v3/proto/proxypb" + "github.com/milvus-io/milvus/pkg/v3/util/merr" ) +func TestProxyAlterCollectionRefreshesRLSEnforcement(t *testing.T) { + ctx := context.Background() + coord := mocks.NewMockMixCoordClient(t) + cache, err := NewMetaCache(coord) + require.NoError(t, err) + shard := shardclient.NewMockShardClientManager(t) + shard.EXPECT().InvalidateShardLeaderCache([]int64{100}).Return().Times(5) + node := &Proxy{metaCache: cache, shardMgr: shard} + node.UpdateStateCode(commonpb.StateCode_Healthy) + + for _, value := range []string{"", "true", "false", "true", ""} { + var properties []*commonpb.KeyValuePair + if value != "" { + properties = []*commonpb.KeyValuePair{{Key: common.RLSEnabledKey, Value: value}} + } + coord.EXPECT().DescribeCollection(mock.Anything, mock.Anything).Return(&milvuspb.DescribeCollectionResponse{ + Status: merr.Success(), CollectionID: 100, DbName: "default", + Schema: &schemapb.CollectionSchema{Name: "coll", Properties: properties}, Properties: properties, + }, nil).Once() + status, err := node.InvalidateCollectionMetaCache(ctx, &proxypb.InvalidateCollMetaCacheRequest{ + Base: &commonpb.MsgBase{MsgType: commonpb.MsgType_AlterCollection}, + DbName: "default", CollectionName: "coll", CollectionID: 100, + }) + require.NoError(t, merr.CheckRPCCall(status, err)) + info, err := cache.GetCollectionInfo(ctx, "default", "coll", 100) + require.NoError(t, err) + require.Equal(t, value == "true", info.RlsEnabled) + fromSchema, err := common.IsRLSEnabled(info.Schema.GetProperties()...) + require.NoError(t, err) + require.Equal(t, info.RlsEnabled, fromSchema) + hit, err := cache.GetCollectionInfo(ctx, "default", "coll", 100) + require.NoError(t, err) + require.Same(t, info, hit) + } +} + func TestProxyRLSInvalidateRemovesSnapshots(t *testing.T) { ctx := context.Background() node := &Proxy{} diff --git a/internal/proxy/rls/manager_test.go b/internal/proxy/rls/manager_test.go index fceba0607c..4d1482a91f 100644 --- a/internal/proxy/rls/manager_test.go +++ b/internal/proxy/rls/manager_test.go @@ -827,8 +827,8 @@ func TestManagerRequestPathLoadsMissingStartupState(t *testing.T) { CollectionId: collectionID, PolicyId: 1, PolicyName: "dept", - PolicyType: milvuspb.RowPolicyType(rlsutil.PolicyTypePermissive), - Actions: []milvuspb.RowPolicyAction{milvuspb.RowPolicyAction(rlsutil.PolicyActionQuery)}, + PolicyType: rlsutil.PolicyTypePermissive, + Actions: []milvuspb.RowPolicyAction{rlsutil.PolicyActionQuery}, UsingExpr: "dept == $current_principal_tags['dept']", }, }, diff --git a/internal/proxy/task.go b/internal/proxy/task.go index 0bafe763b9..9b40ee5f28 100644 --- a/internal/proxy/task.go +++ b/internal/proxy/task.go @@ -2134,9 +2134,6 @@ func (t *alterCollectionTask) PreExecute(ctx context.Context) error { if err := common.ValidateRLSProperties(t.GetProperties()...); err != nil { return err } - if err := common.ValidateRLSEnabledNotAltered(t.GetProperties(), t.GetDeleteKeys()); err != nil { - return err - } for _, key := range t.GetDeleteKeys() { for _, expected := range []string{common.RLSEnabledKey, common.RLSForceKey} { if strings.EqualFold(key, expected) && key != expected { diff --git a/internal/proxy/task_test.go b/internal/proxy/task_test.go index 9397863b91..00a50692db 100644 --- a/internal/proxy/task_test.go +++ b/internal/proxy/task_test.go @@ -5307,7 +5307,7 @@ func TestCollectionRLSEnabledValidation(t *testing.T) { assert.Contains(t, err.Error(), "invalid rls.enabled") }) - t.Run("alter rejects rls.enabled", func(t *testing.T) { + t.Run("alter accepts rls.enabled", func(t *testing.T) { colName := prefix + funcutil.GenRandomStr() createCollection(colName) alterTask := &alterCollectionTask{ @@ -5320,11 +5320,10 @@ func TestCollectionRLSEnabledValidation(t *testing.T) { mixCoord: qc, } err := alterTask.PreExecute(ctx) - assert.ErrorIs(t, err, merr.ErrParameterInvalid) - assert.ErrorContains(t, err, "cannot alter rls.enabled") + assert.NoError(t, err) }) - t.Run("alter rejects standard boolean rls.enabled spelling", func(t *testing.T) { + t.Run("alter accepts standard boolean rls.enabled spelling", func(t *testing.T) { colName := prefix + funcutil.GenRandomStr() createCollection(colName) alterTask := &alterCollectionTask{ @@ -5337,11 +5336,10 @@ func TestCollectionRLSEnabledValidation(t *testing.T) { mixCoord: qc, } err := alterTask.PreExecute(ctx) - assert.ErrorIs(t, err, merr.ErrParameterInvalid) - assert.ErrorContains(t, err, "cannot alter rls.enabled") + assert.NoError(t, err) }) - t.Run("alter rejects deleting rls.enabled", func(t *testing.T) { + t.Run("alter accepts deleting rls.enabled", func(t *testing.T) { colName := prefix + funcutil.GenRandomStr() createCollection(colName) alterTask := &alterCollectionTask{ @@ -5354,8 +5352,7 @@ func TestCollectionRLSEnabledValidation(t *testing.T) { mixCoord: qc, } err := alterTask.PreExecute(ctx) - assert.ErrorIs(t, err, merr.ErrParameterInvalid) - assert.ErrorContains(t, err, "cannot delete rls.enabled") + assert.NoError(t, err) }) t.Run("alter rejects wrong case rls.enabled delete key", func(t *testing.T) { diff --git a/internal/rootcoord/ddl_callbacks_alter_collection_field.go b/internal/rootcoord/ddl_callbacks_alter_collection_field.go index 939a0c9a7a..172f85b5d9 100644 --- a/internal/rootcoord/ddl_callbacks_alter_collection_field.go +++ b/internal/rootcoord/ddl_callbacks_alter_collection_field.go @@ -103,7 +103,8 @@ func (c *Core) broadcastAlterCollectionV2ForAlterCollectionField(ctx context.Con } newFieldProperties := common.NewKeyValuePairs(oldFieldPropertiesMap) - if newFieldProperties.Equal(oldFieldProperties) && desc == nil { + propertiesChanged := !newFieldProperties.Equal(oldFieldProperties) + if !propertiesChanged && desc == nil { // if there's no change, return nil directly to promise idempotent. return errIgnoredAlterCollection } @@ -139,8 +140,10 @@ func (c *Core) broadcastAlterCollectionV2ForAlterCollectionField(ctx context.Con return err } } - if err := validateRLSPoliciesWithSchema(coll.RLSPolicies, schema); err != nil { - return err + if propertiesChanged { + if err := validateRLSPoliciesWithSchema(coll.RLSPolicies, schema); err != nil { + return err + } } cacheExpirations, err := c.getCacheExpireForCollection(ctx, req.GetDbName(), req.GetCollectionName()) if err != nil { diff --git a/internal/rootcoord/ddl_callbacks_alter_collection_field_test.go b/internal/rootcoord/ddl_callbacks_alter_collection_field_test.go index f4c3799e1d..c1f97dab83 100644 --- a/internal/rootcoord/ddl_callbacks_alter_collection_field_test.go +++ b/internal/rootcoord/ddl_callbacks_alter_collection_field_test.go @@ -130,6 +130,28 @@ func TestDDLCallbacksAlterCollectionField(t *testing.T) { assertFieldProperties(t, ctx, core, dbName, collectionName, fieldName, "key1", "value1") assertFieldPropertiesNotFound(t, ctx, core, dbName, collectionName, fieldName, "key2") assertSchemaVersion(t, ctx, core, dbName, collectionName, 3) + + // A description-only edit preserves the existing field properties. + req := &milvuspb.AlterCollectionFieldRequest{ + DbName: dbName, + CollectionName: collectionName, + FieldName: fieldName, + Properties: []*commonpb.KeyValuePair{ + {Key: "key1", Value: "value1"}, + {Key: common.FieldDescriptionKey, Value: "updated description"}, + }, + } + resp, err = core.AlterCollectionField(ctx, req) + require.NoError(t, merr.CheckRPCCall(resp, err)) + coll, err := core.meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp, false) + require.NoError(t, err) + require.Equal(t, "updated description", getAlterCollectionField(coll.ToCollectionSchemaPB(), fieldName).GetDescription()) + assertFieldProperties(t, ctx, core, dbName, collectionName, fieldName, "key1", "value1") + + req.Properties[0].Value = "value2" + resp, err = core.AlterCollectionField(ctx, req) + require.NoError(t, merr.CheckRPCCall(resp, err)) + assertFieldProperties(t, ctx, core, dbName, collectionName, fieldName, "key1", "value2") } func TestDDLCallbacksAlterNestedArrayRootCapacity(t *testing.T) { diff --git a/internal/rootcoord/ddl_callbacks_alter_collection_properties.go b/internal/rootcoord/ddl_callbacks_alter_collection_properties.go index ecc1945208..33cd9229c7 100644 --- a/internal/rootcoord/ddl_callbacks_alter_collection_properties.go +++ b/internal/rootcoord/ddl_callbacks_alter_collection_properties.go @@ -58,9 +58,6 @@ func (c *Core) broadcastAlterCollectionForAlterCollection(ctx context.Context, r if err := common.ValidateRLSProperties(req.GetProperties()...); err != nil { return err } - if err := common.ValidateRLSEnabledNotAltered(req.GetProperties(), req.GetDeleteKeys()); err != nil { - return err - } for _, key := range req.GetDeleteKeys() { for _, expected := range []string{common.RLSEnabledKey, common.RLSForceKey} { if strings.EqualFold(key, expected) && key != expected { @@ -492,11 +489,11 @@ func (c *DDLCallback) alterCollectionV2AckCallback(ctx context.Context, result m return merr.Wrap(err, "failed to update load config") } if err := merr.CheckRPCCall(resp, err); err != nil { - if errors.Is(err, merr.ErrResourceGroupNotFound) { - mlog.Warn(ctx, "failed to update load config due to missing resource group, stop retrying", mlog.Err(err)) - return nil + if !errors.Is(err, merr.ErrResourceGroupNotFound) { + return merr.Wrap(err, "failed to update load config") } - return merr.Wrap(err, "failed to update load config") + // Properties are already persisted; still invalidate Proxy caches. + mlog.Warn(ctx, "failed to update load config due to missing resource group, stop retrying", mlog.Err(err)) } } if err := c.cascadeDropFieldIndexesInline(ctx, result); err != nil { diff --git a/internal/rootcoord/ddl_callbacks_alter_collection_properties_test.go b/internal/rootcoord/ddl_callbacks_alter_collection_properties_test.go index 7265a52645..bd677b9c7c 100644 --- a/internal/rootcoord/ddl_callbacks_alter_collection_properties_test.go +++ b/internal/rootcoord/ddl_callbacks_alter_collection_properties_test.go @@ -36,6 +36,7 @@ import ( "github.com/milvus-io/milvus/internal/streamingcoord/server/balancer/balance" "github.com/milvus-io/milvus/pkg/v3/common" "github.com/milvus-io/milvus/pkg/v3/proto/messagespb" + "github.com/milvus-io/milvus/pkg/v3/proto/proxypb" "github.com/milvus-io/milvus/pkg/v3/streaming/util/message" "github.com/milvus-io/milvus/pkg/v3/streaming/util/types" "github.com/milvus-io/milvus/pkg/v3/util" @@ -110,12 +111,11 @@ func TestDDLCallbacksAlterCollectionProperties(t *testing.T) { }) require.ErrorIs(t, merr.CheckRPCCall(resp, err), merr.ErrParameterInvalid) - // "True" is a valid standard boolean spelling, but rls.enabled itself is - // immutable after collection creation. + // RLS property values are validated before looking up the collection. resp, err = core.AlterCollection(ctx, &milvuspb.AlterCollectionRequest{ DbName: dbName, CollectionName: collectionName, - Properties: []*commonpb.KeyValuePair{{Key: common.RLSEnabledKey, Value: "True"}}, + Properties: []*commonpb.KeyValuePair{{Key: common.RLSEnabledKey, Value: "invalid"}}, }) require.ErrorIs(t, merr.CheckRPCCall(resp, err), merr.ErrParameterInvalid) @@ -159,14 +159,14 @@ func TestDDLCallbacksAlterCollectionProperties(t *testing.T) { createCollectionAndAliasForTest(t, ctx, core, dbName, collectionName) assertReplicaNumber(t, ctx, core, dbName, collectionName, 1) - // RLS can only be enabled when the collection is created. + // Existing collections can enable RLS through the normal property path. resp, err = core.AlterCollection(ctx, &milvuspb.AlterCollectionRequest{ DbName: dbName, CollectionName: collectionName, Properties: []*commonpb.KeyValuePair{{Key: common.RLSEnabledKey, Value: "true"}}, }) alterErr := merr.CheckRPCCall(resp, err) - require.ErrorIs(t, alterErr, merr.ErrParameterInvalid) + require.NoError(t, alterErr) for _, tc := range []struct { name string @@ -442,6 +442,7 @@ func TestDDLCallbacksAlterCollectionV2AckCallback_StopRetryOnResourceGroupNotFou withMeta(meta), withMixCoord(mixc), withValidProxyManager(), + withTsoAllocator(newMockTsoAllocator()), withBroker(&mockBroker{ BroadcastAlteredCollectionFunc: func(ctx context.Context, collectionID int64) error { return nil @@ -449,13 +450,23 @@ func TestDDLCallbacksAlterCollectionV2AckCallback_StopRetryOnResourceGroupNotFou }), ) cb := &DDLCallback{Core: c} + notifyErr := merr.WrapErrServiceUnavailableMsg("proxy temporarily unavailable") + var notifications []*proxypb.InvalidateCollMetaCacheRequest + proxy := newMockProxy() + proxy.InvalidateCollectionMetaCacheFunc = func(_ context.Context, req *proxypb.InvalidateCollMetaCacheRequest) (*commonpb.Status, error) { + notifications = append(notifications, req) + return merr.Status(notifyErr), nil + } + c.proxyClientManager.GetProxyClients().Insert(TestProxyID, proxy) raw := message.NewAlterCollectionMessageBuilderV2(). WithHeader(&messagespb.AlterCollectionMessageHeader{ - CollectionId: 1, + CollectionId: 1, + CacheExpirations: newRLSCacheExpirations("db", "coll", 1, commonpb.MsgType_AlterCollection), }). WithBody(&messagespb.AlterCollectionMessageBody{ Updates: &messagespb.AlterCollectionMessageUpdates{ + Properties: []*commonpb.KeyValuePair{{Key: common.RLSEnabledKey, Value: "true"}}, AlterLoadConfig: &messagespb.AlterLoadConfigOfAlterCollection{ ReplicaNumber: 1, ResourceGroups: []string{"rg_not_exist"}, @@ -470,7 +481,18 @@ func TestDDLCallbacksAlterCollectionV2AckCallback_StopRetryOnResourceGroupNotFou Message: msg, Results: map[string]*message.AppendResult{}, }) + require.ErrorIs(t, err, merr.ErrServiceUnavailable) + require.Len(t, notifications, 1) + require.Equal(t, int64(1), notifications[0].GetCollectionID()) + require.Equal(t, commonpb.MsgType_AlterCollection, notifications[0].GetBase().GetMsgType()) + + notifyErr = nil + err = cb.alterCollectionV2AckCallback(ctx, message.BroadcastResultAlterCollectionMessageV2{ + Message: msg, + Results: map[string]*message.AppendResult{}, + }) require.NoError(t, err) + require.Len(t, notifications, 2) } func TestCore_getAlterLoadConfigOfAlterCollection(t *testing.T) { diff --git a/internal/rootcoord/ddl_callbacks_collection_function.go b/internal/rootcoord/ddl_callbacks_collection_function.go index 5314337946..277c83c34f 100644 --- a/internal/rootcoord/ddl_callbacks_collection_function.go +++ b/internal/rootcoord/ddl_callbacks_collection_function.go @@ -179,8 +179,6 @@ func (c *Core) broadcastAlterCollectionForAlterFunction(ctx context.Context, req if err := rejectExternalCollectionFunctionMutation(oldColl.ToCollectionSchemaPB()); err != nil { return err } - oldFn := findRLSFunctionByName(oldColl, req.GetFunctionSchema().GetName()) - // Only whitelisted params may be altered; function identity (type, name, // input/output fields) is immutable. Skip when the function is absent so the // mutation helper below emits the canonical "not exists" error. @@ -197,15 +195,5 @@ func (c *Core) broadcastAlterCollectionForAlterFunction(ctx context.Context, req if err := alterFunctionGenNewCollection(ctx, req.FunctionSchema, newColl); err != nil { return err } - newFn := findRLSFunctionByName(newColl, req.GetFunctionSchema().GetName()) - if !rlsFunctionKeepsOutputShape(oldFn, newFn) { - if err := validateRLSFunctionOutputNotReferenced(oldColl, oldFn, "altered"); err != nil { - return err - } - } - if err := validateRLSPoliciesWithSchema(oldColl.RLSPolicies, newColl.ToCollectionSchemaPB()); err != nil { - return err - } - return callAlterCollection(ctx, c, broadcaster, oldColl, newColl, req.GetDbName(), req.GetCollectionName()) } diff --git a/internal/rootcoord/ddl_callbacks_collection_function_test.go b/internal/rootcoord/ddl_callbacks_collection_function_test.go index 47992db561..7398980643 100644 --- a/internal/rootcoord/ddl_callbacks_collection_function_test.go +++ b/internal/rootcoord/ddl_callbacks_collection_function_test.go @@ -385,6 +385,7 @@ func (suite *DDLCallbacksCollectionFunctionTestSuite) TestBroadcastAlterCollecti err := suite.core.broadcastAlterCollectionForAlterFunction(context.Background(), req) suite.NoError(err) + mockMeta.AssertNotCalled(suite.T(), "GetRLSMetadata", mock.Anything, mock.Anything, mock.Anything, mock.Anything) }) suite.Run("altering non-whitelisted param rejected", func() { diff --git a/internal/rootcoord/ddl_callbacks_rls.go b/internal/rootcoord/ddl_callbacks_rls.go index 16aebf4138..6f080f26e0 100644 --- a/internal/rootcoord/ddl_callbacks_rls.go +++ b/internal/rootcoord/ddl_callbacks_rls.go @@ -18,9 +18,9 @@ package rootcoord import ( "context" + "slices" "github.com/milvus-io/milvus-proto/go-api/v3/commonpb" - "github.com/milvus-io/milvus-proto/go-api/v3/milvuspb" "github.com/milvus-io/milvus/internal/metastore/model" "github.com/milvus-io/milvus/internal/streamingcoord/server/broadcaster" "github.com/milvus-io/milvus/internal/util/proxyutil" @@ -203,8 +203,8 @@ func marshalRLSPolicyMessage(policy *model.RLSPolicy) *messagespb.RLSPolicyMetad return &messagespb.RLSPolicyMetadata{ PolicyId: policy.PolicyID, PolicyName: policy.PolicyName, - PolicyType: milvuspb.RowPolicyType(policy.PolicyType), - Actions: policyActionsToMilvusProto(policy.Actions), + PolicyType: policy.PolicyType, + Actions: slices.Clone(policy.Actions), UsingExpr: policy.UsingExpr, CheckExpr: policy.CheckExpr, Description: policy.Description, @@ -228,30 +228,14 @@ func unmarshalRLSPolicyMessage(header *message.AlterRLSMetadataMessageHeader, po CollectionID: header.GetCollectionId(), PolicyID: policy.GetPolicyId(), PolicyName: policy.GetPolicyName(), - PolicyType: rlsutil.PolicyType(policy.GetPolicyType()), - Actions: policyActionsFromMilvusProto(policy.GetActions()), + PolicyType: policy.GetPolicyType(), + Actions: slices.Clone(policy.GetActions()), UsingExpr: policy.GetUsingExpr(), CheckExpr: policy.GetCheckExpr(), Description: policy.GetDescription(), } } -func policyActionsToMilvusProto(actions []rlsutil.PolicyAction) []milvuspb.RowPolicyAction { - converted := make([]milvuspb.RowPolicyAction, len(actions)) - for i, action := range actions { - converted[i] = milvuspb.RowPolicyAction(action) - } - return converted -} - -func policyActionsFromMilvusProto(actions []milvuspb.RowPolicyAction) []rlsutil.PolicyAction { - converted := make([]rlsutil.PolicyAction, len(actions)) - for i, action := range actions { - converted[i] = rlsutil.PolicyAction(action) - } - return converted -} - func unmarshalRLSPrincipalMessage(header *message.AlterRLSMetadataMessageHeader, principal *messagespb.RLSPrincipalMetadata) (*model.RLSPrincipal, error) { tags, err := rlsutil.TagsFromJSON(principal.GetTags()) if err != nil { diff --git a/internal/rootcoord/ddl_callbacks_rls_test.go b/internal/rootcoord/ddl_callbacks_rls_test.go index 5fe137354f..1b049fe67f 100644 --- a/internal/rootcoord/ddl_callbacks_rls_test.go +++ b/internal/rootcoord/ddl_callbacks_rls_test.go @@ -26,6 +26,7 @@ import ( "github.com/stretchr/testify/require" "github.com/milvus-io/milvus-proto/go-api/v3/commonpb" + "github.com/milvus-io/milvus-proto/go-api/v3/milvuspb" "github.com/milvus-io/milvus/internal/metastore/model" mockrootcoord "github.com/milvus-io/milvus/internal/rootcoord/mocks" "github.com/milvus-io/milvus/internal/util/proxyutil" @@ -35,8 +36,145 @@ import ( "github.com/milvus-io/milvus/pkg/v3/proto/proxypb" "github.com/milvus-io/milvus/pkg/v3/streaming/util/message" "github.com/milvus-io/milvus/pkg/v3/util/merr" + "github.com/milvus-io/milvus/pkg/v3/util/typeutil" ) +func TestRLSMetadataIndependentOfEnforcement(t *testing.T) { + ctx := context.Background() + core := initStreamingSystemAndCore(t) + const db, name = "rls_switch_db", "rls_switch_collection" + createCollectionForTest(t, ctx, core, db, name) + meta := core.meta.(*MetaTable) + coll, err := meta.GetCollectionByName(ctx, db, name, typeutil.MaxTimestamp, false) + require.NoError(t, err) + id := coll.CollectionID + + // Configure metadata before enabling enforcement, through the WAL callbacks. + createPolicy := &rlsutil.CreateRowPolicyRequest{ + DbName: db, CollectionName: name, PolicyName: "tenant", + PolicyType: rlsutil.PolicyTypePermissive, + Actions: []rlsutil.PolicyAction{rlsutil.PolicyActionQuery}, + UsingExpr: "field1 == $current_principal_tags['tenant']", + } + require.NoError(t, core.broadcastCreateRLSPolicy(ctx, createPolicy)) + setTags := &rlsutil.SetRLSPrincipalTagsRequest{ + DbName: db, CollectionName: name, PrincipalName: "alice", + Tags: map[string]rlsutil.TagValue{"tenant": rlsutil.NewInt64TagValue(1)}, + } + require.NoError(t, core.broadcastSetRLSPrincipalTags(ctx, setTags)) + + for _, enabled := range []string{"True", "false", "true"} { + status, err := core.AlterCollection(ctx, &milvuspb.AlterCollectionRequest{ + DbName: db, CollectionName: name, + Properties: []*commonpb.KeyValuePair{{Key: common.RLSEnabledKey, Value: enabled}}, + }) + require.NoError(t, merr.CheckRPCCall(status, err)) + coll, err = meta.GetCollectionByName(ctx, db, name, typeutil.MaxTimestamp, false) + require.NoError(t, err) + actual, err := common.IsRLSEnabled(coll.Properties...) + require.NoError(t, err) + require.Equal(t, enabled != "false", actual) + require.Len(t, coll.RLSPolicies, 1) + + if enabled == "false" { + // Disabling also permits editing the configuration for the next enable. + require.NoError(t, core.broadcastUpdateRLSPolicy(ctx, &rlsutil.UpdateRowPolicyRequest{ + DbName: db, CollectionName: name, PolicyName: "tenant", + PolicyType: rlsutil.PolicyTypePermissive, + Actions: []rlsutil.PolicyAction{rlsutil.PolicyActionQuery}, + UsingExpr: "field1 == 2", + })) + setTags.Tags["tenant"] = rlsutil.NewInt64TagValue(2) + require.NoError(t, core.broadcastSetRLSPrincipalTags(ctx, setTags)) + } + tags, err := meta.GetRLSPrincipalTags(ctx, &rlsutil.GetRLSPrincipalTagsRequest{ + DbName: db, CollectionName: name, PrincipalName: "alice", + }) + require.NoError(t, err) + require.Equal(t, setTags.Tags, tags) + } + require.Equal(t, "field1 == 2", coll.RLSPolicies["tenant"].UsingExpr) + + status, err := core.AlterCollection(ctx, &milvuspb.AlterCollectionRequest{ + DbName: db, CollectionName: name, + Properties: []*commonpb.KeyValuePair{{Key: common.RLSForceKey, Value: "true"}}, + }) + require.NoError(t, merr.CheckRPCCall(status, err)) + status, err = core.AlterCollection(ctx, &milvuspb.AlterCollectionRequest{ + DbName: db, CollectionName: name, + Properties: []*commonpb.KeyValuePair{{Key: common.RLSEnabledKey, Value: "false"}}, + }) + require.ErrorIs(t, merr.CheckRPCCall(status, err), merr.ErrParameterInvalid) + status, err = core.AlterCollection(ctx, &milvuspb.AlterCollectionRequest{ + DbName: db, CollectionName: name, + DeleteKeys: []string{common.RLSEnabledKey, common.RLSForceKey}, + }) + require.NoError(t, merr.CheckRPCCall(status, err)) + coll, err = meta.GetCollectionByName(ctx, db, name, typeutil.MaxTimestamp, false) + require.NoError(t, err) + enabled, err := common.IsRLSEnabled(coll.Properties...) + require.NoError(t, err) + require.False(t, enabled) + + // Recovery restores metadata even while enforcement is disabled. + coll = coll.Clone() + coll.RLSPolicies = nil + require.NoError(t, meta.reloadCollectionsRLSMetadata(ctx, []*model.Collection{coll})) + require.Len(t, coll.RLSPolicies, 1) + meta.ddLock.Lock() + meta.collID2Meta[id] = coll.Clone() + meta.ddLock.Unlock() + require.ErrorContains(t, core.broadcastCreateRLSPolicy(ctx, createPolicy), "already exists") + require.Equal(t, "field1 == 2", coll.RLSPolicies["tenant"].UsingExpr) + + status, err = core.AlterCollection(ctx, &milvuspb.AlterCollectionRequest{ + DbName: db, CollectionName: name, + Properties: []*commonpb.KeyValuePair{{Key: common.RLSEnabledKey, Value: "true"}}, + }) + require.NoError(t, merr.CheckRPCCall(status, err)) + coll, err = meta.GetCollectionByName(ctx, db, name, typeutil.MaxTimestamp, false) + require.NoError(t, err) + require.Len(t, coll.RLSPolicies, 1) + require.Equal(t, "field1 == 2", coll.RLSPolicies["tenant"].UsingExpr) + status, err = core.AlterCollection(ctx, &milvuspb.AlterCollectionRequest{ + DbName: db, CollectionName: name, + Properties: []*commonpb.KeyValuePair{{Key: common.RLSEnabledKey, Value: "false"}}, + }) + require.NoError(t, merr.CheckRPCCall(status, err)) + coll, err = meta.GetCollectionByName(ctx, db, name, typeutil.MaxTimestamp, false) + require.NoError(t, err) + require.NoError(t, meta.reloadCollectionsRLSMetadata(ctx, []*model.Collection{coll})) + meta.ddLock.Lock() + meta.collID2Meta[id] = coll + meta.ddLock.Unlock() + + require.NoError(t, core.broadcastDeleteRLSPrincipalTags(ctx, &rlsutil.DeleteRLSPrincipalTagsRequest{ + DbName: db, CollectionName: name, PrincipalName: "alice", TagKeys: []string{"tenant"}, + })) + require.NoError(t, core.broadcastDropRLSPolicy(ctx, &rlsutil.DropRowPolicyRequest{ + DbName: db, CollectionName: name, PolicyName: "tenant", + })) + policies, err := meta.ListRLSPolicies(ctx, &rlsutil.ListRowPoliciesRequest{DbName: db, CollectionName: name}) + require.NoError(t, err) + require.Empty(t, policies) + principals, err := meta.ListRLSPrincipals(ctx, &rlsutil.ListRLSPrincipalsRequest{DbName: db, CollectionName: name}) + require.NoError(t, err) + require.Empty(t, principals) + + // Dropping a disabled collection, unlike toggling enforcement, removes metadata. + require.NoError(t, core.broadcastCreateRLSPolicy(ctx, createPolicy)) + require.NoError(t, core.broadcastSetRLSPrincipalTags(ctx, setTags)) + status, err = core.DropCollection(ctx, &milvuspb.DropCollectionRequest{DbName: db, CollectionName: name}) + require.NoError(t, merr.CheckRPCCall(status, err)) + // The production tombstone sweeper performs this after data GC. + require.NoError(t, meta.RemoveCollection(ctx, id, 0)) + stored, err := meta.catalog.ListRLSPolicies(ctx, id) + require.NoError(t, err) + require.Empty(t, stored) + _, err = meta.catalog.GetRLSPrincipal(ctx, id, "alice") + require.ErrorIs(t, err, merr.ErrIoKeyNotFound) +} + func mustMarshalRLSPrincipalMessage(principal *model.RLSPrincipal) *messagespb.RLSPrincipalMetadata { message, err := marshalRLSPrincipalMessage(principal) if err != nil { diff --git a/internal/rootcoord/meta_table.go b/internal/rootcoord/meta_table.go index 21e50bef1b..12c9e671ec 100644 --- a/internal/rootcoord/meta_table.go +++ b/internal/rootcoord/meta_table.go @@ -2570,21 +2570,13 @@ func (mt *MetaTable) resolveRLSCollection(ctx context.Context, dbName string, co if err != nil { return nil, err } - enabled, err := common.IsRLSEnabled(collection.Properties...) - if err != nil { + if _, err := common.IsRLSEnabled(collection.Properties...); err != nil { return nil, merr.WrapErrDataIntegrity(err, "invalid RLS properties for collection %d", collection.CollectionID) } - if !enabled { - return nil, merr.WrapErrParameterInvalidMsg( - "RLS is not enabled for collection %q; set %s=true when creating the collection", - collection.Name, - common.RLSEnabledKey, - ) - } return collection, nil } -func (mt *MetaTable) reloadEnabledCollectionRLSMetadata(ctx context.Context, collection *model.Collection) error { +func (mt *MetaTable) reloadCollectionRLSMetadata(ctx context.Context, collection *model.Collection) error { policies, err := mt.catalog.ListRLSPolicies(ctx, collection.CollectionID) if err != nil { return merr.Wrapf(err, "failed to reload RLS policies for collection %d", collection.CollectionID) @@ -2611,31 +2603,24 @@ func (mt *MetaTable) reloadEnabledCollectionRLSMetadata(ctx context.Context, col } func (mt *MetaTable) reloadCollectionsRLSMetadata(ctx context.Context, collections []*model.Collection) error { - enabledCollections := make([]*model.Collection, 0) for _, collection := range collections { if collection == nil { continue } - enabled, err := common.IsRLSEnabled(collection.Properties...) - if err != nil { + if _, err := common.IsRLSEnabled(collection.Properties...); err != nil { return merr.WrapErrDataIntegrity(err, "invalid RLS properties for collection %d", collection.CollectionID) } - if !enabled { - collection.RLSPolicies = nil - continue - } - enabledCollections = append(enabledCollections, collection) } - if ctx == nil { - ctx = context.TODO() - } group, groupCtx := errgroup.WithContext(ctx) group.SetLimit(rlsRecoveryConcurrency) - for _, collection := range enabledCollections { - collection := collection + // Policies are metadata even while enforcement is disabled. + for _, collection := range collections { + if collection == nil { + continue + } group.Go(func() error { - return mt.reloadEnabledCollectionRLSMetadata(groupCtx, collection) + return mt.reloadCollectionRLSMetadata(groupCtx, collection) }) } return group.Wait() @@ -2651,11 +2636,13 @@ func upsertCollectionRLSPolicy(collection *model.Collection, policy *model.RLSPo collection.RLSPolicies[policy.PolicyName] = model.CloneRLSPolicy(policy) } -func removeCollectionRLSPolicy(collection *model.Collection, policyName string) { +func removeCollectionRLSPolicy(collection *model.Collection, policy *model.RLSPolicy) { if collection == nil { return } - delete(collection.RLSPolicies, policyName) + if cached := collection.RLSPolicies[policy.PolicyName]; cached != nil && cached.PolicyID == policy.PolicyID { + delete(collection.RLSPolicies, policy.PolicyName) + } } func validateRLSCombinedExpressionLength(policies []*model.RLSPolicy) error { @@ -2848,50 +2835,6 @@ func validateRLSNoReferencedFieldDropped(coll *model.Collection, droppedFieldIDs return nil } -func validateRLSFunctionOutputNotReferenced(coll *model.Collection, fn *model.Function, operation string) error { - if fn == nil || len(fn.OutputFieldIDs) == 0 { - return nil - } - fieldRefs, err := collectRLSPolicyFieldRefs(coll) - if err != nil { - return err - } - for _, fieldID := range fn.OutputFieldIDs { - if policies := fieldRefs[fieldID]; len(policies) > 0 { - sort.Strings(policies) - return merr.WrapErrParameterInvalidMsg("function %q cannot be %s because output field %d is referenced by RLS policies %v", fn.Name, operation, fieldID, policies) - } - } - return nil -} - -func findRLSFunctionByName(coll *model.Collection, functionName string) *model.Function { - for _, fn := range coll.Functions { - if fn.Name == functionName { - return fn - } - } - return nil -} - -func rlsFunctionKeepsOutputShape(oldFn *model.Function, newFn *model.Function) bool { - if oldFn == nil || newFn == nil { - return false - } - if oldFn.Type != newFn.Type { - return false - } - if len(oldFn.OutputFieldIDs) != len(newFn.OutputFieldIDs) { - return false - } - for i := range oldFn.OutputFieldIDs { - if oldFn.OutputFieldIDs[i] != newFn.OutputFieldIDs[i] { - return false - } - } - return true -} - func validateRLSPolicyExpression(schemaHelper *typeutil.SchemaHelper, exprKind string, expr string, enforceArrayLiteralLimit bool) error { expr = strings.TrimSpace(expr) if expr == "" { @@ -3140,13 +3083,7 @@ func (mt *MetaTable) ApplyDropRLSPolicy(ctx context.Context, collectionID int64, } mt.ddLock.Lock() defer mt.ddLock.Unlock() - collection = mt.collID2Meta[collectionID] - if collection != nil { - current := collection.RLSPolicies[policyName] - if current != nil && current.PolicyID == policy.PolicyID { - removeCollectionRLSPolicy(collection, policyName) - } - } + removeCollectionRLSPolicy(mt.collID2Meta[collectionID], policy) return nil } @@ -3187,11 +3124,9 @@ func (mt *MetaTable) GetRLSMetadata(ctx context.Context, collectionID int64, kin } loadPrincipals := false switch kind { - case rootcoordpb.RLSMetadataKind_RLS_METADATA_KIND_ALL: - metadata.Policies = model.RLSPolicyMapToSlice(coll.RLSPolicies) - loadPrincipals = true - case rootcoordpb.RLSMetadataKind_RLS_METADATA_KIND_POLICIES: + case rootcoordpb.RLSMetadataKind_RLS_METADATA_KIND_ALL, rootcoordpb.RLSMetadataKind_RLS_METADATA_KIND_POLICIES: metadata.Policies = model.RLSPolicyMapToSlice(coll.RLSPolicies) + loadPrincipals = kind == rootcoordpb.RLSMetadataKind_RLS_METADATA_KIND_ALL case rootcoordpb.RLSMetadataKind_RLS_METADATA_KIND_PRINCIPALS: loadPrincipals = true default: @@ -3200,18 +3135,14 @@ func (mt *MetaTable) GetRLSMetadata(ctx context.Context, collectionID int64, kin } mt.ddLock.RUnlock() - if !loadPrincipals { - return metadata, nil - } - var principals []*model.RLSPrincipal - if principalName == "" { + if loadPrincipals && principalName == "" { var err error principals, err = mt.catalog.ListRLSPrincipals(ctx, collectionID) if err != nil { return nil, merr.Wrap(err, "failed to list RLS principal metadata") } - } else { + } else if loadPrincipals { // TODO: Add a bounded Coord-side principal cache for point reads; // never restore startup full loading. principal, err := mt.catalog.GetRLSPrincipal(ctx, collectionID, principalName) @@ -3251,17 +3182,6 @@ func (mt *MetaTable) GetRLSMetadata(ctx context.Context, collectionID int64, kin return metadata, nil } -func cloneRLSTags(tags map[string]rlsutil.TagValue) map[string]rlsutil.TagValue { - if tags == nil { - return nil - } - cloned := make(map[string]rlsutil.TagValue, len(tags)) - for key, value := range tags { - cloned[key] = value - } - return cloned -} - func (mt *MetaTable) PrepareSetRLSPrincipalTags(ctx context.Context, req *rlsutil.SetRLSPrincipalTagsRequest) (*model.RLSPrincipal, error) { if req == nil { return nil, merr.WrapErrParameterInvalidMsg("set RLS principal tags request is nil") @@ -3289,9 +3209,9 @@ func (mt *MetaTable) PrepareSetRLSPrincipalTags(ctx context.Context, req *rlsuti return nil, merr.Wrap(err, "failed to get RLS principal") } - mergedTags := cloneRLSTags(req.GetTags()) + mergedTags := maps.Clone(req.GetTags()) if !isNew { - mergedTags = cloneRLSTags(existingPrincipal.Tags) + mergedTags = maps.Clone(existingPrincipal.Tags) if mergedTags == nil { mergedTags = make(map[string]rlsutil.TagValue, len(req.GetTags())) } @@ -3334,7 +3254,7 @@ func (mt *MetaTable) GetRLSPrincipalTags(ctx context.Context, req *rlsutil.GetRL } return nil, merr.Wrap(err, "failed to get RLS principal") } - return cloneRLSTags(principal.Tags), nil + return maps.Clone(principal.Tags), nil } func (mt *MetaTable) ListRLSPrincipals(ctx context.Context, req *rlsutil.ListRLSPrincipalsRequest) ([]string, error) { @@ -3403,7 +3323,7 @@ func (mt *MetaTable) PrepareDeleteRLSPrincipalTags(ctx context.Context, req *rls return principal, true, nil } - tags := cloneRLSTags(principal.Tags) + tags := maps.Clone(principal.Tags) for _, key := range tagKeys { delete(tags, key) } diff --git a/internal/rootcoord/meta_table_test.go b/internal/rootcoord/meta_table_test.go index e5abd01014..36d8b3582d 100644 --- a/internal/rootcoord/meta_table_test.go +++ b/internal/rootcoord/meta_table_test.go @@ -20,6 +20,7 @@ import ( "context" "math/rand" "strconv" + "sync" "sync/atomic" "testing" "time" @@ -110,6 +111,7 @@ func newRLSMetaTableForTest(t *testing.T) (*MetaTable, *mocks.RootCoordCatalog) names.insert("db1", "coll1", 20) return &MetaTable{ + ctx: context.Background(), catalog: catalog, names: names, aliases: newNameDb(), @@ -260,7 +262,7 @@ func (mt *MetaTable) DeleteRLSPrincipalTags(ctx context.Context, req *rlsutil.De return principal.CollectionID, nil } -func TestReloadEnabledCollectionRLSMetadataLoadsPoliciesOnly(t *testing.T) { +func TestReloadCollectionRLSMetadataLoadsPoliciesOnly(t *testing.T) { catalog := mocks.NewRootCoordCatalog(t) meta := &MetaTable{catalog: catalog} collection := &model.Collection{ @@ -277,7 +279,7 @@ func TestReloadEnabledCollectionRLSMetadataLoadsPoliciesOnly(t *testing.T) { } catalog.EXPECT().ListRLSPolicies(mock.Anything, int64(20)).Return([]*model.RLSPolicy{nil, stored}, nil).Once() - require.NoError(t, meta.reloadEnabledCollectionRLSMetadata(context.Background(), collection)) + require.NoError(t, meta.reloadCollectionRLSMetadata(context.Background(), collection)) require.Len(t, collection.RLSPolicies, 1) require.Equal(t, int64(11), collection.RLSPolicies["tenant"].DBID) require.Equal(t, int64(10), stored.DBID) @@ -294,12 +296,12 @@ func TestReloadRLSMetadataRejectsDuplicatePolicyNames(t *testing.T) { {CollectionID: 20, PolicyID: 100, PolicyName: "tenant"}, {CollectionID: 20, PolicyID: 101, PolicyName: "tenant"}, }, nil).Once() - require.ErrorIs(t, meta.reloadEnabledCollectionRLSMetadata(context.Background(), collection), merr.ErrDataIntegrity) + require.ErrorIs(t, meta.reloadCollectionRLSMetadata(context.Background(), collection), merr.ErrDataIntegrity) require.Len(t, collection.RLSPolicies, 1) require.Same(t, existing, collection.RLSPolicies["existing"]) } -func TestReloadCollectionsRLSMetadataSkipsDisabledCollection(t *testing.T) { +func TestReloadCollectionsRLSMetadataRestoresDisabledCollection(t *testing.T) { for _, tc := range []struct { name string properties []*commonpb.KeyValuePair @@ -313,16 +315,19 @@ func TestReloadCollectionsRLSMetadataSkipsDisabledCollection(t *testing.T) { }, } { t.Run(tc.name, func(t *testing.T) { - catalog := mocks.NewRootCoordCatalog(t) - meta := &MetaTable{catalog: catalog} - collection := &model.Collection{ - CollectionID: 20, - Properties: tc.properties, - RLSPolicies: map[string]*model.RLSPolicy{"stale": {PolicyName: "stale"}}, - } + meta, catalog := newRLSMetaTableForTest(t) + collection := meta.collID2Meta[20] + collection.Properties = tc.properties + collection.RLSPolicies = map[string]*model.RLSPolicy{"stale": {PolicyName: "stale"}} + catalog.EXPECT().ListRLSPolicies(mock.Anything, int64(20)).Return([]*model.RLSPolicy{{ + CollectionID: 20, PolicyID: 100, PolicyName: "tenant", UsingExpr: `dept == "sales"`, + }}, nil).Once() require.NoError(t, meta.reloadCollectionsRLSMetadata(context.Background(), []*model.Collection{collection})) - require.Empty(t, collection.RLSPolicies) + require.Len(t, collection.RLSPolicies, 1) + require.Equal(t, int64(100), collection.RLSPolicies["tenant"].PolicyID) + require.ErrorContains(t, validateRLSNoReferencedFieldDropped(collection, []int64{101}), "tenant") + require.NoError(t, validateRLSNoReferencedFieldDropped(collection, []int64{102})) }) } } @@ -341,26 +346,178 @@ func TestReloadCollectionsRLSMetadataRejectsInvalidProperty(t *testing.T) { require.ErrorIs(t, err, merr.ErrDataIntegrity) } -func TestResolveRLSCollectionRejectsDisabledCollection(t *testing.T) { +func TestReloadDisabledCollectionRLSMetadataPropagatesCatalogError(t *testing.T) { + meta, catalog := newRLSMetaTableForTest(t) + collection := meta.collID2Meta[20] + collection.Properties = nil + policy := &model.RLSPolicy{PolicyName: "tenant", PolicyID: 100} + collection.RLSPolicies = map[string]*model.RLSPolicy{"tenant": policy} + catalog.EXPECT().ListRLSPolicies(mock.Anything, int64(20)).Return( + nil, merr.WrapErrServiceUnavailableMsg("catalog unavailable")).Once() + + err := meta.reloadCollectionsRLSMetadata(context.Background(), []*model.Collection{collection}) + require.ErrorIs(t, err, merr.ErrServiceUnavailable) + require.Same(t, policy, collection.RLSPolicies["tenant"]) +} + +func TestRLSPoliciesStayResidentWhileTagsLoadOnDemand(t *testing.T) { + ctx := context.Background() + meta, catalog := newRLSMetaTableForTest(t) + collection := meta.collID2Meta[20] + collection.Properties = nil + + policy := &model.RLSPolicy{DBID: 10, CollectionID: 20, PolicyID: 100, PolicyName: "tenant"} + catalog.EXPECT().SaveRLSPolicy(mock.Anything, policy).Return(nil).Once() + require.NoError(t, meta.ApplyAlterRLSPolicy(ctx, policy)) + require.Equal(t, policy, collection.RLSPolicies["tenant"]) + require.NotSame(t, policy, collection.RLSPolicies["tenant"]) + + catalog.EXPECT().GetRLSPrincipal(mock.Anything, int64(20), "alice").Return(nil, merr.ErrIoKeyNotFound).Once() + metadata, err := meta.GetRLSMetadata(ctx, 20, rootcoordpb.RLSMetadataKind_RLS_METADATA_KIND_PRINCIPALS, "alice") + require.NoError(t, err) + require.Empty(t, metadata.Principals) + catalog.AssertNotCalled(t, "ListRLSPolicies", mock.Anything, mock.Anything) + + metadata, err = meta.GetRLSMetadata(ctx, 20, rootcoordpb.RLSMetadataKind_RLS_METADATA_KIND_POLICIES, "") + require.NoError(t, err) + require.Len(t, metadata.Policies, 1) + require.Equal(t, int64(100), metadata.Policies[0].PolicyID) + require.Len(t, collection.RLSPolicies, 1) + catalog.AssertNotCalled(t, "ListRLSPolicies", mock.Anything, mock.Anything) +} + +func TestRLSPolicyMetadataChangesAfterPersistence(t *testing.T) { + ctx := context.Background() + meta, catalog := newRLSMetaTableForTest(t) + collection := meta.collID2Meta[20] + collection.Properties = nil + old := &model.RLSPolicy{CollectionID: 20, PolicyID: 100, PolicyName: "tenant", UsingExpr: "true"} + collection.RLSPolicies = map[string]*model.RLSPolicy{"tenant": old} + updated := model.CloneRLSPolicy(old) + updated.UsingExpr = "false" + + catalog.EXPECT().SaveRLSPolicy(mock.Anything, updated).Return(merr.ErrServiceUnavailable).Once() + require.ErrorIs(t, meta.ApplyAlterRLSPolicy(ctx, updated), merr.ErrServiceUnavailable) + require.Same(t, old, collection.RLSPolicies["tenant"]) + catalog.EXPECT().SaveRLSPolicy(mock.Anything, updated).Return(nil).Once() + require.NoError(t, meta.ApplyAlterRLSPolicy(ctx, updated)) + require.Equal(t, updated, collection.RLSPolicies["tenant"]) + require.NotSame(t, updated, collection.RLSPolicies["tenant"]) + + catalog.EXPECT().DropRLSPolicy(mock.Anything, int64(20), int64(100)).Return(merr.ErrServiceUnavailable).Once() + require.ErrorIs(t, meta.ApplyDropRLSPolicy(ctx, 20, "tenant"), merr.ErrServiceUnavailable) + require.Equal(t, updated, collection.RLSPolicies["tenant"]) + catalog.EXPECT().DropRLSPolicy(mock.Anything, int64(20), int64(100)).Return(nil).Once() + require.NoError(t, meta.ApplyDropRLSPolicy(ctx, 20, "tenant")) + require.Empty(t, collection.RLSPolicies) +} + +func TestNewMetaTableRejectsPolicyRecoveryFailure(t *testing.T) { + catalog := mocks.NewRootCoordCatalog(t) + catalog.EXPECT().ListDatabases(mock.Anything, typeutil.MaxTimestamp). + Return([]*model.Database{model.NewDefaultDatabase(nil)}, nil).Once() + catalog.EXPECT().ListCollections(mock.Anything, util.NonDBID, typeutil.MaxTimestamp). + Return([]*model.Collection{{CollectionID: 20}}, nil).Once() + catalog.EXPECT().ListRLSPolicies(mock.Anything, int64(20)).Return(nil, merr.ErrServiceUnavailable).Once() + + meta, err := NewMetaTable(context.Background(), catalog, nil) + require.ErrorIs(t, err, merr.ErrServiceUnavailable) + require.Nil(t, meta, "do not publish metadata when policy recovery fails") +} + +func TestGetRLSMetadataOwnsSortedPolicySnapshot(t *testing.T) { + meta, _ := newRLSMetaTableForTest(t) + policies := map[string]*model.RLSPolicy{ + "z": {PolicyName: "z", PolicyID: 100, Actions: []rlsutil.PolicyAction{rlsutil.PolicyActionQuery}}, + "a": {PolicyName: "a", PolicyID: 200, Actions: []rlsutil.PolicyAction{rlsutil.PolicyActionQuery}}, + } + meta.collID2Meta[20].RLSPolicies = policies + metadata, err := meta.GetRLSMetadata(context.Background(), 20, rootcoordpb.RLSMetadataKind_RLS_METADATA_KIND_POLICIES, "") + require.NoError(t, err) + require.Len(t, metadata.Policies, 2) + require.Equal(t, "a", metadata.Policies[0].PolicyName) + require.Equal(t, "z", metadata.Policies[1].PolicyName) + metadata.Policies[0].PolicyName = "changed" + metadata.Policies[0].Actions[0] = rlsutil.PolicyActionInsert + require.Equal(t, "a", policies["a"].PolicyName) + require.Equal(t, rlsutil.PolicyActionQuery, policies["a"].Actions[0]) +} + +func TestAlterCollectionPreservesRLSPolicies(t *testing.T) { + for _, test := range []struct { + name string + before string + after string + properties bool + }{ + {"unrelated disabled alter", "false", "true", false}, + {"unrelated enabled alter", "true", "false", false}, + {"enabled property unchanged", "true", "true", true}, + {"enable", "false", "true", true}, + {"disable", "true", "false", true}, + {"remove enable property", "true", "", true}, + } { + t.Run(test.name, func(t *testing.T) { + meta, catalog := newRLSMetaTableForTest(t) + collection := meta.collID2Meta[20] + collection.Properties = common.NewKeyValuePairs(map[string]string{common.RLSEnabledKey: test.before}) + collection.RLSPolicies = map[string]*model.RLSPolicy{ + "tenant": {CollectionID: 20, PolicyID: 100, PolicyName: "tenant"}, + } + catalog.EXPECT().AlterCollection(mock.Anything, mock.Anything, mock.Anything, metastore.MODIFY, uint64(2), false).Return(nil).Once() + paths := []string{message.FieldMaskCollectionDescription} + if test.properties { + paths = append(paths, message.FieldMaskCollectionProperties) + } + updates := &message.AlterCollectionMessageUpdates{Description: "updated"} + if test.after != "" { + updates.Properties = common.NewKeyValuePairs(map[string]string{common.RLSEnabledKey: test.after}) + } + raw := message.NewAlterCollectionMessageBuilderV2(). + WithHeader(&message.AlterCollectionMessageHeader{ + CollectionId: 20, + UpdateMask: &fieldmaskpb.FieldMask{Paths: paths}, + }). + WithBody(&message.AlterCollectionMessageBody{Updates: updates}). + WithBroadcast([]string{"control"}).MustBuildBroadcast() + require.NoError(t, meta.AlterCollection(context.Background(), message.BroadcastResultAlterCollectionMessageV2{ + Message: message.MustAsBroadcastAlterCollectionMessageV2(raw), + Results: map[string]*message.AppendResult{"control": {TimeTick: 2}}, + })) + require.Equal(t, collection.RLSPolicies, meta.collID2Meta[20].RLSPolicies) + catalog.AssertNotCalled(t, "ListRLSPolicies", mock.Anything, mock.Anything) + enabled, err := common.IsRLSEnabled(meta.collID2Meta[20].Properties...) + require.NoError(t, err) + expected := test.before + if test.properties { + expected = test.after + } + require.Equal(t, expected == "true", enabled) + }) + } +} + +func TestResolveRLSCollectionAllowsDisabledCollection(t *testing.T) { meta, _ := newRLSMetaTableForTest(t) for _, properties := range [][]*commonpb.KeyValuePair{ nil, {{Key: common.RLSEnabledKey, Value: "false"}}, } { meta.collID2Meta[20].Properties = properties - _, err := meta.resolveRLSCollection(context.Background(), "db1", "coll1") - require.ErrorIs(t, err, merr.ErrParameterInvalid) - require.ErrorContains(t, err, "RLS is not enabled") + collection, err := meta.resolveRLSCollection(context.Background(), "db1", "coll1") + require.NoError(t, err) + require.Equal(t, int64(20), collection.CollectionID) } meta.collID2Meta[20].Properties = []*commonpb.KeyValuePair{{Key: common.RLSEnabledKey, Value: "invalid"}} _, err := meta.resolveRLSCollection(context.Background(), "db1", "coll1") require.ErrorIs(t, err, merr.ErrDataIntegrity) } -func TestReloadCollectionsRLSMetadataOnlyQueriesEnabledCollections(t *testing.T) { +func TestReloadCollectionsRLSMetadataQueriesAllCollections(t *testing.T) { catalog := mocks.NewRootCoordCatalog(t) meta := &MetaTable{catalog: catalog} collections := []*model.Collection{ + nil, {CollectionID: 10}, { CollectionID: 20, @@ -375,7 +532,9 @@ func TestReloadCollectionsRLSMetadataOnlyQueriesEnabledCollections(t *testing.T) }, }, } - catalog.EXPECT().ListRLSPolicies(mock.Anything, int64(20)).Return(nil, nil).Once() + for _, id := range []int64{10, 20, 30} { + catalog.EXPECT().ListRLSPolicies(mock.Anything, id).Return(nil, nil).Once() + } require.NoError(t, meta.reloadCollectionsRLSMetadata(context.Background(), collections)) } @@ -397,6 +556,7 @@ func TestReloadCollectionsRLSMetadataUsesBoundedConcurrency(t *testing.T) { var inFlight atomic.Int32 var maxInFlight atomic.Int32 release := make(chan struct{}) + unblock := sync.OnceFunc(func() { close(release) }) trackCall := func() { current := inFlight.Add(1) for { @@ -414,15 +574,27 @@ func TestReloadCollectionsRLSMetadataUsesBoundedConcurrency(t *testing.T) { return nil, nil }).Times(collectionCount) - done := make(chan error, 1) + done := make(chan struct{}) + var reloadErr error go func() { - done <- meta.reloadCollectionsRLSMetadata(context.Background(), collections) + reloadErr = meta.reloadCollectionsRLSMetadata(context.Background(), collections) + close(done) }() + t.Cleanup(func() { + unblock() + <-done + }) require.Eventually(t, func() bool { return maxInFlight.Load() == rlsRecoveryConcurrency }, time.Second, time.Millisecond) - close(release) - require.NoError(t, <-done) + select { + case <-done: + t.Fatal("metadata recovery returned before policies loaded") + default: + } + unblock() + <-done + require.NoError(t, reloadErr) assert.LessOrEqual(t, maxInFlight.Load(), int32(rlsRecoveryConcurrency)) } @@ -3581,6 +3753,7 @@ func TestMetaTable_reload(t *testing.T) { opt(catalog) } return &MetaTable{ + ctx: context.Background(), names: newNameDb(), aliases: newNameDb(), catalog: catalog, @@ -3710,7 +3883,7 @@ func TestMetaTable_reload(t *testing.T) { mock.Anything, ).Return(nil, uint64(0), nil) - meta := &MetaTable{catalog: catalog} + meta := &MetaTable{ctx: context.Background(), catalog: catalog} channel.ResetStaticPChannelStatsManager() err := meta.reload() assert.NoError(t, err) diff --git a/internal/rootcoord/rls_api.go b/internal/rootcoord/rls_api.go index 6a4c5c754f..a9e4a890d2 100644 --- a/internal/rootcoord/rls_api.go +++ b/internal/rootcoord/rls_api.go @@ -52,25 +52,11 @@ func (c *Core) UpdateRowPolicy(ctx context.Context, req *milvuspb.UpdateRowPolic } func (c *Core) DropRowPolicy(ctx context.Context, req *milvuspb.DropRowPolicyRequest) (*commonpb.Status, error) { - if req == nil { - return c.dropRowPolicy(ctx, nil) - } - return c.dropRowPolicy(ctx, &rlsutil.DropRowPolicyRequest{ - DbName: req.GetDbName(), - CollectionName: req.GetCollectionName(), - PolicyName: req.GetPolicyName(), - }) + return c.dropRowPolicy(ctx, req) } func (c *Core) ListRowPolicies(ctx context.Context, req *milvuspb.ListRowPoliciesRequest) (*milvuspb.ListRowPoliciesResponse, error) { - var internalReq *rlsutil.ListRowPoliciesRequest - if req != nil { - internalReq = &rlsutil.ListRowPoliciesRequest{ - DbName: req.GetDbName(), - CollectionName: req.GetCollectionName(), - } - } - resp, err := c.listRowPolicies(ctx, internalReq) + resp, err := c.listRowPolicies(ctx, req) if resp == nil { return nil, err } @@ -81,8 +67,8 @@ func (c *Core) ListRowPolicies(ctx context.Context, req *milvuspb.ListRowPolicie } policies = append(policies, &milvuspb.RowPolicy{ PolicyName: policy.PolicyName, - PolicyType: milvuspb.RowPolicyType(policy.PolicyType), - Actions: policyActionsToMilvusProto(policy.Actions), + PolicyType: policy.PolicyType, + Actions: slices.Clone(policy.Actions), UsingExpr: policy.UsingExpr, CheckExpr: policy.CheckExpr, Description: policy.Description, @@ -127,15 +113,7 @@ func setRLSPrincipalTagsRequestFromProto(req *milvuspb.SetRLSPrincipalTagsReques } func (c *Core) GetRLSPrincipalTags(ctx context.Context, req *milvuspb.GetRLSPrincipalTagsRequest) (*milvuspb.GetRLSPrincipalTagsResponse, error) { - var internalReq *rlsutil.GetRLSPrincipalTagsRequest - if req != nil { - internalReq = &rlsutil.GetRLSPrincipalTagsRequest{ - DbName: req.GetDbName(), - CollectionName: req.GetCollectionName(), - PrincipalName: req.GetPrincipalName(), - } - } - resp, err := c.getRLSPrincipalTags(ctx, internalReq) + resp, err := c.getRLSPrincipalTags(ctx, req) if resp == nil { return nil, err } @@ -158,23 +136,7 @@ func (c *Core) GetRLSPrincipalTags(ctx context.Context, req *milvuspb.GetRLSPrin } func (c *Core) ListRLSPrincipals(ctx context.Context, req *milvuspb.ListRLSPrincipalsRequest) (*milvuspb.ListRLSPrincipalsResponse, error) { - var internalReq *rlsutil.ListRLSPrincipalsRequest - if req != nil { - internalReq = &rlsutil.ListRLSPrincipalsRequest{ - DbName: req.GetDbName(), - CollectionName: req.GetCollectionName(), - } - } - resp, err := c.listRLSPrincipals(ctx, internalReq) - if resp == nil { - return nil, err - } - return &milvuspb.ListRLSPrincipalsResponse{ - Status: resp.Status, - PrincipalNames: slices.Clone(resp.PrincipalNames), - DbName: resp.DbName, - CollectionName: resp.CollectionName, - }, err + return c.listRLSPrincipals(ctx, req) } func (c *Core) DeleteRLSPrincipalTags(ctx context.Context, req *milvuspb.DeleteRLSPrincipalTagsRequest) (*commonpb.Status, error) { @@ -199,14 +161,14 @@ func createRowPolicyRequestFromProto(req *milvuspb.CreateRowPolicyRequest) *rlsu } policyType := rlsutil.PolicyTypePermissive if req.PolicyType != nil { - policyType = rlsutil.PolicyType(req.GetPolicyType()) + policyType = req.GetPolicyType() } return &rlsutil.CreateRowPolicyRequest{ DbName: req.GetDbName(), CollectionName: req.GetCollectionName(), PolicyName: req.GetPolicyName(), PolicyType: policyType, - Actions: policyActionsFromMilvusProto(req.GetActions()), + Actions: slices.Clone(req.GetActions()), UsingExpr: req.GetUsingExpr(), CheckExpr: req.GetCheckExpr(), Description: req.GetDescription(), @@ -221,8 +183,8 @@ func updateRowPolicyRequestFromProto(req *milvuspb.UpdateRowPolicyRequest) *rlsu DbName: req.GetDbName(), CollectionName: req.GetCollectionName(), PolicyName: req.GetPolicyName(), - PolicyType: rlsutil.PolicyType(req.GetPolicyType()), - Actions: policyActionsFromMilvusProto(req.GetActions()), + PolicyType: req.GetPolicyType(), + Actions: slices.Clone(req.GetActions()), UsingExpr: req.GetUsingExpr(), CheckExpr: req.GetCheckExpr(), Description: req.GetDescription(), diff --git a/internal/rootcoord/rls_api_test.go b/internal/rootcoord/rls_api_test.go index 2cfd5d964e..1af8c9cb19 100644 --- a/internal/rootcoord/rls_api_test.go +++ b/internal/rootcoord/rls_api_test.go @@ -53,21 +53,21 @@ func TestRLSProtoRequestConversion(t *testing.T) { require.Equal(t, []rlsutil.PolicyAction{rlsutil.PolicyActionQuery, rlsutil.PolicyActionUpsert}, req.GetActions()) defaultTypeReq := createRowPolicyRequestFromProto(&milvuspb.CreateRowPolicyRequest{}) require.Equal(t, rlsutil.PolicyTypePermissive, defaultTypeReq.GetPolicyType()) + explicitUnknown := createRowPolicyRequestFromProto(&milvuspb.CreateRowPolicyRequest{ + PolicyType: milvuspb.RowPolicyType_RowPolicyTypeUnknown.Enum(), + }) + require.Equal(t, rlsutil.PolicyTypeUnknown, explicitUnknown.GetPolicyType()) + require.Equal(t, rlsutil.PolicyTypeUnknown, updateRowPolicyRequestFromProto(&milvuspb.UpdateRowPolicyRequest{}).GetPolicyType()) + actions := []milvuspb.RowPolicyAction{milvuspb.RowPolicyAction_Query} + create := createRowPolicyRequestFromProto(&milvuspb.CreateRowPolicyRequest{Actions: actions}) + update := updateRowPolicyRequestFromProto(&milvuspb.UpdateRowPolicyRequest{Actions: actions}) + actions[0] = milvuspb.RowPolicyAction_Delete + require.Equal(t, []rlsutil.PolicyAction{rlsutil.PolicyActionQuery}, create.Actions) + require.Equal(t, []rlsutil.PolicyAction{rlsutil.PolicyActionQuery}, update.Actions) require.Nil(t, createRowPolicyRequestFromProto(nil)) require.Nil(t, updateRowPolicyRequestFromProto(nil)) } -func TestRLSPolicyActionValuesMatchPublicProto(t *testing.T) { - require.Equal(t, int32(milvuspb.RowPolicyAction_Query), int32(rlsutil.PolicyActionQuery)) - require.Equal(t, int32(milvuspb.RowPolicyAction_Search), int32(rlsutil.PolicyActionSearch)) - require.Equal(t, int32(milvuspb.RowPolicyAction_Insert), int32(rlsutil.PolicyActionInsert)) - require.Equal(t, int32(milvuspb.RowPolicyAction_Delete), int32(rlsutil.PolicyActionDelete)) - require.Equal(t, int32(milvuspb.RowPolicyAction_Upsert), int32(rlsutil.PolicyActionUpsert)) - require.Equal(t, int32(milvuspb.RowPolicyAction_QueryIterator), int32(rlsutil.PolicyActionQueryIterator)) - require.Equal(t, int32(milvuspb.RowPolicyAction_SearchIterator), int32(rlsutil.PolicyActionSearchIterator)) - require.Equal(t, int32(milvuspb.RowPolicyAction_HybridSearch), int32(rlsutil.PolicyActionHybridSearch)) -} - func TestCoreRejectsLegacyRowPolicyRoles(t *testing.T) { core := newTestCore() diff --git a/internal/rootcoord/root_coord.go b/internal/rootcoord/root_coord.go index c312426692..c121d76601 100644 --- a/internal/rootcoord/root_coord.go +++ b/internal/rootcoord/root_coord.go @@ -463,7 +463,6 @@ func (c *Core) initKVCreator() { func (c *Core) initMetaTable(initCtx context.Context) error { fn := func() error { var catalog metastore.RootCoordCatalog - var err error switch Params.MetaStoreCfg.MetaStoreType.GetValue() { case util.MetaStoreTypeEtcd: @@ -482,9 +481,11 @@ func (c *Core) initMetaTable(initCtx context.Context) error { return retry.Unrecoverable(merr.WrapErrServiceInternalMsg("not supported meta store: %s", Params.MetaStoreCfg.MetaStoreType.GetValue())) } - if c.meta, err = NewMetaTable(c.ctx, catalog, c.tsoAllocator); err != nil { + meta, err := NewMetaTable(c.ctx, catalog, c.tsoAllocator) + if err != nil { return err } + c.meta = meta return nil } diff --git a/internal/rootcoord/root_coord_test.go b/internal/rootcoord/root_coord_test.go index 70e3553c3c..7cb73fd330 100644 --- a/internal/rootcoord/root_coord_test.go +++ b/internal/rootcoord/root_coord_test.go @@ -166,6 +166,7 @@ func initStreamingSystemAndCore(t *testing.T) *Core { tso.EXPECT().GenerateTSO(mock.Anything).Return(uint64(1), nil).Maybe() core := newTestCore(withHealthyCode(), withMeta(&MetaTable{ + ctx: context.Background(), catalog: rootcoord.NewCatalog(catalogKV), names: testDB, aliases: newNameDb(), diff --git a/internal/util/rlsutil/metadata.go b/internal/util/rlsutil/metadata.go index 73bf5c4280..b788b344f3 100644 --- a/internal/util/rlsutil/metadata.go +++ b/internal/util/rlsutil/metadata.go @@ -18,6 +18,7 @@ package rlsutil import ( "context" + "slices" "github.com/cockroachdb/errors" "google.golang.org/grpc/codes" @@ -51,14 +52,10 @@ func RowPoliciesFromInfo(collectionID int64, policies []*rootcoordpb.RLSPolicyIn if policy == nil { return nil, merr.WrapErrDataIntegrityMsg("RLS policy metadata at index %d is nil", i) } - actions := make([]PolicyAction, len(policy.GetActions())) - for i, action := range policy.GetActions() { - actions[i] = PolicyAction(action) - } convertedPolicy := &RowPolicy{ PolicyName: policy.GetPolicyName(), - PolicyType: PolicyType(policy.GetPolicyType()), - Actions: actions, + PolicyType: policy.GetPolicyType(), + Actions: slices.Clone(policy.GetActions()), UsingExpr: policy.GetUsingExpr(), CheckExpr: policy.GetCheckExpr(), Description: policy.GetDescription(), diff --git a/internal/util/rlsutil/types.go b/internal/util/rlsutil/types.go index b2f58d1e40..bc1fbc2459 100644 --- a/internal/util/rlsutil/types.go +++ b/internal/util/rlsutil/types.go @@ -25,6 +25,7 @@ import ( "unsafe" "github.com/milvus-io/milvus-proto/go-api/v3/commonpb" + "github.com/milvus-io/milvus-proto/go-api/v3/milvuspb" "github.com/milvus-io/milvus/pkg/v3/util/merr" "github.com/milvus-io/milvus/pkg/v3/util/paramtable" ) @@ -375,61 +376,27 @@ func tagValueToJSON(value TagValue) (any, bool) { } } -type PolicyType int32 +type PolicyType = milvuspb.RowPolicyType const ( - PolicyTypeUnknown PolicyType = 0 - PolicyTypePermissive PolicyType = 1 - PolicyTypeRestrictive PolicyType = 2 + PolicyTypeUnknown = milvuspb.RowPolicyType_RowPolicyTypeUnknown + PolicyTypePermissive = milvuspb.RowPolicyType_RowPolicyTypePermissive + PolicyTypeRestrictive = milvuspb.RowPolicyType_RowPolicyTypeRestrictive ) -func (policyType PolicyType) String() string { - switch policyType { - case PolicyTypePermissive: - return "RowPolicyTypePermissive" - case PolicyTypeRestrictive: - return "RowPolicyTypeRestrictive" - default: - return "RowPolicyTypeUnknown" - } -} - -type PolicyAction int32 +type PolicyAction = milvuspb.RowPolicyAction const ( - PolicyActionQuery PolicyAction = 0 - PolicyActionSearch PolicyAction = 1 - PolicyActionInsert PolicyAction = 2 - PolicyActionDelete PolicyAction = 3 - PolicyActionUpsert PolicyAction = 4 - PolicyActionQueryIterator PolicyAction = 5 - PolicyActionSearchIterator PolicyAction = 6 - PolicyActionHybridSearch PolicyAction = 7 + PolicyActionQuery = milvuspb.RowPolicyAction_Query + PolicyActionSearch = milvuspb.RowPolicyAction_Search + PolicyActionInsert = milvuspb.RowPolicyAction_Insert + PolicyActionDelete = milvuspb.RowPolicyAction_Delete + PolicyActionUpsert = milvuspb.RowPolicyAction_Upsert + PolicyActionQueryIterator = milvuspb.RowPolicyAction_QueryIterator + PolicyActionSearchIterator = milvuspb.RowPolicyAction_SearchIterator + PolicyActionHybridSearch = milvuspb.RowPolicyAction_HybridSearch ) -func (action PolicyAction) String() string { - switch action { - case PolicyActionQuery: - return "Query" - case PolicyActionSearch: - return "Search" - case PolicyActionInsert: - return "Insert" - case PolicyActionDelete: - return "Delete" - case PolicyActionUpsert: - return "Upsert" - case PolicyActionQueryIterator: - return "QueryIterator" - case PolicyActionSearchIterator: - return "SearchIterator" - case PolicyActionHybridSearch: - return "HybridSearch" - default: - return "Unknown" - } -} - func PolicyActionOperation(action PolicyAction) string { switch action { case PolicyActionQuery: @@ -522,51 +489,9 @@ func (request *CreateRowPolicyRequest) GetDescription() string { type UpdateRowPolicyRequest = CreateRowPolicyRequest -type DropRowPolicyRequest struct { - DbName string - CollectionName string - PolicyName string -} +type DropRowPolicyRequest = milvuspb.DropRowPolicyRequest -func (request *DropRowPolicyRequest) GetDbName() string { - if request == nil { - return "" - } - return request.DbName -} - -func (request *DropRowPolicyRequest) GetCollectionName() string { - if request == nil { - return "" - } - return request.CollectionName -} - -func (request *DropRowPolicyRequest) GetPolicyName() string { - if request == nil { - return "" - } - return request.PolicyName -} - -type ListRowPoliciesRequest struct { - DbName string - CollectionName string -} - -func (request *ListRowPoliciesRequest) GetDbName() string { - if request == nil { - return "" - } - return request.DbName -} - -func (request *ListRowPoliciesRequest) GetCollectionName() string { - if request == nil { - return "" - } - return request.CollectionName -} +type ListRowPoliciesRequest = milvuspb.ListRowPoliciesRequest type RowPolicy struct { PolicyName string @@ -655,32 +580,7 @@ func (request *SetRLSPrincipalTagsRequest) GetTags() map[string]TagValue { return request.Tags } -type GetRLSPrincipalTagsRequest struct { - DbName string - CollectionName string - PrincipalName string -} - -func (request *GetRLSPrincipalTagsRequest) GetDbName() string { - if request == nil { - return "" - } - return request.DbName -} - -func (request *GetRLSPrincipalTagsRequest) GetCollectionName() string { - if request == nil { - return "" - } - return request.CollectionName -} - -func (request *GetRLSPrincipalTagsRequest) GetPrincipalName() string { - if request == nil { - return "" - } - return request.PrincipalName -} +type GetRLSPrincipalTagsRequest = milvuspb.GetRLSPrincipalTagsRequest type GetRLSPrincipalTagsResponse struct { Status *commonpb.Status @@ -690,63 +590,8 @@ type GetRLSPrincipalTagsResponse struct { PrincipalName string } -type ListRLSPrincipalsRequest struct { - DbName string - CollectionName string -} +type ListRLSPrincipalsRequest = milvuspb.ListRLSPrincipalsRequest -func (request *ListRLSPrincipalsRequest) GetDbName() string { - if request == nil { - return "" - } - return request.DbName -} +type ListRLSPrincipalsResponse = milvuspb.ListRLSPrincipalsResponse -func (request *ListRLSPrincipalsRequest) GetCollectionName() string { - if request == nil { - return "" - } - return request.CollectionName -} - -type ListRLSPrincipalsResponse struct { - Status *commonpb.Status - PrincipalNames []string - DbName string - CollectionName string -} - -type DeleteRLSPrincipalTagsRequest struct { - DbName string - CollectionName string - PrincipalName string - TagKeys []string -} - -func (request *DeleteRLSPrincipalTagsRequest) GetDbName() string { - if request == nil { - return "" - } - return request.DbName -} - -func (request *DeleteRLSPrincipalTagsRequest) GetCollectionName() string { - if request == nil { - return "" - } - return request.CollectionName -} - -func (request *DeleteRLSPrincipalTagsRequest) GetPrincipalName() string { - if request == nil { - return "" - } - return request.PrincipalName -} - -func (request *DeleteRLSPrincipalTagsRequest) GetTagKeys() []string { - if request == nil { - return nil - } - return request.TagKeys -} +type DeleteRLSPrincipalTagsRequest = milvuspb.DeleteRLSPrincipalTagsRequest diff --git a/internal/util/rlsutil/validation.go b/internal/util/rlsutil/validation.go index 40a1c59316..13f7e0afa7 100644 --- a/internal/util/rlsutil/validation.go +++ b/internal/util/rlsutil/validation.go @@ -81,7 +81,7 @@ func ValidatePolicyRoles(roles []string) error { return nil } -// ValidatePolicyActionCount bounds the raw action list before conversion. +// ValidatePolicyActionCount bounds the raw action list before validation. func ValidatePolicyActionCount(actionCount int) error { if actionCount == 0 { return merr.WrapErrParameterInvalidMsg("RLS policy actions is empty") @@ -138,7 +138,7 @@ func validatePolicy(policyName string, policyType PolicyType, actions []PolicyAc switch policyType { case PolicyTypePermissive, PolicyTypeRestrictive: default: - return merr.WrapErrParameterInvalidMsg("invalid RLS policy type: %s", policyType.String()) + return merr.WrapErrParameterInvalidMsg("invalid RLS policy type: %s", PolicyTypeUnknown.String()) } if err := ValidatePolicyActionCount(len(actions)); err != nil { return err @@ -181,7 +181,7 @@ func validatePolicy(policyName string, policyType PolicyType, actions []PolicyAc needUsingExpr = true needCheckExpr = true default: - return merr.WrapErrParameterInvalidMsg("invalid RLS policy action: %s", action.String()) + return merr.WrapErrParameterInvalidMsg("invalid RLS policy action: Unknown") } } if needUsingExpr && usingExprEmpty { diff --git a/internal/util/rlsutil/validation_test.go b/internal/util/rlsutil/validation_test.go index 7c38dbc47e..d5ce8fefaa 100644 --- a/internal/util/rlsutil/validation_test.go +++ b/internal/util/rlsutil/validation_test.go @@ -46,6 +46,28 @@ func arrayTagValueForTest(values []TagValue) (TagValue, error) { return TagValue{Kind: TagValueKindArray, arrayValue: array}, nil } +func TestPolicyEnumCompatibility(t *testing.T) { + paramtable.Init() + for _, policyType := range []PolicyType{PolicyTypeUnknown, -1, 100} { + err := ValidateStoredPolicy("policy", policyType, []PolicyAction{PolicyActionQuery}, "true", "") + require.ErrorIs(t, err, merr.ErrParameterInvalid) + require.Contains(t, err.Error(), "invalid RLS policy type: RowPolicyTypeUnknown") + } + for _, action := range []PolicyAction{-1, 100} { + err := ValidateStoredPolicy("policy", PolicyTypePermissive, []PolicyAction{action}, "true", "") + require.ErrorIs(t, err, merr.ErrParameterInvalid) + require.Contains(t, err.Error(), "invalid RLS policy action: Unknown") + require.Equal(t, "unknown", PolicyActionOperation(action)) + } + for action, operation := range map[PolicyAction]string{ + PolicyActionQuery: "query", PolicyActionSearch: "search", PolicyActionInsert: "insert", + PolicyActionDelete: "delete", PolicyActionUpsert: "upsert", PolicyActionQueryIterator: "query iterator", + PolicyActionSearchIterator: "search iterator", PolicyActionHybridSearch: "hybrid search", + } { + require.Equal(t, operation, PolicyActionOperation(action)) + } +} + func TestValidatePayloadBounds(t *testing.T) { paramtable.Init() diff --git a/pkg/common/common.go b/pkg/common/common.go index a7ccae14c7..5595621ad3 100644 --- a/pkg/common/common.go +++ b/pkg/common/common.go @@ -793,28 +793,6 @@ func ValidateNamespaceShardingEnabledNotAltered(properties []*commonpb.KeyValueP return nil } -// ValidateRLSEnabledNotAltered rejects attempts to update or delete -// rls.enabled after collection creation. -func ValidateRLSEnabledNotAltered(properties []*commonpb.KeyValuePair, deleteKeys []string) error { - for _, property := range properties { - if property.GetKey() == RLSEnabledKey { - return merr.WrapErrParameterInvalidMsg("cannot alter %s after collection creation", RLSEnabledKey) - } - if strings.EqualFold(property.GetKey(), RLSEnabledKey) { - return merr.WrapErrParameterInvalidMsg("invalid property key %q, did you mean %q?", property.GetKey(), RLSEnabledKey) - } - } - for _, key := range deleteKeys { - if key == RLSEnabledKey { - return merr.WrapErrParameterInvalidMsg("cannot delete %s after collection creation", RLSEnabledKey) - } - if strings.EqualFold(key, RLSEnabledKey) { - return merr.WrapErrParameterInvalidMsg("invalid property key %q, did you mean %q?", key, RLSEnabledKey) - } - } - return nil -} - // IsRLSEnabled extracts rls.enabled from collection properties. // Returns false if not set. func IsRLSEnabled(kvs ...*commonpb.KeyValuePair) (bool, error) { diff --git a/pkg/common/common_test.go b/pkg/common/common_test.go index c12cbd1255..c7ab2f5e75 100644 --- a/pkg/common/common_test.go +++ b/pkg/common/common_test.go @@ -688,35 +688,6 @@ func TestNamespaceShardingEnabled(t *testing.T) { } func TestRLSEnabled(t *testing.T) { - t.Run("rls.enabled is immutable after creation", func(t *testing.T) { - for _, value := range []string{"true", "false", "True", "0"} { - err := ValidateRLSEnabledNotAltered( - []*commonpb.KeyValuePair{{Key: RLSEnabledKey, Value: value}}, - nil, - ) - assert.ErrorIs(t, err, merr.ErrParameterInvalid) - } - assert.ErrorIs(t, ValidateRLSEnabledNotAltered(nil, []string{RLSEnabledKey}), merr.ErrParameterInvalid) - }) - - t.Run("rls.enabled immutable validation rejects wrong case", func(t *testing.T) { - for _, key := range []string{"RLS.ENABLED", "Rls.Enabled"} { - err := ValidateRLSEnabledNotAltered( - []*commonpb.KeyValuePair{{Key: key, Value: "true"}}, - nil, - ) - assert.ErrorContains(t, err, "did you mean") - assert.ErrorContains(t, ValidateRLSEnabledNotAltered(nil, []string{key}), "did you mean") - } - }) - - t.Run("rls.enabled immutable validation accepts unrelated changes", func(t *testing.T) { - assert.NoError(t, ValidateRLSEnabledNotAltered( - []*commonpb.KeyValuePair{{Key: RLSForceKey, Value: "true"}}, - []string{"other.key"}, - )) - }) - t.Run("returns value when set", func(t *testing.T) { enabled, err := IsRLSEnabled(&commonpb.KeyValuePair{Key: RLSEnabledKey, Value: "true"}) assert.NoError(t, err)