Merge pull request #392 from go-faster/feat/absent-over-time

feat(logqlmetric): implement `absent_over_time`

dev/local/ch-logql-compliance/logql-test-queries.yml

@@ -59,13 +59,29 @@ test_cases:
         {job="varlogs"} [{{ .range }}]
       )
     variant_args: ["simpleRangeAggOp", "range"]
+  # Absent over time range aggregation.
+  # NOTE(tdakkota): We do a separate test because `absent_over_time` returns non-empty result
+  #   only if there is NO samples in step and vice versa.
+  - query: |-
+      absent_over_time(
+        {job="varlogs"} [{{ .range }}]
+      )
+    variant_args: ["range"]
+    should_be_empty: true
+  - query: |-
+      absent_over_time(
+        {job="varlogs"} |= "no way line would contain this message" [{{ .range }}]
+      )
+    variant_args: ["range"]
+  # Unwrap sampler.
   - query: |-
       {{ .unwrapRangeAggOp }}(
         {job="varlogs"} | json | unwrap status
         [{{ .range }}]
       )
     variant_args: ["unwrapRangeAggOp", "range"]
     skip_comparison: true # It seems, there is some issues with unwrap.
+
   # Vector aggregation.
   - query: |-
       {{ .simpleVecAggOp }} by (filename) (

internal/chstorage/querier_logs_node.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/go-faster/oteldb/internal/logql"
 	"github.com/go-faster/oteldb/internal/logql/logqlengine"
+	"github.com/go-faster/oteldb/internal/logql/logqlengine/logqlabels"
 	"github.com/go-faster/oteldb/internal/logstorage"
 )
 
@@ -53,7 +54,7 @@ func (n *InputNode) EvalPipeline(ctx context.Context, params logqlengine.EvalPar
 }
 
 func entryMapper(r logstorage.Record) (logqlengine.Entry, error) {
-	set := logqlengine.NewLabelSet()
+	set := logqlabels.NewLabelSet()
 	e := logqlengine.Entry{
 		Timestamp: r.Timestamp,
 		Line:      r.Body,

internal/chstorage/querier_logs_query.go

@@ -21,6 +21,7 @@ import (
 	"github.com/go-faster/oteldb/internal/iterators"
 	"github.com/go-faster/oteldb/internal/logql"
 	"github.com/go-faster/oteldb/internal/logql/logqlengine"
+	"github.com/go-faster/oteldb/internal/logql/logqlengine/logqlabels"
 	"github.com/go-faster/oteldb/internal/logql/logqlengine/logqlmetric"
 	"github.com/go-faster/oteldb/internal/logstorage"
 	"github.com/go-faster/oteldb/internal/otelstorage"
@@ -253,7 +254,7 @@ func (v *SampleQuery) Eval(ctx context.Context, q *Querier) (_ logqlengine.Sampl
 				result = append(result, logqlmetric.SampledEntry{
 					Timestamp: pcommon.NewTimestampFromTime(timestamp),
 					Sample:    sample,
-					Set:       logqlengine.AggregatedLabelsFromMap(labels),
+					Set:       logqlabels.AggregatedLabelsFromMap(labels),
 				})
 			}
 			return nil

internal/logql/logqlengine/decolorize.go

@@ -4,6 +4,7 @@ import (
 	"regexp"
 
 	"github.com/go-faster/oteldb/internal/logql"
+	"github.com/go-faster/oteldb/internal/logql/logqlengine/logqlabels"
 	"github.com/go-faster/oteldb/internal/otelstorage"
 )
 
@@ -20,6 +21,6 @@ const ansiPattern = "[\u001B\u009B][[\\]()#;?]*(?:(?:(?:[a-zA-Z\\d]*(?:;[a-zA-Z\
 var ansiRegex = regexp.MustCompile(ansiPattern)
 
 // Process implements Processor.
-func (d *Decolorize) Process(_ otelstorage.Timestamp, line string, _ LabelSet) (string, bool) {
+func (d *Decolorize) Process(_ otelstorage.Timestamp, line string, _ logqlabels.LabelSet) (string, bool) {
 	return ansiRegex.ReplaceAllString(line, ""), true
 }

internal/logql/logqlengine/decolorize_test.go

@@ -7,6 +7,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/go-faster/oteldb/internal/logql"
+	"github.com/go-faster/oteldb/internal/logql/logqlengine/logqlabels"
 )
 
 func TestDecolorize(t *testing.T) {
@@ -26,7 +27,7 @@ func TestDecolorize(t *testing.T) {
 	for i, tt := range tests {
 		tt := tt
 		t.Run(fmt.Sprintf("Test%d", i+1), func(t *testing.T) {
-			set := NewLabelSet()
+			set := logqlabels.NewLabelSet()
 
 			f, err := buildDecolorize(&logql.DecolorizeExpr{})
 			require.NoError(t, err)

internal/logql/logqlengine/distinct.go

@@ -2,6 +2,7 @@ package logqlengine
 
 import (
 	"github.com/go-faster/oteldb/internal/logql"
+	"github.com/go-faster/oteldb/internal/logql/logqlengine/logqlabels"
 	"github.com/go-faster/oteldb/internal/otelstorage"
 )
 
@@ -24,7 +25,7 @@ func buildDistinctFilter(stage *logql.DistinctFilter) (Processor, error) {
 }
 
 // Process implements Processor.
-func (d *DistinctFilter) Process(_ otelstorage.Timestamp, line string, set LabelSet) (_ string, keep bool) {
+func (d *DistinctFilter) Process(_ otelstorage.Timestamp, line string, set logqlabels.LabelSet) (_ string, keep bool) {
 	for _, label := range d.labels {
 		val, ok := set.GetString(label)
 		if !ok {

internal/logql/logqlengine/drop.go

@@ -4,6 +4,7 @@ import (
 	"go.opentelemetry.io/collector/pdata/pcommon"
 
 	"github.com/go-faster/oteldb/internal/logql"
+	"github.com/go-faster/oteldb/internal/logql/logqlengine/logqlabels"
 	"github.com/go-faster/oteldb/internal/otelstorage"
 )
 
@@ -36,7 +37,7 @@ func buildDropLabels(stage *logql.DropLabelsExpr) (Processor, error) {
 }
 
 // Process implements Processor.
-func (k *DropLabels) Process(_ otelstorage.Timestamp, line string, set LabelSet) (string, bool) {
+func (k *DropLabels) Process(_ otelstorage.Timestamp, line string, set logqlabels.LabelSet) (string, bool) {
 	set.Range(func(label logql.Label, val pcommon.Value) {
 		if k.dropPair(label, val) {
 			set.Delete(label)

internal/logql/logqlengine/drop_test.go

@@ -87,8 +87,7 @@ func TestDropLabels(t *testing.T) {
 			})
 			require.NoError(t, err)
 
-			set := NewLabelSet()
-			set.labels = tt.input
+			set := newLabelSet(tt.input)
 			newLine, ok := e.Process(0, ``, set)
 			// Ensure that processor does not change the line.
 			require.Equal(t, ``, newLine)

internal/logql/logqlengine/engine_plan.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/go-faster/oteldb/internal/iterators"
+	"github.com/go-faster/oteldb/internal/logql/logqlengine/logqlabels"
 	"github.com/go-faster/oteldb/internal/logql/logqlengine/logqlmetric"
 	"github.com/go-faster/oteldb/internal/lokiapi"
 	"github.com/go-faster/oteldb/internal/otelstorage"
@@ -38,7 +39,7 @@ const (
 type Entry struct {
 	Timestamp otelstorage.Timestamp
 	Line      string
-	Set       LabelSet
+	Set       logqlabels.LabelSet
 }
 
 type (

internal/logql/logqlengine/engine_test.go

@@ -17,6 +17,7 @@ import (
 
 	"github.com/go-faster/oteldb/internal/iterators"
 	"github.com/go-faster/oteldb/internal/logql"
+	"github.com/go-faster/oteldb/internal/logql/logqlengine/logqlabels"
 	"github.com/go-faster/oteldb/internal/logstorage"
 	"github.com/go-faster/oteldb/internal/lokiapi"
 	"github.com/go-faster/oteldb/internal/otelstorage"
@@ -161,7 +162,7 @@ func (n *mockPipelineNode) EvalPipeline(ctx context.Context, params EvalParams)
 			}
 		}
 
-		set := NewLabelSet()
+		set := logqlabels.NewLabelSet()
 		set.SetAttrs(
 			otelstorage.Attrs(attrs),
 			otelstorage.Attrs(scopeAttrs),

internal/logql/logqlengine/json.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/go-faster/oteldb/internal/logql"
 	"github.com/go-faster/oteldb/internal/logql/logqlengine/jsonexpr"
+	"github.com/go-faster/oteldb/internal/logql/logqlengine/logqlabels"
 	"github.com/go-faster/oteldb/internal/otelstorage"
 )
 
@@ -52,7 +53,7 @@ func buildJSONExtractor(stage *logql.JSONExpressionParser) (Processor, error) {
 }
 
 // Process implements Processor.
-func (e *JSONExtractor) Process(_ otelstorage.Timestamp, line string, set LabelSet) (string, bool) {
+func (e *JSONExtractor) Process(_ otelstorage.Timestamp, line string, set logqlabels.LabelSet) (string, bool) {
 	var err error
 	switch {
 	case len(e.paths) != 0:
@@ -68,7 +69,7 @@ func (e *JSONExtractor) Process(_ otelstorage.Timestamp, line string, set LabelS
 	return line, true
 }
 
-func extractExprs(paths map[logql.Label]jsonexpr.Path, line string, set LabelSet) error {
+func extractExprs(paths map[logql.Label]jsonexpr.Path, line string, set logqlabels.LabelSet) error {
 	// TODO(tdakkota): allocates buffer for each line.
 	d := decodeStr(line)
 	return jsonexpr.Extract(
@@ -80,7 +81,7 @@ func extractExprs(paths map[logql.Label]jsonexpr.Path, line string, set LabelSet
 	)
 }
 
-func extractSome(labels map[logql.Label]struct{}, line string, set LabelSet) error {
+func extractSome(labels map[logql.Label]struct{}, line string, set logqlabels.LabelSet) error {
 	d := decodeStr(line)
 	return d.ObjBytes(func(d *jx.Decoder, key []byte) error {
 		if _, ok := labels[logql.Label(key)]; !ok {
@@ -101,7 +102,7 @@ func extractSome(labels map[logql.Label]struct{}, line string, set LabelSet) err
 	})
 }
 
-func extractAll(line string, set LabelSet) error {
+func extractAll(line string, set logqlabels.LabelSet) error {
 	d := decodeStr(line)
 	return d.Obj(func(d *jx.Decoder, key string) error {
 		value, ok, err := parseValue(d)

internal/logql/logqlengine/json_test.go

@@ -8,6 +8,7 @@ import (
 	"go.opentelemetry.io/collector/pdata/pcommon"
 
 	"github.com/go-faster/oteldb/internal/logql"
+	"github.com/go-faster/oteldb/internal/logql/logqlengine/logqlabels"
 )
 
 func TestJSONExtractor(t *testing.T) {
@@ -90,7 +91,7 @@ func TestJSONExtractor(t *testing.T) {
 			})
 			require.NoError(t, err)
 
-			set := NewLabelSet()
+			set := logqlabels.NewLabelSet()
 			newLine, ok := e.Process(0, tt.input, set)
 			// Ensure that extractor does not change the line.
 			require.Equal(t, tt.input, newLine)
@@ -104,7 +105,7 @@ func TestJSONExtractor(t *testing.T) {
 			errMsg, ok := set.GetError()
 			require.False(t, ok, "got error: %s", errMsg)
 
-			require.Len(t, set.labels, len(tt.expectLabels))
+			require.Equal(t, len(tt.expectLabels), set.Len())
 			for k, expect := range tt.expectLabels {
 				got, ok := set.Get(k)
 				require.Truef(t, ok, "key %q", k)
@@ -172,7 +173,7 @@ func BenchmarkJSONExtractor(b *testing.B) {
 			p, err := buildJSONExtractor(bb.expr)
 			require.NoError(b, err)
 
-			set := NewLabelSet()
+			set := logqlabels.NewLabelSet()
 			var (
 				line string
 				ok   bool
@@ -182,7 +183,7 @@ func BenchmarkJSONExtractor(b *testing.B) {
 			b.ResetTimer()
 
 			for i := 0; i < b.N; i++ {
-				set.reset()
+				set.Reset()
 				line, ok = p.Process(10, benchdata, set)
 			}
 

internal/logql/logqlengine/keep.go

@@ -4,6 +4,7 @@ import (
 	"go.opentelemetry.io/collector/pdata/pcommon"
 
 	"github.com/go-faster/oteldb/internal/logql"
+	"github.com/go-faster/oteldb/internal/logql/logqlengine/logqlabels"
 	"github.com/go-faster/oteldb/internal/otelstorage"
 )
 
@@ -36,7 +37,7 @@ func buildKeepLabels(stage *logql.KeepLabelsExpr) (Processor, error) {
 }
 
 // Process implements Processor.
-func (k *KeepLabels) Process(_ otelstorage.Timestamp, line string, set LabelSet) (string, bool) {
+func (k *KeepLabels) Process(_ otelstorage.Timestamp, line string, set logqlabels.LabelSet) (string, bool) {
 	set.Range(func(label logql.Label, val pcommon.Value) {
 		if !k.keepPair(label, val) {
 			set.Delete(label)

internal/logql/logqlengine/keep_test.go

@@ -84,8 +84,7 @@ func TestKeepLabels(t *testing.T) {
 			})
 			require.NoError(t, err)
 
-			set := NewLabelSet()
-			set.labels = tt.input
+			set := newLabelSet(tt.input)
 			newLine, ok := e.Process(0, ``, set)
 			// Ensure that processor does not change the line.
 			require.Equal(t, ``, newLine)

internal/logql/logqlengine/label_filter.go

@@ -9,6 +9,7 @@ import (
 	"github.com/go-faster/errors"
 
 	"github.com/go-faster/oteldb/internal/logql"
+	"github.com/go-faster/oteldb/internal/logql/logqlengine/logqlabels"
 	"github.com/go-faster/oteldb/internal/otelstorage"
 )
 
@@ -67,7 +68,7 @@ type AndLabelMatcher struct {
 }
 
 // Process implements Processor.
-func (m *AndLabelMatcher) Process(ts otelstorage.Timestamp, line string, set LabelSet) (_ string, keep bool) {
+func (m *AndLabelMatcher) Process(ts otelstorage.Timestamp, line string, set logqlabels.LabelSet) (_ string, keep bool) {
 	line, keep = m.Left.Process(ts, line, set)
 	if !keep {
 		return line, keep
@@ -82,7 +83,7 @@ type OrLabelMatcher struct {
 }
 
 // Process implements Processor.
-func (m *OrLabelMatcher) Process(ts otelstorage.Timestamp, line string, set LabelSet) (_ string, keep bool) {
+func (m *OrLabelMatcher) Process(ts otelstorage.Timestamp, line string, set logqlabels.LabelSet) (_ string, keep bool) {
 	line, keep = m.Left.Process(ts, line, set)
 	if keep {
 		return line, keep
@@ -109,7 +110,7 @@ func buildLabelMatcher(pred logql.LabelMatcher) (Processor, error) {
 }
 
 // Process implements Processor.
-func (lf *LabelMatcher) Process(_ otelstorage.Timestamp, line string, set LabelSet) (_ string, keep bool) {
+func (lf *LabelMatcher) Process(_ otelstorage.Timestamp, line string, set logqlabels.LabelSet) (_ string, keep bool) {
 	labelValue, _ := set.GetString(lf.name)
 	keep = lf.matcher.Match(labelValue)
 	return line, keep
@@ -160,7 +161,7 @@ func buildDurationLabelFilter(pred *logql.DurationFilter) (Processor, error) {
 }
 
 // Process implements Processor.
-func (lf *DurationLabelFilter[C]) Process(_ otelstorage.Timestamp, line string, set LabelSet) (_ string, keep bool) {
+func (lf *DurationLabelFilter[C]) Process(_ otelstorage.Timestamp, line string, set logqlabels.LabelSet) (_ string, keep bool) {
 	v, ok := set.GetString(lf.name)
 	if !ok {
 		return "", false
@@ -222,7 +223,7 @@ func buildBytesLabelFilter(pred *logql.BytesFilter) (Processor, error) {
 }
 
 // Process implements Processor.
-func (lf *BytesLabelFilter[C]) Process(_ otelstorage.Timestamp, line string, set LabelSet) (_ string, keep bool) {
+func (lf *BytesLabelFilter[C]) Process(_ otelstorage.Timestamp, line string, set logqlabels.LabelSet) (_ string, keep bool) {
 	v, ok := set.GetString(lf.name)
 	if !ok {
 		return "", false
@@ -284,7 +285,7 @@ func buildNumberLabelFilter(pred *logql.NumberFilter) (Processor, error) {
 }
 
 // Process implements Processor.
-func (lf *NumberLabelFilter[C]) Process(_ otelstorage.Timestamp, line string, set LabelSet) (_ string, keep bool) {
+func (lf *NumberLabelFilter[C]) Process(_ otelstorage.Timestamp, line string, set logqlabels.LabelSet) (_ string, keep bool) {
 	switch val, ok, err := set.GetFloat(lf.name); {
 	case err != nil:
 		// Keep the line, but set error label.
@@ -314,7 +315,7 @@ func buildIPLabelFilter(pred *logql.IPFilter) (Processor, error) {
 }
 
 // Process implements Processor.
-func (lf *IPLabelFilter) Process(_ otelstorage.Timestamp, line string, set LabelSet) (_ string, keep bool) {
+func (lf *IPLabelFilter) Process(_ otelstorage.Timestamp, line string, set logqlabels.LabelSet) (_ string, keep bool) {
 	v, ok := set.GetString(lf.name)
 	if !ok {
 		return "", false