fix: Better error message

erezrokah · erezrokah · commit 371be8e45337 · 2023-01-04T17:34:36.000+02:00
diff --git a/plugins/destination/postgresql/client/client_test.go b/plugins/destination/postgresql/client/client_test.go
@@ -60,5 +60,12 @@ func TestPgPluginPrimaryKeyRename(t *testing.T) {
 	// Migrate the table again without the `stale_pk_1` and `stale_pk_2` columns
 	table := testdata.TestTable(tableName)
 	err := p.Migrate(ctx, []*schema.Table{table})
-	require.ErrorContains(t, err, fmt.Sprintf("the following primary keys were removed from the schema [\"stale_pk_1\" \"stale_pk_2\"] for table \"%s\"", tableName))
+	expected := `the following primary keys were removed from the schema ["stale_pk_1" "stale_pk_2"] for table "%s".
+You can migrate the table manually by running:
+-----------------------------------------------------------------------------------------
+alter table "%s" drop constraint if exists "%s_cqpk";
+alter table "%s" alter column "stale_pk_1" drop not null;
+alter table "%s" alter column "stale_pk_2" drop not null;
+-----------------------------------------------------------------------------------------`
+	require.ErrorContains(t, err, fmt.Sprintf(expected, tableName, tableName, tableName, tableName, tableName))
 }
diff --git a/plugins/destination/postgresql/client/migrate.go b/plugins/destination/postgresql/client/migrate.go
@@ -88,6 +88,7 @@ func (c *Client) autoMigrateTable(ctx context.Context, table *schema.Table) erro
 
 	// create the new column as it doesn't exist
 	tableName := pgx.Identifier{table.Name}.Sanitize()
+
 	if pgColumns, err = c.getPgTableColumns(ctx, table.Name); err != nil {
 		return fmt.Errorf("failed to get table %s columns types: %w", table.Name, err)
 	}
@@ -97,9 +98,13 @@ func (c *Client) autoMigrateTable(ctx context.Context, table *schema.Table) erro
 	}
 
 	stalePks := c.getStalePks(pgPKs, table)
+	constraintName := pgx.Identifier{table.Name + "_cqpk"}.Sanitize()
+	dropConstraintSQL := "alter table " + tableName + " drop constraint if exists " + constraintName
 	if len(stalePks) > 0 {
-		message := "the following primary keys were removed from the schema %q for table %q.\nEither drop the columns or remove their \"not null\" constraint"
-		return fmt.Errorf(message, stalePks, table.Name)
+		message := "the following primary keys were removed from the schema %q for table %q.\nYou can migrate the table manually by running:\n%s"
+		sep := strings.Repeat("-", len(dropConstraintSQL)+1)
+		query := fmt.Sprintf("%s\n%s;\n%s\n%s", sep, dropConstraintSQL, getDropNotNullQuery(table, stalePks), sep)
+		return fmt.Errorf(message, stalePks, table.Name, query)
 	}
 
 	reCreatePrimaryKeys := false
@@ -160,16 +165,14 @@ func (c *Client) autoMigrateTable(ctx context.Context, table *schema.Table) erro
 		if err != nil {
 			return fmt.Errorf("failed to begin transaction to recreate primary keys: %w", err)
 		}
-		constraintName := pgx.Identifier{table.Name + "_cqpk"}.Sanitize()
-		sql := "alter table " + tableName + " drop constraint if exists " + constraintName
-		if _, err := tx.Exec(ctx, sql); err != nil {
+		if _, err := tx.Exec(ctx, dropConstraintSQL); err != nil {
 			if err := tx.Rollback(ctx); err != nil {
 				c.logger.Error().Err(err).Msg("failed to rollback transaction")
 			}
 			return fmt.Errorf("failed to drop primary key constraint on table %s: %w", table.Name, err)
 		}
 
-		sql = "alter table " + tableName + " add constraint " + constraintName + " primary key (" + strings.Join(table.PrimaryKeys(), ",") + ")"
+		sql := "alter table " + tableName + " add constraint " + constraintName + " primary key (" + strings.Join(table.PrimaryKeys(), ",") + ")"
 		if _, err := tx.Exec(ctx, sql); err != nil {
 			if err := tx.Rollback(ctx); err != nil {
 				c.logger.Error().Err(err).Msg("failed to rollback transaction")
@@ -194,6 +197,15 @@ func (c *Client) setNotNullOnPks(ctx context.Context, table *schema.Table) error
 	return nil
 }
 
+func getDropNotNullQuery(table *schema.Table, stalePks []string) string {
+	queries := []string{}
+	for _, col := range stalePks {
+		queries = append(queries, "alter table "+pgx.Identifier{table.Name}.Sanitize()+" alter column "+pgx.Identifier{col}.Sanitize()+" drop not null;")
+	}
+
+	return strings.Join(queries, "\n")
+}
+
 func (c *Client) createTableIfNotExist(ctx context.Context, table *schema.Table) error {
 	var sb strings.Builder
 	tableName := pgx.Identifier{table.Name}.Sanitize()
