You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by at...@apache.org on 2026/07/15 18:56:51 UTC
(iceberg-go) branch main updated: WitMaxConcurrency rewrite (#1462)
This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 43989fe8 WitMaxConcurrency rewrite (#1462)
43989fe8 is described below
commit 43989fe895f31eb881451ba1e76f95b9a948b063
Author: Alex Stephen <13...@users.noreply.github.com>
AuthorDate: Wed Jul 15 11:56:46 2026 -0700
WitMaxConcurrency rewrite (#1462)
(I think we should have a discussion on this one)
We had a ScanOption named `WitMaxConcurrency`, which is obviously a
typo. But, the correct name conflicts with an option in
orpan_cleanups.go called `WithMaxConcurrency`.
I've renamed `WithMaxConcurrency` -> `WithScanMaxConcurrency`. Ideally,
this conflict shouldn't happen in the first place, but we'll have to do
some package renames to make that happen.
---
table/orphan_cleanup.go | 4 ++--
table/orphan_cleanup_integration_test.go | 2 +-
table/orphan_cleanup_test.go | 4 ++--
table/rewrite_data_files.go | 7 ++-----
table/scan_options_test.go | 31 +++++++++++++++++++++++++++++++
table/table.go | 12 ++++++++++--
website/src/api.md | 2 +-
website/src/configuration.md | 2 +-
8 files changed, 50 insertions(+), 14 deletions(-)
diff --git a/table/orphan_cleanup.go b/table/orphan_cleanup.go
index 6f8a0db6..586f7988 100644
--- a/table/orphan_cleanup.go
+++ b/table/orphan_cleanup.go
@@ -116,10 +116,10 @@ func WithDeleteFunc(deleteFunc func(string) error) OrphanCleanupOption {
}
}
-// WithMaxConcurrency sets the maximum number of goroutines for parallel deletion.
+// WithCleanupMaxConcurrency sets the maximum number of goroutines for parallel deletion.
// Defaults to a reasonable number based on the system. Only used when deleteFunc is nil or when
// the FileIO doesn't support bulk operations.
-func WithMaxConcurrency(maxWorkers int) OrphanCleanupOption {
+func WithCleanupMaxConcurrency(maxWorkers int) OrphanCleanupOption {
return func(cfg *orphanCleanupConfig) {
if maxWorkers > 0 {
cfg.maxConcurrency = maxWorkers
diff --git a/table/orphan_cleanup_integration_test.go b/table/orphan_cleanup_integration_test.go
index 9acb32d6..0b8e0dc3 100644
--- a/table/orphan_cleanup_integration_test.go
+++ b/table/orphan_cleanup_integration_test.go
@@ -359,7 +359,7 @@ func (s *OrphanCleanupIntegrationSuite) TestOrphanCleanupWithConcurrency() {
// Run orphan cleanup with specific concurrency
result, err := tbl.DeleteOrphanFiles(s.ctx,
- table.WithMaxConcurrency(tc.concurrency),
+ table.WithCleanupMaxConcurrency(tc.concurrency),
table.WithFilesOlderThan(0),
table.WithDryRun(false),
)
diff --git a/table/orphan_cleanup_test.go b/table/orphan_cleanup_test.go
index d517ba52..3278bcd4 100644
--- a/table/orphan_cleanup_test.go
+++ b/table/orphan_cleanup_test.go
@@ -71,7 +71,7 @@ func TestOrphanCleanupOptions(t *testing.T) {
WithDeleteFunc(deleteFunc)(cfg)
assert.NotNil(t, cfg.deleteFunc)
- WithMaxConcurrency(8)(cfg)
+ WithCleanupMaxConcurrency(8)(cfg)
assert.Equal(t, 8, cfg.maxConcurrency)
WithPrefixMismatchMode(PrefixMismatchIgnore)(cfg)
@@ -772,7 +772,7 @@ func TestDeleteOrphanFilesPopulatesOrphanFileSizes(t *testing.T) {
result, err := tbl.DeleteOrphanFiles(context.Background(),
WithDryRun(true),
WithLocation("s3://bucket/table"),
- WithMaxConcurrency(1),
+ WithCleanupMaxConcurrency(1),
WithFilesOlderThan(0), // Consider files created before the scan.
)
diff --git a/table/rewrite_data_files.go b/table/rewrite_data_files.go
index d78c2cb2..4bb07623 100644
--- a/table/rewrite_data_files.go
+++ b/table/rewrite_data_files.go
@@ -189,10 +189,7 @@ func WithCompactionTargetFileSize(size int64) CompactionGroupOption {
// WithCompactionScanConcurrency sets the scan concurrency used when
// reading the group's tasks. Forwarded to [Table.Scan] as
-// [WitMaxConcurrency]. Zero (the default) means runtime.GOMAXPROCS.
-//
-// TODO: the [WitMaxConcurrency] link enshrines a pre-existing typo
-// (missing `h`). Update this reference when that symbol is renamed.
+// [WithMaxConcurrency]. Zero (the default) means runtime.GOMAXPROCS.
func WithCompactionScanConcurrency(n int) CompactionGroupOption {
return func(c *compactionGroupConfig) {
c.scanConcurrency = n
@@ -297,7 +294,7 @@ func ExecuteCompactionGroup(ctx context.Context, tbl *Table, group CompactionTas
var scanOpts []ScanOption
if cfg.scanConcurrency > 0 {
- scanOpts = append(scanOpts, WitMaxConcurrency(cfg.scanConcurrency))
+ scanOpts = append(scanOpts, WithMaxConcurrency(cfg.scanConcurrency))
}
// Preserve row lineage only when every source file in the group carries
diff --git a/table/scan_options_test.go b/table/scan_options_test.go
index ac42d8a8..d8ed5a15 100644
--- a/table/scan_options_test.go
+++ b/table/scan_options_test.go
@@ -35,3 +35,34 @@ func TestWithSelectedFieldsDoesNotAliasCallerSlice(t *testing.T) {
assert.Equal(t, []string{"id", "name"}, scan.selectedFields)
assert.NotEqual(t, fields[0], scan.selectedFields[0])
}
+
+func TestWithMaxConcurrency(t *testing.T) {
+ cases := []struct {
+ name string
+ n int
+ want int
+ }{
+ {"positive sets concurrency", 4, 4},
+ {"zero is a no-op", 0, 0},
+ {"negative is a no-op", -1, 0},
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ scan := &Scan{}
+ WithMaxConcurrency(tc.n)(scan)
+ assert.Equal(t, tc.want, scan.concurrency)
+ })
+ }
+}
+
+// The deprecated WitMaxConcurrency alias must stay in lock-step with the
+// canonical WithMaxConcurrency so callers on the old name don't drift.
+func TestWitMaxConcurrencyAliasParity(t *testing.T) {
+ for _, n := range []int{-1, 0, 1, 8} {
+ canonical, alias := &Scan{}, &Scan{}
+ WithMaxConcurrency(n)(canonical)
+ WitMaxConcurrency(n)(alias)
+ assert.Equal(t, canonical.concurrency, alias.concurrency, "n=%d", n)
+ }
+}
diff --git a/table/table.go b/table/table.go
index 2d5d7be4..c2d2af41 100644
--- a/table/table.go
+++ b/table/table.go
@@ -866,9 +866,9 @@ func WithLimit(n int64) ScanOption {
}
}
-// WitMaxConcurrency sets the maximum concurrency for table scan and plan
+// WithMaxConcurrency sets the maximum concurrency for table scan and plan
// operations. When unset it defaults to runtime.GOMAXPROCS.
-func WitMaxConcurrency(n int) ScanOption {
+func WithMaxConcurrency(n int) ScanOption {
if n <= 0 {
return noopOption
}
@@ -878,6 +878,14 @@ func WitMaxConcurrency(n int) ScanOption {
}
}
+// WitMaxConcurrency is a deprecated alias for [WithMaxConcurrency], kept for
+// backward compatibility with the pre-existing typo'd name.
+//
+// Deprecated: use [WithMaxConcurrency].
+func WitMaxConcurrency(n int) ScanOption {
+ return WithMaxConcurrency(n)
+}
+
func WithOptions(opts iceberg.Properties) ScanOption {
if opts == nil {
return noopOption
diff --git a/website/src/api.md b/website/src/api.md
index d00c3e41..1149fd6a 100644
--- a/website/src/api.md
+++ b/website/src/api.md
@@ -490,7 +490,7 @@ import (
result, err := tbl.DeleteOrphanFiles(ctx,
table.WithFilesOlderThan(72*time.Hour),
table.WithDryRun(false),
- table.WithMaxConcurrency(8),
+ table.WithCleanupMaxConcurrency(8),
)
if err != nil { /* ... */ }
fmt.Printf("removed %d files\n", len(result.DeletedFiles))
diff --git a/website/src/configuration.md b/website/src/configuration.md
index 98ec21e6..1d3b97c6 100644
--- a/website/src/configuration.md
+++ b/website/src/configuration.md
@@ -199,7 +199,7 @@ There is no `PYICEBERG_*`-style env var convention. Use the YAML config file or
| Setting | Source | Effect |
|---|---|---|
| `max-workers` in `~/.iceberg-go.yaml` (`config.EnvConfig.MaxWorkers`) | YAML config | Worker pool size used by parallel column writes, snapshot producers, scan plan, equality-delete writers. Default `5`. |
-| `WitMaxConcurrency(n int) ScanOption` | Code (`table.WitMaxConcurrency`) | Per-scan override. Note: function name is `Wit...` (not `With...`) - this is a pre-existing typo in the public API. |
+| `WithMaxConcurrency(n int) ScanOption` | Code (`table.WithMaxConcurrency`) | Per-scan override. `WitMaxConcurrency` (missing `h`) remains as a deprecated alias. |
| `WithMaxWriteWorkers(n int)` | Code (per-write API on `WriteRecords`) | Per-write override of the worker count. |
| `WithClusteredWrite()` | Code (per-write API on `WriteRecords`) | Forces single-threaded writes. Mutually exclusive with `WithMaxWriteWorkers`. |