fix: improve delete speed when a measurement is part of the predicate

cmd/influxd/launcher/engine.go

@@ -7,6 +7,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/influxdata/influxql"
+
 	"github.com/influxdata/influxdb/v2"
 	"github.com/influxdata/influxdb/v2/http"
 	"github.com/influxdata/influxdb/v2/kit/platform"
@@ -119,8 +121,8 @@ func (t *TemporaryEngine) SeriesCardinality(ctx context.Context, bucketID platfo
 }
 
 // DeleteBucketRangePredicate will delete a bucket from the range and predicate.
-func (t *TemporaryEngine) DeleteBucketRangePredicate(ctx context.Context, orgID, bucketID platform.ID, min, max int64, pred influxdb.Predicate) error {
-	return t.engine.DeleteBucketRangePredicate(ctx, orgID, bucketID, min, max, pred)
+func (t *TemporaryEngine) DeleteBucketRangePredicate(ctx context.Context, orgID, bucketID platform.ID, min, max int64, pred influxdb.Predicate, measurement influxql.Expr) error {
+	return t.engine.DeleteBucketRangePredicate(ctx, orgID, bucketID, min, max, pred, measurement)
 }
 
 func (t *TemporaryEngine) CreateBucket(ctx context.Context, b *influxdb.Bucket) error {

delete.go

@@ -3,6 +3,8 @@ package influxdb
 import (
 	"context"
 
+	"github.com/influxdata/influxql"
+
 	"github.com/influxdata/influxdb/v2/kit/platform"
 )
 
@@ -15,5 +17,5 @@ type Predicate interface {
 
 // DeleteService will delete a bucket from the range and predict.
 type DeleteService interface {
-	DeleteBucketRangePredicate(ctx context.Context, orgID, bucketID platform.ID, min, max int64, pred Predicate) error
+	DeleteBucketRangePredicate(ctx context.Context, orgID, bucketID platform.ID, min, max int64, pred Predicate, measurement influxql.Expr) error
 }

http/delete_handler.go

@@ -5,9 +5,12 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	http "net/http"
 	"time"
 
+	"github.com/influxdata/influxql"
+
 	"github.com/influxdata/httprouter"
 	"github.com/influxdata/influxdb/v2"
 	pcontext "github.com/influxdata/influxdb/v2/context"
@@ -91,7 +94,7 @@ func (h *DeleteHandler) handleDelete(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	dr, err := decodeDeleteRequest(
+	dr, measurement, err := decodeDeleteRequest(
 		ctx, r,
 		h.OrganizationService,
 		h.BucketService,
@@ -121,7 +124,7 @@ func (h *DeleteHandler) handleDelete(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if err := h.DeleteService.DeleteBucketRangePredicate(r.Context(), dr.Org.ID, dr.Bucket.ID, dr.Start, dr.Stop, dr.Predicate); err != nil {
+	if err := h.DeleteService.DeleteBucketRangePredicate(r.Context(), dr.Org.ID, dr.Bucket.ID, dr.Start, dr.Stop, dr.Predicate, measurement); err != nil {
 		h.HandleHTTPError(ctx, &errors.Error{
 			Code: errors.EInternal,
 			Op:   "http/handleDelete",
@@ -139,24 +142,78 @@ func (h *DeleteHandler) handleDelete(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(http.StatusNoContent)
 }
 
-func decodeDeleteRequest(ctx context.Context, r *http.Request, orgSvc influxdb.OrganizationService, bucketSvc influxdb.BucketService) (*deleteRequest, error) {
+func decodeDeleteRequest(ctx context.Context, r *http.Request, orgSvc influxdb.OrganizationService, bucketSvc influxdb.BucketService) (*deleteRequest, influxql.Expr, error) {
 	dr := new(deleteRequest)
-	err := json.NewDecoder(r.Body).Decode(dr)
+	buf, err := io.ReadAll(r.Body)
 	if err != nil {
-		return nil, &errors.Error{
+		je := &errors.Error{
 			Code: errors.EInvalid,
-			Msg:  "invalid request; error parsing request json",
+			Msg:  "error reading json body",
 			Err:  err,
 		}
+		return nil, nil, je
 	}
+	buffer := bytes.NewBuffer(buf)
+	err = json.NewDecoder(buffer).Decode(dr)
+	if err != nil {
+		je := &errors.Error{
+			Code: errors.EInvalid,
+			Msg:  "error decoding json body",
+			Err:  err,
+		}
+		return nil, nil, je
+	}
+
+	var drd deleteRequestDecode
+	err = json.Unmarshal(buf, &drd)
+	if err != nil {
+		je := &errors.Error{
+			Code: errors.EInvalid,
+			Msg:  "error decoding json body for predicate",
+			Err:  err,
+		}
+		return nil, nil, je
+	}
+	var measurementExpr influxql.Expr
+	if drd.Predicate != "" {
+		expr, err := influxql.ParseExpr(drd.Predicate)
+		if err != nil {
+			return nil, nil, &errors.Error{
+				Code: errors.EInvalid,
+				Msg:  "invalid request; error parsing predicate",
+				Err:  err,
+			}
+		}
+		measurementExpr, _, err = influxql.PartitionExpr(influxql.CloneExpr(expr), func(e influxql.Expr) (bool, error) {
+			switch e := e.(type) {
+			case *influxql.BinaryExpr:
+				switch e.Op {
+				case influxql.EQ, influxql.NEQ, influxql.EQREGEX, influxql.NEQREGEX:
+					tag, ok := e.LHS.(*influxql.VarRef)
+					if ok && tag.Val == "_measurement" {
+						return true, nil
+					}
+				}
+			}
+			return false, nil
+		})
+		if err != nil {
+			return nil, nil, &errors.Error{
+				Code: errors.EInvalid,
+				Msg:  "invalid request; error partitioning predicate",
+				Err:  err,
+			}
+		}
+	}
+
 	if dr.Org, err = queryOrganization(ctx, r, orgSvc); err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	if dr.Bucket, err = queryBucket(ctx, dr.Org.ID, r, bucketSvc); err != nil {
-		return nil, err
+		return nil, nil, err
 	}
-	return dr, nil
+	return dr, measurementExpr, nil
 }
 
 type deleteRequest struct {

http/delete_test.go

@@ -72,7 +72,7 @@ func TestDelete(t *testing.T) {
 				contentType: "application/json; charset=utf-8",
 				body: `{
 					"code": "invalid",
-					"message": "invalid request; error parsing request json: invalid RFC3339Nano for field start, please format your time with RFC3339Nano format, example: 2009-01-02T23:00:00Z"
+					"message": "error decoding json body: invalid RFC3339Nano for field start, please format your time with RFC3339Nano format, example: 2009-01-02T23:00:00Z"
 				  }`,
 			},
 		},
@@ -89,7 +89,7 @@ func TestDelete(t *testing.T) {
 				contentType: "application/json; charset=utf-8",
 				body: `{
 					"code": "invalid",
-					"message": "invalid request; error parsing request json: invalid RFC3339Nano for field stop, please format your time with RFC3339Nano format, example: 2009-01-01T23:00:00Z"
+					"message": "error decoding json body: invalid RFC3339Nano for field stop, please format your time with RFC3339Nano format, example: 2009-01-01T23:00:00Z"
 				  }`,
 			},
 		},
@@ -106,7 +106,7 @@ func TestDelete(t *testing.T) {
 				contentType: "application/json; charset=utf-8",
 				body: fmt.Sprintf(`{
 					"code": "invalid",
-					"message": "invalid request; error parsing request json: %s"
+					"message": "error decoding json body: %s"
 				  }`, msgStartTooSoon),
 			},
 		},
@@ -123,7 +123,7 @@ func TestDelete(t *testing.T) {
 				contentType: "application/json; charset=utf-8",
 				body: fmt.Sprintf(`{
 					"code": "invalid",
-					"message": "invalid request; error parsing request json: %s"
+					"message": "error decoding json body: %s"
 				  }`, msgStopTooLate),
 			},
 		},
@@ -321,7 +321,61 @@ func TestDelete(t *testing.T) {
 				statusCode: http.StatusBadRequest,
 				body: `{
 					"code": "invalid",
-					"message": "invalid request; error parsing request json: the logical operator OR is not supported yet at position 25"
+					"message": "error decoding json body: the logical operator OR is not supported yet at position 25"
+				  }`,
+			},
+		},
+		{
+			name: "unsupported delete measurements",
+			args: args{
+				queryParams: map[string][]string{
+					"org":    {"org1"},
+					"bucket": {"buck1"},
+				},
+				body: []byte(`{
+					"start":"2009-01-01T23:00:00Z",
+					"stop":"2019-11-10T01:00:00Z",
+					"predicate": "_measurement=\"cpu\" or _measurement=\"mem\""
+				}`),
+				authorizer: &influxdb.Authorization{
+					UserID: user1ID,
+					Status: influxdb.Active,
+					Permissions: []influxdb.Permission{
+						{
+							Action: influxdb.WriteAction,
+							Resource: influxdb.Resource{
+								Type:  influxdb.BucketsResourceType,
+								ID:    influxtesting.IDPtr(platform.ID(2)),
+								OrgID: influxtesting.IDPtr(platform.ID(1)),
+							},
+						},
+					},
+				},
+			},
+			fields: fields{
+				DeleteService: mock.NewDeleteService(),
+				BucketService: &mock.BucketService{
+					FindBucketFn: func(ctx context.Context, f influxdb.BucketFilter) (*influxdb.Bucket, error) {
+						return &influxdb.Bucket{
+							ID:   platform.ID(2),
+							Name: "bucket1",
+						}, nil
+					},
+				},
+				OrganizationService: &mock.OrganizationService{
+					FindOrganizationF: func(ctx context.Context, f influxdb.OrganizationFilter) (*influxdb.Organization, error) {
+						return &influxdb.Organization{
+							ID:   platform.ID(1),
+							Name: "org1",
+						}, nil
+					},
+				},
+			},
+			wants: wants{
+				statusCode: http.StatusBadRequest,
+				body: `{
+					"code": "invalid",
+					"message": "error decoding json body: the logical operator OR is not supported yet at position 19"
 				  }`,
 			},
 		},
@@ -335,7 +389,7 @@ func TestDelete(t *testing.T) {
 				body: []byte(`{
 					"start":"2009-01-01T23:00:00Z",
 					"stop":"2019-11-10T01:00:00Z",
-					"predicate": "tag1=\"v1\" and (tag2=\"v2\" and tag3=\"v3\")"
+					"predicate": "_measurement=\"testing\" and tag1=\"v1\" and (tag2=\"v2\" and tag3=\"v3\")"
 				}`),
 				authorizer: &influxdb.Authorization{
 					UserID: user1ID,

mock/delete.go

@@ -3,6 +3,8 @@ package mock
 import (
 	"context"
 
+	"github.com/influxdata/influxql"
+
 	"github.com/influxdata/influxdb/v2"
 	"github.com/influxdata/influxdb/v2/kit/platform"
 )
@@ -11,20 +13,20 @@ var _ influxdb.DeleteService = &DeleteService{}
 
 // DeleteService is a mock delete server.
 type DeleteService struct {
-	DeleteBucketRangePredicateF func(tx context.Context, orgID, bucketID platform.ID, min, max int64, pred influxdb.Predicate) error
+	DeleteBucketRangePredicateF func(tx context.Context, orgID, bucketID platform.ID, min, max int64, pred influxdb.Predicate, measurement influxql.Expr) error
 }
 
 // NewDeleteService returns a mock DeleteService where its methods will return
 // zero values.
 func NewDeleteService() DeleteService {
 	return DeleteService{
-		DeleteBucketRangePredicateF: func(tx context.Context, orgID, bucketID platform.ID, min, max int64, pred influxdb.Predicate) error {
+		DeleteBucketRangePredicateF: func(tx context.Context, orgID, bucketID platform.ID, min, max int64, pred influxdb.Predicate, measurement influxql.Expr) error {
 			return nil
 		},
 	}
 }
 
-//DeleteBucketRangePredicate calls DeleteBucketRangePredicateF.
-func (s DeleteService) DeleteBucketRangePredicate(ctx context.Context, orgID, bucketID platform.ID, min, max int64, pred influxdb.Predicate) error {
-	return s.DeleteBucketRangePredicateF(ctx, orgID, bucketID, min, max, pred)
+// DeleteBucketRangePredicate calls DeleteBucketRangePredicateF.
+func (s DeleteService) DeleteBucketRangePredicate(ctx context.Context, orgID, bucketID platform.ID, min, max int64, pred influxdb.Predicate, measurement influxql.Expr) error {
+	return s.DeleteBucketRangePredicateF(ctx, orgID, bucketID, min, max, pred, measurement)
 }

storage/engine.go

@@ -327,7 +327,7 @@ func (e *Engine) DeleteBucket(ctx context.Context, orgID, bucketID platform.ID)
 
 // DeleteBucketRangePredicate deletes data within a bucket from the storage engine. Any data
 // deleted must be in [min, max], and the key must match the predicate if provided.
-func (e *Engine) DeleteBucketRangePredicate(ctx context.Context, orgID, bucketID platform.ID, min, max int64, pred influxdb.Predicate) error {
+func (e *Engine) DeleteBucketRangePredicate(ctx context.Context, orgID, bucketID platform.ID, min, max int64, pred influxdb.Predicate, measurement influxql.Expr) error {
 	span, _ := tracing.StartSpanFromContext(ctx)
 	defer span.Finish()
 
@@ -336,7 +336,7 @@ func (e *Engine) DeleteBucketRangePredicate(ctx context.Context, orgID, bucketID
 	if e.closing == nil {
 		return ErrEngineClosed
 	}
-	return e.tsdbStore.DeleteSeriesWithPredicate(ctx, bucketID.String(), min, max, pred)
+	return e.tsdbStore.DeleteSeriesWithPredicate(ctx, bucketID.String(), min, max, pred, measurement)
 }
 
 // RLockKVStore locks the KV store as well as the engine in preparation for doing a backup.