Delete by range (#426)

* support delete by range

* update partitions max time when needed + tests + debug logs and prints

* only delete item if needed + remove logs and unneeded tests

* fix possible panic

* go fmt

* fix race

* go fmt

* fix hang + debug prints

* add ctl validation

* recalculate server-side aggregations + make deleting of whole partitions parallel

* add maxTime optimization to set as the latest datapoint in the last chunk

* delete entire chunk if no samples are left in the chunk after deleting

* break update expressions to smaller chunks due to engine limitation

* change max time logic in delete

* update v3io-go

* linter stuff

Author: talIguaz

go.mod

@@ -19,7 +19,7 @@ require (
 	github.com/stretchr/testify v1.4.0
 	github.com/tinylib/msgp v1.1.1 // indirect
 	github.com/v3io/frames v0.6.8-v0.9.11
-	github.com/v3io/v3io-go v0.0.5-0.20191205125653-9003ae83f0b6
+	github.com/v3io/v3io-go v0.0.7-0.20200216132233-3b52a325296d
 	github.com/v3io/v3io-go-http v0.0.0-20190415143924-cc2fbcde6663 // indirect
 	github.com/xwb1989/sqlparser v0.0.0-20180606152119-120387863bf2
 	google.golang.org/genproto v0.0.0-20181026194446-8b5d7a19e2d9 // indirect

go.sum

@@ -76,9 +76,9 @@ github.com/prometheus/prometheus v2.5.0+incompatible h1:7QPitgO2kOFG8ecuRn9O/4L9
 github.com/prometheus/prometheus v2.5.0+incompatible/go.mod h1:oAIUtOny2rjMX0OWN5vPR5/q/twIROJvdqnQKDdil/s=
 github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a h1:9ZKAASQSHhDYGoxY8uLVpewe1GDZ2vu2Tr/vTdVAkFQ=
 github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
+github.com/russross/blackfriday v1.5.2 h1:HyvC0ARfnZBqnXwABFeSZHpKvJHJJfPz81GNueLj0oo=
 github.com/russross/blackfriday v1.5.2+incompatible h1:/YIL6L1Deczl4O/cQ7ZVdrdKwuB6y7EWpw9LkD8xofE=
 github.com/russross/blackfriday v1.5.2+incompatible/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
-github.com/russross/blackfriday v1.5.2 h1:HyvC0ARfnZBqnXwABFeSZHpKvJHJJfPz81GNueLj0oo=
 github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72 h1:qLC7fQah7D6K1B0ujays3HV9gkFtllcxhzImRR7ArPQ=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
@@ -104,10 +104,9 @@ github.com/v3io/frames v0.6.8-v0.9.11/go.mod h1:V3j8yjzhNNGXjosCBn7Qf8C8jo25Y+7G
 github.com/v3io/sqlparser v0.0.0-20190306105200-4d7273501871 h1:myF4tU/HdFWU1UzMdf16cHRbownzsyvL7VKIHqkrSvo=
 github.com/v3io/sqlparser v0.0.0-20190306105200-4d7273501871/go.mod h1:QD2Bo64oyTWzeV8RFehXS0hZEDFgOK99/h2a6ErRu6E=
 github.com/v3io/v3io-go v0.0.0-20191024084247-042df6b5ee40eb60996ab7f4e74ec9aa07d996c4/go.mod h1:IFb6dJiyvJnOjXUoCoPJ5UViaYjgVYmqJb4fD1qDeLk=
-github.com/v3io/v3io-go v0.0.0-20191120130819-9003ae83f0b673afb88b862d8f46dcc818684450 h1:3JMzABqziU+dBO4NCoIGRhI/NGYPd6d6Zug68nTXQkU=
 github.com/v3io/v3io-go v0.0.0-20191120130819-9003ae83f0b673afb88b862d8f46dcc818684450/go.mod h1:IFb6dJiyvJnOjXUoCoPJ5UViaYjgVYmqJb4fD1qDeLk=
-github.com/v3io/v3io-go v0.0.5-0.20191205125653-9003ae83f0b6 h1:+52DyMCjcWg6uXAlTe0KgbOsiQqUKrtL9tBPSERhyFg=
-github.com/v3io/v3io-go v0.0.5-0.20191205125653-9003ae83f0b6/go.mod h1:IFb6dJiyvJnOjXUoCoPJ5UViaYjgVYmqJb4fD1qDeLk=
+github.com/v3io/v3io-go v0.0.7-0.20200216132233-3b52a325296d h1:OotbIx7+QYju2DlAAVxWz0QFzBicHLc47u9DJGpVUL4=
+github.com/v3io/v3io-go v0.0.7-0.20200216132233-3b52a325296d/go.mod h1:IFb6dJiyvJnOjXUoCoPJ5UViaYjgVYmqJb4fD1qDeLk=
 github.com/v3io/v3io-go-http v0.0.0-20190221115935-53e2b487c9a2 h1:NJc63wM25iS+ci5z7LVwjWD4QM0QpTQw/fovKzatss0=
 github.com/v3io/v3io-go-http v0.0.0-20190221115935-53e2b487c9a2/go.mod h1:GXYcR9MxgfbE3BJdkXki5EclvtS8Nxu2RQNLA8hMMog=
 github.com/v3io/v3io-go-http v0.0.0-20190415143924-cc2fbcde6663 h1:WZcM/GRBAastacksmv5pODbtr8fJ/0/9EsPDpPfXkRk=

pkg/config/config.go

@@ -62,14 +62,16 @@ const (
 	DefaultUseServerAggregateCoefficient = 3
 
 	// KV attribute names
-	MaxTimeAttrName     = "_maxtime"
-	LabelSetAttrName    = "_lset"
-	EncodingAttrName    = "_enc"
-	OutOfOrderAttrName  = "_ooo"
-	MetricNameAttrName  = "_name"
-	ObjectNameAttrName  = "__name"
-	ChunkAttrPrefix     = "_v"
-	AggregateAttrPrefix = "_v_"
+	MaxTimeAttrName         = "_maxtime"
+	LabelSetAttrName        = "_lset"
+	EncodingAttrName        = "_enc"
+	OutOfOrderAttrName      = "_ooo"
+	MetricNameAttrName      = "_name"
+	ObjectNameAttrName      = "__name"
+	ChunkAttrPrefix         = "_v"
+	AggregateAttrPrefix     = "_v_"
+	MtimeSecsAttributeName  = "__mtime_secs"
+	MtimeNSecsAttributeName = "__mtime_nsecs"
 
 	PrometheusMetricNameAttribute = "__name__"
 

pkg/partmgr/partmgr.go

@@ -198,7 +198,7 @@ func (p *PartitionManager) updateSchema() error {
 			}
 
 			input := &v3io.PutItemInput{Path: schemaFilePath, Attributes: attributes}
-			err := p.container.PutItemSync(input)
+			_, err := p.container.PutItemSync(input)
 
 			if err != nil {
 				outerError = errors.Wrap(err, "failed to update partitions table.")
@@ -238,7 +238,7 @@ func (p *PartitionManager) DeletePartitionsFromSchema(partitionsToDelete []*DBPa
 			deletePartitionExpression.WriteString(");")
 		}
 		expression := deletePartitionExpression.String()
-		err := p.container.UpdateItemSync(&v3io.UpdateItemInput{Path: p.GetSchemaFilePath(), Expression: &expression})
+		_, err := p.container.UpdateItemSync(&v3io.UpdateItemInput{Path: p.GetSchemaFilePath(), Expression: &expression})
 		if err != nil {
 			return err
 		}
@@ -592,6 +592,33 @@ func (p *DBPartition) Time2Bucket(t int64) int {
 	return int((t - p.startTime) / p.rollupTime)
 }
 
+// Return the start time of an aggregation bucket by id
+func (p *DBPartition) GetAggregationBucketStartTime(id int) int64 {
+	return p.startTime + int64(id)*p.rollupTime
+}
+
+// Return the end time of an aggregation bucket by id
+func (p *DBPartition) GetAggregationBucketEndTime(id int) int64 {
+	return p.startTime + int64(id+1)*p.rollupTime - 1
+}
+
+func (p *DBPartition) Times2BucketRange(start, end int64) []int {
+	var buckets []int
+
+	if start > p.GetEndTime() || end < p.startTime {
+		return buckets
+	}
+
+	startingAggrBucket := p.Time2Bucket(start)
+	endAggrBucket := p.Time2Bucket(end)
+
+	for bucketID := startingAggrBucket; bucketID <= endAggrBucket; bucketID++ {
+		buckets = append(buckets, bucketID)
+	}
+
+	return buckets
+}
+
 // Return the nearest chunk start time for the specified time
 func (p *DBPartition) GetChunkMint(t int64) int64 {
 	if t > p.GetEndTime() {
@@ -622,6 +649,37 @@ func (p *DBPartition) TimeToChunkId(tmilli int64) (int, error) {
 	}
 }
 
+// Check if a chunk (by attribute name) is in the given time range.
+func (p *DBPartition) IsChunkInRangeByAttr(attr string, mint, maxt int64) bool {
+
+	// Discard '_v' prefix
+	chunkIDStr := attr[2:]
+	chunkID, err := strconv.ParseInt(chunkIDStr, 10, 64)
+	if err != nil {
+		return false
+	}
+
+	chunkStartTime := p.startTime + (chunkID-1)*p.chunkInterval
+	chunkEndTime := chunkStartTime + p.chunkInterval - 1
+
+	return mint <= chunkStartTime && maxt >= chunkEndTime
+}
+
+// Get a chunk's start time by it's attribute name
+func (p *DBPartition) GetChunkStartTimeByAttr(attr string) (int64, error) {
+
+	// Discard '_v' prefix
+	chunkIDStr := attr[2:]
+	chunkID, err := strconv.ParseInt(chunkIDStr, 10, 64)
+	if err != nil {
+		return 0, err
+	}
+
+	chunkStartTime := p.startTime + (chunkID-1)*p.chunkInterval
+
+	return chunkStartTime, nil
+}
+
 // Check whether the specified time is within the range of this partition
 func (p *DBPartition) InRange(t int64) bool {
 	if p.manager.cyclic {