iceberg: let table properties override the worker config (#10772)

* iceberg: carry snapshot retention in milliseconds

Config stored retention as hours, so any sub-hour value would have to be
truncated to 0 and then clamped back up to the 168 hour default. Keep the
plugin config key in hours and convert once at parse time.

* iceberg: let table properties override the worker config

Every other Iceberg implementation lets a table's own properties win over
engine defaults; the worker ignored them entirely. A writer honouring
write.target-file-size-bytes and a compactor rewriting to the plugin
config's size would rewrite each other's output forever.

Resolved once per job rather than per operation, so compaction committing
new metadata mid-job cannot change the settings underneath it.

* iceberg: clamp the orphan cutoff so it cannot overflow

collectOrphanCandidates converts the cutoff to a time.Duration. Past
roughly 2.5 million hours that multiplication wraps negative, putting the
cutoff in the future so every file walked looks like an orphan and gets
deleted, including data a concurrent writer has not yet committed.

Reachable today through orphan_older_than_hours.

test/s3tables/maintenance/maintenance_integration_test.go

@@ -600,9 +600,9 @@ func testExpireSnapshots(t *testing.T) {
 
 	handler := icebergHandler.NewHandler(nil)
 	config := icebergHandler.Config{
-		SnapshotRetentionHours: 0, // instant expiry — everything eligible
-		MaxSnapshotsToKeep:     1, // keep only the current snapshot
-		MaxCommitRetries:       3,
+		SnapshotRetentionMs: 0, // instant expiry — everything eligible
+		MaxSnapshotsToKeep:  1, // keep only the current snapshot
+		MaxCommitRetries:    3,
 	}
 
 	result, _, err := handler.ExpireSnapshots(context.Background(), client, bucket, path.Join(ns, tbl), config)
@@ -932,9 +932,9 @@ func testFullMaintenanceCycle(t *testing.T) {
 
 	// Step 1: Expire snapshots
 	expireConfig := icebergHandler.Config{
-		SnapshotRetentionHours: 0, // instant expiry
-		MaxSnapshotsToKeep:     1,
-		MaxCommitRetries:       3,
+		SnapshotRetentionMs: 0, // instant expiry
+		MaxSnapshotsToKeep:  1,
+		MaxCommitRetries:    3,
 	}
 	result, _, err := handler.ExpireSnapshots(ctx, client, bucket, tablePath, expireConfig)
 	require.NoError(t, err)

weed/worker/tasks/iceberg/config.go

@@ -2,8 +2,10 @@ package iceberg
 
 import (
 	"fmt"
+	"math"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
@@ -46,9 +48,29 @@ const (
 
 const bytesPerMB int64 = 1024 * 1024
 
+const msPerHour int64 = 3600 * 1000
+
+// maxOrphanOlderThanHours is the largest cutoff that still survives conversion
+// to a time.Duration. Beyond it the multiplication wraps negative, putting the
+// cutoff in the future and making every file look like an orphan.
+const maxOrphanOlderThanHours = int64(math.MaxInt64 / int64(time.Hour))
+
+// hoursToMs converts hours to milliseconds, saturating instead of overflowing.
+func hoursToMs(hours int64) int64 {
+	if hours <= 0 {
+		return 0
+	}
+	if hours > math.MaxInt64/msPerHour {
+		return math.MaxInt64
+	}
+	return hours * msPerHour
+}
+
 // Config holds parsed worker config values.
 type Config struct {
-	SnapshotRetentionHours      int64
+	// Milliseconds rather than hours so sub-hour retentions survive; the
+	// plugin config key stays in hours.
+	SnapshotRetentionMs         int64
 	MaxSnapshotsToKeep          int64
 	OrphanOlderThanHours        int64
 	MaxCommitRetries            int64
@@ -70,7 +92,7 @@ type Config struct {
 // Values are clamped to safe minimums to prevent misconfiguration.
 func ParseConfig(values map[string]*plugin_pb.ConfigValue) Config {
 	cfg := Config{
-		SnapshotRetentionHours:      readInt64Config(values, "snapshot_retention_hours", defaultSnapshotRetentionHours),
+		SnapshotRetentionMs:         hoursToMs(readInt64Config(values, "snapshot_retention_hours", defaultSnapshotRetentionHours)),
 		MaxSnapshotsToKeep:          readInt64Config(values, "max_snapshots_to_keep", defaultMaxSnapshotsToKeep),
 		OrphanOlderThanHours:        readInt64Config(values, "orphan_older_than_hours", defaultOrphanOlderThanHours),
 		MaxCommitRetries:            readInt64Config(values, "max_commit_retries", defaultMaxCommitRetries),
@@ -89,8 +111,8 @@ func ParseConfig(values map[string]*plugin_pb.ConfigValue) Config {
 	}
 
 	// Clamp the fields that are always defaulted by worker config parsing.
-	if cfg.SnapshotRetentionHours <= 0 {
-		cfg.SnapshotRetentionHours = defaultSnapshotRetentionHours
+	if cfg.SnapshotRetentionMs <= 0 {
+		cfg.SnapshotRetentionMs = hoursToMs(defaultSnapshotRetentionHours)
 	}
 	if cfg.MaxSnapshotsToKeep <= 0 {
 		cfg.MaxSnapshotsToKeep = defaultMaxSnapshotsToKeep
@@ -106,6 +128,9 @@ func applyThresholdDefaults(cfg Config) Config {
 	if cfg.OrphanOlderThanHours <= 0 {
 		cfg.OrphanOlderThanHours = defaultOrphanOlderThanHours
 	}
+	if cfg.OrphanOlderThanHours > maxOrphanOlderThanHours {
+		cfg.OrphanOlderThanHours = maxOrphanOlderThanHours
+	}
 	if cfg.TargetFileSizeBytes <= 0 {
 		cfg.TargetFileSizeBytes = defaultTargetFileSizeMB * 1024 * 1024
 	}

weed/worker/tasks/iceberg/detection.go

@@ -127,7 +127,9 @@ func (h *Handler) scanTablesForMaintenance(
 					continue
 				}
 
-				needsWork, err := h.tableNeedsMaintenance(ctx, filerClient, bucketName, tablePath, state, config, ops)
+				effective := resolveTableConfig(config, state.Metadata.Properties())
+
+				needsWork, err := h.tableNeedsMaintenance(ctx, filerClient, bucketName, tablePath, state, effective, ops)
 				if err != nil {
 					glog.V(2).Infof("iceberg maintenance: skipping %s/%s/%s: cannot evaluate maintenance need: %v", bucketName, nsName, tblName, err)
 					continue
@@ -154,8 +156,8 @@ func (h *Handler) scanTablesForMaintenance(
 
 func normalizeDetectionConfig(config Config) Config {
 	config = applyThresholdDefaults(config)
-	if config.SnapshotRetentionHours <= 0 {
-		config.SnapshotRetentionHours = defaultSnapshotRetentionHours
+	if config.SnapshotRetentionMs <= 0 {
+		config.SnapshotRetentionMs = hoursToMs(defaultSnapshotRetentionHours)
 	}
 	if config.MaxSnapshotsToKeep <= 0 {
 		config.MaxSnapshotsToKeep = defaultMaxSnapshotsToKeep
@@ -481,7 +483,7 @@ func needsMaintenance(meta table.Metadata, config Config) bool {
 	}
 
 	// Check oldest snapshot age
-	retentionMs := config.SnapshotRetentionHours * 3600 * 1000
+	retentionMs := config.SnapshotRetentionMs
 	nowMs := time.Now().UnixMilli()
 	for _, snap := range snapshots {
 		if nowMs-snap.TimestampMs > retentionMs {

weed/worker/tasks/iceberg/exec_test.go

@@ -557,10 +557,10 @@ func TestExpireSnapshotsExecution(t *testing.T) {
 
 	handler := NewHandler(nil)
 	config := Config{
-		SnapshotRetentionHours: 0, // expire everything eligible
-		MaxSnapshotsToKeep:     1, // keep only 1
-		MaxCommitRetries:       3,
-		Operations:             "expire_snapshots",
+		SnapshotRetentionMs: hoursToMs(0), // expire everything eligible
+		MaxSnapshotsToKeep:  1,            // keep only 1
+		MaxCommitRetries:    3,
+		Operations:          "expire_snapshots",
 	}
 
 	result, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config)
@@ -598,9 +598,9 @@ func TestExpireSnapshotsNothingToExpire(t *testing.T) {
 
 	handler := NewHandler(nil)
 	config := Config{
-		SnapshotRetentionHours: 24 * 365, // very long retention
-		MaxSnapshotsToKeep:     10,
-		MaxCommitRetries:       3,
+		SnapshotRetentionMs: hoursToMs(24 * 365), // very long retention
+		MaxSnapshotsToKeep:  10,
+		MaxCommitRetries:    3,
 	}
 
 	result, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config)
@@ -639,10 +639,10 @@ func TestMaintenanceOnExternalTableLocation(t *testing.T) {
 
 	handler := NewHandler(nil)
 	config := Config{
-		SnapshotRetentionHours: 0,
-		MaxSnapshotsToKeep:     1,
-		MaxCommitRetries:       3,
-		Operations:             "expire_snapshots",
+		SnapshotRetentionMs: hoursToMs(0),
+		MaxSnapshotsToKeep:  1,
+		MaxCommitRetries:    3,
+		Operations:          "expire_snapshots",
 	}
 
 	result, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config)
@@ -1245,9 +1245,9 @@ func TestDetectWithFakeFiler(t *testing.T) {
 	handler := NewHandler(nil)
 
 	config := Config{
-		SnapshotRetentionHours: 0, // everything is expired
-		MaxSnapshotsToKeep:     2, // 3 > 2, needs maintenance
-		MaxCommitRetries:       3,
+		SnapshotRetentionMs: hoursToMs(0), // everything is expired
+		MaxSnapshotsToKeep:  2,            // 3 > 2, needs maintenance
+		MaxCommitRetries:    3,
 	}
 
 	tables, err := handler.scanTablesForMaintenance(
@@ -1302,9 +1302,9 @@ func TestDetectWithFilters(t *testing.T) {
 
 	handler := NewHandler(nil)
 	config := Config{
-		SnapshotRetentionHours: 0,
-		MaxSnapshotsToKeep:     2,
-		MaxCommitRetries:       3,
+		SnapshotRetentionMs: hoursToMs(0),
+		MaxSnapshotsToKeep:  2,
+		MaxCommitRetries:    3,
 	}
 
 	// Without filter: should find both
@@ -1390,11 +1390,11 @@ func TestDetectSchedulesCompactionWithoutSnapshotPressure(t *testing.T) {
 
 	handler := NewHandler(nil)
 	config := Config{
-		SnapshotRetentionHours: 24 * 365,
-		MaxSnapshotsToKeep:     10,
-		TargetFileSizeBytes:    4096,
-		MinInputFiles:          2,
-		Operations:             "compact",
+		SnapshotRetentionMs: hoursToMs(24 * 365),
+		MaxSnapshotsToKeep:  10,
+		TargetFileSizeBytes: 4096,
+		MinInputFiles:       2,
+		Operations:          "compact",
 	}
 
 	tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
@@ -1495,11 +1495,11 @@ func TestDetectSchedulesCompactionWithDeleteManifestPresent(t *testing.T) {
 
 	handler := NewHandler(nil)
 	config := Config{
-		SnapshotRetentionHours: 24 * 365,
-		MaxSnapshotsToKeep:     10,
-		TargetFileSizeBytes:    4096,
-		MinInputFiles:          2,
-		Operations:             "compact",
+		SnapshotRetentionMs: hoursToMs(24 * 365),
+		MaxSnapshotsToKeep:  10,
+		TargetFileSizeBytes: 4096,
+		MinInputFiles:       2,
+		Operations:          "compact",
 	}
 
 	tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
@@ -1542,9 +1542,9 @@ func TestDetectSchedulesSnapshotExpiryDespiteCompactionEvaluationError(t *testin
 
 	handler := NewHandler(nil)
 	config := Config{
-		SnapshotRetentionHours: 24 * 365, // very long retention so age doesn't trigger
-		MaxSnapshotsToKeep:     1,        // 2 snapshots > 1 triggers expiry
-		Operations:             "compact,expire_snapshots",
+		SnapshotRetentionMs: hoursToMs(24 * 365), // very long retention so age doesn't trigger
+		MaxSnapshotsToKeep:  1,                   // 2 snapshots > 1 triggers expiry
+		Operations:          "compact,expire_snapshots",
 	}
 
 	tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
@@ -1580,10 +1580,10 @@ func TestDetectSchedulesManifestRewriteWithoutSnapshotPressure(t *testing.T) {
 
 	handler := NewHandler(nil)
 	config := Config{
-		SnapshotRetentionHours: 24 * 365,
-		MaxSnapshotsToKeep:     10,
-		MinManifestsToRewrite:  5,
-		Operations:             "rewrite_manifests",
+		SnapshotRetentionMs:   hoursToMs(24 * 365),
+		MaxSnapshotsToKeep:    10,
+		MinManifestsToRewrite: 5,
+		Operations:            "rewrite_manifests",
 	}
 
 	tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
@@ -2029,10 +2029,10 @@ func TestDetectDoesNotScheduleManifestRewriteFromDeleteManifestsOnly(t *testing.
 
 	handler := NewHandler(nil)
 	config := Config{
-		SnapshotRetentionHours: 24 * 365,
-		MaxSnapshotsToKeep:     10,
-		MinManifestsToRewrite:  2,
-		Operations:             "rewrite_manifests",
+		SnapshotRetentionMs:   hoursToMs(24 * 365),
+		MaxSnapshotsToKeep:    10,
+		MinManifestsToRewrite: 2,
+		Operations:            "rewrite_manifests",
 	}
 
 	tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
@@ -2070,10 +2070,10 @@ func TestDetectSchedulesOrphanCleanupWithoutSnapshotPressure(t *testing.T) {
 
 	handler := NewHandler(nil)
 	config := Config{
-		SnapshotRetentionHours: 24 * 365,
-		MaxSnapshotsToKeep:     10,
-		OrphanOlderThanHours:   72,
-		Operations:             "remove_orphans",
+		SnapshotRetentionMs:  hoursToMs(24 * 365),
+		MaxSnapshotsToKeep:   10,
+		OrphanOlderThanHours: 72,
+		Operations:           "remove_orphans",
 	}
 
 	tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
@@ -3043,9 +3043,9 @@ func TestExpireSnapshotsMetrics(t *testing.T) {
 
 	handler := NewHandler(nil)
 	config := Config{
-		SnapshotRetentionHours: 0,
-		MaxSnapshotsToKeep:     1,
-		MaxCommitRetries:       3,
+		SnapshotRetentionMs: hoursToMs(0),
+		MaxSnapshotsToKeep:  1,
+		MaxCommitRetries:    3,
 	}
 
 	_, metrics, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config)
@@ -3082,9 +3082,9 @@ func TestExecuteCompletionOutputValues(t *testing.T) {
 
 	handler := NewHandler(nil)
 	config := Config{
-		SnapshotRetentionHours: 0,
-		MaxSnapshotsToKeep:     1,
-		MaxCommitRetries:       3,
+		SnapshotRetentionMs: hoursToMs(0),
+		MaxSnapshotsToKeep:  1,
+		MaxCommitRetries:    3,
 	}
 
 	_, metrics, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config)

weed/worker/tasks/iceberg/handler.go

@@ -519,6 +519,17 @@ func (h *Handler) Execute(ctx context.Context, request *plugin_pb.ExecuteJobRequ
 	defer conn.Close()
 	filerClient := filer_pb.NewSeaweedFilerClient(conn)
 
+	// Resolve once for the whole job: compaction commits new metadata as it
+	// runs, and a job should not change settings halfway through. Failing to
+	// read the properties has to fail the job rather than fall back to the
+	// plugin config, or the operations would rewrite the table to a size it
+	// did not ask for.
+	state, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
+	if err != nil {
+		return fmt.Errorf("read table properties for %s/%s: %w", bucketName, tablePath, err)
+	}
+	workerConfig = resolveTableConfig(workerConfig, state.Metadata.Properties())
+
 	var results []string
 	var lastErr error
 	totalOps := len(ops)

weed/worker/tasks/iceberg/handler_test.go

@@ -19,8 +19,8 @@ import (
 func TestParseConfig(t *testing.T) {
 	config := ParseConfig(nil)
 
-	if config.SnapshotRetentionHours != defaultSnapshotRetentionHours {
-		t.Errorf("expected SnapshotRetentionHours=%d, got %d", defaultSnapshotRetentionHours, config.SnapshotRetentionHours)
+	if config.SnapshotRetentionMs != hoursToMs(defaultSnapshotRetentionHours) {
+		t.Errorf("expected SnapshotRetentionMs=%d, got %d", hoursToMs(defaultSnapshotRetentionHours), config.SnapshotRetentionMs)
 	}
 	if config.MaxSnapshotsToKeep != defaultMaxSnapshotsToKeep {
 		t.Errorf("expected MaxSnapshotsToKeep=%d, got %d", defaultMaxSnapshotsToKeep, config.MaxSnapshotsToKeep)
@@ -115,8 +115,8 @@ func TestExtractMetadataVersion(t *testing.T) {
 
 func TestNeedsMaintenanceNoSnapshots(t *testing.T) {
 	config := Config{
-		SnapshotRetentionHours: 24,
-		MaxSnapshotsToKeep:     2,
+		SnapshotRetentionMs: hoursToMs(24),
+		MaxSnapshotsToKeep:  2,
 	}
 
 	meta := buildTestMetadata(t, nil)
@@ -127,8 +127,8 @@ func TestNeedsMaintenanceNoSnapshots(t *testing.T) {
 
 func TestNeedsMaintenanceExceedsMaxSnapshots(t *testing.T) {
 	config := Config{
-		SnapshotRetentionHours: 24 * 365, // very long retention
-		MaxSnapshotsToKeep:     2,
+		SnapshotRetentionMs: hoursToMs(24 * 365), // very long retention
+		MaxSnapshotsToKeep:  2,
 	}
 
 	now := time.Now().UnixMilli()
@@ -145,8 +145,8 @@ func TestNeedsMaintenanceExceedsMaxSnapshots(t *testing.T) {
 
 func TestNeedsMaintenanceWithinLimits(t *testing.T) {
 	config := Config{
-		SnapshotRetentionHours: 24 * 365, // very long retention
-		MaxSnapshotsToKeep:     5,
+		SnapshotRetentionMs: hoursToMs(24 * 365), // very long retention
+		MaxSnapshotsToKeep:  5,
 	}
 
 	now := time.Now().UnixMilli()
@@ -162,8 +162,8 @@ func TestNeedsMaintenanceWithinLimits(t *testing.T) {
 func TestNeedsMaintenanceOldSnapshot(t *testing.T) {
 	// Use a retention of 0 hours so that any snapshot is considered "old"
 	config := Config{
-		SnapshotRetentionHours: 0, // instant expiry
-		MaxSnapshotsToKeep:     10,
+		SnapshotRetentionMs: hoursToMs(0), // instant expiry
+		MaxSnapshotsToKeep:  10,
 	}
 
 	now := time.Now().UnixMilli()
@@ -1005,8 +1005,8 @@ func TestNormalizeDetectionConfigUsesSharedDefaults(t *testing.T) {
 	if config.OrphanOlderThanHours != defaultOrphanOlderThanHours {
 		t.Fatalf("expected OrphanOlderThanHours default, got %d", config.OrphanOlderThanHours)
 	}
-	if config.SnapshotRetentionHours != defaultSnapshotRetentionHours {
-		t.Fatalf("expected SnapshotRetentionHours default, got %d", config.SnapshotRetentionHours)
+	if config.SnapshotRetentionMs != hoursToMs(defaultSnapshotRetentionHours) {
+		t.Fatalf("expected SnapshotRetentionMs default, got %d", config.SnapshotRetentionMs)
 	}
 	if config.MaxSnapshotsToKeep != defaultMaxSnapshotsToKeep {
 		t.Fatalf("expected MaxSnapshotsToKeep default, got %d", config.MaxSnapshotsToKeep)

weed/worker/tasks/iceberg/operations.go

@@ -58,7 +58,7 @@ func (h *Handler) expireSnapshots(
 		currentSnapID = currentSnap.SnapshotID
 	}
 
-	retentionMs := config.SnapshotRetentionHours * 3600 * 1000
+	retentionMs := config.SnapshotRetentionMs
 	nowMs := time.Now().UnixMilli()
 
 	// Sort snapshots by timestamp descending (most recent first) so that

weed/worker/tasks/iceberg/table_config.go

@@ -0,0 +1,58 @@
+package iceberg
+
+import (
+	"strconv"
+	"strings"
+
+	"github.com/apache/iceberg-go"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+)
+
+// Iceberg table properties that describe the physical layout maintenance has
+// to produce.
+const (
+	propTargetFileSize       = "write.target-file-size-bytes"
+	propDeleteTargetFileSize = "write.delete.target-file-size-bytes"
+	propMaxSnapshotAgeMs     = "history.expire.max-snapshot-age-ms"
+	propMinSnapshotsToKeep   = "history.expire.min-snapshots-to-keep"
+)
+
+// resolveTableConfig layers a table's own properties over the worker config.
+// Properties win: a writer targeting 512 MiB and a compactor rewriting to
+// 256 MiB would rewrite each other's output forever.
+func resolveTableConfig(base Config, props iceberg.Properties) Config {
+	cfg := base
+	if v, ok := propInt64(props, propTargetFileSize); ok {
+		cfg.TargetFileSizeBytes = v
+	}
+	if v, ok := propInt64(props, propDeleteTargetFileSize); ok {
+		cfg.DeleteTargetFileSizeBytes = v
+	}
+	if v, ok := propInt64(props, propMaxSnapshotAgeMs); ok {
+		cfg.SnapshotRetentionMs = v
+	}
+	if v, ok := propInt64(props, propMinSnapshotsToKeep); ok {
+		cfg.MaxSnapshotsToKeep = v
+	}
+	return applyThresholdDefaults(cfg)
+}
+
+// propInt64 reads a positive integer table property. Anything unset,
+// unparseable or non-positive leaves the worker config in place so a
+// misconfigured table never stops its own maintenance.
+func propInt64(props iceberg.Properties, key string) (int64, bool) {
+	raw, ok := props[key]
+	if !ok {
+		return 0, false
+	}
+	value, err := strconv.ParseInt(strings.TrimSpace(raw), 10, 64)
+	if err != nil {
+		glog.V(1).Infof("iceberg maintenance: ignoring table property %s=%q: not an integer", key, raw)
+		return 0, false
+	}
+	if value <= 0 {
+		glog.V(1).Infof("iceberg maintenance: ignoring table property %s=%d: not positive", key, value)
+		return 0, false
+	}
+	return value, true
+}

weed/worker/tasks/iceberg/table_config_test.go

@@ -0,0 +1,115 @@
+package iceberg
+
+import (
+	"math"
+	"testing"
+	"time"
+
+	"github.com/apache/iceberg-go"
+)
+
+func baseTestConfig() Config {
+	return applyThresholdDefaults(Config{
+		SnapshotRetentionMs: hoursToMs(defaultSnapshotRetentionHours),
+		MaxSnapshotsToKeep:  defaultMaxSnapshotsToKeep,
+	})
+}
+
+func TestResolveTableConfigNoProperties(t *testing.T) {
+	base := baseTestConfig()
+
+	got := resolveTableConfig(base, iceberg.Properties{})
+	if got != base {
+		t.Errorf("expected config untouched, got %+v", got)
+	}
+
+	if got := resolveTableConfig(base, nil); got != base {
+		t.Errorf("expected config untouched for nil properties, got %+v", got)
+	}
+}
+
+func TestResolveTableConfigPropertiesWin(t *testing.T) {
+	base := baseTestConfig()
+	got := resolveTableConfig(base, iceberg.Properties{
+		propTargetFileSize:       "536870912",
+		propDeleteTargetFileSize: "33554432",
+		propMaxSnapshotAgeMs:     "432000000",
+		propMinSnapshotsToKeep:   "1",
+	})
+
+	if got.TargetFileSizeBytes != 536870912 {
+		t.Errorf("expected TargetFileSizeBytes=536870912, got %d", got.TargetFileSizeBytes)
+	}
+	if got.DeleteTargetFileSizeBytes != 33554432 {
+		t.Errorf("expected DeleteTargetFileSizeBytes=33554432, got %d", got.DeleteTargetFileSizeBytes)
+	}
+	if got.SnapshotRetentionMs != 432000000 {
+		t.Errorf("expected SnapshotRetentionMs=432000000, got %d", got.SnapshotRetentionMs)
+	}
+	if got.MaxSnapshotsToKeep != 1 {
+		t.Errorf("expected MaxSnapshotsToKeep=1, got %d", got.MaxSnapshotsToKeep)
+	}
+}
+
+// A sub-hour retention has to survive resolution; truncating it to whole hours
+// would round down to zero and get clamped back up to the default.
+func TestResolveTableConfigSubHourRetention(t *testing.T) {
+	got := resolveTableConfig(baseTestConfig(), iceberg.Properties{propMaxSnapshotAgeMs: "1"})
+	if got.SnapshotRetentionMs != 1 {
+		t.Errorf("expected SnapshotRetentionMs=1, got %d", got.SnapshotRetentionMs)
+	}
+}
+
+func TestResolveTableConfigIgnoresUnusableValues(t *testing.T) {
+	base := baseTestConfig()
+
+	for name, value := range map[string]string{
+		"not a number": "512mb",
+		"empty":        "",
+		"zero":         "0",
+		"negative":     "-1",
+		"overflow":     "99999999999999999999",
+	} {
+		t.Run(name, func(t *testing.T) {
+			got := resolveTableConfig(base, iceberg.Properties{propTargetFileSize: value})
+			if got.TargetFileSizeBytes != base.TargetFileSizeBytes {
+				t.Errorf("expected fallback to %d, got %d", base.TargetFileSizeBytes, got.TargetFileSizeBytes)
+			}
+		})
+	}
+}
+
+func TestResolveTableConfigTrimsWhitespace(t *testing.T) {
+	got := resolveTableConfig(baseTestConfig(), iceberg.Properties{propTargetFileSize: "  536870912\n"})
+	if got.TargetFileSizeBytes != 536870912 {
+		t.Errorf("expected TargetFileSizeBytes=536870912, got %d", got.TargetFileSizeBytes)
+	}
+}
+
+func TestResolveTableConfigLeavesOtherFields(t *testing.T) {
+	base := baseTestConfig()
+	base.Operations = "compact"
+	base.Where = "day = 3"
+	base.MinInputFiles = 9
+
+	got := resolveTableConfig(base, iceberg.Properties{propTargetFileSize: "536870912"})
+	if got.Operations != "compact" || got.Where != "day = 3" || got.MinInputFiles != 9 {
+		t.Errorf("expected unrelated fields preserved, got %+v", got)
+	}
+}
+
+// An orphan cutoff large enough to overflow time.Duration would wrap negative,
+// put the cutoff in the future and make every file look like an orphan.
+func TestApplyThresholdDefaultsClampsOrphanCutoff(t *testing.T) {
+	got := applyThresholdDefaults(Config{OrphanOlderThanHours: math.MaxInt64})
+	if got.OrphanOlderThanHours != maxOrphanOlderThanHours {
+		t.Fatalf("expected the cutoff clamped to %d, got %d", maxOrphanOlderThanHours, got.OrphanOlderThanHours)
+	}
+	if cutoff := time.Duration(got.OrphanOlderThanHours) * time.Hour; cutoff <= 0 {
+		t.Errorf("expected a positive cutoff duration, got %v", cutoff)
+	}
+
+	if got := applyThresholdDefaults(Config{OrphanOlderThanHours: 72}); got.OrphanOlderThanHours != 72 {
+		t.Errorf("expected a normal cutoff untouched, got %d", got.OrphanOlderThanHours)
+	}
+}