diff --git a/provider/pkg/gen/schema.go b/provider/pkg/gen/schema.go
index 6dbf43e318bb..6abfa1c7a28d 100644
--- a/provider/pkg/gen/schema.go
+++ b/provider/pkg/gen/schema.go
@@ -872,6 +872,7 @@ func (g *packageGenerator) genResourceVariant(apiSpec *openapi.ResourceSpec, res
 		RequiredContainers:   requiredContainers,
 		DefaultProperties:    propertyDefaults(module, resource.typeName),
 	}
+
 	g.metadata.Resources[resourceTok] = r
 
 	g.generateExampleReferences(resourceTok, path, swagger)
diff --git a/provider/pkg/provider/provider.go b/provider/pkg/provider/provider.go
index 6317f605def6..1ec12d6976f7 100644
--- a/provider/pkg/provider/provider.go
+++ b/provider/pkg/provider/provider.go
@@ -81,6 +81,7 @@ type azureNativeProvider struct {
 	converter       *convert.SdkShapeConverter
 	customResources map[string]*resources.CustomResource
 	rgLocationMap   map[string]string
+	lookupType      resources.TypeLookupFunc
 }
 
 func makeProvider(host *provider.HostClient, name, version string, schemaBytes []byte,
@@ -101,7 +102,7 @@ func makeProvider(host *provider.HostClient, name, version string, schemaBytes [
 	converter := convert.NewSdkShapeConverterPartial(resourceMap.Types)
 
 	// Return the new provider
-	return &azureNativeProvider{
+	p := &azureNativeProvider{
 		host:          host,
 		name:          name,
 		version:       version,
@@ -111,7 +112,9 @@ func makeProvider(host *provider.HostClient, name, version string, schemaBytes [
 		schemaBytes:   schemaBytes,
 		converter:     &converter,
 		rgLocationMap: map[string]string{},
-	}, nil
+	}
+	p.lookupType = p.lookupTypeDefault
+	return p, nil
 }
 
 // LoadMetadataPartial partially deserializes the provided json byte array into an AzureAPIMetadata
@@ -147,6 +150,14 @@ func (k *azureNativeProvider) getFullMetadata() (*resources.AzureAPIMetadata, er
 	return k.fullResourceMap, nil
 }
 
+// Looks up a type reference, with or without the #/types/ prefix, in the resource map.
+// Typically used via lookupType so it can be overridden for tests.
+func (k *azureNativeProvider) lookupTypeDefault(ref string) (*resources.AzureAPIType, bool, error) {
+	typeName := strings.TrimPrefix(ref, "#/types/")
+	t, ok, err := k.resourceMap.Types.Get(typeName)
+	return &t, ok, err
+}
+
 func (p *azureNativeProvider) Attach(context context.Context, req *rpc.PluginAttach) (*emptypb.Empty, error) {
 	host, err := provider.NewHostClient(req.GetAddress())
 	if err != nil {
@@ -623,7 +634,7 @@ func (k *azureNativeProvider) validateProperty(ctx string, prop *resources.Azure
 
 		// Typed object: validate all properties by looking up its type definition.
 		typeName := strings.TrimPrefix(prop.Ref, "#/types/")
-		typ, ok, err := k.resourceMap.Types.Get(typeName)
+		typ, ok, err := k.lookupTypeDefault(prop.Ref)
 		if err != nil {
 			return append(failures, &rpc.CheckFailure{
 				Reason: fmt.Sprintf("error decoding type spec '%s': %v", typeName, err),
@@ -635,7 +646,7 @@ func (k *azureNativeProvider) validateProperty(ctx string, prop *resources.Azure
 			})
 		}
 
-		failures = append(failures, k.validateType(ctx, &typ, value)...)
+		failures = append(failures, k.validateType(ctx, typ, value)...)
 	case []interface{}:
 		if prop.Type != "array" && !prop.IsStringSet {
 			return append(failures, &rpc.CheckFailure{
@@ -861,7 +872,7 @@ func (k *azureNativeProvider) Create(ctx context.Context, req *rpc.CreateRequest
 		}, nil
 	}
 
-	// Construct ARM REST API body and query from intputs
+	// Construct ARM REST API body and query from inputs
 	id, bodyParams, queryParams, err := k.prepareAzureRESTInputs(
 		res.Path,
 		res.PutParameters,
@@ -908,6 +919,8 @@ func (k *azureNativeProvider) Create(ctx context.Context, req *rpc.CreateRequest
 		ctx, cancel := azureContext(ctx, req.Timeout)
 		defer cancel()
 
+		k.setUnsetSubresourcePropertiesToDefaults(res, bodyParams)
+
 		// Submit the `PUT` against the ARM endpoint
 		response, created, err := k.azureCreateOrUpdate(ctx, id, bodyParams, queryParams, res.UpdateMethod, res.PutAsyncStyle)
 		if err != nil {
@@ -949,6 +962,44 @@ func (k *azureNativeProvider) Create(ctx context.Context, req *rpc.CreateRequest
 	}, nil
 }
 
+// Properties pointing to sub-resources that can be maintained as separate resources might not be
+// present in the inputs because the user wants to manage them as standalone resources. However,
+// auch a property might be required by Azure even if it's not annotated as such in the spec, e.g.,
+// Key Vault's accessPolicies. Therefore, we set these properties to their default value here,
+// an empty array.
+// During create, no sub-resources can exist yet so there's no danger of overwriting existing values.
+//
+// Implementation note: we should make it possible to write custom resources that call code from
+// the default implementation as needed. This would allow us to cleanly implement special logic
+// like for Key Vault into custom resources without duplicating much code. In the Key Vault case,
+// the custom Read() would look like
+//
+//	provider.azureCanCreate(ctx, id, &res)
+//	setUnsetSubresourcePropertiesToDefaults(res, bodyParams) // custom
+//	k.azureCreateOrUpdate
+//	...
+func (k *azureNativeProvider) setUnsetSubresourcePropertiesToDefaults(res resources.AzureAPIResource, bodyParams map[string]interface{}) {
+	unset := k.findUnsetPropertiesToMaintain(&res, bodyParams)
+	for _, p := range unset {
+		cur := bodyParams
+		for _, pathEl := range p.path[:len(p.path)-1] {
+			curObj, ok := cur[pathEl]
+			if !ok {
+				newContainer := map[string]any{}
+				cur[pathEl] = newContainer
+				cur = newContainer
+				continue
+			}
+			cur, ok = curObj.(map[string]any)
+			if !ok {
+				break
+			}
+		}
+
+		cur[p.path[len(p.path)-1]] = []any{}
+	}
+}
+
 // currentResourceStateCheckpoint reads the resource state by ID, converts it to outputs map, and
 // produces a checkpoint with these outputs and given inputs.
 func (k *azureNativeProvider) currentResourceStateCheckpoint(ctx context.Context, id string, res resources.AzureAPIResource, inputs resource.PropertyMap) (*structpb.Struct, error) {
@@ -1120,8 +1171,10 @@ func mappableOldState(res resources.AzureAPIResource, oldState resource.Property
 	return plainOldState
 }
 
+// removeUnsetSubResourceProperties removes sub-resource properties from new outputs which weren't set in the old inputs.
+// If the user didn't specify them inline originally, we don't want to push them into the inputs now.
 func (k *azureNativeProvider) removeUnsetSubResourceProperties(ctx context.Context, urn resource.URN, sdkResponse map[string]interface{}, oldInputs resource.PropertyMap, res *resources.AzureAPIResource) map[string]interface{} {
-	propertiesToRemove := findUnsetSubResourceProperties(res, oldInputs)
+	propertiesToRemove := k.findUnsetPropertiesToMaintain(res, oldInputs.Mappable())
 
 	if len(propertiesToRemove) == 0 {
 		return sdkResponse
@@ -1138,32 +1191,32 @@ func (k *azureNativeProvider) removeUnsetSubResourceProperties(ctx context.Conte
 	}
 
 	for _, prop := range propertiesToRemove {
-		delete(result, prop)
+		deleteFromMap(result, prop.path)
 	}
 	return result
 }
 
-func findUnsetSubResourceProperties(res *resources.AzureAPIResource, oldInputs resource.PropertyMap) []string {
-	var propertiesToRemove []string
-	for _, param := range res.PutParameters {
-		if param.Location != "body" || param.Body == nil {
-			continue
-		}
-		for propName, prop := range param.Body.Properties {
-			if !prop.MaintainSubResourceIfUnset {
-				continue
-			}
-			key := propName
-			if prop.SdkName != "" {
-				key = prop.SdkName
-			}
-			propKey := resource.PropertyKey(key)
-			if !oldInputs.HasValue(propKey) {
-				propertiesToRemove = append(propertiesToRemove, key)
+func deleteFromMap(m map[string]interface{}, path []string) bool {
+	container := m
+	for i, key := range path {
+		if i == len(path)-1 {
+			_, found := container[key]
+			if found {
+				delete(container, key)
 			}
+			return found
+		}
+
+		value, ok := container[key]
+		if !ok {
+			return false
+		}
+		container, ok = value.(map[string]interface{})
+		if !ok {
+			return false
 		}
 	}
-	return propertiesToRemove
+	return false
 }
 
 // Update updates an existing resource with new values.
@@ -1398,9 +1451,14 @@ func (k *azureNativeProvider) Cancel(context.Context, *pbempty.Empty) (*pbempty.
 	return &pbempty.Empty{}, nil
 }
 
+type propertyPath struct {
+	path         []string
+	propertyName string
+}
+
 func (k *azureNativeProvider) maintainSubResourcePropertiesIfNotSet(ctx context.Context, res *resources.AzureAPIResource, id string, bodyParams map[string]interface{}) error {
 	// Identify the properties we need to read
-	missingProperties := findSubResourcePropertiesToMaintain(res, bodyParams)
+	missingProperties := k.findUnsetPropertiesToMaintain(res, bodyParams)
 
 	if len(missingProperties) == 0 {
 		// Everything's already specified explicitly by the user, no need to do read.
@@ -1421,92 +1479,71 @@ func (k *azureNativeProvider) maintainSubResourcePropertiesIfNotSet(ctx context.
 	return nil
 }
 
-func writePropertiesToBody(missingProperties map[string]resources.AzureAPIProperty, bodyParams map[string]interface{}, responseBody map[string]interface{}) map[string]interface{} {
+func writePropertiesToBody(missingProperties []propertyPath, bodyParams map[string]interface{}, remoteState map[string]interface{}) map[string]interface{} {
 	writtenProperties := map[string]interface{}{}
-	for propName, prop := range missingProperties {
+	for _, prop := range missingProperties {
 		currentBodyContainer := bodyParams
-		currentResponseContainer := responseBody
-		for _, containerName := range prop.Containers {
+		currentStateContainer := remoteState
+		for _, containerName := range prop.path {
 			innerBodyContainer, bodyOk := currentBodyContainer[containerName]
-			innerStateContainer, stateOk := currentResponseContainer[containerName]
+			innerStateContainer, stateOk := currentStateContainer[containerName]
 			if !bodyOk {
 				innerBodyContainer = map[string]interface{}{}
 				currentBodyContainer[containerName] = innerBodyContainer
 			}
 			if !stateOk {
 				innerStateContainer = map[string]interface{}{}
-				currentResponseContainer[containerName] = innerStateContainer
+				currentStateContainer[containerName] = innerStateContainer
+			}
+			innerBodyObj, innerBodyIsObject := innerBodyContainer.(map[string]interface{})
+			innerStateObj, innerStateIsObject := innerStateContainer.(map[string]interface{})
+			if !innerBodyIsObject || !innerStateIsObject { // we've reached a leaf node (primitive type)
+				break
 			}
-			currentBodyContainer = innerBodyContainer.(map[string]interface{})
-			currentResponseContainer = innerStateContainer.(map[string]interface{})
+			currentBodyContainer = innerBodyObj
+			currentStateContainer = innerStateObj
 		}
-		responseValue, ok := currentResponseContainer[propName]
+
+		stateValue, ok := currentStateContainer[prop.propertyName]
 		if ok {
-			currentBodyContainer[propName] = responseValue
-			writtenProperties[fmt.Sprintf("%s.%s", strings.Join(prop.Containers, "."), propName)] = responseValue
+			currentBodyContainer[prop.propertyName] = stateValue
+			writtenProperties[fmt.Sprintf("%s.%s", strings.Join(prop.path, "."), prop.propertyName)] = stateValue
 		}
 	}
 	return writtenProperties
 }
 
-func findSubResourcePropertiesToMaintain(res *resources.AzureAPIResource, bodyParams map[string]interface{}) map[string]resources.AzureAPIProperty {
-	// Find all properties which might need to be read
-	subResourceProperties := findSubResourceProperties(res)
-
-	if len(subResourceProperties) == 0 {
-		// No properties to be read
-		return subResourceProperties
-	}
-	// Filter to only properties which the user also hasn't set
-	return findUnsetProperties(subResourceProperties, bodyParams)
-}
-
-func findUnsetProperties(candidateProperties map[string]resources.AzureAPIProperty, bodyParams map[string]interface{}) map[string]resources.AzureAPIProperty {
-	missingProperties := map[string]resources.AzureAPIProperty{}
-	for propName, prop := range candidateProperties {
-		currentContainer := bodyParams
-		containerNotFound := false
-		for _, containerName := range prop.Containers {
-			innerContainer, ok := currentContainer[containerName]
+func (k *azureNativeProvider) findUnsetPropertiesToMaintain(res *resources.AzureAPIResource, bodyParams map[string]interface{}) []propertyPath {
+	missingProperties := []propertyPath{}
+	for _, path := range res.PathsToSubResourcePropertiesToMaintain(true /* includeContainers i.e. API-shape */, k.lookupType) {
+		curBody := bodyParams
+		for i, pathEl := range path {
+			v, ok := curBody[pathEl]
 			if !ok {
-				containerNotFound = true
+				missingProperties = append(missingProperties, propertyPath{
+					path:         path,
+					propertyName: pathEl,
+				})
 				break
 			}
-			currentContainer, ok = innerContainer.(map[string]interface{})
-			if !ok {
-				containerNotFound = true
+
+			// At the end of the path we don't need to go deeper via references and map lookups.
+			if i == len(path)-1 {
 				break
 			}
-		}
-
-		if containerNotFound {
-			// If the containers are not found, the property is also not defined
-			missingProperties[propName] = prop
-			continue
-		}
-		if _, ok := currentContainer[propName]; !ok {
-			missingProperties[propName] = prop
-		}
-	}
-	return missingProperties
-}
 
-func findSubResourceProperties(res *resources.AzureAPIResource) map[string]resources.AzureAPIProperty {
-	subResourceProperties := map[string]resources.AzureAPIProperty{}
-	for _, param := range res.PutParameters {
-		if param.Location != "body" {
-			continue
-		}
-		if param.Body.Properties == nil {
-			continue
-		}
-		for propName, prop := range param.Body.Properties {
-			if prop.MaintainSubResourceIfUnset {
-				subResourceProperties[propName] = prop
+			curBody, ok = v.(map[string]interface{})
+			if !ok {
+				missingProperties = append(missingProperties, propertyPath{
+					path:         path,
+					propertyName: pathEl,
+				})
+				break
 			}
 		}
 	}
-	return subResourceProperties
+
+	return missingProperties
 }
 
 func (k *azureNativeProvider) azureCreateOrUpdate(
diff --git a/provider/pkg/provider/provider_test.go b/provider/pkg/provider/provider_test.go
index 17445dbcd5d1..046e83129d95 100644
--- a/provider/pkg/provider/provider_test.go
+++ b/provider/pkg/provider/provider_test.go
@@ -1,16 +1,18 @@
 package provider
 
 import (
+	"context"
 	"testing"
 
 	"github.com/pulumi/pulumi-azure-native/v2/provider/pkg/resources"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/resource"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestWritePropertiesToBody(t *testing.T) {
 	t.Run("empty", func(t *testing.T) {
-		missingProperties := map[string]resources.AzureAPIProperty{}
+		missingProperties := []propertyPath{}
 		bodyParams := map[string]interface{}{}
 		response := map[string]interface{}{}
 
@@ -19,11 +21,10 @@ func TestWritePropertiesToBody(t *testing.T) {
 	})
 
 	t.Run("top-level", func(t *testing.T) {
-		missingProperties := map[string]resources.AzureAPIProperty{
-			"remote": {
-				Type: "string",
-			},
-		}
+		missingProperties := []propertyPath{{
+			propertyName: "remote",
+			path:         []string{},
+		}}
 		bodyParams := map[string]interface{}{
 			"existing": "value",
 		}
@@ -40,12 +41,10 @@ func TestWritePropertiesToBody(t *testing.T) {
 	})
 
 	t.Run("properties container", func(t *testing.T) {
-		missingProperties := map[string]resources.AzureAPIProperty{
-			"remote": {
-				Type:       "string",
-				Containers: []string{"properties"},
-			},
-		}
+		missingProperties := []propertyPath{{
+			propertyName: "remote",
+			path:         []string{"properties"},
+		}}
 		bodyParams := map[string]interface{}{
 			"properties": map[string]interface{}{
 				"existing": "value",
@@ -66,12 +65,10 @@ func TestWritePropertiesToBody(t *testing.T) {
 	})
 
 	t.Run("existing properties are maintained", func(t *testing.T) {
-		missingProperties := map[string]resources.AzureAPIProperty{
-			"remote": {
-				Type:       "string",
-				Containers: []string{"properties"},
-			},
-		}
+		missingProperties := []propertyPath{{
+			propertyName: "remote",
+			path:         []string{"properties"},
+		}}
 		bodyParams := map[string]interface{}{
 			"properties": map[string]interface{}{
 				"existing": "value",
@@ -94,12 +91,10 @@ func TestWritePropertiesToBody(t *testing.T) {
 	})
 
 	t.Run("properties missed from remote", func(t *testing.T) {
-		missingProperties := map[string]resources.AzureAPIProperty{
-			"remote": {
-				Type:       "string",
-				Containers: []string{"properties"},
-			},
-		}
+		missingProperties := []propertyPath{{
+			propertyName: "remote",
+			path:         []string{"properties"},
+		}}
 		bodyParams := map[string]interface{}{
 			"properties": map[string]interface{}{
 				"existing": "value",
@@ -120,12 +115,10 @@ func TestWritePropertiesToBody(t *testing.T) {
 	})
 
 	t.Run("properties container missing from remote", func(t *testing.T) {
-		missingProperties := map[string]resources.AzureAPIProperty{
-			"remote": {
-				Type:       "string",
-				Containers: []string{"properties"},
-			},
-		}
+		missingProperties := []propertyPath{{
+			propertyName: "remote",
+			path:         []string{"properties"},
+		}}
 		bodyParams := map[string]interface{}{
 			"properties": map[string]interface{}{
 				"existing": "value",
@@ -142,12 +135,10 @@ func TestWritePropertiesToBody(t *testing.T) {
 	})
 
 	t.Run("properties container missing in body", func(t *testing.T) {
-		missingProperties := map[string]resources.AzureAPIProperty{
-			"remote": {
-				Type:       "string",
-				Containers: []string{"properties"},
-			},
-		}
+		missingProperties := []propertyPath{{
+			propertyName: "remote",
+			path:         []string{"properties"},
+		}}
 		bodyParams := map[string]interface{}{}
 		response := map[string]interface{}{
 			"properties": map[string]interface{}{
@@ -164,12 +155,10 @@ func TestWritePropertiesToBody(t *testing.T) {
 	})
 
 	t.Run("empty with container", func(t *testing.T) {
-		missingProperties := map[string]resources.AzureAPIProperty{
-			"remote": {
-				Type:       "string",
-				Containers: []string{"properties"},
-			},
-		}
+		missingProperties := []propertyPath{{
+			propertyName: "remote",
+			path:         []string{"properties"},
+		}}
 		bodyParams := map[string]interface{}{}
 		response := map[string]interface{}{}
 		writePropertiesToBody(missingProperties, bodyParams, response)
@@ -179,94 +168,28 @@ func TestWritePropertiesToBody(t *testing.T) {
 		}
 		assert.Equal(t, expected, bodyParams)
 	})
-}
-
-func TestFindSubResourcePropertiesToMaintain(t *testing.T) {
-	t.Run("empty", func(t *testing.T) {
-		res := &resources.AzureAPIResource{}
-		bodyParams := map[string]interface{}{}
-		actual := findSubResourcePropertiesToMaintain(res, bodyParams)
-		expected := map[string]resources.AzureAPIProperty{}
-		assert.Equal(t, expected, actual)
-	})
-
-	t.Run("single prop", func(t *testing.T) {
-		prop := resources.AzureAPIProperty{
-			Type:                       "string",
-			MaintainSubResourceIfUnset: true,
-		}
-		res := &resources.AzureAPIResource{
-			PutParameters: []resources.AzureAPIParameter{
-				{
-					Location: "body",
-					Body: &resources.AzureAPIType{
-						Properties: map[string]resources.AzureAPIProperty{
-							"prop": prop,
-						},
-					},
-				},
-			},
-		}
-		bodyParams := map[string]interface{}{}
-		actual := findSubResourcePropertiesToMaintain(res, bodyParams)
-		expected := map[string]resources.AzureAPIProperty{
-			"prop": prop,
-		}
-		assert.Equal(t, expected, actual)
-	})
-
-	t.Run("containered missing prop", func(t *testing.T) {
-		prop := resources.AzureAPIProperty{
-			Type:                       "string",
-			MaintainSubResourceIfUnset: true,
-			Containers:                 []string{"properties"},
-		}
-		res := &resources.AzureAPIResource{
-			PutParameters: []resources.AzureAPIParameter{
-				{
-					Location: "body",
-					Body: &resources.AzureAPIType{
-						Properties: map[string]resources.AzureAPIProperty{
-							"prop": prop,
-						},
-					},
-				},
-			},
-		}
-		bodyParams := map[string]interface{}{}
-		actual := findSubResourcePropertiesToMaintain(res, bodyParams)
-		expected := map[string]resources.AzureAPIProperty{
-			"prop": prop,
-		}
-		assert.Equal(t, expected, actual)
-	})
 
-	t.Run("containered existing prop", func(t *testing.T) {
-		prop := resources.AzureAPIProperty{
-			Type:                       "string",
-			MaintainSubResourceIfUnset: true,
-			Containers:                 []string{"properties"},
+	t.Run("Nested array missing from body", func(t *testing.T) {
+		missingProperties := []propertyPath{{
+			propertyName: "accessPolicies",
+			path:         []string{"properties", "accessPolicies"},
+		}}
+		bodyParams := map[string]interface{}{
+			"properties": map[string]interface{}{},
 		}
-		res := &resources.AzureAPIResource{
-			PutParameters: []resources.AzureAPIParameter{
-				{
-					Location: "body",
-					Body: &resources.AzureAPIType{
-						Properties: map[string]resources.AzureAPIProperty{
-							"prop": prop,
-						},
-					},
-				},
+		response := map[string]interface{}{
+			"properties": map[string]interface{}{
+				"accessPolicies": []interface{}{},
 			},
 		}
-		bodyParams := map[string]interface{}{
+		writePropertiesToBody(missingProperties, bodyParams, response)
+		expected := map[string]interface{}{
+			// Container is auto-created
 			"properties": map[string]interface{}{
-				"prop": "value",
+				"accessPolicies": []interface{}{},
 			},
 		}
-		actual := findSubResourcePropertiesToMaintain(res, bodyParams)
-		expected := map[string]resources.AzureAPIProperty{}
-		assert.Equal(t, expected, actual)
+		assert.Equal(t, expected, bodyParams)
 	})
 }
 
@@ -286,34 +209,64 @@ func TestFindUnsetSubResourceProperties(t *testing.T) {
 			},
 		},
 	}
+
+	provider := azureNativeProvider{}
+
 	t.Run("empty", func(t *testing.T) {
 		res := &resources.AzureAPIResource{}
-		oldInputs := resource.PropertyMap{}
-		actual := findUnsetSubResourceProperties(res, oldInputs)
-		var expected []string
+		oldInputs := map[string]any{}
+		actual := provider.findUnsetPropertiesToMaintain(res, oldInputs)
+		expected := []propertyPath{}
 		assert.Equal(t, expected, actual)
 	})
 
 	t.Run("sub-resource not set", func(t *testing.T) {
-		oldInputs := resource.PropertyMap{
-			"existing": resource.NewStringProperty("value"),
+		oldInputs := map[string]any{
+			"existing": "value",
 		}
-		actual := findUnsetSubResourceProperties(resWithSubResource, oldInputs)
-		expected := []string{"subResource"}
+		actual := provider.findUnsetPropertiesToMaintain(resWithSubResource, oldInputs)
+		expected := []propertyPath{{
+			propertyName: "subResource",
+			path:         []string{"subResource"},
+		}}
 		assert.Equal(t, expected, actual)
 	})
 
 	t.Run("sub-resource set", func(t *testing.T) {
-		oldInputs := resource.PropertyMap{
-			"existing":    resource.NewStringProperty("value"),
-			"subResource": resource.NewStringProperty("value"),
+		oldInputs := map[string]any{
+			"existing":    "value",
+			"subResource": "value",
 		}
-		actual := findUnsetSubResourceProperties(resWithSubResource, oldInputs)
-		var expected []string
+		actual := provider.findUnsetPropertiesToMaintain(resWithSubResource, oldInputs)
+		expected := []propertyPath{}
 		assert.Equal(t, expected, actual)
 	})
 }
 
+func TestFindUnsetSubResourcePropertiesFollowingTypeRefs(t *testing.T) {
+	res, provider := setUpResourceWithRefAndProviderWithTypeLookup()
+
+	t.Run("KV accessPolicies is not set", func(t *testing.T) {
+		bodyParams := map[string]interface{}{
+			"properties": map[string]interface{}{},
+		}
+		unset := provider.findUnsetPropertiesToMaintain(res, bodyParams)
+		require.Equal(t, 1, len(unset))
+		assert.Equal(t, "accessPolicies", unset[0].propertyName)
+		assert.Equal(t, []string{"properties", "accessPolicies"}, unset[0].path)
+	})
+
+	t.Run("KV accessPolicies is set", func(t *testing.T) {
+		bodyParams := map[string]interface{}{
+			"properties": map[string]interface{}{
+				"accessPolicies": []interface{}{},
+			},
+		}
+		unset := provider.findUnsetPropertiesToMaintain(res, bodyParams)
+		assert.Empty(t, unset)
+	})
+}
+
 func TestRestoreDefaultInputs(t *testing.T) {
 	inputs := resource.PropertyMap{
 		"unrelated": resource.NewStringProperty("foo"),
@@ -449,3 +402,170 @@ func TestMappableOldStatePreservesDefaultsThatWereNotInputs(t *testing.T) {
 	assert.Contains(t, m, "__inputs")
 	assert.Contains(t, m, "networkRuleSet")
 }
+
+func TestDeleteFromMap(t *testing.T) {
+	m := map[string]any{
+		"a": "scalar",
+		"b": map[string]any{
+			"b.a": "scalar",
+			"b.b": map[string]any{
+				"b.b.a": map[string]any{},
+			},
+		},
+	}
+
+	deleted := deleteFromMap(m, []string{"a"})
+	assert.True(t, deleted)
+	assert.NotContains(t, m, "a")
+	assert.Contains(t, m, "b")
+
+	deleted = deleteFromMap(m, []string{"b", "b.b", "b.b.a"})
+	assert.True(t, deleted)
+	assert.Contains(t, m, "b")
+	b := m["b"].(map[string]any)
+	assert.Contains(t, b, "b.a")
+	assert.Contains(t, b, "b.b")
+	bb := b["b.b"].(map[string]any)
+	assert.NotContains(t, bb, "b.b.a")
+
+	assert.False(t, deleteFromMap(m, []string{"b", "notfound"}))
+}
+
+func TestRemoveUnsetSubResourceProperties(t *testing.T) {
+	ctx := context.Background()
+
+	res, provider := setUpResourceWithRefAndProviderWithTypeLookup()
+
+	t.Run("empty", func(t *testing.T) {
+		empty := &resources.AzureAPIResource{}
+		oldInputs := resource.PropertyMap{}
+		sdkResponse := map[string]any{}
+		actual := provider.removeUnsetSubResourceProperties(ctx, "urn", sdkResponse, oldInputs, empty)
+		expected := map[string]any{}
+		assert.Equal(t, expected, actual)
+	})
+
+	t.Run("remove", func(t *testing.T) {
+		oldInputs := resource.PropertyMap{}
+		sdkResponse := map[string]any{
+			"properties": map[string]any{
+				"accessPolicies": []any{},
+			},
+		}
+		actual := provider.removeUnsetSubResourceProperties(ctx, "urn", sdkResponse, oldInputs, res)
+		expected := map[string]any{"properties": map[string]any{}}
+		assert.Equal(t, expected, actual)
+	})
+
+	t.Run("preserve", func(t *testing.T) {
+		oldInputs := resource.PropertyMap{
+			resource.PropertyKey("properties"): resource.NewObjectProperty(resource.PropertyMap{
+				resource.PropertyKey("accessPolicies"): resource.NewArrayProperty([]resource.PropertyValue{}),
+			}),
+		}
+		sdkResponse := map[string]any{
+			"properties": map[string]any{
+				"accessPolicies": []any{},
+			},
+		}
+		actual := provider.removeUnsetSubResourceProperties(ctx, "urn", sdkResponse, oldInputs, res)
+		expected := sdkResponse
+		assert.Equal(t, expected, actual)
+	})
+}
+
+// Helper to avoid repeating the same setup code in multiple tests. Returns a resource with a
+// "properties" property of type azure-native:keyvault:VaultProperties, which the returned provider
+// will return when asked to look up that type.
+func setUpResourceWithRefAndProviderWithTypeLookup() (*resources.AzureAPIResource, *azureNativeProvider) {
+	res := resources.AzureAPIResource{
+		PutParameters: []resources.AzureAPIParameter{
+			{
+				Location: "body",
+				Body: &resources.AzureAPIType{
+					Properties: map[string]resources.AzureAPIProperty{
+						"properties": {
+							Type: "object",
+							Ref:  "#/types/azure-native:keyvault:VaultProperties",
+						},
+					},
+				},
+			},
+		},
+	}
+
+	provider := azureNativeProvider{
+		// Mock the type lookup to only return the type referenced in the resource above
+		lookupType: func(ref string) (*resources.AzureAPIType, bool, error) {
+			if ref == "#/types/azure-native:keyvault:VaultProperties" {
+				return &resources.AzureAPIType{
+					Properties: map[string]resources.AzureAPIProperty{
+						"accessPolicies": {
+							Type: "array",
+							Items: &resources.AzureAPIProperty{
+								Type: "object",
+								Ref:  "#/types/azure-native:keyvault:AccessPolicyEntry",
+							},
+							MaintainSubResourceIfUnset: true,
+						},
+					},
+				}, true, nil
+			}
+			if ref == "#/types/azure-native:keyvault:AccessPolicyEntry" {
+				return &resources.AzureAPIType{
+					Properties: map[string]resources.AzureAPIProperty{
+						"permissions": {
+							Type: "array",
+							Items: &resources.AzureAPIProperty{
+								Type: "string",
+							},
+							Containers: []string{"container2", "container3"},
+						}},
+				}, true, nil
+			}
+			return nil, false, nil
+		},
+	}
+
+	return &res, &provider
+}
+
+func TestSetUnsetSubresourcePropertiesToDefaults(t *testing.T) {
+	res, provider := setUpResourceWithRefAndProviderWithTypeLookup()
+
+	t.Run("unchanged", func(t *testing.T) {
+		body := map[string]any{
+			"properties": map[string]any{
+				"accessPolicies": []any{},
+			},
+		}
+		provider.setUnsetSubresourcePropertiesToDefaults(*res, body)
+		assert.Equal(t, map[string]any{
+			"properties": map[string]any{
+				"accessPolicies": []any{},
+			},
+		}, body)
+	})
+
+	t.Run("simple missing", func(t *testing.T) {
+		body := map[string]any{
+			"properties": map[string]any{},
+		}
+		provider.setUnsetSubresourcePropertiesToDefaults(*res, body)
+		assert.Equal(t, map[string]any{
+			"properties": map[string]any{
+				"accessPolicies": []any{},
+			},
+		}, body)
+	})
+
+	t.Run("nested missing", func(t *testing.T) {
+		body := map[string]any{}
+		provider.setUnsetSubresourcePropertiesToDefaults(*res, body)
+		assert.Equal(t, map[string]any{
+			"properties": map[string]any{
+				"accessPolicies": []any{},
+			},
+		}, body)
+	})
+}
diff --git a/provider/pkg/resources/resources.go b/provider/pkg/resources/resources.go
index 4a3af821fa9b..0df3254a3a0c 100644
--- a/provider/pkg/resources/resources.go
+++ b/provider/pkg/resources/resources.go
@@ -5,6 +5,7 @@ package resources
 import (
 	"fmt"
 	"regexp"
+	"sort"
 	"strings"
 
 	"github.com/gedex/inflector"
@@ -61,12 +62,10 @@ type AzureAPIProperty struct {
 	IsStringSet bool `json:"isStringSet,omitempty"`
 	// Default is the default value for the parameter, if any
 	Default interface{} `json:"default,omitempty"`
-	// Some resources are nested beneath other resources, but are also made available as a property on the parent.
-	// When updating the parent, if the list of nested resources is not specified as a property on the parent, the child resources will be deleted.
-	// When refreshing the parent, the list of nested resources will be populated from the current state - resulting in an unwanted diff.
-	// We work around this by setting MaintainSubResourceIfUnset to true for the properties containing a list of nested resources.
-	// When updating or refreshing the parent, if the list of nested resources was not originally specified inline, we'll maintain the existing sub-resources transparently.
-	MaintainSubResourceIfUnset bool `json:"maintainSubResourceIfUnset,omitempty"`
+	// Some resources are nested beneath other resources, but are also made available as a property
+	// on the parent. This flag temporarily marks properties that hold such sub-resources, for
+	// later aggregation. For more detail, see AzureAPIResource.SubResourcesToMaintainIfUnset.
+	MaintainSubResourceIfUnset bool `json:"omitempty,maintainSubResourceIfUnset"`
 }
 
 // AzureAPIType represents the shape of an object property.
@@ -111,17 +110,98 @@ type AzureAPIResource struct {
 	DefaultProperties map[string]interface{} `json:"defaultProperties,omitempty"`
 }
 
-func (res *AzureAPIResource) LookupProperty(key string) (AzureAPIProperty, bool) {
-	for _, param := range res.PutParameters {
-		if param.Location == "body" {
-			if prop, ok := param.Body.Properties[key]; ok {
-				return prop, true
+type TypeLookupFunc func(ref string) (*AzureAPIType, bool, error)
+
+func TraverseProperties(props map[string]AzureAPIProperty, lookupType TypeLookupFunc, includeContainers bool, f func(propName string, prop AzureAPIProperty, path []string)) {
+	// Start the traversal with an empty path and an empty set of seen types for cycle detection.
+	traverseProperties(props, lookupType, includeContainers, []string{}, map[string]struct{}{}, f)
+}
+
+func traverseProperties(props map[string]AzureAPIProperty,
+	lookupType TypeLookupFunc,
+	includeContainers bool,
+	path []string,
+	seen map[string]struct{},
+	f func(propName string, prop AzureAPIProperty, path []string)) {
+	// Sort the properties to ensure stable output
+	propNames := make([]string, 0, len(props))
+	for propName := range props {
+		propNames = append(propNames, propName)
+	}
+	sort.StringSlice(propNames).Sort()
+
+	for _, propName := range propNames {
+		prop := props[propName]
+		pathCopy := append([]string{}, path...)
+
+		if includeContainers {
+			pathCopy = append(pathCopy, prop.Containers...)
+		}
+
+		ref := prop.Ref
+		if ref == "" && prop.Items != nil {
+			ref = prop.Items.Ref
+		}
+		if ref != "" && strings.HasPrefix(ref, "#/types/azure-native") {
+			refType, ok, err := lookupType(ref)
+			if !ok || err != nil {
+				fmt.Printf("Cannot traverse properties of %s: failed to find ref %s: %v\n", propName, ref, err)
+				continue
+			}
+			if _, visited := seen[ref]; !visited {
+				seen[ref] = struct{}{}
+				nextPath := append(pathCopy, propName)
+				traverseProperties(refType.Properties, lookupType, includeContainers, nextPath, seen, f)
+			}
+		}
+
+		f(propName, prop, pathCopy)
+	}
+}
+
+func (res *AzureAPIResource) PathsToSubResourcePropertiesToMaintain(includeContainers bool, typeLookup TypeLookupFunc) [][]string {
+	result := [][]string{}
+
+	body, hasBody := res.BodyParameter()
+	if !hasBody {
+		return result
+	}
+
+	TraverseProperties(
+		body.Body.Properties,
+		typeLookup,
+		includeContainers,
+		func(propName string, prop AzureAPIProperty, path []string) {
+			if prop.MaintainSubResourceIfUnset {
+				// make a copy of path since the original might be passed to other callbacks
+				pathToProperty := append([]string{}, path...)
+				pathToProperty = append(pathToProperty, propName)
+				result = append(result, pathToProperty)
 			}
+		})
+
+	return result
+}
+
+func (res *AzureAPIResource) LookupProperty(key string) (AzureAPIProperty, bool) {
+	if body, ok := res.BodyParameter(); ok {
+		if prop, ok := body.Body.Properties[key]; ok {
+			return prop, true
 		}
 	}
 	return AzureAPIProperty{}, false
 }
 
+// BodyParameter returns the body parameter of a resource's PUT parameters, if any.
+func (res *AzureAPIResource) BodyParameter() (*AzureAPIParameter, bool) {
+	for _, param := range res.PutParameters {
+		if param.Location == "body" || param.Body != nil {
+			return &param, true
+		}
+	}
+	return nil, false
+}
+
 // AzureAPIExample provides a pointer to examples relevant to a resource from the Azure REST API spec.
 type AzureAPIExample struct {
 	Description string `json:"description"`
diff --git a/provider/pkg/resources/resources_test.go b/provider/pkg/resources/resources_test.go
index 7aa1be5a5100..af5aa362cac0 100644
--- a/provider/pkg/resources/resources_test.go
+++ b/provider/pkg/resources/resources_test.go
@@ -87,3 +87,114 @@ func TestAutoName(t *testing.T) {
 		assert.Equal(t, expected, string(actual))
 	}
 }
+
+func TestTraverseProperties(t *testing.T) {
+	properties := map[string]AzureAPIProperty{
+		"properties": {
+			Type: "object",
+			Ref:  "#/types/azure-native:keyvault:VaultProperties",
+		},
+		"location": {
+			Type: "string",
+		},
+	}
+
+	res := AzureAPIResource{
+		PutParameters: []AzureAPIParameter{
+			{
+				Location: "body",
+				Name:     "bodyProperties",
+				Body: &AzureAPIType{
+					Properties: properties,
+				},
+			},
+		},
+	}
+
+	// Mock the type lookup to only return the type referenced in the resource above
+	lookupType := func(ref string) (*AzureAPIType, bool, error) {
+		if ref == "#/types/azure-native:keyvault:VaultProperties" {
+			return &AzureAPIType{
+				Properties: map[string]AzureAPIProperty{
+					"accessPolicies": {
+						Type: "array",
+						Items: &AzureAPIProperty{
+							Type: "object",
+							Ref:  "#/types/azure-native:keyvault:AccessPolicyEntry",
+						},
+						Containers: []string{"container"}, // not the case in the real KV spec but we want to test this
+					},
+				},
+			}, true, nil
+		}
+		if ref == "#/types/azure-native:keyvault:AccessPolicyEntry" {
+			return &AzureAPIType{
+				Properties: map[string]AzureAPIProperty{
+					"permissions": {
+						Type: "array",
+						Items: &AzureAPIProperty{
+							Type: "string", // not true in the real KV spec but good enough
+						},
+						Containers:                 []string{"container2", "container3"},
+						MaintainSubResourceIfUnset: true,
+					},
+					"other_array": {
+						Type: "array",
+						Items: &AzureAPIProperty{
+							Type: "string",
+						},
+					},
+				},
+			}, true, nil
+		}
+		return nil, false, nil
+	}
+
+	t.Run("including containers", func(t *testing.T) {
+		visited := map[string][]string{}
+		visitor := func(name string, property AzureAPIProperty, path []string) {
+			visited[name] = path
+		}
+
+		TraverseProperties(properties, lookupType, true, visitor)
+
+		expected := map[string][]string{
+			"properties":     {},
+			"accessPolicies": {"properties", "container"},
+			"permissions":    {"properties", "container", "accessPolicies", "container2", "container3"},
+			"other_array":    {"properties", "container", "accessPolicies"},
+			"location":       {},
+		}
+		assert.Equal(t, expected, visited)
+	})
+
+	t.Run("without containers", func(t *testing.T) {
+		visited := map[string][]string{}
+		visitor := func(name string, property AzureAPIProperty, path []string) {
+			visited[name] = path
+		}
+
+		TraverseProperties(properties, lookupType, false, visitor)
+
+		expected := map[string][]string{
+			"properties":     {},
+			"accessPolicies": {"properties"},
+			"permissions":    {"properties", "accessPolicies"},
+			"other_array":    {"properties", "accessPolicies"},
+			"location":       {},
+		}
+		assert.Equal(t, expected, visited)
+	})
+
+	t.Run("collect subresource properties with containers", func(t *testing.T) {
+		paths := res.PathsToSubResourcePropertiesToMaintain(true, lookupType)
+		assert.Equal(t, 1, len(paths))
+		assert.Equal(t, []string{"properties", "container", "accessPolicies", "container2", "container3", "permissions"}, paths[0])
+	})
+
+	t.Run("collect subresource properties without containers", func(t *testing.T) {
+		paths := res.PathsToSubResourcePropertiesToMaintain(false, lookupType)
+		assert.Equal(t, 1, len(paths))
+		assert.Equal(t, []string{"properties", "accessPolicies", "permissions"}, paths[0])
+	})
+}