direct: Added support for ClassifyChanges (#3690)

## Changes
Added support for ClassifyChanges

## Why
Needed for #3682

Some resources might have different types of changes based on their
remote state. For example, the cluster's resize action becomes an update
if the cluster is stopped. In order to do such classification, we need
access to the remote state, so we are introducing a new ClassifyChange
method.

ClassifyChange is called first, if it returns ActionTypeUnset,
ClassifyByTriggers called

## Tests
<!-- How have you tested the changes? -->

<!-- If your PR needs to be included in the release notes for next
release,
add a separate entry in NEXT_CHANGELOG.md as part of your PR. -->

Author: andrewnester

bundle/direct/bundle_plan.go

@@ -169,7 +169,7 @@ func (b *DeploymentBundle) CalculatePlanForDeploy(ctx context.Context, client *d
 				return false
 			}
 
-			remoteAction, remoteChangeMap = interpretOldStateVsRemoteState(adapter, remoteDiff)
+			remoteAction, remoteChangeMap = interpretOldStateVsRemoteState(ctx, adapter, remoteDiff, remoteState)
 		}
 
 		entry.Action = max(localAction, remoteAction).String()
@@ -237,7 +237,7 @@ func convertChangesToTriggersMap(adapter *dresources.Adapter, diff []structdiff.
 	return action, m
 }
 
-func interpretOldStateVsRemoteState(adapter *dresources.Adapter, diff []structdiff.Change) (deployplan.ActionType, map[string]deployplan.Trigger) {
+func interpretOldStateVsRemoteState(ctx context.Context, adapter *dresources.Adapter, diff []structdiff.Change, remoteState any) (deployplan.ActionType, map[string]deployplan.Trigger) {
 	action := deployplan.ActionTypeSkip
 	m := make(map[string]deployplan.Trigger)
 
@@ -252,7 +252,12 @@ func interpretOldStateVsRemoteState(adapter *dresources.Adapter, diff []structdi
 			}
 			continue
 		}
-		fieldAction := adapter.ClassifyByTriggers(ch)
+		fieldAction, err := adapter.ClassifyChange(ch, remoteState)
+		if err != nil {
+			logdiag.LogError(ctx, fmt.Errorf("internal error: failed to classify changes: %w", err))
+			continue
+		}
+
 		if fieldAction > action {
 			action = fieldAction
 		}

bundle/direct/dresources/adapter.go

@@ -72,6 +72,9 @@ type IResourceNoRefresh interface {
 
 	// [Optional] WaitAfterUpdate waits for the resource to become ready after update.
 	WaitAfterUpdate(ctx context.Context, newState any) error
+
+	// [Optional] ClassifyChange classifies a set of changes using custom logic.
+	ClassifyChange(change structdiff.Change, remoteState any) (deployplan.ActionType, error)
 }
 
 // IResourceWithRefresh is an alternative to IResourceNoRefresh but every method can return remoteState.
@@ -111,6 +114,7 @@ type Adapter struct {
 	doUpdateWithID  *calladapt.BoundCaller
 	waitAfterCreate *calladapt.BoundCaller
 	waitAfterUpdate *calladapt.BoundCaller
+	classifyChange  *calladapt.BoundCaller
 
 	fieldTriggers map[string]deployplan.ActionType
 }
@@ -138,6 +142,7 @@ func NewAdapter(typedNil any, client *databricks.WorkspaceClient) (*Adapter, err
 		doUpdateWithID:  nil,
 		waitAfterCreate: nil,
 		waitAfterUpdate: nil,
+		classifyChange:  nil,
 		fieldTriggers:   map[string]deployplan.ActionType{},
 	}
 
@@ -306,6 +311,10 @@ func (a *Adapter) validate() error {
 		}
 	}
 
+	if a.classifyChange != nil {
+		validations = append(validations, "ClassifyChange changes", a.classifyChange.InTypes[1], remoteType)
+	}
+
 	err = validateTypes(validations...)
 	if err != nil {
 		return err
@@ -501,6 +510,21 @@ func (a *Adapter) WaitAfterUpdate(ctx context.Context, newState any) (any, error
 	}
 }
 
+func (a *Adapter) ClassifyChange(change structdiff.Change, remoteState any) (deployplan.ActionType, error) {
+	// If ClassifyChange is not implemented, use FieldTriggers.
+	if a.classifyChange == nil {
+		return a.ClassifyByTriggers(change), nil
+	}
+
+	outs, err := a.classifyChange.Call(change, remoteState)
+	if err != nil {
+		return deployplan.ActionTypeSkip, err
+	}
+
+	actionType := outs[0].(deployplan.ActionType)
+	return actionType, nil
+}
+
 // prepareCallRequired prepares a call and ensures the method is found.
 func prepareCallRequired(resource any, methodName string) (*calladapt.BoundCaller, error) {
 	caller, err := calladapt.PrepareCall(resource, calladapt.TypeOf[IResource](), methodName)

bundle/direct/dresources/all_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/databricks/cli/bundle/config/resources"
 	"github.com/databricks/cli/bundle/deployplan"
 	"github.com/databricks/cli/libs/structs/structaccess"
+	"github.com/databricks/cli/libs/structs/structdiff"
 	"github.com/databricks/cli/libs/structs/structpath"
 	"github.com/databricks/cli/libs/structs/structwalk"
 	"github.com/databricks/cli/libs/testserver"
@@ -205,6 +206,16 @@ func testCRUD(t *testing.T, group string, adapter *Adapter, client *databricks.W
 	remoteAfterDelete, err := adapter.DoRefresh(ctx, createdID)
 	require.Error(t, err)
 	require.Nil(t, remoteAfterDelete)
+
+	path, err := structpath.Parse("name")
+	require.NoError(t, err)
+
+	_, err = adapter.ClassifyChange(structdiff.Change{
+		Path: path,
+		Old:  nil,
+		New:  "mynewname",
+	}, remote)
+	require.NoError(t, err)
 }
 
 // validateFields uses structwalk to generate all valid field paths and checks membership.