iceberg: harden lifecycle retention planning

other/java/client/src/main/proto/filer.proto

@@ -213,6 +213,7 @@ message UpdateEntryRequest {
     Entry entry = 2;
     bool is_from_other_cluster = 3;
     repeated int32 signatures = 4;
+    map<string, bytes> expected_extended = 5;
 }
 message UpdateEntryResponse {
     SubscribeMetadataResponse metadata_event = 1;

weed/pb/filer.proto

@@ -213,6 +213,7 @@ message UpdateEntryRequest {
     Entry entry = 2;
     bool is_from_other_cluster = 3;
     repeated int32 signatures = 4;
+    map<string, bytes> expected_extended = 5;
 }
 message UpdateEntryResponse {
     SubscribeMetadataResponse metadata_event = 1;

weed/pb/filer_pb/filer.pb.go

@@ -1175,6 +1175,7 @@ type UpdateEntryRequest struct {
 	Entry              *Entry                 `protobuf:"bytes,2,opt,name=entry,proto3" json:"entry,omitempty"`
 	IsFromOtherCluster bool                   `protobuf:"varint,3,opt,name=is_from_other_cluster,json=isFromOtherCluster,proto3" json:"is_from_other_cluster,omitempty"`
 	Signatures         []int32                `protobuf:"varint,4,rep,packed,name=signatures,proto3" json:"signatures,omitempty"`
+	ExpectedExtended   map[string][]byte      `protobuf:"bytes,5,rep,name=expected_extended,json=expectedExtended,proto3" json:"expected_extended,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
 	unknownFields      protoimpl.UnknownFields
 	sizeCache          protoimpl.SizeCache
 }
@@ -1237,6 +1238,13 @@ func (x *UpdateEntryRequest) GetSignatures() []int32 {
 	return nil
 }
 
+func (x *UpdateEntryRequest) GetExpectedExtended() map[string][]byte {
+	if x != nil {
+		return x.ExpectedExtended
+	}
+	return nil
+}
+
 type UpdateEntryResponse struct {
 	state         protoimpl.MessageState     `protogen:"open.v1"`
 	MetadataEvent *SubscribeMetadataResponse `protobuf:"bytes,1,opt,name=metadata_event,json=metadataEvent,proto3" json:"metadata_event,omitempty"`
@@ -4234,7 +4242,7 @@ type LocateBrokerResponse_Resource struct {
 
 func (x *LocateBrokerResponse_Resource) Reset() {
 	*x = LocateBrokerResponse_Resource{}
-	mi := &file_filer_proto_msgTypes[68]
+	mi := &file_filer_proto_msgTypes[69]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -4246,7 +4254,7 @@ func (x *LocateBrokerResponse_Resource) String() string {
 func (*LocateBrokerResponse_Resource) ProtoMessage() {}
 
 func (x *LocateBrokerResponse_Resource) ProtoReflect() protoreflect.Message {
-	mi := &file_filer_proto_msgTypes[68]
+	mi := &file_filer_proto_msgTypes[69]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -4300,7 +4308,7 @@ type FilerConf_PathConf struct {
 
 func (x *FilerConf_PathConf) Reset() {
 	*x = FilerConf_PathConf{}
-	mi := &file_filer_proto_msgTypes[69]
+	mi := &file_filer_proto_msgTypes[70]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -4312,7 +4320,7 @@ func (x *FilerConf_PathConf) String() string {
 func (*FilerConf_PathConf) ProtoMessage() {}
 
 func (x *FilerConf_PathConf) ProtoReflect() protoreflect.Message {
-	mi := &file_filer_proto_msgTypes[69]
+	mi := &file_filer_proto_msgTypes[70]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -4550,14 +4558,18 @@ const file_filer_proto_rawDesc = "" +
 	"\x1bskip_check_parent_directory\x18\x06 \x01(\bR\x18skipCheckParentDirectory\"w\n" +
 	"\x13CreateEntryResponse\x12\x14\n" +
 	"\x05error\x18\x01 \x01(\tR\x05error\x12J\n" +
-	"\x0emetadata_event\x18\x02 \x01(\v2#.filer_pb.SubscribeMetadataResponseR\rmetadataEvent\"\xac\x01\n" +
+	"\x0emetadata_event\x18\x02 \x01(\v2#.filer_pb.SubscribeMetadataResponseR\rmetadataEvent\"\xd2\x02\n" +
 	"\x12UpdateEntryRequest\x12\x1c\n" +
 	"\tdirectory\x18\x01 \x01(\tR\tdirectory\x12%\n" +
 	"\x05entry\x18\x02 \x01(\v2\x0f.filer_pb.EntryR\x05entry\x121\n" +
 	"\x15is_from_other_cluster\x18\x03 \x01(\bR\x12isFromOtherCluster\x12\x1e\n" +
 	"\n" +
 	"signatures\x18\x04 \x03(\x05R\n" +
-	"signatures\"a\n" +
+	"signatures\x12_\n" +
+	"\x11expected_extended\x18\x05 \x03(\v22.filer_pb.UpdateEntryRequest.ExpectedExtendedEntryR\x10expectedExtended\x1aC\n" +
+	"\x15ExpectedExtendedEntry\x12\x10\n" +
+	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
+	"\x05value\x18\x02 \x01(\fR\x05value:\x028\x01\"a\n" +
 	"\x13UpdateEntryResponse\x12J\n" +
 	"\x0emetadata_event\x18\x01 \x01(\v2#.filer_pb.SubscribeMetadataResponseR\rmetadataEvent\"\x80\x01\n" +
 	"\x14AppendToEntryRequest\x12\x1c\n" +
@@ -4865,7 +4877,7 @@ func file_filer_proto_rawDescGZIP() []byte {
 }
 
 var file_filer_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
-var file_filer_proto_msgTypes = make([]protoimpl.MessageInfo, 70)
+var file_filer_proto_msgTypes = make([]protoimpl.MessageInfo, 71)
 var file_filer_proto_goTypes = []any{
 	(SSEType)(0),                                    // 0: filer_pb.SSEType
 	(*LookupDirectoryEntryRequest)(nil),             // 1: filer_pb.LookupDirectoryEntryRequest
@@ -4935,9 +4947,10 @@ var file_filer_proto_goTypes = []any{
 	(*TransferLocksRequest)(nil),                    // 65: filer_pb.TransferLocksRequest
 	(*TransferLocksResponse)(nil),                   // 66: filer_pb.TransferLocksResponse
 	nil,                                             // 67: filer_pb.Entry.ExtendedEntry
-	nil,                                             // 68: filer_pb.LookupVolumeResponse.LocationsMapEntry
-	(*LocateBrokerResponse_Resource)(nil),           // 69: filer_pb.LocateBrokerResponse.Resource
-	(*FilerConf_PathConf)(nil),                      // 70: filer_pb.FilerConf.PathConf
+	nil,                                             // 68: filer_pb.UpdateEntryRequest.ExpectedExtendedEntry
+	nil,                                             // 69: filer_pb.LookupVolumeResponse.LocationsMapEntry
+	(*LocateBrokerResponse_Resource)(nil),           // 70: filer_pb.LocateBrokerResponse.Resource
+	(*FilerConf_PathConf)(nil),                      // 71: filer_pb.FilerConf.PathConf
 }
 var file_filer_proto_depIdxs = []int32{
 	6,  // 0: filer_pb.LookupDirectoryEntryResponse.entry:type_name -> filer_pb.Entry
@@ -4956,77 +4969,78 @@ var file_filer_proto_depIdxs = []int32{
 	6,  // 13: filer_pb.CreateEntryRequest.entry:type_name -> filer_pb.Entry
 	43, // 14: filer_pb.CreateEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse
 	6,  // 15: filer_pb.UpdateEntryRequest.entry:type_name -> filer_pb.Entry
-	43, // 16: filer_pb.UpdateEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse
-	9,  // 17: filer_pb.AppendToEntryRequest.chunks:type_name -> filer_pb.FileChunk
-	43, // 18: filer_pb.DeleteEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse
-	8,  // 19: filer_pb.StreamRenameEntryResponse.event_notification:type_name -> filer_pb.EventNotification
-	29, // 20: filer_pb.AssignVolumeResponse.location:type_name -> filer_pb.Location
-	29, // 21: filer_pb.Locations.locations:type_name -> filer_pb.Location
-	68, // 22: filer_pb.LookupVolumeResponse.locations_map:type_name -> filer_pb.LookupVolumeResponse.LocationsMapEntry
-	31, // 23: filer_pb.CollectionListResponse.collections:type_name -> filer_pb.Collection
-	8,  // 24: filer_pb.SubscribeMetadataResponse.event_notification:type_name -> filer_pb.EventNotification
-	6,  // 25: filer_pb.TraverseBfsMetadataResponse.entry:type_name -> filer_pb.Entry
-	69, // 26: filer_pb.LocateBrokerResponse.resources:type_name -> filer_pb.LocateBrokerResponse.Resource
-	70, // 27: filer_pb.FilerConf.locations:type_name -> filer_pb.FilerConf.PathConf
-	6,  // 28: filer_pb.CacheRemoteObjectToLocalClusterResponse.entry:type_name -> filer_pb.Entry
-	43, // 29: filer_pb.CacheRemoteObjectToLocalClusterResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse
-	64, // 30: filer_pb.TransferLocksRequest.locks:type_name -> filer_pb.Lock
-	28, // 31: filer_pb.LookupVolumeResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations
-	1,  // 32: filer_pb.SeaweedFiler.LookupDirectoryEntry:input_type -> filer_pb.LookupDirectoryEntryRequest
-	3,  // 33: filer_pb.SeaweedFiler.ListEntries:input_type -> filer_pb.ListEntriesRequest
-	13, // 34: filer_pb.SeaweedFiler.CreateEntry:input_type -> filer_pb.CreateEntryRequest
-	15, // 35: filer_pb.SeaweedFiler.UpdateEntry:input_type -> filer_pb.UpdateEntryRequest
-	17, // 36: filer_pb.SeaweedFiler.AppendToEntry:input_type -> filer_pb.AppendToEntryRequest
-	19, // 37: filer_pb.SeaweedFiler.DeleteEntry:input_type -> filer_pb.DeleteEntryRequest
-	21, // 38: filer_pb.SeaweedFiler.AtomicRenameEntry:input_type -> filer_pb.AtomicRenameEntryRequest
-	23, // 39: filer_pb.SeaweedFiler.StreamRenameEntry:input_type -> filer_pb.StreamRenameEntryRequest
-	25, // 40: filer_pb.SeaweedFiler.AssignVolume:input_type -> filer_pb.AssignVolumeRequest
-	27, // 41: filer_pb.SeaweedFiler.LookupVolume:input_type -> filer_pb.LookupVolumeRequest
-	32, // 42: filer_pb.SeaweedFiler.CollectionList:input_type -> filer_pb.CollectionListRequest
-	34, // 43: filer_pb.SeaweedFiler.DeleteCollection:input_type -> filer_pb.DeleteCollectionRequest
-	36, // 44: filer_pb.SeaweedFiler.Statistics:input_type -> filer_pb.StatisticsRequest
-	38, // 45: filer_pb.SeaweedFiler.Ping:input_type -> filer_pb.PingRequest
-	40, // 46: filer_pb.SeaweedFiler.GetFilerConfiguration:input_type -> filer_pb.GetFilerConfigurationRequest
-	44, // 47: filer_pb.SeaweedFiler.TraverseBfsMetadata:input_type -> filer_pb.TraverseBfsMetadataRequest
-	42, // 48: filer_pb.SeaweedFiler.SubscribeMetadata:input_type -> filer_pb.SubscribeMetadataRequest
-	42, // 49: filer_pb.SeaweedFiler.SubscribeLocalMetadata:input_type -> filer_pb.SubscribeMetadataRequest
-	51, // 50: filer_pb.SeaweedFiler.KvGet:input_type -> filer_pb.KvGetRequest
-	53, // 51: filer_pb.SeaweedFiler.KvPut:input_type -> filer_pb.KvPutRequest
-	56, // 52: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:input_type -> filer_pb.CacheRemoteObjectToLocalClusterRequest
-	58, // 53: filer_pb.SeaweedFiler.DistributedLock:input_type -> filer_pb.LockRequest
-	60, // 54: filer_pb.SeaweedFiler.DistributedUnlock:input_type -> filer_pb.UnlockRequest
-	62, // 55: filer_pb.SeaweedFiler.FindLockOwner:input_type -> filer_pb.FindLockOwnerRequest
-	65, // 56: filer_pb.SeaweedFiler.TransferLocks:input_type -> filer_pb.TransferLocksRequest
-	2,  // 57: filer_pb.SeaweedFiler.LookupDirectoryEntry:output_type -> filer_pb.LookupDirectoryEntryResponse
-	4,  // 58: filer_pb.SeaweedFiler.ListEntries:output_type -> filer_pb.ListEntriesResponse
-	14, // 59: filer_pb.SeaweedFiler.CreateEntry:output_type -> filer_pb.CreateEntryResponse
-	16, // 60: filer_pb.SeaweedFiler.UpdateEntry:output_type -> filer_pb.UpdateEntryResponse
-	18, // 61: filer_pb.SeaweedFiler.AppendToEntry:output_type -> filer_pb.AppendToEntryResponse
-	20, // 62: filer_pb.SeaweedFiler.DeleteEntry:output_type -> filer_pb.DeleteEntryResponse
-	22, // 63: filer_pb.SeaweedFiler.AtomicRenameEntry:output_type -> filer_pb.AtomicRenameEntryResponse
-	24, // 64: filer_pb.SeaweedFiler.StreamRenameEntry:output_type -> filer_pb.StreamRenameEntryResponse
-	26, // 65: filer_pb.SeaweedFiler.AssignVolume:output_type -> filer_pb.AssignVolumeResponse
-	30, // 66: filer_pb.SeaweedFiler.LookupVolume:output_type -> filer_pb.LookupVolumeResponse
-	33, // 67: filer_pb.SeaweedFiler.CollectionList:output_type -> filer_pb.CollectionListResponse
-	35, // 68: filer_pb.SeaweedFiler.DeleteCollection:output_type -> filer_pb.DeleteCollectionResponse
-	37, // 69: filer_pb.SeaweedFiler.Statistics:output_type -> filer_pb.StatisticsResponse
-	39, // 70: filer_pb.SeaweedFiler.Ping:output_type -> filer_pb.PingResponse
-	41, // 71: filer_pb.SeaweedFiler.GetFilerConfiguration:output_type -> filer_pb.GetFilerConfigurationResponse
-	45, // 72: filer_pb.SeaweedFiler.TraverseBfsMetadata:output_type -> filer_pb.TraverseBfsMetadataResponse
-	43, // 73: filer_pb.SeaweedFiler.SubscribeMetadata:output_type -> filer_pb.SubscribeMetadataResponse
-	43, // 74: filer_pb.SeaweedFiler.SubscribeLocalMetadata:output_type -> filer_pb.SubscribeMetadataResponse
-	52, // 75: filer_pb.SeaweedFiler.KvGet:output_type -> filer_pb.KvGetResponse
-	54, // 76: filer_pb.SeaweedFiler.KvPut:output_type -> filer_pb.KvPutResponse
-	57, // 77: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:output_type -> filer_pb.CacheRemoteObjectToLocalClusterResponse
-	59, // 78: filer_pb.SeaweedFiler.DistributedLock:output_type -> filer_pb.LockResponse
-	61, // 79: filer_pb.SeaweedFiler.DistributedUnlock:output_type -> filer_pb.UnlockResponse
-	63, // 80: filer_pb.SeaweedFiler.FindLockOwner:output_type -> filer_pb.FindLockOwnerResponse
-	66, // 81: filer_pb.SeaweedFiler.TransferLocks:output_type -> filer_pb.TransferLocksResponse
-	57, // [57:82] is the sub-list for method output_type
-	32, // [32:57] is the sub-list for method input_type
-	32, // [32:32] is the sub-list for extension type_name
-	32, // [32:32] is the sub-list for extension extendee
-	0,  // [0:32] is the sub-list for field type_name
+	68, // 16: filer_pb.UpdateEntryRequest.expected_extended:type_name -> filer_pb.UpdateEntryRequest.ExpectedExtendedEntry
+	43, // 17: filer_pb.UpdateEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse
+	9,  // 18: filer_pb.AppendToEntryRequest.chunks:type_name -> filer_pb.FileChunk
+	43, // 19: filer_pb.DeleteEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse
+	8,  // 20: filer_pb.StreamRenameEntryResponse.event_notification:type_name -> filer_pb.EventNotification
+	29, // 21: filer_pb.AssignVolumeResponse.location:type_name -> filer_pb.Location
+	29, // 22: filer_pb.Locations.locations:type_name -> filer_pb.Location
+	69, // 23: filer_pb.LookupVolumeResponse.locations_map:type_name -> filer_pb.LookupVolumeResponse.LocationsMapEntry
+	31, // 24: filer_pb.CollectionListResponse.collections:type_name -> filer_pb.Collection
+	8,  // 25: filer_pb.SubscribeMetadataResponse.event_notification:type_name -> filer_pb.EventNotification
+	6,  // 26: filer_pb.TraverseBfsMetadataResponse.entry:type_name -> filer_pb.Entry
+	70, // 27: filer_pb.LocateBrokerResponse.resources:type_name -> filer_pb.LocateBrokerResponse.Resource
+	71, // 28: filer_pb.FilerConf.locations:type_name -> filer_pb.FilerConf.PathConf
+	6,  // 29: filer_pb.CacheRemoteObjectToLocalClusterResponse.entry:type_name -> filer_pb.Entry
+	43, // 30: filer_pb.CacheRemoteObjectToLocalClusterResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse
+	64, // 31: filer_pb.TransferLocksRequest.locks:type_name -> filer_pb.Lock
+	28, // 32: filer_pb.LookupVolumeResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations
+	1,  // 33: filer_pb.SeaweedFiler.LookupDirectoryEntry:input_type -> filer_pb.LookupDirectoryEntryRequest
+	3,  // 34: filer_pb.SeaweedFiler.ListEntries:input_type -> filer_pb.ListEntriesRequest
+	13, // 35: filer_pb.SeaweedFiler.CreateEntry:input_type -> filer_pb.CreateEntryRequest
+	15, // 36: filer_pb.SeaweedFiler.UpdateEntry:input_type -> filer_pb.UpdateEntryRequest
+	17, // 37: filer_pb.SeaweedFiler.AppendToEntry:input_type -> filer_pb.AppendToEntryRequest
+	19, // 38: filer_pb.SeaweedFiler.DeleteEntry:input_type -> filer_pb.DeleteEntryRequest
+	21, // 39: filer_pb.SeaweedFiler.AtomicRenameEntry:input_type -> filer_pb.AtomicRenameEntryRequest
+	23, // 40: filer_pb.SeaweedFiler.StreamRenameEntry:input_type -> filer_pb.StreamRenameEntryRequest
+	25, // 41: filer_pb.SeaweedFiler.AssignVolume:input_type -> filer_pb.AssignVolumeRequest
+	27, // 42: filer_pb.SeaweedFiler.LookupVolume:input_type -> filer_pb.LookupVolumeRequest
+	32, // 43: filer_pb.SeaweedFiler.CollectionList:input_type -> filer_pb.CollectionListRequest
+	34, // 44: filer_pb.SeaweedFiler.DeleteCollection:input_type -> filer_pb.DeleteCollectionRequest
+	36, // 45: filer_pb.SeaweedFiler.Statistics:input_type -> filer_pb.StatisticsRequest
+	38, // 46: filer_pb.SeaweedFiler.Ping:input_type -> filer_pb.PingRequest
+	40, // 47: filer_pb.SeaweedFiler.GetFilerConfiguration:input_type -> filer_pb.GetFilerConfigurationRequest
+	44, // 48: filer_pb.SeaweedFiler.TraverseBfsMetadata:input_type -> filer_pb.TraverseBfsMetadataRequest
+	42, // 49: filer_pb.SeaweedFiler.SubscribeMetadata:input_type -> filer_pb.SubscribeMetadataRequest
+	42, // 50: filer_pb.SeaweedFiler.SubscribeLocalMetadata:input_type -> filer_pb.SubscribeMetadataRequest
+	51, // 51: filer_pb.SeaweedFiler.KvGet:input_type -> filer_pb.KvGetRequest
+	53, // 52: filer_pb.SeaweedFiler.KvPut:input_type -> filer_pb.KvPutRequest
+	56, // 53: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:input_type -> filer_pb.CacheRemoteObjectToLocalClusterRequest
+	58, // 54: filer_pb.SeaweedFiler.DistributedLock:input_type -> filer_pb.LockRequest
+	60, // 55: filer_pb.SeaweedFiler.DistributedUnlock:input_type -> filer_pb.UnlockRequest
+	62, // 56: filer_pb.SeaweedFiler.FindLockOwner:input_type -> filer_pb.FindLockOwnerRequest
+	65, // 57: filer_pb.SeaweedFiler.TransferLocks:input_type -> filer_pb.TransferLocksRequest
+	2,  // 58: filer_pb.SeaweedFiler.LookupDirectoryEntry:output_type -> filer_pb.LookupDirectoryEntryResponse
+	4,  // 59: filer_pb.SeaweedFiler.ListEntries:output_type -> filer_pb.ListEntriesResponse
+	14, // 60: filer_pb.SeaweedFiler.CreateEntry:output_type -> filer_pb.CreateEntryResponse
+	16, // 61: filer_pb.SeaweedFiler.UpdateEntry:output_type -> filer_pb.UpdateEntryResponse
+	18, // 62: filer_pb.SeaweedFiler.AppendToEntry:output_type -> filer_pb.AppendToEntryResponse
+	20, // 63: filer_pb.SeaweedFiler.DeleteEntry:output_type -> filer_pb.DeleteEntryResponse
+	22, // 64: filer_pb.SeaweedFiler.AtomicRenameEntry:output_type -> filer_pb.AtomicRenameEntryResponse
+	24, // 65: filer_pb.SeaweedFiler.StreamRenameEntry:output_type -> filer_pb.StreamRenameEntryResponse
+	26, // 66: filer_pb.SeaweedFiler.AssignVolume:output_type -> filer_pb.AssignVolumeResponse
+	30, // 67: filer_pb.SeaweedFiler.LookupVolume:output_type -> filer_pb.LookupVolumeResponse
+	33, // 68: filer_pb.SeaweedFiler.CollectionList:output_type -> filer_pb.CollectionListResponse
+	35, // 69: filer_pb.SeaweedFiler.DeleteCollection:output_type -> filer_pb.DeleteCollectionResponse
+	37, // 70: filer_pb.SeaweedFiler.Statistics:output_type -> filer_pb.StatisticsResponse
+	39, // 71: filer_pb.SeaweedFiler.Ping:output_type -> filer_pb.PingResponse
+	41, // 72: filer_pb.SeaweedFiler.GetFilerConfiguration:output_type -> filer_pb.GetFilerConfigurationResponse
+	45, // 73: filer_pb.SeaweedFiler.TraverseBfsMetadata:output_type -> filer_pb.TraverseBfsMetadataResponse
+	43, // 74: filer_pb.SeaweedFiler.SubscribeMetadata:output_type -> filer_pb.SubscribeMetadataResponse
+	43, // 75: filer_pb.SeaweedFiler.SubscribeLocalMetadata:output_type -> filer_pb.SubscribeMetadataResponse
+	52, // 76: filer_pb.SeaweedFiler.KvGet:output_type -> filer_pb.KvGetResponse
+	54, // 77: filer_pb.SeaweedFiler.KvPut:output_type -> filer_pb.KvPutResponse
+	57, // 78: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:output_type -> filer_pb.CacheRemoteObjectToLocalClusterResponse
+	59, // 79: filer_pb.SeaweedFiler.DistributedLock:output_type -> filer_pb.LockResponse
+	61, // 80: filer_pb.SeaweedFiler.DistributedUnlock:output_type -> filer_pb.UnlockResponse
+	63, // 81: filer_pb.SeaweedFiler.FindLockOwner:output_type -> filer_pb.FindLockOwnerResponse
+	66, // 82: filer_pb.SeaweedFiler.TransferLocks:output_type -> filer_pb.TransferLocksResponse
+	58, // [58:83] is the sub-list for method output_type
+	33, // [33:58] is the sub-list for method input_type
+	33, // [33:33] is the sub-list for extension type_name
+	33, // [33:33] is the sub-list for extension extendee
+	0,  // [0:33] is the sub-list for field type_name
 }
 
 func init() { file_filer_proto_init() }
@@ -5040,7 +5054,7 @@ func file_filer_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_filer_proto_rawDesc), len(file_filer_proto_rawDesc)),
 			NumEnums:      1,
-			NumMessages:   70,
+			NumMessages:   71,
 			NumExtensions: 0,
 			NumServices:   1,
 		},

weed/plugin/worker/iceberg/config.go

@@ -23,15 +23,18 @@ const (
 	defaultOperations             = "all"
 
 	// Metric keys returned by maintenance operations.
-	MetricFilesMerged        = "files_merged"
-	MetricFilesWritten       = "files_written"
-	MetricBins               = "bins"
-	MetricSnapshotsExpired   = "snapshots_expired"
-	MetricFilesDeleted       = "files_deleted"
-	MetricOrphansRemoved     = "orphans_removed"
-	MetricManifestsRewritten = "manifests_rewritten"
-	MetricEntriesTotal       = "entries_total"
-	MetricDurationMs         = "duration_ms"
+	MetricFilesMerged          = "files_merged"
+	MetricFilesWritten         = "files_written"
+	MetricBins                 = "bins"
+	MetricSnapshotsExpired     = "snapshots_expired"
+	MetricSnapshotRefsRemoved  = "snapshot_refs_removed"
+	MetricFilesDeleted         = "files_deleted"
+	MetricOrphansRemoved       = "orphans_removed"
+	MetricOrphansFound         = "orphans_found"
+	MetricMetadataFilesDeleted = "metadata_files_deleted"
+	MetricManifestsRewritten   = "manifests_rewritten"
+	MetricEntriesTotal         = "entries_total"
+	MetricDurationMs           = "duration_ms"
 )
 
 // Config holds parsed worker config values.
@@ -45,6 +48,8 @@ type Config struct {
 	MinManifestsToRewrite  int64
 	Operations             string
 	ApplyDeletes           bool
+	RemoveOrphansDryRun    bool
+	CleanExpiredMetadata   bool
 }
 
 // ParseConfig extracts an iceberg maintenance Config from plugin config values.
@@ -60,6 +65,8 @@ func ParseConfig(values map[string]*plugin_pb.ConfigValue) Config {
 		MinManifestsToRewrite:  readInt64Config(values, "min_manifests_to_rewrite", defaultMinManifestsToRewrite),
 		Operations:             readStringConfig(values, "operations", defaultOperations),
 		ApplyDeletes:           readBoolConfig(values, "apply_deletes", true),
+		RemoveOrphansDryRun:    readBoolConfig(values, "remove_orphans_dry_run", false),
+		CleanExpiredMetadata:   readBoolConfig(values, "clean_expired_metadata", false),
 	}
 
 	// Clamp to safe minimums using the default constants

weed/plugin/worker/iceberg/detection.go

@@ -333,20 +333,15 @@ func hasEligibleCompaction(
 	}
 
 	var dataManifests []iceberg.ManifestFile
-	specIDs := make(map[int32]struct{})
 	for _, mf := range manifests {
 		if mf.ManifestContent() != iceberg.ManifestContentData {
 			continue
 		}
 		dataManifests = append(dataManifests, mf)
-		specIDs[mf.PartitionSpecID()] = struct{}{}
 	}
 	if len(dataManifests) == 0 {
 		return false, nil
 	}
-	if len(specIDs) > 1 {
-		return false, nil
-	}
 
 	var allEntries []iceberg.ManifestEntry
 	for _, mf := range dataManifests {

weed/plugin/worker/iceberg/exec_test.go

@@ -24,6 +24,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 )
 
 // ---------------------------------------------------------------------------
@@ -35,8 +36,9 @@ import (
 type fakeFilerServer struct {
 	filer_pb.UnimplementedSeaweedFilerServer
 
-	mu      sync.Mutex
-	entries map[string]map[string]*filer_pb.Entry // dir → name → entry
+	mu           sync.Mutex
+	entries      map[string]map[string]*filer_pb.Entry // dir → name → entry
+	beforeUpdate func(*fakeFilerServer, *filer_pb.UpdateEntryRequest)
 
 	// Counters for assertions
 	createCalls int
@@ -138,9 +140,33 @@ func (f *fakeFilerServer) CreateEntry(_ context.Context, req *filer_pb.CreateEnt
 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()
 
-	f.putEntry(req.Directory, req.Entry.Name, req.Entry)
+	if beforeUpdate != nil {
+		beforeUpdate(f, req)
+	}
+
+	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 || !bytes.Equal(actualValue, expectedValue) {
+			return nil, status.Errorf(codes.FailedPrecondition, "extended attribute %q changed", key)
+		}
+	}
+
+	dirEntries[req.Entry.Name] = req.Entry
 	return &filer_pb.UpdateEntryResponse{}, nil
 }
 
@@ -237,24 +263,6 @@ func populateTable(t *testing.T, fs *fakeFilerServer, setup tableSetup) table.Me
 	t.Helper()
 
 	meta := buildTestMetadata(t, setup.Snapshots)
-	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": path.Join("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)
@@ -280,10 +288,9 @@ func populateTable(t *testing.T, fs *fakeFilerServer, setup tableSetup) table.Me
 	fs.putEntry(nsPath, setup.TableName, &filer_pb.Entry{
 		Name:        setup.TableName,
 		IsDirectory: true,
-		Extended: map[string][]byte{
-			s3tables.ExtendedKeyMetadata: xattr,
-		},
+		Extended:    map[string][]byte{},
 	})
+	setTableMetadataForTest(t, fs, setup, meta, 1, path.Join("metadata", "v1.metadata.json"))
 
 	// Create metadata/ and data/ directory placeholders
 	metaDir := path.Join(tableFilerPath, "metadata")
@@ -372,9 +379,55 @@ func populateTable(t *testing.T, fs *fakeFilerServer, setup tableSetup) table.Me
 	return meta
 }
 
+func setTableMetadataForTest(t *testing.T, fs *fakeFilerServer, setup tableSetup, meta table.Metadata, metadataVersion int, metadataLocation string) {
+	t.Helper()
+
+	fullMetadataJSON, err := json.Marshal(meta)
+	if err != nil {
+		t.Fatalf("marshal metadata: %v", err)
+	}
+
+	internalMeta := map[string]interface{}{
+		"metadataVersion":  metadataVersion,
+		"metadataLocation": metadataLocation,
+		"metadata": map[string]interface{}{
+			"fullMetadata": json.RawMessage(fullMetadataJSON),
+		},
+	}
+	xattr, err := json.Marshal(internalMeta)
+	if err != nil {
+		t.Fatalf("marshal xattr: %v", err)
+	}
+
+	tableDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.Namespace)
+	entry := fs.getEntry(tableDir, setup.TableName)
+	if entry == nil {
+		t.Fatalf("table entry not found: %s/%s", tableDir, setup.TableName)
+	}
+	updatedEntry := cloneEntryForTest(t, entry)
+	if updatedEntry.Extended == nil {
+		updatedEntry.Extended = make(map[string][]byte)
+	}
+	updatedEntry.Extended[s3tables.ExtendedKeyMetadata] = xattr
+	fs.putEntry(tableDir, setup.TableName, updatedEntry)
+}
+
 func writeCurrentSnapshotManifests(t *testing.T, fs *fakeFilerServer, setup tableSetup, meta table.Metadata, manifestEntries [][]iceberg.ManifestEntry) {
 	t.Helper()
 
+	specs := make([]iceberg.PartitionSpec, len(manifestEntries))
+	for i := range specs {
+		specs[i] = meta.PartitionSpec()
+	}
+	writeCurrentSnapshotManifestsWithSpecs(t, fs, setup, meta, specs, manifestEntries)
+}
+
+func writeCurrentSnapshotManifestsWithSpecs(t *testing.T, fs *fakeFilerServer, setup tableSetup, meta table.Metadata, specs []iceberg.PartitionSpec, manifestEntries [][]iceberg.ManifestEntry) {
+	t.Helper()
+	if len(specs) != len(manifestEntries) {
+		t.Fatalf("spec count %d does not match manifest count %d", len(specs), len(manifestEntries))
+	}
+
 	currentSnap := meta.CurrentSnapshot()
 	if currentSnap == nil {
 		t.Fatal("current snapshot is required")
@@ -383,7 +436,6 @@ func writeCurrentSnapshotManifests(t *testing.T, fs *fakeFilerServer, setup tabl
 	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 {
@@ -393,7 +445,7 @@ func writeCurrentSnapshotManifests(t *testing.T, fs *fakeFilerServer, setup tabl
 			path.Join("metadata", manifestName),
 			&manifestBuf,
 			version,
-			spec,
+			specs[i],
 			schema,
 			currentSnap.SnapshotID,
 			entries,
@@ -533,6 +585,118 @@ func TestExpireSnapshotsNothingToExpire(t *testing.T) {
 	}
 }
 
+func TestExpireSnapshotsPreservesTaggedSnapshot(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 - 3_000},
+			{SnapshotID: 2, TimestampMs: now - 2_000},
+			{SnapshotID: 3, TimestampMs: now - 1_000},
+		},
+	}
+	meta := populateTable(t, fs, setup)
+
+	builder, err := table.MetadataBuilderFromBase(meta, "s3://test-bucket/test-table")
+	if err != nil {
+		t.Fatalf("metadata builder: %v", err)
+	}
+	if err := builder.SetSnapshotRef("audit-tag", 1, table.TagRef, table.WithMaxRefAgeMs(24*3600*1000)); err != nil {
+		t.Fatalf("set snapshot ref: %v", err)
+	}
+	taggedMeta, err := builder.Build()
+	if err != nil {
+		t.Fatalf("build tagged metadata: %v", err)
+	}
+	setTableMetadataForTest(t, fs, setup, taggedMeta, 1, path.Join("metadata", "v1.metadata.json"))
+
+	handler := NewHandler(nil)
+	config := Config{
+		SnapshotRetentionHours: 0,
+		MaxSnapshotsToKeep:     1,
+		MaxCommitRetries:       3,
+	}
+
+	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 1 snapshot") {
+		t.Fatalf("expected one snapshot to expire, got %q", result)
+	}
+
+	currentMeta, _, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
+	if err != nil {
+		t.Fatalf("loadCurrentMetadata: %v", err)
+	}
+	if currentMeta.SnapshotByID(1) == nil {
+		t.Fatal("expected tagged snapshot 1 to be preserved")
+	}
+	if currentMeta.SnapshotByID(2) != nil {
+		t.Fatal("expected unreferenced snapshot 2 to expire")
+	}
+	refs := snapshotRefs(currentMeta)
+	if ref, ok := refs["audit-tag"]; !ok || ref.SnapshotID != 1 {
+		t.Fatalf("expected audit-tag to remain on snapshot 1, got %+v", refs["audit-tag"])
+	}
+}
+
+func TestExpireSnapshotsCleansExpiredMetadata(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)
+
+	metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata")
+	fs.putEntry(metaDir, "v1.metadata.json", &filer_pb.Entry{
+		Name:       "v1.metadata.json",
+		Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Add(-time.Hour).Unix()},
+		Content:    []byte(`{"current":true}`),
+	})
+	fs.putEntry(metaDir, "v0.metadata.json", &filer_pb.Entry{
+		Name:       "v0.metadata.json",
+		Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Add(-2 * time.Hour).Unix()},
+		Content:    []byte(`{"stale":true}`),
+	})
+
+	handler := NewHandler(nil)
+	config := Config{
+		SnapshotRetentionHours: 24 * 365,
+		MaxSnapshotsToKeep:     10,
+		MaxCommitRetries:       3,
+		CleanExpiredMetadata:   true,
+	}
+
+	result, metrics, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config)
+	if err != nil {
+		t.Fatalf("expireSnapshots failed: %v", err)
+	}
+	if !strings.Contains(result, "deleted 1 expired metadata file") {
+		t.Fatalf("expected expired metadata cleanup, got %q", result)
+	}
+	if got := metrics[MetricMetadataFilesDeleted]; got != 1 {
+		t.Fatalf("expected one expired metadata file deleted, got %d", got)
+	}
+	if fs.getEntry(metaDir, "v0.metadata.json") != nil {
+		t.Fatal("expected stale metadata file to be deleted")
+	}
+	if fs.getEntry(metaDir, "v1.metadata.json") == nil {
+		t.Fatal("expected current metadata file to be preserved")
+	}
+}
+
 func TestRemoveOrphansExecution(t *testing.T) {
 	fs, client := startFakeFiler(t)
 
@@ -594,6 +758,52 @@ func TestRemoveOrphansExecution(t *testing.T) {
 	}
 }
 
+func TestRemoveOrphansDryRun(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)
+
+	dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data")
+	oldTime := time.Now().Add(-200 * time.Hour).Unix()
+	fs.putEntry(dataDir, "orphan-data.parquet", &filer_pb.Entry{
+		Name:       "orphan-data.parquet",
+		Attributes: &filer_pb.FuseAttributes{Mtime: oldTime},
+	})
+
+	handler := NewHandler(nil)
+	config := Config{
+		OrphanOlderThanHours: 72,
+		MaxCommitRetries:     3,
+		RemoveOrphansDryRun:  true,
+	}
+
+	result, metrics, err := handler.removeOrphans(context.Background(), client, setup.BucketName, setup.tablePath(), config)
+	if err != nil {
+		t.Fatalf("removeOrphans failed: %v", err)
+	}
+	if !strings.Contains(result, "dry run") {
+		t.Fatalf("expected dry-run summary, got %q", result)
+	}
+	if got := metrics[MetricOrphansFound]; got != 1 {
+		t.Fatalf("expected one orphan found, got %d", got)
+	}
+	if got := metrics[MetricOrphansRemoved]; got != 0 {
+		t.Fatalf("expected zero orphan deletions, got %d", got)
+	}
+	if fs.getEntry(dataDir, "orphan-data.parquet") == nil {
+		t.Fatal("expected orphan file to remain during dry run")
+	}
+}
+
 func TestRemoveOrphansPreservesReferencedFiles(t *testing.T) {
 	fs, client := startFakeFiler(t)
 
@@ -1140,6 +1350,58 @@ func TestDetectSchedulesCompactionWithDeleteManifestPresent(t *testing.T) {
 	}
 }
 
+func TestDetectSchedulesCompactionForSpecEvolvedTable(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)
+
+	spec0 := meta.PartitionSpec()
+	spec1 := iceberg.NewPartitionSpecID(1)
+	partSpecs := map[int32]iceberg.PartitionSpec{
+		0: spec0,
+		1: spec1,
+	}
+	writeCurrentSnapshotManifestsWithSpecs(t, fs, setup, meta,
+		[]iceberg.PartitionSpec{spec0, spec1},
+		[][]iceberg.ManifestEntry{
+			makeTestEntriesWithSpec(t, []testEntrySpec{
+				{path: "data/spec0-small-1.parquet", size: 1024, partition: map[int]any{}, specID: 0},
+				{path: "data/spec0-small-2.parquet", size: 1024, partition: map[int]any{}, specID: 0},
+			}, partSpecs),
+			makeTestEntriesWithSpec(t, []testEntrySpec{
+				{path: "data/spec1-small-1.parquet", size: 1024, partition: map[int]any{}, specID: 1},
+				{path: "data/spec1-small-2.parquet", size: 1024, partition: map[int]any{}, specID: 1},
+			}, partSpecs),
+		},
+	)
+
+	handler := NewHandler(nil)
+	config := Config{
+		SnapshotRetentionHours: 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 for spec-evolved table, got %d", len(tables))
+	}
+}
+
 func TestDetectSchedulesSnapshotExpiryDespiteCompactionEvaluationError(t *testing.T) {
 	fs, client := startFakeFiler(t)
 
@@ -1486,6 +1748,66 @@ func TestMetadataVersionCAS(t *testing.T) {
 	}
 }
 
+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) {
+		entry := f.getEntry(path.Dir(tableDir), path.Base(tableDir))
+		if entry == nil {
+			t.Fatal("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 {
+			t.Fatalf("unmarshal xattr: %v", err)
+		}
+
+		versionJSON, err := json.Marshal(2)
+		if err != nil {
+			t.Fatalf("marshal version: %v", err)
+		}
+		internalMeta["metadataVersion"] = versionJSON
+
+		updatedXattr, err := json.Marshal(internalMeta)
+		if err != nil {
+			t.Fatalf("marshal xattr: %v", err)
+		}
+		updatedEntry.Extended[s3tables.ExtendedKeyMetadata] = updatedXattr
+		f.putEntry(path.Dir(tableDir), path.Base(tableDir), updatedEntry)
+	}
+
+	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
 // ---------------------------------------------------------------------------

weed/plugin/worker/iceberg/filer_io.go

@@ -280,9 +280,10 @@ func deleteFilerFile(ctx context.Context, client filer_pb.SeaweedFilerClient, di
 }
 
 // updateTableMetadataXattr updates the table entry's metadata xattr with
-// the new Iceberg metadata. It performs a compare-and-swap: if the stored
-// metadataVersion does not match expectedVersion, it returns
-// errMetadataVersionConflict so the caller can retry.
+// the new Iceberg metadata. It verifies the stored metadataVersion before
+// writing and passes the previous metadata xattr back to the filer as a
+// server-side precondition so concurrent writers fail with a retryable
+// metadata version conflict.
 // newMetadataLocation is the table-relative path to the new metadata file
 // (e.g. "metadata/v3.metadata.json").
 func updateTableMetadataXattr(ctx context.Context, client filer_pb.SeaweedFilerClient, tableDir string, expectedVersion int, newFullMetadata []byte, newMetadataLocation string) error {
@@ -311,13 +312,8 @@ func updateTableMetadataXattr(ctx context.Context, client filer_pb.SeaweedFilerC
 		return fmt.Errorf("unmarshal existing xattr: %w", err)
 	}
 
-	// Compare-and-swap: verify the stored metadataVersion matches what we expect.
-	// NOTE: This is a client-side CAS — two workers could both read the same
-	// version, pass this check, and race at UpdateEntry (last-write-wins).
-	// The proper fix is server-side precondition support on UpdateEntryRequest
-	// (e.g. expect-version or If-Match semantics). Until then, commitWithRetry
-	// with exponential backoff mitigates but does not eliminate the race.
-	// Avoid scheduling concurrent maintenance on the same table.
+	// Verify the stored metadataVersion matches what we expect before issuing
+	// the conditional UpdateEntry request below.
 	versionRaw, ok := internalMeta["metadataVersion"]
 	if !ok {
 		return fmt.Errorf("%w: metadataVersion field missing from xattr", errMetadataVersionConflict)
@@ -372,8 +368,14 @@ func updateTableMetadataXattr(ctx context.Context, client filer_pb.SeaweedFilerC
 	_, err = client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{
 		Directory: parentDir,
 		Entry:     resp.Entry,
+		ExpectedExtended: map[string][]byte{
+			s3tables.ExtendedKeyMetadata: existingXattr,
+		},
 	})
 	if err != nil {
+		if status.Code(err) == codes.FailedPrecondition {
+			return fmt.Errorf("%w: table metadata changed during update", errMetadataVersionConflict)
+		}
 		return fmt.Errorf("update table entry: %w", err)
 	}
 	return nil

weed/plugin/worker/iceberg/handler.go

@@ -129,6 +129,13 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor {
 							Widget:      plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER,
 							MinValue:    &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1}},
 						},
+						{
+							Name:        "clean_expired_metadata",
+							Label:       "Clean Expired Metadata",
+							Description: "When true, delete older unreferenced metadata JSON files after snapshot expiration planning finishes.",
+							FieldType:   plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_BOOL,
+							Widget:      plugin_pb.ConfigWidget_CONFIG_WIDGET_TOGGLE,
+						},
 					},
 				},
 				{
@@ -174,6 +181,13 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor {
 							Widget:      plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER,
 							MinValue:    &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1}},
 						},
+						{
+							Name:        "remove_orphans_dry_run",
+							Label:       "Dry Run",
+							Description: "When true, report orphan files without deleting them.",
+							FieldType:   plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_BOOL,
+							Widget:      plugin_pb.ConfigWidget_CONFIG_WIDGET_TOGGLE,
+						},
 					},
 				},
 				{
@@ -225,6 +239,8 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor {
 				"max_commit_retries":       {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxCommitRetries}},
 				"operations":               {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: defaultOperations}},
 				"apply_deletes":            {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: true}},
+				"remove_orphans_dry_run":   {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: false}},
+				"clean_expired_metadata":   {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: false}},
 			},
 		},
 		AdminRuntimeDefaults: &plugin_pb.AdminRuntimeDefaults{
@@ -247,6 +263,8 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor {
 			"max_commit_retries":       {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxCommitRetries}},
 			"operations":               {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: defaultOperations}},
 			"apply_deletes":            {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: true}},
+			"remove_orphans_dry_run":   {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: false}},
+			"clean_expired_metadata":   {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: false}},
 		},
 	}
 }

weed/plugin/worker/iceberg/handler_test.go

@@ -761,6 +761,27 @@ func TestParseConfigApplyDeletes(t *testing.T) {
 	}
 }
 
+func TestParseConfigLifecycleFlags(t *testing.T) {
+	config := ParseConfig(nil)
+	if config.RemoveOrphansDryRun {
+		t.Error("expected RemoveOrphansDryRun=false by default")
+	}
+	if config.CleanExpiredMetadata {
+		t.Error("expected CleanExpiredMetadata=false by default")
+	}
+
+	config = ParseConfig(map[string]*plugin_pb.ConfigValue{
+		"remove_orphans_dry_run": {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: true}},
+		"clean_expired_metadata": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: "true"}},
+	})
+	if !config.RemoveOrphansDryRun {
+		t.Error("expected RemoveOrphansDryRun=true when explicitly set")
+	}
+	if !config.CleanExpiredMetadata {
+		t.Error("expected CleanExpiredMetadata=true when set via string 'true'")
+	}
+}
+
 func TestCollectPositionDeletes(t *testing.T) {
 	fs, client := startFakeFiler(t)
 

weed/plugin/worker/iceberg/operations.go

@@ -8,7 +8,6 @@ import (
 	"fmt"
 	"math/rand/v2"
 	"path"
-	"sort"
 	"strings"
 	"time"
 
@@ -51,49 +50,11 @@ func (h *Handler) expireSnapshots(
 		return "no snapshots", nil, nil
 	}
 
-	// Determine which snapshots to expire
-	currentSnap := meta.CurrentSnapshot()
-	var currentSnapID int64
-	if currentSnap != nil {
-		currentSnapID = currentSnap.SnapshotID
-	}
-
-	retentionMs := config.SnapshotRetentionHours * 3600 * 1000
-	nowMs := time.Now().UnixMilli()
-
-	// Sort snapshots by timestamp descending (most recent first) so that
-	// the keep-count logic always preserves the newest snapshots.
-	sorted := make([]table.Snapshot, len(snapshots))
-	copy(sorted, snapshots)
-	sort.Slice(sorted, func(i, j int) bool {
-		return sorted[i].TimestampMs > sorted[j].TimestampMs
-	})
-
-	// Walk from newest to oldest. The current snapshot is always kept.
-	// Among the remaining, keep up to MaxSnapshotsToKeep-1 (since current
-	// counts toward the quota). Expire the rest only if they exceed the
-	// retention window; snapshots within the window are kept regardless.
-	var toExpire []int64
-	var kept int64
-	for _, snap := range sorted {
-		if snap.SnapshotID == currentSnapID {
-			kept++
-			continue
-		}
-		age := nowMs - snap.TimestampMs
-		if kept < config.MaxSnapshotsToKeep {
-			kept++
-			continue
-		}
-		if age > retentionMs {
-			toExpire = append(toExpire, snap.SnapshotID)
-		} else {
-			kept++
-		}
-	}
-
-	if len(toExpire) == 0 {
-		return "no snapshots expired", nil, nil
+	plannedSnapshotIDs := snapshotIDSet(snapshots)
+	plannedRefs := snapshotRefs(meta)
+	toExpire, refsToRemove, err := planSnapshotExpiration(meta, config, start)
+	if err != nil {
+		return "", nil, fmt.Errorf("plan snapshot expiration: %w", err)
 	}
 
 	// Split snapshots into expired and kept sets
@@ -102,7 +63,7 @@ func (h *Handler) expireSnapshots(
 		expireSet[id] = struct{}{}
 	}
 	var expiredSnaps, keptSnaps []table.Snapshot
-	for _, snap := range sorted {
+	for _, snap := range snapshots {
 		if _, ok := expireSet[snap.SnapshotID]; ok {
 			expiredSnaps = append(expiredSnaps, snap)
 		} else {
@@ -110,15 +71,19 @@ func (h *Handler) expireSnapshots(
 		}
 	}
 
-	// Collect all files referenced by each set before modifying metadata.
-	// This lets us determine which files become unreferenced.
-	expiredFiles, err := collectSnapshotFiles(ctx, filerClient, bucketName, tablePath, expiredSnaps)
-	if err != nil {
-		return "", nil, fmt.Errorf("collect expired snapshot files: %w", err)
-	}
-	keptFiles, err := collectSnapshotFiles(ctx, filerClient, bucketName, tablePath, keptSnaps)
-	if err != nil {
-		return "", nil, fmt.Errorf("collect kept snapshot files: %w", err)
+	expiredFiles := make(map[string]struct{})
+	keptFiles := make(map[string]struct{})
+	if len(toExpire) > 0 {
+		// Collect all files referenced by each set before modifying metadata.
+		// This lets us determine which files become unreferenced.
+		expiredFiles, err = collectSnapshotFiles(ctx, filerClient, bucketName, tablePath, expiredSnaps)
+		if err != nil {
+			return "", nil, fmt.Errorf("collect expired snapshot files: %w", err)
+		}
+		keptFiles, err = collectSnapshotFiles(ctx, filerClient, bucketName, tablePath, keptSnaps)
+		if err != nil {
+			return "", nil, fmt.Errorf("collect kept snapshot files: %w", err)
+		}
 	}
 
 	// Normalize kept file paths for consistent comparison
@@ -127,17 +92,25 @@ func (h *Handler) expireSnapshots(
 		normalizedKept[normalizeIcebergPath(f, bucketName, tablePath)] = struct{}{}
 	}
 
-	// Use MetadataBuilder to remove snapshots and create new metadata
-	err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, metadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error {
-		// Guard: verify table head hasn't changed since we planned
-		cs := currentMeta.CurrentSnapshot()
-		if (cs == nil) != (currentSnapID == 0) || (cs != nil && cs.SnapshotID != currentSnapID) {
-			return errStalePlan
+	if len(toExpire) > 0 || len(refsToRemove) > 0 {
+		// Use MetadataBuilder to remove refs/snapshots and create new metadata.
+		err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, metadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error {
+			if !sameSnapshotSet(currentMeta, plannedSnapshotIDs) || !sameSnapshotRefs(currentMeta, plannedRefs) {
+				return errStalePlan
+			}
+			for _, refName := range refsToRemove {
+				if err := builder.RemoveSnapshotRef(refName); err != nil {
+					return err
+				}
+			}
+			if len(toExpire) == 0 {
+				return nil
+			}
+			return builder.RemoveSnapshots(toExpire)
+		})
+		if err != nil {
+			return "", nil, fmt.Errorf("commit snapshot expiration: %w", err)
 		}
-		return builder.RemoveSnapshots(toExpire)
-	})
-	if err != nil {
-		return "", nil, fmt.Errorf("commit snapshot expiration: %w", err)
 	}
 
 	// Delete files exclusively referenced by expired snapshots (best-effort)
@@ -157,12 +130,175 @@ func (h *Handler) expireSnapshots(
 		}
 	}
 
-	metrics := map[string]int64{
-		MetricSnapshotsExpired: int64(len(toExpire)),
-		MetricFilesDeleted:     int64(deletedCount),
-		MetricDurationMs:       time.Since(start).Milliseconds(),
+	metadataFilesDeleted := 0
+	if config.CleanExpiredMetadata {
+		metadataFilesDeleted, err = cleanupExpiredMetadataFiles(ctx, filerClient, bucketName, tablePath)
+		if err != nil {
+			return "", nil, fmt.Errorf("clean expired metadata: %w", err)
+		}
 	}
-	return fmt.Sprintf("expired %d snapshot(s), deleted %d unreferenced file(s)", len(toExpire), deletedCount), metrics, nil
+
+	if len(toExpire) == 0 && len(refsToRemove) == 0 && metadataFilesDeleted == 0 {
+		return "no snapshots expired", nil, nil
+	}
+
+	metrics := map[string]int64{
+		MetricSnapshotsExpired:     int64(len(toExpire)),
+		MetricSnapshotRefsRemoved:  int64(len(refsToRemove)),
+		MetricFilesDeleted:         int64(deletedCount),
+		MetricMetadataFilesDeleted: int64(metadataFilesDeleted),
+		MetricDurationMs:           time.Since(start).Milliseconds(),
+	}
+
+	var summaryParts []string
+	if len(toExpire) > 0 {
+		summaryParts = append(summaryParts, fmt.Sprintf("expired %d snapshot(s)", len(toExpire)))
+	}
+	if len(refsToRemove) > 0 {
+		summaryParts = append(summaryParts, fmt.Sprintf("removed %d snapshot ref(s)", len(refsToRemove)))
+	}
+	if len(toExpire) > 0 {
+		summaryParts = append(summaryParts, fmt.Sprintf("deleted %d unreferenced file(s)", deletedCount))
+	}
+	if metadataFilesDeleted > 0 {
+		summaryParts = append(summaryParts, fmt.Sprintf("deleted %d expired metadata file(s)", metadataFilesDeleted))
+	}
+	return strings.Join(summaryParts, ", "), metrics, nil
+}
+
+func planSnapshotExpiration(meta table.Metadata, config Config, now time.Time) ([]int64, []string, error) {
+	retentionMs := config.SnapshotRetentionHours * 3600 * 1000
+	minSnapshotsToKeep, err := snapshotMinKeep(config.MaxSnapshotsToKeep)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	snapsToKeep := make(map[int64]struct{})
+	var refsToRemove []string
+	nowMs := now.UnixMilli()
+	refCount := 0
+
+	for refName, ref := range meta.Refs() {
+		refCount++
+		snap := meta.SnapshotByID(ref.SnapshotID)
+		if snap == nil {
+			return nil, nil, fmt.Errorf("snapshot ref %q points to missing snapshot %d", refName, ref.SnapshotID)
+		}
+
+		maxRefAgeMs := retentionMs
+		if ref.MaxRefAgeMs != nil {
+			maxRefAgeMs = *ref.MaxRefAgeMs
+		}
+		if refName != table.MainBranch && nowMs-snap.TimestampMs > maxRefAgeMs {
+			refsToRemove = append(refsToRemove, refName)
+			continue
+		}
+
+		if ref.SnapshotRefType != table.BranchRef {
+			snapsToKeep[ref.SnapshotID] = struct{}{}
+			continue
+		}
+
+		maxSnapshotAgeMs := retentionMs
+		if ref.MaxSnapshotAgeMs != nil {
+			maxSnapshotAgeMs = *ref.MaxSnapshotAgeMs
+		}
+		refMinSnapshotsToKeep := minSnapshotsToKeep
+		if ref.MinSnapshotsToKeep != nil && *ref.MinSnapshotsToKeep > 0 {
+			refMinSnapshotsToKeep = *ref.MinSnapshotsToKeep
+		}
+
+		ancestorCount := 0
+		snapshotID := ref.SnapshotID
+		for {
+			snap = meta.SnapshotByID(snapshotID)
+			if snap == nil {
+				return nil, nil, fmt.Errorf("snapshot %d not found while traversing ref %q", snapshotID, refName)
+			}
+
+			snapshotAgeMs := nowMs - snap.TimestampMs
+			if snapshotAgeMs > maxSnapshotAgeMs && ancestorCount >= refMinSnapshotsToKeep {
+				break
+			}
+
+			snapsToKeep[snap.SnapshotID] = struct{}{}
+			if snap.ParentSnapshotID == nil {
+				break
+			}
+
+			snapshotID = *snap.ParentSnapshotID
+			ancestorCount++
+		}
+	}
+
+	if refCount == 0 {
+		if currentSnap := meta.CurrentSnapshot(); currentSnap != nil {
+			snapsToKeep[currentSnap.SnapshotID] = struct{}{}
+		}
+	}
+
+	var toExpire []int64
+	for _, snap := range meta.Snapshots() {
+		if _, ok := snapsToKeep[snap.SnapshotID]; !ok {
+			toExpire = append(toExpire, snap.SnapshotID)
+		}
+	}
+
+	return toExpire, refsToRemove, nil
+}
+
+func snapshotMinKeep(maxSnapshotsToKeep int64) (int, error) {
+	if maxSnapshotsToKeep < 1 {
+		return 1, nil
+	}
+	maxInt := int64(^uint(0) >> 1)
+	if maxSnapshotsToKeep > maxInt {
+		return 0, fmt.Errorf("max snapshots to keep %d exceeds platform int size", maxSnapshotsToKeep)
+	}
+	return int(maxSnapshotsToKeep), nil
+}
+
+func snapshotIDSet(snapshots []table.Snapshot) map[int64]struct{} {
+	ids := make(map[int64]struct{}, len(snapshots))
+	for _, snap := range snapshots {
+		ids[snap.SnapshotID] = struct{}{}
+	}
+	return ids
+}
+
+func sameSnapshotSet(meta table.Metadata, expected map[int64]struct{}) bool {
+	current := meta.Snapshots()
+	if len(current) != len(expected) {
+		return false
+	}
+	for _, snap := range current {
+		if _, ok := expected[snap.SnapshotID]; !ok {
+			return false
+		}
+	}
+	return true
+}
+
+func snapshotRefs(meta table.Metadata) map[string]table.SnapshotRef {
+	refs := make(map[string]table.SnapshotRef)
+	for refName, ref := range meta.Refs() {
+		refs[refName] = ref
+	}
+	return refs
+}
+
+func sameSnapshotRefs(meta table.Metadata, expected map[string]table.SnapshotRef) bool {
+	current := snapshotRefs(meta)
+	if len(current) != len(expected) {
+		return false
+	}
+	for refName, expectedRef := range expected {
+		currentRef, ok := current[refName]
+		if !ok || !currentRef.Equals(expectedRef) {
+			return false
+		}
+	}
+	return true
 }
 
 // collectSnapshotFiles returns all file paths (manifest lists, manifest files,
@@ -234,6 +370,15 @@ func (h *Handler) removeOrphans(
 		return "", nil, fmt.Errorf("collect orphan candidates: %w", err)
 	}
 
+	if config.RemoveOrphansDryRun {
+		metrics := map[string]int64{
+			MetricOrphansFound:   int64(len(orphanCandidates)),
+			MetricOrphansRemoved: 0,
+			MetricDurationMs:     time.Since(start).Milliseconds(),
+		}
+		return fmt.Sprintf("found %d orphan file(s) (dry run)", len(orphanCandidates)), metrics, nil
+	}
+
 	orphanCount := 0
 	for _, candidate := range orphanCandidates {
 		if delErr := deleteFilerFile(ctx, filerClient, candidate.Dir, candidate.Entry.Name); delErr != nil {
@@ -244,12 +389,60 @@ func (h *Handler) removeOrphans(
 	}
 
 	metrics := map[string]int64{
+		MetricOrphansFound:   int64(len(orphanCandidates)),
 		MetricOrphansRemoved: int64(orphanCount),
 		MetricDurationMs:     time.Since(start).Milliseconds(),
 	}
 	return fmt.Sprintf("removed %d orphan file(s)", orphanCount), metrics, nil
 }
 
+func cleanupExpiredMetadataFiles(
+	ctx context.Context,
+	filerClient filer_pb.SeaweedFilerClient,
+	bucketName, tablePath string,
+) (int, error) {
+	meta, currentMetadataFileName, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
+	if err != nil {
+		return 0, fmt.Errorf("load metadata: %w", err)
+	}
+
+	referencedFiles := map[string]struct{}{
+		path.Join("metadata", currentMetadataFileName): {},
+	}
+	for entry := range meta.PreviousFiles() {
+		referencedFiles[normalizeIcebergPath(entry.MetadataFile, bucketName, tablePath)] = struct{}{}
+	}
+
+	currentVersion := extractMetadataVersion(currentMetadataFileName)
+	metadataDir := path.Join(s3tables.TablesPath, bucketName, tablePath, "metadata")
+	entries, err := listFilerEntries(ctx, filerClient, metadataDir, "")
+	if err != nil {
+		return 0, fmt.Errorf("list metadata dir: %w", err)
+	}
+
+	deleted := 0
+	for _, entry := range entries {
+		if entry == nil || entry.IsDirectory || !strings.HasSuffix(entry.Name, ".metadata.json") {
+			continue
+		}
+
+		relPath := path.Join("metadata", entry.Name)
+		if _, ok := referencedFiles[relPath]; ok {
+			continue
+		}
+		if extractMetadataVersion(entry.Name) >= currentVersion {
+			continue
+		}
+		if err := deleteFilerFile(ctx, filerClient, metadataDir, entry.Name); err != nil {
+			glog.Warningf("iceberg maintenance: failed to delete expired metadata file %s/%s: %v", metadataDir, entry.Name, err)
+			continue
+		}
+		deleted++
+	}
+
+	return deleted, nil
+}
+
 func collectOrphanCandidates(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,

weed/server/filer_grpc_server.go

@@ -1,6 +1,7 @@
 package weed_server
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"os"
@@ -17,6 +18,8 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 func (fs *FilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
@@ -202,6 +205,9 @@ func (fs *FilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntr
 	if err != nil {
 		return &filer_pb.UpdateEntryResponse{}, fmt.Errorf("not found %s: %v", fullpath, err)
 	}
+	if err := validateUpdateEntryPreconditions(entry, req.ExpectedExtended); err != nil {
+		return &filer_pb.UpdateEntryResponse{}, err
+	}
 
 	chunks, garbage, err2 := fs.cleanupChunks(ctx, fullpath, entry, req.Entry)
 	if err2 != nil {
@@ -235,6 +241,25 @@ func (fs *FilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntr
 	return resp, err
 }
 
+func validateUpdateEntryPreconditions(entry *filer.Entry, expectedExtended map[string][]byte) error {
+	if len(expectedExtended) == 0 {
+		return nil
+	}
+
+	for key, expectedValue := range expectedExtended {
+		var actualValue []byte
+		var ok bool
+		if entry != nil {
+			actualValue, ok = entry.Extended[key]
+		}
+		if !ok || !bytes.Equal(actualValue, expectedValue) {
+			return status.Errorf(codes.FailedPrecondition, "extended attribute %q changed", key)
+		}
+	}
+
+	return nil
+}
+
 func (fs *FilerServer) cleanupChunks(ctx context.Context, fullpath string, existingEntry *filer.Entry, newEntry *filer_pb.Entry) (chunks, garbage []*filer_pb.FileChunk, err error) {
 
 	// remove old chunks if not included in the new ones