Files
seaweedfs/weed/worker/tasks/iceberg/exec_test.go
T
dependabot[bot]andChris Lu 2d4b730a2f build(deps): bump github.com/twmb/avro from 1.7.2 to 1.8.0 (#11210)
* build(deps): bump github.com/twmb/avro from 1.7.2 to 1.8.0

Bumps [github.com/twmb/avro](https://github.com/twmb/avro) from 1.7.2 to 1.8.0.
- [Commits](https://github.com/twmb/avro/compare/v1.7.2...v1.8.0)

---
updated-dependencies:
- dependency-name: github.com/twmb/avro
  dependency-version: 1.8.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>

* iceberg: adapt to twmb/avro v1.8.0 and iceberg-go defensive copies

avro v1.8.0 changes Schema.Root() to return *SchemaNode, which breaks
iceberg-go v0.6.0's internal avro_schemas.go. The fix (apache/iceberg-go#1843)
is only on iceberg-go's main branch, unreleased, so bump iceberg-go to
that commit (c210509) alongside the avro bump.

That iceberg-go revision also changes two behaviors seaweedfs worked
around:

- It now infers a manifest list's format version from the embedded
  writer schema, so a list missing the "format-version" header entry
  (DuckDB's shape is read as v2, not v1. ReadManifestList's header
  patching is now a redundant safety net; tests updated to expect v2.

- It returns defensive copies from DataFile.Partition(), so the
  ReadManifest shim's in-place partition normalization was silently
  discarded. Rebuild the entry through NewDataFileBuilder when any
  partition value is normalized, copying every other DataFile field so
  manifest round-trips are preserved.

- It converts day-transform partitions to iceberg.Date on read
  (applyDayTransformDates), so the day-partition cases the shim and
  tests guarded now convert without help; tests updated to expect
  iceberg.Date from the raw read.
EOF
)

* iceberg: accept assert-ref-snapshot-id without snapshot-id

iceberg-go's new nullableInt64 parser rejects an assert-ref-snapshot-id
requirement whose "snapshot-id" field is absent from the JSON, even
though the Iceberg REST spec makes it optional (null means the ref must
not already exist). v0.6.0 used a plain *int64, so absent was nil and
accepted. ClickHouse sends the requirement without snapshot-id when
asserting a branch does not yet exist, so its writes fail with
"missing required field \"snapshot-id\"".

normalizeRequirements splices an explicit null into any
assert-ref-snapshot-id requirement missing the field before handing
the JSON to iceberg-go's parser, restoring the v0.6.0 behavior across
both iceberg-go versions.

* iceberg: fix v1 block_size_in_bytes default in rebuilt manifest entries

rebuildManifestEntry set block_size_in_bytes to 0, but the v1 manifest
schema requires the default of 64 MiB ("Always write default in v1").
The original value is not exposed on the DataFile interface, so use the
spec default. Also clarify the fallback comment to note that empty
(zero-record / zero-byte) files also trigger it, not just a nil spec.

Add a round-trip test that writes a rebuilt entry as v1 and verifies
block_size_in_bytes is 64 MiB via Avro decoding.

---------

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>
2026-09-07 15:58:24 -07:00

4451 lines
134 KiB
Go

package iceberg
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net"
"path"
"sort"
"strconv"
"strings"
"sync"
"testing"
"time"
"github.com/apache/iceberg-go"
"github.com/apache/iceberg-go/table"
"github.com/parquet-go/parquet-go"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/proto"
)
// ---------------------------------------------------------------------------
// Fake filer server for execution tests
// ---------------------------------------------------------------------------
// fakeFilerServer is an in-memory filer that implements the gRPC methods used
// by the iceberg maintenance handler.
type fakeFilerServer struct {
filer_pb.UnimplementedSeaweedFilerServer
mu sync.Mutex
entries map[string]map[string]*filer_pb.Entry // dir → name → entry
beforeUpdate func(*fakeFilerServer, *filer_pb.UpdateEntryRequest) error
beforeLookup func(*fakeFilerServer, *filer_pb.LookupDirectoryEntryRequest)
// Set by enableAssign to serve AssignVolume against a fake volume server.
assignVolumeServer string
assignCount int
// Counters for assertions
createCalls int
updateCalls int
deleteCalls int
}
func newFakeFilerServer() *fakeFilerServer {
return &fakeFilerServer{
entries: make(map[string]map[string]*filer_pb.Entry),
}
}
func (f *fakeFilerServer) putEntry(dir, name string, entry *filer_pb.Entry) {
f.mu.Lock()
defer f.mu.Unlock()
if _, ok := f.entries[dir]; !ok {
f.entries[dir] = make(map[string]*filer_pb.Entry)
}
f.entries[dir][name] = entry
}
func (f *fakeFilerServer) getEntry(dir, name string) *filer_pb.Entry {
f.mu.Lock()
defer f.mu.Unlock()
if dirEntries, ok := f.entries[dir]; ok {
return dirEntries[name]
}
return nil
}
func (f *fakeFilerServer) listDir(dir string) []*filer_pb.Entry {
f.mu.Lock()
defer f.mu.Unlock()
dirEntries, ok := f.entries[dir]
if !ok {
return nil
}
result := make([]*filer_pb.Entry, 0, len(dirEntries))
for _, e := range dirEntries {
result = append(result, e)
}
sort.Slice(result, func(i, j int) bool {
return result[i].Name < result[j].Name
})
return result
}
func (f *fakeFilerServer) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
f.mu.Lock()
beforeLookup := f.beforeLookup
f.mu.Unlock()
if beforeLookup != nil {
beforeLookup(f, req)
}
entry := f.getEntry(req.Directory, req.Name)
if entry == nil {
return nil, status.Errorf(codes.NotFound, "entry not found: %s/%s", req.Directory, req.Name)
}
return &filer_pb.LookupDirectoryEntryResponse{Entry: entry}, nil
}
func (f *fakeFilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream grpc.ServerStreamingServer[filer_pb.ListEntriesResponse]) error {
entries := f.listDir(req.Directory)
if entries == nil {
return nil // empty directory
}
var sent uint32
for _, entry := range entries {
if req.Prefix != "" && !strings.HasPrefix(entry.Name, req.Prefix) {
continue
}
if req.StartFromFileName != "" {
if req.InclusiveStartFrom {
if entry.Name < req.StartFromFileName {
continue
}
} else {
if entry.Name <= req.StartFromFileName {
continue
}
}
}
if err := stream.Send(&filer_pb.ListEntriesResponse{Entry: entry}); err != nil {
return err
}
sent++
if req.Limit > 0 && sent >= req.Limit {
break
}
}
return nil
}
func (f *fakeFilerServer) CreateEntry(_ context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
f.mu.Lock()
f.createCalls++
f.mu.Unlock()
f.putEntry(req.Directory, req.Entry.Name, req.Entry)
return &filer_pb.CreateEntryResponse{}, nil
}
func (f *fakeFilerServer) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) {
f.mu.Lock()
f.updateCalls++
beforeUpdate := f.beforeUpdate
f.beforeUpdate = nil
f.mu.Unlock()
if beforeUpdate != nil {
if err := beforeUpdate(f, req); err != nil {
return nil, err
}
}
f.mu.Lock()
defer f.mu.Unlock()
dirEntries, ok := f.entries[req.Directory]
if !ok {
return nil, status.Errorf(codes.NotFound, "entry not found: %s/%s", req.Directory, req.Entry.Name)
}
current := dirEntries[req.Entry.Name]
if current == nil {
return nil, status.Errorf(codes.NotFound, "entry not found: %s/%s", req.Directory, req.Entry.Name)
}
for key, expectedValue := range req.ExpectedExtended {
actualValue, ok := current.Extended[key]
if ok {
if !bytes.Equal(actualValue, expectedValue) {
return nil, status.Errorf(codes.FailedPrecondition, "extended attribute %q changed", key)
}
continue
}
if len(expectedValue) > 0 {
return nil, status.Errorf(codes.FailedPrecondition, "extended attribute %q changed", key)
}
}
dirEntries[req.Entry.Name] = req.Entry
return &filer_pb.UpdateEntryResponse{}, nil
}
func (f *fakeFilerServer) DeleteEntry(_ context.Context, req *filer_pb.DeleteEntryRequest) (*filer_pb.DeleteEntryResponse, error) {
f.mu.Lock()
f.deleteCalls++
if dirEntries, ok := f.entries[req.Directory]; ok {
delete(dirEntries, req.Name)
}
f.mu.Unlock()
return &filer_pb.DeleteEntryResponse{}, nil
}
func (f *fakeFilerServer) AssignVolume(_ context.Context, _ *filer_pb.AssignVolumeRequest) (*filer_pb.AssignVolumeResponse, error) {
f.mu.Lock()
defer f.mu.Unlock()
if f.assignVolumeServer == "" {
return nil, status.Error(codes.Unavailable, "no volume server")
}
f.assignCount++
return &filer_pb.AssignVolumeResponse{
FileId: fmt.Sprintf("1,%08x", f.assignCount),
Location: &filer_pb.Location{Url: f.assignVolumeServer, PublicUrl: f.assignVolumeServer},
Count: 1,
}, nil
}
func (f *fakeFilerServer) Ping(_ context.Context, _ *filer_pb.PingRequest) (*filer_pb.PingResponse, error) {
now := time.Now().UnixNano()
return &filer_pb.PingResponse{
StartTimeNs: now,
RemoteTimeNs: now,
StopTimeNs: now,
}, nil
}
// startFakeFiler starts a gRPC server and returns a connected client.
func startFakeFiler(t *testing.T) (*fakeFilerServer, filer_pb.SeaweedFilerClient) {
t.Helper()
fakeServer, client, _ := startFakeFilerWithAddress(t)
return fakeServer, client
}
func startFakeFilerWithAddress(t *testing.T) (*fakeFilerServer, filer_pb.SeaweedFilerClient, string) {
t.Helper()
fakeServer := newFakeFilerServer()
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen: %v", err)
}
server := grpc.NewServer()
filer_pb.RegisterSeaweedFilerServer(server, fakeServer)
go func() { _ = server.Serve(listener) }()
t.Cleanup(server.GracefulStop)
conn, err := grpc.NewClient(listener.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
t.Fatalf("dial: %v", err)
}
t.Cleanup(func() { conn.Close() })
client := filer_pb.NewSeaweedFilerClient(conn)
deadline := time.Now().Add(5 * time.Second)
for {
pingCtx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
_, err := client.Ping(pingCtx, &filer_pb.PingRequest{})
cancel()
if err == nil {
break
}
if time.Now().After(deadline) {
t.Fatalf("filer not ready: %v", err)
}
code := status.Code(err)
if code != codes.Unavailable && code != codes.DeadlineExceeded && code != codes.Canceled {
t.Fatalf("unexpected filer readiness error: %v", err)
}
time.Sleep(10 * time.Millisecond)
}
// Return the gRPC address in dialable form (host:grpcPort) since
// dialFiler dials the resolved filer_grpc_address verbatim.
return fakeServer, client, listener.Addr().String()
}
// ---------------------------------------------------------------------------
// Helpers to populate the fake filer with Iceberg table state
// ---------------------------------------------------------------------------
// tableSetup holds the state needed to set up a test table in the fake filer.
type tableSetup struct {
BucketName string
Namespace string
TableName string
// DataPath is the bucket-relative directory holding the table's files when
// it differs from the catalog path, as it does for tables an external REST
// client created at their own location (e.g. "ns/table-<uuid>"). Such a
// table records absolute s3:// URIs for every file it references.
DataPath string
// SnapshotSummary is recorded on the table's snapshot, the way the writer
// that created it would record the table's running totals.
SnapshotSummary map[string]string
Snapshots []table.Snapshot
// Refs are branches and tags beyond main, which always points at the last
// snapshot.
Refs map[string]table.SnapshotRef
// Age backdates the whole table so its snapshots can sit outside a
// retention window.
Age time.Duration
// Schema and Spec describe the table when the default unpartitioned test
// table does not fit.
Schema *iceberg.Schema
Spec *iceberg.PartitionSpec
}
func (ts tableSetup) tablePath() string {
return path.Join(ts.Namespace, ts.TableName)
}
func (ts tableSetup) dataPath() string {
if ts.DataPath != "" {
return ts.DataPath
}
return ts.tablePath()
}
// fileRef builds the reference a table records for one of its files.
func (ts tableSetup) fileRef(elem ...string) string {
if ts.DataPath == "" {
return path.Join(elem...)
}
return absoluteIcebergPath(ts.BucketName, append([]string{ts.DataPath}, elem...)...)
}
// populateTable creates the directory hierarchy and metadata entries in the
// fake filer for a table, writes manifest files referenced by snapshots,
// and returns the built metadata.
func populateTable(t *testing.T, fs *fakeFilerServer, setup tableSetup) table.Metadata {
t.Helper()
meta := buildTestMetadata(t, setup.Snapshots, setup.Refs, setup.Age, nil, setup.Schema, setup.Spec)
fullMetadataJSON, err := json.Marshal(meta)
if err != nil {
t.Fatalf("marshal metadata: %v", err)
}
// Build internal metadata xattr
const metadataVersion = 1
internalMeta := map[string]interface{}{
"metadataVersion": metadataVersion,
"metadataLocation": setup.fileRef("metadata", fmt.Sprintf("v%d.metadata.json", metadataVersion)),
"metadata": map[string]interface{}{
"fullMetadata": json.RawMessage(fullMetadataJSON),
},
}
xattr, err := json.Marshal(internalMeta)
if err != nil {
t.Fatalf("marshal xattr: %v", err)
}
bucketsPath := s3tables.TablesPath // "/buckets"
bucketPath := path.Join(bucketsPath, setup.BucketName)
nsPath := path.Join(bucketPath, setup.Namespace)
tableFilerPath := path.Join(bucketPath, setup.dataPath())
// Register bucket entry (marked as table bucket)
fs.putEntry(bucketsPath, setup.BucketName, &filer_pb.Entry{
Name: setup.BucketName,
IsDirectory: true,
Extended: map[string][]byte{
s3tables.ExtendedKeyTableBucket: []byte("true"),
},
})
// Register namespace entry
fs.putEntry(bucketPath, setup.Namespace, &filer_pb.Entry{
Name: setup.Namespace,
IsDirectory: true,
})
// Register table entry with metadata xattr
fs.putEntry(nsPath, setup.TableName, &filer_pb.Entry{
Name: setup.TableName,
IsDirectory: true,
Extended: map[string][]byte{
s3tables.ExtendedKeyMetadata: xattr,
s3tables.ExtendedKeyMetadataVersion: metadataVersionXattr(metadataVersion),
},
})
// Create metadata/ and data/ directory placeholders
metaDir := path.Join(tableFilerPath, "metadata")
dataDir := path.Join(tableFilerPath, "data")
// Write manifest files for each snapshot that has a ManifestList
schema := meta.CurrentSchema()
spec := meta.PartitionSpec()
version := meta.Version()
for _, snap := range setup.Snapshots {
if snap.ManifestList == "" {
continue
}
// Create a minimal manifest with one dummy entry for this snapshot
dfBuilder, err := iceberg.NewDataFileBuilder(
spec,
iceberg.EntryContentData,
setup.fileRef("data", fmt.Sprintf("snap-%d-data.parquet", snap.SnapshotID)),
iceberg.ParquetFile,
map[int]any{},
nil, nil,
10, // recordCount
4096, // fileSizeBytes
)
if err != nil {
t.Fatalf("build data file for snap %d: %v", snap.SnapshotID, err)
}
snapID := snap.SnapshotID
entry := iceberg.NewManifestEntry(
iceberg.EntryStatusADDED,
&snapID,
nil, nil,
dfBuilder.Build(),
)
// Write manifest
manifestFileName := fmt.Sprintf("manifest-%d.avro", snap.SnapshotID)
manifestPath := setup.fileRef("metadata", manifestFileName)
var manifestBuf bytes.Buffer
mf, err := iceberg.WriteManifest(manifestPath, &manifestBuf, version, spec, schema, snap.SnapshotID, []iceberg.ManifestEntry{entry})
if err != nil {
t.Fatalf("write manifest for snap %d: %v", snap.SnapshotID, err)
}
fs.putEntry(metaDir, manifestFileName, &filer_pb.Entry{
Name: manifestFileName,
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Unix(),
FileSize: uint64(manifestBuf.Len()),
},
Content: manifestBuf.Bytes(),
})
// Write manifest list
manifestListFileName := path.Base(snap.ManifestList)
var mlBuf bytes.Buffer
parentSnap := snap.ParentSnapshotID
seqNum := snap.SequenceNumber
if err := iceberg.WriteManifestList(version, &mlBuf, snap.SnapshotID, parentSnap, &seqNum, 0, []iceberg.ManifestFile{mf}); err != nil {
t.Fatalf("write manifest list for snap %d: %v", snap.SnapshotID, err)
}
fs.putEntry(metaDir, manifestListFileName, &filer_pb.Entry{
Name: manifestListFileName,
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Unix(),
FileSize: uint64(mlBuf.Len()),
},
Content: mlBuf.Bytes(),
})
// Write a dummy data file
dataFileName := fmt.Sprintf("snap-%d-data.parquet", snap.SnapshotID)
fs.putEntry(dataDir, dataFileName, &filer_pb.Entry{
Name: dataFileName,
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Unix(),
FileSize: 4096,
},
Content: []byte("fake-parquet-data"),
})
}
return meta
}
func writeCurrentSnapshotManifests(t *testing.T, fs *fakeFilerServer, setup tableSetup, meta table.Metadata, manifestEntries [][]iceberg.ManifestEntry) {
t.Helper()
currentSnap := meta.CurrentSnapshot()
if currentSnap == nil {
t.Fatal("current snapshot is required")
}
metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata")
version := meta.Version()
schema := meta.CurrentSchema()
spec := meta.PartitionSpec()
var manifests []iceberg.ManifestFile
for i, entries := range manifestEntries {
manifestName := fmt.Sprintf("detect-manifest-%d.avro", i+1)
var manifestBuf bytes.Buffer
mf, err := iceberg.WriteManifest(
path.Join("metadata", manifestName),
&manifestBuf,
version,
spec,
schema,
currentSnap.SnapshotID,
entries,
)
if err != nil {
t.Fatalf("write manifest %d: %v", i+1, err)
}
fs.putEntry(metaDir, manifestName, &filer_pb.Entry{
Name: manifestName,
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Unix(),
FileSize: uint64(manifestBuf.Len()),
},
Content: manifestBuf.Bytes(),
})
manifests = append(manifests, mf)
}
var manifestListBuf bytes.Buffer
seqNum := currentSnap.SequenceNumber
if err := iceberg.WriteManifestList(version, &manifestListBuf, currentSnap.SnapshotID, currentSnap.ParentSnapshotID, &seqNum, 0, manifests); err != nil {
t.Fatalf("write current manifest list: %v", err)
}
fs.putEntry(metaDir, path.Base(currentSnap.ManifestList), &filer_pb.Entry{
Name: path.Base(currentSnap.ManifestList),
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Unix(),
FileSize: uint64(manifestListBuf.Len()),
},
Content: manifestListBuf.Bytes(),
})
}
func makeManifestEntries(t *testing.T, specs []testEntrySpec, snapshotID int64) []iceberg.ManifestEntry {
t.Helper()
return makeManifestEntriesWithSnapshot(t, specs, snapshotID, iceberg.EntryStatusADDED)
}
// ---------------------------------------------------------------------------
// Recording senders for Execute tests
// ---------------------------------------------------------------------------
type recordingExecutionSender struct {
mu sync.Mutex
progress []*plugin_pb.JobProgressUpdate
completed *plugin_pb.JobCompleted
}
func (r *recordingExecutionSender) SendProgress(p *plugin_pb.JobProgressUpdate) error {
r.mu.Lock()
defer r.mu.Unlock()
r.progress = append(r.progress, p)
return nil
}
func (r *recordingExecutionSender) SendCompleted(c *plugin_pb.JobCompleted) error {
r.mu.Lock()
defer r.mu.Unlock()
r.completed = c
return nil
}
// ---------------------------------------------------------------------------
// Execution tests
// ---------------------------------------------------------------------------
func TestExpireSnapshotsExecution(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().Add(-10 * time.Second).UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
},
}
populateTable(t, fs, setup)
handler := NewHandler(nil)
config := Config{
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)
if err != nil {
t.Fatalf("expireSnapshots failed: %v", err)
}
if !strings.Contains(result, "expired") {
t.Errorf("expected result to mention expiration, got %q", result)
}
t.Logf("expireSnapshots result: %s", result)
// Verify the metadata was updated (update calls > 0)
fs.mu.Lock()
updates := fs.updateCalls
fs.mu.Unlock()
if updates == 0 {
t.Error("expected at least one UpdateEntry call for xattr update")
}
}
func TestExpireSnapshotsKeepsTaggedSnapshot(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().Add(-10 * time.Second).UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
},
Refs: map[string]table.SnapshotRef{
"release": {SnapshotID: 1, SnapshotRefType: table.TagRef},
},
}
populateTable(t, fs, setup)
handler := NewHandler(nil)
config := Config{
SnapshotRetentionMs: hoursToMs(0),
MaxSnapshotsToKeep: 1,
MaxCommitRetries: 3,
Operations: "expire_snapshots",
}
if _, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config); err != nil {
t.Fatalf("expireSnapshots failed: %v", err)
}
state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
if err != nil {
t.Fatalf("reload metadata: %v", err)
}
remaining := map[int64]bool{}
for _, snap := range state.Metadata.Snapshots() {
remaining[snap.SnapshotID] = true
}
if !remaining[1] {
t.Error("tagged snapshot 1 was expired")
}
if remaining[2] {
t.Error("expected untagged snapshot 2 to be expired")
}
var tagged bool
for name, ref := range state.Metadata.Refs() {
if name == "release" {
tagged = true
if ref.SnapshotID != 1 {
t.Errorf("tag release points at snapshot %d, want 1", ref.SnapshotID)
}
}
}
if !tagged {
t.Error("tag release was dropped from the metadata")
}
metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.dataPath(), "metadata")
if fs.getEntry(metaDir, "snap-1.avro") == nil {
t.Error("manifest list of the tagged snapshot was deleted")
}
if fs.getEntry(metaDir, "snap-2.avro") != nil {
t.Error("expected the expired snapshot's manifest list to be deleted")
}
}
func TestExpireSnapshotsHonorsBranchRetention(t *testing.T) {
fs, client := startFakeFiler(t)
// The branch chain is 1 <- 2 <- 3, with main on 4. Backdating the table by
// Age gives every snapshot a real age; the ten-second spacing keeps them
// distinct and stays inside what iceberg-go accepts at build time.
now := time.Now().Add(-30 * time.Second).UnixMilli()
stepMs := int64(10 * time.Second / time.Millisecond)
first, second := int64(1), int64(2)
minKeep := 2
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
{SnapshotID: 2, TimestampMs: now + stepMs, ManifestList: "metadata/snap-2.avro", ParentSnapshotID: &first, SequenceNumber: 2},
{SnapshotID: 3, TimestampMs: now + 2*stepMs, ManifestList: "metadata/snap-3.avro", ParentSnapshotID: &second, SequenceNumber: 3},
{SnapshotID: 4, TimestampMs: now + 3*stepMs, ManifestList: "metadata/snap-4.avro", SequenceNumber: 4},
},
Refs: map[string]table.SnapshotRef{
"audit": {SnapshotID: 3, SnapshotRefType: table.BranchRef, MinSnapshotsToKeep: &minKeep},
},
Age: 5 * time.Hour,
}
populateTable(t, fs, setup)
handler := NewHandler(nil)
config := Config{
SnapshotRetentionMs: hoursToMs(0),
MaxSnapshotsToKeep: 1,
MaxCommitRetries: 3,
Operations: "expire_snapshots",
}
if _, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config); err != nil {
t.Fatalf("expireSnapshots failed: %v", err)
}
// min-snapshots-to-keep=2 reaches from the branch head back to its parent,
// and stops there: the grandparent is past the count and expires.
assertSnapshots(t, client, setup, []int64{2, 3, 4}, []int64{1})
}
func TestExpireSnapshotsHonorsBranchMaxSnapshotAge(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().Add(-30 * time.Second).UnixMilli()
stepMs := int64(10 * time.Second / time.Millisecond)
first, second := int64(1), int64(2)
// Between the branch head's parent (5h20s) and its grandparent (5h30s), so
// the window has to reach past the head to be observable.
maxAgeMs := int64(5*time.Hour/time.Millisecond) + 25*1000
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
{SnapshotID: 2, TimestampMs: now + stepMs, ManifestList: "metadata/snap-2.avro", ParentSnapshotID: &first, SequenceNumber: 2},
{SnapshotID: 3, TimestampMs: now + 2*stepMs, ManifestList: "metadata/snap-3.avro", ParentSnapshotID: &second, SequenceNumber: 3},
{SnapshotID: 4, TimestampMs: now + 3*stepMs, ManifestList: "metadata/snap-4.avro", SequenceNumber: 4},
},
Refs: map[string]table.SnapshotRef{
"audit": {SnapshotID: 3, SnapshotRefType: table.BranchRef, MaxSnapshotAgeMs: &maxAgeMs},
},
Age: 5 * time.Hour,
}
populateTable(t, fs, setup)
handler := NewHandler(nil)
config := Config{
SnapshotRetentionMs: hoursToMs(0),
MaxSnapshotsToKeep: 1,
MaxCommitRetries: 3,
Operations: "expire_snapshots",
}
if _, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config); err != nil {
t.Fatalf("expireSnapshots failed: %v", err)
}
// The window reaches from the branch head back over its parent but stops
// short of the grandparent.
assertSnapshots(t, client, setup, []int64{2, 3, 4}, []int64{1})
}
// assertSnapshots reloads the table and checks exactly which snapshots survived.
func assertSnapshots(t *testing.T, client filer_pb.SeaweedFilerClient, setup tableSetup, want, gone []int64) {
t.Helper()
state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
if err != nil {
t.Fatalf("reload metadata: %v", err)
}
remaining := map[int64]bool{}
for _, snap := range state.Metadata.Snapshots() {
remaining[snap.SnapshotID] = true
}
for _, id := range want {
if !remaining[id] {
t.Errorf("snapshot %d was expired, want kept", id)
}
}
for _, id := range gone {
if remaining[id] {
t.Errorf("snapshot %d survived, want expired", id)
}
}
}
// A tag created between planning and commit pins a snapshot the plan was going
// to expire. The head has not moved, so only a refs check can catch it.
func TestExpireSnapshotsRejectsPlanWhenARefAppears(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().Add(-30 * time.Second).UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
},
Age: 200 * time.Hour,
}
populateTable(t, fs, setup)
// Re-publish the table with a tag on snapshot 1 while the commit is
// re-reading it, which is the window the guard closes.
tagged := setup
tagged.Refs = map[string]table.SnapshotRef{
"release": {SnapshotID: 1, SnapshotRefType: table.TagRef},
}
lookups := 0
fs.beforeLookup = func(f *fakeFilerServer, req *filer_pb.LookupDirectoryEntryRequest) {
if req.Name != setup.TableName {
return
}
lookups++
if lookups == 2 {
populateTable(t, f, tagged)
}
}
handler := NewHandler(nil)
config := Config{
SnapshotRetentionMs: hoursToMs(0),
MaxSnapshotsToKeep: 1,
MaxCommitRetries: 1,
Operations: "expire_snapshots",
}
_, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config)
if err == nil {
t.Fatal("expireSnapshots() error = nil, want the plan rejected as stale")
}
if !errors.Is(err, errStalePlan) {
t.Fatalf("expireSnapshots() error = %v, want errStalePlan", err)
}
metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.dataPath(), "metadata")
if fs.getEntry(metaDir, "snap-1.avro") == nil {
t.Error("the newly tagged snapshot's manifest list was deleted")
}
}
func TestExpireSnapshotsNothingToExpire(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().Add(-10 * time.Second).UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "ns",
TableName: "tbl",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
},
}
populateTable(t, fs, setup)
handler := NewHandler(nil)
config := Config{
SnapshotRetentionMs: hoursToMs(24 * 365), // very long retention
MaxSnapshotsToKeep: 10,
MaxCommitRetries: 3,
}
result, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config)
if err != nil {
t.Fatalf("expireSnapshots failed: %v", err)
}
if result != "no snapshots expired" {
t.Errorf("expected 'no snapshots expired', got %q", result)
}
}
// A table created through the Iceberg REST catalog can live outside its
// catalog path, and every location it records is then an absolute s3:// URI
// under that other directory. Maintenance must follow the recorded location
// instead of joining the URI onto the catalog path.
func TestMaintenanceOnExternalTableLocation(t *testing.T) {
fs, client := startFakeFiler(t)
const dataPath = "source/events-0cd81bca-b389-4dda-8dae-3502ce680501"
snapAt := func(id int64) string {
return fmt.Sprintf("s3://lake/%s/metadata/snap-%d.avro", dataPath, id)
}
now := time.Now().Add(-10 * time.Second).UnixMilli()
setup := tableSetup{
BucketName: "lake",
Namespace: "source",
TableName: "events",
DataPath: dataPath,
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: snapAt(1)},
{SnapshotID: 2, TimestampMs: now + 1, ManifestList: snapAt(2)},
{SnapshotID: 3, TimestampMs: now + 2, ManifestList: snapAt(3)},
},
}
populateTable(t, fs, setup)
handler := NewHandler(nil)
config := Config{
SnapshotRetentionMs: hoursToMs(0),
MaxSnapshotsToKeep: 1,
MaxCommitRetries: 3,
Operations: "expire_snapshots",
}
result, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config)
if err != nil {
t.Fatalf("expireSnapshots: %v", err)
}
if !strings.Contains(result, "expired 2 snapshot(s)") {
t.Errorf("expected two snapshots expired, got %q", result)
}
// The expired snapshots' files are gone from the table's own directory.
metaDir := path.Join(s3tables.TablesPath, setup.BucketName, dataPath, "metadata")
for _, name := range []string{"snap-1.avro", "snap-2.avro"} {
if fs.getEntry(metaDir, name) != nil {
t.Errorf("expected %s/%s to be deleted", metaDir, name)
}
}
if fs.getEntry(metaDir, "snap-3.avro") == nil {
t.Errorf("expected %s/snap-3.avro to be kept", metaDir)
}
// The new metadata file is written next to the table's files, and the
// catalog now points at it there.
state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
if err != nil {
t.Fatalf("loadCurrentMetadata: %v", err)
}
if fs.getEntry(metaDir, state.MetadataFileName) == nil {
t.Errorf("expected new metadata file %s in %s", state.MetadataFileName, metaDir)
}
if got := len(state.Metadata.Snapshots()); got != 1 {
t.Errorf("expected 1 snapshot left, got %d", got)
}
}
// Compaction of such a table must keep its output with the rest of the table
// instead of splitting it across the catalog path and the table's location.
func TestCompactDataFilesOnExternalTableLocation(t *testing.T) {
fs, client := startFakeFiler(t)
setup := tableSetup{
BucketName: "lake",
Namespace: "source",
TableName: "events",
DataPath: "source/events-0cd81bca-b389-4dda-8dae-3502ce680501",
}
populateTableWithDeleteFiles(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{1, "a"}, {2, "b"}}},
{"d2.parquet", []struct {
ID int64
Name string
}{{3, "c"}}},
},
nil, nil,
)
handler := NewHandler(nil)
config := Config{
TargetFileSizeBytes: 256 * 1024 * 1024,
MinInputFiles: 2,
MaxCommitRetries: 3,
ApplyDeletes: true,
}
result, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil)
if err != nil {
t.Fatalf("compactDataFiles: %v", err)
}
if !strings.Contains(result, "compacted 2 files into 1") {
t.Fatalf("unexpected result: %q", result)
}
state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
if err != nil {
t.Fatalf("loadCurrentMetadata: %v", err)
}
if state.DataPath != setup.DataPath {
t.Fatalf("data path = %q, want %q", state.DataPath, setup.DataPath)
}
// Everything the new snapshot names lives under the table's own location.
wantPrefix := "s3://" + setup.BucketName + "/" + setup.DataPath + "/"
newSnap := state.Metadata.CurrentSnapshot()
if newSnap == nil || !strings.HasPrefix(newSnap.ManifestList, wantPrefix+"metadata/snap-") {
t.Fatalf("manifest list %q should be under %q", newSnap.ManifestList, wantPrefix)
}
manifests, err := loadCurrentManifests(context.Background(), client, setup.BucketName, state.DataPath, state.Metadata)
if err != nil {
t.Fatalf("loadCurrentManifests: %v", err)
}
for _, mf := range manifests {
if !strings.HasPrefix(mf.FilePath(), wantPrefix+"metadata/") {
t.Errorf("manifest %q should be under %q", mf.FilePath(), wantPrefix)
}
manifestData, err := loadFileByIcebergPath(context.Background(), client, setup.BucketName, state.DataPath, mf.FilePath())
if err != nil {
t.Fatalf("load manifest: %v", err)
}
entries, err := iceberg.ReadManifest(mf, bytes.NewReader(manifestData), true)
if err != nil {
t.Fatalf("read manifest: %v", err)
}
for _, entry := range entries {
if !strings.HasPrefix(entry.DataFile().FilePath(), wantPrefix+"data/") {
t.Errorf("data file %q should be under %q", entry.DataFile().FilePath(), wantPrefix)
}
}
}
// Nothing was written to the catalog path.
catalogDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath())
for _, subdir := range []string{"metadata", "data"} {
if entries := fs.listDir(path.Join(catalogDir, subdir)); len(entries) > 0 {
t.Errorf("expected no files under %s/%s, got %d", catalogDir, subdir, len(entries))
}
}
}
func TestRemoveOrphansExecution(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
},
}
populateTable(t, fs, setup)
// Add orphan files (old enough to be removed)
metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata")
dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data")
oldTime := time.Now().Add(-200 * time.Hour).Unix()
fs.putEntry(metaDir, "orphan-old.avro", &filer_pb.Entry{
Name: "orphan-old.avro",
Attributes: &filer_pb.FuseAttributes{Mtime: oldTime},
})
fs.putEntry(dataDir, "orphan-data.parquet", &filer_pb.Entry{
Name: "orphan-data.parquet",
Attributes: &filer_pb.FuseAttributes{Mtime: oldTime},
})
// Add a recent orphan that should NOT be removed (within safety window)
fs.putEntry(dataDir, "recent-orphan.parquet", &filer_pb.Entry{
Name: "recent-orphan.parquet",
Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix()},
})
handler := NewHandler(nil)
config := Config{
OrphanOlderThanHours: 72,
MaxCommitRetries: 3,
}
result, _, err := handler.removeOrphans(context.Background(), client, setup.BucketName, setup.tablePath(), config)
if err != nil {
t.Fatalf("removeOrphans failed: %v", err)
}
if !strings.Contains(result, "removed 2 orphan") {
t.Errorf("expected 2 orphans removed, got %q", result)
}
// Verify orphan files were deleted
if fs.getEntry(metaDir, "orphan-old.avro") != nil {
t.Error("orphan-old.avro should have been deleted")
}
if fs.getEntry(dataDir, "orphan-data.parquet") != nil {
t.Error("orphan-data.parquet should have been deleted")
}
// Recent orphan should still exist
if fs.getEntry(dataDir, "recent-orphan.parquet") == nil {
t.Error("recent-orphan.parquet should NOT have been deleted (within safety window)")
}
}
func TestRemoveOrphansPreservesReferencedFiles(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "ns",
TableName: "tbl",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
},
}
populateTable(t, fs, setup)
handler := NewHandler(nil)
config := Config{
OrphanOlderThanHours: 0, // no safety window — remove immediately
MaxCommitRetries: 3,
}
result, _, err := handler.removeOrphans(context.Background(), client, setup.BucketName, setup.tablePath(), config)
if err != nil {
t.Fatalf("removeOrphans failed: %v", err)
}
if !strings.Contains(result, "removed 0 orphan") {
t.Errorf("expected 0 orphans removed (all files are referenced), got %q", result)
}
// Verify referenced files are still present
metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata")
if fs.getEntry(metaDir, "snap-1.avro") == nil {
t.Error("snap-1.avro (referenced manifest list) should not have been deleted")
}
if fs.getEntry(metaDir, "manifest-1.avro") == nil {
t.Error("manifest-1.avro (referenced manifest) should not have been deleted")
}
}
// DuckDB writes manifest lists carrying no Iceberg header metadata. iceberg-go
// now infers the format version from the embedded writer schema, so a stripped
// v2 list is read as v2 and maintenance operations proceed normally. This used
// to fail because iceberg-go fell back to v1 and rejected every v2 manifest.
func TestMaintenanceOnManifestListWithoutFormatVersion(t *testing.T) {
fs, client := startFakeFiler(t)
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "test",
TableName: "foo",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: time.Now().UnixMilli(), ManifestList: "metadata/snap-1.avro"},
},
}
populateTable(t, fs, setup)
metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata")
dropManifestListFormatVersion(t, fs, metaDir, "snap-1.avro")
handler := NewHandler(nil)
config := Config{
OrphanOlderThanHours: 0, // no safety window — every unreferenced file is an orphan
MaxCommitRetries: 3,
}
result, _, err := handler.removeOrphans(context.Background(), client, setup.BucketName, setup.tablePath(), config)
if err != nil {
t.Fatalf("removeOrphans failed: %v", err)
}
if !strings.Contains(result, "removed 0 orphan") {
t.Errorf("expected every file to still be referenced, got %q", result)
}
if fs.getEntry(metaDir, "manifest-1.avro") == nil {
t.Error("manifest-1.avro (referenced manifest) should not have been deleted")
}
}
// dropManifestListFormatVersion rewrites a manifest list's Avro header the way
// DuckDB writes it, without a "format-version" entry. The key is renamed rather
// than removed so every Avro length prefix stays valid — iceberg-go ignores
// header entries it does not recognise.
func dropManifestListFormatVersion(t *testing.T, fs *fakeFilerServer, dir, name string) {
t.Helper()
entry := fs.getEntry(dir, name)
if entry == nil {
t.Fatalf("manifest list %s/%s not found", dir, name)
}
patched := bytes.Replace(entry.Content, []byte("format-version"), []byte("ignored-header"), 1)
if bytes.Equal(patched, entry.Content) {
t.Fatalf("manifest list %s carries no format-version to drop", name)
}
fs.putEntry(dir, name, &filer_pb.Entry{
Name: entry.Name,
Attributes: entry.Attributes,
Content: patched,
})
// Confirm the fixture: with the entry gone, iceberg-go infers v2 from
// the embedded writer schema (content / sequence_number are v2 fields).
manifests, err := iceberg.ReadManifestList(bytes.NewReader(patched))
if err != nil {
t.Fatalf("read patched manifest list: %v", err)
}
if got := manifests[0].Version(); got != 2 {
t.Fatalf("patched manifest list reports v%d, want v2", got)
}
}
func TestRewriteManifestsExecution(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
// Create a table with a single snapshot — we'll add extra small manifests
// to the manifest list so there's something to rewrite.
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", Summary: &table.Summary{
Operation: table.OpAppend,
Properties: map[string]string{
"total-data-files": "5",
"total-records": "5",
"total-files-size": "5120",
},
}},
},
}
meta := populateTable(t, fs, setup)
schema := meta.CurrentSchema()
spec := meta.PartitionSpec()
version := meta.Version()
// Build 5 small manifests and write them + a manifest list pointing to all of them
metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata")
var allManifests []iceberg.ManifestFile
for i := 1; i <= 5; i++ {
dfBuilder, err := iceberg.NewDataFileBuilder(
spec,
iceberg.EntryContentData,
fmt.Sprintf("data/rewrite-%d.parquet", i),
iceberg.ParquetFile,
map[int]any{},
nil, nil,
1,
1024,
)
if err != nil {
t.Fatalf("build data file %d: %v", i, err)
}
snapID := int64(1)
entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfBuilder.Build())
manifestName := fmt.Sprintf("small-manifest-%d.avro", i)
var buf bytes.Buffer
mf, err := iceberg.WriteManifest(path.Join("metadata", manifestName), &buf, version, spec, schema, 1, []iceberg.ManifestEntry{entry})
if err != nil {
t.Fatalf("write small manifest %d: %v", i, err)
}
fs.putEntry(metaDir, manifestName, &filer_pb.Entry{
Name: manifestName,
Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix()},
Content: buf.Bytes(),
})
allManifests = append(allManifests, mf)
}
// Overwrite the manifest list with all 5 manifests
var mlBuf bytes.Buffer
seqNum := int64(1)
if err := iceberg.WriteManifestList(version, &mlBuf, 1, nil, &seqNum, 0, allManifests); err != nil {
t.Fatalf("write manifest list: %v", err)
}
fs.putEntry(metaDir, "snap-1.avro", &filer_pb.Entry{
Name: "snap-1.avro",
Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix()},
Content: mlBuf.Bytes(),
})
handler := NewHandler(nil)
config := Config{
MinInputFiles: 3, // threshold to trigger rewrite (5 >= 3)
MaxCommitRetries: 3,
}
result, _, err := handler.rewriteManifests(context.Background(), client, setup.BucketName, setup.tablePath(), config)
if err != nil {
t.Fatalf("rewriteManifests failed: %v", err)
}
if !strings.Contains(result, "rewrote 5 manifests into 1") {
t.Errorf("expected '5 manifests into 1', got %q", result)
}
t.Logf("rewriteManifests result: %s", result)
// Verify a new metadata file and merged manifest were written
fs.mu.Lock()
creates := fs.createCalls
updates := fs.updateCalls
fs.mu.Unlock()
if creates < 3 {
// At minimum: merged manifest, manifest list, new metadata file
t.Errorf("expected at least 3 CreateEntry calls, got %d", creates)
}
if updates == 0 {
t.Error("expected at least one UpdateEntry call for xattr update")
}
// The spec requires absolute locations — strict readers (Spark/Trino via
// S3FileIO) reject scheme-less paths, so verify every written location.
wantPrefix := "s3://test-bucket/analytics/events/metadata/"
state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
if err != nil {
t.Fatalf("reload metadata: %v", err)
}
newSnap := state.Metadata.CurrentSnapshot()
if newSnap == nil || !strings.HasPrefix(newSnap.ManifestList, wantPrefix+"snap-") {
t.Fatalf("new snapshot manifest list should be absolute, got %+v", newSnap)
}
foundPreviousEntry := false
for mle := range state.Metadata.PreviousFiles() {
if mle.MetadataFile == wantPrefix+"v1.metadata.json" {
foundPreviousEntry = true
}
}
if !foundPreviousEntry {
t.Error("metadata-log should record the previous metadata file at its absolute location")
}
mlData, err := loadFileByIcebergPath(context.Background(), client, setup.BucketName, setup.tablePath(), newSnap.ManifestList)
if err != nil {
t.Fatalf("load new manifest list: %v", err)
}
newManifests, err := iceberg.ReadManifestList(bytes.NewReader(mlData))
if err != nil {
t.Fatalf("parse new manifest list: %v", err)
}
for _, mf := range newManifests {
if !strings.HasPrefix(mf.FilePath(), wantPrefix+"merged-") {
t.Errorf("merged manifest path should be absolute, got %q", mf.FilePath())
}
}
tableEntry := fs.getEntry(path.Join(s3tables.TablesPath, setup.BucketName, setup.Namespace), setup.TableName)
if tableEntry == nil {
t.Fatal("table entry missing")
}
var xattrMeta struct {
MetadataLocation string `json:"metadataLocation"`
}
if err := json.Unmarshal(tableEntry.Extended[s3tables.ExtendedKeyMetadata], &xattrMeta); err != nil {
t.Fatalf("unmarshal table xattr: %v", err)
}
if !strings.HasPrefix(xattrMeta.MetadataLocation, wantPrefix+"v2-") {
t.Errorf("xattr metadataLocation should be absolute, got %q", xattrMeta.MetadataLocation)
}
// Merging manifests moves no data, so the table's totals are unchanged.
if newSnap.Summary == nil {
t.Fatal("new snapshot has no summary")
}
for key, want := range map[string]string{"total-data-files": "5", "total-records": "5", "total-files-size": "5120"} {
if got := newSnap.Summary.Properties[key]; got != want {
t.Errorf("%s = %q, want %q", key, got, want)
}
}
}
func TestRewriteManifestsBelowThreshold(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "ns",
TableName: "tbl",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
},
}
populateTable(t, fs, setup)
handler := NewHandler(nil)
config := Config{
MinInputFiles: 10,
MinManifestsToRewrite: 10, // threshold higher than actual manifest count (1)
MaxCommitRetries: 3,
}
result, _, err := handler.rewriteManifests(context.Background(), client, setup.BucketName, setup.tablePath(), config)
if err != nil {
t.Fatalf("rewriteManifests failed: %v", err)
}
if !strings.Contains(result, "below threshold") {
t.Errorf("expected 'below threshold', got %q", result)
}
}
func TestFullExecuteFlow(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
},
}
populateTable(t, fs, setup)
// Add an orphan
metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata")
fs.putEntry(metaDir, "orphan.avro", &filer_pb.Entry{
Name: "orphan.avro",
Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Add(-200 * time.Hour).Unix()},
})
handler := NewHandler(nil)
// We need to build the request manually since Execute takes gRPC types
// but we're connecting directly
request := &plugin_pb.ExecuteJobRequest{
Job: &plugin_pb.JobSpec{
JobId: "test-job-1",
JobType: jobType,
Parameters: map[string]*plugin_pb.ConfigValue{
"bucket_name": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: setup.BucketName}},
"namespace": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: setup.Namespace}},
"table_name": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: setup.TableName}},
"table_path": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: setup.tablePath()}},
"filer_address": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: "not-used"}},
},
},
WorkerConfigValues: map[string]*plugin_pb.ConfigValue{
"snapshot_retention_hours": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
"max_snapshots_to_keep": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1}},
"orphan_older_than_hours": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 72}},
"max_commit_retries": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 3}},
"min_input_files": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 100}}, // high threshold to skip rewrite
"operations": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: "expire_snapshots,remove_orphans"}},
},
}
// Execute uses grpc.NewClient internally, but we need to pass our existing
// client. Call operations directly instead of full Execute to avoid the
// grpc.NewClient call which requires a real address.
workerConfig := ParseConfig(request.GetWorkerConfigValues())
ops, err := parseOperations(workerConfig.Operations)
if err != nil {
t.Fatalf("parseOperations: %v", err)
}
var results []string
for _, op := range ops {
var opResult string
var opErr error
switch op {
case "expire_snapshots":
opResult, _, opErr = handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), workerConfig)
case "remove_orphans":
opResult, _, opErr = handler.removeOrphans(context.Background(), client, setup.BucketName, setup.tablePath(), workerConfig)
case "rewrite_manifests":
opResult, _, opErr = handler.rewriteManifests(context.Background(), client, setup.BucketName, setup.tablePath(), workerConfig)
}
if opErr != nil {
t.Fatalf("operation %s failed: %v", op, opErr)
}
results = append(results, fmt.Sprintf("%s: %s", op, opResult))
}
t.Logf("Full execution results: %s", strings.Join(results, "; "))
// Verify snapshots were expired
if !strings.Contains(results[0], "expired") {
t.Errorf("expected snapshot expiration, got %q", results[0])
}
// Verify orphan was removed
if !strings.Contains(results[1], "removed") {
t.Errorf("expected orphan removal, got %q", results[1])
}
if fs.getEntry(metaDir, "orphan.avro") != nil {
t.Error("orphan.avro should have been deleted")
}
}
func TestDetectWithFakeFiler(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
},
Age: 200 * time.Hour, // past the default 7-day retention
}
populateTable(t, fs, setup)
handler := NewHandler(nil)
config := Config{
SnapshotRetentionMs: hoursToMs(0), // everything is expired
MaxSnapshotsToKeep: 2, // 3 > 2, needs maintenance
MaxCommitRetries: 3,
}
tables, err := handler.scanTablesForMaintenance(
context.Background(),
client,
config,
"", "", "", // no filters
0, // no limit
)
if err != nil {
t.Fatalf("scanTablesForMaintenance failed: %v", err)
}
if len(tables) != 1 {
t.Fatalf("expected 1 table needing maintenance, got %d", len(tables))
}
if tables[0].BucketName != setup.BucketName {
t.Errorf("expected bucket %q, got %q", setup.BucketName, tables[0].BucketName)
}
if tables[0].TableName != setup.TableName {
t.Errorf("expected table %q, got %q", setup.TableName, tables[0].TableName)
}
}
func TestDetectWithFilters(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
// Create two tables in different buckets
setup1 := tableSetup{
BucketName: "bucket-a",
Namespace: "ns",
TableName: "table1",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
},
Age: 200 * time.Hour,
}
setup2 := tableSetup{
BucketName: "bucket-b",
Namespace: "ns",
TableName: "table2",
Snapshots: []table.Snapshot{
{SnapshotID: 4, TimestampMs: now + 3, ManifestList: "metadata/snap-4.avro"},
{SnapshotID: 5, TimestampMs: now + 4, ManifestList: "metadata/snap-5.avro"},
{SnapshotID: 6, TimestampMs: now + 5, ManifestList: "metadata/snap-6.avro"},
},
Age: 200 * time.Hour,
}
populateTable(t, fs, setup1)
populateTable(t, fs, setup2)
handler := NewHandler(nil)
config := Config{
SnapshotRetentionMs: hoursToMs(0),
MaxSnapshotsToKeep: 2,
MaxCommitRetries: 3,
}
// Without filter: should find both
tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
if err != nil {
t.Fatalf("scan failed: %v", err)
}
if len(tables) != 2 {
t.Fatalf("expected 2 tables without filter, got %d", len(tables))
}
// With bucket filter: should find only one
tables, err = handler.scanTablesForMaintenance(context.Background(), client, config, "bucket-a", "", "", 0)
if err != nil {
t.Fatalf("scan with filter failed: %v", err)
}
if len(tables) != 1 {
t.Fatalf("expected 1 table with bucket filter, got %d", len(tables))
}
if tables[0].BucketName != "bucket-a" {
t.Errorf("expected bucket-a, got %q", tables[0].BucketName)
}
}
func TestConnectToFilerSkipsUnreachableAddresses(t *testing.T) {
handler := NewHandler(grpc.WithTransportCredentials(insecure.NewCredentials()))
_, _, liveGrpcAddr := startFakeFilerWithAddress(t)
deadListener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen for dead address: %v", err)
}
deadGrpcAddr := deadListener.Addr().String()
_ = deadListener.Close()
grpcAddr, conn, err := handler.connectToFiler(context.Background(), []string{deadGrpcAddr, liveGrpcAddr})
if err != nil {
t.Fatalf("connectToFiler failed: %v", err)
}
defer conn.Close()
if grpcAddr != liveGrpcAddr {
t.Fatalf("expected live address %q, got %q", liveGrpcAddr, grpcAddr)
}
}
func TestConnectToFilerFailsWhenAllAddressesAreUnreachable(t *testing.T) {
handler := NewHandler(grpc.WithTransportCredentials(insecure.NewCredentials()))
deadListener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen for dead address: %v", err)
}
deadGrpcAddr := deadListener.Addr().String()
_ = deadListener.Close()
_, _, err = handler.connectToFiler(context.Background(), []string{deadGrpcAddr})
if err == nil {
t.Fatal("expected connectToFiler to fail")
}
}
func TestDetectSchedulesCompactionWithoutSnapshotPressure(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
},
}
meta := populateTable(t, fs, setup)
writeCurrentSnapshotManifests(t, fs, setup, meta, [][]iceberg.ManifestEntry{
makeManifestEntries(t, []testEntrySpec{
{path: "data/small-1.parquet", size: 1024, partition: map[int]any{}},
{path: "data/small-2.parquet", size: 1024, partition: map[int]any{}},
{path: "data/small-3.parquet", size: 1024, partition: map[int]any{}},
}, 1),
})
handler := NewHandler(nil)
config := Config{
SnapshotRetentionMs: hoursToMs(24 * 365),
MaxSnapshotsToKeep: 10,
TargetFileSizeBytes: 4096,
MinInputFiles: 2,
Operations: "compact",
}
tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
if err != nil {
t.Fatalf("scanTablesForMaintenance failed: %v", err)
}
if len(tables) != 1 {
t.Fatalf("expected 1 compaction candidate, got %d", len(tables))
}
}
func TestDetectSchedulesCompactionWithDeleteManifestPresent(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
},
}
meta := populateTable(t, fs, setup)
currentSnap := meta.CurrentSnapshot()
if currentSnap == nil {
t.Fatal("current snapshot is required")
}
metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata")
version := meta.Version()
schema := meta.CurrentSchema()
spec := meta.PartitionSpec()
dataEntries := makeManifestEntries(t, []testEntrySpec{
{path: "data/small-1.parquet", size: 1024, partition: map[int]any{}},
{path: "data/small-2.parquet", size: 1024, partition: map[int]any{}},
{path: "data/small-3.parquet", size: 1024, partition: map[int]any{}},
}, currentSnap.SnapshotID)
var dataManifestBuf bytes.Buffer
dataManifestName := "detect-manifest-1.avro"
dataManifest, err := iceberg.WriteManifest(
path.Join("metadata", dataManifestName),
&dataManifestBuf,
version,
spec,
schema,
currentSnap.SnapshotID,
dataEntries,
)
if err != nil {
t.Fatalf("write data manifest: %v", err)
}
fs.putEntry(metaDir, dataManifestName, &filer_pb.Entry{
Name: dataManifestName,
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Unix(),
FileSize: uint64(dataManifestBuf.Len()),
},
Content: dataManifestBuf.Bytes(),
})
deleteManifest := iceberg.NewManifestFile(
version,
path.Join("metadata", "detect-delete-manifest.avro"),
0,
int32(spec.ID()),
currentSnap.SnapshotID,
).Content(iceberg.ManifestContentDeletes).
SequenceNum(currentSnap.SequenceNumber, currentSnap.SequenceNumber).
DeletedFiles(1).
DeletedRows(1).
Build()
var manifestListBuf bytes.Buffer
seqNum := currentSnap.SequenceNumber
if err := iceberg.WriteManifestList(
version,
&manifestListBuf,
currentSnap.SnapshotID,
currentSnap.ParentSnapshotID,
&seqNum,
0,
[]iceberg.ManifestFile{dataManifest, deleteManifest},
); err != nil {
t.Fatalf("write manifest list: %v", err)
}
fs.putEntry(metaDir, path.Base(currentSnap.ManifestList), &filer_pb.Entry{
Name: path.Base(currentSnap.ManifestList),
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Unix(),
FileSize: uint64(manifestListBuf.Len()),
},
Content: manifestListBuf.Bytes(),
})
handler := NewHandler(nil)
config := Config{
SnapshotRetentionMs: hoursToMs(24 * 365),
MaxSnapshotsToKeep: 10,
TargetFileSizeBytes: 4096,
MinInputFiles: 2,
Operations: "compact",
}
tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
if err != nil {
t.Fatalf("scanTablesForMaintenance failed: %v", err)
}
if len(tables) != 1 {
t.Fatalf("expected 1 compaction candidate with delete manifest present, got %d", len(tables))
}
}
func TestDetectSchedulesSnapshotExpiryDespiteCompactionEvaluationError(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro", SequenceNumber: 2},
},
Age: 400 * 24 * time.Hour, // past the year-long retention below
}
populateTable(t, fs, setup)
// Corrupt manifest lists so compaction evaluation fails.
metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata")
for _, snap := range setup.Snapshots {
manifestListName := path.Base(snap.ManifestList)
fs.putEntry(metaDir, manifestListName, &filer_pb.Entry{
Name: manifestListName,
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Unix(),
FileSize: uint64(len("not-a-manifest-list")),
},
Content: []byte("not-a-manifest-list"),
})
}
handler := NewHandler(nil)
config := Config{
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)
if err != nil {
t.Fatalf("scanTablesForMaintenance failed: %v", err)
}
if len(tables) != 1 {
t.Fatalf("expected snapshot expiration candidate despite compaction evaluation error, got %d", len(tables))
}
}
func TestDetectSchedulesManifestRewriteWithoutSnapshotPressure(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
},
}
meta := populateTable(t, fs, setup)
manifestEntries := make([][]iceberg.ManifestEntry, 0, 5)
for i := 0; i < 5; i++ {
manifestEntries = append(manifestEntries, makeManifestEntries(t, []testEntrySpec{
{path: fmt.Sprintf("data/rewrite-%d.parquet", i), size: 1024, partition: map[int]any{}},
}, 1))
}
writeCurrentSnapshotManifests(t, fs, setup, meta, manifestEntries)
handler := NewHandler(nil)
config := Config{
SnapshotRetentionMs: hoursToMs(24 * 365),
MaxSnapshotsToKeep: 10,
MinManifestsToRewrite: 5,
Operations: "rewrite_manifests",
}
tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
if err != nil {
t.Fatalf("scanTablesForMaintenance failed: %v", err)
}
if len(tables) != 1 {
t.Fatalf("expected 1 manifest rewrite candidate, got %d", len(tables))
}
}
func TestDetectUsesPlanningIndexForRepeatedCompactionScans(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
},
}
meta := populateTable(t, fs, setup)
writeCurrentSnapshotManifests(t, fs, setup, meta, [][]iceberg.ManifestEntry{
makeManifestEntries(t, []testEntrySpec{
{path: "data/small-1.parquet", size: 1024, partition: map[int]any{}},
{path: "data/small-2.parquet", size: 1024, partition: map[int]any{}},
{path: "data/small-3.parquet", size: 1024, partition: map[int]any{}},
}, 1),
})
handler := NewHandler(nil)
config := Config{
TargetFileSizeBytes: 4096,
MinInputFiles: 2,
Operations: "compact",
}
tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
if err != nil {
t.Fatalf("scanTablesForMaintenance failed: %v", err)
}
if len(tables) != 1 {
t.Fatalf("expected 1 compaction candidate, got %d", len(tables))
}
tableDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.Namespace)
tableEntry := fs.getEntry(tableDir, setup.TableName)
if tableEntry == nil {
t.Fatal("table entry not found")
}
var envelope struct {
PlanningIndex *planningIndex `json:"planningIndex,omitempty"`
}
if err := json.Unmarshal(tableEntry.Extended[s3tables.ExtendedKeyMetadata], &envelope); err != nil {
t.Fatalf("parse table metadata xattr: %v", err)
}
if envelope.PlanningIndex == nil || envelope.PlanningIndex.Compaction == nil {
t.Fatal("expected persisted compaction planning index after first scan")
}
metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata")
fs.putEntry(metaDir, "snap-1.avro", &filer_pb.Entry{
Name: "snap-1.avro",
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Unix(),
FileSize: uint64(len("broken")),
},
Content: []byte("broken"),
})
tables, err = handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
if err != nil {
t.Fatalf("scanTablesForMaintenance with cached planning index failed: %v", err)
}
if len(tables) != 1 {
t.Fatalf("expected cached planning index to preserve 1 compaction candidate, got %d", len(tables))
}
}
func TestDetectInvalidatesPlanningIndexWhenCompactionConfigChanges(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
},
}
meta := populateTable(t, fs, setup)
writeCurrentSnapshotManifests(t, fs, setup, meta, [][]iceberg.ManifestEntry{
makeManifestEntries(t, []testEntrySpec{
{path: "data/small-1.parquet", size: 1024, partition: map[int]any{}},
{path: "data/small-2.parquet", size: 1024, partition: map[int]any{}},
}, 1),
})
handler := NewHandler(nil)
initialConfig := Config{
TargetFileSizeBytes: 4096,
MinInputFiles: 3,
Operations: "compact",
}
tables, err := handler.scanTablesForMaintenance(context.Background(), client, initialConfig, "", "", "", 0)
if err != nil {
t.Fatalf("initial scanTablesForMaintenance failed: %v", err)
}
if len(tables) != 0 {
t.Fatalf("expected no compaction candidates with min_input_files=3, got %d", len(tables))
}
updatedConfig := Config{
TargetFileSizeBytes: 4096,
MinInputFiles: 2,
Operations: "compact",
}
tables, err = handler.scanTablesForMaintenance(context.Background(), client, updatedConfig, "", "", "", 0)
if err != nil {
t.Fatalf("updated scanTablesForMaintenance failed: %v", err)
}
if len(tables) != 1 {
t.Fatalf("expected planning index invalidation to yield 1 compaction candidate, got %d", len(tables))
}
}
func TestDetectPlanningIndexPreservesUnscannedSections(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
},
}
meta := populateTable(t, fs, setup)
writeCurrentSnapshotManifests(t, fs, setup, meta, [][]iceberg.ManifestEntry{
makeManifestEntries(t, []testEntrySpec{
{path: "data/small-1.parquet", size: 1024, partition: map[int]any{}},
{path: "data/small-2.parquet", size: 1024, partition: map[int]any{}},
}, 1),
})
handler := NewHandler(nil)
compactConfig := Config{
TargetFileSizeBytes: 4096,
MinInputFiles: 2,
Operations: "compact",
}
if _, err := handler.scanTablesForMaintenance(context.Background(), client, compactConfig, "", "", "", 0); err != nil {
t.Fatalf("compact scanTablesForMaintenance failed: %v", err)
}
rewriteConfig := Config{
MinManifestsToRewrite: 5,
Operations: "rewrite_manifests",
}
if _, err := handler.scanTablesForMaintenance(context.Background(), client, rewriteConfig, "", "", "", 0); err != nil {
t.Fatalf("rewrite scanTablesForMaintenance failed: %v", err)
}
tableDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.Namespace)
tableEntry := fs.getEntry(tableDir, setup.TableName)
if tableEntry == nil {
t.Fatal("table entry not found")
}
var envelope struct {
PlanningIndex *planningIndex `json:"planningIndex,omitempty"`
}
if err := json.Unmarshal(tableEntry.Extended[s3tables.ExtendedKeyMetadata], &envelope); err != nil {
t.Fatalf("parse table metadata xattr: %v", err)
}
if envelope.PlanningIndex == nil {
t.Fatal("expected persisted planning index")
}
if envelope.PlanningIndex.Compaction == nil {
t.Fatal("expected compaction section to be preserved")
}
if envelope.PlanningIndex.RewriteManifests == nil {
t.Fatal("expected rewrite_manifests section to be added")
}
}
func TestTableNeedsMaintenanceCachesPlanningIndexBuildError(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
},
}
meta := populateTable(t, fs, setup)
writeCurrentSnapshotManifests(t, fs, setup, meta, [][]iceberg.ManifestEntry{
makeManifestEntries(t, []testEntrySpec{
{path: "data/small-1.parquet", size: 1024, partition: map[int]any{}},
{path: "data/small-2.parquet", size: 1024, partition: map[int]any{}},
}, 1),
})
metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata")
fs.putEntry(metaDir, "snap-1.avro", &filer_pb.Entry{
Name: "snap-1.avro",
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Unix(),
FileSize: uint64(len("broken")),
},
Content: []byte("broken"),
})
handler := NewHandler(nil)
config := Config{
TargetFileSizeBytes: 4096,
MinInputFiles: 2,
MinManifestsToRewrite: 2,
Operations: "compact,rewrite_manifests",
}
ops, err := parseOperations(config.Operations)
if err != nil {
t.Fatalf("parseOperations: %v", err)
}
needsWork, err := handler.tableNeedsMaintenance(context.Background(), client, setup.BucketName, setup.tablePath(), &tableState{Metadata: meta, MetadataFileName: "v1.metadata.json", DataPath: setup.dataPath()}, config, ops)
if err == nil {
t.Fatal("expected planning-index build error")
}
if needsWork {
t.Fatal("expected no maintenance result on planning-index build error")
}
if strings.Count(err.Error(), "parse manifest list") != 1 {
t.Fatalf("expected planning-index build error to be reported once, got %q", err)
}
}
func TestTableNeedsMaintenanceScopesPlanningIndexBuildErrorsPerOperation(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
},
}
meta := populateTable(t, fs, setup)
writeCurrentSnapshotManifests(t, fs, setup, meta, [][]iceberg.ManifestEntry{
makeManifestEntries(t, []testEntrySpec{
{path: "data/small-1.parquet", size: 1024, partition: map[int]any{}},
}, 1),
makeManifestEntries(t, []testEntrySpec{
{path: "data/small-2.parquet", size: 1024, partition: map[int]any{}},
}, 1),
})
metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata")
fs.putEntry(metaDir, "detect-manifest-1.avro", &filer_pb.Entry{
Name: "detect-manifest-1.avro",
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Unix(),
FileSize: uint64(len("broken")),
},
Content: []byte("broken"),
})
handler := NewHandler(nil)
config := Config{
TargetFileSizeBytes: 4096,
MinInputFiles: 2,
MinManifestsToRewrite: 2,
Operations: "compact,rewrite_manifests",
}
ops, err := parseOperations(config.Operations)
if err != nil {
t.Fatalf("parseOperations: %v", err)
}
needsWork, err := handler.tableNeedsMaintenance(context.Background(), client, setup.BucketName, setup.tablePath(), &tableState{Metadata: meta, MetadataFileName: "v1.metadata.json", DataPath: setup.dataPath()}, config, ops)
if err != nil {
t.Fatalf("expected rewrite_manifests planning to survive compaction planning error, got %v", err)
}
if !needsWork {
t.Fatal("expected rewrite_manifests maintenance despite compaction planning error")
}
}
func TestPersistPlanningIndexUsesMetadataXattrCASGuard(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
},
}
populateTable(t, fs, setup)
tableDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath())
tableEntry := fs.getEntry(path.Dir(tableDir), path.Base(tableDir))
if tableEntry == nil {
t.Fatal("table entry not found")
}
observedExpectedExtended := make(chan map[string][]byte, 1)
fs.beforeUpdate = func(_ *fakeFilerServer, req *filer_pb.UpdateEntryRequest) error {
cloned := make(map[string][]byte, len(req.ExpectedExtended))
for key, value := range req.ExpectedExtended {
cloned[key] = append([]byte(nil), value...)
}
observedExpectedExtended <- cloned
return nil
}
err := persistPlanningIndex(context.Background(), client, setup.BucketName, setup.tablePath(), &planningIndex{
SnapshotID: 1,
ManifestList: "metadata/snap-1.avro",
UpdatedAtMs: time.Now().UnixMilli(),
})
if err != nil {
t.Fatalf("persistPlanningIndex: %v", err)
}
var expectedExtended map[string][]byte
select {
case expectedExtended = <-observedExpectedExtended:
default:
t.Fatal("expected persistPlanningIndex to issue an UpdateEntry request")
}
if got := expectedExtended[s3tables.ExtendedKeyMetadata]; !bytes.Equal(got, tableEntry.Extended[s3tables.ExtendedKeyMetadata]) {
t.Fatal("expected metadata xattr to be included in ExpectedExtended")
}
if got := expectedExtended[s3tables.ExtendedKeyMetadataVersion]; !bytes.Equal(got, tableEntry.Extended[s3tables.ExtendedKeyMetadataVersion]) {
t.Fatal("expected metadata version xattr to be preserved in ExpectedExtended")
}
}
func TestDetectDoesNotScheduleManifestRewriteFromDeleteManifestsOnly(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
},
}
meta := populateTable(t, fs, setup)
currentSnap := meta.CurrentSnapshot()
if currentSnap == nil {
t.Fatal("current snapshot is required")
}
metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata")
version := meta.Version()
schema := meta.CurrentSchema()
spec := meta.PartitionSpec()
dataEntries := makeManifestEntries(t, []testEntrySpec{
{path: "data/rewrite-0.parquet", size: 1024, partition: map[int]any{}},
}, currentSnap.SnapshotID)
var dataManifestBuf bytes.Buffer
dataManifestName := "detect-rewrite-data.avro"
dataManifest, err := iceberg.WriteManifest(
path.Join("metadata", dataManifestName),
&dataManifestBuf,
version,
spec,
schema,
currentSnap.SnapshotID,
dataEntries,
)
if err != nil {
t.Fatalf("write data manifest: %v", err)
}
fs.putEntry(metaDir, dataManifestName, &filer_pb.Entry{
Name: dataManifestName,
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Unix(),
FileSize: uint64(dataManifestBuf.Len()),
},
Content: dataManifestBuf.Bytes(),
})
manifests := []iceberg.ManifestFile{dataManifest}
for i := 0; i < 4; i++ {
deleteManifest := iceberg.NewManifestFile(
version,
path.Join("metadata", fmt.Sprintf("detect-delete-%d.avro", i)),
0,
int32(spec.ID()),
currentSnap.SnapshotID,
).Content(iceberg.ManifestContentDeletes).
SequenceNum(currentSnap.SequenceNumber, currentSnap.SequenceNumber).
DeletedFiles(1).
DeletedRows(1).
Build()
manifests = append(manifests, deleteManifest)
}
var manifestListBuf bytes.Buffer
seqNum := currentSnap.SequenceNumber
if err := iceberg.WriteManifestList(
version,
&manifestListBuf,
currentSnap.SnapshotID,
currentSnap.ParentSnapshotID,
&seqNum,
0,
manifests,
); err != nil {
t.Fatalf("write manifest list: %v", err)
}
fs.putEntry(metaDir, path.Base(currentSnap.ManifestList), &filer_pb.Entry{
Name: path.Base(currentSnap.ManifestList),
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Unix(),
FileSize: uint64(manifestListBuf.Len()),
},
Content: manifestListBuf.Bytes(),
})
handler := NewHandler(nil)
config := Config{
SnapshotRetentionMs: hoursToMs(24 * 365),
MaxSnapshotsToKeep: 10,
MinManifestsToRewrite: 2,
Operations: "rewrite_manifests",
}
tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
if err != nil {
t.Fatalf("scanTablesForMaintenance failed: %v", err)
}
if len(tables) != 0 {
t.Fatalf("expected no manifest rewrite candidate when only one data manifest exists, got %d", len(tables))
}
}
func TestDetectSchedulesOrphanCleanupWithoutSnapshotPressure(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
},
}
populateTable(t, fs, setup)
dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data")
fs.putEntry(dataDir, "stale-orphan.parquet", &filer_pb.Entry{
Name: "stale-orphan.parquet",
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Add(-200 * time.Hour).Unix(),
FileSize: 100,
},
Content: []byte("orphan"),
})
handler := NewHandler(nil)
config := Config{
SnapshotRetentionMs: hoursToMs(24 * 365),
MaxSnapshotsToKeep: 10,
OrphanOlderThanHours: 72,
Operations: "remove_orphans",
}
tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
if err != nil {
t.Fatalf("scanTablesForMaintenance failed: %v", err)
}
if len(tables) != 1 {
t.Fatalf("expected 1 orphan cleanup candidate, got %d", len(tables))
}
}
func TestDetectSchedulesOrphanCleanupWithoutSnapshots(t *testing.T) {
fs, client := startFakeFiler(t)
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "analytics",
TableName: "events",
}
populateTable(t, fs, setup)
dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data")
fs.putEntry(dataDir, "stale-orphan.parquet", &filer_pb.Entry{
Name: "stale-orphan.parquet",
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Add(-200 * time.Hour).Unix(),
FileSize: 100,
},
Content: []byte("orphan"),
})
handler := NewHandler(nil)
config := Config{
OrphanOlderThanHours: 72,
Operations: "remove_orphans",
}
tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
if err != nil {
t.Fatalf("scanTablesForMaintenance failed: %v", err)
}
if len(tables) != 1 {
t.Fatalf("expected 1 orphan cleanup candidate without snapshots, got %d", len(tables))
}
}
func TestStalePlanGuard(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "ns",
TableName: "tbl",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
},
}
populateTable(t, fs, setup)
handler := NewHandler(nil)
// Call commitWithRetry with a stale plan that expects a different snapshot
config := Config{MaxCommitRetries: 1}
staleSnapshotID := int64(999)
err := handler.commitWithRetry(context.Background(), client, setup.BucketName, setup.tablePath(), "v1.metadata.json", config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error {
cs := currentMeta.CurrentSnapshot()
if cs == nil || cs.SnapshotID != staleSnapshotID {
return errStalePlan
}
return nil
})
if err == nil {
t.Fatal("expected stale plan error")
}
if !strings.Contains(err.Error(), "stale plan") {
t.Errorf("expected stale plan in error, got %q", err)
}
}
func TestMetadataVersionCAS(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "ns",
TableName: "tbl",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
},
}
populateTable(t, fs, setup)
// The table xattr has metadataVersion=1. Try updating with wrong expected version.
tableDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath())
err := updateTableMetadataXattr(context.Background(), client, tableDir, 99, []byte(`{}`), "metadata/v100.metadata.json")
if err == nil {
t.Fatal("expected version conflict error")
}
if !strings.Contains(err.Error(), "metadata version conflict") {
t.Errorf("expected version conflict in error, got %q", err)
}
// Correct expected version should succeed
err = updateTableMetadataXattr(context.Background(), client, tableDir, 1, []byte(`{}`), "metadata/v2.metadata.json")
if err != nil {
t.Fatalf("expected success with correct version, got: %v", err)
}
// Verify version was incremented to 2
entry := fs.getEntry(path.Dir(tableDir), path.Base(tableDir))
if entry == nil {
t.Fatal("table entry not found after update")
}
var internalMeta map[string]json.RawMessage
if err := json.Unmarshal(entry.Extended[s3tables.ExtendedKeyMetadata], &internalMeta); err != nil {
t.Fatalf("unmarshal xattr: %v", err)
}
var version int
if err := json.Unmarshal(internalMeta["metadataVersion"], &version); err != nil {
t.Fatalf("unmarshal version: %v", err)
}
if version != 2 {
t.Errorf("expected version 2 after update, got %d", version)
}
}
func TestMetadataVersionCASDetectsConcurrentUpdate(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "ns",
TableName: "tbl",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
},
}
populateTable(t, fs, setup)
tableDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath())
fs.beforeUpdate = func(f *fakeFilerServer, req *filer_pb.UpdateEntryRequest) error {
entry := f.getEntry(path.Dir(tableDir), path.Base(tableDir))
if entry == nil {
return fmt.Errorf("table entry not found before concurrent update")
}
updatedEntry := cloneEntryForTest(t, entry)
var internalMeta map[string]json.RawMessage
if err := json.Unmarshal(updatedEntry.Extended[s3tables.ExtendedKeyMetadata], &internalMeta); err != nil {
return fmt.Errorf("unmarshal xattr: %w", err)
}
versionJSON, err := json.Marshal(2)
if err != nil {
return fmt.Errorf("marshal version: %w", err)
}
internalMeta["metadataVersion"] = versionJSON
updatedXattr, err := json.Marshal(internalMeta)
if err != nil {
return fmt.Errorf("marshal xattr: %w", err)
}
updatedEntry.Extended[s3tables.ExtendedKeyMetadata] = updatedXattr
updatedEntry.Extended[s3tables.ExtendedKeyMetadataVersion] = metadataVersionXattr(2)
f.putEntry(path.Dir(tableDir), path.Base(tableDir), updatedEntry)
return nil
}
err := updateTableMetadataXattr(context.Background(), client, tableDir, 1, []byte(`{}`), "metadata/v2.metadata.json")
if err == nil {
t.Fatal("expected version conflict error")
}
if !strings.Contains(err.Error(), "metadata version conflict") {
t.Fatalf("expected metadata version conflict, got %q", err)
}
}
func cloneEntryForTest(t *testing.T, entry *filer_pb.Entry) *filer_pb.Entry {
t.Helper()
cloned, ok := proto.Clone(entry).(*filer_pb.Entry)
if !ok {
t.Fatal("clone entry: unexpected type")
}
return cloned
}
// ---------------------------------------------------------------------------
// Avro manifest content patching for tests
// ---------------------------------------------------------------------------
// patchManifestContentToDeletes performs a binary patch on an Avro manifest
// file to change the "content" metadata value from "data" to "deletes".
// This workaround is needed because iceberg-go's WriteManifest API always
// sets content="data" and provides no way to create delete manifests.
// The function validates the pattern was found (bytes.Equal check) and fails
// fast if not, so breakage from encoding changes is caught immediately.
//
// In Avro OCF encoding, strings are stored as zigzag-encoded length + bytes.
// "content" (7 chars) = \x0e + "content", "data" (4 chars) = \x08 + "data",
// "deletes" (7 chars) = \x0e + "deletes".
func patchManifestContentToDeletes(t *testing.T, manifestBytes []byte) []byte {
t.Helper()
// Pattern: zigzag(7)="content" zigzag(4)="data"
old := append([]byte{0x0e}, []byte("content")...)
old = append(old, 0x08)
old = append(old, []byte("data")...)
// Replacement: zigzag(7)="content" zigzag(7)="deletes"
new := append([]byte{0x0e}, []byte("content")...)
new = append(new, 0x0e)
new = append(new, []byte("deletes")...)
result := bytes.Replace(manifestBytes, old, new, 1)
if bytes.Equal(result, manifestBytes) {
t.Fatal("patchManifestContentToDeletes: pattern not found in manifest bytes")
}
return result
}
// ---------------------------------------------------------------------------
// End-to-end compaction tests with deletes
// ---------------------------------------------------------------------------
// writeTestParquetFile creates a Parquet file with id/name columns in the fake filer.
func writeTestParquetFile(t *testing.T, fs *fakeFilerServer, dir, name string, rows []struct {
ID int64
Name string
}) []byte {
t.Helper()
type dataRow struct {
ID int64 `parquet:"id"`
Name string `parquet:"name"`
}
var buf bytes.Buffer
w := parquet.NewWriter(&buf, parquet.SchemaOf(new(dataRow)))
for _, r := range rows {
if err := w.Write(&dataRow{r.ID, r.Name}); err != nil {
t.Fatalf("write parquet row: %v", err)
}
}
if err := w.Close(); err != nil {
t.Fatalf("close parquet writer: %v", err)
}
data := buf.Bytes()
fs.putEntry(dir, name, &filer_pb.Entry{
Name: name,
Content: data,
Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(len(data))},
})
return data
}
// populateTableWithDeleteFiles sets up a table with data files and delete manifest(s)
// for compaction testing. Returns the table metadata.
func populateTableWithDeleteFiles(
t *testing.T,
fs *fakeFilerServer,
setup tableSetup,
dataFiles []struct {
Name string
Rows []struct {
ID int64
Name string
}
},
posDeleteFiles []struct {
Name string
Rows []struct {
FilePath string
Pos int64
}
},
eqDeleteFiles []struct {
Name string
FieldIDs []int
Rows []struct {
ID int64
Name string
}
},
) table.Metadata {
return populateTableWithDeleteFilesAndSortOrder(t, fs, setup, dataFiles, posDeleteFiles, eqDeleteFiles, table.UnsortedSortOrder)
}
func populateTableWithDeleteFilesAndSortOrder(
t *testing.T,
fs *fakeFilerServer,
setup tableSetup,
dataFiles []struct {
Name string
Rows []struct {
ID int64
Name string
}
},
posDeleteFiles []struct {
Name string
Rows []struct {
FilePath string
Pos int64
}
},
eqDeleteFiles []struct {
Name string
FieldIDs []int
Rows []struct {
ID int64
Name string
}
},
sortOrder table.SortOrder,
) table.Metadata {
t.Helper()
schema := newTestSchema()
spec := *iceberg.UnpartitionedSpec
meta, err := table.NewMetadata(schema, &spec, sortOrder, "s3://"+setup.BucketName+"/"+setup.dataPath(), nil)
if err != nil {
t.Fatalf("create metadata: %v", err)
}
bucketsPath := s3tables.TablesPath
bucketPath := path.Join(bucketsPath, setup.BucketName)
nsPath := path.Join(bucketPath, setup.Namespace)
tableFilerPath := path.Join(bucketPath, setup.dataPath())
metaDir := path.Join(tableFilerPath, "metadata")
dataDir := path.Join(tableFilerPath, "data")
version := meta.Version()
// Write data files
var dataManifestEntries []iceberg.ManifestEntry
for _, df := range dataFiles {
data := writeTestParquetFile(t, fs, dataDir, df.Name, df.Rows)
dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, setup.fileRef("data", df.Name), iceberg.ParquetFile, map[int]any{}, nil, nil, int64(len(df.Rows)), int64(len(data)))
if err != nil {
t.Fatalf("build data file %s: %v", df.Name, err)
}
snapID := int64(1)
dataManifestEntries = append(dataManifestEntries, iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build()))
}
// Write data manifest
var dataManifestBuf bytes.Buffer
dataManifestName := "data-manifest-1.avro"
dataMf, err := iceberg.WriteManifest(setup.fileRef("metadata", dataManifestName), &dataManifestBuf, version, spec, schema, 1, dataManifestEntries)
if err != nil {
t.Fatalf("write data manifest: %v", err)
}
fs.putEntry(metaDir, dataManifestName, &filer_pb.Entry{
Name: dataManifestName, Content: dataManifestBuf.Bytes(),
Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(dataManifestBuf.Len())},
})
allManifests := []iceberg.ManifestFile{dataMf}
// Write position delete files and manifests
if len(posDeleteFiles) > 0 {
var posDeleteEntries []iceberg.ManifestEntry
for _, pdf := range posDeleteFiles {
type posRow struct {
FilePath string `parquet:"file_path"`
Pos int64 `parquet:"pos"`
}
var buf bytes.Buffer
w := parquet.NewWriter(&buf, parquet.SchemaOf(new(posRow)))
for _, r := range pdf.Rows {
if err := w.Write(&posRow{r.FilePath, r.Pos}); err != nil {
t.Fatalf("write pos delete: %v", err)
}
}
if err := w.Close(); err != nil {
t.Fatalf("close pos delete: %v", err)
}
fs.putEntry(dataDir, pdf.Name, &filer_pb.Entry{
Name: pdf.Name, Content: buf.Bytes(),
Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(buf.Len())},
})
dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentPosDeletes, setup.fileRef("data", pdf.Name), iceberg.ParquetFile, map[int]any{}, nil, nil, int64(len(pdf.Rows)), int64(buf.Len()))
if err != nil {
t.Fatalf("build pos delete file: %v", err)
}
snapID := int64(1)
posDeleteEntries = append(posDeleteEntries, iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build()))
}
// WriteManifest always sets content="data", so we patch the Avro
// metadata to "deletes" and build a ManifestFile with the right content type.
var posManifestBuf bytes.Buffer
posManifestName := "pos-delete-manifest-1.avro"
posManifestPath := setup.fileRef("metadata", posManifestName)
_, err := iceberg.WriteManifest(posManifestPath, &posManifestBuf, version, spec, schema, 1, posDeleteEntries)
if err != nil {
t.Fatalf("write pos delete manifest: %v", err)
}
patchedBytes := patchManifestContentToDeletes(t, posManifestBuf.Bytes())
fs.putEntry(metaDir, posManifestName, &filer_pb.Entry{
Name: posManifestName, Content: patchedBytes,
Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(len(patchedBytes))},
})
posMf := iceberg.NewManifestFile(version, posManifestPath, int64(len(patchedBytes)), int32(spec.ID()), 1).
Content(iceberg.ManifestContentDeletes).
AddedFiles(int32(len(posDeleteEntries))).
AddedRows(int64(len(posDeleteFiles[0].Rows))).
Build()
allManifests = append(allManifests, posMf)
}
// Write equality delete files and manifests
if len(eqDeleteFiles) > 0 {
var eqDeleteEntries []iceberg.ManifestEntry
for _, edf := range eqDeleteFiles {
type eqRow struct {
ID int64 `parquet:"id"`
Name string `parquet:"name"`
}
var buf bytes.Buffer
w := parquet.NewWriter(&buf, parquet.SchemaOf(new(eqRow)))
for _, r := range edf.Rows {
if err := w.Write(&eqRow{r.ID, r.Name}); err != nil {
t.Fatalf("write eq delete: %v", err)
}
}
if err := w.Close(); err != nil {
t.Fatalf("close eq delete: %v", err)
}
fs.putEntry(dataDir, edf.Name, &filer_pb.Entry{
Name: edf.Name, Content: buf.Bytes(),
Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(buf.Len())},
})
dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentEqDeletes, setup.fileRef("data", edf.Name), iceberg.ParquetFile, map[int]any{}, nil, nil, int64(len(edf.Rows)), int64(buf.Len()))
if err != nil {
t.Fatalf("build eq delete file: %v", err)
}
dfb.EqualityFieldIDs(edf.FieldIDs)
snapID := int64(1)
eqDeleteEntries = append(eqDeleteEntries, iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build()))
}
var eqManifestBuf bytes.Buffer
eqManifestName := "eq-delete-manifest-1.avro"
eqManifestPath := setup.fileRef("metadata", eqManifestName)
_, err := iceberg.WriteManifest(eqManifestPath, &eqManifestBuf, version, spec, schema, 1, eqDeleteEntries)
if err != nil {
t.Fatalf("write eq delete manifest: %v", err)
}
patchedBytes := patchManifestContentToDeletes(t, eqManifestBuf.Bytes())
fs.putEntry(metaDir, eqManifestName, &filer_pb.Entry{
Name: eqManifestName, Content: patchedBytes,
Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(len(patchedBytes))},
})
eqMf := iceberg.NewManifestFile(version, eqManifestPath, int64(len(patchedBytes)), int32(spec.ID()), 1).
Content(iceberg.ManifestContentDeletes).
AddedFiles(int32(len(eqDeleteEntries))).
AddedRows(int64(len(eqDeleteFiles[0].Rows))).
Build()
allManifests = append(allManifests, eqMf)
}
// Write manifest list
var mlBuf bytes.Buffer
seqNum := int64(1)
if err := iceberg.WriteManifestList(version, &mlBuf, 1, nil, &seqNum, 0, allManifests); err != nil {
t.Fatalf("write manifest list: %v", err)
}
fs.putEntry(metaDir, "snap-1.avro", &filer_pb.Entry{
Name: "snap-1.avro", Content: mlBuf.Bytes(),
Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(mlBuf.Len())},
})
// Build final metadata with snapshot
now := time.Now().UnixMilli()
snap := table.Snapshot{SnapshotID: 1, TimestampMs: now, ManifestList: setup.fileRef("metadata", "snap-1.avro")}
if setup.SnapshotSummary != nil {
snap.Summary = &table.Summary{Operation: table.OpAppend, Properties: setup.SnapshotSummary}
}
builder, err := table.MetadataBuilderFromBase(meta, "s3://"+setup.BucketName+"/"+setup.dataPath())
if err != nil {
t.Fatalf("create metadata builder: %v", err)
}
if err := builder.AddSnapshot(&snap); err != nil {
t.Fatalf("add snapshot: %v", err)
}
if err := builder.SetSnapshotRef(table.MainBranch, snap.SnapshotID, table.BranchRef); err != nil {
t.Fatalf("set snapshot ref: %v", err)
}
meta, err = builder.Build()
if err != nil {
t.Fatalf("build metadata: %v", err)
}
// Register table structure
fullMetadataJSON, _ := json.Marshal(meta)
internalMeta := map[string]interface{}{
"metadataVersion": 1,
"metadataLocation": setup.fileRef("metadata", "v1.metadata.json"),
"metadata": map[string]interface{}{"fullMetadata": json.RawMessage(fullMetadataJSON)},
}
xattr, _ := json.Marshal(internalMeta)
fs.putEntry(bucketsPath, 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(nsPath, setup.TableName, &filer_pb.Entry{
Name: setup.TableName, IsDirectory: true,
Extended: map[string][]byte{s3tables.ExtendedKeyMetadata: xattr},
})
return meta
}
func loadLiveDeleteFilePaths(
t *testing.T,
client filer_pb.SeaweedFilerClient,
bucketName, tablePath string,
) (posPaths, eqPaths []string) {
t.Helper()
state, err := loadCurrentMetadata(context.Background(), client, bucketName, tablePath)
if err != nil {
t.Fatalf("loadCurrentMetadata: %v", err)
}
manifests, err := loadCurrentManifests(context.Background(), client, bucketName, state.DataPath, state.Metadata)
if err != nil {
t.Fatalf("loadCurrentManifests: %v", err)
}
for _, mf := range manifests {
if mf.ManifestContent() != iceberg.ManifestContentDeletes {
continue
}
manifestData, err := loadFileByIcebergPath(context.Background(), client, bucketName, tablePath, mf.FilePath())
if err != nil {
t.Fatalf("load delete manifest: %v", err)
}
entries, err := iceberg.ReadManifest(mf, bytes.NewReader(manifestData), true)
if err != nil {
t.Fatalf("read delete manifest: %v", err)
}
for _, entry := range entries {
switch entry.DataFile().ContentType() {
case iceberg.EntryContentPosDeletes:
posPaths = append(posPaths, entry.DataFile().FilePath())
case iceberg.EntryContentEqDeletes:
eqPaths = append(eqPaths, entry.DataFile().FilePath())
}
}
}
sort.Strings(posPaths)
sort.Strings(eqPaths)
return posPaths, eqPaths
}
func rewriteDeleteManifestsAsMixed(
t *testing.T,
fs *fakeFilerServer,
client filer_pb.SeaweedFilerClient,
setup tableSetup,
) {
t.Helper()
state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
if err != nil {
t.Fatalf("loadCurrentMetadata: %v", err)
}
manifests, err := loadCurrentManifests(context.Background(), client, setup.BucketName, state.DataPath, state.Metadata)
if err != nil {
t.Fatalf("loadCurrentManifests: %v", err)
}
var dataManifests []iceberg.ManifestFile
var deleteEntries []iceberg.ManifestEntry
for _, mf := range manifests {
if mf.ManifestContent() == iceberg.ManifestContentData {
dataManifests = append(dataManifests, mf)
continue
}
manifestData, err := loadFileByIcebergPath(context.Background(), client, setup.BucketName, setup.tablePath(), mf.FilePath())
if err != nil {
t.Fatalf("load delete manifest: %v", err)
}
entries, err := iceberg.ReadManifest(mf, bytes.NewReader(manifestData), true)
if err != nil {
t.Fatalf("read delete manifest: %v", err)
}
for _, entry := range entries {
deleteEntries = append(deleteEntries, entry)
}
}
spec := *iceberg.UnpartitionedSpec
version := state.Metadata.Version()
metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata")
manifestName := "mixed-delete-manifest-1.avro"
manifestPath := path.Join("metadata", manifestName)
var manifestBuf bytes.Buffer
_, err = iceberg.WriteManifest(manifestPath, &manifestBuf, version, spec, state.Metadata.CurrentSchema(), 1, deleteEntries)
if err != nil {
t.Fatalf("write mixed delete manifest: %v", err)
}
mixedBytes := patchManifestContentToDeletes(t, manifestBuf.Bytes())
fs.putEntry(metaDir, manifestName, &filer_pb.Entry{
Name: manifestName, Content: mixedBytes,
Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(len(mixedBytes))},
})
mixedManifest := iceberg.NewManifestFile(version, manifestPath, int64(len(mixedBytes)), int32(spec.ID()), 1).
Content(iceberg.ManifestContentDeletes).
AddedFiles(int32(len(deleteEntries))).
Build()
var manifestListBuf bytes.Buffer
seqNum := int64(1)
allManifests := append(dataManifests, mixedManifest)
if err := iceberg.WriteManifestList(version, &manifestListBuf, 1, nil, &seqNum, 0, allManifests); err != nil {
t.Fatalf("write mixed manifest list: %v", err)
}
fs.putEntry(metaDir, "snap-1.avro", &filer_pb.Entry{
Name: "snap-1.avro", Content: manifestListBuf.Bytes(),
Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(manifestListBuf.Len())},
})
}
func readCompactedRows(t *testing.T, fs *fakeFilerServer, setup tableSetup) []struct {
ID int64
Name string
} {
t.Helper()
dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data")
entries := fs.listDir(dataDir)
var mergedContent []byte
for _, e := range entries {
if strings.HasPrefix(e.Name, "compact-") {
mergedContent = e.Content
break
}
}
if mergedContent == nil {
t.Fatal("no merged file found")
}
reader := parquet.NewReader(bytes.NewReader(mergedContent))
defer reader.Close()
type row struct {
ID int64 `parquet:"id"`
Name string `parquet:"name"`
}
var outputRows []struct {
ID int64
Name string
}
for {
var r row
if err := reader.Read(&r); err != nil {
if err == io.EOF {
break
}
t.Fatalf("read: %v", err)
}
outputRows = append(outputRows, struct {
ID int64
Name string
}{ID: r.ID, Name: r.Name})
}
return outputRows
}
func TestCompactDataFilesMetrics(t *testing.T) {
fs, client := startFakeFiler(t)
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
populateTableWithDeleteFiles(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{1, "a"}, {2, "b"}}},
{"d2.parquet", []struct {
ID int64
Name string
}{{3, "c"}}},
},
nil, nil,
)
handler := NewHandler(nil)
config := Config{
TargetFileSizeBytes: 256 * 1024 * 1024,
MinInputFiles: 2,
MaxCommitRetries: 3,
ApplyDeletes: true,
}
// Track progress callbacks
var progressCalls []int
result, metrics, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, func(binIdx, totalBins int) {
progressCalls = append(progressCalls, binIdx)
})
if err != nil {
t.Fatalf("compactDataFiles: %v", err)
}
if !strings.Contains(result, "compacted") {
t.Errorf("expected compaction result, got %q", result)
}
// Verify metrics
if metrics == nil {
t.Fatal("expected non-nil metrics")
}
if metrics[MetricFilesMerged] != 2 {
t.Errorf("expected files_merged=2, got %d", metrics[MetricFilesMerged])
}
if metrics[MetricFilesWritten] != 1 {
t.Errorf("expected files_written=1, got %d", metrics[MetricFilesWritten])
}
if metrics[MetricBins] != 1 {
t.Errorf("expected bins=1, got %d", metrics[MetricBins])
}
if metrics[MetricDurationMs] < 0 {
t.Errorf("expected non-negative duration_ms, got %d", metrics[MetricDurationMs])
}
// Verify progress callback was invoked
if len(progressCalls) != 1 {
t.Errorf("expected 1 progress call, got %d", len(progressCalls))
}
}
// summaryDataFiles is the two-file input every snapshot-summary test compacts.
func summaryDataFiles() []struct {
Name string
Rows []struct {
ID int64
Name string
}
} {
return []struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{1, "a"}, {2, "b"}}},
{"d2.parquet", []struct {
ID int64
Name string
}{{3, "c"}}},
}
}
func snapshotSummaryProps(t *testing.T, client filer_pb.SeaweedFilerClient, setup tableSetup) map[string]string {
t.Helper()
state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
if err != nil {
t.Fatalf("loadCurrentMetadata: %v", err)
}
snap := state.Metadata.CurrentSnapshot()
if snap == nil || snap.Summary == nil {
t.Fatalf("new snapshot has no summary: %+v", snap)
}
return snap.Summary.Properties
}
func requireSummaryValue(t *testing.T, props map[string]string, key, want string) {
t.Helper()
if got := props[key]; got != want {
t.Errorf("%s = %q, want %q", key, got, want)
}
}
func summaryInt(t *testing.T, props map[string]string, key string) int64 {
t.Helper()
raw, ok := props[key]
if !ok {
t.Fatalf("summary is missing %s: %v", key, props)
}
value, err := strconv.ParseInt(raw, 10, 64)
if err != nil {
t.Fatalf("summary %s = %q: %v", key, raw, err)
}
return value
}
// Engines report a table's size from the current snapshot summary, so a
// compaction has to carry the running totals forward the way the writer that
// created the table did.
func TestCompactDataFilesRecordsSnapshotTotals(t *testing.T) {
fs, client := startFakeFiler(t)
const parentFilesSize = 4096
setup := tableSetup{
BucketName: "tb", Namespace: "ns", TableName: "tbl",
SnapshotSummary: map[string]string{
"total-data-files": "2",
"total-records": "3",
"total-files-size": strconv.Itoa(parentFilesSize),
},
}
populateTableWithDeleteFiles(t, fs, setup, summaryDataFiles(), nil, nil)
handler := NewHandler(nil)
config := Config{
TargetFileSizeBytes: 256 * 1024 * 1024,
MinInputFiles: 2,
MaxCommitRetries: 3,
ApplyDeletes: true,
}
if _, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil); err != nil {
t.Fatalf("compactDataFiles: %v", err)
}
props := snapshotSummaryProps(t, client, setup)
requireSummaryValue(t, props, "added-data-files", "1")
requireSummaryValue(t, props, "deleted-data-files", "2")
requireSummaryValue(t, props, "added-records", "3")
requireSummaryValue(t, props, "deleted-records", "3")
// Two files became one, holding the same rows.
requireSummaryValue(t, props, "total-data-files", "1")
requireSummaryValue(t, props, "total-records", "3")
wantSize := parentFilesSize + summaryInt(t, props, "added-files-size") - summaryInt(t, props, "removed-files-size")
requireSummaryValue(t, props, "total-files-size", strconv.FormatInt(wantSize, 10))
// The operation's own labels survive alongside the counters.
requireSummaryValue(t, props, "maintenance", "compact_data_files")
}
// A table whose parent snapshot never recorded totals gets none invented for
// it: "total-records: 0" on a table with rows is worse than no answer.
func TestCompactDataFilesLeavesOutTotalsParentNeverRecorded(t *testing.T) {
fs, client := startFakeFiler(t)
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
populateTableWithDeleteFiles(t, fs, setup, summaryDataFiles(), nil, nil)
handler := NewHandler(nil)
config := Config{
TargetFileSizeBytes: 256 * 1024 * 1024,
MinInputFiles: 2,
MaxCommitRetries: 3,
ApplyDeletes: true,
}
if _, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil); err != nil {
t.Fatalf("compactDataFiles: %v", err)
}
props := snapshotSummaryProps(t, client, setup)
requireSummaryValue(t, props, "added-data-files", "1")
requireSummaryValue(t, props, "deleted-data-files", "2")
for _, key := range []string{"total-data-files", "total-records", "total-files-size"} {
if got, ok := props[key]; ok {
t.Errorf("%s = %q, want it left out", key, got)
}
}
}
func TestRewritePositionDeleteFilesRecordsSnapshotTotals(t *testing.T) {
fs, client := startFakeFiler(t)
setup := tableSetup{
BucketName: "tb", Namespace: "ns", TableName: "tbl",
SnapshotSummary: map[string]string{
"total-data-files": "1",
"total-records": "3",
"total-delete-files": "2",
"total-position-deletes": "3",
},
}
populateTableWithDeleteFiles(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{1, "alice"}, {2, "bob"}, {3, "charlie"}}},
},
[]struct {
Name string
Rows []struct {
FilePath string
Pos int64
}
}{
{"pd1.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 0}, {"data/d1.parquet", 2}}},
{"pd2.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 1}}},
},
nil,
)
handler := NewHandler(nil)
config := Config{
DeleteTargetFileSizeBytes: 64 * 1024 * 1024,
DeleteMinInputFiles: 2,
DeleteMaxFileGroupSizeBytes: 128 * 1024 * 1024,
DeleteMaxOutputFiles: 4,
MaxCommitRetries: 3,
}
if _, _, err := handler.rewritePositionDeleteFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config); err != nil {
t.Fatalf("rewritePositionDeleteFiles: %v", err)
}
props := snapshotSummaryProps(t, client, setup)
requireSummaryValue(t, props, "added-delete-files", "1")
requireSummaryValue(t, props, "removed-delete-files", "2")
requireSummaryValue(t, props, "added-position-delete-files", "1")
requireSummaryValue(t, props, "added-position-deletes", "3")
requireSummaryValue(t, props, "removed-position-deletes", "3")
// Two delete files became one; the deleted rows and the data are untouched.
requireSummaryValue(t, props, "total-delete-files", "1")
requireSummaryValue(t, props, "total-position-deletes", "3")
requireSummaryValue(t, props, "total-data-files", "1")
requireSummaryValue(t, props, "total-records", "3")
}
func TestExpireSnapshotsMetrics(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "ns",
TableName: "tbl",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
},
}
populateTable(t, fs, setup)
handler := NewHandler(nil)
config := Config{
SnapshotRetentionMs: hoursToMs(0),
MaxSnapshotsToKeep: 1,
MaxCommitRetries: 3,
}
_, metrics, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config)
if err != nil {
t.Fatalf("expireSnapshots: %v", err)
}
if metrics == nil {
t.Fatal("expected non-nil metrics")
}
if metrics[MetricSnapshotsExpired] == 0 {
t.Error("expected snapshots_expired > 0")
}
if metrics[MetricDurationMs] < 0 {
t.Errorf("expected non-negative duration_ms, got %d", metrics[MetricDurationMs])
}
}
func TestExecuteCompletionOutputValues(t *testing.T) {
fs, client := startFakeFiler(t)
now := time.Now().UnixMilli()
setup := tableSetup{
BucketName: "test-bucket",
Namespace: "ns",
TableName: "tbl",
Snapshots: []table.Snapshot{
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
},
}
populateTable(t, fs, setup)
handler := NewHandler(nil)
config := Config{
SnapshotRetentionMs: hoursToMs(0),
MaxSnapshotsToKeep: 1,
MaxCommitRetries: 3,
}
_, metrics, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config)
if err != nil {
t.Fatalf("expireSnapshots: %v", err)
}
// Verify metrics have the expected keys
if _, ok := metrics[MetricSnapshotsExpired]; !ok {
t.Error("expected 'snapshots_expired' key in metrics")
}
if _, ok := metrics[MetricFilesDeleted]; !ok {
t.Error("expected 'files_deleted' key in metrics")
}
if _, ok := metrics[MetricDurationMs]; !ok {
t.Error("expected 'duration_ms' key in metrics")
}
}
func TestCompactDataFilesWithPositionDeletes(t *testing.T) {
fs, client := startFakeFiler(t)
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
populateTableWithDeleteFiles(t, fs, setup,
// 3 small data files (to meet min_input_files=2)
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{1, "alice"}, {2, "bob"}, {3, "charlie"}}},
{"d2.parquet", []struct {
ID int64
Name string
}{{4, "dave"}, {5, "eve"}}},
},
// Position deletes: delete row 1 (bob) from d1
[]struct {
Name string
Rows []struct {
FilePath string
Pos int64
}
}{
{"pd1.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 1}}},
},
nil, // no equality deletes
)
handler := NewHandler(nil)
config := Config{
TargetFileSizeBytes: 256 * 1024 * 1024,
MinInputFiles: 2,
MaxCommitRetries: 3,
ApplyDeletes: true,
}
result, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil)
if err != nil {
t.Fatalf("compactDataFiles: %v", err)
}
if !strings.Contains(result, "compacted") {
t.Errorf("expected compaction result, got %q", result)
}
t.Logf("result: %s", result)
// Verify: read the merged output and count rows
// The merged file should have 4 rows (5 total - 1 position delete)
dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data")
entries := fs.listDir(dataDir)
var mergedContent []byte
for _, e := range entries {
if strings.HasPrefix(e.Name, "compact-") {
mergedContent = e.Content
break
}
}
if mergedContent == nil {
t.Fatal("no merged file found")
}
reader := parquet.NewReader(bytes.NewReader(mergedContent))
defer reader.Close()
type row struct {
ID int64 `parquet:"id"`
Name string `parquet:"name"`
}
var outputRows []row
for {
var r row
if err := reader.Read(&r); err != nil {
if err == io.EOF {
break
}
t.Fatalf("read: %v", err)
}
outputRows = append(outputRows, r)
}
if len(outputRows) != 4 {
t.Errorf("expected 4 rows (5 - 1 pos delete), got %d", len(outputRows))
}
for _, r := range outputRows {
if r.Name == "bob" {
t.Error("bob should have been deleted by position delete")
}
}
}
func TestCompactDataFilesWithEqualityDeletes(t *testing.T) {
skipIfEqualityDeleteAvroUnsupported(t)
fs, client := startFakeFiler(t)
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
populateTableWithDeleteFiles(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{1, "alice"}, {2, "bob"}}},
{"d2.parquet", []struct {
ID int64
Name string
}{{3, "charlie"}, {4, "dave"}}},
},
nil, // no position deletes
// Equality deletes: delete rows where name="bob" or name="dave"
[]struct {
Name string
FieldIDs []int
Rows []struct {
ID int64
Name string
}
}{
{"ed1.parquet", []int{2}, []struct {
ID int64
Name string
}{{0, "bob"}, {0, "dave"}}},
},
)
handler := NewHandler(nil)
config := Config{
TargetFileSizeBytes: 256 * 1024 * 1024,
MinInputFiles: 2,
MaxCommitRetries: 3,
ApplyDeletes: true,
}
result, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil)
if err != nil {
t.Fatalf("compactDataFiles: %v", err)
}
if !strings.Contains(result, "compacted") {
t.Errorf("expected compaction result, got %q", result)
}
t.Logf("result: %s", result)
// Verify merged output
dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data")
entries := fs.listDir(dataDir)
var mergedContent []byte
for _, e := range entries {
if strings.HasPrefix(e.Name, "compact-") {
mergedContent = e.Content
break
}
}
if mergedContent == nil {
t.Fatal("no merged file found")
}
reader := parquet.NewReader(bytes.NewReader(mergedContent))
defer reader.Close()
type row struct {
ID int64 `parquet:"id"`
Name string `parquet:"name"`
}
var outputRows []row
for {
var r row
if err := reader.Read(&r); err != nil {
if err == io.EOF {
break
}
t.Fatalf("read: %v", err)
}
outputRows = append(outputRows, r)
}
if len(outputRows) != 2 {
t.Errorf("expected 2 rows (4 - 2 eq deletes), got %d", len(outputRows))
}
for _, r := range outputRows {
if r.Name == "bob" || r.Name == "dave" {
t.Errorf("row %q should have been deleted by equality delete", r.Name)
}
}
}
func TestCompactDataFilesApplyDeletesDisabled(t *testing.T) {
fs, client := startFakeFiler(t)
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
populateTableWithDeleteFiles(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{1, "a"}, {2, "b"}}},
{"d2.parquet", []struct {
ID int64
Name string
}{{3, "c"}}},
},
[]struct {
Name string
Rows []struct {
FilePath string
Pos int64
}
}{
{"pd1.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 0}}},
},
nil,
)
handler := NewHandler(nil)
config := Config{
TargetFileSizeBytes: 256 * 1024 * 1024,
MinInputFiles: 2,
MaxCommitRetries: 3,
ApplyDeletes: false, // disabled
}
result, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil)
if err != nil {
t.Fatalf("compactDataFiles: %v", err)
}
if !strings.Contains(result, "skipped") {
t.Errorf("expected skip when apply_deletes=false, got %q", result)
}
}
func TestCompactDataFilesWithMixedDeletes(t *testing.T) {
skipIfEqualityDeleteAvroUnsupported(t)
fs, client := startFakeFiler(t)
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
populateTableWithDeleteFiles(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{1, "alice"}, {2, "bob"}, {3, "charlie"}}},
{"d2.parquet", []struct {
ID int64
Name string
}{{4, "dave"}, {5, "eve"}}},
},
// Position delete: row 0 (alice) from d1
[]struct {
Name string
Rows []struct {
FilePath string
Pos int64
}
}{
{"pd1.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 0}}},
},
// Equality delete: name="eve"
[]struct {
Name string
FieldIDs []int
Rows []struct {
ID int64
Name string
}
}{
{"ed1.parquet", []int{2}, []struct {
ID int64
Name string
}{{0, "eve"}}},
},
)
handler := NewHandler(nil)
config := Config{
TargetFileSizeBytes: 256 * 1024 * 1024,
MinInputFiles: 2,
MaxCommitRetries: 3,
ApplyDeletes: true,
}
result, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil)
if err != nil {
t.Fatalf("compactDataFiles: %v", err)
}
if !strings.Contains(result, "compacted") {
t.Errorf("expected compaction, got %q", result)
}
// Verify: 5 total - 1 pos delete (alice) - 1 eq delete (eve) = 3 rows
dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data")
entries := fs.listDir(dataDir)
var mergedContent []byte
for _, e := range entries {
if strings.HasPrefix(e.Name, "compact-") {
mergedContent = e.Content
break
}
}
if mergedContent == nil {
t.Fatal("no merged file found")
}
reader := parquet.NewReader(bytes.NewReader(mergedContent))
defer reader.Close()
type row struct {
ID int64 `parquet:"id"`
Name string `parquet:"name"`
}
var outputRows []row
for {
var r row
if err := reader.Read(&r); err != nil {
if err == io.EOF {
break
}
t.Fatalf("read: %v", err)
}
outputRows = append(outputRows, r)
}
if len(outputRows) != 3 {
t.Errorf("expected 3 rows (5 - 1 pos - 1 eq), got %d", len(outputRows))
}
for _, r := range outputRows {
if r.Name == "alice" || r.Name == "eve" {
t.Errorf("%q should have been deleted", r.Name)
}
}
}
func TestCompactDataFilesSortStrategyUsesAscendingTableSortOrder(t *testing.T) {
fs, client := startFakeFiler(t)
sortOrder, err := table.NewSortOrder(1, []table.SortField{{
SourceIDs: []int{1},
Transform: iceberg.IdentityTransform{},
Direction: table.SortASC,
NullOrder: table.NullsFirst,
}})
if err != nil {
t.Fatalf("new sort order: %v", err)
}
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
populateTableWithDeleteFilesAndSortOrder(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{3, "charlie"}, {1, "alice"}}},
{"d2.parquet", []struct {
ID int64
Name string
}{{4, "delta"}, {2, "bravo"}}},
},
nil,
nil,
sortOrder,
)
handler := NewHandler(nil)
config := Config{
TargetFileSizeBytes: 256 * 1024 * 1024,
MinInputFiles: 2,
MaxCommitRetries: 3,
ApplyDeletes: true,
RewriteStrategy: "sort",
}
result, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil)
if err != nil {
t.Fatalf("compactDataFiles: %v", err)
}
if !strings.Contains(result, "using sort") {
t.Fatalf("expected sorted compaction result, got %q", result)
}
rows := readCompactedRows(t, fs, setup)
if len(rows) != 4 {
t.Fatalf("expected 4 compacted rows, got %d", len(rows))
}
for i := 1; i < len(rows); i++ {
if rows[i-1].ID > rows[i].ID {
t.Fatalf("rows are not sorted by id: %+v", rows)
}
}
}
func TestCompactDataFilesSortStrategyUsesTableSortOrder(t *testing.T) {
fs, client := startFakeFiler(t)
sortOrder, err := table.NewSortOrder(1, []table.SortField{{
SourceIDs: []int{2},
Transform: iceberg.IdentityTransform{},
Direction: table.SortDESC,
NullOrder: table.NullsLast,
}})
if err != nil {
t.Fatalf("new sort order: %v", err)
}
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
populateTableWithDeleteFilesAndSortOrder(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{1, "alice"}, {2, "charlie"}}},
{"d2.parquet", []struct {
ID int64
Name string
}{{3, "bravo"}, {4, "delta"}}},
},
nil,
nil,
sortOrder,
)
handler := NewHandler(nil)
config := Config{
TargetFileSizeBytes: 256 * 1024 * 1024,
MinInputFiles: 2,
MaxCommitRetries: 3,
ApplyDeletes: true,
RewriteStrategy: "sort",
}
result, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil)
if err != nil {
t.Fatalf("compactDataFiles: %v", err)
}
if !strings.Contains(result, "name desc") {
t.Fatalf("expected table sort order in result, got %q", result)
}
rows := readCompactedRows(t, fs, setup)
gotNames := make([]string, 0, len(rows))
for _, row := range rows {
gotNames = append(gotNames, row.Name)
}
expectedNames := []string{"delta", "charlie", "bravo", "alice"}
if len(gotNames) != len(expectedNames) {
t.Fatalf("got %d rows, want %d", len(gotNames), len(expectedNames))
}
for i := range expectedNames {
if gotNames[i] != expectedNames[i] {
t.Fatalf("names[%d] = %q, want %q (all names: %v)", i, gotNames[i], expectedNames[i], gotNames)
}
}
}
func TestDetectSkipsSortCompactionBinsAboveCap(t *testing.T) {
fs, client := startFakeFiler(t)
sortOrder, err := table.NewSortOrder(1, []table.SortField{{
SourceIDs: []int{1},
Transform: iceberg.IdentityTransform{},
Direction: table.SortASC,
NullOrder: table.NullsFirst,
}})
if err != nil {
t.Fatalf("new sort order: %v", err)
}
setup := tableSetup{BucketName: "test-bucket", Namespace: "analytics", TableName: "events"}
populateTableWithDeleteFilesAndSortOrder(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"small-1.parquet", []struct {
ID int64
Name string
}{{1, "alpha"}}},
{"small-2.parquet", []struct {
ID int64
Name string
}{{2, "bravo"}}},
},
nil,
nil,
sortOrder,
)
handler := NewHandler(nil)
config := Config{
TargetFileSizeBytes: 4096,
MinInputFiles: 2,
Operations: "compact",
RewriteStrategy: "sort",
SortMaxInputBytes: 1,
}
tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
if err != nil {
t.Fatalf("scanTablesForMaintenance failed: %v", err)
}
if len(tables) != 0 {
t.Fatalf("expected no sort compaction candidates above the input cap, got %d", len(tables))
}
}
func TestDetectSplitsSortCompactionBinsByCap(t *testing.T) {
fs, client := startFakeFiler(t)
sortOrder, err := table.NewSortOrder(1, []table.SortField{{
SourceIDs: []int{1},
Transform: iceberg.IdentityTransform{},
Direction: table.SortASC,
NullOrder: table.NullsFirst,
}})
if err != nil {
t.Fatalf("new sort order: %v", err)
}
setup := tableSetup{BucketName: "test-bucket", Namespace: "analytics", TableName: "events"}
populateTableWithDeleteFilesAndSortOrder(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"small-1.parquet", []struct {
ID int64
Name string
}{{1, "alpha"}}},
{"small-2.parquet", []struct {
ID int64
Name string
}{{2, "bravo"}}},
{"small-3.parquet", []struct {
ID int64
Name string
}{{3, "charlie"}}},
{"small-4.parquet", []struct {
ID int64
Name string
}{{4, "delta"}}},
},
nil,
nil,
sortOrder,
)
state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
if err != nil {
t.Fatalf("loadCurrentMetadata: %v", err)
}
manifests, err := loadCurrentManifests(context.Background(), client, setup.BucketName, state.DataPath, state.Metadata)
if err != nil {
t.Fatalf("loadCurrentManifests: %v", err)
}
var totalSize int64
var maxFileSize int64
for _, mf := range manifests {
if mf.ManifestContent() != iceberg.ManifestContentData {
continue
}
manifestData, err := loadFileByIcebergPath(context.Background(), client, setup.BucketName, setup.tablePath(), mf.FilePath())
if err != nil {
t.Fatalf("load data manifest: %v", err)
}
entries, err := iceberg.ReadManifest(mf, bytes.NewReader(manifestData), true)
if err != nil {
t.Fatalf("read data manifest: %v", err)
}
for _, entry := range entries {
size := entry.DataFile().FileSizeBytes()
totalSize += size
if size > maxFileSize {
maxFileSize = size
}
}
}
if totalSize == 0 || maxFileSize == 0 {
t.Fatal("expected data file sizes to be populated")
}
capBytes := totalSize / 2
if capBytes <= maxFileSize {
capBytes = maxFileSize + 1
}
handler := NewHandler(nil)
config := Config{
TargetFileSizeBytes: totalSize + 1,
MinInputFiles: 2,
Operations: "compact",
RewriteStrategy: "sort",
SortMaxInputBytes: capBytes,
}
tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
if err != nil {
t.Fatalf("scanTablesForMaintenance failed: %v", err)
}
if len(tables) != 1 {
t.Fatalf("expected sort compaction candidate to survive cap-based repartitioning, got %d", len(tables))
}
}
func TestCompactDataFilesSortStrategyRequiresTableSortOrder(t *testing.T) {
fs, client := startFakeFiler(t)
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
populateTableWithDeleteFiles(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{3, "charlie"}, {1, "alice"}}},
{"d2.parquet", []struct {
ID int64
Name string
}{{4, "delta"}, {2, "bravo"}}},
},
nil,
nil,
)
handler := NewHandler(nil)
config := Config{
TargetFileSizeBytes: 256 * 1024 * 1024,
MinInputFiles: 2,
MaxCommitRetries: 3,
ApplyDeletes: true,
RewriteStrategy: "sort",
}
_, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil)
if err == nil || !strings.Contains(err.Error(), "table sort order") {
t.Fatalf("expected missing table sort order error, got %v", err)
}
}
func TestRewritePositionDeleteFilesExecution(t *testing.T) {
fs, client := startFakeFiler(t)
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
populateTableWithDeleteFiles(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{1, "alice"}, {2, "bob"}, {3, "charlie"}}},
},
[]struct {
Name string
Rows []struct {
FilePath string
Pos int64
}
}{
{"pd1.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 0}, {"data/d1.parquet", 2}}},
{"pd2.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 1}}},
},
nil,
)
handler := NewHandler(nil)
config := Config{
DeleteTargetFileSizeBytes: 64 * 1024 * 1024,
DeleteMinInputFiles: 2,
DeleteMaxFileGroupSizeBytes: 128 * 1024 * 1024,
DeleteMaxOutputFiles: 4,
MaxCommitRetries: 3,
}
result, metrics, err := handler.rewritePositionDeleteFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config)
if err != nil {
t.Fatalf("rewritePositionDeleteFiles: %v", err)
}
if !strings.Contains(result, "rewrote 2 position delete files into 1") {
t.Fatalf("unexpected result: %q", result)
}
if metrics[MetricDeleteFilesRewritten] != 2 {
t.Fatalf("expected 2 rewritten files, got %d", metrics[MetricDeleteFilesRewritten])
}
if metrics[MetricDeleteFilesWritten] != 1 {
t.Fatalf("expected 1 written file, got %d", metrics[MetricDeleteFilesWritten])
}
liveDeletePaths, _ := loadLiveDeleteFilePaths(t, client, setup.BucketName, setup.tablePath())
if len(liveDeletePaths) != 1 {
t.Fatalf("expected 1 live rewritten delete file, got %v", liveDeletePaths)
}
if !strings.HasPrefix(liveDeletePaths[0], "s3://tb/ns/tbl/data/rewrite-delete-") {
t.Fatalf("expected rewritten delete file path, got %q", liveDeletePaths[0])
}
}
func TestRewritePositionDeleteFilesDetection(t *testing.T) {
fs, client := startFakeFiler(t)
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
populateTableWithDeleteFiles(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{1, "alice"}, {2, "bob"}}},
},
[]struct {
Name string
Rows []struct {
FilePath string
Pos int64
}
}{
{"pd1.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 0}}},
{"pd2.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 1}}},
},
nil,
)
handler := NewHandler(nil)
config := Config{
Operations: "rewrite_position_delete_files",
DeleteTargetFileSizeBytes: 64 * 1024 * 1024,
DeleteMinInputFiles: 2,
DeleteMaxFileGroupSizeBytes: 128 * 1024 * 1024,
DeleteMaxOutputFiles: 4,
}
tables, err := handler.scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
if err != nil {
t.Fatalf("scanTablesForMaintenance: %v", err)
}
if len(tables) != 1 {
t.Fatalf("expected 1 table needing delete rewrite, got %d", len(tables))
}
}
func TestRewritePositionDeleteFilesSkipsSingleFile(t *testing.T) {
fs, client := startFakeFiler(t)
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
populateTableWithDeleteFiles(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{1, "alice"}, {2, "bob"}}},
},
[]struct {
Name string
Rows []struct {
FilePath string
Pos int64
}
}{
{"pd1.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 0}}},
},
nil,
)
handler := NewHandler(nil)
config := Config{
DeleteTargetFileSizeBytes: 64 * 1024 * 1024,
DeleteMinInputFiles: 2,
DeleteMaxFileGroupSizeBytes: 128 * 1024 * 1024,
DeleteMaxOutputFiles: 4,
MaxCommitRetries: 3,
}
result, _, err := handler.rewritePositionDeleteFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config)
if err != nil {
t.Fatalf("rewritePositionDeleteFiles: %v", err)
}
if !strings.Contains(result, "no position delete files eligible") {
t.Fatalf("unexpected result: %q", result)
}
}
func TestRewritePositionDeleteFilesRespectsMinInputFiles(t *testing.T) {
fs, client := startFakeFiler(t)
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
populateTableWithDeleteFiles(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{1, "alice"}, {2, "bob"}}},
},
[]struct {
Name string
Rows []struct {
FilePath string
Pos int64
}
}{
{"pd1.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 0}}},
{"pd2.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 1}}},
},
nil,
)
handler := NewHandler(nil)
config := Config{
DeleteTargetFileSizeBytes: 64 * 1024 * 1024,
DeleteMinInputFiles: 3,
DeleteMaxFileGroupSizeBytes: 128 * 1024 * 1024,
DeleteMaxOutputFiles: 4,
MaxCommitRetries: 3,
}
result, _, err := handler.rewritePositionDeleteFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config)
if err != nil {
t.Fatalf("rewritePositionDeleteFiles: %v", err)
}
if !strings.Contains(result, "no position delete files eligible") {
t.Fatalf("unexpected result: %q", result)
}
}
func TestRewritePositionDeleteFilesPreservesUnsupportedMultiTargetDeletes(t *testing.T) {
fs, client := startFakeFiler(t)
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
populateTableWithDeleteFiles(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{1, "alice"}, {2, "bob"}, {3, "charlie"}}},
{"d2.parquet", []struct {
ID int64
Name string
}{{4, "diana"}, {5, "eve"}}},
},
[]struct {
Name string
Rows []struct {
FilePath string
Pos int64
}
}{
{"pd1.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 0}}},
{"pd2.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 1}}},
{"pd3.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 2}, {"data/d2.parquet", 0}}},
},
nil,
)
handler := NewHandler(nil)
config := Config{
DeleteTargetFileSizeBytes: 64 * 1024 * 1024,
DeleteMinInputFiles: 2,
DeleteMaxFileGroupSizeBytes: 128 * 1024 * 1024,
DeleteMaxOutputFiles: 4,
MaxCommitRetries: 3,
}
if _, _, err := handler.rewritePositionDeleteFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config); err != nil {
t.Fatalf("rewritePositionDeleteFiles: %v", err)
}
posPaths, _ := loadLiveDeleteFilePaths(t, client, setup.BucketName, setup.tablePath())
if len(posPaths) != 2 {
t.Fatalf("expected rewritten file plus untouched multi-target file, got %v", posPaths)
}
if posPaths[0] != "data/pd3.parquet" && posPaths[1] != "data/pd3.parquet" {
t.Fatalf("expected multi-target delete file to be preserved, got %v", posPaths)
}
if !strings.HasPrefix(posPaths[0], "s3://tb/ns/tbl/data/rewrite-delete-") && !strings.HasPrefix(posPaths[1], "s3://tb/ns/tbl/data/rewrite-delete-") {
t.Fatalf("expected rewritten delete file to remain live, got %v", posPaths)
}
}
func TestRewritePositionDeleteFilesRebuildsMixedDeleteManifests(t *testing.T) {
skipIfEqualityDeleteAvroUnsupported(t)
fs, client := startFakeFiler(t)
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
populateTableWithDeleteFiles(t, fs, setup,
[]struct {
Name string
Rows []struct {
ID int64
Name string
}
}{
{"d1.parquet", []struct {
ID int64
Name string
}{{1, "alice"}, {2, "bob"}, {3, "charlie"}}},
},
[]struct {
Name string
Rows []struct {
FilePath string
Pos int64
}
}{
{"pd1.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 0}}},
{"pd2.parquet", []struct {
FilePath string
Pos int64
}{{"data/d1.parquet", 1}}},
},
[]struct {
Name string
FieldIDs []int
Rows []struct {
ID int64
Name string
}
}{
{"eq1.parquet", []int{1}, []struct {
ID int64
Name string
}{{3, "charlie"}}},
},
)
rewriteDeleteManifestsAsMixed(t, fs, client, setup)
handler := NewHandler(nil)
config := Config{
DeleteTargetFileSizeBytes: 64 * 1024 * 1024,
DeleteMinInputFiles: 2,
DeleteMaxFileGroupSizeBytes: 128 * 1024 * 1024,
DeleteMaxOutputFiles: 4,
MaxCommitRetries: 3,
}
if _, _, err := handler.rewritePositionDeleteFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config); err != nil {
t.Fatalf("rewritePositionDeleteFiles: %v", err)
}
posPaths, eqPaths := loadLiveDeleteFilePaths(t, client, setup.BucketName, setup.tablePath())
if len(posPaths) != 1 || !strings.HasPrefix(posPaths[0], "s3://tb/ns/tbl/data/rewrite-delete-") {
t.Fatalf("expected only the rewritten position delete file to remain live, got %v", posPaths)
}
if len(eqPaths) != 1 || eqPaths[0] != "data/eq1.parquet" {
t.Fatalf("expected equality delete file to be preserved, got %v", eqPaths)
}
}
func TestResolveCompactionRewritePlanFallsBackForUnsupportedSortTransform(t *testing.T) {
sortOrder, err := table.NewSortOrder(1, []table.SortField{{
SourceIDs: []int{1},
Transform: iceberg.BucketTransform{NumBuckets: 16},
Direction: table.SortASC,
NullOrder: table.NullsFirst,
}})
if err != nil {
t.Fatalf("new sort order: %v", err)
}
meta, err := table.NewMetadata(newTestSchema(), iceberg.UnpartitionedSpec, sortOrder, "s3://test-bucket/test-table", nil)
if err != nil {
t.Fatalf("new metadata: %v", err)
}
plan, err := resolveCompactionRewritePlan(Config{RewriteStrategy: "sort"}, meta)
if err != nil {
t.Fatalf("resolveCompactionRewritePlan: %v", err)
}
if plan == nil || plan.strategy != defaultRewriteStrategy {
t.Fatalf("expected fallback to %q for unsupported transform, got %+v", defaultRewriteStrategy, plan)
}
}