Repository: VictoriaMetrics/VictoriaMetrics
Pull request #6314 (2024-05-20 21:24:11)

lib/streamaggr: added aggregation windows
### Describe Your Changes

By default, stream aggregation and deduplication stores a single state per each aggregation output result.
The data for each aggregator is flushed independently once per aggregation interval. But there's no guarantee that
incoming samples with timestamps close to the aggregation interval's end will get into it. For example, when aggregating
with `interval: 1m` a data sample with timestamp 1739473078 (18:57:59) can fall into aggregation round `18:58:00` or `18:59:00`.
It depends on network lag, load, clock synchronization, etc. In most scenarios it doesn't impact aggregation or
deduplication results, which are consistent within margin of error. But for metrics represented as a collection of series,
like [histograms](https://docs.victoriametrics.com/keyconcepts/#histogram), such inaccuracy leads to invalid aggregation results.

For this case, streaming aggregation and deduplication support mode with aggregation windows for current and previous state. With this mode, flush doesn't happen immediately but is shifted by a calculated samples 
lag that improves correctness for delayed data.

![windows](https://github.com/user-attachments/assets/c5354d6d-c524-4b7d-b801-e72ef912561a)

Enabling of this mode has increased resource usage: memory usage is expected to double as aggregation will store two states
instead of one. However, this significantly improves accuracy of calculations. Aggregation windows can be enabled via
the following settings:

- `-streamAggr.enableWindows` at [single-node VictoriaMetrics](https://docs.victoriametrics.com/single-server-victoriametrics/)
  and [vmagent](https://docs.victoriametrics.com/vmagent/). At [vmagent](https://docs.victoriametrics.com/vmagent/)
  `-remoteWrite.streamAggr.enableWindows` flag can be specified individually per each `-remoteWrite.url`.
  If one of these flags is set, then all aggregators will be using fixed windows. In conjunction with `-remoteWrite.streamAggr.dedupInterval` or
  `-streamAggr.dedupInterval` fixed aggregation windows are enabled on deduplicator as well.
 - `enable_windows` option in [aggregation config](https://docs.victoriametrics.com/stream-aggregation/#stream-aggregation-config).
  It allows enabling aggregation windows for a specific aggregator.



### Checklist

The following checks are **mandatory**:

- [ ] My change adheres [VictoriaMetrics contributing guidelines](https://docs.victoriametrics.com/contributing/).


AndrewChubatiuk: > When reviewing this changes I've noticed that all `*State` changes are required in order to keep `aggrStateSize` states for each aggregation state internally. Will it be possible to store multiple aggregation states on aggregator level instead? I remember you've mentioned that some aggregation states require custom logic for the flush depending on which state should be flushed, but I failed to find states where this is needed. If you can point me to an aggregation states which require custom logic when flushing one of the internals states that would be great. This will also mean that this is not possible and you can skip this comment right away.
> 
> Here is what I'm thinking about: An `aggregator` will keep track of `aggrStateSize` states, it will calculate an idx of state which should be used to flush and push samples with the given timestamp and will handle state reset accordingly. This will require each aggregation state to implement a `reset()` to properly reset the state. Compared to amount of changes to support storing `aggrStateSize` states it should require fewer changes and code duplication.

histogram_buckets, rate, total aggregators persist a shared state for all windows, which differs from other aggregators, which persist a state till window is flushed. I'm not against moving output states to an aggregator's level, but first want to be sure once this PR (or similar approach) will be merged not to waist to much time 
hagen1778: Related tickets 
https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6424 
https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4768
https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4580
AndrewChubatiuk: > As far as I understand, mostly those changes is needed only for histograms. Is it possible to implement it only for histogram_bucket instead of changing other aggregations?

this change is needed to apply aggregations against histogram series, not generate histogram aggregation
hagen1778: LGTM. Have to run a few tests. Approx will merge it tomorrow.
hagen1778: Thanks @AndrewChubatiuk for your work!

Patch:
diff --git a/app/vmagent/remotewrite/streamaggr.go b/app/vmagent/remotewrite/streamaggr.go
index a3b2e700e..71255cfc4 100644
--- a/app/vmagent/remotewrite/streamaggr.go
+++ b/app/vmagent/remotewrite/streamaggr.go
@@ -35,6 +35,9 @@ var (
 		"clients pushing data into the vmagent. See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start")
 	streamAggrGlobalDropInputLabels = flagutil.NewArrayString("streamAggr.dropInputLabels", "An optional list of labels to drop from samples for aggregator "+
 		"before stream de-duplication and aggregation . See https://docs.victoriametrics.com/stream-aggregation/#dropping-unneeded-labels")
+	streamAggrGlobalEnableWindows = flag.Bool("streamAggr.enableWindows", false, "Enables aggregation within fixed windows for all global aggregators. "+
+		"This allows to get more precise results, but impacts resource usage as it requires twice more memory to store two states. "+
+		"See https://docs.victoriametrics.com/stream-aggregation/#aggregation-windows.")
 
 	// Per URL config
 	streamAggrConfig = flagutil.NewArrayString("remoteWrite.streamAggr.config", "Optional path to file with stream aggregation config for the corresponding -remoteWrite.url. "+
@@ -59,6 +62,9 @@ var (
 		"before stream de-duplication and aggregation with -remoteWrite.streamAggr.config and -remoteWrite.streamAggr.dedupInterval at the corresponding -remoteWrite.url. "+
 		"Multiple labels per remoteWrite.url must be delimited by '^^': -remoteWrite.streamAggr.dropInputLabels='replica^^az,replica'. "+
 		"See https://docs.victoriametrics.com/stream-aggregation/#dropping-unneeded-labels")
+	streamAggrEnableWindows = flagutil.NewArrayBool("remoteWrite.streamAggr.enableWindows", "Enables aggregation within fixed windows for all remote write's aggregators. "+
+		"This allows to get more precise results, but impacts resource usage as it requires twice more memory to store two states. "+
+		"See https://docs.victoriametrics.com/stream-aggregation/#aggregation-windows.")
 )
 
 // CheckStreamAggrConfigs checks -remoteWrite.streamAggr.config and -streamAggr.config.
@@ -135,7 +141,7 @@ func initStreamAggrConfigGlobal() {
 	}
 	dedupInterval := *streamAggrGlobalDedupInterval
 	if dedupInterval > 0 {
-		deduplicatorGlobal = streamaggr.NewDeduplicator(pushToRemoteStoragesTrackDropped, dedupInterval, *streamAggrGlobalDropInputLabels, "dedup-global")
+		deduplicatorGlobal = streamaggr.NewDeduplicator(pushToRemoteStoragesTrackDropped, *streamAggrGlobalEnableWindows, dedupInterval, *streamAggrGlobalDropInputLabels, "dedup-global")
 	}
 }
 
@@ -161,7 +167,7 @@ func (rwctx *remoteWriteCtx) initStreamAggrConfig() {
 		if streamAggrDropInputLabels.GetOptionalArg(idx) != "" {
 			dropLabels = strings.Split(streamAggrDropInputLabels.GetOptionalArg(idx), "^^")
 		}
-		rwctx.deduplicator = streamaggr.NewDeduplicator(rwctx.pushInternalTrackDropped, dedupInterval, dropLabels, alias)
+		rwctx.deduplicator = streamaggr.NewDeduplicator(rwctx.pushInternalTrackDropped, *streamAggrGlobalEnableWindows, dedupInterval, dropLabels, alias)
 	}
 }
 
@@ -207,6 +213,7 @@ func newStreamAggrConfigGlobal() (*streamaggr.Aggregators, error) {
 		IgnoreOldSamples:     *streamAggrGlobalIgnoreOldSamples,
 		IgnoreFirstIntervals: *streamAggrGlobalIgnoreFirstIntervals,
 		KeepInput:            *streamAggrGlobalKeepInput,
+		EnableWindows:        *streamAggrGlobalEnableWindows,
 	}
 
 	sas, err := streamaggr.LoadFromFile(path, pushToRemoteStoragesTrackDropped, opts, "global")
@@ -240,6 +247,7 @@ func newStreamAggrConfigPerURL(idx int, pushFunc streamaggr.PushFunc) (*streamag
 		IgnoreOldSamples:     streamAggrIgnoreOldSamples.GetOptionalArg(idx),
 		IgnoreFirstIntervals: streamAggrIgnoreFirstIntervals.GetOptionalArg(idx),
 		KeepInput:            streamAggrKeepInput.GetOptionalArg(idx),
+		EnableWindows:        streamAggrEnableWindows.GetOptionalArg(idx),
 	}
 
 	sas, err := streamaggr.LoadFromFile(path, pushFunc, opts, alias)
diff --git a/app/vminsert/common/streamaggr.go b/app/vminsert/common/streamaggr.go
index 3cc649c52..f4f932e36 100644
--- a/app/vminsert/common/streamaggr.go
+++ b/app/vminsert/common/streamaggr.go
@@ -36,6 +36,9 @@ var (
 		"See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples")
 	streamAggrIgnoreFirstIntervals = flag.Int("streamAggr.ignoreFirstIntervals", 0, "Number of aggregation intervals to skip after the start. Increase this value if you observe incorrect aggregation results after restarts. It could be caused by receiving unordered delayed data from clients pushing data into the database. "+
 		"See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start")
+	streamAggrEnableWindows = flag.Bool("streamAggr.enableWindows", false, "Enables aggregation within fixed windows for all aggregators. "+
+		"This allows to get more precise results, but impacts resource usage as it requires twice more memory to store two states. "+
+		"See https://docs.victoriametrics.com/stream-aggregation/#aggregation-windows.")
 )
 
 var (
@@ -62,6 +65,7 @@ func CheckStreamAggrConfig() error {
 		DropInputLabels:      *streamAggrDropInputLabels,
 		IgnoreOldSamples:     *streamAggrIgnoreOldSamples,
 		IgnoreFirstIntervals: *streamAggrIgnoreFirstIntervals,
+		EnableWindows:        *streamAggrEnableWindows,
 	}
 	sas, err := streamaggr.LoadFromFile(*streamAggrConfig, pushNoop, opts, "global")
 	if err != nil {
@@ -78,7 +82,7 @@ func InitStreamAggr() {
 	saCfgReloaderStopCh = make(chan struct{})
 	if *streamAggrConfig == "" {
 		if *streamAggrDedupInterval > 0 {
-			deduplicator = streamaggr.NewDeduplicator(pushAggregateSeries, *streamAggrDedupInterval, *streamAggrDropInputLabels, "global")
+			deduplicator = streamaggr.NewDeduplicator(pushAggregateSeries, *streamAggrEnableWindows, *streamAggrDedupInterval, *streamAggrDropInputLabels, "global")
 		}
 		return
 	}
diff --git a/docs/README.md b/docs/README.md
index 52e85c3bd..86b0f0e34 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -3381,6 +3381,8 @@ Pass `-help` to VictoriaMetrics in order to see the list of supported command-li
      An optional list of labels to drop from samples before stream de-duplication and aggregation . See https://docs.victoriametrics.com/stream-aggregation/#dropping-unneeded-labels
      Supports an array of values separated by comma or specified via multiple flags.
      Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
+  -streamAggr.enableWindows
+      Enables aggregation within fixed windows for all aggregators. This allows to get more precise results, but impacts resource usage as it requires twice more memory to store two states. See https://docs.victoriametrics.com/stream-aggregation/#aggregation-windows.  
   -streamAggr.ignoreFirstIntervals int
      Number of aggregation intervals to skip after the start. Increase this value if you observe incorrect aggregation results after restarts. It could be caused by receiving unordered delayed data from clients pushing data into the database. See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start
   -streamAggr.ignoreOldSamples
diff --git a/docs/changelog/CHANGELOG.md b/docs/changelog/CHANGELOG.md
index 8cb3403b7..4980fb9c4 100644
--- a/docs/changelog/CHANGELOG.md
+++ b/docs/changelog/CHANGELOG.md
@@ -24,6 +24,7 @@ See also [LTS releases](https://docs.victoriametrics.com/lts-releases/).
 * FEATURE: [Single-node VictoriaMetrics](https://docs.victoriametrics.com/) and [vmstorage](https://docs.victoriametrics.com/victoriametrics/): allow disabling per-day indexes for workloads with [low or zero churn rates](https://docs.victoriametrics.com/#index-tuning-for-low-churn-rate) via `-disablePerDayIndex` cmd-line flag. This option works the best for IoT cases and significantly loads resource and disk usage. See [this PR](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6976).
 * FEATURE: [vmui](https://docs.victoriametrics.com/#vmui): look back for recent data for longer time intervals when switching to Table or JSON view. The look back window depends on the `step` setting, which is now set as `end - start` (30m) by default. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8240).
 * FEATURE: [vmui](https://docs.victoriametrics.com/#vmui): preserve user-defined `step` setting when changing the time interval of displayed query. See [this comment](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8240#issuecomment-2647674065).
+* FEATURE: [vmagent]((https://docs.victoriametrics.com/vmagent/) and [vmsingle](https://docs.victoriametrics.com/single-server-victoriametrics/): add support of [aggregation windows](https://docs.victoriametrics.com/stream-aggregation/#aggregation-windows) to improve accuracy of stream aggregation. This feature is especially important for aggregation of [histrograms](https://docs.victoriametrics.com/keyconcepts/#histogram), but enabling this feature requires additional memory. See [this issue]((https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4580).
 
 * BUGFIX: all the VictoriaMetrics components: properly override basic authorization for API endpoints protected with `authKey`. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7345#issuecomment-2662595807) for details.
 * BUGFIX: [vmalert](https://docs.victoriametrics.com/vmalert/): fix polluted alert messages when multiple Alertmanager instances are configured with `alert_relabel_configs`. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8040), and thanks to @evkuzin for [the pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/8258).
diff --git a/docs/stream-aggregation.md b/docs/stream-aggregation.md
index 784cbb357..3a2332914 100644
--- a/docs/stream-aggregation.md
+++ b/docs/stream-aggregation.md
@@ -291,6 +291,7 @@ metrics with different `vmrange` or `le` labels. As they're counters, the applic
 - match: 'http_request_duration_seconds_bucket'
   interval: 1m
   without: [instance]
+  enable_windows: true
   outputs: [total]
 ```
 
@@ -316,6 +317,8 @@ Please note, histograms can be aggregated if their `le` labels are configured id
 [VictoriaMetrics histogram buckets](https://valyala.medium.com/improving-histogram-usability-for-prometheus-and-grafana-bc7e5df0e350)
 have no such requirement.
 
+It's recommended to use [aggregation windows](#aggregation-windows) when aggregating histograms if you observe [accuracy issues](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4580).
+
 See [the list of aggregate output](#aggregation-outputs), which can be specified at `output` field.
 See also [histograms over input metrics](#histograms-over-input-metrics) and [quantiles over input metrics](#quantiles-over-input-metrics).
 
@@ -808,6 +811,11 @@ specified individually per each `-remoteWrite.url`:
   #
   # dedup_interval: 30s
 
+  # enable_windows is a boolean option to enable fixed aggregation windows.
+  # See https://docs.victoriametrics.com/stream-aggregation#aggregation-windows
+  #
+  # enable_windows: true
+
   # staleness_interval is an optional interval for resetting the per-series state if no new samples
   # are received during this interval for the following outputs:
   # - histogram_bucket
@@ -1143,6 +1151,33 @@ Typical use case is to drop `replica` label from samples, which are received fro
 - [Lower than expected values for `total_prometheus` and `increase_prometheus` outputs](#staleness).
 - [High memory usage and CPU usage](#high-resource-usage).
 - [Unexpected results in vmagent cluster mode](#cluster-mode).
+- [Inaccurate aggregation results for histograms](#aggregation-windows)
+
+## Aggregation windows
+
+By default, stream aggregation and deduplication stores a single state per each aggregation output result.
+The data for each aggregator is flushed independently once per aggregation interval. But there's no guarantee that
+incoming samples with timestamps close to the aggregation interval's end will get into it. For example, when aggregating
+with `interval: 1m` a data sample with timestamp 1739473078 (18:57:59) can fall into aggregation round `18:58:00` or `18:59:00`.
+It depends on network lag, load, clock synchronization, etc. In most scenarios it doesn't impact aggregation or
+deduplication results, which are consistent within margin of error. But for metrics represented as a collection of series,
+like [histograms](https://docs.victoriametrics.com/keyconcepts/#histogram), such inaccuracy leads to invalid aggregation results.
+
+For this case, streaming aggregation and deduplication support mode with aggregation windows {{% available_from "#tip" %}}
+for current and previous state. With this mode, flush doesn't happen immediately but is shifted by a calculated samples 
+lag that improves correctness for delayed data.
+
+Enabling of this mode has increased resource usage: memory usage is expected to double as aggregation will store two states
+instead of one. However, this significantly improves accuracy of calculations. Aggregation windows can be enabled via
+the following settings:
+
+- `-streamAggr.enableWindows` at [single-node VictoriaMetrics](https://docs.victoriametrics.com/single-server-victoriametrics/)
+  and [vmagent](https://docs.victoriametrics.com/vmagent/). At [vmagent](https://docs.victoriametrics.com/vmagent/)
+  `-remoteWrite.streamAggr.enableWindows` flag can be specified individually per each `-remoteWrite.url`.
+  If one of these flags is set, then all aggregators will be using fixed windows. In conjunction with `-remoteWrite.streamAggr.dedupInterval` or
+  `-streamAggr.dedupInterval` fixed aggregation windows are enabled on deduplicator as well.
+ - `enable_windows` option in [aggregation config](https://docs.victoriametrics.com/stream-aggregation/#stream-aggregation-config).
+  It allows enabling aggregation windows for a specific aggregator.
 
 ## Staleness
 
diff --git a/docs/vmagent.md b/docs/vmagent.md
index a7ba33220..a63422eca 100644
--- a/docs/vmagent.md
+++ b/docs/vmagent.md
@@ -2270,6 +2270,10 @@ See the docs at https://docs.victoriametrics.com/vmagent/ .
      An optional list of labels to drop from samples before stream de-duplication and aggregation with -remoteWrite.streamAggr.config and -remoteWrite.streamAggr.dedupInterval at the corresponding -remoteWrite.url. Multiple labels per remoteWrite.url must be delimited by '^^': -remoteWrite.streamAggr.dropInputLabels='replica^^az,replica'. See https://docs.victoriametrics.com/stream-aggregation/#dropping-unneeded-labels
      Supports an array of values separated by comma or specified via multiple flags.
      Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
+  -remoteWrite.streamAggr.enableWindows array
+     Enables aggregation within fixed windows for all remote write's aggregators. This allows to get more precise results, but impacts resource usage as it requires twice more memory to store two states. See https://docs.victoriametrics.com/stream-aggregation/#aggregation-windows.
+     Supports array of values separated by comma or specified via multiple flags.
+     Empty values are set to false
   -remoteWrite.streamAggr.ignoreFirstIntervals array
      Number of aggregation intervals to skip after the start for the corresponding -remoteWrite.streamAggr.config at the corresponding -remoteWrite.url. Increase this value if you observe incorrect aggregation results after vmagent restarts. It could be caused by receiving buffered delayed data from clients pushing data into the vmagent. See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start
   -remoteWrite.streamAggr.ignoreOldSamples array
@@ -2329,6 +2333,8 @@ See the docs at https://docs.victoriametrics.com/vmagent/ .
     An optional list of labels to drop from samples for aggregator before stream de-duplication and aggregation . See https://docs.victoriametrics.com/stream-aggregation/#dropping-unneeded-labels
     Supports an array of values separated by comma or specified via multiple flags.
     Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
+  -streamAggr.enableWindows
+     Enables aggregation within fixed windows for all global aggregators. This allows to get more precise results, but impacts resource usage as it requires twice more memory to store two states. See https://docs.victoriametrics.com/stream-aggregation/#aggregation-windows.
   -streamAggr.ignoreFirstIntervals int
     Number of aggregation intervals to skip after the start for aggregator. Increase this value if you observe incorrect aggregation results after vmagent restarts. It could be caused by receiving unordered delayed data from clients pushing data into the vmagent. See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start
   -streamAggr.ignoreOldSamples
diff --git a/lib/streamaggr/avg.go b/lib/streamaggr/avg.go
index 32ab569ac..27e51b408 100644
--- a/lib/streamaggr/avg.go
+++ b/lib/streamaggr/avg.go
@@ -1,80 +1,34 @@
 package streamaggr
 
-import (
-	"sync"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-)
-
-// avgAggrState calculates output=avg, e.g. the average value over input samples.
-type avgAggrState struct {
-	m sync.Map
+type avgAggrValue struct {
+	sum   float64
+	count float64
 }
 
-type avgStateValue struct {
-	mu      sync.Mutex
-	sum     float64
-	count   int64
-	deleted bool
+func (av *avgAggrValue) pushSample(_ aggrConfig, sample *pushSample, _ string, _ int64) {
+	av.sum += sample.value
+	av.count++
 }
 
-func newAvgAggrState() *avgAggrState {
-	return &avgAggrState{}
+func (av *avgAggrValue) flush(_ aggrConfig, ctx *flushCtx, key string) {
+	if av.count > 0 {
+		avg := av.sum / av.count
+		ctx.appendSeries(key, "avg", avg)
+		av.sum = 0
+		av.count = 0
+	}
 }
 
-func (as *avgAggrState) pushSamples(samples []pushSample) {
-	for i := range samples {
-		s := &samples[i]
-		outputKey := getOutputKey(s.key)
-
-	again:
-		v, ok := as.m.Load(outputKey)
-		if !ok {
-			// The entry is missing in the map. Try creating it.
-			v = &avgStateValue{
-				sum:   s.value,
-				count: 1,
-			}
-			outputKey = bytesutil.InternString(outputKey)
-			vNew, loaded := as.m.LoadOrStore(outputKey, v)
-			if !loaded {
-				// The entry has been successfully stored
-				continue
-			}
-			// Update the entry created by a concurrent goroutine.
-			v = vNew
-		}
-		sv := v.(*avgStateValue)
-		sv.mu.Lock()
-		deleted := sv.deleted
-		if !deleted {
-			sv.sum += s.value
-			sv.count++
-		}
-		sv.mu.Unlock()
-		if deleted {
-			// The entry has been deleted by the concurrent call to flushState
-			// Try obtaining and updating the entry again.
-			goto again
-		}
-	}
+func (*avgAggrValue) state() any {
+	return nil
 }
 
-func (as *avgAggrState) flushState(ctx *flushCtx) {
-	m := &as.m
-	m.Range(func(k, v any) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
+func newAvgAggrConfig() aggrConfig {
+	return &avgAggrConfig{}
+}
 
-		sv := v.(*avgStateValue)
-		sv.mu.Lock()
-		avg := sv.sum / float64(sv.count)
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-		sv.deleted = true
-		sv.mu.Unlock()
+type avgAggrConfig struct{}
 
-		key := k.(string)
-		ctx.appendSeries(key, "avg", avg)
-		return true
-	})
+func (*avgAggrConfig) getValue(_ any) aggrValue {
+	return &avgAggrValue{}
 }
diff --git a/lib/streamaggr/count_samples.go b/lib/streamaggr/count_samples.go
index ca2bc709e..3a53593bb 100644
--- a/lib/streamaggr/count_samples.go
+++ b/lib/streamaggr/count_samples.go
@@ -1,77 +1,30 @@
 package streamaggr
 
-import (
-	"sync"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-)
-
-// countSamplesAggrState calculates output=count_samples, e.g. the count of input samples.
-type countSamplesAggrState struct {
-	m sync.Map
+type countSamplesAggrValue struct {
+	count uint64
 }
 
-type countSamplesStateValue struct {
-	mu      sync.Mutex
-	n       uint64
-	deleted bool
+func (av *countSamplesAggrValue) pushSample(_ aggrConfig, _ *pushSample, _ string, _ int64) {
+	av.count++
 }
 
-func newCountSamplesAggrState() *countSamplesAggrState {
-	return &countSamplesAggrState{}
+func (av *countSamplesAggrValue) flush(_ aggrConfig, ctx *flushCtx, key string) {
+	if av.count > 0 {
+		ctx.appendSeries(key, "count_samples", float64(av.count))
+		av.count = 0
+	}
 }
 
-func (as *countSamplesAggrState) pushSamples(samples []pushSample) {
-	for i := range samples {
-		s := &samples[i]
-		outputKey := getOutputKey(s.key)
-
-	again:
-		v, ok := as.m.Load(outputKey)
-		if !ok {
-			// The entry is missing in the map. Try creating it.
-			v = &countSamplesStateValue{
-				n: 1,
-			}
-			outputKey = bytesutil.InternString(outputKey)
-			vNew, loaded := as.m.LoadOrStore(outputKey, v)
-			if !loaded {
-				// The new entry has been successfully created.
-				continue
-			}
-			// Use the entry created by a concurrent goroutine.
-			v = vNew
-		}
-		sv := v.(*countSamplesStateValue)
-		sv.mu.Lock()
-		deleted := sv.deleted
-		if !deleted {
-			sv.n++
-		}
-		sv.mu.Unlock()
-		if deleted {
-			// The entry has been deleted by the concurrent call to flushState
-			// Try obtaining and updating the entry again.
-			goto again
-		}
-	}
+func (*countSamplesAggrValue) state() any {
+	return nil
 }
 
-func (as *countSamplesAggrState) flushState(ctx *flushCtx) {
-	m := &as.m
-	m.Range(func(k, v any) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
+func newCountSamplesAggrConfig() aggrConfig {
+	return &countSamplesAggrConfig{}
+}
 
-		sv := v.(*countSamplesStateValue)
-		sv.mu.Lock()
-		n := sv.n
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-		sv.deleted = true
-		sv.mu.Unlock()
+type countSamplesAggrConfig struct{}
 
-		key := k.(string)
-		ctx.appendSeries(key, "count_samples", float64(n))
-		return true
-	})
+func (*countSamplesAggrConfig) getValue(_ any) aggrValue {
+	return &countSamplesAggrValue{}
 }
diff --git a/lib/streamaggr/count_series.go b/lib/streamaggr/count_series.go
index 816581464..b93230aec 100644
--- a/lib/streamaggr/count_series.go
+++ b/lib/streamaggr/count_series.go
@@ -1,86 +1,42 @@
 package streamaggr
 
 import (
-	"sync"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/cespare/xxhash/v2"
 )
 
-// countSeriesAggrState calculates output=count_series, e.g. the number of unique series.
-type countSeriesAggrState struct {
-	m sync.Map
+type countSeriesAggrValue struct {
+	samples map[uint64]struct{}
 }
 
-type countSeriesStateValue struct {
-	mu      sync.Mutex
-	m       map[uint64]struct{}
-	deleted bool
+func (av *countSeriesAggrValue) pushSample(_ aggrConfig, _ *pushSample, key string, _ int64) {
+	// Count unique hashes over the keys instead of unique key values.
+	// This reduces memory usage at the cost of possible hash collisions for distinct key values.
+	h := xxhash.Sum64(bytesutil.ToUnsafeBytes(key))
+	if _, ok := av.samples[h]; !ok {
+		av.samples[h] = struct{}{}
+	}
 }
 
-func newCountSeriesAggrState() *countSeriesAggrState {
-	return &countSeriesAggrState{}
+func (av *countSeriesAggrValue) flush(_ aggrConfig, ctx *flushCtx, key string) {
+	if len(av.samples) > 0 {
+		ctx.appendSeries(key, "count_series", float64(len(av.samples)))
+		clear(av.samples)
+	}
 }
 
-func (as *countSeriesAggrState) pushSamples(samples []pushSample) {
-	for i := range samples {
-		s := &samples[i]
-		inputKey, outputKey := getInputOutputKey(s.key)
-
-		// Count unique hashes over the inputKeys instead of unique inputKey values.
-		// This reduces memory usage at the cost of possible hash collisions for distinct inputKey values.
-		h := xxhash.Sum64(bytesutil.ToUnsafeBytes(inputKey))
-
-	again:
-		v, ok := as.m.Load(outputKey)
-		if !ok {
-			// The entry is missing in the map. Try creating it.
-			v = &countSeriesStateValue{
-				m: map[uint64]struct{}{
-					h: {},
-				},
-			}
-			outputKey = bytesutil.InternString(outputKey)
-			vNew, loaded := as.m.LoadOrStore(outputKey, v)
-			if !loaded {
-				// The entry has been added to the map.
-				continue
-			}
-			// Update the entry created by a concurrent goroutine.
-			v = vNew
-		}
-		sv := v.(*countSeriesStateValue)
-		sv.mu.Lock()
-		deleted := sv.deleted
-		if !deleted {
-			if _, ok := sv.m[h]; !ok {
-				sv.m[h] = struct{}{}
-			}
-		}
-		sv.mu.Unlock()
-		if deleted {
-			// The entry has been deleted by the concurrent call to flushState
-			// Try obtaining and updating the entry again.
-			goto again
-		}
-	}
+func (*countSeriesAggrValue) state() any {
+	return nil
 }
 
-func (as *countSeriesAggrState) flushState(ctx *flushCtx) {
-	m := &as.m
-	m.Range(func(k, v any) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
+func newCountSeriesAggrConfig() aggrConfig {
+	return &countSeriesAggrConfig{}
+}
 
-		sv := v.(*countSeriesStateValue)
-		sv.mu.Lock()
-		n := len(sv.m)
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-		sv.deleted = true
-		sv.mu.Unlock()
+type countSeriesAggrConfig struct{}
 
-		key := k.(string)
-		ctx.appendSeries(key, "count_series", float64(n))
-		return true
-	})
+func (*countSeriesAggrConfig) getValue(_ any) aggrValue {
+	return &countSeriesAggrValue{
+		samples: make(map[uint64]struct{}),
+	}
 }
diff --git a/lib/streamaggr/dedup.go b/lib/streamaggr/dedup.go
index 7e4df520b..d92a70ec5 100644
--- a/lib/streamaggr/dedup.go
+++ b/lib/streamaggr/dedup.go
@@ -5,6 +5,7 @@ import (
 	"sync/atomic"
 	"unsafe"
 
+	"github.com/VictoriaMetrics/metrics"
 	"github.com/cespare/xxhash/v2"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
@@ -15,7 +16,9 @@ import (
 const dedupAggrShardsCount = 128
 
 type dedupAggr struct {
-	shards []dedupAggrShard
+	shards        []dedupAggrShard
+	flushDuration *metrics.Histogram
+	flushTimeouts *metrics.Counter
 }
 
 type dedupAggrShard struct {
@@ -26,45 +29,53 @@ type dedupAggrShard struct {
 	_ [128 - unsafe.Sizeof(dedupAggrShardNopad{})%128]byte
 }
 
-type dedupAggrShardNopad struct {
-	mu sync.Mutex
-	m  map[string]*dedupAggrSample
-
+type dedupAggrState struct {
+	m          map[string]*dedupAggrSample
+	mu         sync.Mutex
 	samplesBuf []dedupAggrSample
-
 	sizeBytes  atomic.Uint64
 	itemsCount atomic.Uint64
 }
 
+type dedupAggrShardNopad struct {
+	blue  dedupAggrState
+	green dedupAggrState
+}
+
 type dedupAggrSample struct {
 	value     float64
 	timestamp int64
 }
 
 func newDedupAggr() *dedupAggr {
-	shards := make([]dedupAggrShard, dedupAggrShardsCount)
 	return &dedupAggr{
-		shards: shards,
+		shards: make([]dedupAggrShard, dedupAggrShardsCount),
 	}
 }
 
 func (da *dedupAggr) sizeBytes() uint64 {
 	n := uint64(unsafe.Sizeof(*da))
+	var shard *dedupAggrShard
 	for i := range da.shards {
-		n += da.shards[i].sizeBytes.Load()
+		shard = &da.shards[i]
+		n += shard.blue.sizeBytes.Load()
+		n += shard.green.sizeBytes.Load()
 	}
 	return n
 }
 
 func (da *dedupAggr) itemsCount() uint64 {
 	n := uint64(0)
+	var shard *dedupAggrShard
 	for i := range da.shards {
-		n += da.shards[i].itemsCount.Load()
+		shard = &da.shards[i]
+		n += shard.blue.itemsCount.Load()
+		n += shard.green.itemsCount.Load()
 	}
 	return n
 }
 
-func (da *dedupAggr) pushSamples(samples []pushSample) {
+func (da *dedupAggr) pushSamples(samples []pushSample, _ int64, isGreen bool) {
 	pss := getPerShardSamples()
 	shards := pss.shards
 	for _, sample := range samples {
@@ -76,17 +87,20 @@ func (da *dedupAggr) pushSamples(samples []pushSample) {
 		if len(shardSamples) == 0 {
 			continue
 		}
-		da.shards[i].pushSamples(shardSamples)
+		da.shards[i].pushSamples(shardSamples, isGreen)
 	}
 	putPerShardSamples(pss)
 }
 
-func getDedupFlushCtx() *dedupFlushCtx {
+func getDedupFlushCtx(deleteDeadline int64, isGreen bool) *dedupFlushCtx {
 	v := dedupFlushCtxPool.Get()
 	if v == nil {
-		return &dedupFlushCtx{}
+		v = &dedupFlushCtx{}
 	}
-	return v.(*dedupFlushCtx)
+	ctx := v.(*dedupFlushCtx)
+	ctx.deleteDeadline = deleteDeadline
+	ctx.isGreen = isGreen
+	return ctx
 }
 
 func putDedupFlushCtx(ctx *dedupFlushCtx) {
@@ -97,15 +111,18 @@ func putDedupFlushCtx(ctx *dedupFlushCtx) {
 var dedupFlushCtxPool sync.Pool
 
 type dedupFlushCtx struct {
-	samples []pushSample
+	samples        []pushSample
+	deleteDeadline int64
+	isGreen        bool
 }
 
 func (ctx *dedupFlushCtx) reset() {
 	clear(ctx.samples)
 	ctx.samples = ctx.samples[:0]
+	ctx.deleteDeadline = 0
 }
 
-func (da *dedupAggr) flush(f func(samples []pushSample)) {
+func (da *dedupAggr) flush(f aggrPushFunc, deleteDeadline int64, isGreen bool) {
 	var wg sync.WaitGroup
 	for i := range da.shards {
 		flushConcurrencyCh <- struct{}{}
@@ -115,8 +132,7 @@ func (da *dedupAggr) flush(f func(samples []pushSample)) {
 				<-flushConcurrencyCh
 				wg.Done()
 			}()
-
-			ctx := getDedupFlushCtx()
+			ctx := getDedupFlushCtx(deleteDeadline, isGreen)
 			shard.flush(ctx, f)
 			putDedupFlushCtx(ctx)
 		}(&da.shards[i])
@@ -155,18 +171,23 @@ func putPerShardSamples(pss *perShardSamples) {
 
 var perShardSamplesPool sync.Pool
 
-func (das *dedupAggrShard) pushSamples(samples []pushSample) {
-	das.mu.Lock()
-	defer das.mu.Unlock()
+func (das *dedupAggrShard) pushSamples(samples []pushSample, isGreen bool) {
+	var state *dedupAggrState
+
+	if isGreen {
+		state = &das.green
+	} else {
+		state = &das.blue
+	}
 
-	m := das.m
-	if m == nil {
-		m = make(map[string]*dedupAggrSample, len(samples))
-		das.m = m
+	state.mu.Lock()
+	defer state.mu.Unlock()
+	if state.m == nil {
+		state.m = make(map[string]*dedupAggrSample, len(samples))
 	}
-	samplesBuf := das.samplesBuf
+	samplesBuf := state.samplesBuf
 	for _, sample := range samples {
-		s, ok := m[sample.key]
+		s, ok := state.m[sample.key]
 		if !ok {
 			samplesBuf = slicesutil.SetLength(samplesBuf, len(samplesBuf)+1)
 			s = &samplesBuf[len(samplesBuf)-1]
@@ -174,10 +195,10 @@ func (das *dedupAggrShard) pushSamples(samples []pushSample) {
 			s.timestamp = sample.timestamp
 
 			key := bytesutil.InternString(sample.key)
-			m[key] = s
+			state.m[key] = s
 
-			das.itemsCount.Add(1)
-			das.sizeBytes.Add(uint64(len(key)) + uint64(unsafe.Sizeof(key)+unsafe.Sizeof(s)+unsafe.Sizeof(*s)))
+			state.itemsCount.Add(1)
+			state.sizeBytes.Add(uint64(len(key)) + uint64(unsafe.Sizeof(key)+unsafe.Sizeof(s)+unsafe.Sizeof(*s)))
 			continue
 		}
 		if !isDuplicate(s, sample) {
@@ -185,7 +206,7 @@ func (das *dedupAggrShard) pushSamples(samples []pushSample) {
 			s.timestamp = sample.timestamp
 		}
 	}
-	das.samplesBuf = samplesBuf
+	state.samplesBuf = samplesBuf
 }
 
 // isDuplicate returns true if b is duplicate of a
@@ -205,18 +226,24 @@ func isDuplicate(a *dedupAggrSample, b pushSample) bool {
 	return true
 }
 
-func (das *dedupAggrShard) flush(ctx *dedupFlushCtx, f func(samples []pushSample)) {
-	das.mu.Lock()
-
-	m := das.m
-	if len(m) > 0 {
-		das.m = make(map[string]*dedupAggrSample, len(m))
-		das.sizeBytes.Store(0)
-		das.itemsCount.Store(0)
-		das.samplesBuf = make([]dedupAggrSample, 0, len(das.samplesBuf))
+func (das *dedupAggrShard) flush(ctx *dedupFlushCtx, f aggrPushFunc) {
+	var m map[string]*dedupAggrSample
+	var state *dedupAggrState
+	if ctx.isGreen {
+		state = &das.green
+	} else {
+		state = &das.blue
 	}
 
-	das.mu.Unlock()
+	state.mu.Lock()
+	if len(state.m) > 0 {
+		m = state.m
+		state.m = make(map[string]*dedupAggrSample, len(state.m))
+		state.samplesBuf = make([]dedupAggrSample, 0, len(state.samplesBuf))
+		state.sizeBytes.Store(0)
+		state.itemsCount.Store(0)
+	}
+	state.mu.Unlock()
 
 	if len(m) == 0 {
 		return
@@ -232,11 +259,11 @@ func (das *dedupAggrShard) flush(ctx *dedupFlushCtx, f func(samples []pushSample
 
 		// Limit the number of samples per each flush in order to limit memory usage.
 		if len(dstSamples) >= 10_000 {
-			f(dstSamples)
+			f(dstSamples, ctx.deleteDeadline, false)
 			clear(dstSamples)
 			dstSamples = dstSamples[:0]
 		}
 	}
-	f(dstSamples)
+	f(dstSamples, ctx.deleteDeadline, false)
 	ctx.samples = dstSamples
 }
diff --git a/lib/streamaggr/deduplicator.go b/lib/streamaggr/deduplicator.go
index 273fd6eef..4177908e9 100644
--- a/lib/streamaggr/deduplicator.go
+++ b/lib/streamaggr/deduplicator.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"slices"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
@@ -11,26 +12,31 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
 	"github.com/VictoriaMetrics/metrics"
+	"github.com/valyala/histogram"
 )
 
 // Deduplicator deduplicates samples per each time series.
 type Deduplicator struct {
 	da *dedupAggr
 
-	dropLabels []string
+	cs            atomic.Pointer[currentState]
+	enableWindows bool
+	dropLabels    []string
+	interval      time.Duration
+	minDeadline   atomic.Int64
 
 	wg     sync.WaitGroup
 	stopCh chan struct{}
 
 	ms *metrics.Set
-
-	dedupFlushDuration *metrics.Histogram
-	dedupFlushTimeouts *metrics.Counter
+	// time to wait after interval end before flush
+	flushAfter   *histogram.Fast
+	muFlushAfter sync.Mutex
 }
 
 // NewDeduplicator returns new deduplicator, which deduplicates samples per each time series.
 //
-// The de-duplicated samples are passed to pushFunc once per dedupInterval.
+// The de-duplicated samples are passed to pushFunc once per interval.
 //
 // An optional dropLabels list may contain label names, which must be dropped before de-duplicating samples.
 // Common case is to drop `replica`-like labels from samples received from HA datasources.
@@ -38,14 +44,25 @@ type Deduplicator struct {
 // alias is url label used in metrics exposed by the returned Deduplicator.
 //
 // MustStop must be called on the returned deduplicator in order to free up occupied resources.
-func NewDeduplicator(pushFunc PushFunc, dedupInterval time.Duration, dropLabels []string, alias string) *Deduplicator {
+func NewDeduplicator(pushFunc PushFunc, enableWindows bool, interval time.Duration, dropLabels []string, alias string) *Deduplicator {
 	d := &Deduplicator{
-		da:         newDedupAggr(),
-		dropLabels: dropLabels,
-
-		stopCh: make(chan struct{}),
-		ms:     metrics.NewSet(),
+		da:            newDedupAggr(),
+		dropLabels:    dropLabels,
+		interval:      interval,
+		enableWindows: enableWindows,
+		stopCh:        make(chan struct{}),
+		ms:            metrics.NewSet(),
+	}
+	startTime := time.Now()
+	cs := &currentState{
+		maxDeadline: startTime.Add(interval).UnixMilli(),
+	}
+	d.cs.Store(cs)
+	if enableWindows {
+		d.flushAfter = histogram.GetFast()
+		d.minDeadline.Store(startTime.UnixMilli())
 	}
+	d.cs.Store(cs)
 
 	ms := d.ms
 
@@ -58,15 +75,15 @@ func NewDeduplicator(pushFunc PushFunc, dedupInterval time.Duration, dropLabels
 		return float64(d.da.itemsCount())
 	})
 
-	d.dedupFlushDuration = ms.NewHistogram(fmt.Sprintf(`vm_streamaggr_dedup_flush_duration_seconds{%s}`, metricLabels))
-	d.dedupFlushTimeouts = ms.NewCounter(fmt.Sprintf(`vm_streamaggr_dedup_flush_timeouts_total{%s}`, metricLabels))
+	d.da.flushDuration = ms.NewHistogram(fmt.Sprintf(`vm_streamaggr_dedup_flush_duration_seconds{%s}`, metricLabels))
+	d.da.flushTimeouts = ms.NewCounter(fmt.Sprintf(`vm_streamaggr_dedup_flush_timeouts_total{%s}`, metricLabels))
 
 	metrics.RegisterSet(ms)
 
 	d.wg.Add(1)
 	go func() {
 		defer d.wg.Done()
-		d.runFlusher(pushFunc, dedupInterval)
+		d.runFlusher(pushFunc)
 	}()
 
 	return d
@@ -84,9 +101,12 @@ func (d *Deduplicator) MustStop() {
 // Push pushes tss to d.
 func (d *Deduplicator) Push(tss []prompbmarshal.TimeSeries) {
 	ctx := getDeduplicatorPushCtx()
-	pss := ctx.pss
 	labels := &ctx.labels
 	buf := ctx.buf
+	cs := d.cs.Load()
+	nowMsec := time.Now().UnixMilli()
+	minDeadline := d.minDeadline.Load()
+	var maxLagMsec int64
 
 	dropLabels := d.dropLabels
 	for _, ts := range tss {
@@ -104,17 +124,41 @@ func (d *Deduplicator) Push(tss []prompbmarshal.TimeSeries) {
 		buf = lc.Compress(buf, labels.Labels)
 		key := bytesutil.ToUnsafeString(buf[bufLen:])
 		for _, s := range ts.Samples {
-			pss = append(pss, pushSample{
-				key:       key,
-				value:     s.Value,
-				timestamp: s.Timestamp,
-			})
+			if d.enableWindows && minDeadline > s.Timestamp {
+				continue
+			} else if d.enableWindows && s.Timestamp <= cs.maxDeadline == cs.isGreen {
+				ctx.green = append(ctx.green, pushSample{
+					key:       key,
+					value:     s.Value,
+					timestamp: s.Timestamp,
+				})
+			} else {
+				ctx.blue = append(ctx.blue, pushSample{
+					key:       key,
+					value:     s.Value,
+					timestamp: s.Timestamp,
+				})
+			}
+			lagMsec := nowMsec - s.Timestamp
+			if lagMsec > maxLagMsec {
+				maxLagMsec = lagMsec
+			}
 		}
 	}
 
-	d.da.pushSamples(pss)
+	if d.enableWindows && maxLagMsec > 0 {
+		d.muFlushAfter.Lock()
+		d.flushAfter.Update(float64(maxLagMsec))
+		d.muFlushAfter.Unlock()
+	}
+
+	if len(ctx.blue) > 0 {
+		d.da.pushSamples(ctx.blue, 0, false)
+	}
+	if len(ctx.green) > 0 {
+		d.da.pushSamples(ctx.green, 0, true)
+	}
 
-	ctx.pss = pss
 	ctx.buf = buf
 	putDeduplicatorPushCtx(ctx)
 }
@@ -128,75 +172,92 @@ func dropSeriesLabels(dst, src []prompbmarshal.Label, labelNames []string) []pro
 	return dst
 }
 
-func (d *Deduplicator) runFlusher(pushFunc PushFunc, dedupInterval time.Duration) {
-	t := time.NewTicker(dedupInterval)
+func (d *Deduplicator) runFlusher(pushFunc PushFunc) {
+	t := time.NewTicker(d.interval)
+	var fa *histogram.Fast
 	defer t.Stop()
 	for {
 		select {
 		case <-d.stopCh:
 			return
 		case <-t.C:
-			d.flush(pushFunc, dedupInterval)
+			if d.enableWindows {
+				// Calculate delay and wait
+				d.muFlushAfter.Lock()
+				fa, d.flushAfter = d.flushAfter, histogram.GetFast()
+				d.muFlushAfter.Unlock()
+				delay := time.Duration(fa.Quantile(flushQuantile)) * time.Millisecond
+				histogram.PutFast(fa)
+				time.Sleep(delay)
+			}
+			d.flush(pushFunc)
 		}
 	}
 }
 
-func (d *Deduplicator) flush(pushFunc PushFunc, dedupInterval time.Duration) {
+func (d *Deduplicator) flush(pushFunc PushFunc) {
+	cs := d.cs.Load().newState()
+	d.minDeadline.Store(cs.maxDeadline)
 	startTime := time.Now()
-
-	timestamp := startTime.UnixMilli()
-	d.da.flush(func(pss []pushSample) {
+	deadlineTime := time.UnixMilli(cs.maxDeadline)
+	d.da.flush(func(samples []pushSample, _ int64, _ bool) {
 		ctx := getDeduplicatorFlushCtx()
 
 		tss := ctx.tss
 		labels := ctx.labels
-		samples := ctx.samples
-		for _, ps := range pss {
+		dstSamples := ctx.samples
+		for _, ps := range samples {
 			labelsLen := len(labels)
 			labels = decompressLabels(labels, ps.key)
 
-			samplesLen := len(samples)
-			samples = append(samples, prompbmarshal.Sample{
+			dstSamplesLen := len(dstSamples)
+			dstSamples = append(dstSamples, prompbmarshal.Sample{
 				Value:     ps.value,
-				Timestamp: timestamp,
+				Timestamp: ps.timestamp,
 			})
 
 			tss = append(tss, prompbmarshal.TimeSeries{
 				Labels:  labels[labelsLen:],
-				Samples: samples[samplesLen:],
+				Samples: dstSamples[dstSamplesLen:],
 			})
 		}
 		pushFunc(tss)
 
 		ctx.tss = tss
 		ctx.labels = labels
-		ctx.samples = samples
+		ctx.samples = dstSamples
 		putDeduplicatorFlushCtx(ctx)
-	})
+	}, cs.maxDeadline, cs.isGreen)
 
 	duration := time.Since(startTime)
-	d.dedupFlushDuration.Update(duration.Seconds())
-	if duration > dedupInterval {
-		d.dedupFlushTimeouts.Inc()
+	d.da.flushDuration.Update(duration.Seconds())
+	if duration > d.interval {
+		d.da.flushTimeouts.Inc()
 		logger.Warnf("deduplication couldn't be finished in the configured dedupInterval=%s; it took %.03fs; "+
-			"possible solutions: increase dedupInterval; reduce samples' ingestion rate", dedupInterval, duration.Seconds())
+			"possible solutions: increase dedupInterval; reduce samples' ingestion rate", d.interval, duration.Seconds())
 	}
-
+	for time.Now().After(deadlineTime) {
+		deadlineTime = deadlineTime.Add(d.interval)
+	}
+	cs.maxDeadline = deadlineTime.UnixMilli()
+	if d.enableWindows {
+		cs.isGreen = !cs.isGreen
+	}
+	d.cs.Store(cs)
 }
 
 type deduplicatorPushCtx struct {
-	pss    []pushSample
+	blue   []pushSample
+	green  []pushSample
 	labels promutils.Labels
 	buf    []byte
 }
 
 func (ctx *deduplicatorPushCtx) reset() {
-	clear(ctx.pss)
-	ctx.pss = ctx.pss[:0]
-
-	ctx.labels.Reset()
-
+	ctx.blue = ctx.blue[:0]
+	ctx.green = ctx.green[:0]
 	ctx.buf = ctx.buf[:0]
+	ctx.labels.Reset()
 }
 
 func getDeduplicatorPushCtx() *deduplicatorPushCtx {
diff --git a/lib/streamaggr/histogram_bucket.go b/lib/streamaggr/histogram_bucket.go
index fdf7b3fad..1e0ea6e90 100644
--- a/lib/streamaggr/histogram_bucket.go
+++ b/lib/streamaggr/histogram_bucket.go
@@ -1,115 +1,57 @@
 package streamaggr
 
 import (
-	"math"
-	"sync"
-	"time"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/metrics"
 )
 
-// histogramBucketAggrState calculates output=histogram_bucket, e.g. VictoriaMetrics histogram over input samples.
-type histogramBucketAggrState struct {
-	m sync.Map
-
-	stalenessSecs uint64
+// histogramBucketAggrValue calculates output=histogram_bucket, e.g. VictoriaMetrics histogram over input samples.
+type histogramBucketAggrValue struct {
+	h      metrics.Histogram
+	shared *metrics.Histogram
 }
 
-type histogramBucketStateValue struct {
-	mu             sync.Mutex
-	h              metrics.Histogram
-	deleteDeadline uint64
-	deleted        bool
+func (av *histogramBucketAggrValue) pushSample(_ aggrConfig, sample *pushSample, _ string, _ int64) {
+	av.h.Update(sample.value)
 }
 
-func newHistogramBucketAggrState(stalenessInterval time.Duration) *histogramBucketAggrState {
-	stalenessSecs := roundDurationToSecs(stalenessInterval)
-	return &histogramBucketAggrState{
-		stalenessSecs: stalenessSecs,
+func (av *histogramBucketAggrValue) flush(c aggrConfig, ctx *flushCtx, key string) {
+	ac := c.(*histogramBucketAggrConfig)
+	shared := av.shared
+	if ac.useSharedState {
+		shared.Merge(&av.h)
+		av.h.Reset()
+	} else {
+		shared = &av.h
 	}
+	shared.VisitNonZeroBuckets(func(vmrange string, count uint64) {
+		ctx.appendSeriesWithExtraLabel(key, "histogram_bucket", float64(count), "vmrange", vmrange)
+	})
 }
 
-func (as *histogramBucketAggrState) pushSamples(samples []pushSample) {
-	currentTime := fasttime.UnixTimestamp()
-	deleteDeadline := currentTime + as.stalenessSecs
-	for i := range samples {
-		s := &samples[i]
-		outputKey := getOutputKey(s.key)
+func (av *histogramBucketAggrValue) state() any {
+	return av.shared
+}
 
-	again:
-		v, ok := as.m.Load(outputKey)
-		if !ok {
-			// The entry is missing in the map. Try creating it.
-			v = &histogramBucketStateValue{}
-			outputKey = bytesutil.InternString(outputKey)
-			vNew, loaded := as.m.LoadOrStore(outputKey, v)
-			if loaded {
-				// Use the entry created by a concurrent goroutine.
-				v = vNew
-			}
-		}
-		sv := v.(*histogramBucketStateValue)
-		sv.mu.Lock()
-		deleted := sv.deleted
-		if !deleted {
-			sv.h.Update(s.value)
-			sv.deleteDeadline = deleteDeadline
-		}
-		sv.mu.Unlock()
-		if deleted {
-			// The entry has been deleted by the concurrent call to flushState
-			// Try obtaining and updating the entry again.
-			goto again
-		}
+func newHistogramBucketAggrConfig(useSharedState bool) aggrConfig {
+	return &histogramBucketAggrConfig{
+		useSharedState: useSharedState,
 	}
 }
 
-func (as *histogramBucketAggrState) removeOldEntries(currentTime uint64) {
-	m := &as.m
-	m.Range(func(k, v any) bool {
-		sv := v.(*histogramBucketStateValue)
-
-		sv.mu.Lock()
-		deleted := currentTime > sv.deleteDeadline
-		if deleted {
-			// Mark the current entry as deleted
-			sv.deleted = deleted
-		}
-		sv.mu.Unlock()
-
-		if deleted {
-			m.Delete(k)
-		}
-		return true
-	})
+type histogramBucketAggrConfig struct {
+	useSharedState bool
 }
 
-func (as *histogramBucketAggrState) flushState(ctx *flushCtx) {
-	currentTime := fasttime.UnixTimestamp()
-
-	as.removeOldEntries(currentTime)
-
-	m := &as.m
-	m.Range(func(k, v any) bool {
-		sv := v.(*histogramBucketStateValue)
-		sv.mu.Lock()
-		if !sv.deleted {
-			key := k.(string)
-			sv.h.VisitNonZeroBuckets(func(vmrange string, count uint64) {
-				ctx.appendSeriesWithExtraLabel(key, "histogram_bucket", float64(count), "vmrange", vmrange)
-			})
+func (ac *histogramBucketAggrConfig) getValue(s any) aggrValue {
+	var shared *metrics.Histogram
+	if ac.useSharedState {
+		if s == nil {
+			shared = &metrics.Histogram{}
+		} else {
+			shared = s.(*metrics.Histogram)
 		}
-		sv.mu.Unlock()
-		return true
-	})
-}
-
-func roundDurationToSecs(d time.Duration) uint64 {
-	if d < 0 {
-		return 0
 	}
-	secs := d.Seconds()
-	return uint64(math.Ceil(secs))
+	return &histogramBucketAggrValue{
+		shared: shared,
+	}
 }
diff --git a/lib/streamaggr/last.go b/lib/streamaggr/last.go
index 0ae4b9b8c..7425696e7 100644
--- a/lib/streamaggr/last.go
+++ b/lib/streamaggr/last.go
@@ -1,82 +1,34 @@
 package streamaggr
 
-import (
-	"sync"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-)
-
-// lastAggrState calculates output=last, e.g. the last value over input samples.
-type lastAggrState struct {
-	m sync.Map
-}
-
-type lastStateValue struct {
-	mu        sync.Mutex
+type lastAggrValue struct {
 	last      float64
 	timestamp int64
-	deleted   bool
 }
 
-func newLastAggrState() *lastAggrState {
-	return &lastAggrState{}
+func (av *lastAggrValue) pushSample(_ aggrConfig, sample *pushSample, _ string, _ int64) {
+	if sample.timestamp >= av.timestamp {
+		av.last = sample.value
+		av.timestamp = sample.timestamp
+	}
 }
 
-func (as *lastAggrState) pushSamples(samples []pushSample) {
-	for i := range samples {
-		s := &samples[i]
-		outputKey := getOutputKey(s.key)
-
-	again:
-		v, ok := as.m.Load(outputKey)
-		if !ok {
-			// The entry is missing in the map. Try creating it.
-			v = &lastStateValue{
-				last:      s.value,
-				timestamp: s.timestamp,
-			}
-			outputKey = bytesutil.InternString(outputKey)
-			vNew, loaded := as.m.LoadOrStore(outputKey, v)
-			if !loaded {
-				// The new entry has been successfully created.
-				continue
-			}
-			// Use the entry created by a concurrent goroutine.
-			v = vNew
-		}
-		sv := v.(*lastStateValue)
-		sv.mu.Lock()
-		deleted := sv.deleted
-		if !deleted {
-			if s.timestamp >= sv.timestamp {
-				sv.last = s.value
-				sv.timestamp = s.timestamp
-			}
-		}
-		sv.mu.Unlock()
-		if deleted {
-			// The entry has been deleted by the concurrent call to flushState
-			// Try obtaining and updating the entry again.
-			goto again
-		}
+func (av *lastAggrValue) flush(_ aggrConfig, ctx *flushCtx, key string) {
+	if av.timestamp > 0 {
+		ctx.appendSeries(key, "last", av.last)
+		av.timestamp = 0
 	}
 }
 
-func (as *lastAggrState) flushState(ctx *flushCtx) {
-	m := &as.m
-	m.Range(func(k, v any) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
+func (*lastAggrValue) state() any {
+	return nil
+}
+
+func newLastAggrConfig() aggrConfig {
+	return &lastAggrConfig{}
+}
 
-		sv := v.(*lastStateValue)
-		sv.mu.Lock()
-		last := sv.last
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-		sv.deleted = true
-		sv.mu.Unlock()
+type lastAggrConfig struct{}
 
-		key := k.(string)
-		ctx.appendSeries(key, "last", last)
-		return true
-	})
+func (*lastAggrConfig) getValue(_ any) aggrValue {
+	return &lastAggrValue{}
 }
diff --git a/lib/streamaggr/max.go b/lib/streamaggr/max.go
index f4f33f1b7..b9a5fcc2c 100644
--- a/lib/streamaggr/max.go
+++ b/lib/streamaggr/max.go
@@ -1,79 +1,37 @@
 package streamaggr
 
-import (
-	"sync"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-)
-
-// maxAggrState calculates output=max, e.g. the maximum value over input samples.
-type maxAggrState struct {
-	m sync.Map
-}
-
-type maxStateValue struct {
-	mu      sync.Mutex
+type maxAggrValue struct {
 	max     float64
-	deleted bool
+	defined bool
 }
 
-func newMaxAggrState() *maxAggrState {
-	return &maxAggrState{}
+func (av *maxAggrValue) pushSample(_ aggrConfig, sample *pushSample, _ string, _ int64) {
+	if sample.value > av.max || !av.defined {
+		av.max = sample.value
+	}
+	if !av.defined {
+		av.defined = true
+	}
 }
 
-func (as *maxAggrState) pushSamples(samples []pushSample) {
-	for i := range samples {
-		s := &samples[i]
-		outputKey := getOutputKey(s.key)
-
-	again:
-		v, ok := as.m.Load(outputKey)
-		if !ok {
-			// The entry is missing in the map. Try creating it.
-			v = &maxStateValue{
-				max: s.value,
-			}
-			outputKey = bytesutil.InternString(outputKey)
-			vNew, loaded := as.m.LoadOrStore(outputKey, v)
-			if !loaded {
-				// The new entry has been successfully created.
-				continue
-			}
-			// Use the entry created by a concurrent goroutine.
-			v = vNew
-		}
-		sv := v.(*maxStateValue)
-		sv.mu.Lock()
-		deleted := sv.deleted
-		if !deleted {
-			if s.value > sv.max {
-				sv.max = s.value
-			}
-		}
-		sv.mu.Unlock()
-		if deleted {
-			// The entry has been deleted by the concurrent call to flushState
-			// Try obtaining and updating the entry again.
-			goto again
-		}
+func (av *maxAggrValue) flush(_ aggrConfig, ctx *flushCtx, key string) {
+	if av.defined {
+		ctx.appendSeries(key, "max", av.max)
+		av.max = 0
+		av.defined = false
 	}
 }
 
-func (as *maxAggrState) flushState(ctx *flushCtx) {
-	m := &as.m
-	m.Range(func(k, v any) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
+func (*maxAggrValue) state() any {
+	return nil
+}
+
+func newMaxAggrConfig() aggrConfig {
+	return &maxAggrConfig{}
+}
 
-		sv := v.(*maxStateValue)
-		sv.mu.Lock()
-		maxV := sv.max
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-		sv.deleted = true
-		sv.mu.Unlock()
+type maxAggrConfig struct{}
 
-		key := k.(string)
-		ctx.appendSeries(key, "max", maxV)
-		return true
-	})
+func (*maxAggrConfig) getValue(_ any) aggrValue {
+	return &maxAggrValue{}
 }
diff --git a/lib/streamaggr/min.go b/lib/streamaggr/min.go
index b82c9dc4c..8627e1e2c 100644
--- a/lib/streamaggr/min.go
+++ b/lib/streamaggr/min.go
@@ -1,78 +1,37 @@
 package streamaggr
 
-import (
-	"sync"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-)
-
-// minAggrState calculates output=min, e.g. the minimum value over input samples.
-type minAggrState struct {
-	m sync.Map
+type minAggrValue struct {
+	min     float64
+	defined bool
 }
 
-type minStateValue struct {
-	mu      sync.Mutex
-	min     float64
-	deleted bool
+func (av *minAggrValue) pushSample(_ aggrConfig, sample *pushSample, _ string, _ int64) {
+	if sample.value < av.min || !av.defined {
+		av.min = sample.value
+	}
+	if !av.defined {
+		av.defined = true
+	}
 }
 
-func newMinAggrState() *minAggrState {
-	return &minAggrState{}
+func (av *minAggrValue) flush(_ aggrConfig, ctx *flushCtx, key string) {
+	if av.defined {
+		ctx.appendSeries(key, "min", av.min)
+		av.min = 0
+		av.defined = false
+	}
 }
 
-func (as *minAggrState) pushSamples(samples []pushSample) {
-	for i := range samples {
-		s := &samples[i]
-		outputKey := getOutputKey(s.key)
+func (*minAggrValue) state() any {
+	return nil
+}
 
-	again:
-		v, ok := as.m.Load(outputKey)
-		if !ok {
-			// The entry is missing in the map. Try creating it.
-			v = &minStateValue{
-				min: s.value,
-			}
-			outputKey = bytesutil.InternString(outputKey)
-			vNew, loaded := as.m.LoadOrStore(outputKey, v)
-			if !loaded {
-				// The new entry has been successfully created.
-				continue
-			}
-			// Use the entry created by a concurrent goroutine.
-			v = vNew
-		}
-		sv := v.(*minStateValue)
-		sv.mu.Lock()
-		deleted := sv.deleted
-		if !deleted {
-			if s.value < sv.min {
-				sv.min = s.value
-			}
-		}
-		sv.mu.Unlock()
-		if deleted {
-			// The entry has been deleted by the concurrent call to flushState
-			// Try obtaining and updating the entry again.
-			goto again
-		}
-	}
+func newMinAggrConfig() aggrConfig {
+	return &minAggrConfig{}
 }
 
-func (as *minAggrState) flushState(ctx *flushCtx) {
-	m := &as.m
-	m.Range(func(k, v any) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
+type minAggrConfig struct{}
 
-		sv := v.(*minStateValue)
-		sv.mu.Lock()
-		minV := sv.min
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-		sv.deleted = true
-		sv.mu.Unlock()
-		key := k.(string)
-		ctx.appendSeries(key, "min", minV)
-		return true
-	})
+func (*minAggrConfig) getValue(_ any) aggrValue {
+	return &minAggrValue{}
 }
diff --git a/lib/streamaggr/output.go b/lib/streamaggr/output.go
new file mode 100644
index 000000000..440c2578c
--- /dev/null
+++ b/lib/streamaggr/output.go
@@ -0,0 +1,140 @@
+package streamaggr
+
+import (
+	"sync"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/metrics"
+)
+
+type aggrOutputs struct {
+	m              sync.Map
+	useSharedState bool
+	useInputKey    bool
+	configs        []aggrConfig
+	outputSamples  *metrics.Counter
+}
+
+func (ao *aggrOutputs) getInputOutputKey(key string) (string, string) {
+	src := bytesutil.ToUnsafeBytes(key)
+	outputKeyLen, nSize := encoding.UnmarshalVarUint64(src)
+	if nSize <= 0 {
+		logger.Panicf("BUG: cannot unmarshal outputKeyLen from uvarint")
+	}
+	src = src[nSize:]
+	outputKey := src[:outputKeyLen]
+	if !ao.useInputKey {
+		return key, bytesutil.ToUnsafeString(outputKey)
+	}
+	inputKey := src[outputKeyLen:]
+	return bytesutil.ToUnsafeString(inputKey), bytesutil.ToUnsafeString(outputKey)
+}
+
+func (ao *aggrOutputs) pushSamples(samples []pushSample, deleteDeadline int64, isGreen bool) {
+	var inputKey, outputKey string
+	var sample *pushSample
+	var outputs []aggrValue
+	var nv *aggrValues
+	for i := range samples {
+		sample = &samples[i]
+		inputKey, outputKey = ao.getInputOutputKey(sample.key)
+
+	again:
+		v, ok := ao.m.Load(outputKey)
+		if !ok {
+			// The entry is missing in the map. Try creating it.
+			nv = &aggrValues{
+				blue: make([]aggrValue, len(ao.configs)),
+			}
+			if ao.useSharedState {
+				nv.green = make([]aggrValue, len(ao.configs))
+			}
+			for idx, ac := range ao.configs {
+				nv.blue[idx] = ac.getValue(nil)
+				if ao.useSharedState {
+					nv.green[idx] = ac.getValue(nv.blue[idx].state())
+				}
+			}
+			v = nv
+			outputKey = bytesutil.InternString(outputKey)
+			vNew, loaded := ao.m.LoadOrStore(outputKey, v)
+			if loaded {
+				// Use the entry created by a concurrent goroutine.
+				v = vNew
+			}
+		}
+		av := v.(*aggrValues)
+		av.mu.Lock()
+		deleted := av.deleteDeadline < 0
+		if !deleted {
+			if isGreen {
+				outputs = av.green
+			} else {
+				outputs = av.blue
+			}
+			for idx, o := range outputs {
+				o.pushSample(ao.configs[idx], sample, inputKey, deleteDeadline)
+			}
+			av.deleteDeadline = deleteDeadline
+		}
+		av.mu.Unlock()
+		if deleted {
+			// The entry has been deleted by the concurrent call to flush
+			// Try obtaining and updating the entry again.
+			goto again
+		}
+	}
+}
+
+func (ao *aggrOutputs) flushState(ctx *flushCtx) {
+	m := &ao.m
+	var outputs []aggrValue
+	m.Range(func(k, v any) bool {
+		// Atomically delete the entry from the map, so new entry is created for the next flush.
+		av := v.(*aggrValues)
+		av.mu.Lock()
+
+		// check for stale entries
+		deleted := ctx.flushTimestamp > av.deleteDeadline
+		if deleted {
+			// Mark the current entry as deleted
+			av.deleteDeadline = -1
+			av.mu.Unlock()
+			m.Delete(k)
+			return true
+		}
+		outputKey := k.(string)
+		if ctx.isGreen {
+			outputs = av.green
+		} else {
+			outputs = av.blue
+		}
+		for i, o := range outputs {
+			o.flush(ao.configs[i], ctx, outputKey)
+		}
+		av.mu.Unlock()
+		if ctx.isLast {
+			m.Delete(k)
+		}
+		return true
+	})
+}
+
+type aggrValues struct {
+	mu             sync.Mutex
+	blue           []aggrValue
+	green          []aggrValue
+	deleteDeadline int64
+}
+
+type aggrConfig interface {
+	getValue(any) aggrValue
+}
+
+type aggrValue interface {
+	pushSample(aggrConfig, *pushSample, string, int64)
+	flush(aggrConfig, *flushCtx, string)
+	state() any
+}
diff --git a/lib/streamaggr/quantiles.go b/lib/streamaggr/quantiles.go
index 00fefe877..ffdc35b3f 100644
--- a/lib/streamaggr/quantiles.go
+++ b/lib/streamaggr/quantiles.go
@@ -1,91 +1,54 @@
 package streamaggr
 
 import (
-	"strconv"
-	"sync"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/valyala/histogram"
+	"strconv"
 )
 
-// quantilesAggrState calculates output=quantiles, e.g. the given quantiles over the input samples.
-type quantilesAggrState struct {
-	m sync.Map
-
-	phis []float64
+// quantilesAggrValue calculates output=quantiles, e.g. the given quantiles over the input samples.
+type quantilesAggrValue struct {
+	h *histogram.Fast
 }
 
-type quantilesStateValue struct {
-	mu      sync.Mutex
-	h       *histogram.Fast
-	deleted bool
+func (av *quantilesAggrValue) pushSample(_ aggrConfig, sample *pushSample, _ string, _ int64) {
+	if av.h == nil {
+		av.h = histogram.GetFast()
+	}
+	av.h.Update(sample.value)
 }
 
-func newQuantilesAggrState(phis []float64) *quantilesAggrState {
-	return &quantilesAggrState{
-		phis: phis,
+func (av *quantilesAggrValue) flush(c aggrConfig, ctx *flushCtx, key string) {
+	ac := c.(*quantilesAggrConfig)
+	if av.h != nil {
+		ac.quantiles = av.h.Quantiles(ac.quantiles[:0], ac.phis)
+	}
+	histogram.PutFast(av.h)
+	if len(ac.quantiles) > 0 {
+		for i, quantile := range ac.quantiles {
+			ac.b = strconv.AppendFloat(ac.b[:0], ac.phis[i], 'g', -1, 64)
+			phiStr := bytesutil.InternBytes(ac.b)
+			ctx.appendSeriesWithExtraLabel(key, "quantiles", quantile, "quantile", phiStr)
+		}
 	}
 }
 
-func (as *quantilesAggrState) pushSamples(samples []pushSample) {
-	for i := range samples {
-		s := &samples[i]
-		outputKey := getOutputKey(s.key)
+func (*quantilesAggrValue) state() any {
+	return nil
+}
 
-	again:
-		v, ok := as.m.Load(outputKey)
-		if !ok {
-			// The entry is missing in the map. Try creating it.
-			h := histogram.GetFast()
-			v = &quantilesStateValue{
-				h: h,
-			}
-			outputKey = bytesutil.InternString(outputKey)
-			vNew, loaded := as.m.LoadOrStore(outputKey, v)
-			if loaded {
-				// Use the entry created by a concurrent goroutine.
-				histogram.PutFast(h)
-				v = vNew
-			}
-		}
-		sv := v.(*quantilesStateValue)
-		sv.mu.Lock()
-		deleted := sv.deleted
-		if !deleted {
-			sv.h.Update(s.value)
-		}
-		sv.mu.Unlock()
-		if deleted {
-			// The entry has been deleted by the concurrent call to flushState
-			// Try obtaining and updating the entry again.
-			goto again
-		}
+func newQuantilesAggrConfig(phis []float64) aggrConfig {
+	return &quantilesAggrConfig{
+		phis: phis,
 	}
 }
 
-func (as *quantilesAggrState) flushState(ctx *flushCtx) {
-	m := &as.m
-	phis := as.phis
-	var quantiles []float64
-	var b []byte
-	m.Range(func(k, v any) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
-
-		sv := v.(*quantilesStateValue)
-		sv.mu.Lock()
-		quantiles = sv.h.Quantiles(quantiles[:0], phis)
-		histogram.PutFast(sv.h)
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-		sv.deleted = true
-		sv.mu.Unlock()
+type quantilesAggrConfig struct {
+	phis      []float64
+	quantiles []float64
+	b         []byte
+}
 
-		key := k.(string)
-		for i, quantile := range quantiles {
-			b = strconv.AppendFloat(b[:0], phis[i], 'g', -1, 64)
-			phiStr := bytesutil.InternBytes(b)
-			ctx.appendSeriesWithExtraLabel(key, "quantiles", quantile, "quantile", phiStr)
-		}
-		return true
-	})
+func (*quantilesAggrConfig) getValue(_ any) aggrValue {
+	return &quantilesAggrValue{}
 }
diff --git a/lib/streamaggr/rate.go b/lib/streamaggr/rate.go
index 1c3f2fb5c..6b64c85fc 100644
--- a/lib/streamaggr/rate.go
+++ b/lib/streamaggr/rate.go
@@ -1,183 +1,189 @@
 package streamaggr
 
 import (
-	"sync"
-	"time"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
+	"sync"
 )
 
-// rateAggrState calculates output=rate_avg and rate_sum, e.g. the average per-second increase rate for counter metrics.
-type rateAggrState struct {
-	m sync.Map
+var rateAggrSharedValuePool sync.Pool
 
-	// isAvg is set to true if rate_avg() must be calculated instead of rate_sum().
-	isAvg bool
+func putRateAggrSharedValue(v *rateAggrSharedValue) {
+	v.reset()
+	rateAggrSharedValuePool.Put(v)
+}
 
-	// Time series state is dropped if no new samples are received during stalenessSecs.
-	stalenessSecs uint64
+func getRateAggrSharedValue(isGreen bool) *rateAggrSharedValue {
+	v := rateAggrSharedValuePool.Get()
+	if v == nil {
+		v = &rateAggrSharedValue{}
+	}
+	av := v.(*rateAggrSharedValue)
+	if isGreen {
+		av.green = getRateAggrStateValue()
+	} else {
+		av.blue = getRateAggrStateValue()
+	}
+	return av
 }
 
-type rateStateValue struct {
-	mu             sync.Mutex
-	lastValues     map[string]rateLastValueState
-	deleteDeadline uint64
-	deleted        bool
+var rateAggrStateValuePool sync.Pool
+
+func putRateAggrStateValue(v *rateAggrStateValue) {
+	v.timestamp = 0
+	v.increase = 0
+	rateAggrStateValuePool.Put(v)
 }
 
-type rateLastValueState struct {
-	value          float64
-	timestamp      int64
-	deleteDeadline uint64
+func getRateAggrStateValue() *rateAggrStateValue {
+	v := rateAggrStateValuePool.Get()
+	if v == nil {
+		return &rateAggrStateValue{}
+	}
+	return v.(*rateAggrStateValue)
+}
 
-	// increase stores cumulative increase for the current time series on the current aggregation interval
-	increase float64
+// rateAggrSharedValue calculates output=rate_avg and rate_sum, e.g. the average per-second increase rate for counter metrics.
+type rateAggrSharedValue struct {
+	value          float64
+	deleteDeadline int64
 
 	// prevTimestamp is the timestamp of the last registered sample in the previous aggregation interval
 	prevTimestamp int64
+	blue          *rateAggrStateValue
+	green         *rateAggrStateValue
 }
 
-func newRateAggrState(stalenessInterval time.Duration, isAvg bool) *rateAggrState {
-	stalenessSecs := roundDurationToSecs(stalenessInterval)
-	return &rateAggrState{
-		isAvg:         isAvg,
-		stalenessSecs: stalenessSecs,
-	}
-}
-
-func (as *rateAggrState) pushSamples(samples []pushSample) {
-	currentTime := fasttime.UnixTimestamp()
-	deleteDeadline := currentTime + as.stalenessSecs
-	for i := range samples {
-		s := &samples[i]
-		inputKey, outputKey := getInputOutputKey(s.key)
-
-	again:
-		v, ok := as.m.Load(outputKey)
-		if !ok {
-			// The entry is missing in the map. Try creating it.
-			v = &rateStateValue{
-				lastValues: make(map[string]rateLastValueState),
-			}
-			outputKey = bytesutil.InternString(outputKey)
-			vNew, loaded := as.m.LoadOrStore(outputKey, v)
-			if loaded {
-				// Use the entry created by a concurrent goroutine.
-				v = vNew
-			}
-		}
-		sv := v.(*rateStateValue)
-		sv.mu.Lock()
-		deleted := sv.deleted
-		if !deleted {
-			lv, ok := sv.lastValues[inputKey]
-			if ok {
-				if s.timestamp < lv.timestamp {
-					// Skip out of order sample
-					sv.mu.Unlock()
-					continue
-				}
-
-				if s.value >= lv.value {
-					lv.increase += s.value - lv.value
-				} else {
-					// counter reset
-					lv.increase += s.value
-				}
-			} else {
-				lv.prevTimestamp = s.timestamp
-			}
-			lv.value = s.value
-			lv.timestamp = s.timestamp
-			lv.deleteDeadline = deleteDeadline
-
-			inputKey = bytesutil.InternString(inputKey)
-			sv.lastValues[inputKey] = lv
-			sv.deleteDeadline = deleteDeadline
-		}
-		sv.mu.Unlock()
-		if deleted {
-			// The entry has been deleted by the concurrent call to flushState
-			// Try obtaining and updating the entry again.
-			goto again
+func (v *rateAggrSharedValue) getState(isGreen bool) *rateAggrStateValue {
+	if isGreen {
+		if v.green == nil {
+			v.green = getRateAggrStateValue()
 		}
+		return v.green
 	}
+	if v.blue == nil {
+		v.blue = getRateAggrStateValue()
+	}
+	return v.blue
 }
 
-func (as *rateAggrState) flushState(ctx *flushCtx) {
-	currentTime := fasttime.UnixTimestamp()
-
-	suffix := as.getSuffix()
+func (v *rateAggrSharedValue) reset() {
+	v.value = 0
+	v.deleteDeadline = 0
+	v.prevTimestamp = 0
+	putRateAggrStateValue(v.blue)
+	putRateAggrStateValue(v.green)
+	v.blue = nil
+	v.green = nil
+}
 
-	as.removeOldEntries(currentTime)
+type rateAggrStateValue struct {
+	// increase stores cumulative increase for the current time series on the current aggregation interval
+	increase  float64
+	timestamp int64
+}
 
-	m := &as.m
-	m.Range(func(k, v any) bool {
-		sv := v.(*rateStateValue)
+type rateAggrValue struct {
+	shared  map[string]*rateAggrSharedValue
+	isGreen bool
+}
 
-		sv.mu.Lock()
-		lvs := sv.lastValues
-		sumRate := 0.0
-		countSeries := 0
-		for k1, lv := range lvs {
-			d := float64(lv.timestamp-lv.prevTimestamp) / 1000
-			if d > 0 {
-				sumRate += lv.increase / d
-				countSeries++
-			}
-			lv.prevTimestamp = lv.timestamp
-			lv.increase = 0
-			lvs[k1] = lv
+func (av *rateAggrValue) pushSample(_ aggrConfig, sample *pushSample, key string, deleteDeadline int64) {
+	var state *rateAggrStateValue
+	sv, ok := av.shared[key]
+	if ok {
+		state = sv.getState(av.isGreen)
+		if sample.timestamp < state.timestamp {
+			// Skip out of order sample
+			return
 		}
-		deleted := sv.deleted
-		sv.mu.Unlock()
-
-		if countSeries == 0 || deleted {
-			// Nothing to update
-			return true
+		if sample.value >= sv.value {
+			state.increase += sample.value - sv.value
+		} else {
+			// counter reset
+			state.increase += sample.value
 		}
+	} else {
+		sv = getRateAggrSharedValue(av.isGreen)
+		sv.prevTimestamp = sample.timestamp
+		key = bytesutil.InternString(key)
+		av.shared[key] = sv
+		state = sv.getState(av.isGreen)
+	}
+	sv.value = sample.value
+	sv.deleteDeadline = deleteDeadline
+	state.timestamp = sample.timestamp
+}
 
-		result := sumRate
-		if as.isAvg {
-			result /= float64(countSeries)
+func (av *rateAggrValue) flush(c aggrConfig, ctx *flushCtx, key string) {
+	ac := c.(*rateAggrConfig)
+	var state *rateAggrStateValue
+	suffix := ac.getSuffix()
+	rate := 0.0
+	countSeries := 0
+	for sk, sv := range av.shared {
+		if ctx.flushTimestamp > sv.deleteDeadline {
+			delete(av.shared, sk)
+			putRateAggrSharedValue(sv)
+			continue
+		}
+		if sv.prevTimestamp == 0 {
+			continue
+		}
+		if av.isGreen {
+			state = sv.green
+		} else {
+			state = sv.blue
+		}
+		d := float64(state.timestamp-sv.prevTimestamp) / 1000
+		if d > 0 {
+			rate += state.increase / d
+			countSeries++
 		}
+		sv.prevTimestamp = state.timestamp
+		state.timestamp = 0
+		state.increase = 0
+		av.shared[sk] = sv
+	}
+
+	if countSeries == 0 {
+		return
+	}
+	if ac.isAvg {
+		rate /= float64(countSeries)
+	}
+	ctx.appendSeries(key, suffix, rate)
+}
 
-		key := k.(string)
-		ctx.appendSeries(key, suffix, result)
-		return true
-	})
+func (av *rateAggrValue) state() any {
+	return av.shared
 }
 
-func (as *rateAggrState) getSuffix() string {
-	if as.isAvg {
-		return "rate_avg"
+func newRateAggrConfig(isAvg bool) aggrConfig {
+	return &rateAggrConfig{
+		isAvg: isAvg,
 	}
-	return "rate_sum"
 }
 
-func (as *rateAggrState) removeOldEntries(currentTime uint64) {
-	m := &as.m
-	m.Range(func(k, v any) bool {
-		sv := v.(*rateStateValue)
+type rateAggrConfig struct {
+	isAvg bool
+}
 
-		sv.mu.Lock()
-		if currentTime > sv.deleteDeadline {
-			// Mark the current entry as deleted
-			sv.deleted = true
-			sv.mu.Unlock()
-			m.Delete(k)
-			return true
-		}
+func (*rateAggrConfig) getValue(s any) aggrValue {
+	var shared map[string]*rateAggrSharedValue
+	if s == nil {
+		shared = make(map[string]*rateAggrSharedValue)
+	} else {
+		shared = s.(map[string]*rateAggrSharedValue)
+	}
+	return &rateAggrValue{
+		shared:  shared,
+		isGreen: s != nil,
+	}
+}
 
-		// Delete outdated entries in sv.lastValues
-		lvs := sv.lastValues
-		for k1, lv := range lvs {
-			if currentTime > lv.deleteDeadline {
-				delete(lvs, k1)
-			}
-		}
-		sv.mu.Unlock()
-		return true
-	})
+func (ac *rateAggrConfig) getSuffix() string {
+	if ac.isAvg {
+		return "rate_avg"
+	}
+	return "rate_sum"
 }
diff --git a/lib/streamaggr/std.go b/lib/streamaggr/std.go
new file mode 100644
index 000000000..98155d4de
--- /dev/null
+++ b/lib/streamaggr/std.go
@@ -0,0 +1,63 @@
+package streamaggr
+
+import (
+	"math"
+)
+
+// stdAggrValue calculates output=stdvar or stddev, e.g. the average value over input samples.
+type stdAggrValue struct {
+	count float64
+	avg   float64
+	q     float64
+}
+
+func (av *stdAggrValue) pushSample(_ aggrConfig, sample *pushSample, _ string, _ int64) {
+	av.count++
+	avg := av.avg + (sample.value-av.avg)/av.count
+	av.q += (sample.value - av.avg) * (sample.value - avg)
+	av.avg = avg
+}
+
+func (av *stdAggrValue) flush(c aggrConfig, ctx *flushCtx, key string) {
+	ac := c.(*stdAggrConfig)
+	if av.count > 0 {
+		suffix := ac.getSuffix()
+		output := av.q / av.count
+		if ac.isDeviation {
+			output = math.Sqrt(output)
+		}
+		ctx.appendSeries(key, suffix, output)
+		av.count = 0
+		av.avg = 0
+		av.q = 0
+	}
+}
+
+func (ac *stdAggrConfig) getSuffix() string {
+	if ac.isDeviation {
+		return "stddev"
+	}
+	return "stdvar"
+}
+
+func (*stdAggrValue) state() any {
+	return nil
+}
+
+func newStddevAggrConfig() aggrConfig {
+	return &stdAggrConfig{
+		isDeviation: true,
+	}
+}
+
+func newStdvarAggrConfig() aggrConfig {
+	return &stdAggrConfig{}
+}
+
+type stdAggrConfig struct {
+	isDeviation bool
+}
+
+func (*stdAggrConfig) getValue(_ any) aggrValue {
+	return &stdAggrValue{}
+}
diff --git a/lib/streamaggr/stddev.go b/lib/streamaggr/stddev.go
deleted file mode 100644
index 053e0f209..000000000
--- a/lib/streamaggr/stddev.go
+++ /dev/null
@@ -1,80 +0,0 @@
-package streamaggr
-
-import (
-	"math"
-	"sync"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-)
-
-// stddevAggrState calculates output=stddev, e.g. the average value over input samples.
-type stddevAggrState struct {
-	m sync.Map
-}
-
-type stddevStateValue struct {
-	mu      sync.Mutex
-	count   float64
-	avg     float64
-	q       float64
-	deleted bool
-}
-
-func newStddevAggrState() *stddevAggrState {
-	return &stddevAggrState{}
-}
-
-func (as *stddevAggrState) pushSamples(samples []pushSample) {
-	for i := range samples {
-		s := &samples[i]
-		outputKey := getOutputKey(s.key)
-
-	again:
-		v, ok := as.m.Load(outputKey)
-		if !ok {
-			// The entry is missing in the map. Try creating it.
-			v = &stddevStateValue{}
-			outputKey = bytesutil.InternString(outputKey)
-			vNew, loaded := as.m.LoadOrStore(outputKey, v)
-			if loaded {
-				// Use the entry created by a concurrent goroutine.
-				v = vNew
-			}
-		}
-		sv := v.(*stddevStateValue)
-		sv.mu.Lock()
-		deleted := sv.deleted
-		if !deleted {
-			// See `Rapid calculation methods` at https://en.wikipedia.org/wiki/Standard_deviation
-			sv.count++
-			avg := sv.avg + (s.value-sv.avg)/sv.count
-			sv.q += (s.value - sv.avg) * (s.value - avg)
-			sv.avg = avg
-		}
-		sv.mu.Unlock()
-		if deleted {
-			// The entry has been deleted by the concurrent call to flushState
-			// Try obtaining and updating the entry again.
-			goto again
-		}
-	}
-}
-
-func (as *stddevAggrState) flushState(ctx *flushCtx) {
-	m := &as.m
-	m.Range(func(k, v any) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
-
-		sv := v.(*stddevStateValue)
-		sv.mu.Lock()
-		stddev := math.Sqrt(sv.q / sv.count)
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-		sv.deleted = true
-		sv.mu.Unlock()
-
-		key := k.(string)
-		ctx.appendSeries(key, "stddev", stddev)
-		return true
-	})
-}
diff --git a/lib/streamaggr/stdvar.go b/lib/streamaggr/stdvar.go
deleted file mode 100644
index 8170f227a..000000000
--- a/lib/streamaggr/stdvar.go
+++ /dev/null
@@ -1,79 +0,0 @@
-package streamaggr
-
-import (
-	"sync"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-)
-
-// stdvarAggrState calculates output=stdvar, e.g. the average value over input samples.
-type stdvarAggrState struct {
-	m sync.Map
-}
-
-type stdvarStateValue struct {
-	mu      sync.Mutex
-	count   float64
-	avg     float64
-	q       float64
-	deleted bool
-}
-
-func newStdvarAggrState() *stdvarAggrState {
-	return &stdvarAggrState{}
-}
-
-func (as *stdvarAggrState) pushSamples(samples []pushSample) {
-	for i := range samples {
-		s := &samples[i]
-		outputKey := getOutputKey(s.key)
-
-	again:
-		v, ok := as.m.Load(outputKey)
-		if !ok {
-			// The entry is missing in the map. Try creating it.
-			v = &stdvarStateValue{}
-			outputKey = bytesutil.InternString(outputKey)
-			vNew, loaded := as.m.LoadOrStore(outputKey, v)
-			if loaded {
-				// Use the entry created by a concurrent goroutine.
-				v = vNew
-			}
-		}
-		sv := v.(*stdvarStateValue)
-		sv.mu.Lock()
-		deleted := sv.deleted
-		if !deleted {
-			// See `Rapid calculation methods` at https://en.wikipedia.org/wiki/Standard_deviation
-			sv.count++
-			avg := sv.avg + (s.value-sv.avg)/sv.count
-			sv.q += (s.value - sv.avg) * (s.value - avg)
-			sv.avg = avg
-		}
-		sv.mu.Unlock()
-		if deleted {
-			// The entry has been deleted by the concurrent call to flushState
-			// Try obtaining and updating the entry again.
-			goto again
-		}
-	}
-}
-
-func (as *stdvarAggrState) flushState(ctx *flushCtx) {
-	m := &as.m
-	m.Range(func(k, v any) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
-
-		sv := v.(*stdvarStateValue)
-		sv.mu.Lock()
-		stdvar := sv.q / sv.count
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-		sv.deleted = true
-		sv.mu.Unlock()
-
-		key := k.(string)
-		ctx.appendSeries(key, "stdvar", stdvar)
-		return true
-	})
-}
diff --git a/lib/streamaggr/streamaggr.go b/lib/streamaggr/streamaggr.go
index f16f9a10a..ad47461d0 100644
--- a/lib/streamaggr/streamaggr.go
+++ b/lib/streamaggr/streamaggr.go
@@ -23,9 +23,14 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
 	"github.com/VictoriaMetrics/metrics"
+	"github.com/valyala/histogram"
 	"gopkg.in/yaml.v2"
 )
 
+// defines ingested samples lag quantile to determine a time to wait before flush.
+// It's not configurable at the moment.
+const flushQuantile = 0.95
+
 var supportedOutputs = []string{
 	"avg",
 	"count_samples",
@@ -140,6 +145,9 @@ type Options struct {
 	//
 	// By default, aggregates samples are dropped, while the remaining samples are written to the corresponding -remoteWrite.url.
 	KeepInput bool
+
+	// EnableWindows enables aggregation in windows
+	EnableWindows bool
 }
 
 // Config is a configuration for a single stream aggregation.
@@ -247,6 +255,9 @@ type Config struct {
 	// OutputRelabelConfigs is an optional relabeling rules, which are applied
 	// on the aggregated output before being sent to remote storage.
 	OutputRelabelConfigs []promrelabel.RelabelConfig `yaml:"output_relabel_configs,omitempty"`
+
+	// EnableWindows enables aggregation in windows
+	EnableWindows *bool `yaml:"enable_windows,omitempty"`
 }
 
 // Aggregators aggregates metrics passed to Push and calls pushFunc for aggregated data.
@@ -367,17 +378,36 @@ func (a *Aggregators) Push(tss []prompbmarshal.TimeSeries, matchIdxs []byte) []b
 	return matchIdxs
 }
 
+type currentState struct {
+
+	// isGreen defines a state, where data sample with minDeadline < timestamp < maxDeadline will be aggregated
+	// and which is expected to be flushed with maxDeadline timestamp
+	isGreen bool
+
+	// maxDeadline is a threshold for samples from different states, when enableWindows is set
+	maxDeadline int64
+}
+
+func (cs *currentState) newState() *currentState {
+	return &currentState{
+		isGreen:     cs.isGreen,
+		maxDeadline: cs.maxDeadline,
+	}
+}
+
 // aggregator aggregates input series according to the config passed to NewAggregator
 type aggregator struct {
 	match *promrelabel.IfExpression
 
 	dropInputLabels []string
 
-	inputRelabeling  *promrelabel.ParsedConfigs
-	outputRelabeling *promrelabel.ParsedConfigs
+	inputRelabeling   *promrelabel.ParsedConfigs
+	outputRelabeling  *promrelabel.ParsedConfigs
+	stalenessInterval time.Duration
 
 	keepMetricNames  bool
 	ignoreOldSamples bool
+	enableWindows    bool
 
 	by                  []string
 	without             []string
@@ -392,11 +422,18 @@ type aggregator struct {
 	// da is set to non-nil if input samples must be de-duplicated
 	da *dedupAggr
 
+	// cs persists information of state, which is expected to be flushed next
+	cs atomic.Pointer[currentState]
+
+	// minDeadline is used for ignoring old samples when ignoreOldSamples or enableWindows is set
+	minDeadline atomic.Int64
+
 	// aggrOutputs contains aggregate states for the given outputs
-	aggrOutputs []aggrOutput
+	aggrOutputs *aggrOutputs
 
-	// minTimestamp is used for ignoring old samples when ignoreOldSamples is set
-	minTimestamp atomic.Int64
+	// time to wait after interval end before flush
+	flushAfter   *histogram.Fast
+	muFlushAfter sync.Mutex
 
 	// suffix contains a suffix, which should be added to aggregate metric names
 	//
@@ -408,36 +445,20 @@ type aggregator struct {
 	wg     sync.WaitGroup
 	stopCh chan struct{}
 
-	flushDuration      *metrics.Histogram
-	dedupFlushDuration *metrics.Histogram
-	samplesLag         *metrics.Histogram
-
-	flushTimeouts      *metrics.Counter
-	dedupFlushTimeouts *metrics.Counter
-	ignoredOldSamples  *metrics.Counter
-	ignoredNaNSamples  *metrics.Counter
-	matchedSamples     *metrics.Counter
-}
-
-type aggrOutput struct {
-	as aggrState
-
-	outputSamples *metrics.Counter
-}
-
-type aggrState interface {
-	// pushSamples must push samples to the aggrState.
-	//
-	// samples[].key must be cloned by aggrState, since it may change after returning from pushSamples.
-	pushSamples(samples []pushSample)
+	flushDuration *metrics.Histogram
+	samplesLag    *metrics.Histogram
 
-	// flushState must flush aggrState data to ctx.
-	flushState(ctx *flushCtx)
+	flushTimeouts     *metrics.Counter
+	ignoredOldSamples *metrics.Counter
+	ignoredNaNSamples *metrics.Counter
+	matchedSamples    *metrics.Counter
 }
 
 // PushFunc is called by Aggregators when it needs to push its state to metrics storage
 type PushFunc func(tss []prompbmarshal.TimeSeries)
 
+type aggrPushFunc func([]pushSample, int64, bool)
+
 // newAggregator creates new aggregator for the given cfg, which pushes the aggregate data to pushFunc.
 //
 // opts can contain additional options. If opts is nil, then default options are used.
@@ -557,31 +578,44 @@ func newAggregator(cfg *Config, path string, pushFunc PushFunc, ms *metrics.Set,
 		ignoreFirstIntervals = *v
 	}
 
+	// check cfg.
+	enableWindows := opts.EnableWindows
+	if v := cfg.EnableWindows; v != nil {
+		enableWindows = *v
+	}
+
 	// Initialize common metric labels
 	name := cfg.Name
 	if name == "" {
 		name = "none"
 	}
-	metricLabels := fmt.Sprintf(`name=%q,path=%q,url=%q,position="%d"`, name, path, alias, aggrID)
 
 	// initialize aggrOutputs
 	if len(cfg.Outputs) == 0 {
 		return nil, fmt.Errorf("`outputs` list must contain at least a single entry from the list %s; "+
 			"see https://docs.victoriametrics.com/stream-aggregation/", supportedOutputs)
 	}
-	aggrOutputs := make([]aggrOutput, len(cfg.Outputs))
+	useInputKey := dedupInterval <= 0
+	useSharedState := enableWindows && useInputKey
+	aggrOutputs := &aggrOutputs{
+		configs:        make([]aggrConfig, len(cfg.Outputs)),
+		useSharedState: useSharedState,
+		useInputKey:    useInputKey,
+	}
 	outputsSeen := make(map[string]struct{}, len(cfg.Outputs))
 	for i, output := range cfg.Outputs {
-		as, err := newAggrState(output, outputsSeen, stalenessInterval, ignoreFirstSampleInterval)
+		ac, err := newOutputConfig(output, outputsSeen, useSharedState, ignoreFirstSampleInterval)
 		if err != nil {
 			return nil, err
 		}
-		aggrOutputs[i] = aggrOutput{
-			as: as,
-
-			outputSamples: ms.NewCounter(fmt.Sprintf(`vm_streamaggr_output_samples_total{output=%q,%s}`, output, metricLabels)),
-		}
+		aggrOutputs.configs[i] = ac
+	}
+	outputsLabels := make([]string, 0, len(outputsSeen))
+	for o := range outputsSeen {
+		outputsLabels = append(outputsLabels, o)
 	}
+	metricLabels := fmt.Sprintf(`outputs=%q,name=%q,path=%q,url=%q,position="%d"`, strings.Join(outputsLabels, ","), name, path, alias, aggrID)
+	aggrOutputs.outputSamples = ms.NewCounter(fmt.Sprintf(`vm_streamaggr_output_samples_total{%s}`, metricLabels))
 
 	// initialize suffix to add to metric names after aggregation
 	suffix := ":" + cfg.Interval
@@ -601,8 +635,10 @@ func newAggregator(cfg *Config, path string, pushFunc PushFunc, ms *metrics.Set,
 		inputRelabeling:  inputRelabeling,
 		outputRelabeling: outputRelabeling,
 
-		keepMetricNames:  keepMetricNames,
-		ignoreOldSamples: ignoreOldSamples,
+		keepMetricNames:   keepMetricNames,
+		ignoreOldSamples:  ignoreOldSamples,
+		enableWindows:     enableWindows,
+		stalenessInterval: stalenessInterval,
 
 		by:                  by,
 		without:             without,
@@ -617,19 +653,19 @@ func newAggregator(cfg *Config, path string, pushFunc PushFunc, ms *metrics.Set,
 
 		stopCh: make(chan struct{}),
 
-		flushDuration:      ms.NewHistogram(fmt.Sprintf(`vm_streamaggr_flush_duration_seconds{%s}`, metricLabels)),
-		dedupFlushDuration: ms.NewHistogram(fmt.Sprintf(`vm_streamaggr_dedup_flush_duration_seconds{%s}`, metricLabels)),
-		samplesLag:         ms.NewHistogram(fmt.Sprintf(`vm_streamaggr_samples_lag_seconds{%s}`, metricLabels)),
+		flushDuration: ms.NewHistogram(fmt.Sprintf(`vm_streamaggr_flush_duration_seconds{%s}`, metricLabels)),
+		samplesLag:    ms.NewHistogram(fmt.Sprintf(`vm_streamaggr_samples_lag_seconds{%s}`, metricLabels)),
 
-		matchedSamples:     ms.NewCounter(fmt.Sprintf(`vm_streamaggr_matched_samples_total{%s}`, metricLabels)),
-		flushTimeouts:      ms.NewCounter(fmt.Sprintf(`vm_streamaggr_flush_timeouts_total{%s}`, metricLabels)),
-		dedupFlushTimeouts: ms.NewCounter(fmt.Sprintf(`vm_streamaggr_dedup_flush_timeouts_total{%s}`, metricLabels)),
-		ignoredNaNSamples:  ms.NewCounter(fmt.Sprintf(`vm_streamaggr_ignored_samples_total{reason="nan",%s}`, metricLabels)),
-		ignoredOldSamples:  ms.NewCounter(fmt.Sprintf(`vm_streamaggr_ignored_samples_total{reason="too_old",%s}`, metricLabels)),
+		matchedSamples:    ms.NewCounter(fmt.Sprintf(`vm_streamaggr_matched_samples_total{%s}`, metricLabels)),
+		flushTimeouts:     ms.NewCounter(fmt.Sprintf(`vm_streamaggr_flush_timeouts_total{%s}`, metricLabels)),
+		ignoredNaNSamples: ms.NewCounter(fmt.Sprintf(`vm_streamaggr_ignored_samples_total{reason="nan",%s}`, metricLabels)),
+		ignoredOldSamples: ms.NewCounter(fmt.Sprintf(`vm_streamaggr_ignored_samples_total{reason="too_old",%s}`, metricLabels)),
 	}
 
 	if dedupInterval > 0 {
 		a.da = newDedupAggr()
+		a.da.flushTimeouts = ms.NewCounter(fmt.Sprintf(`vm_streamaggr_dedup_flush_timeouts_total{%s}`, metricLabels))
+		a.da.flushDuration = ms.NewHistogram(fmt.Sprintf(`vm_streamaggr_dedup_flush_duration_seconds{%s}`, metricLabels))
 
 		_ = ms.NewGauge(fmt.Sprintf(`vm_streamaggr_dedup_state_size_bytes{%s}`, metricLabels), func() float64 {
 			n := a.da.sizeBytes()
@@ -651,6 +687,26 @@ func newAggregator(cfg *Config, path string, pushFunc PushFunc, ms *metrics.Set,
 		skipIncompleteFlush = !*v
 	}
 
+	startTime := time.Now()
+	minTime := startTime
+	if skipIncompleteFlush && alignFlushToInterval {
+		minTime = minTime.Truncate(a.interval)
+		if !startTime.Equal(minTime) {
+			minTime = minTime.Add(interval)
+		}
+	}
+	if enableWindows {
+		a.flushAfter = histogram.GetFast()
+	}
+	a.minDeadline.Store(minTime.UnixMilli())
+	cs := &currentState{}
+	if a.dedupInterval > 0 {
+		cs.maxDeadline = minTime.Add(a.dedupInterval).UnixMilli()
+	} else {
+		cs.maxDeadline = minTime.Add(a.interval).UnixMilli()
+	}
+	a.cs.Store(cs)
+
 	a.wg.Add(1)
 	go func() {
 		a.runFlusher(pushFunc, alignFlushToInterval, skipIncompleteFlush, ignoreFirstIntervals)
@@ -660,7 +716,7 @@ func newAggregator(cfg *Config, path string, pushFunc PushFunc, ms *metrics.Set,
 	return a, nil
 }
 
-func newAggrState(output string, outputsSeen map[string]struct{}, stalenessInterval, ignoreFirstSampleInterval time.Duration) (aggrState, error) {
+func newOutputConfig(output string, outputsSeen map[string]struct{}, useSharedState bool, ignoreFirstSampleInterval time.Duration) (aggrConfig, error) {
 	// check for duplicated output
 	if _, ok := outputsSeen[output]; ok {
 		return nil, fmt.Errorf("`outputs` list contains duplicate aggregation function: %s", output)
@@ -692,57 +748,62 @@ func newAggrState(output string, outputsSeen map[string]struct{}, stalenessInter
 			return nil, fmt.Errorf("`outputs` list contains duplicated `quantiles()` function, please combine multiple phi* like `quantiles(0.5, 0.9)`")
 		}
 		outputsSeen["quantiles"] = struct{}{}
-		return newQuantilesAggrState(phis), nil
+		return newQuantilesAggrConfig(phis), nil
 	}
+	ignoreFirstSampleIntervalSecs := uint64(ignoreFirstSampleInterval.Seconds())
 
 	switch output {
 	case "avg":
-		return newAvgAggrState(), nil
+		return newAvgAggrConfig(), nil
 	case "count_samples":
-		return newCountSamplesAggrState(), nil
+		return newCountSamplesAggrConfig(), nil
 	case "count_series":
-		return newCountSeriesAggrState(), nil
+		return newCountSeriesAggrConfig(), nil
 	case "histogram_bucket":
-		return newHistogramBucketAggrState(stalenessInterval), nil
+		return newHistogramBucketAggrConfig(useSharedState), nil
 	case "increase":
-		return newTotalAggrState(stalenessInterval, ignoreFirstSampleInterval, true, true), nil
+		return newTotalAggrConfig(ignoreFirstSampleIntervalSecs, true, true), nil
 	case "increase_prometheus":
-		return newTotalAggrState(stalenessInterval, ignoreFirstSampleInterval, true, false), nil
+		return newTotalAggrConfig(ignoreFirstSampleIntervalSecs, true, false), nil
 	case "last":
-		return newLastAggrState(), nil
+		return newLastAggrConfig(), nil
 	case "max":
-		return newMaxAggrState(), nil
+		return newMaxAggrConfig(), nil
 	case "min":
-		return newMinAggrState(), nil
+		return newMinAggrConfig(), nil
 	case "rate_avg":
-		return newRateAggrState(stalenessInterval, true), nil
+		return newRateAggrConfig(true), nil
 	case "rate_sum":
-		return newRateAggrState(stalenessInterval, false), nil
+		return newRateAggrConfig(false), nil
 	case "stddev":
-		return newStddevAggrState(), nil
+		return newStddevAggrConfig(), nil
 	case "stdvar":
-		return newStdvarAggrState(), nil
+		return newStdvarAggrConfig(), nil
 	case "sum_samples":
-		return newSumSamplesAggrState(), nil
+		return newSumSamplesAggrConfig(), nil
 	case "total":
-		return newTotalAggrState(stalenessInterval, ignoreFirstSampleInterval, false, true), nil
+		return newTotalAggrConfig(ignoreFirstSampleIntervalSecs, false, true), nil
 	case "total_prometheus":
-		return newTotalAggrState(stalenessInterval, ignoreFirstSampleInterval, false, false), nil
+		return newTotalAggrConfig(ignoreFirstSampleIntervalSecs, false, false), nil
 	case "unique_samples":
-		return newUniqueSamplesAggrState(), nil
+		return newUniqueSamplesAggrConfig(), nil
 	default:
 		return nil, fmt.Errorf("unsupported output=%q; supported values: %s; see https://docs.victoriametrics.com/stream-aggregation/", output, supportedOutputs)
 	}
 }
 
 func (a *aggregator) runFlusher(pushFunc PushFunc, alignFlushToInterval, skipIncompleteFlush bool, ignoreFirstIntervals int) {
-	alignedSleep := func(d time.Duration) {
-		if !alignFlushToInterval {
+	minTime := time.UnixMilli(a.minDeadline.Load())
+	flushTime := minTime.Add(a.interval)
+	interval := a.interval
+	if a.dedupInterval > 0 {
+		interval = a.dedupInterval
+	}
+	alignedSleep := func() {
+		dSleep := time.Until(minTime)
+		if dSleep <= 0 {
 			return
 		}
-
-		ct := time.Duration(time.Now().UnixNano())
-		dSleep := d - (ct % d)
 		timer := timerpool.Get(dSleep)
 		defer timer.Stop()
 		select {
@@ -751,135 +812,129 @@ func (a *aggregator) runFlusher(pushFunc PushFunc, alignFlushToInterval, skipInc
 		}
 	}
 
-	var flushTimeMsec int64
 	tickerWait := func(t *time.Ticker) bool {
 		select {
 		case <-a.stopCh:
-			flushTimeMsec = time.Now().UnixMilli()
 			return false
-		case ct := <-t.C:
-			flushTimeMsec = ct.UnixMilli()
+		case <-t.C:
 			return true
 		}
 	}
 
-	if a.dedupInterval <= 0 {
-		alignedSleep(a.interval)
-		t := time.NewTicker(a.interval)
-		defer t.Stop()
+	alignedSleep()
 
-		if alignFlushToInterval && skipIncompleteFlush {
-			a.flush(nil, 0)
-			ignoreFirstIntervals--
+	t := time.NewTicker(interval)
+	defer t.Stop()
+
+	var fa *histogram.Fast
+	for tickerWait(t) {
+		pf := pushFunc
+		if a.enableWindows {
+			// Calculate delay and wait
+			a.muFlushAfter.Lock()
+			fa, a.flushAfter = a.flushAfter, histogram.GetFast()
+			a.muFlushAfter.Unlock()
+			delay := time.Duration(fa.Quantile(flushQuantile)) * time.Millisecond
+			histogram.PutFast(fa)
+			time.Sleep(delay)
 		}
 
-		for tickerWait(t) {
+		cs := a.cs.Load().newState()
+		deadlineTime := time.UnixMilli(cs.maxDeadline)
+		a.dedupFlush(deadlineTime, cs)
+
+		if !flushTime.After(deadlineTime) {
+			// It is time to flush the aggregated state
 			if ignoreFirstIntervals > 0 {
-				a.flush(nil, 0)
+				a.flush(nil, flushTime, cs, false)
 				ignoreFirstIntervals--
 			} else {
-				a.flush(pushFunc, flushTimeMsec)
+				a.flush(pf, flushTime, cs, false)
 			}
-
-			if alignFlushToInterval {
-				select {
-				case <-t.C:
-				default:
-				}
+			for time.Now().After(flushTime) {
+				flushTime = flushTime.Add(a.interval)
+			}
+			if a.dedupInterval <= 0 {
+				cs.maxDeadline = flushTime.UnixMilli()
 			}
 		}
-	} else {
-		alignedSleep(a.dedupInterval)
-		t := time.NewTicker(a.dedupInterval)
-		defer t.Stop()
-
-		flushDeadline := time.Now().Add(a.interval)
-		isSkippedFirstFlush := false
-		for tickerWait(t) {
-			a.dedupFlush()
-
-			ct := time.Now()
-			if ct.After(flushDeadline) {
-				// It is time to flush the aggregated state
-				if alignFlushToInterval && skipIncompleteFlush && !isSkippedFirstFlush {
-					a.flush(nil, 0)
-					ignoreFirstIntervals--
-					isSkippedFirstFlush = true
-				} else if ignoreFirstIntervals > 0 {
-					a.flush(nil, 0)
-					ignoreFirstIntervals--
-				} else {
-					a.flush(pushFunc, flushTimeMsec)
-				}
-				for ct.After(flushDeadline) {
-					flushDeadline = flushDeadline.Add(a.interval)
-				}
+		if a.enableWindows {
+			cs.isGreen = !cs.isGreen
+		}
+		a.cs.Store(cs)
+		if alignFlushToInterval {
+			select {
+			case <-t.C:
+			default:
 			}
+		}
+	}
 
-			if alignFlushToInterval {
-				select {
-				case <-t.C:
-				default:
-				}
-			}
+	cs := a.cs.Load()
+	var dedupTime time.Time
+	if alignFlushToInterval {
+		if a.dedupInterval > 0 {
+			dedupTime = time.UnixMilli(cs.maxDeadline)
+		}
+	} else {
+		flushTime = time.Now()
+		if a.dedupInterval > 0 {
+			dedupTime = flushTime
 		}
 	}
 
-	if !skipIncompleteFlush && ignoreFirstIntervals <= 0 {
-		a.dedupFlush()
-		a.flush(pushFunc, flushTimeMsec)
+	a.dedupFlush(dedupTime, cs)
+	pf := pushFunc
+	if skipIncompleteFlush || ignoreFirstIntervals > 0 {
+		pf = nil
 	}
+	a.flush(pf, flushTime, cs, true)
 }
 
-func (a *aggregator) dedupFlush() {
+func (a *aggregator) dedupFlush(dedupTime time.Time, cs *currentState) {
 	if a.dedupInterval <= 0 {
 		// The de-duplication is disabled.
 		return
 	}
 
+	a.minDeadline.Store(cs.maxDeadline)
 	startTime := time.Now()
+	deleteDeadline := dedupTime.Add(a.stalenessInterval)
 
-	a.da.flush(a.pushSamples)
+	a.da.flush(a.aggrOutputs.pushSamples, deleteDeadline.UnixMilli(), cs.isGreen)
 
 	d := time.Since(startTime)
-	a.dedupFlushDuration.Update(d.Seconds())
+	a.da.flushDuration.Update(d.Seconds())
 	if d > a.dedupInterval {
-		a.dedupFlushTimeouts.Inc()
+		a.da.flushTimeouts.Inc()
 		logger.Warnf("deduplication couldn't be finished in the configured dedup_interval=%s; it took %.03fs; "+
 			"possible solutions: increase dedup_interval; use match filter matching smaller number of series; "+
 			"reduce samples' ingestion rate to stream aggregation", a.dedupInterval, d.Seconds())
 	}
+	for time.Now().After(dedupTime) {
+		dedupTime = dedupTime.Add(a.dedupInterval)
+	}
+	cs.maxDeadline = dedupTime.UnixMilli()
 }
 
 // flush flushes aggregator state to pushFunc.
 //
 // If pushFunc is nil, then the aggregator state is just reset.
-func (a *aggregator) flush(pushFunc PushFunc, flushTimeMsec int64) {
+func (a *aggregator) flush(pushFunc PushFunc, flushTime time.Time, cs *currentState, isLast bool) {
+	if a.dedupInterval > 0 {
+		a.minDeadline.Store(cs.maxDeadline)
+	}
+
 	startTime := time.Now()
+	ao := a.aggrOutputs
 
-	// Update minTimestamp before flushing samples to the storage,
-	// since the flush durtion can be quite long.
-	// This should prevent from dropping samples with old timestamps when the flush takes long time.
-	a.minTimestamp.Store(flushTimeMsec - 5_000)
-
-	var wg sync.WaitGroup
-	for i := range a.aggrOutputs {
-		ao := &a.aggrOutputs[i]
-		flushConcurrencyCh <- struct{}{}
-		wg.Add(1)
-		go func(ao *aggrOutput) {
-			defer func() {
-				<-flushConcurrencyCh
-				wg.Done()
-			}()
-
-			ctx := getFlushCtx(a, ao, pushFunc, flushTimeMsec)
-			ao.as.flushState(ctx)
-			ctx.flushSeries()
-			putFlushCtx(ctx)
-		}(ao)
-	}
-	wg.Wait()
+	ctx := getFlushCtx(a, ao, pushFunc, flushTime.UnixMilli(), isLast)
+	if a.dedupInterval <= 0 {
+		ctx.isGreen = cs.isGreen
+	}
+	ao.flushState(ctx)
+	ctx.flushSeries()
+	putFlushCtx(ctx)
 
 	d := time.Since(startTime)
 	a.flushDuration.Update(d.Seconds())
@@ -906,17 +961,21 @@ func (a *aggregator) Push(tss []prompbmarshal.TimeSeries, matchIdxs []byte) {
 	ctx := getPushCtx()
 	defer putPushCtx(ctx)
 
-	samples := ctx.samples
 	buf := ctx.buf
 	labels := &ctx.labels
 	inputLabels := &ctx.inputLabels
 	outputLabels := &ctx.outputLabels
+	now := time.Now()
+	nowMsec := now.UnixMilli()
+	deleteDeadline := now.Add(a.stalenessInterval)
+	deleteDeadlineMsec := deleteDeadline.UnixMilli()
 
+	minDeadline := a.minDeadline.Load()
 	dropLabels := a.dropInputLabels
 	ignoreOldSamples := a.ignoreOldSamples
-	minTimestamp := a.minTimestamp.Load()
+	enableWindows := a.enableWindows
+	cs := a.cs.Load()
 
-	nowMsec := time.Now().UnixMilli()
 	var maxLagMsec int64
 	for idx, ts := range tss {
 		if !a.match.Match(ts.Labels) {
@@ -951,47 +1010,66 @@ func (a *aggregator) Push(tss []prompbmarshal.TimeSeries, matchIdxs []byte) {
 		key := bytesutil.ToUnsafeString(buf[bufLen:])
 		for _, s := range ts.Samples {
 			if math.IsNaN(s.Value) {
-				a.ignoredNaNSamples.Inc()
 				// Skip NaN values
+				a.ignoredNaNSamples.Inc()
 				continue
 			}
-			if ignoreOldSamples && s.Timestamp < minTimestamp {
-				a.ignoredOldSamples.Inc()
+			if (ignoreOldSamples || enableWindows) && s.Timestamp < minDeadline {
 				// Skip old samples outside the current aggregation interval
+				a.ignoredOldSamples.Inc()
 				continue
 			}
 			lagMsec := nowMsec - s.Timestamp
 			if lagMsec > maxLagMsec {
 				maxLagMsec = lagMsec
 			}
-			samples = append(samples, pushSample{
-				key:       key,
-				value:     s.Value,
-				timestamp: s.Timestamp,
-			})
+			if enableWindows && s.Timestamp <= cs.maxDeadline == cs.isGreen {
+				ctx.green = append(ctx.green, pushSample{
+					key:       key,
+					value:     s.Value,
+					timestamp: s.Timestamp,
+				})
+			} else {
+				ctx.blue = append(ctx.blue, pushSample{
+					key:       key,
+					value:     s.Value,
+					timestamp: s.Timestamp,
+				})
+			}
 		}
 	}
-	if len(samples) > 0 {
-		a.matchedSamples.Add(len(samples))
-		a.samplesLag.Update(float64(maxLagMsec) / 1_000)
+	if enableWindows && maxLagMsec > 0 {
+		a.muFlushAfter.Lock()
+		a.flushAfter.Update(float64(maxLagMsec))
+		a.muFlushAfter.Unlock()
 	}
-	ctx.samples = samples
+	a.samplesLag.Update(float64(maxLagMsec) / 1_000)
+
 	ctx.buf = buf
 
+	pushSamples := a.aggrOutputs.pushSamples
 	if a.da != nil {
-		a.da.pushSamples(samples)
-	} else {
-		a.pushSamples(samples)
+		pushSamples = a.da.pushSamples
+	}
+
+	if len(ctx.blue) > 0 {
+		a.matchedSamples.Add(len(ctx.blue))
+		pushSamples(ctx.blue, deleteDeadlineMsec, false)
+	}
+
+	if len(ctx.green) > 0 {
+		a.matchedSamples.Add(len(ctx.green))
+		pushSamples(ctx.green, deleteDeadlineMsec, true)
 	}
 }
 
 func compressLabels(dst []byte, inputLabels, outputLabels []prompbmarshal.Label) []byte {
 	bb := bbPool.Get()
-	bb.B = lc.Compress(bb.B, inputLabels)
+	bb.B = lc.Compress(bb.B, outputLabels)
 	dst = encoding.MarshalVarUint64(dst, uint64(len(bb.B)))
 	dst = append(dst, bb.B...)
 	bbPool.Put(bb)
-	dst = lc.Compress(dst, outputLabels)
+	dst = lc.Compress(dst, inputLabels)
 	return dst
 }
 
@@ -999,37 +1077,9 @@ func decompressLabels(dst []prompbmarshal.Label, key string) []prompbmarshal.Lab
 	return lc.Decompress(dst, bytesutil.ToUnsafeBytes(key))
 }
 
-func getOutputKey(key string) string {
-	src := bytesutil.ToUnsafeBytes(key)
-	inputKeyLen, nSize := encoding.UnmarshalVarUint64(src)
-	if nSize <= 0 {
-		logger.Panicf("BUG: cannot unmarshal inputKeyLen from uvarint")
-	}
-	src = src[nSize:]
-	outputKey := src[inputKeyLen:]
-	return bytesutil.ToUnsafeString(outputKey)
-}
-
-func getInputOutputKey(key string) (string, string) {
-	src := bytesutil.ToUnsafeBytes(key)
-	inputKeyLen, nSize := encoding.UnmarshalVarUint64(src)
-	if nSize <= 0 {
-		logger.Panicf("BUG: cannot unmarshal inputKeyLen from uvarint")
-	}
-	src = src[nSize:]
-	inputKey := src[:inputKeyLen]
-	outputKey := src[inputKeyLen:]
-	return bytesutil.ToUnsafeString(inputKey), bytesutil.ToUnsafeString(outputKey)
-}
-
-func (a *aggregator) pushSamples(samples []pushSample) {
-	for _, ao := range a.aggrOutputs {
-		ao.as.pushSamples(samples)
-	}
-}
-
 type pushCtx struct {
-	samples      []pushSample
+	green        []pushSample
+	blue         []pushSample
 	labels       promutils.Labels
 	inputLabels  promutils.Labels
 	outputLabels promutils.Labels
@@ -1037,9 +1087,8 @@ type pushCtx struct {
 }
 
 func (ctx *pushCtx) reset() {
-	clear(ctx.samples)
-	ctx.samples = ctx.samples[:0]
-
+	ctx.blue = ctx.blue[:0]
+	ctx.green = ctx.green[:0]
 	ctx.labels.Reset()
 	ctx.inputLabels.Reset()
 	ctx.outputLabels.Reset()
@@ -1090,7 +1139,7 @@ func getInputOutputLabels(dstInput, dstOutput, labels []prompbmarshal.Label, by,
 	return dstInput, dstOutput
 }
 
-func getFlushCtx(a *aggregator, ao *aggrOutput, pushFunc PushFunc, flushTimestamp int64) *flushCtx {
+func getFlushCtx(a *aggregator, ao *aggrOutputs, pushFunc PushFunc, flushTimestamp int64, isLast bool) *flushCtx {
 	v := flushCtxPool.Get()
 	if v == nil {
 		v = &flushCtx{}
@@ -1100,6 +1149,7 @@ func getFlushCtx(a *aggregator, ao *aggrOutput, pushFunc PushFunc, flushTimestam
 	ctx.ao = ao
 	ctx.pushFunc = pushFunc
 	ctx.flushTimestamp = flushTimestamp
+	ctx.isLast = isLast
 	return ctx
 }
 
@@ -1112,9 +1162,11 @@ var flushCtxPool sync.Pool
 
 type flushCtx struct {
 	a              *aggregator
-	ao             *aggrOutput
+	ao             *aggrOutputs
 	pushFunc       PushFunc
 	flushTimestamp int64
+	isGreen        bool
+	isLast         bool
 
 	tss     []prompbmarshal.TimeSeries
 	labels  []prompbmarshal.Label
@@ -1125,6 +1177,8 @@ func (ctx *flushCtx) reset() {
 	ctx.a = nil
 	ctx.ao = nil
 	ctx.pushFunc = nil
+	ctx.isGreen = false
+	ctx.isLast = false
 	ctx.flushTimestamp = 0
 	ctx.resetSeries()
 }
diff --git a/lib/streamaggr/sum_samples.go b/lib/streamaggr/sum_samples.go
index 947239bb3..6705663d1 100644
--- a/lib/streamaggr/sum_samples.go
+++ b/lib/streamaggr/sum_samples.go
@@ -1,77 +1,28 @@
 package streamaggr
 
-import (
-	"sync"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-)
-
-// sumSamplesAggrState calculates output=sum_samples, e.g. the sum over input samples.
-type sumSamplesAggrState struct {
-	m sync.Map
+type sumSamplesAggrValue struct {
+	sum float64
 }
 
-type sumSamplesStateValue struct {
-	mu      sync.Mutex
-	sum     float64
-	deleted bool
+func (av *sumSamplesAggrValue) pushSample(_ aggrConfig, sample *pushSample, _ string, _ int64) {
+	av.sum += sample.value
 }
 
-func newSumSamplesAggrState() *sumSamplesAggrState {
-	return &sumSamplesAggrState{}
+func (av *sumSamplesAggrValue) flush(_ aggrConfig, ctx *flushCtx, key string) {
+	ctx.appendSeries(key, "sum_samples", av.sum)
+	av.sum = 0
 }
 
-func (as *sumSamplesAggrState) pushSamples(samples []pushSample) {
-	for i := range samples {
-		s := &samples[i]
-		outputKey := getOutputKey(s.key)
-
-	again:
-		v, ok := as.m.Load(outputKey)
-		if !ok {
-			// The entry is missing in the map. Try creating it.
-			v = &sumSamplesStateValue{
-				sum: s.value,
-			}
-			outputKey = bytesutil.InternString(outputKey)
-			vNew, loaded := as.m.LoadOrStore(outputKey, v)
-			if !loaded {
-				// The new entry has been successfully created.
-				continue
-			}
-			// Use the entry created by a concurrent goroutine.
-			v = vNew
-		}
-		sv := v.(*sumSamplesStateValue)
-		sv.mu.Lock()
-		deleted := sv.deleted
-		if !deleted {
-			sv.sum += s.value
-		}
-		sv.mu.Unlock()
-		if deleted {
-			// The entry has been deleted by the concurrent call to flushState
-			// Try obtaining and updating the entry again.
-			goto again
-		}
-	}
+func (*sumSamplesAggrValue) state() any {
+	return nil
 }
 
-func (as *sumSamplesAggrState) flushState(ctx *flushCtx) {
-	m := &as.m
-	m.Range(func(k, v any) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
+func newSumSamplesAggrConfig() aggrConfig {
+	return &sumSamplesAggrConfig{}
+}
 
-		sv := v.(*sumSamplesStateValue)
-		sv.mu.Lock()
-		sum := sv.sum
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-		sv.deleted = true
-		sv.mu.Unlock()
+type sumSamplesAggrConfig struct{}
 
-		key := k.(string)
-		ctx.appendSeries(key, "sum_samples", sum)
-		return true
-	})
+func (*sumSamplesAggrConfig) getValue(_ any) aggrValue {
+	return &sumSamplesAggrValue{}
 }
diff --git a/lib/streamaggr/total.go b/lib/streamaggr/total.go
index 53dbcdc47..af9e70e24 100644
--- a/lib/streamaggr/total.go
+++ b/lib/streamaggr/total.go
@@ -2,185 +2,122 @@ package streamaggr
 
 import (
 	"math"
-	"sync"
-	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
-// totalAggrState calculates output=total, total_prometheus, increase and increase_prometheus.
-type totalAggrState struct {
-	m sync.Map
-
-	// Whether to reset the output value on every flushState call.
-	resetTotalOnFlush bool
-
-	// Whether to take into account the first sample in new time series when calculating the output value.
-	keepFirstSample bool
-
-	// Time series state is dropped if no new samples are received during stalenessSecs.
-	//
-	// Aslo, the first sample per each new series is ignored during stalenessSecs even if keepFirstSample is set.
-	// see ignoreFirstSampleDeadline for more details.
-	stalenessSecs uint64
-
-	// The first sample per each new series is ignored until this unix timestamp deadline in seconds even if keepFirstSample is set.
-	// This allows avoiding an initial spike of the output values at startup when new time series
-	// cannot be distinguished from already existing series. This is tracked with ignoreFirstSampleDeadline.
-	ignoreFirstSampleDeadline uint64
-}
-
-type totalStateValue struct {
-	mu             sync.Mutex
-	lastValues     map[string]totalLastValueState
-	total          float64
-	deleteDeadline uint64
-	deleted        bool
-}
-
-type totalLastValueState struct {
+type totalLastValue struct {
 	value          float64
 	timestamp      int64
-	deleteDeadline uint64
+	deleteDeadline int64
 }
 
-func newTotalAggrState(stalenessInterval, ignoreFirstSampleInterval time.Duration, resetTotalOnFlush, keepFirstSample bool) *totalAggrState {
-	stalenessSecs := roundDurationToSecs(stalenessInterval)
-	ignoreFirstSampleDeadline := fasttime.UnixTimestamp() + roundDurationToSecs(ignoreFirstSampleInterval)
+type totalAggrValueShared struct {
+	lastValues map[string]totalLastValue
+	total      float64
+}
 
-	return &totalAggrState{
-		resetTotalOnFlush:         resetTotalOnFlush,
-		keepFirstSample:           keepFirstSample,
-		stalenessSecs:             stalenessSecs,
-		ignoreFirstSampleDeadline: ignoreFirstSampleDeadline,
-	}
+type totalAggrValue struct {
+	total  float64
+	shared *totalAggrValueShared
 }
 
-func (as *totalAggrState) pushSamples(samples []pushSample) {
+func (av *totalAggrValue) pushSample(c aggrConfig, sample *pushSample, key string, deleteDeadline int64) {
+	ac := c.(*totalAggrConfig)
 	currentTime := fasttime.UnixTimestamp()
-	deleteDeadline := currentTime + as.stalenessSecs
-	keepFirstSample := as.keepFirstSample && currentTime >= as.ignoreFirstSampleDeadline
-	for i := range samples {
-		s := &samples[i]
-		inputKey, outputKey := getInputOutputKey(s.key)
-
-	again:
-		v, ok := as.m.Load(outputKey)
-		if !ok {
-			// The entry is missing in the map. Try creating it.
-			v = &totalStateValue{
-				lastValues: make(map[string]totalLastValueState),
-			}
-			outputKey = bytesutil.InternString(outputKey)
-			vNew, loaded := as.m.LoadOrStore(outputKey, v)
-			if loaded {
-				// Use the entry created by a concurrent goroutine.
-				v = vNew
-			}
+	keepFirstSample := ac.keepFirstSample && currentTime >= ac.ignoreFirstSampleDeadline
+	lv, ok := av.shared.lastValues[key]
+	if ok || keepFirstSample {
+		if sample.timestamp < lv.timestamp {
+			// Skip out of order sample
+			return
 		}
-		sv := v.(*totalStateValue)
-		sv.mu.Lock()
-		deleted := sv.deleted
-		if !deleted {
-			lv, ok := sv.lastValues[inputKey]
-			if ok || keepFirstSample {
-				if s.timestamp < lv.timestamp {
-					// Skip out of order sample
-					sv.mu.Unlock()
-					continue
-				}
-
-				if s.value >= lv.value {
-					sv.total += s.value - lv.value
-				} else {
-					// counter reset
-					sv.total += s.value
-				}
-			}
-			lv.value = s.value
-			lv.timestamp = s.timestamp
-			lv.deleteDeadline = deleteDeadline
-
-			inputKey = bytesutil.InternString(inputKey)
-			sv.lastValues[inputKey] = lv
-			sv.deleteDeadline = deleteDeadline
+		if sample.value >= lv.value {
+			av.total += sample.value - lv.value
+		} else {
+			// counter reset
+			av.total += sample.value
 		}
-		sv.mu.Unlock()
-		if deleted {
-			// The entry has been deleted by the concurrent call to flushState
-			// Try obtaining and updating the entry again.
-			goto again
+	}
+	lv.value = sample.value
+	lv.timestamp = sample.timestamp
+	lv.deleteDeadline = deleteDeadline
+	key = bytesutil.InternString(key)
+	av.shared.lastValues[key] = lv
+}
+
+func (av *totalAggrValue) flush(c aggrConfig, ctx *flushCtx, key string) {
+	ac := c.(*totalAggrConfig)
+	suffix := ac.getSuffix()
+	// check for stale entries
+	total := av.shared.total + av.total
+	av.total = 0
+	lvs := av.shared.lastValues
+	for lk, lv := range lvs {
+		if ctx.flushTimestamp > lv.deleteDeadline {
+			delete(lvs, lk)
 		}
 	}
+	if ac.resetTotalOnFlush {
+		av.shared.total = 0
+	} else if math.Abs(total) >= (1 << 53) {
+		// It is time to reset the entry, since it starts losing float64 precision
+		av.shared.total = 0
+	} else {
+		av.shared.total = total
+	}
+	ctx.appendSeries(key, suffix, total)
 }
 
-func (as *totalAggrState) flushState(ctx *flushCtx) {
-	currentTime := fasttime.UnixTimestamp()
+func (av *totalAggrValue) state() any {
+	return av.shared
+}
 
-	suffix := as.getSuffix()
+func newTotalAggrConfig(ignoreFirstSampleIntervalSecs uint64, resetTotalOnFlush, keepFirstSample bool) aggrConfig {
+	ignoreFirstSampleDeadline := fasttime.UnixTimestamp() + ignoreFirstSampleIntervalSecs
+	return &totalAggrConfig{
+		keepFirstSample:           keepFirstSample,
+		resetTotalOnFlush:         resetTotalOnFlush,
+		ignoreFirstSampleDeadline: ignoreFirstSampleDeadline,
+	}
+}
 
-	as.removeOldEntries(currentTime)
+type totalAggrConfig struct {
+	resetTotalOnFlush bool
 
-	m := &as.m
-	m.Range(func(k, v any) bool {
-		sv := v.(*totalStateValue)
+	// Whether to take into account the first sample in new time series when calculating the output value.
+	keepFirstSample bool
 
-		sv.mu.Lock()
-		total := sv.total
-		if as.resetTotalOnFlush {
-			sv.total = 0
-		} else if math.Abs(sv.total) >= (1 << 53) {
-			// It is time to reset the entry, since it starts losing float64 precision
-			sv.total = 0
-		}
-		deleted := sv.deleted
-		sv.mu.Unlock()
+	// The first sample per each new series is ignored until this unix timestamp deadline in seconds even if keepFirstSample is set.
+	// This allows avoiding an initial spike of the output values at startup when new time series
+	// cannot be distinguished from already existing series. This is tracked with ignoreFirstSampleDeadline.
+	ignoreFirstSampleDeadline uint64
+}
 
-		if !deleted {
-			key := k.(string)
-			ctx.appendSeries(key, suffix, total)
+func (*totalAggrConfig) getValue(s any) aggrValue {
+	var shared *totalAggrValueShared
+	if s == nil {
+		shared = &totalAggrValueShared{
+			lastValues: make(map[string]totalLastValue),
 		}
-		return true
-	})
+	} else {
+		shared = s.(*totalAggrValueShared)
+	}
+	return &totalAggrValue{
+		shared: shared,
+	}
 }
 
-func (as *totalAggrState) getSuffix() string {
-	// Note: this function is at hot path, so it shouldn't allocate.
-	if as.resetTotalOnFlush {
-		if as.keepFirstSample {
+func (ac *totalAggrConfig) getSuffix() string {
+	if ac.resetTotalOnFlush {
+		if ac.keepFirstSample {
 			return "increase"
 		}
 		return "increase_prometheus"
 	}
-	if as.keepFirstSample {
+	if ac.keepFirstSample {
 		return "total"
 	}
 	return "total_prometheus"
 }
-
-func (as *totalAggrState) removeOldEntries(currentTime uint64) {
-	m := &as.m
-	m.Range(func(k, v any) bool {
-		sv := v.(*totalStateValue)
-
-		sv.mu.Lock()
-		if currentTime > sv.deleteDeadline {
-			// Mark the current entry as deleted
-			sv.deleted = true
-			sv.mu.Unlock()
-			m.Delete(k)
-			return true
-		}
-
-		// Delete outdated entries in sv.lastValues
-		lvs := sv.lastValues
-		for k1, lv := range lvs {
-			if currentTime > lv.deleteDeadline {
-				delete(lvs, k1)
-			}
-		}
-		sv.mu.Unlock()
-		return true
-	})
-}
diff --git a/lib/streamaggr/unique_samples.go b/lib/streamaggr/unique_samples.go
index 93145397c..f351015c6 100644
--- a/lib/streamaggr/unique_samples.go
+++ b/lib/streamaggr/unique_samples.go
@@ -1,81 +1,34 @@
 package streamaggr
 
-import (
-	"sync"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-)
-
-// uniqueSamplesAggrState calculates output=unique_samples, e.g. the number of unique sample values.
-type uniqueSamplesAggrState struct {
-	m sync.Map
+type uniqueSamplesAggrValue struct {
+	samples map[float64]struct{}
 }
 
-type uniqueSamplesStateValue struct {
-	mu      sync.Mutex
-	m       map[float64]struct{}
-	deleted bool
+func (av *uniqueSamplesAggrValue) pushSample(_ aggrConfig, sample *pushSample, _ string, _ int64) {
+	if _, ok := av.samples[sample.value]; !ok {
+		av.samples[sample.value] = struct{}{}
+	}
 }
 
-func newUniqueSamplesAggrState() *uniqueSamplesAggrState {
-	return &uniqueSamplesAggrState{}
+func (av *uniqueSamplesAggrValue) flush(_ aggrConfig, ctx *flushCtx, key string) {
+	if len(av.samples) > 0 {
+		ctx.appendSeries(key, "unique_samples", float64(len(av.samples)))
+		clear(av.samples)
+	}
 }
 
-func (as *uniqueSamplesAggrState) pushSamples(samples []pushSample) {
-	for i := range samples {
-		s := &samples[i]
-		outputKey := getOutputKey(s.key)
-
-	again:
-		v, ok := as.m.Load(outputKey)
-		if !ok {
-			// The entry is missing in the map. Try creating it.
-			v = &uniqueSamplesStateValue{
-				m: map[float64]struct{}{
-					s.value: {},
-				},
-			}
-			outputKey = bytesutil.InternString(outputKey)
-			vNew, loaded := as.m.LoadOrStore(outputKey, v)
-			if !loaded {
-				// The new entry has been successfully created.
-				continue
-			}
-			// Use the entry created by a concurrent goroutine.
-			v = vNew
-		}
-		sv := v.(*uniqueSamplesStateValue)
-		sv.mu.Lock()
-		deleted := sv.deleted
-		if !deleted {
-			if _, ok := sv.m[s.value]; !ok {
-				sv.m[s.value] = struct{}{}
-			}
-		}
-		sv.mu.Unlock()
-		if deleted {
-			// The entry has been deleted by the concurrent call to flushState
-			// Try obtaining and updating the entry again.
-			goto again
-		}
-	}
+func (*uniqueSamplesAggrValue) state() any {
+	return nil
 }
 
-func (as *uniqueSamplesAggrState) flushState(ctx *flushCtx) {
-	m := &as.m
-	m.Range(func(k, v any) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
+func newUniqueSamplesAggrConfig() aggrConfig {
+	return &uniqueSamplesAggrConfig{}
+}
 
-		sv := v.(*uniqueSamplesStateValue)
-		sv.mu.Lock()
-		n := len(sv.m)
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-		sv.deleted = true
-		sv.mu.Unlock()
+type uniqueSamplesAggrConfig struct{}
 
-		key := k.(string)
-		ctx.appendSeries(key, "unique_samples", float64(n))
-		return true
-	})
+func (*uniqueSamplesAggrConfig) getValue(_ any) aggrValue {
+	return &uniqueSamplesAggrValue{
+		samples: make(map[float64]struct{}),
+	}
 }


Test patch:
diff --git a/app/vmagent/remotewrite/remotewrite_test.go b/app/vmagent/remotewrite/remotewrite_test.go
index 01a9fc1f2..6461eda3f 100644
--- a/app/vmagent/remotewrite/remotewrite_test.go
+++ b/app/vmagent/remotewrite/remotewrite_test.go
@@ -53,7 +53,7 @@ func TestGetLabelsHash_Distribution(t *testing.T) {
 }
 
 func TestRemoteWriteContext_TryPush_ImmutableTimeseries(t *testing.T) {
-	f := func(streamAggrConfig, relabelConfig string, dedupInterval time.Duration, keepInput, dropInput bool, input string) {
+	f := func(streamAggrConfig, relabelConfig string, enableWindows bool, dedupInterval time.Duration, keepInput, dropInput bool, input string) {
 		t.Helper()
 		perURLRelabel, err := promrelabel.ParseRelabelConfigsData([]byte(relabelConfig))
 		if err != nil {
@@ -77,12 +77,15 @@ func TestRemoteWriteContext_TryPush_ImmutableTimeseries(t *testing.T) {
 			rowsDroppedByRelabel:   metrics.GetOrCreateCounter(`bar`),
 		}
 		if dedupInterval > 0 {
-			rwctx.deduplicator = streamaggr.NewDeduplicator(nil, dedupInterval, nil, "dedup-global")
+			rwctx.deduplicator = streamaggr.NewDeduplicator(nil, enableWindows, dedupInterval, nil, "dedup-global")
 		}
 
 		if streamAggrConfig != "" {
 			pushNoop := func(_ []prompbmarshal.TimeSeries) {}
-			sas, err := streamaggr.LoadFromData([]byte(streamAggrConfig), pushNoop, nil, "global")
+			opts := streamaggr.Options{
+				EnableWindows: enableWindows,
+			}
+			sas, err := streamaggr.LoadFromData([]byte(streamAggrConfig), pushNoop, &opts, "global")
 			if err != nil {
 				t.Fatalf("cannot load streamaggr configs: %s", err)
 			}
@@ -114,13 +117,13 @@ func TestRemoteWriteContext_TryPush_ImmutableTimeseries(t *testing.T) {
 - action: keep
   source_labels: [env]
   regex: "dev"
-`, 0, false, false, `
+`, false, 0, false, false, `
 metric{env="dev"} 10
 metric{env="bar"} 20
 metric{env="dev"} 15
 metric{env="bar"} 25
 `)
-	f(``, ``, time.Hour, false, false, `
+	f(``, ``, true, time.Hour, false, false, `
 metric{env="dev"} 10
 metric{env="foo"} 20
 metric{env="dev"} 15
@@ -130,7 +133,7 @@ metric{env="foo"} 25
 - action: keep
   source_labels: [env]
   regex: "dev"
-`, time.Hour, false, false, `
+`, true, time.Hour, false, false, `
 metric{env="dev"} 10
 metric{env="bar"} 20
 metric{env="dev"} 15
@@ -140,7 +143,7 @@ metric{env="bar"} 25
 - action: keep
   source_labels: [env]
   regex: "dev"
-`, time.Hour, true, false, `
+`, true, time.Hour, true, false, `
 metric{env="test"} 10
 metric{env="dev"} 20
 metric{env="foo"} 15
@@ -150,7 +153,7 @@ metric{env="dev"} 25
 - action: keep
   source_labels: [env]
   regex: "dev"
-`, time.Hour, false, true, `
+`, true, time.Hour, false, true, `
 metric{env="foo"} 10
 metric{env="dev"} 20
 metric{env="foo"} 15
@@ -160,7 +163,7 @@ metric{env="dev"} 25
 - action: keep
   source_labels: [env]
   regex: "dev"
-`, time.Hour, true, true, `
+`, true, time.Hour, true, true, `
 metric{env="dev"} 10
 metric{env="test"} 20
 metric{env="dev"} 15
diff --git a/lib/streamaggr/dedup_test.go b/lib/streamaggr/dedup_test.go
index 91ce09e13..b97d0a74e 100644
--- a/lib/streamaggr/dedup_test.go
+++ b/lib/streamaggr/dedup_test.go
@@ -5,6 +5,7 @@ import (
 	"reflect"
 	"sync"
 	"testing"
+	"time"
 )
 
 func TestDedupAggrSerial(t *testing.T) {
@@ -12,16 +13,14 @@ func TestDedupAggrSerial(t *testing.T) {
 
 	const seriesCount = 100_000
 	expectedSamplesMap := make(map[string]pushSample)
-	for i := 0; i < 2; i++ {
-		samples := make([]pushSample, seriesCount)
-		for j := range samples {
-			sample := &samples[j]
-			sample.key = fmt.Sprintf("key_%d", j)
-			sample.value = float64(i + j)
-			expectedSamplesMap[sample.key] = *sample
-		}
-		da.pushSamples(samples)
+	samples := make([]pushSample, seriesCount)
+	for j := range samples {
+		sample := &samples[j]
+		sample.key = fmt.Sprintf("key_%d", j)
+		sample.value = float64(j)
+		expectedSamplesMap[sample.key] = *sample
 	}
+	da.pushSamples(samples, 0, false)
 
 	if n := da.sizeBytes(); n > 5_000_000 {
 		t.Fatalf("too big dedupAggr state before flush: %d bytes; it shouldn't exceed 5_000_000 bytes", n)
@@ -32,14 +31,16 @@ func TestDedupAggrSerial(t *testing.T) {
 
 	flushedSamplesMap := make(map[string]pushSample)
 	var mu sync.Mutex
-	flushSamples := func(samples []pushSample) {
+	flushSamples := func(samples []pushSample, _ int64, _ bool) {
 		mu.Lock()
 		for _, sample := range samples {
 			flushedSamplesMap[sample.key] = sample
 		}
 		mu.Unlock()
 	}
-	da.flush(flushSamples)
+
+	flushTimestamp := time.Now().UnixMilli()
+	da.flush(flushSamples, flushTimestamp, false)
 
 	if !reflect.DeepEqual(expectedSamplesMap, flushedSamplesMap) {
 		t.Fatalf("unexpected samples;\ngot\n%v\nwant\n%v", flushedSamplesMap, expectedSamplesMap)
@@ -70,7 +71,7 @@ func TestDedupAggrConcurrent(_ *testing.T) {
 					sample.key = fmt.Sprintf("key_%d", j)
 					sample.value = float64(i + j)
 				}
-				da.pushSamples(samples)
+				da.pushSamples(samples, 0, false)
 			}
 		}()
 	}
diff --git a/lib/streamaggr/dedup_timing_test.go b/lib/streamaggr/dedup_timing_test.go
index 2b6bab25c..8079bf4c7 100644
--- a/lib/streamaggr/dedup_timing_test.go
+++ b/lib/streamaggr/dedup_timing_test.go
@@ -27,7 +27,7 @@ func benchmarkDedupAggr(b *testing.B, samplesPerPush int) {
 	b.RunParallel(func(pb *testing.PB) {
 		for pb.Next() {
 			for i := 0; i < loops; i++ {
-				da.pushSamples(benchSamples)
+				da.pushSamples(benchSamples, 0, false)
 			}
 		}
 	})
diff --git a/lib/streamaggr/deduplicator_test.go b/lib/streamaggr/deduplicator_test.go
index 1d304f6dd..bfbc33c8d 100644
--- a/lib/streamaggr/deduplicator_test.go
+++ b/lib/streamaggr/deduplicator_test.go
@@ -17,24 +17,26 @@ func TestDeduplicator(t *testing.T) {
 		tssResultLock.Unlock()
 	}
 
-	offsetMsecs := time.Now().UnixMilli()
+	offsetMsecs := time.Now().Add(time.Minute).UnixMilli()
 	tss := prompbmarshal.MustParsePromMetrics(`
 foo{instance="x",job="aaa",pod="sdfd-dfdfdfs",node="aosijjewrerfd",namespace="asdff",container="ohohffd"} 123
 bar{instance="x",job="aaa",pod="sdfd-dfdfdfs",node="aosijjewrerfd",namespace="asdff",container="ohohffd"} 34.54
-x 8943 1000
+x 8943 1
 baz_aaa_aaa_fdd{instance="x",job="aaa",pod="sdfd-dfdfdfs",node="aosijjewrerfd",namespace="asdff",container="ohohffd"} -34.34
-x 90984 900
-x 433 1000
+x 90984
+x 433 1
 asfjkldsf{instance="x",job="aaa",pod="sdfd-dfdfdfs",node="aosijjewrerfd",namespace="asdff",container="ohohffd"} 12322
 foo{instance="x",job="aaa",pod="sdfd-dfdfdfs",node="aosijjewrerfd",namespace="asdff",container="ohohffd"} 894
 baz_aaa_aaa_fdd{instance="x",job="aaa",pod="sdfd-dfdfdfs",node="aosijjewrerfd",namespace="asdff",container="ohohffd"} -2.3
 `, offsetMsecs)
 
-	d := NewDeduplicator(pushFunc, time.Hour, []string{"node", "instance"}, "global")
+	dedupInterval := time.Hour
+	d := NewDeduplicator(pushFunc, true, dedupInterval, []string{"node", "instance"}, "global")
 	for i := 0; i < 10; i++ {
 		d.Push(tss)
 	}
-	d.flush(pushFunc, time.Hour)
+
+	d.flush(pushFunc)
 	d.MustStop()
 
 	result := timeSeriessToString(tssResult)
diff --git a/lib/streamaggr/deduplicator_timing_test.go b/lib/streamaggr/deduplicator_timing_test.go
index ed98b638c..479f3e13a 100644
--- a/lib/streamaggr/deduplicator_timing_test.go
+++ b/lib/streamaggr/deduplicator_timing_test.go
@@ -9,7 +9,7 @@ import (
 
 func BenchmarkDeduplicatorPush(b *testing.B) {
 	pushFunc := func(_ []prompbmarshal.TimeSeries) {}
-	d := NewDeduplicator(pushFunc, time.Hour, nil, "global")
+	d := NewDeduplicator(pushFunc, true, time.Hour, nil, "global")
 
 	b.ReportAllocs()
 	b.SetBytes(int64(len(benchSeries)))
diff --git a/lib/streamaggr/streamaggr_test.go b/lib/streamaggr/streamaggr_test.go
index 8338be12c..ccc04ef19 100644
--- a/lib/streamaggr/streamaggr_test.go
+++ b/lib/streamaggr/streamaggr_test.go
@@ -200,11 +200,14 @@ func TestAggregatorsEqual(t *testing.T) {
 		t.Helper()
 
 		pushFunc := func(_ []prompbmarshal.TimeSeries) {}
-		aa, err := LoadFromData([]byte(a), pushFunc, nil, "some_alias")
+		opts := Options{
+			EnableWindows: true,
+		}
+		aa, err := LoadFromData([]byte(a), pushFunc, &opts, "some_alias")
 		if err != nil {
 			t.Fatalf("cannot initialize aggregators: %s", err)
 		}
-		ab, err := LoadFromData([]byte(b), pushFunc, nil, "some_alias")
+		ab, err := LoadFromData([]byte(b), pushFunc, &opts, "some_alias")
 		if err != nil {
 			t.Fatalf("cannot initialize aggregators: %s", err)
 		}
@@ -266,6 +269,7 @@ func TestAggregatorsSuccess(t *testing.T) {
 		opts := &Options{
 			FlushOnShutdown:        true,
 			NoAlignFlushToInterval: true,
+			EnableWindows:          true,
 		}
 		a, err := LoadFromData([]byte(config), pushFunc, opts, "some_alias")
 		if err != nil {
@@ -273,7 +277,7 @@ func TestAggregatorsSuccess(t *testing.T) {
 		}
 
 		// Push the inputMetrics to Aggregators
-		offsetMsecs := time.Now().UnixMilli()
+		offsetMsecs := time.Now().Add(15 * time.Second).UnixMilli()
 		tssInput := prompbmarshal.MustParsePromMetrics(inputMetrics, offsetMsecs)
 		matchIdxs := a.Push(tssInput, nil)
 		a.MustStop()
@@ -305,7 +309,7 @@ func TestAggregatorsSuccess(t *testing.T) {
   outputs: [count_samples, sum_samples, count_series, last]
 `, `
 foo{abc="123"} 4
-bar 5 100
+bar 5 11
 bar 34 10
 foo{abc="123"} 8.5
 foo{abc="456",de="fg"} 8
@@ -586,7 +590,7 @@ foo:1m_total_prometheus 0
 `, `
 foo 123
 bar{baz="qwe"} 1.31
-bar{baz="qwe"} 4.34 1000
+bar{baz="qwe"} 4.34 1
 bar{baz="qwe"} 2
 foo{baz="qwe"} -5
 bar{baz="qwer"} 343
@@ -997,6 +1001,7 @@ func TestAggregatorsWithDedupInterval(t *testing.T) {
 		opts := &Options{
 			DedupInterval:   30 * time.Second,
 			FlushOnShutdown: true,
+			EnableWindows:   true,
 		}
 		a, err := LoadFromData([]byte(config), pushFunc, opts, "some_alias")
 		if err != nil {
@@ -1004,7 +1009,7 @@ func TestAggregatorsWithDedupInterval(t *testing.T) {
 		}
 
 		// Push the inputMetrics to Aggregators
-		offsetMsecs := time.Now().UnixMilli()
+		offsetMsecs := time.Now().Add(15 * time.Second).UnixMilli()
 		tssInput := prompbmarshal.MustParsePromMetrics(inputMetrics, offsetMsecs)
 		matchIdxs := a.Push(tssInput, nil)
 		a.MustStop()