feat(repository): Metadata compression config support for indirect content

internal/server/server_test.go

@@ -210,7 +210,8 @@ func remoteRepositoryTest(ctx context.Context, t *testing.T, rep repo.Repository
 		mustReadObject(ctx, t, w, result, written)
 
 		ow := w.NewObjectWriter(ctx, object.WriterOptions{
-			Prefix: content.ManifestContentPrefix,
+			Prefix:             content.ManifestContentPrefix,
+			MetadataCompressor: "zstd-fastest",
 		})
 
 		_, err := ow.Write([]byte{2, 3, 4})
@@ -258,7 +259,7 @@ func remoteRepositoryTest(ctx context.Context, t *testing.T, rep repo.Repository
 func mustWriteObject(ctx context.Context, t *testing.T, w repo.RepositoryWriter, data []byte) object.ID {
 	t.Helper()
 
-	ow := w.NewObjectWriter(ctx, object.WriterOptions{})
+	ow := w.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 
 	_, err := ow.Write(data)
 	require.NoError(t, err)

repo/blob/storage_extend_test.go

@@ -42,7 +42,7 @@ func (s *formatSpecificTestSuite) TestExtendBlobRetention(t *testing.T) {
 			nro.RetentionPeriod = period
 		},
 	})
-	w := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{})
+	w := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 	io.WriteString(w, "hello world!")
 	w.Result()
 	w.Close()
@@ -103,7 +103,7 @@ func (s *formatSpecificTestSuite) TestExtendBlobRetentionUnsupported(t *testing.
 			nro.RetentionMode = ""
 		},
 	})
-	w := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{})
+	w := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 	io.WriteString(w, "hello world!")
 	w.Result()
 	w.Close()

repo/format/upgrade_lock_test.go

@@ -401,7 +401,7 @@ func TestFormatUpgradeDuringOngoingWriteSessions(t *testing.T) {
 func writeObject(ctx context.Context, t *testing.T, rep repo.RepositoryWriter, data []byte, testCaseID string) {
 	t.Helper()
 
-	w := rep.NewObjectWriter(ctx, object.WriterOptions{})
+	w := rep.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 
 	_, err := w.Write(data)
 	require.NoError(t, err, testCaseID)

repo/grpc_repository_client.go

@@ -529,9 +529,9 @@ func (r *grpcRepositoryClient) NewWriter(ctx context.Context, opt WriteSessionOp
 }
 
 // ConcatenateObjects creates a concatenated objects from the provided object IDs.
-func (r *grpcRepositoryClient) ConcatenateObjects(ctx context.Context, objectIDs []object.ID) (object.ID, error) {
+func (r *grpcRepositoryClient) ConcatenateObjects(ctx context.Context, objectIDs []object.ID, comp compression.Name) (object.ID, error) {
 	//nolint:wrapcheck
-	return r.omgr.Concatenate(ctx, objectIDs)
+	return r.omgr.Concatenate(ctx, objectIDs, comp)
 }
 
 // maybeRetry executes the provided callback with or without automatic retries depending on how

repo/maintenance/blob_gc_test.go

@@ -47,7 +47,7 @@ func (s *formatSpecificTestSuite) TestDeleteUnreferencedBlobs(t *testing.T) {
 			nro.BlockFormat.HMACSecret = testHMACSecret
 		},
 	})
-	w := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{})
+	w := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 	io.WriteString(w, "hello world!")
 	w.Result()
 	w.Close()

repo/maintenance/blob_retain_test.go

@@ -43,7 +43,7 @@ func (s *formatSpecificTestSuite) TestExtendBlobRetentionTime(t *testing.T) {
 			nro.RetentionPeriod = period
 		},
 	})
-	w := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{})
+	w := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 	io.WriteString(w, "hello world!")
 	w.Result()
 	w.Close()
@@ -98,7 +98,7 @@ func (s *formatSpecificTestSuite) TestExtendBlobRetentionTimeDisabled(t *testing
 			nro.BlockFormat.HMACSecret = testHMACSecret
 		},
 	})
-	w := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{})
+	w := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 	io.WriteString(w, "hello world!")
 	w.Result()
 	w.Close()

repo/maintenance/content_rewrite_test.go

@@ -79,7 +79,7 @@ func (s *formatSpecificTestSuite) TestContentRewrite(t *testing.T) {
 			// run N sessions to create N individual pack blobs for each content prefix
 			for range tc.numPContents {
 				require.NoError(t, repo.WriteSession(ctx, env.Repository, repo.WriteSessionOptions{}, func(ctx context.Context, w repo.RepositoryWriter) error {
-					ow := w.NewObjectWriter(ctx, object.WriterOptions{})
+					ow := w.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 					fmt.Fprintf(ow, "%v", uuid.NewString())
 					_, err := ow.Result()
 					return err
@@ -88,7 +88,7 @@ func (s *formatSpecificTestSuite) TestContentRewrite(t *testing.T) {
 
 			for range tc.numQContents {
 				require.NoError(t, repo.WriteSession(ctx, env.Repository, repo.WriteSessionOptions{}, func(ctx context.Context, w repo.RepositoryWriter) error {
-					ow := w.NewObjectWriter(ctx, object.WriterOptions{Prefix: "k"})
+					ow := w.NewObjectWriter(ctx, object.WriterOptions{Prefix: "k", MetadataCompressor: "zstd-fastest"})
 					fmt.Fprintf(ow, "%v", uuid.NewString())
 					_, err := ow.Result()
 					return err

repo/maintenance/maintenance_safety_test.go

@@ -34,7 +34,7 @@ func (s *formatSpecificTestSuite) TestMaintenanceSafety(t *testing.T) {
 
 	// create object that's immediately orphaned since nobody refers to it.
 	require.NoError(t, repo.WriteSession(ctx, env.Repository, repo.WriteSessionOptions{}, func(ctx context.Context, w repo.RepositoryWriter) error {
-		ow := w.NewObjectWriter(ctx, object.WriterOptions{Prefix: "y"})
+		ow := w.NewObjectWriter(ctx, object.WriterOptions{Prefix: "y", MetadataCompressor: "zstd-fastest"})
 		fmt.Fprintf(ow, "hello world")
 		var err error
 		objectID, err = ow.Result()
@@ -43,7 +43,7 @@ func (s *formatSpecificTestSuite) TestMaintenanceSafety(t *testing.T) {
 
 	// create another object in separate pack.
 	require.NoError(t, repo.WriteSession(ctx, env.Repository, repo.WriteSessionOptions{}, func(ctx context.Context, w repo.RepositoryWriter) error {
-		ow := w.NewObjectWriter(ctx, object.WriterOptions{Prefix: "y"})
+		ow := w.NewObjectWriter(ctx, object.WriterOptions{Prefix: "y", MetadataCompressor: "zstd-fastest"})
 		fmt.Fprintf(ow, "hello universe")
 		_, err := ow.Result()
 		return err

repo/object/object_manager.go

@@ -70,6 +70,7 @@ func (om *Manager) NewWriter(ctx context.Context, opt WriterOptions) Writer {
 	w.description = opt.Description
 	w.prefix = opt.Prefix
 	w.compressor = compression.ByName[opt.Compressor]
+	w.metadataCompressor = compression.ByName[opt.MetadataCompressor]
 	w.totalLength = 0
 	w.currentPosition = 0
 
@@ -106,7 +107,7 @@ func (om *Manager) closedWriter(ow *objectWriter) {
 // in parallel utilizing more CPU cores. Because some split points now start at fixed boundaries and not content-specific,
 // this causes some slight loss of deduplication at concatenation points (typically 1-2 contents, usually <10MB),
 // so this method should only be used for very large files where this overhead is relatively small.
-func (om *Manager) Concatenate(ctx context.Context, objectIDs []ID) (ID, error) {
+func (om *Manager) Concatenate(ctx context.Context, objectIDs []ID, metadataComp compression.Name) (ID, error) {
 	if len(objectIDs) == 0 {
 		return EmptyID, errors.Errorf("empty list of objects")
 	}
@@ -131,8 +132,10 @@ func (om *Manager) Concatenate(ctx context.Context, objectIDs []ID) (ID, error)
 	log(ctx).Debugf("concatenated: %v total: %v", concatenatedEntries, totalLength)
 
 	w := om.NewWriter(ctx, WriterOptions{
-		Prefix:      indirectContentPrefix,
-		Description: "CONCATENATED INDEX",
+		Prefix:             indirectContentPrefix,
+		Description:        "CONCATENATED INDEX",
+		Compressor:         metadataComp,
+		MetadataCompressor: metadataComp,
 	})
 	defer w.Close() //nolint:errcheck
 

repo/object/object_manager_test.go

@@ -175,13 +175,15 @@ func TestCompression_ContentCompressionEnabled(t *testing.T) {
 	_, _, om := setupTest(t, cmap)
 
 	w := om.NewWriter(ctx, WriterOptions{
-		Compressor: "gzip",
+		Compressor:         "gzip",
+		MetadataCompressor: "zstd-fastest",
 	})
 	w.Write(bytes.Repeat([]byte{1, 2, 3, 4}, 1000))
 	oid, err := w.Result()
 	require.NoError(t, err)
 
 	cid, isCompressed, ok := oid.ContentID()
+
 	require.True(t, ok)
 	require.False(t, isCompressed) // oid will not indicate compression
 	require.Equal(t, compression.ByName["gzip"].HeaderID(), cmap[cid])
@@ -244,7 +246,8 @@ func TestCompression_ContentCompressionDisabled(t *testing.T) {
 	_, _, om := setupTest(t, nil)
 
 	w := om.NewWriter(ctx, WriterOptions{
-		Compressor: "gzip",
+		Compressor:         "gzip",
+		MetadataCompressor: "zstd-fastest",
 	})
 	w.Write(bytes.Repeat([]byte{1, 2, 3, 4}, 1000))
 	oid, err := w.Result()
@@ -578,7 +581,7 @@ func TestConcatenate(t *testing.T) {
 	}
 
 	for _, tc := range cases {
-		concatenatedOID, err := om.Concatenate(ctx, tc.inputs)
+		concatenatedOID, err := om.Concatenate(ctx, tc.inputs, "zstd-fastest")
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -617,7 +620,7 @@ func TestConcatenate(t *testing.T) {
 		}
 
 		// make sure results of concatenation can be further concatenated.
-		concatenated3OID, err := om.Concatenate(ctx, []ID{concatenatedOID, concatenatedOID, concatenatedOID})
+		concatenated3OID, err := om.Concatenate(ctx, []ID{concatenatedOID, concatenatedOID, concatenatedOID}, "zstd-fastest")
 		if err != nil {
 			t.Fatal(err)
 		}

repo/object/object_writer.go

@@ -68,7 +68,8 @@ type objectWriter struct {
 
 	om *Manager
 
-	compressor compression.Compressor
+	compressor         compression.Compressor
+	metadataCompressor compression.Compressor
 
 	prefix      content.IDPrefix
 	buffer      gather.WriteBuffer
@@ -292,12 +293,13 @@ func (w *objectWriter) checkpointLocked() (ID, error) {
 	}
 
 	iw := &objectWriter{
-		ctx:         w.ctx,
-		om:          w.om,
-		compressor:  nil,
-		description: "LIST(" + w.description + ")",
-		splitter:    w.om.newDefaultSplitter(),
-		prefix:      w.prefix,
+		ctx:                w.ctx,
+		om:                 w.om,
+		compressor:         w.metadataCompressor,
+		metadataCompressor: w.metadataCompressor,
+		description:        "LIST(" + w.description + ")",
+		splitter:           w.om.newDefaultSplitter(),
+		prefix:             w.prefix,
 	}
 
 	if iw.prefix == "" {
@@ -334,9 +336,10 @@ func writeIndirectObject(w io.Writer, entries []IndirectObjectEntry) error {
 
 // WriterOptions can be passed to Repository.NewWriter().
 type WriterOptions struct {
-	Description string
-	Prefix      content.IDPrefix // empty string or a single-character ('g'..'z')
-	Compressor  compression.Name
-	Splitter    string // use particular splitter instead of default
-	AsyncWrites int    // allow up to N content writes to be asynchronous
+	Description        string
+	Prefix             content.IDPrefix // empty string or a single-character ('g'..'z')
+	Compressor         compression.Name
+	MetadataCompressor compression.Name
+	Splitter           string // use particular splitter instead of default
+	AsyncWrites        int    // allow up to N content writes to be asynchronous
 }

repo/repo_benchmarks_test.go

@@ -15,7 +15,7 @@ func BenchmarkWriterDedup1M(b *testing.B) {
 	ctx, env := repotesting.NewEnvironment(b, format.FormatVersion2)
 	dataBuf := make([]byte, 4<<20)
 
-	writer := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{})
+	writer := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 	writer.Write(dataBuf)
 	_, err := writer.Result()
 	require.NoError(b, err)
@@ -25,7 +25,7 @@ func BenchmarkWriterDedup1M(b *testing.B) {
 
 	for range b.N {
 		// write exactly the same data
-		writer := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{})
+		writer := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 		writer.Write(dataBuf)
 		writer.Result()
 		writer.Close()
@@ -45,7 +45,7 @@ func BenchmarkWriterNoDedup1M(b *testing.B) {
 
 	for i := range b.N {
 		// write exactly the same data
-		writer := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{})
+		writer := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 
 		if i+chunkSize > len(dataBuf) {
 			chunkSize++

repo/repository.go

@@ -14,6 +14,7 @@ import (
 	"github.com/kopia/kopia/internal/metrics"
 	"github.com/kopia/kopia/repo/blob"
 	"github.com/kopia/kopia/repo/blob/throttling"
+	"github.com/kopia/kopia/repo/compression"
 	"github.com/kopia/kopia/repo/content"
 	"github.com/kopia/kopia/repo/content/indexblob"
 	"github.com/kopia/kopia/repo/format"
@@ -47,7 +48,7 @@ type RepositoryWriter interface {
 	Repository
 
 	NewObjectWriter(ctx context.Context, opt object.WriterOptions) object.Writer
-	ConcatenateObjects(ctx context.Context, objectIDs []object.ID) (object.ID, error)
+	ConcatenateObjects(ctx context.Context, objectIDs []object.ID, comp compression.Name) (object.ID, error)
 	PutManifest(ctx context.Context, labels map[string]string, payload interface{}) (manifest.ID, error)
 	ReplaceManifests(ctx context.Context, labels map[string]string, payload interface{}) (manifest.ID, error)
 	DeleteManifest(ctx context.Context, id manifest.ID) error
@@ -180,9 +181,9 @@ func (r *directRepository) NewObjectWriter(ctx context.Context, opt object.Write
 }
 
 // ConcatenateObjects creates a concatenated objects from the provided object IDs.
-func (r *directRepository) ConcatenateObjects(ctx context.Context, objectIDs []object.ID) (object.ID, error) {
+func (r *directRepository) ConcatenateObjects(ctx context.Context, objectIDs []object.ID, comp compression.Name) (object.ID, error) {
 	//nolint:wrapcheck
-	return r.omgr.Concatenate(ctx, objectIDs)
+	return r.omgr.Concatenate(ctx, objectIDs, comp)
 }
 
 // DisableIndexRefresh disables index refresh for the duration of the write session.

repo/repository_test.go

@@ -47,7 +47,7 @@ func (s *formatSpecificTestSuite) TestWriters(t *testing.T) {
 	for _, c := range cases {
 		ctx, env := repotesting.NewEnvironment(t, s.formatVersion)
 
-		writer := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{})
+		writer := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 		if _, err := writer.Write(c.data); err != nil {
 			t.Fatalf("write error: %v", err)
 		}
@@ -74,7 +74,7 @@ func (s *formatSpecificTestSuite) TestWriterCompleteChunkInTwoWrites(t *testing.
 	ctx, env := repotesting.NewEnvironment(t, s.formatVersion)
 
 	b := make([]byte, 100)
-	writer := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{})
+	writer := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 	writer.Write(b[0:50])
 	writer.Write(b[0:50])
 	result, err := writer.Result()
@@ -159,7 +159,7 @@ func (s *formatSpecificTestSuite) TestHMAC(t *testing.T) {
 
 	c := bytes.Repeat([]byte{0xcd}, 50)
 
-	w := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{})
+	w := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 	w.Write(c)
 	result, err := w.Result()
 
@@ -185,7 +185,7 @@ func (s *formatSpecificTestSuite) TestReaderStoredBlockNotFound(t *testing.T) {
 func writeObject(ctx context.Context, t *testing.T, rep repo.RepositoryWriter, data []byte, testCaseID string) object.ID {
 	t.Helper()
 
-	w := rep.NewObjectWriter(ctx, object.WriterOptions{})
+	w := rep.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 	if _, err := w.Write(data); err != nil {
 		t.Fatalf("can't write object %q - write failed: %v", testCaseID, err)
 	}
@@ -275,7 +275,7 @@ func TestFormats(t *testing.T) {
 
 		for k, v := range c.oids {
 			bytesToWrite := []byte(k)
-			w := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{})
+			w := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 			w.Write(bytesToWrite)
 
 			oid, err := w.Result()
@@ -555,7 +555,7 @@ func TestObjectWritesWithRetention(t *testing.T) {
 		},
 	})
 
-	writer := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{})
+	writer := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{MetadataCompressor: "zstd-fastest"})
 	_, err := writer.Write([]byte("the quick brown fox jumps over the lazy dog"))
 	require.NoError(t, err)
 
@@ -774,7 +774,8 @@ func TestMetrics_CompressibleData(t *testing.T) {
 
 	for ensureMapEntry(t, env.RepositoryMetrics().Snapshot(false).Counters, "content_write_duration_nanos") < 5e6 {
 		w := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{
-			Compressor: "gzip",
+			Compressor:         "gzip",
+			MetadataCompressor: "zstd-fastest",
 		})
 		w.Write(inputData)