Add interface to clean partition resources

disk_partition.go

@@ -142,3 +142,8 @@ func (d *diskPartition) size() int {
 func (d *diskPartition) active() bool {
 	return false
 }
+
+func (d *diskPartition) clean() error {
+	// FIXME: Remove the directory this partition holds.
+	return nil
+}

fake_partition.go

@@ -32,3 +32,7 @@ func (f *fakePartition) size() int {
 func (f *fakePartition) active() bool {
 	return f.IsActive
 }
+
+func (f *fakePartition) clean() error {
+	return nil
+}

memory_partition.go

@@ -155,6 +155,12 @@ func (m *memoryPartition) active() bool {
 	return m.maxTimestamp()-m.minTimestamp()+1 < m.partitionDuration
 }
 
+func (m *memoryPartition) clean() error {
+	// What all data managed by memoryPartition is on heap that is automatically removed by GC.
+	// So do nothing.
+	return nil
+}
+
 // memoryMetric has a list of ordered data points that belong to the memoryMetric
 type memoryMetric struct {
 	name         string

partition.go

@@ -16,6 +16,8 @@ type partition interface {
 	// If data points older than its min timestamp were given, they won't be
 	// ingested, instead, gave back as a first returned value.
 	insertRows(rows []Row) (outdatedRows []Row, err error)
+	// clean removes everything managed by this partition.
+	clean() error
 
 	// Read operations
 	//

partition_list.go

@@ -98,6 +98,9 @@ func (p *partitionListImpl) remove(target partition) error {
 		}
 
 		// remove the current node.
+		if err := current.value().clean(); err != nil {
+			return fmt.Errorf("failed to clean resources managed by partition to be removed: %w", err)
+		}
 
 		iterator.next()
 		next = iterator.currentNode()