slack-vitess-r14.0.5: slack-cherrypick: revert backup compressor

go.mod

@@ -65,7 +65,6 @@ require (
 	github.com/philhofer/fwd v1.0.0 // indirect
 	github.com/pires/go-proxyproto v0.6.1
 	github.com/pkg/errors v0.9.1 // indirect
-	github.com/planetscale/pargzip v0.0.0-20201116224723-90c7fc03ea8a
 	github.com/planetscale/vtprotobuf v0.3.0
 	github.com/prometheus/client_golang v1.11.1
 	github.com/prometheus/common v0.29.0 // indirect

go.sum

@@ -614,8 +614,6 @@ github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/sftp v1.10.1/go.mod h1:lYOWFsE0bwd1+KfKJaKeuokY15vzFx25BLbzYYoAxZI=
-github.com/planetscale/pargzip v0.0.0-20201116224723-90c7fc03ea8a h1:y0OpQ4+5tKxeh9+H+2cVgASl9yMZYV9CILinKOiKafA=
-github.com/planetscale/pargzip v0.0.0-20201116224723-90c7fc03ea8a/go.mod h1:GJFUzQuXIoB2Kjn1ZfDhJr/42D5nWOqRcIQVgCxTuIE=
 github.com/planetscale/vtprotobuf v0.3.0 h1:oMrOdDFHS1ADc0dHtC2EApxiM5xd0cQkZeibm0WgXiQ=
 github.com/planetscale/vtprotobuf v0.3.0/go.mod h1:wm1N3qk9G/4+VM1WhpkLbvY/d8+0PbwYYpP5P5VhTks=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=

go/vt/mysqlctl/builtinbackupengine.go

@@ -33,7 +33,6 @@ import (
 	"time"
 
 	"github.com/klauspost/pgzip"
-	"github.com/planetscale/pargzip"
 
 	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/sync2"
@@ -498,12 +497,13 @@ func (be *BuiltinBackupEngine) backupFile(ctx context.Context, params BackupPara
 	}
 
 	// Create the gzip compression pipe, if necessary.
-	var gzip *pargzip.Writer
+	var gzip *pgzip.Writer
 	if *backupStorageCompress {
-		gzip = pargzip.NewWriter(writer)
-		gzip.ChunkSize = *backupCompressBlockSize
-		gzip.Parallel = *backupCompressBlocks
-		gzip.CompressionLevel = pargzip.BestSpeed
+		gzip, err = pgzip.NewWriterLevel(writer, pgzip.BestSpeed)
+		if err != nil {
+			return vterrors.Wrap(err, "cannot create gziper")
+		}
+		gzip.SetConcurrency(*backupCompressBlockSize, *backupCompressBlocks)
 		writer = gzip
 	}
 

go/vt/mysqlctl/xtrabackupengine.go

@@ -32,7 +32,6 @@ import (
 	"time"
 
 	"github.com/klauspost/pgzip"
-	"github.com/planetscale/pargzip"
 
 	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/vt/logutil"
@@ -271,18 +270,19 @@ func (be *XtrabackupEngine) backupFiles(ctx context.Context, params BackupParams
 
 	destWriters := []io.Writer{}
 	destBuffers := []*bufio.Writer{}
-	destCompressors := []io.WriteCloser{}
+	destCompressors := []*pgzip.Writer{}
 	for _, file := range destFiles {
 		buffer := bufio.NewWriterSize(file, writerBufferSize)
 		destBuffers = append(destBuffers, buffer)
 		writer := io.Writer(buffer)
 
 		// Create the gzip compression pipe, if necessary.
 		if *backupStorageCompress {
-			compressor := pargzip.NewWriter(writer)
-			compressor.ChunkSize = *backupCompressBlockSize
-			compressor.Parallel = *backupCompressBlocks
-			compressor.CompressionLevel = pargzip.BestSpeed
+			compressor, err := pgzip.NewWriterLevel(writer, pgzip.BestSpeed)
+			if err != nil {
+				return replicationPosition, vterrors.Wrap(err, "cannot create gzip compressor")
+			}
+			compressor.SetConcurrency(*backupCompressBlockSize, *backupCompressBlocks)
 			writer = compressor
 			destCompressors = append(destCompressors, compressor)
 		}
@@ -308,6 +308,7 @@ func (be *XtrabackupEngine) backupFiles(ctx context.Context, params BackupParams
 		capture := false
 		for scanner.Scan() {
 			line := scanner.Text()
+			fmt.Fprintln(stderrBuilder, line)
 			params.Logger.Infof("xtrabackup stderr: %s", line)
 
 			// Wait until we see the first line of the binlog position.
@@ -522,7 +523,7 @@ func (be *XtrabackupEngine) extractFiles(ctx context.Context, logger logutil.Log
 	}()
 
 	srcReaders := []io.Reader{}
-	srcDecompressors := []io.ReadCloser{}
+	srcDecompressors := []*pgzip.Reader{}
 	for _, file := range srcFiles {
 		reader := io.Reader(file)
 
@@ -736,7 +737,7 @@ func copyToStripes(writers []io.Writer, reader io.Reader, blockSize int64) (writ
 	}
 
 	// Read blocks from source and round-robin them to destination writers.
-	// Since we put a buffer in front of the destination file, and pargzip has its
+	// Since we put a buffer in front of the destination file, and pgzip has its
 	// own buffer as well, we are writing into a buffer either way (whether a
 	// compressor is in the chain or not). That means these writes should not
 	// block often, so we shouldn't need separate goroutines here.