fix: configurable part size (#1116)

Co-authored-by: Fabrizio <fabri.feno@gmail.com>
Signed-off-by: ferhat elmas <elmas.ferhat@gmail.com>

Author: ferhatelmas

src/config.ts

@@ -4,6 +4,9 @@ import { SignJWT } from 'jose'
 
 export type StorageBackendType = 'file' | 's3'
 export type IcebergCatalogAuthType = 'sigv4' | 'token'
+const DEFAULT_S3_UPLOAD_PART_SIZE = 16 * 1024 * 1024
+const MIN_S3_UPLOAD_PART_SIZE = 5 * 1024 * 1024
+
 export enum MultitenantMigrationStrategy {
   PROGRESSIVE = 'progressive',
   ON_REQUEST = 'on_request',
@@ -68,6 +71,7 @@ type StorageConfigType = {
   storageS3InternalTracesEnabled?: boolean
   storageS3MaxSockets: number
   storageS3DisableChecksum: boolean
+  storageS3UploadPartSize: number
   storageS3UploadQueueSize: number
   storageS3Bucket: string
   storageS3Endpoint?: string
@@ -374,6 +378,11 @@ export function getConfig(options?: { reload?: boolean }): StorageConfigType {
       10
     ),
     storageS3DisableChecksum: getOptionalConfigFromEnv('STORAGE_S3_DISABLE_CHECKSUM') === 'true',
+    storageS3UploadPartSize: Math.max(
+      envNumber(getOptionalConfigFromEnv('STORAGE_S3_UPLOAD_PART_SIZE')) ??
+        DEFAULT_S3_UPLOAD_PART_SIZE,
+      MIN_S3_UPLOAD_PART_SIZE
+    ),
     storageS3UploadQueueSize:
       envNumber(getOptionalConfigFromEnv('STORAGE_S3_UPLOAD_QUEUE_SIZE')) ?? 2,
     storageS3InternalTracesEnabled:

src/storage/backend/s3/adapter.test.ts

@@ -8,6 +8,8 @@ import { getConfig } from '../../../config'
 import { withOptionalVersion } from '../adapter'
 import { MAX_PUT_OBJECT_SIZE, S3Backend } from './adapter'
 
+const DEFAULT_S3_UPLOAD_PART_SIZE = 16 * 1024 * 1024
+
 vi.mock('@aws-sdk/client-s3', async () => {
   const originalModule =
     await vi.importActual<typeof import('@aws-sdk/client-s3')>('@aws-sdk/client-s3')
@@ -35,6 +37,7 @@ vi.mock('@aws-sdk/s3-request-presigner', () => ({
 }))
 
 type UploadOptionsShape = {
+  partSize?: number
   queueSize?: number
 }
 
@@ -322,6 +325,8 @@ describe('S3Backend', () => {
       )
 
       expect(Upload).toHaveBeenCalledTimes(1)
+      expect(getConfig().storageS3UploadPartSize).toBe(DEFAULT_S3_UPLOAD_PART_SIZE)
+      expect(uploadInstances[0].options.partSize).toBe(getConfig().storageS3UploadPartSize)
       expect(uploadInstances[0].options.queueSize).toBe(getConfig().storageS3UploadQueueSize)
       expect(mockSend).toHaveBeenCalledTimes(1)
       expect(mockSend.mock.calls[0][0]).toBeInstanceOf(HeadObjectCommand)
@@ -348,6 +353,8 @@ describe('S3Backend', () => {
       )
 
       expect(Upload).toHaveBeenCalledTimes(1)
+      expect(getConfig().storageS3UploadPartSize).toBe(DEFAULT_S3_UPLOAD_PART_SIZE)
+      expect(uploadInstances[0].options.partSize).toBe(getConfig().storageS3UploadPartSize)
       expect(uploadInstances[0].options.queueSize).toBe(getConfig().storageS3UploadQueueSize)
       expect(mockSend).not.toHaveBeenCalled()
       expect(result).toMatchObject({

src/storage/backend/s3/adapter.ts

@@ -35,6 +35,7 @@ import {
 } from './../adapter'
 
 const {
+  storageS3UploadPartSize,
   storageS3UploadQueueSize,
   tracingFeatures,
   storageS3MaxSockets,
@@ -249,6 +250,7 @@ export class S3Backend implements StorageBackendAdapter {
 
     const upload = new Upload({
       client: this.client,
+      partSize: storageS3UploadPartSize,
       queueSize: storageS3UploadQueueSize,
       params: {
         Bucket: bucketName,