Let a merge write more than one segment - #16571
Conversation
A merge takes N segments and writes one. This lets it take N and write M, by giving each output a contiguous document range of every input: public boolean isPartitioned() public int[][] getDocRangePartitions(List<CodecReader> readers) The returned array has one entry per input: a non-decreasing array of M + 1 boundaries, where output o takes [b[o], b[o+1]). It is called after the merge readers are open, so boundaries can be placed on values read from the readers rather than on document counts. Returning null is today's behaviour. An index sort is not required. It is what makes an output's documents a range of keys rather than an arbitrary slice, which is what most callers will want, but the split is well defined without one: the documents are shared out by position. Each input is wrapped in a reader presenting everything outside an output's range as deleted, so every format merges through its existing code and none of them learns about partitioning. An ordinary merge becomes the one-output case of the same path, which is why packaging a finished segment and committing a merge's outputs each become one method rather than two. Bytes written are unchanged, since every document is written once into whichever output owns it. Merge inputs are verified once for the whole merge rather than once per output, which otherwise repeats a full read of them per output. Boundaries may not cut inside a document block when a parent field is set: a block's children would end up in a different segment from their parent. Enforced on the offsets, since whether a caller's key-derived offset falls before the children or between them and their parent depends on how the children are indexed. Each output is tested by duelling it against a segment built from exactly its documents in the same order, across all field types, plus sorted and unsorted indexes, blocks both aligned and misaligned, concurrent deletes and doc-values updates arriving mid-merge, soft deletes, rollback, an IO failure while writing one output, malformed partitions, and wrapper preservation. Codecs are randomized.
Masking is enough for correctness but not for cost. Merging a column walks it with nextDoc() and drops whatever the document map sends to -1, having already read it, so each output of a partitioned merge read all of every column and k outputs read them k times. Seeking to the range instead makes the outputs together read each column once. Measured over a 64-way split, against what a single-output merge of the same inputs reads: norms 1.02x, numeric doc values 1.39x, where without this both are linear in the output count. For vectors it is the difference that matters most, since they are most of a vector index: a 768-dimensional float vector is three kilobytes a document against a norm's one byte. Only the iteration is restricted -- ordinals still belong to the reader underneath, so vectorValue(ord) and size() are unchanged, and a caller reaches a value through the ordinal an iterator gave it rather than by counting. Stored fields and term vectors need nothing: their merge is already a random access per live document, so an output only decompresses the chunks its own range falls in, plus the two it straddles. That is a constant penalty rather than one growing with the output count, measured at 1.5x and 3.6x and flat from two outputs to sixty-four. Deduplicating finishMerge() now has to look through the narrowing to the reader underneath, since every output narrows the same one.
The simplest example of what N-to-M merges allow, not the only one: forceMerge(n) says how many segments to leave but nothing about how the documents are shared between them. Merging a hundred equal segments into four leaves 97,000, 1,000, 1,000 and 1,000, and since a search is parallelised by segment the query waits on the one holding 97% of the index. BalancedSegmentsMergePolicy wraps another policy, changes only forced merges, and plans the whole one at once: segments are packed into groups and each group given the outputs its size deserves. The result is several independent merges the scheduler can run concurrently, not one merge of everything, and a group already the right size is skipped, so a nearly balanced index is balanced without being rewritten. Since the output count is per merge, a merge may have a single input: forceMerge(16) on a one-segment index splits it into sixteen. Boundaries are placed on live documents rather than document counts, so deletes -- which the merge drops -- do not eat a share, and are moved to block ends where a parent field is set. Replanning is refused while a plan is running: the plan is made for the whole index, so planning again over what is left of it would give the outputs still being written a second share of the segment count.
|
Interesting. I miss what are the benefits/ use cases for this? |
It wrapped each merge in a plain OneMerge, so a partitioned merge coming from the policy it wraps silently became an ordinary one -- the hazard FilterOneMerge exists for. Wrap with that instead. Reordering is then skipped for a partitioned merge rather than applied: the merge has decided which documents each of its outputs holds, and renumbering them all would undo that. This follows what the policy already does when there is not enough RAM.
Mostly range routing. You route documents by a key, a tenant id for instance, and sort the index on it. The index sort gives you contiguity inside a segment, but a tenant's documents are still spread over all of them, and merging can't fix that. A merge takes N segments and writes one, so the output covers the union of the ranges its inputs covered. It only gets wider, never narrower. With M outputs you can cut on the key instead, so each output owns a disjoint range, and a tenant's data ends up in a bounded set of segments and stays that way as merging continues. That's what makes resharding possible. When an index gets too big for one shard you split it by key range, and if the segments already own those ranges you hand whole segments to each side. Otherwise you copy everything and delete what doesn't belong, on both sides. This is how LSM stores already work. HBase regions are key ranges and a split is just a boundary operation, the daughter regions reference the parent files instead of copying them. Cassandra orders by token so a range can be streamed rather than rebuilt. What makes it work is that a compaction reads N files and writes M, cut on key boundaries. Lucene is the one where a merge always collapses to a single output. The merge policy in the PR is just something concrete to look at, it doesn't need a sort and splits by position. I'm also curious whether a partition following the vector space would help vector search, but I haven't tried it. |
The suite exercised sorted indexes but always cut on document counts, so what it covered was that the outputs are correct, not that a boundary placed on a key means anything. Each input holds its own mix of tenants and is sorted by them, so the cut offsets differ per input. Taking the same tenant in every input gives one output the whole of that key range: the outputs come out individually sorted and holding ranges that do not overlap. Cutting on document counts instead makes it fail with overlapping ranges.
|
Ok, get it. Thanks. |
|
I'm also intersted in this. Range routine in a leveldb/rocksdb like stack could be a big improvement for update throughput at lower cost than approximate membership data structures like bloom filters. Multi-output merge could also be useful for vector indexes where you might like the segments to be arranged by some clustering. |
|
Also, multi-tenant indexing is a use case, although that could be solved with carefully structured ids and range partitioning like describe here. |
michaeljmarshall
left a comment
michaeljmarshall
left a comment
There was a problem hiding this comment.
Posting a partial review. Most comments are optimizations that can be deferred to later.
| FixedBitSet bits = new FixedBitSet(in.maxDoc()); | ||
| if (start < end) { | ||
| // An output can legitimately own no document in this reader -- a key | ||
| // missing here makes two cuts land on the same offset -- and | ||
| // FixedBitSet#set rejects an empty range starting at maxDoc. | ||
| bits.set(start, end); | ||
| } | ||
| Bits existing = in.getLiveDocs(); | ||
| if (existing != null) { | ||
| existing.applyMask(bits, 0); | ||
| } |
There was a problem hiding this comment.
Nit: we could be more memory efficient by adding a RangeBitSet that stores the bounds and a reference to the live docs Bits.
There was a problem hiding this comment.
Done, it stores the bounds and a reference to the reader's live docs now.
| doc = target; | ||
| return target >= start && target < end && values.advanceExact(target); |
There was a problem hiding this comment.
Looks like we can get corrupted state if the advanceExact(target) method is called with a target >= end followed up by a call to nextDoc(). I think we want a defensive check to doc >= end in the nextDoc() method to return DocValuesIterator.NO_MORE_DOCS. Doesn't look like it's used this way, but I don't see anything in the javadocs indicating it is invalid to call advanceExact then nextDoc
There was a problem hiding this comment.
Good catch, it is a real bug and not just undefined. Added the check with a test: without it a reader restricted to [30, 60) returns doc 0 from nextDoc, so a document owned by another output gets merged twice.
| public boolean advanceExact(int target) throws IOException { | ||
| doc = target; | ||
| return target >= start && target < end && values.advanceExact(target); | ||
| } |
There was a problem hiding this comment.
Same observation about advanceExact then nextDoc.
| private final int start; | ||
| private final int end; | ||
|
|
||
| DocRangeCodecReader(CodecReader in, int start, int end) { |
There was a problem hiding this comment.
IIUC, in the special case where start == end, we don't have any docs and can skip the iterating the terms tree. A possible optimization could ensure skip that iteration when the two bounds are equal.
There was a problem hiding this comment.
Done, and points too since a block k-d tree is ordered by value and cannot skip by doc either. Both return null when the range is empty, which is what a reader without the format looks like and the merge already skips those. The rest were already free, the narrowing cursors return NO_MORE_DOCS straight away.
The live documents of an output were materialised as a bit per document of the input, once per output, and held for the whole merge. The ranges of one input partition its documents, so the same answer comes from the two bounds and a reference to the reader's own deletions. Counting the range instead of a cardinality is the same total work, since the ranges partition the input, and allocates nothing. advanceExact() answers a request beyond the range without moving the values it wraps, so a following nextDoc() stepped an iterator still positioned behind the range and handed back a document that had already been merged. Stop instead when the cursor is past the end. An output owning no document in an input returned its postings and points readers anyway, and both are ordered by something other than the document, so the merge read a whole terms dictionary and a whole block k-d tree to discard all of it. Return null, which is what a reader without those formats looks like and which the merge already skips.
The vector cursor kept stepping the iterator underneath after reaching the end of its range, into the documents another output owns, where the doc values and norms cursors now stop. Give it the same guard. The live documents of an output are a view rather than the bit set they replaced, so check they answer identically, for get and for applyMask which the bulk merge paths use, over random ranges, deletions, windows and offsets.
4 participants
A merge takes N segments and writes one. This lets it take N and write M, by giving each output a contiguous document range of every input:
One entry per input: M + 1 non-decreasing boundaries, where output
otakes[b[o], b[o+1]). Called after the merge readers are open, so boundaries can be placed on values read from the readers.nullis the default and keeps today's behaviour. M may be larger or smaller than N, and N may be 1.Each input is wrapped in a reader presenting everything outside an output's range as deleted, so every format merges through its existing code and none of them learns about partitioning. An ordinary merge is the one-output case of the same path. Bytes written are unchanged: each document is written once, into the output that owns it. An index sort is not required; without one, documents are shared out by position. With one, boundaries placed on a key give each output a range of that key, so a merge can keep a tenant's documents in their own segments within a shared index.
Example
BalancedSegmentsMergePolicyinmisc.forceMerge(n)says how many segments to leave but not how the documents are shared between them: a hundred equal segments merged into four leaves 97,000 / 1,000 / 1,000 / 1,000. The policy plans the whole forced merge at once, packing segments into groups and giving each group the outputs its size deserves, so the largest merge is about N/M.forceMerge(16)on a one-segment index splits it into sixteen.Cost
Doc values, norms and vectors seek to an output's range (second commit), so the outputs together read those columns once. Stored fields and term vectors random-access per live document, so an output touches only the chunks its range falls in. Terms dictionary and BKD cannot seek by document and are read once per output. Merge inputs are verified once per merge rather than once per output.
Gaps
OneMerge.getOutputDiagnostics(int)covers this; follow-up.Testing
Each output is duelled against a segment built from exactly its documents in the same order, across all field types, sorted and unsorted indexes, blocks aligned and misaligned to boundaries, concurrent deletes and doc-values updates mid-merge, soft deletes, rollback, an IO failure while writing one output, malformed partitions, and wrapper preservation. Codecs are randomized.