Merge branch '282-compressed-dumps' into 'master'

feat: support compressed plain-text dumps (#282)

Closes #282

See merge request postgres-ai/database-lab!314

Author: agneum

configs/config.example.logical_generic.yml

@@ -244,6 +244,8 @@ retrieval:
         #   database1:
         #     # Dump format. Available formats: directory, custom, plain. Default format: directory.
         #     format: directory
+        #     # Compression (only for plain-text dumps): "gzip", "bzip2", or "no". Default: "no".
+        #     compression: no
         #     # Option for a partial restore. Do not specify the tables section to restore all available tables.
         #     tables:
         #       - table1

configs/config.example.logical_rds_iam.yml

@@ -252,6 +252,8 @@ retrieval:
         #   database1:
         #     # Dump format. Available formats: directory, custom, plain. Default format: directory.
         #     format: directory
+        #     # Compression (only for plain-text dumps): "gzip", "bzip2", or "no". Default: "no".
+        #     compression: no
         #     # Option for a partial restore. Do not specify the tables section to restore all available tables.
         #     tables:
         #       - table1

pkg/retrieval/engine/postgres/logical/archive_types.go

@@ -0,0 +1,45 @@
+/*
+2021 © Postgres.ai
+*/
+
+package logical
+
+import (
+	"path/filepath"
+)
+
+type compressionType string
+
+const (
+	noCompression    compressionType = "no"
+	gzipCompression  compressionType = "gzip"
+	bzip2Compression compressionType = "bzip2"
+)
+
+// getReadingArchiveCommand chooses command to read dump file.
+func getReadingArchiveCommand(compressionType compressionType) string {
+	switch compressionType {
+	case gzipCompression:
+		return "gunzip -c"
+
+	case bzip2Compression:
+		return "bunzip2 -c"
+
+	default:
+		return "cat"
+	}
+}
+
+// getCompressionType returns archive type based on filename extension.
+func getCompressionType(filename string) compressionType {
+	switch filepath.Ext(filename) {
+	case ".gz":
+		return gzipCompression
+
+	case ".bz2":
+		return bzip2Compression
+
+	default:
+		return noCompression
+	}
+}

pkg/retrieval/engine/postgres/logical/archive_types_test.go

@@ -0,0 +1,69 @@
+/*
+2021 © Postgres.ai
+*/
+
+package logical
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestReadingArchiveCommand(t *testing.T) {
+	testCases := []struct {
+		compressionType compressionType
+		expectedCommand string
+	}{
+		{
+			compressionType: gzipCompression,
+			expectedCommand: "gunzip -c",
+		},
+		{
+			compressionType: bzip2Compression,
+			expectedCommand: "bunzip2 -c",
+		},
+		{
+			compressionType: noCompression,
+			expectedCommand: "cat",
+		},
+		{
+			compressionType: compressionType(""),
+			expectedCommand: "cat",
+		},
+	}
+
+	for _, tc := range testCases {
+		command := getReadingArchiveCommand(tc.compressionType)
+		assert.Equal(t, tc.expectedCommand, command)
+	}
+}
+
+func TestArchiveType(t *testing.T) {
+	testCases := []struct {
+		filename                string
+		expectedCompressionType compressionType
+	}{
+		{
+			filename:                "dump.sql.gz",
+			expectedCompressionType: gzipCompression,
+		},
+		{
+			filename:                "dump.bz2",
+			expectedCompressionType: bzip2Compression,
+		},
+		{
+			filename:                "test.dmp",
+			expectedCompressionType: noCompression,
+		},
+		{
+			filename:                "dump",
+			expectedCompressionType: noCompression,
+		},
+	}
+
+	for _, tc := range testCases {
+		compressionType := getCompressionType(tc.filename)
+		assert.Equal(t, tc.expectedCompressionType, compressionType)
+	}
+}

pkg/retrieval/engine/postgres/logical/dump.go

@@ -78,14 +78,14 @@ type DumpJob struct {
 
 // DumpOptions defines a logical dump options.
 type DumpOptions struct {
-	DumpLocation    string                  `yaml:"dumpLocation"`
-	DockerImage     string                  `yaml:"dockerImage"`
-	ContainerConfig map[string]interface{}  `yaml:"containerConfig"`
-	Connection      Connection              `yaml:"connection"`
-	Source          Source                  `yaml:"source"`
-	Databases       map[string]DBDefinition `yaml:"databases"`
-	ParallelJobs    int                     `yaml:"parallelJobs"`
-	Restore         ImmediateRestore        `yaml:"immediateRestore"`
+	DumpLocation    string                    `yaml:"dumpLocation"`
+	DockerImage     string                    `yaml:"dockerImage"`
+	ContainerConfig map[string]interface{}    `yaml:"containerConfig"`
+	Connection      Connection                `yaml:"connection"`
+	Source          Source                    `yaml:"source"`
+	Databases       map[string]DumpDefinition `yaml:"databases"`
+	ParallelJobs    int                       `yaml:"parallelJobs"`
+	Restore         ImmediateRestore          `yaml:"immediateRestore"`
 }
 
 // Source describes source of data to dump.
@@ -95,11 +95,12 @@ type Source struct {
 	RDS        *RDSConfig `yaml:"rdsIam"`
 }
 
-// DBDefinition describes a database for dumping.
-type DBDefinition struct {
-	Tables []string `yaml:"tables"`
-	Format string   `yaml:"format"`
-	dbName string
+// DumpDefinition describes a database for dumping.
+type DumpDefinition struct {
+	Tables      []string        `yaml:"tables"`
+	Format      string          `yaml:"format"`
+	Compression compressionType `yaml:"compression"`
+	dbName      string
 }
 
 type dumpJobConfig struct {
@@ -366,8 +367,8 @@ func (d *DumpJob) Run(ctx context.Context) (err error) {
 	return nil
 }
 
-func (d *DumpJob) getDBList(ctx context.Context) (map[string]DBDefinition, error) {
-	dbList := make(map[string]DBDefinition)
+func (d *DumpJob) getDBList(ctx context.Context) (map[string]DumpDefinition, error) {
+	dbList := make(map[string]DumpDefinition)
 
 	connStr := db.ConnectionString(d.config.db.Host, strconv.Itoa(d.config.db.Port), d.config.db.Username, d.config.db.DBName, d.getPassword())
 
@@ -387,7 +388,7 @@ func (d *DumpJob) getDBList(ctx context.Context) (map[string]DBDefinition, error
 			return nil, errors.Wrap(err, "failed to scan next row in database list result set")
 		}
 
-		dbList[dbName] = DBDefinition{}
+		dbList[dbName] = DumpDefinition{}
 	}
 
 	return dbList, nil
@@ -420,12 +421,12 @@ func (d *DumpJob) cleanupDumpLocation(ctx context.Context, dumpContID string) er
 	return nil
 }
 
-func (d *DumpJob) dumpDatabase(ctx context.Context, dumpContID, dbName string, dbDefinition DBDefinition) error {
-	dumpCommand := d.buildLogicalDumpCommand(dbName, dbDefinition.Tables)
+func (d *DumpJob) dumpDatabase(ctx context.Context, dumpContID, dbName string, dumpDefinition DumpDefinition) error {
+	dumpCommand := d.buildLogicalDumpCommand(dbName, dumpDefinition.Tables)
 	log.Msg("Running dump command: ", dumpCommand)
 
-	if len(dbDefinition.Tables) > 0 {
-		log.Msg("Partial dump will be run. Tables for dumping: ", strings.Join(dbDefinition.Tables, ", "))
+	if len(dumpDefinition.Tables) > 0 {
+		log.Msg("Partial dump will be run. Tables for dumping: ", strings.Join(dumpDefinition.Tables, ", "))
 	}
 
 	if output, err := d.performDumpCommand(ctx, dumpContID, types.ExecConfig{