fix: Bring back plugin validation (#1108)

Fixes #1106

Plugin validation was removed in https://github.com/cloudquery/plugin-sdk/pull/984/files#diff-95f04007d983a3bfdb080b338b9de5a8bbd73a3d65910fbe664f97788a9021b3.

As a result we don't validate tables and columns anymore, see resulted bug in cloudquery/cloudquery#12428.

This PR brings it back and per a better option puts the validation in the `Init` method. Other options:
1. Have each plugin call `.validate` on the plugin side, probably after transformation [here](https://github.com/cloudquery/cloudquery/blob/03fae6d803a4e2ac00ff83d9e90a4169b247c346/plugins/source/aws/resources/plugin/tables.go#L565). Downside that each plugin needs to implement the call
2. Have the validation done as a part of `schema.TransformTables` [here](https://github.com/cloudquery/plugin-sdk/blob/8a6a2cc87e303d66c810e629a497f9acd3c4982c/transformers/tables.go#L30). Downside it will happen before any custom transformation that are executed on tables like [`titleTransformer`](https://github.com/cloudquery/cloudquery/blob/03fae6d803a4e2ac00ff83d9e90a4169b247c346/plugins/source/aws/resources/plugin/tables.go#L567)

> Please note that before it was removed `validate` was called during `NewPlugin` which means it errored right after calling the `serve` command, meaning the plugin would not start. With this fix the plugin will error out during `Init`, which means you need to run `sync/migrate` to see the error

---

Author: erezrokah

plugin/plugin.go

@@ -150,6 +150,11 @@ func (p *Plugin) Init(ctx context.Context, spec []byte, options NewClientOptions
 	if err != nil {
 		return fmt.Errorf("failed to initialize client: %w", err)
 	}
+
+	if err := p.validate(ctx); err != nil {
+		return fmt.Errorf("failed to validate tables: %w", err)
+	}
+
 	p.spec = spec
 
 	return nil

plugin/validate.go

@@ -0,0 +1,39 @@
+package plugin
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/cloudquery/plugin-sdk/v4/schema"
+)
+
+func validateTables(tables schema.Tables) error {
+	if err := tables.ValidateDuplicateTables(); err != nil {
+		return fmt.Errorf("found duplicate tables in plugin: %w", err)
+	}
+
+	if err := tables.ValidateTableNames(); err != nil {
+		return fmt.Errorf("found table with invalid name in plugin: %w", err)
+	}
+
+	if err := tables.ValidateDuplicateColumns(); err != nil {
+		return fmt.Errorf("found duplicate columns in plugin: %w", err)
+	}
+
+	if err := tables.ValidateColumnNames(); err != nil {
+		return fmt.Errorf("found column with invalid name in plugin: %w", err)
+	}
+
+	return nil
+}
+
+func (p *Plugin) validate(ctx context.Context) error {
+	tables, err := p.client.Tables(ctx, TableOptions{Tables: []string{"*"}})
+	// ErrNotImplemented means it's a destination only plugin
+	if err != nil && !errors.Is(err, ErrNotImplemented) {
+		return fmt.Errorf("failed to get tables: %w", err)
+	}
+
+	return validateTables(tables)
+}

schema/table.go

@@ -342,12 +342,13 @@ func (tt Tables) ValidateDuplicateColumns() error {
 }
 
 func (tt Tables) ValidateDuplicateTables() error {
+	tableNames := tt.TableNames()
 	tables := make(map[string]bool, len(tt))
-	for _, t := range tt {
-		if _, ok := tables[t.Name]; ok {
-			return fmt.Errorf("duplicate table %s", t.Name)
+	for _, t := range tableNames {
+		if _, ok := tables[t]; ok {
+			return fmt.Errorf("duplicate table %s", t)
 		}
-		tables[t.Name] = true
+		tables[t] = true
 	}
 	return nil
 }
@@ -405,7 +406,7 @@ func (t *Table) ValidateName() error {
 	if !ok {
 		return fmt.Errorf("table name %q is not valid: table names must contain only lower-case letters, numbers and underscores, and must start with a lower-case letter or underscore", t.Name)
 	}
-	return nil
+	return ValidateTable(t)
 }
 
 func (t *Table) PrimaryKeysIndexes() []int {

schema/table_test.go

@@ -390,3 +390,33 @@ func TestTableToAndFromArrow(t *testing.T) {
 		t.Errorf("diff (+got, -want): %v", diff)
 	}
 }
+
+func TestValidateDuplicateTables(t *testing.T) {
+	tests := []struct {
+		name   string
+		tables Tables
+		err    string
+	}{
+		{
+			name:   "should return error when duplicate tables are found",
+			tables: Tables{{Name: "table1"}, {Name: "table1"}},
+			err:    "duplicate table table1",
+		},
+		{
+			name:   "should return error when duplicate relational tables are found",
+			tables: Tables{{Name: "table1", Relations: []*Table{{Name: "table2"}, {Name: "table2"}}}},
+			err:    "duplicate table table2",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			err := tc.tables.ValidateDuplicateTables()
+			if tc.err != "" {
+				require.ErrorContains(t, err, tc.err)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}