Merge pull request #20 from PDOK/create-index-idempotent

feat: make create search index idempotent, it shouldn't fail when it's already there.

internal/etl/etl_test.go

@@ -37,8 +37,7 @@ func TestCreateSearchIndex(t *testing.T) {
 		t.Error(err)
 	}
 	defer terminateContainer(ctx, t, postgisContainer)
-
-	dbConn := fmt.Sprintf("postgres://postgres:[email protected]:%d/%s?sslmode=disable", dbPort.Int(), "test_db")
+	dbConn := makeDbConnection(dbPort)
 
 	// when/then
 	err = CreateSearchIndex(dbConn, "search_index")
@@ -47,6 +46,31 @@ func TestCreateSearchIndex(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func TestCreateSearchIndexIdempotent(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+	ctx := context.Background()
+
+	// given
+	dbPort, postgisContainer, err := setupPostgis(ctx, t)
+	if err != nil {
+		t.Error(err)
+	}
+	defer terminateContainer(ctx, t, postgisContainer)
+	dbConn := makeDbConnection(dbPort)
+
+	// when/then
+	err = CreateSearchIndex(dbConn, "search_index")
+	assert.NoError(t, err)
+	err = CreateSearchIndex(dbConn, "search_index") // second time, should not fail
+	assert.NoError(t, err)
+}
+
+func makeDbConnection(dbPort nat.Port) string {
+	return fmt.Sprintf("postgres://postgres:[email protected]:%d/%s?sslmode=disable", dbPort.Int(), "test_db")
+}
+
 func TestImportGeoPackage(t *testing.T) {
 	if testing.Short() {
 		t.Skip("Skipping integration test in short mode")
@@ -75,9 +99,7 @@ func TestImportGeoPackage(t *testing.T) {
 		if err != nil {
 			t.Error(err)
 		}
-		defer terminateContainer(ctx, t, postgisContainer)
-
-		dbConn := fmt.Sprintf("postgres://postgres:[email protected]:%d/%s?sslmode=disable", dbPort.Int(), "test_db")
+		dbConn := makeDbConnection(dbPort)
 
 		cfg, err := config.NewConfig(pwd + "/testdata/config.yaml")
 		if err != nil {
@@ -105,9 +127,11 @@ func TestImportGeoPackage(t *testing.T) {
 		assert.NoError(t, err)
 		var count int
 		err = db.QueryRow(ctx, "select count(*) from search_index").Scan(&count)
-		defer db.Close(ctx)
+		db.Close(ctx)
 		assert.NoError(t, err)
 		assert.Equal(t, tt.count, count)
+
+		terminateContainer(ctx, t, postgisContainer)
 	}
 }
 

internal/etl/load/postgres.go

@@ -49,7 +49,14 @@ func (p *Postgres) Load(records []t.SearchIndexRecord, index string) (int64, err
 
 // Init initialize search index
 func (p *Postgres) Init(index string) error {
-	geometryType := `create type geometry_type as enum ('POINT', 'MULTIPOINT', 'LINESTRING', 'MULTILINESTRING', 'POLYGON', 'MULTIPOLYGON');`
+	// since "create type if not exists" isn't supported by Postgres we use a bit
+	// of pl/pgsql to avoid creating the geometry_type when it already exists.
+	geometryType := `
+		do $$ begin
+		    create type geometry_type as enum ('POINT', 'MULTIPOINT', 'LINESTRING', 'MULTILINESTRING', 'POLYGON', 'MULTIPOLYGON');
+		exception
+		    when duplicate_object then null;
+		end $$;`
 	_, err := p.db.Exec(p.ctx, geometryType)
 	if err != nil {
 		return fmt.Errorf("error creating geometry type: %w", err)
@@ -73,14 +80,14 @@ func (p *Postgres) Init(index string) error {
 	}
 
 	fullTextSearchColumn := fmt.Sprintf(`
-		alter table %[1]s add column ts tsvector 
+		alter table %[1]s add column if not exists ts tsvector 
 	    generated always as (to_tsvector('dutch', suggest || display_name )) stored;`, index)
 	_, err = p.db.Exec(p.ctx, fullTextSearchColumn)
 	if err != nil {
 		return fmt.Errorf("error creating full-text search column: %w", err)
 	}
 
-	ginIndex := fmt.Sprintf(`create index ts_idx on  %[1]s using gin(ts);`, index)
+	ginIndex := fmt.Sprintf(`create index if not exists ts_idx on  %[1]s using gin(ts);`, index)
 	_, err = p.db.Exec(p.ctx, ginIndex)
 	if err != nil {
 		return fmt.Errorf("error creating GIN index: %w", err)