use IsEqual, fix run_as

Author: denik

acceptance/bundle/migrate/runas/out.plan.json

@@ -37,7 +37,7 @@
       "changes": {
         "run_as": {
           "action": "skip",
-          "reason": "field_triggers",
+          "reason": "override",
           "old": {
             "service_principal_name": "[UUID]"
           },

bundle/direct/bundle_plan.go

@@ -361,7 +361,7 @@ func prepareChanges(ctx context.Context, adapter *dresources.Adapter, localDiff,
 			}
 		} else {
 			entry.Remote = ch.Old
-			if !reflect.DeepEqual(entry.New, ch.New) {
+			if !structdiff.IsEqual(entry.New, ch.New) {
 				// this is not fatal (may result in unexpected drift or undetected change but not incorrect deploy), but good to log this
 				log.Warnf(ctx, "unexpected local and remote diffs (%T, %T); entry=%v ch=%v", entry.New, ch.New, entry, ch)
 			}
@@ -387,7 +387,7 @@ func addPerFieldActions(ctx context.Context, adapter *dresources.Adapter, change
 			// Note, we only consider struct fields here. Adding/removing elements to/from maps and slices should trigger updates.
 			ch.Action = deployplan.ActionTypeSkipString
 			ch.Reason = deployplan.ReasonServerSideDefault
-		} else if reflect.DeepEqual(ch.Remote, ch.New) {
+		} else if structdiff.IsEqual(ch.Remote, ch.New) {
 			ch.Action = deployplan.ActionTypeSkipString
 			ch.Reason = deployplan.ReasonRemoteAlreadySet
 		} else if action, ok := fieldTriggers[pathString]; ok {

bundle/direct/dresources/pipeline.go

@@ -5,6 +5,8 @@ import (
 
 	"github.com/databricks/cli/bundle/config/resources"
 	"github.com/databricks/cli/bundle/deployplan"
+	"github.com/databricks/cli/libs/structs/structdiff"
+	"github.com/databricks/cli/libs/structs/structpath"
 	"github.com/databricks/cli/libs/utils"
 	"github.com/databricks/databricks-sdk-go"
 	"github.com/databricks/databricks-sdk-go/service/pipelines"
@@ -125,12 +127,22 @@ func (*ResourcePipeline) FieldTriggers() map[string]deployplan.ActionType {
 		"storage":                                   deployplan.ActionTypeRecreate,
 		"ingestion_definition.connection_name":      deployplan.ActionTypeRecreate,
 		"ingestion_definition.ingestion_gateway_id": deployplan.ActionTypeRecreate,
-		"run_as": deployplan.ActionTypeSkip,
 	}
 
 	return result
 }
 
+func (*ResourcePipeline) OverrideChangeDesc(ctx context.Context, path *structpath.PathNode, ch *ChangeDesc, _ *pipelines.GetPipelineResponse) error {
+	if path.String() == "run_as" {
+		if structdiff.IsEqual(ch.Old, ch.New) {
+			ch.Action = deployplan.ActionTypeSkipString
+			ch.Reason = "override"
+		}
+	}
+
+	return nil
+}
+
 // Note, terraform provider either
 // a) reads back state at least once and fails create if state is "failed"
 // b) repeatededly reads state until state is "running" (if spec.Contionous is set).