From 3eac02f676bdf9e2aae3d0785c185d177ef9e2aa Mon Sep 17 00:00:00 2001
From: Klaus Post <klauspost@gmail.com>
Date: Thu, 12 Aug 2021 19:05:24 +0200
Subject: [PATCH] Use metadata reader in ReadVersion (#12942)

Use `readMetadata` when reading version
information without data requested.

Reduces IO on inlined data.

Bonus: Inline compressed data as well when
compression is enabled.
---
 cmd/erasure-object.go | 15 ++++++++++++++-
 cmd/xl-storage.go     | 16 ++++++++++++++--
 2 files changed, 28 insertions(+), 3 deletions(-)

diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go
index 62955329d..3d360f3d4 100644
--- a/cmd/erasure-object.go
+++ b/cmd/erasure-object.go
@@ -869,6 +869,15 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 		} else if shardFileSize < smallFileThreshold/8 {
 			inlineBuffers = make([]*bytes.Buffer, len(onlineDisks))
 		}
+	} else {
+		// If compressed, use actual size to determine.
+		if sz := erasure.ShardFileSize(data.ActualSize()); sz > 0 {
+			if !opts.Versioned && sz < smallFileThreshold {
+				inlineBuffers = make([]*bytes.Buffer, len(onlineDisks))
+			} else if sz < smallFileThreshold/8 {
+				inlineBuffers = make([]*bytes.Buffer, len(onlineDisks))
+			}
+		}
 	}
 	for i, disk := range onlineDisks {
 		if disk == nil {
@@ -876,7 +885,11 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 		}
 
 		if len(inlineBuffers) > 0 {
-			inlineBuffers[i] = bytes.NewBuffer(make([]byte, 0, shardFileSize))
+			sz := shardFileSize
+			if sz < 0 {
+				sz = data.ActualSize()
+			}
+			inlineBuffers[i] = bytes.NewBuffer(make([]byte, 0, sz))
 			writers[i] = newStreamingBitrotWriterBuffer(inlineBuffers[i], DefaultBitrotAlgorithm, erasure.ShardSize())
 			continue
 		}
diff --git a/cmd/xl-storage.go b/cmd/xl-storage.go
index 31207bce0..2127ce7f7 100644
--- a/cmd/xl-storage.go
+++ b/cmd/xl-storage.go
@@ -1068,8 +1068,20 @@ func (s *xlStorage) ReadVersion(ctx context.Context, volume, path, versionID str
 	if err != nil {
 		return fi, err
 	}
-
-	buf, err := s.ReadAll(ctx, volume, pathJoin(path, xlStorageFormatFile))
+	var buf []byte
+	if readData {
+		buf, err = s.ReadAll(ctx, volume, pathJoin(path, xlStorageFormatFile))
+	} else {
+		buf, err = s.readMetadata(pathJoin(volumeDir, path, xlStorageFormatFile))
+		if err != nil {
+			if osIsNotExist(err) {
+				if err = Access(volumeDir); err != nil && osIsNotExist(err) {
+					return fi, errVolumeNotFound
+				}
+			}
+			err = osErrToFileErr(err)
+		}
+	}
 	if err != nil {
 		if err == errFileNotFound {
 			if err = s.renameLegacyMetadata(volumeDir, path); err != nil {