Introduce two key feats to tsdb

banyand/measure/measure_write.go

@@ -61,7 +61,7 @@ func (s *measure) Write(value *measurev1.DataPointValue) error {
 }
 
 func (s *measure) write(shardID common.ShardID, seriesHashKey []byte, value *measurev1.DataPointValue, cb index.CallbackFn) error {
-	t := value.GetTimestamp().AsTime()
+	t := value.GetTimestamp().AsTime().Local()
 	if err := timestamp.Check(t); err != nil {
 		return errors.WithMessage(err, "writing stream")
 	}
@@ -81,7 +81,7 @@ func (s *measure) write(shardID common.ShardID, seriesHashKey []byte, value *mea
 	if err != nil {
 		return err
 	}
-	wp, err := series.Span(timestamp.NewInclusiveTimeRangeDuration(t, 0))
+	wp, err := series.Create(t)
 	if err != nil {
 		if wp != nil {
 			_ = wp.Close()
@@ -189,7 +189,7 @@ func (w *writeCallback) Rev(message bus.Message) (resp bus.Message) {
 	}
 	err := stm.write(common.ShardID(writeEvent.GetShardId()), writeEvent.GetSeriesHash(), writeEvent.GetRequest().GetDataPoint(), nil)
 	if err != nil {
-		w.l.Debug().Err(err).Msg("fail to write entity")
+		w.l.Error().Err(err).Msg("fail to write entity")
 	}
 	return
 }

banyand/measure/metadata.go

@@ -215,6 +215,9 @@ func (s *supplier) OpenDB(groupSchema *commonv1.Group) (tsdb.Database, error) {
 	if opts.SegmentInterval, err = pb_v1.ToIntervalRule(groupSchema.ResourceOpts.SegmentInterval); err != nil {
 		return nil, err
 	}
+	if opts.TTL, err = pb_v1.ToIntervalRule(groupSchema.ResourceOpts.Ttl); err != nil {
+		return nil, err
+	}
 	return tsdb.OpenDatabase(
 		context.WithValue(context.Background(), common.PositionKey, common.Position{
 			Module:   "measure",

banyand/stream/metadata.go

@@ -218,6 +218,9 @@ func (s *supplier) OpenDB(groupSchema *commonv1.Group) (tsdb.Database, error) {
 	if opts.SegmentInterval, err = pb_v1.ToIntervalRule(groupSchema.ResourceOpts.SegmentInterval); err != nil {
 		return nil, err
 	}
+	if opts.TTL, err = pb_v1.ToIntervalRule(groupSchema.ResourceOpts.Ttl); err != nil {
+		return nil, err
+	}
 	return tsdb.OpenDatabase(
 		context.WithValue(context.Background(), common.PositionKey, common.Position{
 			Module:   "stream",

banyand/stream/stream_write.go

@@ -88,7 +88,7 @@ func (s *stream) write(shardID common.ShardID, seriesHashKey []byte, value *stre
 		return err
 	}
 	t := timestamp.MToN(tp)
-	wp, err := series.Span(timestamp.NewInclusiveTimeRangeDuration(t, 0))
+	wp, err := series.Create(t)
 	if err != nil {
 		if wp != nil {
 			_ = wp.Close()
@@ -183,7 +183,7 @@ func (w *writeCallback) Rev(message bus.Message) (resp bus.Message) {
 	}
 	err := stm.write(common.ShardID(writeEvent.GetShardId()), writeEvent.GetSeriesHash(), writeEvent.GetRequest().GetElement(), nil)
 	if err != nil {
-		w.l.Debug().Err(err).Msg("fail to write entity")
+		w.l.Error().Err(err).Msg("fail to write entity")
 	}
 	return
 }

banyand/tsdb/block.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"os"
 	"path"
 	"runtime"
 	"strconv"
@@ -38,6 +39,7 @@ import (
 	"github.com/apache/skywalking-banyandb/pkg/index/lsm"
 	"github.com/apache/skywalking-banyandb/pkg/logger"
 	"github.com/apache/skywalking-banyandb/pkg/timestamp"
+	"github.com/pkg/errors"
 )
 
 const (
@@ -55,6 +57,7 @@ type block struct {
 	suffix     string
 	ref        *atomic.Int32
 	closed     *atomic.Bool
+	deleted    *atomic.Bool
 	lock       sync.RWMutex
 	position   common.Position
 	memSize    int64
@@ -100,6 +103,7 @@ func newBlock(ctx context.Context, opts blockOpts) (b *block, err error) {
 		flushCh:   make(chan struct{}, 1),
 		ref:       &atomic.Int32{},
 		closed:    &atomic.Bool{},
+		deleted:   &atomic.Bool{},
 		queue:     opts.queue,
 	}
 	b.options(ctx)
@@ -136,6 +140,9 @@ func (b *block) options(ctx context.Context) {
 }
 
 func (b *block) open() (err error) {
+	if b.deleted.Load() {
+		return nil
+	}
 	if b.store, err = kv.OpenTimeSeriesStore(
 		0,
 		path.Join(b.path, componentMain),
@@ -181,6 +188,9 @@ func (b *block) open() (err error) {
 }
 
 func (b *block) delegate() (blockDelegate, error) {
+	if b.deleted.Load() {
+		return nil, errors.WithMessagef(ErrBlockAbsent, "block %d is deleted", b.blockID)
+	}
 	if b.incRef() {
 		return &bDelegate{
 			delegate: b,
@@ -272,6 +282,18 @@ func (b *block) stopThenClose() {
 	b.close()
 }
 
+func (b *block) delete() error {
+	if b.deleted.Load() {
+		return nil
+	}
+	b.deleted.Store(true)
+	if b.Reporter != nil {
+		b.Stop()
+	}
+	b.close()
+	return os.RemoveAll(b.path)
+}
+
 func (b *block) Closed() bool {
 	return b.closed.Load()
 }

banyand/tsdb/bucket/queue.go

@@ -27,6 +27,7 @@ type EvictFn func(id interface{})
 
 type Queue interface {
 	Push(id interface{})
+	Remove(id interface{})
 	Len() int
 	All() []interface{}
 }
@@ -108,6 +109,25 @@ func (q *lruQueue) Push(id interface{}) {
 	q.recent.Add(id, nil)
 }
 
+func (q *lruQueue) Remove(id interface{}) {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+
+	if q.frequent.Contains(id) {
+		q.frequent.Remove(id)
+		return
+	}
+
+	if q.recent.Contains(id) {
+		q.recent.Remove(id)
+		return
+	}
+
+	if q.recentEvict.Contains(id) {
+		q.recentEvict.Remove(id)
+	}
+}
+
 func (q *lruQueue) Len() int {
 	q.lock.RLock()
 	defer q.lock.RUnlock()

banyand/tsdb/retention.go

@@ -0,0 +1,103 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package tsdb
+
+import (
+	"sync"
+	"time"
+
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/robfig/cron/v3"
+)
+
+type retentionController struct {
+	segment   *segmentController
+	scheduler cron.Schedule
+	stopped   bool
+	stopMux   sync.Mutex
+	stopCh    chan struct{}
+	duration  time.Duration
+	l         *logger.Logger
+}
+
+func newRetentionController(segment *segmentController, ttl IntervalRule) (*retentionController, error) {
+	var expr string
+	switch ttl.Unit {
+	case HOUR:
+		// Every hour on the 5th minute
+		expr = "5 *"
+	case DAY:
+		// Every day on 00:05
+		expr = "5 0"
+
+	}
+	parser := cron.NewParser(cron.Minute | cron.Hour)
+	scheduler, err := parser.Parse(expr)
+	if err != nil {
+		return nil, err
+	}
+	return &retentionController{
+		segment:   segment,
+		scheduler: scheduler,
+		stopCh:    make(chan struct{}),
+		l:         segment.l.Named("retention-controller"),
+		duration:  ttl.EstimatedDuration(),
+	}, nil
+}
+
+func (rc *retentionController) start() {
+	rc.stopMux.Lock()
+	if rc.stopped {
+		return
+	}
+	rc.stopMux.Unlock()
+	go rc.run()
+}
+
+func (rc *retentionController) run() {
+	rc.l.Info().Msg("start")
+	now := rc.segment.clock.Now()
+	for {
+		next := rc.scheduler.Next(now)
+		timer := rc.segment.clock.Timer(next.Sub(now))
+		for {
+			select {
+			case now = <-timer.C:
+				rc.l.Info().Time("now", now).Msg("wake")
+				if err := rc.segment.remove(now.Add(-rc.duration)); err != nil {
+					rc.l.Error().Err(err)
+				}
+			case <-rc.stopCh:
+				timer.Stop()
+				rc.l.Info().Msg("stop")
+				return
+			}
+			break
+		}
+	}
+}
+
+func (rc *retentionController) stop() {
+	rc.stopMux.Lock()
+	defer rc.stopMux.Unlock()
+	if rc.stopped {
+		return
+	}
+	rc.stopped = true
+	close(rc.stopCh)
+}