feat: Add better changes summary helper (#2240)

#### Summary

Part of https://linear.app/cloudquery/issue/ENG-1828/feat-improved-forced-migration-needed-error-message.

Open for feedback, I tried to keep only the most interesting bit of each change. There's another improvement to be made and that's to pass only the breaking changes to the summary function (at the moment plugins print all changes which adds additional noise)

## Before

```
Loading spec(s) from examples/gcp_to_postgres.yml
Starting migration for: gcp (cloudquery/[email protected]) -> [postgresql (cloudquery/[email protected])]
Error: failed to migrate v3 source gcp: rpc error: code = Internal desc = write failed: tables gcp_compute_routes,gcp_artifactregistry_docker_images with changes [[column: next_hop_inter_region_cost, type: add, current: next_hop_inter_region_cost:int64 column: next_hop_med, type: add, current: next_hop_med:int64 column: next_hop_origin, type: add, current: next_hop_origin:utf8 column: tags, type: update, current: tags:json, previous: tags:list<item: utf8, nullable>] [column: tags, type: update, current: tags:json, previous: tags:list<item: utf8, nullable>]] require migration. Migrate manually or consider using 'migrate_mode: forced'
```

## After

```
Loading spec(s) from examples/gcp_to_postgres.yml
Starting migration for: gcp (cloudquery/[email protected]) -> [postgresql (grpc@localhost:8888)]
Error: failed to migrate v3 source gcp: rpc error: code = Internal desc = write failed: 
Can't migrate tables automatically, migrate manually or consider using 'migrate_mode: forced'. Non auto migratable tables changes:

gcp_artifactregistry_docker_images:
  - Type changed from "list<item: utf8, nullable>" to "json" for column "tags"

gcp_compute_routes:
  - Column "next_hop_inter_region_cost" added with type "int64"
  - Column "next_hop_med" added with type "int64"
  - Column "next_hop_origin" added with type "utf8"
  - Type changed from "list<item: utf8, nullable>" to "json" for column "tags"
```

---

Author: erezrokah

schema/table.go

@@ -7,9 +7,11 @@ import (
 	"fmt"
 	"regexp"
 	"slices"
+	"strings"
 
 	"github.com/apache/arrow-go/v18/arrow"
 	"github.com/cloudquery/plugin-sdk/v4/glob"
+	"github.com/samber/lo"
 	"github.com/thoas/go-funk"
 )
 
@@ -258,6 +260,100 @@ func (t TableColumnChange) String() string {
 	}
 }
 
+func getColumnChangeSummary(change TableColumnChange) string {
+	switch change.Type {
+	case TableColumnChangeTypeAdd:
+		if change.Current.PrimaryKey {
+			return fmt.Sprintf("Column %q added with type %q and primary key constraint", change.ColumnName, change.Current.Type)
+		}
+		if change.Current.Unique {
+			return fmt.Sprintf("Column %q added with type %q and unique constraint", change.ColumnName, change.Current.Type)
+		}
+		if change.Current.NotNull {
+			return fmt.Sprintf("Column %q added with type %q and not null constraint", change.ColumnName, change.Current.Type)
+		}
+		if change.Current.IncrementalKey {
+			return fmt.Sprintf("Column %q added with type %q and as an incremental key", change.ColumnName, change.Current.Type)
+		}
+		return fmt.Sprintf("Column %q added with type %q", change.ColumnName, change.Current.Type)
+	case TableColumnChangeTypeUpdate:
+		if change.Previous.Type != change.Current.Type {
+			return fmt.Sprintf("Type changed from %q to %q for column %q", change.Previous.Type, change.Current.Type, change.ColumnName)
+		}
+		if !change.Previous.PrimaryKey && change.Current.PrimaryKey {
+			return fmt.Sprintf("Primary key constraint added to column %q", change.ColumnName)
+		}
+		if change.Previous.PrimaryKey && !change.Current.PrimaryKey {
+			return fmt.Sprintf("Primary key constraint removed from column %q", change.ColumnName)
+		}
+		if !change.Previous.Unique && change.Current.Unique {
+			return fmt.Sprintf("Unique constraint added to column %q", change.ColumnName)
+		}
+		if change.Previous.Unique && !change.Current.Unique {
+			return fmt.Sprintf("Unique constraint removed from column %q", change.ColumnName)
+		}
+		if !change.Previous.NotNull && change.Current.NotNull {
+			return fmt.Sprintf("Not null constraint added to column %q", change.ColumnName)
+		}
+		if change.Previous.NotNull && !change.Current.NotNull {
+			return fmt.Sprintf("Not null constraint removed from column %q", change.ColumnName)
+		}
+		if !change.Previous.IncrementalKey && change.Current.IncrementalKey {
+			return fmt.Sprintf("Column %q added as an incremental key", change.ColumnName)
+		}
+		if change.Previous.IncrementalKey && !change.Current.IncrementalKey {
+			return fmt.Sprintf("Column %q removed as an incremental key", change.ColumnName)
+		}
+		if !change.Previous.PrimaryKeyComponent && change.Current.PrimaryKeyComponent {
+			return fmt.Sprintf("Primary key component condition added to column %q", change.ColumnName)
+		}
+		if change.Previous.PrimaryKeyComponent && !change.Current.PrimaryKeyComponent {
+			return fmt.Sprintf("Primary key component condition removed from column %q", change.ColumnName)
+		}
+		return fmt.Sprintf("Column %q updated. Previous: %s, Current: %s", change.ColumnName, change.Previous, change.Current)
+	case TableColumnChangeTypeRemove:
+		if change.Previous.PrimaryKey {
+			return fmt.Sprintf("Primary key column %q removed", change.ColumnName)
+		}
+		if change.Previous.Unique {
+			return fmt.Sprintf("Unique column %q removed", change.ColumnName)
+		}
+		if change.Previous.NotNull {
+			return fmt.Sprintf("Not null column %q removed", change.ColumnName)
+		}
+		if change.Previous.IncrementalKey {
+			return fmt.Sprintf("Incremental key column %q removed", change.ColumnName)
+		}
+		return fmt.Sprintf("Column %q with type %q removed", change.ColumnName, change.Previous.Type)
+	case TableColumnChangeTypeRemoveUniqueConstraint:
+		return fmt.Sprintf("Unique constraint removed from column %q", change.ColumnName)
+	case TableColumnChangeTypeMoveToCQOnly:
+		return fmt.Sprintf("Primary key columns removed and replaced with a single column %q with type %q", change.ColumnName, change.Current.Type)
+	default:
+		return fmt.Sprintf("column: %s, type: %s, current: %s, previous: %s", change.ColumnName, change.Type, change.Current, change.Previous)
+	}
+}
+
+func GetChangesSummary(tablesChanges map[string][]TableColumnChange) string {
+	tables := lo.Keys(tablesChanges)
+	slices.Sort(tables)
+	summary := strings.Builder{}
+	for i, table := range tables {
+		summary.WriteString(fmt.Sprintf("%s:\n", table))
+		changes := tablesChanges[table]
+		changesString := lo.Map(changes, func(change TableColumnChange, _ int) string {
+			return fmt.Sprintf("  - %s", getColumnChangeSummary(change))
+		})
+		slices.Sort(changesString)
+		summary.WriteString(strings.Join(changesString, "\n"))
+		if i < len(tables)-1 {
+			summary.WriteString("\n\n")
+		}
+	}
+
+	return summary.String()
+}
+
 func (tt Tables) FilterDfsFunc(include, exclude func(*Table) bool, skipDependentTables bool) Tables {
 	filteredTables := make(Tables, 0, len(tt))
 	for _, t := range tt {