filer leveldb2: saturate size options instead of overflowing int

The option fields take int, so a MiB config value past math.MaxInt/MiB
(e.g. 2048 on 32-bit) wrapped negative and goleveldb would panic sizing
the memtable. Clamp the product instead of the input.

weed/command/scaffold/filer.toml

@@ -33,6 +33,11 @@ recursive_delete = false
 # faster than previous leveldb, recommended.
 enabled = true
 dir = "./filerldb2"                    # directory to store level db files
+# optional tuning, applied to each of the 8 partitioned stores
+# block_cache_size_mb = 32             # sorted-table read cache
+# write_buffer_size_mb = 16            # memtable size before flushing to disk
+# block_size_kb = 4                    # uncompressed block ('page') size inside table files
+# compaction_table_size_mb = 2         # target size of generated table files
 
 [leveldb3]
 # similar to leveldb2.

weed/command/scaffold/notification.toml

@@ -81,16 +81,3 @@ workers = 5                                        # optional: concurrent worker
 buffer_size = 10000                                # optional: event buffer size (default: 10000, range: 100-1000000)
 # event_types = ["create", "update", "delete", "rename"]  # optional: filter by event types (default: all)
 # path_prefixes = ["/important", "/data"]          # optional: filter by path prefixes (default: all)
-#
-# Additional endpoints can be declared as named subsections, each with its own
-# endpoint, filters, retry and worker settings, e.g.:
-# [notification.webhook.file_ingress]
-# enabled = true
-# endpoint = "https://ingress.example.com/hook"
-# event_types = ["create"]
-# path_prefixes = ["/buckets/uploads/data"]
-#
-# [notification.webhook.metadata_extractor]
-# enabled = true
-# endpoint = "https://metadata.example.com/hook"
-# path_prefixes = ["/buckets/repository"]

weed/filer/leveldb2/leveldb2_store.go

@@ -6,6 +6,7 @@ import (
 	"crypto/md5"
 	"fmt"
 	"io"
+	"math"
 	"os"
 
 	"github.com/syndtr/goleveldb/leveldb"
@@ -28,6 +29,11 @@ type LevelDB2Store struct {
 	dbs      []*leveldb.DB
 	dbCount  int
 	ReadOnly bool
+
+	blockCacheSizeMB      int
+	writeBufferSizeMB     int
+	blockSizeKB           int
+	compactionTableSizeMB int
 }
 
 func (store *LevelDB2Store) GetName() string {
@@ -36,9 +42,44 @@ func (store *LevelDB2Store) GetName() string {
 
 func (store *LevelDB2Store) Initialize(configuration weed_util.Configuration, prefix string) (err error) {
 	dir := configuration.GetString(prefix + "dir")
+	store.blockCacheSizeMB = configuration.GetInt(prefix + "block_cache_size_mb")
+	store.writeBufferSizeMB = configuration.GetInt(prefix + "write_buffer_size_mb")
+	store.blockSizeKB = configuration.GetInt(prefix + "block_size_kb")
+	store.compactionTableSizeMB = configuration.GetInt(prefix + "compaction_table_size_mb")
 	return store.initialize(dir, 8)
 }
 
+func (store *LevelDB2Store) buildOptions() *opt.Options {
+	opts := &opt.Options{
+		BlockCacheCapacity: 32 * 1024 * 1024,         // default value is 8MiB
+		WriteBuffer:        16 * 1024 * 1024,         // default value is 4MiB
+		Filter:             filter.NewBloomFilter(8), // false positive rate 0.02
+		ReadOnly:           store.ReadOnly,
+	}
+	if store.blockCacheSizeMB > 0 {
+		opts.BlockCacheCapacity = scaled(store.blockCacheSizeMB, 1<<20)
+	}
+	if store.writeBufferSizeMB > 0 {
+		opts.WriteBuffer = scaled(store.writeBufferSizeMB, 1<<20)
+	}
+	if store.blockSizeKB > 0 {
+		opts.BlockSize = scaled(store.blockSizeKB, 1<<10)
+	}
+	if store.compactionTableSizeMB > 0 {
+		opts.CompactionTableSize = scaled(store.compactionTableSizeMB, 1<<20)
+	}
+	return opts
+}
+
+// scaled converts a size option to bytes, saturating instead of wrapping a
+// huge value into a negative int on 32-bit builds.
+func scaled(v, scale int) int {
+	if v > math.MaxInt/scale {
+		return math.MaxInt
+	}
+	return v * scale
+}
+
 func (store *LevelDB2Store) initialize(dir string, dbCount int) (err error) {
 	glog.Infof("filer store leveldb2 dir: %s", dir)
 	os.MkdirAll(dir, 0755)
@@ -46,12 +87,7 @@ func (store *LevelDB2Store) initialize(dir string, dbCount int) (err error) {
 		return fmt.Errorf("Check Level Folder %s Writable: %s", dir, err)
 	}
 
-	opts := &opt.Options{
-		BlockCacheCapacity: 32 * 1024 * 1024,         // default value is 8MiB
-		WriteBuffer:        16 * 1024 * 1024,         // default value is 4MiB
-		Filter:             filter.NewBloomFilter(8), // false positive rate 0.02
-		ReadOnly:           store.ReadOnly,
-	}
+	opts := store.buildOptions()
 
 	for d := 0; d < dbCount; d++ {
 		dbFolder := fmt.Sprintf("%s/%02d", dir, d)

weed/filer/leveldb2/leveldb2_store_test.go

@@ -2,6 +2,7 @@ package leveldb
 
 import (
 	"context"
+	"math"
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
@@ -62,6 +63,94 @@ func TestCreateAndFind(t *testing.T) {
 
 }
 
+type fakeConfiguration map[string]interface{}
+
+func (c fakeConfiguration) GetString(key string) string {
+	value, _ := c[key].(string)
+	return value
+}
+
+func (c fakeConfiguration) GetBool(key string) bool {
+	value, _ := c[key].(bool)
+	return value
+}
+
+func (c fakeConfiguration) GetInt(key string) int {
+	value, _ := c[key].(int)
+	return value
+}
+
+func (c fakeConfiguration) GetStringSlice(key string) []string {
+	value, _ := c[key].([]string)
+	return value
+}
+
+func (c fakeConfiguration) SetDefault(key string, value interface{}) {
+	if _, found := c[key]; !found {
+		c[key] = value
+	}
+}
+
+func TestInitializeOptions(t *testing.T) {
+	store := &LevelDB2Store{}
+	if err := store.Initialize(fakeConfiguration{
+		"leveldb2.dir":                      t.TempDir(),
+		"leveldb2.block_cache_size_mb":      64,
+		"leveldb2.write_buffer_size_mb":     32,
+		"leveldb2.block_size_kb":            16,
+		"leveldb2.compaction_table_size_mb": 8,
+	}, "leveldb2."); err != nil {
+		t.Fatalf("initialize: %v", err)
+	}
+	defer store.Shutdown()
+
+	opts := store.buildOptions()
+	if opts.BlockCacheCapacity != 64*1024*1024 {
+		t.Errorf("block cache %d, want 64MiB", opts.BlockCacheCapacity)
+	}
+	if opts.WriteBuffer != 32*1024*1024 {
+		t.Errorf("write buffer %d, want 32MiB", opts.WriteBuffer)
+	}
+	if opts.BlockSize != 16*1024 {
+		t.Errorf("block size %d, want 16KiB", opts.BlockSize)
+	}
+	if opts.CompactionTableSize != 8*1024*1024 {
+		t.Errorf("compaction table size %d, want 8MiB", opts.CompactionTableSize)
+	}
+}
+
+func TestInitializeDefaultOptions(t *testing.T) {
+	store := &LevelDB2Store{}
+	if err := store.Initialize(fakeConfiguration{
+		"leveldb2.dir":                  t.TempDir(),
+		"leveldb2.block_cache_size_mb":  -1,
+		"leveldb2.write_buffer_size_mb": 0,
+	}, "leveldb2."); err != nil {
+		t.Fatalf("initialize: %v", err)
+	}
+	defer store.Shutdown()
+
+	opts := store.buildOptions()
+	if opts.BlockCacheCapacity != 32*1024*1024 {
+		t.Errorf("block cache %d, want the 32MiB default", opts.BlockCacheCapacity)
+	}
+	if opts.WriteBuffer != 16*1024*1024 {
+		t.Errorf("write buffer %d, want the 16MiB default", opts.WriteBuffer)
+	}
+}
+
+func TestScaledSaturates(t *testing.T) {
+	if got := scaled(math.MaxInt, 1<<20); got != math.MaxInt {
+		t.Errorf("expected saturation at %d, got %d", math.MaxInt, got)
+	}
+	if got := scaled(64, 1<<20); got != 64*1024*1024 {
+		t.Errorf("expected 64MiB, got %d", got)
+	}
+	if got := scaled(4, 1<<10); got != 4*1024 {
+		t.Errorf("expected 4KiB, got %d", got)
+	}
+}
+
 func TestEmptyRoot(t *testing.T) {
 	testFiler := filer.NewFiler(pb.ServerDiscovery{}, nil, "", "", "", "", "", 255, nil)
 	dir := t.TempDir()

weed/notification/configuration.go

@@ -26,10 +26,10 @@ func LoadConfiguration(config *util.ViperProxy, prefix string) {
 		return
 	}
 
-	validateOneEnabledQueue(config, prefix)
+	validateOneEnabledQueue(config)
 
 	for _, queue := range MessageQueues {
-		if queueEnabled(config, prefix, queue) {
+		if config.GetBool(prefix + queue.GetName() + ".enabled") {
 			if err := queue.Initialize(config, prefix+queue.GetName()+"."); err != nil {
 				glog.Fatalf("Failed to initialize notification for %s: %+v",
 					queue.GetName(), err)
@@ -42,26 +42,10 @@ func LoadConfiguration(config *util.ViperProxy, prefix string) {
 
 }
 
-// queueEnabled reports whether a queue is switched on — by the flat
-// "<prefix><name>.enabled" flag or, for queues accepting named sub-instances
-// ([notification.<queue>.<target>]), by any enabled subconfig.
-func queueEnabled(config util.Configuration, prefix string, queue MessageQueue) bool {
-	key := prefix + queue.GetName() + "."
-	if config.GetBool(key + "enabled") {
-		return true
-	}
-	if q, ok := queue.(interface {
-		HasEnabledSubconfig(util.Configuration, string) bool
-	}); ok {
-		return q.HasEnabledSubconfig(config, key)
-	}
-	return false
-}
-
-func validateOneEnabledQueue(config *util.ViperProxy, prefix string) {
+func validateOneEnabledQueue(config *util.ViperProxy) {
 	enabledQueue := ""
 	for _, queue := range MessageQueues {
-		if queueEnabled(config, prefix, queue) {
+		if config.GetBool(queue.GetName() + ".enabled") {
 			if enabledQueue == "" {
 				enabledQueue = queue.GetName()
 			} else {

weed/notification/webhook/types.go

@@ -3,6 +3,7 @@ package webhook
 import (
 	"fmt"
 	"net/url"
+	"strconv"
 
 	"github.com/seaweedfs/seaweedfs/weed/notification"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -29,6 +30,12 @@ func (e eventType) valid() bool {
 	return notification.ValidEventType(string(e))
 }
 
+var (
+	pubSubHandlerNameTemplate = func(n int) string {
+		return "webhook_handler_" + strconv.Itoa(n)
+	}
+)
+
 type client interface {
 	sendMessage(message *webhookMessage) error
 }

weed/notification/webhook/webhook_queue.go

@@ -4,9 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sort"
-	"strings"
-	"sync"
 	"time"
 
 	"github.com/ThreeDotsLabs/watermill"
@@ -25,145 +22,45 @@ func init() {
 	notification.MessageQueues = append(notification.MessageQueues, &Queue{})
 }
 
-// target is one webhook endpoint with its own filter, http client and
-// concurrency budget. The flat [notification.webhook] section is the unnamed
-// target; each enabled [notification.webhook.<name>] subsection adds another.
-type target struct {
-	name   string
-	config *config
-	filter *filter
-	client client
-	sem    chan struct{}
-	inbox  chan *message.Message
-}
-
-func newTarget(name string, cfg *config) (*target, error) {
-	t := &target{
-		name:   name,
-		config: cfg,
-		filter: newFilter(cfg),
-		sem:    make(chan struct{}, cfg.nWorkers),
-		inbox:  make(chan *message.Message, cfg.bufferSize),
-	}
-	hClient, err := newHTTPClient(cfg)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create webhook http client: %w", err)
-	}
-	t.client = hClient
-	return t, nil
-}
-
-func (t *target) topic() string {
-	if t.name == "" {
-		return pubSubTopicName
-	}
-	return pubSubTopicName + "_" + t.name
-}
-
-func (t *target) handlerName() string {
-	if t.name == "" {
-		return "webhook_handler"
-	}
-	return "webhook_handler_" + t.name
-}
-
 type Queue struct {
-	routers      []*message.Router
+	router       *message.Router
 	queueChannel *gochannel.GoChannel
-	targets      []*target
+	config       *config
+	client       client
+	filter       *filter
 
 	ctx    context.Context
 	cancel context.CancelFunc
+
+	// Semaphore for controlling concurrent webhook requests
+	sem chan struct{}
 }
 
 func (w *Queue) GetName() string {
 	return queueName
 }
 
-// HasEnabledSubconfig reports whether any [notification.webhook.<name>]
-// subsection is enabled, so the queue activates on named targets even when the
-// flat [notification.webhook] section is absent.
-func (w *Queue) HasEnabledSubconfig(configuration util.Configuration, prefix string) bool {
-	return len(namedTargetPrefixes(configuration, prefix)) > 0
-}
-
-// stringMapConfig is implemented by *util.ViperProxy; anything narrower simply
-// gets no named-target discovery.
-type stringMapConfig interface {
-	GetStringMap(key string) map[string]interface{}
-}
-
-func namedTargetPrefixes(configuration util.Configuration, prefix string) []string {
-	mg, ok := configuration.(stringMapConfig)
-	if !ok {
-		return nil
-	}
-	var prefixes []string
-	for k, v := range mg.GetStringMap(strings.TrimSuffix(prefix, ".")) {
-		if _, ok := v.(map[string]interface{}); !ok {
-			continue
-		}
-		p := prefix + k + "."
-		if !configuration.GetBool(p + "enabled") {
-			continue
-		}
-		prefixes = append(prefixes, p)
-	}
-	sort.Strings(prefixes)
-	return prefixes
-}
-
-func buildTargets(configuration util.Configuration, prefix string) ([]*target, error) {
-	var prefixes []string
-	if configuration.GetBool(prefix + "enabled") {
-		prefixes = append(prefixes, prefix)
-	}
-	prefixes = append(prefixes, namedTargetPrefixes(configuration, prefix)...)
-
-	var targets []*target
-	for _, p := range prefixes {
-		cfg := newConfigWithDefaults(configuration, p)
-		if err := cfg.validate(); err != nil {
-			return nil, err
-		}
-		t, err := newTarget(strings.TrimSuffix(strings.TrimPrefix(p, prefix), "."), cfg)
-		if err != nil {
-			return nil, err
-		}
-		targets = append(targets, t)
-	}
-	return targets, nil
-}
-
 func (w *Queue) SendMessage(key string, msg proto.Message) error {
 	eventNotification, ok := msg.(*filer_pb.EventNotification)
 	if !ok {
 		return nil
 	}
 
-	var firstErr error
-	for _, t := range w.targets {
-		if t.filter != nil && !t.filter.shouldPublish(key, eventNotification) {
-			continue
-		}
-		m := newWebhookMessage(key, msg)
-		if m == nil {
-			continue
-		}
-		wMsg, err := m.toWaterMillMessage()
-		if err != nil {
-			return err
-		}
-		wMsg.Metadata.Set("target", t.name)
-		select {
-		case t.inbox <- wMsg:
-		default:
-			if firstErr == nil {
-				firstErr = fmt.Errorf("webhook target %q buffer is full, dropping event for %s", t.name, key)
-			}
-		}
+	if w.filter != nil && !w.filter.shouldPublish(key, eventNotification) {
+		return nil
 	}
-	return firstErr
+
+	m := newWebhookMessage(key, msg)
+	if m == nil {
+		return nil
+	}
+
+	wMsg, err := m.toWaterMillMessage()
+	if err != nil {
+		return err
+	}
+
+	return w.queueChannel.Publish(pubSubTopicName, wMsg)
 }
 
 func (w *webhookMessage) toWaterMillMessage() (*message.Message, error) {
@@ -181,128 +78,105 @@ func (w *webhookMessage) toWaterMillMessage() (*message.Message, error) {
 }
 
 func (w *Queue) Initialize(configuration util.Configuration, prefix string) error {
-	targets, err := buildTargets(configuration, prefix)
-	if err != nil {
+	c := newConfigWithDefaults(configuration, prefix)
+
+	if err := c.validate(); err != nil {
 		return err
 	}
-	if len(targets) == 0 {
-		return fmt.Errorf("webhook endpoint is required")
-	}
 
-	return w.initialize(targets)
+	return w.initialize(c)
 }
 
-func (w *Queue) initialize(targets []*target) error {
+func (w *Queue) initialize(cfg *config) error {
 	w.ctx, w.cancel = context.WithCancel(context.Background())
-	w.targets = targets
+	w.config = cfg
+	w.filter = newFilter(cfg)
 
-	if err := w.setupWatermillQueue(targets); err != nil {
+	// Initialize semaphore for controlling concurrent webhook requests
+	w.sem = make(chan struct{}, cfg.nWorkers)
+
+	hClient, err := newHTTPClient(cfg)
+	if err != nil {
+		return fmt.Errorf("failed to create webhook http client: %w", err)
+	}
+	w.client = hClient
+
+	if err = w.setupWatermillQueue(cfg); err != nil {
 		return fmt.Errorf("failed to setup watermill queue: %w", err)
 	}
-	if err := w.logDeadLetterMessages(); err != nil {
+	if err = w.logDeadLetterMessages(); err != nil {
 		return err
 	}
 
 	return nil
 }
 
-func (w *Queue) setupWatermillQueue(targets []*target) error {
+func (w *Queue) setupWatermillQueue(cfg *config) error {
 	logger := watermill.NewStdLogger(false, false)
-	bufferSize := 0
-	for _, t := range targets {
-		if t.config.bufferSize > bufferSize {
-			bufferSize = t.config.bufferSize
-		}
-	}
 	pubSubConfig := gochannel.Config{
-		OutputChannelBuffer: int64(bufferSize),
+		OutputChannelBuffer: int64(cfg.bufferSize),
 		Persistent:          false,
 	}
 	w.queueChannel = gochannel.NewGoChannel(pubSubConfig, logger)
 
-	var wg sync.WaitGroup
-	for _, t := range targets {
-		// One router per target so each keeps its own retry/backoff settings.
-		// Handlers subscribe to the target's own topic, so a message is
-		// delivered to an endpoint exactly once per matching target.
-		router, err := message.NewRouter(
-			message.RouterConfig{
-				CloseTimeout: 60 * time.Second,
-			},
-			logger,
-		)
-		if err != nil {
-			return fmt.Errorf("failed to create router: %w", err)
-		}
-		w.routers = append(w.routers, router)
+	router, err := message.NewRouter(
+		message.RouterConfig{
+			CloseTimeout: 60 * time.Second,
+		},
+		logger,
+	)
+	if err != nil {
+		return fmt.Errorf("failed to create router: %w", err)
+	}
+	w.router = router
 
-		retryMiddleware := middleware.Retry{
-			MaxRetries:          t.config.maxRetries,
-			InitialInterval:     time.Duration(t.config.backoffSeconds) * time.Second,
-			MaxInterval:         time.Duration(t.config.maxBackoffSeconds) * time.Second,
-			Multiplier:          2.0,
-			RandomizationFactor: 0.3,
-			Logger:              logger,
-		}.Middleware
+	retryMiddleware := middleware.Retry{
+		MaxRetries:          cfg.maxRetries,
+		InitialInterval:     time.Duration(cfg.backoffSeconds) * time.Second,
+		MaxInterval:         time.Duration(cfg.maxBackoffSeconds) * time.Second,
+		Multiplier:          2.0,
+		RandomizationFactor: 0.3,
+		Logger:              logger,
+	}.Middleware
 
-		poisonQueue, err := middleware.PoisonQueue(w.queueChannel, deadLetterTopic)
-		if err != nil {
-			return fmt.Errorf("failed to create poison queue: %w", err)
-		}
-
-		router.AddPlugin(plugin.SignalsHandler)
-		router.AddMiddleware(retryMiddleware, poisonQueue)
-
-		t := t
-		router.AddConsumerHandler(
-			t.handlerName(),
-			t.topic(),
-			w.queueChannel,
-			t.handleWebhook,
-		)
-
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			if err := router.Run(w.ctx); err != nil && !errors.Is(err, context.Canceled) {
-				glog.Errorf("webhook pubsub worker stopped with error: %v", err)
-			}
-			glog.Info("webhook pubsub worker stopped")
-		}()
-
-		go w.forwardTargetMessages(t)
+	poisonQueue, err := middleware.PoisonQueue(w.queueChannel, deadLetterTopic)
+	if err != nil {
+		return fmt.Errorf("failed to create poison queue: %w", err)
 	}
 
-	// the queue context lives until every target's worker has stopped so the
-	// dead letter logger keeps running while any of them is still up
+	router.AddPlugin(plugin.SignalsHandler)
+	router.AddMiddleware(retryMiddleware, poisonQueue)
+
+	// Add a single handler to avoid duplicate message delivery.
+	// With gochannel's default behavior, each handler call creates
+	// a separate subscription, and all subscriptions receive their own copy of each message.
+	// Using a single handler ensures each webhook is sent only once.
+	// Concurrency is controlled via semaphore in handleWebhook based on nWorkers config.
+	router.AddConsumerHandler(
+		"webhook_handler",
+		pubSubTopicName,
+		w.queueChannel,
+		w.handleWebhook,
+	)
+
 	go func() {
-		wg.Wait()
-		w.cancel()
+		// cancels the queue context so the dead letter logger exists in case context not canceled by the shutdown signal already
+		defer w.cancel()
+
+		if err := router.Run(w.ctx); err != nil && !errors.Is(err, context.Canceled) {
+			glog.Errorf("webhook pubsub worker stopped with error: %v", err)
+		}
+
+		glog.Info("webhook pubsub worker stopped")
 	}()
 
 	return nil
 }
 
-// forwardTargetMessages publishes each queued event to the target's topic in
-// order. It owns the potentially blocking Publish call so a saturated target
-// can neither stall the other targets nor the caller of SendMessage.
-func (w *Queue) forwardTargetMessages(t *target) {
-	for {
-		select {
-		case msg := <-t.inbox:
-			if err := w.queueChannel.Publish(t.topic(), msg); err != nil {
-				glog.Errorf("webhook target %q failed to publish message: %v", t.name, err)
-			}
-		case <-w.ctx.Done():
-			return
-		}
-	}
-}
-
-func (t *target) handleWebhook(msg *message.Message) error {
+func (w *Queue) handleWebhook(msg *message.Message) error {
 	// Acquire semaphore slot (blocks if at capacity)
-	t.sem <- struct{}{}
-	defer func() { <-t.sem }()
+	w.sem <- struct{}{}
+	defer func() { <-w.sem }()
 
 	var n filer_pb.EventNotification
 	if err := proto.Unmarshal(msg.Payload, &n); err != nil {
@@ -317,8 +191,8 @@ func (t *target) handleWebhook(msg *message.Message) error {
 		Notification: &n,
 	}
 
-	if err := t.client.sendMessage(webhookMsg); err != nil {
-		glog.Errorf("webhook target %q failed to send message for %s: %v", t.name, webhookMsg.Key, err)
+	if err := w.client.sendMessage(webhookMsg); err != nil {
+		glog.Errorf("failed to send message to webhook %s: %v", webhookMsg.Key, err)
 		return err
 	}
 
@@ -344,12 +218,10 @@ func (w *Queue) logDeadLetterMessages() error {
 					continue
 				}
 				key := "unknown"
-				targetName := ""
 				if msg.Metadata != nil {
 					if keyValue, exists := msg.Metadata["key"]; exists {
 						key = keyValue
 					}
-					targetName = msg.Metadata["target"]
 				}
 				payload := ""
 				if msg.Payload != nil {
@@ -360,7 +232,7 @@ func (w *Queue) logDeadLetterMessages() error {
 						payload = n.String()
 					}
 				}
-				glog.Errorf("received dead letter message: %s, key: %s, target: %s", payload, key, targetName)
+				glog.Errorf("received dead letter message: %s, key: %s", payload, key)
 			case <-w.ctx.Done():
 				return
 			}

weed/notification/webhook/webhook_queue_test.go

@@ -4,7 +4,6 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"strings"
-	"sync/atomic"
 	"testing"
 	"time"
 
@@ -171,29 +170,6 @@ func TestWebhookMessageSerialization(t *testing.T) {
 	}
 }
 
-func newQueueForTest(t *testing.T, cfg *config) *Queue {
-	t.Helper()
-	tgt, err := newTarget("", cfg)
-	if err != nil {
-		t.Fatalf("newTarget() error = %v", err)
-	}
-	q := &Queue{}
-	if err := q.initialize([]*target{tgt}); err != nil {
-		t.Fatalf("initialize() error = %v", err)
-	}
-	return q
-}
-
-func closeQueueForTest(q *Queue) {
-	if q.cancel != nil {
-		q.cancel()
-	}
-	time.Sleep(100 * time.Millisecond)
-	for _, r := range q.routers {
-		r.Close()
-	}
-}
-
 func TestQueueInitialize(t *testing.T) {
 	cfg := &config{
 		endpoint:          "https://example.com/webhook",
@@ -206,19 +182,32 @@ func TestQueueInitialize(t *testing.T) {
 		bufferSize:        100,
 	}
 
-	q := newQueueForTest(t, cfg)
-	defer closeQueueForTest(q)
+	q := &Queue{}
+	err := q.initialize(cfg)
+	if err != nil {
+		t.Errorf("Initialize() error = %v", err)
+	}
 
-	if len(q.routers) != 1 {
+	defer func() {
+		if q.cancel != nil {
+			q.cancel()
+		}
+		time.Sleep(100 * time.Millisecond)
+		if q.router != nil {
+			q.router.Close()
+		}
+	}()
+
+	if q.router == nil {
 		t.Error("Expected router to be initialized")
 	}
 	if q.queueChannel == nil {
 		t.Error("Expected queueChannel to be initialized")
 	}
-	if q.targets[0].client == nil {
+	if q.client == nil {
 		t.Error("Expected client to be initialized")
 	}
-	if q.targets[0].config == nil {
+	if q.config == nil {
 		t.Error("Expected config to be initialized")
 	}
 }
@@ -236,8 +225,21 @@ func TestQueueSendMessage(t *testing.T) {
 		bufferSize:        10,
 	}
 
-	q := newQueueForTest(t, cfg)
-	defer closeQueueForTest(q)
+	q := &Queue{}
+	err := q.initialize(cfg)
+	if err != nil {
+		t.Fatalf("Failed to initialize queue: %v", err)
+	}
+
+	defer func() {
+		if q.cancel != nil {
+			q.cancel()
+		}
+		time.Sleep(100 * time.Millisecond)
+		if q.router != nil {
+			q.router.Close()
+		}
+	}()
 
 	msg := &filer_pb.EventNotification{
 		NewEntry: &filer_pb.Entry{
@@ -245,7 +247,8 @@ func TestQueueSendMessage(t *testing.T) {
 		},
 	}
 
-	if err := q.SendMessage("/test/path", msg); err != nil {
+	err = q.SendMessage("/test/path", msg)
+	if err != nil {
 		t.Errorf("SendMessage() error = %v", err)
 	}
 }
@@ -268,7 +271,7 @@ func TestQueueHandleWebhook(t *testing.T) {
 	}
 
 	client, _ := newHTTPClient(cfg)
-	tgt := &target{
+	q := &Queue{
 		client: client,
 		sem:    make(chan struct{}, cfg.nWorkers),
 	}
@@ -284,7 +287,8 @@ func TestQueueHandleWebhook(t *testing.T) {
 		t.Fatalf("Failed to create watermill message: %v", err)
 	}
 
-	if err := tgt.handleWebhook(wmMsg); err != nil {
+	err = q.handleWebhook(wmMsg)
+	if err != nil {
 		t.Errorf("handleWebhook() error = %v", err)
 	}
 }
@@ -303,8 +307,21 @@ func TestQueueEndToEnd(t *testing.T) {
 		bufferSize:        10,
 	}
 
-	q := newQueueForTest(t, cfg)
-	defer closeQueueForTest(q)
+	q := &Queue{}
+	err := q.initialize(cfg)
+	if err != nil {
+		t.Fatalf("Failed to initialize queue: %v", err)
+	}
+
+	defer func() {
+		if q.cancel != nil {
+			q.cancel()
+		}
+		time.Sleep(100 * time.Millisecond)
+		if q.router != nil {
+			q.router.Close()
+		}
+	}()
 
 	msg := &filer_pb.EventNotification{
 		NewEntry: &filer_pb.Entry{
@@ -312,7 +329,8 @@ func TestQueueEndToEnd(t *testing.T) {
 		},
 	}
 
-	if err := q.SendMessage("/test/path", msg); err != nil {
+	err = q.SendMessage("/test/path", msg)
+	if err != nil {
 		t.Errorf("SendMessage() error = %v", err)
 	}
 }
@@ -329,20 +347,33 @@ func TestQueueRetryMechanism(t *testing.T) {
 		bufferSize:        10,
 	}
 
-	q := newQueueForTest(t, cfg)
-	defer closeQueueForTest(q)
+	q := &Queue{}
+	err := q.initialize(cfg)
+	if err != nil {
+		t.Fatalf("Failed to initialize queue: %v", err)
+	}
+
+	defer func() {
+		if q.cancel != nil {
+			q.cancel()
+		}
+		time.Sleep(100 * time.Millisecond)
+		if q.router != nil {
+			q.router.Close()
+		}
+	}()
 
 	// Verify that the queue is properly configured for retries
-	if q.targets[0].config.maxRetries != 3 {
-		t.Errorf("Expected maxRetries=3, got %d", q.targets[0].config.maxRetries)
+	if q.config.maxRetries != 3 {
+		t.Errorf("Expected maxRetries=3, got %d", q.config.maxRetries)
 	}
 
-	if q.targets[0].config.backoffSeconds != 2 {
-		t.Errorf("Expected backoffSeconds=2, got %d", q.targets[0].config.backoffSeconds)
+	if q.config.backoffSeconds != 2 {
+		t.Errorf("Expected backoffSeconds=2, got %d", q.config.backoffSeconds)
 	}
 
-	if q.targets[0].config.maxBackoffSeconds != 10 {
-		t.Errorf("Expected maxBackoffSeconds=10, got %d", q.targets[0].config.maxBackoffSeconds)
+	if q.config.maxBackoffSeconds != 10 {
+		t.Errorf("Expected maxBackoffSeconds=10, got %d", q.config.maxBackoffSeconds)
 	}
 
 	// Test that we can send a message (retry behavior is handled by Watermill middleware)
@@ -350,7 +381,8 @@ func TestQueueRetryMechanism(t *testing.T) {
 		NewEntry: &filer_pb.Entry{Name: "test.txt"},
 	}
 
-	if err := q.SendMessage("/test/retry", msg); err != nil {
+	err = q.SendMessage("/test/retry", msg)
+	if err != nil {
 		t.Errorf("SendMessage() error = %v", err)
 	}
 }
@@ -387,8 +419,21 @@ func TestQueueNoDuplicateWebhooks(t *testing.T) {
 				bufferSize:        10,
 			}
 
-			q := newQueueForTest(t, cfg)
-			defer closeQueueForTest(q)
+			q := &Queue{}
+			err := q.initialize(cfg)
+			if err != nil {
+				t.Fatalf("Failed to initialize queue: %v", err)
+			}
+
+			defer func() {
+				if q.cancel != nil {
+					q.cancel()
+				}
+				time.Sleep(100 * time.Millisecond)
+				if q.router != nil {
+					q.router.Close()
+				}
+			}()
 
 			// Wait for router and subscriber to be fully ready
 			time.Sleep(200 * time.Millisecond)
@@ -399,7 +444,8 @@ func TestQueueNoDuplicateWebhooks(t *testing.T) {
 				},
 			}
 
-			if err := q.SendMessage("/test/path", msg); err != nil {
+			err = q.SendMessage("/test/path", msg)
+			if err != nil {
 				t.Errorf("SendMessage() error = %v", err)
 			}
 
@@ -561,146 +607,3 @@ func TestQueueSendMessageWithFilter(t *testing.T) {
 		})
 	}
 }
-
-type mapConfig struct {
-	values map[string]interface{}
-}
-
-func (m *mapConfig) GetString(key string) string {
-	v, _ := m.values[key].(string)
-	return v
-}
-
-func (m *mapConfig) GetBool(key string) bool {
-	v, _ := m.values[key].(bool)
-	return v
-}
-
-func (m *mapConfig) GetInt(key string) int {
-	v, _ := m.values[key].(int)
-	return v
-}
-
-func (m *mapConfig) GetStringSlice(key string) []string {
-	v, _ := m.values[key].([]string)
-	return v
-}
-
-func (m *mapConfig) SetDefault(key string, value interface{}) {}
-
-func (m *mapConfig) GetStringMap(key string) map[string]interface{} {
-	v, _ := m.values[key].(map[string]interface{})
-	return v
-}
-
-func TestNamedTargets(t *testing.T) {
-	var hitsA, hitsB atomic.Int32
-	serverA := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		hitsA.Add(1)
-		w.WriteHeader(http.StatusOK)
-	}))
-	defer serverA.Close()
-	serverB := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		hitsB.Add(1)
-		w.WriteHeader(http.StatusOK)
-	}))
-	defer serverB.Close()
-
-	cfg := &mapConfig{values: map[string]interface{}{
-		"notification.webhook": map[string]interface{}{
-			"file_ingress": map[string]interface{}{
-				"enabled":       true,
-				"endpoint":      serverA.URL,
-				"path_prefixes": []string{"/data/"},
-			},
-			"metadata_extractor": map[string]interface{}{
-				"enabled":       true,
-				"endpoint":      serverB.URL,
-				"path_prefixes": []string{"/repository/"},
-			},
-			"disabled_one": map[string]interface{}{
-				"enabled":  false,
-				"endpoint": "https://example.com/never",
-			},
-		},
-		"notification.webhook.file_ingress.enabled":             true,
-		"notification.webhook.file_ingress.endpoint":            serverA.URL,
-		"notification.webhook.file_ingress.path_prefixes":       []string{"/data/"},
-		"notification.webhook.metadata_extractor.enabled":       true,
-		"notification.webhook.metadata_extractor.endpoint":      serverB.URL,
-		"notification.webhook.metadata_extractor.path_prefixes": []string{"/repository/"},
-	}}
-
-	q := &Queue{}
-	if !q.HasEnabledSubconfig(cfg, "notification.webhook.") {
-		t.Fatal("expected named subconfig to be detected as enabled")
-	}
-
-	targets, err := buildTargets(cfg, "notification.webhook.")
-	if err != nil {
-		t.Fatalf("buildTargets() error = %v", err)
-	}
-	if len(targets) != 2 {
-		t.Fatalf("expected 2 targets, got %d", len(targets))
-	}
-	if targets[0].name != "file_ingress" || targets[1].name != "metadata_extractor" {
-		t.Fatalf("unexpected target names: %s, %s", targets[0].name, targets[1].name)
-	}
-
-	if err := q.initialize(targets); err != nil {
-		t.Fatalf("initialize() error = %v", err)
-	}
-	defer closeQueueForTest(q)
-
-	time.Sleep(200 * time.Millisecond)
-
-	msg := &filer_pb.EventNotification{
-		NewEntry: &filer_pb.Entry{Name: "f.txt"},
-	}
-	for _, key := range []string{"/data/f.txt", "/repository/f.txt", "/elsewhere/f.txt"} {
-		if err := q.SendMessage(key, msg); err != nil {
-			t.Errorf("SendMessage(%q) error = %v", key, err)
-		}
-	}
-
-	deadline := time.Now().Add(5 * time.Second)
-	for time.Now().Before(deadline) && (hitsA.Load() == 0 || hitsB.Load() == 0) {
-		time.Sleep(50 * time.Millisecond)
-	}
-	time.Sleep(200 * time.Millisecond)
-
-	if got := hitsA.Load(); got != 1 {
-		t.Errorf("expected file_ingress to receive 1 event, got %d", got)
-	}
-	if got := hitsB.Load(); got != 1 {
-		t.Errorf("expected metadata_extractor to receive 1 event, got %d", got)
-	}
-}
-
-// A disabled flat section keeps no target even when an endpoint is set —
-// the scaffold ships exactly that combination, and a named target must not
-// reactivate it.
-func TestBuildTargetsDisabledFlat(t *testing.T) {
-	cfg := &mapConfig{values: map[string]interface{}{
-		"notification.webhook": map[string]interface{}{
-			"enabled":  false,
-			"endpoint": "https://your-server.com/webhook",
-			"named": map[string]interface{}{
-				"enabled":  true,
-				"endpoint": "https://example.com/hook",
-			},
-		},
-		"notification.webhook.enabled":        false,
-		"notification.webhook.endpoint":       "https://your-server.com/webhook",
-		"notification.webhook.named.enabled":  true,
-		"notification.webhook.named.endpoint": "https://example.com/hook",
-	}}
-
-	targets, err := buildTargets(cfg, "notification.webhook.")
-	if err != nil {
-		t.Fatalf("buildTargets() error = %v", err)
-	}
-	if len(targets) != 1 || targets[0].name != "named" {
-		t.Fatalf("expected only the named target, got %v", targets)
-	}
-}

weed/worker/tasks/iceberg/compact.go

@@ -300,6 +300,14 @@ func (h *Handler) compactDataFiles(
 			glog.Warningf("iceberg compact: failed to merge bin %d (%d files): %v", binIdx, len(bin.Entries), err)
 			goto binDone
 		}
+		// A default-CRS geometry column declares no CRS (geometry_crs.go). A
+		// footer that cannot be rewritten leaves the file as written: valid,
+		// only labelled differently.
+		if fixed, _, crsErr := withDefaultGeometryCRSUnset(mergedData, schema); crsErr != nil {
+			glog.Warningf("iceberg compact: geometry CRS left as written in bin %d: %v", binIdx, crsErr)
+		} else {
+			mergedData = fixed
+		}
 
 		// Write merged file to filer
 		if err := ensureFilerDir(ctx, filerClient, dataDir); err != nil {

weed/worker/tasks/iceberg/geometry_crs.go

@@ -0,0 +1,204 @@
+package iceberg
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"strings"
+
+	"github.com/apache/iceberg-go"
+	"github.com/parquet-go/parquet-go/deprecated"
+	"github.com/parquet-go/parquet-go/encoding/thrift"
+	"github.com/parquet-go/parquet-go/format"
+)
+
+// A GEOMETRY column whose Iceberg type has the default CRS (OGC:CRS84) is
+// written with no CRS in the Parquet logical type, as the input files from
+// most writers declare it. parquet-go spells the default out when it writes
+// such a column, even when the column it read declared none. Both mean the
+// same, but a reader may then see the file column as a different type from
+// the table column (DuckDB casts it and no longer prunes row groups by their
+// statistics). An explicit CRS is left as it is.
+//
+// The footer is changed after the file is written rather than by giving the
+// writer another column type: parquet-go refuses to convert between two
+// GEOMETRY types that both mean the default, and its sorting writer
+// converts. Only the footer's schema elements change; the data, the page
+// indexes and every offset into the file stay byte for byte.
+
+var parquetMagic = []byte("PAR1")
+
+// withDefaultGeometryCRSUnset returns data with the CRS of every GEOMETRY
+// column whose Iceberg type has the default CRS left unset in the footer, and
+// whether anything changed. A file that needs no change comes back as it was;
+// so does one whose footer cannot be read (encrypted, not Parquet), with an
+// error. A changed file is written into data's buffer when it fits, so the
+// caller uses the result in place of data.
+func withDefaultGeometryCRSUnset(data []byte, schema *iceberg.Schema) ([]byte, bool, error) {
+	if schema == nil || !hasDefaultCRSGeometry(schema) {
+		return data, false, nil
+	}
+	n := len(data)
+	if n < 12 || !bytes.Equal(data[n-4:], parquetMagic) || !bytes.Equal(data[:4], parquetMagic) {
+		return data, false, fmt.Errorf("not a plaintext parquet file")
+	}
+	footerLen := int(binary.LittleEndian.Uint32(data[n-8 : n-4]))
+	footerStart := n - 8 - footerLen
+	if footerLen <= 0 || footerStart < 4 {
+		return data, false, fmt.Errorf("footer length %d out of range", footerLen)
+	}
+	var protocol thrift.CompactProtocol
+	var md format.FileMetaData
+	if err := thrift.Unmarshal(&protocol, data[footerStart:n-8], &md); err != nil {
+		return data, false, fmt.Errorf("decode footer: %w", err)
+	}
+	changed := false
+	for i, path := range schemaElementPaths(md.Schema) {
+		el := &md.Schema[i]
+		g, ok := el.LogicalType.Value.(*format.GeometryType)
+		if !ok || g.CRS != iceberg.DefaultGeoCRS {
+			continue
+		}
+		var field iceberg.NestedField
+		if el.FieldID > 0 {
+			field, ok = schema.FindFieldByID(int(el.FieldID))
+		} else {
+			field, ok = schema.FindFieldByName(strings.Join(path, "."))
+		}
+		if !ok {
+			continue
+		}
+		if gt, isGeo := field.Type.(iceberg.GeometryType); !isGeo || gt.CRS() != iceberg.DefaultGeoCRS {
+			continue
+		}
+		el.LogicalType = format.LogicalType{Value: &format.GeometryType{}}
+		changed = true
+	}
+	if !changed {
+		return data, false, nil
+	}
+	footer, err := thrift.Marshal(&protocol, &md)
+	if err != nil {
+		return data, false, fmt.Errorf("encode footer: %w", err)
+	}
+	// The new footer is written over the old one in data's own buffer when it
+	// fits, as it does when a CRS is removed: the file is as large as a
+	// compaction target, and copying it would double the memory it takes.
+	out := data[:footerStart]
+	if cap(data) < footerStart+len(footer)+8 {
+		out = make([]byte, footerStart, footerStart+len(footer)+8)
+		copy(out, data[:footerStart])
+	}
+	out = append(out, footer...)
+	out = binary.LittleEndian.AppendUint32(out, uint32(len(footer)))
+	out = append(out, parquetMagic...)
+	return out, true, nil
+}
+
+func hasDefaultCRSGeometry(schema *iceberg.Schema) bool {
+	for _, f := range schema.Fields() {
+		if typeHasDefaultCRSGeometry(f.Type) {
+			return true
+		}
+	}
+	return false
+}
+
+// typeHasDefaultCRSGeometry reports whether the type is, or nests, a geometry
+// with the default CRS.
+func typeHasDefaultCRSGeometry(t iceberg.Type) bool {
+	switch v := t.(type) {
+	case iceberg.GeometryType:
+		return v.CRS() == iceberg.DefaultGeoCRS
+	case *iceberg.StructType:
+		for _, f := range v.FieldList {
+			if typeHasDefaultCRSGeometry(f.Type) {
+				return true
+			}
+		}
+	case *iceberg.ListType:
+		return typeHasDefaultCRSGeometry(v.Element)
+	case *iceberg.MapType:
+		return typeHasDefaultCRSGeometry(v.KeyType) || typeHasDefaultCRSGeometry(v.ValueType)
+	}
+	return false
+}
+
+// schemaElementPaths is each flattened schema element's Iceberg field path
+// from the root (the root's is empty): the footer lists the tree depth first,
+// every group followed by its num_children children. The repeated group of a
+// LIST or MAP is not an Iceberg field, so it takes its parent's path; under it
+// a list's element is named "element" and a map's two children "key" and
+// "value", as Iceberg names them, whatever the writer called them.
+func schemaElementPaths(elements []format.SchemaElement) [][]string {
+	paths := make([][]string, len(elements))
+	var walk func(i int, path []string) int
+	// children walks the n children of a group whose own path is path,
+	// naming the k-th by name(k) when name is set.
+	children := func(next, n int, path []string, name func(k int, el *format.SchemaElement) string) int {
+		for c := 0; c < n && next < len(elements); c++ {
+			child := path
+			if name != nil {
+				child = append(append([]string{}, path...), name(c, &elements[next]))
+			} else {
+				child = append(append([]string{}, path...), elements[next].Name)
+			}
+			next = walk(next, child)
+		}
+		return next
+	}
+	walk = func(i int, path []string) int {
+		if i >= len(elements) {
+			return i
+		}
+		paths[i] = path
+		el := &elements[i]
+		n := int(el.NumChildren.V)
+		if n != 1 || i+1 >= len(elements) {
+			return children(i+1, n, path, nil)
+		}
+		repeated := &elements[i+1]
+		if rt, ok := repeated.RepetitionType.Get(); !ok || rt != format.Repeated {
+			return children(i+1, n, path, nil)
+		}
+		switch {
+		case isListAnnotated(el):
+			// A repeated group of one field wraps the element, unless its
+			// name marks the legacy form where the group is the element.
+			if rn := int(repeated.NumChildren.V); rn == 1 && repeated.Name != "array" && repeated.Name != el.Name+"_tuple" {
+				paths[i+1] = path
+				return children(i+2, 1, path, func(int, *format.SchemaElement) string { return "element" })
+			}
+			return children(i+1, 1, path, func(int, *format.SchemaElement) string { return "element" })
+		case isMapAnnotated(el):
+			if int(repeated.NumChildren.V) == 2 {
+				paths[i+1] = path
+				return children(i+2, 2, path, func(k int, _ *format.SchemaElement) string {
+					if k == 0 {
+						return "key"
+					}
+					return "value"
+				})
+			}
+		}
+		return children(i+1, n, path, nil)
+	}
+	walk(0, nil)
+	return paths
+}
+
+func isListAnnotated(el *format.SchemaElement) bool {
+	if _, ok := el.LogicalType.Value.(*format.ListType); ok {
+		return true
+	}
+	ct, ok := el.ConvertedType.Get()
+	return ok && ct == deprecated.List
+}
+
+func isMapAnnotated(el *format.SchemaElement) bool {
+	if _, ok := el.LogicalType.Value.(*format.MapType); ok {
+		return true
+	}
+	ct, ok := el.ConvertedType.Get()
+	return ok && (ct == deprecated.Map || ct == deprecated.MapKeyValue)
+}

weed/worker/tasks/iceberg/geometry_crs_test.go

@@ -0,0 +1,251 @@
+package iceberg
+
+import (
+	"bytes"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/table"
+	"github.com/parquet-go/parquet-go"
+	"github.com/parquet-go/parquet-go/compress/zstd"
+	"github.com/parquet-go/parquet-go/format"
+)
+
+// unsetDefaultCRS makes an input file declare the default CRS as no CRS at
+// all, as DuckDB and other writers do; parquet.Geometry("") spells it out.
+func unsetDefaultCRS(t *testing.T, data []byte) []byte {
+	t.Helper()
+	out, changed, err := withDefaultGeometryCRSUnset(data, geoTestSchema(iceberg.GeometryType{}))
+	if err != nil || !changed {
+		t.Fatalf("unset the input's default CRS: changed %v, %v", changed, err)
+	}
+	return out
+}
+
+func fiveGeoFiles() []geoFile {
+	var files []geoFile
+	for f := 0; f < 5; f++ {
+		var rows []geoRow
+		for i := 0; i < 40; i++ {
+			id := int64(f*40 + i)
+			rows = append(rows, geoRow{ID: id, Geom: wkbPoint(float64(id%20)-10, float64(id/20)-5)})
+		}
+		files = append(files, geoFile{Name: fmt.Sprintf("g%d.parquet", f), Rows: rows})
+	}
+	return files
+}
+
+// The Parquet CRS of a default-CRS geometry column survives compaction as the
+// inputs declared it: none. The control shows what parquet-go does with the
+// same schema on its own: it spells the default out.
+func TestCompactWritesDefaultCRSGeometryWithoutCRS(t *testing.T) {
+	control := unsetDefaultCRS(t, writeGeoParquet(t, []geoRow{{1, wkbPoint(1, 2)}}, ""))
+	if crs, ok := footerGeometryCRS(t, control, "geom"); !ok || crs != "" {
+		t.Fatalf("setup: an input file declares CRS %q (geometry %v), want none", crs, ok)
+	}
+	f, err := parquet.OpenFile(bytes.NewReader(control), int64(len(control)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var plain bytes.Buffer
+	w := parquet.NewWriter(&plain, f.Schema())
+	if _, err := w.WriteRows([]parquet.Row{{parquet.ByteArrayValue(wkbPoint(1, 2)).Level(0, 1, 0), parquet.Int64Value(1).Level(0, 0, 1)}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := w.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if crs, _ := footerGeometryCRS(t, plain.Bytes(), "geom"); crs != iceberg.DefaultGeoCRS {
+		t.Fatalf("control: parquet-go copied the schema with CRS %q; the fix is for %q", crs, iceberg.DefaultGeoCRS)
+	}
+
+	for _, strategy := range []string{"binpack", "sort"} {
+		t.Run(strategy, func(t *testing.T) {
+			fs, client := startFakeFiler(t)
+			setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "geo"}
+			opt := geoTableOptions{fixup: unsetDefaultCRS}
+			config := Config{}
+			if strategy == "sort" {
+				so, err := table.NewSortOrder(1, []table.SortField{{SourceIDs: []int{geoTestIDField}, Transform: iceberg.IdentityTransform{}, Direction: table.SortASC, NullOrder: table.NullsFirst}})
+				if err != nil {
+					t.Fatal(err)
+				}
+				opt.sortOrder = &so
+				config.RewriteStrategy, config.SortBufferRows = "sort", minSortBufferRows
+			}
+			files := fiveGeoFiles()
+			files[2].ParquetCRSSet = true // an input that spells the default out
+			populateGeoTable(t, fs, setup, files, opt)
+			if result := compactGeo(t, client, setup, config); !strings.Contains(result, "compacted 5 files") {
+				t.Fatalf("result %q", result)
+			}
+			for name, data := range compactedFiles(t, fs, setup) {
+				if crs, ok := footerGeometryCRS(t, data, "geom"); !ok || crs != "" {
+					t.Errorf("%s: geometry column declares CRS %q (geometry %v), want none", name, crs, ok)
+				}
+				got, err := parquet.Read[struct {
+					ID int64 `parquet:"id"`
+				}](bytes.NewReader(data), int64(len(data)))
+				if err != nil || len(got) != 200 {
+					t.Errorf("%s: the rewritten file reads %d rows, %v", name, len(got), err)
+				}
+			}
+		})
+	}
+}
+
+// An explicit CRS is not the default: it is written as the input declared it.
+func TestCompactKeepsExplicitGeometryCRS(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "geo"}
+	g, err := iceberg.GeometryTypeOf("EPSG:2913")
+	if err != nil {
+		t.Fatal(err)
+	}
+	populateGeoTable(t, fs, setup, fiveGeoFiles(), geoTableOptions{icebergType: g, parquetCRS: "EPSG:2913"})
+	compactGeo(t, client, setup, Config{})
+	for name, data := range compactedFiles(t, fs, setup) {
+		if crs, ok := footerGeometryCRS(t, data, "geom"); !ok || crs != "EPSG:2913" {
+			t.Errorf("%s: geometry CRS %q (geometry %v), want EPSG:2913", name, crs, ok)
+		}
+	}
+}
+
+// A default-CRS geometry nested in a struct is written without a CRS too.
+func TestDefaultCRSUnsetForNestedGeometry(t *testing.T) {
+	schema := parquet.NewSchema("table", parquet.Group{
+		"id":  parquet.FieldID(parquet.Compressed(parquet.Int(64), &zstd.Codec{}), 1),
+		"loc": parquet.FieldID(parquet.Group{"geom": parquet.FieldID(parquet.Optional(parquet.Geometry("")), 3)}, 2),
+	})
+	var buf bytes.Buffer
+	w := parquet.NewGenericWriter[any](&buf, schema)
+	cols := map[string]int{}
+	for i, p := range schema.Columns() {
+		cols[strings.Join(p, ".")] = i
+	}
+	row := make(parquet.Row, 2)
+	row[cols["id"]] = parquet.Int64Value(1).Level(0, 0, cols["id"])
+	row[cols["loc.geom"]] = parquet.ByteArrayValue(wkbPoint(1, 2)).Level(0, 1, cols["loc.geom"])
+	if _, err := w.WriteRows([]parquet.Row{row}); err != nil {
+		t.Fatal(err)
+	}
+	if err := w.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if crs, ok := footerGeometryCRS(t, buf.Bytes(), "geom"); !ok || crs != iceberg.DefaultGeoCRS {
+		t.Fatalf("setup: the nested column declares %q (geometry %v); the control is the spelled-out default", crs, ok)
+	}
+	ice := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Type: iceberg.PrimitiveTypes.Int64, Name: "id", Required: true},
+		iceberg.NestedField{ID: 2, Name: "loc", Type: &iceberg.StructType{FieldList: []iceberg.NestedField{{ID: 3, Name: "geom", Type: iceberg.GeometryType{}}}}})
+	out, changed, err := withDefaultGeometryCRSUnset(buf.Bytes(), ice)
+	if err != nil || !changed {
+		t.Fatalf("changed %v, %v", changed, err)
+	}
+	if crs, ok := footerGeometryCRS(t, out, "geom"); !ok || crs != "" {
+		t.Errorf("the nested geometry declares CRS %q (geometry %v), want none", crs, ok)
+	}
+	got, err := parquet.Read[struct {
+		ID int64 `parquet:"id"`
+	}](bytes.NewReader(out), int64(len(out)))
+	if err != nil || len(got) != 1 || got[0].ID != 1 {
+		t.Errorf("the rewritten file reads %v, %v", got, err)
+	}
+}
+
+// footerGeometryCRSs is the CRS of every GEOMETRY element in the footer, by
+// schema element index and name.
+func footerGeometryCRSs(t *testing.T, data []byte) map[string]string {
+	t.Helper()
+	f, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data)))
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	out := map[string]string{}
+	for i, el := range f.Metadata().Schema {
+		if g, ok := el.LogicalType.Value.(*format.GeometryType); ok {
+			out[fmt.Sprintf("%d:%s", i, el.Name)] = g.CRS
+		}
+	}
+	return out
+}
+
+// A file without field ids is matched to the table by name: through the
+// LIST and MAP wrapper groups too, which Iceberg's field names leave out.
+func TestDefaultCRSUnsetWithoutFieldIDs(t *testing.T) {
+	geom := func() parquet.Node { return parquet.Optional(parquet.Geometry("")) }
+	element := func(id int, typ iceberg.Type) *iceberg.ListType {
+		return &iceberg.ListType{ElementID: id, Element: typ, ElementRequired: false}
+	}
+	cases := []struct {
+		name    string
+		parquet parquet.Group
+		iceberg []iceberg.NestedField
+	}{
+		{"top level", parquet.Group{"id": parquet.Int(64), "geom": geom()},
+			[]iceberg.NestedField{{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}, {ID: 2, Name: "geom", Type: iceberg.GeometryType{}}}},
+		{"list", parquet.Group{"shapes": parquet.Optional(parquet.List(geom()))},
+			[]iceberg.NestedField{{ID: 1, Name: "shapes", Type: element(2, iceberg.GeometryType{})}}},
+		{"list of structs", parquet.Group{"items": parquet.Optional(parquet.List(parquet.Group{"geom": geom()}))},
+			[]iceberg.NestedField{{ID: 1, Name: "items", Type: element(2, &iceberg.StructType{FieldList: []iceberg.NestedField{{ID: 3, Name: "geom", Type: iceberg.GeometryType{}}}})}}},
+		{"map", parquet.Group{"attrs": parquet.Optional(parquet.Map(parquet.String(), geom()))},
+			[]iceberg.NestedField{{ID: 1, Name: "attrs", Type: &iceberg.MapType{KeyID: 2, KeyType: iceberg.PrimitiveTypes.String, ValueID: 3, ValueType: iceberg.GeometryType{}}}}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			var buf bytes.Buffer
+			w := parquet.NewGenericWriter[any](&buf, parquet.NewSchema("table", tc.parquet))
+			if err := w.Close(); err != nil {
+				t.Fatal(err)
+			}
+			before := footerGeometryCRSs(t, buf.Bytes())
+			if len(before) != 1 {
+				t.Fatalf("setup: geometry columns %v, want one", before)
+			}
+			for path, crs := range before {
+				if crs != iceberg.DefaultGeoCRS {
+					t.Fatalf("setup: %s declares %q; the control is the spelled-out default", path, crs)
+				}
+			}
+			out, changed, err := withDefaultGeometryCRSUnset(buf.Bytes(), iceberg.NewSchema(0, tc.iceberg...))
+			if err != nil || !changed {
+				t.Fatalf("changed %v, %v", changed, err)
+			}
+			for path, crs := range footerGeometryCRSs(t, out) {
+				if crs != "" {
+					t.Errorf("%s declares CRS %q, want none", path, crs)
+				}
+			}
+		})
+	}
+}
+
+// The footer is replaced inside the file's own buffer: a compacted file can be
+// as large as the target file size, and a copy of it would double the memory
+// the rewrite needs.
+func TestDefaultCRSUnsetReusesBuffer(t *testing.T) {
+	data := writeGeoParquet(t, []geoRow{{1, wkbPoint(1, 2)}, {2, wkbPoint(3, 4)}}, "")
+	if crs, ok := footerGeometryCRS(t, data, "geom"); !ok || crs != iceberg.DefaultGeoCRS {
+		t.Fatalf("setup: CRS %q (geometry %v); the control is the spelled-out default", crs, ok)
+	}
+	first := &data[0]
+	out, changed, err := withDefaultGeometryCRSUnset(data, geoTestSchema(iceberg.GeometryType{}))
+	if err != nil || !changed {
+		t.Fatalf("changed %v, %v", changed, err)
+	}
+	if &out[0] != first {
+		t.Errorf("the rewrite copied the file into a new buffer")
+	}
+	if crs, ok := footerGeometryCRS(t, out, "geom"); !ok || crs != "" {
+		t.Errorf("geometry declares CRS %q (geometry %v), want none", crs, ok)
+	}
+	got, err := parquet.Read[struct {
+		ID   int64  `parquet:"id"`
+		Geom []byte `parquet:"geom,optional"`
+	}](bytes.NewReader(out), int64(len(out)))
+	if err != nil || len(got) != 2 || got[1].ID != 2 || !bytes.Equal(got[1].Geom, wkbPoint(3, 4)) {
+		t.Errorf("the rewritten file reads %v, %v", got, err)
+	}
+}

weed/worker/tasks/iceberg/geometry_fixture_test.go

@@ -0,0 +1,360 @@
+package iceberg
+
+import (
+	"bytes"
+	"context"
+	"encoding/binary"
+	"encoding/json"
+	"fmt"
+	"math"
+	"path"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/table"
+	"github.com/parquet-go/parquet-go"
+	"github.com/parquet-go/parquet-go/compress/zstd"
+	"github.com/parquet-go/parquet-go/format"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
+)
+
+// Fixtures for v3 tables with a geometry column: Parquet files with field
+// ids, ZSTD and a native GEOMETRY column, and manifests that carry the v3
+// geometry bounds of each file, as DuckDB writes them.
+
+const (
+	geoTestIDField   = 1
+	geoTestGeomField = 2
+)
+
+func wkbPoint(x, y float64) []byte {
+	b := make([]byte, 21)
+	b[0] = 1 // little endian
+	binary.LittleEndian.PutUint32(b[1:], 1)
+	binary.LittleEndian.PutUint64(b[5:], math.Float64bits(x))
+	binary.LittleEndian.PutUint64(b[13:], math.Float64bits(y))
+	return b
+}
+
+// wkbBox is the polygon of one closed ring around the box.
+func wkbBox(xmin, ymin, xmax, ymax float64) []byte {
+	ring := [][2]float64{{xmin, ymin}, {xmax, ymin}, {xmax, ymax}, {xmin, ymax}, {xmin, ymin}}
+	b := make([]byte, 0, 13+16*len(ring))
+	b = append(b, 1)
+	b = binary.LittleEndian.AppendUint32(b, 3)
+	b = binary.LittleEndian.AppendUint32(b, 1)
+	b = binary.LittleEndian.AppendUint32(b, uint32(len(ring)))
+	for _, p := range ring {
+		b = binary.LittleEndian.AppendUint64(b, math.Float64bits(p[0]))
+		b = binary.LittleEndian.AppendUint64(b, math.Float64bits(p[1]))
+	}
+	return b
+}
+
+func geoLE64(v int64) []byte {
+	b := make([]byte, 8)
+	binary.LittleEndian.PutUint64(b, uint64(v))
+	return b
+}
+
+func xyBound(x, y float64) []byte {
+	b := make([]byte, 16)
+	binary.LittleEndian.PutUint64(b, math.Float64bits(x))
+	binary.LittleEndian.PutUint64(b[8:], math.Float64bits(y))
+	return b
+}
+
+func decodeXYBound(t *testing.T, b []byte) (float64, float64) {
+	t.Helper()
+	if len(b) < 16 {
+		t.Fatalf("geometry bound of %d bytes, want 16", len(b))
+	}
+	return math.Float64frombits(binary.LittleEndian.Uint64(b)), math.Float64frombits(binary.LittleEndian.Uint64(b[8:]))
+}
+
+type geoRow struct {
+	ID   int64
+	Geom []byte // nil: a null geometry
+}
+
+type geoFile struct {
+	Name          string
+	Rows          []geoRow
+	ParquetCRSSet bool // declare the default CRS in the Parquet logical type
+}
+
+type geoTableOptions struct {
+	icebergType iceberg.Type     // the geometry column's Iceberg type (default: geometry with the default CRS)
+	parquetCRS  string           // the CRS of the Parquet GEOMETRY type
+	sortOrder   *table.SortOrder // nil: unsorted
+	properties  iceberg.Properties
+	// fixup, when set, rewrites each input file after it is written
+	fixup func(t *testing.T, data []byte) []byte
+}
+
+func geoTestSchema(geomType iceberg.Type) *iceberg.Schema {
+	return iceberg.NewSchema(0,
+		iceberg.NestedField{ID: geoTestIDField, Type: iceberg.PrimitiveTypes.Int64, Name: "id", Required: true},
+		iceberg.NestedField{ID: geoTestGeomField, Type: geomType, Name: "geom", Required: false},
+	)
+}
+
+func writeGeoParquet(t *testing.T, rows []geoRow, crs string) []byte {
+	t.Helper()
+	schema := parquet.NewSchema("table", parquet.Group{
+		"id":   parquet.FieldID(parquet.Compressed(parquet.Int(64), &zstd.Codec{}), geoTestIDField),
+		"geom": parquet.FieldID(parquet.Compressed(parquet.Optional(parquet.Geometry(crs)), &zstd.Codec{}), geoTestGeomField),
+	})
+	geomCol, idCol := -1, -1
+	for i, p := range schema.Columns() {
+		switch strings.Join(p, ".") {
+		case "geom":
+			geomCol = i
+		case "id":
+			idCol = i
+		}
+	}
+	var buf bytes.Buffer
+	w := parquet.NewGenericWriter[any](&buf, schema)
+	for _, r := range rows {
+		row := make(parquet.Row, 2)
+		if r.Geom == nil {
+			row[geomCol] = parquet.NullValue().Level(0, 0, geomCol)
+		} else {
+			row[geomCol] = parquet.ByteArrayValue(r.Geom).Level(0, 1, geomCol)
+		}
+		row[idCol] = parquet.Int64Value(r.ID).Level(0, 0, idCol)
+		if _, err := w.WriteRows([]parquet.Row{row}); err != nil {
+			t.Fatalf("write row: %v", err)
+		}
+	}
+	if err := w.Close(); err != nil {
+		t.Fatalf("close writer: %v", err)
+	}
+	return buf.Bytes()
+}
+
+// footerGeoBox is the union of the file's row-group geometry boxes.
+func footerGeoBox(t *testing.T, data []byte) (xmin, ymin, xmax, ymax float64, ok bool) {
+	t.Helper()
+	f, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data)))
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	xmin, ymin, xmax, ymax = math.Inf(1), math.Inf(1), math.Inf(-1), math.Inf(-1)
+	for _, rg := range f.Metadata().RowGroups {
+		for _, c := range rg.Columns {
+			if strings.Join(c.MetaData.PathInSchema, ".") != "geom" || len(c.MetaData.GeospatialStatistics.GeoSpatialTypes) == 0 {
+				continue
+			}
+			bb := c.MetaData.GeospatialStatistics.BBox
+			xmin, ymin = math.Min(xmin, bb.XMin), math.Min(ymin, bb.YMin)
+			xmax, ymax = math.Max(xmax, bb.XMax), math.Max(ymax, bb.YMax)
+			ok = true
+		}
+	}
+	return
+}
+
+// footerGeometryCRS is the CRS the file's GEOMETRY logical type declares, and
+// whether the column is GEOMETRY at all.
+func footerGeometryCRS(t *testing.T, data []byte, column string) (string, bool) {
+	t.Helper()
+	f, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data)))
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	for _, el := range f.Metadata().Schema {
+		if el.Name != column {
+			continue
+		}
+		if g, ok := el.LogicalType.Value.(*format.GeometryType); ok {
+			return g.CRS, true
+		}
+		return "", false
+	}
+	t.Fatalf("no column %s in the footer", column)
+	return "", false
+}
+
+// populateGeoTable writes a v3 geometry table with one append snapshot and
+// one manifest per file.
+func populateGeoTable(t *testing.T, fs *fakeFilerServer, setup tableSetup, files []geoFile, opt geoTableOptions) table.Metadata {
+	t.Helper()
+	geomType := opt.icebergType
+	if geomType == nil {
+		geomType = iceberg.GeometryType{}
+	}
+	schema := geoTestSchema(geomType)
+	spec := *iceberg.UnpartitionedSpec
+	location := "s3://" + setup.BucketName + "/" + setup.dataPath()
+	props := iceberg.Properties{"format-version": "3"}
+	for k, v := range opt.properties {
+		props[k] = v
+	}
+	sortOrder := table.UnsortedSortOrder
+	if opt.sortOrder != nil {
+		sortOrder = *opt.sortOrder
+	}
+	meta, err := table.NewMetadata(schema, &spec, sortOrder, location, props)
+	if err != nil {
+		t.Fatalf("create metadata: %v", err)
+	}
+	bucketPath := path.Join(s3tables.TablesPath, setup.BucketName)
+	tableFilerPath := path.Join(bucketPath, setup.dataPath())
+	metaDir := path.Join(tableFilerPath, "metadata")
+	dataDir := path.Join(tableFilerPath, "data")
+	putFile := func(dir, name string, data []byte) {
+		fs.putEntry(dir, name, &filer_pb.Entry{Name: name, Content: data,
+			Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(len(data))}})
+	}
+	snapID := int64(1)
+	var manifests []iceberg.ManifestFile
+	for i, f := range files {
+		crs := opt.parquetCRS
+		if f.ParquetCRSSet && crs == "" {
+			crs = iceberg.DefaultGeoCRS
+		}
+		data := writeGeoParquet(t, f.Rows, crs)
+		if opt.fixup != nil && !f.ParquetCRSSet {
+			data = opt.fixup(t, data)
+		}
+		putFile(dataDir, f.Name, data)
+		dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, setup.fileRef("data", f.Name), iceberg.ParquetFile, map[int]any{}, nil, nil, int64(len(f.Rows)), int64(len(data)))
+		if err != nil {
+			t.Fatalf("data file %s: %v", f.Name, err)
+		}
+		lo, hi := int64(math.MaxInt64), int64(math.MinInt64)
+		for _, r := range f.Rows {
+			lo, hi = min(lo, r.ID), max(hi, r.ID)
+		}
+		lower := map[int][]byte{geoTestIDField: geoLE64(lo)}
+		upper := map[int][]byte{geoTestIDField: geoLE64(hi)}
+		if xmin, ymin, xmax, ymax, ok := footerGeoBox(t, data); ok {
+			lower[geoTestGeomField], upper[geoTestGeomField] = xyBound(xmin, ymin), xyBound(xmax, ymax)
+		}
+		dfb.LowerBoundValues(lower)
+		dfb.UpperBoundValues(upper)
+		entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build())
+		name := fmt.Sprintf("data-manifest-%d.avro", i)
+		var buf bytes.Buffer
+		mf, err := iceberg.WriteManifest(setup.fileRef("metadata", name), &buf, 3, spec, schema, snapID, []iceberg.ManifestEntry{entry})
+		if err != nil {
+			t.Fatalf("write manifest: %v", err)
+		}
+		putFile(metaDir, name, buf.Bytes())
+		manifests = append(manifests, mf)
+	}
+	var mlBuf bytes.Buffer
+	mlw, err := iceberg.NewManifestListWriterV3(&mlBuf, snapID, 1, 0, nil)
+	if err != nil {
+		t.Fatalf("manifest list writer: %v", err)
+	}
+	if err := mlw.AddManifests(manifests); err != nil {
+		t.Fatalf("add manifests: %v", err)
+	}
+	if err := mlw.Close(); err != nil {
+		t.Fatalf("close manifest list: %v", err)
+	}
+	putFile(metaDir, "snap-1.avro", mlBuf.Bytes())
+	firstRowID := int64(0)
+	addedRows := *mlw.NextRowID()
+	snap := table.Snapshot{SnapshotID: snapID, SequenceNumber: 1, TimestampMs: time.Now().UnixMilli(),
+		ManifestList: setup.fileRef("metadata", "snap-1.avro"),
+		Summary:      &table.Summary{Operation: table.OpAppend, Properties: map[string]string{}},
+		FirstRowID:   &firstRowID, AddedRows: &addedRows}
+	builder, err := table.MetadataBuilderFromBase(meta, location)
+	if err != nil {
+		t.Fatalf("metadata builder: %v", err)
+	}
+	if err := builder.AddSnapshot(&snap); err != nil {
+		t.Fatalf("add snapshot: %v", err)
+	}
+	if err := builder.SetSnapshotRef(table.MainBranch, snapID, table.BranchRef); err != nil {
+		t.Fatalf("set ref: %v", err)
+	}
+	if meta, err = builder.Build(); err != nil {
+		t.Fatalf("build metadata: %v", err)
+	}
+	full, _ := json.Marshal(meta)
+	xattr, _ := json.Marshal(map[string]interface{}{
+		"metadataVersion":  1,
+		"metadataLocation": setup.fileRef("metadata", "v1.metadata.json"),
+		"metadata":         map[string]interface{}{"fullMetadata": json.RawMessage(full)},
+	})
+	fs.putEntry(s3tables.TablesPath, setup.BucketName, &filer_pb.Entry{
+		Name: setup.BucketName, IsDirectory: true, Extended: map[string][]byte{s3tables.ExtendedKeyTableBucket: []byte("true")},
+	})
+	fs.putEntry(bucketPath, setup.Namespace, &filer_pb.Entry{Name: setup.Namespace, IsDirectory: true})
+	fs.putEntry(path.Join(bucketPath, setup.Namespace), setup.TableName, &filer_pb.Entry{
+		Name: setup.TableName, IsDirectory: true, Extended: map[string][]byte{s3tables.ExtendedKeyMetadata: xattr},
+	})
+	return meta
+}
+
+func compactGeo(t *testing.T, client filer_pb.SeaweedFilerClient, setup tableSetup, config Config) string {
+	t.Helper()
+	if config.TargetFileSizeBytes == 0 {
+		config.TargetFileSizeBytes = 256 * 1024 * 1024
+	}
+	if config.MinInputFiles == 0 {
+		config.MinInputFiles = 2
+	}
+	if config.MaxCommitRetries == 0 {
+		config.MaxCommitRetries = 3
+	}
+	config.ApplyDeletes = true
+	result, _, err := NewHandler(nil).compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil)
+	if err != nil {
+		t.Fatalf("compactDataFiles: %v", err)
+	}
+	return result
+}
+
+// compactedFiles is every compact-*.parquet in the table's data directory.
+func compactedFiles(t *testing.T, fs *fakeFilerServer, setup tableSetup) map[string][]byte {
+	t.Helper()
+	out := map[string][]byte{}
+	for _, e := range fs.listDir(path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data")) {
+		if strings.HasPrefix(e.Name, "compact-") {
+			out[e.Name] = e.Content
+		}
+	}
+	if len(out) == 0 {
+		t.Fatal("no compacted file")
+	}
+	return out
+}
+
+// currentDataEntries is every entry of the current snapshot's data manifests.
+func currentDataEntries(t *testing.T, client filer_pb.SeaweedFilerClient, setup tableSetup) []iceberg.ManifestEntry {
+	t.Helper()
+	ctx := context.Background()
+	state, err := loadCurrentMetadata(ctx, client, setup.BucketName, setup.tablePath())
+	if err != nil {
+		t.Fatalf("loadCurrentMetadata: %v", err)
+	}
+	manifests, err := loadCurrentManifests(ctx, client, setup.BucketName, state.DataPath, state.Metadata)
+	if err != nil {
+		t.Fatalf("loadCurrentManifests: %v", err)
+	}
+	var out []iceberg.ManifestEntry
+	for _, mf := range manifests {
+		if mf.ManifestContent() != iceberg.ManifestContentData {
+			continue
+		}
+		data, err := loadFileByIcebergPath(ctx, client, setup.BucketName, state.DataPath, mf.FilePath())
+		if err != nil {
+			t.Fatalf("load manifest: %v", err)
+		}
+		entries, err := iceberg.ReadManifest(mf, bytes.NewReader(data), true)
+		if err != nil {
+			t.Fatalf("read manifest: %v", err)
+		}
+		out = append(out, entries...)
+	}
+	return out
+}

weed/worker/tasks/iceberg/row_lineage.go

@@ -10,6 +10,8 @@ import (
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/table"
 	"github.com/parquet-go/parquet-go"
+	"github.com/parquet-go/parquet-go/compress"
+	"github.com/parquet-go/parquet-go/compress/zstd"
 )
 
 // Row lineage (Iceberg v3, spec "Row Lineage" and "Snapshot Row IDs").
@@ -170,9 +172,16 @@ func newLineageLayout(reference *parquet.Schema) (*lineageLayout, error) {
 		offsets = append(offsets, leaves)
 		leaves += leafCount(f)
 	}
+	// The lineage columns are compressed like the reference's data columns
+	// (not like a lineage column it already has): a leaf that declares no
+	// codec is written uncompressed, 16 bytes a row.
+	codec := lineageCodec(base)
+	lineageLeaf := func(id int) parquet.Node {
+		return parquet.FieldID(parquet.Compressed(parquet.Optional(parquet.Leaf(parquet.Int64Type)), codec), id)
+	}
 	fields := append(append([]parquet.Field{}, base...),
-		lineageField{Node: parquet.FieldID(parquet.Optional(parquet.Leaf(parquet.Int64Type)), iceberg.RowIDFieldID), name: iceberg.RowIDColumnName},
-		lineageField{Node: parquet.FieldID(parquet.Optional(parquet.Leaf(parquet.Int64Type)), iceberg.LastUpdatedSequenceNumberFieldID), name: iceberg.LastUpdatedSequenceNumberColumnName},
+		lineageField{Node: lineageLeaf(iceberg.RowIDFieldID), name: iceberg.RowIDColumnName},
+		lineageField{Node: lineageLeaf(iceberg.LastUpdatedSequenceNumberFieldID), name: iceberg.LastUpdatedSequenceNumberColumnName},
 	)
 	return &lineageLayout{
 		schema:      parquet.NewSchema(reference.Name(), &orderedGroup{Node: reference, fields: fields}),
@@ -392,3 +401,26 @@ func mergedFileFirstRowID(data []byte) (*int64, error) {
 	}
 	return nil, nil
 }
+
+// lineageCodec is the codec of the first data column leaf that declares one,
+// ZSTD when none does.
+func lineageCodec(base []parquet.Field) compress.Codec {
+	for _, f := range base {
+		if c := columnCodec(f); c != nil {
+			return c
+		}
+	}
+	return &zstd.Codec{}
+}
+
+func columnCodec(n parquet.Node) compress.Codec {
+	if n.Leaf() {
+		return n.Compression()
+	}
+	for _, f := range n.Fields() {
+		if c := columnCodec(f); c != nil {
+			return c
+		}
+	}
+	return nil
+}

weed/worker/tasks/iceberg/row_lineage_test.go

@@ -9,6 +9,8 @@ import (
 
 	"github.com/apache/iceberg-go"
 	"github.com/parquet-go/parquet-go"
+	"github.com/parquet-go/parquet-go/compress/zstd"
+	"github.com/parquet-go/parquet-go/format"
 
 	filer_pb "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
@@ -382,3 +384,59 @@ func TestCompactDataFilesV3RowLineageManifestRead(t *testing.T) {
 		}
 	}
 }
+
+// The lineage columns a v3 compaction adds are compressed like the data
+// columns.
+func TestCompactCompressesLineageColumns(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "geo"}
+	populateGeoTable(t, fs, setup, fiveGeoFiles(), geoTableOptions{})
+	compactGeo(t, client, setup, Config{})
+	for name, data := range compactedFiles(t, fs, setup) {
+		f, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data)))
+		if err != nil {
+			t.Fatal(err)
+		}
+		seen := 0
+		for _, rg := range f.Metadata().RowGroups {
+			for _, c := range rg.Columns {
+				col := strings.Join(c.MetaData.PathInSchema, ".")
+				if col != iceberg.RowIDColumnName && col != iceberg.LastUpdatedSequenceNumberColumnName {
+					continue
+				}
+				seen++
+				if c.MetaData.Codec != format.Zstd {
+					t.Errorf("%s: %s written with codec %v, want ZSTD like the data columns", name, col, c.MetaData.Codec)
+				}
+			}
+		}
+		if seen == 0 {
+			t.Errorf("%s: no lineage column", name)
+		}
+	}
+}
+
+// The lineage codec follows the data columns: an input whose own _row_id
+// comes first and is uncompressed does not leave the new lineage columns
+// uncompressed.
+func TestLineageCodecFollowsDataColumns(t *testing.T) {
+	schema := parquet.NewSchema("table", parquet.Group{
+		iceberg.RowIDColumnName: parquet.FieldID(parquet.Compressed(parquet.Optional(parquet.Int(64)), &parquet.Uncompressed), iceberg.RowIDFieldID),
+		"id":                    parquet.FieldID(parquet.Compressed(parquet.Int(64), &zstd.Codec{}), 1),
+	})
+	if first := schema.Fields()[0].Name(); first != iceberg.RowIDColumnName {
+		t.Fatalf("setup: first column %s, want %s", first, iceberg.RowIDColumnName)
+	}
+	layout, err := newLineageLayout(schema)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, f := range layout.schema.Fields() {
+		if f.Name() != iceberg.RowIDColumnName && f.Name() != iceberg.LastUpdatedSequenceNumberColumnName {
+			continue
+		}
+		if c := f.Compression(); c == nil || c.CompressionCodec() != format.Zstd {
+			t.Errorf("%s compressed with %v, want ZSTD like the data column", f.Name(), c)
+		}
+	}
+}