dialect/sql/schema: allow running migration concurrently without copying the tables (#4475)

dialect/sql/schema/atlas.go

@@ -1078,6 +1078,7 @@ func (a *Atlas) aIndexes(et *Table, at *schema.Table) error {
 // are linked to their indexes, and PKs columns are defined.
 func (a *Atlas) setupTables(tables []*Table) {
 	for _, t := range tables {
+		t.mu.Lock()
 		if t.columns == nil {
 			t.columns = make(map[string]*Column, len(t.Columns))
 		}
@@ -1101,6 +1102,7 @@ func (a *Atlas) setupTables(tables []*Table) {
 				fk.Columns[i].foreign = fk
 			}
 		}
+		t.mu.Unlock()
 	}
 }
 

dialect/sql/schema/migrate_test.go

@@ -11,6 +11,7 @@ import (
 	"path/filepath"
 	"regexp"
 	"strings"
+	"sync"
 	"testing"
 	"text/template"
 	"time"
@@ -444,3 +445,20 @@ func TestAtlas_StateReader(t *testing.T) {
 		},
 	)
 }
+
+func TestAtlas_ParallelCreate(t *testing.T) {
+	var wg sync.WaitGroup
+	wg.Add(10)
+	for i := 0; i < 10; i++ {
+		db, err := sql.Open(dialect.SQLite, fmt.Sprintf("file:test-%d?mode=memory&_fk=1", i))
+		require.NoError(t, err)
+		m, err := NewMigrate(db)
+		require.NoError(t, err)
+		go func() {
+			defer wg.Done()
+			require.NoError(t, m.Create(context.Background(), petsTable))
+			require.NoError(t, db.Close())
+		}()
+	}
+	wg.Wait()
+}

dialect/sql/schema/schema.go

@@ -11,6 +11,7 @@ import (
 	"slices"
 	"strconv"
 	"strings"
+	"sync"
 
 	"ariga.io/atlas/sql/migrate"
 	"ariga.io/atlas/sql/mysql"
@@ -36,6 +37,7 @@ const (
 
 // Table schema definition for SQL dialects.
 type Table struct {
+	mu          sync.Mutex
 	Name        string
 	Schema      string
 	Columns     []*Column