adding IF NOT EXISTS for pg and bq

PeerDB-io · Dec 12, 2023 · 439b176 · 439b176
1 parent bb49389
commit 439b176
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 2 deletions.
diff --git a/flow/connectors/bigquery/bigquery.go b/flow/connectors/bigquery/bigquery.go
@@ -237,7 +237,8 @@ func (c *BigQueryConnector) ReplayTableSchemaDeltas(flowJobName string,
 		}
 
 		for _, addedColumn := range schemaDelta.AddedColumns {
-			_, err := c.client.Query(fmt.Sprintf("ALTER TABLE %s.%s ADD COLUMN `%s` %s", c.datasetID,
+			_, err := c.client.Query(fmt.Sprintf(
+				"ALTER TABLE %s.%s ADD COLUMN IF NOT EXISTS `%s` %s", c.datasetID,
 				schemaDelta.DstTableName, addedColumn.ColumnName,
 				qValueKindToBigQueryType(addedColumn.ColumnType))).Read(c.ctx)
 			if err != nil {

diff --git a/flow/connectors/postgres/postgres.go b/flow/connectors/postgres/postgres.go
@@ -712,7 +712,8 @@ func (c *PostgresConnector) ReplayTableSchemaDeltas(flowJobName string,
 		}
 
 		for _, addedColumn := range schemaDelta.AddedColumns {
-			_, err = tableSchemaModifyTx.Exec(c.ctx, fmt.Sprintf("ALTER TABLE %s ADD COLUMN \"%s\" %s",
+			_, err = tableSchemaModifyTx.Exec(c.ctx, fmt.Sprintf(
+				"ALTER TABLE %s ADD COLUMN IF NOT EXISTS \"%s\" %s",
 				schemaDelta.DstTableName, addedColumn.ColumnName,
 				qValueKindToPostgresType(addedColumn.ColumnType)))
 			if err != nil {