You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@solr.apache.org by "tuannh982 (via GitHub)" <gi...@apache.org> on 2024/02/29 13:10:06 UTC
[PR] Draft - PoC execute FQs in parallel [solr]
tuannh982 opened a new pull request, #2322:
URL: https://github.com/apache/solr/pull/2322
https://issues.apache.org/jira/browse/SOLR-XXXXX
<!--
_(If you are a project committer then you may remove some/all of the following template.)_
Before creating a pull request, please file an issue in the ASF Jira system for Solr:
* https://issues.apache.org/jira/projects/SOLR
For something minor (i.e. that wouldn't be worth putting in release notes), you can skip JIRA.
To create a Jira issue, you will need to create an account there first.
The title of the PR should reference the Jira issue number in the form:
* SOLR-####: <short description of problem or changes>
SOLR must be fully capitalized. A short description helps people scanning pull requests for items they can work on.
Properly referencing the issue in the title ensures that Jira is correctly updated with code review comments and commits. -->
# Description
Please provide a short description of the changes you're making with this pull request.
# Solution
Please provide a short description of the approach taken to implement your solution.
# Tests
Please describe the tests you've developed or run to confirm this patch implements the feature or solves the problem.
# Checklist
Please review the following and check all that apply:
- [ ] I have reviewed the guidelines for [How to Contribute](https://github.com/apache/solr/blob/main/CONTRIBUTING.md) and my code conforms to the standards described there to the best of my ability.
- [ ] I have created a Jira issue and added the issue ID to my pull request title.
- [ ] I have given Solr maintainers [access](https://help.github.com/en/articles/allowing-changes-to-a-pull-request-branch-created-from-a-fork) to contribute to my PR branch. (optional but recommended)
- [ ] I have developed this patch against the `main` branch.
- [ ] I have run `./gradlew check`.
- [ ] I have added tests for my changes.
- [ ] I have added documentation for the [Reference Guide](https://github.com/apache/solr/tree/main/solr/solr-ref-guide)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For additional commands, e-mail: issues-help@solr.apache.org
Re: [PR] Draft - PoC execute FQs in parallel [solr]
Posted by "tuannh982 (via GitHub)" <gi...@apache.org>.
tuannh982 commented on code in PR #2322:
URL: https://github.com/apache/solr/pull/2322#discussion_r1507851868
##########
solr/core/src/java/org/apache/solr/search/SolrIndexSearcher.java:
##########
@@ -1389,6 +1564,9 @@ private DocSet getResult(DocsEnumState deState, int largestPossible) throws IOEx
// query must be positive
protected DocSet getDocSetNC(Query query, DocSet filter) throws IOException {
+ try {
+ Thread.sleep(20);
Review Comment:
instead of indexing millions document, just simulating the slowness here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For additional commands, e-mail: issues-help@solr.apache.org
Re: [PR] Draft - PoC execute FQs in parallel [solr]
Posted by "tuannh982 (via GitHub)" <gi...@apache.org>.
tuannh982 commented on code in PR #2322:
URL: https://github.com/apache/solr/pull/2322#discussion_r1507851176
##########
solr/core/src/java/org/apache/solr/search/SolrIndexSearcher.java:
##########
@@ -1166,6 +1166,181 @@ public static class ProcessedFilter {
* resolved against the filter cache, and populate it.
*/
public ProcessedFilter getProcessedFilter(List<Query> queries) throws IOException {
+ ProcessedFilter pf = new ProcessedFilter();
+ if (queries == null || queries.isEmpty()) {
+ return pf;
+ }
+
+ int length = queries.size();
+
+ // We combine all the filter queries that come from the filter cache into "answer".
+ // This might become pf.answer but not if there are any non-cached filters
+ DocSet[] answer0 = new DocSet[]{null};
+ DocSet answer = null;
+ boolean[] neg = new boolean[length];
+ DocSet[] sets = new DocSet[length];
+ List<ExtendedQuery> notCached = null;
+ List<PostFilter> postFilters = null;
+ AtomicReference<IOException> ioException = new AtomicReference<>(null);
+ List<CompletableFuture<Void>> futures = new ArrayList<>(length);
+
+ int end = 0; // size of "sets" and "neg"; parallel arrays
+ AtomicInteger numDocSets = new AtomicInteger(0);
+
+ for (Query q : queries) {
+ if (q instanceof ExtendedQuery) {
+ ExtendedQuery eq = (ExtendedQuery) q;
+ if (!eq.getCache()) {
+ if (eq.getCost() >= 100 && eq instanceof PostFilter) {
+ if (postFilters == null) postFilters = new ArrayList<>(length - end);
+ postFilters.add((PostFilter) q);
+ } else {
+ if (notCached == null) notCached = new ArrayList<>(length - end);
+ notCached.add((ExtendedQuery) q);
+ }
+ continue;
+ }
+ }
+
+ if (filterCache == null) {
+ // there is no cache: don't pull bitsets
+ if (notCached == null) notCached = new ArrayList<>(length - end);
+ WrappedQuery uncached = new WrappedQuery(q);
+ uncached.setCache(false);
+ notCached.add(uncached);
+ continue;
+ }
+
+ final int index = end;
+ CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
+ Query posQuery = QueryUtils.getAbs(q);
+ if (ioException.get() != null) {
+ return null;
+ }
+ DocSet docSet = null;
+ try {
+ docSet = getPositiveDocSet(posQuery);
+ } catch (IOException e) {
+ ioException.compareAndSet(null, e);
+ return null;
+ }
+ // Negative query if absolute value different from original
+ if (Objects.equals(q, posQuery)) {
+ synchronized (answer0) {
+ // keep track of the smallest positive set; use "answer" for this.
+ if (answer0[0] == null) {
+ answer0[0] = docSet;
+ return null;
+ }
+ // note: assume that size() is cached. It generally comes from the cache, so should be.
+ if (docSet.size() < answer0[0].size()) {
+ // swap answer & docSet so that answer is smallest
+ DocSet tmp = answer0[0];
+ answer0[0] = docSet;
+ docSet = tmp;
+ }
+ }
+ neg[index] = false;
+ } else {
+ neg[index] = true;
+ }
+ sets[index] = docSet;
+ numDocSets.incrementAndGet();
+ return null;
+ });
+ futures.add(future);
+ end++;
+ } // end of queries
+
+ // await futures
+ CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[0]));
+ try {
+ future.get(); // TODO add time limit?
+ answer = answer0[0];
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ if (ioException.get() != null) {
+ throw ioException.get();
+ }
+
+ if (numDocSets.get() > 0) {
+ // Are all of our normal cached filters negative?
+ if (answer == null) {
+ answer = getLiveDocSet();
+ }
+
+ // This optimizes for the case where we have more than 2 filters and instead
+ // of copying the bitsets we make one mutable bitset. We should only do this
+ // for BitDocSet since it clones the backing bitset for andNot and intersection.
+ if (numDocSets.get() > 1 && answer instanceof BitDocSet) {
+ answer = MutableBitDocSet.fromBitDocSet((BitDocSet) answer);
+ }
+
+ // do negative queries first to shrink set size
+ for (int i = 0; i < end; i++) {
+ if (neg[i] && sets[i] != null) answer = answer.andNot(sets[i]);
+ }
+
+ for (int i = 0; i < end; i++) {
+ if (!neg[i] && sets[i] != null) answer = answer.intersection(sets[i]);
+ }
+
+ // Make sure to keep answer as an immutable DocSet if we made it mutable
+ answer = MutableBitDocSet.unwrapIfMutable(answer);
+ }
+
+ // ignore "answer" if it simply matches all docs
+ if (answer != null && answer.size() == numDocs()) {
+ answer = null;
+ }
+
+ // answer is done.
+
+ // If no notCached nor postFilters, we can return now.
+ if (notCached == null && postFilters == null) {
+ // "answer" is the only part of the filter, so set it.
+ if (answer != null) {
+ pf.answer = answer;
+ pf.filter = answer.makeQuery();
+ }
+ return pf;
+ }
+ // pf.answer will remain null ... (our local "answer" var is not the complete answer)
+
+ // Set pf.filter based on combining "answer" and "notCached"
+ if (notCached == null) {
+ if (answer != null) {
+ pf.filter = answer.makeQuery();
+ }
+ } else {
+ notCached.sort(sortByCost); // pointless?
+ final BooleanQuery.Builder builder = new BooleanQuery.Builder();
+ if (answer != null) {
+ builder.add(answer.makeQuery(), Occur.FILTER);
+ }
+ for (ExtendedQuery eq : notCached) {
+ Query q = eq.getCostAppliedQuery();
+ builder.add(q, Occur.FILTER);
+ }
+ pf.filter = builder.build();
+ }
+
+ // Set pf.postFilter
+ if (postFilters != null) {
+ postFilters.sort(sortByCost);
+ for (int i = postFilters.size() - 1; i >= 0; i--) {
+ DelegatingCollector prev = pf.postFilter;
+ pf.postFilter = postFilters.get(i).getFilterCollector(this);
+ if (prev != null) pf.postFilter.setDelegate(prev);
+ }
+ }
+
+ return pf;
+ }
+
+ // the old method
+ public ProcessedFilter getProcessedFilter_bk(List<Query> queries) throws IOException {
Review Comment:
here the old code, just keeping it for later benchmark
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For additional commands, e-mail: issues-help@solr.apache.org
Re: [PR] Draft - PoC execute FQs in parallel [solr]
Posted by "tuannh982 (via GitHub)" <gi...@apache.org>.
tuannh982 commented on code in PR #2322:
URL: https://github.com/apache/solr/pull/2322#discussion_r1507851176
##########
solr/core/src/java/org/apache/solr/search/SolrIndexSearcher.java:
##########
@@ -1166,6 +1166,181 @@ public static class ProcessedFilter {
* resolved against the filter cache, and populate it.
*/
public ProcessedFilter getProcessedFilter(List<Query> queries) throws IOException {
+ ProcessedFilter pf = new ProcessedFilter();
+ if (queries == null || queries.isEmpty()) {
+ return pf;
+ }
+
+ int length = queries.size();
+
+ // We combine all the filter queries that come from the filter cache into "answer".
+ // This might become pf.answer but not if there are any non-cached filters
+ DocSet[] answer0 = new DocSet[]{null};
+ DocSet answer = null;
+ boolean[] neg = new boolean[length];
+ DocSet[] sets = new DocSet[length];
+ List<ExtendedQuery> notCached = null;
+ List<PostFilter> postFilters = null;
+ AtomicReference<IOException> ioException = new AtomicReference<>(null);
+ List<CompletableFuture<Void>> futures = new ArrayList<>(length);
+
+ int end = 0; // size of "sets" and "neg"; parallel arrays
+ AtomicInteger numDocSets = new AtomicInteger(0);
+
+ for (Query q : queries) {
+ if (q instanceof ExtendedQuery) {
+ ExtendedQuery eq = (ExtendedQuery) q;
+ if (!eq.getCache()) {
+ if (eq.getCost() >= 100 && eq instanceof PostFilter) {
+ if (postFilters == null) postFilters = new ArrayList<>(length - end);
+ postFilters.add((PostFilter) q);
+ } else {
+ if (notCached == null) notCached = new ArrayList<>(length - end);
+ notCached.add((ExtendedQuery) q);
+ }
+ continue;
+ }
+ }
+
+ if (filterCache == null) {
+ // there is no cache: don't pull bitsets
+ if (notCached == null) notCached = new ArrayList<>(length - end);
+ WrappedQuery uncached = new WrappedQuery(q);
+ uncached.setCache(false);
+ notCached.add(uncached);
+ continue;
+ }
+
+ final int index = end;
+ CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
+ Query posQuery = QueryUtils.getAbs(q);
+ if (ioException.get() != null) {
+ return null;
+ }
+ DocSet docSet = null;
+ try {
+ docSet = getPositiveDocSet(posQuery);
+ } catch (IOException e) {
+ ioException.compareAndSet(null, e);
+ return null;
+ }
+ // Negative query if absolute value different from original
+ if (Objects.equals(q, posQuery)) {
+ synchronized (answer0) {
+ // keep track of the smallest positive set; use "answer" for this.
+ if (answer0[0] == null) {
+ answer0[0] = docSet;
+ return null;
+ }
+ // note: assume that size() is cached. It generally comes from the cache, so should be.
+ if (docSet.size() < answer0[0].size()) {
+ // swap answer & docSet so that answer is smallest
+ DocSet tmp = answer0[0];
+ answer0[0] = docSet;
+ docSet = tmp;
+ }
+ }
+ neg[index] = false;
+ } else {
+ neg[index] = true;
+ }
+ sets[index] = docSet;
+ numDocSets.incrementAndGet();
+ return null;
+ });
+ futures.add(future);
+ end++;
+ } // end of queries
+
+ // await futures
+ CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[0]));
+ try {
+ future.get(); // TODO add time limit?
+ answer = answer0[0];
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ if (ioException.get() != null) {
+ throw ioException.get();
+ }
+
+ if (numDocSets.get() > 0) {
+ // Are all of our normal cached filters negative?
+ if (answer == null) {
+ answer = getLiveDocSet();
+ }
+
+ // This optimizes for the case where we have more than 2 filters and instead
+ // of copying the bitsets we make one mutable bitset. We should only do this
+ // for BitDocSet since it clones the backing bitset for andNot and intersection.
+ if (numDocSets.get() > 1 && answer instanceof BitDocSet) {
+ answer = MutableBitDocSet.fromBitDocSet((BitDocSet) answer);
+ }
+
+ // do negative queries first to shrink set size
+ for (int i = 0; i < end; i++) {
+ if (neg[i] && sets[i] != null) answer = answer.andNot(sets[i]);
+ }
+
+ for (int i = 0; i < end; i++) {
+ if (!neg[i] && sets[i] != null) answer = answer.intersection(sets[i]);
+ }
+
+ // Make sure to keep answer as an immutable DocSet if we made it mutable
+ answer = MutableBitDocSet.unwrapIfMutable(answer);
+ }
+
+ // ignore "answer" if it simply matches all docs
+ if (answer != null && answer.size() == numDocs()) {
+ answer = null;
+ }
+
+ // answer is done.
+
+ // If no notCached nor postFilters, we can return now.
+ if (notCached == null && postFilters == null) {
+ // "answer" is the only part of the filter, so set it.
+ if (answer != null) {
+ pf.answer = answer;
+ pf.filter = answer.makeQuery();
+ }
+ return pf;
+ }
+ // pf.answer will remain null ... (our local "answer" var is not the complete answer)
+
+ // Set pf.filter based on combining "answer" and "notCached"
+ if (notCached == null) {
+ if (answer != null) {
+ pf.filter = answer.makeQuery();
+ }
+ } else {
+ notCached.sort(sortByCost); // pointless?
+ final BooleanQuery.Builder builder = new BooleanQuery.Builder();
+ if (answer != null) {
+ builder.add(answer.makeQuery(), Occur.FILTER);
+ }
+ for (ExtendedQuery eq : notCached) {
+ Query q = eq.getCostAppliedQuery();
+ builder.add(q, Occur.FILTER);
+ }
+ pf.filter = builder.build();
+ }
+
+ // Set pf.postFilter
+ if (postFilters != null) {
+ postFilters.sort(sortByCost);
+ for (int i = postFilters.size() - 1; i >= 0; i--) {
+ DelegatingCollector prev = pf.postFilter;
+ pf.postFilter = postFilters.get(i).getFilterCollector(this);
+ if (prev != null) pf.postFilter.setDelegate(prev);
+ }
+ }
+
+ return pf;
+ }
+
+ // the old method
+ public ProcessedFilter getProcessedFilter_bk(List<Query> queries) throws IOException {
Review Comment:
here the old code, just keeping it for later benchmark
##########
solr/core/src/java/org/apache/solr/search/SolrIndexSearcher.java:
##########
@@ -1389,6 +1564,9 @@ private DocSet getResult(DocsEnumState deState, int largestPossible) throws IOEx
// query must be positive
protected DocSet getDocSetNC(Query query, DocSet filter) throws IOException {
+ try {
+ Thread.sleep(20);
Review Comment:
instead of indexing millions document, just simulating the slowness here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For additional commands, e-mail: issues-help@solr.apache.org
Re: [PR] Draft - PoC execute FQs in parallel [solr]
Posted by "tuannh982 (via GitHub)" <gi...@apache.org>.
tuannh982 closed pull request #2322: Draft - PoC execute FQs in parallel
URL: https://github.com/apache/solr/pull/2322
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For additional commands, e-mail: issues-help@solr.apache.org