Speed up loading bundles with many included files

.nextchanges/bundles/include-load-performance.md

@@ -0,0 +1 @@
+* Improved configuration load time for bundles with many included files. ([#6195](https://github.com/databricks/cli/pull/6195))

acceptance/bundle/includes/plan_multifile/databricks.yml

@@ -0,0 +1,17 @@
+bundle:
+  name: plan_multifile
+
+# Resources are spread across separate files matched by two globs, plus one job
+# defined directly in the root file. The plan must account for every one of them.
+include:
+  - resources/*.yml
+  - resources/*/*.yml
+
+resources:
+  jobs:
+    root_job:
+      name: root_job
+
+targets:
+  dev:
+    default: true

acceptance/bundle/includes/plan_multifile/output.txt

@@ -0,0 +1,16 @@
+
+>>> [CLI] bundle plan
+Warning: unknown field: unknown_field
+  at resources.jobs.job_d
+  in resources/nested/job_d.yml:8:7
+
+create jobs.job_a
+create jobs.job_b
+create jobs.job_d
+create jobs.root_job
+create pipelines.pipeline_c
+
+Plan: 5 to add, 0 to change, 0 to delete, 0 unchanged
+
+=== job_a merged max_concurrent_runs from the override file
+4

acceptance/bundle/includes/plan_multifile/resources/job_a.yml

@@ -0,0 +1,8 @@
+resources:
+  jobs:
+    job_a:
+      name: job_a
+      tasks:
+        - task_key: main
+          notebook_task:
+            notebook_path: ./nb.py

acceptance/bundle/includes/plan_multifile/resources/job_b.yml

@@ -0,0 +1,8 @@
+resources:
+  jobs:
+    job_b:
+      name: job_b
+      tasks:
+        - task_key: main
+          notebook_task:
+            notebook_path: ./nb.py

acceptance/bundle/includes/plan_multifile/resources/nb.py

@@ -0,0 +1,2 @@
+# Databricks notebook source
+print("hello")

acceptance/bundle/includes/plan_multifile/resources/nested/job_d.yml

@@ -0,0 +1,12 @@
+resources:
+  jobs:
+    job_d:
+      name: job_d
+      # Unknown field in a nested included file. `bundle validate` must report its
+      # location as this file, which verifies that source locations survive the
+      # in-scope include merge this PR introduces for load performance.
+      unknown_field: true
+      tasks:
+        - task_key: main
+          notebook_task:
+            notebook_path: ../nb.py

acceptance/bundle/includes/plan_multifile/resources/override_job_a.yml

@@ -0,0 +1,9 @@
+# Target override for job_a lives in a different file than its base definition.
+# UniqueResourceKeys allows a resource to span files inside a target block, so this
+# must merge into job_a rather than error or produce a second resource.
+targets:
+  dev:
+    resources:
+      jobs:
+        job_a:
+          max_concurrent_runs: 4

acceptance/bundle/includes/plan_multifile/resources/pipeline_c.yml

@@ -0,0 +1,7 @@
+resources:
+  pipelines:
+    pipeline_c:
+      name: pipeline_c
+      libraries:
+        - notebook:
+            path: ./nb.py

acceptance/bundle/includes/plan_multifile/script

@@ -0,0 +1,11 @@
+# Every resource defined across the included files (two globs + the root file) must
+# appear in the plan exactly once, and the cross-file target override must merge into
+# job_a rather than be dropped. The unknown field in resources/nested/job_d.yml is
+# reported against that file, verifying source locations survive the in-scope include
+# merge this PR introduces for load performance.
+trace $CLI bundle plan
+
+title "job_a merged max_concurrent_runs from the override file\n"
+# The located warning is already asserted by the plan above; drop the duplicate here so
+# this step only shows the merged value.
+trace $CLI bundle validate -o json 2>/dev/null | jq '.resources.jobs.job_a.max_concurrent_runs'

bundle/config/loader/process_root_includes.go

@@ -10,6 +10,7 @@ import (
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/dyn"
 )
 
 type processRootIncludes struct{}
@@ -130,13 +131,35 @@ func (m *processRootIncludes) Apply(ctx context.Context, b *bundle.Bundle) diag.
 		}
 	}
 
-	// Swap out the original includes list with the expanded globs.
-	b.Config.Include = files
+	// Swap out the original includes list with the expanded globs. This goes through
+	// Mutate so the dynamic tree is updated too: the includes below are applied without
+	// their own mutator scope, so nothing converts the typed field back into the dynamic
+	// tree afterwards, and the next ToTyped would otherwise restore the raw patterns.
+	err := b.Config.Mutate(func(root dyn.Value) (dyn.Value, error) {
+		// Include is omitempty in the typed configuration, so an empty list must stay
+		// absent from the dynamic tree rather than be written as [].
+		if len(files) == 0 {
+			return dyn.DropKeys(root, []string{"include"})
+		}
+
+		includeValues := make([]dyn.Value, 0, len(files))
+		for _, file := range files {
+			includeValues = append(includeValues, dyn.V(file))
+		}
+		return dyn.Set(root, "include", dyn.NewValue(includeValues, root.Get("include").Locations()))
+	})
+	if err != nil {
+		return diag.FromErr(err)
+	}
 
 	// Track number of bundle YAML (or JSON) files in the configuration. The +1 is there
 	// to account for the root databricks.yaml file.
 	b.Metrics.ConfigurationFileCount = int64(len(files)) + 1
 
-	bundle.ApplySeqContext(ctx, b, out...)
+	// ProcessInclude merges into the configuration via [config.Root.Merge], so it does
+	// not need its own mutator scope. Giving each included file one would re-convert the
+	// whole accumulated configuration per file, making load quadratic in the number of
+	// included files (~20 minutes for 6000 files).
+	bundle.ApplySeqInScopeContext(ctx, b, out...)
 	return nil
 }

bundle/config/loader/process_root_includes_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/databricks/cli/bundle/config/loader"
 	"github.com/databricks/cli/internal/testutil"
 	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/dyn"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -98,6 +99,49 @@ func TestProcessRootIncludesRemoveDups(t *testing.T) {
 	assert.Equal(t, []string{"a.yml"}, b.Config.Include)
 }
 
+// The expanded include list must be visible in both the typed and the dynamic
+// configuration: the per-file includes are applied without their own mutator scope, so
+// nothing converts the typed field back into the dynamic tree afterwards.
+func TestProcessRootIncludesUpdatesDynamicValue(t *testing.T) {
+	b := &bundle.Bundle{
+		BundleRootPath: t.TempDir(),
+		Config: config.Root{
+			Include: []string{
+				"*.yml",
+			},
+		},
+	}
+
+	testutil.Touch(t, b.BundleRootPath, "databricks.yml")
+	testutil.Touch(t, b.BundleRootPath, "a.yml")
+
+	diags := bundle.Apply(t.Context(), b, loader.ProcessRootIncludes())
+	require.NoError(t, diags.Error())
+	assert.Equal(t, []string{"a.yml"}, b.Config.Include)
+
+	assert.Equal(t, []any{"a.yml"}, b.Config.Value().Get("include").AsAny())
+}
+
+// An empty include list must stay absent from the dynamic tree: the typed field is
+// omitempty, so writing [] would add an empty "include" to `bundle validate -o json`.
+func TestProcessRootIncludesEmptyOmitsDynamicValue(t *testing.T) {
+	b := &bundle.Bundle{
+		BundleRootPath: t.TempDir(),
+		Config: config.Root{
+			Include: []string{
+				"*.yml",
+			},
+		},
+	}
+
+	testutil.Touch(t, b.BundleRootPath, "databricks.yml")
+
+	diags := bundle.Apply(t.Context(), b, loader.ProcessRootIncludes())
+	require.NoError(t, diags.Error())
+	assert.Empty(t, b.Config.Include)
+	assert.Equal(t, dyn.KindInvalid, b.Config.Value().Get("include").Kind())
+}
+
 func TestProcessRootIncludesNotExists(t *testing.T) {
 	b := &bundle.Bundle{
 		BundleRootPath: t.TempDir(),

bundle/mutator.go

@@ -104,6 +104,34 @@ func ApplySeqContext(ctx context.Context, b *Bundle, mutators ...Mutator) {
 	}
 }
 
+// ApplySeqInScopeContext applies mutators without opening a mutator scope per mutator,
+// reusing the caller's scope instead.
+//
+// [ApplyContext] converts the whole configuration tree between its typed and dynamic
+// representations on entry and exit (see [config.Root.MarkMutatorEntry]). That cost is
+// proportional to the size of the accumulated configuration, so applying N mutators this
+// way is quadratic in N. For a bundle with thousands of included files that dominates
+// load time, hence this variant.
+//
+// Only use it for mutators that modify the configuration through [config.Root.Mutate]
+// (which keeps both representations in sync). A mutator that assigns to a typed field
+// directly relies on the scope entry to carry that value into the dynamic tree, and
+// would lose it here.
+func ApplySeqInScopeContext(ctx context.Context, b *Bundle, mutators ...Mutator) {
+	for _, m := range mutators {
+		mctx := log.NewContext(ctx, log.GetLogger(ctx).With("mutator", m.Name()))
+		log.Debugf(mctx, "Apply")
+
+		for _, d := range m.Apply(mctx, b) {
+			logdiag.LogDiag(mctx, d)
+		}
+
+		if logdiag.HasError(ctx) {
+			break
+		}
+	}
+}
+
 type funcMutator struct {
 	fn func(context.Context, *Bundle)
 }

bundle/mutator_test.go

@@ -6,12 +6,16 @@ import (
 
 	"github.com/databricks/cli/libs/diag"
 	"github.com/databricks/cli/libs/dyn"
+	"github.com/databricks/cli/libs/logdiag"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 type testMutator struct {
 	applyCalled    int
 	nestedMutators []Mutator
+	// fn, if set, runs inside Apply after the call is counted.
+	fn func(ctx context.Context, b *Bundle) diag.Diagnostics
 }
 
 func (t *testMutator) Name() string {
@@ -20,6 +24,9 @@ func (t *testMutator) Name() string {
 
 func (t *testMutator) Apply(ctx context.Context, b *Bundle) diag.Diagnostics {
 	t.applyCalled++
+	if t.fn != nil {
+		return t.fn(ctx, b)
+	}
 	return ApplySeq(ctx, b, t.nestedMutators...)
 }
 
@@ -45,6 +52,79 @@ func TestMutator(t *testing.T) {
 	assert.Equal(t, 1, nested[1].applyCalled)
 }
 
+// ApplySeqInScopeContext applies each mutator once, in order.
+func TestApplySeqInScopeContext(t *testing.T) {
+	var order []string
+	makeMutator := func(name string) *testMutator {
+		return &testMutator{fn: func(ctx context.Context, b *Bundle) diag.Diagnostics {
+			order = append(order, name)
+			return nil
+		}}
+	}
+	first := makeMutator("first")
+	second := makeMutator("second")
+
+	b := &Bundle{}
+	ctx := logdiag.InitContext(t.Context())
+	logdiag.SetCollect(ctx, true)
+	ApplySeqInScopeContext(ctx, b, first, second)
+
+	assert.Equal(t, 1, first.applyCalled)
+	assert.Equal(t, 1, second.applyCalled)
+	assert.Equal(t, []string{"first", "second"}, order)
+	assert.Empty(t, logdiag.FlushCollected(ctx))
+}
+
+// ApplySeqInScopeContext collects the diagnostics returned by each mutator and
+// stops at the first one that logs an error, without applying later mutators.
+func TestApplySeqInScopeContextStopsOnError(t *testing.T) {
+	failing := &testMutator{fn: func(ctx context.Context, b *Bundle) diag.Diagnostics {
+		return diag.Diagnostics{diag.Diagnostic{Severity: diag.Error, Summary: "boom"}}
+	}}
+	later := &testMutator{}
+
+	b := &Bundle{}
+	ctx := logdiag.InitContext(t.Context())
+	logdiag.SetCollect(ctx, true)
+	ApplySeqInScopeContext(ctx, b, failing, later)
+
+	assert.Equal(t, 1, failing.applyCalled)
+	assert.Equal(t, 0, later.applyCalled, "mutator after an error must not run")
+
+	diags := logdiag.FlushCollected(ctx)
+	require.Len(t, diags, 1)
+	assert.Equal(t, "boom", diags[0].Summary)
+}
+
+// Unlike ApplySeqContext, ApplySeqInScopeContext does not open a mutator scope per
+// mutator, so it must be called from within one. Changes made through Root.Mutate
+// keep the typed and dynamic configuration in sync and survive without a per-mutator
+// scope. This is the property ProcessRootIncludes relies on.
+func TestApplySeqInScopeContextPreservesMutateChanges(t *testing.T) {
+	setHost := &testMutator{fn: func(ctx context.Context, b *Bundle) diag.Diagnostics {
+		err := b.Config.Mutate(func(v dyn.Value) (dyn.Value, error) {
+			return dyn.Set(v, "bundle", dyn.V(map[string]dyn.Value{"name": dyn.V("set-in-scope")}))
+		})
+		require.NoError(t, err)
+		return nil
+	}}
+
+	// Enclosing mutator that applies setHost in its own scope, mirroring how
+	// ProcessRootIncludes applies the per-file includes.
+	outer := &testMutator{fn: func(ctx context.Context, b *Bundle) diag.Diagnostics {
+		ApplySeqInScopeContext(ctx, b, setHost)
+		return nil
+	}}
+
+	b := &Bundle{}
+	diags := Apply(t.Context(), b, outer)
+	require.NoError(t, diags.Error())
+
+	// Visible in both representations once the enclosing scope exits.
+	assert.Equal(t, "set-in-scope", b.Config.Bundle.Name)
+	assert.Equal(t, "set-in-scope", b.Config.Value().Get("bundle").Get("name").MustString())
+}
+
 func TestSafeMutatorName(t *testing.T) {
 	tests := []struct {
 		name     string