diff --git a/CHANGELOG.md b/CHANGELOG.md
index 35d3fd3dbc..72b32fffbc 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -21,6 +21,7 @@
 * [ENHANCEMENT] Validation: Add an optional `reason` field to `limited_queries` rules, aligning them with `blocked_queries`. When set, the reason is included in the client-facing error and the query-frontend's `"query limited"` log line. #16407
 * [ENHANCEMENT] Compactor: Add the experimental `-compactor.scheduler-client.enable-ring-based-cleanup` option, which when disabled stops a scheduler-mode compactor from running the ring-based background blocks cleaner. #16457
 * [FEATURE] Querier: Add experimental per-tenant limit `-querier.max-blocks-per-store-request` to cap the number of blocks a single store-gateway request may reference. Disabled by default. #16292
+* [FEATURE] MQE: Range vector splitting can now also split subqueries, in addition to range vector selectors. Enable with the experimental `-querier.mimir-query-engine.range-vector-splitting.enable-subquery-splitting` flag, in addition to `-querier.mimir-query-engine.range-vector-splitting.enabled`. Disabled by default. #16444
 * [FEATURE] Validation: Add optional `id`, `note`, `created_by`, `created_at`, and `expires_at` fields to `blocked_queries` and `limited_queries` rules, for tooling to attach ownership/context metadata to a rule. For rules with `expires_at` set, the earliest `expires_at` per tenant and `id` (rules without an `id` are grouped together) is exported as the `cortex_blocked_query_rule_expires_at`/`cortex_limited_query_rule_expires_at` metrics, so an alert can fire on stale rules; this is informational only and never affects enforcement. The query-frontend's `"query blocked"` log line now also includes the matched rule's `id` and whether it is expired, and rate-limited queries are now logged with a new `"query limited"` line carrying the same fields. #16395
 * [BUGFIX] Query-frontend: Wait for the querier ring to be populated during startup, up to 30 seconds, before reporting the query-frontend as ready. Previously a query-frontend could become ready before it had seen any querier in the ring and fail every query it received until the ring was populated. Only applies when remote execution is enabled, and can be disabled with the experimental `-query-frontend.wait-for-querier-ring-on-startup=false`. #16333
 * [BUGFIX] Query-frontend: Fail queries with a clear error, rather than planning them against an invalid maximum supported query plan version, when the querier ring contains only unhealthy queriers. #16333
diff --git a/cmd/mimir/config-descriptor.json b/cmd/mimir/config-descriptor.json
index 80a64bf354..cdf6b1f066 100644
--- a/cmd/mimir/config-descriptor.json
+++ b/cmd/mimir/config-descriptor.json
@@ -2986,6 +2986,17 @@
                   ],
                   "fieldValue": null,
                   "fieldDefaultValue": null
+                },
+                {
+                  "kind": "field",
+                  "name": "enable_subquery_splitting",
+                  "required": false,
+                  "desc": "Enable splitting subqueries, in addition to range vector selectors. Requires -querier.mimir-query-engine.range-vector-splitting.enabled and -querier.mimir-query-engine.enable-common-subexpression-elimination to also be enabled.",
+                  "fieldValue": null,
+                  "fieldDefaultValue": false,
+                  "fieldFlag": "querier.mimir-query-engine.range-vector-splitting.enable-subquery-splitting",
+                  "fieldType": "boolean",
+                  "fieldCategory": "experimental"
                 }
               ],
               "fieldValue": null,
diff --git a/cmd/mimir/help-all.txt.tmpl b/cmd/mimir/help-all.txt.tmpl
index 2596ebd526..5edbab89f0 100644
--- a/cmd/mimir/help-all.txt.tmpl
+++ b/cmd/mimir/help-all.txt.tmpl
@@ -2547,6 +2547,8 @@ Usage of ./cmd/mimir/mimir:
     	Backend for intermediate results cache, if not empty. Supported values: memcached.
   -querier.mimir-query-engine.range-vector-splitting.compression string
     	Enable cache compression, if not empty. Supported values are: snappy.
+  -querier.mimir-query-engine.range-vector-splitting.enable-subquery-splitting
+    	[experimental] Enable splitting subqueries, in addition to range vector selectors. Requires -querier.mimir-query-engine.range-vector-splitting.enabled and -querier.mimir-query-engine.enable-common-subexpression-elimination to also be enabled.
   -querier.mimir-query-engine.range-vector-splitting.enabled
     	[experimental] Enable splitting function over range vectors queries into smaller blocks for caching.
   -querier.mimir-query-engine.range-vector-splitting.memcached.addresses comma-separated-list-of-strings
diff --git a/docs/sources/mimir/configure/configuration-parameters/index.md b/docs/sources/mimir/configure/configuration-parameters/index.md
index e13c1f685e..20d3b0feea 100644
--- a/docs/sources/mimir/configure/configuration-parameters/index.md
+++ b/docs/sources/mimir/configure/configuration-parameters/index.md
@@ -2178,6 +2178,14 @@ mimir_query_engine:
       # CLI flag: -querier.mimir-query-engine.range-vector-splitting.compression
       [compression: <string> | default = ""]
 
+    # (experimental) Enable splitting subqueries, in addition to range vector
+    # selectors. Requires
+    # -querier.mimir-query-engine.range-vector-splitting.enabled and
+    # -querier.mimir-query-engine.enable-common-subexpression-elimination to
+    # also be enabled.
+    # CLI flag: -querier.mimir-query-engine.range-vector-splitting.enable-subquery-splitting
+    [enable_subquery_splitting: <boolean> | default = false]
+
   time_splitting_and_caching:
     # (experimental) Enable caching of query results that were not fully
     # consumed by the query. When enabled, if a query stops reading before all
diff --git a/pkg/streamingpromql/config.go b/pkg/streamingpromql/config.go
index 48605324b3..42239a4ebc 100644
--- a/pkg/streamingpromql/config.go
+++ b/pkg/streamingpromql/config.go
@@ -87,6 +87,10 @@ type RangeVectorSplittingConfig struct {
 	//  without caching (e.g. possibly if splitting is extended to range queries in the future, or if we add
 	//  parallelisation and just want to use query splitting for that and not cache).
 	IntermediateResultsCache rangevectorsplittingcache.Config `yaml:"intermediate_results_cache" category:"experimental"`
+
+	// EnableSubquerySplitting enables splitting subqueries, in addition to range vector selectors. Requires
+	// Enabled and EngineOpts.EnableCommonSubexpressionElimination.
+	EnableSubquerySplitting bool `yaml:"enable_subquery_splitting" category:"experimental"`
 }
 
 type RangeQuerySplittingAndCachingConfig struct {
@@ -143,6 +147,7 @@ func (o *EngineOpts) RegisterFlags(f *flag.FlagSet) {
 func (c *RangeVectorSplittingConfig) RegisterFlags(f *flag.FlagSet) {
 	f.BoolVar(&c.Enabled, "querier.mimir-query-engine.range-vector-splitting.enabled", false, "Enable splitting function over range vectors queries into smaller blocks for caching.")
 	f.DurationVar(&c.SplitInterval, "querier.mimir-query-engine.range-vector-splitting.split-interval", 2*time.Hour, "Time interval used for splitting function over range vectors queries into cacheable blocks.")
+	f.BoolVar(&c.EnableSubquerySplitting, "querier.mimir-query-engine.range-vector-splitting.enable-subquery-splitting", false, "Enable splitting subqueries, in addition to range vector selectors. Requires -querier.mimir-query-engine.range-vector-splitting.enabled and -querier.mimir-query-engine.enable-common-subexpression-elimination to also be enabled.")
 	c.IntermediateResultsCache.RegisterFlagsWithPrefix(f, "querier.mimir-query-engine.range-vector-splitting.")
 }
 
@@ -218,6 +223,8 @@ func (c *RangeVectorSplittingConfig) Validate() error {
 		if err := c.IntermediateResultsCache.Validate(); err != nil {
 			return errors.Wrap(err, "invalid intermediate results cache config")
 		}
+	} else if c.EnableSubquerySplitting {
+		return fmt.Errorf("range vector splitting subqueries is enabled but range vector splitting is not enabled")
 	}
 	return nil
 }
diff --git a/pkg/streamingpromql/engine.go b/pkg/streamingpromql/engine.go
index 457a94f0b7..f69346110a 100644
--- a/pkg/streamingpromql/engine.go
+++ b/pkg/streamingpromql/engine.go
@@ -100,7 +100,7 @@ func NewEngineWithCache(opts EngineOpts, metrics *stats.QueryMetrics, planner *Q
 		planning.NODE_TYPE_NUMBER_LITERAL:        planning.NodeMaterializerFunc[*core.NumberLiteral](core.MaterializeNumberLiteral),
 		planning.NODE_TYPE_STRING_LITERAL:        planning.NodeMaterializerFunc[*core.StringLiteral](core.MaterializeStringLiteral),
 		planning.NODE_TYPE_UNARY_EXPRESSION:      planning.NodeMaterializerFunc[*core.UnaryExpression](core.MaterializeUnaryExpression),
-		planning.NODE_TYPE_SUBQUERY:              planning.NodeMaterializerFunc[*core.Subquery](core.MaterializeSubquery),
+		planning.NODE_TYPE_SUBQUERY:              planning.RangeAwareNodeMaterializerFunc[*core.Subquery](core.MaterializeSubquery),
 		planning.NODE_TYPE_DEDUPLICATE_AND_MERGE: planning.NodeMaterializerFunc[*core.DeduplicateAndMerge](core.MaterializeDeduplicateAndMerge),
 		planning.NODE_TYPE_DROP_NAME:             planning.NodeMaterializerFunc[*core.DropName](core.MaterializeDropName),
 		planning.NODE_TYPE_NO_OP:                 planning.NodeMaterializerFunc[*core.NoOp](core.MaterializeNoOp),
@@ -113,7 +113,7 @@ func NewEngineWithCache(opts EngineOpts, metrics *stats.QueryMetrics, planner *Q
 		planning.NODE_TYPE_MULTI_AGGREGATION_GROUP:    planning.NodeMaterializerFunc[*multiaggregation.MultiAggregationGroup](multiaggregation.MaterializeMultiAggregationGroup),
 		planning.NODE_TYPE_MULTI_AGGREGATION_INSTANCE: planning.NodeMaterializerFunc[*multiaggregation.MultiAggregationInstance](multiaggregation.MaterializeMultiAggregationInstance),
 
-		planning.NODE_TYPE_SPLIT_FUNCTION_OVER_RANGE_VECTOR: rangevectorsplitting.NewMaterializer(opts.RangeVectorSplitting.Enabled, opts.RangeVectorSplitting.SplitInterval, opts.Limits, opts.TimeNow, intermediateCache, opts.CommonOpts.Reg, opts.Logger),
+		planning.NODE_TYPE_SPLIT_FUNCTION_OVER_RANGE_VECTOR: rangevectorsplitting.NewMaterializer(opts.RangeVectorSplitting.Enabled, opts.RangeVectorSplitting.SplitInterval, opts.RangeVectorSplitting.EnableSubquerySplitting, opts.Limits, opts.TimeNow, intermediateCache, opts.CommonOpts.Reg, opts.Logger),
 		planning.NODE_TYPE_TIME_RANGE_SPLIT:                 splitandcache.NewTimeRangeSplitMaterializer(opts.RangeQuerySplittingAndCaching.SplitEnabled, opts.CommonOpts.Reg),
 		planning.NODE_TYPE_CACHE: splitandcache.NewCacheMaterializer(
 			opts.RangeQuerySplittingAndCaching.CacheEnabled,
diff --git a/pkg/streamingpromql/optimize/plan/commonsubexpressionelimination/optimization_pass.go b/pkg/streamingpromql/optimize/plan/commonsubexpressionelimination/optimization_pass.go
index e15c86e463..0af13ba15e 100644
--- a/pkg/streamingpromql/optimize/plan/commonsubexpressionelimination/optimization_pass.go
+++ b/pkg/streamingpromql/optimize/plan/commonsubexpressionelimination/optimization_pass.go
@@ -95,6 +95,13 @@ func (e *OptimizationPass) Apply(ctx context.Context, plan *planning.QueryPlan,
 		return nil, err
 	}
 
+	// Range vector splitting can materialize and execute a nested Subquery/StepInvariantExpression more than
+	// once, once per split block, insert Duplicate nodes here.
+	splitSubqueryDuplicatesInserted, err := e.insertSplitSubqueryDuplicates(plan.Root)
+	if err != nil {
+		return nil, err
+	}
+
 	e.selectorsInspected.Add(float64(len(paths)))
 	e.duplicateSelectorsEliminated.Add(float64(stats.duplicateSelectorsEliminated))
 	e.subsetSelectorsEliminated.Add(float64(stats.subsetSelectorsEliminated))
@@ -105,6 +112,7 @@ func (e *OptimizationPass) Apply(ctx context.Context, plan *planning.QueryPlan,
 		"selectors_inspected", len(paths),
 		"duplicate_selectors_eliminated", stats.duplicateSelectorsEliminated,
 		"subset_selectors_eliminated", stats.subsetSelectorsEliminated,
+		"split_subquery_duplicates_inserted", splitSubqueryDuplicatesInserted,
 	)
 
 	return plan, nil
@@ -930,6 +938,112 @@ func isDuplicateNode(node planning.Node) bool {
 	return isDuplicate
 }
 
+// insertSplitSubqueryDuplicates finds SplitFunctionCall nodes wrapping a subquery, and inserts Duplicate nodes
+// (via insertDuplicatesAcrossSplitBlocks) around any Subquery/StepInvariantExpression nested inside that
+// subquery's inner expression. Returns the number of Duplicate nodes introduced.
+func (e *OptimizationPass) insertSplitSubqueryDuplicates(n planning.Node) (int, error) {
+	introduced := 0
+
+	if splitCall, ok := n.(*rangevectorsplitting.SplitFunctionCall); ok {
+		if splitCall.Inner.ChildCount() != 1 {
+			return 0, fmt.Errorf("expected SplitFunctionCall's inner function call to have exactly one child, got %d", splitCall.Inner.ChildCount())
+		}
+
+		if subquery, isSubquery := unwrapDuplicate(splitCall.Inner.Child(0)).(*core.Subquery); isSubquery {
+			count, err := e.insertDuplicatesAcrossSplitBlocks(subquery.Child(0))
+			if err != nil {
+				return 0, err
+			}
+
+			introduced += count
+		}
+	}
+
+	for child := range planning.ChildrenIter(n) {
+		count, err := e.insertSplitSubqueryDuplicates(child)
+		if err != nil {
+			return 0, err
+		}
+
+		introduced += count
+	}
+
+	return introduced, nil
+}
+
+func unwrapDuplicate(n planning.Node) planning.Node {
+	switch n := n.(type) {
+	case *Duplicate:
+		return unwrapDuplicate(n.Inner)
+	case *DuplicateFilter:
+		return unwrapDuplicate(n.Inner)
+	default:
+		return n
+	}
+}
+
+// insertDuplicatesAcrossSplitBlocks wraps the child of every core.Subquery/core.StepInvariantExpression in n's
+// subtree in a Duplicate node, at any nesting depth, so different split blocks can safely materialize it more
+// than once (see hour_collision_metric test case in range_vector_splitting_2h.test).
+// Returns the number of Duplicate nodes introduced.
+func (e *OptimizationPass) insertDuplicatesAcrossSplitBlocks(n planning.Node) (int, error) {
+	if isSubqueryOrStepInvariantExpression(n) {
+		if n.ChildCount() != 1 {
+			return 0, fmt.Errorf("expected node of type %s to have exactly one child, got %d", n.NodeType(), n.ChildCount())
+		}
+
+		child := n.Child(0)
+
+		if isDuplicateNode(child) {
+			// keep recursing since a further nested Subquery/StepInvariantExpression inside it may still need its own Duplicate.
+			return e.insertDuplicatesAcrossSplitBlocks(child)
+		}
+
+		// Result type is always Vector or Scalar in practice (see planning.go's StepInvariantExpr handling).
+		// This is a defensive check in case that invariant is ever broken.
+		if resultType, err := child.ResultType(); err != nil {
+			return 0, err
+		} else if resultType != parser.ValueTypeVector && resultType != parser.ValueTypeScalar {
+			return 0, fmt.Errorf("cannot insert a Duplicate node for %s node (%s) across split blocks: unexpected result type %s", n.NodeType(), n.Describe(), resultType)
+		}
+
+		introduced, err := e.insertDuplicatesAcrossSplitBlocks(child)
+		if err != nil {
+			return 0, err
+		}
+
+		duplicate := &Duplicate{Inner: child, DuplicateDetails: &DuplicateDetails{}}
+		e.duplicationNodesIntroduced.Inc()
+
+		if err := n.ReplaceChild(0, duplicate); err != nil {
+			return 0, err
+		}
+
+		return introduced + 1, nil
+	}
+
+	introduced := 0
+	for child := range planning.ChildrenIter(n) {
+		count, err := e.insertDuplicatesAcrossSplitBlocks(child)
+		if err != nil {
+			return 0, err
+		}
+
+		introduced += count
+	}
+
+	return introduced, nil
+}
+
+func isSubqueryOrStepInvariantExpression(n planning.Node) bool {
+	switch n.(type) {
+	case *core.Subquery, *core.StepInvariantExpression:
+		return true
+	default:
+		return false
+	}
+}
+
 type path []pathElement
 
 type pathElement struct {
diff --git a/pkg/streamingpromql/optimize/plan/rangevectorsplitting/node.go b/pkg/streamingpromql/optimize/plan/rangevectorsplitting/node.go
index 590849f33e..1f9e0da64d 100644
--- a/pkg/streamingpromql/optimize/plan/rangevectorsplitting/node.go
+++ b/pkg/streamingpromql/optimize/plan/rangevectorsplitting/node.go
@@ -68,21 +68,41 @@ func (s *SplitFunctionCall) ExpressionPosition() (posrange.PositionRange, error)
 }
 
 func (s *SplitFunctionCall) MinimumRequiredPlanVersion(types.QueryTimeRange) (planning.QueryPlanVersion, error) {
+	if containsSubquery(s.Inner) {
+		return planning.QueryPlanV21, nil
+	}
+
 	return planning.QueryPlanV18, nil
 }
 
+// containsSubquery returns true if n or any of its descendants is a Subquery.
+func containsSubquery(n planning.Node) bool {
+	if _, ok := n.(*core.Subquery); ok {
+		return true
+	}
+
+	for child := range planning.ChildrenIter(n) {
+		if containsSubquery(child) {
+			return true
+		}
+	}
+
+	return false
+}
+
 // limitsProvider provides the tenant limits needed to compute split ranges at materialize time.
 type limitsProvider interface {
 	GetMaxOutOfOrderTimeWindow(ctx context.Context) (time.Duration, error)
 }
 
 type Materializer struct {
-	enabled       bool
-	splitInterval time.Duration
-	limits        limitsProvider
-	timeNow       func() time.Time
-	cache         *cache.CacheFactory
-	logger        log.Logger
+	enabled                 bool
+	splitInterval           time.Duration
+	enableSubquerySplitting bool
+	limits                  limitsProvider
+	timeNow                 func() time.Time
+	cache                   *cache.CacheFactory
+	logger                  log.Logger
 
 	nodesSplit   prometheus.Counter
 	nodesUnsplit *prometheus.CounterVec
@@ -90,18 +110,19 @@ type Materializer struct {
 
 var _ planning.NodeMaterializer = &Materializer{}
 
-func NewMaterializer(enabled bool, splitInterval time.Duration, limits limitsProvider, timeNow func() time.Time, cache *cache.CacheFactory, reg prometheus.Registerer, logger log.Logger) *Materializer {
+func NewMaterializer(enabled bool, splitInterval time.Duration, enableSubquerySplitting bool, limits limitsProvider, timeNow func() time.Time, cache *cache.CacheFactory, reg prometheus.Registerer, logger log.Logger) *Materializer {
 	if timeNow == nil {
 		timeNow = time.Now
 	}
 
 	return &Materializer{
-		enabled:       enabled,
-		splitInterval: splitInterval,
-		limits:        limits,
-		timeNow:       timeNow,
-		cache:         cache,
-		logger:        logger,
+		enabled:                 enabled,
+		splitInterval:           splitInterval,
+		enableSubquerySplitting: enableSubquerySplitting,
+		limits:                  limits,
+		timeNow:                 timeNow,
+		cache:                   cache,
+		logger:                  logger,
 		nodesSplit: promauto.With(reg).NewCounter(prometheus.CounterOpts{
 			Name: "cortex_mimir_query_engine_range_vector_splitting_nodes_materialized_split_total",
 			Help: "Total number of range vector splitting nodes materialized as split operators.",
@@ -143,6 +164,12 @@ func (m Materializer) Materialize(ctx context.Context, n planning.Node, material
 		return nil, fmt.Errorf("inner node of split function call does not implement SplitNode: %T", innerNode)
 	}
 
+	if containsSubquery(innerNode) && !m.enableSubquerySplitting {
+		level.Warn(m.logger).Log("msg", "split function node wraps a subquery but subquery splitting is disabled, falling back to unsplit execution; this can happen if subquery splitting is enabled on the query-frontend but not yet on the querier")
+		m.nodesUnsplit.WithLabelValues("subquery_splitting_disabled").Inc()
+		return materializer.FactoryForNode(ctx, s.Inner, timeRange)
+	}
+
 	// The split ranges are computed here, at materialize time, rather than at planning time, because they depend on
 	// the querier's current time and the tenant's out-of-order window. If the ranges turn out not to be worth
 	// splitting (e.g. there's no complete cacheable block, or every block falls within the out-of-order window), fall
diff --git a/pkg/streamingpromql/optimize/plan/rangevectorsplitting/node_test.go b/pkg/streamingpromql/optimize/plan/rangevectorsplitting/node_test.go
index d55a5d15dd..a568639449 100644
--- a/pkg/streamingpromql/optimize/plan/rangevectorsplitting/node_test.go
+++ b/pkg/streamingpromql/optimize/plan/rangevectorsplitting/node_test.go
@@ -164,7 +164,7 @@ func TestMaterializer_computeRanges(t *testing.T) {
 	fixedNow := timestamp.Time(100 * hourInMs)
 
 	newMaterializer := func(oooWindow time.Duration) *Materializer {
-		return NewMaterializer(true, 2*time.Hour, staticLimits{oooWindow: oooWindow}, func() time.Time { return fixedNow }, nil, nil, nil)
+		return NewMaterializer(true, 2*time.Hour, true, staticLimits{oooWindow: oooWindow}, func() time.Time { return fixedNow }, nil, nil, nil)
 	}
 
 	// inner builds an inner matrix selector with the given range and offset.
diff --git a/pkg/streamingpromql/optimize/plan/rangevectorsplitting/operator.go b/pkg/streamingpromql/optimize/plan/rangevectorsplitting/operator.go
index 26e193cbd8..baf1685006 100644
--- a/pkg/streamingpromql/optimize/plan/rangevectorsplitting/operator.go
+++ b/pkg/streamingpromql/optimize/plan/rangevectorsplitting/operator.go
@@ -383,21 +383,19 @@ func (m *FunctionOverRangeVectorSplit[T]) mergeSplitsMetadata(ctx context.Contex
 				}
 				seriesToSplits = append(seriesToSplits, nil)
 			} else {
-				if seriesMetadata.DropName != mergedMetadata[mergedIdx].DropName {
-					// This shouldn't happen for range vector selectors, DropName will always be false at this point.
-					// TODO: There is a problematic edge case if subquery splitting is supported and delayed name
-					//  removal is enabled:
-					//  rate(foo[1d]) or label_replace(bar{}, "__name__", "foo", "", "")
-					//  Left: {__name__="foo"} + DropName=true (from rate)
-					//  Right: {__name__="foo"} + DropName=false (no functions to set DropName=true)
-					//  If the left is missing from some splits, we get inconsistent DropNames.
-					//  DeduplicateAndMerge will take the DropName value from the LHS, if results exist for the LHS at
-					//  any point. Otherwise the RHS DropName is used.
-					//  In the split case, if there are splits that don't have the LHS, we can get inconsistent
-					//  DropNames across splits. The splits don't know whether there were samples from the LHS or not so
-					//  cannot always reproduce the non-split behaviour.
-					return nil, nil, fmt.Errorf("series %s has conflicting DropName values across splits (split %d has %t, merged has %t)", seriesMetadata.Labels.String(), splitIdx, seriesMetadata.DropName, mergedMetadata[mergedIdx].DropName)
-				}
+				// This shouldn't happen for range vector selectors, DropName will always be false at this point.
+				// There is a problematic edge case if subquery splitting and delayed name removal are enabled:
+				//  rate(foo[1d]) or label_replace(bar{}, "__name__", "foo", "", "")
+				//  Left: {__name__="foo"} + DropName=true (from rate)
+				//  Right: {__name__="foo"} + DropName=false (no functions to set DropName=true)
+				//  If the left is missing from some splits, we get inconsistent DropNames.
+				//  DeduplicateAndMerge will take the DropName value from the LHS, if results exist for the LHS at
+				//  any point. Otherwise the RHS DropName is used.
+				//  In the split case, if there are splits that don't have the LHS, we can get inconsistent
+				//  DropNames across splits. The splits don't know whether there were samples from the LHS or not so
+				//  cannot always reproduce the non-split behaviour.
+				// We handle this by keeping the DropName value from whichever split first introduced the series,
+				// rather than erroring, even though this can differ from the non-split behaviour.
 				m.MemoryConsumptionTracker.DecreaseMemoryConsumptionForLabels(seriesMetadata.Labels)
 			}
 			seriesToSplits[mergedIdx] = append(seriesToSplits[mergedIdx], SplitSeries{
diff --git a/pkg/streamingpromql/optimize/plan/rangevectorsplitting/optimization_pass.go b/pkg/streamingpromql/optimize/plan/rangevectorsplitting/optimization_pass.go
index 15b2215bb2..b177801a0a 100644
--- a/pkg/streamingpromql/optimize/plan/rangevectorsplitting/optimization_pass.go
+++ b/pkg/streamingpromql/optimize/plan/rangevectorsplitting/optimization_pass.go
@@ -20,7 +20,8 @@ import (
 )
 
 type OptimizationPass struct {
-	splitInterval time.Duration
+	splitInterval           time.Duration
+	enableSubquerySplitting bool
 
 	splitNodesIntroduced   prometheus.Counter
 	functionNodesInspected prometheus.Counter
@@ -29,9 +30,10 @@ type OptimizationPass struct {
 	logger log.Logger
 }
 
-func NewOptimizationPass(splitInterval time.Duration, reg prometheus.Registerer, logger log.Logger) *OptimizationPass {
+func NewOptimizationPass(splitInterval time.Duration, enableSubquerySplitting bool, reg prometheus.Registerer, logger log.Logger) *OptimizationPass {
 	return &OptimizationPass{
-		splitInterval: splitInterval,
+		splitInterval:           splitInterval,
+		enableSubquerySplitting: enableSubquerySplitting,
 		splitNodesIntroduced: promauto.With(reg).NewCounter(prometheus.CounterOpts{
 			Name: "cortex_mimir_query_engine_range_vector_splitting_nodes_introduced_total",
 			Help: "Total number of SplitFunctionCall nodes introduced by the range vector splitting optimization pass.",
@@ -57,8 +59,12 @@ func (o *OptimizationPass) Apply(ctx context.Context, plan *planning.QueryPlan,
 		return plan, nil
 	}
 
+	// Splitting a subquery requires all queriers to support SplitFunctionCall nodes that wrap a Subquery, which is
+	// only guaranteed from V21 onwards.
+	enableSubquerySplitting := o.enableSubquerySplitting && maximumSupportedQueryPlanVersion >= planning.QueryPlanV21
+
 	var err error
-	plan.Root, err = o.wrapSplitRangeVectorFunctions(ctx, plan.Root, plan.Parameters.TimeRange)
+	plan.Root, err = o.wrapSplitRangeVectorFunctions(ctx, plan.Root, plan.Parameters.TimeRange, enableSubquerySplitting)
 	if err != nil {
 		return nil, err
 	}
@@ -66,7 +72,7 @@ func (o *OptimizationPass) Apply(ctx context.Context, plan *planning.QueryPlan,
 	return plan, nil
 }
 
-func (o *OptimizationPass) wrapSplitRangeVectorFunctions(ctx context.Context, n planning.Node, timeRange types.QueryTimeRange) (planning.Node, error) {
+func (o *OptimizationPass) wrapSplitRangeVectorFunctions(ctx context.Context, n planning.Node, timeRange types.QueryTimeRange, enableSubquerySplitting bool) (planning.Node, error) {
 	logger := spanlogger.FromContext(ctx, o.logger)
 
 	// Skip processing children of subqueries - range vectors inside subqueries
@@ -77,7 +83,7 @@ func (o *OptimizationPass) wrapSplitRangeVectorFunctions(ctx context.Context, n
 
 	if functionCall, isFunctionCall := n.(*core.FunctionCall); isFunctionCall {
 		o.functionNodesInspected.Inc()
-		wrappedNode, notAppliedReason, err := o.trySplitFunction(functionCall, timeRange)
+		wrappedNode, notAppliedReason, err := o.trySplitFunction(functionCall, timeRange, enableSubquerySplitting)
 		if err != nil {
 			o.functionNodesUnsplit.WithLabelValues("error").Inc()
 			return nil, err
@@ -94,7 +100,7 @@ func (o *OptimizationPass) wrapSplitRangeVectorFunctions(ctx context.Context, n
 
 	for i := range n.ChildCount() {
 		child := n.Child(i)
-		newChild, err := o.wrapSplitRangeVectorFunctions(ctx, child, timeRange)
+		newChild, err := o.wrapSplitRangeVectorFunctions(ctx, child, timeRange, enableSubquerySplitting)
 		if err != nil {
 			return nil, err
 		}
@@ -116,7 +122,7 @@ func (o *OptimizationPass) wrapSplitRangeVectorFunctions(ctx context.Context, n
 // computed at materialize time (see Materializer.computeRanges), because they depend on the querier's current time and
 // the tenant's out-of-order window, as well as the exact time range being evaluated (which can vary if splitting
 // and caching applies).
-func (o *OptimizationPass) trySplitFunction(functionCall *core.FunctionCall, timeRange types.QueryTimeRange) (planning.Node, string, error) {
+func (o *OptimizationPass) trySplitFunction(functionCall *core.FunctionCall, timeRange types.QueryTimeRange, enableSubquerySplitting bool) (planning.Node, string, error) {
 	// For now, only support instant queries (range queries are more complex)
 	if !timeRange.IsInstant {
 		return nil, "range_query", nil
@@ -131,7 +137,16 @@ func (o *OptimizationPass) trySplitFunction(functionCall *core.FunctionCall, tim
 	}
 
 	inner, ok := functionCall.Child(0).(planning.SplitNode)
-	if !ok || !inner.IsSplittable() {
+	if !ok {
+		return nil, "unsupported_inner_node", nil
+	}
+
+	_, isSubquery := inner.(*core.Subquery)
+	if isSubquery && !enableSubquerySplitting {
+		return nil, "unsupported_inner_node", nil
+	}
+
+	if !inner.IsSplittable() {
 		return nil, "unsupported_inner_node", nil
 	}
 
diff --git a/pkg/streamingpromql/optimize/plan/rangevectorsplitting/promqltest_test.go b/pkg/streamingpromql/optimize/plan/rangevectorsplitting/promqltest_test.go
index dd00e0af31..2cf2bfc025 100644
--- a/pkg/streamingpromql/optimize/plan/rangevectorsplitting/promqltest_test.go
+++ b/pkg/streamingpromql/optimize/plan/rangevectorsplitting/promqltest_test.go
@@ -156,6 +156,11 @@ func skipUnsupportedTests(t *testing.T, testContent string, testFile string) str
   expect warn msg: PromQL warning: conflicting counter resets during histogram aggregation (1:31)
   expect no_info
   {} 10.5`,
+
+			// TODO: Precision is lost calculating avg_over_time across a block boundary, in cases
+			// of huge opposite-sign values (eg. +1e100/-1e100) that should cancel out exactly.
+			`eval instant at 6m avg_over_time(histogram_sum_over_time_incremental_4[7m:1m])
+    {} {{schema:0 count:3.9967044783747367e+307 sum:0.9}}`,
 		}
 
 	default:
diff --git a/pkg/streamingpromql/optimize/plan/rangevectorsplitting/range_vector_splitting_test.go b/pkg/streamingpromql/optimize/plan/rangevectorsplitting/range_vector_splitting_test.go
index 1c0dfd257b..7a854bb137 100644
--- a/pkg/streamingpromql/optimize/plan/rangevectorsplitting/range_vector_splitting_test.go
+++ b/pkg/streamingpromql/optimize/plan/rangevectorsplitting/range_vector_splitting_test.go
@@ -1528,6 +1528,7 @@ func createSplittingEngine(t *testing.T, registry *prometheus.Registry, splitInt
 	opts.Limits = limits
 	opts.RangeVectorSplitting.Enabled = true
 	opts.RangeVectorSplitting.SplitInterval = splitInterval
+	opts.RangeVectorSplitting.EnableSubquerySplitting = true
 	opts.CommonOpts.Reg = registry
 	if !enableEliminateDeduplicateAndMerge {
 		opts.EnableEliminateDeduplicateAndMerge = false
@@ -1613,6 +1614,7 @@ func defaultSplittingOpts() streamingpromql.EngineOpts {
 	opts := streamingpromql.NewTestEngineOpts()
 	opts.RangeVectorSplitting.Enabled = true
 	opts.RangeVectorSplitting.SplitInterval = 2 * time.Hour
+	opts.RangeVectorSplitting.EnableSubquerySplitting = true
 	return opts
 }
 
diff --git a/pkg/streamingpromql/optimize/plan/rangevectorsplitting/subquery_splitting_test.go b/pkg/streamingpromql/optimize/plan/rangevectorsplitting/subquery_splitting_test.go
new file mode 100644
index 0000000000..26fbe97b7c
--- /dev/null
+++ b/pkg/streamingpromql/optimize/plan/rangevectorsplitting/subquery_splitting_test.go
@@ -0,0 +1,344 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package rangevectorsplitting_test
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/model/timestamp"
+	"github.com/prometheus/prometheus/promql"
+	"github.com/prometheus/prometheus/promql/promqltest"
+	"github.com/stretchr/testify/require"
+
+	"github.com/grafana/mimir/pkg/querier/stats"
+	"github.com/grafana/mimir/pkg/streamingpromql"
+	"github.com/grafana/mimir/pkg/streamingpromql/planning"
+	"github.com/grafana/mimir/pkg/streamingpromql/testutils"
+	"github.com/grafana/mimir/pkg/streamingpromql/types"
+)
+
+func TestSubquery_IsSplittable(t *testing.T) {
+	planner, err := streamingpromql.NewQueryPlanner(defaultSplittingOpts(), streamingpromql.NewMaximumSupportedVersionQueryPlanVersionProvider())
+	require.NoError(t, err)
+
+	testCases := map[string]struct {
+		expr       string
+		splittable bool
+	}{
+		"plain selector nested inside the subquery": {
+			expr:       `sum_over_time(test_metric[5h:1h])`,
+			splittable: true,
+		},
+		"step-invariant expression with no selector nested inside the subquery": {
+			expr:       `sum_over_time(vector(1)[5h:1h])`,
+			splittable: true,
+		},
+		"smoothed matrix selector nested inside the subquery": {
+			expr:       `sum_over_time(rate(test_metric[3m] smoothed)[5h:1h])`,
+			splittable: false,
+		},
+		"smoothed vector selector nested inside the subquery": {
+			expr:       `sum_over_time((test_metric smoothed)[5h:1h])`,
+			splittable: false,
+		},
+		"anchored selector nested inside the subquery": {
+			expr:       `sum_over_time(rate(test_metric[3m] anchored)[5h:1h])`,
+			splittable: false,
+		},
+		"positive offset selector nested inside the subquery": {
+			expr:       `sum_over_time(rate(test_metric[3m] offset 10m)[5h:1h])`,
+			splittable: true,
+		},
+		"negative offset selector nested inside the subquery": {
+			expr:       `sum_over_time(rate(test_metric[3m] offset -10m)[5h:1h])`,
+			splittable: false,
+		},
+		"@ modifier selector nested inside the subquery": {
+			expr:       `sum_over_time((test_metric @ 100)[5h:1h])`,
+			splittable: false,
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			plan, err := planner.NewQueryPlan(t.Context(), tc.expr, types.NewInstantQueryTimeRange(timestamp.Time(0).Add(24*time.Hour)),
+				streamingpromql.DefaultLookbackDelta, false, &streamingpromql.NoopPlanningObserver{})
+			require.NoError(t, err)
+
+			require.Equal(t, tc.splittable, strings.Contains(plan.String(), "SplitFunctionCall"), "plan:\n%s", plan.String())
+		})
+	}
+}
+
+// TestQuerySplitting_InsertDuplicatesAcrossSplitBlocks checks that insertDuplicatesAcrossSplitBlocks (see commonsubexpressionelimination/optimization_pass.go)
+// wraps exactly the nodes it needs to in a Duplicate node. core.Subquery or core.StepInvariantExpression nested
+// below a split target's own child, at any depth, but not the split target's own child itself.
+func TestQuerySplitting_InsertDuplicatesAcrossSplitBlocks(t *testing.T) {
+	planner, err := streamingpromql.NewQueryPlanner(defaultSplittingOpts(), streamingpromql.NewMaximumSupportedVersionQueryPlanVersionProvider())
+	require.NoError(t, err)
+
+	testCases := map[string]struct {
+		expr         string
+		expectedPlan string
+	}{
+		"no nested subquery or step-invariant expression: nothing is wrapped": {
+			expr: `sum_over_time(max_over_time(test_metric[10m])[5h:1h])`,
+			expectedPlan: `
+				- SplitFunctionCall
+					- FunctionCall: sum_over_time(...)
+						- Subquery: [5h0m0s:1h0m0s]
+							- FunctionCall: max_over_time(...)
+								- MatrixSelector: {__name__="test_metric"}[10m0s]
+			`,
+		},
+		"subquery nested one level below the split target: only the nested subquery's own child is wrapped": {
+			expr: `count_over_time(sum_over_time(min_over_time(test_metric[2h])[20h:2h])[5h:12h])`,
+			expectedPlan: `
+				- SplitFunctionCall
+					- FunctionCall: count_over_time(...)
+						- Subquery: [5h0m0s:12h0m0s]
+							- FunctionCall: sum_over_time(...)
+								- Subquery: [20h0m0s:2h0m0s]
+									- Duplicate
+										- FunctionCall: min_over_time(...)
+											- MatrixSelector: {__name__="test_metric"}[2h0m0s]
+			`,
+		},
+		"binary expression with a constant nested below the split target: the whole expression is wrapped": {
+			expr: `count_over_time(sum_over_time((test_metric / 2)[3h:1h])[5h:12h])`,
+			expectedPlan: `
+				- SplitFunctionCall
+					- FunctionCall: count_over_time(...)
+						- Subquery: [5h0m0s:12h0m0s]
+							- FunctionCall: sum_over_time(...)
+								- Subquery: [3h0m0s:1h0m0s]
+									- Duplicate
+										- DeduplicateAndMerge
+											- BinaryExpression: LHS / RHS
+												- LHS: VectorSelector: {__name__="test_metric"}
+												- RHS: NumberLiteral: 2
+			`,
+		},
+		"subquery nested two levels below the split target: every nested level's own child is wrapped": {
+			expr: `count_over_time(sum_over_time(avg_over_time(min_over_time(test_metric[1h])[3h:30m])[10h:1h])[5h:12h])`,
+			expectedPlan: `
+				- SplitFunctionCall
+					- FunctionCall: count_over_time(...)
+						- Subquery: [5h0m0s:12h0m0s]
+							- FunctionCall: sum_over_time(...)
+								- Subquery: [10h0m0s:1h0m0s]
+									- Duplicate
+										- FunctionCall: avg_over_time(...)
+											- Subquery: [3h0m0s:30m0s]
+												- Duplicate
+													- FunctionCall: min_over_time(...)
+														- MatrixSelector: {__name__="test_metric"}[1h0m0s]
+			`,
+		},
+		"step-invariant expression nested below the split target: its child is wrapped": {
+			expr: `count_over_time(vector(1)[5h:3h])`,
+			expectedPlan: `
+				- DeduplicateAndMerge
+					- SplitFunctionCall
+						- FunctionCall: count_over_time(...)
+							- Subquery: [5h0m0s:3h0m0s]
+								- StepInvariantExpression
+									- Duplicate
+										- FunctionCall: vector(...)
+											- NumberLiteral: 1
+			`,
+		},
+		"step-invariant expression (vector(1)) as one operand of a binary expression: only that operand is wrapped": {
+			expr: `sum_over_time((vector(1) + on() test_metric)[5h:1h])`,
+			expectedPlan: `
+				- SplitFunctionCall
+					- FunctionCall: sum_over_time(...)
+						- Subquery: [5h0m0s:1h0m0s]
+							- BinaryExpression: LHS + on () RHS
+								- LHS: StepInvariantExpression
+									- Duplicate
+										- FunctionCall: vector(...)
+											- NumberLiteral: 1
+								- RHS: VectorSelector: {__name__="test_metric"}
+			`,
+		},
+		"nested subquery shared by subset selector elimination: its child is still wrapped exactly once": {
+			expr: `count_over_time((sum_over_time(max_over_time(dedupe_filter_metric{a="1"}[1h])[5h:1h]) / ignoring(a) min_over_time(max_over_time(dedupe_filter_metric[1h])[5h:1h]))[10h:12h])`,
+			expectedPlan: `
+				- SplitFunctionCall
+					- FunctionCall: count_over_time(...)
+						- Subquery: [10h0m0s:12h0m0s]
+							- BinaryExpression: LHS / ignoring (a) RHS, hints exclude (a)
+								- LHS: FunctionCall: sum_over_time(...)
+									- DuplicateFilter: {a="1"}, subset index: 0
+										- ref#1 Duplicate
+											- Subquery: [5h0m0s:1h0m0s]
+												- Duplicate
+													- FunctionCall: max_over_time(...)
+														- MatrixSelector: {__name__="dedupe_filter_metric"}[1h0m0s], subsets: {a="1"} ({__name__="dedupe_filter_metric", a="1"})
+								- RHS: FunctionCall: min_over_time(...)
+									- ref#1 Duplicate ...
+			`,
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			plan, err := planner.NewQueryPlan(t.Context(), tc.expr, types.NewInstantQueryTimeRange(timestamp.Time(0).Add(24*time.Hour)),
+				streamingpromql.DefaultLookbackDelta, false, &streamingpromql.NoopPlanningObserver{})
+			require.NoError(t, err)
+
+			require.Equal(t, testutils.TrimIndent(tc.expectedPlan), plan.String())
+		})
+	}
+}
+
+// TestQuerySplitting_MinimumRequiredPlanVersion verifies that a SplitFunctionCall reports QueryPlanV18 when it
+// wraps a plain selector and QueryPlanV21 when it wraps a subquery, in each case regardless of whether CSE has
+// inserted a Duplicate node between the SplitFunctionCall and what it wraps.
+func TestQuerySplitting_MinimumRequiredPlanVersion(t *testing.T) {
+	planner, err := streamingpromql.NewQueryPlanner(defaultSplittingOpts(), streamingpromql.NewMaximumSupportedVersionQueryPlanVersionProvider())
+	require.NoError(t, err)
+
+	testCases := map[string]struct {
+		expr            string
+		expectedPlan    string
+		expectedVersion planning.QueryPlanVersion
+	}{
+		"selector, no CSE duplication": {
+			expr: `sum_over_time(test_metric[5h])`,
+			expectedPlan: `
+				- SplitFunctionCall
+					- FunctionCall: sum_over_time(...)
+						- MatrixSelector: {__name__="test_metric"}[5h0m0s]
+			`,
+			expectedVersion: planning.QueryPlanV18,
+		},
+		"selector, CSE inserts Duplicate below SplitFunctionCall": {
+			expr: `sum_over_time(test_metric[5h]) / count_over_time(test_metric[5h])`,
+			expectedPlan: `
+				- BinaryExpression: LHS / RHS, hints exclude ()
+					- LHS: SplitFunctionCall
+						- FunctionCall: sum_over_time(...)
+							- ref#1 Duplicate
+								- MatrixSelector: {__name__="test_metric"}[5h0m0s]
+					- RHS: SplitFunctionCall
+						- FunctionCall: count_over_time(...)
+							- ref#1 Duplicate ...
+			`,
+			expectedVersion: planning.QueryPlanV18,
+		},
+		"subquery, no CSE duplication": {
+			expr: `sum_over_time(max_over_time(test_metric[10m])[5h:1h])`,
+			expectedPlan: `
+				- SplitFunctionCall
+					- FunctionCall: sum_over_time(...)
+						- Subquery: [5h0m0s:1h0m0s]
+							- FunctionCall: max_over_time(...)
+								- MatrixSelector: {__name__="test_metric"}[10m0s]
+			`,
+			expectedVersion: planning.QueryPlanV21,
+		},
+		"subquery, CSE inserts Duplicate below SplitFunctionCall": {
+			expr: `sum_over_time(max_over_time(test_metric[10m])[5h:1h]) / count_over_time(max_over_time(test_metric[10m])[5h:1h])`,
+			expectedPlan: `
+				- BinaryExpression: LHS / RHS, hints exclude ()
+					- LHS: SplitFunctionCall
+						- FunctionCall: sum_over_time(...)
+							- ref#1 Duplicate
+								- Subquery: [5h0m0s:1h0m0s]
+									- FunctionCall: max_over_time(...)
+										- MatrixSelector: {__name__="test_metric"}[10m0s]
+					- RHS: SplitFunctionCall
+						- FunctionCall: count_over_time(...)
+							- ref#1 Duplicate ...
+			`,
+			expectedVersion: planning.QueryPlanV21,
+		},
+		"subquery, CSE inserts Duplicate above SplitFunctionCall": {
+			expr: `sum_over_time(max_over_time(test_metric[10m])[5h:1h]) + sum_over_time(max_over_time(test_metric[10m])[5h:1h])`,
+			expectedPlan: `
+				- BinaryExpression: LHS + RHS, hints exclude ()
+					- LHS: ref#1 Duplicate
+						- SplitFunctionCall
+							- FunctionCall: sum_over_time(...)
+								- Subquery: [5h0m0s:1h0m0s]
+									- FunctionCall: max_over_time(...)
+										- MatrixSelector: {__name__="test_metric"}[10m0s]
+					- RHS: ref#1 Duplicate ...
+			`,
+			expectedVersion: planning.QueryPlanV21,
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			plan, err := planner.NewQueryPlan(t.Context(), tc.expr, types.NewInstantQueryTimeRange(timestamp.Time(0).Add(6*time.Hour)),
+				streamingpromql.DefaultLookbackDelta, false, &streamingpromql.NoopPlanningObserver{})
+			require.NoError(t, err)
+
+			require.Equal(t, testutils.TrimIndent(tc.expectedPlan), plan.String())
+			require.Equal(t, tc.expectedVersion, plan.Version)
+		})
+	}
+}
+
+// TestQuerySplitting_ConflictingDropNameAcrossSplits checks that FunctionOverRangeVectorSplit#mergeSplitsMetadata doesn't error
+// when subquery splitting and delayed name removal combine to produce different DropName values for the same series across splits.
+func TestQuerySplitting_ConflictingDropNameAcrossSplits(t *testing.T) {
+	testCases := map[string]struct {
+		expr                string
+		expectedPlainMetric labels.Labels
+		expectedSplitMetric labels.Labels
+	}{
+		"count_over_time: DropSeriesName overwrites the conflict, split matches unsplit": {
+			expr:                `count_over_time((rate(drop_name_foo[1h]) or label_replace(drop_name_bar, "__name__", "drop_name_foo", "", ""))[10h:1h])`,
+			expectedPlainMetric: labels.FromStrings("env", "prod"),
+			expectedSplitMetric: labels.FromStrings("env", "prod"),
+		},
+		"last_over_time: metadata passes through unchanged, split's __name__ diverges from unsplit": {
+			expr:                `last_over_time((rate(drop_name_foo[1h]) or label_replace(drop_name_bar, "__name__", "drop_name_foo", "", ""))[10h:1h])`,
+			expectedPlainMetric: labels.FromStrings("env", "prod"),
+			expectedSplitMetric: labels.FromStrings("__name__", "drop_name_foo", "env", "prod"),
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			opts := defaultSplittingOpts()
+			limits := streamingpromql.NewStaticQueryLimitsProvider()
+			limits.EnableDelayedNameRemoval = true
+			opts.Limits = limits
+
+			_, splitEngine := setupEngineAndCacheWithOpts(t, opts)
+
+			storage := promqltest.LoadedStorage(t, `
+				load 1h
+					drop_name_foo{env="prod"} _ _ _ _ _ 1 1 1 1 1 1
+					drop_name_bar{env="prod"} 1 1 1 1 1 1 1 1 1 1 1
+			`)
+			t.Cleanup(func() { require.NoError(t, storage.Close()) })
+
+			ts := timestamp.Time(0).Add(10 * time.Hour)
+
+			splitResult, _ := runInstantQuery(t, splitEngine, storage, tc.expr, ts)
+			require.NoError(t, splitResult.Err)
+
+			plainOpts := streamingpromql.NewTestEngineOpts()
+			plainOpts.Limits = limits
+			plainPlanner, err := streamingpromql.NewQueryPlanner(plainOpts, streamingpromql.NewMaximumSupportedVersionQueryPlanVersionProvider())
+			require.NoError(t, err)
+			plainEngine, err := streamingpromql.NewEngine(plainOpts, stats.NewQueryMetrics(plainOpts.CommonOpts.Reg), plainPlanner)
+			require.NoError(t, err)
+
+			plainResult, _ := runInstantQuery(t, plainEngine, storage, tc.expr, ts)
+			require.NoError(t, plainResult.Err)
+
+			require.Equal(t, tc.expectedPlainMetric, plainResult.Value.(promql.Vector)[0].Metric)
+			require.Equal(t, tc.expectedSplitMetric, splitResult.Value.(promql.Vector)[0].Metric)
+		})
+	}
+}
diff --git a/pkg/streamingpromql/optimize/plan/rangevectorsplitting/testdata/range_vector_splitting_2h.test b/pkg/streamingpromql/optimize/plan/rangevectorsplitting/testdata/range_vector_splitting_2h.test
index 26069ffbfb..c5d1486234 100644
--- a/pkg/streamingpromql/optimize/plan/rangevectorsplitting/testdata/range_vector_splitting_2h.test
+++ b/pkg/streamingpromql/optimize/plan/rangevectorsplitting/testdata/range_vector_splitting_2h.test
@@ -798,3 +798,233 @@ eval instant at 6h histogram_count(increase(hist_bucket_only_reset[6h]))
   {env="prod"} 12
 
 clear
+
+# ===== subqueries =====
+
+load 1h
+  subquery_metric{env="prod"} 1 2 3 4 5 6 7 8 9 10
+
+eval instant at 6h sum_over_time(max_over_time(subquery_metric[1h])[5h:1h])
+  {env="prod"} 25
+
+eval instant at 6h sum_over_time(max_over_time(subquery_metric[1h])[5h:1h])
+  {env="prod"} 25
+
+eval instant at 9h sum_over_time(max_over_time(subquery_metric[1h])[5h:1h])
+  {env="prod"} 40
+
+eval instant at 9h sum_over_time(max_over_time(subquery_metric[1h])[5h:1h])
+  {env="prod"} 40
+
+clear
+
+load 1h
+  subquery_metric_mod{env="prod"} 1 2 3 4 5 6 7
+
+eval instant at 6h sum_over_time(max_over_time(subquery_metric_mod[1h])[5h:1h])
+  {env="prod"} 25
+
+eval instant at 6h sum_over_time(max_over_time(subquery_metric_mod[1h])[5h:1h])
+  {env="prod"} 25
+
+eval instant at 7h sum_over_time(max_over_time(subquery_metric_mod[1h])[5h:1h] offset 1h)
+  {env="prod"} 25
+
+eval instant at 7h sum_over_time(max_over_time(subquery_metric_mod[1h])[5h:1h] offset 1h)
+  {env="prod"} 25
+
+clear
+
+# A subquery whose own inner expression is itself a subquery.
+
+load 1h
+  nested_metric{env="prod"} 1 2 3 4 5 6 7 8 9 10 11 12
+
+eval instant at 9h sum_over_time(min_over_time(max_over_time(nested_metric[1h])[3h:1h])[6h:1h])
+  {env="prod"} 33
+
+eval instant at 9h sum_over_time(min_over_time(max_over_time(nested_metric[1h])[3h:1h])[6h:1h])
+  {env="prod"} 33
+
+clear
+
+# A subquery nested inside another subquery, with a step that doesn't line up with the split interval.
+load 1h
+  nested_metric2{env="prod"} 1 2 3 4 5 6 7 8 9
+
+eval instant at 8h count_over_time(min_over_time(max_over_time(nested_metric2[1h])[3h:1h])[7h:3h])
+  {env="prod"} 2
+
+eval instant at 8h count_over_time(min_over_time(max_over_time(nested_metric2[1h])[3h:1h])[7h:3h])
+  {env="prod"} 2
+
+clear
+
+# Same shape, but with a step equal to the split interval.
+load 1h
+  nested_metric3{env="prod"} 1 2 3 4 5 6 7 8 9
+
+eval instant at 8h max_over_time(sum_over_time(min_over_time(nested_metric3[1h])[2h:1h])[6h:2h])
+  {env="prod"} 17
+
+eval instant at 8h max_over_time(sum_over_time(min_over_time(nested_metric3[1h])[2h:1h])[6h:2h])
+  {env="prod"} 17
+
+clear
+
+# query with subset selector elimination
+load 1h
+  sse_subquery_metric{env="prod", code="ok"} 1 2 3 4 5 6 7 8 9 10
+  sse_subquery_metric{env="prod", code="err"} 100 200 300 400 500 600 700 800 900 1000
+
+eval instant at 6h sum_over_time(max_over_time(sse_subquery_metric{code!="err"}[1h])[5h:1h]) / min_over_time(max_over_time(sse_subquery_metric[1h])[5h:1h])
+  {env="prod", code="ok"} 8.33333333333333
+
+eval instant at 6h sum_over_time(max_over_time(sse_subquery_metric{code!="err"}[1h])[5h:1h]) / min_over_time(max_over_time(sse_subquery_metric[1h])[5h:1h])
+  {env="prod", code="ok"} 8.33333333333333
+
+clear
+
+# count_over_time( sum_over_time( min_over_time(hour_collision_metric[2h]) [20h:2h] ) [5h:12h] )
+#
+#  S1, outer subquery:  sum_over_time(S2)                        [5h:12h]  range 5h,  step 12h
+#  S2, inner subquery:  min_over_time(hour_collision_metric[2h]) [20h:2h]  range 20h, step 2h
+#
+#      19h        20h                                 24h
+#      |-- Head --|----- Block1 + Block2, 2 steps -----|
+#        (~1h)     (20h->22h step, then 22h->24h step)
+#
+# S1 12h step aligns Head end (19h59m59.999s) and Block1+Block2 end (23h59m59.999s) to
+# the same instant, 24h. That shared instant becomes S2's own reference point, so S2 computes the identical 10-step
+# window regardless of which piece asked. Head range has zero steps of its own, so it never actually reads through this path,
+# but materialization happens regardless of that, so the same S2 operator still gets materialized for Head as for Block1+Block2.
+# To mitigate this, rvs wraps S2's own child (not S2 itself) in a Duplicate node.
+load 1h
+  hour_collision_metric{env="prod"} 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30
+
+eval instant at 24h count_over_time(sum_over_time(min_over_time(hour_collision_metric[2h])[20h:2h])[5h:12h])
+  {env="prod"} 1
+
+eval instant at 24h count_over_time(sum_over_time(min_over_time(hour_collision_metric[2h])[20h:2h])[5h:12h])
+  {env="prod"} 1
+
+clear
+
+# Different split blocks share the same step-invariant expression (eg. from the @ modifier).
+load 1h
+  step_invariant_metric{env="prod"} 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30
+
+eval instant at 24h count_over_time(sum_over_time(step_invariant_metric[2h] @ 10h)[5h:3h])
+  {env="prod"} 2
+
+eval instant at 24h count_over_time(sum_over_time(step_invariant_metric[2h] @ 10h)[5h:3h])
+  {env="prod"} 2
+
+# @ applied directly to the selector feeding the subquery, with no function call in between.
+eval instant at 24h count_over_time((step_invariant_metric @ 10h)[5h:3h])
+  {env="prod"} 2
+
+eval instant at 24h count_over_time((step_invariant_metric @ 10h)[5h:3h])
+  {env="prod"} 2
+
+# Step-invariant expression nested two levels deep, inside a subquery nested inside another split subquery.
+eval instant at 24h count_over_time(sum_over_time(avg_over_time(step_invariant_metric[1h] @ 10h)[3h:30m])[5h:12h])
+  {env="prod"} 1
+
+eval instant at 24h count_over_time(sum_over_time(avg_over_time(step_invariant_metric[1h] @ 10h)[3h:30m])[5h:12h])
+  {env="prod"} 1
+
+clear
+
+# vector(1) is step-invariant without an @ modifier
+load 1h
+  vector_binop_metric{env="prod"} 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30
+
+eval instant at 24h sum_over_time((vector(1) + on() vector_binop_metric)[5h:1h])
+  {} 120
+
+eval instant at 24h sum_over_time((vector(1) + on() vector_binop_metric)[5h:1h])
+  {} 120
+
+clear
+
+# shared, split subquery deduplicated by common subexpression elimination
+load 1h
+  cse_subquery_metric{env="prod", code="ok"} 1 2 3 4 5 6 7 8 9 10
+
+eval instant at 6h sum_over_time(max_over_time(cse_subquery_metric[1h])[5h:1h]) / count_over_time(max_over_time(cse_subquery_metric[1h])[5h:1h])
+  {env="prod", code="ok"} 5
+
+eval instant at 6h sum_over_time(max_over_time(cse_subquery_metric[1h])[5h:1h]) / count_over_time(max_over_time(cse_subquery_metric[1h])[5h:1h])
+  {env="prod", code="ok"} 5
+
+clear
+
+# Binary operation over a shared, step-invariant subquery: CSE sharing and split-block sharing together.
+load 1h
+  cse_stepinvariant_metric{env="prod", code="ok"} 1 2 3 4 5 6 7 8 9 10
+
+eval instant at 6h sum_over_time(max_over_time(cse_stepinvariant_metric[1h] @ 3h)[5h:1h]) / count_over_time(max_over_time(cse_stepinvariant_metric[1h] @ 3h)[5h:1h])
+  {env="prod", code="ok"} 4
+
+eval instant at 6h sum_over_time(max_over_time(cse_stepinvariant_metric[1h] @ 3h)[5h:1h]) / count_over_time(max_over_time(cse_stepinvariant_metric[1h] @ 3h)[5h:1h])
+  {env="prod", code="ok"} 4
+
+clear
+
+load 1h
+  sse_above_subquery_metric{env="prod", a="1"} 1 2 3 4 5 6 7 8 9 10
+  sse_above_subquery_metric{env="prod", a="2"} 100 200 300 400 500 600 700 800 900 1000
+
+eval instant at 6h sum_over_time(max_over_time(sse_above_subquery_metric{a="1"}[1h])[5h:1h]) / sum_over_time(max_over_time(sse_above_subquery_metric[1h])[5h:1h])
+  {env="prod", a="1"} 1
+
+eval instant at 6h sum_over_time(max_over_time(sse_above_subquery_metric{a="1"}[1h])[5h:1h]) / sum_over_time(max_over_time(sse_above_subquery_metric[1h])[5h:1h])
+  {env="prod", a="1"} 1
+
+clear
+
+# A DuplicateFilter ends up directly under a Subquery, shared between two split subqueries.
+load 1h
+  dup_filter_metric{a="1"} 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30
+
+eval instant at 24h count_over_time(sum_over_time((dup_filter_metric{a="1"})[20h:2h])[5h:12h]) + on() sum(sum_over_time((dup_filter_metric)[20h:2h]))
+  {} 151
+
+eval instant at 24h count_over_time(sum_over_time((dup_filter_metric{a="1"})[20h:2h])[5h:12h]) + on() sum(sum_over_time((dup_filter_metric)[20h:2h]))
+  {} 151
+
+clear
+
+# The subquery here emits unsorted series (RHS before LHS), check FunctionOverRangeVectorSplit#mergeSplitsMetadata handles that.
+load 1h
+  unsorted_merge_a{id="a"} 1 2 3 4 5
+  unsorted_merge_b{id="b"} 10 20 30 40 50
+
+eval instant at 5h sum_over_time((unsorted_merge_a or unsorted_merge_b)[5h:1h])
+  {id="a"} 14
+  {id="b"} 140
+
+eval instant at 5h sum_over_time((unsorted_merge_a or unsorted_merge_b)[5h:1h])
+  {id="a"} 14
+  {id="b"} 140
+
+clear
+
+# Same as above, but with three chained "or"
+load 1h
+  unsorted_merge_3way_a{id="a"} 1 2 3 4 5
+  unsorted_merge_3way_b{id="b"} 10 20 30 40 50
+  unsorted_merge_3way_c{id="c"} 100 200 300 400 500
+
+eval instant at 5h sum_over_time((unsorted_merge_3way_a or unsorted_merge_3way_b or unsorted_merge_3way_c)[5h:1h])
+  {id="a"} 14
+  {id="b"} 140
+  {id="c"} 1400
+
+eval instant at 5h sum_over_time((unsorted_merge_3way_a or unsorted_merge_3way_b or unsorted_merge_3way_c)[5h:1h])
+  {id="a"} 14
+  {id="b"} 140
+  {id="c"} 1400
+
+clear
diff --git a/pkg/streamingpromql/planning.go b/pkg/streamingpromql/planning.go
index 9949fa17d1..0cd575b77f 100644
--- a/pkg/streamingpromql/planning.go
+++ b/pkg/streamingpromql/planning.go
@@ -122,7 +122,11 @@ func NewQueryPlanner(opts EngineOpts, versionProvider QueryPlanVersionProvider)
 			return nil, errors.New("range vector splitting and common subexpression elimination are enabled but range query range vector common subexpression elimination is not enabled")
 		}
 
-		planner.RegisterQueryPlanOptimizationPass(rangevectorsplitting.NewOptimizationPass(splitInterval, opts.CommonOpts.Reg, opts.Logger))
+		if opts.RangeVectorSplitting.EnableSubquerySplitting && !opts.EnableCommonSubexpressionElimination {
+			return nil, errors.New("cannot enable subquery splitting in range vector splitting without common subexpression elimination")
+		}
+
+		planner.RegisterQueryPlanOptimizationPass(rangevectorsplitting.NewOptimizationPass(splitInterval, opts.RangeVectorSplitting.EnableSubquerySplitting, opts.CommonOpts.Reg, opts.Logger))
 	}
 
 	// This optimization pass must be registered before common subexpression elimination, if that is enabled.
diff --git a/pkg/streamingpromql/planning/core/subquery.go b/pkg/streamingpromql/planning/core/subquery.go
index ba6c390b13..847fe1d274 100644
--- a/pkg/streamingpromql/planning/core/subquery.go
+++ b/pkg/streamingpromql/planning/core/subquery.go
@@ -54,6 +54,49 @@ func (s *Subquery) ChildrenTimeRange(timeRange types.QueryTimeRange) types.Query
 	return SubqueryChildrenTimeRange(timeRange, s.Range, s.Step, s.Offset, s.Timestamp)
 }
 
+func (s *Subquery) IsSplittable() bool {
+	return !hasUnsafeTimeModifier(s.Inner)
+}
+
+// hasUnsafeTimeModifier reports whether subtree contains a selector whose time range isn't safe to split:
+// a negative offset, an `@` timestamp, or a `smoothed`/`anchored` matrix selector.
+// Splitting a subquery containing one of these can produce a cache entry that goes stale once matching data lands.
+func hasUnsafeTimeModifier(node planning.Node) bool {
+	switch n := node.(type) {
+	case *MatrixSelector:
+		return n.Anchored || n.Smoothed || n.Offset < 0 || n.Timestamp != nil
+	case *VectorSelector:
+		return n.Smoothed || n.Offset < 0 || n.Timestamp != nil
+	case *Subquery:
+		if n.Offset < 0 || n.Timestamp != nil {
+			return true
+		}
+	}
+
+	for child := range planning.ChildrenIter(node) {
+		if hasUnsafeTimeModifier(child) {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (s *Subquery) GetRangeParams() planning.RangeParams {
+	params := planning.RangeParams{
+		IsSet:  true,
+		Range:  s.Range,
+		Offset: s.Offset,
+	}
+	if s.Timestamp != nil {
+		params.HasTimestamp = true
+		params.Timestamp = *s.Timestamp
+	}
+	return params
+}
+
+var _ planning.SplitNode = &Subquery{}
+
 // SubqueryChildrenTimeRange computes the time range used by the children of a subquery with the given
 // range, step, offset and @ timestamp (ts, nil if the subquery does not use the @ modifier), when the
 // subquery is evaluated over parentTimeRange.
@@ -118,14 +161,29 @@ func (s *Subquery) MergeHints(_ planning.Node) error {
 	return nil
 }
 
-func MaterializeSubquery(ctx context.Context, s *Subquery, materializer *planning.Materializer, timeRange types.QueryTimeRange, params *planning.OperatorParameters) (planning.OperatorFactory, error) {
-	innerTimeRange := s.ChildrenTimeRange(timeRange)
+func MaterializeSubquery(ctx context.Context, s *Subquery, materializer *planning.Materializer, timeRange types.QueryTimeRange, params *planning.OperatorParameters, overrideRangeParams planning.RangeParams) (planning.OperatorFactory, error) {
+	subqueryRange := s.Range
+	subqueryOffset := s.Offset
+	subqueryTimestamp := s.Timestamp
+
+	if overrideRangeParams.IsSet {
+		subqueryRange = overrideRangeParams.Range
+		subqueryOffset = overrideRangeParams.Offset
+		if overrideRangeParams.HasTimestamp {
+			subqueryTimestamp = &overrideRangeParams.Timestamp
+		} else {
+			subqueryTimestamp = nil
+		}
+	}
+
+	innerTimeRange := SubqueryChildrenTimeRange(timeRange, subqueryRange, s.Step, subqueryOffset, subqueryTimestamp)
+
 	inner, err := materializer.ConvertNodeToInstantVectorOperator(ctx, s.Inner, innerTimeRange)
 	if err != nil {
 		return nil, fmt.Errorf("could not create inner operator for Subquery: %w", err)
 	}
 
-	o, err := operators.NewSubquery(inner, timeRange, innerTimeRange, TimestampFromTime(s.Timestamp), s.Offset, s.Range, s.GetExpressionPosition().ToPrometheusType(), params.MemoryConsumptionTracker)
+	o, err := operators.NewSubquery(inner, timeRange, innerTimeRange, TimestampFromTime(subqueryTimestamp), subqueryOffset, subqueryRange, s.GetExpressionPosition().ToPrometheusType(), params.MemoryConsumptionTracker)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/streamingpromql/planning/plan.go b/pkg/streamingpromql/planning/plan.go
index d32b82ab40..388f153424 100644
--- a/pkg/streamingpromql/planning/plan.go
+++ b/pkg/streamingpromql/planning/plan.go
@@ -104,7 +104,11 @@ const QueryPlanV19 = QueryPlanVersion(19)
 // incorrect results.
 const QueryPlanV20 = QueryPlanVersion(20)
 
-var MaximumSupportedQueryPlanVersion = QueryPlanV20
+// QueryPlanV21 introduces support for splitting subqueries in range vector splitting, in addition
+// to range vector selectors.
+const QueryPlanV21 = QueryPlanVersion(21)
+
+var MaximumSupportedQueryPlanVersion = QueryPlanV21
 
 type QueryPlan struct {
 	Root       Node