diff --git a/.github/workflows/maven-publish.yml b/.github/workflows/maven-publish.yml index 539765c..0301a3a 100644 --- a/.github/workflows/maven-publish.yml +++ b/.github/workflows/maven-publish.yml @@ -43,7 +43,7 @@ jobs: shell: bash run: | cd lucene - ./gradlew assemble -x test -x errorprone publishToMavenLocal + ./gradlew assemble -x test -x errorprone publishToMavenLocal -x signJarsPublication mvn deploy:deploy-file -DgroupId=org.lucene -DartifactId=lucene-core -Dversion=9.0.0-SNAPSHOT -DgeneratePom=true -Dpackaging=jar -DrepositoryId=mchv-release-distribution -Durl=https://mvn.mchv.eu/repository/mchv-snapshot -Dfile=lucene/core/build/libs/lucene-core-9.0.0-SNAPSHOT.jar -Djavadoc=lucene/core/build/libs/lucene-core-9.0.0-SNAPSHOT-javadoc.jar -Dsources=lucene/core/build/libs/lucene-core-9.0.0-SNAPSHOT-sources.jar mvn deploy:deploy-file -DgroupId=org.lucene -DartifactId=lucene-join -Dversion=9.0.0-SNAPSHOT -DgeneratePom=true -Dpackaging=jar -DrepositoryId=mchv-release-distribution -Durl=https://mvn.mchv.eu/repository/mchv-snapshot -Dfile=lucene/join/build/libs/lucene-join-9.0.0-SNAPSHOT.jar -Djavadoc=lucene/join/build/libs/lucene-join-9.0.0-SNAPSHOT-javadoc.jar -Dsources=lucene/join/build/libs/lucene-join-9.0.0-SNAPSHOT-sources.jar mvn deploy:deploy-file -DgroupId=org.lucene -DartifactId=lucene-analysis-common -Dversion=9.0.0-SNAPSHOT -DgeneratePom=true -Dpackaging=jar -DrepositoryId=mchv-release-distribution -Durl=https://mvn.mchv.eu/repository/mchv-snapshot -Dfile=lucene/analysis/common/build/libs/lucene-analysis-common-9.0.0-SNAPSHOT.jar -Djavadoc=lucene/analysis/common/build/libs/lucene-analysis-common-9.0.0-SNAPSHOT-javadoc.jar -Dsources=lucene/analysis/common/build/libs/lucene-analysis-common-9.0.0-SNAPSHOT-sources.jar diff --git a/pom.xml b/pom.xml index 6169e63..d2b7b83 100644 --- a/pom.xml +++ b/pom.xml @@ -522,7 +522,7 @@ it.cavallium data-generator - 0.9.69 + 0.9.71 generate-lucene-query-sources diff --git a/src/main/java/it/cavallium/dbengine/client/HitEntry.java b/src/main/java/it/cavallium/dbengine/client/HitEntry.java new file mode 100644 index 0000000..701de4a --- /dev/null +++ b/src/main/java/it/cavallium/dbengine/client/HitEntry.java @@ -0,0 +1,13 @@ +package it.cavallium.dbengine.client; + +import org.jetbrains.annotations.NotNull; +import reactor.core.publisher.Mono; + +public record HitEntry(T key, U value, float score) + implements Comparable> { + + @Override + public int compareTo(@NotNull HitEntry o) { + return Float.compare(o.score, this.score); + } +} diff --git a/src/main/java/it/cavallium/dbengine/client/HitKey.java b/src/main/java/it/cavallium/dbengine/client/HitKey.java new file mode 100644 index 0000000..2c9157a --- /dev/null +++ b/src/main/java/it/cavallium/dbengine/client/HitKey.java @@ -0,0 +1,18 @@ +package it.cavallium.dbengine.client; + +import java.util.Comparator; +import java.util.function.Function; +import org.jetbrains.annotations.NotNull; +import reactor.core.publisher.Mono; + +public record HitKey(T key, float score) implements Comparable> { + + public Mono> withValue(Function> valueGetter) { + return valueGetter.apply(key).map(value -> new HitEntry<>(key, value, score)); + } + + @Override + public int compareTo(@NotNull HitKey o) { + return Float.compare(o.score, this.score); + } +} diff --git a/src/main/java/it/cavallium/dbengine/client/Hits.java b/src/main/java/it/cavallium/dbengine/client/Hits.java new file mode 100644 index 0000000..e8b3918 --- /dev/null +++ b/src/main/java/it/cavallium/dbengine/client/Hits.java @@ -0,0 +1,126 @@ +package it.cavallium.dbengine.client; + +import io.net5.buffer.api.Drop; +import io.net5.buffer.api.Owned; +import io.net5.buffer.api.Send; +import it.cavallium.dbengine.client.query.current.data.TotalHitsCount; +import io.net5.buffer.api.internal.ResourceSupport; +import it.cavallium.dbengine.database.collections.ValueGetter; +import it.cavallium.dbengine.database.collections.ValueTransformer; +import java.util.function.Function; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.util.function.Tuples; + +public final class Hits extends ResourceSupport, Hits> { + + private static final Drop> DROP = new Drop<>() { + @Override + public void drop(Hits obj) { + if (obj.onClose != null) { + obj.onClose.run(); + } + } + + @Override + public Drop> fork() { + return this; + } + + @Override + public void attach(Hits obj) { + + } + }; + + private Flux results; + private TotalHitsCount totalHitsCount; + private Runnable onClose; + + @SuppressWarnings({"unchecked", "rawtypes"}) + public Hits(Flux results, TotalHitsCount totalHitsCount, Runnable onClose) { + super((Drop>) (Drop) DROP); + this.results = results; + this.totalHitsCount = totalHitsCount; + this.onClose = onClose; + } + + public static Hits empty() { + return new Hits<>(Flux.empty(), TotalHitsCount.of(0, true), null); + } + + public static Hits> withValuesLazy(Hits> hits, + ValueGetter valuesGetter) { + var hitsEntry = hits.results().map(hitKey -> hitKey.withValue(valuesGetter::get)); + + return new Hits<>(hitsEntry, hits.totalHitsCount, hits::close); + } + + public static Hits> withValues(Hits> hits, ValueGetter valuesGetter) { + var hitsEntry = hits.results().flatMap(hitKey -> hitKey.withValue(valuesGetter::get)); + + return new Hits<>(hitsEntry, hits.totalHitsCount, hits::close); + } + + public static Function>>, Send>>> generateMapper( + ValueGetter valueGetter) { + return resultToReceive -> { + var result = resultToReceive.receive(); + var hitsToTransform = result.results() + .map(hit -> new LazyHitEntry<>(Mono.just(hit.key()), valueGetter.get(hit.key()), hit.score())); + return new Hits<>(hitsToTransform, result.totalHitsCount(), result::close).send(); + }; + } + + public static Function>>, Send>>> generateMapper( + ValueTransformer valueTransformer) { + return resultToReceive -> { + var result = resultToReceive.receive(); + var hitsToTransform = result.results().map(hit -> Tuples.of(hit.score(), hit.key())); + var transformed = valueTransformer + .transform(hitsToTransform) + .filter(tuple3 -> tuple3.getT3().isPresent()) + .map(tuple3 -> new LazyHitEntry<>(Mono.just(tuple3.getT2()), + Mono.just(tuple3.getT3().orElseThrow()), + tuple3.getT1() + )); + return new Hits<>(transformed, result.totalHitsCount(), result::close).send(); + }; + } + + public Flux results() { + return results; + } + + public TotalHitsCount totalHitsCount() { + return totalHitsCount; + } + + @Override + public String toString() { + return "Hits[" + "results=" + results + ", " + "totalHitsCount=" + totalHitsCount + ']'; + } + + @Override + protected RuntimeException createResourceClosedException() { + return new IllegalStateException("Closed"); + } + + @Override + protected Owned> prepareSend() { + var results = this.results; + var totalHitsCount = this.totalHitsCount; + var onClose = this.onClose; + return drop -> { + var instance = new Hits<>(results, totalHitsCount, onClose); + drop.attach(instance); + return instance; + }; + } + + protected void makeInaccessible() { + this.results = null; + this.totalHitsCount = null; + this.onClose = null; + } +} diff --git a/src/main/java/it/cavallium/dbengine/client/LazyHitEntry.java b/src/main/java/it/cavallium/dbengine/client/LazyHitEntry.java new file mode 100644 index 0000000..feb94d8 --- /dev/null +++ b/src/main/java/it/cavallium/dbengine/client/LazyHitEntry.java @@ -0,0 +1,11 @@ +package it.cavallium.dbengine.client; + +import org.jetbrains.annotations.NotNull; +import reactor.core.publisher.Mono; + +public record LazyHitEntry(Mono key, Mono value, float score) { + + public Mono> resolve() { + return Mono.zip(key, value, (k, v) -> new HitEntry<>(k, v, score)); + } +} diff --git a/src/main/java/it/cavallium/dbengine/client/LazyHitKey.java b/src/main/java/it/cavallium/dbengine/client/LazyHitKey.java new file mode 100644 index 0000000..c6c101f --- /dev/null +++ b/src/main/java/it/cavallium/dbengine/client/LazyHitKey.java @@ -0,0 +1,19 @@ +package it.cavallium.dbengine.client; + +import java.util.function.Function; +import reactor.core.publisher.Mono; + +public record LazyHitKey(Mono key, float score) { + + public LazyHitEntry withValue(Function> valueGetter) { + return new LazyHitEntry<>(key, key.flatMap(valueGetter), score); + } + + public Mono> resolve() { + return key.map(k -> new HitKey<>(k, score)); + } + + public Mono> resolveWithValue(Function> valueGetter) { + return resolve().flatMap(key -> key.withValue(valueGetter)); + } +} diff --git a/src/main/java/it/cavallium/dbengine/client/LuceneIndex.java b/src/main/java/it/cavallium/dbengine/client/LuceneIndex.java index 19832a1..9609d19 100644 --- a/src/main/java/it/cavallium/dbengine/client/LuceneIndex.java +++ b/src/main/java/it/cavallium/dbengine/client/LuceneIndex.java @@ -9,16 +9,11 @@ import it.cavallium.dbengine.database.LLSnapshottable; import it.cavallium.dbengine.database.collections.ValueGetter; import it.cavallium.dbengine.database.collections.ValueTransformer; import java.util.Map.Entry; -import java.util.Optional; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import reactor.util.function.Tuple2; -import reactor.util.function.Tuple3; -import reactor.util.function.Tuples; -@SuppressWarnings("unused") public interface LuceneIndex extends LLSnapshottable { Mono addDocument(T key, U value); @@ -53,32 +48,10 @@ public interface LuceneIndex extends LLSnapshottable { Mono deleteAll(); - Mono>> moreLikeThis(ClientQueryParams> queryParams, T key, U mltDocumentValue); + Mono>>> moreLikeThis(ClientQueryParams queryParams, T key, + U mltDocumentValue); - default Mono>> moreLikeThisWithValues(ClientQueryParams> queryParams, - T key, - U mltDocumentValue, - ValueGetter valueGetter) { - return this.moreLikeThisWithTransformer(queryParams, - key, - mltDocumentValue, - getValueGetterTransformer(valueGetter)); - } - - Mono>> moreLikeThisWithTransformer(ClientQueryParams> queryParams, - T key, - U mltDocumentValue, - ValueTransformer valueTransformer); - - Mono>> search(ClientQueryParams> queryParams); - - default Mono>> searchWithValues(ClientQueryParams> queryParams, - ValueGetter valueGetter) { - return this.searchWithTransformer(queryParams, getValueGetterTransformer(valueGetter)); - } - - Mono>> searchWithTransformer(ClientQueryParams> queryParams, - ValueTransformer valueTransformer); + Mono>>> search(ClientQueryParams queryParams); Mono count(@Nullable CompositeSnapshot snapshot, Query query); @@ -89,17 +62,4 @@ public interface LuceneIndex extends LLSnapshottable { Mono flush(); Mono refresh(boolean force); - - private static ValueTransformer getValueGetterTransformer(ValueGetter valueGetter) { - return new ValueTransformer() { - @Override - public Flux>> transform(Flux> keys) { - return keys.flatMapSequential(key -> valueGetter - .get(key.getT2()) - .map(result -> Tuples.of(key.getT1(), key.getT2(), Optional.of(result))) - .switchIfEmpty(Mono.fromSupplier(() -> Tuples.of(key.getT1(), key.getT2(), Optional.empty()))) - ); - } - }; - } } diff --git a/src/main/java/it/cavallium/dbengine/client/LuceneIndexImpl.java b/src/main/java/it/cavallium/dbengine/client/LuceneIndexImpl.java index d0f0f07..34b65e2 100644 --- a/src/main/java/it/cavallium/dbengine/client/LuceneIndexImpl.java +++ b/src/main/java/it/cavallium/dbengine/client/LuceneIndexImpl.java @@ -19,7 +19,6 @@ import it.cavallium.dbengine.database.LLLuceneIndex; import it.cavallium.dbengine.database.LLSearchResultShard; import it.cavallium.dbengine.database.LLSnapshot; import it.cavallium.dbengine.database.LLTerm; -import it.cavallium.dbengine.database.collections.ValueGetter; import it.cavallium.dbengine.database.collections.ValueTransformer; import java.lang.ref.Cleaner; import java.util.Map; @@ -36,11 +35,9 @@ import reactor.core.publisher.Sinks; import reactor.core.publisher.Sinks.EmitResult; import reactor.core.publisher.Sinks.Empty; import reactor.core.publisher.Sinks.Many; -import reactor.core.publisher.Sinks.One; import reactor.core.scheduler.Schedulers; import reactor.util.concurrent.Queues; import reactor.util.function.Tuple2; -import reactor.util.function.Tuples; public class LuceneIndexImpl implements LuceneIndex { @@ -202,140 +199,47 @@ public class LuceneIndexImpl implements LuceneIndex { return Mono.create(sink -> emitActionOptimistically(new IndexAction.DeleteAll(sink))); } - private Mono>> transformLuceneResultWithTransformer( - Mono> llSearchResultMono) { - return llSearchResultMono.map(llSearchResultToReceive -> { - var llSearchResult = llSearchResultToReceive.receive(); - return new SearchResultKeys<>(llSearchResult.results() - .map(signal -> new SearchResultKey<>(Mono - .fromCallable(signal::key) - .map(indicizer::getKey), signal.score())), - llSearchResult.totalHitsCount(), - llSearchResult::close - ).send(); - }); - } - - private Mono>> transformLuceneResultWithValues( - Mono> llSearchResultMono, - ValueGetter valueGetter) { - return llSearchResultMono.map(llSearchResultToReceive -> { - var llSearchResult = llSearchResultToReceive.receive(); - return new SearchResult<>(llSearchResult.results().map(signal -> { - var key = Mono.fromCallable(signal::key).map(indicizer::getKey); - return new SearchResultItem<>(key, key.flatMap(valueGetter::get), signal.score()); - }), llSearchResult.totalHitsCount(), llSearchResult::close).send(); - }); - } - - private Mono>> transformLuceneResultWithTransformer( - Mono> llSearchResultMono, - ValueTransformer valueTransformer) { - return llSearchResultMono - .map(llSearchResultToReceive -> { - var llSearchResult = llSearchResultToReceive.receive(); - var scoresWithKeysFlux = llSearchResult - .results() - .flatMapSequential(signal -> Mono - .fromCallable(signal::key) - .map(indicizer::getKey) - .map(key -> Tuples.of(signal.score(), key)) - ); - var resultItemsFlux = valueTransformer - .transform(scoresWithKeysFlux) - .filter(tuple3 -> tuple3.getT3().isPresent()) - .map(tuple3 -> new SearchResultItem<>(Mono.just(tuple3.getT2()), - Mono.just(tuple3.getT3().orElseThrow()), - tuple3.getT1() - )); - return new SearchResult<>(resultItemsFlux, llSearchResult.totalHitsCount(), llSearchResult::close).send(); - }); - } - @Override - public Mono>> moreLikeThis(ClientQueryParams> queryParams, + public Mono>>> moreLikeThis(ClientQueryParams queryParams, T key, U mltDocumentValue) { Flux>> mltDocumentFields = indicizer.getMoreLikeThisDocumentFields(key, mltDocumentValue); - return luceneIndex - .moreLikeThis(resolveSnapshot(queryParams.snapshot()), queryParams.toQueryParams(), indicizer.getKeyFieldName(), mltDocumentFields) - .transform(this::transformLuceneResultWithTransformer); - } - - @Override - public Mono>> moreLikeThisWithValues(ClientQueryParams> queryParams, - T key, - U mltDocumentValue, - ValueGetter valueGetter) { - Flux>> mltDocumentFields - = indicizer.getMoreLikeThisDocumentFields(key, mltDocumentValue); return luceneIndex .moreLikeThis(resolveSnapshot(queryParams.snapshot()), queryParams.toQueryParams(), indicizer.getKeyFieldName(), mltDocumentFields ) - .transform(llSearchResult -> this.transformLuceneResultWithValues(llSearchResult, - valueGetter)); + .map(this::mapResults) + .single(); } @Override - public Mono>> moreLikeThisWithTransformer(ClientQueryParams> queryParams, - T key, - U mltDocumentValue, - ValueTransformer valueTransformer) { - Flux>> mltDocumentFields - = indicizer.getMoreLikeThisDocumentFields(key, mltDocumentValue); - return luceneIndex - .moreLikeThis(resolveSnapshot(queryParams.snapshot()), - queryParams.toQueryParams(), - indicizer.getKeyFieldName(), - mltDocumentFields - ) - .transform(llSearchResult -> this.transformLuceneResultWithTransformer(llSearchResult, - valueTransformer)); - } - - @Override - public Mono>> search(ClientQueryParams> queryParams) { + public Mono>>> search(ClientQueryParams queryParams) { return luceneIndex .search(resolveSnapshot(queryParams.snapshot()), queryParams.toQueryParams(), indicizer.getKeyFieldName() ) - .single() - .transform(this::transformLuceneResultWithTransformer) + .map(this::mapResults) .single(); } - @Override - public Mono>> searchWithValues( - ClientQueryParams> queryParams, - ValueGetter valueGetter) { - return luceneIndex - .search(resolveSnapshot(queryParams.snapshot()), queryParams.toQueryParams(), - indicizer.getKeyFieldName()) - .transform(llSearchResult -> this.transformLuceneResultWithValues(llSearchResult, - valueGetter)); - } + private Send>> mapResults(Send llSearchResultToReceive) { + var llSearchResult = llSearchResultToReceive.receive(); + var scoresWithKeysFlux = llSearchResult + .results() + .map(hit -> new HitKey<>(indicizer.getKey(hit.key()), hit.score())); - @Override - public Mono>> searchWithTransformer( - ClientQueryParams> queryParams, - ValueTransformer valueTransformer) { - return luceneIndex - .search(resolveSnapshot(queryParams.snapshot()), queryParams.toQueryParams(), - indicizer.getKeyFieldName()) - .transform(llSearchResult -> this.transformLuceneResultWithTransformer(llSearchResult, - valueTransformer)); + return new Hits<>(scoresWithKeysFlux, llSearchResult.totalHitsCount(), llSearchResult::close).send(); } @Override public Mono count(@Nullable CompositeSnapshot snapshot, Query query) { return this - .search(ClientQueryParams.>builder().snapshot(snapshot).query(query).limit(0).build()) + .search(ClientQueryParams.>builder().snapshot(snapshot).query(query).limit(0).build()) .single() .map(searchResultKeysSend -> { try (var searchResultKeys = searchResultKeysSend.receive()) { diff --git a/src/main/java/it/cavallium/dbengine/client/SearchResult.java b/src/main/java/it/cavallium/dbengine/client/SearchResult.java deleted file mode 100644 index 99b3607..0000000 --- a/src/main/java/it/cavallium/dbengine/client/SearchResult.java +++ /dev/null @@ -1,81 +0,0 @@ -package it.cavallium.dbengine.client; - -import io.net5.buffer.api.Drop; -import io.net5.buffer.api.Owned; -import it.cavallium.dbengine.client.query.current.data.TotalHitsCount; -import io.net5.buffer.api.internal.ResourceSupport; -import reactor.core.publisher.Flux; - -public final class SearchResult extends ResourceSupport, SearchResult> { - - private static final Drop> DROP = new Drop<>() { - @Override - public void drop(SearchResult obj) { - if (obj.onClose != null) { - obj.onClose.run(); - } - } - - @Override - public Drop> fork() { - return this; - } - - @Override - public void attach(SearchResult obj) { - - } - }; - - private Flux> results; - private TotalHitsCount totalHitsCount; - private Runnable onClose; - - @SuppressWarnings({"unchecked", "rawtypes"}) - public SearchResult(Flux> results, TotalHitsCount totalHitsCount, Runnable onClose) { - super((Drop>) (Drop) DROP); - this.results = results; - this.totalHitsCount = totalHitsCount; - this.onClose = onClose; - } - - public static SearchResult empty() { - return new SearchResult(Flux.empty(), TotalHitsCount.of(0, true), null); - } - - public Flux> results() { - return results; - } - - public TotalHitsCount totalHitsCount() { - return totalHitsCount; - } - - @Override - public String toString() { - return "SearchResult[" + "results=" + results + ", " + "totalHitsCount=" + totalHitsCount + ']'; - } - - @Override - protected RuntimeException createResourceClosedException() { - return new IllegalStateException("Closed"); - } - - @Override - protected Owned> prepareSend() { - var results = this.results; - var totalHitsCount = this.totalHitsCount; - var onClose = this.onClose; - return drop -> { - var instance = new SearchResult<>(results, totalHitsCount, onClose); - drop.attach(instance); - return instance; - }; - } - - protected void makeInaccessible() { - this.results = null; - this.totalHitsCount = null; - this.onClose = null; - } -} diff --git a/src/main/java/it/cavallium/dbengine/client/SearchResultItem.java b/src/main/java/it/cavallium/dbengine/client/SearchResultItem.java deleted file mode 100644 index aac0e12..0000000 --- a/src/main/java/it/cavallium/dbengine/client/SearchResultItem.java +++ /dev/null @@ -1,13 +0,0 @@ -package it.cavallium.dbengine.client; - -import org.jetbrains.annotations.NotNull; -import reactor.core.publisher.Mono; - -public record SearchResultItem(Mono key, Mono value, float score) - implements Comparable> { - - @Override - public int compareTo(@NotNull SearchResultItem o) { - return Float.compare(o.score, this.score); - } -} diff --git a/src/main/java/it/cavallium/dbengine/client/SearchResultKey.java b/src/main/java/it/cavallium/dbengine/client/SearchResultKey.java deleted file mode 100644 index 6193f3d..0000000 --- a/src/main/java/it/cavallium/dbengine/client/SearchResultKey.java +++ /dev/null @@ -1,7 +0,0 @@ -package it.cavallium.dbengine.client; - -import java.util.Objects; -import java.util.StringJoiner; -import reactor.core.publisher.Mono; - -public record SearchResultKey(Mono key, float score) {} diff --git a/src/main/java/it/cavallium/dbengine/client/SearchResultKeys.java b/src/main/java/it/cavallium/dbengine/client/SearchResultKeys.java deleted file mode 100644 index d017f04..0000000 --- a/src/main/java/it/cavallium/dbengine/client/SearchResultKeys.java +++ /dev/null @@ -1,96 +0,0 @@ -package it.cavallium.dbengine.client; - -import io.net5.buffer.api.Drop; -import io.net5.buffer.api.Owned; -import it.cavallium.dbengine.client.query.current.data.TotalHitsCount; -import io.net5.buffer.api.internal.ResourceSupport; -import it.cavallium.dbengine.database.collections.ValueGetter; -import org.warp.commonutils.log.Logger; -import org.warp.commonutils.log.LoggerFactory; -import reactor.core.publisher.Flux; - -@SuppressWarnings("unused") -public final class SearchResultKeys extends ResourceSupport, SearchResultKeys> { - - private static final Logger logger = LoggerFactory.getLogger(SearchResultKeys.class); - - private static final Drop> DROP = new Drop<>() { - @Override - public void drop(SearchResultKeys obj) { - if (obj.onClose != null) { - obj.onClose.run(); - } - } - - @Override - public Drop> fork() { - return this; - } - - @Override - public void attach(SearchResultKeys obj) { - - } - }; - - private Flux> results; - private TotalHitsCount totalHitsCount; - private Runnable onClose; - - @SuppressWarnings({"unchecked", "rawtypes"}) - public SearchResultKeys(Flux> results, TotalHitsCount totalHitsCount, Runnable onClose) { - super((Drop>) (Drop) DROP); - this.results = results; - this.totalHitsCount = totalHitsCount; - this.onClose = onClose; - } - - public static SearchResultKeys empty() { - return new SearchResultKeys<>(Flux.empty(), TotalHitsCount.of(0, true), null); - } - - public SearchResult withValues(ValueGetter valuesGetter) { - return new SearchResult<>(results.map(item -> new SearchResultItem<>(item.key(), - item.key().flatMap(valuesGetter::get), - item.score() - )), totalHitsCount, this::close); - } - - public Flux> results() { - return results; - } - - public TotalHitsCount totalHitsCount() { - return totalHitsCount; - } - - @Override - public String toString() { - return "SearchResultKeys[" + "results=" + results + ", " + "totalHitsCount=" + totalHitsCount + ']'; - } - - @Override - protected RuntimeException createResourceClosedException() { - return new IllegalStateException("Closed"); - } - - @Override - protected Owned> prepareSend() { - var results = this.results; - var totalHitsCount = this.totalHitsCount; - var onClose = this.onClose; - makeInaccessible(); - return drop -> { - var instance = new SearchResultKeys<>(results, totalHitsCount, onClose); - drop.attach(instance); - return instance; - }; - } - - protected void makeInaccessible() { - this.results = null; - this.totalHitsCount = null; - this.onClose = null; - } - -} diff --git a/src/main/java/it/cavallium/dbengine/client/query/ClientQueryParams.java b/src/main/java/it/cavallium/dbengine/client/query/ClientQueryParams.java index df7007b..8ec2560 100644 --- a/src/main/java/it/cavallium/dbengine/client/query/ClientQueryParams.java +++ b/src/main/java/it/cavallium/dbengine/client/query/ClientQueryParams.java @@ -12,17 +12,17 @@ import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @RecordBuilder -public final record ClientQueryParams(@Nullable CompositeSnapshot snapshot, - @NotNull Query query, - long offset, - long limit, - @Nullable Float minCompetitiveScore, - @Nullable Sort sort, - boolean complete) { +public final record ClientQueryParams(@Nullable CompositeSnapshot snapshot, + @NotNull Query query, + long offset, + long limit, + @Nullable Float minCompetitiveScore, + @Nullable Sort sort, + boolean complete) { - public static ClientQueryParamsBuilder builder() { + public static ClientQueryParamsBuilder builder() { return ClientQueryParamsBuilder - .builder() + .builder() .snapshot(null) .offset(0) .limit(Long.MAX_VALUE) diff --git a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingle.java b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingle.java index 5103ec7..e4cda43 100644 --- a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingle.java +++ b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingle.java @@ -7,7 +7,6 @@ import io.net5.buffer.api.Send; import io.net5.buffer.api.internal.ResourceSupport; import it.cavallium.dbengine.client.BadBlock; import it.cavallium.dbengine.client.CompositeSnapshot; -import it.cavallium.dbengine.client.SearchResultKeys; import it.cavallium.dbengine.database.Delta; import it.cavallium.dbengine.database.LLDictionary; import it.cavallium.dbengine.database.LLDictionaryResultType; diff --git a/src/main/java/it/cavallium/dbengine/database/disk/LLLocalReactiveRocksIterator.java b/src/main/java/it/cavallium/dbengine/database/disk/LLLocalReactiveRocksIterator.java index 7698d08..1856e1a 100644 --- a/src/main/java/it/cavallium/dbengine/database/disk/LLLocalReactiveRocksIterator.java +++ b/src/main/java/it/cavallium/dbengine/database/disk/LLLocalReactiveRocksIterator.java @@ -4,22 +4,14 @@ import static it.cavallium.dbengine.database.LLUtils.MARKER_ROCKSDB; import static it.cavallium.dbengine.database.disk.LLLocalDictionary.getRocksIterator; import io.net5.buffer.api.Buffer; -import io.net5.buffer.api.BufferAllocator; import io.net5.buffer.api.Drop; import io.net5.buffer.api.Owned; -import io.net5.buffer.api.Resource; import io.net5.buffer.api.Send; import io.net5.buffer.api.internal.ResourceSupport; -import io.net5.util.IllegalReferenceCountException; -import it.cavallium.dbengine.client.SearchResult; import it.cavallium.dbengine.database.LLRange; import it.cavallium.dbengine.database.LLUtils; -import it.cavallium.dbengine.netty.NullableBuffer; -import java.util.concurrent.atomic.AtomicBoolean; import org.jetbrains.annotations.Nullable; -import org.rocksdb.ColumnFamilyHandle; import org.rocksdb.ReadOptions; -import org.rocksdb.RocksDB; import org.rocksdb.RocksDBException; import org.warp.commonutils.log.Logger; import org.warp.commonutils.log.LoggerFactory; diff --git a/src/test/java/it/cavallium/dbengine/TestLuceneSearches.java b/src/test/java/it/cavallium/dbengine/TestLuceneSearches.java index 94b3c3f..a88de11 100644 --- a/src/test/java/it/cavallium/dbengine/TestLuceneSearches.java +++ b/src/test/java/it/cavallium/dbengine/TestLuceneSearches.java @@ -10,10 +10,11 @@ import static org.junit.jupiter.api.Assertions.fail; import io.net5.buffer.PooledByteBufAllocator; import it.cavallium.dbengine.DbTestUtils.TempDb; import it.cavallium.dbengine.DbTestUtils.TestAllocator; +import it.cavallium.dbengine.client.HitKey; +import it.cavallium.dbengine.client.Hits; import it.cavallium.dbengine.client.LuceneIndex; import it.cavallium.dbengine.client.Sort; -import it.cavallium.dbengine.client.SearchResultKey; -import it.cavallium.dbengine.client.SearchResultKeys; +import it.cavallium.dbengine.client.LazyHitKey; import it.cavallium.dbengine.client.query.ClientQueryParams; import it.cavallium.dbengine.client.query.ClientQueryParamsBuilder; import it.cavallium.dbengine.client.query.current.data.BooleanQuery; @@ -237,7 +238,7 @@ public class TestLuceneSearches { } private boolean supportsPreciseHitsCount(LocalSearcher searcher, - ClientQueryParams> query) { + ClientQueryParams query) { var sorted = query.isSorted(); if (searcher instanceof UnsortedUnscoredStreamingMultiSearcher) { return false; @@ -248,7 +249,7 @@ public class TestLuceneSearches { } } - public void testSearch(ClientQueryParamsBuilder> queryParamsBuilder, + public void testSearch(ClientQueryParamsBuilder queryParamsBuilder, ExpectedQueryType expectedQueryType) throws Throwable { runSearchers(expectedQueryType, searcher -> { @@ -291,7 +292,7 @@ public class TestLuceneSearches { @MethodSource("provideQueryArgumentsScoreModeAndSort") public void testSearchNoDocs(boolean shards, Sort multiSort) throws Throwable { var queryBuilder = ClientQueryParams - .>builder() + .>builder() .query(new MatchNoDocsQuery()) .snapshot(null) .complete(true) @@ -305,7 +306,7 @@ public class TestLuceneSearches { @MethodSource("provideQueryArgumentsScoreModeAndSort") public void testSearchAllDocs(boolean shards, Sort multiSort) throws Throwable { var queryBuilder = ClientQueryParams - .>builder() + .>builder() .query(new MatchAllDocsQuery()) .snapshot(null) .complete(true) @@ -319,7 +320,7 @@ public class TestLuceneSearches { @MethodSource("provideQueryArgumentsScoreModeAndSort") public void testSearchAdvancedText(boolean shards, Sort multiSort) throws Throwable { var queryBuilder = ClientQueryParams - .>builder() + .>builder() .query(new BooleanQuery(List.of( new BooleanQueryPart(new BoostQuery(new TermQuery(new Term("text", "hello")), 3), new OccurShould()), new BooleanQueryPart(new TermQuery(new Term("text", "world")), new OccurShould()), @@ -357,14 +358,10 @@ public class TestLuceneSearches { assertEquals(new TotalHitsCount(expectedCount, true), hits); } - private List getResults(SearchResultKeys results) { + private List getResults(Hits> results) { return run(results .results() - .flatMapSequential(searchResultKey -> searchResultKey - .key() - .single() - .map(key -> new Scored(key, searchResultKey.score())) - ) + .map(key -> new Scored(key.key(), key.score())) .collectList()); }