Enable reading deployment state from DMS

acceptance/bundle/dms/read/databricks.yml

@@ -0,0 +1,18 @@
+# pipelines.delete_me is recorded in DMS (see script) but intentionally absent
+# here, so the plan deletes it. Pipelines are used (not jobs) because their ids
+# are strings: the `databricks api` command rounds large int64 job ids in its
+# output, which would break the id round-trip this test relies on.
+bundle:
+  name: dms-read
+
+experimental:
+  record_deployment_history: true
+
+resources:
+  pipelines:
+    # Not recorded in DMS -> plan should create it.
+    create_me:
+      name: create-me
+    # Recorded in DMS with a different name -> plan should update it.
+    update_me:
+      name: update-me-new

acceptance/bundle/dms/read/output.txt

@@ -0,0 +1,119 @@
+
+=== bundle summary reflects resources read from DMS (update_me resolves its id; create_me is not deployed)
+>>> [CLI] bundle summary
+Name: dms-read
+Target: default
+Workspace:
+  User: [USERNAME]
+  Path: /Workspace/Users/[USERNAME]/.bundle/dms-read/default
+Resources:
+  Pipelines:
+    create_me:
+      Name: create-me
+      URL:  (not deployed)
+    delete_me:
+      Name: 
+      URL:  [DATABRICKS_URL]/pipelines/[UUID]?w=[NUMID]
+    update_me:
+      Name: update-me-new
+      URL:  [DATABRICKS_URL]/pipelines/[UUID]?w=[NUMID]
+
+=== bundle plan derives actions from DMS state: create create_me, update update_me, delete delete_me
+>>> [CLI] bundle plan
+create pipelines.create_me
+delete pipelines.delete_me
+update pipelines.update_me
+
+Plan: 1 to add, 1 to change, 1 to delete, 0 unchanged
+
+=== full plan as JSON
+>>> [CLI] bundle plan -o json
+{
+  "plan_version": 2,
+  "cli_version": "[DEV_VERSION]",
+  "lineage": "synthetic-dep",
+  "serial": 1,
+  "plan": {
+    "resources.pipelines.create_me": {
+      "action": "create",
+      "new_state": {
+        "value": {
+          "channel": "CURRENT",
+          "deployment": {
+            "kind": "BUNDLE",
+            "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-read/default/state/metadata.json"
+          },
+          "edition": "ADVANCED",
+          "name": "create-me"
+        }
+      }
+    },
+    "resources.pipelines.delete_me": {
+      "action": "delete",
+      "remote_state": {
+        "creator_user_name": "[USERNAME]",
+        "effective_publishing_mode": "DEFAULT_PUBLISHING_MODE",
+        "id": "[UUID]",
+        "last_modified": [UNIX_TIME_MILLIS][0],
+        "name": "delete-me",
+        "pipeline_id": "[UUID]",
+        "run_as_user_name": "[USERNAME]",
+        "state": "IDLE",
+        "storage": "dbfs:/pipelines/[UUID]"
+      }
+    },
+    "resources.pipelines.update_me": {
+      "action": "update",
+      "new_state": {
+        "value": {
+          "channel": "CURRENT",
+          "deployment": {
+            "kind": "BUNDLE",
+            "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-read/default/state/metadata.json"
+          },
+          "edition": "ADVANCED",
+          "name": "update-me-new"
+        }
+      },
+      "remote_state": {
+        "creator_user_name": "[USERNAME]",
+        "effective_publishing_mode": "DEFAULT_PUBLISHING_MODE",
+        "id": "[UUID]",
+        "last_modified": [UNIX_TIME_MILLIS][1],
+        "name": "update-me-old",
+        "pipeline_id": "[UUID]",
+        "run_as_user_name": "[USERNAME]",
+        "state": "IDLE",
+        "storage": "dbfs:/pipelines/[UUID]"
+      },
+      "changes": {
+        "channel": {
+          "action": "update",
+          "new": "CURRENT"
+        },
+        "deployment": {
+          "action": "update",
+          "new": {
+            "kind": "BUNDLE",
+            "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-read/default/state/metadata.json"
+          }
+        },
+        "edition": {
+          "action": "update",
+          "new": "ADVANCED"
+        },
+        "name": {
+          "action": "update",
+          "old": "update-me-old",
+          "new": "update-me-new",
+          "remote": "update-me-old"
+        },
+        "storage": {
+          "action": "skip",
+          "reason": "backend_default",
+          "remote": "dbfs:/pipelines/[UUID]"
+        }
+      }
+    }
+  }
+}

acceptance/bundle/dms/read/script

@@ -0,0 +1,28 @@
+# Read deployment state from DMS: build synthetic DMS state for a deployment
+# whose resources, compared with databricks.yml, exercise each plan action
+#   - create_me: in config, not in DMS  -> create
+#   - update_me: in both, names differ   -> update
+#   - delete_me: in DMS, not in config   -> delete
+# then verify both `bundle summary` and `bundle plan` read it from DMS.
+
+update_id=$($CLI pipelines create -o json --json '{"name":"update-me-old"}' | jq -r .pipeline_id)
+delete_id=$($CLI pipelines create -o json --json '{"name":"delete-me"}' | jq -r .pipeline_id)
+
+# DMS owns the state only once a version has completed successfully, so record one
+# for the deployment before seeding its resources.
+$CLI api post "/api/2.0/bundle/deployments/synthetic-dep/versions?version_id=1" --json '{"version_type":"VERSION_TYPE_DEPLOY"}' > /dev/null
+$CLI api post "/api/2.0/bundle/deployments/synthetic-dep/versions/1/complete" --json '{"completion_reason":"VERSION_COMPLETE_SUCCESS"}' > /dev/null
+
+$CLI api post "/api/2.0/bundle/deployments/synthetic-dep/versions/1/operations" --json "{\"resource_key\":\"pipelines.update_me\",\"resource_id\":\"$update_id\",\"action_type\":\"OPERATION_ACTION_TYPE_CREATE\",\"status\":\"OPERATION_STATUS_SUCCEEDED\",\"state\":{\"name\":\"update-me-old\"}}" > /dev/null
+$CLI api post "/api/2.0/bundle/deployments/synthetic-dep/versions/1/operations" --json "{\"resource_key\":\"pipelines.delete_me\",\"resource_id\":\"$delete_id\",\"action_type\":\"OPERATION_ACTION_TYPE_CREATE\",\"status\":\"OPERATION_STATUS_SUCCEEDED\",\"state\":{\"name\":\"delete-me\"}}" > /dev/null
+
+$CLI api post "/api/2.0/workspace-files/import-file/Workspace/Users/$CURRENT_USER_NAME/.bundle/dms-read/default/state/resources.json?overwrite=true" --json '{"state_version":2,"lineage":"synthetic-dep","serial":1,"state":{}}' > /dev/null
+
+title "bundle summary reflects resources read from DMS (update_me resolves its id; create_me is not deployed)"
+trace $CLI bundle summary
+
+title "bundle plan derives actions from DMS state: create create_me, update update_me, delete delete_me"
+trace $CLI bundle plan
+
+title "full plan as JSON"
+trace $CLI bundle plan -o json

acceptance/bundle/dms/read/test.toml

@@ -0,0 +1,2 @@
+# DMS state reading is a direct-engine feature; pin the engine for stable output.
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]

bundle/statemgmt/statereader.go

@@ -0,0 +1,217 @@
+package statemgmt
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/fs"
+	"os"
+	"path/filepath"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/direct/dstate"
+	"github.com/databricks/databricks-sdk-go/apierr"
+	sdkbundle "github.com/databricks/databricks-sdk-go/service/bundle"
+)
+
+// StateReader loads a deployment's state into the in-memory DeploymentState that
+// the direct engine reads during plan/apply.
+//
+// Deployment state has two parts:
+//
+//   - Identity: the lineage (the deployment id) and serial. These always live in
+//     the local resources.json file and identify the deployment across runs.
+//   - Resources: each deployed resource's id and last-deployed config.
+//
+// Where the resources come from depends on whether the bundle records deployment
+// history, which is what the two implementations capture:
+//
+//   - fileStateReader: identity and resources both come from resources.json.
+//     This is the default and matches the behavior before DMS.
+//   - dmsStateReader: identity still comes from resources.json (its lineage is the
+//     DMS deployment id), but the resource set is read from the deployment
+//     metadata service (DMS).
+//
+// Both leave the DB open for reading; the plan path may later upgrade it to write
+// mode (see dstate.DeploymentState.UpgradeToWrite).
+type StateReader interface {
+	// Load populates db with the deployment's state, leaving it open for reading.
+	Load(ctx context.Context, db *dstate.DeploymentState) error
+}
+
+// fileStateReader loads both identity and resources from the local resources.json
+// file. "local" means the file in the bundle's state cache that PullResourcesState
+// has already synced from the workspace.
+type fileStateReader struct {
+	path string
+}
+
+// NewFileStateReader returns a StateReader that reads both identity and resources
+// from the local resources.json file at path.
+func NewFileStateReader(path string) StateReader {
+	return &fileStateReader{path: path}
+}
+
+func (r *fileStateReader) Load(ctx context.Context, db *dstate.DeploymentState) error {
+	// Open reads resources.json (identity + resources) into db. Recovery replays a
+	// leftover write-ahead log from a crashed deploy, which only the file-backed
+	// state can have.
+	return db.Open(ctx, r.path, dstate.WithRecovery(true), dstate.WithWrite(false))
+}
+
+// dmsStateReader loads the identity (lineage/serial) from the local resources.json
+// file and the resource set from DMS, replacing whatever resources.json held.
+// NewStateReader only selects this reader once DMS is the authoritative source for
+// the deployment, so it trusts the DMS resource set even when it is empty (a
+// successful deploy with no resources). It never falls back to resources.json for
+// resource state unless the user opts out of DMS by disabling
+// record_deployment_history.
+type dmsStateReader struct {
+	client       sdkbundle.BundleInterface
+	deploymentID string
+	path         string // local resources.json path; supplies the identity (lineage/serial)
+}
+
+// NewDMSStateReader returns a StateReader that reads identity from the local
+// resources.json at path and the resource set from DMS for deploymentID (which is
+// the deployment's lineage).
+func NewDMSStateReader(client sdkbundle.BundleInterface, deploymentID, path string) StateReader {
+	return &dmsStateReader{client: client, deploymentID: deploymentID, path: path}
+}
+
+func (r *dmsStateReader) Load(ctx context.Context, db *dstate.DeploymentState) error {
+	// Identity (lineage/serial) always comes from resources.json: the lineage is
+	// the DMS deployment id and a later deploy must reuse it rather than mint a new
+	// one. resources.json keeps recording resources too, so the bundle can be
+	// migrated back to file-based state.
+	data, err := readLocalDatabase(r.path)
+	if err != nil {
+		return err
+	}
+
+	data.State, err = fetchDeploymentResources(ctx, r.client, r.deploymentID)
+	if err != nil {
+		return err
+	}
+
+	db.OpenWithData(r.path, data)
+	return nil
+}
+
+// deploymentHasSuccessfulVersion reports whether DMS holds a successfully
+// completed version for the deployment. That is the signal that DMS owns the
+// state: if the deployment was never recorded to DMS, or its initial DMS deploy
+// did not complete successfully, DMS state is absent or partial and callers
+// should fall back to the local file.
+func deploymentHasSuccessfulVersion(ctx context.Context, client sdkbundle.BundleInterface, deploymentID string) (bool, error) {
+	versions, err := client.ListVersionsAll(ctx, sdkbundle.ListVersionsRequest{
+		Parent: "deployments/" + deploymentID,
+	})
+	if err != nil {
+		// A deployment that was never recorded to DMS is not an error here: it just
+		// means DMS is not (yet) the source of truth.
+		if errors.Is(err, apierr.ErrNotFound) {
+			return false, nil
+		}
+		return false, fmt.Errorf("listing versions from deployment metadata service: %w", err)
+	}
+
+	for _, v := range versions {
+		if v.Status == sdkbundle.VersionStatusVersionStatusCompleted &&
+			v.CompletionReason == sdkbundle.VersionCompleteVersionCompleteSuccess {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+// fetchDeploymentResources lists every resource recorded for the deployment in
+// DMS and maps them into state entries keyed by the fully-qualified resource key.
+func fetchDeploymentResources(ctx context.Context, client sdkbundle.BundleInterface, deploymentID string) (map[string]dstate.ResourceEntry, error) {
+	resources, err := client.ListResourcesAll(ctx, sdkbundle.ListResourcesRequest{
+		Parent: "deployments/" + deploymentID,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("listing resources from deployment metadata service: %w", err)
+	}
+
+	out := make(map[string]dstate.ResourceEntry, len(resources))
+	for _, res := range resources {
+		// DMS reports resource keys without the "resources." prefix (e.g.
+		// "jobs.foo"), but the local state DB keys are fully qualified
+		// ("resources.jobs.foo"), so prepend it here.
+		key := "resources." + res.ResourceKey
+
+		var state json.RawMessage
+		if res.State != nil {
+			state = *res.State
+		}
+
+		out[key] = dstate.ResourceEntry{
+			ID:    res.ResourceId,
+			State: state,
+		}
+	}
+	return out, nil
+}
+
+// NewStateReader picks the reader for the bundle. Four cases:
+//
+//  1. record_deployment_history off: read everything from resources.json.
+//  2. on, but resources.json has no lineage (nothing deployed yet): read the
+//     (empty) local file. The lineage is minted lazily on the first deploy, when
+//     the state DB is first opened for write (dstate.DeploymentState.UpgradeToWrite);
+//     reads open it write-disabled and never mint one.
+//  3. on, with a lineage, and DMS has a successfully-completed version for it:
+//     DMS owns the state, so read resources from DMS.
+//  4. on, with a lineage, but DMS has no successfully-completed version (DMS not
+//     initialized for this deployment yet, e.g. record_deployment_history was just
+//     enabled on an existing deployment, or the initial DMS deploy failed): defer
+//     to the local file so existing resources are neither re-created nor lost.
+//
+// The lineage comes from the local resources.json, so PullResourcesState must
+// have synced it before this is called.
+func NewStateReader(ctx context.Context, b *bundle.Bundle, path string) (StateReader, error) {
+	recordHistory := b.Config.Experimental != nil && b.Config.Experimental.RecordDeploymentHistory
+	if !recordHistory {
+		return NewFileStateReader(path), nil // case 1
+	}
+
+	local, err := readLocalDatabase(path)
+	if err != nil {
+		return nil, err
+	}
+	if local.Lineage == "" {
+		return NewFileStateReader(path), nil // case 2
+	}
+
+	client := b.WorkspaceClient(ctx).Bundle
+	authoritative, err := deploymentHasSuccessfulVersion(ctx, client, local.Lineage)
+	if err != nil {
+		return nil, err
+	}
+	if !authoritative {
+		return NewFileStateReader(path), nil // case 4
+	}
+
+	return NewDMSStateReader(client, local.Lineage, path), nil // case 3
+}
+
+// readLocalDatabase parses the local resources.json file. A missing file yields
+// an empty database (no lineage), which callers read as "nothing deployed yet".
+func readLocalDatabase(path string) (dstate.Database, error) {
+	content, err := os.ReadFile(path)
+	if errors.Is(err, fs.ErrNotExist) {
+		return dstate.NewDatabase("", 0), nil
+	}
+	if err != nil {
+		return dstate.Database{}, err
+	}
+
+	var db dstate.Database
+	if err := json.Unmarshal(content, &db); err != nil {
+		return dstate.Database{}, fmt.Errorf("parsing %s: %w", filepath.ToSlash(path), err)
+	}
+	return db, nil
+}