Repository: cortexproject/cortex
Pull request #4934 (2022-10-21 23:04:32)

Remove duplicate log message
Signed-off-by: Daniel Deluiggi <ddeluigg@amazon.com>

<!--  Thanks for sending a pull request!  Before submitting:

1. Read our CONTRIBUTING.md guide
2. Rebase your PR if it gets out of sync with master
-->

**What this PR does**:
We accidently add a duplicate log line on ingester in #4803

**Checklist**
- [ ] Tests updated
- [ ] Documentation added
- [ ] `CHANGELOG.md` updated - the order of entries should be `[CHANGE]`, `[FEATURE]`, `[ENHANCEMENT]`, `[BUGFIX]`



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 359cae53ac..62fd1bd335 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -41,6 +41,8 @@
 * [CHANGE] Disables TSDB isolation. #4825
 * [CHANGE] Drops support Prometheus 1.x rule format on configdb. #4826
 * [CHANGE] Removes `-ingester.stream-chunks-when-using-blocks` experimental flag and stream chunks by default when `querier.ingester-streaming` is enabled. #4864
+* [CHANGE] Compactor: Added `cortex_compactor_runs_interrupted_total` to separate compaction interruptions from failures
+* [CHANGE] Enable PromQL `@` modifier, negative offset always. #4927
 * [ENHANCEMENT] AlertManager: Retrying AlertManager Get Requests (Get Alertmanager status, Get Alertmanager Receivers) on next replica on error #4840
 * [ENHANCEMENT] Querier/Ruler: Retry store-gateway in case of unexpected failure, instead of failing the query. #4532 #4839
 * [ENHANCEMENT] Ring: DoBatch prioritize 4xx errors when failing. #4783
@@ -68,6 +70,7 @@
 * [BUGFIX] Respecting `-tracing.otel.sample-ratio` configuration when enabling OpenTelemetry tracing with X-ray. #4862
 * [BUGFIX] QueryFrontend: fixed query_range requests when query has `start` equals to `end`. #4877
 * [BUGFIX] AlertManager: fixed issue introduced by #4495 where templates files were being deleted when using alertmanager local store. #4890
+* [BUGFIX] Ingester: fixed incorrect logging at the start of ingester block shipping logic. #4934
 
 ## 1.13.0 2022-07-14
 
diff --git a/docs/blocks-storage/querier.md b/docs/blocks-storage/querier.md
index 3c8cbdc3d6..f199fe8815 100644
--- a/docs/blocks-storage/querier.md
+++ b/docs/blocks-storage/querier.md
@@ -136,10 +136,6 @@ querier:
   # CLI flag: -querier.query-store-for-labels-enabled
   [query_store_for_labels_enabled: <boolean> | default = false]
 
-  # Enable the @ modifier in PromQL.
-  # CLI flag: -querier.at-modifier-enabled
-  [at_modifier_enabled: <boolean> | default = false]
-
   # Enable returning samples stats per steps in query response.
   # CLI flag: -querier.per-step-stats-enabled
   [per_step_stats_enabled: <boolean> | default = false]
diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md
index e97d4cdb6f..e717e6ec7a 100644
--- a/docs/configuration/config-file-reference.md
+++ b/docs/configuration/config-file-reference.md
@@ -814,10 +814,6 @@ The `querier_config` configures the Cortex querier.
 # CLI flag: -querier.query-store-for-labels-enabled
 [query_store_for_labels_enabled: <boolean> | default = false]
 
-# Enable the @ modifier in PromQL.
-# CLI flag: -querier.at-modifier-enabled
-[at_modifier_enabled: <boolean> | default = false]
-
 # Enable returning samples stats per steps in query response.
 # CLI flag: -querier.per-step-stats-enabled
 [per_step_stats_enabled: <boolean> | default = false]
diff --git a/docs/configuration/v1-guarantees.md b/docs/configuration/v1-guarantees.md
index 5ce5760d46..2dc2db88bf 100644
--- a/docs/configuration/v1-guarantees.md
+++ b/docs/configuration/v1-guarantees.md
@@ -9,7 +9,7 @@ For the v1.0 release, we want to provide the following guarantees:
 
 ## Flags, Config and minor version upgrades
 
-Upgrading cortex from one minor version to the next should "just work"; that being said, we don't want to bump the major version every time we remove a flag, so we will will keep deprecated flags around for 2 minor release.  There is a metric (`cortex_deprecated_flags_inuse_total`) you can alert on to find out if you're using a deprecated  flag.
+Upgrading cortex from one minor version to the next should "just work"; that being said, we don't want to bump the major version every time we remove a flag, so we will will keep deprecated flags around for 2 minor release.  There is a metric (`deprecated_flags_inuse_total`) you can alert on to find out if you're using a deprecated  flag.
 
 Similarly to flags, minor version upgrades using config files should "just work".  If we do need to change config, we will keep the old way working for two minor version.  There will be a metric you can alert on for this too.
 
diff --git a/docs/proposals/timeseries-partitioning-in-compactor.md b/docs/proposals/timeseries-partitioning-in-compactor.md
index f61e25192c..5b733717b3 100644
--- a/docs/proposals/timeseries-partitioning-in-compactor.md
+++ b/docs/proposals/timeseries-partitioning-in-compactor.md
@@ -122,6 +122,37 @@ Now that the planner has produced a compaction plan for the T1-T3 compaction gro
 ![Grouping](/images/proposals/timeseries-partitioning-in-compactor-compact.png)
 T1 - Partition 1-2 was created with hash % 2 == 0, and in order to avoid having duplication information in blocks produced by compaction group 3-8, compaction group 5-8, and compaction group 7-8, we need apply the filter the `%8 == 0` hash, as that’s the hash of the highest partition count.
 
+### Compaction Workflow
+
+1. Compactor initializes Grouper and Planner.
+2. Compactor retrieves block's meta.json and call Grouper to group blocks for compaction.
+3. Grouper generates partitioned compaction groups:
+   1. Grouper groups source blocks into unpartitioned groups.
+   2. For each unpartitioned group:
+      1. Generates partitioned compaction group ID which is hash of min and max time of result block.
+      2. If the ID exists under the tenant directory in block storage, continue on next unpartitioned group.
+      3. Calculates number of partitions. Number of partitions indicates how many partitions one unpartitioned group would be partitioned into based on the total size of indices and number of time series from each source blocks in the unpartitioned group.
+      4. Assign source blocks into each partition with partition ID (value is in range from 0 to number_of_partitions - 1). Note that one source block could be used in multiple partitions (explanation in [Planning the compaction](#planning-the-compaction) and [Compaction](#compaction)). So multiple partition ID could be assigned to same source block. Check more partitioning examples in [Compaction Partitioning Examples](#compaction-partitioning-examples)
+      5. Generates partitioned compaction group that indicates which partition ID each blocks got assigned.
+      6. Partitioned compaction group information would be stored in block storage under the tenant directory it belongs to and the stored file can be picked up by cleaner later. Partitioned compaction group information contains partitioned compaction group ID, number of partitions, list of partitions which has partition ID and list of source blocks.
+      7. Store partitioned compaction group ID in block storage under each blocks' directory that are used by the generated partitioned compaction group.
+4. Grouper returns partitioned compaction groups to Compactor. Each returned group would have partition ID, number of partitions, and list of source blocks in memory.
+5. Compactor iterates over each partitioned compaction group. For each iteration, calls Planner to make sure the group is ready for compaction.
+6. Planner collects partitioned compaction group which is ready for compaction.
+   1. For each partitions in the group and for each blocks in the partition:
+      1. Make sure all source blocks fit within the time range of the group.
+      2. Make sure each source block with assigned partition IDs is currently not used by another ongoing compaction. This could utilize visit marker file that is introduced in #4805 by expanding it for each partition ID of the source block.
+      3. If all blocks in the partition are ready to be compacted,
+         1. mark status of those blocks with assigned partition ID as `pending`.
+         2. The status information of each partition ID would be stored in block storage under the corresponding block directory in order for cleaner to pick it up later.
+      4. If not all blocks in the partition are ready, continue on next partition
+7. Return all ready partitions to Compactor.
+8. Compactor starts compacting partitioned blocks. Once compaction completed, Compactor would mark status of all blocks along with assigned partition ID in the group as `completed`. Compactor should use partitioned compaction group ID to retrieve partitioned compaction group information from block storage to get all partition IDs assigned to each block. Then, retrieve status information of each partition ID this assigned to block under current block directory in block storage. If all assigned partition ID of the block have status set to `completed`, upload deletion marker for this block. Otherwise, no deletion marker would be uploaded.
+
+### Clean up Workflow
+
+Cleaner would periodically check any tenants having deletion marker. If there is a deletion marker for the tenant, Cleaner should remove all blocks and then clean up other files including partitioned group information files after tenant clean up delay. If there is no deletion marker for tenant, Clean should scan any source blocks having a deletion marker. If there is a deletion marker for the block, Cleaner should delete it.
+
 ## Performance
 
 Currently a 400M timeseries takes 12 hours to compact, without taking block download into consideration. If we have a partition count of 2, we can reduce this down to 6 hours, and a partition count of 10 is 3 hours. The scaling is not linear, and I’m still attempting to find out why. The initial result is promising enough to continue though.
@@ -264,4 +295,160 @@ T2 partition 4 - Hash(timeseries label) % 4 == 3 && % 8 == 7
 T3 partition 8 - Hash(timeseries label) % 8 == 7
 ```
 
+### Compaction Partitioning Examples
+
+#### Scenario: All source blocks were compacted by partitioning compaction (Idea case)
+
+All source blocks were previously compacted through partitioning compaction. In this case for each time range, the number of blocks belong to same time range would be 2^x if multiplier is set to 2.
+
+```
+Time ranges:
+T1, T2, T3
+
+Source blocks:
+T1: B1, B2
+T2: B3, B4, B5, B6
+T3: B7, B8, B9, B10, B11, B12, B13, B14
+
+Total indices size of all source blocks:
+200G
+```
+
+Number of Partitions = (200G / 64G = 3.125) => round up to next 2^x = 4
+
+Partitioning:
+* For T1, there are only 2 blocks which is < 4. So
+    * B1 (index 0 in the time range) can be grouped with other blocks having N % 4 == 0 or 2. Because 0 % 2 == 0.
+    * B2 (index 1 in the time range) can be grouped with other blocks having N % 4 == 1 or 3. Because 1 % 2 == 1.
+* For T2,
+    * B3 (index 0 in the time range) can be grouped with other blocks having N % 4 == 0.
+    * B4 (index 1 in the time range) can be grouped with other blocks having N % 4 == 1.
+    * B5 (index 2 in the time range) can be grouped with other blocks having N % 4 == 2.
+    * B6 (index 3 in the time range) can be grouped with other blocks having N % 4 == 3.
+* For T3,
+    * B7 (index 0 in the time range) can be grouped with other blocks having N % 4 == 0.
+    * B8 (index 1 in the time range) can be grouped with other blocks having N % 4 == 1.
+    * B9 (index 2 in the time range) can be grouped with other blocks having N % 4 == 2.
+    * B10 (index 3 in the time range) can be grouped with other blocks having N % 4 == 3.
+    * B11 (index 4 in the time range) can be grouped with other blocks having N % 4 == 0.
+    * B12 (index 5 in the time range) can be grouped with other blocks having N % 4 == 1.
+    * B13 (index 6 in the time range) can be grouped with other blocks having N % 4 == 2.
+    * B14 (index 7 in the time range) can be grouped with other blocks having N % 4 == 3.
+
+Partitions in Partitioned Compaction Group:
+* Partition ID: 0 \
+  Number of Partitions: 4 \
+  Blocks: B1, B3, B7, B11
+* Partition ID: 1 \
+  Number of Partitions: 4 \
+  Blocks: B2, B4, B8, B12
+* Partition ID: 2 \
+  Number of Partitions: 4 \
+  Blocks: B1, B5, B9, B13
+* Partition ID: 3 \
+  Number of Partitions: 4 \
+  Blocks: B2, B6, B10, B14
+
+---
+
+#### Scenario: All source blocks are level 1 blocks
+
+All source blocks are level 1 blocks. Since number of level 1 blocks in one time range is not guaranteed to be 2^x, all blocks need to be included in each partition.
+
+```
+Time ranges:
+T1
+
+Source blocks:
+T1: B1, B2, B3
+
+Total indices size of all source blocks:
+100G
+```
+
+Number of Partitions = (100G / 64G = 1.5625) => round up to next 2^x = 2
+
+Partitioning: There is only one time range from all source blocks which means it is compacting level 1 blocks. Partitioning needs to include all source blocks in each partition.
+
+Partitions in Partitioned Compaction Group:
+* Partition ID: 0 \
+  Number of Partitions: 2 \
+  Blocks: B1, B2, B3
+* Partition ID: 1 \
+  Number of Partitions: 2 \
+  Blocks: B1, B2, B3
+
+---
+
+#### Scenario: All source blocks are with compaction level > 1 and were generated by compactor without partitioning compaction
+
+If source block was generated by compactor without partitioning compaction, there should be only one block per time range. Since there is only one block in one time range, that one block would be included in all partitions.
+
+```
+Time ranges:
+T1, T2, T3
+
+Source blocks:
+T1: B1
+T2: B2
+T3: B3
+
+Total indices size of all source blocks:
+100G
+```
+
+Number of Partitions = (100G / 64G = 1.5625) => round up to next 2^x = 2
+
+Partitioning:
+* For T1, there is only one source block. Include B1 in all partitions.
+* For T2, there is only one source block. Include B2 in all partitions.
+* For T3, there is only one source block. Include B3 in all partitions.
+
+Partitions in Partitioned Compaction Group:
+* Partition ID: 0 \
+  Number of Partitions: 2 \
+  Blocks: B1, B2, B3
+* Partition ID: 1 \
+  Number of Partitions: 2 \
+  Blocks: B1, B2, B3
+
+---
+
+#### Scenario: All source blocks are with compaction level > 1 and some of them were generated by compactor with partitioning compaction
+
+Blocks generated by compactor without partitioning compaction would be included in all partitions. Blocks generated with partitioning compaction would be partitioned based on multiplier.
+
+```
+Time ranges:
+T1, T2, T3
+
+Source blocks:
+T1: B1 (unpartitioned)
+T2: B2, B3
+T3: B4, B5, B6, B7
+
+Total indices size of all source blocks:
+100G
+```
+
+Number of Partitions = (100G / 64G = 1.5625) => round up to next 2^x = 2
+
+Partitioning:
+* For T1, there is only one source block. Include B1 in all partitions.
+* For T2,
+    * B2 (index 0 in the time range) can be grouped with other blocks having N % 2 == 0.
+    * B3 (index 1 in the time range) can be grouped with other blocks having N % 2 == 1.
+* For T3,
+    * B4 (index 0 in the time range) can be grouped with other blocks having N % 2 == 0.
+    * B5 (index 1 in the time range) can be grouped with other blocks having N % 2 == 1.
+    * B6 (index 2 in the time range) can be grouped with other blocks having N % 2 == 0.
+    * B7 (index 3 in the time range) can be grouped with other blocks having N % 2 == 1.
+
+Partitions in Partitioned Compaction Group:
+* Partition ID: 0 \
+  Number of Partitions: 2 \
+  Blocks: B1, B2, B4, B6
+* Partition ID: 1 \
+  Number of Partitions: 2 \
+  Blocks: B1, B3, B5, B7
 
diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go
index c36ee8224d..75732e2200 100644
--- a/pkg/compactor/compactor.go
+++ b/pkg/compactor/compactor.go
@@ -313,6 +313,7 @@ type Compactor struct {
 
 	// Metrics.
 	compactionRunsStarted          prometheus.Counter
+	compactionRunsInterrupted      prometheus.Counter
 	compactionRunsCompleted        prometheus.Counter
 	compactionRunsFailed           prometheus.Counter
 	compactionRunsLastSuccess      prometheus.Gauge
@@ -399,6 +400,10 @@ func newCompactor(
 			Name: "cortex_compactor_runs_started_total",
 			Help: "Total number of compaction runs started.",
 		}),
+		compactionRunsInterrupted: promauto.With(registerer).NewCounter(prometheus.CounterOpts{
+			Name: "cortex_compactor_runs_interrupted_total",
+			Help: "Total number of compaction runs interrupted.",
+		}),
 		compactionRunsCompleted: promauto.With(registerer).NewCounter(prometheus.CounterOpts{
 			Name: "cortex_compactor_runs_completed_total",
 			Help: "Total number of compaction runs successfully completed.",
@@ -598,16 +603,23 @@ func (c *Compactor) running(ctx context.Context) error {
 }
 
 func (c *Compactor) compactUsers(ctx context.Context) {
-	succeeded := false
-	compactionErrorCount := 0
+	failed := false
+	interrupted := false
 
 	c.compactionRunsStarted.Inc()
 
 	defer func() {
-		if succeeded && compactionErrorCount == 0 {
+		// interruptions and successful runs are considered
+		// mutually exclusive but we consider a run failed if any
+		// tenant runs failed even if later runs are interrupted
+		if !interrupted && !failed {
 			c.compactionRunsCompleted.Inc()
 			c.compactionRunsLastSuccess.SetToCurrentTime()
-		} else {
+		}
+		if interrupted {
+			c.compactionRunsInterrupted.Inc()
+		}
+		if failed {
 			c.compactionRunsFailed.Inc()
 		}
 
@@ -621,6 +633,7 @@ func (c *Compactor) compactUsers(ctx context.Context) {
 	level.Info(c.logger).Log("msg", "discovering users from bucket")
 	users, err := c.discoverUsersWithRetries(ctx)
 	if err != nil {
+		failed = true
 		level.Error(c.logger).Log("msg", "failed to discover users from bucket", "err", err)
 		return
 	}
@@ -640,7 +653,8 @@ func (c *Compactor) compactUsers(ctx context.Context) {
 	for _, userID := range users {
 		// Ensure the context has not been canceled (ie. compactor shutdown has been triggered).
 		if ctx.Err() != nil {
-			level.Info(c.logger).Log("msg", "interrupting compaction of user blocks", "err", err)
+			interrupted = true
+			level.Info(c.logger).Log("msg", "interrupting compaction of user blocks", "user", userID)
 			return
 		}
 
@@ -670,8 +684,15 @@ func (c *Compactor) compactUsers(ctx context.Context) {
 		level.Info(c.logger).Log("msg", "starting compaction of user blocks", "user", userID)
 
 		if err = c.compactUserWithRetries(ctx, userID); err != nil {
+			// TODO: patch thanos error types to support errors.Is(err, context.Canceled) here
+			if ctx.Err() != nil && ctx.Err() == context.Canceled {
+				interrupted = true
+				level.Info(c.logger).Log("msg", "interrupting compaction of user blocks", "user", userID)
+				return
+			}
+
 			c.compactionRunFailedTenants.Inc()
-			compactionErrorCount++
+			failed = true
 			level.Error(c.logger).Log("msg", "failed to compact user blocks", "user", userID, "err", err)
 			continue
 		}
@@ -706,8 +727,6 @@ func (c *Compactor) compactUsers(ctx context.Context) {
 			}
 		}
 	}
-
-	succeeded = true
 }
 
 func (c *Compactor) compactUserWithRetries(ctx context.Context, userID string) error {
diff --git a/pkg/cortex/modules.go b/pkg/cortex/modules.go
index 3e66efd51d..ba573fddbd 100644
--- a/pkg/cortex/modules.go
+++ b/pkg/cortex/modules.go
@@ -533,14 +533,15 @@ func (t *Cortex) initRuler() (serv services.Service, err error) {
 	if t.Cfg.ExternalPusher != nil && t.Cfg.ExternalQueryable != nil {
 		rulerRegisterer := prometheus.WrapRegistererWith(prometheus.Labels{"engine": "ruler"}, prometheus.DefaultRegisterer)
 		engine := promql.NewEngine(promql.EngineOpts{
-			Logger:             util_log.Logger,
-			Reg:                rulerRegisterer,
-			ActiveQueryTracker: createActiveQueryTracker(t.Cfg.Querier, util_log.Logger),
-			MaxSamples:         t.Cfg.Querier.MaxSamples,
-			Timeout:            t.Cfg.Querier.Timeout,
-			LookbackDelta:      t.Cfg.Querier.LookbackDelta,
-			EnablePerStepStats: t.Cfg.Querier.EnablePerStepStats,
-			EnableAtModifier:   t.Cfg.Querier.AtModifierEnabled,
+			Logger:               util_log.Logger,
+			Reg:                  rulerRegisterer,
+			ActiveQueryTracker:   createActiveQueryTracker(t.Cfg.Querier, util_log.Logger),
+			MaxSamples:           t.Cfg.Querier.MaxSamples,
+			Timeout:              t.Cfg.Querier.Timeout,
+			LookbackDelta:        t.Cfg.Querier.LookbackDelta,
+			EnablePerStepStats:   t.Cfg.Querier.EnablePerStepStats,
+			EnableAtModifier:     true,
+			EnableNegativeOffset: true,
 			NoStepSubqueryIntervalFn: func(int64) int64 {
 				return t.Cfg.Querier.DefaultEvaluationInterval.Milliseconds()
 			},
diff --git a/pkg/ingester/ingester.go b/pkg/ingester/ingester.go
index 3f0fca4703..91ad41f0e7 100644
--- a/pkg/ingester/ingester.go
+++ b/pkg/ingester/ingester.go
@@ -2102,7 +2102,6 @@ func (i *Ingester) shipBlocks(ctx context.Context, allowed *util.AllowedTenants)
 	// particularly important for the JOINING state because there could
 	// be a blocks transfer in progress (from another ingester) and if we
 	// run the shipper in such state we could end up with race conditions.
-	level.Info(logutil.WithContext(ctx, i.logger)).Log("msg", "TSDB blocks shipping has been skipped because of the current ingester state")
 	if i.lifecycler != nil {
 		if ingesterState := i.lifecycler.GetState(); ingesterState == ring.PENDING || ingesterState == ring.JOINING {
 			level.Info(logutil.WithContext(ctx, i.logger)).Log("msg", "TSDB blocks shipping has been skipped because of the current ingester state", "state", ingesterState)
diff --git a/pkg/querier/querier.go b/pkg/querier/querier.go
index d17ab287e7..2c950f28d4 100644
--- a/pkg/querier/querier.go
+++ b/pkg/querier/querier.go
@@ -30,7 +30,9 @@ import (
 	seriesset "github.com/cortexproject/cortex/pkg/querier/series"
 	"github.com/cortexproject/cortex/pkg/tenant"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/flagext"
 	"github.com/cortexproject/cortex/pkg/util/limiter"
+	util_log "github.com/cortexproject/cortex/pkg/util/log"
 	"github.com/cortexproject/cortex/pkg/util/spanlogger"
 	"github.com/cortexproject/cortex/pkg/util/validation"
 )
@@ -46,7 +48,7 @@ type Config struct {
 	MaxSamples                int           `yaml:"max_samples"`
 	QueryIngestersWithin      time.Duration `yaml:"query_ingesters_within"`
 	QueryStoreForLabels       bool          `yaml:"query_store_for_labels_enabled"`
-	AtModifierEnabled         bool          `yaml:"at_modifier_enabled"`
+	AtModifierEnabled         bool          `yaml:"at_modifier_enabled" doc:"hidden"`
 	EnablePerStepStats        bool          `yaml:"per_step_stats_enabled"`
 
 	// QueryStoreAfter the time after which queries should also be sent to the store and not just ingesters.
@@ -82,6 +84,9 @@ var (
 
 // RegisterFlags adds the flags required to config this to the given FlagSet.
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
+	//lint:ignore faillint Need to pass the global logger like this for warning on deprecated methods
+	flagext.DeprecatedFlag(f, "querier.at-modifier-enabled", "This flag is no longer functional; at-modifier is always enabled now.", util_log.Logger)
+
 	cfg.StoreGatewayClient.RegisterFlagsWithPrefix("querier.store-gateway-client", f)
 	f.IntVar(&cfg.MaxConcurrent, "querier.max-concurrent", 20, "The maximum number of concurrent queries.")
 	f.DurationVar(&cfg.Timeout, "querier.timeout", 2*time.Minute, "The timeout for a query.")
@@ -92,7 +97,6 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.IntVar(&cfg.MaxSamples, "querier.max-samples", 50e6, "Maximum number of samples a single query can load into memory.")
 	f.DurationVar(&cfg.QueryIngestersWithin, "querier.query-ingesters-within", 0, "Maximum lookback beyond which queries are not sent to ingester. 0 means all queries are sent to ingester.")
 	f.BoolVar(&cfg.QueryStoreForLabels, "querier.query-store-for-labels-enabled", false, "Query long-term store for series, label values and label names APIs. Works only with blocks engine.")
-	f.BoolVar(&cfg.AtModifierEnabled, "querier.at-modifier-enabled", false, "Enable the @ modifier in PromQL.")
 	f.BoolVar(&cfg.EnablePerStepStats, "querier.per-step-stats-enabled", false, "Enable returning samples stats per steps in query response.")
 	f.DurationVar(&cfg.MaxQueryIntoFuture, "querier.max-query-into-future", 10*time.Minute, "Maximum duration into the future you can query. 0 to disable.")
 	f.DurationVar(&cfg.DefaultEvaluationInterval, "querier.default-evaluation-interval", time.Minute, "The default evaluation interval or step size for subqueries.")
@@ -163,14 +167,15 @@ func New(cfg Config, limits *validation.Overrides, distributor Distributor, stor
 	})
 
 	engine := promql.NewEngine(promql.EngineOpts{
-		Logger:             logger,
-		Reg:                reg,
-		ActiveQueryTracker: createActiveQueryTracker(cfg, logger),
-		MaxSamples:         cfg.MaxSamples,
-		Timeout:            cfg.Timeout,
-		LookbackDelta:      cfg.LookbackDelta,
-		EnablePerStepStats: cfg.EnablePerStepStats,
-		EnableAtModifier:   cfg.AtModifierEnabled,
+		Logger:               logger,
+		Reg:                  reg,
+		ActiveQueryTracker:   createActiveQueryTracker(cfg, logger),
+		MaxSamples:           cfg.MaxSamples,
+		Timeout:              cfg.Timeout,
+		LookbackDelta:        cfg.LookbackDelta,
+		EnablePerStepStats:   cfg.EnablePerStepStats,
+		EnableAtModifier:     true,
+		EnableNegativeOffset: true,
 		NoStepSubqueryIntervalFn: func(int64) int64 {
 			return cfg.DefaultEvaluationInterval.Milliseconds()
 		},


Test patch:
diff --git a/pkg/compactor/compactor_test.go b/pkg/compactor/compactor_test.go
index 62e815cb4e..52e4cf3828 100644
--- a/pkg/compactor/compactor_test.go
+++ b/pkg/compactor/compactor_test.go
@@ -1796,3 +1796,64 @@ func TestCompactor_ShouldFailCompactionOnTimeout(t *testing.T) {
 		`level=error component=compactor msg="compactor failed to become ACTIVE in the ring" err="context deadline exceeded"`,
 	}, removeIgnoredLogs(strings.Split(strings.TrimSpace(logs.String()), "\n")))
 }
+
+func TestCompactor_ShouldNotTreatInterruptionsAsErrors(t *testing.T) {
+	bucketClient := objstore.NewInMemBucket()
+	id := ulid.MustNew(ulid.Now(), rand.Reader)
+	require.NoError(t, bucketClient.Upload(context.Background(), "user-1/"+id.String()+"/meta.json", strings.NewReader(mockBlockMetaJSON(id.String()))))
+
+	b1 := createTSDBBlock(t, bucketClient, "user-1", 10, 20, map[string]string{"__name__": "Teste"})
+	b2 := createTSDBBlock(t, bucketClient, "user-1", 20, 30, map[string]string{"__name__": "Teste"})
+
+	c, tsdbCompactor, tsdbPlanner, logs, registry := prepare(t, prepareConfig(), bucketClient, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	tsdbCompactor.On("Compact", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(ulid.ULID{}, context.Canceled).Run(func(args mock.Arguments) {
+		cancel()
+	})
+	tsdbPlanner.On("Plan", mock.Anything, mock.Anything).Return([]*metadata.Meta{
+		{
+			BlockMeta: tsdb.BlockMeta{
+				ULID:    b1,
+				MinTime: 10,
+				MaxTime: 20,
+			},
+		},
+		{
+			BlockMeta: tsdb.BlockMeta{
+				ULID:    b2,
+				MinTime: 20,
+				MaxTime: 30,
+			},
+		},
+	}, nil)
+	require.NoError(t, services.StartAndAwaitRunning(ctx, c))
+
+	cortex_testutil.Poll(t, 1*time.Second, 1.0, func() interface{} {
+		return prom_testutil.ToFloat64(c.compactionRunsInterrupted)
+	})
+
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), c))
+
+	assert.NoError(t, prom_testutil.GatherAndCompare(registry, strings.NewReader(`
+		# TYPE cortex_compactor_runs_completed_total counter
+		# HELP cortex_compactor_runs_completed_total Total number of compaction runs successfully completed.
+		cortex_compactor_runs_completed_total 0
+
+		# TYPE cortex_compactor_runs_interrupted_total counter
+		# HELP cortex_compactor_runs_interrupted_total Total number of compaction runs interrupted.
+		cortex_compactor_runs_interrupted_total 1
+
+		# TYPE cortex_compactor_runs_failed_total counter
+		# HELP cortex_compactor_runs_failed_total Total number of compaction runs failed.
+		cortex_compactor_runs_failed_total 0
+	`),
+		"cortex_compactor_runs_completed_total",
+		"cortex_compactor_runs_interrupted_total",
+		"cortex_compactor_runs_failed_total",
+	))
+
+	lines := strings.Split(logs.String(), "\n")
+	require.Contains(t, lines, `level=info component=compactor msg="interrupting compaction of user blocks" user=user-1`)
+	require.NotContains(t, logs.String(), `level=error`)
+}