package it.cavallium.dbengine.client; import it.cavallium.dbengine.database.LLLuceneIndex; import it.cavallium.dbengine.database.LLScoreMode; import it.cavallium.dbengine.database.LLSearchResult; import it.cavallium.dbengine.database.LLSnapshot; import it.cavallium.dbengine.database.LLSnapshottable; import it.cavallium.dbengine.database.LLSort; import it.cavallium.dbengine.database.LLTerm; import it.cavallium.dbengine.database.LuceneUtils; import it.cavallium.dbengine.database.collections.Joiner.ValueGetter; import it.cavallium.dbengine.lucene.serializer.Query; import java.util.Map; import java.util.Map.Entry; import java.util.Set; import org.jetbrains.annotations.Nullable; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.util.function.Tuple2; public class LuceneIndex implements LLSnapshottable { private final LLLuceneIndex luceneIndex; private final Indicizer indicizer; public LuceneIndex(LLLuceneIndex luceneIndex, Indicizer indicizer) { this.luceneIndex = luceneIndex; this.indicizer = indicizer; } private LLSnapshot resolveSnapshot(CompositeSnapshot snapshot) { if (snapshot == null) { return null; } else { return snapshot.getSnapshot(luceneIndex); } } public Mono addDocument(T key, U value) { return indicizer .toDocument(key, value) .flatMap(doc -> luceneIndex.addDocument(indicizer.toIndex(key), doc)); } public Mono addDocuments(Flux> entries) { return luceneIndex .addDocuments(entries .flatMap(entry -> indicizer .toDocument(entry.getKey(), entry.getValue()) .map(doc -> Map.entry(indicizer.toIndex(entry.getKey()), doc))) .groupBy(Entry::getKey, Entry::getValue) ); } public Mono deleteDocument(T key) { LLTerm id = indicizer.toIndex(key); return luceneIndex.deleteDocument(id); } public Mono updateDocument(T key, U value) { return indicizer .toDocument(key, value) .flatMap(doc -> luceneIndex.updateDocument(indicizer.toIndex(key), doc)); } public Mono updateDocuments(Flux> entries) { return luceneIndex .updateDocuments(entries .flatMap(entry -> indicizer .toDocument(entry.getKey(), entry.getValue()) .map(doc -> Map.entry(indicizer.toIndex(entry.getKey()), doc))) .groupBy(Entry::getKey, Entry::getValue) ); } public Mono deleteAll() { return luceneIndex.deleteAll(); } private SearchResultKeys transformLuceneResult(LLSearchResult llSearchResult, @Nullable MultiSort> sort, @Nullable Integer limit) { var mappedKeys = llSearchResult .results() .map(flux -> flux.map(item -> new SearchResultKey<>(indicizer.getKey(item.getKey()), item.getScore()))); var sortedKeys = LuceneUtils.mergeStream(mappedKeys, sort, limit); return new SearchResultKeys<>(llSearchResult.totalHitsCount(), sortedKeys); } private SearchResult transformLuceneResultWithValues(LLSearchResult llSearchResult, @Nullable MultiSort> sort, @Nullable Integer limit, ValueGetter valueGetter) { var mappedKeys = llSearchResult .results() .map(flux -> flux.flatMap(item -> { var key = indicizer.getKey(item.getKey()); return valueGetter.get(key).map(value -> new SearchResultItem<>(key, value, item.getScore())); })); var sortedKeys = LuceneUtils.mergeStream(mappedKeys, sort, limit); return new SearchResult<>(llSearchResult.totalHitsCount(), sortedKeys); } /** * * @param limit the limit is valid for each lucene instance. * If you have 15 instances, the number of elements returned * can be at most limit * 15 * @return the collection has one or more flux */ public Mono> moreLikeThis(@Nullable CompositeSnapshot snapshot, U mltDocumentValue, int limit) { Flux>> mltDocumentFields = indicizer.getMoreLikeThisDocumentFields(mltDocumentValue); return luceneIndex .moreLikeThis(resolveSnapshot(snapshot), mltDocumentFields, limit, indicizer.getKeyFieldName()) .map(llSearchResult -> this.transformLuceneResult(llSearchResult, MultiSort.topScore(), limit)); } /** * * @param limit the limit is valid for each lucene instance. * If you have 15 instances, the number of elements returned * can be at most limit * 15 * @return the collection has one or more flux */ public Mono> moreLikeThisWithValues(@Nullable CompositeSnapshot snapshot, U mltDocumentValue, int limit, ValueGetter valueGetter) { Flux>> mltDocumentFields = indicizer.getMoreLikeThisDocumentFields(mltDocumentValue); return luceneIndex .moreLikeThis(resolveSnapshot(snapshot), mltDocumentFields, limit, indicizer.getKeyFieldName()) .map(llSearchResult -> this.transformLuceneResultWithValues(llSearchResult, MultiSort.topScoreWithValues(), limit, valueGetter)); } /** * * @param limit the limit is valid for each lucene instance. * If you have 15 instances, the number of elements returned * can be at most limit * 15 * @return the collection has one or more flux */ public Mono> search(@Nullable CompositeSnapshot snapshot, Query query, int limit, @Nullable MultiSort> sort, LLScoreMode scoreMode) { LLSort querySort = sort != null ? sort.getQuerySort() : null; return luceneIndex .search(resolveSnapshot(snapshot), query, limit, querySort, scoreMode, indicizer.getKeyFieldName()) .map(llSearchResult -> this.transformLuceneResult(llSearchResult, sort, limit)); } /** * * @param limit the limit is valid for each lucene instance. * If you have 15 instances, the number of elements returned * can be at most limit * 15 * @return the collection has one or more flux */ public Mono> searchWithValues(@Nullable CompositeSnapshot snapshot, Query query, int limit, @Nullable MultiSort> sort, LLScoreMode scoreMode, ValueGetter valueGetter) { LLSort querySort = sort != null ? sort.getQuerySort() : null; return luceneIndex .search(resolveSnapshot(snapshot), query, limit, querySort, scoreMode, indicizer.getKeyFieldName()) .map(llSearchResult -> this.transformLuceneResultWithValues(llSearchResult, sort, limit, valueGetter)); } public Mono count(@Nullable CompositeSnapshot snapshot, Query query) { return this.search(snapshot, query, 0, null, null) .flatMap(SearchResultKeys::totalHitsCount) .single(); } public boolean isLowMemoryMode() { return luceneIndex.isLowMemoryMode(); } public Mono close() { return luceneIndex.close(); } /** * Flush writes to disk */ public Mono flush() { return luceneIndex.flush(); } /** * Refresh index searcher */ public Mono refresh() { return luceneIndex.refresh(); } @Override public Mono takeSnapshot() { return luceneIndex.takeSnapshot(); } @Override public Mono releaseSnapshot(LLSnapshot snapshot) { return luceneIndex.releaseSnapshot(snapshot); } }