Merge pull request #23 from solarisdb/filter_with_intervals

Filter with intervals

docker-compose.yaml

@@ -11,6 +11,8 @@ services:
     ports:
       - 50051:50051
       - 8080:8080
+    volumes:
+      - fs:/app/slogs
     healthcheck:
       test: [ "CMD", "/bin/grpc_health_probe", "-addr=:50051" ]
       interval: 10s
@@ -34,5 +36,7 @@ services:
       retries: 3
 
 volumes:
+  fs:
+    driver: local
   db:
     driver: local

golibs/datetime/date.go

@@ -107,6 +107,7 @@ var (
 		"YYYY/M/DD",
 
 		// yyyy-mm-ddThh
+		"YYYY-MM-DDTHH:mm:ss.SSSZZZZZ",
 		"YYYY-MM-DDTHH:mm:ss.SSSZZZZ",
 		"YYYY-MM-DDTHH:mm:ss.SSSZ",
 		"YYYY-MM-DDTHH:mm:ssZZZZZ",
@@ -161,7 +162,7 @@ var (
 		{"m", "4", "\\d{1,2}"},
 		{"ss", "05", "\\d{2}"},
 		{"s", "5", "\\d{1,2}"},
-		{".SSS", ".999999999", ".\\d{3,}"},
+		{".SSS", ".999999999", ".\\d+"},
 		{"P", "PM", "(?:am|AM|pm|PM)"},
 		{"ZZZZZ", "-07:00", "[+-][0-9]{2}:[0-9]{2}"},
 		{"ZZZZ", "-0700", "[+-][0-9]{4}"},

golibs/ulidutils/ulidutils.go

@@ -11,6 +11,7 @@
 // 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 ulidutils
 
 import (
@@ -20,6 +21,11 @@ import (
 	"github.com/oklog/ulid/v2"
 )
 
+var (
+	ZeroULID   = ulid.ULID{}
+	MaxULID, _ = ulid.Parse("7ZZZZZZZZZZZZZZZZZZZZZZZZZ")
+)
+
 // New returns new ulid.ULID.
 func New() ulid.ULID {
 	return ulid.Make()

pkg/api/rest/rest.go

@@ -50,35 +50,35 @@ func (r *Rest) RegisterEPs(g *gin.Engine) error {
 
 func (r *Rest) CreateLog(c *gin.Context) {
 	var rReq restapi.CreateLogRequest
-	if r.errorRespnse(c, BindAppJson(c, &rReq), "") {
+	if r.errorResponse(c, BindAppJson(c, &rReq), "") {
 		return
 	}
 	sLog, err := r.svc.CreateLog(c, &solaris.Log{Tags: rReq.Tags})
-	if r.errorRespnse(c, err, "") {
+	if r.errorResponse(c, err, "") {
 		return
 	}
 	c.JSON(http.StatusCreated, logToRest(sLog))
 }
 
 func (r *Rest) UpdateLog(c *gin.Context, logId restapi.LogId) {
 	var rReq restapi.UpdateLogRequest
-	if r.errorRespnse(c, BindAppJson(c, &rReq), "") {
+	if r.errorResponse(c, BindAppJson(c, &rReq), "") {
 		return
 	}
 	sLog, err := r.svc.UpdateLog(c, &solaris.Log{ID: logId, Tags: rReq.Tags})
-	if r.errorRespnse(c, err, "") {
+	if r.errorResponse(c, err, "") {
 		return
 	}
 	c.JSON(http.StatusOK, logToRest(sLog))
 }
 
 func (r *Rest) DeleteLogs(c *gin.Context) {
 	var rReq restapi.DeleteLogsRequest
-	if r.errorRespnse(c, BindAppJson(c, &rReq), "") {
+	if r.errorResponse(c, BindAppJson(c, &rReq), "") {
 		return
 	}
 	sRes, err := r.svc.DeleteLogs(c, &solaris.DeleteLogsRequest{Condition: rReq.FilterCondition})
-	if r.errorRespnse(c, err, "") {
+	if r.errorResponse(c, err, "") {
 		return
 	}
 	c.JSON(http.StatusOK, restapi.DeleteLogsResponse{Deleted: len(sRes.DeletedIDs)})
@@ -91,7 +91,7 @@ func (r *Rest) QueryLogs(c *gin.Context, params restapi.QueryLogsParams) {
 	sReq.PageID = cast.String(params.FromPageId, "")
 
 	sRes, err := r.svc.QueryLogs(c, sReq)
-	if r.errorRespnse(c, err, "") {
+	if r.errorResponse(c, err, "") {
 		return
 	}
 	rRes := restapi.QueryLogsResult{Items: logsToRest(sRes.Logs), Total: int(sRes.Total)}
@@ -103,14 +103,14 @@ func (r *Rest) QueryLogs(c *gin.Context, params restapi.QueryLogsParams) {
 
 func (r *Rest) CreateRecords(c *gin.Context, logId restapi.LogId) {
 	var rReq restapi.CreateRecordsRequest
-	if r.errorRespnse(c, BindAppJson(c, &rReq), "") {
+	if r.errorResponse(c, BindAppJson(c, &rReq), "") {
 		return
 	}
 	sReq := new(solaris.AppendRecordsRequest)
 	sReq.LogID = logId
 	sReq.Records = createRecsToSvc(rReq.Records)
 	sRes, err := r.svc.AppendRecords(c, sReq)
-	if r.errorRespnse(c, err, "") {
+	if r.errorResponse(c, err, "") {
 		return
 	}
 	c.JSON(http.StatusCreated, restapi.CreateRecordsResponse{Added: int(sRes.Added)})
@@ -126,16 +126,14 @@ func (r *Rest) QueryRecords(c *gin.Context, params restapi.QueryRecordsParams) {
 	sReq.Limit = int64(cast.Int(params.Limit, 0))
 
 	sResQ, err := r.svc.QueryRecords(c, sReq)
-	if r.errorRespnse(c, err, "") {
+	if r.errorResponse(c, err, "") {
 		return
 	}
-	// TODO: uncomment after svc.CountRecords is implemented
-	//sResC, err := r.svc.CountRecords(c, sReq)
-	//if r.errorRespnse(c, err, "") {
-	//	return
-	//}
-	//rRes := restapi.QueryRecordsResult{Items: recsToRest(sResQ.Records), Total: int(sResC.Total)}
-	rRes := restapi.QueryRecordsResult{Items: recsToRest(sResQ.Records), Total: len(sResQ.Records)}
+	sResC, err := r.svc.CountRecords(c, sReq)
+	if r.errorResponse(c, err, "") {
+		return
+	}
+	rRes := restapi.QueryRecordsResult{Items: recsToRest(sResQ.Records), Total: int(sResC.Count)}
 	if len(sResQ.NextPageID) > 0 {
 		rRes.NextPageId = cast.Ptr(sResQ.NextPageID)
 	}
@@ -146,7 +144,7 @@ func (r *Rest) Ping(c *gin.Context) {
 	c.String(http.StatusOK, "pong")
 }
 
-func (r *Rest) errorRespnse(c *gin.Context, err error, msg string) bool {
+func (r *Rest) errorResponse(c *gin.Context, err error, msg string) bool {
 	if err == nil {
 		return false
 	}

pkg/intervals/intervals.go

@@ -206,8 +206,8 @@ func (b Basis[T]) Union(i1, i2 Interval[T]) (Interval[T], bool) {
 }
 
 // After returns true if the L border of the i1 interval
-// is lesser than the R border of the i2 interval.
-// NOTE: The `(L=1, R=3)` and `(L=2, R=4)` intervals are
+// is greater than the R border of the i2 interval.
+// NOTE: The `(L=2, R=4)` and `(L=1, R=3)` intervals are
 // considered to have an intersection (see Intersection description)
 // and the `(L=2, R=4).After.((L=1, R=3))` call returns false.
 func (b Basis[T]) After(i1, i2 Interval[T]) bool {

pkg/ql/datetime.go

@@ -67,6 +67,7 @@ var dateTimeParser = datetime.NewParser([]string{
 	"YYYY/M/DD",
 
 	// yyyy-mm-ddThh
+	"YYYY-MM-DDTHH:mm:ss.SSSZZZZZ",
 	"YYYY-MM-DDTHH:mm:ss.SSSZZZZ",
 	"YYYY-MM-DDTHH:mm:ss.SSSZ",
 	"YYYY-MM-DDTHH:mm:ssZZZZZ",

pkg/ql/intervalbuilder.go

@@ -46,6 +46,7 @@ func NewParamIntervalBuilder[T, K any](basis intervals.Basis[T], dialect Dialect
 }
 
 // Build returns a list of intervals built from the AST expression.
+// Returned intervals are sorted by the L border.
 func (ib *ParamIntervalBuilder[T, K]) Build(expr *Expression) ([]intervals.Interval[T], error) {
 	var res []intervals.Interval[T]
 	for _, or := range expr.Or {
@@ -57,8 +58,7 @@ func (ib *ParamIntervalBuilder[T, K]) Build(expr *Expression) ([]intervals.Inter
 			res = append(res, tt...)
 		}
 	}
-	res = ib.union(res)
-	return res, nil
+	return ib.union(res), nil
 }
 
 func (ib *ParamIntervalBuilder[T, K]) buildOR(or *OrCondition) ([]intervals.Interval[T], error) {
@@ -105,9 +105,6 @@ func (ib *ParamIntervalBuilder[T, K]) buildXCond(and *XCondition) ([]intervals.I
 func (ib *ParamIntervalBuilder[T, K]) buildCond(cond *Condition) ([]intervals.Interval[T], error) {
 	// param1
 	p1 := cond.FirstParam
-	if p1.Name(false) != ib.param { // skip not the param we look for
-		return nil, nil
-	}
 	dp1, ok := ib.dialect[p1.ID()]
 	if !ok {
 		return nil, fmt.Errorf("the parameter %s must be known: %w", p1.Name(false), errors.ErrInvalid)
@@ -118,15 +115,15 @@ func (ib *ParamIntervalBuilder[T, K]) buildCond(cond *Condition) ([]intervals.In
 	if dp1.Flags&PfNop != 0 {
 		return nil, fmt.Errorf("the parameter %s must allow operation (%s): %w", p1.Name(false), cond.Op, errors.ErrInvalid)
 	}
+	if p1.Name(false) != ib.param { // skip not the param we look for
+		return nil, nil
+	}
 
 	// param2
 	p2 := cond.SecondParam
 	if p2 == nil {
 		return nil, fmt.Errorf("the second parameter must be specified for the parameter %s and the operation %q: %w", p1.Name(false), cond.Op, errors.ErrInvalid)
 	}
-	if p2.Const == nil { // skip not a constant param
-		return nil, nil
-	}
 	dp2, ok := ib.dialect[p2.ID()]
 	if !ok {
 		return nil, fmt.Errorf("the second parameter %s must be known: %w", p2.Name(false), errors.ErrInvalid)
@@ -137,6 +134,9 @@ func (ib *ParamIntervalBuilder[T, K]) buildCond(cond *Condition) ([]intervals.In
 	if dp2.Flags&PfNop != 0 {
 		return nil, fmt.Errorf("the second parameter %s must allow operation (%s): %w", p2.Name(false), cond.Op, errors.ErrInvalid)
 	}
+	if p2.Const == nil { // skip not a constant param
+		return nil, nil
+	}
 
 	// operation
 	if !ib.ops[cond.Op] { // skip not the ops we look for

pkg/storage/buntdb/buntdb.go

@@ -58,8 +58,6 @@ type (
 	}
 )
 
-const maxULID = "7ZZZZZZZZZZZZZZZZZZZZZZZZZ"
-
 // NewStorage creates new logs meta storage based on BuntDB
 func NewStorage(cfg Config) *Storage {
 	return &Storage{cfg: &cfg}
@@ -381,7 +379,7 @@ func (s *Storage) queryLogsByCondition(ctx context.Context, qr storage.QueryLogs
 	tx := mustBeginTx(s.db, false)
 	defer mustRollback(tx)
 
-	if err = tx.AscendRange("", logKey(qr.Page), logKey(maxULID), iter); err != nil {
+	if err = tx.AscendRange("", logKey(qr.Page), logKey(ulidutils.MaxULID.String()), iter); err != nil {
 		return nil, fmt.Errorf("iteration failed: %w", err)
 	}
 	if iterErr != nil {