feat(dataobj-compactor): IndexMerge executor over postings + stats sections - #22233
Conversation
loki-gh-app
Bot
added
the
type/docs
|
💻 Deploy preview available (feat(engine/executor,compactor): IndexMerge executor over postings + stats sections): |
…stats sections
Replaces the zero-byte IndexMerge stub with the real K-way merge over
index sections (postings + stats), producing one merged index object
per task, and wires the compaction worker to drive it end to end.
Executor (executeIndexMerge):
1. Existence-check on OutputIndexPath via bucket.Exists, so retries
short-circuit idempotently.
2. Per-task TTL deadline via context.WithTimeout(node.TaskTTL).
3. classifyRuns scans every section of each referenced source object
(deduped by ObjectPath, opened once) and routes each by type to a
postings or stats pile, skipping streams/pointers/indexPointers.
SectionIndex is the planner's stable tiebreaker, not a dispatch
hint, so dispatch must be by section type. A pre-flight pass
asserts all stats sections share one SortSchema (the K-way merge
comparator requires a total order).
4. K-way merge over postings piles using the util/loser tournament
tree (the same primitive sortmerge uses for log sections), with a
last-wins reducer that records full-key collisions via xcap.
5. K-way merge over stats piles with an aggregating reducer:
min(min)/max(max)/sum(row_count)/sum(uncompressed_size) on
equal-key rows.
6. Build via indexobj.MergeBuilder and stream-upload the resulting
object directly (no intermediate buffering). On ErrBuilderEmpty,
upload a zero-byte sentinel so retries short-circuit on Exists.
executor.Config grows ScratchStore + IndexobjCfg (grouped by
use-case: query tuning / shared plumbing / query-side / compaction-
side). Both are zero-value-safe for query-only callers.
Wiring: the compaction worker threads ScratchStore (from the operator's
configured scratch backend) and the new Cfg.DataObj.Compaction.
IndexobjBuilder config end-to-end through enginecompactor.WorkerParams,
engine/worker, internal/worker, and the per-task executor.Config.
DataObjCompactionWorker now depends on the ScratchStore module so the
store is initialized first.
The coordinator end-to-end test seeds real index objects so the real
executor has sections to merge and asserts the merged outputs exist
after the ToC swap.
A per-task execution deadline is a task-orchestration concern, not part of the physical plan the executor consumes. Carrying it on the node and enforcing it in executeIndexMerge promotes orchestration policy into the compute layer. Remove the context.WithTimeout(node.TaskTTL) guard (and its dedicated test) from the executor. The deadline will be reintroduced at the orchestration layer (the wire Task envelope, alongside MaxTimeRange, which already carries execution-control metadata outside the fragment) as separate work covering both IndexMerge and LogMerge uniformly. The TaskTTL field remains on physical.IndexMerge for now; its removal and the orchestration-layer replacement are tracked separately.
5acee0d to
a2ae6c8
Compare
rfratto
left a comment
rfratto
left a comment
There was a problem hiding this comment.
first review pass, I need to take a much closer look at the index_merge*.go files
| defaults := NewConfig() | ||
| if cfg.MaxRunningCompactionTasks == 0 && cfg.IndexobjBuilder.TargetObjectSize == 0 { | ||
| cfg.MaxRunningCompactionTasks = defaults.MaxRunningCompactionTasks | ||
| cfg.Scheduler.Endpoint = defaults.Scheduler.Endpoint | ||
| cfg.IndexobjBuilder = defaults.IndexobjBuilder | ||
| } |
There was a problem hiding this comment.
We've never really done a NewConfig pattern before, and it also doesn't work here: defining a flag will pre-set the pointer with the default value. I'd suggest getting rid of NewConfig and letting flag registration drive defaults, and find other solutions for the other call sites which are calling NewConfig().
| @@ -37,14 +39,16 @@ type StreamFilterer interface { | |||
| } | |||
|
|
|||
| type Config struct { | |||
There was a problem hiding this comment.
The added comments here are a little wonky, now the doc comment for BatchSize is "Query execution tuning" which is certainly strange.
It looks like they're meant to be markers for the struct field groupings rather than actual documentation on fields. Can we add blank lines after them?
Also, I don't know what the difference between "Query execution tuning" and "Query-side plumbing" is lol
There was a problem hiding this comment.
yeah, there's no real distinction there other than the "plumbing" was added later. I'll clean these up. The goal is to call out which is used by compaction vs query to point out the start of a smell I was getting these belong in different types, but I don't think the lines are clear enough/warranted in this PR.
| // IndexMerge statistics. Count the number of equal-key collisions observed | ||
| // during a K-way index merge. A non-zero value indicates the upstream | ||
| // invariant, that each (ObjectPath, SectionIndex) is referenced by exactly one | ||
| // source index, has been violated — the merge tolerates this via last-wins, but | ||
| // we want to know how often it happens. | ||
| var ( | ||
| StatIndexMergeDuplicatePostings = NewStatisticInt64("index.merge.duplicate.postings", AggregationTypeSum) | ||
| StatIndexMergeDuplicateStats = NewStatisticInt64("index.merge.duplicate.stats", AggregationTypeSum) | ||
| ) |
There was a problem hiding this comment.
nit: defining stats centrally in xcap is a pattern I'm trying to get rid of; have an agent look at what pkg/engine does now for defining + reporting stats.
trevorwhitney
changed the title
rfratto
left a comment
rfratto
left a comment
There was a problem hiding this comment.
Here's the other review pass :)
| } | ||
|
|
||
| // statsPileReader reads stats.Stat records from a stats section in order. | ||
| type statsPileReader struct { |
There was a problem hiding this comment.
This threw me off so much and it took me a while to figure out why. I think it's because the pile term is being introduced here, but the API being implemented has nothing to do with piles, it's purely a row-based iterator over the columnar reader.
I'd suggest two things:
-
Remove
pileIdxfrom the type, since it's only being stored to carry around state on behalf of one specific call site. -
Move this to the stats package, calling it either
stats.RowReader,stats.Iterator,stats.RowAdapter, or something that makes it more clear specifically what logic is being implemented.
Same comment applies to the type in index_merge_postings.go.
There was a problem hiding this comment.
yup, I removed all mention of pile here and moved these to row readers in their respective package. reads much better now, very helpful feedback!
| // are reduced into a single record before being yielded. | ||
| // | ||
| // on return (success or error) every pile's Close is called. | ||
| func merge[R any]( |
There was a problem hiding this comment.
The generics made this a bit hard to understand, but I think I have a few suggestions for this:
The reducer logic kicks in when the current value is the same as the previous value. That can be done at the call site; the merge doesn't need to be doing that.
Removing the reducer logic from merge should simplify the state quite a bit. After you've refactored, you might find that we don't need to have a merge function at all. But if you think a merge function is still justifiable, I'd ask that we make some internal type definitions for cmp and the return iterator, which IMO would make the signature easier to understand.
BTW: In my last comment I suggested removing pileIdx from the row iterator adapter. You can encapsulate that here instead, since pileIdx matches up with the index in piles:
type positionedSeq struct {
Seq sequence[R]
Index int
}
posSequences := make([]positionedSeq, len(seq))
for i := range seq {
posSeqeunces[i] = positionedSeq{Seq: seq[i], Index: i}
}
// use posSequences in loser tree instead so you can access
// their index for tie breaking. Though you actually might not need to use index as a tie breaker. I'd ask Bryan if it's necessary or if loser tree handles that automatically and guarantees stable ordering if there's equal keys.
There was a problem hiding this comment.
after removing the reduce function that was just a closure around metrics anyway, this was greatly simplified to the point of being able to inline it and just invoke the loser tree directly at both call sites. @bboreham any thoughts on the need for the index as a tie breaker?
There was a problem hiding this comment.
I don't think it's stable; this line implies to me that two sequences that compare equal will alternate.
However it also doesn't seem hard to add a line there to use the index (as already known to loser tree) if it had a tri-state compare function. Since this version is inside Loki it's low-impact to try the change.
There was a problem hiding this comment.
seems worth doing at some point but I'm kicking that can for now, but I asked AI to create an issue #22285 (and it created a very verbose one, surprise surprise)
Spec and plan for moving the index-merge pile row iterators into the postings/stats section packages as reusable RowReader types, with a thin indexedSeq adapter in the executor to carry the merge pile index.
The executor's bespoke rowReader[R] interface duplicated the existing pkg/iter/v2.CloseIterator[T] (Next/At/Err/Close). Rename RowReader.Value to At so postings.RowReader and stats.RowReader implement iter.CloseIterator, and have pileSequence embed it rather than redeclaring the same method set.
|
💻 Deploy preview deleted (feat(dataobj-compactor): IndexMerge executor over postings + stats sections). |
rfratto
left a comment
rfratto
left a comment
There was a problem hiding this comment.
Much easier to follow :) Up to you if you want to try stable sorting in the loser tree directly or merge as is!
3 participants
What this does
Part 5 (final) of splitting up #21972 (IndexMerge executor) into reviewable PRs.
Replaces the zero-byte
IndexMergestub with the real K-way merge over index sections (postings + stats), producing one merged index object per task, and wires the compaction worker to drive it end to end.Executor (
executeIndexMerge)OutputIndexPathviabucket.Exists— retries short-circuit idempotently, making compaction re-entrant.context.WithTimeout(node.TaskTTL).classifyRunsscans every section of each referenced source object (deduped byObjectPath, opened once) and routes each by section type to a postings or stats pile, skipping streams/pointers/indexPointers.SectionIndexis the planner's stable tiebreaker, not a dispatch hint. A pre-flight pass asserts all stats sections share oneSortSchema(the K-way merge comparator requires a total order).util/losertournament tree (same primitivesortmergeuses for log sections), with a last-wins reducer that records full-key collisions viaxcap.min/max/sum/sumon equal-key rows.indexobj.MergeBuilderdirectly (no intermediate buffering). OnErrBuilderEmpty, upload a zero-byte sentinel.Wiring
executor.ConfiggrowsScratchStore+IndexobjCfg(grouped by use-case), both zero-value-safe for query-only callers.ScratchStore(operator's configured scratch backend) and the newCfg.DataObj.Compaction.IndexobjBuilderconfig end-to-end throughenginecompactor.WorkerParams→engine/worker→internal/worker→ per-taskexecutor.Config.DataObjCompactionWorkernow depends on theScratchStoremodule so the store initializes first.