diff --git a/.nextchanges/cli/pipelines-history-id.md b/.nextchanges/cli/pipelines-history-id.md new file mode 100644 index 00000000000..3859e1a9cdb --- /dev/null +++ b/.nextchanges/cli/pipelines-history-id.md @@ -0,0 +1 @@ +* Support pipeline IDs in `pipelines history`. ([#6995](https://github.com/databricks/cli/pull/6995)) diff --git a/.nextchanges/cli/pipelines-public-api.md b/.nextchanges/cli/pipelines-public-api.md new file mode 100644 index 00000000000..ed5f00a098e --- /dev/null +++ b/.nextchanges/cli/pipelines-public-api.md @@ -0,0 +1 @@ +* Show a hint for previewing more rows in `pipelines show`. ([#6995](https://github.com/databricks/cli/pull/6995)) diff --git a/acceptance/pipelines/history/basic/databricks.yml b/acceptance/pipelines/history/basic/databricks.yml new file mode 100644 index 00000000000..88084d31062 --- /dev/null +++ b/acceptance/pipelines/history/basic/databricks.yml @@ -0,0 +1,2 @@ +# invalid databricks.yml to show that `pipelines history PIPELINE_ID` bypasses the bundle +bundle: invalid diff --git a/acceptance/pipelines/history/basic/out.test.toml b/acceptance/pipelines/history/basic/out.test.toml new file mode 100644 index 00000000000..0938e678987 --- /dev/null +++ b/acceptance/pipelines/history/basic/out.test.toml @@ -0,0 +1,2 @@ +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/pipelines/history/basic/output.txt b/acceptance/pipelines/history/basic/output.txt new file mode 100644 index 00000000000..cc9db675fb4 --- /dev/null +++ b/acceptance/pipelines/history/basic/output.txt @@ -0,0 +1,59 @@ + +=== default history by pipeline id + +>>> [CLI] pipelines history [PIPELINE_ID] +Updates Summary for pipeline [PIPELINE_ID]: +Update ID: update-3 + State: COMPLETED + Creation Time: [TIMESTAMP] + Full Refresh: false + Validate Only: false +Update ID: update-2 + State: FAILED + Creation Time: [TIMESTAMP] + Full Refresh: false + Validate Only: true +Update ID: update-1 + State: CANCELED + Creation Time: [TIMESTAMP] + Full Refresh: false + Validate Only: false + +=== json output + +>>> [CLI] pipelines history [PIPELINE_ID] --output json +{ + "Key": "[PIPELINE_ID]", + "Updates": [ + { + "creation_time": 1640995200000, + "state": "COMPLETED", + "update_id": "update-3", + "validate_only": false + }, + { + "creation_time": 1640908800000, + "state": "FAILED", + "update_id": "update-2", + "validate_only": true + }, + { + "creation_time": 1640822400000, + "state": "CANCELED", + "update_id": "update-1", + "validate_only": false + } + ] +} + +=== wire request uses pipeline id + +>>> print_requests.py //api/2.0/pipelines/[PIPELINE_ID]/updates --get +{ + "method": "GET", + "path": "/api/2.0/pipelines/[PIPELINE_ID]/updates" +} +{ + "method": "GET", + "path": "/api/2.0/pipelines/[PIPELINE_ID]/updates" +} diff --git a/acceptance/pipelines/history/basic/script b/acceptance/pipelines/history/basic/script new file mode 100644 index 00000000000..27a44231923 --- /dev/null +++ b/acceptance/pipelines/history/basic/script @@ -0,0 +1,10 @@ +add_repl 3fb8e5a1-0d2c-4a6b-9f1e-2c7d8e9f0a1b PIPELINE_ID + +title "default history by pipeline id\n" +trace $CLI pipelines history 3fb8e5a1-0d2c-4a6b-9f1e-2c7d8e9f0a1b + +title "json output\n" +trace $CLI pipelines history 3fb8e5a1-0d2c-4a6b-9f1e-2c7d8e9f0a1b --output json + +title "wire request uses pipeline id\n" +trace print_requests.py //api/2.0/pipelines/3fb8e5a1-0d2c-4a6b-9f1e-2c7d8e9f0a1b/updates --get diff --git a/acceptance/pipelines/history/basic/test.toml b/acceptance/pipelines/history/basic/test.toml new file mode 100644 index 00000000000..3f23481ca91 --- /dev/null +++ b/acceptance/pipelines/history/basic/test.toml @@ -0,0 +1,16 @@ +Cloud = false + +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] +RecordRequests = true + +[[Server]] +Pattern = "GET /api/2.0/pipelines/{pipeline_id}/updates" +Response.Body = ''' +{ + "updates": [ + {"update_id": "update-3", "state": "COMPLETED", "creation_time": 1640995200000, "validate_only": false}, + {"update_id": "update-2", "state": "FAILED", "creation_time": 1640908800000, "validate_only": true}, + {"update_id": "update-1", "state": "CANCELED", "creation_time": 1640822400000, "validate_only": false} + ] +} +''' diff --git a/acceptance/pipelines/show/basic/output.txt b/acceptance/pipelines/show/basic/output.txt index 87fe70aeeee..5acd64d832a 100644 --- a/acceptance/pipelines/show/basic/output.txt +++ b/acceptance/pipelines/show/basic/output.txt @@ -23,12 +23,22 @@ id name } ] +=== explicit limit does not show a hint + +>>> [CLI] pipelines show main.default.people --limit 9 +id name +-- ----- +1 alice +2 bob + +2 rows + === invalid table name >>> [CLI] pipelines show people Error: invalid table name "people": expected catalog.schema.table (Unity Catalog) or schema.table (legacy Hive metastore) -=== wire SQL pins quoting and default limit +=== wire SQL pins quoting and limits >>> print_requests.py //api/2.0/sql/statements { @@ -38,7 +48,7 @@ Error: invalid table name "people": expected catalog.schema.table (Unity Catalog "disposition": "INLINE", "format": "JSON_ARRAY", "on_wait_timeout": "CONTINUE", - "statement": "SELECT * FROM `main`.`default`.`people` LIMIT 10", + "statement": "SELECT * FROM `main`.`default`.`people` LIMIT 11", "wait_timeout": "10s", "warehouse_id": "wh-test" } @@ -55,3 +65,15 @@ Error: invalid table name "people": expected catalog.schema.table (Unity Catalog "warehouse_id": "wh-test" } } +{ + "method": "POST", + "path": "/api/2.0/sql/statements", + "body": { + "disposition": "INLINE", + "format": "JSON_ARRAY", + "on_wait_timeout": "CONTINUE", + "statement": "SELECT * FROM `main`.`default`.`people` LIMIT 9", + "wait_timeout": "10s", + "warehouse_id": "wh-test" + } +} diff --git a/acceptance/pipelines/show/basic/script b/acceptance/pipelines/show/basic/script index 04f40545d6c..b63cfc73e23 100644 --- a/acceptance/pipelines/show/basic/script +++ b/acceptance/pipelines/show/basic/script @@ -4,8 +4,11 @@ trace $CLI pipelines show main.default.people title "json output\n" trace $CLI pipelines show main.default.people --output json +title "explicit limit does not show a hint\n" +trace $CLI pipelines show main.default.people --limit 9 + title "invalid table name\n" musterr trace $CLI pipelines show people -title "wire SQL pins quoting and default limit\n" +title "wire SQL pins quoting and limits\n" trace print_requests.py //api/2.0/sql/statements diff --git a/acceptance/pipelines/show/exact-limit/out.test.toml b/acceptance/pipelines/show/exact-limit/out.test.toml new file mode 100644 index 00000000000..18b1a88417e --- /dev/null +++ b/acceptance/pipelines/show/exact-limit/out.test.toml @@ -0,0 +1 @@ +Cloud = false diff --git a/acceptance/pipelines/show/exact-limit/output.txt b/acceptance/pipelines/show/exact-limit/output.txt new file mode 100644 index 00000000000..dbdb65f350b --- /dev/null +++ b/acceptance/pipelines/show/exact-limit/output.txt @@ -0,0 +1,34 @@ + +=== default limit does not show a hint when exactly 10 rows are available + +>>> [CLI] pipelines show main.default.people +id +-- +1 +2 +3 +4 +5 +6 +7 +8 +9 +10 + +10 rows + +=== wire SQL fetches one extra row + +>>> print_requests.py //api/2.0/sql/statements +{ + "method": "POST", + "path": "/api/2.0/sql/statements", + "body": { + "disposition": "INLINE", + "format": "JSON_ARRAY", + "on_wait_timeout": "CONTINUE", + "statement": "SELECT * FROM `main`.`default`.`people` LIMIT 11", + "wait_timeout": "10s", + "warehouse_id": "wh-test" + } +} diff --git a/acceptance/pipelines/show/exact-limit/script b/acceptance/pipelines/show/exact-limit/script new file mode 100644 index 00000000000..1b74a228537 --- /dev/null +++ b/acceptance/pipelines/show/exact-limit/script @@ -0,0 +1,5 @@ +title "default limit does not show a hint when exactly 10 rows are available\n" +trace $CLI pipelines show main.default.people + +title "wire SQL fetches one extra row\n" +trace print_requests.py //api/2.0/sql/statements diff --git a/acceptance/pipelines/show/exact-limit/test.toml b/acceptance/pipelines/show/exact-limit/test.toml new file mode 100644 index 00000000000..eec09b87ca7 --- /dev/null +++ b/acceptance/pipelines/show/exact-limit/test.toml @@ -0,0 +1,14 @@ +RecordRequests = true + +Env.DATABRICKS_WAREHOUSE_ID = "wh-test" + +[[Server]] +Pattern = "POST /api/2.0/sql/statements" +Response.Body = ''' +{ + "statement_id": "s1", + "status": {"state": "SUCCEEDED"}, + "manifest": {"schema": {"columns": [{"name": "id"}]}, "total_chunk_count": 1}, + "result": {"data_array": [["1"], ["2"], ["3"], ["4"], ["5"], ["6"], ["7"], ["8"], ["9"], ["10"]]} +} +''' diff --git a/acceptance/pipelines/show/limit-hint/out.test.toml b/acceptance/pipelines/show/limit-hint/out.test.toml new file mode 100644 index 00000000000..18b1a88417e --- /dev/null +++ b/acceptance/pipelines/show/limit-hint/out.test.toml @@ -0,0 +1 @@ +Cloud = false diff --git a/acceptance/pipelines/show/limit-hint/output.txt b/acceptance/pipelines/show/limit-hint/output.txt new file mode 100644 index 00000000000..5a359f7b992 --- /dev/null +++ b/acceptance/pipelines/show/limit-hint/output.txt @@ -0,0 +1,35 @@ + +=== default limit shows a hint when more rows are available + +>>> [CLI] pipelines show main.default.people +id +-- +1 +2 +3 +4 +5 +6 +7 +8 +9 +10 + +10 rows +Use --limit to preview more rows. + +=== wire SQL fetches one extra row + +>>> print_requests.py //api/2.0/sql/statements +{ + "method": "POST", + "path": "/api/2.0/sql/statements", + "body": { + "disposition": "INLINE", + "format": "JSON_ARRAY", + "on_wait_timeout": "CONTINUE", + "statement": "SELECT * FROM `main`.`default`.`people` LIMIT 11", + "wait_timeout": "10s", + "warehouse_id": "wh-test" + } +} diff --git a/acceptance/pipelines/show/limit-hint/script b/acceptance/pipelines/show/limit-hint/script new file mode 100644 index 00000000000..7f066a8fe9c --- /dev/null +++ b/acceptance/pipelines/show/limit-hint/script @@ -0,0 +1,5 @@ +title "default limit shows a hint when more rows are available\n" +trace $CLI pipelines show main.default.people + +title "wire SQL fetches one extra row\n" +trace print_requests.py //api/2.0/sql/statements diff --git a/acceptance/pipelines/show/limit-hint/test.toml b/acceptance/pipelines/show/limit-hint/test.toml new file mode 100644 index 00000000000..9887d749d23 --- /dev/null +++ b/acceptance/pipelines/show/limit-hint/test.toml @@ -0,0 +1,14 @@ +RecordRequests = true + +Env.DATABRICKS_WAREHOUSE_ID = "wh-test" + +[[Server]] +Pattern = "POST /api/2.0/sql/statements" +Response.Body = ''' +{ + "statement_id": "s1", + "status": {"state": "SUCCEEDED"}, + "manifest": {"schema": {"columns": [{"name": "id"}]}, "total_chunk_count": 1}, + "result": {"data_array": [["1"], ["2"], ["3"], ["4"], ["5"], ["6"], ["7"], ["8"], ["9"], ["10"], ["11"]]} +} +''' diff --git a/cmd/pipelines/history.go b/cmd/pipelines/history.go index 6c30d1e72bd..a5fc42a4f61 100644 --- a/cmd/pipelines/history.go +++ b/cmd/pipelines/history.go @@ -5,18 +5,21 @@ import ( "github.com/databricks/cli/cmd/bundle/utils" "github.com/databricks/cli/cmd/root" + "github.com/databricks/cli/libs/cmdctx" "github.com/databricks/cli/libs/cmdgroup" "github.com/databricks/cli/libs/cmdio" + databricks "github.com/databricks/databricks-sdk-go" "github.com/databricks/databricks-sdk-go/service/pipelines" + "github.com/google/uuid" "github.com/spf13/cobra" ) func historyCommand() *cobra.Command { cmd := &cobra.Command{ - Use: "history [flags] [KEY]", + Use: "history [flags] [KEY|PIPELINE_ID]", Args: root.MaximumNArgs(1), Short: "Retrieve past runs for a pipeline", - Long: `Retrieve past runs for a pipeline identified by KEY, the unique name of the pipeline as defined in its YAML file.`, + Long: `Retrieve past runs for a pipeline identified by KEY, the unique name of the pipeline as defined in its YAML file, or by PIPELINE_ID, the unique identifier of the pipeline.`, } var startTimeStr string @@ -34,25 +37,45 @@ func historyCommand() *cobra.Command { wrappedCmd.AddFlagGroup(historyGroup) cmd.RunE = func(cmd *cobra.Command, args []string) error { - b, err := utils.ProcessBundle(cmd, utils.ProcessOptions{ - InitIDs: true, - }) - if err != nil { - return err + var key, pipelineId string + var w *databricks.WorkspaceClient + if len(args) == 1 { + id, err := uuid.Parse(args[0]) + if err == nil { + pipelineId = id.String() + } } - ctx := cmd.Context() - key, err := resolvePipelineArgument(ctx, b, args) - if err != nil { - return err - } + if pipelineId != "" { + if cmd.Flag("target").Value.String() == "" { + cmd.SetContext(root.SkipLoadBundle(cmd.Context())) + } + if err := root.MustWorkspaceClient(cmd, args); err != nil { + return err + } + key = pipelineId + w = cmdctx.WorkspaceClient(cmd.Context()) + } else { + b, err := utils.ProcessBundle(cmd, utils.ProcessOptions{ + InitIDs: true, + }) + if err != nil { + return err + } + ctx := cmd.Context() - pipelineId, err := resolvePipelineIdFromKey(ctx, b, key) - if err != nil { - return err - } + key, err = resolvePipelineArgument(ctx, b, args) + if err != nil { + return err + } - w := b.WorkspaceClient(ctx) + pipelineId, err = resolvePipelineIdFromKey(ctx, b, key) + if err != nil { + return err + } + w = b.WorkspaceClient(ctx) + } + ctx := cmd.Context() startTimePtr, err := parseTimeToUnixMillis(startTimeStr) if err != nil { diff --git a/cmd/pipelines/history_test.go b/cmd/pipelines/history_test.go index c1baec58327..e2c4f948bdd 100644 --- a/cmd/pipelines/history_test.go +++ b/cmd/pipelines/history_test.go @@ -250,9 +250,9 @@ func TestFilterUpdates(t *testing.T) { expectedLast: 400, }, { - name: "no updates - empty slice", + name: "no updates with time filter", updates: noUpdates, - startTime: nil, + startTime: new(int64(500)), endTime: nil, expectedCount: 0, expectedFirst: 0, diff --git a/cmd/pipelines/show.go b/cmd/pipelines/show.go index 0e115e0508d..af8fd732d4c 100644 --- a/cmd/pipelines/show.go +++ b/cmd/pipelines/show.go @@ -53,8 +53,14 @@ schema.table (legacy Hive metastore).`, return err } + checkForMoreRows := root.OutputType(cmd) == flags.OutputText && !cmd.Flags().Changed("limit") + queryLimit := limit + if checkForMoreRows { + queryLimit++ + } + client := sqlexec.New(w.StatementExecution, warehouseID) - statement := fmt.Sprintf("SELECT * FROM %s LIMIT %d", quoted, limit) + statement := fmt.Sprintf("SELECT * FROM %s LIMIT %d", quoted, queryLimit) result, err := client.Execute(ctx, statement) if err != nil { if ctx.Err() != nil { @@ -64,7 +70,19 @@ schema.table (legacy Hive metastore).`, return err } - return render(ctx, cmd, result.Columns, result.Rows) + rows := result.Rows + hasMoreRows := checkForMoreRows && len(rows) > limit + if hasMoreRows { + rows = rows[:limit] + } + + if err := render(ctx, cmd, result.Columns, rows); err != nil { + return err + } + if hasMoreRows { + fmt.Fprintln(cmd.OutOrStdout(), "Use --limit to preview more rows.") + } + return nil }, } diff --git a/cmd/pipelines/utils.go b/cmd/pipelines/utils.go index be2f481db05..39cc74f3681 100644 --- a/cmd/pipelines/utils.go +++ b/cmd/pipelines/utils.go @@ -279,6 +279,10 @@ func updatesAfter(updates []pipelines.UpdateInfo, ts int64) []pipelines.UpdateIn // assuming updates are in descending order, largest CreationTime first. // Time is in milliseconds since epoch. If time is nil, it is ignored. func filterUpdates(updates []pipelines.UpdateInfo, startTime, endTime *int64) ([]pipelines.UpdateInfo, error) { + if len(updates) == 0 { + return updates, nil + } + if (endTime == nil || updates[0].CreationTime <= *endTime) && (startTime == nil || updates[len(updates)-1].CreationTime >= *startTime) { return updates, nil }