(skywalking-banyandb) 04/08: Add PeekOldestSegmentEndTimeInGroup method to data and standalone services

[ha...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

hanahmily pushed a commit to branch tsdb/force-cleanup
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git

commit ce42351d84bc4ae1ee012a4fbda8e1df08884321
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Mon Sep 15 06:58:35 2025 +0800

    Add PeekOldestSegmentEndTimeInGroup method to data and standalone services
    
    This commit introduces the PeekOldestSegmentEndTimeInGroup method to the data and standalone services, allowing retrieval of the end time of the oldest segment in a specified group. The DiskMonitor is updated to utilize this new method for improved segment management. Additionally, the DeleteOldestSegmentInGroup method is enhanced to check for database support before attempting deletion, improving error handling and service robustness.
---
 banyand/internal/storage/disk_monitor.go | 23 ++++++++++++-------
 banyand/measure/svc_data.go              | 38 +++++++++++++++++++++++++++++---
 banyand/measure/svc_standalone.go        | 38 +++++++++++++++++++++++++++++---
 banyand/stream/svc_standalone.go         | 38 +++++++++++++++++++++++++++++---
 banyand/trace/svc_standalone.go          | 38 +++++++++++++++++++++++++++++---
 5 files changed, 155 insertions(+), 20 deletions(-)

diff --git a/banyand/internal/storage/disk_monitor.go b/banyand/internal/storage/disk_monitor.go
index 115a2bd8..11f01c96 100644
--- a/banyand/internal/storage/disk_monitor.go
+++ b/banyand/internal/storage/disk_monitor.go
@@ -38,6 +38,9 @@ type RetentionService interface {
 	GetSnapshotDir() string
 	// LoadAllGroups returns all groups managed by this service
 	LoadAllGroups() []resourceSchema.Group
+	// PeekOldestSegmentEndTimeInGroup returns the end time of the oldest segment in the specified group
+	// Returns zero time and false if no segments exist or group not found
+	PeekOldestSegmentEndTimeInGroup(group string) (time.Time, bool)
 	// DeleteOldestSegmentInGroup deletes the oldest segment in the specified group
 	// Returns true if a segment was deleted, false if no segments to delete
 	DeleteOldestSegmentInGroup(group string) (bool, error)
@@ -289,21 +292,25 @@ func (dm *DiskMonitor) findGroupWithOldestSegment(groups []resourceSchema.Group)
 
 	var candidates []groupSegmentTime
 
-	// For now, we'll implement a simple round-robin approach across groups
-	// since we don't have the storage API extensions yet to peek segment times
-	// TODO: Replace with actual oldest-segment-finding logic when storage APIs are ready
+	// Query each group's oldest segment end time
 	for _, group := range groups {
-		candidates = append(candidates, groupSegmentTime{
-			groupName: group.GetSchema().Metadata.Name,
-			endTime:   time.Now(), // Placeholder - will be replaced with actual segment time
-		})
+		groupName := group.GetSchema().Metadata.Name
+		endTime, hasSegments := dm.service.PeekOldestSegmentEndTimeInGroup(groupName)
+
+		// Only consider groups that have segments
+		if hasSegments {
+			candidates = append(candidates, groupSegmentTime{
+				groupName: groupName,
+				endTime:   endTime,
+			})
+		}
 	}
 
 	if len(candidates) == 0 {
 		return ""
 	}
 
-	// Sort by end time to find oldest
+	// Sort by end time to find oldest (earliest end time = oldest segment)
 	sort.Slice(candidates, func(i, j int) bool {
 		return candidates[i].endTime.Before(candidates[j].endTime)
 	})
diff --git a/banyand/measure/svc_data.go b/banyand/measure/svc_data.go
index 7099a573..65943672 100644
--- a/banyand/measure/svc_data.go
+++ b/banyand/measure/svc_data.go
@@ -106,10 +106,42 @@ func (s *dataSVC) LoadAllGroups() []resourceSchema.Group {
 	return s.schemaRepo.LoadAllGroups()
 }
 
+func (s *dataSVC) PeekOldestSegmentEndTimeInGroup(group string) (time.Time, bool) {
+	g, ok := s.schemaRepo.LoadGroup(group)
+	if !ok {
+		return time.Time{}, false
+	}
+
+	db := g.SupplyTSDB()
+	if db == nil {
+		return time.Time{}, false
+	}
+
+	// Type assert to the storage interface that has PeekOldestSegmentEndTime
+	if dbWithPeek, ok := db.(interface{ PeekOldestSegmentEndTime() (time.Time, bool) }); ok {
+		return dbWithPeek.PeekOldestSegmentEndTime()
+	}
+
+	return time.Time{}, false
+}
+
 func (s *dataSVC) DeleteOldestSegmentInGroup(group string) (bool, error) {
-	// TODO: Implement actual segment deletion when storage APIs are ready
-	// For now, return false to indicate no deletion performed
-	s.l.Debug().Str("group", group).Msg("DeleteOldestSegmentInGroup not yet implemented")
+	g, ok := s.schemaRepo.LoadGroup(group)
+	if !ok {
+		return false, nil
+	}
+
+	db := g.SupplyTSDB()
+	if db == nil {
+		return false, nil
+	}
+
+	// Type assert to the storage interface that has DeleteOldestSegment
+	if dbWithDelete, ok := db.(interface{ DeleteOldestSegment() (bool, error) }); ok {
+		return dbWithDelete.DeleteOldestSegment()
+	}
+
+	s.l.Debug().Str("group", group).Msg("database does not support DeleteOldestSegment")
 	return false, nil
 }
 
diff --git a/banyand/measure/svc_standalone.go b/banyand/measure/svc_standalone.go
index f697c9b4..4f946d2f 100644
--- a/banyand/measure/svc_standalone.go
+++ b/banyand/measure/svc_standalone.go
@@ -110,10 +110,42 @@ func (s *standalone) LoadAllGroups() []resourceSchema.Group {
 	return s.schemaRepo.LoadAllGroups()
 }
 
+func (s *standalone) PeekOldestSegmentEndTimeInGroup(group string) (time.Time, bool) {
+	g, ok := s.schemaRepo.LoadGroup(group)
+	if !ok {
+		return time.Time{}, false
+	}
+
+	db := g.SupplyTSDB()
+	if db == nil {
+		return time.Time{}, false
+	}
+
+	// Type assert to the storage interface that has PeekOldestSegmentEndTime
+	if dbWithPeek, ok := db.(interface{ PeekOldestSegmentEndTime() (time.Time, bool) }); ok {
+		return dbWithPeek.PeekOldestSegmentEndTime()
+	}
+
+	return time.Time{}, false
+}
+
 func (s *standalone) DeleteOldestSegmentInGroup(group string) (bool, error) {
-	// TODO: Implement actual segment deletion when storage APIs are ready
-	// For now, return false to indicate no deletion performed
-	s.l.Debug().Str("group", group).Msg("DeleteOldestSegmentInGroup not yet implemented")
+	g, ok := s.schemaRepo.LoadGroup(group)
+	if !ok {
+		return false, nil
+	}
+
+	db := g.SupplyTSDB()
+	if db == nil {
+		return false, nil
+	}
+
+	// Type assert to the storage interface that has DeleteOldestSegment
+	if dbWithDelete, ok := db.(interface{ DeleteOldestSegment() (bool, error) }); ok {
+		return dbWithDelete.DeleteOldestSegment()
+	}
+
+	s.l.Debug().Str("group", group).Msg("database does not support DeleteOldestSegment")
 	return false, nil
 }
 
diff --git a/banyand/stream/svc_standalone.go b/banyand/stream/svc_standalone.go
index 1bc0f8fe..31f58843 100644
--- a/banyand/stream/svc_standalone.go
+++ b/banyand/stream/svc_standalone.go
@@ -109,10 +109,42 @@ func (s *standalone) LoadAllGroups() []resourceSchema.Group {
 	return s.schemaRepo.LoadAllGroups()
 }
 
+func (s *standalone) PeekOldestSegmentEndTimeInGroup(group string) (time.Time, bool) {
+	g, ok := s.schemaRepo.LoadGroup(group)
+	if !ok {
+		return time.Time{}, false
+	}
+
+	db := g.SupplyTSDB()
+	if db == nil {
+		return time.Time{}, false
+	}
+
+	// Type assert to the storage interface that has PeekOldestSegmentEndTime
+	if dbWithPeek, ok := db.(interface{ PeekOldestSegmentEndTime() (time.Time, bool) }); ok {
+		return dbWithPeek.PeekOldestSegmentEndTime()
+	}
+
+	return time.Time{}, false
+}
+
 func (s *standalone) DeleteOldestSegmentInGroup(group string) (bool, error) {
-	// TODO: Implement actual segment deletion when storage APIs are ready
-	// For now, return false to indicate no deletion performed
-	s.l.Debug().Str("group", group).Msg("DeleteOldestSegmentInGroup not yet implemented")
+	g, ok := s.schemaRepo.LoadGroup(group)
+	if !ok {
+		return false, nil
+	}
+
+	db := g.SupplyTSDB()
+	if db == nil {
+		return false, nil
+	}
+
+	// Type assert to the storage interface that has DeleteOldestSegment
+	if dbWithDelete, ok := db.(interface{ DeleteOldestSegment() (bool, error) }); ok {
+		return dbWithDelete.DeleteOldestSegment()
+	}
+
+	s.l.Debug().Str("group", group).Msg("database does not support DeleteOldestSegment")
 	return false, nil
 }
 
diff --git a/banyand/trace/svc_standalone.go b/banyand/trace/svc_standalone.go
index bd3e7e05..49065764 100644
--- a/banyand/trace/svc_standalone.go
+++ b/banyand/trace/svc_standalone.go
@@ -212,10 +212,42 @@ func (s *standalone) LoadAllGroups() []resourceSchema.Group {
 	return s.schemaRepo.LoadAllGroups()
 }
 
+func (s *standalone) PeekOldestSegmentEndTimeInGroup(group string) (time.Time, bool) {
+	g, ok := s.schemaRepo.LoadGroup(group)
+	if !ok {
+		return time.Time{}, false
+	}
+
+	db := g.SupplyTSDB()
+	if db == nil {
+		return time.Time{}, false
+	}
+
+	// Type assert to the storage interface that has PeekOldestSegmentEndTime
+	if dbWithPeek, ok := db.(interface{ PeekOldestSegmentEndTime() (time.Time, bool) }); ok {
+		return dbWithPeek.PeekOldestSegmentEndTime()
+	}
+
+	return time.Time{}, false
+}
+
 func (s *standalone) DeleteOldestSegmentInGroup(group string) (bool, error) {
-	// TODO: Implement actual segment deletion when storage APIs are ready
-	// For now, return false to indicate no deletion performed
-	s.l.Debug().Str("group", group).Msg("DeleteOldestSegmentInGroup not yet implemented")
+	g, ok := s.schemaRepo.LoadGroup(group)
+	if !ok {
+		return false, nil
+	}
+
+	db := g.SupplyTSDB()
+	if db == nil {
+		return false, nil
+	}
+
+	// Type assert to the storage interface that has DeleteOldestSegment
+	if dbWithDelete, ok := db.(interface{ DeleteOldestSegment() (bool, error) }); ok {
+		return dbWithDelete.DeleteOldestSegment()
+	}
+
+	s.l.Debug().Str("group", group).Msg("database does not support DeleteOldestSegment")
 	return false, nil
 }