Add Pipelines Show Command

NOTICE

@@ -111,6 +111,10 @@ golang.org/x/sys - https://github.com/golang/sys
 Copyright 2009 The Go Authors.
 License - https://github.com/golang/sys/blob/master/LICENSE
 
+golang.org/x/term - https://github.com/golang/term
+Copyright 2009 The Go Authors.
+License - https://github.com/golang/term/blob/master/LICENSE
+
 golang.org/x/text - https://github.com/golang/text
 Copyright 2009 The Go Authors.
 License - https://github.com/golang/text/blob/master/LICENSE

acceptance/pipelines/show/basic/output.txt

@@ -0,0 +1,57 @@
+
+=== table output
+
+>>> [CLI] pipelines show main.default.people
+id  name
+--  -----
+1   alice
+2   bob
+
+2 rows
+
+=== json output
+
+>>> [CLI] pipelines show main.default.people --output json
+[
+  {
+    "id": "1",
+    "name": "alice"
+  },
+  {
+    "id": "2",
+    "name": "bob"
+  }
+]
+
+=== invalid table name
+
+>>> [CLI] pipelines show people
+Error: invalid table name "people": expected catalog.schema.table or schema.table
+
+=== wire SQL pins quoting and default limit
+
+>>> 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 100",
+    "wait_timeout": "10s",
+    "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 100",
+    "wait_timeout": "10s",
+    "warehouse_id": "wh-test"
+  }
+}

acceptance/pipelines/show/basic/script

@@ -0,0 +1,11 @@
+title "table output\n"
+trace $CLI pipelines show main.default.people
+
+title "json output\n"
+trace $CLI pipelines show main.default.people --output json
+
+title "invalid table name\n"
+musterr trace $CLI pipelines show people
+
+title "wire SQL pins quoting and default limit\n"
+trace print_requests.py //api/2.0/sql/statements

acceptance/pipelines/show/basic/test.toml

@@ -0,0 +1,21 @@
+RecordRequests = true
+
+[EnvMatrix]
+DATABRICKS_BUNDLE_ENGINE = []
+
+[Env]
+DATABRICKS_WAREHOUSE_ID = "wh-test"
+
+# The testserver's SQL Statement Execution API is matcher-driven and fails
+# unmatched statements by default; stub a canned result so the golden shows real
+# columns and rows. The SDK POSTs to the path without a trailing slash.
+[[Server]]
+Pattern = "POST /api/2.0/sql/statements"
+Response.Body = '''
+{
+  "statement_id": "s1",
+  "status": {"state": "SUCCEEDED"},
+  "manifest": {"schema": {"columns": [{"name": "id"}, {"name": "name"}]}, "total_chunk_count": 1},
+  "result": {"data_array": [["1", "alice"], ["2", "bob"]]}
+}
+'''

acceptance/pipelines/show/help/output.txt

@@ -0,0 +1,22 @@
+
+=== help
+>>> [CLI] pipelines show --help
+Preview a table's columns and sample rows on a SQL warehouse.
+
+TABLE is a fully-qualified name: catalog.schema.table (Unity Catalog) or
+schema.table (legacy Hive metastore).
+
+Usage:
+  databricks pipelines show TABLE [flags]
+
+Flags:
+  -h, --help                  help for show
+  -n, --limit int             Maximum number of rows to fetch. (default 100)
+      --warehouse-id string   SQL warehouse to run the query on. Defaults to DATABRICKS_WAREHOUSE_ID or a workspace default.
+
+Global Flags:
+      --debug            enable debug logging
+  -o, --output type      output type: text or json (default text)
+  -p, --profile string   ~/.databrickscfg profile
+  -t, --target string    bundle target to use (if applicable)
+      --var strings      set values for variables defined in project config. Example: --var="foo=bar"

acceptance/pipelines/show/help/script

@@ -0,0 +1,4 @@
+# Pin the command tree so any change to the command or its help shows up as a diff here.
+
+title "help"
+trace $CLI pipelines show --help

acceptance/pipelines/show/help/test.toml

@@ -0,0 +1,3 @@
+# --help prints without authenticating, so no server stubs are needed.
+[EnvMatrix]
+DATABRICKS_BUNDLE_ENGINE = []

cmd/pipelines/commands.go

@@ -20,5 +20,6 @@ func Commands() []*cobra.Command {
 		historyCommand(),
 		logsCommand(),
 		openCommand(),
+		showCommand(),
 	}
 }

cmd/pipelines/show.go

@@ -0,0 +1,101 @@
+package pipelines
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/databricks/cli/cmd/root"
+	"github.com/databricks/cli/libs/cmdctx"
+	"github.com/databricks/cli/libs/databrickscfg/cfgpickers"
+	"github.com/databricks/cli/libs/flags"
+	"github.com/databricks/cli/libs/sqlexec"
+	"github.com/databricks/cli/libs/tableprint"
+	"github.com/databricks/databricks-sdk-go"
+	"github.com/spf13/cobra"
+)
+
+const defaultLimit = 100
+
+func showCommand() *cobra.Command {
+	var warehouseID string
+	var limit int
+
+	cmd := &cobra.Command{
+		Use:   "show TABLE",
+		Short: "Preview a table's columns and sample rows",
+		Long: `Preview a table's columns and sample rows on a SQL warehouse.
+
+TABLE is a fully-qualified name: catalog.schema.table (Unity Catalog) or
+schema.table (legacy Hive metastore).`,
+		Args:    root.ExactArgs(1),
+		PreRunE: root.MustWorkspaceClient,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			ctx := cmd.Context()
+			w := cmdctx.WorkspaceClient(ctx)
+
+			if limit <= 0 {
+				return errors.New("--limit must be a positive integer")
+			}
+
+			quoted, err := quoteTableName(args[0])
+			if err != nil {
+				return err
+			}
+
+			warehouseID, err = resolveWarehouseID(ctx, w, warehouseID)
+			if err != nil {
+				return err
+			}
+
+			client := sqlexec.New(w.StatementExecution, warehouseID)
+			statement := fmt.Sprintf("SELECT * FROM %s LIMIT %d", quoted, limit)
+			result, err := client.Execute(ctx, statement)
+			if err != nil {
+				return err
+			}
+
+			return render(ctx, cmd, result.Columns, result.Rows)
+		},
+	}
+
+	cmd.Flags().StringVar(&warehouseID, "warehouse-id", "", "SQL warehouse to run the query on. Defaults to DATABRICKS_WAREHOUSE_ID or a workspace default.")
+	cmd.Flags().IntVarP(&limit, "limit", "n", defaultLimit, "Maximum number of rows to fetch.")
+
+	return cmd
+}
+
+// warehouse to query either --warehouse-id flag, DATABRICKS_WAREHOUSE_ID config,
+// or the workspace default warehouse.
+func resolveWarehouseID(ctx context.Context, w *databricks.WorkspaceClient, flag string) (string, error) {
+	if flag != "" {
+		return flag, nil
+	}
+	if w.Config.WarehouseID != "" {
+		return w.Config.WarehouseID, nil
+	}
+	warehouse, err := cfgpickers.GetDefaultWarehouse(ctx, w)
+	if errors.Is(err, cfgpickers.ErrNoCompatibleWarehouses) {
+		return "", errors.New("no SQL warehouse available; pass --warehouse-id or set DATABRICKS_WAREHOUSE_ID")
+	}
+	if err != nil {
+		return "", err
+	}
+	return warehouse.Id, nil
+}
+
+// writes the result as JSON (--output json) or a terminal-width-aware table.
+func render(ctx context.Context, cmd *cobra.Command, columns []string, rows [][]string) error {
+	out := cmd.OutOrStdout()
+	switch root.OutputType(cmd) {
+	case flags.OutputJSON:
+		return renderJSON(out, columns, rows)
+	case flags.OutputText:
+		if len(columns) == 0 {
+			return nil
+		}
+		return tableprint.Render(ctx, out, columns, rows, tableprint.DetectWidth(out))
+	default:
+		return fmt.Errorf("unknown output type %s", root.OutputType(cmd))
+	}
+}

cmd/pipelines/show_identifier.go

@@ -0,0 +1,25 @@
+package pipelines
+
+import (
+	"fmt"
+	"strings"
+)
+
+// validates a fully-qualified table name and returns it with each
+// segment backtick-quoted. Accepts two-part (schema.table, Hive)
+// and three-part (catalog.schema.table, Unity Catalog) names.
+func quoteTableName(name string) (string, error) {
+	parts := strings.Split(name, ".")
+	if len(parts) < 2 || len(parts) > 3 {
+		return "", fmt.Errorf("invalid table name %q: expected catalog.schema.table or schema.table", name)
+	}
+
+	quoted := make([]string, len(parts))
+	for i, part := range parts {
+		if part == "" {
+			return "", fmt.Errorf("invalid table name %q: empty identifier segment", name)
+		}
+		quoted[i] = "`" + strings.ReplaceAll(part, "`", "``") + "`"
+	}
+	return strings.Join(quoted, "."), nil
+}

cmd/pipelines/show_identifier_test.go

@@ -0,0 +1,54 @@
+package pipelines
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestQuoteTableName(t *testing.T) {
+	tests := []struct {
+		name string
+		in   string
+		want string
+	}{
+		{"three part", "main.sales.orders", "`main`.`sales`.`orders`"},
+		{"two part", "sales.orders", "`sales`.`orders`"},
+		{"hive metastore", "hive_metastore.sales.orders", "`hive_metastore`.`sales`.`orders`"},
+		{"leading digit segment", "cat.sch.2020_sales", "`cat`.`sch`.`2020_sales`"},
+		{"hyphen segment", "cat.sch.order-items", "`cat`.`sch`.`order-items`"},
+		{"space segment", "cat.sch.my table", "`cat`.`sch`.`my table`"},
+		{"reserved word segment", "cat.sch.order", "`cat`.`sch`.`order`"},
+		{"internal backtick doubled", "cat.sch.wei`rd", "`cat`.`sch`.`wei``rd`"},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := quoteTableName(tc.in)
+			require.NoError(t, err)
+			assert.Equal(t, tc.want, got)
+		})
+	}
+}
+
+func TestQuoteTableNameErrors(t *testing.T) {
+	tests := []struct {
+		name string
+		in   string
+	}{
+		{"one part", "orders"},
+		{"four part", "a.b.c.d"},
+		{"empty", ""},
+		{"empty middle segment", "cat..table"},
+		{"trailing dot", "cat.sch."},
+		{"leading dot", ".sch.table"},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := quoteTableName(tc.in)
+			assert.Error(t, err)
+		})
+	}
+}

cmd/pipelines/show_render.go

@@ -0,0 +1,56 @@
+package pipelines
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+)
+
+// marshals a result row as a JSON object with keys in column order.
+type orderedRow struct {
+	columns []string
+	values  []string
+}
+
+func (r orderedRow) MarshalJSON() ([]byte, error) {
+	var buf bytes.Buffer
+	buf.WriteByte('{')
+	for i, col := range r.columns {
+		if i > 0 {
+			buf.WriteByte(',')
+		}
+		key, err := json.Marshal(col)
+		if err != nil {
+			return nil, err
+		}
+		buf.Write(key)
+		buf.WriteByte(':')
+		var val []byte
+		if i < len(r.values) {
+			val, err = json.Marshal(r.values[i])
+		} else {
+			val = []byte("null")
+		}
+		if err != nil {
+			return nil, err
+		}
+		buf.Write(val)
+	}
+	buf.WriteByte('}')
+	return buf.Bytes(), nil
+}
+
+// writes rows as an indented JSON array of column-ordered objects.
+func renderJSON(w io.Writer, columns []string, rows [][]string) error {
+	objects := make([]orderedRow, len(rows))
+	for i, row := range rows {
+		objects[i] = orderedRow{columns: columns, values: row}
+	}
+	output, err := json.MarshalIndent(objects, "", "  ")
+	if err != nil {
+		return fmt.Errorf("marshal results: %w", err)
+	}
+	fmt.Fprintf(w, "%s\n", output)
+	return nil
+}