You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by rn...@apache.org on 2024/06/12 21:34:15 UTC
(couchdb) branch decaffeinated updated (a29153b8c -> 7d284d4bf)
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a change to branch decaffeinated
in repository https://gitbox.apache.org/repos/asf/couchdb.git
discard a29153b8c support concurrency across multiple segments
discard 2c9d29836 Use our own logic for index management
omit 1c6e83624 Use the minSeq check in more places and add a test
new 307bae07b Use our own logic for index management
new 7d284d4bf support concurrency across multiple segments
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (a29153b8c)
\
N -- N -- N refs/heads/decaffeinated (7d284d4bf)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../couchdb/nouveau/health/IndexHealthCheck.java | 1 -
.../couchdb/nouveau/lucene9/Lucene9IndexTest.java | 21 ---------------------
2 files changed, 22 deletions(-)
(couchdb) 01/02: Use our own logic for index management
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch decaffeinated
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 307bae07b5868048ee305764a5ffcf6ac0c3a355
Author: Robert Newson <Ro...@ibm.com>
AuthorDate: Wed Jun 12 14:54:26 2024 +0100
Use our own logic for index management
Caffeine complaines when eviction is slow (because we might commit the index).
Build our own logic for opening and closing indexes which allows opening and
closing to be very slow, while ensuring we don't close an index that's in use.
---
nouveau/build.gradle | 1 -
.../apache/couchdb/nouveau/NouveauApplication.java | 4 +-
.../org/apache/couchdb/nouveau/core/Index.java | 17 -
.../apache/couchdb/nouveau/core/IndexManager.java | 347 +++++++++++----------
.../nouveau/health/IndexHealthCheckTest.java | 6 +-
5 files changed, 189 insertions(+), 186 deletions(-)
diff --git a/nouveau/build.gradle b/nouveau/build.gradle
index 11836ead8..f3f65e446 100644
--- a/nouveau/build.gradle
+++ b/nouveau/build.gradle
@@ -20,7 +20,6 @@ dependencies {
implementation 'io.dropwizard:dropwizard-core'
implementation 'io.dropwizard:dropwizard-http2'
implementation 'io.dropwizard.metrics:metrics-core'
- implementation 'io.dropwizard.metrics:metrics-caffeine'
implementation 'io.dropwizard.metrics:metrics-jersey2'
testImplementation 'io.dropwizard:dropwizard-testing'
diff --git a/nouveau/src/main/java/org/apache/couchdb/nouveau/NouveauApplication.java b/nouveau/src/main/java/org/apache/couchdb/nouveau/NouveauApplication.java
index 4373509ce..7179eadc0 100644
--- a/nouveau/src/main/java/org/apache/couchdb/nouveau/NouveauApplication.java
+++ b/nouveau/src/main/java/org/apache/couchdb/nouveau/NouveauApplication.java
@@ -13,7 +13,6 @@
package org.apache.couchdb.nouveau;
-import com.github.benmanes.caffeine.cache.Scheduler;
import io.dropwizard.core.Application;
import io.dropwizard.core.setup.Environment;
import io.swagger.v3.jaxrs2.integration.resources.OpenApiResource;
@@ -54,13 +53,12 @@ public class NouveauApplication extends Application<NouveauApplicationConfigurat
indexManager.setCommitIntervalSeconds(configuration.getCommitIntervalSeconds());
indexManager.setIdleSeconds(configuration.getIdleSeconds());
indexManager.setMaxIndexesOpen(configuration.getMaxIndexesOpen());
- indexManager.setMetricRegistry(environment.metrics());
final ScheduledExecutorService schedulerExecutorService = environment
.lifecycle()
.scheduledExecutorService("index-manager-%d")
.threads(configuration.getSchedulerThreadCount())
.build();
- indexManager.setScheduler(Scheduler.forScheduledExecutorService(schedulerExecutorService));
+ indexManager.setScheduledExecutorService(schedulerExecutorService);
indexManager.setSearcherFactory(new ParallelSearcherFactory(ForkJoinPool.commonPool()));
indexManager.setObjectMapper(environment.getObjectMapper());
indexManager.setRootDir(configuration.getRootDir());
diff --git a/nouveau/src/main/java/org/apache/couchdb/nouveau/core/Index.java b/nouveau/src/main/java/org/apache/couchdb/nouveau/core/Index.java
index b278fe6a4..69c9bd48f 100644
--- a/nouveau/src/main/java/org/apache/couchdb/nouveau/core/Index.java
+++ b/nouveau/src/main/java/org/apache/couchdb/nouveau/core/Index.java
@@ -17,7 +17,6 @@ import jakarta.ws.rs.WebApplicationException;
import jakarta.ws.rs.core.Response.Status;
import java.io.Closeable;
import java.io.IOException;
-import java.util.concurrent.Semaphore;
import org.apache.couchdb.nouveau.api.DocumentDeleteRequest;
import org.apache.couchdb.nouveau.api.DocumentUpdateRequest;
import org.apache.couchdb.nouveau.api.IndexInfo;
@@ -39,21 +38,12 @@ public abstract class Index implements Closeable {
private long updateSeq;
private long purgeSeq;
private boolean deleteOnClose = false;
- private final Semaphore semaphore = new Semaphore(Integer.MAX_VALUE);
protected Index(final long updateSeq, final long purgeSeq) {
this.updateSeq = updateSeq;
this.purgeSeq = purgeSeq;
}
- public final boolean tryAcquire() {
- return semaphore.tryAcquire();
- }
-
- public final void release() {
- semaphore.release();
- }
-
public final IndexInfo info() throws IOException {
final int numDocs = doNumDocs();
final long diskSize = doDiskSize();
@@ -130,10 +120,7 @@ public abstract class Index implements Closeable {
@Override
public final void close() throws IOException {
- // Ensures exclusive access to the index before closing.
- semaphore.acquireUninterruptibly(Integer.MAX_VALUE);
doClose();
- // Never release semaphore.
}
protected abstract void doClose() throws IOException;
@@ -146,10 +133,6 @@ public abstract class Index implements Closeable {
this.deleteOnClose = deleteOnClose;
}
- public final boolean isActive() {
- return semaphore.availablePermits() < Integer.MAX_VALUE || semaphore.hasQueuedThreads();
- }
-
protected final void assertUpdateSeqProgress(final long matchSeq, final long updateSeq)
throws UpdatesOutOfOrderException {
assert Thread.holdsLock(this);
diff --git a/nouveau/src/main/java/org/apache/couchdb/nouveau/core/IndexManager.java b/nouveau/src/main/java/org/apache/couchdb/nouveau/core/IndexManager.java
index f434d5169..2365b9619 100644
--- a/nouveau/src/main/java/org/apache/couchdb/nouveau/core/IndexManager.java
+++ b/nouveau/src/main/java/org/apache/couchdb/nouveau/core/IndexManager.java
@@ -13,18 +13,7 @@
package org.apache.couchdb.nouveau.core;
-import static com.codahale.metrics.MetricRegistry.name;
-
-import com.codahale.metrics.MetricRegistry;
-import com.codahale.metrics.caffeine.MetricsStatsCounter;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.github.benmanes.caffeine.cache.AsyncCacheLoader;
-import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
-import com.github.benmanes.caffeine.cache.Caffeine;
-import com.github.benmanes.caffeine.cache.RemovalCause;
-import com.github.benmanes.caffeine.cache.RemovalListener;
-import com.github.benmanes.caffeine.cache.Scheduler;
-import com.github.benmanes.caffeine.cache.Weigher;
import io.dropwizard.lifecycle.Managed;
import jakarta.validation.constraints.PositiveOrZero;
import jakarta.ws.rs.WebApplicationException;
@@ -34,12 +23,16 @@ import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
-import java.time.Duration;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executor;
+import java.util.Objects;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Stream;
import org.apache.couchdb.nouveau.api.IndexDefinition;
import org.apache.couchdb.nouveau.lucene9.Lucene9AnalyzerFactory;
@@ -52,7 +45,6 @@ import org.apache.lucene.search.SearcherFactory;
import org.apache.lucene.search.SearcherManager;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.FSDirectory;
-import org.checkerframework.checker.index.qual.NonNegative;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -62,11 +54,51 @@ import org.slf4j.LoggerFactory;
*/
public final class IndexManager implements Managed {
+ private class LRUMap extends LinkedHashMap<String, IndexHolder> {
+
+ private int capacity;
+
+ private LRUMap(final int capacity) {
+ super(capacity, 0.75f, true);
+ this.capacity = capacity;
+ }
+
+ @Override
+ protected boolean removeEldestEntry(final java.util.Map.Entry<String, IndexHolder> eldest) {
+ if (size() > capacity) {
+ schedulerExecutorService.execute(() -> {
+ try {
+ unload(eldest.getKey(), false);
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOGGER.warn("Interrupted while evicting eldest entry {}", eldest.getKey(), e);
+ } catch (final IOException e) {
+ LOGGER.warn("I/O exception while evicting eldest entry {}", eldest.getKey(), e);
+ }
+ });
+ }
+ return false;
+ }
+ }
+
@FunctionalInterface
public interface IndexFunction<V, R> {
R apply(final V value) throws IOException;
}
+ private enum HolderState {
+ NOT_LOADED,
+ LOADED,
+ UNLOADED
+ }
+
+ private static class IndexHolder {
+ private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
+ private ScheduledFuture<?> commitFuture;
+ private HolderState state = HolderState.NOT_LOADED;
+ private Index index;
+ }
+
private static final Logger LOGGER = LoggerFactory.getLogger(IndexManager.class);
@PositiveOrZero
@@ -82,43 +114,104 @@ public final class IndexManager implements Managed {
private ObjectMapper objectMapper;
- private MetricRegistry metricRegistry;
-
- private Scheduler scheduler;
+ private ScheduledExecutorService schedulerExecutorService;
private SearcherFactory searcherFactory;
- private AsyncLoadingCache<String, Index> cache;
+ private LRUMap cache;
- private StripedLock<String> lock;
+ private StripedLock<String> createLock;
public <R> R with(final String name, final IndexFunction<Index, R> indexFun)
throws IOException, InterruptedException {
+ retry:
while (true) {
if (!exists(name)) {
throw new WebApplicationException("Index does not exist", Status.NOT_FOUND);
}
-
- final CompletableFuture<Index> future = cache.get(name);
- final Index index = future.join();
-
- if (index.tryAcquire()) {
- try {
- return indexFun.apply(index);
- } finally {
- index.release();
+ final IndexHolder holder;
+ synchronized (cache) {
+ holder = cache.computeIfAbsent(name, (k) -> new IndexHolder());
+ }
+ holder.lock.readLock().lock();
+ try {
+ // Load if not already loaded.
+ if (holder.state == HolderState.NOT_LOADED) {
+ holder.lock.readLock().unlock();
+ holder.lock.writeLock().lock();
+ try {
+ if (holder.state == HolderState.NOT_LOADED) {
+ holder.index = load(name);
+ holder.commitFuture = this.schedulerExecutorService.scheduleWithFixedDelay(
+ commitFun(name, holder),
+ commitIntervalSeconds,
+ commitIntervalSeconds,
+ TimeUnit.SECONDS);
+ holder.state = HolderState.LOADED;
+ }
+ } finally {
+ holder.lock.readLock().lock();
+ holder.lock.writeLock().unlock();
+ }
+ }
+ switch (holder.state) {
+ case NOT_LOADED:
+ throw new IllegalStateException();
+ case UNLOADED:
+ Thread.sleep(1000);
+ continue retry;
+ case LOADED:
+ return indexFun.apply(holder.index);
}
+ } finally {
+ holder.lock.readLock().unlock();
}
}
}
+ public void unload(final String name, final boolean forceDelete) throws IOException, InterruptedException {
+ final IndexHolder holder;
+ synchronized (cache) {
+ holder = cache.computeIfAbsent(name, (k) -> new IndexHolder());
+ }
+ holder.lock.writeLock().lock();
+ try {
+ switch (holder.state) {
+ case LOADED:
+ if (forceDelete) {
+ holder.index.setDeleteOnClose(true);
+ }
+ LOGGER.info("closing {}", name);
+ try {
+ close(name, holder);
+ } catch (final IOException e) {
+ LOGGER.error("I/O exception when evicting {}", name, e);
+ }
+ holder.state = HolderState.UNLOADED;
+ holder.index = null;
+ synchronized (cache) {
+ cache.remove(name, holder);
+ }
+ break;
+ case NOT_LOADED:
+ case UNLOADED:
+ if (forceDelete) {
+ IOUtils.rm(indexRootPath(name));
+ }
+ break;
+ }
+ } finally {
+ holder.lock.writeLock().unlock();
+ }
+ }
+
public void create(final String name, IndexDefinition indexDefinition) throws IOException {
if (exists(name)) {
assertSame(indexDefinition, loadIndexDefinition(name));
return;
}
- final Lock lock = this.lock.writeLock(name);
+ final Lock lock = this.createLock.writeLock(name);
lock.lock();
try {
if (exists(name)) {
@@ -164,8 +257,8 @@ public final class IndexManager implements Managed {
final String relativeName = rootDir.relativize(p).toString();
try {
deleteIndex(relativeName);
- } catch (final IOException e) {
- LOGGER.error("I/O exception deleting " + p, e);
+ } catch (final IOException | InterruptedException e) {
+ LOGGER.error("Exception deleting {}", p, e);
}
// Clean any newly empty directories.
do {
@@ -180,18 +273,8 @@ public final class IndexManager implements Managed {
}
}
- private void deleteIndex(final String name) throws IOException {
- final CompletableFuture<Index> future = cache.asMap().remove(name);
- if (future == null) {
- return;
- }
- final Index index = future.getNow(null);
- if (index != null) {
- index.setDeleteOnClose(true);
- close(name, index);
- } else {
- IOUtils.rm(indexRootPath(name));
- }
+ private void deleteIndex(final String name) throws IOException, InterruptedException {
+ unload(name, true);
}
public void setMaxIndexesOpen(int maxIndexesOpen) {
@@ -214,12 +297,8 @@ public final class IndexManager implements Managed {
this.objectMapper = objectMapper;
}
- public void setMetricRegistry(final MetricRegistry metricRegistry) {
- this.metricRegistry = metricRegistry;
- }
-
- public void setScheduler(final Scheduler scheduler) {
- this.scheduler = scheduler;
+ public void setScheduledExecutorService(ScheduledExecutorService schedulerExecutorService) {
+ this.schedulerExecutorService = schedulerExecutorService;
}
public void setSearcherFactory(final SearcherFactory searcherFactory) {
@@ -228,27 +307,38 @@ public final class IndexManager implements Managed {
@Override
public void start() throws IOException {
- cache = Caffeine.newBuilder()
- .recordStats(() -> new MetricsStatsCounter(metricRegistry, name(IndexManager.class, "cache")))
- .initialCapacity(maxIndexesOpen)
- .maximumWeight(maxIndexesOpen)
- .weigher(new IndexWeigher())
- .expireAfterAccess(Duration.ofSeconds(idleSeconds))
- .refreshAfterWrite(Duration.ofSeconds(commitIntervalSeconds))
- .scheduler(scheduler)
- .evictionListener(new IndexEvictionListener())
- .buildAsync(new AsyncIndexLoader());
- lock = new StripedLock<String>(100);
+ Objects.requireNonNull(this.objectMapper, "objectMapper must not be null");
+ Objects.requireNonNull(this.rootDir, "rootDir must not be null");
+ Objects.requireNonNull(this.searcherFactory, "searcherFactory must not be null");
+ Objects.requireNonNull(this.schedulerExecutorService, "schedulerExecutorService must not be null");
+
+ this.cache = new LRUMap(this.maxIndexesOpen);
+ this.createLock = new StripedLock<String>(100);
+ }
+
+ private Runnable commitFun(final String name, final IndexHolder holder) {
+ return () -> {
+ holder.lock.readLock().lock();
+ try {
+ if (holder.index.commit()) {
+ LOGGER.info("committed {}", name);
+ }
+ } catch (final IOException e) {
+ LOGGER.warn("I/O exception while committing " + name, e);
+ } finally {
+ holder.lock.readLock().unlock();
+ }
+ };
}
@Override
public void stop() throws IOException, InterruptedException {
- final var it = cache.asMap().entrySet().iterator();
- while (it.hasNext()) {
- var e = it.next();
- LOGGER.info("closing {} during shutdown", e.getKey());
- close(e.getKey(), e.getValue());
- it.remove();
+ synchronized (cache) {
+ var names = new HashSet<String>(cache.keySet());
+ for (var name : names) {
+ unload(name, false);
+ }
+ cache.clear();
}
}
@@ -276,115 +366,50 @@ public final class IndexManager implements Managed {
throw new WebApplicationException(name + " attempts to escape from index root directory", Status.BAD_REQUEST);
}
- private class IndexEvictionListener implements RemovalListener<String, Index> {
-
- public void onRemoval(String name, Index index, RemovalCause cause) {
- LOGGER.info("closing {} for cause {}", name, cause);
- try {
- close(name, index);
- } catch (final IOException e) {
- LOGGER.error("I/O exception when evicting " + name, e);
- }
- }
+ private Index load(final String name) throws IOException {
+ LOGGER.info("opening {}", name);
+ final Path path = indexPath(name);
+ final IndexDefinition indexDefinition = loadIndexDefinition(name);
+ final Analyzer analyzer = Lucene9AnalyzerFactory.fromDefinition(indexDefinition);
+ final Directory dir = new DirectIODirectory(FSDirectory.open(path.resolve("9")));
+ final IndexWriterConfig config = new IndexWriterConfig(analyzer);
+ config.setUseCompoundFile(false);
+ final IndexWriter writer = new IndexWriter(dir, config);
+ final long updateSeq = getSeq(writer, "update_seq");
+ final long purgeSeq = getSeq(writer, "purge_seq");
+ final SearcherManager searcherManager = new SearcherManager(writer, searcherFactory);
+ return new Lucene9Index(analyzer, writer, updateSeq, purgeSeq, searcherManager);
}
- private class IndexWeigher implements Weigher<String, Index> {
-
- @Override
- public @NonNegative int weigh(String key, Index value) {
- // Pin active indexes
- return value.isActive() ? 0 : 1;
- }
- }
-
- private class AsyncIndexLoader implements AsyncCacheLoader<String, Index> {
-
- @Override
- public CompletableFuture<? extends Index> asyncLoad(String name, Executor executor) throws Exception {
- final CompletableFuture<Index> future = new CompletableFuture<Index>();
-
- executor.execute(() -> {
- LOGGER.info("opening {}", name);
- final Path path = indexPath(name);
- Index result;
- try {
- final IndexDefinition indexDefinition = loadIndexDefinition(name);
- final Analyzer analyzer = Lucene9AnalyzerFactory.fromDefinition(indexDefinition);
- final Directory dir = new DirectIODirectory(FSDirectory.open(path.resolve("9")));
- final IndexWriterConfig config = new IndexWriterConfig(analyzer);
- config.setUseCompoundFile(false);
- final IndexWriter writer = new IndexWriter(dir, config);
- final long updateSeq = getSeq(writer, "update_seq");
- final long purgeSeq = getSeq(writer, "purge_seq");
- final SearcherManager searcherManager = new SearcherManager(writer, searcherFactory);
- result = new Lucene9Index(analyzer, writer, updateSeq, purgeSeq, searcherManager);
- future.complete(result);
- } catch (IOException e) {
- future.completeExceptionally(e);
- }
- });
-
- return future;
- }
-
- @Override
- public CompletableFuture<? extends Index> asyncReload(String name, Index index, Executor executor)
- throws Exception {
- executor.execute(() -> {
- if (index.tryAcquire()) {
- try {
- if (index.commit()) {
- LOGGER.info("committed {}", name);
- }
- } catch (final IOException e) {
- LOGGER.warn("I/O exception while committing " + name, e);
- } finally {
- index.release();
- }
- }
- });
- return CompletableFuture.completedFuture(index);
+ private long getSeq(final IndexWriter writer, final String key) throws IOException {
+ final Iterable<Map.Entry<String, String>> commitData = writer.getLiveCommitData();
+ if (commitData == null) {
+ return 0L;
}
-
- private long getSeq(final IndexWriter writer, final String key) throws IOException {
- final Iterable<Map.Entry<String, String>> commitData = writer.getLiveCommitData();
- if (commitData == null) {
- return 0L;
+ for (Map.Entry<String, String> entry : commitData) {
+ if (entry.getKey().equals(key)) {
+ return Long.parseLong(entry.getValue());
}
- for (Map.Entry<String, String> entry : commitData) {
- if (entry.getKey().equals(key)) {
- return Long.parseLong(entry.getValue());
- }
- }
- return 0L;
}
+ return 0L;
}
- private void close(final String name, final CompletableFuture<Index> future) throws IOException {
- final Index index = future.getNow(null);
- if (index != null) {
- close(name, index);
+ private void close(final String name, final IndexHolder holder) throws IOException {
+ if (!holder.lock.isWriteLockedByCurrentThread()) {
+ throw new IllegalStateException();
}
- }
-
- private void close(final String name, final Index index) throws IOException {
+ holder.commitFuture.cancel(true);
IOUtils.runAll(
() -> {
- if (index.tryAcquire()) {
- try {
- if (index.commit()) {
- LOGGER.debug("committed {} before close", name);
- }
- } finally {
- index.release();
- }
+ if (holder.index.commit()) {
+ LOGGER.debug("committed {} before close", name);
}
},
() -> {
- index.close();
+ holder.index.close();
},
() -> {
- if (index.isDeleteOnClose()) {
+ if (holder.index.isDeleteOnClose()) {
IOUtils.rm(indexRootPath(name));
}
});
diff --git a/nouveau/src/test/java/org/apache/couchdb/nouveau/health/IndexHealthCheckTest.java b/nouveau/src/test/java/org/apache/couchdb/nouveau/health/IndexHealthCheckTest.java
index d282de568..0c777fbab 100644
--- a/nouveau/src/test/java/org/apache/couchdb/nouveau/health/IndexHealthCheckTest.java
+++ b/nouveau/src/test/java/org/apache/couchdb/nouveau/health/IndexHealthCheckTest.java
@@ -15,10 +15,9 @@ package org.apache.couchdb.nouveau.health;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import com.codahale.metrics.MetricRegistry;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.github.benmanes.caffeine.cache.Scheduler;
import java.nio.file.Path;
+import java.util.concurrent.Executors;
import org.apache.couchdb.nouveau.core.IndexManager;
import org.apache.couchdb.nouveau.resources.IndexResource;
import org.apache.lucene.search.SearcherFactory;
@@ -33,10 +32,9 @@ public class IndexHealthCheckTest {
manager.setCommitIntervalSeconds(30);
manager.setIdleSeconds(60);
manager.setMaxIndexesOpen(1);
- manager.setMetricRegistry(new MetricRegistry());
manager.setObjectMapper(new ObjectMapper());
manager.setRootDir(tempDir);
- manager.setScheduler(Scheduler.systemScheduler());
+ manager.setScheduledExecutorService(Executors.newScheduledThreadPool(2));
manager.setSearcherFactory(new SearcherFactory());
manager.start();
(couchdb) 02/02: support concurrency across multiple segments
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch decaffeinated
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 7d284d4bfee0451d35253c95a8bd4772f33e7409
Author: Robert Newson <Ro...@ibm.com>
AuthorDate: Wed Jun 12 16:09:46 2024 +0100
support concurrency across multiple segments
---
.../src/main/java/org/apache/couchdb/nouveau/lucene9/Lucene9Index.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/nouveau/src/main/java/org/apache/couchdb/nouveau/lucene9/Lucene9Index.java b/nouveau/src/main/java/org/apache/couchdb/nouveau/lucene9/Lucene9Index.java
index a83724a9f..a9734f31d 100644
--- a/nouveau/src/main/java/org/apache/couchdb/nouveau/lucene9/Lucene9Index.java
+++ b/nouveau/src/main/java/org/apache/couchdb/nouveau/lucene9/Lucene9Index.java
@@ -232,7 +232,7 @@ public class Lucene9Index extends Index {
fieldDoc = null;
}
- return new TopFieldCollectorManager(sort, searchRequest.getLimit(), fieldDoc, 1000, false);
+ return new TopFieldCollectorManager(sort, searchRequest.getLimit(), fieldDoc, 1000, true);
}
private SortField getLastSortField(final Sort sort) {