Merge branch 'main' into feature/complex-variables

bundle/config/experimental.go

@@ -23,6 +23,22 @@ type Experimental struct {
 	// be removed in the future once we have a proper workaround like allowing IS_OWNER
 	// as a top-level permission in the DAB.
 	UseLegacyRunAs bool `json:"use_legacy_run_as,omitempty"`
+
+	// PyDABs determines whether to load the 'databricks-pydabs' package.
+	//
+	// PyDABs allows to define bundle configuration using Python.
+	PyDABs PyDABs `json:"pydabs,omitempty"`
+}
+
+type PyDABs struct {
+	// Enabled is a flag to enable the feature.
+	Enabled bool `json:"enabled,omitempty"`
+
+	// VEnvPath is path to the virtual environment.
+	//
+	// Required if PyDABs is enabled. PyDABs will load the code in the specified
+	// environment.
+	VEnvPath string `json:"venv_path,omitempty"`
 }
 
 type Command string

bundle/config/generate/job.go

@@ -17,7 +17,7 @@ func ConvertJobToValue(job *jobs.Job) (dyn.Value, error) {
 		for _, task := range job.Settings.Tasks {
 			v, err := convertTaskToValue(task, taskOrder)
 			if err != nil {
-				return dyn.NilValue, err
+				return dyn.InvalidValue, err
 			}
 			tasks = append(tasks, v)
 		}

bundle/config/mutator/environments_compat.go

@@ -32,18 +32,18 @@ func (m *environmentsToTargets) Apply(ctx context.Context, b *bundle.Bundle) dia
 		targets := v.Get("targets")
 
 		// Return an error if both "environments" and "targets" are set.
-		if environments != dyn.NilValue && targets != dyn.NilValue {
-			return dyn.NilValue, fmt.Errorf(
+		if environments != dyn.InvalidValue && targets != dyn.InvalidValue {
+			return dyn.InvalidValue, fmt.Errorf(
 				"both 'environments' and 'targets' are specified; only 'targets' should be used: %s",
 				environments.Location().String(),
 			)
 		}
 
 		// Rewrite "environments" to "targets".
-		if environments != dyn.NilValue && targets == dyn.NilValue {
+		if environments != dyn.InvalidValue && targets == dyn.InvalidValue {
 			nv, err := dyn.Set(v, "targets", environments)
 			if err != nil {
-				return dyn.NilValue, err
+				return dyn.InvalidValue, err
 			}
 			// Drop the "environments" key.
 			return dyn.Walk(nv, func(p dyn.Path, v dyn.Value) (dyn.Value, error) {

bundle/config/mutator/merge_job_clusters.go

@@ -21,7 +21,7 @@ func (m *mergeJobClusters) Name() string {
 
 func (m *mergeJobClusters) jobClusterKey(v dyn.Value) string {
 	switch v.Kind() {
-	case dyn.KindNil:
+	case dyn.KindInvalid, dyn.KindNil:
 		return ""
 	case dyn.KindString:
 		return v.MustString()

bundle/config/mutator/merge_job_tasks.go

@@ -21,7 +21,7 @@ func (m *mergeJobTasks) Name() string {
 
 func (m *mergeJobTasks) taskKeyString(v dyn.Value) string {
 	switch v.Kind() {
-	case dyn.KindNil:
+	case dyn.KindInvalid, dyn.KindNil:
 		return ""
 	case dyn.KindString:
 		return v.MustString()

bundle/config/mutator/merge_pipeline_clusters.go

@@ -22,7 +22,7 @@ func (m *mergePipelineClusters) Name() string {
 
 func (m *mergePipelineClusters) clusterLabel(v dyn.Value) string {
 	switch v.Kind() {
-	case dyn.KindNil:
+	case dyn.KindInvalid, dyn.KindNil:
 		// Note: the cluster label is optional and defaults to 'default'.
 		// We therefore ALSO merge all clusters without a label.
 		return "default"

bundle/config/mutator/mutator.go

@@ -4,6 +4,7 @@ import (
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/bundle/config/loader"
+	pythonmutator "github.com/databricks/cli/bundle/config/mutator/python"
 	"github.com/databricks/cli/bundle/scripts"
 )
 
@@ -24,5 +25,6 @@ func DefaultMutators() []bundle.Mutator {
 		InitializeVariables(),
 		DefineDefaultTarget(),
 		LoadGitDetails(),
+		pythonmutator.PythonMutator(pythonmutator.PythonMutatorPhaseLoad),
 	}
 }

bundle/config/mutator/process_target_mode.go

@@ -9,8 +9,8 @@ import (
 	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/libs/auth"
 	"github.com/databricks/cli/libs/diag"
-	"github.com/databricks/cli/libs/dyn"
 	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/databricks-sdk-go/service/catalog"
 	"github.com/databricks/databricks-sdk-go/service/jobs"
 	"github.com/databricks/databricks-sdk-go/service/ml"
 )
@@ -33,10 +33,8 @@ func (m *processTargetMode) Name() string {
 func transformDevelopmentMode(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
 	if !b.Config.Bundle.Deployment.Lock.IsExplicitlyEnabled() {
 		log.Infof(ctx, "Development mode: disabling deployment lock since bundle.deployment.lock.enabled is not set to true")
-		err := disableDeploymentLock(b)
-		if err != nil {
-			return diag.FromErr(err)
-		}
+		disabled := false
+		b.Config.Bundle.Deployment.Lock.Enabled = &disabled
 	}
 
 	r := b.Config.Resources
@@ -105,15 +103,16 @@ func transformDevelopmentMode(ctx context.Context, b *bundle.Bundle) diag.Diagno
 		// (registered models in Unity Catalog don't yet support tags)
 	}
 
-	return nil
-}
+	for i := range r.QualityMonitors {
+		// Remove all schedules from monitors, since they don't support pausing/unpausing.
+		// Quality monitors might support the "pause" property in the future, so at the
+		// CLI level we do respect that property if it is set to "unpaused".
+		if r.QualityMonitors[i].Schedule != nil && r.QualityMonitors[i].Schedule.PauseStatus != catalog.MonitorCronSchedulePauseStatusUnpaused {
+			r.QualityMonitors[i].Schedule = nil
+		}
+	}
 
-func disableDeploymentLock(b *bundle.Bundle) error {
-	return b.Config.Mutate(func(v dyn.Value) (dyn.Value, error) {
-		return dyn.Map(v, "bundle.deployment.lock", func(_ dyn.Path, v dyn.Value) (dyn.Value, error) {
-			return dyn.Set(v, "enabled", dyn.V(false))
-		})
-	})
+	return nil
 }
 
 func validateDevelopmentMode(b *bundle.Bundle) diag.Diagnostics {

bundle/config/mutator/process_target_mode_test.go

@@ -99,6 +99,20 @@ func mockBundle(mode config.Mode) *bundle.Bundle {
 				},
 				QualityMonitors: map[string]*resources.QualityMonitor{
 					"qualityMonitor1": {CreateMonitor: &catalog.CreateMonitor{TableName: "qualityMonitor1"}},
+					"qualityMonitor2": {
+						CreateMonitor: &catalog.CreateMonitor{
+							TableName: "qualityMonitor2",
+							Schedule:  &catalog.MonitorCronSchedule{},
+						},
+					},
+					"qualityMonitor3": {
+						CreateMonitor: &catalog.CreateMonitor{
+							TableName: "qualityMonitor3",
+							Schedule: &catalog.MonitorCronSchedule{
+								PauseStatus: catalog.MonitorCronSchedulePauseStatusUnpaused,
+							},
+						},
+					},
 				},
 			},
 		},
@@ -151,6 +165,8 @@ func TestProcessTargetModeDevelopment(t *testing.T) {
 
 	// Quality Monitor 1
 	assert.Equal(t, "qualityMonitor1", b.Config.Resources.QualityMonitors["qualityMonitor1"].TableName)
+	assert.Nil(t, b.Config.Resources.QualityMonitors["qualityMonitor2"].Schedule)
+	assert.Equal(t, catalog.MonitorCronSchedulePauseStatusUnpaused, b.Config.Resources.QualityMonitors["qualityMonitor3"].Schedule.PauseStatus)
 }
 
 func TestProcessTargetModeDevelopmentTagNormalizationForAws(t *testing.T) {
@@ -314,7 +330,7 @@ func TestDisableLocking(t *testing.T) {
 	ctx := context.Background()
 	b := mockBundle(config.Development)
 
-	err := transformDevelopmentMode(ctx, b)
+	err := bundle.Apply(ctx, b, ProcessTargetMode())
 	require.Nil(t, err)
 	assert.False(t, b.Config.Bundle.Deployment.Lock.IsEnabled())
 }
@@ -325,7 +341,7 @@ func TestDisableLockingDisabled(t *testing.T) {
 	explicitlyEnabled := true
 	b.Config.Bundle.Deployment.Lock.Enabled = &explicitlyEnabled
 
-	err := transformDevelopmentMode(ctx, b)
+	err := bundle.Apply(ctx, b, ProcessTargetMode())
 	require.Nil(t, err)
 	assert.True(t, b.Config.Bundle.Deployment.Lock.IsEnabled(), "Deployment lock should remain enabled in development mode when explicitly enabled")
 }

bundle/config/mutator/python/log_writer.go

@@ -0,0 +1,42 @@
+package python
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"io"
+
+	"github.com/databricks/cli/libs/log"
+)
+
+type logWriter struct {
+	ctx    context.Context
+	prefix string
+	buf    bytes.Buffer
+}
+
+// newLogWriter creates a new io.Writer that writes to log with specified prefix.
+func newLogWriter(ctx context.Context, prefix string) io.Writer {
+	return &logWriter{
+		ctx:    ctx,
+		prefix: prefix,
+	}
+}
+
+func (p *logWriter) Write(bytes []byte) (n int, err error) {
+	p.buf.Write(bytes)
+
+	scanner := bufio.NewScanner(&p.buf)
+
+	for scanner.Scan() {
+		line := scanner.Text()
+
+		log.Debugf(p.ctx, "%s%s", p.prefix, line)
+	}
+
+	remaining := p.buf.Bytes()
+	p.buf.Reset()
+	p.buf.Write(remaining)
+
+	return len(bytes), nil
+}