Index: lucene/CHANGES.txt
===================================================================
--- lucene/CHANGES.txt	(revision 1214418)
+++ lucene/CHANGES.txt	(working copy)
@@ -751,6 +751,12 @@
   where they would create invalid offsets in some situations, leading to problems
   in highlighting. (Max Beutel via Robert Muir)
 
+* LUCENE-3639: TopDocs.merge was incorrectly setting TopDocs.maxScore to
+  Float.MIN_VALUE when it should be Float.NaN, when there were 0
+  hits.  Improved age calculation in SearcherLifetimeManager, to have
+  double precision and to compute age to be how long ago the searcher
+  was replaced with a new searcher (Mike McCandless)
+
 Documentation
 
 * LUCENE-3597: Fixed incorrect grouping documentation. (Martijn van Groningen, Robert Muir)
Index: lucene/src/test/org/apache/lucene/search/TestShardSearching.java
===================================================================
--- lucene/src/test/org/apache/lucene/search/TestShardSearching.java	(revision 0)
+++ lucene/src/test/org/apache/lucene/search/TestShardSearching.java	(working copy)
@@ -0,0 +1,393 @@
+package org.apache.lucene.search;
+
+/**
+ * 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.
+ */
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.MultiFields;
+import org.apache.lucene.index.MultiReader;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.index.TermsEnum;
+import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.LuceneTestCase.UseNoMemoryExpensiveCodec;
+import org.apache.lucene.util._TestUtil;
+
+// TODO
+//   - other queries besides PrefixQuery & TermQuery (but:
+//     FuzzyQ will be problematic... the top N terms it
+//     takes means results will differ)
+//   - NRQ/F
+//   - BQ, negated clauses, negated prefix clauses
+//   - test pulling docs in 2nd round trip...
+//   - filter too
+
+@UseNoMemoryExpensiveCodec
+public class TestShardSearching extends ShardSearchingTestBase {
+
+  private static class PreviousSearchState {
+    public final long searchTimeNanos;
+    public final long[] versions;
+    public final ScoreDoc searchAfterLocal;
+    public final ScoreDoc searchAfterShard;
+    public final Sort sort;
+    public final Query query;
+    public final int numHitsPaged;
+
+    public PreviousSearchState(Query query, Sort sort, ScoreDoc searchAfterLocal, ScoreDoc searchAfterShard, long[] versions, int numHitsPaged) {
+      this.versions = versions.clone();
+      this.searchAfterLocal = searchAfterLocal;
+      this.searchAfterShard = searchAfterShard;
+      this.sort = sort;
+      this.query = query;
+      this.numHitsPaged = numHitsPaged;
+      searchTimeNanos = System.nanoTime();
+    }
+  }
+
+  public void testSimple() throws Exception {
+     final int numNodes = _TestUtil.nextInt(random, 1, 10);
+
+    final double runTimeSec = 5.0 * RANDOM_MULTIPLIER;
+
+    final int minDocsToMakeTerms = _TestUtil.nextInt(random, 5, 20);
+
+    final int maxSearcherAgeSeconds = _TestUtil.nextInt(random, 1, 4);
+
+    if (VERBOSE) {
+      System.out.println("TEST: numNodes=" + numNodes + " runTimeSec=" + runTimeSec + " maxSearcherAgeSeconds=" + maxSearcherAgeSeconds);
+    }
+
+    start(_TestUtil.getTempDir("TestShardSearching").toString(),
+          numNodes,
+          runTimeSec,
+          maxSearcherAgeSeconds
+          );
+
+    final List<PreviousSearchState> priorSearches = new ArrayList<PreviousSearchState>();
+    List<BytesRef> terms = null;
+    while (System.nanoTime() < endTimeNanos) {
+
+      final boolean doFollowon = priorSearches.size() > 0 && random.nextInt(7) == 1;
+
+      // Pick a random node; we will run the query on this node:
+      final int myNodeID = random.nextInt(numNodes);
+
+      final NodeState.ShardIndexSearcher localShardSearcher;
+
+      final PreviousSearchState prevSearchState;
+
+      if (doFollowon) {
+        // Pretend user issued a followon query:
+        prevSearchState = priorSearches.get(random.nextInt(priorSearches.size()));
+
+        if (VERBOSE) {
+          System.out.println("\nTEST: follow-on query age=" + ((System.nanoTime() - prevSearchState.searchTimeNanos)/1000000000.0));
+        }
+
+        try {
+          localShardSearcher = nodes[myNodeID].acquire(prevSearchState.versions);
+        } catch (SearcherExpiredException see) {
+          // Expected, sometimes; in a "real" app we would
+          // either forward this error to the user ("too
+          // much time has passed; please re-run your
+          // search") or sneakily just switch to newest
+          // searcher w/o telling them...
+          if (VERBOSE) {
+            System.out.println("  searcher expired during local shard searcher init: " + see);
+          }
+          priorSearches.remove(prevSearchState);
+          continue;
+        }
+      } else {
+        if (VERBOSE) {
+          System.out.println("\nTEST: fresh query");
+        }
+        // Do fresh query:
+        localShardSearcher = nodes[myNodeID].acquire();
+        prevSearchState = null;
+      }
+
+      final IndexReader[] subs = new IndexReader[numNodes];
+
+      PreviousSearchState searchState = null;
+
+      try {
+
+        // Mock: now make a single reader (MultiReader) from all node
+        // searchers.  In a real shard env you can't do this... we
+        // do it to confirm results from the shard searcher
+        // are correct:
+        int docCount = 0;
+        try {
+          for(int nodeID=0;nodeID<numNodes;nodeID++) {
+            final long subVersion = localShardSearcher.nodeVersions[nodeID];
+            final IndexSearcher sub = nodes[nodeID].searchers.acquire(subVersion);
+            if (sub == null) {
+              nodeID--;
+              while(nodeID >= 0) {
+                subs[nodeID].decRef();
+                subs[nodeID] = null;
+                nodeID--;
+              }
+              throw new SearcherExpiredException("nodeID=" + nodeID + " version=" + subVersion);
+            }
+            subs[nodeID] = sub.getIndexReader();
+            docCount += subs[nodeID].maxDoc();
+          }
+        } catch (SearcherExpiredException see) {
+          // Expected
+          if (VERBOSE) {
+            System.out.println("  searcher expired during mock reader init: " + see);
+          }
+          continue;
+        }
+
+        final IndexReader mockReader = new MultiReader(subs);
+        final IndexSearcher mockSearcher = new IndexSearcher(mockReader);
+
+        Query query;
+        Sort sort;
+
+        if (prevSearchState != null) {
+          query = prevSearchState.query;
+          sort = prevSearchState.sort;
+        } else {
+          if (terms == null && docCount > minDocsToMakeTerms) {
+            // TODO: try to "focus" on high freq terms sometimes too
+            // TODO: maybe also periodically reset the terms...?
+            final TermsEnum termsEnum = MultiFields.getTerms(mockReader, "body").iterator(null);
+            terms = new ArrayList<BytesRef>();
+            while(termsEnum.next() != null) {
+              terms.add(BytesRef.deepCopyOf(termsEnum.term()));
+            }
+            if (VERBOSE) {
+              System.out.println("TEST: init terms: " + terms.size() + " terms");
+            }
+            if (terms.size() == 0) {
+              terms = null;
+            }
+          }
+
+          if (VERBOSE) {
+            System.out.println("  maxDoc=" + mockReader.maxDoc());
+          }
+
+          if (terms != null) {
+            if (random.nextBoolean()) {
+              query = new TermQuery(new Term("body", terms.get(random.nextInt(terms.size()))));
+            } else {
+              final String t = terms.get(random.nextInt(terms.size())).utf8ToString();
+              final String prefix;
+              if (t.length() <= 1) {
+                prefix = t;
+              } else {
+                prefix = t.substring(0, _TestUtil.nextInt(random, 1, 2));
+              }
+              query = new PrefixQuery(new Term("body", prefix));
+            }
+            
+            if (random.nextBoolean()) {
+              sort = null;
+            } else {
+              // TODO: sort by more than 1 field
+              final int what = random.nextInt(3);
+              if (what == 0) {
+                sort = new Sort(SortField.FIELD_SCORE);
+              } else if (what == 1) {
+                // TODO: this sort doesn't merge
+                // correctly... it's tricky because you
+                // could have > 2.1B docs across all shards: 
+                //sort = new Sort(SortField.FIELD_DOC);
+                sort = null;
+              } else if (what == 2) {
+                sort = new Sort(new SortField[] {new SortField("docid", SortField.Type.INT, random.nextBoolean())});
+              } else {
+                sort = new Sort(new SortField[] {new SortField("title", SortField.Type.STRING, random.nextBoolean())});
+              }
+            }
+          } else {
+            query = null;
+            sort = null;
+          }
+        }
+
+        if (query != null) {
+
+          try {
+            searchState = assertSame(mockSearcher, localShardSearcher, query, sort, prevSearchState);
+          } catch (SearcherExpiredException see) {
+            // Expected; in a "real" app we would
+            // either forward this error to the user ("too
+            // much time has passed; please re-run your
+            // search") or sneakily just switch to newest
+            // searcher w/o telling them...
+            if (VERBOSE) {
+              System.out.println("  searcher expired during search: " + see);
+              see.printStackTrace(System.out);
+            }
+            assert prevSearchState != null;
+            priorSearches.remove(prevSearchState);
+          }
+        }
+      } finally {
+        nodes[myNodeID].release(localShardSearcher);
+        for(IndexReader sub : subs) {
+          if (sub != null) {
+            sub.decRef();
+          }
+        }
+      }
+
+      if (searchState != null && searchState.searchAfterLocal != null && random.nextInt(5) == 3) {
+        priorSearches.add(searchState);
+        if (priorSearches.size() > 200) {
+          Collections.shuffle(priorSearches, random);
+          priorSearches.subList(100, priorSearches.size()).clear();
+        }
+      }
+    }
+
+    finish();
+  }
+
+  private PreviousSearchState assertSame(IndexSearcher mockSearcher, NodeState.ShardIndexSearcher shardSearcher, Query q, Sort sort, PreviousSearchState state) throws IOException {
+
+    int numHits = _TestUtil.nextInt(random, 1, 100);
+    if (state != null && state.searchAfterLocal == null) {
+      // In addition to what we last searched:
+      numHits += state.numHitsPaged;
+    }
+
+    if (VERBOSE) {
+      System.out.println("TEST: query=" + q + " sort=" + sort + " numHits=" + numHits);
+      if (state != null) {
+        System.out.println("  prev: searchAfterLocal=" + state.searchAfterLocal + " searchAfterShard=" + state.searchAfterShard + " numHitsPaged=" + state.numHitsPaged);
+      }
+    }
+
+    // Single (mock local) searcher:
+    final TopDocs hits;
+    if (sort == null) {
+      if (state != null && state.searchAfterLocal != null) {
+        hits = mockSearcher.searchAfter(state.searchAfterLocal, q, numHits);
+      } else {
+        hits = mockSearcher.search(q, numHits);
+      }
+    } else {
+      hits = mockSearcher.search(q, numHits, sort);
+    }
+
+    // Shard searcher
+    final TopDocs shardHits;
+    if (sort == null) {
+      if (state != null && state.searchAfterShard != null) {
+        shardHits = shardSearcher.searchAfter(state.searchAfterShard, q, numHits);
+      } else {
+        shardHits = shardSearcher.search(q, numHits);
+      }
+    } else {
+      shardHits = shardSearcher.search(q, numHits, sort);
+    }
+
+    final int numNodes = shardSearcher.nodeVersions.length;
+    int[] base = new int[numNodes];
+    final IndexReader[] subs = mockSearcher.getIndexReader().getSequentialSubReaders();
+    assertEquals(numNodes, subs.length);
+
+    int docCount = 0;
+    for(int nodeID=0;nodeID<numNodes;nodeID++) {
+      base[nodeID] = docCount;
+      docCount += subs[nodeID].maxDoc();
+    }
+
+    if (VERBOSE) {
+      /*
+      for(int shardID=0;shardID<shardSearchers.length;shardID++) {
+        System.out.println("  shard=" + shardID + " maxDoc=" + shardSearchers[shardID].searcher.getIndexReader().maxDoc());
+      }
+      */
+      System.out.println("  single searcher: " + hits.totalHits + " totalHits maxScore=" + hits.getMaxScore());
+      for(int i=0;i<hits.scoreDocs.length;i++) {
+        final ScoreDoc sd = hits.scoreDocs[i];
+        System.out.println("    doc=" + sd.doc + " score=" + sd.score);
+      }
+      System.out.println("  shard searcher: " + shardHits.totalHits + " totalHits maxScore=" + shardHits.getMaxScore());
+      for(int i=0;i<shardHits.scoreDocs.length;i++) {
+        final ScoreDoc sd = shardHits.scoreDocs[i];
+        System.out.println("    doc=" + sd.doc + " (rebased: " + (sd.doc + base[sd.shardIndex]) + ") score=" + sd.score + " shard=" + sd.shardIndex);
+      }
+    }
+
+    int numHitsPaged;
+    if (state != null && state.searchAfterLocal != null) {
+      numHitsPaged = hits.scoreDocs.length; 
+      if (state != null) {
+        numHitsPaged += state.numHitsPaged;
+      }
+    } else {
+      numHitsPaged = hits.scoreDocs.length;
+    }
+
+    final boolean moreHits;
+
+    final ScoreDoc bottomHit;
+    final ScoreDoc bottomHitShards;
+
+    if (numHitsPaged < hits.totalHits) {
+      // More hits to page through
+      moreHits = true;
+      if (sort == null) {
+        bottomHit = hits.scoreDocs[hits.scoreDocs.length-1];
+        final ScoreDoc sd = shardHits.scoreDocs[shardHits.scoreDocs.length-1];
+        // Must copy because below we rebase:
+        bottomHitShards = new ScoreDoc(sd.doc, sd.score, sd.shardIndex);
+        if (VERBOSE) {
+          System.out.println("  save bottomHit=" + bottomHit);
+        }
+      } else {
+        bottomHit = null;
+        bottomHitShards = null;
+      }
+
+    } else {
+      assertEquals(hits.totalHits, numHitsPaged);
+      bottomHit = null;
+      bottomHitShards = null;
+      moreHits = false;
+    }
+
+    // Must rebase so assertEquals passes:
+    for(int hitID=0;hitID<shardHits.scoreDocs.length;hitID++) {
+      final ScoreDoc sd = shardHits.scoreDocs[hitID];
+      sd.doc += base[sd.shardIndex];
+    }
+
+    _TestUtil.assertEquals(hits, shardHits);
+
+    if (moreHits) {
+      // Return a continuation:
+      return new PreviousSearchState(q, sort, bottomHit, bottomHitShards, shardSearcher.nodeVersions, numHitsPaged);
+    } else {
+      return null;
+    }
+  }
+}

Property changes on: lucene/src/test/org/apache/lucene/search/TestShardSearching.java
___________________________________________________________________
Added: svn:eol-style
## -0,0 +1 ##
+native
Index: lucene/src/java/org/apache/lucene/search/SearcherLifetimeManager.java
===================================================================
--- lucene/src/java/org/apache/lucene/search/SearcherLifetimeManager.java	(revision 1214418)
+++ lucene/src/java/org/apache/lucene/search/SearcherLifetimeManager.java	(working copy)
@@ -23,7 +23,6 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.TimeUnit;
 
 import org.apache.lucene.search.NRTManager;        // javadocs
 import org.apache.lucene.index.IndexReader;        // javadocs
@@ -101,9 +100,11 @@
 
 public class SearcherLifetimeManager implements Closeable {
 
+  static final double NANOS_PER_SEC = 1000000000.0;
+
   private static class SearcherTracker implements Comparable<SearcherTracker>, Closeable {
     public final IndexSearcher searcher;
-    public final long recordTimeSec;
+    public final double recordTimeSec;
     public final long version;
 
     public SearcherTracker(IndexSearcher searcher) {
@@ -112,7 +113,7 @@
       searcher.getIndexReader().incRef();
       // Use nanoTime not currentTimeMillis since it [in
       // theory] reduces risk from clock shift
-      recordTimeSec = TimeUnit.NANOSECONDS.toSeconds(System.nanoTime());
+      recordTimeSec = System.nanoTime() / NANOS_PER_SEC;
     }
 
     // Newer searchers are sort before older ones:
@@ -170,6 +171,7 @@
     final long version = searcher.getIndexReader().getVersion();
     SearcherTracker tracker = searchers.get(version);
     if (tracker == null) {
+      //System.out.println("RECORD version=" + version + " ms=" + System.currentTimeMillis());
       tracker = new SearcherTracker(searcher);
       if (searchers.putIfAbsent(version, tracker) != null) {
         // Another thread beat us -- must decRef to undo
@@ -217,29 +219,28 @@
   /** See {@link #prune}. */
   public interface Pruner {
     /** Return true if this searcher should be removed. 
-     *  @param ageSec how long ago this searcher was
-     *         recorded vs the most recently recorded
-     *         searcher
+     *  @param ageSec how much time has passed since this
+     *         searcher was the current (live) searcher
      *  @param searcher Searcher
      **/
-    public boolean doPrune(int ageSec, IndexSearcher searcher);
+    public boolean doPrune(double ageSec, IndexSearcher searcher);
   }
 
   /** Simple pruner that drops any searcher older by
    *  more than the specified seconds, than the newest
    *  searcher. */
   public final static class PruneByAge implements Pruner {
-    private final int maxAgeSec;
+    private final double maxAgeSec;
 
-    public PruneByAge(int maxAgeSec) {
-      if (maxAgeSec < 1) {
+    public PruneByAge(double maxAgeSec) {
+      if (maxAgeSec < 0) {
         throw new IllegalArgumentException("maxAgeSec must be > 0 (got " + maxAgeSec + ")");
       }
       this.maxAgeSec = maxAgeSec;
     }
 
     @Override
-    public boolean doPrune(int ageSec, IndexSearcher searcher) {
+    public boolean doPrune(double ageSec, IndexSearcher searcher) {
       return ageSec > maxAgeSec;
     }
   }
@@ -261,14 +262,25 @@
       trackers.add(tracker);
     }
     Collections.sort(trackers);
-    final long newestSec = trackers.isEmpty() ? 0L : trackers.get(0).recordTimeSec;
+    double lastRecordTimeSec = 0.0;
+    final double now = System.nanoTime()/NANOS_PER_SEC;
     for (SearcherTracker tracker: trackers) {
-      final int ageSec = (int) (newestSec - tracker.recordTimeSec);
-      assert ageSec >= 0;
+      final double ageSec;
+      if (lastRecordTimeSec == 0.0) {
+        ageSec = 0.0;
+      } else {
+        ageSec = now - lastRecordTimeSec;
+      }
+      // First tracker is always age 0.0 sec, since it's
+      // still "live"; second tracker's age (= seconds since
+      // it was "live") is now minus first tracker's
+      // recordTime, etc:
       if (pruner.doPrune(ageSec, tracker.searcher)) {
+        //System.out.println("PRUNE version=" + tracker.version + " age=" + ageSec + " ms=" + System.currentTimeMillis());
         searchers.remove(tracker.version);
         tracker.close();
       }
+      lastRecordTimeSec = tracker.recordTimeSec;
     }
   }
 
Index: lucene/src/java/org/apache/lucene/search/TopDocs.java
===================================================================
--- lucene/src/java/org/apache/lucene/search/TopDocs.java	(revision 1214418)
+++ lucene/src/java/org/apache/lucene/search/TopDocs.java	(working copy)
@@ -216,8 +216,10 @@
     float maxScore = Float.MIN_VALUE;
     for(int shardIDX=0;shardIDX<shardHits.length;shardIDX++) {
       final TopDocs shard = shardHits[shardIDX];
+      // totalHits can be non-zero even if no hits were
+      // collected, when searchAfter was used:
+      totalHitCount += shard.totalHits;
       if (shard.scoreDocs != null && shard.scoreDocs.length > 0) {
-        totalHitCount += shard.totalHits;
         availHitCount += shard.scoreDocs.length;
         queue.add(new ShardRef(shardIDX));
         maxScore = Math.max(maxScore, shard.getMaxScore());
@@ -225,6 +227,10 @@
       }
     }
 
+    if (availHitCount == 0) {
+      maxScore = Float.NaN;
+    }
+
     final ScoreDoc[] hits = new ScoreDoc[Math.min(topN, availHitCount)];
 
     int hitUpto = 0;
Index: lucene/src/java/org/apache/lucene/search/ScoreDoc.java
===================================================================
--- lucene/src/java/org/apache/lucene/search/ScoreDoc.java	(revision 1214418)
+++ lucene/src/java/org/apache/lucene/search/ScoreDoc.java	(working copy)
@@ -46,6 +46,6 @@
   // A convenience method for debugging.
   @Override
   public String toString() {
-    return "doc=" + doc + " score=" + score;
+    return "doc=" + doc + " score=" + score + " shardIndex=" + shardIndex;
   }
 }
Index: lucene/src/test-framework/java/org/apache/lucene/search/ShardSearchingTestBase.java
===================================================================
--- lucene/src/test-framework/java/org/apache/lucene/search/ShardSearchingTestBase.java	(revision 0)
+++ lucene/src/test-framework/java/org/apache/lucene/search/ShardSearchingTestBase.java	(working copy)
@@ -0,0 +1,575 @@
+package org.apache.lucene.search;
+
+/**
+ * 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.
+ */
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriterConfig;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.analysis.MockAnalyzer;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.util.LineFileDocs;
+import org.apache.lucene.util.LuceneTestCase;
+import org.apache.lucene.util.TermContext;
+
+// TODO
+//   - doc blocks?  so we can test joins/grouping...
+//   - controlled consistency (NRTMgr)
+
+public abstract class ShardSearchingTestBase extends LuceneTestCase {
+
+  // TODO: maybe SLM should throw this instead of returning null...
+  public static class SearcherExpiredException extends RuntimeException {
+    public SearcherExpiredException(String message) {
+      super(message);
+    }
+  }
+
+  private static class FieldAndShardVersion {
+    private final long version;
+    private final int nodeID;
+    private final String field;
+
+    public FieldAndShardVersion(int nodeID, long version, String field) {
+      this.nodeID = nodeID;
+      this.version = version;
+      this.field = field;
+    }
+
+    @Override
+    public int hashCode() {
+      return (int) (version * nodeID + field.hashCode());
+    }
+
+    @Override
+    public boolean equals(Object _other) {
+      if (!(_other instanceof FieldAndShardVersion)) {
+        return false;
+      }
+
+      final FieldAndShardVersion other = (FieldAndShardVersion) _other;
+
+      return field.equals(other.field) && version == other.version && nodeID == other.nodeID;
+    }
+
+    @Override
+    public String toString() {
+      return "FieldAndShardVersion(field=" + field + " nodeID=" + nodeID + " version=" + version+ ")";
+    }
+  }
+
+  private static class TermAndShardVersion {
+    private final long version;
+    private final int nodeID;
+    private final Term term;
+
+    public TermAndShardVersion(int nodeID, long version, Term term) {
+      this.nodeID = nodeID;
+      this.version = version;
+      this.term = term;
+    }
+
+    @Override
+    public int hashCode() {
+      return (int) (version * nodeID + term.hashCode());
+    }
+
+    @Override
+    public boolean equals(Object _other) {
+      if (!(_other instanceof TermAndShardVersion)) {
+        return false;
+      }
+
+      final TermAndShardVersion other = (TermAndShardVersion) _other;
+
+      return term.equals(other.term) && version == other.version && nodeID == other.nodeID;
+    }
+  }
+
+  // We share collection stats for these fields on each node
+  // reopen:
+  private final String[] fieldsToShare = new String[] {"body", "title"};
+
+  // Called by one node once it has reopened, to notify all
+  // other nodes.  This is just a mock (since it goes and
+  // directly updates all other nodes, in RAM)... in a real
+  // env this would hit the wire, sending version &
+  // collection stats to all other nodes:
+  void broadcastNodeReopen(int nodeID, long version, IndexSearcher newSearcher) throws IOException {
+
+    if (VERBOSE) {
+      System.out.println("REOPEN: nodeID=" + nodeID + " version=" + version + " maxDoc=" + newSearcher.getIndexReader().maxDoc());
+    }
+
+    // Broadcast new collection stats for this node to all
+    // other nodes:
+    for(String field : fieldsToShare) {
+      final CollectionStatistics stats = newSearcher.collectionStatistics(field);
+      for (NodeState node : nodes) {
+        // Don't put my own collection stats into the cache;
+        // we pull locally:
+        if (node.myNodeID != nodeID) {
+          node.collectionStatsCache.put(new FieldAndShardVersion(nodeID, version, field), stats);
+        }
+      }
+    }
+    for (NodeState node : nodes) {
+      node.updateNodeVersion(nodeID, version);
+    }
+  }
+
+  // MOCK: in a real env you have to hit the wire
+  // (send this query to all remote nodes
+  // concurrently):
+  TopDocs searchNode(int nodeID, long[] nodeVersions, Query q, Sort sort, int numHits, ScoreDoc searchAfter) throws IOException {
+    final NodeState.ShardIndexSearcher s = nodes[nodeID].acquire(nodeVersions);
+    try {
+      if (sort == null) {
+        if (searchAfter != null) {
+          return s.localSearchAfter(searchAfter, q, numHits);
+        } else {
+          return s.localSearch(q, numHits);
+        }
+      } else {
+        assert searchAfter == null;  // not supported yet
+        return s.localSearch(q, numHits, sort);
+      }
+    } finally {
+      nodes[nodeID].release(s);
+    }
+  }
+
+  // Mock: in a real env, this would hit the wire and get
+  // term stats from remote node
+  Map<Term,TermStatistics> getNodeTermStats(Set<Term> terms, int nodeID, long version) throws IOException {
+    final NodeState node = nodes[nodeID];
+    final Map<Term,TermStatistics> stats = new HashMap<Term,TermStatistics>();
+    final IndexSearcher s = node.searchers.acquire(version);
+    if (s == null) {
+      throw new SearcherExpiredException("node=" + nodeID + " version=" + version);
+    }
+    try {
+      for(Term term : terms) {
+        final TermContext termContext = TermContext.build(s.getIndexReader().getTopReaderContext(), term, false);
+        stats.put(term, s.termStatistics(term, termContext));
+      }
+    } finally {
+      node.searchers.release(s);
+    }
+    return stats;
+  }
+
+  protected final class NodeState implements Closeable {
+    public final Directory dir;
+    public final IndexWriter writer;
+    public final SearcherLifetimeManager searchers;
+    public final SearcherManager mgr;
+    public final int myNodeID;
+    public final long[] currentNodeVersions;
+
+    // TODO: nothing evicts from here!!!  Somehow, on searcher
+    // expiration on remote nodes we must evict from our
+    // local cache...?
+
+    private final Map<FieldAndShardVersion,CollectionStatistics> collectionStatsCache = new ConcurrentHashMap<FieldAndShardVersion,CollectionStatistics>();
+    private final Map<TermAndShardVersion,TermStatistics> termStatsCache = new ConcurrentHashMap<TermAndShardVersion,TermStatistics>();
+
+    /** Matches docs in the local shard but scores based on
+     *  aggregated stats ("mock distributed scoring") from all
+     *  nodes. */ 
+
+    public class ShardIndexSearcher extends IndexSearcher {
+      // Version for the node searchers we search:
+      public final long[] nodeVersions;
+      public final int myNodeID;
+
+      public ShardIndexSearcher(long[] nodeVersions, IndexReader localReader, int nodeID) {
+        super(localReader);
+        this.nodeVersions = nodeVersions;
+        myNodeID = nodeID;
+        assert myNodeID == NodeState.this.myNodeID: "myNodeID=" + nodeID + " NodeState.this.myNodeID=" + NodeState.this.myNodeID;
+      }
+
+      @Override
+      public Query rewrite(Query original) throws IOException {
+        final Query rewritten = super.rewrite(original);
+        final Set<Term> terms = new HashSet<Term>();
+        rewritten.extractTerms(terms);
+
+        // Make a single request to remote nodes for term
+        // stats:
+        for(int nodeID=0;nodeID<nodeVersions.length;nodeID++) {
+          if (nodeID == myNodeID) {
+            continue;
+          }
+
+          final Set<Term> missing = new HashSet<Term>();
+          for(Term term : terms) {
+            final TermAndShardVersion key = new TermAndShardVersion(nodeID, nodeVersions[nodeID], term);
+            if (!termStatsCache.containsKey(key)) {
+              missing.add(term);
+            }
+          }
+          if (missing.size() != 0) {
+            for(Map.Entry<Term,TermStatistics> ent : getNodeTermStats(missing, nodeID, nodeVersions[nodeID]).entrySet()) {
+              final TermAndShardVersion key = new TermAndShardVersion(nodeID, nodeVersions[nodeID], ent.getKey());
+              termStatsCache.put(key, ent.getValue());
+            }
+          }
+        }
+
+        return rewritten;
+      }
+
+      @Override
+      public TermStatistics termStatistics(Term term, TermContext context) throws IOException {
+        assert term != null;
+        int docFreq = 0;
+        long totalTermFreq = 0;
+        for(int nodeID=0;nodeID<nodeVersions.length;nodeID++) {
+
+          final TermStatistics subStats;
+          if (nodeID == myNodeID) {
+            subStats = super.termStatistics(term, context);
+          } else {
+            final TermAndShardVersion key = new TermAndShardVersion(nodeID, nodeVersions[nodeID], term);
+            subStats = termStatsCache.get(key);
+            // We pre-cached during rewrite so all terms
+            // better be here...
+            assert subStats != null;
+          }
+        
+          docFreq += subStats.docFreq();
+          totalTermFreq += subStats.totalTermFreq();
+        }
+
+        return new TermStatistics(term.bytes(), docFreq, totalTermFreq);
+      }
+
+      @Override
+      public CollectionStatistics collectionStatistics(String field) throws IOException {
+        // TODO: we could compute this on init and cache,
+        // since we are re-inited whenever any nodes have a
+        // new reader
+        int docCount = 0;
+        long sumTotalTermFreq = 0;
+        long sumDocFreq = 0;
+        int maxDoc = 0;
+
+        for(int nodeID=0;nodeID<nodeVersions.length;nodeID++) {
+          final FieldAndShardVersion key = new FieldAndShardVersion(nodeID, nodeVersions[nodeID], field);
+          final CollectionStatistics nodeStats;
+          if (nodeID == myNodeID) {
+            nodeStats = super.collectionStatistics(field);
+          } else {
+            nodeStats = collectionStatsCache.get(key);
+          }
+          if (nodeStats == null) {
+            System.out.println("coll stats myNodeID=" + myNodeID + ": " + collectionStatsCache.keySet());
+          }
+          // Collection stats are pre-shared on reopen, so,
+          // we better not have a cache miss:
+          assert nodeStats != null: "myNodeID=" + myNodeID + " nodeID=" + nodeID + " version=" + nodeVersions[nodeID] + " field=" + field;
+          docCount += nodeStats.docCount();
+          sumTotalTermFreq += nodeStats.sumTotalTermFreq();
+          sumDocFreq += nodeStats.sumDocFreq();
+          maxDoc += nodeStats.maxDoc();
+        }
+
+        return new CollectionStatistics(field, maxDoc, docCount, sumTotalTermFreq, sumDocFreq);
+      }
+
+      @Override
+      public TopDocs search(Query query, int numHits) throws IOException {
+        final TopDocs[] shardHits = new TopDocs[nodeVersions.length];
+        for(int nodeID=0;nodeID<nodeVersions.length;nodeID++) {
+          if (nodeID == myNodeID) {
+            // My node; run using local shard searcher we
+            // already aquired:
+            shardHits[nodeID] = localSearch(query, numHits);
+          } else {
+            shardHits[nodeID] = searchNode(nodeID, nodeVersions, query, null, numHits, null);
+          }
+        }
+
+        // Merge:
+        return TopDocs.merge(null, numHits, shardHits);
+      }
+
+      public TopDocs localSearch(Query query, int numHits) throws IOException {
+        return super.search(query, numHits);
+      }
+
+      @Override
+      public TopDocs searchAfter(ScoreDoc after, Query query, int numHits) throws IOException {
+        final TopDocs[] shardHits = new TopDocs[nodeVersions.length];
+        ScoreDoc shardAfter = new ScoreDoc(after.doc, after.score);
+        for(int nodeID=0;nodeID<nodeVersions.length;nodeID++) {
+          if (nodeID < after.shardIndex) {
+            // If score is tied then no docs in this shard
+            // should be collected:
+            shardAfter.doc = Integer.MAX_VALUE;
+          } else if (nodeID == after.shardIndex) {
+            // If score is tied then we break according to
+            // docID (like normal):  
+            shardAfter.doc = after.doc;
+          } else {
+            // If score is tied then all docs in this shard
+            // should be collected, because they come after
+            // the previous bottom:
+            shardAfter.doc = -1;
+          }
+          if (nodeID == myNodeID) {
+            // My node; run using local shard searcher we
+            // already aquired:
+            shardHits[nodeID] = localSearchAfter(shardAfter, query, numHits);
+          } else {
+            shardHits[nodeID] = searchNode(nodeID, nodeVersions, query, null, numHits, shardAfter);
+          }
+          //System.out.println("  node=" + nodeID + " totHits=" + shardHits[nodeID].totalHits);
+        }
+
+        // Merge:
+        return TopDocs.merge(null, numHits, shardHits);
+      }
+
+      public TopDocs localSearchAfter(ScoreDoc after, Query query, int numHits) throws IOException {
+        return super.searchAfter(after, query, numHits);
+      }
+
+      @Override
+      public TopFieldDocs search(Query query, int numHits, Sort sort) throws IOException {
+        assert sort != null;
+        final TopDocs[] shardHits = new TopDocs[nodeVersions.length];
+        for(int nodeID=0;nodeID<nodeVersions.length;nodeID++) {
+          if (nodeID == myNodeID) {
+            // My node; run using local shard searcher we
+            // already aquired:
+            shardHits[nodeID] = localSearch(query, numHits, sort);
+          } else {
+            shardHits[nodeID] = searchNode(nodeID, nodeVersions, query, sort, numHits, null);
+          }
+        }
+
+        // Merge:
+        return (TopFieldDocs) TopDocs.merge(sort, numHits, shardHits);
+      }
+
+      public TopFieldDocs localSearch(Query query, int numHits, Sort sort) throws IOException {
+        return super.search(query, numHits, sort);
+      }
+
+    }
+
+    private volatile ShardIndexSearcher currentShardSearcher;
+
+    public NodeState(Random random, String baseDir, int nodeID, int numNodes) throws IOException {
+      myNodeID = nodeID;
+      dir = newFSDirectory(new File(baseDir + "." + myNodeID));
+      // TODO: set warmer
+      writer = new IndexWriter(dir, new IndexWriterConfig(TEST_VERSION_CURRENT, new MockAnalyzer(random)));
+      mgr = new SearcherManager(writer, true, null);
+      searchers = new SearcherLifetimeManager();
+
+      // Init w/ 0s... caller above will do initial
+      // "broadcast" by calling initSearcher:
+      currentNodeVersions = new long[numNodes];
+    }
+
+    public void initSearcher(long[] nodeVersions) {
+      assert currentShardSearcher == null;
+      System.arraycopy(nodeVersions, 0, currentNodeVersions, 0, currentNodeVersions.length);
+      currentShardSearcher = new ShardIndexSearcher(currentNodeVersions.clone(),
+                                                    mgr.acquire().getIndexReader(),
+                                                    myNodeID);
+    }
+
+    public void updateNodeVersion(int nodeID, long version) throws IOException {
+      currentNodeVersions[nodeID] = version;
+      if (currentShardSearcher != null) {
+        currentShardSearcher.getIndexReader().decRef();
+      }
+      currentShardSearcher = new ShardIndexSearcher(currentNodeVersions.clone(), 
+                                                    mgr.acquire().getIndexReader(),
+                                                    myNodeID);
+    }
+
+    // Get the current (fresh) searcher for this node
+    public ShardIndexSearcher acquire() {
+      final ShardIndexSearcher s = currentShardSearcher;
+      // TODO: this isn't thread safe.... in theory the
+      // reader could get decRef'd to 0 before we have a
+      // chance to incRef, ie if a reopen happens right
+      // after the above line, this thread gets stalled, and
+      // the old IR is closed.  But because we use SLM in
+      // this test, this will be exceptionally rare:
+      s.getIndexReader().incRef();
+      return s;
+    }
+
+    public void release(ShardIndexSearcher s) throws IOException {
+      s.getIndexReader().decRef();
+    }
+
+    // Get and old searcher matching the specified versions:
+    public ShardIndexSearcher acquire(long[] nodeVersions) {
+      final IndexSearcher s = searchers.acquire(nodeVersions[myNodeID]);
+      if (s == null) {
+        throw new SearcherExpiredException("nodeID=" + myNodeID + " version=" + nodeVersions[myNodeID]);
+      }
+      return new ShardIndexSearcher(nodeVersions, s.getIndexReader(), myNodeID);
+    }
+
+    // Reopen local reader
+    public void reopen() throws IOException {
+      final IndexSearcher before = mgr.acquire();
+      mgr.release(before);
+
+      mgr.maybeReopen();
+      final IndexSearcher after = mgr.acquire();
+      try {
+        if (after != before) {
+          // New searcher was opened
+          final long version = searchers.record(after);
+          searchers.prune(new SearcherLifetimeManager.PruneByAge(maxSearcherAgeSeconds));
+          broadcastNodeReopen(myNodeID, version, after);
+        }
+      } finally {
+        mgr.release(after);
+      }
+    }
+
+    @Override
+    public void close() throws IOException {
+      if (currentShardSearcher != null) {
+        currentShardSearcher.getIndexReader().decRef();
+      }
+      searchers.close();
+      mgr.close();
+      writer.close();
+      dir.close();
+    }
+  }
+
+  // TODO: make this more realistic, ie, each node should
+  // have its own thread, so we have true node to node
+  // concurrency
+  private final class ChangeIndices extends Thread {
+    @Override
+    public void run() {
+      try {
+        final LineFileDocs docs = new LineFileDocs(random);
+        int numDocs = 0;
+        while (System.nanoTime() < endTimeNanos) {
+          final int what = random.nextInt(3);
+          final NodeState node = nodes[random.nextInt(nodes.length)];
+          if (numDocs == 0 || what == 0) {
+            node.writer.addDocument(docs.nextDoc());
+            numDocs++;
+          } else if (what == 1) {
+            node.writer.updateDocument(new Term("docid", ""+random.nextInt(numDocs)),
+                                        docs.nextDoc());
+            numDocs++;
+          } else {
+            node.writer.deleteDocuments(new Term("docid", ""+random.nextInt(numDocs)));
+          }
+          // TODO: doc blocks too
+
+          if (random.nextInt(17) == 12) {
+            node.writer.commit();
+          }
+
+          if (random.nextInt(17) == 12) {
+            nodes[random.nextInt(nodes.length)].reopen();
+          }
+        }
+      } catch (Throwable t) {
+        System.out.println("FAILED:");
+        t.printStackTrace(System.out);
+        throw new RuntimeException(t);
+      }
+    }
+  }
+
+  protected NodeState[] nodes;
+  int maxSearcherAgeSeconds;
+  long endTimeNanos;
+  private Thread changeIndicesThread;
+
+  protected void start(String baseDirName, int numNodes, double runTimeSec, int maxSearcherAgeSeconds) throws IOException {
+
+    endTimeNanos = System.nanoTime() + (long) (runTimeSec*1000000000);
+    this.maxSearcherAgeSeconds = maxSearcherAgeSeconds;
+
+    nodes = new NodeState[numNodes];
+    for(int nodeID=0;nodeID<numNodes;nodeID++) {
+      nodes[nodeID] = new NodeState(random, baseDirName, nodeID, numNodes);
+    }
+
+    long[] nodeVersions = new long[nodes.length];
+    for(int nodeID=0;nodeID<numNodes;nodeID++) {
+      final IndexSearcher s = nodes[nodeID].mgr.acquire();
+      try {
+        nodeVersions[nodeID] = nodes[nodeID].searchers.record(s);
+      } finally {
+        nodes[nodeID].mgr.release(s);
+      }
+    }
+
+    for(int nodeID=0;nodeID<numNodes;nodeID++) {
+      final IndexSearcher s = nodes[nodeID].mgr.acquire();
+      assert nodeVersions[nodeID] == nodes[nodeID].searchers.record(s);
+      assert s != null;
+      try {
+        broadcastNodeReopen(nodeID, nodeVersions[nodeID], s);
+      } finally {
+        nodes[nodeID].mgr.release(s);
+      }
+    }
+
+    changeIndicesThread = new ChangeIndices();
+    changeIndicesThread.start();
+  }
+
+  protected void finish() throws InterruptedException, IOException {
+    changeIndicesThread.join();
+    for(NodeState node : nodes) {
+      node.close();
+    }
+  }
+
+  protected static class SearcherAndVersion {
+    public final IndexSearcher searcher;
+    public final long version;
+
+    public SearcherAndVersion(IndexSearcher searcher, long version) {
+      this.searcher = searcher;
+      this.version = version;
+    }
+  }
+}

Property changes on: lucene/src/test-framework/java/org/apache/lucene/search/ShardSearchingTestBase.java
___________________________________________________________________
Added: svn:eol-style
## -0,0 +1 ##
+native