ChrisHegarty commented on code in PR #16570: URL: https://github.com/apache/lucene/pull/16570#discussion_r3880976868
########## lucene/core/src/test/org/apache/lucene/index/TestMergeCarryOverFromDisk.java: ########## @@ -0,0 +1,290 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.lucene.index; + +import java.io.IOException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.lucene.document.BinaryDocValuesField; +import org.apache.lucene.document.Document; +import org.apache.lucene.document.NumericDocValuesField; +import org.apache.lucene.document.StringField; +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.FilterDirectory; +import org.apache.lucene.store.IOContext; +import org.apache.lucene.store.IndexOutput; +import org.apache.lucene.tests.analysis.MockAnalyzer; +import org.apache.lucene.tests.util.LuceneTestCase; +import org.apache.lucene.util.Bits; +import org.apache.lucene.util.BytesRef; + +/** + * Deterministically exercises the doc-values-update merge carry-over path: an update that resolves + * onto a segment <em>while it is being merged</em> must appear on the merged segment. That + * carry-over is reconstructed from the source segments at merge commit; the test blocks a merge + * mid-flight (only its own output, via {@link IOContext#mergeInfo}, so the update's flush is free), + * applies + resolves updates, then releases the merge. + */ +public class TestMergeCarryOverFromDisk extends LuceneTestCase { + + /** Wraps a directory, pausing the first merge output write until the test releases it. */ + private static class MergePausingDirectory extends FilterDirectory { + final CountDownLatch mergeStarted = new CountDownLatch(1); + final CountDownLatch resumeMerge = new CountDownLatch(1); + final AtomicReference<Throwable> failure = new AtomicReference<>(); + + MergePausingDirectory(Directory in) { + super(in); + } + + @Override + public IndexOutput createOutput(String name, IOContext context) throws IOException { + if (context != null && context.mergeInfo() != null && mergeStarted.getCount() > 0) { + mergeStarted.countDown(); + try { + resumeMerge.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + } + return super.createOutput(name, context); + } + } + + private static Document doc(int id, long ndv, String bdv) { + Document d = new Document(); + d.add(new StringField("id", Integer.toString(id), StringField.Store.NO)); + d.add(new NumericDocValuesField("ndv", ndv)); + d.add(new BinaryDocValuesField("bdv", new BytesRef(bdv))); + return d; + } + + public void testUpdatesResolvedDuringMergeAreCarriedOver() throws Exception { + MergePausingDirectory dir = new MergePausingDirectory(newDirectory()); + IndexWriterConfig conf = + newIndexWriterConfig(new MockAnalyzer(random())) + .setMergeScheduler(new ConcurrentMergeScheduler()); + IndexWriter writer = new IndexWriter(dir, conf); + + // Two segments; the default TieredMergePolicy won't auto-merge two small segments. + for (int i = 0; i < 4; i++) { + writer.addDocument(doc(i, i, "v" + i)); + } + writer.commit(); + for (int i = 4; i < 8; i++) { + writer.addDocument(doc(i, i, "v" + i)); + } + writer.commit(); + + // Force the merge on a background thread so we can inject updates while it is paused + // mid-flight. + Thread merger = + new Thread( + () -> { + try { + writer.forceMerge(1); + } catch (Throwable t) { + dir.failure.compareAndSet(null, t); + } + }, + "forceMerge"); + merger.start(); + + // Wait until the merge has opened its readers (its on-disk baseline is fixed) and begun writing + // output. + dir.mergeStarted.await(); + + // Update numeric + binary values on docs in the now-merging segments, and reset one, then Review Comment: The comment says "and reset one" but the test only updates values, nothing is reset. Could you either add a reset here or drop that phrase from the comment? ########## lucene/core/src/java/org/apache/lucene/index/ReadersAndUpdates.java: ########## @@ -193,6 +171,18 @@ public synchronized long getNumDVUpdates() { return count; } + /** + * Copies of the currently-pending (resolved but not yet written) doc-values updates, keyed by + * field, so callers can iterate without holding the lock. Review Comment: Small javadoc nit: "without holding the lock" reads like callers can use this unsynchronized. What we really mean is callers can do the long disk I/O in buildMappedDVUpdatesFromDisk without holding the RaU lock. They still need IndexWriter’s coordination for correctness. Maybe reword to something like "so callers can iterate pending updates without holding this monitor during I/O" ? ########## lucene/core/src/java/org/apache/lucene/index/IndexWriter.java: ########## @@ -4487,96 +4479,235 @@ private synchronized ReadersAndUpdates commitMergedDeletesAndUpdates( merge.getMergeReader().get(i).hardLiveDocs, rld.getHardLiveDocs(), segDocMap); + } - // Now carry over all doc values updates that were resolved while we were merging, remapping - // the docIDs to the newly merged docIDs. - // We only carry over packets that finished resolving; if any are still running (concurrently) - // they will detect that our merge completed - // and re-resolve against the newly merged segment: - Map<String, List<DocValuesFieldUpdates>> mergingDVUpdates = rld.getMergingDVUpdates(); - for (Map.Entry<String, List<DocValuesFieldUpdates>> ent : mergingDVUpdates.entrySet()) { - - String field = ent.getKey(); + // Carry over the doc-values updates that resolved while merging, remapping to the merged + // docIDs. + // Updates still resolving concurrently are skipped; they re-resolve against the merged segment + // once they complete. + Map<String, LongObjectHashMap<DocValuesFieldUpdates>> mappedDVUpdates = + buildMappedDVUpdatesFromDisk(merge, docMaps, minGen); - LongObjectHashMap<DocValuesFieldUpdates> mappedField = mappedDVUpdates.get(field); - if (mappedField == null) { - mappedField = new LongObjectHashMap<>(); - mappedDVUpdates.put(field, mappedField); + boolean anyDVUpdates = false; + if (mappedDVUpdates.isEmpty() == false) { + // Persist the merged DV updates onto the RAU for the merged segment (already finished): + for (LongObjectHashMap<DocValuesFieldUpdates> d : mappedDVUpdates.values()) { + for (ObjectCursor<DocValuesFieldUpdates> updates : d.values()) { + mergedDeletesAndUpdates.addDVUpdate(updates.value); + anyDVUpdates = true; } + } + } + + if (infoStream.isEnabled("IW")) { + String msg = mergedDeletesAndUpdates.getDelCount() - numDeletesBefore + " new deletes"; + if (anyDVUpdates) { + msg += " and " + mergedDeletesAndUpdates.getNumDVUpdates() + " new field updates"; + msg += " (" + mergedDeletesAndUpdates.ramBytesUsed.get() + ") bytes"; + } + msg += " since merge started"; + infoStream.message("IW", msg); + } - for (DocValuesFieldUpdates updates : ent.getValue()) { + merge.info.setBufferedDeletesGen(minGen); - if (bufferedUpdatesStream.stillRunning(updates.delGen)) { - continue; - } + return mergedDeletesAndUpdates; + } - // sanity check: - assert field.equals(updates.field); + /** Sentinel for "no doc-values value" (a reset, or a doc that simply has no value). */ + private static final Object DV_NO_VALUE = new Object(); - DocValuesFieldUpdates mappedUpdates = mappedField.get(updates.delGen); - if (mappedUpdates == null) { - switch (updates.type) { - case NUMERIC: - mappedUpdates = - new NumericDocValuesFieldUpdates( - updates.delGen, updates.field, merge.info.info.maxDoc()); - break; - case BINARY: - mappedUpdates = - new BinaryDocValuesFieldUpdates( - updates.delGen, updates.field, merge.info.info.maxDoc()); - break; - case NONE: - case SORTED: - case SORTED_SET: - case SORTED_NUMERIC: - default: - throw new AssertionError(); + /** + * Builds the doc-values-update merge carry-over from the source segments' on-disk state plus + * their residual (resolved but not-yet-written) pending updates, remapping to merged docIDs. + * Returns {@code field -> delGen -> updates} for the merged {@link ReadersAndUpdates} to apply. + * + * <p>Per source segment and updated field, the written changes are read back from disk + * (current-vs-baseline diff) and collapsed into one packet, while the residual updates keep their + * real delGens. Flushing advances {@code completedDelGen} as a prefix, so every residual delGen + * is greater than any written one; the collapsed packet therefore uses {@code minResidualDelGen - + * 1} (or {@code completedDelGen} when there is no residual), a gen below all residual and + * still-running ones so newest-wins ordering holds on the merged segment. + */ + private Map<String, LongObjectHashMap<DocValuesFieldUpdates>> buildMappedDVUpdatesFromDisk( + MergePolicy.OneMerge merge, MergeState.DocMap[] docMaps, long minGen) throws IOException { + final int mergedMaxDoc = merge.info.info.maxDoc(); + Map<String, LongObjectHashMap<DocValuesFieldUpdates>> mapped = new HashMap<>(); + for (int i = 0; i < merge.segments.size(); i++) { + SegmentCommitInfo info = merge.segments.get(i); + final ReadersAndUpdates rld = getPooledInstance(info, false); + final MergeState.DocMap segDocMap = docMaps[i]; + final CodecReader baseline = merge.getMergeReader().get(i).codecReader; + final SegmentReader current = rld.getReader(IOContext.DEFAULT); + try { + // Residual finished-but-unflushed updates for this segment, per field, skipping any packet + // still globally running (those re-resolve against the merged segment). + Map<String, List<DocValuesFieldUpdates>> residualByField = new HashMap<>(); + for (Map.Entry<String, List<DocValuesFieldUpdates>> e : + rld.getPendingDVUpdatesSnapshot().entrySet()) { + List<DocValuesFieldUpdates> keep = new ArrayList<>(); + for (DocValuesFieldUpdates u : e.getValue()) { + if (bufferedUpdatesStream.stillRunning(u.delGen) == false) { + keep.add(u); } - mappedField.put(updates.delGen, mappedUpdates); } + if (keep.isEmpty() == false) { + residualByField.put(e.getKey(), keep); + } + } + // Updated fields = those whose on-disk DV generation advanced vs the merge-reader baseline + // (gen advances for both classic and overlay writes), plus any field with residual updates. + Set<String> updatedFields = new HashSet<>(residualByField.keySet()); + FieldInfos baseFieldInfos = baseline.getFieldInfos(); + for (FieldInfo fi : current.getFieldInfos()) { + DocValuesType t = fi.getDocValuesType(); + if (t != DocValuesType.NUMERIC && t != DocValuesType.BINARY) { + continue; + } + FieldInfo baseFi = baseFieldInfos.fieldInfo(fi.name); + if (baseFi == null || baseFi.getDocValuesGen() != fi.getDocValuesGen()) { + updatedFields.add(fi.name); + } + } - DocValuesFieldUpdates.Iterator it = updates.iterator(); - int doc; - while ((doc = it.nextDoc()) != NO_MORE_DOCS) { - int mappedDoc = segDocMap.get(doc); - if (mappedDoc != -1) { + for (String field : updatedFields) { + List<DocValuesFieldUpdates> residual = residualByField.getOrDefault(field, List.of()); + long minResidual = Long.MAX_VALUE; + for (DocValuesFieldUpdates u : residual) { + minResidual = Math.min(minResidual, u.delGen); + } + final long diskDelGen = + minResidual != Long.MAX_VALUE + ? minResidual - 1 + : bufferedUpdatesStream.getCompletedDelGen(); + LongObjectHashMap<DocValuesFieldUpdates> byGen = + mapped.computeIfAbsent(field, _ -> new LongObjectHashMap<>()); + + // (A) flushed changes since baseline, collapsed at diskDelGen. Only when the field has + // on-disk doc values; a field whose only updates are still pending is absent from the + // reader (e.g. a soft-deletes field on its first, not-yet-written update). + FieldInfo fi = current.getFieldInfos().fieldInfo(field); + if (fi != null) { + addDiskDiffToPacket(fi, baseline, current, segDocMap, diskDelGen, mergedMaxDoc, byGen); + assert byGen.containsKey(diskDelGen) == false || diskDelGen > minGen + : "diskDelGen " + diskDelGen + " <= minGen " + minGen; + } + + // (B) residual updates at their real delGens, typed from the packet itself since the + // field + // need not exist on disk yet. + for (DocValuesFieldUpdates u : residual) { + DocValuesFieldUpdates.Iterator it = u.iterator(); + int doc; + while ((doc = it.nextDoc()) != NO_MORE_DOCS) { + int md = segDocMap.get(doc); + if (md == -1) { + continue; + } + DocValuesFieldUpdates p = byGen.get(u.delGen); + if (p == null) { + p = + u.type == DocValuesType.BINARY + ? new BinaryDocValuesFieldUpdates(u.delGen, u.field, mergedMaxDoc) + : new NumericDocValuesFieldUpdates(u.delGen, u.field, mergedMaxDoc); + byGen.put(u.delGen, p); + } if (it.hasValue()) { - // not deleted - mappedUpdates.add(mappedDoc, it); + p.add(md, it); } else { - mappedUpdates.reset(mappedDoc); + p.reset(md); } - anyDVUpdates = true; } } } + } finally { + current.decRef(); } } - - if (anyDVUpdates) { - // Persist the merged DV updates onto the RAU for the merged segment: - for (LongObjectHashMap<DocValuesFieldUpdates> d : mappedDVUpdates.values()) { - for (ObjectCursor<DocValuesFieldUpdates> updates : d.values()) { - updates.value.finish(); - mergedDeletesAndUpdates.addDVUpdate(updates.value); - } + for (LongObjectHashMap<DocValuesFieldUpdates> byGen : mapped.values()) { + for (ObjectCursor<DocValuesFieldUpdates> c : byGen.values()) { + c.value.finish(); } } + return mapped; + } - if (infoStream.isEnabled("IW")) { - String msg = mergedDeletesAndUpdates.getDelCount() - numDeletesBefore + " new deletes"; - if (anyDVUpdates) { - msg += " and " + mergedDeletesAndUpdates.getNumDVUpdates() + " new field updates"; - msg += " (" + mergedDeletesAndUpdates.ramBytesUsed.get() + ") bytes"; - } - msg += " since merge started"; - infoStream.message("IW", msg); + private static DocValuesFieldUpdates getOrCreatePacket( + LongObjectHashMap<DocValuesFieldUpdates> byGen, long delGen, FieldInfo fi, int mergedMaxDoc) { + DocValuesFieldUpdates p = byGen.get(delGen); + if (p == null) { + p = + fi.getDocValuesType() == DocValuesType.BINARY + ? new BinaryDocValuesFieldUpdates(delGen, fi.name, mergedMaxDoc) + : new NumericDocValuesFieldUpdates(delGen, fi.name, mergedMaxDoc); + byGen.put(delGen, p); } + return p; + } - merge.info.setBufferedDeletesGen(minGen); + /** + * Emits, into a single {@code diskDelGen} packet, each merged doc of {@code fi} whose current + * on-disk value differs from the merge-reader baseline (i.e. an update flushed during the merge), + * remapping source docIDs to merged docIDs. + */ + private static void addDiskDiffToPacket( + FieldInfo fi, + CodecReader baseline, + CodecReader current, + MergeState.DocMap segDocMap, + long diskDelGen, + int mergedMaxDoc, + LongObjectHashMap<DocValuesFieldUpdates> byGen) + throws IOException { + final int maxDoc = current.maxDoc(); + final boolean binary = fi.getDocValuesType() == DocValuesType.BINARY; + BinaryDocValues curB = binary ? current.getBinaryDocValues(fi.name) : null; + BinaryDocValues baseB = binary ? baseline.getBinaryDocValues(fi.name) : null; + NumericDocValues curN = binary ? null : current.getNumericDocValues(fi.name); + NumericDocValues baseN = binary ? null : baseline.getNumericDocValues(fi.name); + for (int doc = 0; doc < maxDoc; doc++) { + int md = segDocMap.get(doc); + if (md == -1) { + continue; + } + Object curVal; + Object baseVal; + if (binary) { + curVal = + (curB != null && curB.advanceExact(doc)) + ? BytesRef.deepCopyOf(curB.binaryValue()) Review Comment: Could we optimize this later by comparing without copying when bytes are equal? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
