vigyasharma commented on code in PR #15378:
URL: https://github.com/apache/lucene/pull/15378#discussion_r2480245898
##########
lucene/core/src/java/org/apache/lucene/index/MergePolicy.java:
##########
@@ -939,4 +939,96 @@ static final class MergeReader {
this.hardLiveDocs = hardLiveDocs;
}
}
+
+ /**
+ * Observer for merge operations returned by {@link
IndexWriter#forceMergeDeletes(boolean)}.
+ * Provides methods to query merge status and wait for completion.
+ *
+ * <p>When no merges are needed, {@link #hasNewMerges()} returns {@code
false} and {@link
+ * #numMerges()} returns 0. In this case, {@link #await()} returns {@code
true} immediately since
+ * there is nothing to wait for.
+ *
+ * @lucene.experimental
+ */
+ public static final class MergeObserver {
+ private final MergePolicy.MergeSpecification spec;
+
+ MergeObserver(MergePolicy.MergeSpecification spec) {
+ this.spec = spec;
+ }
+
+ /**
+ * Returns the number of merges in this specification.
+ *
+ * @return number of merges, or 0 if no merges were scheduled
+ */
+ public int numMerges() {
+ return spec == null ? 0 : spec.merges.size();
+ }
+
+ /**
+ * Returns whether any new merges were scheduled.
+ *
+ * @return {@code true} if merges were scheduled, {@code false} if no
merges needed
+ */
+ public boolean hasNewMerges() {
+ return spec != null;
+ }
+
+ /**
+ * Waits for all merges in this specification to complete. Returns
immediately if no merges were
+ * scheduled.
+ *
+ * @return {@code true} if all merges completed successfully or no merges
were needed, {@code
+ * false} on error
+ */
+ public boolean await() {
+ return spec == null || spec.await();
+ }
+
+ /**
+ * Waits for all merges in this specification to complete, with timeout.
Returns immediately if
+ * no merges were scheduled.
+ *
+ * @param timeout maximum time to wait
+ * @param unit time unit for timeout
+ * @return {@code true} if all merges completed within timeout or no
merges were needed, {@code
+ * false} on timeout or error
+ */
+ public boolean await(long timeout, TimeUnit unit) {
+ return spec == null || spec.await(timeout, unit);
+ }
+
+ /**
+ * Returns a {@link CompletableFuture} that completes when all merges
finish. Returns an
+ * already-completed future if no merges were scheduled.
+ *
+ * @return future that completes when merges finish
+ */
+ public CompletableFuture<Void> awaitAsync() {
+ return spec == null
+ ? CompletableFuture.completedFuture(null)
+ : spec.getMergeCompletedFutures();
+ }
+
+ @Override
+ public String toString() {
+ return spec == null ? "MergeObserver: no merges" : spec.toString();
+ }
+
+ /**
+ * Returns the merge at the specified index. Caller must ensure {@link
#hasNewMerges()} returns
+ * {@code true} and index is within bounds.
+ *
+ * @param i merge index (0 to {@link #numMerges()} - 1)
+ * @return the merge at index i
+ * @throws IndexOutOfBoundsException if index is invalid or no merges exist
+ */
+ public MergePolicy.OneMerge getMerge(int i) {
Review Comment:
I'm not sure if providing this method is really useful. The caller can't
know the "index" location of a OneMerge unless they already have access to the
MergeSpec.
##########
lucene/core/src/java/org/apache/lucene/index/MergePolicy.java:
##########
@@ -939,4 +939,96 @@ static final class MergeReader {
this.hardLiveDocs = hardLiveDocs;
}
}
+
+ /**
+ * Observer for merge operations returned by {@link
IndexWriter#forceMergeDeletes(boolean)}.
+ * Provides methods to query merge status and wait for completion.
+ *
+ * <p>When no merges are needed, {@link #hasNewMerges()} returns {@code
false} and {@link
+ * #numMerges()} returns 0. In this case, {@link #await()} returns {@code
true} immediately since
+ * there is nothing to wait for.
+ *
+ * @lucene.experimental
+ */
+ public static final class MergeObserver {
+ private final MergePolicy.MergeSpecification spec;
+
+ MergeObserver(MergePolicy.MergeSpecification spec) {
+ this.spec = spec;
+ }
+
+ /**
+ * Returns the number of merges in this specification.
+ *
+ * @return number of merges, or 0 if no merges were scheduled
+ */
+ public int numMerges() {
Review Comment:
How about we also expose the no. of completed merges? You will need to run
through all the OneMerge objects in `merges` and check for `mergeCompleted`. It
could be useful to track overall merge progress.
##########
lucene/core/src/test/org/apache/lucene/index/TestIndexWriterMerging.java:
##########
@@ -333,6 +337,330 @@ public synchronized void merge(MergeSource mergeSource,
MergeTrigger trigger)
public void close() {}
}
+ public void testForceMergeDeletesWithObserver() throws IOException {
+ Directory dir = newDirectory();
+ IndexWriter indexer =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new MockAnalyzer(random()))
+ .setMaxBufferedDocs(2)
+ .setRAMBufferSizeMB(IndexWriterConfig.DISABLE_AUTO_FLUSH));
+
+ for (int i = 0; i < 10; i++) {
+ Document doc = new Document();
+ Field idField = newStringField("id", "" + i, Field.Store.NO);
+ doc.add(idField);
+ indexer.addDocument(doc);
+ }
+ indexer.close();
+
+ IndexReader beforeDeleteReader = DirectoryReader.open(dir);
+ assertEquals(10, beforeDeleteReader.maxDoc());
+ assertEquals(10, beforeDeleteReader.numDocs());
+ beforeDeleteReader.close();
+
+ IndexWriter deleter =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new MockAnalyzer(random()))
+ .setMergePolicy(NoMergePolicy.INSTANCE));
+ for (int i = 0; i < 10; i++) {
+ if (i % 2 == 0) {
+ deleter.deleteDocuments(new Term("id", "" + i));
+ }
+ }
+ deleter.close();
+
+ IndexReader afterDeleteReader = DirectoryReader.open(dir);
+ assertEquals(10, afterDeleteReader.maxDoc());
+ assertEquals(5, afterDeleteReader.numDocs());
+ afterDeleteReader.close();
+
+ IndexWriter iw =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new
MockAnalyzer(random())).setMergePolicy(newLogMergePolicy()));
+ assertEquals(10, iw.getDocStats().maxDoc);
+ assertEquals(5, iw.getDocStats().numDocs);
+ MergePolicy.MergeObserver observer = iw.forceMergeDeletes(false);
+
+ assertTrue(observer.hasNewMerges());
+ assertTrue(observer.numMerges() > 0);
+
+ // Measure time to detect stuck merges
+ long startNanos = System.nanoTime();
+ observer.await();
+ long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNanos);
+
+ assertTrue(
+ "Merge took too long: " + elapsedMillis + "ms (expected < 30000ms)",
+ elapsedMillis < 30_000);
+
+ assertEquals(5, iw.getDocStats().maxDoc);
+ assertEquals(5, iw.getDocStats().numDocs);
+
+ iw.waitForMerges();
+ iw.close();
+ dir.close();
+ }
+
+ public void testMergeObserverGetMerges() throws IOException {
Review Comment:
We can remove this test if we remove the getMerges api.
##########
lucene/core/src/java/org/apache/lucene/index/MergePolicy.java:
##########
@@ -939,4 +939,96 @@ static final class MergeReader {
this.hardLiveDocs = hardLiveDocs;
}
}
+
+ /**
+ * Observer for merge operations returned by {@link
IndexWriter#forceMergeDeletes(boolean)}.
+ * Provides methods to query merge status and wait for completion.
+ *
+ * <p>When no merges are needed, {@link #hasNewMerges()} returns {@code
false} and {@link
+ * #numMerges()} returns 0. In this case, {@link #await()} returns {@code
true} immediately since
+ * there is nothing to wait for.
+ *
+ * @lucene.experimental
+ */
+ public static final class MergeObserver {
+ private final MergePolicy.MergeSpecification spec;
+
+ MergeObserver(MergePolicy.MergeSpecification spec) {
+ this.spec = spec;
+ }
+
+ /**
+ * Returns the number of merges in this specification.
+ *
+ * @return number of merges, or 0 if no merges were scheduled
+ */
+ public int numMerges() {
+ return spec == null ? 0 : spec.merges.size();
+ }
+
+ /**
+ * Returns whether any new merges were scheduled.
+ *
+ * @return {@code true} if merges were scheduled, {@code false} if no
merges needed
+ */
+ public boolean hasNewMerges() {
Review Comment:
I think this naming can get confusing: is it still a "new" merge once the
merge is underway, does it count completed ones, etc.? Instead of this API, can
we just use `numMerges() > 0` ? Or how about an `isEmpty` api?
##########
lucene/core/src/test/org/apache/lucene/index/TestIndexWriterMerging.java:
##########
@@ -333,6 +337,330 @@ public synchronized void merge(MergeSource mergeSource,
MergeTrigger trigger)
public void close() {}
}
+ public void testForceMergeDeletesWithObserver() throws IOException {
+ Directory dir = newDirectory();
+ IndexWriter indexer =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new MockAnalyzer(random()))
+ .setMaxBufferedDocs(2)
+ .setRAMBufferSizeMB(IndexWriterConfig.DISABLE_AUTO_FLUSH));
+
+ for (int i = 0; i < 10; i++) {
+ Document doc = new Document();
+ Field idField = newStringField("id", "" + i, Field.Store.NO);
+ doc.add(idField);
+ indexer.addDocument(doc);
+ }
+ indexer.close();
+
+ IndexReader beforeDeleteReader = DirectoryReader.open(dir);
+ assertEquals(10, beforeDeleteReader.maxDoc());
+ assertEquals(10, beforeDeleteReader.numDocs());
+ beforeDeleteReader.close();
+
+ IndexWriter deleter =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new MockAnalyzer(random()))
+ .setMergePolicy(NoMergePolicy.INSTANCE));
+ for (int i = 0; i < 10; i++) {
+ if (i % 2 == 0) {
+ deleter.deleteDocuments(new Term("id", "" + i));
+ }
+ }
+ deleter.close();
+
+ IndexReader afterDeleteReader = DirectoryReader.open(dir);
+ assertEquals(10, afterDeleteReader.maxDoc());
+ assertEquals(5, afterDeleteReader.numDocs());
+ afterDeleteReader.close();
+
+ IndexWriter iw =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new
MockAnalyzer(random())).setMergePolicy(newLogMergePolicy()));
+ assertEquals(10, iw.getDocStats().maxDoc);
+ assertEquals(5, iw.getDocStats().numDocs);
+ MergePolicy.MergeObserver observer = iw.forceMergeDeletes(false);
+
+ assertTrue(observer.hasNewMerges());
+ assertTrue(observer.numMerges() > 0);
+
+ // Measure time to detect stuck merges
+ long startNanos = System.nanoTime();
+ observer.await();
+ long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNanos);
+
+ assertTrue(
+ "Merge took too long: " + elapsedMillis + "ms (expected < 30000ms)",
+ elapsedMillis < 30_000);
+
+ assertEquals(5, iw.getDocStats().maxDoc);
+ assertEquals(5, iw.getDocStats().numDocs);
+
+ iw.waitForMerges();
+ iw.close();
+ dir.close();
+ }
+
+ public void testMergeObserverGetMerges() throws IOException {
+ Directory dir = newDirectory();
+ IndexWriter indexer =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new MockAnalyzer(random()))
+ .setMergePolicy(NoMergePolicy.INSTANCE));
+
+ for (int i = 0; i < 10; i++) {
+ Document doc = new Document();
+ Field idField = newStringField("id", "" + i, Field.Store.NO);
+ doc.add(idField);
+ indexer.addDocument(doc);
+ }
+ indexer.close();
+
+ IndexWriter deleter =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new
MockAnalyzer(random())).setMergePolicy(newLogMergePolicy()));
+ for (int i = 0; i < 2; i++) {
+ deleter.deleteDocuments(new Term("id", "" + i));
+ }
+
+ MergePolicy.MergeObserver observer = deleter.forceMergeDeletes(false);
+ assertTrue(observer.hasNewMerges());
+ assertTrue(observer.numMerges() > 0);
+
+ int numMerges = observer.numMerges();
+ for (int j = 0; j < numMerges; j++) {
+ MergePolicy.OneMerge oneMerge = observer.getMerge(j);
+ assertNotNull(oneMerge);
+ }
+
+ try {
+ observer.getMerge(-1);
+ fail("Should throw IndexOutOfBoundsException");
+ } catch (IndexOutOfBoundsException expected) {
+ String message = expected.getMessage();
+ assertTrue(
+ "Message should mention 'out of bounds', got: " + message,
+ message.contains("Index -1 out of bounds for length 1"));
+ }
+
+ try {
+ observer.getMerge(numMerges);
+ fail("Should throw IndexOutOfBoundsException");
+ } catch (IndexOutOfBoundsException expected) {
+ String message = expected.getMessage();
+ assertTrue(
+ "Message should mention 'out of bounds', got: " + message,
+ message.contains("Index " + numMerges + " out of bounds for length
1"));
+ }
+
+ deleter.waitForMerges();
+ deleter.close();
+ dir.close();
+ }
+
+ public void testMergeObserverNoMerges() throws IOException {
+ Directory dir = newDirectory();
+ IndexWriter writer =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new MockAnalyzer(random()))
+ .setMergePolicy(NoMergePolicy.INSTANCE));
+
+ Document doc = new Document();
+ doc.add(newStringField("id", "1", Field.Store.NO));
+ writer.addDocument(doc);
+ writer.commit();
+
+ MergePolicy.MergeObserver observer = writer.forceMergeDeletes(false);
+
+ assertFalse("Should have no merges when no deletions",
observer.hasNewMerges());
+ assertEquals("Should have zero merges", 0, observer.numMerges());
+
+ try {
+ observer.getMerge(0);
+ fail("Should throw IndexOutOfBoundsException when no merges");
+ } catch (IndexOutOfBoundsException expected) {
+ String message = expected.getMessage();
+ assertTrue(
+ "Message should mention 'no merges', got: " + message,
+ message.contains("No merges available"));
+ }
+
+ writer.close();
+ dir.close();
+ }
+
+ public void testMergeObserverAwaitWithTimeout() throws Exception {
+ Directory dir = newDirectory();
+ IndexWriter iw =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new
MockAnalyzer(random())).setMergePolicy(newLogMergePolicy()));
+
+ for (int i = 0; i < 10; i++) {
+ Document doc = new Document();
+ doc.add(newStringField("id", "" + i, Field.Store.NO));
+ iw.addDocument(doc);
+ }
+ iw.commit();
+
+ iw.deleteDocuments(new Term("id", "0"));
+ iw.deleteDocuments(new Term("id", "1"));
+ iw.deleteDocuments(new Term("id", "2"));
+ iw.commit();
+
+ MergePolicy.MergeObserver observer = iw.forceMergeDeletes(false);
+
+ // Measure time to detect stuck merges
+ long startNanos = System.nanoTime();
+ assertTrue("await should complete before timeout", observer.await(10,
TimeUnit.MINUTES));
+ long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNanos);
+
+ assertTrue(
+ "Merge took too long: " + elapsedMillis + "ms (expected < 30000ms)",
+ elapsedMillis < 30_000);
+
+ iw.waitForMerges();
+ iw.close();
+ dir.close();
+ }
+
+ @SuppressForbidden(reason = "Thread sleep")
+ public void testMergeObserverAwaitTimeout() throws Exception {
+ Directory dir = newDirectory();
+
+ CountDownLatch mergeStarted = new CountDownLatch(1);
+ CountDownLatch allowMergeToFinish = new CountDownLatch(1);
+
+ ConcurrentMergeScheduler mergeScheduler =
+ new ConcurrentMergeScheduler() {
+ @Override
+ protected void doMerge(MergeSource mergeSource, MergePolicy.OneMerge
merge)
+ throws IOException {
+ try {
+ mergeStarted.countDown();
+ // Block until test allows completion
+ allowMergeToFinish.await();
+ super.doMerge(mergeSource, merge);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException(e);
+ }
+ }
+ };
+
+ IndexWriter indexer =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new MockAnalyzer(random()))
+ .setMergePolicy(newLogMergePolicy())
+ .setMergeScheduler(mergeScheduler));
+
+ for (int i = 0; i < 20; i++) {
+ Document doc = new Document();
+ doc.add(newStringField("id", "" + i, Field.Store.NO));
+ indexer.addDocument(doc);
+ }
+ indexer.commit();
+
+ for (int i = 0; i < 10; i++) {
+ indexer.deleteDocuments(new Term("id", "" + i));
+ }
+ indexer.commit();
+
+ MergePolicy.MergeObserver observer = indexer.forceMergeDeletes(false);
+
+ if (observer.hasNewMerges()) {
+ mergeStarted.await();
+ assertFalse("await should timeout", observer.await(10,
TimeUnit.MILLISECONDS));
+ allowMergeToFinish.countDown();
+ }
+
+ indexer.waitForMerges();
+ indexer.close();
+ dir.close();
+ }
+
+ public void testForceMergeDeletesBlockingWithObserver() throws IOException {
+ Directory dir = newDirectory();
+ IndexWriter indexer =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new MockAnalyzer(random()))
+ .setMaxBufferedDocs(2)
+ .setRAMBufferSizeMB(IndexWriterConfig.DISABLE_AUTO_FLUSH));
+
+ for (int i = 0; i < 10; i++) {
+ Document doc = new Document();
+ Field idField = newStringField("id", "" + i, Field.Store.NO);
+ doc.add(idField);
+ indexer.addDocument(doc);
+ }
+ indexer.close();
+
+ IndexWriter deleter =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new MockAnalyzer(random()))
+ .setMergePolicy(NoMergePolicy.INSTANCE));
+ for (int i = 0; i < 10; i++) {
+ if (i % 2 == 0) {
+ deleter.deleteDocuments(new Term("id", "" + i));
+ }
+ }
+ deleter.close();
+
+ IndexWriter iw =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new
MockAnalyzer(random())).setMergePolicy(newLogMergePolicy()));
+ assertEquals(10, iw.getDocStats().maxDoc);
+ assertEquals(5, iw.getDocStats().numDocs);
+
+ MergePolicy.MergeObserver observer = iw.forceMergeDeletes(true);
+ assertTrue("Should have scheduled merges", observer.hasNewMerges());
+ assertTrue("Should have completed merges", observer.numMerges() > 0);
Review Comment:
You can add an assert here that tests that numMerges == completedMerges
##########
lucene/core/src/test/org/apache/lucene/index/TestIndexWriterMerging.java:
##########
@@ -333,6 +337,330 @@ public synchronized void merge(MergeSource mergeSource,
MergeTrigger trigger)
public void close() {}
}
Review Comment:
Appreciate the thorough tests below, nice work!
##########
lucene/CHANGES.txt:
##########
@@ -69,6 +69,12 @@ Improvements
* GITHUB#15124: Use RamUsageEstimator to calculate size for non-accountable
queries. (Sagar Upadhyaya)
+* GITHUB#14515: IndexWriter.forceMergeDeletes() now returns
MergePolicy.MergeObserver,
Review Comment:
This change can go into 10.4. Let's move this entry to the "API Changes"
section under 10.4?
##########
lucene/core/src/test/org/apache/lucene/index/TestIndexWriterMerging.java:
##########
@@ -333,6 +337,330 @@ public synchronized void merge(MergeSource mergeSource,
MergeTrigger trigger)
public void close() {}
}
+ public void testForceMergeDeletesWithObserver() throws IOException {
+ Directory dir = newDirectory();
+ IndexWriter indexer =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new MockAnalyzer(random()))
+ .setMaxBufferedDocs(2)
+ .setRAMBufferSizeMB(IndexWriterConfig.DISABLE_AUTO_FLUSH));
+
+ for (int i = 0; i < 10; i++) {
+ Document doc = new Document();
+ Field idField = newStringField("id", "" + i, Field.Store.NO);
+ doc.add(idField);
+ indexer.addDocument(doc);
+ }
+ indexer.close();
+
+ IndexReader beforeDeleteReader = DirectoryReader.open(dir);
+ assertEquals(10, beforeDeleteReader.maxDoc());
+ assertEquals(10, beforeDeleteReader.numDocs());
+ beforeDeleteReader.close();
+
+ IndexWriter deleter =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new MockAnalyzer(random()))
+ .setMergePolicy(NoMergePolicy.INSTANCE));
+ for (int i = 0; i < 10; i++) {
+ if (i % 2 == 0) {
+ deleter.deleteDocuments(new Term("id", "" + i));
+ }
+ }
+ deleter.close();
+
+ IndexReader afterDeleteReader = DirectoryReader.open(dir);
+ assertEquals(10, afterDeleteReader.maxDoc());
+ assertEquals(5, afterDeleteReader.numDocs());
+ afterDeleteReader.close();
+
+ IndexWriter iw =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new
MockAnalyzer(random())).setMergePolicy(newLogMergePolicy()));
+ assertEquals(10, iw.getDocStats().maxDoc);
+ assertEquals(5, iw.getDocStats().numDocs);
+ MergePolicy.MergeObserver observer = iw.forceMergeDeletes(false);
+
+ assertTrue(observer.hasNewMerges());
+ assertTrue(observer.numMerges() > 0);
+
+ // Measure time to detect stuck merges
+ long startNanos = System.nanoTime();
+ observer.await();
+ long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNanos);
+
+ assertTrue(
+ "Merge took too long: " + elapsedMillis + "ms (expected < 30000ms)",
+ elapsedMillis < 30_000);
+
+ assertEquals(5, iw.getDocStats().maxDoc);
+ assertEquals(5, iw.getDocStats().numDocs);
+
+ iw.waitForMerges();
+ iw.close();
+ dir.close();
+ }
+
+ public void testMergeObserverGetMerges() throws IOException {
+ Directory dir = newDirectory();
+ IndexWriter indexer =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new MockAnalyzer(random()))
+ .setMergePolicy(NoMergePolicy.INSTANCE));
+
+ for (int i = 0; i < 10; i++) {
+ Document doc = new Document();
+ Field idField = newStringField("id", "" + i, Field.Store.NO);
+ doc.add(idField);
+ indexer.addDocument(doc);
+ }
+ indexer.close();
+
+ IndexWriter deleter =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new
MockAnalyzer(random())).setMergePolicy(newLogMergePolicy()));
+ for (int i = 0; i < 2; i++) {
+ deleter.deleteDocuments(new Term("id", "" + i));
+ }
+
+ MergePolicy.MergeObserver observer = deleter.forceMergeDeletes(false);
+ assertTrue(observer.hasNewMerges());
+ assertTrue(observer.numMerges() > 0);
+
+ int numMerges = observer.numMerges();
+ for (int j = 0; j < numMerges; j++) {
+ MergePolicy.OneMerge oneMerge = observer.getMerge(j);
+ assertNotNull(oneMerge);
+ }
+
+ try {
+ observer.getMerge(-1);
+ fail("Should throw IndexOutOfBoundsException");
+ } catch (IndexOutOfBoundsException expected) {
+ String message = expected.getMessage();
+ assertTrue(
+ "Message should mention 'out of bounds', got: " + message,
+ message.contains("Index -1 out of bounds for length 1"));
+ }
+
+ try {
+ observer.getMerge(numMerges);
+ fail("Should throw IndexOutOfBoundsException");
+ } catch (IndexOutOfBoundsException expected) {
+ String message = expected.getMessage();
+ assertTrue(
+ "Message should mention 'out of bounds', got: " + message,
+ message.contains("Index " + numMerges + " out of bounds for length
1"));
+ }
+
+ deleter.waitForMerges();
+ deleter.close();
+ dir.close();
+ }
+
+ public void testMergeObserverNoMerges() throws IOException {
+ Directory dir = newDirectory();
+ IndexWriter writer =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new MockAnalyzer(random()))
+ .setMergePolicy(NoMergePolicy.INSTANCE));
+
+ Document doc = new Document();
+ doc.add(newStringField("id", "1", Field.Store.NO));
+ writer.addDocument(doc);
+ writer.commit();
+
+ MergePolicy.MergeObserver observer = writer.forceMergeDeletes(false);
+
+ assertFalse("Should have no merges when no deletions",
observer.hasNewMerges());
+ assertEquals("Should have zero merges", 0, observer.numMerges());
+
+ try {
+ observer.getMerge(0);
+ fail("Should throw IndexOutOfBoundsException when no merges");
+ } catch (IndexOutOfBoundsException expected) {
+ String message = expected.getMessage();
+ assertTrue(
+ "Message should mention 'no merges', got: " + message,
+ message.contains("No merges available"));
+ }
+
+ writer.close();
+ dir.close();
+ }
+
+ public void testMergeObserverAwaitWithTimeout() throws Exception {
+ Directory dir = newDirectory();
+ IndexWriter iw =
+ new IndexWriter(
+ dir,
+ newIndexWriterConfig(new
MockAnalyzer(random())).setMergePolicy(newLogMergePolicy()));
+
+ for (int i = 0; i < 10; i++) {
+ Document doc = new Document();
+ doc.add(newStringField("id", "" + i, Field.Store.NO));
+ iw.addDocument(doc);
+ }
+ iw.commit();
+
+ iw.deleteDocuments(new Term("id", "0"));
+ iw.deleteDocuments(new Term("id", "1"));
+ iw.deleteDocuments(new Term("id", "2"));
+ iw.commit();
+
+ MergePolicy.MergeObserver observer = iw.forceMergeDeletes(false);
+
+ // Measure time to detect stuck merges
+ long startNanos = System.nanoTime();
+ assertTrue("await should complete before timeout", observer.await(10,
TimeUnit.MINUTES));
+ long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNanos);
+
+ assertTrue(
+ "Merge took too long: " + elapsedMillis + "ms (expected < 30000ms)",
+ elapsedMillis < 30_000);
Review Comment:
Is this 30_000 a random magic number for timeout upperbound? Why don't we
just use it in the await timeout value? Then `assertTrue` will fail if we don't
meet this limit.
##########
lucene/core/src/java/org/apache/lucene/index/MergePolicy.java:
##########
@@ -939,4 +939,96 @@ static final class MergeReader {
this.hardLiveDocs = hardLiveDocs;
}
}
+
+ /**
+ * Observer for merge operations returned by {@link
IndexWriter#forceMergeDeletes(boolean)}.
+ * Provides methods to query merge status and wait for completion.
+ *
+ * <p>When no merges are needed, {@link #hasNewMerges()} returns {@code
false} and {@link
+ * #numMerges()} returns 0. In this case, {@link #await()} returns {@code
true} immediately since
+ * there is nothing to wait for.
+ *
+ * @lucene.experimental
+ */
+ public static final class MergeObserver {
+ private final MergePolicy.MergeSpecification spec;
+
+ MergeObserver(MergePolicy.MergeSpecification spec) {
+ this.spec = spec;
+ }
+
+ /**
+ * Returns the number of merges in this specification.
+ *
+ * @return number of merges, or 0 if no merges were scheduled
+ */
+ public int numMerges() {
+ return spec == null ? 0 : spec.merges.size();
+ }
+
+ /**
+ * Returns whether any new merges were scheduled.
+ *
+ * @return {@code true} if merges were scheduled, {@code false} if no
merges needed
+ */
+ public boolean hasNewMerges() {
+ return spec != null;
+ }
+
+ /**
+ * Waits for all merges in this specification to complete. Returns
immediately if no merges were
+ * scheduled.
+ *
+ * @return {@code true} if all merges completed successfully or no merges
were needed, {@code
+ * false} on error
+ */
+ public boolean await() {
+ return spec == null || spec.await();
+ }
+
+ /**
+ * Waits for all merges in this specification to complete, with timeout.
Returns immediately if
+ * no merges were scheduled.
+ *
+ * @param timeout maximum time to wait
+ * @param unit time unit for timeout
+ * @return {@code true} if all merges completed within timeout or no
merges were needed, {@code
+ * false} on timeout or error
+ */
+ public boolean await(long timeout, TimeUnit unit) {
+ return spec == null || spec.await(timeout, unit);
+ }
+
+ /**
+ * Returns a {@link CompletableFuture} that completes when all merges
finish. Returns an
+ * already-completed future if no merges were scheduled.
+ *
+ * @return future that completes when merges finish
+ */
+ public CompletableFuture<Void> awaitAsync() {
+ return spec == null
+ ? CompletableFuture.completedFuture(null)
+ : spec.getMergeCompletedFutures();
+ }
+
+ @Override
+ public String toString() {
+ return spec == null ? "MergeObserver: no merges" : spec.toString();
+ }
+
+ /**
+ * Returns the merge at the specified index. Caller must ensure {@link
#hasNewMerges()} returns
+ * {@code true} and index is within bounds.
+ *
+ * @param i merge index (0 to {@link #numMerges()} - 1)
+ * @return the merge at index i
+ * @throws IndexOutOfBoundsException if index is invalid or no merges exist
+ */
+ public MergePolicy.OneMerge getMerge(int i) {
Review Comment:
Also, I don't think that the `OneMerge` object is immutable?
--
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]
