Added support for pipeline environment field (#3153)

## Changes
Added support for pipeline `environment` field

## Why
This allows setting libraries as dependencies for pipeline serverless
environments in the form of
```
  pipelines:
    test_pipeline:
      name: "Test Pipeline"
      environment:
        dependencies:
          - "dist/*.whl"
```

## Tests
Added an acceptance test

<!-- 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

NEXT_CHANGELOG.md

@@ -11,6 +11,7 @@
 * Fixed an issue where running `databricks auth login` would remove the `cluster_id` field from profiles in `.databrickscfg`. The login process now preserves the `cluster_id` field. ([#2988](https://github.com/databricks/cli/pull/2988))
 
 ### Bundles
+* Added support for pipeline environment field ([#3153](https://github.com/databricks/cli/pull/3153))
 * "bundle summary" now prints diagnostic warnings to stderr ([#3123](https://github.com/databricks/cli/pull/3123))
 
 ### API Changes

acceptance/bundle/environments/dependencies/databricks.yml

@@ -29,3 +29,16 @@ resources:
               - "beautifulsoup4>=1.0.0,~=1.2.0,<2.0.0"
               - "beautifulsoup4[security, tests] ~= 4.12.3"
               - "requests[security] @ https://github.com/psf/requests/archive/refs/heads/main.zip"
+
+  pipelines:
+    test_pipeline:
+      name: "Test Pipeline"
+      environment:
+        dependencies:
+          - "dist/*.whl"
+          - "/Workspace/Users/test@databricks.com/test-package.whl"
+          - "test_package"
+          - "test_package==2.0.1"
+          - "test_package>=2.0.1"
+          - "beautifulsoup4>=1.0.0,~=1.2.0,<2.0.0"
+          - "requests[security] @ https://github.com/psf/requests/archive/refs/heads/main.zip"

acceptance/bundle/environments/dependencies/output.txt

@@ -45,6 +45,32 @@ Deployment complete!
   }
 ]
 
+>>> jq -s .[] | select(.path=="/api/2.0/pipelines") out.requests.txt
+{
+  "method": "POST",
+  "path": "/api/2.0/pipelines",
+  "body": {
+    "channel": "CURRENT",
+    "deployment": {
+      "kind": "BUNDLE",
+      "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dependencies/default/state/metadata.json"
+    },
+    "edition": "ADVANCED",
+    "environment": {
+      "dependencies": [
+        "/Workspace/Users/[USERNAME]/.bundle/dependencies/default/artifacts/.internal/test.whl",
+        "/Workspace/Users/test@databricks.com/test-package.whl",
+        "test_package",
+        "test_package==2.0.1",
+        "test_package>=2.0.1",
+        "beautifulsoup4>=1.0.0,~=1.2.0,<2.0.0",
+        "requests[security] @ https://github.com/psf/requests/archive/refs/heads/main.zip"
+      ]
+    },
+    "name": "Test Pipeline"
+  }
+}
+
 >>> [CLI] bundle validate -o json
 [
   {
@@ -74,3 +100,16 @@ Deployment complete!
     }
   }
 ]
+
+>>> [CLI] bundle validate -o json
+{
+  "dependencies": [
+    "./dist/*.whl",
+    "/Workspace/Users/test@databricks.com/test-package.whl",
+    "test_package",
+    "test_package==2.0.1",
+    "test_package>=2.0.1",
+    "beautifulsoup4>=1.0.0,~=1.2.0,<2.0.0",
+    "requests[security] @ https://github.com/psf/requests/archive/refs/heads/main.zip"
+  ]
+}

acceptance/bundle/environments/dependencies/script

@@ -5,5 +5,10 @@ trace $CLI bundle validate
 
 trace $CLI bundle deploy
 trace jq -s '.[] | select(.path=="/api/2.2/jobs/create") | .body.environments' out.requests.txt
+
+trace jq -s '.[] | select(.path=="/api/2.0/pipelines")' out.requests.txt
+
 trace $CLI bundle validate -o json | jq '.resources.jobs.test_job.environments'
+trace $CLI bundle validate -o json | jq '.resources.pipelines.test_pipeline.environment'
+
 rm out.requests.txt

bundle/config/mutator/paths/pipeline_paths_visitor.go

@@ -1,12 +1,17 @@
 package paths
 
 import (
+	"github.com/databricks/cli/bundle/libraries"
 	"github.com/databricks/cli/libs/dyn"
 )
 
 type pipelineRewritePattern struct {
 	pattern dyn.Pattern
 	mode    TranslateMode
+
+	// If function defined in skipRewrite returns true, we skip rewriting the path.
+	// For example, for environment dependencies, we skip rewriting if the path is not a local library.
+	skipRewrite func(string) bool
 }
 
 func pipelineRewritePatterns() []pipelineRewritePattern {
@@ -17,25 +22,49 @@ func pipelineRewritePatterns() []pipelineRewritePattern {
 		dyn.AnyKey(),
 	)
 
+	pipelineEnvironmentsPatterns := []pipelineRewritePattern{
+		{
+			pattern: dyn.NewPattern(
+				dyn.Key("resources"),
+				dyn.Key("pipelines"),
+				dyn.AnyKey(),
+				dyn.Key("environment"),
+				dyn.Key("dependencies"),
+				dyn.AnyIndex(),
+			),
+			mode: TranslateModeLocalRelativeWithPrefix,
+			skipRewrite: func(s string) bool {
+				return !libraries.IsLibraryLocal(s)
+			},
+		},
+	}
+
 	// Compile list of configuration paths to rewrite.
-	return []pipelineRewritePattern{
+	allPatterns := []pipelineRewritePattern{
 		{
-			pattern: base.Append(dyn.Key("libraries"), dyn.AnyIndex(), dyn.Key("notebook"), dyn.Key("path")),
-			mode:    TranslateModeNotebook,
+			pattern:     base.Append(dyn.Key("libraries"), dyn.AnyIndex(), dyn.Key("notebook"), dyn.Key("path")),
+			mode:        TranslateModeNotebook,
+			skipRewrite: noSkipRewrite,
 		},
 		{
-			pattern: base.Append(dyn.Key("libraries"), dyn.AnyIndex(), dyn.Key("file"), dyn.Key("path")),
-			mode:    TranslateModeFile,
+			pattern:     base.Append(dyn.Key("libraries"), dyn.AnyIndex(), dyn.Key("file"), dyn.Key("path")),
+			mode:        TranslateModeFile,
+			skipRewrite: noSkipRewrite,
 		},
 		{
-			pattern: base.Append(dyn.Key("libraries"), dyn.AnyIndex(), dyn.Key("glob"), dyn.Key("include")),
-			mode:    TranslateModeGlob,
+			pattern:     base.Append(dyn.Key("libraries"), dyn.AnyIndex(), dyn.Key("glob"), dyn.Key("include")),
+			mode:        TranslateModeGlob,
+			skipRewrite: noSkipRewrite,
 		},
 		{
-			pattern: base.Append(dyn.Key("root_path")),
-			mode:    TranslateModeDirectory,
+			pattern:     base.Append(dyn.Key("root_path")),
+			mode:        TranslateModeDirectory,
+			skipRewrite: noSkipRewrite,
 		},
 	}
+
+	allPatterns = append(allPatterns, pipelineEnvironmentsPatterns...)
+	return allPatterns
 }
 
 func VisitPipelinePaths(value dyn.Value, fn VisitFunc) (dyn.Value, error) {
@@ -44,6 +73,14 @@ func VisitPipelinePaths(value dyn.Value, fn VisitFunc) (dyn.Value, error) {
 
 	for _, rewritePattern := range pipelineRewritePatterns() {
 		newValue, err = dyn.MapByPattern(newValue, rewritePattern.pattern, func(p dyn.Path, v dyn.Value) (dyn.Value, error) {
+			sv, ok := v.AsString()
+			if !ok {
+				return v, nil
+			}
+			if rewritePattern.skipRewrite(sv) {
+				return v, nil
+			}
+
 			return fn(p, rewritePattern.mode, v)
 		})
 		if err != nil {

bundle/config/mutator/paths/pipeline_paths_visitor_test.go

@@ -34,6 +34,11 @@ func TestVisitPipelinePaths(t *testing.T) {
 								},
 							},
 						},
+						Environment: &pipelines.PipelinesEnvironment{
+							Dependencies: []string{
+								"src/foo.whl",
+							},
+						},
 					},
 				},
 			},
@@ -46,6 +51,7 @@ func TestVisitPipelinePaths(t *testing.T) {
 		dyn.MustPathFromString("resources.pipelines.pipeline0.libraries[1].notebook.path"),
 		dyn.MustPathFromString("resources.pipelines.pipeline0.libraries[2].glob.include"),
 		dyn.MustPathFromString("resources.pipelines.pipeline0.root_path"),
+		dyn.MustPathFromString("resources.pipelines.pipeline0.environment.dependencies[0]"),
 	}
 
 	assert.ElementsMatch(t, expected, actual)

bundle/libraries/expand_glob_references.go

@@ -165,6 +165,14 @@ var envDepsPattern = dyn.NewPattern(
 	dyn.Key("dependencies"),
 )
 
+var pipelineEnvDepsPattern = dyn.NewPattern(
+	dyn.Key("resources"),
+	dyn.Key("pipelines"),
+	dyn.AnyKey(),
+	dyn.Key("environment"),
+	dyn.Key("dependencies"),
+)
+
 func (e *expand) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
 	expanders := []expandPattern{
 		{
@@ -179,6 +187,10 @@ func (e *expand) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
 			pattern: envDepsPattern,
 			fn:      expandEnvironmentDeps,
 		},
+		{
+			pattern: pipelineEnvDepsPattern,
+			fn:      expandEnvironmentDeps,
+		},
 	}
 
 	var diags diag.Diagnostics

bundle/libraries/same_name_libraries.go

@@ -18,6 +18,7 @@ var patterns = []dyn.Pattern{
 	forEachTaskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("whl")),
 	forEachTaskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("jar")),
 	envDepsPattern.Append(dyn.AnyIndex()),
+	pipelineEnvDepsPattern.Append(dyn.AnyIndex()),
 }
 
 type libData struct {

bundle/libraries/upload.go

@@ -59,6 +59,7 @@ func collectLocalLibraries(b *bundle.Bundle) (map[string][]configLocation, error
 		forEachTaskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("whl")),
 		forEachTaskLibrariesPattern.Append(dyn.AnyIndex(), dyn.Key("jar")),
 		envDepsPattern.Append(dyn.AnyIndex()),
+		pipelineEnvDepsPattern.Append(dyn.AnyIndex()),
 	}
 
 	for _, pattern := range patterns {