initial spillover shard impl

Author: mikemccand

scripts/indexTaxis.py

@@ -1,3 +1,4 @@
+import pprint
 import random
 import gzip
 import socket
@@ -259,8 +260,7 @@ def main():
     if len(replicaPorts) > 0:
       refreshSec = 1.0
     else:
-      # Turn off refreshes to maximize indexing throughput:
-      refreshSec = 100000.0
+      refreshSec = 5.0
     send(LOCALHOST, primaryPorts[0], "liveSettings", {'indexName': 'index', 'index.ramBufferSizeMB': 256., 'maxRefreshSec': refreshSec})
 
     fields = {'indexName': 'index',
@@ -361,7 +361,7 @@ def main():
       dps = float(dps)
 
     if USE_JSON:
-       c = ChunkedHTTPSend(LOCALHOST, primaryPorts[0], 'bulkAddDocument2', {'indexName': 'index'})
+      c = ChunkedHTTPSend(LOCALHOST, primaryPorts[0], 'bulkAddDocument2', {'indexName': 'index'})
     elif BINARY_CSV:
       b1 = BinarySend(LOCALHOST, primaryPorts[1], 'bulkCSVAddDocument')
       b1.add(b',index\n')
@@ -421,7 +421,10 @@ def main():
             print('%6.1f sec: %.2fM hits on replica, %.2f M docs... %.1f docs/sec, %.1f MB/sec' % (delay, x['totalHits']/1000000., id/1000000., dps, (totBytes/1024./1024.)/delay))
           else:
             print('%6.1f sec: %.2f M docs... %.1f docs/sec, %.1f MB/sec' % (delay, id/1000000., dps, (totBytes/1024./1024.)/delay))
-
+          result = json.loads(send(LOCALHOST, primaryPorts[0], 'stats', {'indexName': 'index'}))
+          print(json.dumps(result, sort_keys=True, indent=4))
+          x = json.loads(send(LOCALHOST, primaryPorts[0], 'search2', {'indexNames': ['index'], 'queryText': '*:*'}));
+          print('%d hits...' % x['totalHits'])
           while nextPrint <= id:
             nextPrint += 250000
 

src/java/org/apache/lucene/server/GlobalState.java

@@ -53,6 +53,7 @@
 
 import org.apache.lucene.facet.taxonomy.SearcherTaxonomyManager.SearcherAndTaxonomy;
 import org.apache.lucene.index.Term;
+import org.apache.lucene.search.MatchAllDocsQuery;
 import org.apache.lucene.search.ScoreDoc;
 import org.apache.lucene.search.TermQuery;
 import org.apache.lucene.search.TimeLimitingCollector;
@@ -168,12 +169,14 @@ private void _run() throws Exception {
           }
           //System.out.println("N" + StringHelper.idToString(nodeID) + ": run query Q" + query.id);
 
-          IndexState indexState = get(query.indexName);
+          IndexState indexState = getIndex(query.indexName);
           ShardState shardState = indexState.getShard(query.shardOrd);
           SearcherAndTaxonomy searcher = shardState.acquire();
+          shardState.slm.record(searcher.searcher);
+          System.out.println("searcher for " + shardState);
           TopDocs hits;
           try {
-            hits = searcher.searcher.search(new TermQuery(new Term("title", query.text)), 10);
+            hits = searcher.searcher.search(new MatchAllDocsQuery(), 10);
           } finally {
             shardState.release(searcher);
           }
@@ -282,8 +285,7 @@ public Map<String,Handler> getHandlers() {
   }
 
   /** Get the {@link IndexState} by index name. */
-  // nocommit rename to getIndex
-  public IndexState get(String name) throws IOException {
+  public IndexState getIndex(String name) throws IOException {
     synchronized(indices) {
       IndexState state = indices.get(name);
       if (state == null) {

src/java/org/apache/lucene/server/IndexState.java

@@ -74,6 +74,7 @@
 import org.apache.lucene.index.IndexWriterConfig;
 import org.apache.lucene.index.IndexableField;
 import org.apache.lucene.index.KeepOnlyLastCommitDeletionPolicy;
+import org.apache.lucene.index.MergePolicy;
 import org.apache.lucene.index.PersistentSnapshotDeletionPolicy;
 import org.apache.lucene.index.SegmentInfos;
 import org.apache.lucene.index.SimpleMergedSegmentWarmer;
@@ -183,7 +184,9 @@ public class IndexState implements Closeable {
   /** {@link Bindings} to pass when evaluating expressions. */
   public final Bindings exprBindings = new FieldDefBindings(fields);
 
-  private final Map<Integer,ShardState> shards = new ConcurrentHashMap<>();
+  public final Map<Integer,ShardState> shards = new ConcurrentHashMap<>();
+
+  private ShardState writableShard;
 
   /** Built suggest implementations */
   public final Map<String,Lookup> suggesters = new ConcurrentHashMap<String,Lookup>();
@@ -362,6 +365,7 @@ public IndexState(GlobalState globalState, String name, Path rootDir, boolean do
     this.globalState = globalState;
     this.name = name;
     this.rootDir = rootDir;
+    // nocommit require rootDir != null!  no RAMDirectory!
     if (rootDir != null) {
       if (Files.exists(rootDir) == false) {
         Files.createDirectories(rootDir);
@@ -435,7 +439,7 @@ public boolean hasCommit() throws IOException {
     return saveLoadState.getNextWriteGen() != 0;
   }
 
-  IndexWriterConfig getIndexWriterConfig(OpenMode openMode, Directory origIndexDir) throws IOException {
+  IndexWriterConfig getIndexWriterConfig(OpenMode openMode, Directory origIndexDir, int shardOrd) throws IOException {
     IndexWriterConfig iwc = new IndexWriterConfig(indexAnalyzer);
     iwc.setOpenMode(openMode);
     if (getBooleanSetting("index.verbose")) {
@@ -452,7 +456,15 @@ IndexWriterConfig getIndexWriterConfig(OpenMode openMode, Directory origIndexDir
     // nocommit in primary case we can't do this?
     iwc.setMergedSegmentWarmer(new SimpleMergedSegmentWarmer(iwc.getInfoStream()));
 
-    ConcurrentMergeScheduler cms = (ConcurrentMergeScheduler) iwc.getMergeScheduler();
+    ConcurrentMergeScheduler cms = new ConcurrentMergeScheduler() {
+        @Override
+        public synchronized MergeThread getMergeThread(IndexWriter writer, MergePolicy.OneMerge merge) throws IOException {
+          MergeThread thread = super.getMergeThread(writer, merge);
+          thread.setName(thread.getName() + " [" + name + ":" + shardOrd + "]");
+          return thread;
+        }
+      };
+    iwc.setMergeScheduler(cms);
 
     if (getBooleanSetting("index.merge.scheduler.auto_throttle")) {
       cms.enableAutoIOThrottle();
@@ -710,8 +722,12 @@ public void setIndexSort(Sort sort, Object saveState) {
 
   public void start() throws Exception {
     // start all local shards
+    
     for(ShardState shard : shards.values()) {
       shard.start();
+      if (writableShard == null || shard.shardOrd > writableShard.shardOrd) {
+        writableShard = shard;
+      }
     }
 
     if (suggesterSettings != null) {
@@ -1009,10 +1025,36 @@ public ShardState addShard(int shardOrd, boolean doCreate) {
       throw new IllegalArgumentException("shardOrd=" + shardOrd + " already exists in index + \"" + name + "\"");
     }
     ShardState shard = new ShardState(this, shardOrd, doCreate);
+    // nocommit fail if there is already a shard here?
     shards.put(shardOrd, shard);
     return shard;
   }
 
+  /** Returns the shard that should be used for writing new documents, possibly creating and starting a new shard if the current one is full. */
+  public ShardState getWritableShard() throws Exception {
+    verifyStarted(null);
+
+    synchronized (this) {
+
+      // nocommit make this tunable:
+      if (writableShard.writer.maxDoc() > 50000000) {
+        System.out.println("NEW SHARD ORD " + (writableShard.shardOrd+1));
+        ShardState nextShard = addShard(writableShard.shardOrd+1, true);
+
+        // nocommit hmm we can't do this now ... we need to get all in-flight ops to finish first
+        writableShard.finishWriting();
+        // nocommit we should close IW after forceMerge finishes
+
+        nextShard.start();
+        writableShard = nextShard;
+      }
+
+      writableShard.startIndexingChunk();
+    }
+
+    return writableShard;
+  }
+
   public ShardState getShard(int shardOrd) {
     ShardState shardState = shards.get(shardOrd);
     if (shardState == null) {

src/java/org/apache/lucene/server/Server.java

@@ -443,7 +443,7 @@ private void _handle(HttpExchange x) throws Exception {
           IndexState state;
           if (handler.requiresIndexName) {
             String indexName = request.getString("indexName");
-            state = globalState.get(indexName);
+            state = globalState.getIndex(indexName);
           } else {
             state = null;
           }
@@ -903,6 +903,7 @@ private class BinaryServer extends Thread {
     private volatile boolean stop;
 
     public BinaryServer(String host, int port) throws IOException {
+      setName("BinaryServer " + host + ":" + port);
       serverSocket = new ServerSocket();
       serverSocket.bind(new InetSocketAddress(host, port));
       actualPort = serverSocket.getLocalPort();

src/java/org/apache/lucene/server/ShardState.java

@@ -96,7 +96,7 @@ public class ShardState implements Closeable {
   public final Path rootDir;
 
   /** Which shard we are in this index */
-  private final int shardOrd;
+  public final int shardOrd;
 
   /** Base directory */
   public Directory origIndexDir;
@@ -434,7 +434,7 @@ protected IndexWriter openIndexWriter(Directory dir, IndexWriterConfig iwc) thro
           }
         };
 
-      writer = new IndexWriter(indexDir, indexState.getIndexWriterConfig(openMode, origIndexDir));
+      writer = new IndexWriter(indexDir, indexState.getIndexWriterConfig(openMode, origIndexDir, shardOrd));
       snapshots = (PersistentSnapshotDeletionPolicy) writer.getConfig().getIndexDeletionPolicy();
 
       // NOTE: must do this after writer, because SDP only
@@ -586,7 +586,7 @@ public synchronized void startPrimary(long primaryGen) throws Exception {
 
       boolean verbose = indexState.getBooleanSetting("index.verbose");
 
-      writer = new IndexWriter(indexDir, indexState.getIndexWriterConfig(openMode, origIndexDir));
+      writer = new IndexWriter(indexDir, indexState.getIndexWriterConfig(openMode, origIndexDir, shardOrd));
       snapshots = (PersistentSnapshotDeletionPolicy) writer.getConfig().getIndexDeletionPolicy();
 
       // NOTE: must do this after writer, because SDP only
@@ -995,4 +995,93 @@ public OrdRange getOrdRange(String dim) {
       return ssdvState;
     }
   }
+
+  @Override
+  public String toString() {
+    return indexState.name + ":" + shardOrd;
+  }
+
+  private final AtomicInteger runningChunks = new AtomicInteger();
+  private volatile int finishWriting;
+
+  public void startIndexingChunk() {
+    if (finishWriting != 0) {
+      throw new IllegalStateException("cannot start indexing when finish is called");
+    }
+    runningChunks.incrementAndGet();
+  }
+
+  public void finishIndexingChunk() throws InterruptedException {
+    if (runningChunks.decrementAndGet() == 0) {
+      if (finishWriting == 1) {
+        finishIndexing();
+      }
+    }
+  }
+
+  public String getState() {
+    if (finishWriting == 2) {
+      return "readonly";
+    } else if (finishWriting == 1) {
+      return "finishing";
+    } else {
+      return "active";
+    }
+  }
+
+  public void finishIndexing() throws InterruptedException {
+    System.out.println(this + ": now finish indexing");
+    indexState.globalState.submitIndexingTask(new Runnable() {
+        @Override
+        public void run() {
+          // nocommit make this tunable
+          try {
+            System.out.println(ShardState.this + ": now force merge");
+            writer.forceMerge(1);
+            System.out.println(ShardState.this + ": done force merge; now refresh");
+            reopenThread.close();
+            maybeRefreshBlocking();
+            System.out.println(ShardState.this + ": done force merge; now close writers");
+            finishWriting = 2;
+
+            // nocommit register this, somehow, in a long running task
+            // nocommit close searcher pruning thread too
+            
+            writer.close();
+            taxoWriter.close();
+            System.out.println(ShardState.this + ": done force merge; done close writers");
+            // nocommit what about refreshing readers after this?  can i "reopen to last reader" somehow?
+            indexState.globalState.indexingJobsRunning.release();
+          } catch (Throwable t) {
+            System.out.println("ShardState.finishIndexing " + this + " FAILED:");
+            t.printStackTrace(System.out);
+            throw new RuntimeException(t);
+          }
+        }
+      });
+  }
+
+  public int maxDoc() throws IOException {
+    synchronized (this) {
+      if (finishWriting == 0) {
+        return writer.maxDoc();
+      }
+    }
+
+    SearcherAndTaxonomy searcher = acquire();
+    try {
+      return searcher.searcher.getIndexReader().maxDoc();
+    } finally {
+      release(searcher);
+    }
+  }
+
+  public void finishWriting() throws InterruptedException {
+    synchronized(this) {
+      finishWriting = 1;
+    }
+    if (runningChunks.get() == 0) {
+      finishIndexing();
+    }
+  }
 }

src/java/org/apache/lucene/server/handlers/AddReplicaHandler.java

@@ -50,7 +50,7 @@ public boolean binaryRequest() {
   @Override
   public void handleBinary(InputStream streamIn, DataInput in, DataOutput out, OutputStream streamOut) throws Exception {
     String indexName = in.readString();
-    IndexState indexState = globalState.get(indexName);
+    IndexState indexState = globalState.getIndex(indexName);
     ShardState shardState = indexState.getShard(0);
     if (shardState.isPrimary() == false) {
       throw new IllegalArgumentException("index \"" + indexName + "\" was not started or is not a primary");