direct: Fix continuous pipeline restart on every deploy from channel/edition case

.nextchanges/bundles/pipeline-normalize-case.md

@@ -0,0 +1 @@
+* direct: Fix a continuous pipeline being cancelled and restarted on every `bundle deploy` when its `channel`, `edition`, or `catalog` differs only in letter case from the backend's canonical value. ([#6747](https://github.com/databricks/cli/pull/6747))

acceptance/bundle/invariant/configs/pipeline_normalize_case.yml.tmpl

@@ -0,0 +1,19 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+resources:
+  pipelines:
+    foo:
+      name: test-pipeline-$UNIQUE_NAME
+      # channel/edition are stored by the backend in canonical upper case and catalog in
+      # lower case. Declaring them in a different case here must still converge
+      # (normalize_case), otherwise every deploy replays an update -- which cancels a
+      # continuous pipeline's running update.
+      catalog: MAIN
+      target: test_schema_$UNIQUE_NAME
+      continuous: true
+      channel: current
+      edition: pro
+      libraries:
+        - file:
+            path: pipeline.py

acceptance/bundle/invariant/migrate/test.toml

@@ -49,6 +49,11 @@ EnvMatrixExclude.no_pydabs_1000_tasks = ["INPUT_CONFIG=job_pydabs_1000_tasks.yml
 # volume schema_name ("inconsistent final plan"). Covered by no_drift on direct.
 EnvMatrixExclude.no_volume_uppercase = ["INPUT_CONFIG=volume_uppercase_name.yml.tmpl"]
 
+# migrate deploys via Terraform first, and for a continuous pipeline the TF provider blocks
+# until the pipeline's update reaches RUNNING, which the local mock never reports (it stays
+# IDLE), so the deploy hangs until the test times out. Covered by no_drift on direct.
+EnvMatrixExclude.no_pipeline_normalize_case = ["INPUT_CONFIG=pipeline_normalize_case.yml.tmpl"]
+
 EnvMatrixExclude.no_secret = ["INPUT_CONFIG=secret.yml.tmpl"]
 
 # Terraform types sampling_fraction as an integer and truncates 0.5; covered by no_drift.

acceptance/bundle/invariant/test.toml

@@ -53,6 +53,7 @@ EnvMatrix.INPUT_CONFIG = [
     "model_serving_endpoint_telemetry.yml.tmpl",
     "pipeline.yml.tmpl",
     "pipeline_allow_duplicate_names.yml.tmpl",
+    "pipeline_normalize_case.yml.tmpl",
     "pipeline_apply_policy_default_values.yml.tmpl",
     "pipeline_config_dots.yml.tmpl",
     "postgres_branch.yml.tmpl",

bundle/direct/bundle_plan.go

@@ -666,11 +666,12 @@ func classifyIDField(cfg *dresources.ResourceLifecycleConfig, path *structpath.P
 	return deployplan.Undefined, "", false
 }
 
-// shouldSkipNormalized skips a change that is a false diff caused by UC API
-// normalization: the API strips trailing slashes from storage URLs
-// (normalize_slash). The direct engine saves local config to state, so without
-// this the next plan sees the original value against the normalized remote value
-// and triggers a spurious recreate/update.
+// shouldSkipNormalized skips a change that is a false diff caused by backend
+// normalization: the UC API strips trailing slashes from storage URLs
+// (normalize_slash), and some fields are stored in a canonical case such as
+// upper-case enums (normalize_case). The direct engine saves local config to
+// state, so without this the next plan sees the original value against the
+// normalized remote value and triggers a spurious recreate/update.
 func shouldSkipNormalized(cfg *dresources.ResourceLifecycleConfig, path *structpath.PathNode, ch *deployplan.ChangeDesc) (string, bool) {
 	if cfg == nil {
 		return "", false
@@ -680,7 +681,17 @@ func shouldSkipNormalized(cfg *dresources.ResourceLifecycleConfig, path *structp
 	if !newOk || !remoteOk {
 		return "", false
 	}
-	if reason, ok := findMatchingRule(path, cfg.NormalizeSlash); ok && strings.TrimRight(newStr, "/") == strings.TrimRight(remoteStr, "/") {
+	// normalize_slash strips trailing slashes and normalize_case folds letter case. A field
+	// under both rules is normalized on both axes, so trim first and then fold: a value
+	// differing by slash and case at once still converges.
+	newVal, remoteVal := newStr, remoteStr
+	if reason, ok := findMatchingRule(path, cfg.NormalizeSlash); ok {
+		newVal, remoteVal = strings.TrimRight(newVal, "/"), strings.TrimRight(remoteVal, "/")
+		if newVal == remoteVal {
+			return reason, true
+		}
+	}
+	if reason, ok := findMatchingRule(path, cfg.NormalizeCase); ok && strings.EqualFold(newVal, remoteVal) {
 		return reason, true
 	}
 	return "", false

bundle/direct/bundle_plan_test.go

@@ -604,3 +604,108 @@ func TestLadderBackendDefaultBeforeRemoteAddition(t *testing.T) {
 	assert.Equal(t, deployplan.Skip, changes["enable_elastic_disk"].Action)
 	assert.Equal(t, deployplan.ReasonBackendDefault, changes["enable_elastic_disk"].Reason)
 }
+
+func TestShouldSkipNormalized(t *testing.T) {
+	slashPat, err := structpath.ParsePattern("storage_root")
+	require.NoError(t, err)
+	casePat, err := structpath.ParsePattern("channel")
+	require.NoError(t, err)
+	bothPat, err := structpath.ParsePattern("weird")
+	require.NoError(t, err)
+
+	cfg := &dresources.ResourceLifecycleConfig{
+		NormalizeSlash: []dresources.FieldRule{{Field: slashPat, Reason: "slash"}, {Field: bothPat, Reason: "slash"}},
+		NormalizeCase:  []dresources.FieldRule{{Field: casePat, Reason: "case"}, {Field: bothPat, Reason: "case"}},
+	}
+
+	tests := []struct {
+		name       string
+		path       string
+		newVal     any
+		remoteVal  any
+		wantReason string
+		wantSkip   bool
+	}{
+		{
+			name:       "slash rule, slash-only diff",
+			path:       "storage_root",
+			newVal:     "s3://b/x/",
+			remoteVal:  "s3://b/x",
+			wantReason: "slash",
+			wantSkip:   true,
+		},
+		{
+			name:       "slash rule stays case-sensitive",
+			path:       "storage_root",
+			newVal:     "s3://b/X",
+			remoteVal:  "s3://b/x",
+			wantReason: "",
+			wantSkip:   false,
+		},
+		{
+			name:       "case rule, case-only diff",
+			path:       "channel",
+			newVal:     "current",
+			remoteVal:  "CURRENT",
+			wantReason: "case",
+			wantSkip:   true,
+		},
+		{
+			name:       "case rule, genuine diff",
+			path:       "channel",
+			newVal:     "current",
+			remoteVal:  "preview",
+			wantReason: "",
+			wantSkip:   false,
+		},
+		{
+			name:       "both rules, slash-only diff",
+			path:       "weird",
+			newVal:     "Foo/",
+			remoteVal:  "Foo",
+			wantReason: "slash",
+			wantSkip:   true,
+		},
+		{
+			name:       "both rules, case-only diff",
+			path:       "weird",
+			newVal:     "PRO",
+			remoteVal:  "pro",
+			wantReason: "case",
+			wantSkip:   true,
+		},
+		{
+			name:       "both rules, slash and case diff",
+			path:       "weird",
+			newVal:     "Foo/",
+			remoteVal:  "foo",
+			wantReason: "case",
+			wantSkip:   true,
+		},
+		{
+			name:       "both rules, genuine diff",
+			path:       "weird",
+			newVal:     "foo",
+			remoteVal:  "bar",
+			wantReason: "",
+			wantSkip:   false,
+		},
+		{
+			name:       "non-string change",
+			path:       "channel",
+			newVal:     1,
+			remoteVal:  2,
+			wantReason: "",
+			wantSkip:   false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			path, err := structpath.ParsePath(tt.path)
+			require.NoError(t, err)
+			reason, ok := shouldSkipNormalized(cfg, path, &deployplan.ChangeDesc{New: tt.newVal, Remote: tt.remoteVal})
+			assert.Equal(t, tt.wantSkip, ok)
+			assert.Equal(t, tt.wantReason, reason)
+		})
+	}
+}

bundle/direct/dresources/config.go

@@ -91,6 +91,11 @@ type ResourceLifecycleConfig struct {
 	// A change is skipped when local and remote differ only by trailing slashes.
 	NormalizeSlash []FieldRule `yaml:"normalize_slash,omitempty"`
 
+	// NormalizeCase: string field patterns the backend stores in a canonical case
+	// (e.g. upper-case enums like a pipeline's channel/edition). A change is skipped
+	// when local and remote differ only by case.
+	NormalizeCase []FieldRule `yaml:"normalize_case,omitempty"`
+
 	// IgnoreRemoteAdditions: objects whose fields the backend may add to when a gate field
 	// is set. A field that is absent from both old and new state but present in the remote
 	// is skipped; a disagreement between config and remote is still an update.
@@ -138,6 +143,7 @@ var empty = ResourceLifecycleConfig{
 	ProvidedIDFields:      nil,
 	UpdatableIDFields:     nil,
 	NormalizeSlash:        nil,
+	NormalizeCase:         nil,
 	IgnoreRemoteAdditions: nil,
 	BackendDefaults:       nil,
 	HashedFields:          nil,

bundle/direct/dresources/configs/pipelines.yml

@@ -41,6 +41,19 @@ ignore_remote_changes:
   - field: allow_duplicate_names
     reason: input_only
 
+# The backend stores these in a canonical case and normalizes whatever the request sent,
+# so a config value in a different case reads back changed and never converges. The
+# terraform provider suppresses the same diffs (channel via SetSuppressDiff; edition and
+# catalog via EqualFoldDiffSuppress), so this is not a regression when migrating to direct.
+# https://github.com/databricks/terraform-provider-databricks/blob/4eba541abe1a9f50993ea7b9dd83874207e224a1/pipelines/resource_pipeline.go#L176-L181
+normalize_case:
+  - field: channel
+    reason: backend_normalizes_case
+  - field: edition
+    reason: backend_normalizes_case
+  - field: catalog
+    reason: backend_normalizes_case
+
 ignore_local_changes:
   - field: deployment.version_id
     reason: auto

libs/testserver/pipelines.go

@@ -96,6 +96,13 @@ func setSpecDefaults(spec *pipelines.PipelineSpec, pipelineId string) {
 	if spec.Storage == "" && spec.Catalog == "" {
 		spec.Storage = "dbfs:/pipelines/" + pipelineId
 	}
+	// The backend stores channel and edition as canonical upper-case enums (CURRENT/PREVIEW,
+	// CORE/PRO/ADVANCED) and catalog as a lower-case UC identifier, normalizing whatever case
+	// the request used. A deploy that sends a different case then reads back the canonical
+	// value; without case-insensitive diffing this drifts on every plan. Mirror that here.
+	spec.Channel = strings.ToUpper(spec.Channel)
+	spec.Edition = strings.ToUpper(spec.Edition)
+	spec.Catalog = strings.ToLower(spec.Catalog)
 }
 
 func (s *FakeWorkspace) PipelineUpdate(req Request, pipelineId string) Response {