Added Parallelism configuration option to lakectl

cmd/lakectl/cmd/fs_download.go

@@ -105,6 +105,7 @@ var fsDownloadCmd = &cobra.Command{
 
 		s := local.NewSyncManager(ctx, client, getHTTPClient(), local.Config{
 			SyncFlags:           syncFlags,
+			Parallelism:         cfg.Options.Parallelism,
 			SkipNonRegularFiles: cfg.Local.SkipNonRegularFiles,
 			IncludePerm:         false,
 		})

cmd/lakectl/cmd/fs_upload.go

@@ -63,6 +63,7 @@ var fsUploadCmd = &cobra.Command{
 		}()
 		s := local.NewSyncManager(ctx, client, getHTTPClient(), local.Config{
 			SyncFlags:           syncFlags,
+			Parallelism:         cfg.Options.Parallelism,
 			SkipNonRegularFiles: cfg.Local.SkipNonRegularFiles,
 			IncludePerm:         false,
 		})

cmd/lakectl/cmd/local_checkout.go

@@ -59,6 +59,7 @@ func localCheckout(cmd *cobra.Command, localPath string, specifiedRef string, co
 	sigCtx := localHandleSyncInterrupt(cmd.Context(), idx, string(checkoutOperation))
 	syncMgr := local.NewSyncManager(sigCtx, client, getHTTPClient(), local.Config{
 		SyncFlags:           syncFlags,
+		Parallelism:         cfg.Options.Parallelism,
 		SkipNonRegularFiles: cfg.Local.SkipNonRegularFiles,
 		IncludePerm:         cfg.Experimental.Local.POSIXPerm.Enabled,
 		IncludeUID:          cfg.Experimental.Local.POSIXPerm.IncludeUID,

cmd/lakectl/cmd/local_clone.go

@@ -88,6 +88,7 @@ var localCloneCmd = &cobra.Command{
 		sigCtx := localHandleSyncInterrupt(ctx, idx, string(cloneOperation))
 		s := local.NewSyncManager(sigCtx, client, getHTTPClient(), local.Config{
 			SyncFlags:           syncFlags,
+			Parallelism:         cfg.Options.Parallelism,
 			SkipNonRegularFiles: cfg.Local.SkipNonRegularFiles,
 			IncludePerm:         cfg.Experimental.Local.POSIXPerm.Enabled,
 			IncludeUID:          cfg.Experimental.Local.POSIXPerm.IncludeUID,

cmd/lakectl/cmd/local_commit.go

@@ -174,6 +174,7 @@ var localCommitCmd = &cobra.Command{
 		sigCtx := localHandleSyncInterrupt(cmd.Context(), idx, string(commitOperation))
 		s := local.NewSyncManager(sigCtx, client, getHTTPClient(), local.Config{
 			SyncFlags:           syncFlags,
+			Parallelism:         cfg.Options.Parallelism,
 			SkipNonRegularFiles: cfg.Local.SkipNonRegularFiles,
 			IncludePerm:         cfg.Experimental.Local.POSIXPerm.Enabled,
 			IncludeUID:          cfg.Experimental.Local.POSIXPerm.IncludeUID,

cmd/lakectl/cmd/local_pull.go

@@ -68,6 +68,7 @@ var localPullCmd = &cobra.Command{
 		sigCtx := localHandleSyncInterrupt(cmd.Context(), idx, string(pullOperation))
 		s := local.NewSyncManager(sigCtx, client, getHTTPClient(), local.Config{
 			SyncFlags:           syncFlags,
+			Parallelism:         cfg.Options.Parallelism,
 			SkipNonRegularFiles: cfg.Local.SkipNonRegularFiles,
 			IncludePerm:         cfg.Experimental.Local.POSIXPerm.Enabled,
 			IncludeUID:          cfg.Experimental.Local.POSIXPerm.IncludeUID,

cmd/lakectl/cmd/root.go

@@ -72,6 +72,9 @@ type Configuration struct {
 		EndpointURL lakefsconfig.OnlyString `mapstructure:"endpoint_url"`
 		Retries     RetriesCfg              `mapstructure:"retries"`
 	} `mapstructure:"server"`
+	Options struct {
+		Parallelism int `mapstructure:"parallelism"`
+	} `mapstructure:"options"`
 	Metastore struct {
 		Type lakefsconfig.OnlyString `mapstructure:"type"`
 		Hive struct {
@@ -153,9 +156,9 @@ const (
 	parallelismFlagName   = "parallelism"
 	noProgressBarFlagName = "no-progress"
 
-	defaultSyncParallelism = 25
-	defaultSyncPresign     = true
-	defaultNoProgress      = false
+	defaultParallelism = 25
+	defaultSyncPresign = true
+	defaultNoProgress  = false
 
 	myRepoExample   = "lakefs://my-repo"
 	myBucketExample = "s3://my-bucket"
@@ -178,7 +181,7 @@ func withRecursiveFlag(cmd *cobra.Command, usage string) {
 }
 
 func withParallelismFlag(cmd *cobra.Command) {
-	cmd.Flags().IntP(parallelismFlagName, "p", defaultSyncParallelism,
+	cmd.Flags().IntP(parallelismFlagName, "p", defaultParallelism,
 		"Max concurrent operations to perform")
 }
 
@@ -245,9 +248,11 @@ func getSyncFlags(cmd *cobra.Command, client *apigen.ClientWithResponses) local.
 	if parallelism < 1 {
 		DieFmt("Invalid value for parallelism (%d), minimum is 1.\n", parallelism)
 	}
+	setParallelism := cmd.Flags().Changed(parallelismFlagName)
 
 	presignMode := getPresignMode(cmd, client)
 	return local.SyncFlags{
+		SetParallelism:   setParallelism,
 		Parallelism:      parallelism,
 		Presign:          presignMode.Enabled,
 		PresignMultipart: presignMode.Multipart,
@@ -574,6 +579,7 @@ func initConfig() {
 	viper.SetDefault("server.retries.min_wait_interval", defaultMinRetryInterval)
 	viper.SetDefault("experimental.local.posix_permissions.enabled", false)
 	viper.SetDefault("local.skip_non_regular_files", false)
+	viper.SetDefault("options.parallelism", defaultParallelism)
 
 	cfgErr = viper.ReadInConfig()
 }

pkg/local/config.go

@@ -9,6 +9,7 @@ const (
 )
 
 type SyncFlags struct {
+	SetParallelism   bool
 	Parallelism      int
 	Presign          bool
 	PresignMultipart bool
@@ -17,6 +18,7 @@ type SyncFlags struct {
 
 type Config struct {
 	SyncFlags
+	Parallelism int
 	// SkipNonRegularFiles - By default lakectl local fails if local directory contains irregular files. When set, lakectl will skip these files instead.
 	SkipNonRegularFiles bool
 	// IncludePerm - Experimental: preserve Unix file permissions

pkg/local/sync.go

@@ -74,7 +74,11 @@ func (s *SyncManager) Sync(rootPath string, remote *uri.URI, changeSet <-chan *C
 	defer s.progressBar.Stop()
 
 	wg, ctx := errgroup.WithContext(s.ctx)
-	for i := 0; i < s.cfg.SyncFlags.Parallelism; i++ {
+	parallelism := s.cfg.Parallelism
+	if s.cfg.SyncFlags.SetParallelism {
+		parallelism = s.cfg.SyncFlags.Parallelism
+	}
+	for i := 0; i < parallelism; i++ {
 		wg.Go(func() error {
 			for change := range changeSet {
 				if err := s.apply(ctx, rootPath, remote, change); err != nil {