timing broken

2025-12-19 18:12:43 -05:00 · 2024-09-13 15:49:23 +01:00
parent 08b210797a
commit 5fd5ae0740
34 changed files with 84 additions and 504 deletions
--- a/cmd/login.go
+++ b/cmd/login.go
@@ -105,7 +105,7 @@ func displayLoginMessage(ctx context.Context, token string) {
 	error_helpers.FailOnError(sperr.WrapWithMessage(err, "failed to read user name"))

 	fmt.Println()
-	fmt.Printf("Logged in as: %s\n", constants.Bold(userName))
+	fmt.Printf("Logged in as: %s\n", pconstants.Bold(userName))
 	fmt.Println()
 }

--- a/cmd/plugin.go
+++ b/cmd/plugin.go
@@ -252,7 +252,7 @@ func runPluginInstallCmd(cmd *cobra.Command, args []string) {
 	// - ghcr.io/turbot/steampipe/plugins/turbot/aws:1.0.0
 	plugins := append([]string{}, args...)
 	showProgress := viper.GetBool(pconstants.ArgProgress)
-	installReports := make(display.PluginInstallReports, 0, len(plugins))
+	installReports := make(pplugin.PluginInstallReports, 0, len(plugins))

 	if len(plugins) == 0 {
 		if len(steampipeconfig.GlobalConfig.Plugins) == 0 {
@@ -279,7 +279,7 @@ func runPluginInstallCmd(cmd *cobra.Command, args []string) {
 	fmt.Println()
 	progressBars := uiprogress.New()
 	installWaitGroup := &sync.WaitGroup{}
-	reportChannel := make(chan *display.PluginInstallReport, len(plugins))
+	reportChannel := make(chan *pplugin.PluginInstallReport, len(plugins))

 	if showProgress {
 		progressBars.Start()
@@ -295,7 +295,7 @@ func runPluginInstallCmd(cmd *cobra.Command, args []string) {
 		if ref.IsFromTurbotHub() {
 			rpv, err := pplugin.GetLatestPluginVersionByConstraint(ctx, state.InstallationID, org, name, constraint)
 			if err != nil || rpv == nil {
-				report := &display.PluginInstallReport{
+				report := &pplugin.PluginInstallReport{
 					Plugin:         pluginName,
 					Skipped:        true,
 					SkipReason:     pconstants.InstallMessagePluginNotFound,
@@ -343,14 +343,14 @@ func runPluginInstallCmd(cmd *cobra.Command, args []string) {

 		statushooks.Done(ctx)
 	}
-	display.PrintInstallReports(installReports, false)
+	pplugin.PrintInstallReports(installReports, false)

 	// a concluding blank line - since we always output multiple lines
 	fmt.Println()
 }

-func doPluginInstall(ctx context.Context, bar *uiprogress.Bar, pluginName string, resolvedPlugin pplugin.ResolvedPluginVersion, wg *sync.WaitGroup, returnChannel chan *display.PluginInstallReport) {
-	var report *display.PluginInstallReport
+func doPluginInstall(ctx context.Context, bar *uiprogress.Bar, pluginName string, resolvedPlugin pplugin.ResolvedPluginVersion, wg *sync.WaitGroup, returnChannel chan *pplugin.PluginInstallReport) {
+	var report *pplugin.PluginInstallReport

 	pluginAlreadyInstalled, _ := pplugin.Exists(ctx, pluginName)
 	if pluginAlreadyInstalled {
@@ -361,7 +361,7 @@ func doPluginInstall(ctx context.Context, bar *uiprogress.Bar, pluginName string
 		bar.AppendFunc(func(b *uiprogress.Bar) string {
 			return helpers.Resize(pconstants.InstallMessagePluginAlreadyInstalled, 20)
 		})
-		report = &display.PluginInstallReport{
+		report = &pplugin.PluginInstallReport{
 			Plugin:         pluginName,
 			Skipped:        true,
 			SkipReason:     pconstants.InstallMessagePluginAlreadyInstalled,
@@ -422,7 +422,7 @@ func runPluginUpdateCmd(cmd *cobra.Command, args []string) {
 		// improve the response to wrong argument "steampipe plugin update all"
 		fmt.Println()
 		exitCode = constants.ExitCodeInsufficientOrWrongInputs
-		error_helpers.ShowError(ctx, fmt.Errorf("Did you mean %s?", constants.Bold("--all")))
+		error_helpers.ShowError(ctx, fmt.Errorf("Did you mean %s?", pconstants.Bold("--all")))
 		fmt.Println()
 		return
 	}
@@ -438,7 +438,7 @@ func runPluginUpdateCmd(cmd *cobra.Command, args []string) {
 	pluginVersions := steampipeconfig.GlobalConfig.PluginVersions

 	var runUpdatesFor []*versionfile.InstalledVersion
-	updateResults := make(display.PluginInstallReports, 0, len(plugins))
+	updateResults := make(pplugin.PluginInstallReports, 0, len(plugins))

 	// a leading blank line - since we always output multiple lines
 	fmt.Println()
@@ -467,7 +467,7 @@ func runPluginUpdateCmd(cmd *cobra.Command, args []string) {
 				}
 			} else {
 				exitCode = constants.ExitCodePluginNotFound
-				updateResults = append(updateResults, &display.PluginInstallReport{
+				updateResults = append(updateResults, &pplugin.PluginInstallReport{
 					Skipped:        true,
 					Plugin:         p,
 					SkipReason:     pconstants.InstallMessagePluginNotInstalled,
@@ -481,7 +481,7 @@ func runPluginUpdateCmd(cmd *cobra.Command, args []string) {
 		// we have report for all
 		// this may happen if all given plugins are
 		// not installed
-		display.PrintInstallReports(updateResults, true)
+		pplugin.PrintInstallReports(updateResults, true)
 		fmt.Println()
 		return
 	}
@@ -501,7 +501,7 @@ func runPluginUpdateCmd(cmd *cobra.Command, args []string) {
 	}

 	updateWaitGroup := &sync.WaitGroup{}
-	reportChannel := make(chan *display.PluginInstallReport, len(reports))
+	reportChannel := make(chan *pplugin.PluginInstallReport, len(reports))
 	progressBars := uiprogress.New()
 	if showProgress {
 		progressBars.Start()
@@ -530,14 +530,14 @@ func runPluginUpdateCmd(cmd *cobra.Command, args []string) {
 		progressBars.Stop()
 	}

-	display.PrintInstallReports(updateResults, true)
+	pplugin.PrintInstallReports(updateResults, true)

 	// a concluding blank line - since we always output multiple lines
 	fmt.Println()
 }

-func doPluginUpdate(ctx context.Context, bar *uiprogress.Bar, pvr pplugin.PluginVersionCheckReport, wg *sync.WaitGroup, returnChannel chan *display.PluginInstallReport) {
-	var report *display.PluginInstallReport
+func doPluginUpdate(ctx context.Context, bar *uiprogress.Bar, pvr pplugin.PluginVersionCheckReport, wg *sync.WaitGroup, returnChannel chan *pplugin.PluginInstallReport) {
+	var report *pplugin.PluginInstallReport

 	if pplugin.UpdateRequired(pvr) {
 		// update required, resolve version and install update
@@ -559,7 +559,7 @@ func doPluginUpdate(ctx context.Context, bar *uiprogress.Bar, pvr pplugin.Plugin
 		})
 		// set the progress bar to the maximum
 		bar.Set(len(pluginInstallSteps))
-		report = &display.PluginInstallReport{
+		report = &pplugin.PluginInstallReport{
 			Plugin:         fmt.Sprintf("%s@%s", pvr.CheckResponse.Name, pvr.CheckResponse.Constraint),
 			Skipped:        true,
 			SkipReason:     pconstants.InstallMessagePluginLatestAlreadyInstalled,
@@ -579,7 +579,7 @@ func createProgressBar(plugin string, parentProgressBars *uiprogress.Progress) *
 	return bar
 }

-func installPlugin(ctx context.Context, resolvedPlugin pplugin.ResolvedPluginVersion, isUpdate bool, bar *uiprogress.Bar) *display.PluginInstallReport {
+func installPlugin(ctx context.Context, resolvedPlugin pplugin.ResolvedPluginVersion, isUpdate bool, bar *uiprogress.Bar) *pplugin.PluginInstallReport {
 	// start a channel for progress publications from plugin.Install
 	progress := make(chan struct{}, 5)
 	defer func() {
@@ -606,7 +606,7 @@ func installPlugin(ctx context.Context, resolvedPlugin pplugin.ResolvedPluginVer
 		} else {
 			msg = err.Error()
 		}
-		return &display.PluginInstallReport{
+		return &pplugin.PluginInstallReport{
 			Plugin:         fmt.Sprintf("%s@%s", name, constraint),
 			Skipped:        true,
 			SkipReason:     msg,
@@ -624,7 +624,7 @@ func installPlugin(ctx context.Context, resolvedPlugin pplugin.ResolvedPluginVer
 	if !image.ImageRef.IsFromTurbotHub() {
 		docURL = fmt.Sprintf("https://%s/%s", org, name)
 	}
-	return &display.PluginInstallReport{
+	return &pplugin.PluginInstallReport{
 		Plugin:         fmt.Sprintf("%s@%s", name, resolvedPlugin.Constraint),
 		Skipped:        false,
 		Version:        versionString,
@@ -642,12 +642,12 @@ func resolveUpdatePluginsFromArgs(args []string) ([]string, error) {

 	if len(plugins) == 0 && !(cmdconfig.Viper().GetBool("all")) {
 		// either plugin name(s) or "all" must be provided
-		return nil, fmt.Errorf("you need to provide at least one plugin to update or use the %s flag", constants.Bold("--all"))
+		return nil, fmt.Errorf("you need to provide at least one plugin to update or use the %s flag", pconstants.Bold("--all"))
 	}

 	if len(plugins) > 0 && cmdconfig.Viper().GetBool(pconstants.ArgAll) {
 		// we can't allow update and install at the same time
-		return nil, fmt.Errorf("%s cannot be used when updating specific plugins", constants.Bold("`--all`"))
+		return nil, fmt.Errorf("%s cannot be used when updating specific plugins", pconstants.Bold("`--all`"))
 	}

 	return plugins, nil
--- a/cmd/service.go
+++ b/cmd/service.go
@@ -648,7 +648,7 @@ To keep the service running after the %s session completes, use %s.
 			msg,
 			fmt.Sprintf("steampipe %s", dbState.Invoker),
 			dbState.Invoker,
-			constants.Bold("steampipe service start"),
+			pconstants.Bold("steampipe service start"),
 		)
 	}

@@ -662,7 +662,7 @@ To keep the service running after the %s session completes, use %s.
 Service is running, but the Plugin Manager cannot be recovered.
 Please use %s to recover the service
 `,
-			constants.Bold("steampipe service restart"),
+			pconstants.Bold("steampipe service restart"),
 		))
 	}
 }
@@ -675,7 +675,7 @@ To force stop the service, use %s

 `,
 		fmt.Sprintf("steampipe %s", invoker),
-		constants.Bold("steampipe service stop --force"),
+		pconstants.Bold("steampipe service stop --force"),
 	)
 }

@@ -687,7 +687,7 @@ Cannot stop service since there are clients connected to the service.
 To force stop the service, use %s

 `,
-		constants.Bold("steampipe service stop --force"),
+		pconstants.Bold("steampipe service stop --force"),
 	)
 }

--- a/pkg/cmdconfig/cmd_hooks.go
+++ b/pkg/cmdconfig/cmd_hooks.go
@@ -476,5 +476,5 @@ func displayDeprecationWarnings(errorsAndWarnings perror_helpers.ErrorAndWarning
 }

 func displayPpDeprecationWarning() {
-	fmt.Fprintf(color.Error, "\n%s Steampipe mods and dashboards have been moved to %s. This command %s in a future version. Migration guide - https://powerpipe.io/blog/migrating-from-steampipe \n", color.YellowString("Deprecation warning:"), constants.Bold("Powerpipe"), constants.Bold("will be removed"))
+	fmt.Fprintf(color.Error, "\n%s Steampipe mods and dashboards have been moved to %s. This command %s in a future version. Migration guide - https://powerpipe.io/blog/migrating-from-steampipe \n", color.YellowString("Deprecation warning:"), pconstants.Bold("Powerpipe"), pconstants.Bold("will be removed"))
 }
--- a/pkg/connection/refresh_connections_state.go
+++ b/pkg/connection/refresh_connections_state.go
@@ -301,7 +301,7 @@ func (s *refreshConnectionState) addMissingPluginWarnings() {
 			len(connectionNames),
 			utils.Pluralize("connection", len(connectionNames)),
 			utils.Pluralize("is", len(pluginNames)),
-			constants.Bold(fmt.Sprintf("steampipe plugin install %s", strings.Join(pluginNames, " ")))))
+			pconstants.Bold(fmt.Sprintf("steampipe plugin install %s", strings.Join(pluginNames, " ")))))
 	}
 }

--- a/pkg/constants/ascii.go
+++ b/pkg/constants/ascii.go
@@ -1,12 +0,0 @@
-package constants
-
-var (
-	// OptLeftArrowASCIICode ::
-	OptLeftArrowASCIICode = []byte{0x1b, 0x62}
-	// OptRightArrowASCIICode ::
-	OptRightArrowASCIICode = []byte{0x1b, 0x66}
-	// AltLeftArrowASCIICode ::
-	AltLeftArrowASCIICode = []byte{0x1b, 0x1b, 0x5B, 0x44}
-	// AltRightArrowASCIICode ::
-	AltRightArrowASCIICode = []byte{0x1b, 0x1b, 0x5B, 0x43}
-)
--- a/pkg/constants/colors.go
+++ b/pkg/constants/colors.go
@@ -1,163 +0,0 @@
-package constants
-
-import (
-	"github.com/fatih/color"
-	"github.com/logrusorgru/aurora"
-)
-
-var (
-	ColoredErr  = color.RedString("Error")
-	ColoredWarn = color.YellowString("Warning")
-)
-
-// Colors is a map of string to aurora colour value
-var Colors = map[string]func(arg interface{}) aurora.Value{
-	"bold":       Bold,
-	"italic":     Italic,
-	"underline":  Underline,
-	"slow-blink": SlowBlink,
-
-	"black":   Black,
-	"red":     Red,
-	"green":   Green,
-	"yellow":  Yellow,
-	"blue":    Blue,
-	"magenta": Magenta,
-	"cyan":    Cyan,
-	"white":   White,
-
-	"bold-black":   BoldBlack,
-	"bold-red":     BoldRed,
-	"bold-green":   BoldGreen,
-	"bold-yellow":  BoldYellow,
-	"bold-blue":    BoldBlue,
-	"bold-magenta": BoldMagenta,
-	"bold-cyan":    BoldCyan,
-	"bold-white":   BoldWhite,
-
-	"bright-black":   BrightBlack,
-	"bright-red":     BrightRed,
-	"bright-green":   BrightGreen,
-	"bright-yellow":  BrightYellow,
-	"bright-blue":    BrightBlue,
-	"bright-magenta": BrightMagenta,
-	"bright-cyan":    BrightCyan,
-	"bright-white":   BrightWhite,
-
-	"bold-bright-black":   BoldBrightBlack,
-	"bold-bright-red":     BoldBrightRed,
-	"bold-bright-green":   BoldBrightGreen,
-	"bold-bright-yellow":  BoldBrightYellow,
-	"bold-bright-blue":    BoldBrightBlue,
-	"bold-bright-magenta": BoldBrightMagenta,
-	"bold-bright-cyan":    BoldBrightCyan,
-	"bold-bright-white":   BoldBrightWhite,
-
-	"gray1": Gray1,
-	"gray2": Gray2,
-	"gray3": Gray3,
-	"gray4": Gray4,
-	"gray5": Gray5,
-}
-
-var Bold = aurora.Bold
-var Italic = aurora.Italic
-var Underline = aurora.Underline
-var SlowBlink = aurora.SlowBlink
-var Blink = aurora.Blink
-
-var Black = aurora.Black
-var Red = aurora.Red
-var Green = aurora.Green
-var Yellow = aurora.Yellow
-var Blue = aurora.Blue
-var Magenta = aurora.Magenta
-var Cyan = aurora.Cyan
-var White = aurora.White
-
-// bright colors
-
-var BrightBlack = aurora.BrightBlack
-var BrightRed = aurora.BrightRed
-var BrightGreen = aurora.BrightGreen
-var BrightYellow = aurora.BrightYellow
-var BrightBlue = aurora.BrightBlue
-var BrightMagenta = aurora.BrightMagenta
-var BrightCyan = aurora.BrightCyan
-var BrightWhite = aurora.BrightWhite
-
-// bold colors
-
-func BoldBlack(arg interface{}) aurora.Value {
-	return Bold(Black(arg))
-}
-func BoldRed(arg interface{}) aurora.Value {
-	return Bold(Red(arg))
-}
-func BoldGreen(arg interface{}) aurora.Value {
-	return Bold(Green(arg))
-}
-func BoldYellow(arg interface{}) aurora.Value {
-	return Bold(Yellow(arg))
-}
-func BoldBlue(arg interface{}) aurora.Value {
-	return Bold(Blue(arg))
-}
-func BoldMagenta(arg interface{}) aurora.Value {
-	return Bold(Magenta(arg))
-}
-func BoldCyan(arg interface{}) aurora.Value {
-	return Bold(Cyan(arg))
-}
-func BoldWhite(arg interface{}) aurora.Value {
-	return Bold(White(arg))
-}
-
-// bold bright  colors
-
-func BoldBrightBlack(arg interface{}) aurora.Value {
-	return Bold(BrightBlack(arg))
-}
-func BoldBrightRed(arg interface{}) aurora.Value {
-	return Bold(BrightRed(arg))
-}
-func BoldBrightGreen(arg interface{}) aurora.Value {
-	return Bold(BrightGreen(arg))
-}
-func BoldBrightYellow(arg interface{}) aurora.Value {
-	return Bold(BrightYellow(arg))
-}
-func BoldBrightBlue(arg interface{}) aurora.Value {
-	return Bold(BrightBlue(arg))
-}
-func BoldBrightMagenta(arg interface{}) aurora.Value {
-	return Bold(BrightMagenta(arg))
-}
-func BoldBrightCyan(arg interface{}) aurora.Value {
-	return Bold(BrightCyan(arg))
-}
-func BoldBrightWhite(arg interface{}) aurora.Value {
-	return Bold(BrightWhite(arg))
-}
-
-// various preset grays - lower number is a darker grey
-
-func Gray1(arg interface{}) aurora.Value {
-	return aurora.Gray(6, arg)
-}
-
-func Gray2(arg interface{}) aurora.Value {
-	return aurora.Gray(10, arg)
-}
-
-func Gray3(arg interface{}) aurora.Value {
-	return aurora.Gray(14, arg)
-}
-
-func Gray4(arg interface{}) aurora.Value {
-	return aurora.Gray(16, arg)
-}
-
-func Gray5(arg interface{}) aurora.Value {
-	return aurora.Gray(20, arg)
-}
--- a/pkg/db/db_client/db_client_execute.go
+++ b/pkg/db/db_client/db_client_execute.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	queryresult2 "github.com/turbot/pipe-fittings/queryresult"
 	"log"
 	"net/netip"
 	"strings"
@@ -67,7 +68,7 @@ func (c *DbClient) ExecuteSyncInSession(ctx context.Context, session *db_common.
 		}
 	}
 	if c.shouldFetchTiming() {
-		syncResult.TimingResult = <-result.TimingResult
+		syncResult.Timing = <-result.Timing
 	}

 	return syncResult, err
@@ -147,7 +148,7 @@ func (c *DbClient) ExecuteInSession(ctx context.Context, session *db_common.Data
 		// define a callback which fetches the timing information
 		// this will be invoked after reading rows is complete but BEFORE closing the rows object (which closes the connection)
 		timingCallback := func() {
-			c.getQueryTiming(ctxExecute, startTime, session, result.TimingResult)
+			c.getQueryTiming(ctxExecute, startTime, session, result.Timing)
 		}

 		// read in the rows and stream to the query result object
@@ -331,7 +332,7 @@ Loop:
 	}
 }

-func readRow(rows pgx.Rows, cols []*queryresult.ColumnDef) ([]interface{}, error) {
+func readRow(rows pgx.Rows, cols []*queryresult2.ColumnDef) ([]interface{}, error) {
 	columnValues, err := rows.Values()
 	if err != nil {
 		return nil, error_helpers.WrapError(err)
@@ -339,7 +340,7 @@ func readRow(rows pgx.Rows, cols []*queryresult.ColumnDef) ([]interface{}, error
 	return populateRow(columnValues, cols)
 }

-func populateRow(columnValues []interface{}, cols []*queryresult.ColumnDef) ([]interface{}, error) {
+func populateRow(columnValues []interface{}, cols []*queryresult2.ColumnDef) ([]interface{}, error) {
 	result := make([]interface{}, len(columnValues))
 	for i, columnValue := range columnValues {
 		if columnValue != nil {
--- a/pkg/db/db_client/pgx_types.go
+++ b/pkg/db/db_client/pgx_types.go
@@ -2,13 +2,13 @@ package db_client

 import (
 	"fmt"
+	"github.com/turbot/pipe-fittings/queryresult"
 	"strconv"
 	"strings"

 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/turbot/pipe-fittings/utils"
-	"github.com/turbot/steampipe/pkg/query/queryresult"
 )

 // ColumnTypeDatabaseTypeName returns the database system type name. If the name is unknown the OID is returned.
--- a/pkg/db/db_common/client.go
+++ b/pkg/db/db_common/client.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgxpool"
+	pqueryresult "github.com/turbot/pipe-fittings/queryresult"
 	"github.com/turbot/steampipe/pkg/query/queryresult"
 )

@@ -21,7 +22,7 @@ type Client interface {
 	AcquireSession(context.Context) *AcquireSessionResult

 	ExecuteSync(context.Context, string, ...any) (*queryresult.SyncQueryResult, error)
-	Execute(context.Context, string, ...any) (*queryresult.Result, error)
+	Execute(context.Context, string, ...any) (*pqueryresult.Result[queryresult.TimingResultStream], error)

 	ExecuteSyncInSession(context.Context, *DatabaseSession, string, ...any) (*queryresult.SyncQueryResult, error)
 	ExecuteInSession(context.Context, *DatabaseSession, func(), string, ...any) (*queryresult.Result, error)
--- a/pkg/db/db_common/execute.go
+++ b/pkg/db/db_common/execute.go
@@ -2,9 +2,9 @@ package db_common

 import (
 	"context"
+	"github.com/turbot/steampipe/pkg/query/queryresult"

 	"github.com/turbot/pipe-fittings/utils"
-	"github.com/turbot/steampipe/pkg/query/queryresult"
 )

 // ExecuteQuery executes a single query. If shutdownAfterCompletion is true, shutdown the client after completion
--- a/pkg/db/db_local/install.go
+++ b/pkg/db/db_local/install.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/turbot/pipe-fittings/app_specific"
+	constants2 "github.com/turbot/pipe-fittings/constants"
 	"log"
 	"os"
 	"os/exec"
@@ -65,7 +66,7 @@ func EnsureDBInstalled(ctx context.Context) (err error) {
 		return err
 	}
 	if dbState != nil {
-		return fmt.Errorf("cannot install service - a previous version of the Steampipe service is still running. To stop running services, use %s ", constants.Bold("steampipe service stop"))
+		return fmt.Errorf("cannot install service - a previous version of the Steampipe service is still running. To stop running services, use %s ", constants2.Bold("steampipe service stop"))
 	}

 	log.Println("[TRACE] calling removeRunningInstanceInfo")
@@ -209,7 +210,7 @@ func prepareDb(ctx context.Context) error {
 		// get the message renderer from the context
 		// this allows the interactive client init to inject a custom renderer
 		messageRenderer := statushooks.MessageRendererFromContext(ctx)
-		messageRenderer("%s updated to %s.", constants.Bold("steampipe-postgres-fdw"), constants.Bold(constants.FdwVersion))
+		messageRenderer("%s updated to %s.", constants2.Bold("steampipe-postgres-fdw"), constants2.Bold(constants.FdwVersion))
 	}

 	if needsInit() {
@@ -246,7 +247,7 @@ func installFDW(ctx context.Context, firstSetup bool) (string, error) {
 			}
 		}()
 	}
-	statushooks.SetStatus(ctx, fmt.Sprintf("Download & install %s…", constants.Bold("steampipe-postgres-fdw")))
+	statushooks.SetStatus(ctx, fmt.Sprintf("Download & install %s…", constants2.Bold("steampipe-postgres-fdw")))
 	return ociinstaller.InstallFdw(ctx, filepaths.GetDatabaseLocation())
 }

--- a/pkg/db/db_local/service.go
+++ b/pkg/db/db_local/service.go
@@ -2,6 +2,7 @@ package db_local

 import (
 	"fmt"
+	"github.com/turbot/pipe-fittings/constants"
 	"log"
 	"os"
 	"strconv"
@@ -9,7 +10,6 @@ import (

 	filehelpers "github.com/turbot/go-kit/files"
 	"github.com/turbot/pipe-fittings/utils"
-	"github.com/turbot/steampipe/pkg/constants"
 	"github.com/turbot/steampipe/pkg/filepaths"
 )

--- a/pkg/db/db_local/start_services.go
+++ b/pkg/db/db_local/start_services.go
@@ -261,7 +261,7 @@ func startDB(ctx context.Context, listenAddresses []string, port int, invoker co
 	}

 	if err := putils.IsPortBindable(putils.GetFirstListenAddress(listenAddresses), port); err != nil {
-		return res.SetError(fmt.Errorf("cannot listen on port %d and %s %s. To check if there's any other steampipe services running, use %s", constants.Bold(port), putils.Pluralize("address", len(listenAddresses)), constants.Bold(strings.Join(listenAddresses, ",")), constants.Bold("steampipe service status --all")))
+		return res.SetError(fmt.Errorf("cannot listen on port %d and %s %s. To check if there's any other steampipe services running, use %s", pconstants.Bold(port), putils.Pluralize("address", len(listenAddresses)), pconstants.Bold(strings.Join(listenAddresses, ",")), pconstants.Bold("steampipe service status --all")))
 	}

 	if err := migrateLegacyPasswordFile(); err != nil {
--- a/pkg/display/column.go
+++ b/pkg/display/column.go
@@ -3,11 +3,11 @@ package display
 import (
 	"encoding/json"
 	"fmt"
+	"github.com/turbot/pipe-fittings/queryresult"
 	"time"

 	typeHelpers "github.com/turbot/go-kit/types"
 	"github.com/turbot/steampipe/pkg/constants"
-	"github.com/turbot/steampipe/pkg/query/queryresult"
 )

 // columnNames builds a list of name from a slice of column defs - respecting the original name if present
--- a/pkg/display/display.go
+++ b/pkg/display/display.go
@@ -7,6 +7,7 @@ import (
 	"encoding/csv"
 	"encoding/json"
 	"fmt"
+	pqueryresult "github.com/turbot/pipe-fittings/queryresult"
 	"io"
 	"log"
 	"os"
@@ -22,7 +23,6 @@ import (
 	"github.com/spf13/viper"
 	"github.com/turbot/go-kit/helpers"
 	pconstants "github.com/turbot/pipe-fittings/constants"
-	pqueryresult "github.com/turbot/pipe-fittings/queryresult"
 	"github.com/turbot/steampipe/pkg/cmdconfig"
 	"github.com/turbot/steampipe/pkg/constants"
 	"github.com/turbot/steampipe/pkg/error_helpers"
@@ -431,7 +431,7 @@ func getTiming(result *queryresult.Result, count int) *queryresult.TimingResult
 		return nil
 	}
 	// now we have iterated the rows, get the timing
-	timingResult := <-result.TimingResult
+	timingResult := <-result.Timing
 	// set rows returned
 	timingResult.RowsReturned = int64(count)

--- a/pkg/display/plugin_install.go
+++ b/pkg/display/plugin_install.go
@@ -1,164 +0,0 @@
-package display
-
-import (
-	"fmt"
-	"sort"
-	"strings"
-
-	pconstants "github.com/turbot/pipe-fittings/constants"
-	"github.com/turbot/pipe-fittings/ociinstaller"
-	"github.com/turbot/pipe-fittings/utils"
-	"github.com/turbot/steampipe/pkg/constants"
-)
-
-type PluginInstallReports []*PluginInstallReport
-
-// making the type compatible with sort.Interface so that we can use the sort package utilities
-func (i PluginInstallReports) Len() int                 { return len(i) }
-func (i PluginInstallReports) Swap(lIdx, rIdx int)      { i[lIdx], i[rIdx] = i[rIdx], i[lIdx] }
-func (i PluginInstallReports) Less(lIdx, rIdx int) bool { return i[lIdx].Plugin < i[rIdx].Plugin }
-
-type PluginInstallReport struct {
-	Skipped        bool
-	Plugin         string
-	SkipReason     string
-	DocURL         string
-	Version        string
-	IsUpdateReport bool
-}
-
-func (i *PluginInstallReport) skipString() string {
-	ref := ociinstaller.NewImageRef(i.Plugin)
-	_, name, constraint := ref.GetOrgNameAndStream()
-
-	return fmt.Sprintf("Plugin:   %s\nReason:   %s", fmt.Sprintf("%s@%s", name, constraint), i.SkipReason)
-}
-
-func (i *PluginInstallReport) installString() string {
-	thisReport := []string{}
-	if i.IsUpdateReport {
-		thisReport = append(
-			thisReport,
-			fmt.Sprintf("Updated plugin: %s%s", constants.Bold(i.Plugin), i.Version),
-		)
-		if len(i.DocURL) > 0 {
-			thisReport = append(
-				thisReport,
-				fmt.Sprintf("Documentation:  %s", i.DocURL),
-			)
-		}
-	} else {
-		thisReport = append(
-			thisReport,
-			fmt.Sprintf("Installed plugin: %s%s", constants.Bold(i.Plugin), i.Version),
-		)
-		if len(i.DocURL) > 0 {
-			thisReport = append(
-				thisReport,
-				fmt.Sprintf("Documentation:    %s", i.DocURL),
-			)
-		}
-	}
-
-	return strings.Join(thisReport, "\n")
-}
-
-func (i *PluginInstallReport) String() string {
-	if !i.Skipped {
-		return i.installString()
-	} else {
-		return i.skipString()
-	}
-}
-
-// PrintInstallReports Prints out the installation reports onto the console
-func PrintInstallReports(reports PluginInstallReports, isUpdateReport bool) {
-	installedOrUpdated := PluginInstallReports{}
-	canBeInstalled := PluginInstallReports{}
-	canBeUpdated := PluginInstallReports{}
-	notFound := PluginInstallReports{}
-
-	for _, report := range reports {
-		report.IsUpdateReport = isUpdateReport
-		if !report.Skipped {
-			installedOrUpdated = append(installedOrUpdated, report)
-		} else if report.SkipReason == pconstants.InstallMessagePluginNotInstalled {
-			canBeInstalled = append(canBeInstalled, report)
-		} else if report.SkipReason == pconstants.InstallMessagePluginAlreadyInstalled {
-			canBeUpdated = append(canBeUpdated, report)
-		} else if report.SkipReason == pconstants.InstallMessagePluginNotFound {
-			notFound = append(notFound, report)
-		}
-	}
-
-	// sort the report
-	sort.Stable(reports)
-	// sort the individual chunks
-	sort.Stable(installedOrUpdated)
-	sort.Stable(canBeInstalled)
-	sort.Stable(canBeUpdated)
-	sort.Stable(notFound)
-
-	if len(installedOrUpdated) > 0 {
-		fmt.Println()
-		asString := []string{}
-		for _, report := range installedOrUpdated {
-			asString = append(asString, report.installString())
-		}
-		fmt.Println(strings.Join(asString, "\n\n"))
-	}
-
-	if len(installedOrUpdated) < len(reports) {
-		installSkipReports := []string{}
-		for _, report := range reports {
-			showReport := true
-			if report.SkipReason == pconstants.InstallMessagePluginAlreadyInstalled || report.SkipReason == pconstants.InstallMessagePluginLatestAlreadyInstalled {
-				showReport = false
-			}
-			if report.Skipped && showReport {
-				installSkipReports = append(installSkipReports, report.skipString())
-			}
-		}
-
-		skipCount := len(installSkipReports)
-		if (len(installSkipReports)) > 0 {
-			fmt.Printf(
-				"\nSkipped the following %s:\n\n%s\n",
-				utils.Pluralize("plugin", skipCount),
-				strings.Join(installSkipReports, "\n\n"),
-			)
-		}
-
-		if len(canBeInstalled) > 0 {
-			pluginList := []string{}
-			for _, r := range canBeInstalled {
-				pluginList = append(pluginList, r.Plugin)
-			}
-			fmt.Println()
-			fmt.Printf(
-				"To install %s which %s not installed, please run %s\n",
-				utils.Pluralize("plugin", len(canBeInstalled)),
-				utils.Pluralize("is", len(canBeInstalled)),
-				constants.Bold(fmt.Sprintf(
-					"steampipe plugin install %s",
-					strings.Join(pluginList, " "),
-				)),
-			)
-		}
-
-		if len(canBeUpdated) > 0 {
-			pluginList := []string{}
-			for _, r := range canBeUpdated {
-				pluginList = append(pluginList, r.Plugin)
-			}
-			fmt.Println()
-			fmt.Printf(
-				"To update %s %s: %s\nTo update all plugins: %s",
-				utils.Pluralize("this", len(pluginList)),
-				utils.Pluralize("plugin", len(pluginList)),
-				constants.Bold(fmt.Sprintf("steampipe plugin update %s", strings.Join(pluginList, " "))),
-				constants.Bold(fmt.Sprintln("steampipe plugin update --all")),
-			)
-		}
-	}
-}
--- a/pkg/error_helpers/errors.go
+++ b/pkg/error_helpers/errors.go
@@ -3,8 +3,7 @@ package error_helpers
 import (
 	"errors"
 	"fmt"
-
-	"github.com/turbot/steampipe/pkg/constants"
+	"github.com/turbot/pipe-fittings/constants"
 )

 var MissingCloudTokenError = fmt.Errorf("Not authenticated for Turbot Pipes.\nPlease run %s or setup a token.", constants.Bold("steampipe login"))
--- a/pkg/error_helpers/utils.go
+++ b/pkg/error_helpers/utils.go
@@ -12,7 +12,6 @@ import (
 	"github.com/shiena/ansicolor"
 	"github.com/spf13/viper"
 	pconstants "github.com/turbot/pipe-fittings/constants"
-	"github.com/turbot/steampipe/pkg/constants"
 	"github.com/turbot/steampipe/pkg/statushooks"
 )

@@ -47,7 +46,7 @@ func ShowError(ctx context.Context, err error) {
 	}
 	err = HandleCancelError(err)
 	statushooks.Done(ctx)
-	fmt.Fprintf(color.Error, "%s: %v\n", constants.ColoredErr, TransformErrorToSteampipe(err))
+	fmt.Fprintf(color.Error, "%s: %v\n", pconstants.ColoredErr, TransformErrorToSteampipe(err))
 }

 // ShowErrorWithMessage displays the given error nicely with the given message
@@ -57,7 +56,7 @@ func ShowErrorWithMessage(ctx context.Context, err error, message string) {
 	}
 	err = HandleCancelError(err)
 	statushooks.Done(ctx)
-	fmt.Fprintf(color.Error, "%s: %s - %v\n", constants.ColoredErr, message, TransformErrorToSteampipe(err))
+	fmt.Fprintf(color.Error, "%s: %s - %v\n", pconstants.ColoredErr, message, TransformErrorToSteampipe(err))
 }

 // TransformErrorToSteampipe removes the pq: and rpc error prefixes along
@@ -110,7 +109,7 @@ func ShowWarning(warning string) {
 	if len(warning) == 0 {
 		return
 	}
-	fmt.Fprintf(color.Error, "%s: %v\n", constants.ColoredWarn, warning)
+	fmt.Fprintf(color.Error, "%s: %v\n", pconstants.ColoredWarn, warning)
 }

 func CombineErrorsWithPrefix(prefix string, errors ...error) error {
--- a/pkg/interactive/interactive_client.go
+++ b/pkg/interactive/interactive_client.go
@@ -132,7 +132,7 @@ func (c *InteractiveClient) InteractivePrompt(parentContext context.Context) {
 	statushooks.Message(
 		ctx,
 		fmt.Sprintf("Welcome to Steampipe v%s", version.SteampipeVersion.String()),
-		fmt.Sprintf("For more information, type %s", constants.Bold(".help")),
+		fmt.Sprintf("For more information, type %s", pconstants.Bold(".help")),
 	)

 	// run the prompt in a goroutine, so we can also detect async initialisation errors
@@ -288,22 +288,22 @@ func (c *InteractiveClient) runInteractivePrompt(ctx context.Context) {
 		}),
 		// Opt+LeftArrow
 		prompt.OptionAddASCIICodeBind(prompt.ASCIICodeBind{
-			ASCIICode: constants.OptLeftArrowASCIICode,
+			ASCIICode: pconstants.OptLeftArrowASCIICode,
 			Fn:        prompt.GoLeftWord,
 		}),
 		// Opt+RightArrow
 		prompt.OptionAddASCIICodeBind(prompt.ASCIICodeBind{
-			ASCIICode: constants.OptRightArrowASCIICode,
+			ASCIICode: pconstants.OptRightArrowASCIICode,
 			Fn:        prompt.GoRightWord,
 		}),
 		// Alt+LeftArrow
 		prompt.OptionAddASCIICodeBind(prompt.ASCIICodeBind{
-			ASCIICode: constants.AltLeftArrowASCIICode,
+			ASCIICode: pconstants.AltLeftArrowASCIICode,
 			Fn:        prompt.GoLeftWord,
 		}),
 		// Alt+RightArrow
 		prompt.OptionAddASCIICodeBind(prompt.ASCIICodeBind{
-			ASCIICode: constants.AltRightArrowASCIICode,
+			ASCIICode: pconstants.AltRightArrowASCIICode,
 			Fn:        prompt.GoRightWord,
 		}),
 		prompt.OptionBufferPreHook(func(input string) (modifiedInput string, ignore bool) {
--- a/pkg/interactive/metaquery/handler_cache.go
+++ b/pkg/interactive/metaquery/handler_cache.go
@@ -11,7 +11,6 @@ import (
 	"github.com/spf13/viper"
 	pconstants "github.com/turbot/pipe-fittings/constants"
 	"github.com/turbot/steampipe-plugin-sdk/v5/sperr"
-	"github.com/turbot/steampipe/pkg/constants"
 	"github.com/turbot/steampipe/pkg/db/db_common"
 )

@@ -99,9 +98,9 @@ func showCache(_ context.Context, input *HandlerInput) error {

 	fmt.Printf(
 		`Caching is %s. To turn it %s, type %s`,
-		constants.Bold(currentStatusString),
-		constants.Bold(action),
-		constants.Bold(fmt.Sprintf(".cache %s", action)),
+		pconstants.Bold(currentStatusString),
+		pconstants.Bold(action),
+		pconstants.Bold(fmt.Sprintf(".cache %s", action)),
 	)

 	// add an empty line here so that the rendering buffer can start from the next line
--- a/pkg/interactive/metaquery/handler_help.go
+++ b/pkg/interactive/metaquery/handler_help.go
@@ -3,6 +3,7 @@ package metaquery
 import (
 	"context"
 	"fmt"
+	constants2 "github.com/turbot/pipe-fittings/constants"
 	"sort"
 	"strings"

@@ -26,7 +27,7 @@ func doHelp(_ context.Context, _ *HandlerInput) error {
 	fmt.Printf("Welcome to Steampipe shell.\n\nTo start, simply enter your SQL query at the prompt:\n\n  select * from aws_iam_user\n\nCommon commands:\n\n%s\n\nAdvanced commands:\n\n%s\n\nDocumentation available at %s\n",
 		buildTable(commonCmdRows, true),
 		buildTable(advanceCmdRows, true),
-		constants.Bold("https://steampipe.io/docs"))
+		constants2.Bold("https://steampipe.io/docs"))
 	fmt.Println()
 	return nil
 }
--- a/pkg/interactive/metaquery/handler_inspect.go
+++ b/pkg/interactive/metaquery/handler_inspect.go
@@ -3,6 +3,7 @@ package metaquery
 import (
 	"context"
 	"fmt"
+	constants2 "github.com/turbot/pipe-fittings/constants"
 	"log"
 	"regexp"
 	"sort"
@@ -109,7 +110,7 @@ func listTables(ctx context.Context, input *HandlerInput) error {
 		fmt.Printf(`
 To get information about the columns in a table, run %s
 	
-`, constants.Bold(".inspect {connection}.{table}"))
+`, constants2.Bold(".inspect {connection}.{table}"))
 	} else {
 		// could be one of connectionName and {string}*
 		arg := input.args()[0]
@@ -193,7 +194,7 @@ func listConnections(ctx context.Context, input *HandlerInput) error {
 To get information about the tables in a connection, run %s
 To get information about the columns in a table, run %s

-`, constants.Bold(".inspect {connection}"), constants.Bold(".inspect {connection}.{table}"))
+`, constants2.Bold(".inspect {connection}"), constants2.Bold(".inspect {connection}.{table}"))

 	return nil
 }
--- a/pkg/interactive/metaquery/handler_inspect_legacy.go
+++ b/pkg/interactive/metaquery/handler_inspect_legacy.go
@@ -3,7 +3,7 @@ package metaquery
 import (
 	"context"
 	"fmt"
-	"github.com/turbot/steampipe/pkg/constants"
+	"github.com/turbot/pipe-fittings/constants"
 	"github.com/turbot/steampipe/pkg/display"
 	"sort"
 	"strings"
--- a/pkg/interactive/metaquery/handlers.go
+++ b/pkg/interactive/metaquery/handlers.go
@@ -55,8 +55,8 @@ func showTimingFlag() {
 	timing := cmdconfig.Viper().GetString(pconstants.ArgTiming)

 	fmt.Printf(`Timing is %s. Available options are: %s`,
-		constants.Bold(timing),
-		constants.Bold(strings.Join(maps.Keys(constants.QueryTimingValueLookup), ", ")))
+		pconstants.Bold(timing),
+		pconstants.Bold(strings.Join(maps.Keys(constants.QueryTimingValueLookup), ", ")))
 	// add an empty line here so that the rendering buffer can start from the next line
 	fmt.Println()

--- a/pkg/interactive/metaquery/validators.go
+++ b/pkg/interactive/metaquery/validators.go
@@ -71,9 +71,9 @@ func booleanValidator(metaquery string, validators ...validator) validator {
 			return ValidationResult{
 				Message: fmt.Sprintf(`%s mode is %s. You can %s it with: %s.`,
 					title,
-					constants.Bold(currentStatusString),
+					pconstants.Bold(currentStatusString),
 					actionString,
-					constants.Bold(fmt.Sprintf("%s %s", metaquery, newStatusString))),
+					pconstants.Bold(fmt.Sprintf("%s %s", metaquery, newStatusString))),
 			}
 		}
 		if numArgs > 1 {
--- a/pkg/interactive/run.go
+++ b/pkg/interactive/run.go
@@ -2,7 +2,6 @@ package interactive

 import (
 	"context"
-
 	"github.com/turbot/steampipe/pkg/constants"
 	"github.com/turbot/steampipe/pkg/db/db_local"
 	"github.com/turbot/steampipe/pkg/error_helpers"
--- a/pkg/query/queryresult/column_def.go
+++ b/pkg/query/queryresult/column_def.go
@@ -1,28 +0,0 @@
-package queryresult
-
-import "reflect"
-
-// ColumnDef is a struct used to store column information from query results
-type ColumnDef struct {
-	Name         string `json:"name"`
-	DataType     string `json:"data_type"`
-	isScalar     *bool
-	OriginalName string `json:"original_name"`
-}
-
-// IsScalar checks if the given value is a scalar value
-// it also mutates the containing ColumnDef so that it doesn't have to reflect
-// for all values in a column
-func (c *ColumnDef) IsScalar(v any) bool {
-	if c.isScalar == nil {
-		var scalar bool
-		switch reflect.ValueOf(v).Kind() {
-		case reflect.Array, reflect.Map, reflect.Slice, reflect.Struct:
-			scalar = false
-		default:
-			scalar = true
-		}
-		c.isScalar = &scalar
-	}
-	return *c.isScalar
-}
--- a/pkg/query/queryresult/result.go
+++ b/pkg/query/queryresult/result.go
@@ -1,5 +1,9 @@
 package queryresult

+import "github.com/turbot/pipe-fittings/queryresult"
+
+type TimingResultStream chan *TimingResult
+
 type TimingResult struct {
 	DurationMs          int64              `json:"duration_ms"`
 	Scans               []*ScanMetadataRow `json:"scans"`
@@ -21,42 +25,16 @@ func (r *TimingResult) Initialise(summary *QueryRowSummary, scans []*ScanMetadat
 	r.Scans = scans
 }

-type RowResult struct {
-	Data  []interface{}
-	Error error
-}
-type Result struct {
-	RowChan      *chan *RowResult
-	Cols         []*ColumnDef
-	TimingResult chan *TimingResult
+type Result = queryresult.Result[TimingResultStream]
+
+func NewResult(cols []*queryresult.ColumnDef) *Result {
+	return queryresult.NewResult[TimingResultStream](cols)
 }

-func NewResult(cols []*ColumnDef) *Result {
-	rowChan := make(chan *RowResult)
-	return &Result{
-		RowChan:      &rowChan,
-		Cols:         cols,
-		TimingResult: make(chan *TimingResult, 1),
-	}
-}
+type SyncQueryResult = queryresult.SyncQueryResult[*TimingResult]

-// IsExportSourceData implements ExportSourceData
-func (*Result) IsExportSourceData() {}
+type ResultStreamer = queryresult.ResultStreamer[TimingResultStream]

-// Close closes the row channel
-func (r *Result) Close() {
-	close(*r.RowChan)
-}
-
-func (r *Result) StreamRow(rowResult []interface{}) {
-	*r.RowChan <- &RowResult{Data: rowResult}
-}
-func (r *Result) StreamError(err error) {
-	*r.RowChan <- &RowResult{Error: err}
-}
-
-type SyncQueryResult struct {
-	Rows         []interface{}
-	Cols         []*ColumnDef
-	TimingResult *TimingResult
+func NewResultStreamer() *ResultStreamer {
+	return queryresult.NewResultStreamer[TimingResultStream]()
 }
--- a/pkg/query/queryresult/result_streamer.go
+++ b/pkg/query/queryresult/result_streamer.go
@@ -1,31 +0,0 @@
-package queryresult
-
-type ResultStreamer struct {
-	Results            chan *Result
-	allResultsReceived chan string
-}
-
-func NewResultStreamer() *ResultStreamer {
-	return &ResultStreamer{
-		// make buffered channel so we can always stream a single result
-		Results:            make(chan *Result, 1),
-		allResultsReceived: make(chan string, 1),
-	}
-}
-
-// StreamResult streams result on the Results channel, then waits for them to be read
-func (q *ResultStreamer) StreamResult(result *Result) {
-	q.Results <- result
-	// wait for the result to be read
-	<-q.allResultsReceived
-}
-
-// Close closes the result stream
-func (q *ResultStreamer) Close() {
-	close(q.Results)
-}
-
-// AllResultsRead is a signal that indicates the all results have been read from the stream
-func (q *ResultStreamer) AllResultsRead() {
-	q.allResultsReceived <- ""
-}
--- a/pkg/statushooks/spinner.go
+++ b/pkg/statushooks/spinner.go
@@ -2,6 +2,7 @@ package statushooks

 import (
 	"fmt"
+	"github.com/turbot/pipe-fittings/constants"
 	"os"
 	"strings"
 	"time"
@@ -9,7 +10,6 @@ import (
 	"github.com/briandowns/spinner"
 	"github.com/fatih/color"
 	"github.com/karrick/gows"
-	"github.com/turbot/steampipe/pkg/constants"
 )

 // spinner format:
--- a/pkg/steampipeconfig/connection_plugin.go
+++ b/pkg/steampipeconfig/connection_plugin.go
@@ -12,7 +12,6 @@ import (
 	sdkgrpc "github.com/turbot/steampipe-plugin-sdk/v5/grpc"
 	sdkproto "github.com/turbot/steampipe-plugin-sdk/v5/grpc/proto"
 	sdkplugin "github.com/turbot/steampipe-plugin-sdk/v5/plugin"
-	"github.com/turbot/steampipe/pkg/constants"
 	"github.com/turbot/steampipe/pkg/error_helpers"
 	"github.com/turbot/steampipe/pkg/pluginmanager_service/grpc/proto"
 	pluginshared "github.com/turbot/steampipe/pkg/pluginmanager_service/grpc/shared"
@@ -206,7 +205,7 @@ func handleGetFailures(getResponse *proto.GetResponse, res *RefreshConnectionRes
 			utils.Pluralize("plugin", pluginCount),
 			compatibilityErrorConnectionCount,
 			utils.Pluralize("connection", compatibilityErrorConnectionCount),
-			constants.Bold(fmt.Sprintf("steampipe plugin update %s", strings.Join(maps.Keys(pluginsWithCompatibilityError), " "))))
+			pconstants.Bold(fmt.Sprintf("steampipe plugin update %s", strings.Join(maps.Keys(pluginsWithCompatibilityError), " "))))
 		res.AddWarning(compatibilityWarning)
 	}
 }
--- a/pkg/steampipeconfig/connection_updates_validate.go
+++ b/pkg/steampipeconfig/connection_updates_validate.go
@@ -2,12 +2,12 @@ package steampipeconfig

 import (
 	"fmt"
+	"github.com/turbot/pipe-fittings/constants"
 	"log"
 	"strings"

 	"github.com/turbot/pipe-fittings/utils"
 	sdkversion "github.com/turbot/steampipe-plugin-sdk/v5/version"
-	"github.com/turbot/steampipe/pkg/constants"
 )

 func (u *ConnectionUpdates) validate() {
--- a/pkg/task/available_versions.go
+++ b/pkg/task/available_versions.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"github.com/turbot/pipe-fittings/constants"
 	plugin2 "github.com/turbot/pipe-fittings/plugin"
 	"os"
 	"sort"
@@ -12,7 +13,6 @@ import (
 	"github.com/fatih/color"
 	"github.com/olekukonko/tablewriter"
 	"github.com/turbot/pipe-fittings/utils"
-	"github.com/turbot/steampipe/pkg/constants"
 )

 type AvailableVersionCache struct {