feat(metrics): reset-exact cumulative rate/increase via counter epochs

Cumulative counters gain an epoch-aware temporal aggregation pipeline
behind the per-org flag use_counter_epochs. The collector stamps every
cumulative monotonic sample with its normalized start_ts (epoch); within
an epoch the counter is monotone, so per-(bucket, epoch) min/max are
first/last observations and increase is the sum of per-epoch growth —
exact under any number of resets, identical at every step interval, and
computable from both the raw table and the 5m/30m rollups (which carry
per-epoch min/max as mergeable minMap/maxMap states; collector migration
1012).

start_ts = 0 rows (pre-rollout data, sources without a start time) flow
through the same statement with the legacy negative-diff semantics, so a
query over old data reproduces today's results, new data is reset-exact,
and ranges spanning the rollout seam the two per row with no watermark.
Multiple-temporality metrics become a union of the delta aggregation and
the epoch pipeline. The querier bucket-cache fingerprint includes the
pipeline state so legacy and epoch buckets never stitch.

Verification: tests/integration/testdata/counter_reset_epochs/ — a
one-day, 15-scenario dataset with three independent reference
implementations cross-asserted, then the builder-emitted SQL compared
row-for-row on ClickHouse (raw + rollup paths, pre/post-merge, 11 query
cases), plus whale-scale (50k series / 144M samples), shard-pushdown,
and layer-profile benchmarks. Design doc: docs/counter-reset-epochs.md.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

docs/counter-reset-epochs.md

@@ -0,0 +1,332 @@
+# Counter-reset epochs: read side
+
+Status: implemented behind the per-org feature flag `use_counter_epochs`.
+Write side: `docs/counter-reset-epochs.md` in the signoz-otel-collector repo
+(`start_ts` normalizer, schema migration 1012).
+Verification: `tests/integration/testdata/counter_reset_epochs/`.
+
+## Summary
+
+Cumulative rate/increase in the v5 query builder gains a reset-exact pipeline.
+The collector stamps every cumulative monotonic sample with its **epoch** (the
+normalized OTLP `start_time_unix_nano`, ms, column `start_ts`; `0` = unknown).
+Within an epoch the counter is monotone, so per-(step bucket, epoch) min/max
+are the first/last observations and the increase over any window is the sum of
+per-epoch growth — exact under any number of resets, identical at every step
+interval, computable from the raw table and from the 5m/30m rollups (which
+carry per-epoch min/max as mergeable `minMap`/`maxMap` states).
+
+What this fixes, concretely (issues #6973, #7069, and the auto-vs-1d step
+inconsistency reports):
+
+| problem | legacy | epochs |
+|---|---|---|
+| reset inside a step bucket | undercounted (`max` hides it) | exact |
+| reset regrowing past the previous value | invisible, silent undercount | exact |
+| N resets inside a 1-day step | miscounted (one negative diff at most) | exact |
+| zoom consistency (auto vs 86400) | day totals disagree (harness: 3969 vs 242) | identical at every step |
+| rollup tables (5m/30m) | loss baked in at write | reset-exact from rollups |
+| first-ever data point | invisible (`nan`) | visible (counts from 0) |
+| two writers sharing a fingerprint | sawtooth garbage | each epoch tracked, growth summed |
+
+The irreducible losses (shared with created-timestamp-enabled Prometheus):
+growth between an epoch's last sample and its death, epochs that live and die
+between two exports, and broken-start-time sources — which degrade to exactly
+today's behavior, never worse.
+
+## The pipeline
+
+`buildTemporalAggCumulativeEpochs` (pkg/telemetrymetrics/epoch_statement.go)
+replaces the temporal aggregation CTE for cumulative rate/increase when the
+flag is on. Four layers, innermost first:
+
+```
+L1  per (fingerprint, bucket):    __first_by_epoch / __last_by_epoch maps
+      raw tables:   minMap/maxMap(map(start_ts, value)), stale rows filtered
+      5m/30m:       minMap(min_value_by_start_ts) / maxMap(max_value_by_start_ts);
+                    a bucket whose merged maps are empty (all rows pre-migration)
+                    falls back to the scalar min/max as key 0
+L2  per-series bucket window:     previous bucket ts + overall max, last known
+                                  key-0 value, ever-had-key-0, row numbers
+L3  ARRAY JOIN over epochs:       per-(series, epoch) lag, then the
+                                  contribution multiIf below
+L4  regroup per (series, bucket): increase = sum of finite contributions
+                                  (NaN if none); rate divides by the distance
+                                  to the series' previous bucket (step for the
+                                  first); buckets before the display start are
+                                  cut (they exist only as bases)
+```
+
+Spatial aggregation is untouched — epochs are summed inside the series before
+anything crosses series boundaries, which is why this does not repeat the
+start-time-in-fingerprint mistake (series identity never changes; no
+cardinality or billing impact).
+
+Contribution rules per (bucket, epoch) row, in `multiIf` branch order:
+
+| # | condition | contribution | meaning |
+|---|---|---|---|
+| 1 | epoch ≠ 0, seen in an earlier bucket | `last − previous last of this epoch` (clamped to `last` if negative) | continuation; the clamp only fires if upstream broke the monotonicity invariant |
+| 2 | epoch ≠ 0, first appearance, bucket has current/earlier key-0 rows | `last − (first ≥ seam ? seam : 0)`, seam = same-bucket key-0 value, else last key-0 value | the rollout seam: continuation subtracts the legacy value, a drop keeps base 0 |
+| 3 | epoch ≠ 0, first appearance, epoch born ≥ fetched-range start | `last` | the counter started from 0 inside the range; makes single-shot scripts visible |
+| 4 | epoch ≠ 0, first appearance, born before the range | `last − first` | only the growth visible inside the bucket; the pre-range part is unknowable |
+| 5 | epoch = 0, series' first bucket | `NaN` | legacy first-point behavior |
+| 6 | epoch = 0, value < previous bucket's overall max | `value` | legacy negative-diff rule |
+| 7 | epoch = 0, otherwise | `value − previous bucket's overall max` | legacy pair rule; "overall max" (not key-0-only) also seams the epoch→0 downgrade direction |
+
+## How queries behave across the rollout — the three regimes
+
+The **same statement** serves all three regimes; they differ only in which
+branches fire, row by row. There is no watermark, no config, no per-time-range
+switching.
+
+### Regime 1 — no rollout in range (all rows have `start_ts = 0`)
+
+Every row lands in the maps under key 0. Branches 5–7 are the only ones that
+fire, and they are exactly the legacy semantics (bucket max, `nan` first
+point, negative-diff clamp). Harness assertions A5/A6 prove byte-equality of
+results with the legacy pipeline on epoch-less and reset-free data, and the
+`l_*` query cases pin the legacy pipeline itself. Flag on + old data =
+identical charts.
+
+### Regime 2 — fully rolled out (every cumulative row has a real epoch)
+
+Maps carry real epochs; branches 1–4 fire. Per bucket, each epoch contributes
+its observed growth; a reset inside a bucket simply means two epochs
+contribute. The result is exact at every step: the harness's day totals at
+steps 60/300/1800/86400 are identical per scenario and equal the independent
+per-point truth (A1/A3), from raw and rollup tables alike.
+
+Worked example (the research doc's numbers), one series, buckets of 5m, reset
+at 11:38:01, epoch A pre-range, epoch B born 11:38:01:
+
+```
+bucket        maps {epoch: last}         contributions            increase
+11:30–11:35   {A: 90}                    A: rule 4 → 90−84 = 6         6
+11:35–11:40   {A: 107, B: 23}            A: rule 1 → 17; B: rule 3 → 23   40
+11:40–11:45   {B: 37}                    B: rule 1 → 37−23 = 14        14
+one 15m bucket {A: 107, B: 37}           A: 107−84 = 23; B: 37         60 = 6+40+14
+```
+
+### Regime 3 — the overlap (range spans the rollout boundary, or a mixed fleet)
+
+Old buckets are key-0, new buckets carry epochs, and a mid-bucket switch puts
+both in one bucket. The seam is branch 2: when an epoch first appears for a
+series that has key-0 history, its base is the nearest legacy value — the
+same-bucket key-0 value if the switch happened mid-bucket, otherwise the last
+key-0 value from earlier buckets — with the legacy pair rule applied (a drop
+at the seam keeps base 0, so a reset coinciding with the rollout is still
+counted correctly). Continuity: a counter at 100 under key-0 rows that
+continues at 101 under epoch rows contributes 1, not 101.
+
+The reverse seam (epoch → key-0, i.e. a collector rollback) flows through
+branch 7 against the previous bucket's overall max.
+
+Covered in the harness by `s06_transition` (key-0 half-day → epoch half-day →
+a real reset at 18:00), `s15_aggold` (rollup rows from before migration 1012,
+whose empty maps fall back to scalar min/max as key 0), and the mixed-bucket
+path inside s06. Exactness during the overlap is legacy-grade at the seam and
+epoch-grade everywhere else; it converges to regime 2 as old rows age out
+(30d TTL).
+
+One operational caveat (also in the collector doc): a series whose samples
+*alternate* between old and new collector **versions** for an extended period
+(LB across a half-upgraded fleet) can double-count across repeated key-0↔epoch
+seams. Upgrade a series' path atomically where possible and keep the
+version-mix window short. Replica count and load balancing among
+*same-version* collectors are not a concern: the normalizer only ever emits
+validated wire values or 0, so every replica assigns the same epoch to a
+spec-compliant series (see the collector doc's replica analysis).
+
+## Feature flag, caching, alerts
+
+- `use_counter_epochs` (registry default disabled). Routing:
+  temporality Cumulative + rate/increase → epoch pipeline; temporality
+  Multiple → a UNION of the delta fast aggregation and the epoch pipeline
+  (replacing the single multiplexed window expression, which could not be made
+  reset-exact); everything else (delta, gauges, Unspecified, latest/sum/avg/…)
+  is unchanged.
+- The querier's bucket-cache fingerprint gains `counterepochs=v1` for affected
+  specs when the flag is on, so cached legacy buckets and epoch buckets can
+  never stitch into one series. Flipping the flag is a cache-key change for
+  cumulative rate/increase queries only.
+- Threshold alert rules evaluate through the same v5 querier and pick the
+  pipeline up from the flag; PromQL and the v3/v4 builders are non-goals (see
+  below).
+
+## Verification
+
+`tests/integration/testdata/counter_reset_epochs/` — one-day dataset, 15
+scenarios (steady, mid-bucket reset, 11 resets/day, regrow-past-previous,
+pure-legacy, rollout transition, gap, single point, slow reporter, boundary
+reset, duplicates+out-of-order, two-writer overlap, UpDownCounter, pre-
+migration rollup rows, stale markers). Three independent implementations
+(bucket rules, legacy rules, per-point truth) cross-assert in
+`generate_dataset.py`; the builder-emitted SQL (goldens in
+`pkg/telemetrymetrics/testdata/`, emitter `TestEmitHarnessQueries`) is then
+compared row-for-row against them on ClickHouse 26.4 via `run.sh`: 11 query
+cases × pre/post-merge, raw and rollup paths, all passing. The harness runs
+the migration-1012 MV SQL verbatim, so collector-side rollup correctness is
+covered by the same run.
+
+## Performance
+
+Cost model: an L1 scan+aggregate identical in shape to legacy (maps have ~1
+entry per bucket — `1 + resets inside the bucket`; the write-side churn guard
+bounds the degenerate case), then `O(series × buckets)` window-layer work at a
+**~3–4× heavier per-window-row constant** than legacy's single window. The
+`ARRAY JOIN` itself is a flatMap over the already-collapsed set with ~1.0–1.05×
+expansion and is not the cost; micro-variants (tuple-free array join, stripping
+the entire seam machinery) were measured at noise to ~15%. Query cost scales
+with the *queried metric's* rows and series, not total ingest.
+
+Measured, clickhouse local on an M-series laptop (relative numbers are the
+point; `bench.sh` / `bench_large.sh` in the harness):
+
+Small metric — 2,000 series × 24h × 30s (5.76M rows, 576k window rows):
+
+| query | legacy | epochs | ratio |
+|---|---|---|---|
+| increase, step=300, raw | 0.51 s | 0.79 s | 1.5× |
+| increase, step=60, raw | 0.91 s | 2.24 s | 2.5× |
+
+Whale metric — 50,000 series × 24h × 30s (144M rows, 14.4M window rows), the
+realistic worst case for a single query at a ~1B samples/day tenant:
+
+| query | legacy | epochs | ratio |
+|---|---|---|---|
+| increase, step=300, raw | 5.4 s | 20.4 s | 3.8× |
+| increase, step=300, agg_5m | — | 16.6 s | windows dominate; rollups don't rescue this shape |
+| increase, step=1800, agg_30m (2.4M window rows) | — | 2.1 s | linear in window rows |
+| peak RSS (step=300 raw) | 2.7 GB | 4.3 GB | +60% |
+
+Interpretation: typical panels (≤ 5k series × ≤ 400 points ⇒ ≤ 2M window
+rows) pay ~1.5× and sub-second absolutes. Tenant-wide whale panels pay 3–4×
+on a query that is already slow under legacy, and the bucket cache makes the
+steady-state refresh incremental (only new buckets recompute), so the full
+cost is a first-load/backfill event. The flag is per-org, so whale-heavy
+tenants can hold until shard-local pushdown lands (below).
+
+### Where the cost lives (layer profile, whale case)
+
+Progressive-stage timings (`bench_profile.sh`, FORMAT Null, best-of-3; laptop
+numbers are noisy — read the shape, not the decimals):
+
+| stage | time | delta = layer cost |
+|---|---|---|
+| L1 legacy (scan 144M → 14.4M groups, scalar `max`) | ~5.0 s | baseline |
+| L1 epoch (same, `minMap`/`maxMap` states) | ~13–16 s | **map aggregation ≈ +8–11 s — the dominant cost** |
+| + L2 window | ~12 s (noise-overlapped) | |
+| + L3 explode + epoch window | ~18 s | window passes ≈ +5–6 s |
+| full epoch (+L4 regroup + spatial) | ~21 s | ≈ +2 s |
+| full legacy | ~6 s | legacy window ≈ +1 s |
+
+So roughly **half the overhead is the Map aggregate states in L1** (per-row
+1-entry map construction plus per-group Map-state merges — a hash-and-arrays
+operation where legacy does one float compare), ~a third is the two window
+passes over map-carrying rows, and the rest is the extra regroup. Two findings
+kill the obvious rewrites: a two-level L1 (scalar states grouped by
+`(series, bucket, start_ts)`, maps built over the 10×-smaller set) saves only
+~10% — the 144M-row group-by machinery costs what the allocations save — and
+the rollup path at step == bucket width (pure 1:1 map *merges*, zero map
+*construction*) is just as slow, so merge and construction cost about the
+same. The cost is the Map datatype in the aggregation/sort path, not any
+single expression. Hence the levers that actually work are the ones measured
+below (pushdown; rollups when step ≫ bucket; the bucket cache) rather than
+SQL micro-tuning. A per-bucket single-epoch scalar fast path would need
+per-group aggregation-strategy switching that SQL cannot express; if this
+stage ever needs another 2×, that is an engine-level (or two-plan pre-check)
+project.
+
+### Shard-local pushdown (measured)
+
+The sharding key hashes the fingerprint, so shards hold disjoint series — and
+nothing in L1–L4 crosses series until the spatial sum, which decomposes into
+per-shard partials. Simulated with 4 shard databases each holding a disjoint
+quarter of the whale dataset (per-shard wall time measured with the full
+machine, since real shards run in parallel on their own hardware;
+`bench_shard.sh`):
+
+| whale query, step=300, raw | single node (today) | per-shard wall (4 shards) | projected cluster wall |
+|---|---|---|---|
+| epoch pipeline | 20.4 s | 3.3–3.8 s | **~3.6 s** (5.9×) |
+| legacy pipeline | 5.4 s | 1.2 s | **~1.2 s** (4.6×) |
+
+The merged 4-shard partials equal the single-node output **exactly (276 rows,
+0 mismatches)** — empirical proof the decomposition is lossless, not
+approximate. Initiator merge is a GROUP BY over `shards × (buckets × groups)`
+rows (60 ms even in unoptimized python). Speedups are slightly superlinear
+because the per-shard window working set fits cache (the single-node run
+carried a 4.3 GB working set).
+
+Implementation shape: wrap the exact statement the builder emits today —
+with local table names; the time-series join already reads local tables —
+in `cluster('{cluster}', view(...))` and re-aggregate the group columns at
+the initiator. Requirements: the spatial aggregation must decompose
+(sum/min/max do; avg needs sum+count partials; the histogram-quantile path
+already sums per `le` before quantiling at the outermost level). This
+benefits the legacy pipeline equally and is the right lever for whale
+tenants regardless of epochs.
+
+Write side, from the ingest A/B (144M rows through the full MV chain, fresh
+db each):
+
+| | legacy MVs | migration-1012 MVs |
+|---|---|---|
+| ingest wall time | 34.7 s (4.1M rows/s) | 77.1 s (1.9M rows/s) |
+| agg_5m on disk | 66.5 MiB | 111.2 MiB (+67%) |
+| agg_30m on disk | 6.8 MiB | 13.1 MiB (+92%) |
+| samples_v4 on disk | 20.8 MiB | 20.8 MiB (+~0; `start_ts` delta-compresses away) |
+
+The rollup-MV stage of ingestion costs ~2.2× its previous CPU. At 1B
+samples/day (~12k rows/s average), even a single laptop-class node retains
+>100× headroom on this stage, but budget the MV share of ingest CPU
+accordingly on hot clusters. Rollup tables are a small fraction of metrics
+storage (raw dominates on real data), so the +67–92% there nets out to a few
+percent overall.
+
+## Audit notes (what was found and addressed, what remains)
+
+Found during the audit and fixed:
+
+- **Stale markers on the raw path**: legacy `max()` was immune to
+  no-recorded-value rows; `minMap` is not (a 0 would become an epoch's "first"
+  and overstate its first bucket). The epoch L1 filters `bitAnd(flags,1) = 0`
+  on raw/buffer tables, matching the rollup MVs (`s16_stale`).
+- **Stale markers vs the normalizer**: a marker's fake 0 would read as a
+  counter reset and mint a phantom epoch; markers now bypass the normalizer
+  entirely.
+- **Cache stitching**: the bucket-cache fingerprint ignored the pipeline
+  semantics; flag state is now part of the cache identity for affected specs.
+- **Lookback leak**: the epoch pipeline produces real values where legacy
+  produced `nan` (dropped), so the lookback bucket is cut explicitly from the
+  output; the delta side of the Multiple union reads only the display range
+  (this also fixes a pre-existing lookback-row leak in the legacy Multiple
+  path).
+
+Known limitations, by design:
+
+- **Step below the reporting interval**: with a one-step lookback, growth
+  between the last pre-range sample and the first in-range sample is
+  unattributable when the series reports less often than the step (s09 at
+  step=60). Legacy loses the same information (`nan` first point); Prometheus
+  returns nothing at all for such windows. The eventual fix is a
+  reporting-interval-aware lookback (existing TODO in
+  `pkg/querybuilder/time.go`).
+- **Unknown-epoch data stays legacy**: key-0 rows are zoom-inconsistent
+  exactly as today; nothing can recover epochs that were never recorded.
+- **Transition-hour rollup buckets** mixing pre- and post-migration rows use
+  the map rows plus scalar fallback approximately for the pre-migration part
+  (bounded to buckets written while the migration itself was running).
+- **Non-goals**: PromQL path (parity with the Prometheus engine is its
+  contract; the rollup schema was deliberately shaped so a future PromQL
+  rollup effort can derive its own value-pair-rule quantities from the same
+  maps), v3/v4 query builders, meter metrics, exponential histograms
+  (cumulative exp-hist is not ingested today).
+
+## Rollout order
+
+1. Collector schema migration 1012.
+2. Exporters: `enable_start_ts: true`.
+3. Per-org flag `use_counter_epochs` — safe immediately (regime 1 ≡ legacy);
+   full effect accrues as epoch data accumulates and the seam ages out.

pkg/flagger/flaggertest/flaggertest.go

@@ -27,6 +27,24 @@ func New(t *testing.T) flagger.Flagger {
 	return fl
 }
 
+// WithBooleans returns a Flagger with the given boolean flags overriding the
+// registry defaults.
+func WithBooleans(t *testing.T, flags map[string]bool) flagger.Flagger {
+	t.Helper()
+	registry := flagger.MustNewRegistry()
+	cfg := flagger.Config{}
+	cfg.Config.Boolean = flags
+	fl, err := flagger.New(
+		context.Background(),
+		instrumentationtest.New().ToProviderSettings(),
+		cfg,
+		registry,
+		configflagger.NewFactory(registry),
+	)
+	require.NoError(t, err)
+	return fl
+}
+
 // WithUseJSONBody returns a Flagger with use_json_body set to the given value.
 func WithUseJSONBody(t *testing.T, enabled bool) flagger.Flagger {
 	t.Helper()

pkg/flagger/registry.go

@@ -15,6 +15,7 @@ var (
 	FeatureEnableAIObservability  = featuretypes.MustNewName("enable_ai_observability")
 	FeatureEnableMetricsReduction = featuretypes.MustNewName("enable_metrics_reduction")
 	FeatureUseInfraMonitoringV2   = featuretypes.MustNewName("use_infra_monitoring_v2")
+	FeatureUseCounterEpochs       = featuretypes.MustNewName("use_counter_epochs")
 )
 
 func MustNewRegistry() featuretypes.Registry {
@@ -115,6 +116,14 @@ func MustNewRegistry() featuretypes.Registry {
 			DefaultVariant: featuretypes.MustNewName("disabled"),
 			Variants:       featuretypes.NewBooleanVariants(),
 		},
+		&featuretypes.Feature{
+			Name:           FeatureUseCounterEpochs,
+			Kind:           featuretypes.KindBoolean,
+			Stage:          featuretypes.StageExperimental,
+			Description:    "Controls whether cumulative rate/increase use the reset-exact start_ts epoch pipeline (requires collector schema migration 1012)",
+			DefaultVariant: featuretypes.MustNewName("disabled"),
+			Variants:       featuretypes.NewBooleanVariants(),
+		},
 	)
 	if err != nil {
 		panic(err)

pkg/querier/builder_query.go

@@ -15,6 +15,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/telemetrytraces"
 	"github.com/SigNoz/signoz/pkg/types/ctxtypes"
 	"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
+	"github.com/SigNoz/signoz/pkg/types/metrictypes"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
@@ -42,6 +43,23 @@ var _ qbtypes.StatementProvider = (*builderQuery[any])(nil)
 
 type builderConfig struct {
 	logTraceIDWindowPaddingMS uint64
+	// fingerprintSuffix distinguishes cache entries produced by semantically
+	// different pipelines for the same spec (e.g. the counter-epoch rollout)
+	fingerprintSuffix string
+}
+
+// metricSpecUsesCounterEpochs reports whether the epoch pipeline would change
+// this spec's results: cumulative (or mixed) rate/increase aggregations.
+func metricSpecUsesCounterEpochs(spec qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]) bool {
+	for _, agg := range spec.Aggregations {
+		if agg.Temporality == metrictypes.Delta || agg.Temporality == metrictypes.Unspecified {
+			continue
+		}
+		if agg.TimeAggregation == metrictypes.TimeAggregationRate || agg.TimeAggregation == metrictypes.TimeAggregationIncrease {
+			return true
+		}
+	}
+	return false
 }
 
 func newBuilderQuery[T any](
@@ -171,6 +189,10 @@ func (q *builderQuery[T]) Fingerprint() string {
 		parts = append(parts, fmt.Sprintf("shiftby=%d", q.spec.ShiftBy))
 	}
 
+	if q.builderConfig.fingerprintSuffix != "" {
+		parts = append(parts, q.builderConfig.fingerprintSuffix)
+	}
+
 	return strings.Join(parts, "&")
 }
 

pkg/querier/preview.go

@@ -78,7 +78,7 @@ func (q *querier) QueryRangePreview(
 			skip[name] = true
 		}
 	}
-	providers, buildErrs := q.buildPreviewProviders(orgID, req, dependencyQueries, missingMetricQuerySet, skip)
+	providers, buildErrs := q.buildPreviewProviders(ctx, orgID, req, dependencyQueries, missingMetricQuerySet, skip)
 
 	// Render each executing query's statement and collect the ClickHouse-bound
 	// analysis work to run concurrently.
@@ -192,6 +192,7 @@ func missingMetricNames(env qbtypes.QueryEnvelope) []string {
 }
 
 func (q *querier) buildPreviewProviders(
+	ctx context.Context,
 	orgID valuer.UUID,
 	req *qbtypes.QueryRangeRequest,
 	dependencyQueries map[string]bool,
@@ -231,7 +232,7 @@ func (q *querier) buildPreviewProviders(
 			sub.CompositeQuery = qbtypes.CompositeQuery{Queries: []qbtypes.QueryEnvelope{query}}
 		}
 
-		built, _, bErr := q.buildQueries(orgID, &sub, deps, missingMetricQuerySet, event)
+		built, _, bErr := q.buildQueries(ctx, orgID, &sub, deps, missingMetricQuerySet, event)
 		if bErr != nil {
 			errs[name] = bErr
 			continue

pkg/querier/querier.go

@@ -23,6 +23,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/querybuilder"
 	"github.com/SigNoz/signoz/pkg/telemetrystore"
 	"github.com/SigNoz/signoz/pkg/types/ctxtypes"
+	"github.com/SigNoz/signoz/pkg/types/featuretypes"
 	"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
 	"github.com/SigNoz/signoz/pkg/types/metrictypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -132,7 +133,7 @@ func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtype
 		missingMetricQuerySet[name] = true
 	}
 
-	queries, steps, err := q.buildQueries(orgID, req, dependencyQueries, missingMetricQuerySet, event)
+	queries, steps, err := q.buildQueries(ctx, orgID, req, dependencyQueries, missingMetricQuerySet, event)
 	if err != nil {
 		return nil, err
 	}
@@ -176,6 +177,7 @@ func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtype
 }
 
 func (q *querier) buildQueries(
+	ctx context.Context,
 	orgID valuer.UUID,
 	req *qbtypes.QueryRangeRequest,
 	dependencyQueries map[string]bool,
@@ -266,7 +268,16 @@ func (q *querier) buildQueries(
 					event.Source = telemetrytypes.SourceMeter.StringValue()
 					bq = newBuilderQuery(q.logger, q.telemetryStore, orgID, q.meterStmtBuilder, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
 				} else {
-					bq = newBuilderQuery(q.logger, q.telemetryStore, orgID, q.metricStmtBuilder, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
+					// the epoch pipeline changes cumulative rate/increase
+					// results, so its rollout state must be part of the cache
+					// identity: cached legacy buckets and fresh epoch buckets
+					// must never stitch into one series
+					cfg := builderConfig{}
+					if metricSpecUsesCounterEpochs(spec) &&
+						q.fl.BooleanOrEmpty(ctx, flagger.FeatureUseCounterEpochs, featuretypes.NewFlaggerEvaluationContext(orgID)) {
+						cfg.fingerprintSuffix = "counterepochs=v1"
+					}
+					bq = newBuilderQuery(q.logger, q.telemetryStore, orgID, q.metricStmtBuilder, spec, timeRange, req.RequestType, tmplVars, cfg)
 				}
 
 				queries[spec.Name] = bq

pkg/telemetrymetrics/epoch_harness_emit_test.go

@@ -0,0 +1,123 @@
+package telemetrymetrics
+
+// Harness emitter: writes the exact SQL the statement builder produces (args
+// inlined) into the counter-reset verification harness
+// (tests/integration/testdata/counter_reset_epochs/). Skips unless
+// EPOCH_HARNESS_DIR is set; run it after builder changes to refresh the
+// harness queries. The pinned goldens live in epoch_stmt_builder_test.go.
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/flagger"
+	"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
+	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
+	"github.com/SigNoz/signoz/pkg/types/metrictypes"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
+	"github.com/SigNoz/signoz/pkg/valuer"
+)
+
+const harnessDay = uint64(1784332800000) // 2026-07-18T00:00:00Z
+
+func inlineArgs(q string, args []any) string {
+	for _, a := range args {
+		var s string
+		switch v := a.(type) {
+		case string:
+			s = "'" + strings.ReplaceAll(v, "'", "\\'") + "'"
+		default:
+			s = fmt.Sprintf("%v", v)
+		}
+		q = strings.Replace(q, "?", s, 1)
+	}
+	return q
+}
+
+func TestEmitHarnessQueries(t *testing.T) {
+	outDir := os.Getenv("EPOCH_HARNESS_DIR")
+	if outDir == "" {
+		t.Skip("EPOCH_HARNESS_DIR not set")
+	}
+	if err := os.MkdirAll(filepath.Join(outDir, "queries"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+
+	fm := NewFieldMapper()
+	cb := NewConditionBuilder(fm)
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	keys, err := telemetrytypestest.LoadFieldKeysFromJSON("testdata/keys_map.json")
+	if err != nil {
+		t.Fatalf("failed to load field keys: %v", err)
+	}
+	mockMetadataStore.KeysMap = keys
+
+	baseQuery := func(step time.Duration, timeAgg metrictypes.TimeAggregation, hints *metrictypes.MetricTableHints) qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation] {
+		return qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+			Signal:       telemetrytypes.SignalMetrics,
+			StepInterval: qbtypes.Step{Duration: step},
+			Aggregations: []qbtypes.MetricAggregation{{
+				MetricName:       "it_counter_total",
+				Type:             metrictypes.SumType,
+				Temporality:      metrictypes.Cumulative,
+				TimeAggregation:  timeAgg,
+				SpaceAggregation: metrictypes.SpaceAggregationSum,
+				TableHints:       hints,
+			}},
+			GroupBy: []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "scenario"}}},
+		}
+	}
+
+	agg5m := &metrictypes.MetricTableHints{SamplesTableName: SamplesV4Agg5mTableName}
+	agg30m := &metrictypes.MetricTableHints{SamplesTableName: SamplesV4Agg30mTableName}
+
+	cases := []struct {
+		name    string
+		epochs  bool
+		step    time.Duration
+		timeAgg metrictypes.TimeAggregation
+		hints   *metrictypes.MetricTableHints
+	}{
+		{"e_inc_60_raw", true, 60 * time.Second, metrictypes.TimeAggregationIncrease, nil},
+		{"e_inc_300_raw", true, 300 * time.Second, metrictypes.TimeAggregationIncrease, nil},
+		{"e_inc_300_agg5m", true, 300 * time.Second, metrictypes.TimeAggregationIncrease, agg5m},
+		{"e_inc_1800_agg30m", true, 1800 * time.Second, metrictypes.TimeAggregationIncrease, agg30m},
+		{"e_inc_86400_raw", true, 86400 * time.Second, metrictypes.TimeAggregationIncrease, nil},
+		{"e_inc_86400_agg30m", true, 86400 * time.Second, metrictypes.TimeAggregationIncrease, agg30m},
+		{"e_rate_300_raw", true, 300 * time.Second, metrictypes.TimeAggregationRate, nil},
+		{"l_inc_60_raw", false, 60 * time.Second, metrictypes.TimeAggregationIncrease, nil},
+		{"l_inc_300_raw", false, 300 * time.Second, metrictypes.TimeAggregationIncrease, nil},
+		{"l_inc_86400_raw", false, 86400 * time.Second, metrictypes.TimeAggregationIncrease, nil},
+		{"l_rate_300_raw", false, 300 * time.Second, metrictypes.TimeAggregationRate, nil},
+	}
+
+	for _, tc := range cases {
+		fl := flaggertest.WithBooleans(t, map[string]bool{
+			flagger.FeatureUseCounterEpochs.String(): tc.epochs,
+		})
+		b := NewMetricQueryStatementBuilder(
+			instrumentationtest.New().ToProviderSettings(),
+			mockMetadataStore, fm, cb, fl,
+		)
+		stmt, err := b.Build(
+			context.Background(), valuer.UUID{},
+			harnessDay, harnessDay+86400000,
+			qbtypes.RequestTypeTimeSeries, baseQuery(tc.step, tc.timeAgg, tc.hints), nil,
+		)
+		if err != nil {
+			t.Fatalf("%s: %v", tc.name, err)
+		}
+		sql := inlineArgs(stmt.Query, stmt.Args)
+		out := fmt.Sprintf("-- %s (epochs=%v step=%s agg=%s)\nSELECT toUnixTimestamp(ts) AS ts_s, `scenario`, value FROM (\n%s\n) ORDER BY scenario, ts_s FORMAT CSV;\n", tc.name, tc.epochs, tc.step, tc.timeAgg.StringValue(), sql)
+		if err := os.WriteFile(filepath.Join(outDir, "queries", tc.name+".sql"), []byte(out), 0o644); err != nil {
+			t.Fatal(err)
+		}
+	}
+}

pkg/telemetrymetrics/epoch_statement.go

@@ -0,0 +1,306 @@
+package telemetrymetrics
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/SigNoz/signoz/pkg/querybuilder"
+	"github.com/SigNoz/signoz/pkg/types/metrictypes"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/huandu/go-sqlbuilder"
+)
+
+// Epoch-aware temporal aggregation for cumulative counters.
+//
+// A cumulative monotonic series is split by the collector into epochs: runs of
+// samples sharing a normalized start_ts, monotonic non-decreasing within each
+// run. Per (series, step bucket, epoch) the min and max value are therefore
+// the first and last observations, and the increase over any window is the sum
+// of per-epoch growth — exact regardless of how many resets happened inside a
+// bucket, and identical at every step interval.
+//
+// start_ts = 0 is the "unknown epoch": rows written before the collector
+// rollout, sources that never send a start time, and every non-cumulative
+// sample. Key-0 rows flow through the same pipeline with the legacy
+// negative-diff pair rule, so a query over old data reproduces today's
+// results, a query over new data is reset-exact, and a query spanning the
+// rollout boundary seams the two together per row without any watermark
+// configuration.
+//
+// The pipeline has four layers, built innermost first:
+//
+//	L1  per (fingerprint, bucket): __first_by_epoch / __last_by_epoch maps
+//	    (raw tables build them from map(start_ts, value); the 5m/30m rollups
+//	    merge their pre-aggregated map states, falling back to the scalar
+//	    min/max columns as key 0 for rows written before migration 1012)
+//	L2  per-series bucket window: previous bucket ts / overall max, the last
+//	    known key-0 value, and row numbers
+//	L3  ARRAY JOIN over epochs + per-(series, epoch) window: previous value of
+//	    THIS epoch, then the per-epoch contribution (the multiIf below)
+//	L4  regroup to (fingerprint, bucket): increase = sum of finite
+//	    contributions, NaN when none; rate divides by the distance to the
+//	    previous bucket of the series (step for the first)
+//
+// Contribution rules, in branch order:
+//
+//	epoch != 0, seen before        -> last - previous last of this epoch
+//	                                  (negative-diff clamp only as a guard for
+//	                                  invariant violations upstream)
+//	epoch != 0, first time, series
+//	has current/earlier key-0 rows -> pair rule against the nearest key-0
+//	                                  value (the rollout seam: continuation
+//	                                  subtracts it, a drop keeps base 0)
+//	epoch != 0, first time, epoch
+//	born inside the fetched range  -> full value (the counter started at 0
+//	                                  after the range began; this is what
+//	                                  makes a single-shot script visible)
+//	epoch != 0, first time, born
+//	before the range               -> last - first within the bucket (growth
+//	                                  we can actually see; the pre-range part
+//	                                  is unknowable)
+//	epoch  = 0, first bucket       -> NaN (legacy first-point behavior)
+//	epoch  = 0, otherwise          -> legacy pair rule against the previous
+//	                                  bucket's overall max (covers pure legacy
+//	                                  data and the epoch->0 downgrade seam)
+const (
+	// L2: bucket-level series window. %s placeholders: group-by column list
+	// (with trailing comma or empty), inner query.
+	epochBucketWindowTmpl = `SELECT
+  fingerprint,
+  ts,
+  %s__first_by_epoch,
+  __last_by_epoch,
+  mapContains(__last_by_epoch, toInt64(0)) AS __has0,
+  __last_by_epoch[toInt64(0)] AS __v0,
+  anyLastIf(__v0, __has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __prev_v0,
+  max(__has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __ever_had0,
+  arrayMax(mapValues(__last_by_epoch)) AS __bucket_max,
+  lagInFrame(__bucket_max, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_max,
+  lagInFrame(ts, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_ts,
+  row_number() OVER (PARTITION BY fingerprint ORDER BY ts) AS __bucket_rn
+FROM (%s)`
+
+	// L3: explode epochs and compute per-epoch contributions. %s / %d
+	// placeholders: group-by list, L2 query, fetch-range start in ms.
+	epochContribTmpl = `SELECT
+  fingerprint,
+  ts,
+  %s__prev_bucket_ts,
+  __bucket_rn,
+  __kv.1 AS __epoch,
+  __kv.2 AS __lval,
+  __first_by_epoch[__kv.1] AS __fval,
+  lagInFrame(__lval, 1) OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_lval,
+  row_number() OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts) AS __epoch_rn,
+  multiIf(
+    __epoch != 0 AND __epoch_rn > 1, if(__lval >= __prev_lval, __lval - __prev_lval, __lval),
+    __epoch != 0 AND (__has0 OR __ever_had0), __lval - if(__fval >= if(__has0, __v0, __prev_v0), if(__has0, __v0, __prev_v0), 0.),
+    __epoch != 0 AND __epoch >= %d, __lval,
+    __epoch != 0, __lval - __fval,
+    __bucket_rn = 1, nan,
+    __lval < __prev_bucket_max, __lval,
+    __lval - __prev_bucket_max
+  ) AS __contrib
+FROM (%s) ARRAY JOIN arrayZip(mapKeys(__last_by_epoch), mapValues(__last_by_epoch)) AS __kv`
+
+	// L4 for increase: regroup to (series, bucket). Buckets before the display
+	// start exist only to provide bases (the one-step lookback the range was
+	// extended by); they are cut so the output shape matches the legacy path,
+	// which always turned them into NaN. %s/%d: group-by list, L3 query,
+	// display start (s), group-by list for GROUP BY.
+	epochIncreaseTmpl = `SELECT
+  ts,
+  %sif(countIf(isNaN(__contrib) = 0) > 0, sumIf(__contrib, isNaN(__contrib) = 0), nan) AS per_series_value
+FROM (%s)
+WHERE ts >= toDateTime(%d)
+GROUP BY fingerprint, ts%s`
+
+	// L4 for rate: same, divided by the distance to the previous bucket of the
+	// series (falls back to the step for its first bucket). %d: step seconds.
+	epochRateTmpl = `SELECT
+  ts,
+  %sif(
+    countIf(isNaN(__contrib) = 0) > 0,
+    sumIf(__contrib, isNaN(__contrib) = 0) / if(any(__bucket_rn) > 1, ts - any(__prev_bucket_ts), %d),
+    nan
+  ) AS per_series_value
+FROM (%s)
+WHERE ts >= toDateTime(%d)
+GROUP BY fingerprint, ts%s`
+)
+
+// epochMapExpressions returns the L1 select expressions building
+// __first_by_epoch / __last_by_epoch for the given samples table.
+func epochMapExpressions(samplesTable string) []string {
+	if samplesTable == SamplesV4Agg5mTableName || samplesTable == SamplesV4Agg30mTableName {
+		// merge the rollup map states; rows written before migration 1012 have
+		// empty maps, so a bucket made only of such rows falls back to the
+		// scalar min/max columns as the unknown epoch
+		return []string{
+			"minMap(min_value_by_start_ts) AS __first_map_merged",
+			"maxMap(max_value_by_start_ts) AS __last_map_merged",
+			"if(length(__first_map_merged) = 0, map(toInt64(0), min(min)), __first_map_merged) AS __first_by_epoch",
+			"if(length(__last_map_merged) = 0, map(toInt64(0), max(max)), __last_map_merged) AS __last_by_epoch",
+		}
+	}
+	return []string{
+		"minMap(map(start_ts, value)) AS __first_by_epoch",
+		"maxMap(map(start_ts, value)) AS __last_by_epoch",
+	}
+}
+
+// epochGroupByList renders group-by columns as a quoted, comma-terminated
+// prefix ("`a`, `b`, ") for interpolation into the layer templates.
+func epochGroupByList(groupBy []qbtypes.GroupByKey) string {
+	if len(groupBy) == 0 {
+		return ""
+	}
+	var sb strings.Builder
+	for _, g := range groupBy {
+		fmt.Fprintf(&sb, "`%s`, ", g.Name)
+	}
+	return sb.String()
+}
+
+// buildTemporalAggCumulativeEpochs builds the reset-exact temporal aggregation
+// CTE body for cumulative rate/increase. extraSamplesFilter is an optional raw
+// WHERE fragment (used by the mixed-temporality union to split delta rows from
+// the rest); wrapCTE controls whether the result is wrapped as
+// __temporal_aggregation_cte.
+func (b *MetricQueryStatementBuilder) buildTemporalAggCumulativeEpochs(
+	start, end uint64,
+	query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
+	samplesTable string,
+	timeSeriesCTE string,
+	timeSeriesCTEArgs []any,
+	extraSamplesFilter string,
+	wrapCTE bool,
+) (string, []any, error) {
+	stepSec := int64(query.StepInterval.Seconds())
+
+	// L1: per (fingerprint, bucket) epoch maps
+	baseSb := sqlbuilder.NewSelectBuilder()
+	baseSb.Select("fingerprint")
+	baseSb.SelectMore(fmt.Sprintf(
+		"toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(%d)) AS ts",
+		stepSec,
+	))
+	for _, g := range query.GroupBy {
+		baseSb.SelectMore(fmt.Sprintf("`%s`", g.Name))
+	}
+	for _, expr := range epochMapExpressions(samplesTable) {
+		baseSb.SelectMore(expr)
+	}
+	baseSb.From(fmt.Sprintf("%s.%s AS points", DBName, samplesTable))
+	baseSb.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.fingerprint = filtered_time_series.fingerprint")
+	baseSb.Where(
+		baseSb.In("metric_name", query.Aggregations[0].MetricName),
+		baseSb.GTE("unix_milli", start),
+		baseSb.LT("unix_milli", end),
+	)
+	if samplesTable != SamplesV4Agg5mTableName && samplesTable != SamplesV4Agg30mTableName {
+		// the rollup MVs already exclude stale markers; the raw path must too:
+		// a no-recorded-value row would drag an epoch's first observation to 0
+		// and overstate its first bucket (legacy max() was immune, minMap isn't)
+		baseSb.Where("bitAnd(flags, 1) = 0")
+	}
+	if extraSamplesFilter != "" {
+		baseSb.Where(extraSamplesFilter)
+	}
+	baseSb.GroupBy("fingerprint", "ts")
+	baseSb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
+	baseSb.OrderBy("fingerprint", "ts")
+	l1Query, l1Args := baseSb.BuildWithFlavor(sqlbuilder.ClickHouse, timeSeriesCTEArgs...)
+
+	gbList := epochGroupByList(query.GroupBy)
+
+	// L2: bucket-level series window
+	l2Query := fmt.Sprintf(epochBucketWindowTmpl, gbList, l1Query)
+
+	// L3: per-epoch contributions; epochs born at/after the fetched range
+	// start are complete in the data, so base 0 is exact for them
+	l3Query := fmt.Sprintf(epochContribTmpl, gbList, int64(start), l2Query)
+
+	// L4: per-series value per bucket. The fetched range was extended one step
+	// behind the display range for bases (AdjustedMetricTimeRange); those
+	// lookback buckets are cut from the output.
+	displayStartSec := (int64(start) + stepSec*1000) / 1000
+	gbGroupSuffix := ""
+	if len(query.GroupBy) > 0 {
+		gbGroupSuffix = ", " + strings.TrimSuffix(gbList, ", ")
+	}
+	var l4Query string
+	if query.Aggregations[0].TimeAggregation == metrictypes.TimeAggregationRate {
+		l4Query = fmt.Sprintf(epochRateTmpl, gbList, stepSec, l3Query, displayStartSec, gbGroupSuffix)
+	} else {
+		l4Query = fmt.Sprintf(epochIncreaseTmpl, gbList, l3Query, displayStartSec, gbGroupSuffix)
+	}
+
+	if !wrapCTE {
+		return l4Query, l1Args, nil
+	}
+	return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", l4Query), l1Args, nil
+}
+
+// buildTemporalAggMultiTemporalityEpochs handles metrics reporting under both
+// delta and cumulative temporality when the epoch pipeline is enabled: the two
+// populations get their own correct aggregation and are unioned into one
+// temporal CTE. (The legacy path multiplexes them through a single windowed
+// expression instead, which cannot be made reset-exact.)
+func (b *MetricQueryStatementBuilder) buildTemporalAggMultiTemporalityEpochs(
+	start, end uint64,
+	query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
+	samplesTable string,
+	timeSeriesCTE string,
+	timeSeriesCTEArgs []any,
+) (string, []any, error) {
+	stepSec := int64(query.StepInterval.Seconds())
+
+	// delta side: per-bucket sum, no windowing needed
+	deltaSb := sqlbuilder.NewSelectBuilder()
+	deltaSb.Select(fmt.Sprintf(
+		"toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(%d)) AS ts",
+		stepSec,
+	))
+	for _, g := range query.GroupBy {
+		deltaSb.SelectMore(fmt.Sprintf("`%s`", g.Name))
+	}
+	deltaAggCol, err := AggregationColumnForSamplesTable(samplesTable, metrictypes.Delta, query.Aggregations[0].TimeAggregation)
+	if err != nil {
+		return "", nil, err
+	}
+	if query.Aggregations[0].TimeAggregation == metrictypes.TimeAggregationRate {
+		deltaAggCol = fmt.Sprintf("%s/%d", deltaAggCol, stepSec)
+	}
+	deltaSb.SelectMore(fmt.Sprintf("%s AS per_series_value", deltaAggCol))
+	deltaSb.From(fmt.Sprintf("%s.%s AS points", DBName, samplesTable))
+	deltaSb.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.fingerprint = filtered_time_series.fingerprint")
+	// the range was extended one step behind the display start for cumulative
+	// bases; delta needs no base, so it reads only the display range
+	deltaSb.Where(
+		deltaSb.In("metric_name", query.Aggregations[0].MetricName),
+		deltaSb.GTE("unix_milli", uint64(int64(start)+stepSec*1000)),
+		deltaSb.LT("unix_milli", end),
+		"LOWER(temporality) LIKE LOWER('Delta')",
+	)
+	deltaSb.GroupBy("fingerprint", "ts")
+	deltaSb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
+	deltaQuery, deltaArgs := deltaSb.BuildWithFlavor(sqlbuilder.ClickHouse, timeSeriesCTEArgs...)
+
+	// cumulative (and unspecified) side: the epoch pipeline
+	cumulativeQuery, cumulativeArgs, err := b.buildTemporalAggCumulativeEpochs(
+		start, end, query, samplesTable, timeSeriesCTE, timeSeriesCTEArgs,
+		"NOT (LOWER(temporality) LIKE LOWER('Delta'))",
+		false,
+	)
+	if err != nil {
+		return "", nil, err
+	}
+
+	frag := fmt.Sprintf(
+		"__temporal_aggregation_cte AS (SELECT * FROM (%s) UNION ALL SELECT * FROM (%s))",
+		deltaQuery, cumulativeQuery,
+	)
+	args := append(append([]any{}, deltaArgs...), cumulativeArgs...)
+	return frag, args, nil
+}

pkg/telemetrymetrics/epoch_stmt_builder_test.go

@@ -0,0 +1,134 @@
+package telemetrymetrics
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/flagger"
+	"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
+	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
+	"github.com/SigNoz/signoz/pkg/types/metrictypes"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
+	"github.com/SigNoz/signoz/pkg/valuer"
+	"github.com/stretchr/testify/require"
+)
+
+// TestStatementBuilderCounterEpochs pins the SQL of the epoch-aware cumulative
+// pipeline (use_counter_epochs). The golden files in testdata/ are the exact
+// statements validated row-for-row against the reference implementations and a
+// real ClickHouse in the counter-reset harness (see the design doc
+// docs/counter-reset-epochs.md); regenerate them deliberately, not to make a
+// failing test pass.
+func TestStatementBuilderCounterEpochs(t *testing.T) {
+	cases := []struct {
+		name       string
+		goldenFile string
+		start, end uint64
+		query      qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]
+		args       []any
+	}{
+		{
+			name:       "cumulative_rate_raw",
+			goldenFile: "epoch_cumulative_rate_raw.sql",
+			start:      1747947419000,
+			end:        1747983448000,
+			query: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+				Signal:       telemetrytypes.SignalMetrics,
+				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
+				Aggregations: []qbtypes.MetricAggregation{{
+					MetricName:       "signoz_calls_total",
+					Type:             metrictypes.SumType,
+					Temporality:      metrictypes.Cumulative,
+					TimeAggregation:  metrictypes.TimeAggregationRate,
+					SpaceAggregation: metrictypes.SpaceAggregationSum,
+				}},
+				Filter:  &qbtypes.Filter{Expression: "service.name = 'cartservice'"},
+				GroupBy: []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "service.name"}}},
+			},
+			args: []any{"signoz_calls_total", uint64(1747936800000), uint64(1747983420000), "cumulative", "cartservice", "signoz_calls_total", uint64(1747947360000), uint64(1747983420000), 0},
+		},
+		{
+			name:       "cumulative_increase_agg5m",
+			goldenFile: "epoch_cumulative_increase_agg5m.sql",
+			start:      1747800000000,
+			end:        1747983448000,
+			query: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+				Signal:       telemetrytypes.SignalMetrics,
+				StepInterval: qbtypes.Step{Duration: 300 * time.Second},
+				Aggregations: []qbtypes.MetricAggregation{{
+					MetricName:       "signoz_calls_total",
+					Type:             metrictypes.SumType,
+					Temporality:      metrictypes.Cumulative,
+					TimeAggregation:  metrictypes.TimeAggregationIncrease,
+					SpaceAggregation: metrictypes.SpaceAggregationSum,
+				}},
+			},
+			args: []any{"signoz_calls_total", uint64(1747785600000), uint64(1747983420000), "cumulative", "signoz_calls_total", uint64(1747799700000), uint64(1747983420000), 0},
+		},
+		{
+			name:       "multi_temporality_rate_raw",
+			goldenFile: "epoch_multi_temporality_rate_raw.sql",
+			start:      1747947419000,
+			end:        1747983448000,
+			query: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+				Signal:       telemetrytypes.SignalMetrics,
+				StepInterval: qbtypes.Step{Duration: 60 * time.Second},
+				Aggregations: []qbtypes.MetricAggregation{{
+					MetricName:       "signoz_calls_total",
+					Type:             metrictypes.SumType,
+					Temporality:      metrictypes.Multiple,
+					TimeAggregation:  metrictypes.TimeAggregationRate,
+					SpaceAggregation: metrictypes.SpaceAggregationSum,
+				}},
+			},
+			// delta side reads only the display range (no base needed); the
+			// cumulative side keeps the one-step lookback for bases
+			args: []any{
+				"signoz_calls_total", uint64(1747936800000), uint64(1747983420000),
+				"signoz_calls_total", uint64(1747947360000), uint64(1747983420000),
+				"signoz_calls_total", uint64(1747936800000), uint64(1747983420000),
+				"signoz_calls_total", uint64(1747947300000), uint64(1747983420000),
+				0,
+			},
+		},
+	}
+
+	fm := NewFieldMapper()
+	cb := NewConditionBuilder(fm)
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	keys, err := telemetrytypestest.LoadFieldKeysFromJSON("testdata/keys_map.json")
+	require.NoError(t, err)
+	mockMetadataStore.KeysMap = keys
+
+	fl := flaggertest.WithBooleans(t, map[string]bool{
+		flagger.FeatureUseCounterEpochs.String(): true,
+	})
+	statementBuilder := NewMetricQueryStatementBuilder(
+		instrumentationtest.New().ToProviderSettings(),
+		mockMetadataStore,
+		fm,
+		cb,
+		fl,
+	)
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			q, err := statementBuilder.Build(context.Background(), valuer.UUID{}, c.start, c.end, qbtypes.RequestTypeTimeSeries, c.query, nil)
+			require.NoError(t, err)
+
+			goldenPath := filepath.Join("testdata", c.goldenFile)
+			if os.Getenv("UPDATE_GOLDEN") != "" {
+				require.NoError(t, os.WriteFile(goldenPath, []byte(q.Query+"\n"), 0o644))
+			}
+			golden, err := os.ReadFile(goldenPath)
+			require.NoError(t, err)
+			require.Equal(t, string(golden), q.Query+"\n")
+			require.Equal(t, c.args, q.Args)
+		})
+	}
+}

pkg/telemetrymetrics/statement_builder.go

@@ -9,6 +9,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/factory"
 	"github.com/SigNoz/signoz/pkg/flagger"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
+	"github.com/SigNoz/signoz/pkg/types/featuretypes"
 	"github.com/SigNoz/signoz/pkg/types/metrictypes"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -202,6 +203,10 @@ func (b *MetricQueryStatementBuilder) buildPipelineStatement(
 		return nil, err
 	}
 
+	// the reset-exact epoch pipeline needs the start_ts column and epoch map
+	// rollup columns from collector schema migration 1012
+	epochsEnabled := b.flagger.BooleanOrEmpty(ctx, flagger.FeatureUseCounterEpochs, featuretypes.NewFlaggerEvaluationContext(orgID))
+
 	if qbtypes.CanShortCircuitDelta(query.Aggregations[0]) {
 		// spatial_aggregation_cte directly for certain delta queries
 		if frag, args, err := b.buildTemporalAggDeltaFastPath(start, end, query, samplesTable, timeSeriesCTE, timeSeriesCTEArgs); err != nil {
@@ -212,7 +217,7 @@ func (b *MetricQueryStatementBuilder) buildPipelineStatement(
 		}
 	} else {
 		// temporal_aggregation_cte
-		if frag, args, err := b.buildTemporalAggregationCTE(ctx, start, end, query, keys, samplesTable, timeSeriesCTE, timeSeriesCTEArgs); err != nil {
+		if frag, args, err := b.buildTemporalAggregationCTE(ctx, start, end, query, keys, samplesTable, timeSeriesCTE, timeSeriesCTEArgs, epochsEnabled); err != nil {
 			return nil, err
 		} else if frag != "" {
 			cteFragments = append(cteFragments, frag)
@@ -575,10 +580,26 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggregationCTE(
 	samplesTable string,
 	timeSeriesCTE string,
 	timeSeriesCTEArgs []any,
+	epochsEnabled bool,
 ) (string, []any, error) {
-	if query.Aggregations[0].Temporality == metrictypes.Delta {
+	agg := query.Aggregations[0]
+	if agg.Temporality == metrictypes.Delta {
 		return b.buildTemporalAggDelta(ctx, start, end, query, samplesTable, timeSeriesCTE, timeSeriesCTEArgs)
-	} else if query.Aggregations[0].Temporality != metrictypes.Multiple {
+	}
+
+	// reset-exact pipeline for cumulative counters; Unspecified temporality
+	// has no counter semantics and stays on the legacy path
+	isRateOrIncrease := agg.TimeAggregation == metrictypes.TimeAggregationRate || agg.TimeAggregation == metrictypes.TimeAggregationIncrease
+	if epochsEnabled && isRateOrIncrease {
+		if agg.Temporality == metrictypes.Cumulative {
+			return b.buildTemporalAggCumulativeEpochs(start, end, query, samplesTable, timeSeriesCTE, timeSeriesCTEArgs, "", true)
+		}
+		if agg.Temporality == metrictypes.Multiple {
+			return b.buildTemporalAggMultiTemporalityEpochs(start, end, query, samplesTable, timeSeriesCTE, timeSeriesCTEArgs)
+		}
+	}
+
+	if agg.Temporality != metrictypes.Multiple {
 		return b.buildTemporalAggCumulativeOrUnspecified(ctx, start, end, query, samplesTable, timeSeriesCTE, timeSeriesCTEArgs)
 	}
 	return b.buildTemporalAggForMultipleTemporalities(ctx, start, end, query, samplesTable, timeSeriesCTE, timeSeriesCTEArgs)

pkg/telemetrymetrics/testdata/epoch_cumulative_increase_agg5m.sql

@@ -0,0 +1,38 @@
+WITH __temporal_aggregation_cte AS (SELECT
+  ts,
+  if(countIf(isNaN(__contrib) = 0) > 0, sumIf(__contrib, isNaN(__contrib) = 0), nan) AS per_series_value
+FROM (SELECT
+  fingerprint,
+  ts,
+  __prev_bucket_ts,
+  __bucket_rn,
+  __kv.1 AS __epoch,
+  __kv.2 AS __lval,
+  __first_by_epoch[__kv.1] AS __fval,
+  lagInFrame(__lval, 1) OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_lval,
+  row_number() OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts) AS __epoch_rn,
+  multiIf(
+    __epoch != 0 AND __epoch_rn > 1, if(__lval >= __prev_lval, __lval - __prev_lval, __lval),
+    __epoch != 0 AND (__has0 OR __ever_had0), __lval - if(__fval >= if(__has0, __v0, __prev_v0), if(__has0, __v0, __prev_v0), 0.),
+    __epoch != 0 AND __epoch >= 1747799700000, __lval,
+    __epoch != 0, __lval - __fval,
+    __bucket_rn = 1, nan,
+    __lval < __prev_bucket_max, __lval,
+    __lval - __prev_bucket_max
+  ) AS __contrib
+FROM (SELECT
+  fingerprint,
+  ts,
+  __first_by_epoch,
+  __last_by_epoch,
+  mapContains(__last_by_epoch, toInt64(0)) AS __has0,
+  __last_by_epoch[toInt64(0)] AS __v0,
+  anyLastIf(__v0, __has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __prev_v0,
+  max(__has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __ever_had0,
+  arrayMax(mapValues(__last_by_epoch)) AS __bucket_max,
+  lagInFrame(__bucket_max, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_max,
+  lagInFrame(ts, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_ts,
+  row_number() OVER (PARTITION BY fingerprint ORDER BY ts) AS __bucket_rn
+FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, minMap(min_value_by_start_ts) AS __first_map_merged, maxMap(max_value_by_start_ts) AS __last_map_merged, if(length(__first_map_merged) = 0, map(toInt64(0), min(min)), __first_map_merged) AS __first_by_epoch, if(length(__last_map_merged) = 0, map(toInt64(0), max(max)), __last_map_merged) AS __last_by_epoch FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts)) ARRAY JOIN arrayZip(mapKeys(__last_by_epoch), mapValues(__last_by_epoch)) AS __kv)
+WHERE ts >= toDateTime(1747800000)
+GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts

pkg/telemetrymetrics/testdata/epoch_cumulative_rate_raw.sql

@@ -0,0 +1,42 @@
+WITH __temporal_aggregation_cte AS (SELECT
+  ts,
+  `service.name`, if(
+    countIf(isNaN(__contrib) = 0) > 0,
+    sumIf(__contrib, isNaN(__contrib) = 0) / if(any(__bucket_rn) > 1, ts - any(__prev_bucket_ts), 30),
+    nan
+  ) AS per_series_value
+FROM (SELECT
+  fingerprint,
+  ts,
+  `service.name`, __prev_bucket_ts,
+  __bucket_rn,
+  __kv.1 AS __epoch,
+  __kv.2 AS __lval,
+  __first_by_epoch[__kv.1] AS __fval,
+  lagInFrame(__lval, 1) OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_lval,
+  row_number() OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts) AS __epoch_rn,
+  multiIf(
+    __epoch != 0 AND __epoch_rn > 1, if(__lval >= __prev_lval, __lval - __prev_lval, __lval),
+    __epoch != 0 AND (__has0 OR __ever_had0), __lval - if(__fval >= if(__has0, __v0, __prev_v0), if(__has0, __v0, __prev_v0), 0.),
+    __epoch != 0 AND __epoch >= 1747947360000, __lval,
+    __epoch != 0, __lval - __fval,
+    __bucket_rn = 1, nan,
+    __lval < __prev_bucket_max, __lval,
+    __lval - __prev_bucket_max
+  ) AS __contrib
+FROM (SELECT
+  fingerprint,
+  ts,
+  `service.name`, __first_by_epoch,
+  __last_by_epoch,
+  mapContains(__last_by_epoch, toInt64(0)) AS __has0,
+  __last_by_epoch[toInt64(0)] AS __v0,
+  anyLastIf(__v0, __has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __prev_v0,
+  max(__has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __ever_had0,
+  arrayMax(mapValues(__last_by_epoch)) AS __bucket_max,
+  lagInFrame(__bucket_max, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_max,
+  lagInFrame(ts, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_ts,
+  row_number() OVER (PARTITION BY fingerprint ORDER BY ts) AS __bucket_rn
+FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `service.name`, minMap(map(start_ts, value)) AS __first_by_epoch, maxMap(map(start_ts, value)) AS __last_by_epoch FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `service.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND JSONExtractString(labels, 'service.name') = ? GROUP BY fingerprint, `service.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? AND bitAnd(flags, 1) = 0 GROUP BY fingerprint, ts, `service.name` ORDER BY fingerprint, ts)) ARRAY JOIN arrayZip(mapKeys(__last_by_epoch), mapValues(__last_by_epoch)) AS __kv)
+WHERE ts >= toDateTime(1747947390)
+GROUP BY fingerprint, ts, `service.name`), __spatial_aggregation_cte AS (SELECT ts, `service.name`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `service.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `service.name`, ts

pkg/telemetrymetrics/testdata/epoch_multi_temporality_rate_raw.sql

@@ -0,0 +1,42 @@
+WITH __temporal_aggregation_cte AS (SELECT * FROM (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(60)) AS ts, sum(value)/60 AS per_series_value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? AND LOWER(temporality) LIKE LOWER('Delta') GROUP BY fingerprint, ts) UNION ALL SELECT * FROM (SELECT
+  ts,
+  if(
+    countIf(isNaN(__contrib) = 0) > 0,
+    sumIf(__contrib, isNaN(__contrib) = 0) / if(any(__bucket_rn) > 1, ts - any(__prev_bucket_ts), 60),
+    nan
+  ) AS per_series_value
+FROM (SELECT
+  fingerprint,
+  ts,
+  __prev_bucket_ts,
+  __bucket_rn,
+  __kv.1 AS __epoch,
+  __kv.2 AS __lval,
+  __first_by_epoch[__kv.1] AS __fval,
+  lagInFrame(__lval, 1) OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_lval,
+  row_number() OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts) AS __epoch_rn,
+  multiIf(
+    __epoch != 0 AND __epoch_rn > 1, if(__lval >= __prev_lval, __lval - __prev_lval, __lval),
+    __epoch != 0 AND (__has0 OR __ever_had0), __lval - if(__fval >= if(__has0, __v0, __prev_v0), if(__has0, __v0, __prev_v0), 0.),
+    __epoch != 0 AND __epoch >= 1747947300000, __lval,
+    __epoch != 0, __lval - __fval,
+    __bucket_rn = 1, nan,
+    __lval < __prev_bucket_max, __lval,
+    __lval - __prev_bucket_max
+  ) AS __contrib
+FROM (SELECT
+  fingerprint,
+  ts,
+  __first_by_epoch,
+  __last_by_epoch,
+  mapContains(__last_by_epoch, toInt64(0)) AS __has0,
+  __last_by_epoch[toInt64(0)] AS __v0,
+  anyLastIf(__v0, __has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __prev_v0,
+  max(__has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __ever_had0,
+  arrayMax(mapValues(__last_by_epoch)) AS __bucket_max,
+  lagInFrame(__bucket_max, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_max,
+  lagInFrame(ts, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_ts,
+  row_number() OVER (PARTITION BY fingerprint ORDER BY ts) AS __bucket_rn
+FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(60)) AS ts, minMap(map(start_ts, value)) AS __first_by_epoch, maxMap(map(start_ts, value)) AS __last_by_epoch FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? AND bitAnd(flags, 1) = 0 AND NOT (LOWER(temporality) LIKE LOWER('Delta')) GROUP BY fingerprint, ts ORDER BY fingerprint, ts)) ARRAY JOIN arrayZip(mapKeys(__last_by_epoch), mapValues(__last_by_epoch)) AS __kv)
+WHERE ts >= toDateTime(1747947360)
+GROUP BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts

tests/integration/testdata/counter_reset_epochs/README.md

@@ -0,0 +1,61 @@
+# Counter-reset epochs: one-day verification dataset
+
+Self-contained correctness harness for the epoch-aware cumulative rate/increase
+pipeline (`use_counter_epochs` feature flag + collector schema migration 1012).
+Runs against `clickhouse local` — no cluster needed. See
+`docs/counter-reset-epochs.md` for the design.
+
+## What it proves
+
+- The **builder-emitted SQL** (both the epoch pipeline and the legacy pipeline)
+  reproduces its reference implementation **row-for-row** on a one-day dataset,
+  from the raw table and from the 5m/30m rollups, before and after part merges.
+- The **migration-1012 materialized views** (epoch map columns) are exercised
+  verbatim: raw inserts → 5m MV → chained 30m MV.
+- **Rollup fidelity**: the rollup path equals the raw path at every step.
+- **Replay safety**: shuffled insert order and a duplicated insert batch change
+  nothing.
+- The reference implementations are cross-checked in `generate_dataset.py`
+  (assertions A1–A6, written to `report.txt`):
+  - A1  bucket rules == independent per-point truth, per bucket and in total
+  - A3  zoom consistency: identical day totals at 60s/300s/1800s/86400s steps
+  - A4  the legacy pipeline's zoom **in**consistency, reproduced (the bug)
+  - A5  on epoch-less data the epoch rules reproduce legacy exactly
+  - A6  on reset-free data the two pipelines agree exactly
+
+## Scenarios (one series each, label `scenario`)
+
+| scenario | what it covers |
+|---|---|
+| s01_steady | no resets — must equal legacy exactly |
+| s02_midreset | one reset inside a 5m bucket |
+| s03_multireset | resets every 2h; legacy day total 242 vs true 3969 |
+| s04_regrow | reset that regrows past the previous value — invisible to value-based detection |
+| s05_legacy | `start_ts = 0` everywhere (pre-rollout data) — must equal legacy exactly |
+| s06_transition | key-0 first half of day, epochs second half, then a real reset (rollout seam) |
+| s07_gap | idle series with a 10-minute gap (false-spike regression) |
+| s08_singlepoint | one sample ever, epoch in range — visible with epochs, absent in legacy |
+| s09_slow | 300s reporting interval (step < interval limitation documented) |
+| s10_boundaryreset | reset exactly on a bucket boundary |
+| s11_dupooo | duplicated + out-of-order ingestion |
+| s12_twowriter | two epochs overlapping in time on one fingerprint (broken series identity) |
+| s14_updown | non-monotonic cumulative, key-0 — must equal legacy exactly |
+| s15_aggold | pre-migration rollup rows (empty epoch maps) — key-0 fallback in agg tables |
+| s16_stale | NoRecordedValue markers (flags bit 0) — must be invisible |
+
+## Running
+
+```bash
+python3 generate_dataset.py          # regenerates inserts.sql, expected/, report.txt
+./run.sh                             # schema + inserts + queries + comparisons
+```
+
+`queries/*.sql` are the exact statements produced by
+`MetricQueryStatementBuilder` (args inlined). Regenerate them after builder
+changes with:
+
+```bash
+EPOCH_HARNESS_DIR=$PWD go test ./pkg/telemetrymetrics/ -run TestEmitHarnessQueries
+```
+
+(from the repo root, pointing the env var here).

tests/integration/testdata/counter_reset_epochs/bench.sh

@@ -0,0 +1,66 @@
+#!/bin/bash
+# Relative cost of the epoch pipeline vs legacy at scale.
+# 2000 cumulative series, 24h at 30s interval = 5.76M samples, each series
+# resetting every ~6h (4 epochs/day). Uses the same db as the harness but a
+# separate metric name so correctness runs stay isolated.
+set -eu
+cd "$(dirname "$0")"
+CH="clickhouse local --path chdata"
+
+B=1784332800000
+
+echo "== generating bench data in-db (5.76M samples) =="
+$CH -q "
+INSERT INTO signoz_metrics.samples_v4
+SELECT
+    'default' AS env,
+    'Cumulative' AS temporality,
+    'bench_counter' AS metric_name,
+    10000 + intDiv(number, 2880) AS fingerprint,
+    $B - 3600000 + (number % 2880) * 30000 AS unix_milli,
+    -- value grows 2/point within each 6h epoch, resets to 0 at epoch change
+    2.0 * ((number % 2880) % 720) AS value,
+    0 AS flags,
+    $B AS inserted_at_unix_milli,
+    -- epoch = start of the series' current 6h segment
+    $B - 3600000 + intDiv((number % 2880), 720) * 720 * 30000 AS start_ts
+FROM numbers(2000 * 2880);
+
+INSERT INTO signoz_metrics.time_series_v4_1day
+SELECT 'default', 'Cumulative', 'bench_counter', 10000 + number,
+       ts, '{\"scenario\":\"bench\"}'
+FROM numbers(2000) ARRAY JOIN [toInt64($B - 86400000), toInt64($B)] AS ts;
+"
+
+$CH -q "SELECT 'bench samples', count() FROM signoz_metrics.samples_v4 WHERE metric_name = 'bench_counter'"
+
+# adapt the emitted harness queries to the bench metric
+mkdir -p bench_queries
+for f in e_inc_300_raw l_inc_300_raw e_inc_60_raw l_inc_60_raw; do
+  sed "s/it_counter_total/bench_counter/g" "queries/$f.sql" > "bench_queries/$f.sql"
+done
+
+echo "== timing (3 runs each, seconds) =="
+for f in l_inc_60_raw e_inc_60_raw l_inc_300_raw e_inc_300_raw; do
+  printf "%-16s" "$f"
+  for i in 1 2 3; do
+    t0=$(python3 -c 'import time; print(time.time())')
+    $CH --queries-file "bench_queries/$f.sql" > /dev/null
+    t1=$(python3 -c 'import time; print(time.time())')
+    printf " %.2f" "$(python3 -c "print($t1 - $t0)")"
+  done
+  echo
+done
+
+echo "== row counts through the pipeline (step=300) =="
+$CH -q "
+SELECT 'raw rows scanned' AS stage, count() AS rows FROM signoz_metrics.samples_v4 WHERE metric_name = 'bench_counter'
+UNION ALL
+SELECT 'L1 series-bucket groups', count() FROM (
+  SELECT fingerprint, intDiv(unix_milli, 300000) FROM signoz_metrics.samples_v4
+  WHERE metric_name = 'bench_counter' GROUP BY 1, 2)
+UNION ALL
+SELECT 'L3 rows after ARRAY JOIN', sum(n) FROM (
+  SELECT length(maxMap(map(start_ts, value))) AS n FROM signoz_metrics.samples_v4
+  WHERE metric_name = 'bench_counter' GROUP BY fingerprint, intDiv(unix_milli, 300000))
+ORDER BY stage"

tests/integration/testdata/counter_reset_epochs/bench_large.sh

@@ -0,0 +1,92 @@
+#!/bin/bash
+# Large-scale benchmark: one big metric at a ~1B samples/day tenant.
+#   50,000 series x 24h x 30s = 144M raw samples, a reset every 6h per series
+#   (mid-bucket, so buckets at reset boundaries genuinely carry 2 epochs).
+# Measures:
+#   1. ingest throughput WITH migration-1012 MVs vs the legacy MVs (write cost)
+#   2. query latency: legacy vs epoch pipeline, raw and rollup paths
+#   3. peak RSS of the query process for the step=300 raw case
+set -eu
+cd "$(dirname "$0")"
+
+B=1784332800000
+N=$((50000 * 2880))
+
+gen_insert() {
+  local db_dir="$1"
+  clickhouse local --path "$db_dir" -q "
+INSERT INTO signoz_metrics.samples_v4
+SELECT
+    'default', 'Cumulative', 'bench2_counter',
+    200000 + intDiv(number, 2880) AS fingerprint,
+    $B - 3600000 + (number % 2880) * 30000 AS unix_milli,
+    2.0 * ((number % 2880) - segfirst) AS value,
+    0 AS flags,
+    $B AS inserted_at_unix_milli,
+    if(seg = 0, $B - 90000000, $B - 3600000 + segfirst * 30000 - 15000) AS start_ts
+FROM (
+  SELECT number,
+    if(number % 2880 < 3, 0, intDiv(number % 2880 - 3, 720) + 1) AS seg,
+    if(seg = 0, 0, 3 + (seg - 1) * 720) AS segfirst
+  FROM numbers($N)
+)
+SETTINGS max_insert_threads = 4, max_partitions_per_insert_block = 100;
+
+INSERT INTO signoz_metrics.time_series_v4_1day
+SELECT 'default', 'Cumulative', 'bench2_counter', 200000 + number,
+       ts, '{\"scenario\":\"bench2\"}'
+FROM numbers(50000) ARRAY JOIN [toInt64($B - 86400000), toInt64($B)] AS ts;
+"
+}
+
+echo "== A/B ingest: legacy MVs (fresh db) =="
+rm -rf chdata_legacy
+clickhouse local --path chdata_legacy --queries-file schema_legacy.sql
+t0=$(python3 -c 'import time; print(time.time())')
+gen_insert chdata_legacy
+t1=$(python3 -c 'import time; print(time.time())')
+LEG_INGEST=$(python3 -c "print(f'{$t1-$t0:.1f}')")
+echo "legacy-MV ingest of ${N} rows: ${LEG_INGEST}s"
+
+echo "== A/B ingest: migration-1012 MVs (fresh db) =="
+rm -rf chdata_large
+clickhouse local --path chdata_large --queries-file schema.sql
+t0=$(python3 -c 'import time; print(time.time())')
+gen_insert chdata_large
+t1=$(python3 -c 'import time; print(time.time())')
+NEW_INGEST=$(python3 -c "print(f'{$t1-$t0:.1f}')")
+echo "epoch-MV ingest of ${N} rows: ${NEW_INGEST}s"
+
+echo "== storage =="
+for d in chdata_legacy chdata_large; do
+  clickhouse local --path $d -q "
+    SELECT '$d', table, formatReadableSize(sum(bytes_on_disk)) AS size, sum(rows) AS rows
+    FROM system.parts WHERE database = 'signoz_metrics' AND active AND table LIKE 'samples%'
+    GROUP BY table ORDER BY table"
+done
+
+echo "== queries (large db, 2 runs each, seconds) =="
+mkdir -p bench_queries
+for f in e_inc_300_raw l_inc_300_raw e_inc_300_agg5m e_inc_1800_agg30m; do
+  sed "s/it_counter_total/bench2_counter/g" "queries/$f.sql" > "bench_queries/big_$f.sql"
+done
+# legacy comparison for the rollup path: reuse the epoch agg query's table by
+# swapping metric in the legacy raw query is not enough — build legacy agg via
+# the l_inc_300 shape but agg table (max(max)); simplest honest proxy: time the
+# epoch agg query and the legacy raw query; legacy-agg sits between them.
+for f in big_l_inc_300_raw big_e_inc_300_raw big_e_inc_300_agg5m big_e_inc_1800_agg30m; do
+  printf "%-26s" "$f"
+  for i in 1 2; do
+    t0=$(python3 -c 'import time; print(time.time())')
+    clickhouse local --path chdata_large --queries-file "bench_queries/$f.sql" > /dev/null
+    t1=$(python3 -c 'import time; print(time.time())')
+    printf " %.2f" "$(python3 -c "print($t1 - $t0)")"
+  done
+  echo
+done
+
+echo "== peak RSS (step=300 raw) =="
+for f in big_l_inc_300_raw big_e_inc_300_raw; do
+  rss=$( { /usr/bin/time -l clickhouse local --path chdata_large --queries-file "bench_queries/$f.sql" > /dev/null; } 2>&1 | grep "maximum resident" | awk '{print $1}')
+  echo "$f peak_rss=$(python3 -c "print(f'{$rss/1e9:.2f} GB')")"
+done

tests/integration/testdata/counter_reset_epochs/bench_profile.sh

@@ -0,0 +1,64 @@
+#!/bin/bash
+# Layer-by-layer cost decomposition of the epoch pipeline on the whale dataset
+# (50k series, 144M rows, step=300, 14.4M series-buckets). All queries FORMAT
+# Null (compute only). Deltas between successive stages = per-layer cost.
+set -eu
+cd "$(dirname "$0")"
+CH="clickhouse local --path chdata_large"
+
+TS_JOIN="INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'scenario') AS scenario FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN ('bench2_counter') AND unix_milli >= 1784246400000 AND unix_milli <= 1784419200000 AND LOWER(temporality) LIKE LOWER('cumulative') GROUP BY fingerprint, scenario) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint"
+WHERE="WHERE metric_name IN ('bench2_counter') AND unix_milli >= 1784332500000 AND unix_milli < 1784419200000 AND bitAnd(flags, 1) = 0"
+BUCKET="toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts"
+
+L1_LEGACY="SELECT fingerprint, $BUCKET, scenario, max(value) AS per_series_value FROM signoz_metrics.samples_v4 AS points $TS_JOIN $WHERE GROUP BY fingerprint, ts, scenario ORDER BY fingerprint, ts"
+
+L1_EPOCH="SELECT fingerprint, $BUCKET, scenario, minMap(map(start_ts, value)) AS __first_by_epoch, maxMap(map(start_ts, value)) AS __last_by_epoch FROM signoz_metrics.samples_v4 AS points $TS_JOIN $WHERE GROUP BY fingerprint, ts, scenario ORDER BY fingerprint, ts"
+
+# two-level alternative: scalar states grouped by (series, bucket, epoch),
+# maps built over the 10x smaller collapsed set
+L1_TWOLEVEL="SELECT fingerprint, ts, scenario, minMap(map(start_ts, mn)) AS __first_by_epoch, maxMap(map(start_ts, mx)) AS __last_by_epoch FROM (SELECT fingerprint, $BUCKET, scenario, start_ts, min(value) AS mn, max(value) AS mx FROM signoz_metrics.samples_v4 AS points $TS_JOIN $WHERE GROUP BY fingerprint, ts, scenario, start_ts) GROUP BY fingerprint, ts, scenario ORDER BY fingerprint, ts"
+
+L2="SELECT fingerprint, ts, scenario, __first_by_epoch, __last_by_epoch, mapContains(__last_by_epoch, toInt64(0)) AS __has0, __last_by_epoch[toInt64(0)] AS __v0, anyLastIf(__v0, __has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __prev_v0, max(__has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __ever_had0, arrayMax(mapValues(__last_by_epoch)) AS __bucket_max, lagInFrame(__bucket_max, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_max, lagInFrame(ts, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_ts, row_number() OVER (PARTITION BY fingerprint ORDER BY ts) AS __bucket_rn FROM (%s)"
+
+L3="SELECT fingerprint, ts, scenario, __prev_bucket_ts, __bucket_rn, __kv.1 AS __epoch, __kv.2 AS __lval, __first_by_epoch[__kv.1] AS __fval, lagInFrame(__lval, 1) OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_lval, row_number() OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts) AS __epoch_rn, multiIf(__epoch != 0 AND __epoch_rn > 1, if(__lval >= __prev_lval, __lval - __prev_lval, __lval), __epoch != 0 AND (__has0 OR __ever_had0), __lval - if(__fval >= if(__has0, __v0, __prev_v0), if(__has0, __v0, __prev_v0), 0.), __epoch != 0 AND __epoch >= 1784332500000, __lval, __epoch != 0, __lval - __fval, __bucket_rn = 1, nan, __lval < __prev_bucket_max, __lval, __lval - __prev_bucket_max) AS __contrib FROM (%s) ARRAY JOIN arrayZip(mapKeys(__last_by_epoch), mapValues(__last_by_epoch)) AS __kv"
+
+L4="SELECT ts, scenario, if(countIf(isNaN(__contrib) = 0) > 0, sumIf(__contrib, isNaN(__contrib) = 0), nan) AS per_series_value FROM (%s) WHERE ts >= toDateTime(1784332800) GROUP BY fingerprint, ts, scenario"
+
+SPATIAL="SELECT ts, scenario, sum(per_series_value) AS value FROM (%s) WHERE isNaN(per_series_value) = 0 GROUP BY ts, scenario"
+
+run() {
+  local label="$1"; local sql="$2"
+  printf "%-34s" "$label"
+  for i in 1 2 3; do
+    t0=$(python3 -c 'import time; print(time.time())')
+    $CH -q "$sql FORMAT Null"
+    t1=$(python3 -c 'import time; print(time.time())')
+    printf " %6.2f" "$(python3 -c "print($t1 - $t0)")"
+  done
+  echo
+}
+
+printf -v Q_L2 "$L2" "$L1_EPOCH"
+printf -v Q_L3 "$L3" "$Q_L2"
+printf -v Q_L4 "$L4" "$Q_L3"
+printf -v Q_FULL "$SPATIAL" "$Q_L4"
+printf -v Q_LEGACY_FULL "$SPATIAL" "SELECT ts, scenario, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value, per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) AS per_series_value FROM ($L1_LEGACY) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)"
+
+printf -v Q_L2_TL "$L2" "$L1_TWOLEVEL"
+printf -v Q_L3_TL "$L3" "$Q_L2_TL"
+printf -v Q_L4_TL "$L4" "$Q_L3_TL"
+printf -v Q_FULL_TL "$SPATIAL" "$Q_L4_TL"
+
+echo "== layer decomposition (3 runs each, seconds; deltas = layer cost) =="
+run "A. L1 legacy (scalar max)"       "$L1_LEGACY"
+run "B. L1 epoch (per-row maps)"      "$L1_EPOCH"
+run "C. L1 two-level (scalars->maps)" "$L1_TWOLEVEL"
+run "D. L1e + L2 window"              "$Q_L2"
+run "E. L1e + L2 + L3 explode+window" "$Q_L3"
+run "F. full epoch (=E+L4+spatial)"   "$Q_FULL"
+run "G. full epoch w/ two-level L1"   "$Q_FULL_TL"
+run "H. full legacy"                  "$Q_LEGACY_FULL"
+
+echo "== sanity: two-level L1 produces identical final results =="
+$CH -q "SELECT count(), round(sum(value), 3) FROM ($Q_FULL)"
+$CH -q "SELECT count(), round(sum(value), 3) FROM ($Q_FULL_TL)"

tests/integration/testdata/counter_reset_epochs/bench_shard.sh

@@ -0,0 +1,91 @@
+#!/bin/bash
+# Shard-local pushdown simulation for the whale case (50k series, 144M rows).
+#
+# Model: fingerprints are shard-disjoint (the sharding key hashes fingerprint),
+# and nothing in the epoch pipeline crosses series until the spatial sum, which
+# decomposes. So a 4-shard cluster runs the ENTIRE per-series pipeline on each
+# shard over its quarter of the data in parallel, and the initiator merges tiny
+# per-(ts, group) partials. On one laptop, per-shard wall time is measured by
+# running one shard's query with the full machine (each real shard has its own
+# hardware); cluster wall ~= shard wall + merge.
+#
+# Correctness: the 4 shard outputs, merged by summing per (ts, scenario), must
+# equal the single-node result exactly.
+set -eu
+cd "$(dirname "$0")"
+
+B=1784332800000
+
+for k in 0 1 2 3; do
+  if [ ! -d "chdata_shard$k" ]; then
+    echo "== building shard $k (quarter of series, full MV chain) =="
+    clickhouse local --path "chdata_shard$k" --queries-file schema.sql
+    clickhouse local --path "chdata_shard$k" -q "
+INSERT INTO signoz_metrics.samples_v4
+SELECT
+    'default', 'Cumulative', 'bench2_counter',
+    200000 + intDiv(number, 2880) AS fingerprint,
+    $B - 3600000 + (number % 2880) * 30000 AS unix_milli,
+    2.0 * ((number % 2880) - segfirst) AS value,
+    0 AS flags,
+    $B AS inserted_at_unix_milli,
+    if(seg = 0, $B - 90000000, $B - 3600000 + segfirst * 30000 - 15000) AS start_ts
+FROM (
+  SELECT number,
+    if(number % 2880 < 3, 0, intDiv(number % 2880 - 3, 720) + 1) AS seg,
+    if(seg = 0, 0, 3 + (seg - 1) * 720) AS segfirst
+  FROM numbers(144000000)
+  WHERE intDiv(number, 2880) % 4 = $k
+)
+SETTINGS max_insert_threads = 4, max_partitions_per_insert_block = 100;
+
+INSERT INTO signoz_metrics.time_series_v4_1day
+SELECT 'default', 'Cumulative', 'bench2_counter', 200000 + number,
+       ts, '{\"scenario\":\"bench2\"}'
+FROM numbers(50000) ARRAY JOIN [toInt64($B - 86400000), toInt64($B)] AS ts
+WHERE number % 4 = $k;
+"
+  fi
+done
+
+clickhouse local --path chdata_shard0 -q "SELECT 'shard0 samples', count() FROM signoz_metrics.samples_v4 WHERE metric_name = 'bench2_counter'"
+
+echo "== per-shard wall time, full machine (= cluster wall time modulo merge) =="
+for q in big_e_inc_300_raw big_l_inc_300_raw; do
+  printf "%-22s" "$q"
+  for i in 1 2 3; do
+    t0=$(python3 -c 'import time; print(time.time())')
+    clickhouse local --path chdata_shard0 --queries-file "bench_queries/$q.sql" > /dev/null
+    t1=$(python3 -c 'import time; print(time.time())')
+    printf " %.2f" "$(python3 -c "print($t1 - $t0)")"
+  done
+  echo
+done
+
+echo "== correctness: merged 4-shard output == single-node output =="
+clickhouse local --path chdata_large --queries-file bench_queries/big_e_inc_300_raw.sql > /tmp/single_e.csv
+for k in 0 1 2 3; do
+  clickhouse local --path "chdata_shard$k" --queries-file bench_queries/big_e_inc_300_raw.sql > "/tmp/shard_e_$k.csv"
+done
+t0=$(python3 -c 'import time; print(time.time())')
+python3 - <<'PYEOF'
+import csv, math, sys
+def load(p):
+    d = {}
+    with open(p, newline="") as f:
+        for r in csv.reader(f):
+            if r:
+                d[(int(r[0]), r[1])] = d.get((int(r[0]), r[1]), 0.0) + float(r[2])
+    return d
+merged = {}
+for k in range(4):
+    for key, v in load(f"/tmp/shard_e_{k}.csv").items():
+        merged[key] = merged.get(key, 0.0) + v
+single = load("/tmp/single_e.csv")
+bad = [k for k in set(merged) | set(single)
+       if not math.isclose(merged.get(k, float("nan")), single.get(k, float("nan")), rel_tol=1e-9, abs_tol=1e-6)]
+print(f"merge check: {len(single)} rows, {len(bad)} mismatches")
+sys.exit(1 if bad else 0)
+PYEOF
+t1=$(python3 -c 'import time; print(time.time())')
+python3 -c "print(f'initiator merge cost (python, unoptimized): {$t1-$t0:.2f}s over 4 x per-(ts,group) partials')"

tests/integration/testdata/counter_reset_epochs/compare.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python3
+"""Compare a query's actual CSV output against the expected CSV.
+Rows keyed by (ts_s, scenario); float tolerance 1e-6 relative."""
+import csv
+import math
+import sys
+
+
+def load(path):
+    out = {}
+    with open(path, newline="") as f:
+        for row in csv.reader(f):
+            if not row:
+                continue
+            out[(int(row[0]), row[1])] = float(row[2])
+    return out
+
+
+def main(expected_path, actual_path):
+    exp, act = load(expected_path), load(actual_path)
+    bad = []
+    for k in sorted(set(exp) | set(act)):
+        e, a = exp.get(k), act.get(k)
+        if e is None:
+            bad.append(f"unexpected row {k}: actual={a}")
+        elif a is None:
+            bad.append(f"missing row {k}: expected={e}")
+        elif not math.isclose(e, a, rel_tol=1e-6, abs_tol=1e-6):
+            bad.append(f"mismatch {k}: expected={e} actual={a}")
+    if bad:
+        print(f"FAIL ({len(bad)} problems, {len(exp)} expected rows)")
+        for b in bad[:20]:
+            print("  " + b)
+        if len(bad) > 20:
+            print(f"  ... and {len(bad) - 20} more")
+        return 1
+    print(f"OK ({len(exp)} rows)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1], sys.argv[2]))

tests/integration/testdata/counter_reset_epochs/expected/e_inc_1800_agg30m.csv

@@ -0,0 +1,673 @@
+1784332800,s01_steady,120.0
+1784334600,s01_steady,120.0
+1784336400,s01_steady,120.0
+1784338200,s01_steady,120.0
+1784340000,s01_steady,120.0
+1784341800,s01_steady,120.0
+1784343600,s01_steady,120.0
+1784345400,s01_steady,120.0
+1784347200,s01_steady,120.0
+1784349000,s01_steady,120.0
+1784350800,s01_steady,120.0
+1784352600,s01_steady,120.0
+1784354400,s01_steady,120.0
+1784356200,s01_steady,120.0
+1784358000,s01_steady,120.0
+1784359800,s01_steady,120.0
+1784361600,s01_steady,120.0
+1784363400,s01_steady,120.0
+1784365200,s01_steady,120.0
+1784367000,s01_steady,120.0
+1784368800,s01_steady,120.0
+1784370600,s01_steady,120.0
+1784372400,s01_steady,120.0
+1784374200,s01_steady,120.0
+1784376000,s01_steady,120.0
+1784377800,s01_steady,120.0
+1784379600,s01_steady,120.0
+1784381400,s01_steady,120.0
+1784383200,s01_steady,120.0
+1784385000,s01_steady,120.0
+1784386800,s01_steady,120.0
+1784388600,s01_steady,120.0
+1784390400,s01_steady,120.0
+1784392200,s01_steady,120.0
+1784394000,s01_steady,120.0
+1784395800,s01_steady,120.0
+1784397600,s01_steady,120.0
+1784399400,s01_steady,120.0
+1784401200,s01_steady,120.0
+1784403000,s01_steady,120.0
+1784404800,s01_steady,120.0
+1784406600,s01_steady,120.0
+1784408400,s01_steady,120.0
+1784410200,s01_steady,120.0
+1784412000,s01_steady,120.0
+1784413800,s01_steady,120.0
+1784415600,s01_steady,120.0
+1784417400,s01_steady,120.0
+1784332800,s02_midreset,180.0
+1784334600,s02_midreset,180.0
+1784336400,s02_midreset,180.0
+1784338200,s02_midreset,180.0
+1784340000,s02_midreset,180.0
+1784341800,s02_midreset,180.0
+1784343600,s02_midreset,180.0
+1784345400,s02_midreset,180.0
+1784347200,s02_midreset,180.0
+1784349000,s02_midreset,180.0
+1784350800,s02_midreset,180.0
+1784352600,s02_midreset,180.0
+1784354400,s02_midreset,184.0
+1784356200,s02_midreset,180.0
+1784358000,s02_midreset,180.0
+1784359800,s02_midreset,180.0
+1784361600,s02_midreset,180.0
+1784363400,s02_midreset,180.0
+1784365200,s02_midreset,180.0
+1784367000,s02_midreset,180.0
+1784368800,s02_midreset,180.0
+1784370600,s02_midreset,180.0
+1784372400,s02_midreset,180.0
+1784374200,s02_midreset,180.0
+1784376000,s02_midreset,180.0
+1784377800,s02_midreset,180.0
+1784379600,s02_midreset,180.0
+1784381400,s02_midreset,180.0
+1784383200,s02_midreset,180.0
+1784385000,s02_midreset,180.0
+1784386800,s02_midreset,180.0
+1784388600,s02_midreset,180.0
+1784390400,s02_midreset,180.0
+1784392200,s02_midreset,180.0
+1784394000,s02_midreset,180.0
+1784395800,s02_midreset,180.0
+1784397600,s02_midreset,180.0
+1784399400,s02_midreset,180.0
+1784401200,s02_midreset,180.0
+1784403000,s02_midreset,180.0
+1784404800,s02_midreset,180.0
+1784406600,s02_midreset,180.0
+1784408400,s02_midreset,180.0
+1784410200,s02_midreset,180.0
+1784412000,s02_midreset,180.0
+1784413800,s02_midreset,180.0
+1784415600,s02_midreset,180.0
+1784417400,s02_midreset,180.0
+1784332800,s03_multireset,60.0
+1784334600,s03_multireset,60.0
+1784336400,s03_multireset,60.0
+1784338200,s03_multireset,60.0
+1784340000,s03_multireset,159.0
+1784341800,s03_multireset,60.0
+1784343600,s03_multireset,60.0
+1784345400,s03_multireset,60.0
+1784347200,s03_multireset,159.0
+1784349000,s03_multireset,60.0
+1784350800,s03_multireset,60.0
+1784352600,s03_multireset,60.0
+1784354400,s03_multireset,159.0
+1784356200,s03_multireset,60.0
+1784358000,s03_multireset,60.0
+1784359800,s03_multireset,60.0
+1784361600,s03_multireset,159.0
+1784363400,s03_multireset,60.0
+1784365200,s03_multireset,60.0
+1784367000,s03_multireset,60.0
+1784368800,s03_multireset,159.0
+1784370600,s03_multireset,60.0
+1784372400,s03_multireset,60.0
+1784374200,s03_multireset,60.0
+1784376000,s03_multireset,159.0
+1784377800,s03_multireset,60.0
+1784379600,s03_multireset,60.0
+1784381400,s03_multireset,60.0
+1784383200,s03_multireset,159.0
+1784385000,s03_multireset,60.0
+1784386800,s03_multireset,60.0
+1784388600,s03_multireset,60.0
+1784390400,s03_multireset,159.0
+1784392200,s03_multireset,60.0
+1784394000,s03_multireset,60.0
+1784395800,s03_multireset,60.0
+1784397600,s03_multireset,159.0
+1784399400,s03_multireset,60.0
+1784401200,s03_multireset,60.0
+1784403000,s03_multireset,60.0
+1784404800,s03_multireset,159.0
+1784406600,s03_multireset,60.0
+1784408400,s03_multireset,60.0
+1784410200,s03_multireset,60.0
+1784412000,s03_multireset,159.0
+1784413800,s03_multireset,60.0
+1784415600,s03_multireset,60.0
+1784417400,s03_multireset,60.0
+1784332800,s04_regrow,150.0
+1784334600,s04_regrow,150.0
+1784336400,s04_regrow,150.0
+1784338200,s04_regrow,150.0
+1784340000,s04_regrow,150.0
+1784341800,s04_regrow,150.0
+1784343600,s04_regrow,150.0
+1784345400,s04_regrow,150.0
+1784347200,s04_regrow,150.0
+1784349000,s04_regrow,150.0
+1784350800,s04_regrow,150.0
+1784352600,s04_regrow,150.0
+1784354400,s04_regrow,150.0
+1784356200,s04_regrow,150.0
+1784358000,s04_regrow,150.0
+1784359800,s04_regrow,150.0
+1784361600,s04_regrow,150.0
+1784363400,s04_regrow,150.0
+1784365200,s04_regrow,150.0
+1784367000,s04_regrow,150.0
+1784368800,s04_regrow,150.0
+1784370600,s04_regrow,150.0
+1784372400,s04_regrow,150.0
+1784374200,s04_regrow,150.0
+1784376000,s04_regrow,5605.0
+1784377800,s04_regrow,1500.0
+1784379600,s04_regrow,1500.0
+1784381400,s04_regrow,1500.0
+1784383200,s04_regrow,1500.0
+1784385000,s04_regrow,1500.0
+1784386800,s04_regrow,1500.0
+1784388600,s04_regrow,1500.0
+1784390400,s04_regrow,1500.0
+1784392200,s04_regrow,1500.0
+1784394000,s04_regrow,1500.0
+1784395800,s04_regrow,1500.0
+1784397600,s04_regrow,1500.0
+1784399400,s04_regrow,1500.0
+1784401200,s04_regrow,1500.0
+1784403000,s04_regrow,1500.0
+1784404800,s04_regrow,1500.0
+1784406600,s04_regrow,1500.0
+1784408400,s04_regrow,1500.0
+1784410200,s04_regrow,1500.0
+1784412000,s04_regrow,1500.0
+1784413800,s04_regrow,1500.0
+1784415600,s04_regrow,1500.0
+1784417400,s04_regrow,1500.0
+1784332800,s05_legacy,120.0
+1784334600,s05_legacy,120.0
+1784336400,s05_legacy,120.0
+1784338200,s05_legacy,120.0
+1784340000,s05_legacy,120.0
+1784341800,s05_legacy,120.0
+1784343600,s05_legacy,120.0
+1784345400,s05_legacy,120.0
+1784347200,s05_legacy,120.0
+1784349000,s05_legacy,120.0
+1784350800,s05_legacy,120.0
+1784352600,s05_legacy,120.0
+1784354400,s05_legacy,120.0
+1784356200,s05_legacy,120.0
+1784358000,s05_legacy,120.0
+1784359800,s05_legacy,120.0
+1784361600,s05_legacy,6.0
+1784363400,s05_legacy,237.0
+1784365200,s05_legacy,120.0
+1784367000,s05_legacy,120.0
+1784368800,s05_legacy,120.0
+1784370600,s05_legacy,120.0
+1784372400,s05_legacy,120.0
+1784374200,s05_legacy,120.0
+1784376000,s05_legacy,120.0
+1784377800,s05_legacy,120.0
+1784379600,s05_legacy,120.0
+1784381400,s05_legacy,120.0
+1784383200,s05_legacy,120.0
+1784385000,s05_legacy,120.0
+1784386800,s05_legacy,120.0
+1784388600,s05_legacy,120.0
+1784390400,s05_legacy,6.0
+1784392200,s05_legacy,243.0
+1784394000,s05_legacy,120.0
+1784395800,s05_legacy,120.0
+1784397600,s05_legacy,120.0
+1784399400,s05_legacy,120.0
+1784401200,s05_legacy,120.0
+1784403000,s05_legacy,120.0
+1784404800,s05_legacy,120.0
+1784406600,s05_legacy,120.0
+1784408400,s05_legacy,120.0
+1784410200,s05_legacy,120.0
+1784412000,s05_legacy,120.0
+1784413800,s05_legacy,120.0
+1784415600,s05_legacy,120.0
+1784417400,s05_legacy,120.0
+1784332800,s06_transition,120.0
+1784334600,s06_transition,120.0
+1784336400,s06_transition,120.0
+1784338200,s06_transition,120.0
+1784340000,s06_transition,120.0
+1784341800,s06_transition,120.0
+1784343600,s06_transition,120.0
+1784345400,s06_transition,120.0
+1784347200,s06_transition,120.0
+1784349000,s06_transition,120.0
+1784350800,s06_transition,120.0
+1784352600,s06_transition,120.0
+1784354400,s06_transition,120.0
+1784356200,s06_transition,120.0
+1784358000,s06_transition,120.0
+1784359800,s06_transition,120.0
+1784361600,s06_transition,120.0
+1784363400,s06_transition,120.0
+1784365200,s06_transition,120.0
+1784367000,s06_transition,120.0
+1784368800,s06_transition,120.0
+1784370600,s06_transition,120.0
+1784372400,s06_transition,120.0
+1784374200,s06_transition,120.0
+1784376000,s06_transition,120.0
+1784377800,s06_transition,120.0
+1784379600,s06_transition,120.0
+1784381400,s06_transition,120.0
+1784383200,s06_transition,120.0
+1784385000,s06_transition,120.0
+1784386800,s06_transition,120.0
+1784388600,s06_transition,120.0
+1784390400,s06_transition,120.0
+1784392200,s06_transition,120.0
+1784394000,s06_transition,120.0
+1784395800,s06_transition,120.0
+1784397600,s06_transition,121.0
+1784399400,s06_transition,120.0
+1784401200,s06_transition,120.0
+1784403000,s06_transition,120.0
+1784404800,s06_transition,120.0
+1784406600,s06_transition,120.0
+1784408400,s06_transition,120.0
+1784410200,s06_transition,120.0
+1784412000,s06_transition,120.0
+1784413800,s06_transition,120.0
+1784415600,s06_transition,120.0
+1784417400,s06_transition,120.0
+1784332800,s07_gap,0.0
+1784334600,s07_gap,0.0
+1784336400,s07_gap,0.0
+1784338200,s07_gap,0.0
+1784340000,s07_gap,0.0
+1784341800,s07_gap,0.0
+1784343600,s07_gap,0.0
+1784345400,s07_gap,0.0
+1784347200,s07_gap,0.0
+1784349000,s07_gap,0.0
+1784350800,s07_gap,0.0
+1784352600,s07_gap,0.0
+1784354400,s07_gap,0.0
+1784356200,s07_gap,0.0
+1784358000,s07_gap,0.0
+1784359800,s07_gap,0.0
+1784361600,s07_gap,0.0
+1784363400,s07_gap,0.0
+1784365200,s07_gap,0.0
+1784367000,s07_gap,0.0
+1784368800,s07_gap,0.0
+1784370600,s07_gap,0.0
+1784372400,s07_gap,0.0
+1784374200,s07_gap,0.0
+1784376000,s07_gap,0.0
+1784377800,s07_gap,0.0
+1784379600,s07_gap,0.0
+1784381400,s07_gap,0.0
+1784383200,s07_gap,0.0
+1784385000,s07_gap,0.0
+1784386800,s07_gap,0.0
+1784388600,s07_gap,0.0
+1784390400,s07_gap,0.0
+1784392200,s07_gap,0.0
+1784394000,s07_gap,0.0
+1784395800,s07_gap,0.0
+1784397600,s07_gap,0.0
+1784399400,s07_gap,0.0
+1784401200,s07_gap,0.0
+1784403000,s07_gap,0.0
+1784404800,s07_gap,0.0
+1784406600,s07_gap,0.0
+1784408400,s07_gap,0.0
+1784410200,s07_gap,0.0
+1784412000,s07_gap,0.0
+1784413800,s07_gap,0.0
+1784415600,s07_gap,0.0
+1784417400,s07_gap,0.0
+1784386800,s08_singlepoint,42.0
+1784332800,s09_slow,60.0
+1784334600,s09_slow,60.0
+1784336400,s09_slow,60.0
+1784338200,s09_slow,60.0
+1784340000,s09_slow,60.0
+1784341800,s09_slow,60.0
+1784343600,s09_slow,60.0
+1784345400,s09_slow,60.0
+1784347200,s09_slow,60.0
+1784349000,s09_slow,60.0
+1784350800,s09_slow,60.0
+1784352600,s09_slow,60.0
+1784354400,s09_slow,60.0
+1784356200,s09_slow,60.0
+1784358000,s09_slow,60.0
+1784359800,s09_slow,60.0
+1784361600,s09_slow,60.0
+1784363400,s09_slow,60.0
+1784365200,s09_slow,60.0
+1784367000,s09_slow,60.0
+1784368800,s09_slow,60.0
+1784370600,s09_slow,60.0
+1784372400,s09_slow,60.0
+1784374200,s09_slow,60.0
+1784376000,s09_slow,60.0
+1784377800,s09_slow,60.0
+1784379600,s09_slow,60.0
+1784381400,s09_slow,60.0
+1784383200,s09_slow,60.0
+1784385000,s09_slow,60.0
+1784386800,s09_slow,60.0
+1784388600,s09_slow,60.0
+1784390400,s09_slow,60.0
+1784392200,s09_slow,60.0
+1784394000,s09_slow,60.0
+1784395800,s09_slow,60.0
+1784397600,s09_slow,60.0
+1784399400,s09_slow,60.0
+1784401200,s09_slow,60.0
+1784403000,s09_slow,60.0
+1784404800,s09_slow,54.0
+1784406600,s09_slow,60.0
+1784408400,s09_slow,60.0
+1784410200,s09_slow,60.0
+1784412000,s09_slow,60.0
+1784413800,s09_slow,60.0
+1784415600,s09_slow,60.0
+1784417400,s09_slow,60.0
+1784332800,s10_boundaryreset,240.0
+1784334600,s10_boundaryreset,240.0
+1784336400,s10_boundaryreset,240.0
+1784338200,s10_boundaryreset,240.0
+1784340000,s10_boundaryreset,240.0
+1784341800,s10_boundaryreset,240.0
+1784343600,s10_boundaryreset,240.0
+1784345400,s10_boundaryreset,240.0
+1784347200,s10_boundaryreset,238.0
+1784349000,s10_boundaryreset,240.0
+1784350800,s10_boundaryreset,240.0
+1784352600,s10_boundaryreset,240.0
+1784354400,s10_boundaryreset,240.0
+1784356200,s10_boundaryreset,240.0
+1784358000,s10_boundaryreset,240.0
+1784359800,s10_boundaryreset,240.0
+1784361600,s10_boundaryreset,240.0
+1784363400,s10_boundaryreset,240.0
+1784365200,s10_boundaryreset,240.0
+1784367000,s10_boundaryreset,240.0
+1784368800,s10_boundaryreset,240.0
+1784370600,s10_boundaryreset,240.0
+1784372400,s10_boundaryreset,240.0
+1784374200,s10_boundaryreset,240.0
+1784376000,s10_boundaryreset,240.0
+1784377800,s10_boundaryreset,240.0
+1784379600,s10_boundaryreset,240.0
+1784381400,s10_boundaryreset,240.0
+1784383200,s10_boundaryreset,240.0
+1784385000,s10_boundaryreset,240.0
+1784386800,s10_boundaryreset,240.0
+1784388600,s10_boundaryreset,240.0
+1784390400,s10_boundaryreset,240.0
+1784392200,s10_boundaryreset,240.0
+1784394000,s10_boundaryreset,240.0
+1784395800,s10_boundaryreset,240.0
+1784397600,s10_boundaryreset,240.0
+1784399400,s10_boundaryreset,240.0
+1784401200,s10_boundaryreset,240.0
+1784403000,s10_boundaryreset,240.0
+1784404800,s10_boundaryreset,240.0
+1784406600,s10_boundaryreset,240.0
+1784408400,s10_boundaryreset,240.0
+1784410200,s10_boundaryreset,240.0
+1784412000,s10_boundaryreset,240.0
+1784413800,s10_boundaryreset,240.0
+1784415600,s10_boundaryreset,240.0
+1784417400,s10_boundaryreset,240.0
+1784332800,s11_dupooo,120.0
+1784334600,s11_dupooo,120.0
+1784336400,s11_dupooo,120.0
+1784338200,s11_dupooo,120.0
+1784340000,s11_dupooo,120.0
+1784341800,s11_dupooo,120.0
+1784343600,s11_dupooo,120.0
+1784345400,s11_dupooo,120.0
+1784347200,s11_dupooo,120.0
+1784349000,s11_dupooo,120.0
+1784350800,s11_dupooo,120.0
+1784352600,s11_dupooo,120.0
+1784354400,s11_dupooo,120.0
+1784356200,s11_dupooo,120.0
+1784358000,s11_dupooo,120.0
+1784359800,s11_dupooo,120.0
+1784361600,s11_dupooo,120.0
+1784363400,s11_dupooo,120.0
+1784365200,s11_dupooo,124.0
+1784367000,s11_dupooo,120.0
+1784368800,s11_dupooo,120.0
+1784370600,s11_dupooo,120.0
+1784372400,s11_dupooo,120.0
+1784374200,s11_dupooo,120.0
+1784376000,s11_dupooo,120.0
+1784377800,s11_dupooo,120.0
+1784379600,s11_dupooo,120.0
+1784381400,s11_dupooo,120.0
+1784383200,s11_dupooo,120.0
+1784385000,s11_dupooo,120.0
+1784386800,s11_dupooo,120.0
+1784388600,s11_dupooo,120.0
+1784390400,s11_dupooo,120.0
+1784392200,s11_dupooo,120.0
+1784394000,s11_dupooo,120.0
+1784395800,s11_dupooo,120.0
+1784397600,s11_dupooo,120.0
+1784399400,s11_dupooo,120.0
+1784401200,s11_dupooo,120.0
+1784403000,s11_dupooo,120.0
+1784404800,s11_dupooo,120.0
+1784406600,s11_dupooo,120.0
+1784408400,s11_dupooo,120.0
+1784410200,s11_dupooo,120.0
+1784412000,s11_dupooo,120.0
+1784413800,s11_dupooo,120.0
+1784415600,s11_dupooo,120.0
+1784417400,s11_dupooo,120.0
+1784332800,s12_twowriter,180.0
+1784334600,s12_twowriter,180.0
+1784336400,s12_twowriter,180.0
+1784338200,s12_twowriter,180.0
+1784340000,s12_twowriter,180.0
+1784341800,s12_twowriter,180.0
+1784343600,s12_twowriter,180.0
+1784345400,s12_twowriter,180.0
+1784347200,s12_twowriter,180.0
+1784349000,s12_twowriter,180.0
+1784350800,s12_twowriter,180.0
+1784352600,s12_twowriter,180.0
+1784354400,s12_twowriter,180.0
+1784356200,s12_twowriter,180.0
+1784358000,s12_twowriter,180.0
+1784359800,s12_twowriter,180.0
+1784361600,s12_twowriter,180.0
+1784363400,s12_twowriter,180.0
+1784365200,s12_twowriter,180.0
+1784367000,s12_twowriter,180.0
+1784368800,s12_twowriter,180.0
+1784370600,s12_twowriter,180.0
+1784372400,s12_twowriter,180.0
+1784374200,s12_twowriter,180.0
+1784376000,s12_twowriter,180.0
+1784377800,s12_twowriter,180.0
+1784379600,s12_twowriter,180.0
+1784381400,s12_twowriter,180.0
+1784383200,s12_twowriter,180.0
+1784385000,s12_twowriter,180.0
+1784386800,s12_twowriter,180.0
+1784388600,s12_twowriter,180.0
+1784390400,s12_twowriter,180.0
+1784392200,s12_twowriter,180.0
+1784394000,s12_twowriter,180.0
+1784395800,s12_twowriter,180.0
+1784397600,s12_twowriter,180.0
+1784399400,s12_twowriter,180.0
+1784401200,s12_twowriter,180.0
+1784403000,s12_twowriter,180.0
+1784404800,s12_twowriter,180.0
+1784406600,s12_twowriter,180.0
+1784408400,s12_twowriter,180.0
+1784410200,s12_twowriter,180.0
+1784412000,s12_twowriter,180.0
+1784413800,s12_twowriter,180.0
+1784415600,s12_twowriter,180.0
+1784417400,s12_twowriter,180.0
+1784332800,s14_updown,0.0
+1784334600,s14_updown,0.0
+1784336400,s14_updown,0.0
+1784338200,s14_updown,0.0
+1784340000,s14_updown,0.0
+1784341800,s14_updown,0.0
+1784343600,s14_updown,0.0
+1784345400,s14_updown,0.0
+1784347200,s14_updown,0.0
+1784349000,s14_updown,0.0
+1784350800,s14_updown,0.0
+1784352600,s14_updown,0.0
+1784354400,s14_updown,0.0
+1784356200,s14_updown,0.0
+1784358000,s14_updown,0.0
+1784359800,s14_updown,0.0
+1784361600,s14_updown,0.0
+1784363400,s14_updown,0.0
+1784365200,s14_updown,0.0
+1784367000,s14_updown,0.0
+1784368800,s14_updown,0.0
+1784370600,s14_updown,0.0
+1784372400,s14_updown,0.0
+1784374200,s14_updown,0.0
+1784376000,s14_updown,0.0
+1784377800,s14_updown,0.0
+1784379600,s14_updown,0.0
+1784381400,s14_updown,0.0
+1784383200,s14_updown,0.0
+1784385000,s14_updown,0.0
+1784386800,s14_updown,0.0
+1784388600,s14_updown,0.0
+1784390400,s14_updown,0.0
+1784392200,s14_updown,0.0
+1784394000,s14_updown,0.0
+1784395800,s14_updown,0.0
+1784397600,s14_updown,0.0
+1784399400,s14_updown,0.0
+1784401200,s14_updown,0.0
+1784403000,s14_updown,0.0
+1784404800,s14_updown,0.0
+1784406600,s14_updown,0.0
+1784408400,s14_updown,0.0
+1784410200,s14_updown,0.0
+1784412000,s14_updown,0.0
+1784413800,s14_updown,0.0
+1784415600,s14_updown,0.0
+1784417400,s14_updown,0.0
+1784332800,s15_aggold,30.0
+1784334600,s15_aggold,30.0
+1784336400,s15_aggold,30.0
+1784338200,s15_aggold,30.0
+1784340000,s15_aggold,30.0
+1784341800,s15_aggold,30.0
+1784343600,s15_aggold,30.0
+1784345400,s15_aggold,30.0
+1784347200,s15_aggold,30.0
+1784349000,s15_aggold,30.0
+1784350800,s15_aggold,30.0
+1784352600,s15_aggold,30.0
+1784354400,s15_aggold,30.0
+1784356200,s15_aggold,30.0
+1784358000,s15_aggold,30.0
+1784359800,s15_aggold,30.0
+1784361600,s15_aggold,30.0
+1784363400,s15_aggold,30.0
+1784365200,s15_aggold,30.0
+1784367000,s15_aggold,30.0
+1784368800,s15_aggold,30.0
+1784370600,s15_aggold,30.0
+1784372400,s15_aggold,30.0
+1784374200,s15_aggold,30.0
+1784376000,s15_aggold,30.0
+1784377800,s15_aggold,30.0
+1784379600,s15_aggold,30.0
+1784381400,s15_aggold,30.0
+1784383200,s15_aggold,5.0
+1784385000,s15_aggold,53.0
+1784386800,s15_aggold,30.0
+1784388600,s15_aggold,30.0
+1784390400,s15_aggold,30.0
+1784392200,s15_aggold,30.0
+1784394000,s15_aggold,30.0
+1784395800,s15_aggold,30.0
+1784397600,s15_aggold,30.0
+1784399400,s15_aggold,30.0
+1784401200,s15_aggold,30.0
+1784403000,s15_aggold,30.0
+1784404800,s15_aggold,30.0
+1784406600,s15_aggold,30.0
+1784408400,s15_aggold,30.0
+1784410200,s15_aggold,30.0
+1784412000,s15_aggold,30.0
+1784413800,s15_aggold,30.0
+1784415600,s15_aggold,30.0
+1784417400,s15_aggold,30.0
+1784332800,s16_stale,120.0
+1784334600,s16_stale,120.0
+1784336400,s16_stale,120.0
+1784338200,s16_stale,120.0
+1784340000,s16_stale,120.0
+1784341800,s16_stale,120.0
+1784343600,s16_stale,120.0
+1784345400,s16_stale,120.0
+1784347200,s16_stale,120.0
+1784349000,s16_stale,120.0
+1784350800,s16_stale,120.0
+1784352600,s16_stale,120.0
+1784354400,s16_stale,120.0
+1784356200,s16_stale,120.0
+1784358000,s16_stale,120.0
+1784359800,s16_stale,120.0
+1784361600,s16_stale,120.0
+1784363400,s16_stale,120.0
+1784365200,s16_stale,120.0
+1784367000,s16_stale,120.0
+1784368800,s16_stale,120.0
+1784370600,s16_stale,120.0
+1784372400,s16_stale,120.0
+1784374200,s16_stale,120.0
+1784376000,s16_stale,120.0
+1784377800,s16_stale,120.0
+1784379600,s16_stale,120.0
+1784381400,s16_stale,120.0
+1784383200,s16_stale,120.0
+1784385000,s16_stale,120.0
+1784386800,s16_stale,120.0
+1784388600,s16_stale,120.0
+1784390400,s16_stale,120.0
+1784392200,s16_stale,120.0
+1784394000,s16_stale,120.0
+1784395800,s16_stale,120.0
+1784397600,s16_stale,120.0
+1784399400,s16_stale,120.0
+1784401200,s16_stale,120.0
+1784403000,s16_stale,120.0
+1784404800,s16_stale,120.0
+1784406600,s16_stale,120.0
+1784408400,s16_stale,120.0
+1784410200,s16_stale,120.0
+1784412000,s16_stale,120.0
+1784413800,s16_stale,120.0
+1784415600,s16_stale,120.0
+1784417400,s16_stale,120.0

tests/integration/testdata/counter_reset_epochs/expected/e_inc_86400_agg30m.csv

@@ -0,0 +1,15 @@
+1784332800,s01_steady,5760.0
+1784332800,s02_midreset,8644.0
+1784332800,s03_multireset,3969.0
+1784332800,s04_regrow,43705.0
+1784332800,s05_legacy,1926.0
+1784332800,s06_transition,5761.0
+1784332800,s07_gap,0.0
+1784332800,s08_singlepoint,42.0
+1784332800,s09_slow,2874.0
+1784332800,s10_boundaryreset,11518.0
+1784332800,s11_dupooo,5764.0
+1784332800,s12_twowriter,8640.0
+1784332800,s14_updown,0.0
+1784332800,s15_aggold,845.0
+1784332800,s16_stale,5760.0

tests/integration/testdata/counter_reset_epochs/expected/e_inc_86400_raw.csv

@@ -0,0 +1,15 @@
+1784332800,s01_steady,5760.0
+1784332800,s02_midreset,8644.0
+1784332800,s03_multireset,3969.0
+1784332800,s04_regrow,43705.0
+1784332800,s05_legacy,1926.0
+1784332800,s06_transition,5761.0
+1784332800,s07_gap,0.0
+1784332800,s08_singlepoint,42.0
+1784332800,s09_slow,2874.0
+1784332800,s10_boundaryreset,11518.0
+1784332800,s11_dupooo,5764.0
+1784332800,s12_twowriter,8640.0
+1784332800,s14_updown,0.0
+1784332800,s15_aggold,845.0
+1784332800,s16_stale,5760.0

tests/integration/testdata/counter_reset_epochs/expected/l_inc_86400_raw.csv

@@ -0,0 +1,14 @@
+1784332800,s01_steady,5760.0
+1784332800,s02_midreset,5612.0
+1784332800,s03_multireset,242.0
+1784332800,s04_regrow,39605.0
+1784332800,s05_legacy,1926.0
+1784332800,s06_transition,4322.0
+1784332800,s07_gap,0.0
+1784332800,s09_slow,2410.0
+1784332800,s10_boundaryreset,8322.0
+1784332800,s11_dupooo,3112.0
+1784332800,s12_twowriter,7200.0
+1784332800,s14_updown,0.0
+1784332800,s15_aggold,845.0
+1784332800,s16_stale,5760.0

tests/integration/testdata/counter_reset_epochs/generate_dataset.py

@@ -0,0 +1,620 @@
+#!/usr/bin/env python3
+"""One-day counter-reset verification dataset + independent ground truth.
+
+Generates:
+  inserts.sql            raw samples (shuffled chunks, one duplicated chunk) + time series rows
+  expected/<case>.csv    per-query expected output (ts_s, scenario, value)
+  report.txt             cross-implementation assertions (bucket-spec vs point-truth,
+                         zoom consistency, legacy parity/inconsistency demos)
+
+Three independent reference implementations:
+  bucket_spec  : the epoch contribution rules from the design (per step-bucket maps)
+  legacy_spec  : the current production semantics (bucket max + negative-diff pair rule)
+  point_truth  : per-point epoch-aware accumulation, knows nothing about buckets
+Exact equality between bucket_spec and point_truth on every pure-epoch scenario is
+the core correctness claim; the SQL pipelines are then compared against these CSVs.
+"""
+
+import csv
+import math
+import os
+import random
+import sys
+
+B = 1784332800000  # 2026-07-18T00:00:00Z in ms
+H = 3600000
+DAY = 86400000
+END = B + DAY
+METRIC = "it_counter_total"
+
+E_OLD = B - 90000000  # an epoch that began ~25h before the day starts
+
+
+def rng(a, b, n):
+    return [a + i * b for i in range(n)]
+
+
+# ---------------------------------------------------------------- scenarios --
+# each sample: (t_ms, value, start_ts_ms)
+
+def s01_steady():
+    pts = []
+    t = B - H
+    i = 0
+    while t < END + 300000:
+        pts.append((t, 1000.0 + 2 * i, E_OLD))
+        t += 30000
+        i += 1
+    return pts
+
+
+def s02_midreset():
+    pts = []
+    reset_at = B + 6 * H + 130000  # 06:02:10, mid 5m and mid 60s bucket
+    e2 = reset_at
+    t = B - H
+    i = 0
+    while t < reset_at:
+        pts.append((t, 500.0 + 3 * i, E_OLD))
+        t += 30000
+        i += 1
+    t = B + 6 * H + 150000  # 06:02:30
+    j = 0
+    while t < END + 300000:
+        pts.append((t, 7.0 + 3 * j, e2))
+        t += 30000
+        j += 1
+    return pts
+
+
+def s03_multireset():
+    pts = []
+    resets = [B + 2 * H * i + 45000 for i in range(1, 12)]  # 02:00:45 .. 22:00:45
+    t = B - H
+    epoch = E_OLD
+    k = 0
+    while t < END + 300000:
+        while resets and t >= resets[0]:
+            epoch = resets.pop(0)
+            k = 0
+        pts.append((t, 100.0 + 1 * k, epoch))
+        t += 30000
+        k += 1
+    return pts
+
+
+def s04_regrow():
+    pts = []
+    t = B - H
+    i = 0
+    last = 0.0
+    while t <= B + 12 * H:
+        last = 200.0 + 5 * i
+        pts.append((t, last, E_OLD))
+        t += 60000
+        i += 1
+    e2 = B + 12 * H + 30000
+    t = B + 12 * H + 60000
+    j = 0
+    while t < END + 300000:
+        # first post-reset export already above the pre-reset last value:
+        # invisible to value-based detection
+        pts.append((t, last + 100.0 + 50 * j, e2))
+        t += 60000
+        j += 1
+    return pts
+
+
+def s05_legacy():
+    pts = []
+    t = B - H
+    v = 300.0
+    while t < END + 300000:
+        if t == B + 8 * H + 90000:
+            v = 5.0
+        elif t == B + 16 * H + 90000:
+            v = 11.0
+        pts.append((t, v, 0))
+        v += 2
+        t += 30000
+    return pts
+
+
+def s06_transition():
+    pts = []
+    e6 = B - 50000000
+    e6b = B + 18 * H + 15000
+    t = B - H
+    v = 700.0
+    while t < END + 300000:
+        if t < B + 12 * H:
+            s = 0
+        elif t < B + 18 * H + 15000:
+            s = e6
+        else:
+            if s != e6b:
+                v = 3.0  # the real reset at 18:00:15
+            s = e6b
+        pts.append((t, v, s))
+        v += 2
+        t += 30000
+    return pts
+
+
+def s07_gap():
+    pts = []
+    t = B - H
+    while t < END + 300000:
+        if not (B + 10 * H <= t < B + 10 * H + 600000):
+            pts.append((t, 1997.0, E_OLD))
+        t += 60000
+    return pts
+
+
+def s08_singlepoint():
+    return [(B + 15 * H, 42.0, B + 15 * H - 60000)]
+
+
+def s09_slow():
+    pts = []
+    t = B - H
+    i = 0
+    while t <= B + 20 * H:
+        pts.append((t, 50.0 + 10 * i, E_OLD))
+        t += 300000
+        i += 1
+    e2 = B + 20 * H + 150000
+    t = B + 20 * H + 300000
+    j = 0
+    while t < END + 300000:
+        pts.append((t, 4.0 + 10 * j, e2))
+        t += 300000
+        j += 1
+    return pts
+
+
+def s10_boundaryreset():
+    pts = []
+    t = B - H
+    i = 0
+    while t <= B + 4 * H - 30000:
+        pts.append((t, 800.0 + 4 * i, E_OLD))
+        t += 30000
+        i += 1
+    e2 = B + 4 * H - 5000
+    t = B + 4 * H  # first post-reset sample exactly on the bucket boundary
+    j = 0
+    while t < END + 300000:
+        pts.append((t, 2.0 + 4 * j, e2))
+        t += 30000
+        j += 1
+    return pts
+
+
+def s11_dupooo():
+    pts = []
+    reset_at = B + 9 * H + 45000
+    e2 = reset_at
+    t = B - H
+    i = 0
+    while t < reset_at:
+        pts.append((t, 250.0 + 2 * i, E_OLD))
+        t += 30000
+        i += 1
+    t = B + 9 * H + 60000
+    j = 0
+    while t < END + 300000:
+        pts.append((t, 6.0 + 2 * j, e2))
+        t += 30000
+        j += 1
+    return pts
+
+
+def s12_twowriter():
+    ea, eb = E_OLD, B - 80000000
+    pts = []
+    t = B - H
+    i = 0
+    while t < END + 300000:
+        pts.append((t, 100.0 + 1 * i, ea))
+        t += 60000
+        i += 1
+    t = B - H + 30000
+    i = 0
+    while t < END + 300000:
+        pts.append((t, 5000.0 + 5 * i, eb))
+        t += 60000
+        i += 1
+    return sorted(pts)
+
+
+def s14_updown():
+    pts = []
+    t = B - H
+    i = 0
+    while t < END + 300000:
+        k = i % 40
+        tri = k if k < 20 else 40 - k
+        pts.append((t, 500.0 + tri * 10.0, 0))
+        t += 30000
+        i += 1
+    return pts
+
+
+def s15_aggold():
+    # rollup rows written BEFORE migration 1012: they exist only in the agg
+    # tables, with empty epoch maps. One reset (a value drop) at 14:05.
+    pts = []
+    t = B - H
+    v = 10.0
+    while t < END + 300000:
+        if t == B + 14 * H + 300000:
+            v = 3.0
+        pts.append((t, v, 0))
+        v += 5
+        t += 300000
+    return pts
+
+
+def s16_stale():
+    # steady counter with no-recorded-value markers (flags bit 0 set, value 0)
+    # in the middle: the markers must be completely invisible to rate/increase
+    pts = []
+    t = B - H
+    i = 0
+    while t < END + 300000:
+        pts.append((t, 400.0 + 2 * i, E_OLD))
+        t += 30000
+        i += 1
+    return pts
+
+
+# (t_ms,) marker rows for s16, inserted with flags=1, value=0, start_ts=0
+S16_MARKERS = [B + 11 * H + 15000, B + 11 * H + 45000]
+
+
+SCENARIOS = {
+    "s01_steady": (101, s01_steady()),
+    "s02_midreset": (102, s02_midreset()),
+    "s03_multireset": (103, s03_multireset()),
+    "s04_regrow": (104, s04_regrow()),
+    "s05_legacy": (105, s05_legacy()),
+    "s06_transition": (106, s06_transition()),
+    "s07_gap": (107, s07_gap()),
+    "s08_singlepoint": (108, s08_singlepoint()),
+    "s09_slow": (109, s09_slow()),
+    "s10_boundaryreset": (110, s10_boundaryreset()),
+    "s11_dupooo": (111, s11_dupooo()),
+    "s12_twowriter": (112, s12_twowriter()),
+    "s14_updown": (114, s14_updown()),
+    "s15_aggold": (115, s15_aggold()),
+    "s16_stale": (116, s16_stale()),
+}
+
+PURE_EPOCH = [
+    "s01_steady", "s02_midreset", "s03_multireset", "s04_regrow", "s07_gap",
+    "s08_singlepoint", "s09_slow", "s10_boundaryreset", "s11_dupooo", "s12_twowriter",
+    "s16_stale",
+]
+ALL_KEY0 = ["s05_legacy", "s14_updown"]
+# rollup rows from before the migration exist only in the agg tables
+AGG_ONLY = {"s15_aggold"}
+# cases whose SQL reads a rollup table: the explicitly hinted ones, plus the
+# step-86400 "raw" cases whose 48h fetch range (day + day lookback) makes
+# WhichSamplesTableToUse pick agg_5m automatically
+AGG_CASES = {
+    "e_inc_300_agg5m", "e_inc_1800_agg30m", "e_inc_86400_agg30m",
+    "e_inc_86400_raw", "l_inc_86400_raw",
+}
+
+
+# ------------------------------------------------------- reference engines --
+
+def buckets_of(points, step_s, adj_start, adj_end):
+    """bucket_ts(s) -> {epoch: (fval, lval)} plus key-0 presence, from raw points."""
+    out = {}
+    for (t, v, s) in points:
+        if not (adj_start <= t < adj_end):
+            continue
+        b = (t // 1000) // step_s * step_s
+        d = out.setdefault(b, {})
+        if s in d:
+            f, l = d[s]
+            d[s] = (min(f, v), max(l, v))
+        else:
+            d[s] = (v, v)
+    return out
+
+
+def bucket_spec(points, step_s, adj_start, adj_end, display_start, time_agg):
+    """The epoch contribution rules from the design. Returns {ts_s: value}, NaN rows omitted."""
+    bks = buckets_of(points, step_s, adj_start, adj_end)
+    res = {}
+    prev_lval = {}
+    prev_v0 = None
+    ever_had0 = False
+    prev_bucket_max = None
+    prev_bucket_ts = None
+    rn = 0
+    for b in sorted(bks):
+        rn += 1
+        d = bks[b]
+        has0 = 0 in d
+        v0 = d[0][1] if has0 else None
+        bucket_max = max(l for (_, l) in d.values())
+        contribs = []
+        for epoch, (fval, lval) in d.items():
+            if epoch != 0:
+                if epoch in prev_lval:
+                    c = lval - prev_lval[epoch] if lval >= prev_lval[epoch] else lval
+                elif has0 or ever_had0:
+                    seam = v0 if has0 else prev_v0
+                    c = lval - (seam if fval >= seam else 0.0)
+                elif epoch >= adj_start:
+                    c = lval
+                else:
+                    c = lval - fval
+            else:
+                if rn == 1:
+                    c = math.nan
+                elif v0 < prev_bucket_max:
+                    c = v0
+                else:
+                    c = v0 - prev_bucket_max
+            contribs.append(c)
+        finite = [c for c in contribs if not math.isnan(c)]
+        inc = sum(finite) if finite else math.nan
+        if time_agg == "rate" and not math.isnan(inc):
+            denom = (b - prev_bucket_ts) if rn > 1 else step_s
+            inc = inc / denom
+        if b >= display_start // 1000 and not math.isnan(inc):
+            res[b] = inc
+        # state updates (after the whole bucket)
+        for epoch, (_, lval) in d.items():
+            if epoch != 0:
+                prev_lval[epoch] = lval
+        if has0:
+            prev_v0 = v0
+            ever_had0 = True
+        prev_bucket_max = bucket_max
+        prev_bucket_ts = b
+    return res
+
+
+def legacy_spec(points, step_s, adj_start, adj_end, display_start, time_agg):
+    """Current production semantics: bucket max + negative-diff pair rule."""
+    bks = buckets_of(points, step_s, adj_start, adj_end)
+    res = {}
+    prev = None
+    prev_ts = None
+    rn = 0
+    for b in sorted(bks):
+        rn += 1
+        v = max(l for (_, l) in bks[b].values())
+        if rn == 1:
+            val = math.nan
+        elif v < prev:
+            val = v if time_agg == "increase" else v / (b - prev_ts)
+        else:
+            val = (v - prev) if time_agg == "increase" else (v - prev) / (b - prev_ts)
+        if b >= display_start // 1000 and not math.isnan(val):
+            res[b] = val
+        prev = v
+        prev_ts = b
+    return res
+
+
+def point_truth(points, step_s, adj_start, adj_end, display_start):
+    """Per-point epoch-aware increase, attributed to the later point's bucket.
+    Independent of the bucket machinery entirely. Only valid for series whose
+    non-zero epochs are trustworthy (pure-epoch scenarios)."""
+    last = {}
+    res = {}
+    for (t, v, s) in sorted(points):
+        if not (adj_start <= t < adj_end):
+            continue
+        b = (t // 1000) // step_s * step_s
+        c = None
+        if s in last:
+            c = max(0.0, v - last[s])
+        else:
+            if s >= adj_start:
+                c = v  # epoch born inside the fetched range: counts from 0
+            # else: first observation of a pre-range epoch, no attributable growth
+        last[s] = v
+        if c is not None and b >= display_start // 1000:
+            res[b] = res.get(b, 0.0) + c
+    return res
+
+
+# ------------------------------------------------------------------- emit --
+
+CASES = [
+    # name, engine, step_s, time_agg
+    ("e_inc_60_raw", "epoch", 60, "increase"),
+    ("e_inc_300_raw", "epoch", 300, "increase"),
+    ("e_inc_300_agg5m", "epoch", 300, "increase"),
+    ("e_inc_1800_agg30m", "epoch", 1800, "increase"),
+    ("e_inc_86400_raw", "epoch", 86400, "increase"),
+    ("e_inc_86400_agg30m", "epoch", 86400, "increase"),
+    ("e_rate_300_raw", "epoch", 300, "rate"),
+    ("l_inc_60_raw", "legacy", 60, "increase"),
+    ("l_inc_300_raw", "legacy", 300, "increase"),
+    ("l_inc_86400_raw", "legacy", 86400, "increase"),
+    ("l_rate_300_raw", "legacy", 300, "rate"),
+]
+
+
+def query_window(step_s):
+    aligned_start = B - B % (step_s * 1000)
+    adj_start = aligned_start - step_s * 1000
+    adj_end = END - END % (min(step_s, 60) * 1000)
+    return adj_start, adj_end, aligned_start  # display start = aligned start
+
+
+def main(outdir):
+    os.makedirs(os.path.join(outdir, "expected"), exist_ok=True)
+    random.seed(42)
+
+    # ---- inserts.sql
+    lines = []
+    lines.append("SET max_partitions_per_insert_block = 1000;")
+    chunks = []
+    for name, (fp, pts) in SCENARIOS.items():
+        if name in AGG_ONLY:
+            continue
+        rows = [
+            f"('default','Cumulative','{METRIC}',{fp},{t},{v!r},0,{B},{s})"
+            for (t, v, s) in pts
+        ]
+        # 500-row chunks; separate INSERTs create separate parts
+        for i in range(0, len(rows), 500):
+            chunks.append((name, rows[i:i + 500]))
+    # no-recorded-value markers: flags=1, value 0, no epoch
+    s16fp = SCENARIOS["s16_stale"][0]
+    chunks.append(("s16_markers", [
+        f"('default','Cumulative','{METRIC}',{s16fp},{t},0.0,1,{B},0)" for t in S16_MARKERS
+    ]))
+    random.shuffle(chunks)
+    # duplicate one s11 chunk to prove replay-safety
+    dup = next(c for c in chunks if c[0] == "s11_dupooo")
+    chunks.append(dup)
+    for _, rows in chunks:
+        lines.append(
+            "INSERT INTO signoz_metrics.samples_v4 "
+            "(env, temporality, metric_name, fingerprint, unix_milli, value, flags, inserted_at_unix_milli, start_ts) VALUES "
+            + ",".join(rows) + ";"
+        )
+    # pre-migration rollup rows: written straight to the agg tables with empty
+    # epoch maps (the state migration 1012 leaves behind for old data)
+    s15fp, s15pts = SCENARIOS["s15_aggold"]
+    rows5m = [
+        f"('default','Cumulative','{METRIC}',{s15fp},{t},{v!r},{v!r},{v!r},{v!r},1)"
+        for (t, v, _) in s15pts
+    ]
+    # the 30m rows come from the chained samples_v4_agg_30m_mv, which fires on
+    # direct agg_5m inserts exactly as it does for MV-produced ones
+    lines.append(
+        "INSERT INTO signoz_metrics.samples_v4_agg_5m "
+        "(env, temporality, metric_name, fingerprint, unix_milli, last, min, max, sum, count) VALUES "
+        + ",".join(rows5m) + ";"
+    )
+    for name, (fp, _) in SCENARIOS.items():
+        labels = (
+            '{"__name__":"' + METRIC + '","scenario":"' + name + '"}'
+        ).replace("'", "\\'")
+        for day in (B - DAY, B):
+            lines.append(
+                "INSERT INTO signoz_metrics.time_series_v4_1day "
+                "(env, temporality, metric_name, fingerprint, unix_milli, labels) VALUES "
+                f"('default','Cumulative','{METRIC}',{fp},{day},'{labels}');"
+            )
+    with open(os.path.join(outdir, "inserts.sql"), "w") as f:
+        f.write("\n".join(lines) + "\n")
+
+    # ---- expected CSVs
+    engines = {"epoch": bucket_spec, "legacy": legacy_spec}
+    for case, engine, step_s, time_agg in CASES:
+        adj_start, adj_end, display = query_window(step_s)
+        with open(os.path.join(outdir, "expected", case + ".csv"), "w", newline="") as f:
+            w = csv.writer(f)
+            for name in sorted(SCENARIOS):
+                if name in AGG_ONLY and case not in AGG_CASES:
+                    continue
+                fp, pts = SCENARIOS[name]
+                res = engines[engine](pts, step_s, adj_start, adj_end, display, time_agg)
+                for b in sorted(res):
+                    w.writerow([b, name, repr(res[b])])
+
+    # ---- cross-implementation assertions
+    failures = []
+    report = []
+
+    # A1: bucket_spec total == point_truth total on pure-epoch scenarios, per step
+    for step_s in (60, 300, 1800, 86400):
+        adj_start, adj_end, display = query_window(step_s)
+        for name in PURE_EPOCH:
+            fp, pts = SCENARIOS[name]
+            spec = bucket_spec(pts, step_s, adj_start, adj_end, display, "increase")
+            truth = point_truth(pts, step_s, adj_start, adj_end, display)
+            st, tt = sum(spec.values()), sum(truth.values())
+            ok = math.isclose(st, tt, rel_tol=1e-9, abs_tol=1e-6)
+            report.append(f"A1 step={step_s:<6} {name:<18} bucket_spec={st:<12g} point_truth={tt:<12g} {'OK' if ok else 'FAIL'}")
+            if not ok:
+                failures.append(f"A1 {name} step={step_s}: {st} != {tt}")
+            # per-bucket equality as well (min/max per epoch are lossless)
+            for b in sorted(set(spec) | set(truth)):
+                sv, tv = spec.get(b, 0.0), truth.get(b, 0.0)
+                if not math.isclose(sv, tv, rel_tol=1e-9, abs_tol=1e-6):
+                    failures.append(f"A1b {name} step={step_s} ts={b}: bucket_spec={sv} point_truth={tv}")
+
+    # A3: zoom consistency of the epoch rules across steps. Holds when the
+    # one-step lookback covers the series' reporting interval (step >= interval);
+    # below that, the growth between the last pre-range sample and the first
+    # in-range sample is unattributable — for the epoch rules AND for legacy AND
+    # for Prometheus (rate over a window shorter than the scrape interval).
+    # s09_slow reports every 300s, so step=60 is excluded there.
+    zoom_steps = {name: (60, 300, 1800, 86400) for name in PURE_EPOCH}
+    zoom_steps["s09_slow"] = (300, 1800, 86400)
+    totals = {}
+    for name in PURE_EPOCH:
+        fp, pts = SCENARIOS[name]
+        for step_s in zoom_steps[name]:
+            adj_start, adj_end, display = query_window(step_s)
+            res = bucket_spec(pts, step_s, adj_start, adj_end, display, "increase")
+            totals.setdefault(name, {})[step_s] = sum(res.values())
+    for name, per_step in totals.items():
+        vals = list(per_step.values())
+        ok = all(math.isclose(v, vals[0], rel_tol=1e-9, abs_tol=1e-6) for v in vals)
+        report.append(f"A3 zoom {name:<18} totals={ {k: round(v, 6) for k, v in per_step.items()} } {'OK' if ok else 'FAIL'}")
+        if not ok:
+            failures.append(f"A3 {name}: totals differ across steps: {per_step}")
+
+    # A4 (informational): legacy zoom inconsistency on reset scenarios (the Matti bug)
+    for name in ("s03_multireset", "s04_regrow"):
+        fp, pts = SCENARIOS[name]
+        per_step = {}
+        for step_s in (60, 86400):
+            adj_start, adj_end, display = query_window(step_s)
+            per_step[step_s] = sum(legacy_spec(pts, step_s, adj_start, adj_end, display, "increase").values())
+        report.append(f"A4 legacy zoom {name}: step60={per_step[60]:g} step86400={per_step[86400]:g} (inconsistent by design of legacy)")
+
+    # A5: all-key-0 scenarios: epoch rules must reproduce legacy exactly
+    for step_s in (60, 300, 86400):
+        adj_start, adj_end, display = query_window(step_s)
+        for name in ALL_KEY0:
+            fp, pts = SCENARIOS[name]
+            a = bucket_spec(pts, step_s, adj_start, adj_end, display, "increase")
+            b_ = legacy_spec(pts, step_s, adj_start, adj_end, display, "increase")
+            if set(a) != set(b_) or any(not math.isclose(a[k], b_[k], rel_tol=1e-12) for k in a):
+                failures.append(f"A5 {name} step={step_s}: epoch rules != legacy on key-0 data")
+            else:
+                report.append(f"A5 step={step_s:<6} {name:<18} epoch==legacy OK ({len(a)} buckets)")
+
+    # A6: no-reset / gap scenarios: epoch == legacy (regression safety)
+    for step_s in (60, 300):
+        adj_start, adj_end, display = query_window(step_s)
+        for name in ("s01_steady", "s07_gap"):
+            fp, pts = SCENARIOS[name]
+            a = bucket_spec(pts, step_s, adj_start, adj_end, display, "increase")
+            b_ = legacy_spec(pts, step_s, adj_start, adj_end, display, "increase")
+            if set(a) != set(b_) or any(not math.isclose(a[k], b_[k], rel_tol=1e-12) for k in a):
+                failures.append(f"A6 {name} step={step_s}: epoch != legacy on clean data")
+            else:
+                report.append(f"A6 step={step_s:<6} {name:<18} epoch==legacy OK ({len(a)} buckets)")
+
+    with open(os.path.join(outdir, "report.txt"), "w") as f:
+        f.write("\n".join(report) + "\n")
+        if failures:
+            f.write("\nFAILURES:\n" + "\n".join(failures) + "\n")
+
+    print("\n".join(report))
+    if failures:
+        print("\nFAILURES:")
+        print("\n".join(failures))
+        sys.exit(1)
+    print(f"\ngenerated OK: {sum(len(p) for _, (f_, p) in SCENARIOS.items())} samples, {len(CASES)} expected CSVs")
+
+
+if __name__ == "__main__":
+    main(sys.argv[1] if len(sys.argv) > 1 else os.path.dirname(os.path.abspath(__file__)))

tests/integration/testdata/counter_reset_epochs/queries/e_inc_1800_agg30m.sql

@@ -0,0 +1,41 @@
+-- e_inc_1800_agg30m (epochs=true step=30m0s agg=increase)
+SELECT toUnixTimestamp(ts) AS ts_s, `scenario`, value FROM (
+WITH __temporal_aggregation_cte AS (SELECT
+  ts,
+  `scenario`, if(countIf(isNaN(__contrib) = 0) > 0, sumIf(__contrib, isNaN(__contrib) = 0), nan) AS per_series_value
+FROM (SELECT
+  fingerprint,
+  ts,
+  `scenario`, __prev_bucket_ts,
+  __bucket_rn,
+  __kv.1 AS __epoch,
+  __kv.2 AS __lval,
+  __first_by_epoch[__kv.1] AS __fval,
+  lagInFrame(__lval, 1) OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_lval,
+  row_number() OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts) AS __epoch_rn,
+  multiIf(
+    __epoch != 0 AND __epoch_rn > 1, if(__lval >= __prev_lval, __lval - __prev_lval, __lval),
+    __epoch != 0 AND (__has0 OR __ever_had0), __lval - if(__fval >= if(__has0, __v0, __prev_v0), if(__has0, __v0, __prev_v0), 0.),
+    __epoch != 0 AND __epoch >= 1784331000000, __lval,
+    __epoch != 0, __lval - __fval,
+    __bucket_rn = 1, nan,
+    __lval < __prev_bucket_max, __lval,
+    __lval - __prev_bucket_max
+  ) AS __contrib
+FROM (SELECT
+  fingerprint,
+  ts,
+  `scenario`, __first_by_epoch,
+  __last_by_epoch,
+  mapContains(__last_by_epoch, toInt64(0)) AS __has0,
+  __last_by_epoch[toInt64(0)] AS __v0,
+  anyLastIf(__v0, __has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __prev_v0,
+  max(__has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __ever_had0,
+  arrayMax(mapValues(__last_by_epoch)) AS __bucket_max,
+  lagInFrame(__bucket_max, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_max,
+  lagInFrame(ts, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_ts,
+  row_number() OVER (PARTITION BY fingerprint ORDER BY ts) AS __bucket_rn
+FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(1800)) AS ts, `scenario`, minMap(min_value_by_start_ts) AS __first_map_merged, maxMap(max_value_by_start_ts) AS __last_map_merged, if(length(__first_map_merged) = 0, map(toInt64(0), min(min)), __first_map_merged) AS __first_by_epoch, if(length(__last_map_merged) = 0, map(toInt64(0), max(max)), __last_map_merged) AS __last_by_epoch FROM signoz_metrics.distributed_samples_v4_agg_30m AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'scenario') AS `scenario` FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784246400000 AND unix_milli <= 1784419200000 AND LOWER(temporality) LIKE LOWER('cumulative') GROUP BY fingerprint, `scenario`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784331000000 AND unix_milli < 1784419200000 GROUP BY fingerprint, ts, `scenario` ORDER BY fingerprint, ts)) ARRAY JOIN arrayZip(mapKeys(__last_by_epoch), mapValues(__last_by_epoch)) AS __kv)
+WHERE ts >= toDateTime(1784332800)
+GROUP BY fingerprint, ts, `scenario`), __spatial_aggregation_cte AS (SELECT ts, `scenario`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = 0 GROUP BY ts, `scenario`) SELECT * FROM __spatial_aggregation_cte ORDER BY `scenario`, ts
+) ORDER BY scenario, ts_s FORMAT CSV;

tests/integration/testdata/counter_reset_epochs/queries/e_inc_300_agg5m.sql

@@ -0,0 +1,41 @@
+-- e_inc_300_agg5m (epochs=true step=5m0s agg=increase)
+SELECT toUnixTimestamp(ts) AS ts_s, `scenario`, value FROM (
+WITH __temporal_aggregation_cte AS (SELECT
+  ts,
+  `scenario`, if(countIf(isNaN(__contrib) = 0) > 0, sumIf(__contrib, isNaN(__contrib) = 0), nan) AS per_series_value
+FROM (SELECT
+  fingerprint,
+  ts,
+  `scenario`, __prev_bucket_ts,
+  __bucket_rn,
+  __kv.1 AS __epoch,
+  __kv.2 AS __lval,
+  __first_by_epoch[__kv.1] AS __fval,
+  lagInFrame(__lval, 1) OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_lval,
+  row_number() OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts) AS __epoch_rn,
+  multiIf(
+    __epoch != 0 AND __epoch_rn > 1, if(__lval >= __prev_lval, __lval - __prev_lval, __lval),
+    __epoch != 0 AND (__has0 OR __ever_had0), __lval - if(__fval >= if(__has0, __v0, __prev_v0), if(__has0, __v0, __prev_v0), 0.),
+    __epoch != 0 AND __epoch >= 1784332500000, __lval,
+    __epoch != 0, __lval - __fval,
+    __bucket_rn = 1, nan,
+    __lval < __prev_bucket_max, __lval,
+    __lval - __prev_bucket_max
+  ) AS __contrib
+FROM (SELECT
+  fingerprint,
+  ts,
+  `scenario`, __first_by_epoch,
+  __last_by_epoch,
+  mapContains(__last_by_epoch, toInt64(0)) AS __has0,
+  __last_by_epoch[toInt64(0)] AS __v0,
+  anyLastIf(__v0, __has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __prev_v0,
+  max(__has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __ever_had0,
+  arrayMax(mapValues(__last_by_epoch)) AS __bucket_max,
+  lagInFrame(__bucket_max, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_max,
+  lagInFrame(ts, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_ts,
+  row_number() OVER (PARTITION BY fingerprint ORDER BY ts) AS __bucket_rn
+FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, `scenario`, minMap(min_value_by_start_ts) AS __first_map_merged, maxMap(max_value_by_start_ts) AS __last_map_merged, if(length(__first_map_merged) = 0, map(toInt64(0), min(min)), __first_map_merged) AS __first_by_epoch, if(length(__last_map_merged) = 0, map(toInt64(0), max(max)), __last_map_merged) AS __last_by_epoch FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'scenario') AS `scenario` FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784246400000 AND unix_milli <= 1784419200000 AND LOWER(temporality) LIKE LOWER('cumulative') GROUP BY fingerprint, `scenario`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784332500000 AND unix_milli < 1784419200000 GROUP BY fingerprint, ts, `scenario` ORDER BY fingerprint, ts)) ARRAY JOIN arrayZip(mapKeys(__last_by_epoch), mapValues(__last_by_epoch)) AS __kv)
+WHERE ts >= toDateTime(1784332800)
+GROUP BY fingerprint, ts, `scenario`), __spatial_aggregation_cte AS (SELECT ts, `scenario`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = 0 GROUP BY ts, `scenario`) SELECT * FROM __spatial_aggregation_cte ORDER BY `scenario`, ts
+) ORDER BY scenario, ts_s FORMAT CSV;

tests/integration/testdata/counter_reset_epochs/queries/e_inc_300_raw.sql

@@ -0,0 +1,41 @@
+-- e_inc_300_raw (epochs=true step=5m0s agg=increase)
+SELECT toUnixTimestamp(ts) AS ts_s, `scenario`, value FROM (
+WITH __temporal_aggregation_cte AS (SELECT
+  ts,
+  `scenario`, if(countIf(isNaN(__contrib) = 0) > 0, sumIf(__contrib, isNaN(__contrib) = 0), nan) AS per_series_value
+FROM (SELECT
+  fingerprint,
+  ts,
+  `scenario`, __prev_bucket_ts,
+  __bucket_rn,
+  __kv.1 AS __epoch,
+  __kv.2 AS __lval,
+  __first_by_epoch[__kv.1] AS __fval,
+  lagInFrame(__lval, 1) OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_lval,
+  row_number() OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts) AS __epoch_rn,
+  multiIf(
+    __epoch != 0 AND __epoch_rn > 1, if(__lval >= __prev_lval, __lval - __prev_lval, __lval),
+    __epoch != 0 AND (__has0 OR __ever_had0), __lval - if(__fval >= if(__has0, __v0, __prev_v0), if(__has0, __v0, __prev_v0), 0.),
+    __epoch != 0 AND __epoch >= 1784332500000, __lval,
+    __epoch != 0, __lval - __fval,
+    __bucket_rn = 1, nan,
+    __lval < __prev_bucket_max, __lval,
+    __lval - __prev_bucket_max
+  ) AS __contrib
+FROM (SELECT
+  fingerprint,
+  ts,
+  `scenario`, __first_by_epoch,
+  __last_by_epoch,
+  mapContains(__last_by_epoch, toInt64(0)) AS __has0,
+  __last_by_epoch[toInt64(0)] AS __v0,
+  anyLastIf(__v0, __has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __prev_v0,
+  max(__has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __ever_had0,
+  arrayMax(mapValues(__last_by_epoch)) AS __bucket_max,
+  lagInFrame(__bucket_max, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_max,
+  lagInFrame(ts, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_ts,
+  row_number() OVER (PARTITION BY fingerprint ORDER BY ts) AS __bucket_rn
+FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, `scenario`, minMap(map(start_ts, value)) AS __first_by_epoch, maxMap(map(start_ts, value)) AS __last_by_epoch FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'scenario') AS `scenario` FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784246400000 AND unix_milli <= 1784419200000 AND LOWER(temporality) LIKE LOWER('cumulative') GROUP BY fingerprint, `scenario`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784332500000 AND unix_milli < 1784419200000 AND bitAnd(flags, 1) = 0 GROUP BY fingerprint, ts, `scenario` ORDER BY fingerprint, ts)) ARRAY JOIN arrayZip(mapKeys(__last_by_epoch), mapValues(__last_by_epoch)) AS __kv)
+WHERE ts >= toDateTime(1784332800)
+GROUP BY fingerprint, ts, `scenario`), __spatial_aggregation_cte AS (SELECT ts, `scenario`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = 0 GROUP BY ts, `scenario`) SELECT * FROM __spatial_aggregation_cte ORDER BY `scenario`, ts
+) ORDER BY scenario, ts_s FORMAT CSV;

tests/integration/testdata/counter_reset_epochs/queries/e_inc_60_raw.sql

@@ -0,0 +1,41 @@
+-- e_inc_60_raw (epochs=true step=1m0s agg=increase)
+SELECT toUnixTimestamp(ts) AS ts_s, `scenario`, value FROM (
+WITH __temporal_aggregation_cte AS (SELECT
+  ts,
+  `scenario`, if(countIf(isNaN(__contrib) = 0) > 0, sumIf(__contrib, isNaN(__contrib) = 0), nan) AS per_series_value
+FROM (SELECT
+  fingerprint,
+  ts,
+  `scenario`, __prev_bucket_ts,
+  __bucket_rn,
+  __kv.1 AS __epoch,
+  __kv.2 AS __lval,
+  __first_by_epoch[__kv.1] AS __fval,
+  lagInFrame(__lval, 1) OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_lval,
+  row_number() OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts) AS __epoch_rn,
+  multiIf(
+    __epoch != 0 AND __epoch_rn > 1, if(__lval >= __prev_lval, __lval - __prev_lval, __lval),
+    __epoch != 0 AND (__has0 OR __ever_had0), __lval - if(__fval >= if(__has0, __v0, __prev_v0), if(__has0, __v0, __prev_v0), 0.),
+    __epoch != 0 AND __epoch >= 1784332740000, __lval,
+    __epoch != 0, __lval - __fval,
+    __bucket_rn = 1, nan,
+    __lval < __prev_bucket_max, __lval,
+    __lval - __prev_bucket_max
+  ) AS __contrib
+FROM (SELECT
+  fingerprint,
+  ts,
+  `scenario`, __first_by_epoch,
+  __last_by_epoch,
+  mapContains(__last_by_epoch, toInt64(0)) AS __has0,
+  __last_by_epoch[toInt64(0)] AS __v0,
+  anyLastIf(__v0, __has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __prev_v0,
+  max(__has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __ever_had0,
+  arrayMax(mapValues(__last_by_epoch)) AS __bucket_max,
+  lagInFrame(__bucket_max, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_max,
+  lagInFrame(ts, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_ts,
+  row_number() OVER (PARTITION BY fingerprint ORDER BY ts) AS __bucket_rn
+FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(60)) AS ts, `scenario`, minMap(map(start_ts, value)) AS __first_by_epoch, maxMap(map(start_ts, value)) AS __last_by_epoch FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'scenario') AS `scenario` FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784246400000 AND unix_milli <= 1784419200000 AND LOWER(temporality) LIKE LOWER('cumulative') GROUP BY fingerprint, `scenario`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784332740000 AND unix_milli < 1784419200000 AND bitAnd(flags, 1) = 0 GROUP BY fingerprint, ts, `scenario` ORDER BY fingerprint, ts)) ARRAY JOIN arrayZip(mapKeys(__last_by_epoch), mapValues(__last_by_epoch)) AS __kv)
+WHERE ts >= toDateTime(1784332800)
+GROUP BY fingerprint, ts, `scenario`), __spatial_aggregation_cte AS (SELECT ts, `scenario`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = 0 GROUP BY ts, `scenario`) SELECT * FROM __spatial_aggregation_cte ORDER BY `scenario`, ts
+) ORDER BY scenario, ts_s FORMAT CSV;

tests/integration/testdata/counter_reset_epochs/queries/e_inc_86400_agg30m.sql

@@ -0,0 +1,41 @@
+-- e_inc_86400_agg30m (epochs=true step=24h0m0s agg=increase)
+SELECT toUnixTimestamp(ts) AS ts_s, `scenario`, value FROM (
+WITH __temporal_aggregation_cte AS (SELECT
+  ts,
+  `scenario`, if(countIf(isNaN(__contrib) = 0) > 0, sumIf(__contrib, isNaN(__contrib) = 0), nan) AS per_series_value
+FROM (SELECT
+  fingerprint,
+  ts,
+  `scenario`, __prev_bucket_ts,
+  __bucket_rn,
+  __kv.1 AS __epoch,
+  __kv.2 AS __lval,
+  __first_by_epoch[__kv.1] AS __fval,
+  lagInFrame(__lval, 1) OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_lval,
+  row_number() OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts) AS __epoch_rn,
+  multiIf(
+    __epoch != 0 AND __epoch_rn > 1, if(__lval >= __prev_lval, __lval - __prev_lval, __lval),
+    __epoch != 0 AND (__has0 OR __ever_had0), __lval - if(__fval >= if(__has0, __v0, __prev_v0), if(__has0, __v0, __prev_v0), 0.),
+    __epoch != 0 AND __epoch >= 1784246400000, __lval,
+    __epoch != 0, __lval - __fval,
+    __bucket_rn = 1, nan,
+    __lval < __prev_bucket_max, __lval,
+    __lval - __prev_bucket_max
+  ) AS __contrib
+FROM (SELECT
+  fingerprint,
+  ts,
+  `scenario`, __first_by_epoch,
+  __last_by_epoch,
+  mapContains(__last_by_epoch, toInt64(0)) AS __has0,
+  __last_by_epoch[toInt64(0)] AS __v0,
+  anyLastIf(__v0, __has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __prev_v0,
+  max(__has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __ever_had0,
+  arrayMax(mapValues(__last_by_epoch)) AS __bucket_max,
+  lagInFrame(__bucket_max, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_max,
+  lagInFrame(ts, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_ts,
+  row_number() OVER (PARTITION BY fingerprint ORDER BY ts) AS __bucket_rn
+FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(86400)) AS ts, `scenario`, minMap(min_value_by_start_ts) AS __first_map_merged, maxMap(max_value_by_start_ts) AS __last_map_merged, if(length(__first_map_merged) = 0, map(toInt64(0), min(min)), __first_map_merged) AS __first_by_epoch, if(length(__last_map_merged) = 0, map(toInt64(0), max(max)), __last_map_merged) AS __last_by_epoch FROM signoz_metrics.distributed_samples_v4_agg_30m AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'scenario') AS `scenario` FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784246400000 AND unix_milli <= 1784419200000 AND LOWER(temporality) LIKE LOWER('cumulative') GROUP BY fingerprint, `scenario`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784246400000 AND unix_milli < 1784419200000 GROUP BY fingerprint, ts, `scenario` ORDER BY fingerprint, ts)) ARRAY JOIN arrayZip(mapKeys(__last_by_epoch), mapValues(__last_by_epoch)) AS __kv)
+WHERE ts >= toDateTime(1784332800)
+GROUP BY fingerprint, ts, `scenario`), __spatial_aggregation_cte AS (SELECT ts, `scenario`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = 0 GROUP BY ts, `scenario`) SELECT * FROM __spatial_aggregation_cte ORDER BY `scenario`, ts
+) ORDER BY scenario, ts_s FORMAT CSV;

tests/integration/testdata/counter_reset_epochs/queries/e_inc_86400_raw.sql

@@ -0,0 +1,41 @@
+-- e_inc_86400_raw (epochs=true step=24h0m0s agg=increase)
+SELECT toUnixTimestamp(ts) AS ts_s, `scenario`, value FROM (
+WITH __temporal_aggregation_cte AS (SELECT
+  ts,
+  `scenario`, if(countIf(isNaN(__contrib) = 0) > 0, sumIf(__contrib, isNaN(__contrib) = 0), nan) AS per_series_value
+FROM (SELECT
+  fingerprint,
+  ts,
+  `scenario`, __prev_bucket_ts,
+  __bucket_rn,
+  __kv.1 AS __epoch,
+  __kv.2 AS __lval,
+  __first_by_epoch[__kv.1] AS __fval,
+  lagInFrame(__lval, 1) OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_lval,
+  row_number() OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts) AS __epoch_rn,
+  multiIf(
+    __epoch != 0 AND __epoch_rn > 1, if(__lval >= __prev_lval, __lval - __prev_lval, __lval),
+    __epoch != 0 AND (__has0 OR __ever_had0), __lval - if(__fval >= if(__has0, __v0, __prev_v0), if(__has0, __v0, __prev_v0), 0.),
+    __epoch != 0 AND __epoch >= 1784246400000, __lval,
+    __epoch != 0, __lval - __fval,
+    __bucket_rn = 1, nan,
+    __lval < __prev_bucket_max, __lval,
+    __lval - __prev_bucket_max
+  ) AS __contrib
+FROM (SELECT
+  fingerprint,
+  ts,
+  `scenario`, __first_by_epoch,
+  __last_by_epoch,
+  mapContains(__last_by_epoch, toInt64(0)) AS __has0,
+  __last_by_epoch[toInt64(0)] AS __v0,
+  anyLastIf(__v0, __has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __prev_v0,
+  max(__has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __ever_had0,
+  arrayMax(mapValues(__last_by_epoch)) AS __bucket_max,
+  lagInFrame(__bucket_max, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_max,
+  lagInFrame(ts, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_ts,
+  row_number() OVER (PARTITION BY fingerprint ORDER BY ts) AS __bucket_rn
+FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(86400)) AS ts, `scenario`, minMap(min_value_by_start_ts) AS __first_map_merged, maxMap(max_value_by_start_ts) AS __last_map_merged, if(length(__first_map_merged) = 0, map(toInt64(0), min(min)), __first_map_merged) AS __first_by_epoch, if(length(__last_map_merged) = 0, map(toInt64(0), max(max)), __last_map_merged) AS __last_by_epoch FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'scenario') AS `scenario` FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784246400000 AND unix_milli <= 1784419200000 AND LOWER(temporality) LIKE LOWER('cumulative') GROUP BY fingerprint, `scenario`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784246400000 AND unix_milli < 1784419200000 GROUP BY fingerprint, ts, `scenario` ORDER BY fingerprint, ts)) ARRAY JOIN arrayZip(mapKeys(__last_by_epoch), mapValues(__last_by_epoch)) AS __kv)
+WHERE ts >= toDateTime(1784332800)
+GROUP BY fingerprint, ts, `scenario`), __spatial_aggregation_cte AS (SELECT ts, `scenario`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = 0 GROUP BY ts, `scenario`) SELECT * FROM __spatial_aggregation_cte ORDER BY `scenario`, ts
+) ORDER BY scenario, ts_s FORMAT CSV;

tests/integration/testdata/counter_reset_epochs/queries/e_rate_300_raw.sql

@@ -0,0 +1,45 @@
+-- e_rate_300_raw (epochs=true step=5m0s agg=rate)
+SELECT toUnixTimestamp(ts) AS ts_s, `scenario`, value FROM (
+WITH __temporal_aggregation_cte AS (SELECT
+  ts,
+  `scenario`, if(
+    countIf(isNaN(__contrib) = 0) > 0,
+    sumIf(__contrib, isNaN(__contrib) = 0) / if(any(__bucket_rn) > 1, ts - any(__prev_bucket_ts), 300),
+    nan
+  ) AS per_series_value
+FROM (SELECT
+  fingerprint,
+  ts,
+  `scenario`, __prev_bucket_ts,
+  __bucket_rn,
+  __kv.1 AS __epoch,
+  __kv.2 AS __lval,
+  __first_by_epoch[__kv.1] AS __fval,
+  lagInFrame(__lval, 1) OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_lval,
+  row_number() OVER (PARTITION BY fingerprint, __kv.1 ORDER BY ts) AS __epoch_rn,
+  multiIf(
+    __epoch != 0 AND __epoch_rn > 1, if(__lval >= __prev_lval, __lval - __prev_lval, __lval),
+    __epoch != 0 AND (__has0 OR __ever_had0), __lval - if(__fval >= if(__has0, __v0, __prev_v0), if(__has0, __v0, __prev_v0), 0.),
+    __epoch != 0 AND __epoch >= 1784332500000, __lval,
+    __epoch != 0, __lval - __fval,
+    __bucket_rn = 1, nan,
+    __lval < __prev_bucket_max, __lval,
+    __lval - __prev_bucket_max
+  ) AS __contrib
+FROM (SELECT
+  fingerprint,
+  ts,
+  `scenario`, __first_by_epoch,
+  __last_by_epoch,
+  mapContains(__last_by_epoch, toInt64(0)) AS __has0,
+  __last_by_epoch[toInt64(0)] AS __v0,
+  anyLastIf(__v0, __has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __prev_v0,
+  max(__has0) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS __ever_had0,
+  arrayMax(mapValues(__last_by_epoch)) AS __bucket_max,
+  lagInFrame(__bucket_max, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_max,
+  lagInFrame(ts, 1) OVER (PARTITION BY fingerprint ORDER BY ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS __prev_bucket_ts,
+  row_number() OVER (PARTITION BY fingerprint ORDER BY ts) AS __bucket_rn
+FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, `scenario`, minMap(map(start_ts, value)) AS __first_by_epoch, maxMap(map(start_ts, value)) AS __last_by_epoch FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'scenario') AS `scenario` FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784246400000 AND unix_milli <= 1784419200000 AND LOWER(temporality) LIKE LOWER('cumulative') GROUP BY fingerprint, `scenario`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784332500000 AND unix_milli < 1784419200000 AND bitAnd(flags, 1) = 0 GROUP BY fingerprint, ts, `scenario` ORDER BY fingerprint, ts)) ARRAY JOIN arrayZip(mapKeys(__last_by_epoch), mapValues(__last_by_epoch)) AS __kv)
+WHERE ts >= toDateTime(1784332800)
+GROUP BY fingerprint, ts, `scenario`), __spatial_aggregation_cte AS (SELECT ts, `scenario`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = 0 GROUP BY ts, `scenario`) SELECT * FROM __spatial_aggregation_cte ORDER BY `scenario`, ts
+) ORDER BY scenario, ts_s FORMAT CSV;

tests/integration/testdata/counter_reset_epochs/queries/l_inc_300_raw.sql

@@ -0,0 +1,4 @@
+-- l_inc_300_raw (epochs=false step=5m0s agg=increase)
+SELECT toUnixTimestamp(ts) AS ts_s, `scenario`, value FROM (
+WITH __temporal_aggregation_cte AS (SELECT ts, `scenario`, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value, per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, `scenario`, max(value) AS per_series_value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'scenario') AS `scenario` FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784246400000 AND unix_milli <= 1784419200000 AND LOWER(temporality) LIKE LOWER('cumulative') GROUP BY fingerprint, `scenario`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784332500000 AND unix_milli < 1784419200000 GROUP BY fingerprint, ts, `scenario` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `scenario`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = 0 GROUP BY ts, `scenario`) SELECT * FROM __spatial_aggregation_cte ORDER BY `scenario`, ts
+) ORDER BY scenario, ts_s FORMAT CSV;

tests/integration/testdata/counter_reset_epochs/queries/l_inc_60_raw.sql

@@ -0,0 +1,4 @@
+-- l_inc_60_raw (epochs=false step=1m0s agg=increase)
+SELECT toUnixTimestamp(ts) AS ts_s, `scenario`, value FROM (
+WITH __temporal_aggregation_cte AS (SELECT ts, `scenario`, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value, per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(60)) AS ts, `scenario`, max(value) AS per_series_value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'scenario') AS `scenario` FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784246400000 AND unix_milli <= 1784419200000 AND LOWER(temporality) LIKE LOWER('cumulative') GROUP BY fingerprint, `scenario`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784332740000 AND unix_milli < 1784419200000 GROUP BY fingerprint, ts, `scenario` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `scenario`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = 0 GROUP BY ts, `scenario`) SELECT * FROM __spatial_aggregation_cte ORDER BY `scenario`, ts
+) ORDER BY scenario, ts_s FORMAT CSV;

tests/integration/testdata/counter_reset_epochs/queries/l_inc_86400_raw.sql

@@ -0,0 +1,4 @@
+-- l_inc_86400_raw (epochs=false step=24h0m0s agg=increase)
+SELECT toUnixTimestamp(ts) AS ts_s, `scenario`, value FROM (
+WITH __temporal_aggregation_cte AS (SELECT ts, `scenario`, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value, per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(86400)) AS ts, `scenario`, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'scenario') AS `scenario` FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784246400000 AND unix_milli <= 1784419200000 AND LOWER(temporality) LIKE LOWER('cumulative') GROUP BY fingerprint, `scenario`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784246400000 AND unix_milli < 1784419200000 GROUP BY fingerprint, ts, `scenario` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `scenario`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = 0 GROUP BY ts, `scenario`) SELECT * FROM __spatial_aggregation_cte ORDER BY `scenario`, ts
+) ORDER BY scenario, ts_s FORMAT CSV;

tests/integration/testdata/counter_reset_epochs/queries/l_rate_300_raw.sql

@@ -0,0 +1,4 @@
+-- l_rate_300_raw (epochs=false step=5m0s agg=rate)
+SELECT toUnixTimestamp(ts) AS ts_s, `scenario`, value FROM (
+WITH __temporal_aggregation_cte AS (SELECT ts, `scenario`, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, `scenario`, max(value) AS per_series_value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'scenario') AS `scenario` FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784246400000 AND unix_milli <= 1784419200000 AND LOWER(temporality) LIKE LOWER('cumulative') GROUP BY fingerprint, `scenario`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN ('it_counter_total') AND unix_milli >= 1784332500000 AND unix_milli < 1784419200000 GROUP BY fingerprint, ts, `scenario` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `scenario`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = 0 GROUP BY ts, `scenario`) SELECT * FROM __spatial_aggregation_cte ORDER BY `scenario`, ts
+) ORDER BY scenario, ts_s FORMAT CSV;

tests/integration/testdata/counter_reset_epochs/report.txt

@@ -0,0 +1,67 @@
+A1 step=60     s01_steady         bucket_spec=5760         point_truth=5760         OK
+A1 step=60     s02_midreset       bucket_spec=8644         point_truth=8644         OK
+A1 step=60     s03_multireset     bucket_spec=3969         point_truth=3969         OK
+A1 step=60     s04_regrow         bucket_spec=43705        point_truth=43705        OK
+A1 step=60     s07_gap            bucket_spec=0            point_truth=0            OK
+A1 step=60     s08_singlepoint    bucket_spec=42           point_truth=42           OK
+A1 step=60     s09_slow           bucket_spec=2864         point_truth=2864         OK
+A1 step=60     s10_boundaryreset  bucket_spec=11518        point_truth=11518        OK
+A1 step=60     s11_dupooo         bucket_spec=5764         point_truth=5764         OK
+A1 step=60     s12_twowriter      bucket_spec=8640         point_truth=8640         OK
+A1 step=60     s16_stale          bucket_spec=5760         point_truth=5760         OK
+A1 step=300    s01_steady         bucket_spec=5760         point_truth=5760         OK
+A1 step=300    s02_midreset       bucket_spec=8644         point_truth=8644         OK
+A1 step=300    s03_multireset     bucket_spec=3969         point_truth=3969         OK
+A1 step=300    s04_regrow         bucket_spec=43705        point_truth=43705        OK
+A1 step=300    s07_gap            bucket_spec=0            point_truth=0            OK
+A1 step=300    s08_singlepoint    bucket_spec=42           point_truth=42           OK
+A1 step=300    s09_slow           bucket_spec=2874         point_truth=2874         OK
+A1 step=300    s10_boundaryreset  bucket_spec=11518        point_truth=11518        OK
+A1 step=300    s11_dupooo         bucket_spec=5764         point_truth=5764         OK
+A1 step=300    s12_twowriter      bucket_spec=8640         point_truth=8640         OK
+A1 step=300    s16_stale          bucket_spec=5760         point_truth=5760         OK
+A1 step=1800   s01_steady         bucket_spec=5760         point_truth=5760         OK
+A1 step=1800   s02_midreset       bucket_spec=8644         point_truth=8644         OK
+A1 step=1800   s03_multireset     bucket_spec=3969         point_truth=3969         OK
+A1 step=1800   s04_regrow         bucket_spec=43705        point_truth=43705        OK
+A1 step=1800   s07_gap            bucket_spec=0            point_truth=0            OK
+A1 step=1800   s08_singlepoint    bucket_spec=42           point_truth=42           OK
+A1 step=1800   s09_slow           bucket_spec=2874         point_truth=2874         OK
+A1 step=1800   s10_boundaryreset  bucket_spec=11518        point_truth=11518        OK
+A1 step=1800   s11_dupooo         bucket_spec=5764         point_truth=5764         OK
+A1 step=1800   s12_twowriter      bucket_spec=8640         point_truth=8640         OK
+A1 step=1800   s16_stale          bucket_spec=5760         point_truth=5760         OK
+A1 step=86400  s01_steady         bucket_spec=5760         point_truth=5760         OK
+A1 step=86400  s02_midreset       bucket_spec=8644         point_truth=8644         OK
+A1 step=86400  s03_multireset     bucket_spec=3969         point_truth=3969         OK
+A1 step=86400  s04_regrow         bucket_spec=43705        point_truth=43705        OK
+A1 step=86400  s07_gap            bucket_spec=0            point_truth=0            OK
+A1 step=86400  s08_singlepoint    bucket_spec=42           point_truth=42           OK
+A1 step=86400  s09_slow           bucket_spec=2874         point_truth=2874         OK
+A1 step=86400  s10_boundaryreset  bucket_spec=11518        point_truth=11518        OK
+A1 step=86400  s11_dupooo         bucket_spec=5764         point_truth=5764         OK
+A1 step=86400  s12_twowriter      bucket_spec=8640         point_truth=8640         OK
+A1 step=86400  s16_stale          bucket_spec=5760         point_truth=5760         OK
+A3 zoom s01_steady         totals={60: 5760.0, 300: 5760.0, 1800: 5760.0, 86400: 5760.0} OK
+A3 zoom s02_midreset       totals={60: 8644.0, 300: 8644.0, 1800: 8644.0, 86400: 8644.0} OK
+A3 zoom s03_multireset     totals={60: 3969.0, 300: 3969.0, 1800: 3969.0, 86400: 3969.0} OK
+A3 zoom s04_regrow         totals={60: 43705.0, 300: 43705.0, 1800: 43705.0, 86400: 43705.0} OK
+A3 zoom s07_gap            totals={60: 0.0, 300: 0.0, 1800: 0.0, 86400: 0.0} OK
+A3 zoom s08_singlepoint    totals={60: 42.0, 300: 42.0, 1800: 42.0, 86400: 42.0} OK
+A3 zoom s09_slow           totals={300: 2874.0, 1800: 2874.0, 86400: 2874.0} OK
+A3 zoom s10_boundaryreset  totals={60: 11518.0, 300: 11518.0, 1800: 11518.0, 86400: 11518.0} OK
+A3 zoom s11_dupooo         totals={60: 5764.0, 300: 5764.0, 1800: 5764.0, 86400: 5764.0} OK
+A3 zoom s12_twowriter      totals={60: 8640.0, 300: 8640.0, 1800: 8640.0, 86400: 8640.0} OK
+A3 zoom s16_stale          totals={60: 5760.0, 300: 5760.0, 1800: 5760.0, 86400: 5760.0} OK
+A4 legacy zoom s03_multireset: step60=3969 step86400=242 (inconsistent by design of legacy)
+A4 legacy zoom s04_regrow: step60=39605 step86400=39605 (inconsistent by design of legacy)
+A5 step=60     s05_legacy         epoch==legacy OK (1440 buckets)
+A5 step=60     s14_updown         epoch==legacy OK (1440 buckets)
+A5 step=300    s05_legacy         epoch==legacy OK (288 buckets)
+A5 step=300    s14_updown         epoch==legacy OK (288 buckets)
+A5 step=86400  s05_legacy         epoch==legacy OK (1 buckets)
+A5 step=86400  s14_updown         epoch==legacy OK (1 buckets)
+A6 step=60     s01_steady         epoch==legacy OK (1440 buckets)
+A6 step=60     s07_gap            epoch==legacy OK (1430 buckets)
+A6 step=300    s01_steady         epoch==legacy OK (288 buckets)
+A6 step=300    s07_gap            epoch==legacy OK (286 buckets)

tests/integration/testdata/counter_reset_epochs/run.sh

@@ -0,0 +1,62 @@
+#!/bin/bash
+# Epoch harness: real schema + MVs, one-day dataset, builder-emitted SQL,
+# expected-vs-actual comparison (pre- and post-merge).
+set -u
+cd "$(dirname "$0")"
+
+DB_DIR="chdata"
+CH="clickhouse local --path $DB_DIR"
+
+echo "== reset =="
+rm -rf "$DB_DIR" actual && mkdir -p actual
+
+echo "== schema =="
+$CH --queries-file schema.sql || exit 1
+
+echo "== inserts (MVs fire per insert block) =="
+$CH --queries-file inserts.sql || exit 1
+
+$CH -q "SELECT 'samples', count() FROM signoz_metrics.samples_v4
+UNION ALL SELECT 'agg_5m rows', count() FROM signoz_metrics.samples_v4_agg_5m
+UNION ALL SELECT 'agg_30m rows', count() FROM signoz_metrics.samples_v4_agg_30m"
+
+run_queries() {
+  local phase="$1"
+  local fails=0
+  for f in queries/*.sql; do
+    name="$(basename "$f" .sql)"
+    out="actual/${name}_${phase}.csv"
+    if ! $CH --queries-file "$f" > "$out" 2> "actual/${name}_${phase}.err"; then
+      echo "  $name: QUERY ERROR"
+      sed 's/^/    /' "actual/${name}_${phase}.err" | head -5
+      fails=$((fails+1))
+      continue
+    fi
+    printf "  %-22s %s " "$name" "$phase"
+    if ! python3 compare.py "expected/${name}.csv" "$out"; then
+      fails=$((fails+1))
+    fi
+  done
+  return $fails
+}
+
+echo "== compare (pre-merge parts) =="
+run_queries premerge
+pre_fails=$?
+
+echo "== optimize final (forced merges of agg states) =="
+$CH -q "OPTIMIZE TABLE signoz_metrics.samples_v4_agg_5m FINAL;
+OPTIMIZE TABLE signoz_metrics.samples_v4_agg_30m FINAL;
+OPTIMIZE TABLE signoz_metrics.samples_v4 FINAL;"
+
+echo "== compare (post-merge) =="
+run_queries postmerge
+post_fails=$?
+
+echo
+if [ $((pre_fails + post_fails)) -eq 0 ]; then
+  echo "ALL COMPARISONS PASSED"
+else
+  echo "FAILURES: pre=$pre_fails post=$post_fails"
+  exit 1
+fi

tests/integration/testdata/counter_reset_epochs/schema.sql

@@ -0,0 +1,152 @@
+-- Epoch harness schema: the real signoz_metrics tables after collector
+-- migration 1012, with distributed_* as plain views (single-node harness).
+-- MV queries are verbatim copies of migration 1012.
+
+CREATE DATABASE IF NOT EXISTS signoz_metrics;
+
+CREATE TABLE signoz_metrics.samples_v4
+(
+    env LowCardinality(String) DEFAULT 'default',
+    temporality LowCardinality(String) DEFAULT 'Unspecified',
+    metric_name LowCardinality(String),
+    fingerprint UInt64 CODEC(Delta(8), ZSTD(1)),
+    unix_milli Int64 CODEC(DoubleDelta, ZSTD(1)),
+    value Float64 CODEC(Gorilla, ZSTD(1)),
+    flags UInt32 DEFAULT 0 CODEC(ZSTD(1)),
+    inserted_at_unix_milli Int64 CODEC(ZSTD(1)),
+    start_ts Int64 DEFAULT 0 CODEC(DoubleDelta, ZSTD(1))
+)
+ENGINE = MergeTree
+PARTITION BY toDate(unix_milli / 1000)
+ORDER BY (env, temporality, metric_name, fingerprint, unix_milli)
+TTL toDateTime(unix_milli / 1000) + toIntervalSecond(2592000)
+SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1;
+
+CREATE TABLE signoz_metrics.samples_v4_agg_5m
+(
+    env LowCardinality(String) DEFAULT 'default',
+    temporality LowCardinality(String) DEFAULT 'Unspecified',
+    metric_name LowCardinality(String),
+    fingerprint UInt64 CODEC(ZSTD(1)),
+    unix_milli Int64 CODEC(Delta(8), ZSTD(1)),
+    last SimpleAggregateFunction(anyLast, Float64) CODEC(ZSTD(1)),
+    min SimpleAggregateFunction(min, Float64) CODEC(ZSTD(1)),
+    max SimpleAggregateFunction(max, Float64) CODEC(ZSTD(1)),
+    sum SimpleAggregateFunction(sum, Float64) CODEC(ZSTD(1)),
+    count SimpleAggregateFunction(sum, UInt64) CODEC(ZSTD(1)),
+    min_unix_milli_by_start_ts SimpleAggregateFunction(minMap, Map(Int64, Int64)) CODEC(ZSTD(1)),
+    min_value_by_start_ts SimpleAggregateFunction(minMap, Map(Int64, Float64)) CODEC(ZSTD(1)),
+    max_unix_milli_by_start_ts SimpleAggregateFunction(maxMap, Map(Int64, Int64)) CODEC(ZSTD(1)),
+    max_value_by_start_ts SimpleAggregateFunction(maxMap, Map(Int64, Float64)) CODEC(ZSTD(1))
+)
+ENGINE = AggregatingMergeTree
+PARTITION BY toDate(unix_milli / 1000)
+ORDER BY (env, temporality, metric_name, fingerprint, unix_milli)
+TTL toDateTime(unix_milli / 1000) + toIntervalSecond(2592000)
+SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1;
+
+CREATE TABLE signoz_metrics.samples_v4_agg_30m
+(
+    env LowCardinality(String) DEFAULT 'default',
+    temporality LowCardinality(String) DEFAULT 'Unspecified',
+    metric_name LowCardinality(String),
+    fingerprint UInt64 CODEC(ZSTD(1)),
+    unix_milli Int64 CODEC(Delta(8), ZSTD(1)),
+    last SimpleAggregateFunction(anyLast, Float64) CODEC(ZSTD(1)),
+    min SimpleAggregateFunction(min, Float64) CODEC(ZSTD(1)),
+    max SimpleAggregateFunction(max, Float64) CODEC(ZSTD(1)),
+    sum SimpleAggregateFunction(sum, Float64) CODEC(ZSTD(1)),
+    count SimpleAggregateFunction(sum, UInt64) CODEC(ZSTD(1)),
+    min_unix_milli_by_start_ts SimpleAggregateFunction(minMap, Map(Int64, Int64)) CODEC(ZSTD(1)),
+    min_value_by_start_ts SimpleAggregateFunction(minMap, Map(Int64, Float64)) CODEC(ZSTD(1)),
+    max_unix_milli_by_start_ts SimpleAggregateFunction(maxMap, Map(Int64, Int64)) CODEC(ZSTD(1)),
+    max_value_by_start_ts SimpleAggregateFunction(maxMap, Map(Int64, Float64)) CODEC(ZSTD(1))
+)
+ENGINE = AggregatingMergeTree
+PARTITION BY toDate(unix_milli / 1000)
+ORDER BY (env, temporality, metric_name, fingerprint, unix_milli)
+TTL toDateTime(unix_milli / 1000) + toIntervalSecond(2592000)
+SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1;
+
+-- migration 1012 MV: 5m rollup
+CREATE MATERIALIZED VIEW signoz_metrics.samples_v4_agg_5m_mv
+TO signoz_metrics.samples_v4_agg_5m
+AS SELECT
+    env,
+    temporality,
+    metric_name,
+    fingerprint,
+    bucket_unix_milli AS unix_milli,
+    anyLast(value) AS last,
+    min(value) AS min,
+    max(value) AS max,
+    sum(value) AS sum,
+    count(*) AS count,
+    minMapIf(map(start_ts, sample_unix_milli), temporality = 'Cumulative') AS min_unix_milli_by_start_ts,
+    minMapIf(map(start_ts, value), temporality = 'Cumulative') AS min_value_by_start_ts,
+    maxMapIf(map(start_ts, sample_unix_milli), temporality = 'Cumulative') AS max_unix_milli_by_start_ts,
+    maxMapIf(map(start_ts, value), temporality = 'Cumulative') AS max_value_by_start_ts
+FROM (
+    SELECT
+        env,
+        temporality,
+        metric_name,
+        fingerprint,
+        start_ts,
+        unix_milli AS sample_unix_milli,
+        intDiv(unix_milli, 300000) * 300000 AS bucket_unix_milli,
+        value,
+        flags
+    FROM signoz_metrics.samples_v4
+)
+WHERE bitAnd(flags, 1) = 0
+GROUP BY
+    env,
+    temporality,
+    metric_name,
+    fingerprint,
+    bucket_unix_milli;
+
+-- migration 1012 MV: 30m rollup chained off 5m
+CREATE MATERIALIZED VIEW signoz_metrics.samples_v4_agg_30m_mv
+TO signoz_metrics.samples_v4_agg_30m
+AS SELECT
+    env,
+    temporality,
+    metric_name,
+    fingerprint,
+    intDiv(unix_milli, 1800000) * 1800000 AS unix_milli,
+    anyLast(last) AS last,
+    min(min) AS min,
+    max(max) AS max,
+    sum(sum) AS sum,
+    sum(count) AS count,
+    minMap(min_unix_milli_by_start_ts) AS min_unix_milli_by_start_ts,
+    minMap(min_value_by_start_ts) AS min_value_by_start_ts,
+    maxMap(max_unix_milli_by_start_ts) AS max_unix_milli_by_start_ts,
+    maxMap(max_value_by_start_ts) AS max_value_by_start_ts
+FROM signoz_metrics.samples_v4_agg_5m
+GROUP BY
+    env,
+    temporality,
+    metric_name,
+    fingerprint,
+    unix_milli;
+
+-- the time series catalog table the 1-day queries join against
+CREATE TABLE signoz_metrics.time_series_v4_1day
+(
+    env LowCardinality(String) DEFAULT 'default',
+    temporality LowCardinality(String) DEFAULT 'Unspecified',
+    metric_name LowCardinality(String),
+    fingerprint UInt64 CODEC(Delta(8), ZSTD(1)),
+    unix_milli Int64 CODEC(Delta(8), ZSTD(1)),
+    labels String CODEC(ZSTD(5))
+)
+ENGINE = ReplacingMergeTree
+ORDER BY (env, temporality, metric_name, fingerprint, unix_milli);
+
+-- single-node stand-ins for the distributed tables the querier reads
+CREATE VIEW signoz_metrics.distributed_samples_v4 AS SELECT * FROM signoz_metrics.samples_v4;
+CREATE VIEW signoz_metrics.distributed_samples_v4_agg_5m AS SELECT * FROM signoz_metrics.samples_v4_agg_5m;
+CREATE VIEW signoz_metrics.distributed_samples_v4_agg_30m AS SELECT * FROM signoz_metrics.samples_v4_agg_30m;

tests/integration/testdata/counter_reset_epochs/schema_legacy.sql

@@ -0,0 +1,100 @@
+-- Epoch harness schema: the real signoz_metrics tables after collector
+-- migration 1012, with distributed_* as plain views (single-node harness).
+-- MV queries are verbatim copies of migration 1012.
+
+CREATE DATABASE IF NOT EXISTS signoz_metrics;
+
+CREATE TABLE signoz_metrics.samples_v4
+(
+    env LowCardinality(String) DEFAULT 'default',
+    temporality LowCardinality(String) DEFAULT 'Unspecified',
+    metric_name LowCardinality(String),
+    fingerprint UInt64 CODEC(Delta(8), ZSTD(1)),
+    unix_milli Int64 CODEC(DoubleDelta, ZSTD(1)),
+    value Float64 CODEC(Gorilla, ZSTD(1)),
+    flags UInt32 DEFAULT 0 CODEC(ZSTD(1)),
+    inserted_at_unix_milli Int64 CODEC(ZSTD(1)),
+    start_ts Int64 DEFAULT 0 CODEC(DoubleDelta, ZSTD(1))
+)
+ENGINE = MergeTree
+PARTITION BY toDate(unix_milli / 1000)
+ORDER BY (env, temporality, metric_name, fingerprint, unix_milli)
+TTL toDateTime(unix_milli / 1000) + toIntervalSecond(2592000)
+SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1;
+
+CREATE TABLE signoz_metrics.samples_v4_agg_5m
+(
+    env LowCardinality(String) DEFAULT 'default',
+    temporality LowCardinality(String) DEFAULT 'Unspecified',
+    metric_name LowCardinality(String),
+    fingerprint UInt64 CODEC(ZSTD(1)),
+    unix_milli Int64 CODEC(Delta(8), ZSTD(1)),
+    last SimpleAggregateFunction(anyLast, Float64) CODEC(ZSTD(1)),
+    min SimpleAggregateFunction(min, Float64) CODEC(ZSTD(1)),
+    max SimpleAggregateFunction(max, Float64) CODEC(ZSTD(1)),
+    sum SimpleAggregateFunction(sum, Float64) CODEC(ZSTD(1)),
+    count SimpleAggregateFunction(sum, UInt64) CODEC(ZSTD(1)),
+)
+ENGINE = AggregatingMergeTree
+PARTITION BY toDate(unix_milli / 1000)
+ORDER BY (env, temporality, metric_name, fingerprint, unix_milli)
+TTL toDateTime(unix_milli / 1000) + toIntervalSecond(2592000)
+SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1;
+
+CREATE TABLE signoz_metrics.samples_v4_agg_30m
+(
+    env LowCardinality(String) DEFAULT 'default',
+    temporality LowCardinality(String) DEFAULT 'Unspecified',
+    metric_name LowCardinality(String),
+    fingerprint UInt64 CODEC(ZSTD(1)),
+    unix_milli Int64 CODEC(Delta(8), ZSTD(1)),
+    last SimpleAggregateFunction(anyLast, Float64) CODEC(ZSTD(1)),
+    min SimpleAggregateFunction(min, Float64) CODEC(ZSTD(1)),
+    max SimpleAggregateFunction(max, Float64) CODEC(ZSTD(1)),
+    sum SimpleAggregateFunction(sum, Float64) CODEC(ZSTD(1)),
+    count SimpleAggregateFunction(sum, UInt64) CODEC(ZSTD(1)),
+)
+ENGINE = AggregatingMergeTree
+PARTITION BY toDate(unix_milli / 1000)
+ORDER BY (env, temporality, metric_name, fingerprint, unix_milli)
+TTL toDateTime(unix_milli / 1000) + toIntervalSecond(2592000)
+SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1;
+
+CREATE MATERIALIZED VIEW signoz_metrics.samples_v4_agg_5m_mv
+TO signoz_metrics.samples_v4_agg_5m
+AS SELECT
+    env, temporality, metric_name, fingerprint,
+    intDiv(unix_milli, 300000) * 300000 as unix_milli,
+    anyLast(value) as last, min(value) as min, max(value) as max,
+    sum(value) as sum, count(*) as count
+FROM signoz_metrics.samples_v4
+WHERE bitAnd(flags, 1) = 0
+GROUP BY env, temporality, metric_name, fingerprint, unix_milli;
+
+CREATE MATERIALIZED VIEW signoz_metrics.samples_v4_agg_30m_mv
+TO signoz_metrics.samples_v4_agg_30m
+AS SELECT
+    env, temporality, metric_name, fingerprint,
+    intDiv(unix_milli, 1800000) * 1800000 AS unix_milli,
+    anyLast(last) AS last, min(min) AS min, max(max) AS max,
+    sum(sum) AS sum, sum(count) AS count
+FROM signoz_metrics.samples_v4_agg_5m
+GROUP BY env, temporality, metric_name, fingerprint, unix_milli;
+
+-- the time series catalog table the 1-day queries join against
+CREATE TABLE signoz_metrics.time_series_v4_1day
+(
+    env LowCardinality(String) DEFAULT 'default',
+    temporality LowCardinality(String) DEFAULT 'Unspecified',
+    metric_name LowCardinality(String),
+    fingerprint UInt64 CODEC(Delta(8), ZSTD(1)),
+    unix_milli Int64 CODEC(Delta(8), ZSTD(1)),
+    labels String CODEC(ZSTD(5))
+)
+ENGINE = ReplacingMergeTree
+ORDER BY (env, temporality, metric_name, fingerprint, unix_milli);
+
+-- single-node stand-ins for the distributed tables the querier reads
+CREATE VIEW signoz_metrics.distributed_samples_v4 AS SELECT * FROM signoz_metrics.samples_v4;
+CREATE VIEW signoz_metrics.distributed_samples_v4_agg_5m AS SELECT * FROM signoz_metrics.samples_v4_agg_5m;
+CREATE VIEW signoz_metrics.distributed_samples_v4_agg_30m AS SELECT * FROM signoz_metrics.samples_v4_agg_30m;