PostgreSQL: SQL injection and nil transaction rollback in migrate.go

Bug 1: SQL injection in dropTable and addColumn

File: plugins/destination/postgresql/client/migrate.go:236, 341

Problem: Table names are interpolated directly into SQL without sanitization. An attacker-controlled table name (e.g., from a source plugin) can inject arbitrary SQL.

Current code:

func (c *Client) dropTable(ctx context.Context, tableName string) error {
    // ...
    sql := "drop table " + tableName  // SQL injection
    if _, err := c.conn.Exec(ctx, sql); err != nil { ... }
}

func (c *Client) addColumn(ctx context.Context, tableName string, column schema.Column) error {
    // ...
    sql := "alter table " + tableName + " add column " + columnName + " " + columnType  // SQL injection
    if _, err := c.conn.Exec(ctx, sql); err != nil { ... }
}

Fix: Use pgx.Identifier{}.Sanitize():

sql := "drop table " + pgx.Identifier{tableName}.Sanitize()
sql := "alter table " + pgx.Identifier{tableName}.Sanitize() + " add column " + columnName + " " + columnType

Bug 2: Nil transaction rollback in migrateToCQID

File: plugins/destination/postgresql/client/migrate.go:267

Problem: The deferred tx.Rollback(ctx) call will panic if BeginTx fails and tx is nil. This can happen in the CockroachDB path where BeginTx is called conditionally.

Current code:

defer func() {
    if err == nil {
        err = tx.Commit(ctx)
        // ...
    }
    // if tx is nil (BeginTx failed), tx.Rollback will panic
    tx.Rollback(ctx)
}()

Fix: Add nil guard:

defer func() {
    if tx == nil {
        return
    }
    // ...
}()

Reproduction

1. SQL injection: Use a source plugin that syncs to a table named "my_table; DELETE FROM public.some_table; --" — the delete will execute.
2. Nil tx rollback: Use CockroachDB destination and trigger migrateToCQID in a scenario where BeginTx fails — the deferred rollback panics instead of propagating the error.