fullDocs() {
+ return new PqFullDocs<>(this.pq, new TotalHits(totalHits, totalHitsRelation));
+ }
+
+ @Override
+ public void close() throws Exception {
+ pq.close();
+ }
+}
\ No newline at end of file
diff --git a/src/main/java/it/cavallium/dbengine/lucene/collector/HitsThresholdChecker.java b/src/main/java/it/cavallium/dbengine/lucene/collector/HitsThresholdChecker.java
new file mode 100644
index 0000000..768be13
--- /dev/null
+++ b/src/main/java/it/cavallium/dbengine/lucene/collector/HitsThresholdChecker.java
@@ -0,0 +1,119 @@
+/*
+ * 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 it.cavallium.dbengine.lucene.collector;
+
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.lucene.search.ScoreMode;
+
+/** Used for defining custom algorithms to allow searches to early terminate */
+abstract class HitsThresholdChecker {
+ /** Implementation of HitsThresholdChecker which allows global hit counting */
+ private static class GlobalHitsThresholdChecker extends HitsThresholdChecker {
+ private final int totalHitsThreshold;
+ private final AtomicLong globalHitCount;
+
+ public GlobalHitsThresholdChecker(int totalHitsThreshold) {
+
+ if (totalHitsThreshold < 0) {
+ throw new IllegalArgumentException(
+ "totalHitsThreshold must be >= 0, got " + totalHitsThreshold);
+ }
+
+ this.totalHitsThreshold = totalHitsThreshold;
+ this.globalHitCount = new AtomicLong();
+ }
+
+ @Override
+ public void incrementHitCount() {
+ globalHitCount.incrementAndGet();
+ }
+
+ @Override
+ public boolean isThresholdReached() {
+ return globalHitCount.getAcquire() > totalHitsThreshold;
+ }
+
+ @Override
+ public ScoreMode scoreMode() {
+ return totalHitsThreshold == Integer.MAX_VALUE ? ScoreMode.COMPLETE : ScoreMode.TOP_SCORES;
+ }
+
+ @Override
+ public int getHitsThreshold() {
+ return totalHitsThreshold;
+ }
+ }
+
+ /** Default implementation of HitsThresholdChecker to be used for single threaded execution */
+ private static class LocalHitsThresholdChecker extends HitsThresholdChecker {
+ private final int totalHitsThreshold;
+ private int hitCount;
+
+ public LocalHitsThresholdChecker(int totalHitsThreshold) {
+
+ if (totalHitsThreshold < 0) {
+ throw new IllegalArgumentException(
+ "totalHitsThreshold must be >= 0, got " + totalHitsThreshold);
+ }
+
+ this.totalHitsThreshold = totalHitsThreshold;
+ }
+
+ @Override
+ public void incrementHitCount() {
+ ++hitCount;
+ }
+
+ @Override
+ public boolean isThresholdReached() {
+ return hitCount > totalHitsThreshold;
+ }
+
+ @Override
+ public ScoreMode scoreMode() {
+ return totalHitsThreshold == Integer.MAX_VALUE ? ScoreMode.COMPLETE : ScoreMode.TOP_SCORES;
+ }
+
+ @Override
+ public int getHitsThreshold() {
+ return totalHitsThreshold;
+ }
+ }
+
+ /*
+ * Returns a threshold checker that is useful for single threaded searches
+ */
+ public static HitsThresholdChecker create(final int totalHitsThreshold) {
+ return new LocalHitsThresholdChecker(totalHitsThreshold);
+ }
+
+ /*
+ * Returns a threshold checker that is based on a shared counter
+ */
+ public static HitsThresholdChecker createShared(final int totalHitsThreshold) {
+ return new GlobalHitsThresholdChecker(totalHitsThreshold);
+ }
+
+ public abstract void incrementHitCount();
+
+ public abstract ScoreMode scoreMode();
+
+ public abstract int getHitsThreshold();
+
+ public abstract boolean isThresholdReached();
+}
\ No newline at end of file
diff --git a/src/main/java/it/cavallium/dbengine/lucene/collector/LMDBFullScoreDocCollector.java b/src/main/java/it/cavallium/dbengine/lucene/collector/LMDBFullScoreDocCollector.java
new file mode 100644
index 0000000..d0d3289
--- /dev/null
+++ b/src/main/java/it/cavallium/dbengine/lucene/collector/LMDBFullScoreDocCollector.java
@@ -0,0 +1,238 @@
+/*
+ * 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 it.cavallium.dbengine.lucene.collector;
+
+import it.cavallium.dbengine.database.disk.LLTempLMDBEnv;
+import it.cavallium.dbengine.lucene.FullDocs;
+import it.cavallium.dbengine.lucene.LLScoreDoc;
+import it.cavallium.dbengine.lucene.LLScoreDocCodec;
+import it.cavallium.dbengine.lucene.LMDBPriorityQueue;
+import it.cavallium.dbengine.lucene.MaxScoreAccumulator;
+import java.io.IOException;
+import java.util.Collection;
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.search.Collector;
+import org.apache.lucene.search.CollectorManager;
+import org.apache.lucene.search.IndexSearcher;
+import org.apache.lucene.search.LeafCollector;
+import it.cavallium.dbengine.lucene.MaxScoreAccumulator.DocAndScore;
+import org.apache.lucene.search.Scorable;
+import org.apache.lucene.search.ScoreMode;
+import org.apache.lucene.search.TotalHits;
+
+/**
+ * A {@link Collector} implementation that collects the top-scoring hits, returning them as a {@link
+ * FullDocs}. This is used by {@link IndexSearcher} to implement {@link FullDocs}-based search. Hits
+ * are sorted by score descending and then (when the scores are tied) docID ascending. When you
+ * create an instance of this collector you should know in advance whether documents are going to be
+ * collected in doc Id order or not.
+ *
+ * NOTE: The values {@link Float#NaN} and {@link Float#NEGATIVE_INFINITY} are not valid
+ * scores. This collector will not properly collect hits with such scores.
+ */
+public abstract class LMDBFullScoreDocCollector extends FullDocsCollector {
+
+ /** Scorable leaf collector */
+ public abstract static class ScorerLeafCollector implements LeafCollector {
+
+ protected Scorable scorer;
+
+ @Override
+ public void setScorer(Scorable scorer) throws IOException {
+ this.scorer = scorer;
+ }
+ }
+
+ private static class SimpleLMDBFullScoreDocCollector extends LMDBFullScoreDocCollector {
+
+ SimpleLMDBFullScoreDocCollector(LLTempLMDBEnv env,
+ HitsThresholdChecker hitsThresholdChecker, MaxScoreAccumulator minScoreAcc) {
+ super(env, hitsThresholdChecker, minScoreAcc);
+ }
+
+ @Override
+ public LeafCollector getLeafCollector(LeafReaderContext context) {
+ // reset the minimum competitive score
+ docBase = context.docBase;
+ return new ScorerLeafCollector() {
+
+ @Override
+ public void setScorer(Scorable scorer) throws IOException {
+ super.setScorer(scorer);
+ minCompetitiveScore = 0f;
+ updateMinCompetitiveScore(scorer);
+ if (minScoreAcc != null) {
+ updateGlobalMinCompetitiveScore(scorer);
+ }
+ }
+
+ @Override
+ public void collect(int doc) throws IOException {
+ float score = scorer.score();
+
+ // This collector relies on the fact that scorers produce positive values:
+ assert score >= 0; // NOTE: false for NaN
+
+ totalHits++;
+ hitsThresholdChecker.incrementHitCount();
+
+ if (minScoreAcc != null && (totalHits & minScoreAcc.modInterval) == 0) {
+ updateGlobalMinCompetitiveScore(scorer);
+ }
+
+ var pqTop = pq.top();
+ if (pqTop != null) {
+ if (score <= pqTop.score()) {
+ if (totalHitsRelation == TotalHits.Relation.EQUAL_TO) {
+ // we just reached totalHitsThreshold, we can start setting the min
+ // competitive score now
+ updateMinCompetitiveScore(scorer);
+ }
+ // Since docs are returned in-order (i.e., increasing doc Id), a document
+ // with equal score to pqTop.score cannot compete since HitQueue favors
+ // documents with lower doc Ids. Therefore reject those docs too.
+ return;
+ }
+ }
+ pq.add(new LLScoreDoc(doc + docBase, score, -1));
+ pq.updateTop();
+ updateMinCompetitiveScore(scorer);
+ }
+ };
+ }
+ }
+
+ /**
+ * Creates a new {@link LMDBFullScoreDocCollector} given the number of hits to collect and the number
+ * of hits to count accurately.
+ *
+ * NOTE: If the total hit count of the top docs is less than or exactly {@code
+ * totalHitsThreshold} then this value is accurate. On the other hand, if the {@link
+ * FullDocs#totalHits} value is greater than {@code totalHitsThreshold} then its value is a lower
+ * bound of the hit count. A value of {@link Integer#MAX_VALUE} will make the hit count accurate
+ * but will also likely make query processing slower.
+ *
+ *
NOTE: The instances returned by this method pre-allocate a full array of length
+ * numHits, and fill the array with sentinel objects.
+ */
+ public static LMDBFullScoreDocCollector create(LLTempLMDBEnv env, int totalHitsThreshold) {
+ return create(env, HitsThresholdChecker.create(totalHitsThreshold), null);
+ }
+
+ static LMDBFullScoreDocCollector create(
+ LLTempLMDBEnv env,
+ HitsThresholdChecker hitsThresholdChecker,
+ MaxScoreAccumulator minScoreAcc) {
+
+ if (hitsThresholdChecker == null) {
+ throw new IllegalArgumentException("hitsThresholdChecker must be non null");
+ }
+
+ return new SimpleLMDBFullScoreDocCollector(env, hitsThresholdChecker, minScoreAcc);
+ }
+
+ /**
+ * Create a CollectorManager which uses a shared hit counter to maintain number of hits and a
+ * shared {@link MaxScoreAccumulator} to propagate the minimum score accross segments
+ */
+ public static CollectorManager> createSharedManager(
+ LLTempLMDBEnv env,
+ int totalHitsThreshold) {
+ return new CollectorManager<>() {
+
+ private final HitsThresholdChecker hitsThresholdChecker =
+ HitsThresholdChecker.createShared(totalHitsThreshold);
+ private final MaxScoreAccumulator minScoreAcc = new MaxScoreAccumulator();
+
+ @Override
+ public LMDBFullScoreDocCollector newCollector() {
+ return LMDBFullScoreDocCollector.create(env, hitsThresholdChecker, minScoreAcc);
+ }
+
+ @Override
+ public FullDocs reduce(Collection collectors) {
+ @SuppressWarnings("unchecked")
+ final FullDocs[] fullDocs = new FullDocs[collectors.size()];
+ int i = 0;
+ for (LMDBFullScoreDocCollector collector : collectors) {
+ fullDocs[i++] = collector.fullDocs();
+ }
+ return FullDocs.merge(null, fullDocs);
+ }
+ };
+ }
+
+ int docBase;
+ final HitsThresholdChecker hitsThresholdChecker;
+ final MaxScoreAccumulator minScoreAcc;
+ float minCompetitiveScore;
+
+ // prevents instantiation
+ LMDBFullScoreDocCollector(LLTempLMDBEnv env,
+ HitsThresholdChecker hitsThresholdChecker, MaxScoreAccumulator minScoreAcc) {
+ super(new LMDBPriorityQueue<>(env, new LLScoreDocCodec()));
+ assert hitsThresholdChecker != null;
+
+ this.hitsThresholdChecker = hitsThresholdChecker;
+ this.minScoreAcc = minScoreAcc;
+ }
+
+ @Override
+ public ScoreMode scoreMode() {
+ return hitsThresholdChecker.scoreMode();
+ }
+
+ protected void updateGlobalMinCompetitiveScore(Scorable scorer) throws IOException {
+ assert minScoreAcc != null;
+ DocAndScore maxMinScore = minScoreAcc.get();
+ if (maxMinScore != null) {
+ // since we tie-break on doc id and collect in doc id order we can require
+ // the next float if the global minimum score is set on a document id that is
+ // smaller than the ids in the current leaf
+ float score =
+ docBase > maxMinScore.docID ? Math.nextUp(maxMinScore.score) : maxMinScore.score;
+ if (score > minCompetitiveScore) {
+ assert hitsThresholdChecker.isThresholdReached();
+ scorer.setMinCompetitiveScore(score);
+ minCompetitiveScore = score;
+ totalHitsRelation = TotalHits.Relation.GREATER_THAN_OR_EQUAL_TO;
+ }
+ }
+ }
+
+ protected void updateMinCompetitiveScore(Scorable scorer) throws IOException {
+ var pqTop = pq.top();
+ if (hitsThresholdChecker.isThresholdReached()
+ && pqTop != null
+ && pqTop.score() != Float.NEGATIVE_INFINITY) { // -Infinity is the score of sentinels
+ // since we tie-break on doc id and collect in doc id order, we can require
+ // the next float
+ float localMinScore = Math.nextUp(pqTop.score());
+ if (localMinScore > minCompetitiveScore) {
+ scorer.setMinCompetitiveScore(localMinScore);
+ totalHitsRelation = TotalHits.Relation.GREATER_THAN_OR_EQUAL_TO;
+ minCompetitiveScore = localMinScore;
+ if (minScoreAcc != null) {
+ // we don't use the next float but we register the document
+ // id so that other leaves can require it if they are after
+ // the current maximum
+ minScoreAcc.accumulate(pqTop.doc(), pqTop.score());
+ }
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/src/main/java/it/cavallium/dbengine/lucene/UnscoredCollector.java b/src/main/java/it/cavallium/dbengine/lucene/collector/UnscoredCollector.java
similarity index 98%
rename from src/main/java/it/cavallium/dbengine/lucene/UnscoredCollector.java
rename to src/main/java/it/cavallium/dbengine/lucene/collector/UnscoredCollector.java
index d6caaf7..898a8b0 100644
--- a/src/main/java/it/cavallium/dbengine/lucene/UnscoredCollector.java
+++ b/src/main/java/it/cavallium/dbengine/lucene/collector/UnscoredCollector.java
@@ -1,4 +1,4 @@
-package it.cavallium.dbengine.lucene;
+package it.cavallium.dbengine.lucene.collector;
import static it.cavallium.dbengine.lucene.searcher.PaginationInfo.ALLOW_UNSCORED_PAGINATION_MODE;
diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/AdaptiveLuceneLocalSearcher.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/AdaptiveLuceneLocalSearcher.java
index ea3a6ac..db315d0 100644
--- a/src/main/java/it/cavallium/dbengine/lucene/searcher/AdaptiveLuceneLocalSearcher.java
+++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/AdaptiveLuceneLocalSearcher.java
@@ -35,6 +35,12 @@ public class AdaptiveLuceneLocalSearcher implements LuceneLocalSearcher {
true);
}
}
+
+ @Override
+ public String getName() {
+ return "adaptivelocal";
+ }
+
public Mono> transformedCollect(Mono> indexSearcher,
LocalQueryParams queryParams,
String keyFieldName,
diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/AdaptiveLuceneMultiSearcher.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/AdaptiveLuceneMultiSearcher.java
index d6b65df..8476a95 100644
--- a/src/main/java/it/cavallium/dbengine/lucene/searcher/AdaptiveLuceneMultiSearcher.java
+++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/AdaptiveLuceneMultiSearcher.java
@@ -4,15 +4,16 @@ import io.net5.buffer.api.Send;
import it.cavallium.dbengine.database.LLUtils;
import it.cavallium.dbengine.database.disk.LLIndexSearchers;
import it.cavallium.dbengine.lucene.searcher.LLSearchTransformer.TransformerInput;
+import java.io.Closeable;
+import java.io.IOException;
import reactor.core.publisher.Mono;
-public class AdaptiveLuceneMultiSearcher implements LuceneMultiSearcher {
+public class AdaptiveLuceneMultiSearcher implements LuceneMultiSearcher, Closeable {
private static final LuceneMultiSearcher count
= new SimpleUnsortedUnscoredLuceneMultiSearcher(new CountLuceneLocalSearcher());
- private static final LuceneMultiSearcher scoredSimple
- = new ScoredSimpleLuceneShardSearcher();
+ private static final LuceneMultiSearcher scoredSimple = new ScoredSimpleLuceneMultiSearcher();
private static final LuceneMultiSearcher unsortedUnscoredPaged
= new SimpleUnsortedUnscoredLuceneMultiSearcher(new SimpleLuceneLocalSearcher());
@@ -20,6 +21,12 @@ public class AdaptiveLuceneMultiSearcher implements LuceneMultiSearcher {
private static final LuceneMultiSearcher unsortedUnscoredContinuous
= new UnsortedUnscoredContinuousLuceneMultiSearcher();
+ private final UnsortedScoredFullLuceneMultiSearcher scoredFull;
+
+ public AdaptiveLuceneMultiSearcher() throws IOException {
+ scoredFull = new UnsortedScoredFullLuceneMultiSearcher();
+ }
+
@Override
public Mono> collectMulti(Mono> indexSearchersMono,
LocalQueryParams queryParams,
@@ -47,7 +54,11 @@ public class AdaptiveLuceneMultiSearcher implements LuceneMultiSearcher {
if (queryParams.limit() == 0) {
return count.collectMulti(indexSearchersMono, queryParams, keyFieldName, transformer);
} else if (queryParams.isSorted() || queryParams.isScored()) {
- return scoredSimple.collectMulti(indexSearchersMono, queryParams, keyFieldName, transformer);
+ if (queryParams.isSorted() || realLimit <= (long) queryParams.pageLimits().getPageLimit(0)) {
+ return scoredSimple.collectMulti(indexSearchersMono, queryParams, keyFieldName, transformer);
+ } else {
+ return scoredFull.collectMulti(indexSearchersMono, queryParams, keyFieldName, transformer);
+ }
} else if (realLimit <= (long) queryParams.pageLimits().getPageLimit(0)) {
// Run single-page searches using the paged multi searcher
return unsortedUnscoredPaged.collectMulti(indexSearchersMono, queryParams, keyFieldName, transformer);
@@ -57,4 +68,14 @@ public class AdaptiveLuceneMultiSearcher implements LuceneMultiSearcher {
}
}, true);
}
+
+ @Override
+ public void close() throws IOException {
+ scoredFull.close();
+ }
+
+ @Override
+ public String getName() {
+ return "adaptivemulti";
+ }
}
diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/CalculatedResults.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/CalculatedResults.java
new file mode 100644
index 0000000..f4cef12
--- /dev/null
+++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/CalculatedResults.java
@@ -0,0 +1,7 @@
+package it.cavallium.dbengine.lucene.searcher;
+
+import it.cavallium.dbengine.client.query.current.data.TotalHitsCount;
+import it.cavallium.dbengine.database.LLKeyScore;
+import reactor.core.publisher.Flux;
+
+record CalculatedResults(TotalHitsCount totalHitsCount, Flux firstPageHitsFlux) {}
diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/CountLuceneLocalSearcher.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/CountLuceneLocalSearcher.java
index 6342eef..ddbd9ba 100644
--- a/src/main/java/it/cavallium/dbengine/lucene/searcher/CountLuceneLocalSearcher.java
+++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/CountLuceneLocalSearcher.java
@@ -42,4 +42,9 @@ public class CountLuceneLocalSearcher implements LuceneLocalSearcher {
.map(count -> new LuceneSearchResult(TotalHitsCount.of(count, true), Flux.empty(), null).send())
.doOnDiscard(Send.class, Send::close);
}
+
+ @Override
+ public String getName() {
+ return "count";
+ }
}
diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/LuceneLocalSearcher.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/LuceneLocalSearcher.java
index dfc5bb5..1a57498 100644
--- a/src/main/java/it/cavallium/dbengine/lucene/searcher/LuceneLocalSearcher.java
+++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/LuceneLocalSearcher.java
@@ -16,4 +16,10 @@ public interface LuceneLocalSearcher {
LocalQueryParams queryParams,
String keyFieldName,
LLSearchTransformer transformer);
+
+ /**
+ * Get the name of this searcher type
+ * @return searcher type name
+ */
+ String getName();
}
diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/ScoredSimpleLuceneShardSearcher.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/ScoredSimpleLuceneMultiSearcher.java
similarity index 91%
rename from src/main/java/it/cavallium/dbengine/lucene/searcher/ScoredSimpleLuceneShardSearcher.java
rename to src/main/java/it/cavallium/dbengine/lucene/searcher/ScoredSimpleLuceneMultiSearcher.java
index 5001683..1e63219 100644
--- a/src/main/java/it/cavallium/dbengine/lucene/searcher/ScoredSimpleLuceneShardSearcher.java
+++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/ScoredSimpleLuceneMultiSearcher.java
@@ -1,34 +1,38 @@
package it.cavallium.dbengine.lucene.searcher;
-import static it.cavallium.dbengine.lucene.searcher.PaginationInfo.FIRST_PAGE_LIMIT;
+import static it.cavallium.dbengine.lucene.searcher.CurrentPageInfo.EMPTY_STATUS;
import static it.cavallium.dbengine.lucene.searcher.PaginationInfo.MAX_SINGLE_SEARCH_LIMIT;
import io.net5.buffer.api.Send;
import it.cavallium.dbengine.database.LLKeyScore;
import it.cavallium.dbengine.database.LLUtils;
-import it.cavallium.dbengine.database.disk.LLIndexSearcher;
import it.cavallium.dbengine.database.disk.LLIndexSearchers;
-import it.cavallium.dbengine.database.disk.LLLocalGroupedReactiveRocksIterator;
import it.cavallium.dbengine.lucene.LuceneUtils;
import it.cavallium.dbengine.lucene.searcher.LLSearchTransformer.TransformerInput;
+import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.lucene.search.FieldDoc;
import org.apache.lucene.search.IndexSearcher;
+import org.apache.lucene.search.ScoreDoc;
import org.apache.lucene.search.Sort;
+import org.apache.lucene.search.TopDocs;
+import org.apache.lucene.search.TotalHits;
+import org.apache.lucene.search.TotalHits.Relation;
+import org.jetbrains.annotations.Nullable;
import org.warp.commonutils.log.Logger;
import org.warp.commonutils.log.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
-public class ScoredSimpleLuceneShardSearcher implements LuceneMultiSearcher {
+public class ScoredSimpleLuceneMultiSearcher implements LuceneMultiSearcher {
- protected static final Logger logger = LoggerFactory.getLogger(ScoredSimpleLuceneShardSearcher.class);
+ protected static final Logger logger = LoggerFactory.getLogger(ScoredSimpleLuceneMultiSearcher.class);
- public ScoredSimpleLuceneShardSearcher() {
+ public ScoredSimpleLuceneMultiSearcher() {
}
@Override
@@ -64,11 +68,7 @@ public class ScoredSimpleLuceneShardSearcher implements LuceneMultiSearcher {
}
private Sort getSort(LocalQueryParams queryParams) {
- Sort luceneSort = queryParams.sort();
- if (luceneSort == null) {
- luceneSort = Sort.RELEVANCE;
- }
- return luceneSort;
+ return queryParams.sort();
}
/**
@@ -175,8 +175,8 @@ public class ScoredSimpleLuceneShardSearcher implements LuceneMultiSearcher {
if (resultsOffset < 0) {
throw new IndexOutOfBoundsException(resultsOffset);
}
- if ((s.pageIndex() == 0 || s.last() != null) && s.remainingLimit() > 0) {
- var sort = getSort(queryParams);
+ if (s.pageIndex() == 0 || (s.last() != null && s.remainingLimit() > 0)) {
+ @Nullable var sort = getSort(queryParams);
var pageLimit = pageLimits.getPageLimit(s.pageIndex());
var after = (FieldDoc) s.last();
var totalHitsThreshold = LuceneUtils.totalHitsThreshold();
@@ -211,4 +211,9 @@ public class ScoredSimpleLuceneShardSearcher implements LuceneMultiSearcher {
}))
);
}
+
+ @Override
+ public String getName() {
+ return "scoredsimplemulti";
+ }
}
diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/ScoringShardsCollectorManager.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/ScoringShardsCollectorManager.java
index 92011ad..5805e4b 100644
--- a/src/main/java/it/cavallium/dbengine/lucene/searcher/ScoringShardsCollectorManager.java
+++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/ScoringShardsCollectorManager.java
@@ -18,6 +18,7 @@ import reactor.core.scheduler.Schedulers;
public class ScoringShardsCollectorManager implements CollectorManager {
+ @Nullable
private final Sort sort;
private final int numHits;
private final FieldDoc after;
@@ -26,7 +27,7 @@ public class ScoringShardsCollectorManager implements CollectorManager sharedCollectorManager;
- public ScoringShardsCollectorManager(final Sort sort,
+ public ScoringShardsCollectorManager(@Nullable final Sort sort,
final int numHits,
final FieldDoc after,
final int totalHitsThreshold,
@@ -35,7 +36,7 @@ public class ScoringShardsCollectorManager implements CollectorManager new CurrentPageInfo(null, limit, 0))
- .handle((s, sink) -> this.searchPageSync(queryParams, indexSearchers, pagination, resultsOffset, s, sink));
+ .just(currentPageInfo)
+ .handle((s, sink) -> this.searchPageSync(queryParams, indexSearchers, pagination, resultsOffset, s, sink))
+ //defaultIfEmpty(new PageData(new TopDocs(new TotalHits(0, Relation.EQUAL_TO), new ScoreDoc[0]), currentPageInfo))
+ .single();
}
/**
@@ -108,7 +119,7 @@ public class SimpleLuceneLocalSearcher implements LuceneLocalSearcher {
CurrentPageInfo nextPageInfo = firstPageData.nextPageInfo();
return new FirstPageResults(totalHitsCount, firstPageHitsFlux, nextPageInfo);
- });
+ }).single();
}
private Mono> computeOtherResults(Mono firstResultMono,
@@ -125,7 +136,7 @@ public class SimpleLuceneLocalSearcher implements LuceneLocalSearcher {
Flux combinedFlux = firstPageHitsFlux.concatWith(nextHitsFlux);
return new LuceneSearchResult(totalHitsCount, combinedFlux, onClose).send();
- });
+ }).single();
}
/**
@@ -162,7 +173,18 @@ public class SimpleLuceneLocalSearcher implements LuceneLocalSearcher {
throw new IndexOutOfBoundsException(resultsOffset);
}
var currentPageLimit = queryParams.pageLimits().getPageLimit(s.pageIndex());
- if ((s.pageIndex() == 0 || s.last() != null) && s.remainingLimit() > 0) {
+ if (s.pageIndex() == 0 && s.remainingLimit() == 0) {
+ int count;
+ try {
+ count = indexSearchers.get(0).count(queryParams.query());
+ } catch (IOException e) {
+ sink.error(e);
+ return EMPTY_STATUS;
+ }
+ var nextPageInfo = new CurrentPageInfo(null, 0, 1);
+ sink.next(new PageData(new TopDocs(new TotalHits(count, Relation.EQUAL_TO), new ScoreDoc[0]), nextPageInfo));
+ return EMPTY_STATUS;
+ } else if (s.pageIndex() == 0 || (s.last() != null && s.remainingLimit() > 0)) {
TopDocs pageTopDocs;
try {
TopDocsCollector collector = TopDocsSearcher.getTopDocsCollector(queryParams.sort(),
diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/SimpleUnsortedUnscoredLuceneMultiSearcher.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/SimpleUnsortedUnscoredLuceneMultiSearcher.java
index 6b8bea3..e3e085d 100644
--- a/src/main/java/it/cavallium/dbengine/lucene/searcher/SimpleUnsortedUnscoredLuceneMultiSearcher.java
+++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/SimpleUnsortedUnscoredLuceneMultiSearcher.java
@@ -100,4 +100,9 @@ public class SimpleUnsortedUnscoredLuceneMultiSearcher implements LuceneMultiSea
queryParams.scoreMode()
);
}
+
+ @Override
+ public String getName() {
+ return "simpleunsortedunscoredmulti";
+ }
}
diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/TopDocsSearcher.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/TopDocsSearcher.java
index 372d574..60171d8 100644
--- a/src/main/java/it/cavallium/dbengine/lucene/searcher/TopDocsSearcher.java
+++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/TopDocsSearcher.java
@@ -2,28 +2,13 @@ package it.cavallium.dbengine.lucene.searcher;
import static it.cavallium.dbengine.lucene.searcher.PaginationInfo.ALLOW_UNSCORED_PAGINATION_MODE;
-import it.cavallium.dbengine.lucene.UnscoredCollector;
-import java.io.IOException;
-import org.apache.lucene.index.LeafReaderContext;
-import org.apache.lucene.index.NumericDocValues;
-import org.apache.lucene.misc.search.DiversifiedTopDocsCollector;
-import org.apache.lucene.search.BulkScorer;
-import org.apache.lucene.search.Collector;
+import it.cavallium.dbengine.lucene.collector.UnscoredCollector;
import org.apache.lucene.search.FieldDoc;
-import org.apache.lucene.search.HitQueue;
-import org.apache.lucene.search.IndexSearcher;
-import org.apache.lucene.search.LeafCollector;
-import org.apache.lucene.search.Query;
-import org.apache.lucene.search.Scorable;
import org.apache.lucene.search.ScoreDoc;
-import org.apache.lucene.search.ScoreMode;
import org.apache.lucene.search.Sort;
-import org.apache.lucene.search.TopDocs;
import org.apache.lucene.search.TopDocsCollector;
import org.apache.lucene.search.TopFieldCollector;
import org.apache.lucene.search.TopScoreDocCollector;
-import org.apache.lucene.search.TotalHits.Relation;
-import reactor.core.scheduler.Schedulers;
class TopDocsSearcher {
diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedScoredFullLuceneMultiSearcher.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedScoredFullLuceneMultiSearcher.java
new file mode 100644
index 0000000..8fb6770
--- /dev/null
+++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedScoredFullLuceneMultiSearcher.java
@@ -0,0 +1,120 @@
+package it.cavallium.dbengine.lucene.searcher;
+
+import io.net5.buffer.api.Send;
+import it.cavallium.dbengine.database.LLKeyScore;
+import it.cavallium.dbengine.database.LLUtils;
+import it.cavallium.dbengine.database.disk.LLIndexSearchers;
+import it.cavallium.dbengine.database.disk.LLTempLMDBEnv;
+import it.cavallium.dbengine.lucene.LuceneUtils;
+import it.cavallium.dbengine.lucene.FullDocs;
+import it.cavallium.dbengine.lucene.LLScoreDoc;
+import it.cavallium.dbengine.lucene.collector.LMDBFullScoreDocCollector;
+import it.cavallium.dbengine.lucene.searcher.LLSearchTransformer.TransformerInput;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Objects;
+import org.apache.lucene.search.IndexSearcher;
+import org.apache.lucene.search.Sort;
+import org.warp.commonutils.log.Logger;
+import org.warp.commonutils.log.LoggerFactory;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class UnsortedScoredFullLuceneMultiSearcher implements LuceneMultiSearcher, Closeable {
+
+ protected static final Logger logger = LoggerFactory.getLogger(UnsortedScoredFullLuceneMultiSearcher.class);
+
+ private final LLTempLMDBEnv env;
+
+ public UnsortedScoredFullLuceneMultiSearcher() throws IOException {
+ this.env = new LLTempLMDBEnv();
+ }
+
+ @Override
+ public Mono> collectMulti(Mono> indexSearchersMono,
+ LocalQueryParams queryParams,
+ String keyFieldName,
+ LLSearchTransformer transformer) {
+ Mono queryParamsMono;
+ if (transformer == LLSearchTransformer.NO_TRANSFORMATION) {
+ queryParamsMono = Mono.just(queryParams);
+ } else {
+ queryParamsMono = LLUtils.usingSendResource(indexSearchersMono, indexSearchers -> transformer.transform(Mono
+ .fromSupplier(() -> new TransformerInput(indexSearchers, queryParams))), true);
+ }
+
+ return queryParamsMono.flatMap(queryParams2 -> {
+ Objects.requireNonNull(queryParams2.scoreMode(), "ScoreMode must not be null");
+ if (queryParams2.sort() != null && queryParams2.sort() != Sort.RELEVANCE) {
+ throw new IllegalArgumentException(UnsortedScoredFullLuceneMultiSearcher.this.getClass().getSimpleName()
+ + " doesn't support sorted queries");
+ }
+
+ return LLUtils.usingSendResource(indexSearchersMono, indexSearchers -> this
+ // Search results
+ .search(indexSearchers.shards(), queryParams2)
+ // Compute the results
+ .transform(fullDocsMono -> this.computeResults(fullDocsMono, indexSearchers,
+ keyFieldName, queryParams2))
+ // Ensure that one LuceneSearchResult is always returned
+ .single(),
+ false);
+ });
+ }
+
+ /**
+ * Search effectively the raw results
+ */
+ private Mono> search(Iterable indexSearchers,
+ LocalQueryParams queryParams) {
+ return Mono
+ .fromCallable(() -> {
+ LLUtils.ensureBlocking();
+ var totalHitsThreshold = LuceneUtils.totalHitsThreshold();
+ return LMDBFullScoreDocCollector.createSharedManager(env, totalHitsThreshold);
+ })
+ .flatMap(sharedManager -> Flux
+ .fromIterable(indexSearchers)
+ .flatMap(shard -> Mono.fromCallable(() -> {
+ LLUtils.ensureBlocking();
+ var collector = sharedManager.newCollector();
+ shard.search(queryParams.query(), collector);
+ return collector;
+ }))
+ .collectList()
+ .flatMap(collectors -> Mono.fromCallable(() -> {
+ LLUtils.ensureBlocking();
+ return sharedManager.reduce(collectors);
+ }))
+ );
+ }
+
+ /**
+ * Compute the results, extracting useful data
+ */
+ private Mono> computeResults(Mono> dataMono,
+ LLIndexSearchers indexSearchers,
+ String keyFieldName,
+ LocalQueryParams queryParams) {
+ return dataMono.map(data -> {
+ var totalHitsCount = LuceneUtils.convertTotalHitsCount(data.totalHits());
+
+ Flux hitsFlux = LuceneUtils
+ .convertHits(data.iterate(queryParams.offset()).map(LLScoreDoc::toScoreDoc),
+ indexSearchers.shards(), keyFieldName, true)
+ .take(queryParams.limit(), true);
+
+ return new LuceneSearchResult(totalHitsCount, hitsFlux, indexSearchers::close).send();
+ });
+ }
+
+ @Override
+ public void close() throws IOException {
+ env.close();
+ }
+
+ @Override
+ public String getName() {
+ return "scoredfullmulti";
+ }
+}
diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedUnscoredContinuousLuceneMultiSearcher.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedUnscoredContinuousLuceneMultiSearcher.java
index 378b7ea..3be653b 100644
--- a/src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedUnscoredContinuousLuceneMultiSearcher.java
+++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedUnscoredContinuousLuceneMultiSearcher.java
@@ -114,4 +114,9 @@ public class UnsortedUnscoredContinuousLuceneMultiSearcher implements LuceneMult
queryParams.scoreMode()
);
}
+
+ @Override
+ public String getName() {
+ return "unsortedunscoredcontinuousmulti";
+ }
}
diff --git a/src/main/java/org/lmdbjava/Net5ByteBufProxy.java b/src/main/java/org/lmdbjava/Net5ByteBufProxy.java
new file mode 100644
index 0000000..7eea25d
--- /dev/null
+++ b/src/main/java/org/lmdbjava/Net5ByteBufProxy.java
@@ -0,0 +1,154 @@
+/*-
+ * #%L
+ * LmdbJava
+ * %%
+ * Copyright (C) 2016 - 2021 The LmdbJava Open Source Project
+ * %%
+ * Licensed 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.
+ * #L%
+ */
+
+package org.lmdbjava;
+
+import static io.net5.buffer.PooledByteBufAllocator.DEFAULT;
+import static java.lang.Class.forName;
+import static org.lmdbjava.UnsafeAccess.UNSAFE;
+
+import io.net5.buffer.ByteBuf;
+import java.lang.reflect.Field;
+
+import io.net5.buffer.ByteBuf;
+import io.net5.buffer.PooledByteBufAllocator;
+import jnr.ffi.Pointer;
+
+/**
+ * A buffer proxy backed by Netty's {@link ByteBuf}.
+ *
+ *
+ * This class requires {@link UnsafeAccess} and netty-buffer must be in the
+ * classpath.
+ */
+public final class Net5ByteBufProxy extends BufferProxy {
+
+ /**
+ * A proxy for using Netty {@link ByteBuf}. Guaranteed to never be null,
+ * although a class initialization exception will occur if an attempt is made
+ * to access this field when Netty is unavailable.
+ */
+ public static final BufferProxy PROXY_NETTY = new Net5ByteBufProxy();
+
+ private static final int BUFFER_RETRIES = 10;
+ private static final String FIELD_NAME_ADDRESS = "memoryAddress";
+ private static final String FIELD_NAME_LENGTH = "length";
+ private static final String NAME = "io.net5.buffer.PooledUnsafeDirectByteBuf";
+ private final long lengthOffset;
+ private final long addressOffset;
+
+ private final PooledByteBufAllocator nettyAllocator;
+
+ private Net5ByteBufProxy() {
+ this(DEFAULT);
+ }
+
+ public Net5ByteBufProxy(final PooledByteBufAllocator allocator) {
+ this.nettyAllocator = allocator;
+
+ try {
+ final ByteBuf initBuf = this.allocate();
+ initBuf.release();
+ final Field address = findField(NAME, FIELD_NAME_ADDRESS);
+ final Field length = findField(NAME, FIELD_NAME_LENGTH);
+ addressOffset = UNSAFE.objectFieldOffset(address);
+ lengthOffset = UNSAFE.objectFieldOffset(length);
+ } catch (final SecurityException e) {
+ throw new LmdbException("Field access error", e);
+ }
+ }
+
+ static Field findField(final String c, final String name) {
+ Class clazz;
+ try {
+ clazz = forName(c);
+ } catch (final ClassNotFoundException e) {
+ throw new LmdbException(c + " class unavailable", e);
+ }
+ do {
+ try {
+ final Field field = clazz.getDeclaredField(name);
+ field.setAccessible(true);
+ return field;
+ } catch (final NoSuchFieldException e) {
+ clazz = clazz.getSuperclass();
+ }
+ } while (clazz != null);
+ throw new LmdbException(name + " not found");
+ }
+
+ @Override
+ protected ByteBuf allocate() {
+ for (int i = 0; i < BUFFER_RETRIES; i++) {
+ final ByteBuf bb = nettyAllocator.directBuffer();
+ if (NAME.equals(bb.getClass().getName())) {
+ return bb;
+ } else {
+ bb.release();
+ }
+ }
+ throw new IllegalStateException("Netty buffer must be " + NAME);
+ }
+
+ @Override
+ protected int compare(final ByteBuf o1, final ByteBuf o2) {
+ return o1.compareTo(o2);
+ }
+
+ @Override
+ protected void deallocate(final ByteBuf buff) {
+ buff.release();
+ }
+
+ @Override
+ protected byte[] getBytes(final ByteBuf buffer) {
+ final byte[] dest = new byte[buffer.capacity()];
+ buffer.getBytes(0, dest);
+ return dest;
+ }
+
+ @Override
+ protected void in(final ByteBuf buffer, final Pointer ptr, final long ptrAddr) {
+ UNSAFE.putLong(ptrAddr + STRUCT_FIELD_OFFSET_SIZE,
+ buffer.writerIndex() - buffer.readerIndex());
+ UNSAFE.putLong(ptrAddr + STRUCT_FIELD_OFFSET_DATA,
+ buffer.memoryAddress() + buffer.readerIndex());
+ }
+
+ @Override
+ protected void in(final ByteBuf buffer, final int size, final Pointer ptr,
+ final long ptrAddr) {
+ UNSAFE.putLong(ptrAddr + STRUCT_FIELD_OFFSET_SIZE,
+ size);
+ UNSAFE.putLong(ptrAddr + STRUCT_FIELD_OFFSET_DATA,
+ buffer.memoryAddress() + buffer.readerIndex());
+ }
+
+ @Override
+ protected ByteBuf out(final ByteBuf buffer, final Pointer ptr,
+ final long ptrAddr) {
+ final long addr = UNSAFE.getLong(ptrAddr + STRUCT_FIELD_OFFSET_DATA);
+ final long size = UNSAFE.getLong(ptrAddr + STRUCT_FIELD_OFFSET_SIZE);
+ UNSAFE.putLong(buffer, addressOffset, addr);
+ UNSAFE.putInt(buffer, lengthOffset, (int) size);
+ buffer.writerIndex((int) size).readerIndex(0);
+ return buffer;
+ }
+}
diff --git a/src/test/java/it/cavallium/dbengine/DbTestUtils.java b/src/test/java/it/cavallium/dbengine/DbTestUtils.java
index 1af7174..dd7faad 100644
--- a/src/test/java/it/cavallium/dbengine/DbTestUtils.java
+++ b/src/test/java/it/cavallium/dbengine/DbTestUtils.java
@@ -9,9 +9,12 @@ import io.net5.buffer.api.pool.MetricUtils;
import io.net5.buffer.api.pool.PoolArenaMetric;
import io.net5.buffer.api.pool.PooledBufferAllocator;
import io.net5.util.internal.PlatformDependent;
+import it.cavallium.dbengine.client.LuceneIndex;
+import it.cavallium.dbengine.client.LuceneIndexImpl;
import it.cavallium.dbengine.database.LLDatabaseConnection;
import it.cavallium.dbengine.database.LLDictionary;
import it.cavallium.dbengine.database.LLKeyValueDatabase;
+import it.cavallium.dbengine.database.LLLuceneIndex;
import it.cavallium.dbengine.database.UpdateMode;
import it.cavallium.dbengine.database.collections.DatabaseMapDictionary;
import it.cavallium.dbengine.database.collections.DatabaseMapDictionaryDeep;
@@ -23,6 +26,7 @@ import it.cavallium.dbengine.database.collections.SubStageGetterMap;
import it.cavallium.dbengine.database.disk.MemorySegmentUtils;
import it.cavallium.dbengine.database.serialization.Serializer;
import it.cavallium.dbengine.database.serialization.SerializerFixedBinaryLength;
+import it.cavallium.dbengine.lucene.StringIndicizer;
import java.nio.file.Path;
import java.util.Map;
import java.util.Objects;
@@ -121,6 +125,9 @@ public class DbTestUtils {
}
public static record TempDb(TestAllocator allocator, LLDatabaseConnection connection, LLKeyValueDatabase db,
+ LLLuceneIndex luceneSingle,
+ LLLuceneIndex luceneMulti,
+ SwappableLuceneSearcher swappableLuceneSearcher,
Path path) {}
static boolean computeCanUseNettyDirect() {
@@ -166,6 +173,10 @@ public class DbTestUtils {
return database.getDictionary(name, updateMode);
}
+ public static Mono> tempLuceneIndex(LLLuceneIndex index) {
+ return Mono.fromCallable(() -> new LuceneIndexImpl<>(index, new StringIndicizer()));
+ }
+
public enum MapType {
MAP,
diff --git a/src/test/java/it/cavallium/dbengine/ExpectedQueryType.java b/src/test/java/it/cavallium/dbengine/ExpectedQueryType.java
new file mode 100644
index 0000000..a468571
--- /dev/null
+++ b/src/test/java/it/cavallium/dbengine/ExpectedQueryType.java
@@ -0,0 +1,3 @@
+package it.cavallium.dbengine;
+
+record ExpectedQueryType(boolean shard, boolean sorted, boolean scored, boolean unlimited, boolean onlyCount) {}
diff --git a/src/test/java/it/cavallium/dbengine/LocalTemporaryDbGenerator.java b/src/test/java/it/cavallium/dbengine/LocalTemporaryDbGenerator.java
index 8204a0a..3fe96aa 100644
--- a/src/test/java/it/cavallium/dbengine/LocalTemporaryDbGenerator.java
+++ b/src/test/java/it/cavallium/dbengine/LocalTemporaryDbGenerator.java
@@ -5,15 +5,26 @@ import static it.cavallium.dbengine.DbTestUtils.ensureNoLeaks;
import it.cavallium.dbengine.DbTestUtils.TempDb;
import it.cavallium.dbengine.DbTestUtils.TestAllocator;
import it.cavallium.dbengine.client.DatabaseOptions;
+import it.cavallium.dbengine.client.IndicizerAnalyzers;
+import it.cavallium.dbengine.client.IndicizerSimilarities;
+import it.cavallium.dbengine.client.LuceneOptions;
+import it.cavallium.dbengine.client.NRTCachingOptions;
import it.cavallium.dbengine.database.Column;
import it.cavallium.dbengine.database.LLKeyValueDatabase;
import it.cavallium.dbengine.database.disk.LLLocalDatabaseConnection;
+import it.cavallium.dbengine.database.lucene.LuceneHacks;
+import it.cavallium.dbengine.lucene.analyzer.TextFieldsAnalyzer;
+import it.cavallium.dbengine.lucene.analyzer.TextFieldsSimilarity;
+import it.cavallium.dbengine.lucene.searcher.AdaptiveLuceneLocalSearcher;
+import it.cavallium.dbengine.lucene.searcher.AdaptiveLuceneMultiSearcher;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.time.Duration;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.CompletionException;
import java.util.concurrent.atomic.AtomicInteger;
import reactor.core.publisher.Mono;
@@ -23,6 +34,10 @@ public class LocalTemporaryDbGenerator implements TemporaryDbGenerator {
private static final AtomicInteger dbId = new AtomicInteger(0);
+ private static final Optional NRT = Optional.empty();
+ private static final LuceneOptions LUCENE_OPTS = new LuceneOptions(Map.of(), Duration.ofSeconds(5), Duration.ofSeconds(5),
+ false, true, Optional.empty(), true, NRT, -1, true, true);
+
@Override
public Mono openTempDb(TestAllocator allocator) {
boolean canUseNettyDirect = DbTestUtils.computeCanUseNettyDirect();
@@ -44,13 +59,33 @@ public class LocalTemporaryDbGenerator implements TemporaryDbGenerator {
})
.subscribeOn(Schedulers.boundedElastic())
.then(new LLLocalDatabaseConnection(allocator.allocator(), wrkspcPath).connect())
- .flatMap(conn -> conn
- .getDatabase("testdb",
- List.of(Column.dictionary("testmap"), Column.special("ints"), Column.special("longs")),
- new DatabaseOptions(Map.of(), true, false, true, false, true, canUseNettyDirect, canUseNettyDirect, -1)
- )
- .map(db -> new TempDb(allocator, conn, db, wrkspcPath))
- );
+ .flatMap(conn -> {
+ SwappableLuceneSearcher searcher = new SwappableLuceneSearcher();
+ var luceneHacks = new LuceneHacks(() -> searcher, () -> searcher);
+ return Mono.zip(
+ conn.getDatabase("testdb",
+ List.of(Column.dictionary("testmap"), Column.special("ints"), Column.special("longs")),
+ new DatabaseOptions(Map.of(), true, false, true, false,
+ true, canUseNettyDirect, canUseNettyDirect, -1)
+ ),
+ conn.getLuceneIndex("testluceneindex1",
+ 1,
+ IndicizerAnalyzers.of(TextFieldsAnalyzer.WordSimple),
+ IndicizerSimilarities.of(TextFieldsSimilarity.Boolean),
+ LUCENE_OPTS,
+ luceneHacks
+ ),
+ conn.getLuceneIndex("testluceneindex16",
+ 1,
+ IndicizerAnalyzers.of(TextFieldsAnalyzer.WordSimple),
+ IndicizerSimilarities.of(TextFieldsSimilarity.Boolean),
+ LUCENE_OPTS,
+ luceneHacks
+ ),
+ Mono.just(searcher)
+ )
+ .map(tuple -> new TempDb(allocator, conn, tuple.getT1(), tuple.getT2(), tuple.getT3(), tuple.getT4(), wrkspcPath));
+ });
});
}
diff --git a/src/test/java/it/cavallium/dbengine/MemoryTemporaryDbGenerator.java b/src/test/java/it/cavallium/dbengine/MemoryTemporaryDbGenerator.java
index 3001728..86b81ab 100644
--- a/src/test/java/it/cavallium/dbengine/MemoryTemporaryDbGenerator.java
+++ b/src/test/java/it/cavallium/dbengine/MemoryTemporaryDbGenerator.java
@@ -3,25 +3,59 @@ package it.cavallium.dbengine;
import it.cavallium.dbengine.DbTestUtils.TempDb;
import it.cavallium.dbengine.DbTestUtils.TestAllocator;
import it.cavallium.dbengine.client.DatabaseOptions;
+import it.cavallium.dbengine.client.IndicizerAnalyzers;
+import it.cavallium.dbengine.client.IndicizerSimilarities;
+import it.cavallium.dbengine.client.LuceneOptions;
+import it.cavallium.dbengine.client.NRTCachingOptions;
import it.cavallium.dbengine.database.Column;
+import it.cavallium.dbengine.database.lucene.LuceneHacks;
import it.cavallium.dbengine.database.memory.LLMemoryDatabaseConnection;
+import it.cavallium.dbengine.lucene.analyzer.TextFieldsAnalyzer;
+import it.cavallium.dbengine.lucene.analyzer.TextFieldsSimilarity;
+import java.time.Duration;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import reactor.core.publisher.Mono;
public class MemoryTemporaryDbGenerator implements TemporaryDbGenerator {
+ private static final Optional NRT = Optional.empty();
+ private static final LuceneOptions LUCENE_OPTS = new LuceneOptions(Map.of(), Duration.ofSeconds(5), Duration.ofSeconds(5),
+ false, true, Optional.empty(), true, NRT, -1, true, true);
+
@Override
public Mono openTempDb(TestAllocator allocator) {
boolean canUseNettyDirect = DbTestUtils.computeCanUseNettyDirect();
return Mono
.fromCallable(() -> new LLMemoryDatabaseConnection(allocator.allocator()))
- .flatMap(conn -> conn
- .getDatabase("testdb",
- List.of(Column.dictionary("testmap"), Column.special("ints"), Column.special("longs")),
- new DatabaseOptions(Map.of(), true, false, true, false, true, canUseNettyDirect, canUseNettyDirect, -1)
- )
- .map(db -> new TempDb(allocator, conn, db, null)));
+ .flatMap(conn -> {
+ SwappableLuceneSearcher searcher = new SwappableLuceneSearcher();
+ var luceneHacks = new LuceneHacks(() -> searcher, () -> searcher);
+ return Mono
+ .zip(
+ conn.getDatabase("testdb",
+ List.of(Column.dictionary("testmap"), Column.special("ints"), Column.special("longs")),
+ new DatabaseOptions(Map.of(), true, false, true, false, true, canUseNettyDirect, canUseNettyDirect, -1)
+ ),
+ conn.getLuceneIndex("testluceneindex1",
+ 1,
+ IndicizerAnalyzers.of(TextFieldsAnalyzer.WordSimple),
+ IndicizerSimilarities.of(TextFieldsSimilarity.Boolean),
+ LUCENE_OPTS,
+ luceneHacks
+ ),
+ conn.getLuceneIndex("testluceneindex16",
+ 1,
+ IndicizerAnalyzers.of(TextFieldsAnalyzer.WordSimple),
+ IndicizerSimilarities.of(TextFieldsSimilarity.Boolean),
+ LUCENE_OPTS,
+ luceneHacks
+ ),
+ Mono.just(searcher)
+ )
+ .map(tuple -> new TempDb(allocator, conn, tuple.getT1(), tuple.getT2(), tuple.getT3(), tuple.getT4(), null));
+ });
}
@Override
diff --git a/src/test/java/it/cavallium/dbengine/Scored.java b/src/test/java/it/cavallium/dbengine/Scored.java
new file mode 100644
index 0000000..745a041
--- /dev/null
+++ b/src/test/java/it/cavallium/dbengine/Scored.java
@@ -0,0 +1,3 @@
+package it.cavallium.dbengine;
+
+record Scored(String key, float score) {}
diff --git a/src/test/java/it/cavallium/dbengine/TestLuceneIndex.java b/src/test/java/it/cavallium/dbengine/TestLuceneIndex.java
new file mode 100644
index 0000000..c3f93ca
--- /dev/null
+++ b/src/test/java/it/cavallium/dbengine/TestLuceneIndex.java
@@ -0,0 +1,385 @@
+package it.cavallium.dbengine;
+
+import static it.cavallium.dbengine.DbTestUtils.destroyAllocator;
+import static it.cavallium.dbengine.DbTestUtils.ensureNoLeaks;
+import static it.cavallium.dbengine.DbTestUtils.newAllocator;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import it.cavallium.dbengine.DbTestUtils.TempDb;
+import it.cavallium.dbengine.DbTestUtils.TestAllocator;
+import it.cavallium.dbengine.client.LuceneIndex;
+import it.cavallium.dbengine.client.MultiSort;
+import it.cavallium.dbengine.client.SearchResultKey;
+import it.cavallium.dbengine.client.SearchResultKeys;
+import it.cavallium.dbengine.client.query.ClientQueryParams;
+import it.cavallium.dbengine.client.query.ClientQueryParamsBuilder;
+import it.cavallium.dbengine.client.query.QueryParser;
+import it.cavallium.dbengine.client.query.current.data.MatchAllDocsQuery;
+import it.cavallium.dbengine.client.query.current.data.MatchNoDocsQuery;
+import it.cavallium.dbengine.client.query.current.data.NoSort;
+import it.cavallium.dbengine.client.query.current.data.TotalHitsCount;
+import it.cavallium.dbengine.database.LLLuceneIndex;
+import it.cavallium.dbengine.database.LLScoreMode;
+import it.cavallium.dbengine.database.LLUtils;
+import it.cavallium.dbengine.lucene.searcher.AdaptiveLuceneLocalSearcher;
+import it.cavallium.dbengine.lucene.searcher.AdaptiveLuceneMultiSearcher;
+import it.cavallium.dbengine.lucene.searcher.CountLuceneLocalSearcher;
+import it.cavallium.dbengine.lucene.searcher.LuceneLocalSearcher;
+import it.cavallium.dbengine.lucene.searcher.LuceneMultiSearcher;
+import it.cavallium.dbengine.lucene.searcher.UnsortedScoredFullLuceneMultiSearcher;
+import it.cavallium.dbengine.lucene.searcher.ScoredSimpleLuceneMultiSearcher;
+import it.cavallium.dbengine.lucene.searcher.SimpleLuceneLocalSearcher;
+import it.cavallium.dbengine.lucene.searcher.SimpleUnsortedUnscoredLuceneMultiSearcher;
+import it.cavallium.dbengine.lucene.searcher.UnsortedUnscoredContinuousLuceneMultiSearcher;
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Stream;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.warp.commonutils.log.Logger;
+import org.warp.commonutils.log.LoggerFactory;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink.OverflowStrategy;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+import reactor.util.function.Tuples;
+
+public class TestLuceneIndex {
+
+ private final Logger log = LoggerFactory.getLogger(this.getClass());
+
+ private TestAllocator allocator;
+ private TempDb tempDb;
+ private LLLuceneIndex luceneSingle;
+ private LLLuceneIndex luceneMulti;
+
+ protected TemporaryDbGenerator getTempDbGenerator() {
+ return new MemoryTemporaryDbGenerator();
+ }
+
+ @BeforeEach
+ public void beforeEach() {
+ this.allocator = newAllocator();
+ ensureNoLeaks(allocator.allocator(), false, false);
+ tempDb = Objects.requireNonNull(getTempDbGenerator().openTempDb(allocator).block(), "TempDB");
+ luceneSingle = tempDb.luceneSingle();
+ luceneMulti = tempDb.luceneMulti();
+ }
+
+ public static Stream provideArguments() {
+ return Stream.of(false, true).map(Arguments::of);
+ }
+
+ private static final Flux multi = Flux.just(false, true);
+ private static final Flux scoreModes = Flux.just(LLScoreMode.NO_SCORES,
+ LLScoreMode.TOP_SCORES,
+ LLScoreMode.COMPLETE_NO_SCORES,
+ LLScoreMode.COMPLETE
+ );
+ private static final Flux>> multiSort = Flux.just(MultiSort.topScore(),
+ MultiSort.randomSortField(),
+ MultiSort.noSort(),
+ MultiSort.docSort(),
+ MultiSort.numericSort("longsort", false),
+ MultiSort.numericSort("longsort", true)
+ );
+
+ private static Flux getSearchers(ExpectedQueryType info) {
+ return Flux.push(sink -> {
+ try {
+ if (info.shard()) {
+ sink.next(new AdaptiveLuceneMultiSearcher());
+ if (info.onlyCount()) {
+ sink.next(new SimpleUnsortedUnscoredLuceneMultiSearcher(new CountLuceneLocalSearcher()));
+ } else {
+ sink.next(new ScoredSimpleLuceneMultiSearcher());
+ if (!info.sorted()) {
+ sink.next(new UnsortedScoredFullLuceneMultiSearcher());
+ }
+ if (!info.scored() && !info.sorted()) {
+ sink.next(new SimpleUnsortedUnscoredLuceneMultiSearcher(new SimpleLuceneLocalSearcher()));
+ sink.next(new UnsortedUnscoredContinuousLuceneMultiSearcher());
+ }
+ }
+ } else {
+ sink.next(new AdaptiveLuceneLocalSearcher());
+ if (info.onlyCount()) {
+ sink.next(new CountLuceneLocalSearcher());
+ } else {
+ sink.next(new SimpleLuceneLocalSearcher());
+ }
+ }
+ sink.complete();
+ } catch (IOException e) {
+ sink.error(e);
+ }
+ }, OverflowStrategy.BUFFER);
+ }
+
+ public static Stream provideQueryArgumentsScoreMode() {
+ return multi
+ .concatMap(shard -> scoreModes.map(scoreMode -> Tuples.of(shard, scoreMode)))
+ .map(tuple -> Arguments.of(tuple.toArray()))
+ .toStream();
+ }
+
+ public static Stream provideQueryArgumentsSort() {
+ return multi
+ .concatMap(shard -> multiSort.map(multiSort -> Tuples.of(shard, multiSort)))
+ .map(tuple -> Arguments.of(tuple.toArray()))
+ .toStream();
+ }
+
+ public static Stream provideQueryArgumentsScoreModeAndSort() {
+ return multi
+ .concatMap(shard -> scoreModes.map(scoreMode -> Tuples.of(shard, scoreMode)))
+ .concatMap(tuple -> multiSort.map(multiSort -> Tuples.of(tuple.getT1(), tuple.getT2(), multiSort)))
+ .map(tuple -> Arguments.of(tuple.toArray()))
+ .toStream();
+ }
+
+ @AfterEach
+ public void afterEach() {
+ getTempDbGenerator().closeTempDb(tempDb).block();
+ ensureNoLeaks(allocator.allocator(), true, false);
+ destroyAllocator(allocator);
+ }
+
+ private LuceneIndex getLuceneIndex(boolean shards, @Nullable LuceneLocalSearcher customSearcher) {
+ LuceneIndex index = run(DbTestUtils.tempLuceneIndex(shards ? luceneSingle : luceneMulti));
+ index.updateDocument("test-key-1", "0123456789").block();
+ index.updateDocument("test-key-2", "test 0123456789 test word").block();
+ index.updateDocument("test-key-3", "0123456789 test example string").block();
+ index.updateDocument("test-key-4", "hello world the quick brown fox jumps over the lazy dog").block();
+ index.updateDocument("test-key-5", "hello the quick brown fox jumps over the lazy dog").block();
+ index.updateDocument("test-key-6", "hello the quick brown fox jumps over the world dog").block();
+ index.updateDocument("test-key-7", "the quick brown fox jumps over the world dog").block();
+ index.updateDocument("test-key-8", "the quick brown fox jumps over the lazy dog").block();
+ index.updateDocument("test-key-9", "Example1").block();
+ index.updateDocument("test-key-10", "Example2").block();
+ index.updateDocument("test-key-11", "Example3").block();
+ index.updateDocument("test-key-12", "-234").block();
+ index.updateDocument("test-key-13", "2111").block();
+ index.updateDocument("test-key-14", "2999").block();
+ index.updateDocument("test-key-15", "3902").block();
+ Flux.range(1, 1000).concatMap(i -> index.updateDocument("test-key-" + (15 + i), "" + i)).blockLast();
+ tempDb.swappableLuceneSearcher().setSingle(new CountLuceneLocalSearcher());
+ tempDb.swappableLuceneSearcher().setMulti(new SimpleUnsortedUnscoredLuceneMultiSearcher(new CountLuceneLocalSearcher()));
+ assertCount(index, 1000 + 15);
+ try {
+ if (customSearcher != null) {
+ tempDb.swappableLuceneSearcher().setSingle(customSearcher);
+ if (shards) {
+ if (customSearcher instanceof LuceneMultiSearcher multiSearcher) {
+ tempDb.swappableLuceneSearcher().setMulti(multiSearcher);
+ } else {
+ throw new IllegalArgumentException("Expected a LuceneMultiSearcher, got a LuceneLocalSearcher: " + customSearcher.getName());
+ }
+ }
+ } else {
+ tempDb.swappableLuceneSearcher().setSingle(new AdaptiveLuceneLocalSearcher());
+ tempDb.swappableLuceneSearcher().setMulti(new AdaptiveLuceneMultiSearcher());
+ }
+ } catch (IOException e) {
+ fail(e);
+ }
+ return index;
+ }
+
+ private void run(Flux publisher) {
+ publisher.subscribeOn(Schedulers.immediate()).blockLast();
+ }
+
+ private void runVoid(Mono publisher) {
+ publisher.then().subscribeOn(Schedulers.immediate()).block();
+ }
+
+ private T run(Mono publisher) {
+ return publisher.subscribeOn(Schedulers.immediate()).block();
+ }
+
+ private T run(boolean shouldFail, Mono publisher) {
+ return publisher.subscribeOn(Schedulers.immediate()).transform(mono -> {
+ if (shouldFail) {
+ return mono.onErrorResume(ex -> Mono.empty());
+ } else {
+ return mono;
+ }
+ }).block();
+ }
+
+ private void runVoid(boolean shouldFail, Mono publisher) {
+ publisher.then().subscribeOn(Schedulers.immediate()).transform(mono -> {
+ if (shouldFail) {
+ return mono.onErrorResume(ex -> Mono.empty());
+ } else {
+ return mono;
+ }
+ }).block();
+ }
+
+ private void assertCount(LuceneIndex luceneIndex, long expected) {
+ Assertions.assertEquals(expected, getCount(luceneIndex));
+ }
+
+ private long getCount(LuceneIndex luceneIndex) {
+ luceneIndex.refresh(true).block();
+ var totalHitsCount = run(luceneIndex.count(null, new MatchAllDocsQuery()));
+ Assertions.assertTrue(totalHitsCount.exact(), "Can't get count because the total hits count is not exact");
+ return totalHitsCount.value();
+ }
+
+ @Test
+ public void testNoOp() {
+ }
+
+ @Test
+ public void testNoOpAllocation() {
+ for (int i = 0; i < 10; i++) {
+ var a = allocator.allocator().allocate(i * 512);
+ a.send().receive().close();
+ }
+ }
+
+ @ParameterizedTest
+ @MethodSource("provideArguments")
+ public void testGetLuceneIndex(boolean shards) {
+ var luceneIndex = getLuceneIndex(shards, null);
+ Assertions.assertNotNull(luceneIndex);
+ }
+
+ @ParameterizedTest
+ @MethodSource("provideArguments")
+ public void testDeleteAll(boolean shards) {
+ var luceneIndex = getLuceneIndex(shards, null);
+ runVoid(luceneIndex.deleteAll());
+ assertCount(luceneIndex, 0);
+ }
+
+ @ParameterizedTest
+ @MethodSource("provideArguments")
+ public void testDelete(boolean shards) {
+ var luceneIndex = getLuceneIndex(shards, null);
+ var prevCount = getCount(luceneIndex);
+ runVoid(luceneIndex.deleteDocument("test-key-1"));
+ assertCount(luceneIndex, prevCount - 1);
+ }
+
+ @ParameterizedTest
+ @MethodSource("provideArguments")
+ public void testUpdateSameDoc(boolean shards) {
+ var luceneIndex = getLuceneIndex(shards, null);
+ var prevCount = getCount(luceneIndex);
+ runVoid(luceneIndex.updateDocument("test-key-1", "new-value"));
+ assertCount(luceneIndex, prevCount );
+ }
+
+ @ParameterizedTest
+ @MethodSource("provideArguments")
+ public void testUpdateNewDoc(boolean shards) {
+ var luceneIndex = getLuceneIndex(shards, null);
+ var prevCount = getCount(luceneIndex);
+ runVoid(luceneIndex.updateDocument("test-key-new", "new-value"));
+ assertCount(luceneIndex, prevCount + 1);
+ }
+
+ @ParameterizedTest
+ @MethodSource("provideQueryArgumentsScoreModeAndSort")
+ public void testSearchNoDocs(boolean shards, LLScoreMode scoreMode, MultiSort> multiSort) {
+ var searchers = run(getSearchers(new ExpectedQueryType(shards, isSorted(multiSort), isScored(scoreMode, multiSort), true, false)).collectList());
+ for (LuceneLocalSearcher searcher : searchers) {
+ log.info("Using searcher \"{}\"", searcher.getName());
+
+ var luceneIndex = getLuceneIndex(shards, searcher);
+ ClientQueryParamsBuilder> queryBuilder = ClientQueryParams.builder();
+ queryBuilder.query(new MatchNoDocsQuery());
+ queryBuilder.snapshot(null);
+ queryBuilder.scoreMode(scoreMode);
+ queryBuilder.sort(multiSort);
+ var query = queryBuilder.build();
+ try (var results = run(luceneIndex.search(query)).receive()) {
+ var hits = results.totalHitsCount();
+ if (supportsPreciseHitsCount(searcher, query)) {
+ assertEquals(new TotalHitsCount(0, true), hits);
+ }
+
+ var keys = getResults(results);
+ assertEquals(List.of(), keys);
+ }
+ }
+ }
+
+ private boolean supportsPreciseHitsCount(LuceneLocalSearcher searcher,
+ ClientQueryParams> query) {
+ if (searcher instanceof UnsortedUnscoredContinuousLuceneMultiSearcher) {
+ return false;
+ }
+ var scored = isScored(query.scoreMode(), Objects.requireNonNullElse(query.sort(), MultiSort.noSort()));
+ var sorted = isSorted(Objects.requireNonNullElse(query.sort(), MultiSort.noSort()));
+ if (!sorted && !scored) {
+ if (searcher instanceof AdaptiveLuceneMultiSearcher || searcher instanceof AdaptiveLuceneLocalSearcher) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ @ParameterizedTest
+ @MethodSource("provideQueryArgumentsScoreModeAndSort")
+ public void testSearchAllDocs(boolean shards, LLScoreMode scoreMode, MultiSort> multiSort) {
+ var searchers = run(getSearchers(new ExpectedQueryType(shards, isSorted(multiSort), isScored(scoreMode, multiSort), true, false)).collectList());
+ for (LuceneLocalSearcher searcher : searchers) {
+ log.info("Using searcher \"{}\"", searcher.getName());
+
+ var luceneIndex = getLuceneIndex(shards, searcher);
+ ClientQueryParamsBuilder> queryBuilder = ClientQueryParams.builder();
+ queryBuilder.query(new MatchNoDocsQuery());
+ queryBuilder.snapshot(null);
+ queryBuilder.scoreMode(scoreMode);
+ queryBuilder.sort(multiSort);
+ var query = queryBuilder.build();
+ try (var results = run(luceneIndex.search(query)).receive()) {
+ var hits = results.totalHitsCount();
+ if (supportsPreciseHitsCount(searcher, query)) {
+ assertEquals(new TotalHitsCount(0, true), hits);
+ }
+
+ var keys = getResults(results);
+ assertEquals(List.of(), keys);
+ }
+ }
+ }
+
+ private boolean isSorted(MultiSort> multiSort) {
+ return !(multiSort.getQuerySort() instanceof NoSort);
+ }
+
+ private boolean isScored(LLScoreMode scoreMode, MultiSort> multiSort) {
+ var needsScores = LLUtils.toScoreMode(scoreMode).needsScores();
+ var sort =QueryParser.toSort(multiSort.getQuerySort());
+ if (sort != null) {
+ needsScores |= sort.needsScores();
+ }
+ return needsScores;
+ }
+
+ private List getResults(SearchResultKeys results) {
+ return run(results
+ .results()
+ .flatMapSequential(searchResultKey -> searchResultKey
+ .key()
+ .single()
+ .map(key -> new Scored(key, searchResultKey.score()))
+ )
+ .collectList());
+ }
+
+}