diff --git a/.nextchanges/bundles/gateway-pipeline-backend-fields.md b/.nextchanges/bundles/gateway-pipeline-backend-fields.md new file mode 100644 index 00000000000..9baef66b464 --- /dev/null +++ b/.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)) diff --git a/acceptance/bundle/resources/pipelines/gateway/databricks.yml.tmpl b/acceptance/bundle/resources/pipelines/gateway/databricks.yml.tmpl new file mode 100644 index 00000000000..0dd87f2556a --- /dev/null +++ b/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} diff --git a/acceptance/bundle/resources/pipelines/gateway/out.test.toml b/acceptance/bundle/resources/pipelines/gateway/out.test.toml new file mode 100644 index 00000000000..669a6b593c4 --- /dev/null +++ b/acceptance/bundle/resources/pipelines/gateway/out.test.toml @@ -0,0 +1,3 @@ +Cloud = true +CloudEnvs.gcp = false +EnvMatrix.DMS = ["", "true"] diff --git a/acceptance/bundle/resources/pipelines/gateway/output.txt b/acceptance/bundle/resources/pipelines/gateway/output.txt new file mode 100644 index 00000000000..b2fb4211393 --- /dev/null +++ b/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 diff --git a/acceptance/bundle/resources/pipelines/gateway/script b/acceptance/bundle/resources/pipelines/gateway/script new file mode 100644 index 00000000000..804c4c8f39d --- /dev/null +++ b/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 diff --git a/acceptance/bundle/resources/pipelines/gateway/test.toml b/acceptance/bundle/resources/pipelines/gateway/test.toml new file mode 100644 index 00000000000..dfd08e8cc41 --- /dev/null +++ b/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"] diff --git a/bundle/direct/dresources/configs/pipelines.yml b/bundle/direct/dresources/configs/pipelines.yml index 59b92f268d6..b029f039cfe 100644 --- a/bundle/direct/dresources/configs/pipelines.yml +++ b/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 diff --git a/libs/testserver/connections.go b/libs/testserver/connections.go new file mode 100644 index 00000000000..8f52ec6d00e --- /dev/null +++ b/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} +} diff --git a/libs/testserver/fake_workspace.go b/libs/testserver/fake_workspace.go index b73803751b6..b32496cf4c6 100644 --- a/libs/testserver/fake_workspace.go +++ b/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{}, diff --git a/libs/testserver/handlers.go b/libs/testserver/handlers.go index a56e2dc4c0b..8fa193d384a 100644 --- a/libs/testserver/handlers.go +++ b/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 { diff --git a/libs/testserver/pipelines.go b/libs/testserver/pipelines.go index 0aedc44c5d3..95a9df988e7 100644 --- a/libs/testserver/pipelines.go +++ b/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{}