From a1d3fe236f474c4134fc4125eec3416a0fed1832 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 16 Aug 2026 09:16:05 -0700 Subject: [PATCH] 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. --- .../maintenance_integration_test.go | 12 +- weed/worker/tasks/iceberg/config.go | 33 ++++- weed/worker/tasks/iceberg/detection.go | 10 +- weed/worker/tasks/iceberg/exec_test.go | 96 +++++++-------- weed/worker/tasks/iceberg/handler.go | 11 ++ weed/worker/tasks/iceberg/handler_test.go | 24 ++-- weed/worker/tasks/iceberg/operations.go | 2 +- weed/worker/tasks/iceberg/table_config.go | 58 +++++++++ .../worker/tasks/iceberg/table_config_test.go | 115 ++++++++++++++++++ 9 files changed, 286 insertions(+), 75 deletions(-) create mode 100644 weed/worker/tasks/iceberg/table_config.go create mode 100644 weed/worker/tasks/iceberg/table_config_test.go diff --git a/test/s3tables/maintenance/maintenance_integration_test.go b/test/s3tables/maintenance/maintenance_integration_test.go index 52bfb2340..4a5b76349 100644 --- a/test/s3tables/maintenance/maintenance_integration_test.go +++ b/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) diff --git a/weed/worker/tasks/iceberg/config.go b/weed/worker/tasks/iceberg/config.go index 22efef049..2eaa8cc21 100644 --- a/weed/worker/tasks/iceberg/config.go +++ b/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 } diff --git a/weed/worker/tasks/iceberg/detection.go b/weed/worker/tasks/iceberg/detection.go index c06d36156..320709583 100644 --- a/weed/worker/tasks/iceberg/detection.go +++ b/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 { diff --git a/weed/worker/tasks/iceberg/exec_test.go b/weed/worker/tasks/iceberg/exec_test.go index dbe4994b7..70033b4f4 100644 --- a/weed/worker/tasks/iceberg/exec_test.go +++ b/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) diff --git a/weed/worker/tasks/iceberg/handler.go b/weed/worker/tasks/iceberg/handler.go index 95fdcfc85..2830a4a49 100644 --- a/weed/worker/tasks/iceberg/handler.go +++ b/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) diff --git a/weed/worker/tasks/iceberg/handler_test.go b/weed/worker/tasks/iceberg/handler_test.go index 78af84739..8a3bf7f23 100644 --- a/weed/worker/tasks/iceberg/handler_test.go +++ b/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) diff --git a/weed/worker/tasks/iceberg/operations.go b/weed/worker/tasks/iceberg/operations.go index fcfa698f6..d00202c22 100644 --- a/weed/worker/tasks/iceberg/operations.go +++ b/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 diff --git a/weed/worker/tasks/iceberg/table_config.go b/weed/worker/tasks/iceberg/table_config.go new file mode 100644 index 000000000..457d4518e --- /dev/null +++ b/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 +} diff --git a/weed/worker/tasks/iceberg/table_config_test.go b/weed/worker/tasks/iceberg/table_config_test.go new file mode 100644 index 000000000..c94a8fb6c --- /dev/null +++ b/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) + } +}