Test/esti lakectl large upload

.github/workflows/lakectl-compatibility-tests.yaml

@@ -125,6 +125,7 @@ jobs:
         env:
           AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
           AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
+          AWS_REGION: us-east-1
           ESTI_AWS_ACCESS_KEY_ID: ${{ secrets.ESTI_AWS_ACCESS_KEY_ID }}
           ESTI_AWS_SECRET_ACCESS_KEY: ${{ secrets.ESTI_AWS_SECRET_ACCESS_KEY }}
           ESTI_BINARIES_DIR: ${{ github.workspace }}/bin

esti/adapter_utils.go

@@ -35,6 +35,9 @@ type ClientAdapter interface {
 	// Download returns a Reader to download data from physicalAddress.  The Close method
 	// of that Reader can return errors!
 	Download(ctx context.Context, physicalAddress *url.URL) (io.ReadCloser, error)
+
+	// Delete removes the object at physicalAddress from the backing store.
+	Delete(ctx context.Context, physicalAddress *url.URL) error
 }
 
 type AdapterFactory map[string]func(context.Context) (ClientAdapter, error)
@@ -110,3 +113,16 @@ func (s *s3Adapter) Download(ctx context.Context, physicalAddress *url.URL) (io.
 	}
 	return getObjectResponse.Body, nil
 }
+
+func (s *s3Adapter) Delete(ctx context.Context, physicalAddress *url.URL) error {
+	if physicalAddress.Scheme != s3Scheme {
+		return fmt.Errorf("%s: %w", s3Scheme, helpers.ErrUnsupportedProtocol)
+	}
+	bucket := physicalAddress.Hostname()
+	key := strings.TrimPrefix(physicalAddress.Path, "/")
+	_, err := s.svc.DeleteObject(ctx, &s3.DeleteObjectInput{
+		Bucket: aws.String(bucket),
+		Key:    aws.String(key),
+	})
+	return err
+}

esti/lakectl_test.go

@@ -2,11 +2,13 @@ package esti
 
 import (
 	"fmt"
+	"net/url"
 	"os"
 	"strings"
 	"testing"
 	"time"
 
+	"github.com/go-openapi/swag"
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/require"
 	"github.com/treeverse/lakefs/pkg/api/apigen"
@@ -162,6 +164,62 @@ func TestLakectlPreSignUpload(t *testing.T) {
 	})
 }
 
+func TestLakectlFsUploadLarge(t *testing.T) {
+	RequireBlockstoreType(t, block.BlockstoreTypeS3)
+	ctx := t.Context()
+
+	repoName := GenerateUniqueRepositoryName()
+	storage := GenerateUniqueStorageNamespace(repoName)
+	vars := map[string]string{
+		"REPO":    repoName,
+		"STORAGE": storage,
+		"BRANCH":  mainBranch,
+	}
+	RunCmdAndVerifySuccessWithFile(t, Lakectl()+" repo create lakefs://"+repoName+" "+storage, false, "lakectl_repo_create", vars)
+
+	// S3 rejects single-part uploads above 5GiB, so one byte more forces
+	// the multipart path (#9284). A sparse file keeps disk usage low.
+	const largeFileSize = 5<<30 + 1
+	f, err := os.CreateTemp("", "lakectl-large-upload-")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, os.Remove(f.Name()))
+	})
+	require.NoError(t, f.Truncate(largeFileSize))
+	require.NoError(t, f.Close())
+
+	uploadAndVerify := func(t *testing.T, objPath, extraArgs string) {
+		t.Helper()
+		// deleting the repository does not remove its data, so remove the large object from the storage
+		t.Cleanup(func() {
+			statResp, err := client.StatObjectWithResponse(ctx, repoName, mainBranch, &apigen.StatObjectParams{Path: objPath})
+			require.NoError(t, err)
+			if statResp.JSON200 == nil {
+				return
+			}
+			physicalAddress, err := url.Parse(statResp.JSON200.PhysicalAddress)
+			require.NoError(t, err)
+			adapter, err := NewAdapter(ctx, physicalAddress.Scheme)
+			require.NoError(t, err)
+			require.NoError(t, adapter.Delete(ctx, physicalAddress))
+		})
+
+		RunCmdAndVerifyContainsText(t, Lakectl()+" fs upload -s "+f.Name()+" lakefs://"+repoName+"/"+mainBranch+"/"+objPath+extraArgs, false, fmt.Sprintf("Size: %d bytes", largeFileSize), vars)
+
+		statResp, err := client.StatObjectWithResponse(ctx, repoName, mainBranch, &apigen.StatObjectParams{Path: objPath})
+		require.NoError(t, err)
+		require.NotNil(t, statResp.JSON200, "stat %s: %s", objPath, statResp.Status())
+		require.Equal(t, int64(largeFileSize), swag.Int64Value(statResp.JSON200.SizeBytes))
+	}
+
+	t.Run("pre-sign", func(t *testing.T) {
+		uploadAndVerify(t, "large_file_presign", " --pre-sign")
+	})
+	t.Run("no pre-sign", func(t *testing.T) {
+		uploadAndVerify(t, "large_file", "")
+	})
+}
+
 func TestLakectlCommit(t *testing.T) {
 	repoName := GenerateUniqueRepositoryName()
 	storage := GenerateUniqueStorageNamespace(repoName)