direct: fix perpetual update on ingestion gateway pipelines

.nextchanges/bundles/gateway-pipeline-backend-fields.md

@@ -0,0 +1 @@
+* Fix `bundle deploy` failing on ingestion gateway pipelines that do not declare `continuous`, `catalog` and `schema`, which the backend sets. ([#7016](https://github.com/databricks/cli/pull/7016))

acceptance/bundle/resources/pipelines/gateway/databricks.yml.tmpl

@@ -0,0 +1,17 @@
+bundle:
+  name: acc-$UNIQUE_NAME
+
+resources:
+  schemas:
+    storage:
+      catalog_name: main
+      name: gateway_storage_$UNIQUE_NAME
+
+  pipelines:
+    gw:
+      name: test-gateway-$UNIQUE_NAME
+      # continuous, catalog and schema are deliberately not declared: the backend sets them.
+      gateway_definition:
+        connection_name: conn_$UNIQUE_NAME
+        gateway_storage_catalog: main
+        gateway_storage_schema: ${resources.schemas.storage.name}

acceptance/bundle/resources/pipelines/gateway/out.test.toml

@@ -0,0 +1,3 @@
+Cloud = true
+CloudEnvs.gcp = false
+EnvMatrix.DMS = ["", "true"]

acceptance/bundle/resources/pipelines/gateway/output.txt

@@ -0,0 +1,59 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/acc-[UNIQUE_NAME]/default/files...
+Created pipelines.gw
+Created schemas.storage
+Files: 2 uploaded, 0 deleted
+Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
+
+=== The backend fills in continuous, catalog and schema
+>>> [CLI] pipelines get [GW_ID]
+{
+  "continuous": true,
+  "catalog": "main",
+  "schema": "gateway_storage_[UNIQUE_NAME]"
+}
+
+=== Fields the backend added are not drift
+>>> [CLI] bundle plan -o json
+{
+  "continuous": {
+    "action": "skip",
+    "reason": "remote_addition",
+    "remote": true
+  },
+  "catalog": {
+    "action": "skip",
+    "reason": "remote_addition",
+    "remote": "main"
+  },
+  "schema": {
+    "action": "skip",
+    "reason": "remote_addition",
+    "remote": "gateway_storage_[UNIQUE_NAME]"
+  }
+}
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/acc-[UNIQUE_NAME]/default/files...
+Files: 0 uploaded, 0 deleted
+Resources: 0 created, 0 changed, 0 deleted, 2 unchanged
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.pipelines.gw
+  delete resources.schemas.storage
+
+This action will result in the deletion of the following UC schemas. Any underlying data may be lost:
+  delete resources.schemas.storage
+
+This action will result in the deletion of the following Lakeflow Spark Declarative Pipelines along with the
+Streaming Tables (STs) and Materialized Views (MVs) managed by them. Set 'cascade_on_destroy: false' on a pipeline to retain datasets on pipeline deletion:
+  delete resources.pipelines.gw
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/acc-[UNIQUE_NAME]/default
+
+Destroy: 2 deleted

acceptance/bundle/resources/pipelines/gateway/script

@@ -0,0 +1,26 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+    $CLI connections delete conn_$UNIQUE_NAME
+}
+trap cleanup EXIT
+
+# The connection host is a placeholder: the gateway never reaches a real source.
+$CLI connections create --json '{"name": "conn_'$UNIQUE_NAME'", "connection_type": "SQLSERVER", "options": {"host": "sqlserver.example.test", "port": "1433", "user": "u", "password": "p"}}' > /dev/null
+
+trace $CLI bundle deploy
+
+gateway_fields() {
+    jq '.spec | {continuous, catalog, schema}'
+}
+
+title "The backend fills in continuous, catalog and schema"
+ppid=$(read_id.py gw)
+trace $CLI pipelines get $ppid | gateway_fields
+
+title "Fields the backend added are not drift"
+# Remote values prove the backend set them; the config declares none of them.
+trace $CLI bundle plan -o json | jq '.plan["resources.pipelines.gw"].changes | {continuous, catalog, schema}'
+trace $CLI bundle plan | contains.py "0 to change"
+trace $CLI bundle deploy

acceptance/bundle/resources/pipelines/gateway/test.toml

@@ -0,0 +1,6 @@
+# TODO: add `Cloud = true` once the -cli test envs can create the SQL Server connection and run
+# gateway pipelines (CREATE CONNECTION on the metastore, Lakeflow Connect gateway ingestion enabled).
+# Until then the cloud run fails with "User does not have CREATE CONNECTION". Verified manually on
+# aws-prod-ucws and azure-prod-ucws, which have both, with identical output.
+
+Ignore = ["databricks.yml"]

bundle/direct/dresources/configs/pipelines.yml

@@ -70,3 +70,11 @@ backend_defaults:
   # s.SchemaPath("event_log", "schema").SetComputed()
   - field: event_log.catalog
   - field: event_log.schema
+
+# An ingestion gateway pipeline is always continuous, and the backend copies
+# gateway_definition.gateway_storage_catalog/schema into the top-level catalog/schema, even when
+# the config declares none of them. Without this, every deploy plans an update that unsets them
+# and fails ("Cannot unset 'schema' field"). A value the config does declare still diffs.
+# Pinned by acceptance/bundle/resources/pipelines/gateway.
+ignore_remote_additions:
+  - when_set: gateway_definition

libs/testserver/connections.go

@@ -0,0 +1,49 @@
+package testserver
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+
+	"github.com/databricks/databricks-sdk-go/service/catalog"
+)
+
+func (s *FakeWorkspace) ConnectionsCreate(req Request) Response {
+	defer s.LockUnlock()()
+
+	var createRequest catalog.CreateConnection
+	if err := json.Unmarshal(req.Body, &createRequest); err != nil {
+		return Response{
+			Body:       fmt.Sprintf("internal error: %s", err),
+			StatusCode: http.StatusInternalServerError,
+		}
+	}
+
+	if _, ok := s.Connections[createRequest.Name]; ok {
+		return Response{
+			StatusCode: http.StatusConflict,
+			Body: map[string]string{
+				"error_code": "CONNECTION_ALREADY_EXISTS",
+				"message":    fmt.Sprintf("Connection '%s' already exists.", createRequest.Name),
+			},
+		}
+	}
+
+	info := catalog.ConnectionInfo{
+		Name:           createRequest.Name,
+		FullName:       createRequest.Name,
+		ConnectionId:   nextUUID(),
+		ConnectionType: createRequest.ConnectionType,
+		Comment:        createRequest.Comment,
+		Options:        createRequest.Options,
+		ReadOnly:       createRequest.ReadOnly,
+		Owner:          s.CurrentUser().UserName,
+		CreatedAt:      nowMilli(),
+		CreatedBy:      s.CurrentUser().UserName,
+	}
+	info.UpdatedAt = info.CreatedAt
+	info.UpdatedBy = info.CreatedBy
+
+	s.Connections[createRequest.Name] = info
+	return Response{Body: info}
+}

libs/testserver/fake_workspace.go

@@ -212,6 +212,7 @@ type FakeWorkspace struct {
 	InstancePools         map[string]compute.GetInstancePool
 	ClusterPolicies       map[string]compute.Policy
 	Catalogs              map[string]catalog.CatalogInfo
+	Connections           map[string]catalog.ConnectionInfo
 	ExternalLocations     map[string]catalog.ExternalLocationInfo
 	RegisteredModels      map[string]catalog.RegisteredModelInfo
 	ModelServices         map[string]catalog.ModelService
@@ -491,6 +492,7 @@ func NewFakeWorkspace(url, token string) *FakeWorkspace {
 		Monitors:              map[string]catalog.MonitorInfo{},
 		Apps:                  map[string]apps.App{},
 		Catalogs:              map[string]catalog.CatalogInfo{},
+		Connections:           map[string]catalog.ConnectionInfo{},
 		ExternalLocations:     map[string]catalog.ExternalLocationInfo{},
 		Schemas:               map[string]catalog.SchemaInfo{},
 		RegisteredModels:      map[string]catalog.RegisteredModelInfo{},

libs/testserver/handlers.go

@@ -569,6 +569,19 @@ func AddDefaultHandlers(server *Server) {
 		return req.Workspace.GrantsGet(req, req.Vars["securable_type"], req.Vars["full_name"])
 	})
 
+	// Connections:
+	server.Handle("POST", "/api/2.1/unity-catalog/connections", func(req Request) any {
+		return req.Workspace.ConnectionsCreate(req)
+	})
+
+	server.Handle("GET", "/api/2.1/unity-catalog/connections/{name}", func(req Request) any {
+		return MapGetUC(req.Workspace, req.Workspace.Connections, req.Vars["name"], "Connection")
+	})
+
+	server.Handle("DELETE", "/api/2.1/unity-catalog/connections/{name}", func(req Request) any {
+		return MapDelete(req.Workspace, req.Workspace.Connections, req.Vars["name"])
+	})
+
 	// Catalogs:
 
 	server.Handle("GET", "/api/2.1/unity-catalog/catalogs/{name}", func(req Request) any {

libs/testserver/pipelines.go

@@ -79,6 +79,7 @@ func (s *FakeWorkspace) PipelineCreate(req Request) Response {
 	r.EffectivePublishingMode = pipelines.PublishingModeDefaultPublishingMode
 
 	setSpecDefaults(&spec, pipelineId)
+	setGatewayDefaults(&spec)
 	s.Pipelines[pipelineId] = r
 
 	return Response{
@@ -88,6 +89,19 @@ func (s *FakeWorkspace) PipelineCreate(req Request) Response {
 	}
 }
 
+// setGatewayDefaults mirrors the backend for ingestion gateway pipelines: they are always
+// continuous, and the gateway storage catalog/schema are echoed as the top-level catalog/schema
+// (observed on aws-prod-ucws and azure-prod-ucws), even when the request omits them.
+func setGatewayDefaults(spec *pipelines.PipelineSpec) {
+	gw := spec.GatewayDefinition
+	if gw == nil {
+		return
+	}
+	spec.Continuous = true //nolint:staticcheck // SA1019: pipeline continuous is deprecated in the SDK but remains a supported bundle config field
+	spec.Catalog = gw.GatewayStorageCatalog
+	spec.Schema = gw.GatewayStorageSchema
+}
+
 func setSpecDefaults(spec *pipelines.PipelineSpec, pipelineId string) {
 	spec.Id = pipelineId
 	// If the pipeline definition does not specify a catalog, it switches to Hive metastore mode
@@ -127,6 +141,17 @@ func (s *FakeWorkspace) PipelineUpdate(req Request, pipelineId string) Response
 		}
 	}
 
+	// Observed on aws-prod-ucws: the backend refuses to clear a schema once the spec has one.
+	if item.Spec != nil && item.Spec.Schema != "" && spec.Schema == "" {
+		return Response{
+			StatusCode: 400,
+			Body: map[string]string{
+				"error_code": "INVALID_PARAMETER_VALUE",
+				"message":    "Specified 'schema' field in the pipeline settings is illegal. Reason: Cannot unset 'schema' field once it's defined in the pipeline spec. Please create a new DLT pipeline. For more information about publishing modes, see https://docs.databricks.com/en/dlt/migrate-to-dpm.html.",
+			},
+		}
+	}
+
 	item.Spec = &spec
 	item.Parameters = edit.Parameters
 	// The backend echoes the spec name on GetPipelineResponse.Name; mirror that so a
@@ -138,6 +163,7 @@ func (s *FakeWorkspace) PipelineUpdate(req Request, pipelineId string) Response
 		item.RunAs = edit.RunAs
 	}
 	setSpecDefaults(&spec, pipelineId)
+	setGatewayDefaults(&spec)
 	s.Pipelines[pipelineId] = item
 
 	return Response{}