update parallel

Author: gouhongshen

pkg/fileservice/file_service.go

@@ -97,11 +97,6 @@ type IOVector struct {
 
 	// Caches indicates extra caches to operate on
 	Caches []IOVectorCache
-
-	// DisableParallel controls whether to skip parallel multipart uploads even if supported.
-	DisableParallel bool
-	// ForceParallel controls whether to try parallel multipart uploads when possible.
-	ForceParallel bool
 }
 
 type IOEntry struct {

pkg/fileservice/get.go

@@ -53,6 +53,16 @@ func Get[T any](fs FileService, name string) (res T, err error) {
 
 var NoDefaultCredentialsForETL = os.Getenv("MO_NO_DEFAULT_CREDENTIALS") != ""
 
+func etlParallelMode(ctx context.Context) ParallelMode {
+	if mode, ok := parallelModeFromContext(ctx); ok {
+		return mode
+	}
+	if mode, ok := parseParallelMode(strings.TrimSpace(os.Getenv("MO_ETL_PARALLEL_MODE"))); ok {
+		return mode
+	}
+	return ParallelOff
+}
+
 // GetForETL get or creates a FileService instance for ETL operations
 // if service part of path is empty, a LocalETLFS will be created
 // if service part of path is not empty, a ETLFileService typed instance will be extracted from fs argument
@@ -110,6 +120,7 @@ func GetForETL(ctx context.Context, fs FileService, path string) (res ETLFileSer
 					KeySecret:          accessSecret,
 					KeyPrefix:          keyPrefix,
 					Name:               name,
+					ParallelMode:       etlParallelMode(ctx),
 				},
 				DisabledCacheConfig,
 				nil,
@@ -143,6 +154,7 @@ func GetForETL(ctx context.Context, fs FileService, path string) (res ETLFileSer
 					Bucket:             bucket,
 					KeyPrefix:          keyPrefix,
 					Name:               name,
+					ParallelMode:       etlParallelMode(ctx),
 				},
 				DisabledCacheConfig,
 				nil,
@@ -157,6 +169,7 @@ func GetForETL(ctx context.Context, fs FileService, path string) (res ETLFileSer
 			}
 			args.NoBucketValidation = true
 			args.IsHDFS = fsPath.Service == "hdfs"
+			args.ParallelMode = etlParallelMode(ctx)
 			res, err = NewS3FS(
 				ctx,
 				args,
@@ -198,6 +211,7 @@ func GetForETL(ctx context.Context, fs FileService, path string) (res ETLFileSer
 					KeyPrefix:          keyPrefix,
 					Name:               name,
 					IsMinio:            true,
+					ParallelMode:       etlParallelMode(ctx),
 				},
 				DisabledCacheConfig,
 				nil,

pkg/fileservice/object_storage_arguments.go

@@ -28,12 +28,13 @@ import (
 
 type ObjectStorageArguments struct {
 	// misc
-	Name                 string `toml:"name"`
-	KeyPrefix            string `toml:"key-prefix"`
-	SharedConfigProfile  string `toml:"shared-config-profile"`
-	NoDefaultCredentials bool   `toml:"no-default-credentials"`
-	NoBucketValidation   bool   `toml:"no-bucket-validation"`
-	Concurrency          int64  `toml:"concurrency"`
+	Name                 string       `toml:"name"`
+	KeyPrefix            string       `toml:"key-prefix"`
+	SharedConfigProfile  string       `toml:"shared-config-profile"`
+	NoDefaultCredentials bool         `toml:"no-default-credentials"`
+	NoBucketValidation   bool         `toml:"no-bucket-validation"`
+	Concurrency          int64        `toml:"concurrency"`
+	ParallelMode         ParallelMode `toml:"parallel-mode"`
 
 	// s3
 	Bucket    string   `toml:"bucket"`
@@ -101,6 +102,10 @@ func (o *ObjectStorageArguments) SetFromString(arguments []string) error {
 			if err == nil {
 				o.Concurrency = n
 			}
+		case "parallel-mode", "parallel":
+			if mode, ok := parseParallelMode(value); ok {
+				o.ParallelMode = mode
+			}
 
 		case "bucket":
 			o.Bucket = value

pkg/fileservice/parallel_mode.go

@@ -0,0 +1,62 @@
+// Copyright 2025 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package fileservice
+
+import (
+	"context"
+	"strings"
+)
+
+// ParallelMode controls when multipart parallel uploads are used.
+type ParallelMode uint8
+
+const (
+	ParallelOff ParallelMode = iota
+	ParallelAuto
+	ParallelForce
+)
+
+func parseParallelMode(s string) (ParallelMode, bool) {
+	switch strings.ToLower(s) {
+	case "off", "false", "0", "":
+		return ParallelOff, true
+	case "auto":
+		return ParallelAuto, true
+	case "force", "on", "true", "1":
+		return ParallelForce, true
+	default:
+		return ParallelOff, false
+	}
+}
+
+type parallelModeKey struct{}
+
+// WithParallelMode sets a per-call parallel mode override on context.
+func WithParallelMode(ctx context.Context, mode ParallelMode) context.Context {
+	return context.WithValue(ctx, parallelModeKey{}, mode)
+}
+
+// parallelModeFromContext retrieves a parallel mode override if present.
+func parallelModeFromContext(ctx context.Context) (ParallelMode, bool) {
+	if ctx == nil {
+		return ParallelOff, false
+	}
+	if v := ctx.Value(parallelModeKey{}); v != nil {
+		if mode, ok := v.(ParallelMode); ok {
+			return mode, true
+		}
+	}
+	return ParallelOff, false
+}

pkg/fileservice/parallel_mode_test.go

@@ -0,0 +1,74 @@
+// Copyright 2025 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package fileservice
+
+import (
+	"context"
+	"testing"
+)
+
+func TestParseParallelModeVariants(t *testing.T) {
+	cases := []struct {
+		in     string
+		expect ParallelMode
+		ok     bool
+	}{
+		{"", ParallelOff, true},
+		{"off", ParallelOff, true},
+		{"false", ParallelOff, true},
+		{"0", ParallelOff, true},
+		{"auto", ParallelAuto, true},
+		{"force", ParallelForce, true},
+		{"on", ParallelForce, true},
+		{"1", ParallelForce, true},
+		{"xxx", ParallelOff, false},
+	}
+
+	for _, c := range cases {
+		mode, ok := parseParallelMode(c.in)
+		if mode != c.expect || ok != c.ok {
+			t.Fatalf("input %s got (%v,%v) expect (%v,%v)", c.in, mode, ok, c.expect, c.ok)
+		}
+	}
+}
+
+func TestWithParallelModeOnContext(t *testing.T) {
+	base := context.Background()
+	ctx := WithParallelMode(base, ParallelForce)
+	mode, ok := parallelModeFromContext(ctx)
+	if !ok || mode != ParallelForce {
+		t.Fatalf("expected force override, got %v %v", mode, ok)
+	}
+	if _, ok := parallelModeFromContext(base); ok {
+		t.Fatalf("unexpected mode on base context")
+	}
+}
+
+func TestETLParallelModePriority(t *testing.T) {
+	t.Setenv("MO_ETL_PARALLEL_MODE", "off")
+	if m := etlParallelMode(context.Background()); m != ParallelOff {
+		t.Fatalf("expect off from env, got %v", m)
+	}
+
+	ctx := WithParallelMode(context.Background(), ParallelForce)
+	if m := etlParallelMode(ctx); m != ParallelForce {
+		t.Fatalf("ctx override should win, got %v", m)
+	}
+
+	t.Setenv("MO_ETL_PARALLEL_MODE", "auto")
+	if m := etlParallelMode(context.Background()); m != ParallelAuto {
+		t.Fatalf("env auto not applied, got %v", m)
+	}
+}