Skip to content

[DQE-1] Core Planning#7688

Open
eeldaly wants to merge 1 commit into
cortexproject:masterfrom
eeldaly:dqe-split/1-core-planning
Open

[DQE-1] Core Planning#7688
eeldaly wants to merge 1 commit into
cortexproject:masterfrom
eeldaly:dqe-split/1-core-planning

Conversation

@eeldaly

@eeldaly eeldaly commented Jul 14, 2026

Copy link
Copy Markdown
Contributor

This PR is part 1 of 6 to get the changes made here by Ruby (https://github.com/rubywtl/cortex/commits?author=rubywtl) upstreamed and working for the distributed query engine path.


Add the core building blocks for distributed query execution (fragment_metadata, deduplicate_node/ShardedRemoteExecutions, distributed_optimizer, codec) and the QFE-level -querier.distributed-exec-shard-count flag that drives how many shards the optimizer fans a shardable aggregation into (0 disables it).

What this PR does:

Which issue(s) this PR fixes:
Fixes #

Checklist

  • Tests updated
  • Documentation added
  • CHANGELOG.md updated - the order of entries should be [CHANGE], [FEATURE], [ENHANCEMENT], [BUGFIX]
  • docs/configuration/v1-guarantees.md updated if this PR introduces experimental flags

// port notes): the node has no JSON (un)marshaling for codec round-tripping,
// and the shard matcher injected by the optimizer does not match the merged
// metric-name-hash storage sharding mechanism.
type ShardedRemoteExecutions struct {

@harry671003 harry671003 Jul 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need a separate node for this?

I'd expect sum(metric) etc to be optimized into:

sum(
  coalesce(
    sum(metric{shard0}),
    sum(metric{shard1}),
  )
)

subqueries := newRemoteAggregation(aggr, shardCount)
*current = &logicalplan.Aggregation{
Op: parser.SUM,
Expr: &ShardedRemoteExecutions{Expressions: subqueries},

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Comment thread pkg/distributed_execution/distributed_optimizer.go Outdated
// fragment into the context. This metadata is propagated from the query-scheduler to
// the querier so that the querier knows which fragment it is executing, where to pull
// its child fragments' results from, and whether it is the root (coordinator) fragment.
func InjectFragmentMetaData(ctx context.Context, fragmentID uint64, queryID uint64, isRoot bool, childIDToAddr map[uint64]string) context.Context {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is context the right place to put this? I'd like to understand the usecase better.

…hard-count flag

Add the core building blocks for distributed query execution (fragment_metadata,
deduplicate_node/ShardedRemoteExecutions, distributed_optimizer, codec) and the
QFE-level -querier.distributed-exec-shard-count flag that drives how many shards
the optimizer fans a shardable aggregation into (0 disables it). The optimizer
tags per-shard subqueries with the __CORTEX_DQE_SHARD__ selector.

Signed-off-by: Essam Eldaly <eeldaly@amazon.com>
@eeldaly
eeldaly force-pushed the dqe-split/1-core-planning branch from 236eb67 to 990b5e8 Compare July 17, 2026 21:00
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants