Add auto-parallelism for RDS refresh dump and restore

Resolve optimal -j values automatically: use EC2 DescribeInstanceTypes
to determine vCPU count of the RDS clone (for pg_dump parallelism) and
runtime.NumCPU for the local machine (for pg_restore parallelism).
Pass both values through the existing ConfigProjection when updating
DBLab config during refresh.

https://claude.ai/code/session_01AhnBVCBWjk24T7BBQtmkbq

Author: claude

engine/go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/aws/aws-sdk-go v1.55.8
 	github.com/aws/aws-sdk-go-v2 v1.41.5
 	github.com/aws/aws-sdk-go-v2/config v1.32.14
+	github.com/aws/aws-sdk-go-v2/service/ec2 v1.297.0
 	github.com/aws/aws-sdk-go-v2/service/rds v1.117.1
 	github.com/containerd/errdefs v1.0.0
 	github.com/docker/cli v28.5.2+incompatible

engine/go.sum

@@ -28,6 +28,8 @@ github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.21 h1:PEgGVtPoB6NTpPrBgq
 github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.21/go.mod h1:p+hz+PRAYlY3zcpJhPwXlLC4C+kqn70WIHwnzAfs6ps=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.6 h1:qYQ4pzQ2Oz6WpQ8T3HvGHnZydA72MnLuFK9tJwmrbHw=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.6/go.mod h1:O3h0IK87yXci+kg6flUKzJnWeziQUKciKrLjcatSNcY=
+github.com/aws/aws-sdk-go-v2/service/ec2 v1.297.0 h1:A+7NViqbMUCoTQFWjbSXdbzE4K5Ziu2zWJtZzAusm+A=
+github.com/aws/aws-sdk-go-v2/service/ec2 v1.297.0/go.mod h1:R+2BNtUfTfhPY0RH18oL02q116bakeBWjanrbnVBqkM=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.7 h1:5EniKhLZe4xzL7a+fU3C2tfUN4nWIqlLesfrjkuPFTY=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.7/go.mod h1:x0nZssQ3qZSnIcePWLvcoFisRXJzcTVvYpAAdYX8+GI=
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21 h1:c31//R3xgIJMSC8S6hEVq+38DcvUlgFY0FM6mSI5oto=

engine/internal/rdsrefresh/dblab.go

@@ -180,6 +180,10 @@ type SourceConfigUpdate struct {
 	Password string
 	// RDSIAMDBInstance is the RDS DB instance identifier for IAM auth. When empty, this field is omitted from the config update.
 	RDSIAMDBInstance string
+	// DumpParallelJobs sets the -j flag for pg_dump. When zero, the existing value is preserved.
+	DumpParallelJobs int
+	// RestoreParallelJobs sets the -j flag for pg_restore. When zero, the existing value is preserved.
+	RestoreParallelJobs int
 }
 
 // UpdateSourceConfig updates the source database connection in DBLab config.
@@ -198,6 +202,16 @@ func (c *DBLabClient) UpdateSourceConfig(ctx context.Context, update SourceConfi
 		proj.RDSIAMDBInstance = &update.RDSIAMDBInstance
 	}
 
+	if update.DumpParallelJobs > 0 {
+		dumpJobs := int64(update.DumpParallelJobs)
+		proj.DumpParallelJobs = &dumpJobs
+	}
+
+	if update.RestoreParallelJobs > 0 {
+		restoreJobs := int64(update.RestoreParallelJobs)
+		proj.RestoreParallelJobs = &restoreJobs
+	}
+
 	nested := map[string]interface{}{}
 
 	// defensive error check: StoreJSON only fails if target is not an addressable struct,

engine/internal/rdsrefresh/dblab_test.go

@@ -191,6 +191,70 @@ func TestDBLabClientUpdateSourceConfig(t *testing.T) {
 		assert.Nil(t, receivedConfig.RDSIAMDBInstance)
 	})
 
+	t.Run("successful with parallelism settings", func(t *testing.T) {
+		var receivedConfig models.ConfigProjection
+
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			var nested map[string]interface{}
+			err := json.NewDecoder(r.Body).Decode(&nested)
+			require.NoError(t, err)
+
+			err = projection.LoadJSON(&receivedConfig, nested, projection.LoadOptions{
+				Groups: []string{"default", "sensitive"},
+			})
+			require.NoError(t, err)
+
+			w.WriteHeader(http.StatusOK)
+		}))
+		defer server.Close()
+
+		client, err := NewDBLabClient(&DBLabConfig{APIEndpoint: server.URL, Token: "test-token"})
+		require.NoError(t, err)
+
+		err = client.UpdateSourceConfig(context.Background(), SourceConfigUpdate{
+			Host: "clone-host.rds.amazonaws.com", Port: 5432, DBName: "postgres",
+			Username: "dbuser", Password: "dbpass",
+			DumpParallelJobs: 4, RestoreParallelJobs: 8,
+		})
+		require.NoError(t, err)
+
+		require.NotNil(t, receivedConfig.DumpParallelJobs)
+		assert.Equal(t, int64(4), *receivedConfig.DumpParallelJobs)
+		require.NotNil(t, receivedConfig.RestoreParallelJobs)
+		assert.Equal(t, int64(8), *receivedConfig.RestoreParallelJobs)
+	})
+
+	t.Run("omits parallelism when zero", func(t *testing.T) {
+		var receivedConfig models.ConfigProjection
+
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			var nested map[string]interface{}
+			err := json.NewDecoder(r.Body).Decode(&nested)
+			require.NoError(t, err)
+
+			err = projection.LoadJSON(&receivedConfig, nested, projection.LoadOptions{
+				Groups: []string{"default", "sensitive"},
+			})
+			require.NoError(t, err)
+
+			w.WriteHeader(http.StatusOK)
+		}))
+		defer server.Close()
+
+		client, err := NewDBLabClient(&DBLabConfig{APIEndpoint: server.URL, Token: "test-token"})
+		require.NoError(t, err)
+
+		err = client.UpdateSourceConfig(context.Background(), SourceConfigUpdate{
+			Host: "host.rds.amazonaws.com", Port: 5432, DBName: "postgres",
+			Username: "dbuser", Password: "dbpass",
+			DumpParallelJobs: 0, RestoreParallelJobs: 0,
+		})
+		require.NoError(t, err)
+
+		assert.Nil(t, receivedConfig.DumpParallelJobs)
+		assert.Nil(t, receivedConfig.RestoreParallelJobs)
+	})
+
 	t.Run("error on non-2xx status", func(t *testing.T) {
 		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 			w.WriteHeader(http.StatusBadRequest)

engine/internal/rdsrefresh/parallelism.go

@@ -0,0 +1,139 @@
+/*
+2026 © PostgresAI
+*/
+
+package rdsrefresh
+
+import (
+	"context"
+	"fmt"
+	"runtime"
+	"strings"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awsconfig "github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/service/ec2"
+	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
+
+	"gitlab.com/postgres-ai/database-lab/v3/pkg/log"
+)
+
+const (
+	// rdsInstanceClassPrefix is stripped to derive the EC2 instance type.
+	rdsInstanceClassPrefix = "db."
+
+	// minParallelJobs is the minimum parallelism level.
+	minParallelJobs = 1
+)
+
+// EC2API defines the interface for EC2 client operations used for vCPU lookup.
+type EC2API interface {
+	DescribeInstanceTypes(ctx context.Context, params *ec2.DescribeInstanceTypesInput, optFns ...func(*ec2.Options)) (*ec2.DescribeInstanceTypesOutput, error)
+}
+
+// ParallelismConfig holds the computed parallelism levels for dump and restore.
+type ParallelismConfig struct {
+	DumpJobs    int
+	RestoreJobs int
+}
+
+// ResolveParallelism determines the optimal parallelism levels for pg_dump and pg_restore.
+// dump parallelism is based on the vCPU count of the RDS clone instance class.
+// restore parallelism is based on the vCPU count of the local machine.
+func ResolveParallelism(ctx context.Context, cfg *Config) (*ParallelismConfig, error) {
+	dumpJobs, err := resolveRDSInstanceVCPUs(ctx, cfg)
+	if err != nil {
+		return nil, fmt.Errorf("failed to resolve RDS instance vCPUs: %w", err)
+	}
+
+	restoreJobs := resolveLocalVCPUs()
+
+	log.Msg("auto-parallelism: dump jobs =", dumpJobs, "(RDS clone vCPUs), restore jobs =", restoreJobs, "(local vCPUs)")
+
+	return &ParallelismConfig{
+		DumpJobs:    dumpJobs,
+		RestoreJobs: restoreJobs,
+	}, nil
+}
+
+// resolveRDSInstanceVCPUs looks up the vCPU count for the configured RDS instance class
+// by querying the EC2 DescribeInstanceTypes API.
+func resolveRDSInstanceVCPUs(ctx context.Context, cfg *Config) (int, error) {
+	ec2Client, err := newEC2Client(ctx, cfg)
+	if err != nil {
+		return 0, fmt.Errorf("failed to create EC2 client: %w", err)
+	}
+
+	return lookupInstanceVCPUs(ctx, ec2Client, cfg.RDSClone.InstanceClass)
+}
+
+func newEC2Client(ctx context.Context, cfg *Config) (EC2API, error) {
+	awsCfg, err := awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(cfg.AWS.Region))
+	if err != nil {
+		return nil, fmt.Errorf("failed to load AWS config: %w", err)
+	}
+
+	var opts []func(*ec2.Options)
+	if cfg.AWS.Endpoint != "" {
+		opts = append(opts, func(o *ec2.Options) {
+			o.BaseEndpoint = aws.String(cfg.AWS.Endpoint)
+		})
+	}
+
+	return ec2.NewFromConfig(awsCfg, opts...), nil
+}
+
+// lookupInstanceVCPUs queries EC2 for the vCPU count of the given RDS instance class.
+func lookupInstanceVCPUs(ctx context.Context, client EC2API, rdsInstanceClass string) (int, error) {
+	ec2InstanceType, err := rdsClassToEC2Type(rdsInstanceClass)
+	if err != nil {
+		return 0, err
+	}
+
+	result, err := client.DescribeInstanceTypes(ctx, &ec2.DescribeInstanceTypesInput{
+		InstanceTypes: []ec2types.InstanceType{ec2types.InstanceType(ec2InstanceType)},
+	})
+	if err != nil {
+		return 0, fmt.Errorf("failed to describe EC2 instance type %q: %w", ec2InstanceType, err)
+	}
+
+	if len(result.InstanceTypes) == 0 {
+		return 0, fmt.Errorf("EC2 instance type %q not found", ec2InstanceType)
+	}
+
+	info := result.InstanceTypes[0]
+	if info.VCpuInfo == nil || info.VCpuInfo.DefaultVCpus == nil {
+		return 0, fmt.Errorf("vCPU info not available for instance type %q", ec2InstanceType)
+	}
+
+	vcpus := int(*info.VCpuInfo.DefaultVCpus)
+	if vcpus < minParallelJobs {
+		return minParallelJobs, nil
+	}
+
+	return vcpus, nil
+}
+
+// rdsClassToEC2Type converts an RDS instance class (e.g. "db.m5.xlarge") to an EC2 instance type ("m5.xlarge").
+func rdsClassToEC2Type(rdsClass string) (string, error) {
+	if !strings.HasPrefix(rdsClass, rdsInstanceClassPrefix) {
+		return "", fmt.Errorf("invalid RDS instance class %q: expected %q prefix", rdsClass, rdsInstanceClassPrefix)
+	}
+
+	ec2Type := strings.TrimPrefix(rdsClass, rdsInstanceClassPrefix)
+	if ec2Type == "" {
+		return "", fmt.Errorf("invalid RDS instance class %q: empty after removing prefix", rdsClass)
+	}
+
+	return ec2Type, nil
+}
+
+// resolveLocalVCPUs returns the number of logical CPUs available on the local machine.
+func resolveLocalVCPUs() int {
+	cpus := runtime.NumCPU()
+	if cpus < minParallelJobs {
+		return minParallelJobs
+	}
+
+	return cpus
+}