[PLAT-108920] adding another dedup iterator to merge samples instead of skipping them (#41)

Author: jnyi

cmd/thanos/query.go

@@ -127,6 +127,9 @@ func registerQuery(app *extkingpin.App) {
 	queryReplicaLabels := cmd.Flag("query.replica-label", "Labels to treat as a replica indicator along which data is deduplicated. Still you will be able to query without deduplication using 'dedup=false' parameter. Data includes time series, recording rules, and alerting rules.").
 		Strings()
 
+	enableDedupMerge := cmd.Flag("query.dedup-merge", "Enable deduplication merge of multiple time series with the same labels.").
+		Default("false").Bool()
+
 	instantDefaultMaxSourceResolution := extkingpin.ModelDuration(cmd.Flag("query.instant.default.max_source_resolution", "default value for max_source_resolution for instant queries. If not set, defaults to 0s only taking raw resolution into account. 1h can be a good value if you use instant queries over time ranges that incorporate times outside of your raw-retention.").Default("0s").Hidden())
 
 	defaultMetadataTimeRange := cmd.Flag("query.metadata.default-time-range", "The default metadata time range duration for retrieving labels through Labels and Series API when the range parameters are not specified. The zero value means range covers the time since the beginning.").Default("0s").Duration()
@@ -374,6 +377,7 @@ func registerQuery(app *extkingpin.App) {
 			*enforceTenancy,
 			*tenantLabel,
 			*enableGroupReplicaPartialStrategy,
+			*enableDedupMerge,
 		)
 	})
 }
@@ -457,6 +461,7 @@ func runQuery(
 	enforceTenancy bool,
 	tenantLabel string,
 	groupReplicaPartialResponseStrategy bool,
+	enableDedupMerge bool,
 ) error {
 	if alertQueryURL == "" {
 		lastColon := strings.LastIndex(httpBindAddr, ":")
@@ -566,24 +571,19 @@ func runQuery(
 		exemplarsProxy   = exemplars.NewProxy(logger, endpoints.GetExemplarsStores, selectorLset)
 		queryableCreator query.QueryableCreator
 	)
-	if groupReplicaPartialResponseStrategy {
-		level.Info(logger).Log("msg", "Enabled group-replica partial response strategy")
-		queryableCreator = query.NewQueryableCreatorWithGroupReplicaPartialResponseStrategy(
-			logger,
-			extprom.WrapRegistererWithPrefix("thanos_query_", reg),
-			proxy,
-			maxConcurrentSelects,
-			queryTimeout,
-		)
-	} else {
-		queryableCreator = query.NewQueryableCreator(
-			logger,
-			extprom.WrapRegistererWithPrefix("thanos_query_", reg),
-			proxy,
-			maxConcurrentSelects,
-			queryTimeout,
-		)
+	opts := query.Options{
+		GroupReplicaPartialResponseStrategy: groupReplicaPartialResponseStrategy,
+		EnableDedupMerge:                    enableDedupMerge,
 	}
+	level.Info(logger).Log("msg", "databricks querier features", "opts", opts)
+	queryableCreator = query.NewQueryableCreatorWithOptions(
+		logger,
+		extprom.WrapRegistererWithPrefix("thanos_query_", reg),
+		proxy,
+		maxConcurrentSelects,
+		queryTimeout,
+		opts,
+	)
 
 	// Run File Service Discovery and update the store set when the files are modified.
 	if fileSD != nil {

pkg/dedup/iter.go

@@ -11,13 +11,20 @@ import (
 	"github.com/prometheus/prometheus/storage"
 	"github.com/prometheus/prometheus/tsdb/chunkenc"
 	"github.com/prometheus/prometheus/util/annotations"
-
 	"github.com/thanos-io/thanos/pkg/store/storepb"
 )
 
+// For the series we didn't pick, add a penalty twice as high as the delta of the last two
+// samples to the next seek against it.
+// This ensures that we don't pick a sample too close, which would increase the overall
+// sample frequency. It also guards against clock drift and inaccuracies during
+// timestamp assignment.
+// If we don't know a delta yet, we pick 5000 as a constant, which is based on the knowledge
+// that timestamps are in milliseconds and sampling frequencies typically multiple seconds long.
+const initialPenalty = 5000
+
 type dedupSeriesSet struct {
-	set       storage.SeriesSet
-	isCounter bool
+	set storage.SeriesSet
 
 	replicas []storage.Series
 
@@ -103,12 +110,12 @@ func (o *overlapSplitSet) Err() error {
 // NewSeriesSet returns seriesSet that deduplicates the same series.
 // The series in series set are expected be sorted by all labels.
 func NewSeriesSet(set storage.SeriesSet, f string) storage.SeriesSet {
-	// TODO: remove dependency on knowing whether it is a counter.
-	s := &dedupSeriesSet{set: set, isCounter: isCounter(f), f: f}
+	s := &dedupSeriesSet{set: set, f: f}
 	s.ok = s.set.Next()
 	if s.ok {
 		s.peek = s.set.At()
 	}
+
 	return s
 }
 
@@ -153,7 +160,10 @@ func (s *dedupSeriesSet) At() storage.Series {
 	// Clients may store the series, so we must make a copy of the slice before advancing.
 	repl := make([]storage.Series, len(s.replicas))
 	copy(repl, s.replicas)
-
+	if s.f == UseMergedSeries {
+		// merge all samples which are ingested via receiver, no skips.
+		return NewMergedSeries(s.lset, repl)
+	}
 	return newDedupSeries(s.lset, repl, s.f)
 }
 
@@ -337,15 +347,6 @@ func (it *dedupSeriesIterator) Next() chunkenc.ValueType {
 
 	it.useA = ta <= tb
 
-	// For the series we didn't pick, add a penalty twice as high as the delta of the last two
-	// samples to the next seek against it.
-	// This ensures that we don't pick a sample too close, which would increase the overall
-	// sample frequency. It also guards against clock drift and inaccuracies during
-	// timestamp assignment.
-	// If we don't know a delta yet, we pick 5000 as a constant, which is based on the knowledge
-	// that timestamps are in milliseconds and sampling frequencies typically multiple seconds long.
-	const initialPenalty = 5000
-
 	if it.useA {
 		if it.lastT != math.MinInt64 {
 			it.penB = 2 * (ta - it.lastT)

pkg/dedup/merge_iter.go

@@ -0,0 +1,109 @@
+// Copyright (c) The Thanos Authors.
+// Licensed under the Apache License 2.0.
+
+package dedup
+
+import (
+	"math"
+
+	"github.com/prometheus/prometheus/model/histogram"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/storage"
+	"github.com/prometheus/prometheus/tsdb/chunkenc"
+)
+
+const UseMergedSeries = "use_merged_series"
+
+// mergedSeries is a storage.Series that implements a simple merge sort algorithm.
+// when replicas has conflict values at the same timestamp, the first replica will be selected.
+type mergedSeries struct {
+	lset     labels.Labels
+	replicas []storage.Series
+}
+
+func NewMergedSeries(lset labels.Labels, replicas []storage.Series) storage.Series {
+	return &mergedSeries{
+		lset:     lset,
+		replicas: replicas,
+	}
+}
+
+func (m *mergedSeries) Labels() labels.Labels {
+	return m.lset
+}
+func (m *mergedSeries) Iterator(_ chunkenc.Iterator) chunkenc.Iterator {
+	iters := make([]chunkenc.Iterator, 0, len(m.replicas))
+	oks := make([]bool, 0, len(m.replicas))
+	for _, r := range m.replicas {
+		it := r.Iterator(nil)
+		ok := it.Next() != chunkenc.ValNone // iterate to the first value.
+		iters = append(iters, it)
+		oks = append(oks, ok)
+	}
+	return &mergedSeriesIterator{
+		iters:    iters,
+		oks:      oks,
+		lastT:    math.MinInt64,
+		lastIter: nil, // behavior is undefined if At() is called before Next(), here we panic if it happens.
+	}
+}
+
+type mergedSeriesIterator struct {
+	iters []chunkenc.Iterator
+	oks   []bool
+
+	lastT    int64
+	lastIter chunkenc.Iterator
+}
+
+func (m *mergedSeriesIterator) Next() chunkenc.ValueType {
+	return m.Seek(m.lastT + initialPenalty) // apply penalty to avoid selecting samples too close
+}
+
+func (m *mergedSeriesIterator) Seek(t int64) chunkenc.ValueType {
+	if len(m.iters) == 0 {
+		return chunkenc.ValNone
+	}
+
+	picked := int64(math.MaxInt64)
+	for i, it := range m.iters {
+		if !m.oks[i] {
+			continue
+		}
+		if it == m.lastIter || it.AtT() <= m.lastT {
+			m.oks[i] = it.Seek(t) != chunkenc.ValNone // move forward for last iterator.
+			if !m.oks[i] {
+				continue
+			}
+		}
+		currT := it.AtT()
+		if currT >= t && currT < picked {
+			picked = currT
+			m.lastIter = it
+		}
+	}
+	if picked == math.MaxInt64 {
+		return chunkenc.ValNone
+	}
+	m.lastT = picked
+	return chunkenc.ValFloat
+}
+func (m *mergedSeriesIterator) At() (t int64, v float64) {
+	return m.lastIter.At()
+}
+
+func (it *mergedSeriesIterator) AtHistogram(h *histogram.Histogram) (int64, *histogram.Histogram) {
+	return it.lastIter.AtHistogram(h)
+}
+
+func (it *mergedSeriesIterator) AtFloatHistogram(fh *histogram.FloatHistogram) (int64, *histogram.FloatHistogram) {
+	return it.lastIter.AtFloatHistogram(fh)
+}
+
+func (it *mergedSeriesIterator) AtT() int64 {
+	return it.lastT
+}
+
+func (m *mergedSeriesIterator) Err() error {
+	return m.lastIter.Err()
+}