fix(deps): Update module github.com/aws/aws-sdk-go-v2/feature/s3/manager to v1.22.0 (#22029)

cq-bot · web-flow · commit e21a6939e8e6 · 2026-02-16T11:34:33.000Z
This PR contains the following updates: | Package | Type | Update | Change | |---|---|---|---| | [github.com/aws/aws-sdk-go-v2/feature/s3/manager](https://redirect.github.com/aws/aws-sdk-go-v2) | require | minor | `v1.21.1` -> `v1.22.0` | --- > [!WARNING] > Some dependencies could not be looked up. Check the Dependency Dashboard for more information. --- ### Configuration 📅 **Schedule**: Branch creation - "before 3am on Saturday" (UTC), Automerge - At any time (no schedule defined). 🚦 **Automerge**: Disabled by config. Please merge this manually once you are satisfied. ♻ **Rebasing**: Whenever PR becomes conflicted, or you tick the rebase/retry checkbox. 🔕 **Ignore**: Close this PR and you won't be reminded about this update again. --- - [ ] If you want to rebase/retry this PR, check this box --- This PR has been generated by [Renovate Bot](https://redirect.github.com/renovatebot/renovate).
diff --git a/plugins/destination/s3/client/client.go b/plugins/destination/s3/client/client.go
@@ -17,7 +17,8 @@ import (
 	awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http"
 	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/credentials/stscreds"
-	"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
+	"github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager"
+	tmtypes "github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager/types"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/aws/aws-sdk-go-v2/service/sts"
 	"github.com/cloudquery/cloudquery/plugins/destination/s3/v7/client/spec"
@@ -41,7 +42,8 @@ type Client struct {
 	*filetypes.Client
 	writer *streamingbatchwriter.StreamingBatchWriter
 
-	s3Client *s3.Client
+	s3Client        *s3.Client
+	transferManager *transfermanager.Client
 
 	initializedTablesLock sync.Mutex
 	initializedTables     map[string]string
@@ -145,23 +147,28 @@ func New(ctx context.Context, logger zerolog.Logger, s []byte, opts plugin.NewCl
 		return nil, errRetrievingCredentials
 	}
 
+	// Initialize the transfer manager
+	c.transferManager = transfermanager.New(c.s3Client, func(o *transfermanager.Options) {
+		o.PartSizeBytes = *c.spec.PartSize
+	})
+
 	if *c.spec.TestWrite {
 		// we want to run this test because we want it to fail early if the bucket is not accessible
 		timeNow := time.Now().UTC()
 
-		params := &s3.PutObjectInput{
+		params := &transfermanager.UploadObjectInput{
 			Bucket: aws.String(c.spec.Bucket),
 			Key:    aws.String(c.spec.ReplacePathVariables("TEST_TABLE", "TEST_UUID", timeNow, c.syncID)),
 			Body:   bytes.NewReader([]byte("")),
 		}
 
 		sseConfiguration := c.spec.ServerSideEncryptionConfiguration
 		if sseConfiguration != nil {
-			params.SSEKMSKeyId = &sseConfiguration.SSEKMSKeyId
-			params.ServerSideEncryption = sseConfiguration.ServerSideEncryption
+			params.SSEKMSKeyID = &sseConfiguration.SSEKMSKeyId
+			params.ServerSideEncryption = tmtypes.ServerSideEncryption(sseConfiguration.ServerSideEncryption)
 		}
 
-		if _, err := manager.NewUploader(c.s3Client).Upload(ctx, params); err != nil {
+		if _, err := c.transferManager.UploadObject(ctx, params); err != nil {
 			return nil, errors.Join(errTestWriteFailed, err)
 		}
 	}
diff --git a/plugins/destination/s3/client/read.go b/plugins/destination/s3/client/read.go
@@ -12,7 +12,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/aws"
 	awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http"
 	"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
-	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager"
 	"github.com/aws/smithy-go"
 	"github.com/cloudquery/plugin-sdk/v4/schema"
 	"github.com/google/uuid"
@@ -30,11 +30,11 @@ func (c *Client) Read(ctx context.Context, table *schema.Table, res chan<- arrow
 
 	name := c.spec.ReplacePathVariables(table.Name, uuid.NewString(), time.Time{}, c.syncID)
 	writerAtBuffer := manager.NewWriteAtBuffer(make([]byte, 0, maxFileSize))
-	_, err := manager.NewDownloader(c.s3Client).Download(ctx,
-		writerAtBuffer,
-		&s3.GetObjectInput{
-			Bucket: aws.String(c.spec.Bucket),
-			Key:    aws.String(name),
+	_, err := c.transferManager.DownloadObject(ctx,
+		&transfermanager.DownloadObjectInput{
+			Bucket:   aws.String(c.spec.Bucket),
+			Key:      aws.String(name),
+			WriterAt: writerAtBuffer,
 		})
 
 	if err != nil {
diff --git a/plugins/destination/s3/client/write.go b/plugins/destination/s3/client/write.go
@@ -1,7 +1,6 @@
 package client
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
@@ -13,9 +12,8 @@ import (
 	"github.com/apache/arrow-go/v18/arrow/array"
 	"github.com/apache/arrow-go/v18/arrow/memory"
 	"github.com/aws/aws-sdk-go-v2/aws"
-	"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
-	"github.com/aws/aws-sdk-go-v2/service/s3"
-	awstypes "github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager"
+	tmtypes "github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager/types"
 	"github.com/cloudquery/filetypes/v4"
 	"github.com/cloudquery/plugin-sdk/v4/message"
 	"github.com/cloudquery/plugin-sdk/v4/schema"
@@ -25,54 +23,26 @@ import (
 
 var reInvalidJSONKey = regexp.MustCompile(`\W`)
 
-// If we don't use a reader that supports seeking, the S3 SDK will allocate a 5MB buffer each time it reads a chunk
-// While this can work for large files, it's not optimal for small files, based on our tests we're mostly uploading small files
-// And depending on source concurrency and destination batch settings, we can upload quite a bit of small files at the same time
-// For example we've seen 220 concurrent uploads where most files are only a few dozen KB, this means we're allocating ~1GB of memory, where we could be allocating only a few MBs
-// The memory is only cleared after the upload is finished, which makes it even worse
-// Please note that for large files the memory we read is capped by the batch size setting which defaults to 50MB
-func (c *Client) adjustReaderIfNeeded(r io.Reader) (io.Reader, error) {
-	// We don't use a reader that supports seeking if NoRotate is set since that means we'll be holding the entire file in memory until the table has finished resolving and the file is uploaded.
-	// This has the downside of potentially using a lot of memory for tables with small files.
-	// But it's a tradeoff we're willing to make for now until https://github.com/aws/aws-sdk-go-v2/issues/2694 is resolved, as it has a workaround that we can use (reducing source concurrency).
-	if c.spec.NoRotate {
-		return r, nil
-	}
-
-	allData, err := io.ReadAll(r)
-	if err != nil {
-		return nil, err
-	}
-	readerSeeker := bytes.NewReader(allData)
-	return readerSeeker, nil
-}
-
 func (c *Client) createObject(ctx context.Context, table *schema.Table, objKey string) (*filetypes.Stream, error) {
 	s, err := c.Client.StartStream(table, func(r io.Reader) error {
-		adjustedReader, err := c.adjustReaderIfNeeded(r)
-		if err != nil {
-			return err
-		}
-		params := &s3.PutObjectInput{
+		params := &transfermanager.UploadObjectInput{
 			Bucket:      aws.String(c.spec.Bucket),
 			Key:         aws.String(objKey),
-			Body:        adjustedReader,
+			Body:        r,
 			ContentType: aws.String(c.spec.GetContentType()),
 		}
 
 		if c.spec.ACL != "" {
-			params.ACL = awstypes.ObjectCannedACL(c.spec.ACL)
+			params.ACL = tmtypes.ObjectCannedACL(c.spec.ACL)
 		}
 
 		sseConfiguration := c.spec.ServerSideEncryptionConfiguration
 		if sseConfiguration != nil {
-			params.SSEKMSKeyId = &sseConfiguration.SSEKMSKeyId
-			params.ServerSideEncryption = sseConfiguration.ServerSideEncryption
+			params.SSEKMSKeyID = &sseConfiguration.SSEKMSKeyId
+			params.ServerSideEncryption = tmtypes.ServerSideEncryption(sseConfiguration.ServerSideEncryption)
 		}
 
-		_, err = manager.NewUploader(c.s3Client, func(uploader *manager.Uploader) {
-			uploader.PartSize = *c.spec.PartSize
-		}).Upload(ctx, params)
+		_, err := c.transferManager.UploadObject(ctx, params)
 		return err
 	})
 	return s, err
diff --git a/plugins/destination/s3/go.mod b/plugins/destination/s3/go.mod
@@ -7,7 +7,8 @@ require (
 	github.com/aws/aws-sdk-go-v2 v1.41.1
 	github.com/aws/aws-sdk-go-v2/config v1.32.7
 	github.com/aws/aws-sdk-go-v2/credentials v1.19.7
-	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.21.1
+	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.22.0
+	github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.1.2
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.96.0
 	github.com/aws/aws-sdk-go-v2/service/sts v1.41.6
 	github.com/aws/smithy-go v1.24.0
diff --git a/plugins/destination/s3/go.sum b/plugins/destination/s3/go.sum
@@ -21,8 +21,10 @@ github.com/aws/aws-sdk-go-v2/credentials v1.19.7 h1:tHK47VqqtJxOymRrNtUXN5SP/zUT
 github.com/aws/aws-sdk-go-v2/credentials v1.19.7/go.mod h1:qOZk8sPDrxhf+4Wf4oT2urYJrYt3RejHSzgAquYeppw=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.17 h1:I0GyV8wiYrP8XpA70g1HBcQO1JlQxCMTW9npl5UbDHY=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.17/go.mod h1:tyw7BOl5bBe/oqvoIeECFJjMdzXoa/dfVz3QQ5lgHGA=
-github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.21.1 h1:1hWFp+52Vq8Fevy/KUhbW/1MEApMz7uitCF/PQXRJpk=
-github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.21.1/go.mod h1:sIec8j802/rCkCKgZV678HFR0s7lhQUYXT77tIvlaa4=
+github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.22.0 h1:MpkX8EjkwuvyuX9B7+Zgk5M4URb2WQ84Y6jM81n5imw=
+github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.22.0/go.mod h1:4V9Pv5sFfMPWQF0Q0zYN6BlV/504dFGaTeogallRqQw=
+github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.1.2 h1:1q8/WwEqZnM/vO4q1gx2g7lHYmyN+o4P7G6EW4zKbRQ=
+github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.1.2/go.mod h1:owKRexW+Ir5ACD2UTesmjkQ+w7mcmknLNfwOiKfVLTg=
 github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.17 h1:xOLELNKGp2vsiteLsvLPwxC+mYmO6OZ8PYgiuPJzF8U=
 github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.17/go.mod h1:5M5CI3D12dNOtH3/mk6minaRwI2/37ifCURZISxA/IQ=
 github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.17 h1:WWLqlh79iO48yLkj1v3ISRNiv+3KdQoZ6JWyfcsyQik=
