This is definitely a half-baked patch, I want to see if this is the right track for moving this issue forward. I created a mergeGroups function to parallel the mergeIds function used to handle 'normal' document responses. The added test is failing currently (though not raising any exceptions) because of an issue with how documents (and subsequently, groups) are sorted, but I'm stumbling in trying to fix the issue and thought it better to submit the patch and get advice now.

Any feedback is really appreciated, I want to move this issue forward as quickly as possible.

Harish - I haven't looked at this patch, but since you are not mentioning SOLR-1682, I'm assuming your patch doesn't build on top of what's in trunk (which is what SOLR-1682 provided)..... although I see it applies to today's trunk.

If the above is correct, would it be possible for you to make a patch that adds support for distributed search to trunk?

Also, here's feedback from somebody who applied and tried your patch today:
"The patch applied cleanly against the nightly build from last night, but a query resulted in solr hanging, then running out of heap and dumping."

Hi Otis - Sorry for the late reply, I've been travelling. I'll have a look at SOLR-1682 and try to build on top of that patch (I did build on top of the trunk originally, but it may be easier to work with the CollapseComponent). This patch was incomplete, I'd just wanted to get advice on implementation before moving forward with it whole hog.

Updated the patch, so that is up to date with the trunk. The code compiles, but the test still fails.

In order to support distributed grouping properly we need a new distributed stage. This stage uses only the first phase collector and only retrieves the top groups.

All the top groups from the different shards can be merged with the utilities Lucene grouping module provides (GroupMerger). The groups that survive the merge are send to back to the shard where the came from in the next phase.

After the merge the normal query stages kick in. But on the shard side only the second phase collector is used.

Attached a rough initial approach for better distributed grouping. This solves distributed grouping in 3 steps. First round trip for getting top groups, second round trip to group documents inside a top group and a third round trip to get the stored fields.

The following parameters will never give correct with this approach:

• group.truncate
• group.ngroups

Some current issues / todos:

• The group.field parameter is now the only supported command.
• In distributed mode group.format=grouped is only the supported format.
• There is some issue with sorting now when a group sort / normal sort has null values.
• Need more tests.
• The code is a bit hacked together now, so there is needs to be cleaned up and restructured.

Other than that it seems to work.

Minor update. Small code cleanup.

Fixed the sorting issue with groups and inside groups when a sorting value is null.

Updated patch to latest trunk changes.

I think the next steps are to support the following parameters:

• group.query
• qroup.main
• group.formay (simple)

After that commit it and port it back to 3x. The group.func parameter can then be addressed after it ported back, since it is only a 4.0 feature.

Martijn, would it be possible to support group.ngroups for distributed grouping? I noticed that when merging TopGroups it's stated that totalGroupCount cannot be merged, but wouldn't it just be a matter of adding up the numbers (since groups shouldn't overlap across shards)? I'm probably missing something, but having a total number of groups - also across shards - is a tremendous help for paging; despite the performance penalty involved.

I noticed that when merging TopGroups it's stated that totalGroupCount cannot be merged, but wouldn't it just be a matter of adding up the numbers (since groups shouldn't overlap across shards)

If group.ngroups is specified we can just add the total group count together, but the Solr user is then responsible for making sure that the documents belonging to a group end up in the same shard. If this is not done, then the group counts will be incorrect.

I'm probably missing something, but having a total number of groups - also across shards - is a tremendous help for paging; despite the performance penalty involved.

All shards need to send all the unique groups over the wire to the shard the is responsible for the group merging. For indices with a lot a of unique groups this will be very expensive. In the current solution only the top N groups (N being equals to rows param) are sent over the wire from the individual shards to the shard that is responsible for the group merging.

Found two issues with this patch:
1. Using Faceting with the combination of sharding and grouping in our queries an error occurs.
2. When a shard returns no results and other shards do an error occurs.

Thanks
Matt.

Thanks for reporting these issues Matt! I'll update the patch soon.

Updated patch.

• Fixes for the errors that Matt reported.
• If group.ngroups is specified the groupCount is also merged. It is important that all documents of one group are in the same shard. Otherwise the groupCount will be incorrect.
• A lot of renames and refactorings.

The changes to modules/grouping look good...

There's a small typo in the comment ("In case groups where indexed..."
-> "In case groups were indexed..."). Also I wouldn't call this
"properly"? Ie, it's OK if the same group value appears on more than
one shard? It just means we can't compute the exact total group count
(we can compute an upper bound).

Can you update the .merge javadocs? Now it says it cannot compute
totalGroupCount; I think just change that to say that it'll produce
the sum of the total group counts, which is an upper bound on the true
value? And this will be exact if you know the groups are orthogonal
across shards?

Maybe, for better readability, change this:

  if (groupValue != null ? !groupValue.equals(that.groupValue) : that.groupValue != null) return false;

to this:

  if (groupValue == null) {
    if (that.groupValue != null) {
      return false;
    }
  } else if (!groupValue.equals(other.groupValue)) {
    return false;
  }

?

Thanks for the review! I'll update the patch.
The SearchGroup#equals method was generated, so that is why it isn't very readable.

And this will be exact if you know the groups are orthogonal across shards?

Yes, if all the documents of a group reside on the same shard, the groupCount will be correct.

I think just change that to say that it'll produce the sum of the total group counts, which is an upper bound on the true value?

If documents of a group do reside on more than one shard then the count will always be higher then it would be if all documents of a group stay on one shard. So it is an upper bound.

Thanks for adding the ngroups feature Martijn - it's really much appreciated; with it I can move forward with some of the functionality that I've been looking to add to a project of mine. The only thing is that I haven't gotten it to work quite yet. When I do:

http://localhost:8983/solr/foo/select?wt=xml&rows=2&group=true&group.field=dcterms_source&group.ngroups=true

I get an "ngroup"-count along with the "matches"-count just fine, but when I then do:

http://localhost:8983/solr/foo/select?wt=xml&rows=2&group=true&group.field=dcterms_source&group.ngroups=true&shards=localhost:8383/solr/foo,localhost:8983/solr/bar

It's not added in. Any thoughts?

I noticed this also yesterday. I do merge the group count, but I forgot the add it to the final response...
I'll update this patch later today.

Updated patch.

• Group merging works now as expected.
• Included Micheal's review comments.
• Using group.format=simple also works when in distributed mode.

Great - ngroups is coming through now! Another thing I noticed is that highlighting doesn't work on account of the resultIds not getting set in the ResponseBuilder. It only happens in combination with distributed grouping - so it works when I do this:

http://localhost:8983/solr/foo/select?wt=json&rows=2&group=true&group.field=dcterms_source&group.ngroups=true&hl=true

Or this:

http://localhost:8983/solr/foo/select?wt=json&rows=2&shards=localhost:8983/solr/foo,localhost:8983/solr/bar&hl=true

But not this:

http://localhost:8983/solr/foo/select?wt=json&rows=2&group=true&group.field=dcterms_source&group.ngroups=true&shards=localhost:8983/solr/foo,localhost:8983/solr/bar&hl=true

Stacktrace:

SEVERE: java.lang.NullPointerException
	at org.apache.solr.handler.component.HighlightComponent.finishStage(HighlightComponent.java:156)
	at org.apache.solr.handler.component.SearchHandler.handleRequestBody(SearchHandler.java:298)
	at org.apache.solr.handler.RequestHandlerBase.handleRequest(RequestHandlerBase.java:129)
	at org.apache.solr.core.SolrCore.execute(SolrCore.java:1407)
	at org.apache.solr.servlet.SolrDispatchFilter.execute(SolrDispatchFilter.java:353)
	at org.apache.solr.servlet.SolrDispatchFilter.doFilter(SolrDispatchFilter.java:248)
	at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:235)
	at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206)
	at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:233)
	at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:191)
	at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:127)
	at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:102)
	at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:109)
	at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:298)
	at org.apache.coyote.http11.Http11Processor.process(Http11Processor.java:859)
	at org.apache.coyote.http11.Http11Protocol$Http11ConnectionHandler.process(Http11Protocol.java:588)
	at org.apache.tomcat.util.net.JIoEndpoint$Worker.run(JIoEndpoint.java:489)
	at java.lang.Thread.run(Thread.java:680)

Good catch Jasper! I've updated the patch that fixes this issue. I also included highlighting + distributed grouping to the tests.

Edit: It is working once there's data in the index!

It was giving me the following exception on initialization when the collection was still empty, rather:

java.lang.NullPointerException at org.apache.solr.search.grouping.firstphase.SearchGroupShardResponseProcessor.process(SearchGroupShardResponseProcessor.java:88)
  at org.apache.solr.handler.component.QueryComponent.handleGroupedResponses(QueryComponent.java:679)
  at org.apache.solr.handler.component.QueryComponent.handleResponses(QueryComponent.java:662)
  at org.apache.solr.handler.component.SearchHandler.handleRequestBody(SearchHandler.java:292)
  at org.apache.solr.handler.RequestHandlerBase.handleRequest(RequestHandlerBase.java:129)
  at org.apache.solr.core.SolrCore.execute(SolrCore.java:1407)
  at org.apache.solr.servlet.SolrDispatchFilter.execute(SolrDispatchFilter.java:353)
  at org.apache.solr.servlet.SolrDispatchFilter.doFilter(SolrDispatchFilter.java:248)
  at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:235)
  at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206)
  at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:233)
  at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:191)
  at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:127)
  at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:102)
  at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:109)
  at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:298)
  at org.apache.coyote.http11.Http11Processor.process(Http11Processor.java:859)
  at org.apache.coyote.http11.Http11Protocol$Http11ConnectionHandler.process(Http11Protocol.java:588)
  at org.apache.tomcat.util.net.JIoEndpoint$Worker.run(JIoEndpoint.java:489)
  at java.lang.Thread.run(Thread.java:680)

Thanks for testing the patch Jasper! I've updated the patch that prevents the NPE when querying empty shards and returns an empty response instead.

You're more than welcome! Having distributed grouping will be a great addition to have. As for the patch, could it be that you've modified a previous version rather than the latest one that includes the highlighting fixes? I'm getting the same highlighting-related stacktrace as before.

SEVERE: java.lang.NullPointerException
	at org.apache.solr.handler.component.HighlightComponent.finishStage(HighlightComponent.java:156)
	at org.apache.solr.handler.component.SearchHandler.handleRequestBody(SearchHandler.java:298)
	at org.apache.solr.handler.RequestHandlerBase.handleRequest(RequestHandlerBase.java:129)
	at org.apache.solr.core.SolrCore.execute(SolrCore.java:1407)
	at org.apache.solr.servlet.SolrDispatchFilter.execute(SolrDispatchFilter.java:353)
	at org.apache.solr.servlet.SolrDispatchFilter.doFilter(SolrDispatchFilter.java:248)
	at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:235)
	at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206)
	at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:233)
	at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:191)
	at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:127)
	at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:102)
	at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:109)
	at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:298)
	at org.apache.coyote.http11.Http11Processor.process(Http11Processor.java:859)
	at org.apache.coyote.http11.Http11Protocol$Http11ConnectionHandler.process(Http11Protocol.java:588)
	at org.apache.tomcat.util.net.JIoEndpoint$Worker.run(JIoEndpoint.java:489)
	at java.lang.Thread.run(Thread.java:680)

Jasper, does the exception occur for the same queries? I did add a test for this. Can you run the TestDistributedSearch test?

Updated patch

• group.query works in distributed search
• group.main works in distributed search
• Many refactorings

I think the feature needs to be committed. Maybe besides some jdocs the patch is ready. I'll commit this feature in the coming days. In the mean time I will start working on the patch for the 3x branch.

Martijn, the problem seems to be with highlighting combined with empty result sets. When I modify the TestDistributedGrouping test as follows:

// Test distributed grouping with empty indices
query("q", "*:*", "rows", 100, "fl", "id," + i1, "group", "true", "group.field", i1, "group.limit", 10, "sort", i1 + " asc, id asc");
query("q", "*:*", "rows", 100, "fl", "id," + i1, "group", "true", "group.field", i1, "group.limit", 10, "sort", i1 + " asc, id asc", "hl","true","hl.fl",t1);

I can reproduce the exact stacktrace. The exception doesn't occur with a populated index.

Thanks Jasper! I added your test case and I also added a bug fix for it. So it shouldn't occur any more!

Apologies for continuing to bring up issues but it seems that facets also cause problems with empty result sets. Modify the TestDistributedGrouping test as follows and you'll see:

// Test distributed grouping with empty indices
query("q", "*:*", "rows", 100, "fl", "id," + i1, "group", "true", "group.field", i1, "group.limit", 10, "sort", i1 + " asc, id asc");
query("q", "*:*", "rows", 100, "fl", "id," + i1, "group", "true", "group.field", i1, "group.limit", 10, "sort", i1 + " asc, id asc", "hl","true","hl.fl",t1);
query("q", "*:*", "rows", 100, "fl", "id," + i1, "group", "true", "group.field", i1, "group.limit", 10, "sort", i1 + " asc, id asc", "facet", "true", "facet.field", t1);

Stacktrace:

    [junit] Caused by: org.apache.solr.common.SolrException: null  java.lang.NullPointerException
 	at org.apache.solr.handler.component.QueryComponent.process(QueryComponent.java:409)
 	at org.apache.solr.handler.component.SearchHandler.handleRequestBody(SearchHandler.java:194)
 	at org.apache.solr.handler.RequestHandlerBase.handleRequest(RequestHandlerBase.java:129)
 	at org.apache.solr.core.SolrCore.execute(SolrCore.java:1407)
 	at org.apache.solr.servlet.SolrDispatchFilter.execute(SolrDispatchFilter.java:353)
 	at org.apache.solr.servlet.SolrDispatchFilter.doFilter(SolrDispatchFilter.java:248)
 	at org.mortbay.jetty.servlet.ServletHandler$CachedChain.doFilter(ServletHandler.java:1212)
 	at org.mortbay.jetty.servlet.ServletHandler.handle(ServletHandler.java:399)
 	at org.mortbay.jetty.servlet.SessionHandler.handle(SessionHandler.java:182)
 	at org.mortbay.jetty.handler.ContextHandler.handle(ContextHandler.java:766)
 	at org.mortbay.jetty.handler.HandlerWrapper.handle(HandlerWrapper.java:152)
 	at org.mortbay.jetty.Server.handle(Server.java:326)
 	at org.mortbay.jetty.HttpConnection.handleRequest(HttpConnection.java:542)
 	at org.mortbay.jetty.HttpConnection$RequestHandler.content(HttpConnection.java:945)
 	at org.mortbay.jetty.HttpParser.parseNext(HttpParser.java:756)
 	at org.mortbay.jetty.HttpParser.parseAvailable(HttpParser.java:218)
 	at org.mortbay.jetty.HttpConnection.handle(HttpConnection.java:404)
 	at org.mortbay.jetty.bio.SocketConnector$Connection.run(SocketConnector.java:228)
 	at org.mortbay.thread.QueuedThreadPool$PoolThread.run(QueuedThreadPool.java:582)  null  java.lang.NullPointerException
 	at org.apache.solr.handler.component.QueryComponent.process(QueryComponent.java:409)
 	at org.apache.solr.handler.component.SearchHandler.handleRequestBody(SearchHandler.java:194)
 	at org.apache.solr.handler.RequestHandlerBase.handleRequest(RequestHandlerBase.java:129)
 	at org.apache.solr.core.SolrCore.execute(SolrCore.java:1407)
 	at org.apache.solr.servlet.SolrDispatchFilter.execute(SolrDispatchFilter.java:353)
 	at org.apache.solr.servlet.SolrDispatchFilter.doFilter(SolrDispatchFilter.

NPE and empty results! Thanks for noticing this. Better now then when distributed grouping is released!

I took a quick peek at this, and I see some changes to how distrib search works (ShardRequestFactory).
Could you give a brief explanation about the need for that and how it works? Maybe changes like this should be in their own issue so it's easy to tell other refactoring vs what's needed just for grouping.

Martijn, would group.truncate be something that can also be added in with the same restrictions as group.ngroups (documents within the same group must reside on the same shard)? Would this be particularly tricky to add to the patch? It would definitely be nice to have, as people seem to rather care a lot about exact facet counts.

Martijn, would group.truncate be something that can also be added in with the same restrictions as group.ngroups (documents within the same group must reside on the same shard)? Would this be particularly tricky to add to the patch? It would definitely be nice to have, as people seem to rather care a lot about exact facet counts.

No that is not that difficult. I was planning on adding it anyway.

I took a quick peek at this, and I see some changes to how distrib search works (ShardRequestFactory).
Could you give a brief explanation about the need for that and how it works? Maybe changes like this should be in their own issue so it's easy to tell other refactoring vs what's needed just for grouping.

Sure! In the initial versions of the patch most of the code was in the QueryComponent. The class became very large and I found it very difficult to change / maintain. So I decided to move responsibilities to behind interfaces and have different concrete implementations for different behavior. During coding I also noticed that a few concepts (like ShardRequestFactory, ShardResponseProcessor and ShardResultTransformer) are very useable for non grouping distributed search. I chose not to refactor the current regular distributed grouping, because otherwise this patch becomes to large. I think after this is committed that we should refactor the non grouped distributed search to use these concepts or similar.
Currently the regular distributed search implementation hasn't changed.

Concepts that can also be used for non grouped distributed searches:

• ShardRequestFactory is responsible for creating requests to shards in the cluster based on the incoming request from the client.
• ShardResultTransformer. Transforming a NamedList response from the client in for example SearchGroup or TopDocs instance.
• ShardResponseProcessor. Basically merges the shard responses. The ShardReponseProcessor uses a ShardResultTransformer to transform the shard response into a native structure (SearchGroup / TopGroups).

Grouping related concepts:

• EndResultTransformer. Basically form the end result structure. There're three implementations for grouped, simple and main results.
• Command. Same responsibility as in Grouping.Command. Currently more focussed on distributed search.

+1 on "Concepts that can also be used for non grouped distributed searches" in a separate issue. The Solr distributed search code is overly complicated.

Attached a patch to support distributed grouping for 3x. Changes between trunk are minimal. The most obvious difference is that this patch uses String for groups whereis the trunk uses BytesRef.

I chose not to refactor the current regular distributed grouping, because otherwise this patch becomes to large.

Ah, ok. So this currently just looks like an implementation detail for QueryComponent (and only used when grouping now), and not some change to the distrib search architecture.

Yes it is an implementation detail of the QueryComponent for grouping. But I think we should open a new issue and look what we can refactor, based on the concepts in this issue.

Committed.
trunk: 1171970
3x branch: 1171968

Bulk close after 3.5 is released