Repository: xataio/pgstream
Pull request #401 (2025-06-10 13:14:13)

[Snapshot] Support multi-schema snapshots
This PR updates the snapshot interface to support providing multiple schemas per snapshot. This will allow to perform the pg_dump of different schemas together to prevent missing dependencies, since it currently performs separate dumps that will fail if they are dependent on each other. 

This fixes https://github.com/xataio/xata-actions-demo/actions/runs/15486471181/job/43602182511.

The main changes are the switch from handling a single schema/tables to handling multiple. Most of the logic remains the same. Main changes:

- The wal snapshot adapter logic is moved to the data snapshot generator, since it's the one that will now parallelise the schema processing
- The pgdump/pgrestore schema snapshot generator is updated to provide the options for all schemas and their tables
- The snapshot requests remain at the schema level to maintain readability
- The snapshot errors are updated to register per schema errors

It's a pretty large PR, since the `Snapshot` type has been modified, so the commits are split to update each of the different dependent implementations separately. 


Patch:
diff --git a/cmd/config/config_env.go b/cmd/config/config_env.go
index 763fa66..2ad7600 100644
--- a/cmd/config/config_env.go
+++ b/cmd/config/config_env.go
@@ -196,14 +196,14 @@ func parseSnapshotListenerConfig() *snapshotbuilder.SnapshotListenerConfig {
 func parseSnapshotConfig(pgURL string) *snapshotbuilder.SnapshotListenerConfig {
 	cfg := &snapshotbuilder.SnapshotListenerConfig{
 		Generator: pgsnapshotgenerator.Config{
-			URL:           pgURL,
-			BatchPageSize: viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_BATCH_PAGE_SIZE"),
-			SchemaWorkers: viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_SCHEMA_WORKERS"),
-			TableWorkers:  viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_TABLE_WORKERS"),
+			URL:             pgURL,
+			BatchPageSize:   viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_BATCH_PAGE_SIZE"),
+			SchemaWorkers:   viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_SCHEMA_WORKERS"),
+			TableWorkers:    viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_TABLE_WORKERS"),
+			SnapshotWorkers: viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_WORKERS"),
 		},
 		Adapter: adapter.SnapshotConfig{
-			Tables:          viper.GetStringSlice("PGSTREAM_POSTGRES_SNAPSHOT_TABLES"),
-			SnapshotWorkers: viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_WORKERS"),
+			Tables: viper.GetStringSlice("PGSTREAM_POSTGRES_SNAPSHOT_TABLES"),
 		},
 		Schema: parseSchemaSnapshotConfig(pgURL),
 	}
diff --git a/cmd/config/config_yaml.go b/cmd/config/config_yaml.go
index f34f068..d1f890e 100644
--- a/cmd/config/config_yaml.go
+++ b/cmd/config/config_yaml.go
@@ -433,8 +433,7 @@ func (c *YAMLConfig) parseSnapshotConfig() (*snapshotbuilder.SnapshotListenerCon
 
 	streamCfg := &snapshotbuilder.SnapshotListenerConfig{
 		Adapter: adapter.SnapshotConfig{
-			Tables:          snapshotConfig.Tables,
-			SnapshotWorkers: uint(snapshotConfig.SnapshotWorkers),
+			Tables: snapshotConfig.Tables,
 		},
 	}
 
@@ -456,10 +455,11 @@ func (c *YAMLConfig) parseSnapshotConfig() (*snapshotbuilder.SnapshotListenerCon
 
 	if snapshotConfig.Mode == fullSnapshotMode || snapshotConfig.Mode == dataSnapshotMode {
 		streamCfg.Generator = pgsnapshotgenerator.Config{
-			URL:           c.Source.Postgres.URL,
-			BatchPageSize: uint(snapshotConfig.Data.BatchPageSize),
-			SchemaWorkers: uint(snapshotConfig.Data.SchemaWorkers),
-			TableWorkers:  uint(snapshotConfig.Data.TableWorkers),
+			URL:             c.Source.Postgres.URL,
+			BatchPageSize:   uint(snapshotConfig.Data.BatchPageSize),
+			SchemaWorkers:   uint(snapshotConfig.Data.SchemaWorkers),
+			TableWorkers:    uint(snapshotConfig.Data.TableWorkers),
+			SnapshotWorkers: uint(snapshotConfig.SnapshotWorkers),
 		}
 	}
 
diff --git a/pkg/snapshot/errors.go b/pkg/snapshot/errors.go
index af00add..c452763 100644
--- a/pkg/snapshot/errors.go
+++ b/pkg/snapshot/errors.go
@@ -7,98 +7,136 @@ import (
 	"strings"
 )
 
-type Errors struct {
-	SnapshotErrMsgs []string     `json:"snapshot,omitempty"`
-	Tables          []TableError `json:"tables,omitempty"`
+type (
+	Errors       map[string]*SchemaErrors
+	SchemaErrors struct {
+		Schema       string            `json:"schema,omitempty"`
+		GlobalErrors []string          `json:"snapshot,omitempty"`
+		TableErrors  map[string]string `json:"tables,omitempty"`
+	}
+)
+
+func NewErrors(schema string, err error) Errors {
+	if err == nil {
+		return nil
+	}
+	errs := make(Errors)
+	if errors.As(err, &errs) {
+		return errs
+	}
+	return Errors{
+		schema: NewSchemaErrors(schema, err),
+	}
 }
 
-type TableError struct {
-	Table    string `json:"table"`
-	ErrorMsg string `json:"error"`
+func (e Errors) GetSchemaErrors(schema string) *SchemaErrors {
+	if e == nil {
+		return nil
+	}
+
+	if errs, found := e[schema]; found {
+		return errs
+	}
+	return nil
 }
 
-func NewErrors(err error) *Errors {
+func (e Errors) AddError(schema string, err error) {
+	if e == nil {
+		e = make(Errors)
+	}
+
+	schemaErrs, found := e[schema]
+	if found {
+		schemaErrs.AddGlobalError(err)
+		return
+	}
+	e[schema] = NewSchemaErrors(schema, err)
+}
+
+func (e Errors) Error() string {
+	var errMsg strings.Builder
+	for _, errs := range e {
+		errMsg.WriteString(errs.Error())
+	}
+	return errMsg.String()
+}
+
+func NewSchemaErrors(schema string, err error) *SchemaErrors {
 	if err == nil {
 		return nil
 	}
-	var snapshotErrs *Errors
-	if errors.As(err, &snapshotErrs) {
-		return snapshotErrs
+	var schemaErrs *SchemaErrors
+	if errors.As(err, &schemaErrs) {
+		return schemaErrs
 	}
-	return &Errors{
-		SnapshotErrMsgs: []string{err.Error()},
+	return &SchemaErrors{
+		Schema:       schema,
+		GlobalErrors: []string{err.Error()},
 	}
 }
 
-func (e *Errors) Error() string {
+func (e *SchemaErrors) Error() string {
 	if e == nil {
 		return ""
 	}
-	errMsgs := make([]string, 0, len(e.SnapshotErrMsgs))
-	errMsgs = append(errMsgs, e.SnapshotErrMsgs...)
+	errMsgs := make([]string, 0, len(e.GlobalErrors))
+	errMsgs = append(errMsgs, e.GlobalErrors...)
 
-	for _, table := range e.Tables {
-		errMsgs = append(errMsgs, table.Error())
+	for table, errMsg := range e.TableErrors {
+		errMsgs = append(errMsgs, table+": "+errMsg)
 	}
-	return strings.Join(errMsgs, ";")
+	return e.Schema + ": " + strings.Join(errMsgs, ";")
 }
 
-func (e *Errors) AddSnapshotError(err error) {
+func (e *SchemaErrors) AddGlobalError(err error) {
 	if e == nil {
 		return
 	}
-	if e.SnapshotErrMsgs == nil {
-		e.SnapshotErrMsgs = []string{err.Error()}
+
+	e.GlobalErrors = append(e.GlobalErrors, err.Error())
+}
+
+func (e *SchemaErrors) AddTableError(table string, err error) {
+	if e == nil || err == nil {
 		return
 	}
-	e.SnapshotErrMsgs = append(e.SnapshotErrMsgs, err.Error())
+
+	if e.TableErrors == nil {
+		e.TableErrors = make(map[string]string)
+	}
+
+	e.TableErrors[table] = err.Error()
 }
 
-func (e *Errors) IsSnapshotError() bool {
-	return e != nil && len(e.SnapshotErrMsgs) > 0
+func (e *SchemaErrors) IsGlobalError() bool {
+	return e != nil && len(e.GlobalErrors) > 0
 }
 
-func (e *Errors) IsTableError(table string) bool {
+func (e *SchemaErrors) IsTableError(table string) bool {
 	if e == nil {
 		return false
 	}
 
-	if e.SnapshotErrMsgs != nil {
+	if len(e.GlobalErrors) > 0 {
 		return true
 	}
 
 	// treat wildcard table errors as true if there are any table errors.
-	if table == "*" && len(e.Tables) > 0 {
+	if table == "*" && len(e.TableErrors) > 0 {
 		return true
 	}
 
-	for _, tableErr := range e.Tables {
-		if tableErr.Table == table {
-			return true
-		}
-	}
-	return false
+	_, found := e.TableErrors[table]
+	return found
 }
 
-func (e *Errors) GetFailedTables() []string {
+func (e *SchemaErrors) GetFailedTables() []string {
 	if e == nil {
 		return nil
 	}
-	failedTables := make([]string, 0, len(e.Tables))
-	for _, table := range e.Tables {
-		failedTables = append(failedTables, table.Table)
+	failedTables := make([]string, 0, len(e.TableErrors))
+	for table := range e.TableErrors {
+		failedTables = append(failedTables, table)
 	}
 	return failedTables
 }
-
-func NewTableError(table string, err error) TableError {
-	errMsg := ""
-	if err != nil {
-		errMsg = err.Error()
-	}
-	return TableError{Table: table, ErrorMsg: errMsg}
-}
-
-func (e TableError) Error() string {
-	return e.Table + ": " + e.ErrorMsg
-}
diff --git a/pkg/snapshot/generator/instrumentation/instrumented_snapshot_generator.go b/pkg/snapshot/generator/instrumentation/instrumented_snapshot_generator.go
index da8a38b..4508af2 100644
--- a/pkg/snapshot/generator/instrumentation/instrumented_snapshot_generator.go
+++ b/pkg/snapshot/generator/instrumentation/instrumented_snapshot_generator.go
@@ -87,11 +87,11 @@ func (i *SnapshotGenerator) snapshotAttributes(s *snapshot.Snapshot) []attribute
 	return []attribute.KeyValue{
 		{
 			Key:   snapshotSchemaAttributeKey,
-			Value: attribute.StringValue(s.SchemaName),
+			Value: attribute.StringSliceValue(s.GetSchemas()),
 		},
 		{
 			Key:   snapshotTablesAttributeKey,
-			Value: attribute.StringSliceValue(s.TableNames),
+			Value: attribute.StringSliceValue(s.GetTables()),
 		},
 	}
 }
diff --git a/pkg/snapshot/generator/postgres/data/config.go b/pkg/snapshot/generator/postgres/data/config.go
index 2881627..9bb797e 100644
--- a/pkg/snapshot/generator/postgres/data/config.go
+++ b/pkg/snapshot/generator/postgres/data/config.go
@@ -8,6 +8,10 @@ type Config struct {
 	// BatchPageSize represents the size of the table page range that will be
 	// processed concurrently by the table workers. Defaults to 1000.
 	BatchPageSize uint
+	// SnapshotWorkers represents the number of snapshots the generator will
+	// process concurrently. This doesn't affect the parallelism of the tables
+	// within each individual snapshot request. It defaults to 1.
+	SnapshotWorkers uint
 	// SchemaWorkers represents the number of tables the snapshot generator will
 	// process concurrently per schema. Defaults to 4.
 	SchemaWorkers uint
@@ -17,9 +21,10 @@ type Config struct {
 }
 
 const (
-	defaultBatchPageSize = 1000
-	defaultTableWorkers  = 4
-	defaultSchemaWorkers = 4
+	defaultBatchPageSize   = 1000
+	defaultTableWorkers    = 4
+	defaultSchemaWorkers   = 4
+	defaultSnapshotWorkers = 1
 )
 
 func (c *Config) batchPageSize() uint {
@@ -42,3 +47,10 @@ func (c *Config) tableWorkers() uint {
 	}
 	return defaultTableWorkers
 }
+
+func (c *Config) snapshotWorkers() uint {
+	if c.SnapshotWorkers > 0 {
+		return c.SnapshotWorkers
+	}
+	return defaultSnapshotWorkers
+}
diff --git a/pkg/snapshot/generator/postgres/data/pg_snapshot_generator.go b/pkg/snapshot/generator/postgres/data/pg_snapshot_generator.go
index eafe73e..90397ea 100644
--- a/pkg/snapshot/generator/postgres/data/pg_snapshot_generator.go
+++ b/pkg/snapshot/generator/postgres/data/pg_snapshot_generator.go
@@ -22,7 +22,11 @@ type SnapshotGenerator struct {
 	conn   pglib.Querier
 	mapper mapper
 
+	// workers per snapshot, parallelise the snapshot creation for each schema
+	snapshotWorkers uint
+	// workers per schema, parallelise the snapshot creation for each table
 	schemaWorkers uint
+	// workers per table, parallelise the snapshot creation for each page range
 	tableWorkers  uint
 	batchPageSize uint
 
@@ -40,6 +44,11 @@ type pageRange struct {
 	end   uint
 }
 
+type schemaTables struct {
+	schema string
+	tables []string
+}
+
 type snapshotTableFn func(ctx context.Context, snapshotID string, schema, table string) error
 
 type Option func(sg *SnapshotGenerator)
@@ -51,13 +60,14 @@ func NewSnapshotGenerator(ctx context.Context, cfg *Config, rowsProcessor snapsh
 	}
 
 	sg := &SnapshotGenerator{
-		logger:        loglib.NewNoopLogger(),
-		mapper:        pglib.NewMapper(conn),
-		conn:          conn,
-		rowsProcessor: rowsProcessor,
-		batchPageSize: cfg.batchPageSize(),
-		tableWorkers:  cfg.tableWorkers(),
-		schemaWorkers: cfg.schemaWorkers(),
+		logger:          loglib.NewNoopLogger(),
+		mapper:          pglib.NewMapper(conn),
+		conn:            conn,
+		rowsProcessor:   rowsProcessor,
+		batchPageSize:   cfg.batchPageSize(),
+		tableWorkers:    cfg.tableWorkers(),
+		schemaWorkers:   cfg.schemaWorkers(),
+		snapshotWorkers: cfg.snapshotWorkers(),
 	}
 
 	sg.tableSnapshotGenerator = sg.snapshotTable
@@ -97,6 +107,48 @@ func (sg *SnapshotGenerator) CreateSnapshot(ctx context.Context, ss *snapshot.Sn
 		// It will wait until all rows are processed before returning.
 		sg.rowsProcessor.Close()
 	}()
+
+	// parallelise the snapshot creation for each schema as configured by the snapshot workers.
+	errGroup, ctx := errgroup.WithContext(ctx)
+	schemaTablesChan := make(chan *schemaTables)
+	schemaErrs := make(map[string]error, len(ss.SchemaTables))
+	for i := uint(0); i < sg.snapshotWorkers; i++ {
+		errGroup.Go(func() error {
+			for schemaTables := range schemaTablesChan {
+				sg.logger.Info("creating snapshot for schema", loglib.Fields{"schema": schemaTables.schema, "tables": schemaTables.tables})
+				if err := sg.createSchemaSnapshot(ctx, schemaTables); err != nil {
+					sg.logger.Error(err, "creating snapshot for schema", loglib.Fields{"schema": schemaTables.schema, "tables": schemaTables.tables, "error": err.Error()})
+					schemaErrs[schemaTables.schema] = err
+				}
+			}
+			return nil
+		})
+	}
+	for schema, tables := range ss.SchemaTables {
+		if len(tables) == 0 {
+			sg.logger.Debug("skipping empty schema", loglib.Fields{"schema": schema})
+			continue
+		}
+		schemaTablesChan <- &schemaTables{
+			schema: schema,
+			tables: tables,
+		}
+	}
+	close(schemaTablesChan)
+
+	if err := errGroup.Wait(); err != nil {
+		return err
+	}
+
+	// collect all schema errors and return them as a single error
+	return sg.collectSchemaErrors(schemaErrs)
+}
+
+func (sg *SnapshotGenerator) Close() error {
+	return sg.conn.Close(context.Background())
+}
+
+func (sg *SnapshotGenerator) createSchemaSnapshot(ctx context.Context, schemaTables *schemaTables) error {
 	// use a transaction snapshot to ensure the table rows can be parallelised.
 	// The transaction snapshot is available for use only until the end of the
 	// transaction that exported it.
@@ -104,42 +156,38 @@ func (sg *SnapshotGenerator) CreateSnapshot(ctx context.Context, ss *snapshot.Sn
 	return sg.conn.ExecInTxWithOptions(ctx, func(tx pglib.Tx) error {
 		snapshotID, err := sg.exportSnapshot(ctx, tx)
 		if err != nil {
-			return &snapshot.Errors{SnapshotErrMsgs: []string{err.Error()}}
+			return snapshot.NewSchemaErrors(schemaTables.schema, err)
 		}
 
-		tableChan := make(chan string, len(ss.TableNames))
+		tableChan := make(chan string, len(schemaTables.tables))
 		// a map of table errors per worker to avoid race conditions
 		workerTableErrs := make([]map[string]error, sg.schemaWorkers)
 		wg := &sync.WaitGroup{}
 		// start as many go routines as configured concurrent workers per schema
 		for i := uint(0); i < sg.schemaWorkers; i++ {
 			wg.Add(1)
-			workerTableErrs[i] = make(map[string]error, len(ss.TableNames))
-			go sg.createSnapshotWorker(ctx, wg, ss, snapshotID, tableChan, workerTableErrs[i])
+			workerTableErrs[i] = make(map[string]error, len(schemaTables.tables))
+			go sg.createSnapshotWorker(ctx, wg, schemaTables.schema, snapshotID, tableChan, workerTableErrs[i])
 		}
 
-		for _, table := range ss.TableNames {
+		for _, table := range schemaTables.tables {
 			tableChan <- table
 		}
 
 		close(tableChan)
 		wg.Wait()
 
-		return sg.collectSnapshotTableErrors(workerTableErrs)
+		return sg.collectTableErrors(schemaTables.schema, workerTableErrs)
 	}, snapshotTxOptions())
 }
 
-func (sg *SnapshotGenerator) Close() error {
-	return sg.conn.Close(context.Background())
-}
-
-func (sg *SnapshotGenerator) createSnapshotWorker(ctx context.Context, wg *sync.WaitGroup, ss *snapshot.Snapshot, snapshotID string, tableChan <-chan string, tableErrMap map[string]error) {
+func (sg *SnapshotGenerator) createSnapshotWorker(ctx context.Context, wg *sync.WaitGroup, schema, snapshotID string, tableChan <-chan string, tableErrMap map[string]error) {
 	defer wg.Done()
 	for table := range tableChan {
-		logFields := loglib.Fields{"schema": ss.SchemaName, "table": table, "snapshotID": snapshotID}
+		logFields := loglib.Fields{"schema": schema, "table": table, "snapshotID": snapshotID}
 		sg.logger.Info("snapshotting table", logFields)
 
-		if err := sg.tableSnapshotGenerator(ctx, snapshotID, ss.SchemaName, table); err != nil {
+		if err := sg.tableSnapshotGenerator(ctx, snapshotID, schema, table); err != nil {
 			sg.logger.Error(err, "snapshotting table", logFields)
 			// errors will get notified unless the table doesn't exist
 			if !errors.Is(err, pglib.ErrNoRows) {
@@ -150,16 +198,38 @@ func (sg *SnapshotGenerator) createSnapshotWorker(ctx context.Context, wg *sync.
 	}
 }
 
-func (sg *SnapshotGenerator) collectSnapshotTableErrors(workerTableErrs []map[string]error) error {
-	var tableErrs []snapshot.TableError
+func (sg *SnapshotGenerator) collectTableErrors(schema string, workerTableErrs []map[string]error) error {
+	var schemaErrs *snapshot.SchemaErrors
 	for _, worker := range workerTableErrs {
 		for table, err := range worker {
-			tableErrs = append(tableErrs, snapshot.NewTableError(table, err))
+			if err == nil {
+				continue
+			}
+			if schemaErrs == nil {
+				schemaErrs = &snapshot.SchemaErrors{
+					Schema: schema,
+				}
+			}
+			schemaErrs.AddTableError(table, err)
 		}
 	}
+	if schemaErrs != nil {
+		return schemaErrs
+	}
 
-	if len(tableErrs) > 0 {
-		return &snapshot.Errors{Tables: tableErrs}
+	return nil
+}
+
+func (sg *SnapshotGenerator) collectSchemaErrors(workerSchemaErrs map[string]error) error {
+	snapshotErrs := make(snapshot.Errors, len(workerSchemaErrs))
+	for schema, err := range workerSchemaErrs {
+		if err == nil {
+			continue
+		}
+		snapshotErrs.AddError(schema, snapshot.NewSchemaErrors(schema, err))
+	}
+	if len(snapshotErrs) > 0 {
+		return snapshotErrs
 	}
 
 	return nil
diff --git a/pkg/snapshot/generator/postgres/schema/pgdumprestore/snapshot_pg_dump_restore_generator.go b/pkg/snapshot/generator/postgres/schema/pgdumprestore/snapshot_pg_dump_restore_generator.go
index ce49a0c..b64e2b5 100644
--- a/pkg/snapshot/generator/postgres/schema/pgdumprestore/snapshot_pg_dump_restore_generator.go
+++ b/pkg/snapshot/generator/postgres/schema/pgdumprestore/snapshot_pg_dump_restore_generator.go
@@ -124,15 +124,21 @@ func WithInstrumentation(i *otel.Instrumentation) Option {
 }
 
 func (s *SnapshotGenerator) CreateSnapshot(ctx context.Context, ss *snapshot.Snapshot) error {
-	s.logger.Info("creating schema snapshot", loglib.Fields{"schema": ss.SchemaName, "tables": ss.TableNames})
-	// when only the schema filter is provided to pg_dump, it will try to create
-	// it. If there are no tables in the snapshot request, we can skip the
-	// snapshot for public schema, since it already exists by default.
-	if ss.SchemaName == publicSchema && len(ss.TableNames) == 0 {
+	s.logger.Info("creating schema snapshot", loglib.Fields{"schemaTables": ss.SchemaTables})
+
+	// make sure any empty schemas are filtered out
+	dumpSchemas := make(map[string][]string, len(ss.SchemaTables))
+	for schema, tables := range ss.SchemaTables {
+		if len(tables) > 0 {
+			dumpSchemas[schema] = tables
+		}
+	}
+	// nothing to dump
+	if len(dumpSchemas) == 0 {
 		return nil
 	}
 
-	dump, err := s.dumpSchema(ctx, ss)
+	dump, err := s.dumpSchema(ctx, dumpSchemas)
 	if err != nil {
 		return err
 	}
@@ -149,14 +155,14 @@ func (s *SnapshotGenerator) CreateSnapshot(ctx context.Context, ss *snapshot.Sna
 	// if there's no further snapshotting happening, we can apply the full dump,
 	// no need to apply the constraints/indices separately.
 	if s.generator == nil {
-		return s.restoreDump(ctx, ss, append(dump.full, sequenceDump...))
+		return s.restoreDump(ctx, dumpSchemas, append(dump.full, sequenceDump...))
 	}
 
 	// otherwise, we need to apply the filtered schema dump first, then call the
 	// wrapped snapshot generator, and apply the indices and constraints last.
 	// This will make the data snapshot faster, since there will be no
 	// constraints to be updated/checked on each insert.
-	if err := s.restoreDump(ctx, ss, dump.filtered); err != nil {
+	if err := s.restoreDump(ctx, dumpSchemas, dump.filtered); err != nil {
 		return err
 	}
 
@@ -164,9 +170,9 @@ func (s *SnapshotGenerator) CreateSnapshot(ctx context.Context, ss *snapshot.Sna
 		return err
 	}
 
-	s.logger.Info("restoring schema indices and constraints", loglib.Fields{"schema": ss.SchemaName, "tables": ss.TableNames})
+	s.logger.Info("restoring schema indices and constraints", loglib.Fields{"schemaTables": ss.SchemaTables})
 	// apply the indices and constraints when the wrapped generator has finished
-	return s.restoreDump(ctx, ss, append(dump.indicesAndConstraints, sequenceDump...))
+	return s.restoreDump(ctx, dumpSchemas, append(dump.indicesAndConstraints, sequenceDump...))
 }
 
 func (s *SnapshotGenerator) Close() error {
@@ -181,8 +187,8 @@ func (s *SnapshotGenerator) Close() error {
 	return nil
 }
 
-func (s *SnapshotGenerator) dumpSchema(ctx context.Context, ss *snapshot.Snapshot) (*dump, error) {
-	pgdumpOpts, err := s.pgdumpOptions(ctx, ss)
+func (s *SnapshotGenerator) dumpSchema(ctx context.Context, schemaTables map[string][]string) (*dump, error) {
+	pgdumpOpts, err := s.pgdumpOptions(ctx, schemaTables)
 	if err != nil {
 		return nil, fmt.Errorf("preparing pg_dump options: %w", err)
 	}
@@ -210,13 +216,15 @@ func (s *SnapshotGenerator) dumpSequenceValues(ctx context.Context, sequences []
 	return s.pgDumpFn(ctx, *opts)
 }
 
-func (s *SnapshotGenerator) restoreDump(ctx context.Context, ss *snapshot.Snapshot, dump []byte) error {
+func (s *SnapshotGenerator) restoreDump(ctx context.Context, schemaTables map[string][]string, dump []byte) error {
 	// if we use table filtering in the pg_dump command, the schema creation
 	// will not be dumped, so it needs to be created explicitly (except for
 	// public schema)
-	if len(ss.TableNames) > 0 && ss.SchemaName != publicSchema {
-		if err := s.createSchemaIfNotExists(ctx, ss.SchemaName); err != nil {
-			return err
+	for schema, tables := range schemaTables {
+		if len(tables) > 0 && schema != publicSchema {
+			if err := s.createSchemaIfNotExists(ctx, schema); err != nil {
+				return err
+			}
 		}
 	}
 
@@ -239,8 +247,10 @@ func (s *SnapshotGenerator) restoreDump(ctx context.Context, ss *snapshot.Snapsh
 	// sure the schema_log table is updated accordingly with the schema view so
 	// that replication can work as expected if configured.
 	if s.schemalogStore != nil {
-		if _, err := s.schemalogStore.Insert(ctx, ss.SchemaName); err != nil {
-			return fmt.Errorf("inserting schemalog entry after schema snapshot: %w", err)
+		for schema := range schemaTables {
+			if _, err := s.schemalogStore.Insert(ctx, schema); err != nil {
+				return fmt.Errorf("inserting schemalog entry after schema snapshot: %w", err)
+			}
 		}
 	}
 
@@ -267,12 +277,16 @@ func (s *SnapshotGenerator) pgrestoreOptions() pglib.PGRestoreOptions {
 	}
 }
 
-func (s *SnapshotGenerator) pgdumpOptions(ctx context.Context, ss *snapshot.Snapshot) (*pglib.PGDumpOptions, error) {
+func (s *SnapshotGenerator) pgdumpOptions(ctx context.Context, schemaTables map[string][]string) (*pglib.PGDumpOptions, error) {
+	schemas := make([]string, 0, len(schemaTables))
+	for schema := range schemaTables {
+		schemas = append(schemas, schema)
+	}
 	opts := &pglib.PGDumpOptions{
 		ConnectionString: s.sourceURL,
 		Format:           "p",
 		SchemaOnly:       true,
-		Schemas:          []string{ss.SchemaName},
+		Schemas:          schemas,
 		Clean:            s.cleanTargetDB,
 		Create:           s.createTargetDB,
 	}
@@ -288,25 +302,25 @@ func (s *SnapshotGenerator) pgdumpOptions(ctx context.Context, ss *snapshot.Snap
 		// created. pg_dump will not include them when using the schema filter,
 		// since they do not belong to the schema.
 		var err error
-		opts.ExcludeSchemas, err = s.pgdumpExcludedSchemas(ctx, ss.SchemaName)
+		opts.ExcludeSchemas, err = s.pgdumpExcludedSchemas(ctx, schemas)
 		if err != nil {
 			return nil, err
 		}
 		opts.Schemas = nil
 	}
 
-	// wildcard means all tables in the schema, so no table filter required
-	if slices.Contains(ss.TableNames, wildcard) {
-		return opts, nil
-	}
-
 	// we use the excluded tables flag to make sure we still dump non table
 	// objects for the schema in question. If we use the tables filter, only
 	// those tables are dumped, and any related non table objects will not be
 	// dumped, causing the restore to fail due to missing related objects.
-	if len(ss.TableNames) > 0 {
+	for schema, tables := range schemaTables {
+		if hasWildcardTable(tables) {
+			// if there's the wildcard table, we don't need to add excluded
+			// tables, since they are all included.
+			continue
+		}
 		var err error
-		opts.ExcludeTables, err = s.pgdumpExcludedTables(ctx, ss.SchemaName, ss.TableNames)
+		opts.ExcludeTables, err = s.pgdumpExcludedTables(ctx, schema, tables)
 		if err != nil {
 			return nil, err
 		}
@@ -315,6 +329,8 @@ func (s *SnapshotGenerator) pgdumpOptions(ctx context.Context, ss *snapshot.Snap
 	return opts, nil
 }
 
+var selectTablesQuery = "SELECT tablename FROM pg_tables WHERE schemaname = '%s' AND tablename NOT IN (%s)"
+
 func (s *SnapshotGenerator) pgdumpExcludedTables(ctx context.Context, schemaName string, includeTables []string) ([]string, error) {
 	conn, err := s.connBuilder(ctx, s.sourceURL)
 	if err != nil {
@@ -330,7 +346,7 @@ func (s *SnapshotGenerator) pgdumpExcludedTables(ctx context.Context, schemaName
 	}
 
 	// get all tables in the schema that are not in the include list
-	query := fmt.Sprintf("SELECT tablename FROM pg_tables WHERE schemaname = '%s' AND tablename NOT IN (%s)", schemaName, strings.Join(paramRefs, ","))
+	query := fmt.Sprintf(selectTablesQuery, schemaName, strings.Join(paramRefs, ","))
 	rows, err := conn.Query(ctx, query, tableParams...)
 	if err != nil {
 		return nil, fmt.Errorf("retrieving tables from schema: %w", err)
@@ -353,18 +369,27 @@ func (s *SnapshotGenerator) pgdumpExcludedTables(ctx context.Context, schemaName
 	return excludeTables, nil
 }
 
-func (s *SnapshotGenerator) pgdumpExcludedSchemas(ctx context.Context, schemaName string) ([]string, error) {
+var selectSchemasQuery = "SELECT schema_name FROM information_schema.schemata WHERE schema_name NOT IN (%s)"
+
+func (s *SnapshotGenerator) pgdumpExcludedSchemas(ctx context.Context, includeSchemas []string) ([]string, error) {
 	conn, err := s.connBuilder(ctx, s.sourceURL)
 	if err != nil {
 		return nil, err
 	}
 	defer conn.Close(ctx)
 
+	paramRefs := make([]string, 0, len(includeSchemas))
+	schemaParams := make([]any, 0, len(includeSchemas))
+	for i, schema := range includeSchemas {
+		schemaParams = append(schemaParams, schema)
+		paramRefs = append(paramRefs, fmt.Sprintf("$%d", i+1))
+	}
+
 	// get all schemas in the database that are not in the snapshot request
-	const query = "SELECT schema_name FROM information_schema.schemata WHERE schema_name != $1"
-	rows, err := conn.Query(ctx, query, schemaName)
+	query := fmt.Sprintf(selectSchemasQuery, strings.Join(paramRefs, ","))
+	rows, err := conn.Query(ctx, query, schemaParams...)
 	if err != nil {
-		return nil, fmt.Errorf("retrieving tables from schema: %w", err)
+		return nil, fmt.Errorf("retrieving schemas: %w", err)
 	}
 	defer rows.Close()
 
@@ -398,6 +423,8 @@ func (s *SnapshotGenerator) initialiseSchemaLogStore(ctx context.Context) error
 	return err
 }
 
+const existsTableQuery = "SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_schema = $1 AND table_name = $2)"
+
 func (s *SnapshotGenerator) schemalogExists(ctx context.Context) (bool, error) {
 	conn, err := s.connBuilder(ctx, s.sourceURL)
 	if err != nil {
@@ -407,9 +434,8 @@ func (s *SnapshotGenerator) schemalogExists(ctx context.Context) (bool, error) {
 
 	// check if the pgstream.schema_log table exists, if not, we can skip the initialisation
 	// of the schemalog store
-	existsQuery := "SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_schema = $1 AND table_name = $2)"
 	var exists bool
-	err = conn.QueryRow(ctx, existsQuery, schemalog.SchemaName, schemalog.TableName).Scan(&exists)
+	err = conn.QueryRow(ctx, existsTableQuery, schemalog.SchemaName, schemalog.TableName).Scan(&exists)
 	if err != nil {
 		return false, fmt.Errorf("checking schemalog table existence: %w", err)
 	}
@@ -479,3 +505,7 @@ func (s *SnapshotGenerator) parseDump(d []byte) *dump {
 		sequences:             sequenceNames,
 	}
 }
+
+func hasWildcardTable(tables []string) bool {
+	return slices.Contains(tables, "*")
+}
diff --git a/pkg/snapshot/generator/postgres/schema/schemalog/snapshot_schemalog_generator.go b/pkg/snapshot/generator/postgres/schema/schemalog/snapshot_schemalog_generator.go
index da9f247..2ae6086 100644
--- a/pkg/snapshot/generator/postgres/schema/schemalog/snapshot_schemalog_generator.go
+++ b/pkg/snapshot/generator/postgres/schema/schemalog/snapshot_schemalog_generator.go
@@ -42,21 +42,27 @@ func WithSnapshotGenerator(g generator.SnapshotGenerator) Option {
 }
 
 func (s *SnapshotGenerator) CreateSnapshot(ctx context.Context, ss *snapshot.Snapshot) error {
-	err := func() error {
-		logEntry, err := s.schemalogStore.Insert(ctx, ss.SchemaName)
+	snapshotErrs := make(snapshot.Errors)
+	for schema := range ss.SchemaTables {
+		err := func() error {
+			logEntry, err := s.schemalogStore.Insert(ctx, schema)
+			if err != nil {
+				return err
+			}
+
+			row, err := s.logEntryToSnapshotRow(logEntry)
+			if err != nil {
+				return err
+			}
+
+			return s.processRow(ctx, row)
+		}()
 		if err != nil {
-			return err
+			snapshotErrs.AddError(schema, snapshot.NewSchemaErrors(schema, err))
 		}
-
-		row, err := s.logEntryToSnapshotRow(logEntry)
-		if err != nil {
-			return err
-		}
-
-		return s.processRow(ctx, row)
-	}()
-	if err != nil {
-		return snapshot.NewErrors(err)
+	}
+	if len(snapshotErrs) > 0 {
+		return snapshotErrs
 	}
 
 	if s.generator != nil {
diff --git a/pkg/snapshot/generator/postgres/tablefinder/pg_snapshot_table_finder.go b/pkg/snapshot/generator/postgres/tablefinder/pg_snapshot_table_finder.go
index 1f71a5b..0b34c16 100644
--- a/pkg/snapshot/generator/postgres/tablefinder/pg_snapshot_table_finder.go
+++ b/pkg/snapshot/generator/postgres/tablefinder/pg_snapshot_table_finder.go
@@ -64,11 +64,13 @@ func WithInstrumentation(i *otel.Instrumentation) Option {
 }
 
 func (s *SnapshotTableFinder) CreateSnapshot(ctx context.Context, ss *snapshot.Snapshot) error {
-	if slices.Contains(ss.TableNames, wildcard) {
-		var err error
-		ss.TableNames, err = s.discoveryFn(ctx, s.conn, ss.SchemaName)
-		if err != nil {
-			return err
+	for schema, tables := range ss.SchemaTables {
+		if slices.Contains(tables, wildcard) {
+			var err error
+			ss.SchemaTables[schema], err = s.discoveryFn(ctx, s.conn, schema)
+			if err != nil {
+				return err
+			}
 		}
 	}
 	return s.wrapped.CreateSnapshot(ctx, ss)
diff --git a/pkg/snapshot/generator/snapshot_generator_recorder.go b/pkg/snapshot/generator/snapshot_generator_recorder.go
index 15b5da0..337ea07 100644
--- a/pkg/snapshot/generator/snapshot_generator_recorder.go
+++ b/pkg/snapshot/generator/snapshot_generator_recorder.go
@@ -4,11 +4,13 @@ package generator
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
 	"github.com/xataio/pgstream/pkg/snapshot"
 	snapshotstore "github.com/xataio/pgstream/pkg/snapshot/store"
+	"golang.org/x/sync/errgroup"
 )
 
 // SnapshotRecorder is a decorator around a snapshot generator that will record
@@ -33,28 +35,26 @@ func NewSnapshotRecorder(store snapshotstore.Store, generator SnapshotGenerator,
 }
 
 func (s *SnapshotRecorder) CreateSnapshot(ctx context.Context, ss *snapshot.Snapshot) error {
-	filteredTables, err := s.filterOutExistingSnapshots(ctx, ss.SchemaName, ss.TableNames)
+	var err error
+	ss.SchemaTables, err = s.filterOutExistingSnapshots(ctx, ss.SchemaTables)
 	if err != nil {
-		return snapshot.NewErrors(err)
+		return err
 	}
 
 	// no tables to snapshot
-	if len(filteredTables) == 0 {
+	if !ss.HasTables() {
 		return nil
 	}
 
-	ss.TableNames = filteredTables
-	req := &snapshot.Request{
-		Snapshot: *ss,
-	}
+	requests := s.createRequests(ss)
 
-	if err := s.markSnapshotInProgress(ctx, req); err != nil {
-		return snapshot.NewErrors(err)
+	if err := s.markSnapshotInProgress(ctx, requests); err != nil {
+		return err
 	}
 
 	err = s.wrapped.CreateSnapshot(ctx, ss)
 
-	return s.markSnapshotCompleted(ctx, req, err)
+	return s.markSnapshotCompleted(ctx, requests, err)
 }
 
 func (s *SnapshotRecorder) Close() error {
@@ -62,16 +62,36 @@ func (s *SnapshotRecorder) Close() error {
 	return s.wrapped.Close()
 }
 
-func (s *SnapshotRecorder) markSnapshotInProgress(ctx context.Context, req *snapshot.Request) error {
-	if err := s.store.CreateSnapshotRequest(ctx, req); err != nil {
-		return err
+func (s *SnapshotRecorder) createRequests(ss *snapshot.Snapshot) []*snapshot.Request {
+	requests := make([]*snapshot.Request, 0, len(ss.SchemaTables))
+	for schema, tables := range ss.SchemaTables {
+		req := &snapshot.Request{
+			Schema: schema,
+			Tables: tables,
+		}
+		requests = append(requests, req)
+	}
+	return requests
+}
+
+func (s *SnapshotRecorder) markSnapshotInProgress(ctx context.Context, requests []*snapshot.Request) error {
+	eg, ctx := errgroup.WithContext(ctx)
+	// create one request per schema
+	for _, req := range requests {
+		eg.Go(func() error {
+			if err := s.store.CreateSnapshotRequest(ctx, req); err != nil {
+				return err
+			}
+			// the snapshot will start immediately
+			req.MarkInProgress()
+			return s.store.UpdateSnapshotRequest(ctx, req)
+		})
 	}
-	// the snapshot will start immediately
-	req.MarkInProgress()
-	return s.store.UpdateSnapshotRequest(ctx, req)
+
+	return eg.Wait()
 }
 
-func (s *SnapshotRecorder) markSnapshotCompleted(ctx context.Context, req *snapshot.Request, err error) error {
+func (s *SnapshotRecorder) markSnapshotCompleted(ctx context.Context, requests []*snapshot.Request, err error) error {
 	// make sure we can update the request status in the store regardless of
 	// context cancelations
 	if ctx.Err() != nil {
@@ -80,24 +100,58 @@ func (s *SnapshotRecorder) markSnapshotCompleted(ctx context.Context, req *snaps
 	ctx, cancel := context.WithTimeout(ctx, updateTimeout)
 	defer cancel()
 
-	req.MarkCompleted(err)
-	if updateErr := s.store.UpdateSnapshotRequest(ctx, req); updateErr != nil {
+	getSchemaErrors := func(schema string, err error) *snapshot.SchemaErrors {
 		if err == nil {
-			return snapshot.NewErrors(updateErr)
+			return nil
+		}
+		snapshotErr := &snapshot.Errors{}
+		if errors.As(err, &snapshotErr) {
+			return snapshotErr.GetSchemaErrors(schema)
 		}
-		snapshotErr := snapshot.NewErrors(err)
-		snapshotErr.AddSnapshotError(updateErr)
-		return snapshotErr
+		return snapshot.NewSchemaErrors(schema, err)
+	}
+
+	eg, ctx := errgroup.WithContext(ctx)
+	for _, req := range requests {
+		eg.Go(func() error {
+			schemaErrs := getSchemaErrors(req.Schema, err)
+			req.MarkCompleted(req.Schema, schemaErrs)
+			if updateErr := s.store.UpdateSnapshotRequest(ctx, req); updateErr != nil {
+				if err == nil {
+					return updateErr
+				}
+				schemaErrs.AddGlobalError(updateErr)
+			}
+			return nil
+		})
+	}
+
+	if err := eg.Wait(); err != nil {
+		return err
 	}
 	return err
 }
 
-func (s *SnapshotRecorder) filterOutExistingSnapshots(ctx context.Context, schema string, tables []string) ([]string, error) {
+func (s *SnapshotRecorder) filterOutExistingSnapshots(ctx context.Context, schemaTables map[string][]string) (map[string][]string, error) {
 	// if we want to be able to repeat snapshots, we don't filter out existing ones
 	if s.repeatableSnapshots {
-		return tables, nil
+		return schemaTables, nil
 	}
 
+	filteredSchemaTables := make(map[string][]string, len(schemaTables))
+	for schema, tables := range schemaTables {
+		filteredTables, err := s.filterOutExistingSchemaTables(ctx, schema, tables)
+		if err != nil {
+			return nil, fmt.Errorf("filtering existing snapshots for schema %s: %w", schema, err)
+		}
+		if len(filteredTables) > 0 {
+			filteredSchemaTables[schema] = filteredTables
+		}
+	}
+	return filteredSchemaTables, nil
+}
+
+func (s *SnapshotRecorder) filterOutExistingSchemaTables(ctx context.Context, schema string, tables []string) ([]string, error) {
 	snapshotRequests, err := s.store.GetSnapshotRequestsBySchema(ctx, schema)
 	if err != nil {
 		return nil, fmt.Errorf("retrieving existing snapshots for schema: %w", err)
@@ -109,7 +163,7 @@ func (s *SnapshotRecorder) filterOutExistingSnapshots(ctx context.Context, schem
 
 	existingRequests := map[string]*snapshot.Request{}
 	for _, req := range snapshotRequests {
-		for _, table := range req.Snapshot.TableNames {
+		for _, table := range req.Tables {
 			existingRequests[table] = req
 		}
 	}
@@ -120,7 +174,7 @@ func (s *SnapshotRecorder) filterOutExistingSnapshots(ctx context.Context, schem
 		wildCardReq, wildcardFound := existingRequests[wildcard]
 		switch table {
 		case wildcard:
-			if wildcardFound && !wildCardReq.Errors.IsSnapshotError() {
+			if wildcardFound && !wildCardReq.Errors.IsGlobalError() {
 				filteredTables = append(filteredTables, wildCardReq.Errors.GetFailedTables()...)
 				break
 			}
diff --git a/pkg/snapshot/snapshot.go b/pkg/snapshot/snapshot.go
index 724798e..5030f4e 100644
--- a/pkg/snapshot/snapshot.go
+++ b/pkg/snapshot/snapshot.go
@@ -7,14 +7,14 @@ import (
 )
 
 type Snapshot struct {
-	SchemaName string
-	TableNames []string
+	SchemaTables map[string][]string
 }
 
 type Request struct {
-	Snapshot Snapshot
-	Status   Status
-	Errors   *Errors
+	Schema string
+	Tables []string
+	Status Status
+	Errors *SchemaErrors
 }
 
 type Row struct {
@@ -44,13 +44,48 @@ const (
 	StatusCompleted  = Status("completed")
 )
 
-func (s *Snapshot) IsValid() bool {
-	return s != nil && s.SchemaName != "" && len(s.TableNames) > 0
+func (s *Snapshot) GetSchemas() []string {
+	if s == nil {
+		return nil
+	}
+
+	schemas := make([]string, 0, len(s.SchemaTables))
+	for schema := range s.SchemaTables {
+		schemas = append(schemas, schema)
+	}
+	return schemas
+}
+
+func (s *Snapshot) GetTables() []string {
+	if s == nil {
+		return nil
+	}
+
+	tables := []string{}
+	for schema, tables := range s.SchemaTables {
+		for _, table := range tables {
+			tables = append(tables, schema+"."+table)
+		}
+	}
+	return tables
+}
+
+func (s *Snapshot) HasTables() bool {
+	if s == nil {
+		return false
+	}
+
+	for _, tables := range s.SchemaTables {
+		if len(tables) > 0 {
+			return true
+		}
+	}
+	return false
 }
 
-func (r *Request) MarkCompleted(err error) {
+func (r *Request) MarkCompleted(schema string, err error) {
 	r.Status = StatusCompleted
-	r.Errors = NewErrors(err)
+	r.Errors = NewSchemaErrors(schema, err)
 }
 
 func (r *Request) MarkInProgress() {
diff --git a/pkg/snapshot/store/instrumentation/instrumented_snapshot_store.go b/pkg/snapshot/store/instrumentation/instrumented_snapshot_store.go
index 241a2f5..90eac3e 100644
--- a/pkg/snapshot/store/instrumentation/instrumented_snapshot_store.go
+++ b/pkg/snapshot/store/instrumentation/instrumented_snapshot_store.go
@@ -65,8 +65,8 @@ func (i *Store) Close() error {
 
 func (i *Store) requestAttributes(req *snapshot.Request) []attribute.KeyValue {
 	return []attribute.KeyValue{
-		{Key: "schema", Value: attribute.StringValue(req.Snapshot.SchemaName)},
-		{Key: "tables", Value: attribute.StringSliceValue(req.Snapshot.TableNames)},
+		{Key: "schema", Value: attribute.StringValue(req.Schema)},
+		{Key: "tables", Value: attribute.StringSliceValue(req.Tables)},
 		{Key: "status", Value: attribute.StringValue(string(req.Status))},
 	}
 }
diff --git a/pkg/snapshot/store/postgres/pg_snapshot_store.go b/pkg/snapshot/store/postgres/pg_snapshot_store.go
index d56a94a..a0e205b 100644
--- a/pkg/snapshot/store/postgres/pg_snapshot_store.go
+++ b/pkg/snapshot/store/postgres/pg_snapshot_store.go
@@ -43,20 +43,22 @@ func (s *Store) Close() error {
 func (s *Store) CreateSnapshotRequest(ctx context.Context, req *snapshot.Request) error {
 	query := fmt.Sprintf(`INSERT INTO %s (schema_name, table_names, created_at, updated_at, status)
 	VALUES($1, $2, now(), now(),'requested')`, snapshotsTable())
-	_, err := s.conn.Exec(ctx, query, req.Snapshot.SchemaName, pq.StringArray(req.Snapshot.TableNames))
+	_, err := s.conn.Exec(ctx, query, req.Schema, pq.StringArray(req.Tables))
 	if err != nil {
 		return fmt.Errorf("error creating snapshot request: %w", err)
 	}
+
 	return nil
 }
 
 func (s *Store) UpdateSnapshotRequest(ctx context.Context, req *snapshot.Request) error {
 	query := fmt.Sprintf(`UPDATE %s SET status = $1, errors = $2, updated_at = now()
 	WHERE schema_name = $3 and table_names = $4 and status != 'completed'`, snapshotsTable())
-	_, err := s.conn.Exec(ctx, query, req.Status, req.Errors, req.Snapshot.SchemaName, pq.StringArray(req.Snapshot.TableNames))
+	_, err := s.conn.Exec(ctx, query, req.Status, req.Errors, req.Schema, pq.StringArray(req.Tables))
 	if err != nil {
 		return fmt.Errorf("error updating snapshot request: %w", err)
 	}
+
 	return nil
 }
 
@@ -72,7 +74,7 @@ func (s *Store) GetSnapshotRequestsByStatus(ctx context.Context, status snapshot
 	snapshotRequests := []*snapshot.Request{}
 	for rows.Next() {
 		req := &snapshot.Request{}
-		if err := rows.Scan(&req.Snapshot.SchemaName, &req.Snapshot.TableNames, &req.Status, &req.Errors); err != nil {
+		if err := rows.Scan(&req.Schema, &req.Tables, &req.Status, &req.Errors); err != nil {
 			return nil, fmt.Errorf("scanning snapshot row: %w", err)
 		}
 
@@ -94,7 +96,7 @@ func (s *Store) GetSnapshotRequestsBySchema(ctx context.Context, schema string)
 	snapshotRequests := []*snapshot.Request{}
 	for rows.Next() {
 		req := &snapshot.Request{}
-		if err := rows.Scan(&req.Snapshot.SchemaName, &req.Snapshot.TableNames, &req.Status, &req.Errors); err != nil {
+		if err := rows.Scan(&req.Schema, &req.Tables, &req.Status, &req.Errors); err != nil {
 			return nil, fmt.Errorf("scanning snapshot request row: %w", err)
 		}
 
diff --git a/pkg/wal/listener/snapshot/adapter/config.go b/pkg/wal/listener/snapshot/adapter/config.go
index 1365df8..f149beb 100644
--- a/pkg/wal/listener/snapshot/adapter/config.go
+++ b/pkg/wal/listener/snapshot/adapter/config.go
@@ -8,14 +8,8 @@ import (
 
 type SnapshotConfig struct {
 	Tables []string
-	// SnapshotWorkers represents the number of snapshots the generator will
-	// process concurrently. This doesn't affect the parallelism of the tables
-	// within each individual snapshot request. It defaults to 1.
-	SnapshotWorkers uint
 }
 
-const defaultSnapshotWorkers = 1
-
 const publicSchema = "public"
 
 func (c *SnapshotConfig) schemaTableMap() map[string][]string {
@@ -32,10 +26,3 @@ func (c *SnapshotConfig) schemaTableMap() map[string][]string {
 	}
 	return schemaTableMap
 }
-
-func (c *SnapshotConfig) snapshotWorkers() uint {
-	if c.SnapshotWorkers > 0 {
-		return c.SnapshotWorkers
-	}
-	return defaultSnapshotWorkers
-}
diff --git a/pkg/wal/listener/snapshot/adapter/wal_snapshot_generator_adapter.go b/pkg/wal/listener/snapshot/adapter/wal_snapshot_generator_adapter.go
index b0e2108..d9e25e5 100644
--- a/pkg/wal/listener/snapshot/adapter/wal_snapshot_generator_adapter.go
+++ b/pkg/wal/listener/snapshot/adapter/wal_snapshot_generator_adapter.go
@@ -9,25 +9,22 @@ import (
 	loglib "github.com/xataio/pgstream/pkg/log"
 	"github.com/xataio/pgstream/pkg/snapshot"
 	"github.com/xataio/pgstream/pkg/snapshot/generator"
-	"golang.org/x/sync/errgroup"
 )
 
 // SnapshotGeneratorAdapter adapts a snapshot generator to work with WAL events
 type SnapshotGeneratorAdapter struct {
-	logger          loglib.Logger
-	generator       generator.SnapshotGenerator
-	schemaTables    map[string][]string
-	snapshotWorkers uint
+	logger       loglib.Logger
+	generator    generator.SnapshotGenerator
+	schemaTables map[string][]string
 }
 
 type Option func(a *SnapshotGeneratorAdapter)
 
 func NewSnapshotGeneratorAdapter(cfg *SnapshotConfig, generator generator.SnapshotGenerator, opts ...Option) *SnapshotGeneratorAdapter {
 	a := &SnapshotGeneratorAdapter{
-		generator:       generator,
-		schemaTables:    cfg.schemaTableMap(),
-		logger:          loglib.NewNoopLogger(),
-		snapshotWorkers: cfg.snapshotWorkers(),
+		generator:    generator,
+		schemaTables: cfg.schemaTableMap(),
+		logger:       loglib.NewNoopLogger(),
 	}
 
 	for _, opt := range opts {
@@ -51,29 +48,15 @@ func (s *SnapshotGeneratorAdapter) CreateSnapshot(ctx context.Context) (err erro
 		s.logger.Info("snapshot generation completed", loglib.Fields{"err": err, "duration": time.Since(startTime).String()})
 	}()
 
-	errGroup, ctx := errgroup.WithContext(ctx)
-	snapshotChan := make(chan *snapshot.Snapshot)
-	for i := uint(0); i < s.snapshotWorkers; i++ {
-		errGroup.Go(func() error {
-			for snapshot := range snapshotChan {
-				s.logger.Info("creating snapshot for schema", loglib.Fields{"schema": snapshot.SchemaName, "tables": snapshot.TableNames})
-				if err := s.generator.CreateSnapshot(ctx, snapshot); err != nil {
-					s.logger.Error(err, "creating snapshot for schema", loglib.Fields{"schema": snapshot.SchemaName, "tables": snapshot.TableNames})
-					return err
-				}
-			}
-			return nil
-		})
+	snapshot := &snapshot.Snapshot{
+		SchemaTables: s.schemaTables,
 	}
-	for schema, tables := range s.schemaTables {
-		snapshotChan <- &snapshot.Snapshot{
-			SchemaName: schema,
-			TableNames: tables,
-		}
+	if err := s.generator.CreateSnapshot(ctx, snapshot); err != nil {
+		s.logger.Error(err, "creating snapshot", loglib.Fields{"schemas": snapshot.GetSchemas(), "tables": snapshot.GetTables()})
+		return err
 	}
-	close(snapshotChan)
 
-	return errGroup.Wait()
+	return nil
 }
 
 func (s *SnapshotGeneratorAdapter) Close() error {


Test patch:
diff --git a/cmd/config/helper_test.go b/cmd/config/helper_test.go
index a998014..12bcf9f 100644
--- a/cmd/config/helper_test.go
+++ b/cmd/config/helper_test.go
@@ -20,7 +20,7 @@ func validateTestStreamConfig(t *testing.T, streamConfig *stream.Config) {
 	assert.Equal(t, "pgstream_mydatabase_slot", streamConfig.Listener.Postgres.Replication.ReplicationSlotName)
 	assert.NotNil(t, streamConfig.Listener.Postgres.Snapshot)
 	assert.ElementsMatch(t, []string{"test", "test_schema.Test", "another_schema.*"}, streamConfig.Listener.Postgres.Snapshot.Adapter.Tables)
-	assert.Equal(t, uint(4), streamConfig.Listener.Postgres.Snapshot.Adapter.SnapshotWorkers)
+	assert.Equal(t, uint(4), streamConfig.Listener.Postgres.Snapshot.Generator.SnapshotWorkers)
 	assert.Equal(t, uint(4), streamConfig.Listener.Postgres.Snapshot.Generator.SchemaWorkers)
 	assert.Equal(t, uint(4), streamConfig.Listener.Postgres.Snapshot.Generator.TableWorkers)
 	assert.Equal(t, uint(1000), streamConfig.Listener.Postgres.Snapshot.Generator.BatchPageSize)
diff --git a/pkg/snapshot/errors_test.go b/pkg/snapshot/errors_test.go
index bfee697..a70e5e5 100644
--- a/pkg/snapshot/errors_test.go
+++ b/pkg/snapshot/errors_test.go
@@ -23,11 +23,13 @@ var errTest = errors.New("oh noes")
 func Test_NewErrors(t *testing.T) {
 	t.Parallel()
 
+	testSchema := "test-schema"
+
 	tests := []struct {
 		name string
 		err  error
 
-		wantErr *Errors
+		wantErr Errors
 	}{
 		{
 			name: "nil error",
@@ -39,18 +41,27 @@ func Test_NewErrors(t *testing.T) {
 			name: "normal error",
 			err:  errTest,
 
-			wantErr: &Errors{
-				SnapshotErrMsgs: []string{errTest.Error()},
+			wantErr: Errors{
+				testSchema: &SchemaErrors{
+					Schema:       testSchema,
+					GlobalErrors: []string{errTest.Error()},
+				},
 			},
 		},
 		{
-			name: "snapshot error",
-			err: &Errors{
-				SnapshotErrMsgs: []string{"test"},
+			name: "global error",
+			err: Errors{
+				testSchema: &SchemaErrors{
+					Schema:       testSchema,
+					GlobalErrors: []string{errTest.Error()},
+				},
 			},
 
-			wantErr: &Errors{
-				SnapshotErrMsgs: []string{"test"},
+			wantErr: Errors{
+				testSchema: &SchemaErrors{
+					Schema:       testSchema,
+					GlobalErrors: []string{errTest.Error()},
+				},
 			},
 		},
 	}
@@ -59,7 +70,7 @@ func Test_NewErrors(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			gotErr := NewErrors(tc.err)
+			gotErr := NewErrors(testSchema, tc.err)
 			require.Equal(t, gotErr, tc.wantErr)
 		})
 	}
@@ -68,6 +79,8 @@ func Test_NewErrors(t *testing.T) {
 func TestErrors_Error(t *testing.T) {
 	t.Parallel()
 
+	testSchema := "test-schema"
+
 	tests := []struct {
 		name string
 		err  error
@@ -76,64 +89,79 @@ func TestErrors_Error(t *testing.T) {
 	}{
 		{
 			name: "nil error",
-			err:  (*Errors)(nil),
+			err:  (Errors)(nil),
 
 			wantStr: "",
 		},
 		{
 			name: "empty error",
-			err:  &Errors{},
+			err:  Errors{},
 
 			wantStr: "",
 		},
 		{
 			name: "with one snapshot error",
-			err: &Errors{
-				SnapshotErrMsgs: []string{err1},
+			err: Errors{
+				testSchema: &SchemaErrors{
+					Schema:       testSchema,
+					GlobalErrors: []string{err1},
+				},
 			},
 
-			wantStr: err1,
+			wantStr: testSchema + ": " + err1,
 		},
 		{
 			name: "with multiple snapshot errors",
-			err: &Errors{
-				SnapshotErrMsgs: []string{err1, err2},
+			err: Errors{
+				testSchema: &SchemaErrors{
+					Schema:       testSchema,
+					GlobalErrors: []string{err1, err2},
+				},
 			},
 
-			wantStr: fmt.Sprintf("%s;%s", err1, err2),
+			wantStr: fmt.Sprintf("%s: %s;%s", testSchema, err1, err2),
 		},
 		{
 			name: "with one table error",
-			err: &Errors{
-				Tables: []TableError{
-					{ErrorMsg: err1, Table: table1},
+			err: Errors{
+				testSchema: &SchemaErrors{
+					Schema: testSchema,
+					TableErrors: map[string]string{
+						table1: err1,
+					},
 				},
 			},
 
-			wantStr: fmt.Sprintf("%s: %s", table1, err1),
+			wantStr: fmt.Sprintf("%s: %s: %s", testSchema, table1, err1),
 		},
 		{
 			name: "with multiple table errors",
-			err: &Errors{
-				Tables: []TableError{
-					{ErrorMsg: err1, Table: table1},
-					{ErrorMsg: err2, Table: table2},
+			err: Errors{
+				testSchema: &SchemaErrors{
+					Schema: testSchema,
+					TableErrors: map[string]string{
+						table1: err1,
+						table2: err2,
+					},
 				},
 			},
 
-			wantStr: fmt.Sprintf("%s: %s;%s: %s", table1, err1, table2, err2),
+			wantStr: fmt.Sprintf("%s: %s: %s;%s: %s", testSchema, table1, err1, table2, err2),
 		},
 		{
 			name: "with snapshot and table errors",
-			err: &Errors{
-				SnapshotErrMsgs: []string{err1, err2},
-				Tables: []TableError{
-					NewTableError(table1, errors.New(err1)),
-					NewTableError(table2, errors.New(err2)),
+			err: Errors{
+				testSchema: &SchemaErrors{
+					Schema:       testSchema,
+					GlobalErrors: []string{err1, err2},
+					TableErrors: map[string]string{
+						table1: err1,
+						table2: err2,
+					},
 				},
 			},
 
-			wantStr: fmt.Sprintf("%s;%s;%s: %s;%s: %s", err1, err2, table1, err1, table2, err2),
+			wantStr: fmt.Sprintf("%s: %s;%s;%s: %s;%s: %s", testSchema, err1, err2, table1, err1, table2, err2),
 		},
 	}
 
@@ -147,12 +175,65 @@ func TestErrors_Error(t *testing.T) {
 	}
 }
 
-func TestErrors_AddSnapshotErrors(t *testing.T) {
+func TestErrors_GetSchemaErrors(t *testing.T) {
+	t.Parallel()
+
+	testSchema := "test-schema"
+
+	tests := []struct {
+		name string
+		err  Errors
+
+		wantErrs *SchemaErrors
+	}{
+		{
+			name: "ok",
+			err: Errors{
+				testSchema: &SchemaErrors{
+					Schema:       testSchema,
+					GlobalErrors: []string{err1, err2},
+				},
+			},
+			wantErrs: &SchemaErrors{
+				Schema:       testSchema,
+				GlobalErrors: []string{err1, err2},
+			},
+		},
+		{
+			name: "ok - no schema errors",
+			err: Errors{
+				"another-schema": &SchemaErrors{
+					Schema:       testSchema,
+					GlobalErrors: []string{err1, err2},
+				},
+			},
+			wantErrs: nil,
+		},
+		{
+			name:     "ok - nil errors",
+			err:      nil,
+			wantErrs: nil,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			schemaErrs := tc.err.GetSchemaErrors(testSchema)
+			require.Equal(t, tc.wantErrs, schemaErrs)
+		})
+	}
+}
+
+func TestErrors_AddError(t *testing.T) {
 	t.Parallel()
 
+	testSchema := "test-schema"
+
 	tests := []struct {
 		name   string
-		err    *Errors
+		err    Errors
 		addErr error
 
 		wantErr error
@@ -162,26 +243,35 @@ func TestErrors_AddSnapshotErrors(t *testing.T) {
 			err:    nil,
 			addErr: errTest,
 
-			wantErr: (*Errors)(nil),
+			wantErr: (Errors)(nil),
 		},
 		{
 			name:   "error",
-			err:    &Errors{},
+			err:    Errors{},
 			addErr: errTest,
 
-			wantErr: &Errors{
-				SnapshotErrMsgs: []string{errTest.Error()},
+			wantErr: Errors{
+				testSchema: &SchemaErrors{
+					Schema:       testSchema,
+					GlobalErrors: []string{errTest.Error()},
+				},
 			},
 		},
 		{
-			name: "with snapshot errors",
-			err: &Errors{
-				SnapshotErrMsgs: []string{err1, err2},
+			name: "with global errors",
+			err: Errors{
+				testSchema: &SchemaErrors{
+					Schema:       testSchema,
+					GlobalErrors: []string{err1, err2},
+				},
 			},
 			addErr: errTest,
 
-			wantErr: &Errors{
-				SnapshotErrMsgs: []string{err1, err2, errTest.Error()},
+			wantErr: Errors{
+				testSchema: &SchemaErrors{
+					Schema:       testSchema,
+					GlobalErrors: []string{err1, err2, errTest.Error()},
+				},
 			},
 		},
 	}
@@ -190,19 +280,72 @@ func TestErrors_AddSnapshotErrors(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			tc.err.AddSnapshotError(tc.addErr)
+			tc.err.AddError(testSchema, tc.addErr)
 			require.Equal(t, tc.wantErr, tc.err)
 		})
 	}
 }
 
-func TestErrors_IsTableError(t *testing.T) {
+func Test_NewSchemaErrors(t *testing.T) {
 	t.Parallel()
 
+	testSchema := "test-schema"
+	otherSchema := "other-schema"
+
+	tests := []struct {
+		name     string
+		schema   string
+		err      error
+		wantErrs *SchemaErrors
+	}{
+		{
+			name:     "nil error",
+			schema:   testSchema,
+			err:      nil,
+			wantErrs: nil,
+		},
+		{
+			name:   "normal error",
+			schema: testSchema,
+			err:    errTest,
+			wantErrs: &SchemaErrors{
+				Schema:       testSchema,
+				GlobalErrors: []string{errTest.Error()},
+			},
+		},
+		{
+			name:   "error is already SchemaErrors",
+			schema: testSchema,
+			err: &SchemaErrors{
+				Schema:       otherSchema,
+				GlobalErrors: []string{"err1", "err2"},
+			},
+			wantErrs: &SchemaErrors{
+				Schema:       otherSchema,
+				GlobalErrors: []string{"err1", "err2"},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			got := NewSchemaErrors(tc.schema, tc.err)
+			require.Equal(t, tc.wantErrs, got)
+		})
+	}
+}
+
+func TestSchemaErrors_IsTableError(t *testing.T) {
+	t.Parallel()
+
+	testSchema := "test-schema"
+
 	tests := []struct {
 		name  string
 		table string
-		err   *Errors
+		err   *SchemaErrors
 
 		wantTableErr bool
 	}{
@@ -215,15 +358,16 @@ func TestErrors_IsTableError(t *testing.T) {
 		},
 		{
 			name:  "no errors",
-			err:   &Errors{},
+			err:   &SchemaErrors{},
 			table: table1,
 
 			wantTableErr: false,
 		},
 		{
 			name: "with snapshto errors",
-			err: &Errors{
-				SnapshotErrMsgs: []string{err1},
+			err: &SchemaErrors{
+				Schema:       testSchema,
+				GlobalErrors: []string{err1},
 			},
 			table: table1,
 
@@ -231,9 +375,10 @@ func TestErrors_IsTableError(t *testing.T) {
 		},
 		{
 			name: "with table errors",
-			err: &Errors{
-				Tables: []TableError{
-					{Table: table1, ErrorMsg: err1},
+			err: &SchemaErrors{
+				Schema: testSchema,
+				TableErrors: map[string]string{
+					table1: err1,
 				},
 			},
 			table: table1,
@@ -242,9 +387,10 @@ func TestErrors_IsTableError(t *testing.T) {
 		},
 		{
 			name: "wildcard table",
-			err: &Errors{
-				Tables: []TableError{
-					{Table: table1, ErrorMsg: err1},
+			err: &SchemaErrors{
+				Schema: testSchema,
+				TableErrors: map[string]string{
+					table1: err1,
 				},
 			},
 			table: "*",
@@ -263,12 +409,14 @@ func TestErrors_IsTableError(t *testing.T) {
 	}
 }
 
-func TestErrors_GetFailedTables(t *testing.T) {
+func TestSchemaErrors_GetFailedTables(t *testing.T) {
 	t.Parallel()
 
+	testSchema := "test-schema"
+
 	tests := []struct {
 		name string
-		err  *Errors
+		err  *SchemaErrors
 
 		wantTables []string
 	}{
@@ -280,16 +428,17 @@ func TestErrors_GetFailedTables(t *testing.T) {
 		},
 		{
 			name: "no failed tables",
-			err:  &Errors{},
+			err:  &SchemaErrors{},
 
 			wantTables: []string{},
 		},
 		{
 			name: "with failed tables",
-			err: &Errors{
-				Tables: []TableError{
-					{Table: table1, ErrorMsg: err1},
-					{Table: table2, ErrorMsg: err2},
+			err: &SchemaErrors{
+				Schema: testSchema,
+				TableErrors: map[string]string{
+					table1: err1,
+					table2: err2,
 				},
 			},
 
@@ -302,43 +451,122 @@ func TestErrors_GetFailedTables(t *testing.T) {
 			t.Parallel()
 
 			tables := tc.err.GetFailedTables()
-			require.Equal(t, tc.wantTables, tables)
+			require.ElementsMatch(t, tc.wantTables, tables)
 		})
 	}
 }
 
-func TestErrors_IsSnapshotError(t *testing.T) {
+func TestSchemaErrors_AddTableError(t *testing.T) {
 	t.Parallel()
 
+	testSchema := "test-schema"
+
+	tests := []struct {
+		name   string
+		err    *SchemaErrors
+		table  string
+		addErr error
+
+		wantErrs *SchemaErrors
+	}{
+		{
+			name:   "nil schema errors",
+			err:    nil,
+			table:  table1,
+			addErr: errTest,
+
+			wantErrs: nil,
+		},
+		{
+			name:   "nil add error",
+			err:    &SchemaErrors{},
+			table:  table1,
+			addErr: nil,
+
+			wantErrs: &SchemaErrors{},
+		},
+		{
+			name: "new table error",
+			err: &SchemaErrors{
+				Schema: testSchema,
+			},
+			table:  table1,
+			addErr: errTest,
+
+			wantErrs: &SchemaErrors{
+				Schema: testSchema,
+				TableErrors: map[string]string{
+					table1: errTest.Error(),
+				},
+			},
+		},
+		{
+			name: "existing table errors",
+			err: &SchemaErrors{
+				Schema: testSchema,
+				TableErrors: map[string]string{
+					table1: errTest.Error(),
+				},
+			},
+			table:  table2,
+			addErr: errTest,
+
+			wantErrs: &SchemaErrors{
+				Schema: testSchema,
+				TableErrors: map[string]string{
+					table1: errTest.Error(),
+					table2: errTest.Error(),
+				},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			tc.err.AddTableError(tc.table, tc.addErr)
+			require.Equal(t, tc.wantErrs, tc.err)
+		})
+	}
+}
+
+func TestSchemaErrors_IsGlobalError(t *testing.T) {
+	t.Parallel()
+
+	testSchema := "test-schema"
+
 	tests := []struct {
 		name string
-		err  *Errors
+		err  *SchemaErrors
 
-		wantIsSnapshotErr bool
+		wantIsGlobalErr bool
 	}{
 		{
 			name: "nil error",
 			err:  nil,
 
-			wantIsSnapshotErr: false,
+			wantIsGlobalErr: false,
 		},
 		{
 			name: "with snapshot error",
-			err: &Errors{
-				SnapshotErrMsgs: []string{err1, err2},
+			err: &SchemaErrors{
+				Schema:       testSchema,
+				GlobalErrors: []string{err1, err2},
 			},
 
-			wantIsSnapshotErr: true,
+			wantIsGlobalErr: true,
 		},
 		{
 			name: "with table error",
-			err: &Errors{
-				Tables: []TableError{
-					NewTableError(table1, errTest),
+			err: &SchemaErrors{
+				Schema: testSchema,
+				TableErrors: map[string]string{
+					table1: err1,
 				},
 			},
 
-			wantIsSnapshotErr: false,
+			wantIsGlobalErr: false,
 		},
 	}
 
@@ -346,8 +574,8 @@ func TestErrors_IsSnapshotError(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			got := tc.err.IsSnapshotError()
-			require.Equal(t, tc.wantIsSnapshotErr, got)
+			got := tc.err.IsGlobalError()
+			require.Equal(t, tc.wantIsGlobalErr, got)
 		})
 	}
 }
diff --git a/pkg/snapshot/generator/postgres/data/pg_snapshot_generator_integration_test.go b/pkg/snapshot/generator/postgres/data/pg_snapshot_generator_integration_test.go
index 204a85d..8da7f09 100644
--- a/pkg/snapshot/generator/postgres/data/pg_snapshot_generator_integration_test.go
+++ b/pkg/snapshot/generator/postgres/data/pg_snapshot_generator_integration_test.go
@@ -45,8 +45,9 @@ func Test_PostgresSnapshotGenerator(t *testing.T) {
 
 	go func() {
 		err = generator.CreateSnapshot(ctx, &snapshot.Snapshot{
-			SchemaName: "public",
-			TableNames: []string{testTable},
+			SchemaTables: map[string][]string{
+				"public": {testTable},
+			},
 		})
 		require.NoError(t, err)
 		mockProcessor.Close()
diff --git a/pkg/snapshot/generator/postgres/data/pg_snapshot_generator_test.go b/pkg/snapshot/generator/postgres/data/pg_snapshot_generator_test.go
index 7e97ba5..aca90ad 100644
--- a/pkg/snapshot/generator/postgres/data/pg_snapshot_generator_test.go
+++ b/pkg/snapshot/generator/postgres/data/pg_snapshot_generator_test.go
@@ -6,7 +6,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sort"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -28,9 +27,11 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 
 	testTable1 := "test-table-1"
 	testTable2 := "test-table-2"
+	testSchema := "test-schema"
 	testSnapshot := &snapshot.Snapshot{
-		SchemaName: "test-schema",
-		TableNames: []string{testTable1},
+		SchemaTables: map[string][]string{
+			testSchema: {testTable1},
+		},
 	}
 
 	txOptions := pglib.TxOptions{
@@ -48,7 +49,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 
 	testRow := func(tableName string, columns []snapshot.Column) *snapshot.Row {
 		return &snapshot.Row{
-			Schema:  testSnapshot.SchemaName,
+			Schema:  testSchema,
 			Table:   tableName,
 			Columns: columns,
 		}
@@ -96,7 +97,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 							},
 							QueryRowFn: func(ctx context.Context, query string, args ...any) pglib.Row {
 								require.Equal(t, "SELECT c.relpages FROM pg_class c JOIN pg_namespace n ON c.relnamespace=n.oid WHERE c.relname=$1 and n.nspname=$2", query)
-								require.Equal(t, []any{testTable1, testSnapshot.SchemaName}, args)
+								require.Equal(t, []any{testTable1, testSchema}, args)
 								return &pgmocks.Row{
 									ScanFn: func(args ...any) error {
 										require.Len(t, args, 1)
@@ -117,7 +118,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 								return pglib.CommandTag{}, nil
 							},
 							QueryFn: func(ctx context.Context, query string, args ...any) (pglib.Rows, error) {
-								require.Equal(t, fmt.Sprintf("SELECT * FROM %q.%q WHERE ctid BETWEEN '(%d,0)' AND '(%d,0)'", testSnapshot.SchemaName, testTable1, 0, 10), query)
+								require.Equal(t, fmt.Sprintf("SELECT * FROM %q.%q WHERE ctid BETWEEN '(%d,0)' AND '(%d,0)'", testSchema, testTable1, 0, 10), query)
 								require.Len(t, args, 0)
 								return &pgmocks.Rows{
 									CloseFn: func() {},
@@ -204,8 +205,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				},
 			},
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSnapshot.SchemaName,
-				TableNames: []string{testTable1, testTable2},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable1, testTable2},
+				},
 			},
 			schemaWorkers: 2,
 
@@ -248,7 +250,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 							},
 							QueryRowFn: func(ctx context.Context, query string, args ...any) pglib.Row {
 								require.Equal(t, "SELECT c.relpages FROM pg_class c JOIN pg_namespace n ON c.relnamespace=n.oid WHERE c.relname=$1 and n.nspname=$2", query)
-								require.Equal(t, []any{testTable1, testSnapshot.SchemaName}, args)
+								require.Equal(t, []any{testTable1, testSchema}, args)
 								return &pgmocks.Row{
 									ScanFn: func(args ...any) error {
 										require.Len(t, args, 1)
@@ -269,7 +271,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 								return pglib.CommandTag{}, nil
 							},
 							QueryFn: func(ctx context.Context, query string, args ...any) (pglib.Rows, error) {
-								require.Equal(t, fmt.Sprintf("SELECT * FROM %q.%q WHERE ctid BETWEEN '(%d,0)' AND '(%d,0)'", testSnapshot.SchemaName, testTable1, 0, 10), query)
+								require.Equal(t, fmt.Sprintf("SELECT * FROM %q.%q WHERE ctid BETWEEN '(%d,0)' AND '(%d,0)'", testSchema, testTable1, 0, 10), query)
 								require.Len(t, args, 0)
 								return &pgmocks.Rows{
 									CloseFn: func() {},
@@ -329,7 +331,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 							},
 							QueryRowFn: func(ctx context.Context, query string, args ...any) pglib.Row {
 								require.Equal(t, "SELECT c.relpages FROM pg_class c JOIN pg_namespace n ON c.relnamespace=n.oid WHERE c.relname=$1 and n.nspname=$2", query)
-								require.Equal(t, []any{testTable1, testSnapshot.SchemaName}, args)
+								require.Equal(t, []any{testTable1, testSchema}, args)
 								return &pgmocks.Row{
 									ScanFn: func(args ...any) error {
 										require.Len(t, args, 1)
@@ -350,7 +352,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 								return pglib.CommandTag{}, nil
 							},
 							QueryFn: func(ctx context.Context, query string, args ...any) (pglib.Rows, error) {
-								require.Equal(t, fmt.Sprintf("SELECT * FROM %q.%q WHERE ctid BETWEEN '(%d,0)' AND '(%d,0)'", testSnapshot.SchemaName, testTable1, 0, 10), query)
+								require.Equal(t, fmt.Sprintf("SELECT * FROM %q.%q WHERE ctid BETWEEN '(%d,0)' AND '(%d,0)'", testSchema, testTable1, 0, 10), query)
 								require.Len(t, args, 0)
 								return &pgmocks.Rows{
 									CloseFn:             func() {},
@@ -395,7 +397,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				},
 			},
 
-			wantErr:  snapshot.NewErrors(fmt.Errorf("exporting snapshot: %w", errTest)),
+			wantErr:  snapshot.NewErrors(testSchema, fmt.Errorf("exporting snapshot: %w", errTest)),
 			wantRows: []*snapshot.Row{},
 		},
 		{
@@ -435,11 +437,11 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				},
 			},
 
-			wantErr: &snapshot.Errors{
-				Tables: []snapshot.TableError{
-					{
-						Table:    testTable1,
-						ErrorMsg: fmt.Sprintf("setting transaction snapshot: %v", errTest),
+			wantErr: snapshot.Errors{
+				testSchema: &snapshot.SchemaErrors{
+					Schema: testSchema,
+					TableErrors: map[string]string{
+						testTable1: fmt.Sprintf("setting transaction snapshot: %v", errTest),
 					},
 				},
 			},
@@ -476,7 +478,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 							},
 							QueryRowFn: func(ctx context.Context, query string, args ...any) pglib.Row {
 								require.Equal(t, "SELECT c.relpages FROM pg_class c JOIN pg_namespace n ON c.relnamespace=n.oid WHERE c.relname=$1 and n.nspname=$2", query)
-								require.Equal(t, []any{testTable1, testSnapshot.SchemaName}, args)
+								require.Equal(t, []any{testTable1, testSchema}, args)
 								return &pgmocks.Row{
 									ScanFn: func(args ...any) error {
 										return errTest
@@ -491,11 +493,11 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				},
 			},
 
-			wantErr: &snapshot.Errors{
-				Tables: []snapshot.TableError{
-					{
-						Table:    testTable1,
-						ErrorMsg: fmt.Sprintf("getting page count for table test-schema.test-table-1: %v", errTest),
+			wantErr: snapshot.Errors{
+				testSchema: &snapshot.SchemaErrors{
+					Schema: testSchema,
+					TableErrors: map[string]string{
+						testTable1: fmt.Sprintf("getting page count for table test-schema.test-table-1: %v", errTest),
 					},
 				},
 			},
@@ -532,7 +534,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 							},
 							QueryRowFn: func(ctx context.Context, query string, args ...any) pglib.Row {
 								require.Equal(t, "SELECT c.relpages FROM pg_class c JOIN pg_namespace n ON c.relnamespace=n.oid WHERE c.relname=$1 and n.nspname=$2", query)
-								require.Equal(t, []any{testTable1, testSnapshot.SchemaName}, args)
+								require.Equal(t, []any{testTable1, testSchema}, args)
 								return &pgmocks.Row{
 									ScanFn: func(args ...any) error {
 										require.Len(t, args, 1)
@@ -560,11 +562,11 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				},
 			},
 
-			wantErr: &snapshot.Errors{
-				Tables: []snapshot.TableError{
-					{
-						Table:    testTable1,
-						ErrorMsg: fmt.Sprintf("setting transaction snapshot: %v", errTest),
+			wantErr: snapshot.Errors{
+				testSchema: &snapshot.SchemaErrors{
+					Schema: testSchema,
+					TableErrors: map[string]string{
+						testTable1: fmt.Sprintf("setting transaction snapshot: %v", errTest),
 					},
 				},
 			},
@@ -601,7 +603,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 							},
 							QueryRowFn: func(ctx context.Context, query string, args ...any) pglib.Row {
 								require.Equal(t, "SELECT c.relpages FROM pg_class c JOIN pg_namespace n ON c.relnamespace=n.oid WHERE c.relname=$1 and n.nspname=$2", query)
-								require.Equal(t, []any{testTable1, testSnapshot.SchemaName}, args)
+								require.Equal(t, []any{testTable1, testSchema}, args)
 								return &pgmocks.Row{
 									ScanFn: func(args ...any) error {
 										require.Len(t, args, 1)
@@ -622,7 +624,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 								return pglib.CommandTag{}, nil
 							},
 							QueryFn: func(ctx context.Context, query string, args ...any) (pglib.Rows, error) {
-								require.Equal(t, fmt.Sprintf("SELECT * FROM %q.%q WHERE ctid BETWEEN '(%d,0)' AND '(%d,0)'", testSnapshot.SchemaName, testTable1, 0, 10), query)
+								require.Equal(t, fmt.Sprintf("SELECT * FROM %q.%q WHERE ctid BETWEEN '(%d,0)' AND '(%d,0)'", testSchema, testTable1, 0, 10), query)
 								require.Len(t, args, 0)
 								return nil, errTest
 							},
@@ -634,11 +636,11 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				},
 			},
 
-			wantErr: &snapshot.Errors{
-				Tables: []snapshot.TableError{
-					{
-						Table:    testTable1,
-						ErrorMsg: fmt.Sprintf("querying table rows: %v", errTest),
+			wantErr: snapshot.Errors{
+				testSchema: &snapshot.SchemaErrors{
+					Schema: testSchema,
+					TableErrors: map[string]string{
+						testTable1: fmt.Sprintf("querying table rows: %v", errTest),
 					},
 				},
 			},
@@ -675,7 +677,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 							},
 							QueryRowFn: func(ctx context.Context, query string, args ...any) pglib.Row {
 								require.Equal(t, "SELECT c.relpages FROM pg_class c JOIN pg_namespace n ON c.relnamespace=n.oid WHERE c.relname=$1 and n.nspname=$2", query)
-								require.Equal(t, []any{testTable1, testSnapshot.SchemaName}, args)
+								require.Equal(t, []any{testTable1, testSchema}, args)
 								return &pgmocks.Row{
 									ScanFn: func(args ...any) error {
 										require.Len(t, args, 1)
@@ -712,11 +714,11 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				},
 			},
 
-			wantErr: &snapshot.Errors{
-				Tables: []snapshot.TableError{
-					{
-						Table:    testTable1,
-						ErrorMsg: fmt.Sprintf("retrieving rows values: %v", errTest),
+			wantErr: snapshot.Errors{
+				testSchema: &snapshot.SchemaErrors{
+					Schema: testSchema,
+					TableErrors: map[string]string{
+						testTable1: fmt.Sprintf("retrieving rows values: %v", errTest),
 					},
 				},
 			},
@@ -753,7 +755,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 							},
 							QueryRowFn: func(ctx context.Context, query string, args ...any) pglib.Row {
 								require.Equal(t, "SELECT c.relpages FROM pg_class c JOIN pg_namespace n ON c.relnamespace=n.oid WHERE c.relname=$1 and n.nspname=$2", query)
-								require.Equal(t, []any{testTable1, testSnapshot.SchemaName}, args)
+								require.Equal(t, []any{testTable1, testSchema}, args)
 								return &pgmocks.Row{
 									ScanFn: func(args ...any) error {
 										require.Len(t, args, 1)
@@ -790,11 +792,11 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				},
 			},
 
-			wantErr: &snapshot.Errors{
-				Tables: []snapshot.TableError{
-					{
-						Table:    testTable1,
-						ErrorMsg: errTest.Error(),
+			wantErr: snapshot.Errors{
+				testSchema: &snapshot.SchemaErrors{
+					Schema: testSchema,
+					TableErrors: map[string]string{
+						testTable1: errTest.Error(),
 					},
 				},
 			},
@@ -839,20 +841,18 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				},
 			},
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSnapshot.SchemaName,
-				TableNames: []string{testTable1, testTable2},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable1, testTable2},
+				},
 			},
 			schemaWorkers: 2,
 
-			wantErr: &snapshot.Errors{
-				Tables: []snapshot.TableError{
-					{
-						Table:    testTable1,
-						ErrorMsg: fmt.Sprintf("getting page count for table %s.%s: %v", testSnapshot.SchemaName, testTable1, errTest),
-					},
-					{
-						Table:    testTable2,
-						ErrorMsg: fmt.Sprintf("getting page count for table %s.%s: %v", testSnapshot.SchemaName, testTable2, errTest),
+			wantErr: snapshot.Errors{
+				testSchema: &snapshot.SchemaErrors{
+					Schema: testSchema,
+					TableErrors: map[string]string{
+						testTable1: fmt.Sprintf("getting page count for table %s.%s: %v", testSchema, testTable1, errTest),
+						testTable2: fmt.Sprintf("getting page count for table %s.%s: %v", testSchema, testTable2, errTest),
 					},
 				},
 			},
@@ -877,9 +877,10 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 						return nil
 					},
 				},
-				schemaWorkers: 1,
-				tableWorkers:  1,
-				batchPageSize: 10,
+				schemaWorkers:   1,
+				tableWorkers:    1,
+				batchPageSize:   10,
+				snapshotWorkers: 1,
 			}
 			sg.tableSnapshotGenerator = sg.snapshotTable
 
@@ -893,7 +894,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 			}
 
 			err := sg.CreateSnapshot(context.Background(), s)
-			require.Equal(t, tc.wantErr, sortSnapshotTableErrors(err))
+			require.Equal(t, tc.wantErr, err)
 			close(rowChan)
 
 			rows := []*snapshot.Row{}
@@ -907,14 +908,3 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		})
 	}
 }
-
-func sortSnapshotTableErrors(err error) error {
-	var snapshotErrs *snapshot.Errors
-	if errors.As(err, &snapshotErrs) {
-		sort.Slice(snapshotErrs.Tables, func(i, j int) bool {
-			return snapshotErrs.Tables[i].Table < snapshotErrs.Tables[j].Table
-		})
-		return snapshotErrs
-	}
-	return err
-}
diff --git a/pkg/snapshot/generator/postgres/schema/pgdumprestore/snapshot_pg_dump_restore_generator_test.go b/pkg/snapshot/generator/postgres/schema/pgdumprestore/snapshot_pg_dump_restore_generator_test.go
index 7529309..debb464 100644
--- a/pkg/snapshot/generator/postgres/schema/pgdumprestore/snapshot_pg_dump_restore_generator_test.go
+++ b/pkg/snapshot/generator/postgres/schema/pgdumprestore/snapshot_pg_dump_restore_generator_test.go
@@ -34,6 +34,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 	excludedSchema := "excluded_test_schema"
 	errTest := errors.New("oh noes")
 	testSequence := "test_sequence"
+	testRole := "test_role"
 
 	validQuerier := func() *mocks.Querier {
 		return &mocks.Querier{
@@ -43,7 +44,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 			},
 			QueryFn: func(ctx context.Context, query string, args ...any) (pglib.Rows, error) {
 				switch query {
-				case "SELECT schema_name FROM information_schema.schemata WHERE schema_name != $1":
+				case fmt.Sprintf(selectSchemasQuery, "$1"):
 					require.Equal(t, []any{testSchema}, args)
 					return &mocks.Rows{
 						CloseFn: func() {},
@@ -57,7 +58,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 						},
 						ErrFn: func() error { return nil },
 					}, nil
-				case "SELECT tablename FROM pg_tables WHERE schemaname = 'test_schema' AND tablename NOT IN ($1)":
+				case fmt.Sprintf(selectTablesQuery, testSchema, "$1"):
 					require.Equal(t, []any{testTable}, args)
 					return &mocks.Rows{
 						CloseFn: func() {},
@@ -87,14 +88,16 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		pgrestoreFn    pglib.PGRestoreFn
 		schemalogStore schemalog.Store
 		generator      generator.SnapshotGenerator
+		role           string
 
 		wantErr error
 	}{
 		{
 			name: "ok",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{testTable},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable},
+				},
 			},
 			conn: validQuerier(),
 			pgdumpFn: newMockPgdump(func(_ context.Context, i uint, po pglib.PGDumpOptions) ([]byte, error) {
@@ -106,6 +109,8 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 						SchemaOnly:       true,
 						ExcludeSchemas:   []string{pglib.QuoteIdentifier(excludedSchema)},
 						ExcludeTables:    []string{pglib.QuoteIdentifier(excludedTable)},
+						Role:             testRole,
+						NoOwner:          true,
 					}, po)
 					return schemaDump, nil
 
@@ -129,14 +134,16 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				require.Equal(t, append(schemaDump, sequenceDump...), dump)
 				return "", nil
 			},
+			role: testRole,
 
 			wantErr: nil,
 		},
 		{
 			name: "ok - no sequence dump",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{testTable},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable},
+				},
 			},
 			conn: validQuerier(),
 			pgdumpFn: newMockPgdump(func(_ context.Context, i uint, po pglib.PGDumpOptions) ([]byte, error) {
@@ -168,8 +175,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		{
 			name: "ok - with generator",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{testTable},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable},
+				},
 			},
 			conn: validQuerier(),
 			pgdumpFn: newMockPgdump(func(_ context.Context, i uint, po pglib.PGDumpOptions) ([]byte, error) {
@@ -213,8 +221,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 			generator: &generatormocks.Generator{
 				CreateSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
 					require.Equal(t, &snapshot.Snapshot{
-						SchemaName: testSchema,
-						TableNames: []string{testTable},
+						SchemaTables: map[string][]string{
+							testSchema: {testTable},
+						},
 					}, ss)
 					return nil
 				},
@@ -225,8 +234,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		{
 			name: "ok - wildcard",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{"*"},
+				SchemaTables: map[string][]string{
+					testSchema: {"*"},
+				},
 			},
 			conn: validQuerier(),
 			pgdumpFn: newMockPgdump(func(_ context.Context, i uint, po pglib.PGDumpOptions) ([]byte, error) {
@@ -265,10 +275,14 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		{
 			name: "ok - no tables in public schema",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: publicSchema,
-				TableNames: []string{},
+				SchemaTables: map[string][]string{
+					publicSchema: {},
+				},
 			},
 			conn: &mocks.Querier{
+				QueryFn: func(ctx context.Context, query string, args ...any) (pglib.Rows, error) {
+					return nil, fmt.Errorf("unexpected query: %s", query)
+				},
 				ExecFn: func(ctx context.Context, i uint, query string, args ...any) (pglib.CommandTag, error) {
 					return pglib.CommandTag{}, errors.New("ExecFn: should not be called")
 				},
@@ -285,8 +299,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		{
 			name: "error - querying excluded schemas",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{testTable},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable},
+				},
 			},
 			conn: &mocks.Querier{
 				QueryFn: func(ctx context.Context, query string, args ...any) (pglib.Rows, error) {
@@ -305,13 +320,14 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		{
 			name: "error - scanning excluded tables",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{testTable},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable},
+				},
 			},
 			conn: &mocks.Querier{
 				QueryFn: func(ctx context.Context, query string, args ...any) (pglib.Rows, error) {
 					switch query {
-					case "SELECT schema_name FROM information_schema.schemata WHERE schema_name != $1":
+					case fmt.Sprintf(selectSchemasQuery, "$1"):
 						require.Equal(t, []any{testSchema}, args)
 						return &mocks.Rows{
 							CloseFn: func() {},
@@ -325,7 +341,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 							},
 							ErrFn: func() error { return nil },
 						}, nil
-					case "SELECT tablename FROM pg_tables WHERE schemaname = 'test_schema' AND tablename NOT IN ($1)":
+					case fmt.Sprintf(selectTablesQuery, testSchema, "$1"):
 						require.Equal(t, []any{testTable}, args)
 						return &mocks.Rows{
 							CloseFn: func() {},
@@ -352,8 +368,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		{
 			name: "error - excluded tables rows",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{testTable},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable},
+				},
 			},
 			conn: &mocks.Querier{
 				ExecFn: func(ctx context.Context, i uint, query string, args ...any) (pglib.CommandTag, error) {
@@ -362,7 +379,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				},
 				QueryFn: func(ctx context.Context, query string, args ...any) (pglib.Rows, error) {
 					switch query {
-					case "SELECT schema_name FROM information_schema.schemata WHERE schema_name != $1":
+					case fmt.Sprintf(selectSchemasQuery, "$1"):
 						require.Equal(t, []any{testSchema}, args)
 						return &mocks.Rows{
 							CloseFn: func() {},
@@ -376,7 +393,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 							},
 							ErrFn: func() error { return nil },
 						}, nil
-					case "SELECT tablename FROM pg_tables WHERE schemaname = 'test_schema' AND tablename NOT IN ($1)":
+					case fmt.Sprintf(selectTablesQuery, testSchema, "$1"):
 						require.Equal(t, []any{testTable}, args)
 						return &mocks.Rows{
 							CloseFn: func() {},
@@ -407,8 +424,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		{
 			name: "error - dumping schema",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{testTable},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable},
+				},
 			},
 			conn: validQuerier(),
 			pgdumpFn: newMockPgdump(func(_ context.Context, i uint, po pglib.PGDumpOptions) ([]byte, error) {
@@ -428,8 +446,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		{
 			name: "error - dumping sequence values",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{testTable},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable},
+				},
 			},
 			conn: validQuerier(),
 			pgdumpFn: newMockPgdump(func(_ context.Context, i uint, po pglib.PGDumpOptions) ([]byte, error) {
@@ -451,8 +470,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		{
 			name: "error - restoring schema dump",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{testTable},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable},
+				},
 			},
 			conn: validQuerier(),
 			pgdumpFn: func(_ context.Context, po pglib.PGDumpOptions) ([]byte, error) {
@@ -467,8 +487,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		{
 			name: "error - restoring filtered dump",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{testTable},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable},
+				},
 			},
 			conn: validQuerier(),
 			pgdumpFn: newMockPgdump(func(_ context.Context, i uint, po pglib.PGDumpOptions) ([]byte, error) {
@@ -509,8 +530,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 			generator: &generatormocks.Generator{
 				CreateSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
 					require.Equal(t, &snapshot.Snapshot{
-						SchemaName: testSchema,
-						TableNames: []string{testTable},
+						SchemaTables: map[string][]string{
+							testSchema: {testTable},
+						},
 					}, ss)
 					return errors.New("CreateSnapshotFn: should not be called")
 				},
@@ -521,8 +543,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		{
 			name: "error - calling internal generator CreateSnapshot",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{testTable},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable},
+				},
 			},
 			conn: validQuerier(),
 			pgdumpFn: newMockPgdump(func(_ context.Context, i uint, po pglib.PGDumpOptions) ([]byte, error) {
@@ -571,8 +594,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		{
 			name: "error - performing pgrestore output critical error",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{testTable},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable},
+				},
 			},
 			conn: validQuerier(),
 			pgdumpFn: func(_ context.Context, po pglib.PGDumpOptions) ([]byte, error) {
@@ -587,8 +611,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		{
 			name: "error - performing pgrestore output ignored errors",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{testTable},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable},
+				},
 			},
 			conn: validQuerier(),
 			pgdumpFn: func(_ context.Context, po pglib.PGDumpOptions) ([]byte, error) {
@@ -603,8 +628,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		{
 			name: "error - getting target conn",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{testTable},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable},
+				},
 			},
 			connBuilder: func(ctx context.Context, s string) (pglib.Querier, error) {
 				if s == "target-url" {
@@ -624,8 +650,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		{
 			name: "error - creating schema",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{testTable},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable},
+				},
 			},
 			conn: func() *mocks.Querier {
 				c := validQuerier()
@@ -646,8 +673,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 		{
 			name: "error - schemalog insertion fails",
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{testTable},
+				SchemaTables: map[string][]string{
+					testSchema: {testTable},
+				},
 			},
 			conn: validQuerier(),
 			pgdumpFn: newMockPgdump(func(_ context.Context, i uint, po pglib.PGDumpOptions) ([]byte, error) {
@@ -688,6 +716,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				logger:                 log.NewNoopLogger(),
 				includeGlobalDBObjects: true,
 				generator:              tc.generator,
+				role:                   tc.role,
 			}
 
 			if tc.connBuilder != nil {
diff --git a/pkg/snapshot/generator/postgres/schema/schemalog/snapshot_schemalog_generator_test.go b/pkg/snapshot/generator/postgres/schema/schemalog/snapshot_schemalog_generator_test.go
index 0c0ec7b..f104f27 100644
--- a/pkg/snapshot/generator/postgres/schema/schemalog/snapshot_schemalog_generator_test.go
+++ b/pkg/snapshot/generator/postgres/schema/schemalog/snapshot_schemalog_generator_test.go
@@ -24,8 +24,9 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 	testSchemaName := "test-schema"
 	testTable := "test-table"
 	testSnapshot := &snapshot.Snapshot{
-		SchemaName: testSchemaName,
-		TableNames: []string{testTable},
+		SchemaTables: map[string][]string{
+			testSchemaName: {testTable},
+		},
 	}
 
 	now := schemalog.NewSchemaCreatedAtTimestamp(time.Now())
@@ -123,7 +124,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				return errors.New("processRow: should not be called")
 			},
 
-			wantErr: snapshot.NewErrors(errTest),
+			wantErr: snapshot.NewErrors(testSchemaName, errTest),
 		},
 		{
 			name: "error - processing schema row",
@@ -137,7 +138,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				return errTest
 			},
 
-			wantErr: snapshot.NewErrors(errTest),
+			wantErr: snapshot.NewErrors(testSchemaName, errTest),
 		},
 		{
 			name: "error - converting log entry to row",
@@ -151,7 +152,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 			},
 			marshaler: func(a any) ([]byte, error) { return nil, errTest },
 
-			wantErr: snapshot.NewErrors(fmt.Errorf("marshaling log entry schema into json: %w", errTest)),
+			wantErr: snapshot.NewErrors(testSchemaName, fmt.Errorf("marshaling log entry schema into json: %w", errTest)),
 		},
 	}
 
@@ -172,7 +173,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 			}
 
 			err := g.CreateSnapshot(context.Background(), testSnapshot)
-			require.Equal(t, err, tc.wantErr)
+			require.Equal(t, tc.wantErr, err)
 
 			if tc.generator != nil {
 				require.Equal(t, tc.wantGeneratorCalls, tc.generator.CreateSnapshotCalls())
diff --git a/pkg/snapshot/generator/postgres/tablefinder/pg_snapshot_table_finder_test.go b/pkg/snapshot/generator/postgres/tablefinder/pg_snapshot_table_finder_test.go
index 9be99f0..69244ba 100644
--- a/pkg/snapshot/generator/postgres/tablefinder/pg_snapshot_table_finder_test.go
+++ b/pkg/snapshot/generator/postgres/tablefinder/pg_snapshot_table_finder_test.go
@@ -23,8 +23,9 @@ func TestSnapshotTableFinder_CreateSnapshot(t *testing.T) {
 
 	newTestSnapshot := func(tables []string) *snapshot.Snapshot {
 		return &snapshot.Snapshot{
-			SchemaName: testSchema,
-			TableNames: tables,
+			SchemaTables: map[string][]string{
+				testSchema: tables,
+			},
 		}
 	}
 
@@ -70,8 +71,9 @@ func TestSnapshotTableFinder_CreateSnapshot(t *testing.T) {
 				},
 			},
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{wildcard},
+				SchemaTables: map[string][]string{
+					testSchema: {wildcard},
+				},
 			},
 			generator: &mocks.Generator{
 				CreateSnapshotFn: func(ctx context.Context, snapshot *snapshot.Snapshot) error {
@@ -103,8 +105,9 @@ func TestSnapshotTableFinder_CreateSnapshot(t *testing.T) {
 				},
 			},
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{"table-a", "table-b", wildcard},
+				SchemaTables: map[string][]string{
+					testSchema: {"table-a", "table-b", wildcard},
+				},
 			},
 			generator: &mocks.Generator{
 				CreateSnapshotFn: func(ctx context.Context, snapshot *snapshot.Snapshot) error {
@@ -125,8 +128,9 @@ func TestSnapshotTableFinder_CreateSnapshot(t *testing.T) {
 				},
 			},
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{"table-a", "table-b", wildcard},
+				SchemaTables: map[string][]string{
+					testSchema: {"table-a", "table-b", wildcard},
+				},
 			},
 			generator: &mocks.Generator{
 				CreateSnapshotFn: func(ctx context.Context, snapshot *snapshot.Snapshot) error {
@@ -153,8 +157,9 @@ func TestSnapshotTableFinder_CreateSnapshot(t *testing.T) {
 				},
 			},
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{"table-a", "table-b", wildcard},
+				SchemaTables: map[string][]string{
+					testSchema: {"table-a", "table-b", wildcard},
+				},
 			},
 			generator: &mocks.Generator{
 				CreateSnapshotFn: func(ctx context.Context, snapshot *snapshot.Snapshot) error {
@@ -178,8 +183,9 @@ func TestSnapshotTableFinder_CreateSnapshot(t *testing.T) {
 				},
 			},
 			snapshot: &snapshot.Snapshot{
-				SchemaName: testSchema,
-				TableNames: []string{"table-a", "table-b", wildcard},
+				SchemaTables: map[string][]string{
+					testSchema: {"table-a", "table-b", wildcard},
+				},
 			},
 			generator: &mocks.Generator{
 				CreateSnapshotFn: func(ctx context.Context, snapshot *snapshot.Snapshot) error {
diff --git a/pkg/snapshot/generator/snapshot_generator_recorder_test.go b/pkg/snapshot/generator/snapshot_generator_recorder_test.go
index ed8eba7..d9f962f 100644
--- a/pkg/snapshot/generator/snapshot_generator_recorder_test.go
+++ b/pkg/snapshot/generator/snapshot_generator_recorder_test.go
@@ -17,9 +17,12 @@ import (
 func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 	t.Parallel()
 
+	testSchema := "test-schema"
+	testTables := []string{"table1", "table2"}
 	testSnapshot := snapshot.Snapshot{
-		SchemaName: "test-schema",
-		TableNames: []string{"table1", "table2"},
+		SchemaTables: map[string][]string{
+			testSchema: testTables,
+		},
 	}
 
 	newTestSnapshot := func() *snapshot.Snapshot {
@@ -45,7 +48,8 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 				},
 				CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
 					require.Equal(t, &snapshot.Request{
-						Snapshot: testSnapshot,
+						Schema: testSchema,
+						Tables: testTables,
 					}, r)
 					return nil
 				},
@@ -53,14 +57,16 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 					switch i {
 					case 1:
 						require.Equal(t, &snapshot.Request{
-							Snapshot: testSnapshot,
-							Status:   snapshot.StatusInProgress,
+							Schema: testSchema,
+							Tables: testTables,
+							Status: snapshot.StatusInProgress,
 						}, r)
 						return nil
 					case 2:
 						require.Equal(t, &snapshot.Request{
-							Snapshot: testSnapshot,
-							Status:   snapshot.StatusCompleted,
+							Schema: testSchema,
+							Tables: testTables,
+							Status: snapshot.StatusCompleted,
 						}, r)
 						return nil
 					default:
@@ -83,7 +89,11 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 			store: &snapshotstoremocks.Store{
 				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
 					return []*snapshot.Request{
-						{Snapshot: testSnapshot, Status: snapshot.StatusCompleted},
+						{
+							Schema: testSchema,
+							Tables: testTables,
+							Status: snapshot.StatusCompleted,
+						},
 					}, nil
 				},
 			},
@@ -100,7 +110,7 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 			},
 			generator: &mockGenerator{},
 
-			wantErr: snapshot.NewErrors(fmt.Errorf("retrieving existing snapshots for schema: %w", errTest)),
+			wantErr: errTest,
 		},
 		{
 			name: "error - snapshot error on wrapped generator",
@@ -110,7 +120,8 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 				},
 				CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
 					require.Equal(t, &snapshot.Request{
-						Snapshot: testSnapshot,
+						Schema: testSchema,
+						Tables: testTables,
 					}, r)
 					return nil
 				},
@@ -118,15 +129,17 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 					switch i {
 					case 1:
 						require.Equal(t, &snapshot.Request{
-							Snapshot: testSnapshot,
-							Status:   snapshot.StatusInProgress,
+							Schema: testSchema,
+							Tables: testTables,
+							Status: snapshot.StatusInProgress,
 						}, r)
 						return nil
 					case 2:
 						require.Equal(t, &snapshot.Request{
-							Snapshot: testSnapshot,
-							Status:   snapshot.StatusCompleted,
-							Errors:   snapshot.NewErrors(errTest),
+							Schema: testSchema,
+							Tables: testTables,
+							Status: snapshot.StatusCompleted,
+							Errors: snapshot.NewSchemaErrors(testSchema, errTest),
 						}, r)
 						return nil
 					default:
@@ -137,11 +150,11 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 			},
 			generator: &mockGenerator{
 				createSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
-					return snapshot.NewErrors(errTest)
+					return errTest
 				},
 			},
 
-			wantErr: snapshot.NewErrors(errTest),
+			wantErr: errTest,
 		},
 		{
 			name: "error - recording snapshot request",
@@ -159,7 +172,7 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 				},
 			},
 
-			wantErr: snapshot.NewErrors(errTest),
+			wantErr: errTest,
 		},
 		{
 			name: "error - updating snapshot request in progress",
@@ -174,8 +187,9 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 					switch i {
 					case 1:
 						require.Equal(t, &snapshot.Request{
-							Snapshot: testSnapshot,
-							Status:   snapshot.StatusInProgress,
+							Schema: testSchema,
+							Tables: testTables,
+							Status: snapshot.StatusInProgress,
 						}, r)
 						return errTest
 					default:
@@ -186,7 +200,7 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 			},
 			generator: &mockGenerator{},
 
-			wantErr: snapshot.NewErrors(errTest),
+			wantErr: errTest,
 		},
 		{
 			name: "error - updating snapshot request completed without errors",
@@ -196,7 +210,8 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 				},
 				CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
 					require.Equal(t, &snapshot.Request{
-						Snapshot: testSnapshot,
+						Schema: testSchema,
+						Tables: testTables,
 					}, r)
 					return nil
 				},
@@ -204,14 +219,16 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 					switch i {
 					case 1:
 						require.Equal(t, &snapshot.Request{
-							Snapshot: testSnapshot,
-							Status:   snapshot.StatusInProgress,
+							Schema: testSchema,
+							Tables: testTables,
+							Status: snapshot.StatusInProgress,
 						}, r)
 						return nil
 					case 2:
 						require.Equal(t, &snapshot.Request{
-							Snapshot: testSnapshot,
-							Status:   snapshot.StatusCompleted,
+							Schema: testSchema,
+							Tables: testTables,
+							Status: snapshot.StatusCompleted,
 						}, r)
 						return errTest
 					default:
@@ -227,7 +244,7 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 				},
 			},
 
-			wantErr: snapshot.NewErrors(errTest),
+			wantErr: errTest,
 		},
 		{
 			name: "error - updating snapshot request completed with snapshot and table errors",
@@ -237,7 +254,8 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 				},
 				CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
 					require.Equal(t, &snapshot.Request{
-						Snapshot: testSnapshot,
+						Schema: testSchema,
+						Tables: testTables,
 					}, r)
 					return nil
 				},
@@ -245,18 +263,21 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 					switch i {
 					case 1:
 						require.Equal(t, &snapshot.Request{
-							Snapshot: testSnapshot,
-							Status:   snapshot.StatusInProgress,
+							Schema: testSchema,
+							Tables: testTables,
+							Status: snapshot.StatusInProgress,
 						}, r)
 						return nil
 					case 2:
 						require.Equal(t, &snapshot.Request{
-							Snapshot: testSnapshot,
-							Status:   snapshot.StatusCompleted,
-							Errors: &snapshot.Errors{
-								SnapshotErrMsgs: []string{errTest.Error()},
-								Tables: []snapshot.TableError{
-									{Table: "table1", ErrorMsg: errTest.Error()},
+							Schema: testSchema,
+							Tables: testTables,
+							Status: snapshot.StatusCompleted,
+							Errors: &snapshot.SchemaErrors{
+								Schema:       testSchema,
+								GlobalErrors: []string{errTest.Error()},
+								TableErrors: map[string]string{
+									"table1": errTest.Error(),
 								},
 							},
 						}, r)
@@ -271,18 +292,24 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 				createSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
 					require.Equal(t, newTestSnapshot(), ss)
 					return &snapshot.Errors{
-						SnapshotErrMsgs: []string{errTest.Error()},
-						Tables: []snapshot.TableError{
-							{Table: "table1", ErrorMsg: errTest.Error()},
+						testSchema: &snapshot.SchemaErrors{
+							Schema:       testSchema,
+							GlobalErrors: []string{errTest.Error()},
+							TableErrors: map[string]string{
+								"table1": errTest.Error(),
+							},
 						},
 					}
 				},
 			},
 
 			wantErr: &snapshot.Errors{
-				SnapshotErrMsgs: []string{errTest.Error(), updateErr.Error()},
-				Tables: []snapshot.TableError{
-					{Table: "table1", ErrorMsg: errTest.Error()},
+				testSchema: &snapshot.SchemaErrors{
+					Schema:       testSchema,
+					GlobalErrors: []string{errTest.Error(), updateErr.Error()},
+					TableErrors: map[string]string{
+						"table1": errTest.Error(),
+					},
 				},
 			},
 		},
@@ -294,7 +321,8 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 				},
 				CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
 					require.Equal(t, &snapshot.Request{
-						Snapshot: testSnapshot,
+						Schema: testSchema,
+						Tables: testTables,
 					}, r)
 					return nil
 				},
@@ -302,17 +330,20 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 					switch i {
 					case 1:
 						require.Equal(t, &snapshot.Request{
-							Snapshot: testSnapshot,
-							Status:   snapshot.StatusInProgress,
+							Schema: testSchema,
+							Tables: testTables,
+							Status: snapshot.StatusInProgress,
 						}, r)
 						return nil
 					case 2:
 						require.Equal(t, &snapshot.Request{
-							Snapshot: testSnapshot,
-							Status:   snapshot.StatusCompleted,
-							Errors: &snapshot.Errors{
-								Tables: []snapshot.TableError{
-									{Table: "table1", ErrorMsg: errTest.Error()},
+							Schema: testSchema,
+							Tables: testTables,
+							Status: snapshot.StatusCompleted,
+							Errors: &snapshot.SchemaErrors{
+								Schema: testSchema,
+								TableErrors: map[string]string{
+									"table1": errTest.Error(),
 								},
 							},
 						}, r)
@@ -327,17 +358,23 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 				createSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
 					require.Equal(t, newTestSnapshot(), ss)
 					return &snapshot.Errors{
-						Tables: []snapshot.TableError{
-							{Table: "table1", ErrorMsg: errTest.Error()},
+						testSchema: &snapshot.SchemaErrors{
+							Schema: testSchema,
+							TableErrors: map[string]string{
+								"table1": errTest.Error(),
+							},
 						},
 					}
 				},
 			},
 
 			wantErr: &snapshot.Errors{
-				SnapshotErrMsgs: []string{updateErr.Error()},
-				Tables: []snapshot.TableError{
-					{Table: "table1", ErrorMsg: errTest.Error()},
+				testSchema: &snapshot.SchemaErrors{
+					Schema:       testSchema,
+					GlobalErrors: []string{updateErr.Error()},
+					TableErrors: map[string]string{
+						"table1": errTest.Error(),
+					},
 				},
 			},
 		},
@@ -351,7 +388,9 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 			defer sr.Close()
 
 			err := sr.CreateSnapshot(context.Background(), newTestSnapshot())
-			require.Equal(t, tc.wantErr, err)
+			if !errors.Is(err, tc.wantErr) {
+				require.Equal(t, tc.wantErr, err)
+			}
 		})
 	}
 }
@@ -369,7 +408,7 @@ func TestSnapshotRecorder_filterOutExistingSnapshots(t *testing.T) {
 		tables              []string
 		repeatableSnapshots bool
 
-		wantTables []string
+		wantTables map[string][]string
 		wantErr    error
 	}{
 		{
@@ -379,7 +418,9 @@ func TestSnapshotRecorder_filterOutExistingSnapshots(t *testing.T) {
 					return []*snapshot.Request{}, nil
 				},
 			},
-			wantTables: testTables,
+			wantTables: map[string][]string{
+				testSchema: testTables,
+			},
 		},
 		{
 			name: "ok - no existing snapshots with wildcard",
@@ -387,17 +428,17 @@ func TestSnapshotRecorder_filterOutExistingSnapshots(t *testing.T) {
 				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
 					return []*snapshot.Request{
 						{
-							Snapshot: snapshot.Snapshot{
-								SchemaName: testSchema,
-								TableNames: []string{"table2"},
-							},
+							Schema: testSchema,
+							Tables: []string{"table2"},
 							Status: snapshot.StatusCompleted,
 						},
 					}, nil
 				},
 			},
-			tables:     []string{"*"},
-			wantTables: []string{"*"},
+			tables: []string{"*"},
+			wantTables: map[string][]string{
+				testSchema: {"*"},
+			},
 		},
 		{
 			name: "ok - existing snapshots",
@@ -405,16 +446,16 @@ func TestSnapshotRecorder_filterOutExistingSnapshots(t *testing.T) {
 				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
 					return []*snapshot.Request{
 						{
-							Snapshot: snapshot.Snapshot{
-								SchemaName: testSchema,
-								TableNames: []string{"table2"},
-							},
+							Schema: testSchema,
+							Tables: []string{"table2"},
 							Status: snapshot.StatusCompleted,
 						},
 					}, nil
 				},
 			},
-			wantTables: []string{"table1"},
+			wantTables: map[string][]string{
+				testSchema: {"table1"},
+			},
 		},
 		{
 			name: "ok - existing snapshots with repeatable snapshots",
@@ -424,7 +465,9 @@ func TestSnapshotRecorder_filterOutExistingSnapshots(t *testing.T) {
 				},
 			},
 			repeatableSnapshots: true,
-			wantTables:          []string{"table1", "table2"},
+			wantTables: map[string][]string{
+				testSchema: {"table1", "table2"},
+			},
 		},
 		{
 			name: "ok - existing wildcard snapshot with wildcard table",
@@ -432,17 +475,15 @@ func TestSnapshotRecorder_filterOutExistingSnapshots(t *testing.T) {
 				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
 					return []*snapshot.Request{
 						{
-							Snapshot: snapshot.Snapshot{
-								SchemaName: testSchema,
-								TableNames: []string{"*"},
-							},
+							Schema: testSchema,
+							Tables: []string{"*"},
 							Status: snapshot.StatusCompleted,
 						},
 					}, nil
 				},
 			},
 			tables:     []string{"*"},
-			wantTables: []string{},
+			wantTables: map[string][]string{},
 		},
 		{
 			name: "ok - existing wildcard snapshot",
@@ -450,16 +491,14 @@ func TestSnapshotRecorder_filterOutExistingSnapshots(t *testing.T) {
 				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
 					return []*snapshot.Request{
 						{
-							Snapshot: snapshot.Snapshot{
-								SchemaName: testSchema,
-								TableNames: []string{"*"},
-							},
+							Schema: testSchema,
+							Tables: []string{"*"},
 							Status: snapshot.StatusCompleted,
 						},
 					}, nil
 				},
 			},
-			wantTables: []string{},
+			wantTables: map[string][]string{},
 		},
 		{
 			name: "ok - existing failed snapshots",
@@ -467,17 +506,17 @@ func TestSnapshotRecorder_filterOutExistingSnapshots(t *testing.T) {
 				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
 					return []*snapshot.Request{
 						{
-							Snapshot: snapshot.Snapshot{
-								SchemaName: testSchema,
-								TableNames: []string{"table2"},
-							},
-							Errors: snapshot.NewErrors(errTest),
+							Schema: testSchema,
+							Tables: []string{"table2"},
+							Errors: snapshot.NewSchemaErrors(testSchema, errTest),
 							Status: snapshot.StatusCompleted,
 						},
 					}, nil
 				},
 			},
-			wantTables: []string{"table1", "table2"},
+			wantTables: map[string][]string{
+				testSchema: {"table1", "table2"},
+			},
 		},
 		{
 			name: "ok - existing failed wildcard snapshot",
@@ -485,18 +524,23 @@ func TestSnapshotRecorder_filterOutExistingSnapshots(t *testing.T) {
 				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
 					return []*snapshot.Request{
 						{
-							Snapshot: snapshot.Snapshot{
-								SchemaName: testSchema,
-								TableNames: []string{"*"},
+							Schema: testSchema,
+							Tables: []string{"*"},
+							Errors: &snapshot.SchemaErrors{
+								Schema: testSchema,
+								TableErrors: map[string]string{
+									"table2": errTest.Error(),
+								},
 							},
-							Errors: &snapshot.Errors{Tables: []snapshot.TableError{{Table: "table2", ErrorMsg: errTest.Error()}}},
 							Status: snapshot.StatusCompleted,
 						},
 					}, nil
 				},
 			},
-			tables:     []string{"*"},
-			wantTables: []string{"table2"},
+			tables: []string{"*"},
+			wantTables: map[string][]string{
+				testSchema: {"table2"},
+			},
 		},
 		{
 			name: "ok - existing failed table on wildcard snapshot",
@@ -504,17 +548,22 @@ func TestSnapshotRecorder_filterOutExistingSnapshots(t *testing.T) {
 				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
 					return []*snapshot.Request{
 						{
-							Snapshot: snapshot.Snapshot{
-								SchemaName: testSchema,
-								TableNames: []string{"*"},
+							Schema: testSchema,
+							Tables: []string{"*"},
+							Errors: &snapshot.SchemaErrors{
+								Schema: testSchema,
+								TableErrors: map[string]string{
+									"table2": errTest.Error(),
+								},
 							},
-							Errors: &snapshot.Errors{Tables: []snapshot.TableError{{Table: "table2", ErrorMsg: errTest.Error()}}},
 							Status: snapshot.StatusCompleted,
 						},
 					}, nil
 				},
 			},
-			wantTables: []string{"table2"},
+			wantTables: map[string][]string{
+				testSchema: {"table2"},
+			},
 		},
 		{
 			name: "error - retrieving existing snapshots",
@@ -542,7 +591,7 @@ func TestSnapshotRecorder_filterOutExistingSnapshots(t *testing.T) {
 				tables = tc.tables
 			}
 
-			filteredTables, err := sr.filterOutExistingSnapshots(context.Background(), testSchema, tables)
+			filteredTables, err := sr.filterOutExistingSnapshots(context.Background(), map[string][]string{testSchema: tables})
 			require.ErrorIs(t, err, tc.wantErr)
 			require.Equal(t, tc.wantTables, filteredTables)
 		})
diff --git a/pkg/snapshot/store/postgres/pg_snapshot_store_test.go b/pkg/snapshot/store/postgres/pg_snapshot_store_test.go
index bfcc0e3..7b0b5c7 100644
--- a/pkg/snapshot/store/postgres/pg_snapshot_store_test.go
+++ b/pkg/snapshot/store/postgres/pg_snapshot_store_test.go
@@ -19,12 +19,11 @@ import (
 func TestStore_CreateSnapshotRequest(t *testing.T) {
 	t.Parallel()
 
-	testSnapshot := snapshot.Snapshot{
-		SchemaName: "test-schema",
-		TableNames: []string{"test-table-1", "test-table-2"},
-	}
+	testSchema := "test-schema"
+	testTables := []string{"test-table-1", "test-table-2"}
 	testSnapshotRequest := &snapshot.Request{
-		Snapshot: testSnapshot,
+		Schema: testSchema,
+		Tables: testTables,
 	}
 
 	errTest := errors.New("oh noes")
@@ -42,7 +41,7 @@ func TestStore_CreateSnapshotRequest(t *testing.T) {
 					wantQuery := fmt.Sprintf(`INSERT INTO %s (schema_name, table_names, created_at, updated_at, status)
 	VALUES($1, $2, now(), now(),'requested')`, snapshotsTable())
 					require.Equal(t, wantQuery, s)
-					wantAttr := []any{testSnapshot.SchemaName, pq.StringArray(testSnapshot.TableNames)}
+					wantAttr := []any{testSchema, pq.StringArray(testTables)}
 					require.Equal(t, wantAttr, a)
 					return postgres.CommandTag{}, nil
 				},
@@ -78,17 +77,16 @@ func TestStore_CreateSnapshotRequest(t *testing.T) {
 func TestStore_UpdateSnapshotRequest(t *testing.T) {
 	t.Parallel()
 
-	testSnapshot := snapshot.Snapshot{
-		SchemaName: "test-schema",
-		TableNames: []string{"test-table-1", "test-table-2"},
-	}
+	testSchema := "test-schema"
+	testTables := []string{"test-table-1", "test-table-2"}
 	testSnapshotRequest := snapshot.Request{
-		Snapshot: testSnapshot,
-		Status:   snapshot.StatusInProgress,
+		Schema: testSchema,
+		Tables: testTables,
+		Status: snapshot.StatusInProgress,
 	}
 
 	errTest := errors.New("oh noes")
-	testSnapshotErr := snapshot.NewErrors(errTest)
+	testSchemaErr := snapshot.NewSchemaErrors(testSchema, errTest)
 
 	tests := []struct {
 		name    string
@@ -105,7 +103,7 @@ func TestStore_UpdateSnapshotRequest(t *testing.T) {
 	WHERE schema_name = $3 and table_names = $4 and status != 'completed'`,
 						snapshotsTable())
 					require.Equal(t, wantQuery, s)
-					require.Equal(t, []any{snapshot.StatusInProgress, (*snapshot.Errors)(nil), testSnapshot.SchemaName, pq.StringArray(testSnapshot.TableNames)}, a)
+					require.Equal(t, []any{snapshot.StatusInProgress, (*snapshot.SchemaErrors)(nil), testSchema, pq.StringArray(testTables)}, a)
 					return postgres.CommandTag{}, nil
 				},
 			},
@@ -121,13 +119,13 @@ func TestStore_UpdateSnapshotRequest(t *testing.T) {
 	WHERE schema_name = $3 and table_names = $4 and status != 'completed'`,
 						snapshotsTable())
 					require.Equal(t, wantQuery, s)
-					require.Equal(t, []any{snapshot.StatusInProgress, testSnapshotErr, testSnapshot.SchemaName, pq.StringArray(testSnapshot.TableNames)}, a)
+					require.Equal(t, []any{snapshot.StatusInProgress, testSchemaErr, testSchema, pq.StringArray(testTables)}, a)
 					return postgres.CommandTag{}, nil
 				},
 			},
 			req: func() *snapshot.Request {
 				s := testSnapshotRequest
-				s.Errors = testSnapshotErr
+				s.Errors = testSchemaErr
 				return &s
 			}(),
 
@@ -162,15 +160,14 @@ func TestStore_UpdateSnapshotRequest(t *testing.T) {
 func TestStore_GetSnapshotRequestsBySchema(t *testing.T) {
 	t.Parallel()
 
-	testSnapshot := snapshot.Snapshot{
-		SchemaName: "test-schema",
-		TableNames: []string{"test-table-1", "test-table-2"},
-	}
+	testSchema := "test-schema"
+	testTables := []string{"test-table-1", "test-table-2"}
 	errTest := errors.New("oh noes")
 	testSnapshotRequest := snapshot.Request{
-		Snapshot: testSnapshot,
-		Status:   snapshot.StatusInProgress,
-		Errors:   snapshot.NewErrors(errTest),
+		Schema: testSchema,
+		Tables: testTables,
+		Status: snapshot.StatusInProgress,
+		Errors: snapshot.NewSchemaErrors(testSchema, errTest),
 	}
 
 	tests := []struct {
@@ -186,7 +183,7 @@ func TestStore_GetSnapshotRequestsBySchema(t *testing.T) {
 				QueryFn: func(ctx context.Context, query string, args ...any) (postgres.Rows, error) {
 					wantQuery := fmt.Sprintf(`SELECT schema_name,table_names,status,errors FROM %s
 	WHERE schema_name = $1 ORDER BY req_id ASC LIMIT %d`, snapshotsTable(), queryLimit)
-					require.Equal(t, []any{testSnapshot.SchemaName}, args)
+					require.Equal(t, []any{testSchema}, args)
 					require.Equal(t, wantQuery, query)
 					return &postgresmocks.Rows{
 						CloseFn: func() {},
@@ -204,7 +201,7 @@ func TestStore_GetSnapshotRequestsBySchema(t *testing.T) {
 				QueryFn: func(ctx context.Context, query string, args ...any) (postgres.Rows, error) {
 					wantQuery := fmt.Sprintf(`SELECT schema_name,table_names,status,errors FROM %s
 	WHERE schema_name = $1 ORDER BY req_id ASC LIMIT %d`, snapshotsTable(), queryLimit)
-					require.Equal(t, []any{testSnapshot.SchemaName}, args)
+					require.Equal(t, []any{testSchema}, args)
 					require.Equal(t, wantQuery, query)
 					return &postgresmocks.Rows{
 						CloseFn: func() {},
@@ -213,17 +210,17 @@ func TestStore_GetSnapshotRequestsBySchema(t *testing.T) {
 							require.Len(t, dest, 4)
 							schemaName, ok := dest[0].(*string)
 							require.True(t, ok)
-							*schemaName = testSnapshot.SchemaName
+							*schemaName = testSchema
 
 							tableNames, ok := dest[1].(*[]string)
 							require.True(t, ok)
-							*tableNames = testSnapshot.TableNames
+							*tableNames = testTables
 
 							status, ok := dest[2].(*snapshot.Status)
 							require.True(t, ok)
 							*status = testSnapshotRequest.Status
 
-							errs, ok := dest[3].(**snapshot.Errors)
+							errs, ok := dest[3].(**snapshot.SchemaErrors)
 							require.True(t, ok)
 							*errs = testSnapshotRequest.Errors
 							return nil
@@ -254,7 +251,7 @@ func TestStore_GetSnapshotRequestsBySchema(t *testing.T) {
 				QueryFn: func(ctx context.Context, query string, args ...any) (postgres.Rows, error) {
 					wantQuery := fmt.Sprintf(`SELECT schema_name,table_names,status,errors FROM %s
 	WHERE schema_name = $1 ORDER BY req_id ASC LIMIT %d`, snapshotsTable(), queryLimit)
-					require.Equal(t, []any{testSnapshot.SchemaName}, args)
+					require.Equal(t, []any{testSchema}, args)
 					require.Equal(t, wantQuery, query)
 					return &postgresmocks.Rows{
 						CloseFn: func() {},
@@ -278,7 +275,7 @@ func TestStore_GetSnapshotRequestsBySchema(t *testing.T) {
 			store := &Store{
 				conn: tc.querier,
 			}
-			requests, err := store.GetSnapshotRequestsBySchema(context.Background(), testSnapshot.SchemaName)
+			requests, err := store.GetSnapshotRequestsBySchema(context.Background(), testSchema)
 			require.ErrorIs(t, err, tc.wantErr)
 			require.Equal(t, tc.wantRequests, requests)
 		})
@@ -288,15 +285,14 @@ func TestStore_GetSnapshotRequestsBySchema(t *testing.T) {
 func TestStore_GetSnapshotRequestsByStatus(t *testing.T) {
 	t.Parallel()
 
-	testSnapshot := snapshot.Snapshot{
-		SchemaName: "test-schema",
-		TableNames: []string{"test-table-1", "test-table-2"},
-	}
+	testSchema := "test-schema"
+	testTables := []string{"test-table-1", "test-table-2"}
 	errTest := errors.New("oh noes")
 	testSnapshotRequest := snapshot.Request{
-		Snapshot: testSnapshot,
-		Status:   snapshot.StatusInProgress,
-		Errors:   snapshot.NewErrors(errTest),
+		Schema: testSchema,
+		Tables: testTables,
+		Status: snapshot.StatusInProgress,
+		Errors: snapshot.NewSchemaErrors(testSchema, errTest),
 	}
 
 	tests := []struct {
@@ -337,17 +333,17 @@ func TestStore_GetSnapshotRequestsByStatus(t *testing.T) {
 							require.Len(t, dest, 4)
 							schemaName, ok := dest[0].(*string)
 							require.True(t, ok)
-							*schemaName = testSnapshot.SchemaName
+							*schemaName = testSchema
 
 							tableNames, ok := dest[1].(*[]string)
 							require.True(t, ok)
-							*tableNames = testSnapshot.TableNames
+							*tableNames = testTables
 
 							status, ok := dest[2].(*snapshot.Status)
 							require.True(t, ok)
 							*status = testSnapshotRequest.Status
 
-							errs, ok := dest[3].(**snapshot.Errors)
+							errs, ok := dest[3].(**snapshot.SchemaErrors)
 							require.True(t, ok)
 							*errs = testSnapshotRequest.Errors
 							return nil
diff --git a/pkg/wal/listener/snapshot/adapter/wal_snapshot_generator_adapter_test.go b/pkg/wal/listener/snapshot/adapter/wal_snapshot_generator_adapter_test.go
index 7d86396..53376f3 100644
--- a/pkg/wal/listener/snapshot/adapter/wal_snapshot_generator_adapter_test.go
+++ b/pkg/wal/listener/snapshot/adapter/wal_snapshot_generator_adapter_test.go
@@ -31,8 +31,9 @@ func TestSnapshotGeneratorAdapter_CreateSnapshot(t *testing.T) {
 			generator: &generatormocks.Generator{
 				CreateSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
 					require.Equal(t, &snapshot.Snapshot{
-						SchemaName: publicSchema,
-						TableNames: []string{"*"},
+						SchemaTables: map[string][]string{
+							publicSchema: {"*"},
+						},
 					}, ss)
 					return nil
 				},
@@ -63,10 +64,9 @@ func TestSnapshotGeneratorAdapter_CreateSnapshot(t *testing.T) {
 			t.Parallel()
 
 			ga := SnapshotGeneratorAdapter{
-				logger:          log.NewNoopLogger(),
-				generator:       tc.generator,
-				schemaTables:    tc.schemaTables,
-				snapshotWorkers: 1,
+				logger:       log.NewNoopLogger(),
+				generator:    tc.generator,
+				schemaTables: tc.schemaTables,
 			}
 			defer ga.Close()