Clean up direct-engine state entries for remotely deleted resources removed from config

acceptance/bundle/resources/jobs/remote_delete/removed_from_config/bar.py

@@ -0,0 +1 @@
+print("hello")

acceptance/bundle/resources/jobs/remote_delete/removed_from_config/databricks.yml

@@ -0,0 +1,14 @@
+bundle:
+  name: test-bundle
+
+resources:
+  jobs:
+    foo:
+      name: foo
+
+  pipelines:
+    bar:
+      name: bar
+      libraries:
+        - file:
+            path: ./bar.py

acceptance/bundle/resources/jobs/remote_delete/removed_from_config/empty.yml

@@ -0,0 +1,4 @@
+bundle:
+  name: test-bundle
+
+resources: {}

acceptance/bundle/resources/jobs/remote_delete/removed_from_config/out.deploy_removed.direct.txt

@@ -0,0 +1,6 @@
+
+>>> [CLI] bundle deploy --auto-approve
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!

acceptance/bundle/resources/jobs/remote_delete/removed_from_config/out.deploy_removed.terraform.txt

@@ -0,0 +1,6 @@
+
+>>> [CLI] bundle deploy --auto-approve
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!

acceptance/bundle/resources/jobs/remote_delete/removed_from_config/out.plan_removed.direct.txt

@@ -0,0 +1,6 @@
+
+>>> [CLI] bundle plan
+delete jobs.foo
+delete pipelines.bar
+
+Plan: 0 to add, 0 to change, 2 to delete, 0 unchanged

acceptance/bundle/resources/jobs/remote_delete/removed_from_config/out.plan_removed.terraform.txt

@@ -0,0 +1,3 @@
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 0 unchanged

acceptance/bundle/resources/jobs/remote_delete/removed_from_config/out.summary.direct.txt

@@ -0,0 +1,8 @@
+
+>>> [CLI] bundle summary
+Name: test-bundle
+Target: default
+Workspace:
+  User: [USERNAME]
+  Path: /Workspace/Users/[USERNAME]/.bundle/test-bundle/default
+Resources:

acceptance/bundle/resources/jobs/remote_delete/removed_from_config/out.summary.terraform.txt

@@ -0,0 +1,16 @@
+
+>>> [CLI] bundle summary
+Name: test-bundle
+Target: default
+Workspace:
+  User: [USERNAME]
+  Path: /Workspace/Users/[USERNAME]/.bundle/test-bundle/default
+Resources:
+  Jobs:
+    foo:
+      Name: 
+      URL:  [DATABRICKS_URL]/jobs/[NUMID]?w=[NUMID]
+  Pipelines:
+    bar:
+      Name: 
+      URL:  [DATABRICKS_URL]/pipelines/[UUID]?w=[NUMID]

acceptance/bundle/resources/jobs/remote_delete/removed_from_config/output.txt

@@ -0,0 +1,24 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] jobs delete [NUMID]
+
+>>> [CLI] pipelines delete [UUID]
+
+=== Remove resources from config, plan and deploy
+=== Plan and deploy again, then summary
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 0 unchanged
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+=== No delete API calls for resources that are already gone remotely
+>>> print_requests.py //jobs/delete //pipelines/

acceptance/bundle/resources/jobs/remote_delete/removed_from_config/script

@@ -0,0 +1,21 @@
+trace $CLI bundle deploy
+job_id=`$CLI bundle summary -o json | jq -r .resources.jobs.foo.id`
+pipeline_id=`$CLI bundle summary -o json | jq -r .resources.pipelines.bar.id`
+trace $CLI jobs delete $job_id
+trace $CLI pipelines delete $pipeline_id
+
+title "Remove resources from config, plan and deploy"
+# Engines diverge: direct plans deletes for the stale state entries; terraform keeps them in state, so its summary still lists them.
+cp empty.yml databricks.yml
+# Drain requests recorded so far; only requests made after the config removal matter below.
+print_requests.py --get // &> LOG.requests_setup
+trace $CLI bundle plan &> out.plan_removed.$DATABRICKS_BUNDLE_ENGINE.txt
+trace $CLI bundle deploy --auto-approve &> out.deploy_removed.$DATABRICKS_BUNDLE_ENGINE.txt
+
+title "Plan and deploy again, then summary"
+trace $CLI bundle plan
+trace $CLI bundle deploy
+trace $CLI bundle summary &> out.summary.$DATABRICKS_BUNDLE_ENGINE.txt
+
+title "No delete API calls for resources that are already gone remotely"
+trace print_requests.py //jobs/delete //pipelines/

acceptance/bundle/resources/jobs/remote_delete/removed_from_config/test.toml

@@ -0,0 +1 @@
+RecordRequests = true

bundle/deploy/metadata/compute.go

@@ -48,7 +48,9 @@ func (m *compute) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics
 		// root
 		l := b.Config.GetLocation("resources.jobs." + name)
 		if l.File == "" {
-			// b.Config.Resources.Jobs may include a job that only exists in state but not in config
+			// Skip resources that exist only in the deployment state: statemgmt.Load,
+			// which runs before this mutator, injects them into the config without a
+			// file location.
 			continue
 		}
 
@@ -72,6 +74,13 @@ func (m *compute) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics
 		// Compute config file path the pipeline is defined in, relative to the bundle
 		// root
 		l := b.Config.GetLocation("resources.pipelines." + name)
+		if l.File == "" {
+			// Skip resources that exist only in the deployment state: statemgmt.Load,
+			// which runs before this mutator, injects them into the config without a
+			// file location.
+			continue
+		}
+
 		relativePath, err := filepath.Rel(b.BundleRootPath, l.File)
 		if err != nil {
 			return diag.Errorf("failed to compute relative path for pipeline %s: %v", name, err)
@@ -90,6 +99,13 @@ func (m *compute) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics
 		// Compute config file path the dashboard is defined in, relative to the bundle
 		// root
 		l := b.Config.GetLocation("resources.dashboards." + name)
+		if l.File == "" {
+			// Skip resources that exist only in the deployment state: statemgmt.Load,
+			// which runs before this mutator, injects them into the config without a
+			// file location.
+			continue
+		}
+
 		relativePath, err := filepath.Rel(b.BundleRootPath, l.File)
 		if err != nil {
 			return diag.Errorf("failed to compute relative path for dashboard %s: %v", name, err)

bundle/deploy/metadata/compute_test.go

@@ -148,6 +148,39 @@ func TestComputeMetadataMutator(t *testing.T) {
 	assert.Equal(t, expectedMetadata, b.Metadata)
 }
 
+func TestComputeMetadataMutatorStateOnlyResources(t *testing.T) {
+	// State-only resources (in state but not in config) have no location and must be skipped, not error.
+	b := &bundle.Bundle{
+		BundleRootPath: "/tmp/some/root",
+		Config: config.Root{
+			Resources: config.Resources{
+				Jobs: map[string]*resources.Job{
+					"state-only-job": {
+						BaseResource: resources.BaseResource{ID: "1111"},
+					},
+				},
+				Pipelines: map[string]*resources.Pipeline{
+					"state-only-pipeline": {
+						BaseResource: resources.BaseResource{ID: "2222"},
+					},
+				},
+				Dashboards: map[string]*resources.Dashboard{
+					"state-only-dashboard": {
+						BaseResource: resources.BaseResource{ID: "3333"},
+					},
+				},
+			},
+		},
+	}
+
+	diags := bundle.Apply(t.Context(), b, Compute())
+	require.NoError(t, diags.Error())
+
+	assert.Empty(t, b.Metadata.Config.Resources.Jobs)
+	assert.Empty(t, b.Metadata.Config.Resources.Pipelines)
+	assert.Empty(t, b.Metadata.Config.Resources.Dashboards)
+}
+
 func TestComputeMetadataMutatorSourceLinked(t *testing.T) {
 	syncRootPath := "/Users/shreyas.goenka@databricks.com/source"
 	enabled := true

bundle/deployplan/action.go

@@ -9,6 +9,9 @@ type Action struct {
 	// Full resource key, e.g. "resources.jobs.foo" or "resources.jobs.foo.permissions"
 	ResourceKey string
 	ActionType  ActionType
+	// Gone mirrors PlanEntry.Gone: the delete is a state-only cleanup because the
+	// resource no longer exists remotely.
+	Gone bool
 }
 
 func (a Action) String() string {

bundle/deployplan/plan.go

@@ -73,9 +73,13 @@ func LoadPlanFromFile(path string) (*Plan, error) {
 }
 
 type PlanEntry struct {
-	ID          string                   `json:"id,omitempty"`
-	DependsOn   []DependsOnEntry         `json:"depends_on,omitempty"`
-	Action      ActionType               `json:"action,omitempty"`
+	ID        string           `json:"id,omitempty"`
+	DependsOn []DependsOnEntry `json:"depends_on,omitempty"`
+	Action    ActionType       `json:"action,omitempty"`
+	// Gone is set on Delete entries when planning confirmed the resource no longer
+	// exists remotely. Applying such an entry only removes it from the state, without
+	// calling the delete API, and approval prompts do not list it as a deletion.
+	Gone        bool                     `json:"gone,omitempty"`
 	NewState    *structvar.StructVarJSON `json:"new_state,omitempty"`
 	RemoteState any                      `json:"remote_state,omitempty"`
 	Changes     Changes                  `json:"changes,omitempty"`
@@ -152,6 +156,7 @@ func (p *Plan) GetActions() []Action {
 		actions = append(actions, Action{
 			ResourceKey: key,
 			ActionType:  entry.Action,
+			Gone:        entry.Gone,
 		})
 	}
 
@@ -195,12 +200,6 @@ func (p *Plan) ReadUnlockEntry(resourceKey string) {
 	p.lockmap.RUnlock(resourceKey)
 }
 
-func (p *Plan) RemoveEntry(resourceKey string) {
-	p.mutex.Lock()
-	defer p.mutex.Unlock()
-	delete(p.Plan, resourceKey)
-}
-
 // FilterToSelected reduces the plan to the nodes in selected (format "type.name",
 // e.g. "jobs.my_job") plus their transitive dependencies as recorded in each
 // entry's DependsOn field. Nodes not reachable from the selected set are removed.

bundle/direct/bundle_apply.go

@@ -96,7 +96,13 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 				}
 				return true
 			}
-			err = d.Destroy(ctx, &b.StateDB)
+			if entry.Gone {
+				// Planning confirmed the resource is already deleted remotely; only
+				// remove it from the state, without calling the delete API.
+				err = b.StateDB.DeleteState(resourceKey)
+			} else {
+				err = d.Destroy(ctx, &b.StateDB)
+			}
 			if err != nil {
 				logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err))
 				return false

bundle/direct/bundle_plan.go

@@ -162,8 +162,10 @@ func (b *DeploymentBundle) CalculatePlan(ctx context.Context, client *databricks
 			})
 			if err != nil {
 				if isResourceGone(err) {
-					// no such resource
-					plan.RemoveEntry(resourceKey)
+					// The resource is already deleted remotely. Keep the Delete entry so
+					// that applying it removes the stale state entry, but mark it Gone so
+					// apply skips the delete call and prompts don't list it as a deletion.
+					entry.Gone = true
 				} else {
 					log.Warnf(ctx, "reading %s id=%q: %s", resourceKey, id, err)
 					// This is not an error during deletion, so don't return false here

bundle/phases/deploy.go

@@ -3,6 +3,7 @@ package phases
 import (
 	"context"
 	"errors"
+	"slices"
 
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/bundle/artifacts"
@@ -43,6 +44,10 @@ var deployApprovalGroups = []approvalGroup{
 func approvalForDeploy(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan) (bool, error) {
 	actions := plan.GetActions()
 
+	// Deletes of resources that are already gone remotely only clean up the state,
+	// so they don't count as destructive actions and need no approval.
+	actions = slices.DeleteFunc(actions, func(a deployplan.Action) bool { return a.Gone })
+
 	err := checkForPreventDestroy(b, actions)
 	if err != nil {
 		return false, err

bundle/phases/destroy.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"net/http"
+	"slices"
 
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/bundle/config/engine"
@@ -47,6 +48,10 @@ var destroyApprovalGroups = []approvalGroup{
 func approvalForDestroy(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan) (bool, error) {
 	deleteActions := plan.GetActions()
 
+	// Deletes of resources that are already gone remotely only clean up the state,
+	// so they don't count as destructive actions and are not listed as deletions.
+	deleteActions = slices.DeleteFunc(deleteActions, func(a deployplan.Action) bool { return a.Gone })
+
 	err := checkForPreventDestroy(b, deleteActions)
 	if err != nil {
 		return false, err