[Performance] Cache SCIM and permission-assignment list calls to reduce API requests during plan

For large IAM deployments (thousands of databricks_group_member,
databricks_group, databricks_user, and mws_permission_assignment
resources), terraform plan was issuing one API call per resource Read —
resulting in thousands of redundant requests and 429 rate-limit errors.

Each resource type now uses a shared in-memory cache backed by
sync.RWMutex (concurrent warm reads) + singleflight (deduplicated
cold-cache fetches). A single list call populates the cache for all
resources of that type; subsequent reads are served from memory.
Mutations (Create/Update/Delete) invalidate the relevant cache entry.

Changes:
- access/resource_permission_assignment.go: permAssignmentCache
  (1 List call per host instead of 1 per resource)
- mws/resource_mws_permission_assignment.go: workspaceAssignmentsCache
  (1 ListByWorkspaceId per workspace_id instead of 1 per resource)
- scim/resource_group.go + scim/groups.go: groupsListCache
  (1 GET /Groups?attributes=id,... instead of N GET /Groups/{id})
- scim/resource_user.go + scim/users.go: usersListCache
  (1 GET /Users?attributes=id,... instead of N GET /Users/{id})
- scim/resource_group_member.go: bulk groupCache
  (1 GET /Groups?attributes=id,members instead of N per-group reads,
  eliminating concurrent request storms that caused 429 errors)

Author: Tagir Bakirov

access/resource_permission_assignment.go

@@ -8,6 +8,9 @@ import (
 	"net/http"
 	"slices"
 	"strconv"
+	"sync"
+
+	"golang.org/x/sync/singleflight"
 
 	"github.com/databricks/databricks-sdk-go/apierr"
 	"github.com/databricks/terraform-provider-databricks/common"
@@ -18,6 +21,98 @@ func NewPermissionAssignmentAPI(ctx context.Context, m any) PermissionAssignment
 	return PermissionAssignmentAPI{m.(*common.DatabricksClient), ctx}
 }
 
+// permAssignmentEntry holds the cached list result for one workspace.
+type permAssignmentEntry struct {
+	mu          sync.RWMutex
+	initialized bool
+	list        permissionAssignmentResponse
+	// sg deduplicates concurrent in-flight API calls when the cache is cold.
+	// All goroutines that arrive while a fetch is in-flight join the same call
+	// and are woken simultaneously when it completes, rather than serialising
+	// through a write lock.
+	sg singleflight.Group
+}
+
+// permAssignmentCache caches the permission assignments list per workspace host
+// so that N databricks_permission_assignment resources in the same workspace
+// only issue a single API call during a terraform plan/apply cycle.
+type permAssignmentCache struct {
+	mu    sync.Mutex
+	cache map[string]*permAssignmentEntry
+}
+
+func newPermAssignmentCache() *permAssignmentCache {
+	return &permAssignmentCache{
+		cache: make(map[string]*permAssignmentEntry),
+	}
+}
+
+func (c *permAssignmentCache) getOrCreate(host string) *permAssignmentEntry {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if entry, ok := c.cache[host]; ok {
+		return entry
+	}
+	entry := &permAssignmentEntry{}
+	c.cache[host] = entry
+	return entry
+}
+
+func (c *permAssignmentCache) list(api PermissionAssignmentAPI) (permissionAssignmentResponse, error) {
+	host := api.client.Config.Host
+	entry := c.getOrCreate(host)
+
+	// Fast path: warm cache. Many goroutines can hold a read-lock simultaneously.
+	entry.mu.RLock()
+	if entry.initialized {
+		l := entry.list
+		entry.mu.RUnlock()
+		return l, nil
+	}
+	entry.mu.RUnlock()
+
+	// Slow path: cache is cold. Use singleflight so exactly one API call is made
+	// regardless of how many goroutines arrive concurrently; all share the result.
+	v, err, _ := entry.sg.Do("fetch", func() (interface{}, error) {
+		// Double-check now that we are the singleflight leader.
+		entry.mu.RLock()
+		if entry.initialized {
+			l := entry.list
+			entry.mu.RUnlock()
+			return &l, nil
+		}
+		entry.mu.RUnlock()
+
+		l, err := api.List()
+		if err != nil {
+			return nil, err
+		}
+		entry.mu.Lock()
+		entry.list = l
+		entry.initialized = true
+		entry.mu.Unlock()
+		return &l, nil
+	})
+	if err != nil {
+		return permissionAssignmentResponse{}, err
+	}
+	return *v.(*permissionAssignmentResponse), nil
+}
+
+func (c *permAssignmentCache) invalidate(host string) {
+	c.mu.Lock()
+	entry, ok := c.cache[host]
+	c.mu.Unlock()
+	if ok {
+		entry.mu.Lock()
+		entry.initialized = false
+		entry.list = permissionAssignmentResponse{}
+		entry.mu.Unlock()
+	}
+}
+
+var globalPermAssignmentCache = newPermAssignmentCache()
+
 type PermissionAssignmentAPI struct {
 	client  *common.DatabricksClient
 	context context.Context
@@ -156,6 +251,7 @@ func ResourcePermissionAssignment() common.Resource {
 			if err != nil {
 				return err
 			}
+			defer globalPermAssignmentCache.invalidate(c.Config.Host)
 			var assignment permissionAssignmentEntity
 			common.DataToStructPointer(d, s, &assignment)
 			api := NewPermissionAssignmentAPI(ctx, c)
@@ -188,7 +284,7 @@ func ResourcePermissionAssignment() common.Resource {
 			if err != nil {
 				return err
 			}
-			list, err := NewPermissionAssignmentAPI(ctx, c).List()
+			list, err := globalPermAssignmentCache.list(NewPermissionAssignmentAPI(ctx, c))
 			if err != nil {
 				return err
 			}
@@ -203,6 +299,7 @@ func ResourcePermissionAssignment() common.Resource {
 			if err != nil {
 				return err
 			}
+			defer globalPermAssignmentCache.invalidate(c.Config.Host)
 			var assignment permissionAssignmentEntity
 			common.DataToStructPointer(d, s, &assignment)
 			api := NewPermissionAssignmentAPI(ctx, c)
@@ -214,6 +311,7 @@ func ResourcePermissionAssignment() common.Resource {
 			if err != nil {
 				return err
 			}
+			defer globalPermAssignmentCache.invalidate(c.Config.Host)
 			return NewPermissionAssignmentAPI(ctx, c).Remove(d.Id())
 		},
 	}

mws/resource_mws_permission_assignment.go

@@ -3,13 +3,107 @@ package mws
 import (
 	"context"
 	"fmt"
+	"sync"
+
+	"golang.org/x/sync/singleflight"
 
 	"github.com/databricks/databricks-sdk-go/apierr"
 	"github.com/databricks/databricks-sdk-go/service/iam"
 	"github.com/databricks/terraform-provider-databricks/common"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 )
 
+// workspaceAssignmentsEntry holds the cached ListByWorkspaceId result for one workspace.
+type workspaceAssignmentsEntry struct {
+	mu          sync.RWMutex
+	initialized bool
+	list        *iam.PermissionAssignments
+	// sg deduplicates concurrent in-flight API calls when the cache is cold.
+	// All goroutines that arrive while a fetch is in-flight join the same call
+	// and are woken simultaneously when it completes, rather than serialising
+	// through a write lock.
+	sg singleflight.Group
+}
+
+// workspaceAssignmentsCache caches ListByWorkspaceId results per workspace ID so
+// that N databricks_mws_permission_assignment resources sharing the same
+// workspace_id only issue a single API call during a terraform plan/apply cycle.
+type workspaceAssignmentsCache struct {
+	mu    sync.Mutex
+	cache map[int64]*workspaceAssignmentsEntry
+}
+
+func newWorkspaceAssignmentsCache() *workspaceAssignmentsCache {
+	return &workspaceAssignmentsCache{
+		cache: make(map[int64]*workspaceAssignmentsEntry),
+	}
+}
+
+func (c *workspaceAssignmentsCache) getOrCreate(workspaceId int64) *workspaceAssignmentsEntry {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if entry, ok := c.cache[workspaceId]; ok {
+		return entry
+	}
+	entry := &workspaceAssignmentsEntry{}
+	c.cache[workspaceId] = entry
+	return entry
+}
+
+func (c *workspaceAssignmentsCache) list(ctx context.Context, api iam.WorkspaceAssignmentInterface, workspaceId int64) (*iam.PermissionAssignments, error) {
+	entry := c.getOrCreate(workspaceId)
+
+	// Fast path: warm cache. Many goroutines can hold a read-lock simultaneously.
+	entry.mu.RLock()
+	if entry.initialized {
+		l := entry.list
+		entry.mu.RUnlock()
+		return l, nil
+	}
+	entry.mu.RUnlock()
+
+	// Slow path: cache is cold. Use singleflight so exactly one API call is made
+	// regardless of how many goroutines arrive concurrently; all share the result.
+	v, err, _ := entry.sg.Do("fetch", func() (interface{}, error) {
+		// Double-check now that we are the singleflight leader.
+		entry.mu.RLock()
+		if entry.initialized {
+			l := entry.list
+			entry.mu.RUnlock()
+			return l, nil
+		}
+		entry.mu.RUnlock()
+
+		list, err := api.ListByWorkspaceId(ctx, workspaceId)
+		if err != nil {
+			return nil, err
+		}
+		entry.mu.Lock()
+		entry.list = list
+		entry.initialized = true
+		entry.mu.Unlock()
+		return list, nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return v.(*iam.PermissionAssignments), nil
+}
+
+func (c *workspaceAssignmentsCache) invalidate(workspaceId int64) {
+	c.mu.Lock()
+	entry, ok := c.cache[workspaceId]
+	c.mu.Unlock()
+	if ok {
+		entry.mu.Lock()
+		entry.initialized = false
+		entry.list = nil
+		entry.mu.Unlock()
+	}
+}
+
+var globalWorkspaceAssignmentsCache = newWorkspaceAssignmentsCache()
+
 func getPermissionsByPrincipal(list iam.PermissionAssignments, principalId int64) (res iam.UpdateWorkspaceAssignments, err error) {
 	for _, v := range list.PermissionAssignments {
 		if v.Principal.PrincipalId != principalId {
@@ -56,6 +150,7 @@ func ResourceMwsPermissionAssignment() common.Resource {
 			if err != nil {
 				return err
 			}
+			globalWorkspaceAssignmentsCache.invalidate(assignment.WorkspaceId)
 			pair.Pack(d)
 			return nil
 		},
@@ -68,7 +163,7 @@ func ResourceMwsPermissionAssignment() common.Resource {
 			if err != nil {
 				return fmt.Errorf("parse id: %w", err)
 			}
-			list, err := acc.WorkspaceAssignment.ListByWorkspaceId(ctx, common.MustInt64(workspaceId))
+			list, err := globalWorkspaceAssignmentsCache.list(ctx, acc.WorkspaceAssignment, common.MustInt64(workspaceId))
 			if err != nil {
 				return err
 			}
@@ -89,7 +184,9 @@ func ResourceMwsPermissionAssignment() common.Resource {
 			if err != nil {
 				return fmt.Errorf("parse id: %w", err)
 			}
-			return acc.WorkspaceAssignment.DeleteByWorkspaceIdAndPrincipalId(ctx, common.MustInt64(workspaceId), common.MustInt64(principalId))
+			err = acc.WorkspaceAssignment.DeleteByWorkspaceIdAndPrincipalId(ctx, common.MustInt64(workspaceId), common.MustInt64(principalId))
+			globalWorkspaceAssignmentsCache.invalidate(common.MustInt64(workspaceId))
+			return err
 		},
 	}
 }

mws/resource_mws_permission_assignment_test.go

@@ -10,6 +10,7 @@ import (
 )
 
 func TestPermissionAssignmentCreate(t *testing.T) {
+	globalWorkspaceAssignmentsCache = newWorkspaceAssignmentsCache()
 	qa.ResourceFixture{
 		MockAccountClientFunc: func(m *mocks.MockAccountClient) {
 			e := m.GetMockWorkspaceAssignmentAPI().EXPECT()
@@ -46,6 +47,7 @@ func TestPermissionAssignmentCreate(t *testing.T) {
 }
 
 func TestPermissionAssignmentRead(t *testing.T) {
+	globalWorkspaceAssignmentsCache = newWorkspaceAssignmentsCache()
 	qa.ResourceFixture{
 		MockAccountClientFunc: func(m *mocks.MockAccountClient) {
 			e := m.GetMockWorkspaceAssignmentAPI().EXPECT()
@@ -79,6 +81,7 @@ func TestPermissionAssignmentRead(t *testing.T) {
 }
 
 func TestPermissionAssignmentReadNotFound(t *testing.T) {
+	globalWorkspaceAssignmentsCache = newWorkspaceAssignmentsCache()
 	qa.ResourceFixture{
 		MockAccountClientFunc: func(m *mocks.MockAccountClient) {
 			e := m.GetMockWorkspaceAssignmentAPI().EXPECT()
@@ -102,6 +105,7 @@ func TestPermissionAssignmentReadNotFound(t *testing.T) {
 }
 
 func TestPermissionAssignmentDelete(t *testing.T) {
+	globalWorkspaceAssignmentsCache = newWorkspaceAssignmentsCache()
 	qa.ResourceFixture{
 		MockAccountClientFunc: func(m *mocks.MockAccountClient) {
 			e := m.GetMockWorkspaceAssignmentAPI().EXPECT()
@@ -115,19 +119,22 @@ func TestPermissionAssignmentDelete(t *testing.T) {
 }
 
 func TestPermissionAssignmentFuzz_NoAccountID(t *testing.T) {
+	globalWorkspaceAssignmentsCache = newWorkspaceAssignmentsCache()
 	qa.ResourceCornerCases(t, ResourceMwsPermissionAssignment(),
 		qa.CornerCaseID("123|456"),
 		qa.CornerCaseExpectError("invalid Databricks Account configuration"))
 }
 
 func TestPermissionAssignmentFuzz_InvalidID(t *testing.T) {
+	globalWorkspaceAssignmentsCache = newWorkspaceAssignmentsCache()
 	qa.ResourceCornerCases(t, ResourceMwsPermissionAssignment(),
 		qa.CornerCaseExpectError("parse id: invalid ID: x"),
 		qa.CornerCaseSkipCRUD("create"),
 		qa.CornerCaseAccountID("abc"))
 }
 
 func TestPermissionAssignmentFuzz_ApiErrors(t *testing.T) {
+	globalWorkspaceAssignmentsCache = newWorkspaceAssignmentsCache()
 	qa.ResourceCornerCases(t, ResourceMwsPermissionAssignment(),
 		qa.CornerCaseAccountID("abc"),
 		qa.CornerCaseID("123|456"))

scim/groups.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"strconv"
 
 	"github.com/databricks/terraform-provider-databricks/common"
 )
@@ -55,6 +56,32 @@ func (a GroupsAPI) Filter(filter string, attributes string) (GroupList, error) {
 	return groups, err
 }
 
+// ListAll retrieves all groups with the given attributes, handling SCIM pagination.
+func (a GroupsAPI) ListAll(attributes string) ([]Group, error) {
+	startIndex := 1
+	var result []Group
+	for {
+		req := map[string]string{
+			"count":      "10000",
+			"startIndex": strconv.Itoa(startIndex),
+		}
+		if attributes != "" {
+			req["attributes"] = attributes
+		}
+		var page GroupList
+		err := a.client.Scim(a.context, http.MethodGet, "/preview/scim/v2/Groups", req, &page, a.ApiLevel)
+		if err != nil {
+			return nil, err
+		}
+		result = append(result, page.Resources...)
+		if len(page.Resources) == 0 || int32(len(result)) >= page.TotalResults {
+			break
+		}
+		startIndex += len(page.Resources)
+	}
+	return result, nil
+}
+
 func (a GroupsAPI) ReadByDisplayName(displayName, attributes string) (group Group, err error) {
 	groupList, err := a.Filter(fmt.Sprintf(`displayName eq "%s"`, displayName), attributes)
 	if err != nil {