pipelines run: enrich output of pipelines

acceptance/pipelines/e2e/output.txt

@@ -23,6 +23,9 @@ View your pipeline my_project_pipeline here: [DATABRICKS_URL]/pipelines/[UUID]?o
 Update URL: [DATABRICKS_URL]/#joblist/pipelines/[UUID]/updates/[UUID]
 
 Update ID: [UUID]
+Update for pipeline completed successfully.
+Pipeline configurations for this update:
+• All tables are refreshed
 
 === Edit project by creating and running a new second pipeline
 >>> [PIPELINES] deploy
@@ -60,6 +63,9 @@ View your pipeline my_project_pipeline_2 here: [DATABRICKS_URL]/pipelines/[UUID]
 Update URL: [DATABRICKS_URL]/#joblist/pipelines/[UUID]/updates/[UUID]
 
 Update ID: [UUID]
+Update for pipeline completed successfully.
+Pipeline configurations for this update:
+• All tables are refreshed
 
 === Stop both pipelines before destroy
 >>> [PIPELINES] stop my_project_pipeline

acceptance/pipelines/run/no-wait/output.txt

@@ -11,3 +11,6 @@ View your pipeline my_pipeline here: [DATABRICKS_URL]/pipelines/[UUID]?o=[NUMID]
 Update URL: [DATABRICKS_URL]/#joblist/pipelines/[UUID]/updates/[UUID]
 
 Update ID: [UUID]
+Update for pipeline completed successfully.
+Pipeline configurations for this update:
+• All tables are refreshed

acceptance/pipelines/run/refresh-flags/output.txt

@@ -11,6 +11,9 @@ View your pipeline my_pipeline here: [DATABRICKS_URL]/pipelines/[UUID]?o=[NUMID]
 Update URL: [DATABRICKS_URL]/#joblist/pipelines/[UUID]/updates/[UUID]
 
 Update ID: [UUID]
+Update for pipeline completed successfully.
+Pipeline configurations for this update:
+• All tables are refreshed
 
 >>> print_requests
 {
@@ -29,6 +32,9 @@ Update ID: [UUID]
 Update URL: [DATABRICKS_URL]/#joblist/pipelines/[UUID]/updates/[UUID]
 
 Update ID: [UUID]
+Update for pipeline completed successfully.
+Pipeline configurations for this update:
+• All tables are refreshed
 
 >>> print_requests
 {
@@ -44,6 +50,9 @@ Update ID: [UUID]
 Update URL: [DATABRICKS_URL]/#joblist/pipelines/[UUID]/updates/[UUID]
 
 Update ID: [UUID]
+Update for pipeline completed successfully.
+Pipeline configurations for this update:
+• All tables are refreshed
 
 >>> print_requests
 {
@@ -62,6 +71,9 @@ Update ID: [UUID]
 Update URL: [DATABRICKS_URL]/#joblist/pipelines/[UUID]/updates/[UUID]
 
 Update ID: [UUID]
+Update for pipeline completed successfully.
+Pipeline configurations for this update:
+• All tables are refreshed
 
 >>> print_requests
 {

acceptance/pipelines/run/restart/output.txt

@@ -11,6 +11,9 @@ View your pipeline my_pipeline here: [DATABRICKS_URL]/pipelines/[UUID]?o=[NUMID]
 Update URL: [DATABRICKS_URL]/#joblist/pipelines/[UUID]/updates/[UUID]
 
 Update ID: [UUID]
+Update for pipeline completed successfully.
+Pipeline configurations for this update:
+• All tables are refreshed
 
 >>> print_requests
 {

acceptance/pipelines/run/run-pipeline/output.txt

@@ -11,9 +11,15 @@ View your pipeline my_pipeline here: [DATABRICKS_URL]/pipelines/[UUID]?o=[NUMID]
 Update URL: [DATABRICKS_URL]/#joblist/pipelines/[UUID]/updates/[UUID]
 
 Update ID: [UUID]
+Update for pipeline completed successfully.
+Pipeline configurations for this update:
+• All tables are refreshed
 
 === Run pipeline with KEY, expect same output as without KEY
 >>> [PIPELINES] run my_pipeline
 Update URL: [DATABRICKS_URL]/#joblist/pipelines/[UUID]/updates/[UUID]
 
 Update ID: [UUID]
+Update for pipeline completed successfully.
+Pipeline configurations for this update:
+• All tables are refreshed

acceptance/pipelines/stop/output.txt

@@ -10,6 +10,9 @@ View your pipeline my_pipeline here: [DATABRICKS_URL]/pipelines/[UUID]?o=[NUMID]
 Update URL: [DATABRICKS_URL]/#joblist/pipelines/[UUID]/updates/[UUID]
 
 Update ID: [UUID]
+Update for pipeline completed successfully.
+Pipeline configurations for this update:
+• All tables are refreshed
 
 === Stopping pipeline
 >>> [PIPELINES] stop

cmd/pipelines/run.go

@@ -3,25 +3,104 @@
 package pipelines
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
+	"time"
 
 	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config/resources"
 	"github.com/databricks/cli/bundle/deploy/terraform"
 	"github.com/databricks/cli/bundle/phases"
-	"github.com/databricks/cli/bundle/resources"
+	bundleresources "github.com/databricks/cli/bundle/resources"
 	"github.com/databricks/cli/bundle/run"
-	"github.com/databricks/cli/bundle/run/output"
+	bundlerunoutput "github.com/databricks/cli/bundle/run/output"
 	"github.com/databricks/cli/bundle/statemgmt"
 	"github.com/databricks/cli/cmd/bundle/utils"
 	"github.com/databricks/cli/cmd/root"
 	"github.com/databricks/cli/libs/cmdgroup"
+	"github.com/databricks/cli/libs/cmdio"
 	"github.com/databricks/cli/libs/flags"
 	"github.com/databricks/cli/libs/logdiag"
+	"github.com/databricks/databricks-sdk-go/service/pipelines"
 	"github.com/spf13/cobra"
 	"golang.org/x/exp/maps"
 )
 
+type PipelineUpdateData struct {
+	PipelineId    string
+	Update        pipelines.UpdateInfo
+	LastEventTime string
+}
+
+func fetchAndDisplayPipelineUpdate(ctx context.Context, bundle *bundle.Bundle, ref bundleresources.Reference, updateId string) error {
+	w := bundle.WorkspaceClient()
+
+	pipelineResource := ref.Resource.(*resources.Pipeline)
+	pipelineID := pipelineResource.ID
+	if pipelineID == "" {
+		return errors.New("unable to get pipeline ID from pipeline")
+	}
+
+	getUpdateResponse, err := w.Pipelines.GetUpdate(ctx, pipelines.GetUpdateRequest{
+		PipelineId: pipelineID,
+		UpdateId:   updateId,
+	})
+	if err != nil {
+		return err
+	}
+
+	if getUpdateResponse.Update == nil {
+		return err
+	}
+
+	latestUpdate := *getUpdateResponse.Update
+
+	params := &PipelineEventsQueryParams{
+		Filter:  fmt.Sprintf("update_id='%s' AND event_type='update_progress'", updateId),
+		OrderBy: "timestamp asc",
+	}
+
+	events, err := fetchAllPipelineEvents(ctx, w, pipelineID, params)
+	if err != nil {
+		return err
+	}
+
+	if latestUpdate.State == pipelines.UpdateInfoStateCompleted {
+		err = displayPipelineUpdate(ctx, latestUpdate, pipelineID, events)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// getLastEventTime returns the timestamp of the last progress event.
+// Expects that the events are already sorted by timestamp in ascending order.
+func getLastEventTime(events []pipelines.PipelineEvent) string {
+	if len(events) == 0 {
+		return ""
+	}
+	lastEvent := events[len(events)-1]
+	parsedTime, err := time.Parse(time.RFC3339Nano, lastEvent.Timestamp)
+	if err != nil {
+		return ""
+	}
+	return parsedTime.Format("2006-01-02T15:04:05Z")
+}
+
+func displayPipelineUpdate(ctx context.Context, update pipelines.UpdateInfo, pipelineID string, events []pipelines.PipelineEvent) error {
+	data := PipelineUpdateData{
+		PipelineId:    pipelineID,
+		Update:        update,
+		LastEventTime: getLastEventTime(events),
+	}
+
+	return cmdio.RenderWithTemplate(ctx, data, "", pipelineUpdateTemplate)
+}
+
 func runCommand() *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "run [flags] [KEY]",
@@ -100,20 +179,20 @@ Refreshes all tables in the pipeline unless otherwise specified.`,
 			NoWait: noWait,
 		}
 
-		var output output.RunOutput
+		var runOutput bundlerunoutput.RunOutput
 		if restart {
-			output, err = runner.Restart(ctx, &runOptions)
+			runOutput, err = runner.Restart(ctx, &runOptions)
 		} else {
-			output, err = runner.Run(ctx, &runOptions)
+			runOutput, err = runner.Run(ctx, &runOptions)
 		}
 		if err != nil {
 			return err
 		}
 
-		if output != nil {
+		if runOutput != nil {
 			switch root.OutputType(cmd) {
 			case flags.OutputText:
-				resultString, err := output.String()
+				resultString, err := runOutput.String()
 				if err != nil {
 					return err
 				}
@@ -122,7 +201,7 @@ Refreshes all tables in the pipeline unless otherwise specified.`,
 					return err
 				}
 			case flags.OutputJSON:
-				b, err := json.MarshalIndent(output, "", "  ")
+				b, err := json.MarshalIndent(runOutput, "", "  ")
 				if err != nil {
 					return err
 				}
@@ -135,6 +214,18 @@ Refreshes all tables in the pipeline unless otherwise specified.`,
 				return fmt.Errorf("unknown output type %s", root.OutputType(cmd))
 			}
 		}
+		ref, err := bundleresources.Lookup(b, key, run.IsRunnable)
+		if err != nil {
+			return err
+		}
+		if ref.Description.SingularName == "pipeline" && runOutput != nil {
+			if pipelineOutput, ok := runOutput.(*bundlerunoutput.PipelineOutput); ok && pipelineOutput.UpdateId != "" {
+				err = fetchAndDisplayPipelineUpdate(ctx, b, ref, pipelineOutput.UpdateId)
+				if err != nil {
+					return err
+				}
+			}
+		}
 		return nil
 	}
 
@@ -151,7 +242,7 @@ Refreshes all tables in the pipeline unless otherwise specified.`,
 		}
 
 		if len(args) == 0 {
-			completions := resources.Completions(b, run.IsRunnable)
+			completions := bundleresources.Completions(b, run.IsRunnable)
 			return maps.Keys(completions), cobra.ShellCompDirectiveNoFileComp
 		} else {
 			// If we know the resource to run, we can complete additional positional arguments.