direct: Add DoUpdateWithChanges (#3901)

## Why
Useful for model serving endpoint where we need to call multiple API
endpoints based on the fields that were modified or had a remote drift.

relevant TF code:
https://github.com/databricks/terraform-provider-databricks/blob/5501f662eb0a8a31bd5f94ba9e9130288b9314a6/serving/resource_model_serving.go#L370

Author: shreyas-goenka

bundle/direct/apply.go

@@ -32,7 +32,7 @@ func (d *DeploymentUnit) Destroy(ctx context.Context, db *dstate.DeploymentState
 	return nil
 }
 
-func (d *DeploymentUnit) Deploy(ctx context.Context, db *dstate.DeploymentState, newState any, actionType deployplan.ActionType) error {
+func (d *DeploymentUnit) Deploy(ctx context.Context, db *dstate.DeploymentState, newState any, actionType deployplan.ActionType, changes *deployplan.Changes) error {
 	if actionType == deployplan.ActionTypeCreate {
 		return d.Create(ctx, db, newState)
 	}
@@ -51,7 +51,7 @@ func (d *DeploymentUnit) Deploy(ctx context.Context, db *dstate.DeploymentState,
 	case deployplan.ActionTypeRecreate:
 		return d.Recreate(ctx, db, oldID, newState)
 	case deployplan.ActionTypeUpdate:
-		return d.Update(ctx, db, oldID, newState)
+		return d.Update(ctx, db, oldID, newState, changes)
 	case deployplan.ActionTypeUpdateWithID:
 		return d.UpdateWithID(ctx, db, oldID, newState)
 	case deployplan.ActionTypeResize:
@@ -107,8 +107,16 @@ func (d *DeploymentUnit) Recreate(ctx context.Context, db *dstate.DeploymentStat
 	return d.Create(ctx, db, newState)
 }
 
-func (d *DeploymentUnit) Update(ctx context.Context, db *dstate.DeploymentState, id string, newState any) error {
-	remoteState, err := d.Adapter.DoUpdate(ctx, id, newState)
+func (d *DeploymentUnit) Update(ctx context.Context, db *dstate.DeploymentState, id string, newState any, changes *deployplan.Changes) error {
+	var remoteState any
+	var err error
+
+	if d.Adapter.HasDoUpdateWithChanges() && changes != nil {
+		remoteState, err = d.Adapter.DoUpdateWithChanges(ctx, id, newState, changes)
+	} else {
+		remoteState, err = d.Adapter.DoUpdate(ctx, id, newState)
+	}
+
 	if err != nil {
 		return fmt.Errorf("updating id=%s: %w", id, err)
 	}

bundle/direct/bundle_apply.go

@@ -97,7 +97,7 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 
 			// TODO: redo calcDiff to downgrade planned action if possible (?)
 
-			err = d.Deploy(ctx, &b.StateDB, entry.NewState.Config, at)
+			err = d.Deploy(ctx, &b.StateDB, entry.NewState.Config, at, entry.Changes)
 			if err != nil {
 				logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err))
 				return false

bundle/direct/dresources/adapter.go

@@ -72,6 +72,9 @@ type IResourceNoRefresh interface {
 	// Example: func (r *ResourceJob) DoUpdate(ctx context.Context, id string, newState *jobs.JobSettings) error
 	DoUpdate(ctx context.Context, id string, newState any) error
 
+	// [Optional] DoUpdateWithChanges updates the resource with information about changes computed during plan.
+	DoUpdateWithChanges(ctx context.Context, id string, newState any, changes *deployplan.Changes) error
+
 	// [Optional] DoUpdateWithID performs an update that may result in resource having a new ID
 	// Example: func (r *ResourceVolume) DoUpdateWithID(ctx, id string, newState *catalog.CreateVolumeRequestContent) (string, error)
 	DoUpdateWithID(ctx context.Context, id string, newState any) (string, error)
@@ -104,6 +107,10 @@ type IResourceWithRefresh interface {
 	// Example: func (r *ResourceSchema) DoUpdate(ctx context.Context, id string, newState *catalog.CreateSchema) (*catalog.SchemaInfo, error)
 	DoUpdate(ctx context.Context, id string, newState any) (remoteState any, e error)
 
+	// [Optional] DoUpdateWithChanges updates the resource with information about changes computed during plan. Returns remote state.
+	// Example: func (r *ResourceModelServingEndpoint) DoUpdateWithChanges(ctx context.Context, id string, newState *serving.CreateServingEndpoint, changes *deployplan.Changes) (*serving.ServingEndpointInfo, error)
+	DoUpdateWithChanges(ctx context.Context, id string, newState any, changes *deployplan.Changes) (remoteState any, e error)
+
 	// Optional: updates that may change ID. Returns new id and remote state when available.
 	DoUpdateWithID(ctx context.Context, id string, newState any) (newID string, remoteState any, e error)
 
@@ -126,11 +133,12 @@ type Adapter struct {
 	doUpdate     *calladapt.BoundCaller
 
 	// Optional:
-	doUpdateWithID  *calladapt.BoundCaller
-	waitAfterCreate *calladapt.BoundCaller
-	waitAfterUpdate *calladapt.BoundCaller
-	classifyChange  *calladapt.BoundCaller
-	doResize        *calladapt.BoundCaller
+	doUpdateWithChanges *calladapt.BoundCaller
+	doUpdateWithID      *calladapt.BoundCaller
+	waitAfterCreate     *calladapt.BoundCaller
+	waitAfterUpdate     *calladapt.BoundCaller
+	classifyChange      *calladapt.BoundCaller
+	doResize            *calladapt.BoundCaller
 
 	fieldTriggersLocal  map[string]deployplan.ActionType
 	fieldTriggersRemote map[string]deployplan.ActionType
@@ -156,6 +164,7 @@ func NewAdapter(typedNil any, client *databricks.WorkspaceClient) (*Adapter, err
 		doDelete:            nil,
 		doCreate:            nil,
 		doUpdate:            nil,
+		doUpdateWithChanges: nil,
 		doUpdateWithID:      nil,
 		doResize:            nil,
 		waitAfterCreate:     nil,
@@ -260,6 +269,11 @@ func (a *Adapter) initMethods(resource any) error {
 
 	// Optional methods:
 
+	a.doUpdateWithChanges, err = prepareCallFromTwoVariants(resource, "DoUpdateWithChanges")
+	if err != nil {
+		return err
+	}
+
 	a.doUpdateWithID, err = prepareCallFromTwoVariants(resource, "DoUpdateWithID")
 	if err != nil {
 		return err
@@ -345,6 +359,13 @@ func (a *Adapter) validate() error {
 		validations = append(validations, "DoUpdate remoteState return", a.doUpdate.OutTypes[0], remoteType)
 	}
 
+	if a.doUpdateWithChanges != nil {
+		validations = append(validations, "DoUpdateWithChanges newState", a.doUpdateWithChanges.InTypes[2], stateType)
+		if len(a.doUpdateWithChanges.OutTypes) == 2 {
+			validations = append(validations, "DoUpdateWithChanges remoteState return", a.doUpdateWithChanges.OutTypes[0], remoteType)
+		}
+	}
+
 	if a.doResize != nil {
 		validations = append(validations, "DoResize newState", a.doResize.InTypes[2], stateType)
 	}
@@ -497,6 +518,32 @@ func (a *Adapter) DoUpdate(ctx context.Context, id string, newState any) (any, e
 	}
 }
 
+// HasDoUpdateWithChanges returns true if the resource implements DoUpdateWithChanges method.
+func (a *Adapter) HasDoUpdateWithChanges() bool {
+	return a.doUpdateWithChanges != nil
+}
+
+// DoUpdateWithChanges updates the resource with information about changes computed during plan.
+// If the implementation returns remote state, it will be returned as the first value; otherwise it will be nil.
+func (a *Adapter) DoUpdateWithChanges(ctx context.Context, id string, newState any, changes *deployplan.Changes) (any, error) {
+	if a.doUpdateWithChanges == nil {
+		return nil, errors.New("internal error: DoUpdateWithChanges not found")
+	}
+
+	outs, err := a.doUpdateWithChanges.Call(ctx, id, newState, changes)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(outs) == 1 {
+		// WithRefresh version
+		return outs[0], nil
+	} else {
+		// NoRefresh version
+		return nil, nil
+	}
+}
+
 // HasDoUpdateWithID returns true if the resource implements DoUpdateWithID method.
 func (a *Adapter) HasDoUpdateWithID() bool {
 	return a.doUpdateWithID != nil