Refactor docker tests into their own subpackage

This commit refactors out all the tests that rely on a running instance
of the database into their own package end_to_end_tests, and breaks up
migrate_test.go into multiple files, whose names are more descriptive
of their roles. Creating a separate end_to_end_tests package has a
number of advantages:

  1. The other tests run faster as they no longer need to start a
     docker container.

  2. The end-to-end tests are more restricted as to what they can rely
     on, ensuring the correct encapsulation.

  3. It is clearer which tests need docker, and which can run without.

  4. There are a significant number of end-to-end tests, and this
     allows us to keep them together in a single subdirectory.

Performing this refactor found two things that are part of the public
API of pgmodel that were mistakenly made private: metricLabelName and
ErrNoMetricName. The former is defacto public as it is written to the
database, and users can easily query based on it. The latter should be
public as that is the only users can check for that specific error.

.github/workflows/go.yml

@@ -37,7 +37,7 @@ jobs:
       run: go test -race ./...
 
     - name: Test W/O Extension
-      run: go test -race ./pkg/pgmodel -use-extension=false
+      run: go test -race ./pkg/pgmodel/end_to_end_tests/ -use-extension=false
 
     - name: Generated
       run: |

pkg/internal/testhelpers/containers_test.go

@@ -19,8 +19,8 @@ import (
 )
 
 var (
-	useDocker = flag.Bool("use-docker", true, "start database using a docker container")
-	database  = flag.String("database", "tmp_db_timescale_migrate_test", "database to run integration tests on")
+	useDocker    = flag.Bool("use-docker", true, "start database using a docker container")
+	testDatabase = flag.String("database", "tmp_db_timescale_migrate_test", "database to run integration tests on")
 )
 
 func TestPGConnection(t *testing.T) {
@@ -46,7 +46,7 @@ func TestWithDB(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test")
 	}
-	WithDB(t, *database, Superuser, func(db *pgxpool.Pool, t testing.TB, connectURL string) {
+	WithDB(t, *testDatabase, Superuser, func(db *pgxpool.Pool, t testing.TB, connectURL string) {
 		var res int
 		err := db.QueryRow(context.Background(), "SELECT 1").Scan(&res)
 		if err != nil {

pkg/pgmodel/end_to_end_tests/concurrent_sql_test.go

@@ -0,0 +1,110 @@
+// This file and its contents are licensed under the Apache License 2.0.
+// Please see the included NOTICE for copyright information and
+// LICENSE for a copy of the license.
+package end_to_end_tests
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/jackc/pgx/v4/pgxpool"
+
+	_ "github.com/jackc/pgx/v4/stdlib"
+)
+
+func testConcurrentMetricTable(t testing.TB, db *pgxpool.Pool, metricName string) int64 {
+	var id *int64
+	var name *string
+	err := db.QueryRow(context.Background(), "SELECT id, table_name FROM _prom_catalog.create_metric_table($1)", metricName).Scan(&id, &name)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if id == nil || name == nil {
+		t.Fatalf("NULL found")
+	}
+	return *id
+}
+
+func testConcurrentNewLabel(t testing.TB, db *pgxpool.Pool, labelName string) int64 {
+	var id *int64
+	err := db.QueryRow(context.Background(), "SELECT _prom_catalog.get_new_label_id($1, $1)", labelName).Scan(&id)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if id == nil {
+		t.Fatalf("NULL found")
+	}
+	return *id
+}
+
+func testConcurrentCreateSeries(t testing.TB, db *pgxpool.Pool, index int) int64 {
+	var id *int64
+	err := db.QueryRow(context.Background(), "SELECT _prom_catalog.create_series((SELECT id FROM _prom_catalog.metric WHERE metric_name=$3), $1, array[(SELECT id FROM _prom_catalog.label WHERE key = '__name__' AND value=$3), $2::int])",
+		fmt.Sprintf("metric_%d", index), index, fmt.Sprintf("metric_%d", index)).Scan(&id)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if id == nil {
+		t.Fatalf("NULL found")
+	}
+	return *id
+}
+
+func TestConcurrentSQL(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	withDB(t, *testDatabase, func(db *pgxpool.Pool, t testing.TB) {
+		for i := 0; i < 10; i++ {
+			name := fmt.Sprintf("metric_%d", i)
+			var id1, id2 int64
+			var wg sync.WaitGroup
+			wg.Add(2)
+			go func() {
+				defer wg.Done()
+				id1 = testConcurrentMetricTable(t, db, name)
+			}()
+			go func() {
+				defer wg.Done()
+				id2 = testConcurrentMetricTable(t, db, name)
+			}()
+			wg.Wait()
+
+			if id1 != id2 {
+				t.Fatalf("ids aren't equal: %d != %d", id1, id2)
+			}
+
+			wg.Add(2)
+			go func() {
+				defer wg.Done()
+				id1 = testConcurrentNewLabel(t, db, name)
+			}()
+			go func() {
+				defer wg.Done()
+				id2 = testConcurrentNewLabel(t, db, name)
+			}()
+			wg.Wait()
+
+			if id1 != id2 {
+				t.Fatalf("ids aren't equal: %d != %d", id1, id2)
+			}
+
+			wg.Add(2)
+			go func() {
+				defer wg.Done()
+				id1 = testConcurrentCreateSeries(t, db, i)
+			}()
+			go func() {
+				defer wg.Done()
+				id2 = testConcurrentCreateSeries(t, db, i)
+			}()
+			wg.Wait()
+
+			if id1 != id2 {
+				t.Fatalf("ids aren't equal: %d != %d", id1, id2)
+			}
+		}
+	})
+}