Ignore unique constraint violation during CREATE IF NOT EXISTS (#905)

https://stackoverflow.com/questions/29900845/create-schema-if-not-exists-raises-duplicate-key-error Ran into this while looking to make all tests run in parallel
PeerDB-io · Dec 26, 2023 · 2f18f76 · 2f18f76
1 parent 67ab274
commit 2f18f76
Show file tree

Hide file tree

Showing 4 changed files with 14 additions and 3 deletions.
diff --git a/flow/connectors/external_metadata/store.go b/flow/connectors/external_metadata/store.go
@@ -2,13 +2,16 @@ package connmetadata
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 
 	"github.com/PeerDB-io/peer-flow/connectors/utils"
 	cc "github.com/PeerDB-io/peer-flow/connectors/utils/catalog"
 	"github.com/PeerDB-io/peer-flow/generated/protos"
 	"github.com/PeerDB-io/peer-flow/shared"
+	"github.com/jackc/pgerrcode"
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/jackc/pgx/v5/pgxpool"
 )
@@ -17,6 +20,11 @@ const (
 	lastSyncStateTableName = "last_sync_state"
 )
 
+func isUniqueError(err error) bool {
+	var pgerr *pgconn.PgError
+	return errors.As(err, &pgerr) && pgerr.Code == pgerrcode.UniqueViolation
+}
+
 type PostgresMetadataStore struct {
 	ctx        context.Context
 	config     *protos.PostgresConfig
@@ -106,7 +114,7 @@ func (p *PostgresMetadataStore) SetupMetadata() error {
 
 	// create the schema
 	_, err = tx.Exec(p.ctx, "CREATE SCHEMA IF NOT EXISTS "+p.schemaName)
-	if err != nil {
+	if err != nil && !isUniqueError(err) {
 		p.logger.Error("failed to create schema", slog.Any("error", err))
 		return err
 	}
@@ -120,7 +128,7 @@ func (p *PostgresMetadataStore) SetupMetadata() error {
 			sync_batch_id BIGINT NOT NULL
 		)
 	`)
-	if err != nil {
+	if err != nil && !isUniqueError(err) {
 		p.logger.Error("failed to create last sync state table", slog.Any("error", err))
 		return err
 	}

diff --git a/flow/go.mod b/flow/go.mod
@@ -15,6 +15,7 @@ require (
 	github.com/google/uuid v1.5.0
 	github.com/grafana/pyroscope-go v1.0.4
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.18.1
+	github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa
 	github.com/jackc/pglogrepl v0.0.0-20231111135425-1627ab1b5780
 	github.com/jackc/pgx/v5 v5.5.1
 	github.com/jmoiron/sqlx v1.3.5

diff --git a/flow/go.sum b/flow/go.sum
@@ -255,6 +255,8 @@ github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c h1:6rhixN/i8
 github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c/go.mod h1:NMPJylDgVpX0MLRlPy15sqSwOFv/U1GZ2m21JhFfek0=
 github.com/hexops/gotextdiff v1.0.3 h1:gitA9+qJrrTCsiCl7+kh75nPqQt1cx4ZkudSTLoUqJM=
 github.com/hexops/gotextdiff v1.0.3/go.mod h1:pSWU5MAI3yDq+fZBTazCSJysOMbxWL1BSow5/V2vxeg=
+github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa h1:s+4MhCQ6YrzisK6hFJUX53drDT4UsSW3DEhKn0ifuHw=
+github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds=
 github.com/jackc/pgio v1.0.0 h1:g12B9UwVnzGhueNavwioyEEpAmqMe1E/BN9ES+8ovkE=
 github.com/jackc/pgio v1.0.0/go.mod h1:oP+2QK2wFfUWgr+gxjoBH9KGBb31Eio69xUb0w5bYf8=
 github.com/jackc/pglogrepl v0.0.0-20231111135425-1627ab1b5780 h1:pNK2AKKIRC1MMMvpa6UiNtdtOebpiIloX7q2JZDkfsk=

diff --git a/flow/workflows/setup_flow.go b/flow/workflows/setup_flow.go
@@ -273,7 +273,7 @@ func (s *SetupFlowExecution) executeSetupFlow(
 func SetupFlowWorkflow(ctx workflow.Context,
 	config *protos.FlowConnectionConfigs,
 ) (*protos.FlowConnectionConfigs, error) {
-	tblNameMapping := make(map[string]string)
+	tblNameMapping := make(map[string]string, len(config.TableMappings))
 	for _, v := range config.TableMappings {
 		tblNameMapping[v.SourceTableIdentifier] = v.DestinationTableIdentifier
 	}