Updates for steampipe_plugin_column table. Closes #4022

* Fix inconsistent value for plugin in steampipe_plugin_column table - always use long name.
* Rename `plugin_name` column to `plugin`
* Ensure empty col values

pkg/introspection/plugin_column_table_sql.go

@@ -13,7 +13,7 @@ import (
 func GetPluginColumnTableCreateSql() db_common.QueryWithArgs {
 	return db_common.QueryWithArgs{
 		Query: fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.%s (
-				plugin_name TEXT NOT NULL,
+				plugin TEXT NOT NULL,
 				table_name TEXT NOT NULL,
 				name TEXT NOT NULL,
 				type TEXT NOT NULL,
@@ -61,10 +61,10 @@ func GetPluginColumnTablePopulateSql(
 		}
 	}
 
-	var listConfig, getConfig keyColumn
+	var listConfig, getConfig *keyColumn
 
 	if getKeyColumn != nil {
-		getConfig = newKeyColumn(getKeyColumn.Operators, getConfig.Required, getKeyColumn.CacheMatch)
+		getConfig = newKeyColumn(getKeyColumn.Operators, getKeyColumn.Require, getKeyColumn.CacheMatch)
 	}
 	if listKeyColumn != nil {
 		listConfig = newKeyColumn(listKeyColumn.Operators, listKeyColumn.Require, listKeyColumn.CacheMatch)
@@ -74,9 +74,14 @@ func GetPluginColumnTablePopulateSql(
 	if s, ok := defaultValue.(string); ok {
 		defaultValue = fmt.Sprintf(`"%s"`, s)
 	}
+	var hydrate any = nil
+	if columnSchema.Hydrate != "" {
+		hydrate = columnSchema.Hydrate
+	}
+
 	q := db_common.QueryWithArgs{
 		Query: fmt.Sprintf(`INSERT INTO %s.%s (
-				plugin_name,
+				plugin,
 				table_name ,
 				name,
 				type,
@@ -95,7 +100,7 @@ func GetPluginColumnTablePopulateSql(
 			description,
 			listConfig,
 			getConfig,
-			columnSchema.Hydrate,
+			hydrate,
 			defaultValue,
 		},
 	}
@@ -117,7 +122,7 @@ func GetPluginColumnTableDeletePluginSql(plugin string) db_common.QueryWithArgs
 	return db_common.QueryWithArgs{
 		Query: fmt.Sprintf(
 			`DELETE FROM %s.%s
-WHERE plugin_name = $1;`,
+WHERE plugin = $1;`,
 			constants.InternalSchema,
 			constants.PluginColumnTable,
 		),
@@ -138,14 +143,14 @@ func GetPluginColumnTableGrantSql() db_common.QueryWithArgs {
 
 type keyColumn struct {
 	Operators  []string `json:"operators,omitempty"`
-	Required   string   `json:"required,omitempty"`
+	Require    string   `json:"require,omitempty"`
 	CacheMatch string   `json:"cache_match,omitempty"`
 }
 
-func newKeyColumn(operators []string, required string, cacheMatch string) keyColumn {
-	return keyColumn{
+func newKeyColumn(operators []string, require string, cacheMatch string) *keyColumn {
+	return &keyColumn{
 		Operators:  cleanOperators(operators),
-		Required:   required,
+		Require:    require,
 		CacheMatch: cacheMatch,
 	}
 }
@@ -159,13 +164,13 @@ func cleanOperators(operators []string) []string {
 		case "<>":
 			operator = "!="
 		case ">":
-			operator = "GT"
+			operator = "gt"
 		case "<":
-			operator = "LT"
+			operator = "lt"
 		case ">=":
-			operator = "GE"
+			operator = "ge"
 		case "<=":
-			operator = "LE"
+			operator = "le"
 		}
 		res[i] = operator
 	}

pkg/pluginmanager_service/plugin_manager_connection_config.go

@@ -56,7 +56,6 @@ func (m *PluginManager) sendUpdateConnectionConfigs(requestMap map[string]*sdkpr
 			errors = append(errors, err)
 			continue
 		}
-
 		err = pluginClient.UpdateConnectionConfigs(req)
 		if err != nil {
 			errors = append(errors, err)

pkg/pluginmanager_service/plugin_manager_plugin_columns.go

@@ -2,6 +2,7 @@ package pluginmanager_service
 
 import (
 	"context"
+	"fmt"
 	sdkgrpc "github.com/turbot/steampipe-plugin-sdk/v5/grpc"
 	"github.com/turbot/steampipe-plugin-sdk/v5/grpc/proto"
 	sdkplugin "github.com/turbot/steampipe-plugin-sdk/v5/plugin"
@@ -12,30 +13,67 @@ import (
 	pb "github.com/turbot/steampipe/pkg/pluginmanager_service/grpc/proto"
 	"golang.org/x/exp/maps"
 	"log"
+	"slices"
+	"strings"
 )
 
 func (m *PluginManager) initialisePluginColumns(ctx context.Context) error {
-	pluginColumnTableExists, err := m.tableExists(ctx, constants.InternalSchema, constants.PluginColumnTable)
-	if err != nil {
-		return err
-	}
-
-	if !pluginColumnTableExists {
+	if m.shouldBootstrapPluginColumnTable(ctx) {
 		return m.bootstrapPluginColumnTable(ctx)
 	}
 	return nil
 }
 
+func (m *PluginManager) shouldBootstrapPluginColumnTable(ctx context.Context) bool {
+	pluginColumnTableExists, err := m.tableExists(ctx, constants.InternalSchema, constants.PluginColumnTable)
+	if err != nil || !pluginColumnTableExists {
+		return true
+	}
+	// check columns match
+	query := fmt.Sprintf(`SELECT column_name
+  FROM information_schema.columns
+ WHERE table_schema = '%s'
+   AND table_name   = '%s'`, constants.InternalSchema, constants.PluginColumnTable)
+
+	rows, err := m.pool.Query(ctx, query)
+	if err != nil {
+		return true
+	}
+	defer rows.Close()
+	var columns []string
+	// Iterate through the rows
+	for rows.Next() {
+		var s string
+		err := rows.Scan(&s)
+		if err != nil {
+			return true
+		}
+		columns = append(columns, s)
+	}
+
+	// Check for errors from iterating over rows
+	if err = rows.Err(); err != nil {
+		return true
+	}
+
+	expectedColumns := []string{"plugin_", "table_name", "name", "type", " description", "list_config", "get_config", "hydrate_name", "default_value"}
+	return !slices.Equal(columns, expectedColumns)
+}
+
 func (m *PluginManager) bootstrapPluginColumnTable(ctx context.Context) error {
 	schemas, err := m.loadPluginSchemas(m.getPluginExemplarConnections())
 	if err != nil {
+		log.Printf("[WARN] loadPluginSchemas error: %s", err.Error())
 		return err
 	}
 
 	if err := m.createPluginColumnsTable(ctx); err != nil {
+		log.Printf("[WARN] createPluginColumnsTable error: %s", err.Error())
 		return err
 	}
 	// now populate the table
+	log.Printf("[INFO] bootstrapPluginColumnTable loaded schema for plugins:  %s", strings.Join(maps.Keys(schemas), ","))
+
 	return m.populatePluginColumnsTable(ctx, schemas)
 }
 
@@ -57,6 +95,11 @@ func (m *PluginManager) createPluginColumnsTable(ctx context.Context) error {
 }
 
 func (m *PluginManager) populatePluginColumnsTable(ctx context.Context, schemas map[string]*proto.Schema) error {
+	if len(schemas) == 0 {
+		log.Printf("[INFO] populatePluginColumnsTable : no updates to plugin_columns table")
+		return nil
+	}
+	log.Printf("[INFO] populating plugin_columns table for plugins %s", strings.Join(maps.Keys(schemas), ","))
 	var queries []db_common.QueryWithArgs
 	for plugin, schema := range schemas {
 		// drop entries for this plugin

pkg/steampipeconfig/connection_updates.go

@@ -154,8 +154,8 @@ func populateConnectionUpdates(ctx context.Context, pool *pgxpool.Pool, pluginMa
 			// we need to refetch the rate limiters for this plugin
 			if res.pluginBinaryChanged {
 				// store map item of plugin name to connection name (so we only have one entry per plugin)
-				pluginShortName := GlobalConfig.Connections[requiredConnectionState.ConnectionName].PluginAlias
-				updates.PluginsWithUpdatedBinary[pluginShortName] = requiredConnectionState.ConnectionName
+				pluginLogName := GlobalConfig.Connections[requiredConnectionState.ConnectionName].Plugin
+				updates.PluginsWithUpdatedBinary[pluginLogName] = requiredConnectionState.ConnectionName
 			}
 		}
 	}