This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 3fb910983 fix(compaction): rewrite complete source files (#2288)
3fb910983 is described below
commit 3fb910983b9674efc6092418af90a1c2b821dd42
Author: Matt Faltyn <[email protected]>
AuthorDate: Mon Oct 26 08:38:46 2026 -0700
fix(compaction): rewrite complete source files (#2288)
* fix(compaction): rewrite complete source files
Signed-off-by: Matt Faltyn <[email protected]>
* test(compaction): format validation callbacks
Signed-off-by: Matt Faltyn <[email protected]>
* fix(compaction): validate split metadata and clear residuals
Signed-off-by: Matt Faltyn <[email protected]>
---------
Signed-off-by: Matt Faltyn <[email protected]>
---
table/compaction/compaction.go | 50 ++++++-
table/compaction/compaction_file_grain_test.go | 163 +++++++++++++++++++++++
table/compaction_file_grain_test.go | 173 +++++++++++++++++++++++++
table/rewrite_data_files.go | 25 ++++
website/src/api.md | 2 +
5 files changed, 410 insertions(+), 3 deletions(-)
diff --git a/table/compaction/compaction.go b/table/compaction/compaction.go
index 31c8ba8e5..c1c5d026f 100644
--- a/table/compaction/compaction.go
+++ b/table/compaction/compaction.go
@@ -21,6 +21,8 @@ package compaction
import (
"bytes"
"fmt"
+ "reflect"
+ "slices"
"github.com/apache/iceberg-go"
"github.com/apache/iceberg-go/internal"
@@ -188,16 +190,44 @@ type Plan struct {
// PlanCompaction analyzes the given scan tasks and produces a Plan.
// Tasks are grouped by partition, classified as candidates or skipped,
// and candidates are bin-packed into groups targeting TargetFileSizeBytes.
+// Tasks should come from one coherent scan plan; split tasks for the same file
+// are collapsed, retaining the first task's per-file delete and lineage data.
+// Residual filters select files to rewrite, but are not applied to file
contents.
func (cfg Config) PlanCompaction(tasks []table.FileScanTask) (Plan, error) {
if err := cfg.Validate(); err != nil {
return Plan{}, fmt.Errorf("invalid compaction config: %w", err)
}
+ // Compaction operates on whole physical files. A scan planner may
split a
+ // file into multiple tasks, so collapse tasks by path before counting,
+ // selecting, or packing them. Keep the first task's delete and lineage
+ // metadata while replacing its range with the complete file range.
+ wholeFileTasks := make([]table.FileScanTask, 0, len(tasks))
+ seenFiles := make(map[string]table.FileScanTask, len(tasks))
+ for _, task := range tasks {
+ if task.File == nil {
+ return Plan{}, fmt.Errorf("%w: compaction task has no
data file", table.ErrInvalidOperation)
+ }
+ path := task.File.FilePath()
+ if first, ok := seenFiles[path]; ok {
+ if !sameCompactionMetadata(first, task) {
+ return Plan{}, fmt.Errorf("%w: split tasks for
%q have different delete or lineage metadata", table.ErrInvalidOperation, path)
+ }
+
+ continue
+ }
+ seenFiles[path] = task
+ task.Start = 0
+ task.Length = task.File.FileSizeBytes()
+ task.Residual = nil
+ wholeFileTasks = append(wholeFileTasks, task)
+ }
+
plan := Plan{
- TotalInputFiles: len(tasks),
+ TotalInputFiles: len(wholeFileTasks),
}
- for _, t := range tasks {
+ for _, t := range wholeFileTasks {
plan.TotalInputBytes += t.File.FileSizeBytes()
}
@@ -211,7 +241,7 @@ func (cfg Config) PlanCompaction(tasks
[]table.FileScanTask) (Plan, error) {
partitionIndices := make(map[string]int)
var partitionBuckets []partitionBucket
- for _, t := range tasks {
+ for _, t := range wholeFileTasks {
if !cfg.isCandidate(t) {
plan.SkippedFiles++
@@ -273,6 +303,20 @@ func (cfg Config) PlanCompaction(tasks
[]table.FileScanTask) (Plan, error) {
return plan, nil
}
+func sameCompactionMetadata(a, b table.FileScanTask) bool {
+ equalDataFiles := func(a, b []iceberg.DataFile) bool {
+ return slices.EqualFunc(a, b, func(a, b iceberg.DataFile) bool {
+ return reflect.DeepEqual(a, b)
+ })
+ }
+
+ return reflect.DeepEqual(a.FirstRowID, b.FirstRowID) &&
+ reflect.DeepEqual(a.DataSequenceNumber, b.DataSequenceNumber) &&
+ equalDataFiles(a.DeleteFiles, b.DeleteFiles) &&
+ equalDataFiles(a.EqualityDeleteFiles, b.EqualityDeleteFiles) &&
+ equalDataFiles(a.DeletionVectorFiles, b.DeletionVectorFiles)
+}
+
// isCandidate returns true if a file should be considered for compaction.
func (cfg Config) isCandidate(t table.FileScanTask) bool {
size := t.File.FileSizeBytes()
diff --git a/table/compaction/compaction_file_grain_test.go
b/table/compaction/compaction_file_grain_test.go
new file mode 100644
index 000000000..40a57193c
--- /dev/null
+++ b/table/compaction/compaction_file_grain_test.go
@@ -0,0 +1,163 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package compaction_test
+
+import (
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/table"
+ "github.com/apache/iceberg-go/table/compaction"
+ "github.com/stretchr/testify/require"
+)
+
+type offsetDataFile struct {
+ iceberg.DataFile
+ offset int64
+}
+
+func (f offsetDataFile) ContentOffset() *int64 { return &f.offset }
+
+func TestPlanCompactionCountsWholeFiles(t *testing.T) {
+ file := newDataFile("split.parquet", 10)
+ firstRowID, sequence := int64(20), int64(3)
+ first := makeTask(file, 1, 1)
+ first.DeletionVectorFiles =
[]iceberg.DataFile{newScopedDeleteFile("delete.puffin", file.path, 1)}
+ first.FirstRowID, first.DataSequenceNumber = &firstRowID, &sequence
+ first.Residual = iceberg.EqualTo(iceberg.Reference("id"), int64(1))
+ first.Length = file.size / 2
+ second := first
+ second.Start, second.Length = first.Length, file.size-first.Length
+ second.FirstRowID = new(int64)
+ *second.FirstRowID = firstRowID
+ second.DataSequenceNumber = new(int64)
+ *second.DataSequenceNumber = sequence
+ second.DeletionVectorFiles =
[]iceberg.DataFile{newScopedDeleteFile("delete.puffin", file.path, 1)}
+ duplicate := *file
+ first.File = &duplicate
+ tasks := []table.FileScanTask{second, first,
makeTask(newDataFile("other.parquet", 10), 0, 0)}
+ original := append([]table.FileScanTask(nil), tasks...)
+
+ cfg := compaction.DefaultConfig()
+ cfg.MinInputFiles = 1
+ plan, err := cfg.PlanCompaction(tasks)
+ require.NoError(t, err)
+ require.Equal(t, 2, plan.TotalInputFiles)
+ require.Equal(t, 2*file.size, plan.TotalInputBytes)
+ require.Len(t, plan.Groups, 1)
+ require.Len(t, plan.Groups[0].Tasks, 2)
+ require.Equal(t, 2*file.size, plan.Groups[0].TotalSizeBytes)
+ require.Equal(t, 3, plan.Groups[0].DeleteFileCount)
+ got := plan.Groups[0].Tasks[0]
+ require.Same(t, second.File, got.File)
+ require.Same(t, second.FirstRowID, got.FirstRowID)
+ require.Same(t, second.DataSequenceNumber, got.DataSequenceNumber)
+ require.Same(t, second.DeletionVectorFiles[0],
got.DeletionVectorFiles[0])
+ require.Equal(t, int64(0), got.Start)
+ require.Equal(t, file.size, got.Length)
+ require.Nil(t, got.Residual, "compaction rewrites all rows in the
selected file")
+ require.Equal(t, original, tasks, "normalization must not mutate caller
tasks")
+}
+
+func TestPlanCompactionRejectsConflictingSplitMetadata(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ change func(*table.FileScanTask)
+ }{
+ {"first row ID", func(task *table.FileScanTask) { value :=
int64(21); task.FirstRowID = &value }},
+ {"missing first row ID", func(task *table.FileScanTask) {
task.FirstRowID = nil }},
+ {"data sequence number", func(task *table.FileScanTask) { value
:= int64(4); task.DataSequenceNumber = &value }},
+ {"missing data sequence number", func(task *table.FileScanTask)
{ task.DataSequenceNumber = nil }},
+ {"positional deletes", func(task *table.FileScanTask) {
task.DeleteFiles = []iceberg.DataFile{newDeleteFile("other.pos")} }},
+ {"same delete path, different metadata", func(task
*table.FileScanTask) {
+ task.DeleteFiles =
[]iceberg.DataFile{newDeleteFile("same.pos")}
+ }},
+ {"delete content offset", func(task *table.FileScanTask) {
+ base := newDeleteFile("same.puffin")
+ task.DeleteFiles =
[]iceberg.DataFile{offsetDataFile{DataFile: base, offset: 2}}
+ }},
+ {"equality deletes", func(task *table.FileScanTask) {
+ task.EqualityDeleteFiles =
[]iceberg.DataFile{newDeleteFile("other.eq")}
+ }},
+ {"deletion vectors", func(task *table.FileScanTask) {
+ task.DeletionVectorFiles =
[]iceberg.DataFile{newScopedDeleteFile("other.puffin", "split.parquet", 1)}
+ }},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ file := newDataFile("split.parquet", 10)
+ first := makeTask(file, 0, 0)
+ first.DeleteFiles = nil
+ if tc.name == "same delete path, different metadata" {
+ first.DeleteFiles =
[]iceberg.DataFile{newDeleteFile("same.pos")}
+ first.DeleteFiles[0] = &testDataFile{path:
"same.pos", size: 1024, records: 5, content: iceberg.EntryContentPosDeletes}
+ }
+ first.FirstRowID, first.DataSequenceNumber =
new(int64), new(int64)
+ if tc.name == "delete content offset" {
+ first.DeleteFiles =
[]iceberg.DataFile{offsetDataFile{DataFile: newDeleteFile("same.puffin"),
offset: 1}}
+ }
+ second := first
+ second.Start, second.Length = file.size/2, file.size/2
+ tc.change(&second)
+ _, err :=
compaction.DefaultConfig().PlanCompaction([]table.FileScanTask{first, second})
+ require.ErrorIs(t, err, table.ErrInvalidOperation)
+ })
+ }
+}
+
+func TestPlanCompactionTreatsNilAndEmptyDeleteListsEqually(t *testing.T) {
+ file := newDataFile("split.parquet", 10)
+ first := makeTask(file, 0, 0)
+ first.DeleteFiles = nil
+ second := first
+ second.Start, second.Length = file.size/2, file.size/2
+ second.DeleteFiles = []iceberg.DataFile{}
+ _, err :=
compaction.DefaultConfig().PlanCompaction([]table.FileScanTask{first, second})
+ require.NoError(t, err)
+}
+
+func TestPlanCompactionDoesNotCountRangesTowardMinimumFiles(t *testing.T) {
+ file := newDataFile("split.parquet", 10)
+ tasks := make([]table.FileScanTask, 5)
+ for i := range tasks {
+ tasks[i] = table.FileScanTask{File: file, Start: int64(i) *
file.size / 5, Length: file.size / 5}
+ }
+ plan, err := compaction.DefaultConfig().PlanCompaction(tasks)
+ require.NoError(t, err)
+ require.Equal(t, 1, plan.TotalInputFiles)
+ require.Equal(t, file.size, plan.TotalInputBytes)
+ require.Equal(t, 1, plan.SkippedFiles)
+ require.Empty(t, plan.Groups, "five ranges are not five source files")
+}
+
+func TestPlanCompactionCountsSkippedFileOnce(t *testing.T) {
+ file := newDataFile("optimal.parquet", 512)
+ plan, err :=
compaction.DefaultConfig().PlanCompaction([]table.FileScanTask{
+ {File: file, Length: file.size / 2},
+ {File: file, Start: file.size / 2, Length: file.size / 2},
+ })
+ require.NoError(t, err)
+ require.Equal(t, 1, plan.TotalInputFiles)
+ require.Equal(t, file.size, plan.TotalInputBytes)
+ require.Equal(t, 1, plan.SkippedFiles)
+ require.Empty(t, plan.Groups)
+}
+
+func TestPlanCompactionRejectsNilFile(t *testing.T) {
+ _, err :=
compaction.DefaultConfig().PlanCompaction([]table.FileScanTask{{}})
+ require.ErrorIs(t, err, table.ErrInvalidOperation)
+}
diff --git a/table/compaction_file_grain_test.go
b/table/compaction_file_grain_test.go
new file mode 100644
index 000000000..5631297d0
--- /dev/null
+++ b/table/compaction_file_grain_test.go
@@ -0,0 +1,173 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table_test
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "strconv"
+ "strings"
+ "sync/atomic"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table"
+ "github.com/apache/iceberg-go/table/compaction"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestCompactionPreservesSplitFileRows(t *testing.T) {
+ for _, split := range []bool{false, true} {
+ for _, partial := range []bool{false, true} {
+ for _, filtered := range []bool{false, true} {
+
t.Run(fmt.Sprintf("split=%t/partial=%t/filtered=%t", split, partial, filtered),
func(t *testing.T) {
+ tbl := newRewriteTestTable(t)
+ sc, err :=
table.SchemaToArrowSchema(tbl.Schema(), nil, false, false)
+ require.NoError(t, err)
+ var rows []string
+ var want []int64
+ for i := range 12 {
+ rows = append(rows,
fmt.Sprintf(`{"id":%d,"data":"a"}`, i+1))
+ want = append(want, int64(i+1))
+ }
+ path := tbl.Location() +
"/data/split.parquet"
+ writeParquetFileWithProperties(t, path,
sc, "["+strings.Join(rows, ",")+"]", 1, nil)
+ splitSize := "999999999"
+ if split {
+ splitSize = "1"
+ }
+ tx := tbl.NewTransaction()
+ require.NoError(t,
tx.SetProperties(iceberg.Properties{table.ReadSplitTargetSizeKey: splitSize}))
+ require.NoError(t,
tx.AddFiles(t.Context(), []string{path}, nil, false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ scan := tbl.Scan()
+ filter :=
iceberg.GreaterThan(iceberg.Reference("id"), int64(6))
+ if filtered {
+ scan =
tbl.Scan(table.WithRowFilter(filter))
+ }
+ tasks, err :=
scan.PlanFiles(t.Context())
+ require.NoError(t, err)
+ if filtered {
+ // Exercise a residual a remote
planner may leave after scan planning.
+ for i := range tasks {
+ tasks[i].Residual =
filter
+ }
+ require.NotNil(t,
tasks[0].Residual)
+ require.False(t,
tasks[0].Residual.Equals(iceberg.AlwaysTrue{}))
+ }
+ if split {
+ require.Len(t, tasks, 12)
+ } else {
+ require.Len(t, tasks, 1)
+ }
+ size := tasks[0].File.FileSizeBytes()
+ cfg := defaultTestCompactionCfg
+ cfg.TargetFileSizeBytes,
cfg.MinFileSizeBytes, cfg.MaxFileSizeBytes = size+1, size+1, size*2
+ cfg.MinInputFiles = 1
+ var groups []compaction.Group
+ if filtered {
+ plan, planErr :=
cfg.PlanCompaction(tasks)
+ require.NoError(t, planErr)
+ groups = plan.Groups
+ } else {
+ plan, planErr :=
compaction.Analyze(t.Context(), tbl, cfg)
+ require.NoError(t, planErr)
+ groups = plan.Groups
+ }
+ result, err :=
tbl.RewriteDataFiles(t.Context(), toTaskGroups(groups),
table.RewriteDataFilesOptions{
+ PartialProgress: partial,
MaxCommits: 12,
+ })
+ assert.NoError(t, err)
+ require.NotNil(t, result)
+ require.NotNil(t, result.Table)
+ require.Equal(t, 1,
result.RewrittenGroups)
+ require.ElementsMatch(t, want,
scanIDs(t, result.Table), "a later batch failure must not lose unread source
ranges")
+ })
+ }
+ }
+ }
+}
+
+func TestExecuteCompactionGroupRejectsIncompleteFilesBeforeIO(t *testing.T) {
+ base := newRewriteTestTable(t)
+ builder, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec,
iceberg.EntryContentData,
+ "missing.parquet", iceberg.ParquetFile, nil, nil, nil, 1, 100)
+ require.NoError(t, err)
+ file := builder.Build()
+ whole := table.FileScanTask{File: file, Length: 100}
+ filtered := whole
+ filtered.Residual = iceberg.EqualTo(iceberg.Reference("id"), int64(1))
+ cases := []struct {
+ name string
+ tasks []table.FileScanTask
+ }{
+ {"nil file", []table.FileScanTask{{}}},
+ {"nonzero start", []table.FileScanTask{{File: file, Start: 1,
Length: 99}}},
+ {"negative start", []table.FileScanTask{{File: file, Start: -1,
Length: 100}}},
+ {"short range", []table.FileScanTask{{File: file, Length: 99}}},
+ {"long range", []table.FileScanTask{{File: file, Length: 101}}},
+ {"negative length", []table.FileScanTask{{File: file, Length:
-1}}},
+ {"duplicate path", []table.FileScanTask{whole, whole}},
+ {"residual filter", []table.FileScanTask{filtered}},
+ {"false residual", []table.FileScanTask{{File: file, Length:
100, Residual: iceberg.AlwaysFalse{}}}},
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ var calls atomic.Int32
+ tbl := table.New(base.Identifier(), base.Metadata(),
base.MetadataLocation(),
+ func(context.Context) (iceio.IO, error) {
+ calls.Add(1)
+
+ return nil, errors.New("unexpected IO")
+ }, nil)
+ result, err :=
table.ExecuteCompactionGroup(t.Context(), tbl, table.CompactionTaskGroup{Tasks:
tc.tasks})
+ require.ErrorIs(t, err, table.ErrInvalidOperation)
+ require.Zero(t, calls.Load(), "invalid groups must fail
before opening input or output files")
+ require.Empty(t, result.NewDataFiles)
+ })
+ }
+}
+
+func TestExecuteCompactionGroupAcceptsWholeFileRanges(t *testing.T) {
+ base := newRewriteTestTable(t)
+ builder, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec,
iceberg.EntryContentData,
+ "missing.parquet", iceberg.ParquetFile, nil, nil, nil, 1, 100)
+ require.NoError(t, err)
+ file := builder.Build()
+ for _, length := range []int64{0, 100} {
+ t.Run(strconv.FormatInt(length, 10), func(t *testing.T) {
+ var calls atomic.Int32
+ ioErr := errors.New("input reached")
+ tbl := table.New(base.Identifier(), base.Metadata(),
base.MetadataLocation(),
+ func(context.Context) (iceio.IO, error) {
+ calls.Add(1)
+
+ return nil, ioErr
+ }, nil)
+ _, err := table.ExecuteCompactionGroup(t.Context(),
tbl, table.CompactionTaskGroup{Tasks: []table.FileScanTask{
+ {File: file, Length: length, Residual:
iceberg.AlwaysTrue{}},
+ }})
+ require.ErrorIs(t, err, ioErr)
+ require.Equal(t, int32(1), calls.Load())
+ })
+ }
+}
diff --git a/table/rewrite_data_files.go b/table/rewrite_data_files.go
index a1a4f7159..d198b4bea 100644
--- a/table/rewrite_data_files.go
+++ b/table/rewrite_data_files.go
@@ -418,6 +418,11 @@ func (t *Transaction) RewriteDataFiles(ctx
context.Context, groups []CompactionT
// Empty groups return a zero [CompactionGroupResult] without doing
// any I/O.
//
+// Every task must represent one complete physical data file with no residual
+// row filter. Compaction applies the attached deletes, then replaces each
+// source file in full; partial ranges or residual filters would otherwise
+// silently discard rows when the original task is replaced.
+//
// In-process callers should prefer [Transaction.RewriteDataFiles],
// which drives this and the commit step in one call.
//
@@ -429,6 +434,26 @@ func ExecuteCompactionGroup(ctx context.Context, tbl
*Table, group CompactionTas
return CompactionGroupResult{PartitionKey: group.PartitionKey},
nil
}
+ seenFiles := make(map[string]struct{}, len(group.Tasks))
+ for _, task := range group.Tasks {
+ if task.File == nil {
+ return CompactionGroupResult{}, fmt.Errorf("%w:
compaction task has no data file", ErrInvalidOperation)
+ }
+ path := task.File.FilePath()
+ if _, ok := seenFiles[path]; ok {
+ return CompactionGroupResult{}, fmt.Errorf("%w:
compaction group contains duplicate data file %q", ErrInvalidOperation, path)
+ }
+ seenFiles[path] = struct{}{}
+
+ fileSize := task.File.FileSizeBytes()
+ if task.Start != 0 || task.Length < 0 || (task.Length > 0 &&
task.Length != fileSize) {
+ return CompactionGroupResult{}, fmt.Errorf("%w:
compaction task for %q must cover the whole data file", ErrInvalidOperation,
path)
+ }
+ if task.Residual != nil &&
!task.Residual.Equals(iceberg.AlwaysTrue{}) {
+ return CompactionGroupResult{}, fmt.Errorf("%w:
compaction task for %q has a residual row filter", ErrInvalidOperation, path)
+ }
+ }
+
cfg := compactionGroupConfig{}
for _, opt := range opts {
opt(&cfg)
diff --git a/website/src/api.md b/website/src/api.md
index 85504ee5a..bf854e5a2 100644
--- a/website/src/api.md
+++ b/website/src/api.md
@@ -520,6 +520,8 @@ fmt.Printf("rewrote %d files into %d (%d -> %d bytes)\n",
The `table/compaction` subpackage provides bin-packing planning. The `iceberg
compact analyze` and `compact run` CLI commands wrap the same machinery - see
[CLI](./cli.md).
+Compaction tasks represent complete data files, not independent scan splits.
The planner collapses split tasks for the same file into one whole-file task,
rejecting inconsistent delete or lineage metadata and dropping residual filters
because they select files but do not filter rows during compaction. Direct
calls to `ExecuteCompactionGroup` reject partial ranges and residual row
filters. Attached delete files are still applied when rewriting each source
file.
+
### Expiring snapshots
```go
