Add configurable PartSize for PutObject (#1093)

api-put-object-common.go

@@ -67,20 +67,41 @@ func isReadAt(reader io.Reader) (ok bool) {
 //  minPartSize - 64MiB
 //  maxMultipartPutObjectSize - 5TiB
 //
-func optimalPartInfo(objectSize int64) (totalPartsCount int, partSize int64, lastPartSize int64, err error) {
+func optimalPartInfo(objectSize int64, configuredPartSize uint64) (totalPartsCount int, partSize int64, lastPartSize int64, err error) {
 	// object size is '-1' set it to 5TiB.
 	if objectSize == -1 {
 		objectSize = maxMultipartPutObjectSize
 	}
+
 	// object size is larger than supported maximum.
 	if objectSize > maxMultipartPutObjectSize {
 		err = ErrEntityTooLarge(objectSize, maxMultipartPutObjectSize, "", "")
 		return
 	}
+
+	if int64(configuredPartSize) > objectSize {
+		err = ErrEntityTooLarge(int64(configuredPartSize), objectSize, "", "")
+		return
+	}
+
+	if objectSize > (int64(configuredPartSize) * maxPartsCount) {
+		err = ErrInvalidArgument("Part size * max_parts(10000) is lesser than input objectSize.")
+	}
+
+	if configuredPartSize < absMinPartSize {
+		err = ErrInvalidArgument("Input part size is smaller than allowed minimum of 5MiB.")
+		return
+	}
+
+	if configuredPartSize > maxPartSize {
+		err = ErrInvalidArgument("Input part size is bigger than allowed maximum of 5GiB.")
+		return
+	}
+
 	// Use floats for part size for all calculations to avoid
 	// overflows during float64 to int64 conversions.
 	partSizeFlt := math.Ceil(float64(objectSize / maxPartsCount))
-	partSizeFlt = math.Ceil(partSizeFlt/minPartSize) * minPartSize
+	partSizeFlt = math.Ceil(partSizeFlt/float64(configuredPartSize)) * float64(configuredPartSize)
 	// Total parts count.
 	totalPartsCount = int(math.Ceil(float64(objectSize) / partSizeFlt))
 	// Part size.

api-put-object-multipart.go

@@ -73,7 +73,7 @@ func (c Client) putObjectMultipartNoStream(ctx context.Context, bucketName, obje
 	var complMultipartUpload completeMultipartUpload
 
 	// Calculate the optimal parts info for a given size.
-	totalPartsCount, partSize, _, err := optimalPartInfo(-1)
+	totalPartsCount, partSize, _, err := optimalPartInfo(-1, opts.PartSize)
 	if err != nil {
 		return 0, err
 	}

api-put-object-streaming.go

@@ -97,7 +97,7 @@ func (c Client) putObjectMultipartStreamFromReadAt(ctx context.Context, bucketNa
 	}
 
 	// Calculate the optimal parts info for a given size.
-	totalPartsCount, partSize, lastPartSize, err := optimalPartInfo(size)
+	totalPartsCount, partSize, lastPartSize, err := optimalPartInfo(size, opts.PartSize)
 	if err != nil {
 		return 0, err
 	}
@@ -240,7 +240,7 @@ func (c Client) putObjectMultipartStreamNoChecksum(ctx context.Context, bucketNa
 	}
 
 	// Calculate the optimal parts info for a given size.
-	totalPartsCount, partSize, lastPartSize, err := optimalPartInfo(size)
+	totalPartsCount, partSize, lastPartSize, err := optimalPartInfo(size, opts.PartSize)
 	if err != nil {
 		return 0, err
 	}

api-put-object.go

@@ -44,6 +44,7 @@ type PutObjectOptions struct {
 	NumThreads              uint
 	StorageClass            string
 	WebsiteRedirectLocation string
+	PartSize                uint64
 }
 
 // getNumThreads - gets the number of threads to be used in the multipart
@@ -147,8 +148,12 @@ func (c Client) putObjectCommon(ctx context.Context, bucketName, objectName stri
 		return c.putObjectNoChecksum(ctx, bucketName, objectName, reader, size, opts)
 	}
 
+	if opts.PartSize == 0 {
+		opts.PartSize = minPartSize
+	}
+
 	if c.overrideSignerType.IsV2() {
-		if size >= 0 && size < minPartSize {
+		if size >= 0 && size < int64(opts.PartSize) {
 			return c.putObjectNoChecksum(ctx, bucketName, objectName, reader, size, opts)
 		}
 		return c.putObjectMultipart(ctx, bucketName, objectName, reader, size, opts)
@@ -157,9 +162,10 @@ func (c Client) putObjectCommon(ctx context.Context, bucketName, objectName stri
 		return c.putObjectMultipartStreamNoLength(ctx, bucketName, objectName, reader, opts)
 	}
 
-	if size < minPartSize {
+	if size < int64(opts.PartSize) {
 		return c.putObjectNoChecksum(ctx, bucketName, objectName, reader, size, opts)
 	}
+
 	// For all sizes greater than 64MiB do multipart.
 	return c.putObjectMultipartStream(ctx, bucketName, objectName, reader, size, opts)
 }
@@ -181,7 +187,7 @@ func (c Client) putObjectMultipartStreamNoLength(ctx context.Context, bucketName
 	var complMultipartUpload completeMultipartUpload
 
 	// Calculate the optimal parts info for a given size.
-	totalPartsCount, partSize, _, err := optimalPartInfo(-1)
+	totalPartsCount, partSize, _, err := optimalPartInfo(-1, opts.PartSize)
 	if err != nil {
 		return 0, err
 	}

api_unit_test.go

@@ -116,11 +116,11 @@ func TestBucketPolicyTypes(t *testing.T) {
 
 // Tests optimal part size.
 func TestPartSize(t *testing.T) {
-	_, _, _, err := optimalPartInfo(5000000000000000000)
+	_, _, _, err := optimalPartInfo(5000000000000000000, minPartSize)
 	if err == nil {
 		t.Fatal("Error: should fail")
 	}
-	totalPartsCount, partSize, lastPartSize, err := optimalPartInfo(5497558138880)
+	totalPartsCount, partSize, lastPartSize, err := optimalPartInfo(5497558138880, minPartSize)
 	if err != nil {
 		t.Fatal("Error: ", err)
 	}
@@ -133,14 +133,14 @@ func TestPartSize(t *testing.T) {
 	if lastPartSize != 134217728 {
 		t.Fatalf("Error: expecting last part size of 241172480: got %v instead", lastPartSize)
 	}
-	_, partSize, _, err = optimalPartInfo(5000000000)
+	_, partSize, _, err = optimalPartInfo(5000000000, minPartSize)
 	if err != nil {
 		t.Fatal("Error:", err)
 	}
 	if partSize != minPartSize {
 		t.Fatalf("Error: expecting part size of %v: got %v instead", minPartSize, partSize)
 	}
-	totalPartsCount, partSize, lastPartSize, err = optimalPartInfo(-1)
+	totalPartsCount, partSize, lastPartSize, err = optimalPartInfo(-1, minPartSize)
 	if err != nil {
 		t.Fatal("Error:", err)
 	}