/*
 * 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.
 */
package org.apache.lucene.index;


import java.io.Closeable;
import java.io.IOException;
import java.nio.file.DirectoryStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Pattern;

import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.document.LongPoint;
import org.apache.lucene.document.NumericDocValuesField;
import org.apache.lucene.search.FieldDoc;
import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.MatchAllDocsQuery;
import org.apache.lucene.search.ScoreDoc;
import org.apache.lucene.search.Sort;
import org.apache.lucene.search.SortField;
import org.apache.lucene.search.TopDocs;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.FSDirectory;
import org.apache.lucene.store.IOContext;
import org.apache.lucene.store.MockDirectoryWrapper.Throttling;
import org.apache.lucene.store.MockDirectoryWrapper;
import org.apache.lucene.util.Bits;
import org.apache.lucene.util.IOUtils;
import org.apache.lucene.util.LuceneTestCase;
import org.apache.lucene.util.StringHelper;
import org.apache.lucene.util.TestUtil;
import org.apache.lucene.util.Version;

// TODO:
//   - old parallel indices are only pruned on commit/close; can we do it on refresh?

/** Simple example showing how to use ParallelLeafReader to index new
 *  stuff (postings, DVs, etc.) from previously stored fields, on the
 *  fly (during NRT reader reopen), after the  initial indexing.  The
 *  test indexes just a single stored field with text "content X" (X is
 *  a number embedded in the text).
 *
 *  Then, on reopen, for any newly created segments (flush or merge), it
 *  builds a new parallel segment by loading all stored docs, parsing
 *  out that X, and adding it as DV and numeric indexed (trie) field.
 *
 *  Finally, for searching, it builds a top-level MultiReader, with
 *  ParallelLeafReader for each segment, and then tests that random
 *  numeric range queries, and sorting by the new DV field, work
 *  correctly.
 *
 *  Each per-segment index lives in a private directory next to the main
 *  index, and they are deleted once their segments are removed from the
 *  index.  They are "volatile", meaning if e.g. the index is replicated to
 *  another machine, it's OK to not copy parallel segments indices,
 *  since they will just be regnerated (at a cost though). */

// @SuppressSysoutChecks(bugUrl="we print stuff")
// See: https://issues.apache.org/jira/browse/SOLR-12028 Tests cannot remove files on Windows machines occasionally
public class TestDemoParallelLeafReader extends LuceneTestCase {

  static final boolean DEBUG = false;

  static abstract class ReindexingReader implements Closeable {

    /** Key used to store the current schema gen in the SegmentInfo diagnostics */
    public final static String SCHEMA_GEN_KEY = "schema_gen";

    public final IndexWriter w;
    public final ReaderManager mgr;

    private final Directory indexDir;
    private final Path root;
    private final Path segsPath;

    /** Which segments have been closed, but their parallel index is not yet not removed. */
    private final Set<SegmentIDAndGen> closedSegments = Collections.newSetFromMap(new ConcurrentHashMap<SegmentIDAndGen,Boolean>());

    /** Holds currently open parallel readers for each segment. */
    private final Map<SegmentIDAndGen,LeafReader> parallelReaders = new ConcurrentHashMap<>();

    void printRefCounts() {
      System.out.println("All refCounts:");
      for(Map.Entry<SegmentIDAndGen,LeafReader> ent : parallelReaders.entrySet()) {
        System.out.println("  " + ent.getKey() + " " + ent.getValue() + " refCount=" + ent.getValue().getRefCount());
      }
    }

    public ReindexingReader(Path root) throws IOException {
      this.root = root;

      // Normal index is stored under "index":
      indexDir = openDirectory(root.resolve("index"));

      // Per-segment parallel indices are stored under subdirs "segs":
      segsPath = root.resolve("segs");
      Files.createDirectories(segsPath);

      IndexWriterConfig iwc = getIndexWriterConfig();
      iwc.setMergePolicy(new ReindexingMergePolicy(iwc.getMergePolicy()));
      if (DEBUG) {
        System.out.println("TEST: use IWC:\n" + iwc);
      }
      w = new IndexWriter(indexDir, iwc);

      w.getConfig().setMergedSegmentWarmer((reader) -> {
        // This will build the parallel index for the merged segment before the merge becomes visible, so reopen delay is only due to
        // newly flushed segments:
        if (DEBUG) System.out.println(Thread.currentThread().getName() +": TEST: now warm " + reader);
        // TODO: it's not great that we pass false here; it means we close the reader & reopen again for NRT reader; still we did "warm" by
        // building the parallel index, if necessary
        getParallelLeafReader(reader, false, getCurrentSchemaGen());
      });

      // start with empty commit:
      w.commit();
      mgr = new ReaderManager(new ParallelLeafDirectoryReader(DirectoryReader.open(w)));
    }

    protected abstract IndexWriterConfig getIndexWriterConfig() throws IOException;

    /** Optional method to validate that the provided parallell reader in fact reflects the changes in schemaGen. */
    protected void checkParallelReader(LeafReader reader, LeafReader parallelReader, long schemaGen) throws IOException {
    }

    /** Override to customize Directory impl. */
    protected Directory openDirectory(Path path) throws IOException {
      return FSDirectory.open(path);
    }

    public void commit() throws IOException {
      w.commit();
    }
    
    LeafReader getCurrentReader(LeafReader reader, long schemaGen) throws IOException {
      LeafReader parallelReader = getParallelLeafReader(reader, true, schemaGen);
      if (parallelReader != null) {

        // We should not be embedding one ParallelLeafReader inside another:
        assertFalse(parallelReader instanceof ParallelLeafReader);
        assertFalse(reader instanceof ParallelLeafReader);

        // NOTE: important that parallelReader is first, so if there are field name overlaps, because changes to the schema
        // overwrote existing field names, it wins:
        LeafReader newReader = new ParallelLeafReader(false, parallelReader, reader) {
          @Override
          public Bits getLiveDocs() {
            return getParallelReaders()[1].getLiveDocs();
          }
          @Override
          public int numDocs() {
            return getParallelReaders()[1].numDocs();
          }
        };

        // Because ParallelLeafReader does its own (extra) incRef:
        parallelReader.decRef();

        return newReader;

      } else {
        // This segment was already current as of currentSchemaGen:
        return reader;
      }
    }

    private class ParallelLeafDirectoryReader extends FilterDirectoryReader {
      public ParallelLeafDirectoryReader(DirectoryReader in) throws IOException {
        super(in, new FilterDirectoryReader.SubReaderWrapper() {
            final long currentSchemaGen = getCurrentSchemaGen();
            @Override
            public LeafReader wrap(LeafReader reader) {
              try {
                return getCurrentReader(reader, currentSchemaGen);
              } catch (IOException ioe) {
                // TODO: must close on exc here:
                throw new RuntimeException(ioe);
              }
            }
          });
      }

      @Override
      protected DirectoryReader doWrapDirectoryReader(DirectoryReader in) throws IOException {
        return new ParallelLeafDirectoryReader(in);
      }

      @Override
      protected void doClose() throws IOException {
        Throwable firstExc = null;
        for (final LeafReader r : getSequentialSubReaders()) {
          if (r instanceof ParallelLeafReader) {
            // try to close each reader, even if an exception is thrown
            try {
              r.decRef();
            } catch (Throwable t) {
              if (firstExc == null) {
                firstExc = t;
              }
            }
          }
        }
        // Also close in, so it decRef's the SegmentInfos
        try {
          in.doClose();
        } catch (Throwable t) {
          if (firstExc == null) {
            firstExc = t;
          }
        }
        
        // throw the first exception
        if (firstExc != null) {
          throw IOUtils.rethrowAlways(firstExc);
        }
      }

      @Override
      public CacheHelper getReaderCacheHelper() {
        return null;
      }
    }

    @Override
    public void close() throws IOException {
      w.close();
      if (DEBUG) System.out.println("TEST: after close writer index=" + SegmentInfos.readLatestCommit(indexDir));

      /*
      DirectoryReader r = mgr.acquire();
      try {
        TestUtil.checkReader(r);
      } finally {
        mgr.release(r);
      }
      */
      mgr.close();
      pruneOldSegments(true);
      assertNoExtraSegments();
      indexDir.close();
    }

    // Make sure we deleted all parallel indices for segments that are no longer in the main index: 
    private void assertNoExtraSegments() throws IOException {
      Set<String> liveIDs = new HashSet<String>();
      for(SegmentCommitInfo info : SegmentInfos.readLatestCommit(indexDir)) {
        String idString = StringHelper.idToString(info.info.getId());
        liveIDs.add(idString);
      }

      // At this point (closing) the only segments in closedSegments should be the still-live ones:
      for(SegmentIDAndGen segIDGen : closedSegments) {
        assertTrue(liveIDs.contains(segIDGen.segID));
      }

      boolean fail = false;
      for(Path path : segSubDirs(segsPath)) {
        SegmentIDAndGen segIDGen = new SegmentIDAndGen(path.getFileName().toString());
        if (liveIDs.contains(segIDGen.segID) == false) {
          if (DEBUG) System.out.println("TEST: fail seg=" + path.getFileName() + " is not live but still has a parallel index");
          fail = true;
        }
      }
      assertFalse(fail);
    }

    private static class SegmentIDAndGen {
      public final String segID;
      public final long schemaGen;

      public SegmentIDAndGen(String segID, long schemaGen) {
        this.segID = segID;
        this.schemaGen = schemaGen;
      }

      public SegmentIDAndGen(String s) {
        String[] parts = s.split("_");
        if (parts.length != 2) {
          throw new IllegalArgumentException("invalid SegmentIDAndGen \"" + s + "\"");
        }
        // TODO: better checking of segID?
        segID = parts[0];
        schemaGen = Long.parseLong(parts[1]);
      }

      @Override
      public int hashCode() {
        return (int) (segID.hashCode() * schemaGen);
      }

      @Override
      public boolean equals(Object _other) {
        if (_other instanceof SegmentIDAndGen) {
          SegmentIDAndGen other = (SegmentIDAndGen) _other;
          return segID.equals(other.segID) && schemaGen == other.schemaGen;
        } else {
          return false;
        }
      }

      @Override
      public String toString() {
        return segID + "_" + schemaGen;
      }
    }

    private class ParallelReaderClosed implements IndexReader.ClosedListener {
      private final SegmentIDAndGen segIDGen;
      private final Directory dir;

      public ParallelReaderClosed(SegmentIDAndGen segIDGen, Directory dir) {
        this.segIDGen = segIDGen;
        this.dir = dir;
      }

      @Override
      public void onClose(IndexReader.CacheKey ignored) {
        try {
          // TODO: make this sync finer, i.e. just the segment + schemaGen
          synchronized(ReindexingReader.this) {
            if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST: now close parallel parLeafReader dir=" + dir + " segIDGen=" + segIDGen);
            parallelReaders.remove(segIDGen);
            dir.close();
            closedSegments.add(segIDGen);
          }
        } catch (IOException ioe) {
          System.out.println("TEST: hit IOExc closing dir=" + dir);
          ioe.printStackTrace(System.out);
          throw new RuntimeException(ioe);
        }
      }
    }

    // Returns a ref
    LeafReader getParallelLeafReader(final LeafReader leaf, boolean doCache, long schemaGen) throws IOException {
      assert leaf instanceof SegmentReader;
      SegmentInfo info = ((SegmentReader) leaf).getSegmentInfo().info;

      long infoSchemaGen = getSchemaGen(info);

      if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST: getParallelLeafReader: " + leaf + " infoSchemaGen=" + infoSchemaGen + " vs schemaGen=" + schemaGen + " doCache=" + doCache);

      if (infoSchemaGen == schemaGen) {
        if (DEBUG) System.out.println(Thread.currentThread().getName()+ ": TEST: segment is already current schemaGen=" + schemaGen + "; skipping");
        return null;
      }

      if (infoSchemaGen > schemaGen) {
        throw new IllegalStateException("segment infoSchemaGen (" + infoSchemaGen + ") cannot be greater than requested schemaGen (" + schemaGen + ")");
      }

      final SegmentIDAndGen segIDGen = new SegmentIDAndGen(StringHelper.idToString(info.getId()), schemaGen);

      // While loop because the parallel reader may be closed out from under us, so we must retry:
      while (true) {

        // TODO: make this sync finer, i.e. just the segment + schemaGen
        synchronized (this) {
          LeafReader parReader = parallelReaders.get(segIDGen);
      
          assert doCache || parReader == null;

          if (parReader == null) {

            Path leafIndex = segsPath.resolve(segIDGen.toString());

            final Directory dir = openDirectory(leafIndex);

            if (slowFileExists(dir, "done") == false) {
              if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST: build segment index for " + leaf + " " + segIDGen + " (source: " + info.getDiagnostics().get("source") + ") dir=" + leafIndex);

              if (dir.listAll().length != 0) {
                // It crashed before finishing last time:
                if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST: remove old incomplete index files: " + leafIndex);
                IOUtils.rm(leafIndex);
              }

              reindex(infoSchemaGen, schemaGen, leaf, dir);

              // Marker file, telling us this index is in fact done.  This way if we crash while doing the reindexing for a given segment, we will
              // later try again:
              dir.createOutput("done", IOContext.DEFAULT).close();
            } else {
              if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST: segment index already exists for " + leaf + " " + segIDGen + " (source: " + info.getDiagnostics().get("source") + ") dir=" + leafIndex);
            }

            if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST: now check index " + dir);
            //TestUtil.checkIndex(dir);

            SegmentInfos infos = SegmentInfos.readLatestCommit(dir);
            assert infos.size() == 1;
            final LeafReader parLeafReader = new SegmentReader(infos.info(0), Version.LATEST.major, IOContext.DEFAULT);

            //checkParallelReader(leaf, parLeafReader, schemaGen);

            if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST: opened parallel reader: " + parLeafReader);
            if (doCache) {
              parallelReaders.put(segIDGen, parLeafReader);

              // Our id+gen could have been previously closed, e.g. if it was a merged segment that was warmed, so we must clear this else
              // the pruning may remove our directory:
              closedSegments.remove(segIDGen);

              parLeafReader.getReaderCacheHelper().addClosedListener(new ParallelReaderClosed(segIDGen, dir));

            } else {
              // Used only for merged segment warming:
              // Messy: we close this reader now, instead of leaving open for reuse:
              if (DEBUG) System.out.println("TEST: now decRef non cached refCount=" + parLeafReader.getRefCount());
              parLeafReader.decRef();
              dir.close();

              // Must do this after dir is closed, else another thread could "rm -rf" while we are closing (which makes MDW.close's
              // checkIndex angry):
              closedSegments.add(segIDGen);
              parReader = null;
            }
            parReader = parLeafReader;

          } else {
            if (parReader.tryIncRef() == false) {
              // We failed: this reader just got closed by another thread, e.g. refresh thread opening a new reader, so this reader is now
              // closed and we must try again.
              if (DEBUG) System.out.println(Thread.currentThread().getName()+ ": TEST: tryIncRef failed for " + parReader + "; retry");
              parReader = null;
              continue;
            }
            if (DEBUG) System.out.println(Thread.currentThread().getName()+ ": TEST: use existing already opened parReader=" + parReader + " refCount=" + parReader.getRefCount());
            //checkParallelReader(leaf, parReader, schemaGen);
          }

          // We return the new reference to caller
          return parReader;
        }
      }
    }

    // TODO: we could pass a writer already opened...?
    protected abstract void reindex(long oldSchemaGen, long newSchemaGen, LeafReader reader, Directory parallelDir) throws IOException;

    /** Returns the gen for the current schema. */
    protected abstract long getCurrentSchemaGen();

    /** Returns the gen that should be merged, meaning those changes will be folded back into the main index. */
    protected long getMergingSchemaGen() {
      return getCurrentSchemaGen();
    }

    /** Removes the parallel index that are no longer in the last commit point.  We can't
     *  remove this when the parallel reader is closed because it may still be referenced by
     *  the last commit. */
    private void pruneOldSegments(boolean removeOldGens) throws IOException {
      SegmentInfos lastCommit = SegmentInfos.readLatestCommit(indexDir);
      if (DEBUG) System.out.println("TEST: prune");

      Set<String> liveIDs = new HashSet<String>();
      for(SegmentCommitInfo info : lastCommit) {
        String idString = StringHelper.idToString(info.info.getId());
        liveIDs.add(idString);
      }

      long currentSchemaGen = getCurrentSchemaGen();

      if (Files.exists(segsPath)) {
        for (Path path : segSubDirs(segsPath)) {
          if (Files.isDirectory(path)) {
            SegmentIDAndGen segIDGen = new SegmentIDAndGen(path.getFileName().toString());
            assert segIDGen.schemaGen <= currentSchemaGen;
            if (liveIDs.contains(segIDGen.segID) == false && (closedSegments.contains(segIDGen) || (removeOldGens && segIDGen.schemaGen < currentSchemaGen))) {
              if (DEBUG) System.out.println("TEST: remove " + segIDGen);
              try {
                IOUtils.rm(path);
                closedSegments.remove(segIDGen);
              } catch (IOException ioe) {
                // OK, we'll retry later
                if (DEBUG) System.out.println("TEST: ignore ioe during delete " + path + ":" + ioe);
              }
            }
          }
        }
      }
    }

    /** Just replaces the sub-readers with parallel readers, so reindexed fields are merged into new segments. */
    private class ReindexingMergePolicy extends FilterMergePolicy {

      class ReindexingOneMerge extends OneMerge {

        final List<ParallelLeafReader> parallelReaders = new ArrayList<>();
        final long schemaGen;

        ReindexingOneMerge(List<SegmentCommitInfo> segments) {
          super(segments);
          // Commit up front to which schemaGen we will merge; we don't want a schema change sneaking in for some of our leaf readers but not others:
          schemaGen = getMergingSchemaGen();
          long currentSchemaGen = getCurrentSchemaGen();

          // Defensive sanity check:
          if (schemaGen > currentSchemaGen) {
            throw new IllegalStateException("currentSchemaGen (" + currentSchemaGen + ") must always be >= mergingSchemaGen (" + schemaGen + ")");
          }
        }

        @Override
        public CodecReader wrapForMerge(CodecReader reader) throws IOException {
          LeafReader wrapped = getCurrentReader(reader, schemaGen);
          if (wrapped instanceof ParallelLeafReader) {
            parallelReaders.add((ParallelLeafReader) wrapped);
          }
          return SlowCodecReaderWrapper.wrap(wrapped);
        }

        @Override
        public void mergeFinished(boolean success, boolean segmentDropped) throws IOException {
          super.mergeFinished(success, segmentDropped);
          Throwable th = null;
          for (ParallelLeafReader r : parallelReaders) {
            try {
              r.decRef();
            } catch (Throwable t) {
              if (th == null) {
                th = t;
              }
            }
          }

          if (th != null) {
            throw IOUtils.rethrowAlways(th);
          }
        }

        @Override
        public void setMergeInfo(SegmentCommitInfo info) {
          // Record that this merged segment is current as of this schemaGen:
          Map<String, String> copy = new HashMap<>(info.info.getDiagnostics());
          copy.put(SCHEMA_GEN_KEY, Long.toString(schemaGen));
          info.info.setDiagnostics(copy);
          super.setMergeInfo(info);
        }

      }

      class ReindexingMergeSpecification extends MergeSpecification {
        @Override
        public void add(OneMerge merge) {
          super.add(new ReindexingOneMerge(merge.segments));
        }

        @Override
        public String segString(Directory dir) {
          return "ReindexingMergeSpec(" + super.segString(dir) + ")";
        }
      }

      MergeSpecification wrap(MergeSpecification spec) {
        MergeSpecification wrapped = null;
        if (spec != null) {
          wrapped = new ReindexingMergeSpecification();
          for (OneMerge merge : spec.merges) {
            wrapped.add(merge);
          }
        }
        return wrapped;
      }

      /** Create a new {@code MergePolicy} that sorts documents with the given {@code sort}. */
      public ReindexingMergePolicy(MergePolicy in) {
        super(in);
      }

      @Override
      public MergeSpecification findMerges(MergeTrigger mergeTrigger,
                                           SegmentInfos segmentInfos, MergeContext mergeContext) throws IOException {
        return wrap(in.findMerges(mergeTrigger, segmentInfos, mergeContext));
      }

      @Override
      public MergeSpecification findForcedMerges(SegmentInfos segmentInfos,
                                                 int maxSegmentCount, Map<SegmentCommitInfo,Boolean> segmentsToMerge, MergeContext mergeContext)
        throws IOException {
        // TODO: do we need to force-force this?  Ie, wrapped MP may think index is already optimized, yet maybe its schemaGen is old?  need test!
        return wrap(in.findForcedMerges(segmentInfos, maxSegmentCount, segmentsToMerge, mergeContext));
      }

      @Override
      public MergeSpecification findForcedDeletesMerges(SegmentInfos segmentInfos, MergeContext mergeContext)
        throws IOException {
        return wrap(in.findForcedDeletesMerges(segmentInfos, mergeContext));
      }

      @Override
      public boolean useCompoundFile(SegmentInfos segments,
                                     SegmentCommitInfo newSegment, MergeContext mergeContext) throws IOException {
        return in.useCompoundFile(segments, newSegment, mergeContext);
      }

      @Override
      public String toString() {
        return "ReindexingMergePolicy(" + in + ")";
      }
    }

    static long getSchemaGen(SegmentInfo info) {
      String s = info.getDiagnostics().get(SCHEMA_GEN_KEY);
      if (s == null) {
        return -1;
      } else {
        return Long.parseLong(s);
      }
    }
  }

  private ReindexingReader getReindexer(Path root) throws IOException {
    return new ReindexingReader(root) {
      @Override
      protected IndexWriterConfig getIndexWriterConfig() throws IOException {
        IndexWriterConfig iwc = newIndexWriterConfig();
        TieredMergePolicy tmp = new TieredMergePolicy();
        // We write tiny docs, so we need tiny floor to avoid O(N^2) merging:
        tmp.setFloorSegmentMB(.01);
        iwc.setMergePolicy(tmp);
        return iwc;
      }

      @Override
      protected Directory openDirectory(Path path) throws IOException {
        MockDirectoryWrapper dir = newMockFSDirectory(path);
        dir.setUseSlowOpenClosers(false);
        dir.setThrottling(Throttling.NEVER);
        return dir;
      }

      @Override
      protected void reindex(long oldSchemaGen, long newSchemaGen, LeafReader reader, Directory parallelDir) throws IOException {
        IndexWriterConfig iwc = newIndexWriterConfig();

        // The order of our docIDs must precisely matching incoming reader:
        iwc.setMergePolicy(new LogByteSizeMergePolicy());
        IndexWriter w = new IndexWriter(parallelDir, iwc);
        int maxDoc = reader.maxDoc();

        // Slowly parse the stored field into a new doc values field:
        for(int i=0;i<maxDoc;i++) {
          // TODO: is this still O(blockSize^2)?
          Document oldDoc = reader.document(i);
          Document newDoc = new Document();
          long value = Long.parseLong(oldDoc.get("text").split(" ")[1]);
          newDoc.add(new NumericDocValuesField("number", value));
          newDoc.add(new LongPoint("number", value));
          w.addDocument(newDoc);
        }

        w.forceMerge(1);

        w.close();
      }

      @Override
      protected long getCurrentSchemaGen() {
        return 0;
      }
    };
  }

  /** Schema change by adding a new number_<schemaGen> DV field each time. */
  private ReindexingReader getReindexerNewDVFields(Path root, final AtomicLong currentSchemaGen) throws IOException {
    return new ReindexingReader(root) {
      @Override
      protected IndexWriterConfig getIndexWriterConfig() throws IOException {
        IndexWriterConfig iwc = newIndexWriterConfig();
        TieredMergePolicy tmp = new TieredMergePolicy();
        // We write tiny docs, so we need tiny floor to avoid O(N^2) merging:
        tmp.setFloorSegmentMB(.01);
        iwc.setMergePolicy(tmp);
        return iwc;
      }

      @Override
      protected Directory openDirectory(Path path) throws IOException {
        MockDirectoryWrapper dir = newMockFSDirectory(path);
        dir.setUseSlowOpenClosers(false);
        dir.setThrottling(Throttling.NEVER);
        return dir;
      }

      @Override
      protected void reindex(long oldSchemaGen, long newSchemaGen, LeafReader reader, Directory parallelDir) throws IOException {
        IndexWriterConfig iwc = newIndexWriterConfig();

        // The order of our docIDs must precisely matching incoming reader:
        iwc.setMergePolicy(new LogByteSizeMergePolicy());
        IndexWriter w = new IndexWriter(parallelDir, iwc);
        int maxDoc = reader.maxDoc();

        if (oldSchemaGen <= 0) {
          // Must slowly parse the stored field into a new doc values field:
          for(int i=0;i<maxDoc;i++) {
            // TODO: is this still O(blockSize^2)?
            Document oldDoc = reader.document(i);
            Document newDoc = new Document();
            long value = Long.parseLong(oldDoc.get("text").split(" ")[1]);
            newDoc.add(new NumericDocValuesField("number_" + newSchemaGen, value));
            newDoc.add(new LongPoint("number", value));
            w.addDocument(newDoc);
          }
        } else {
          // Just carry over doc values from previous field:
          NumericDocValues oldValues = reader.getNumericDocValues("number_" + oldSchemaGen);
          assertNotNull("oldSchemaGen=" + oldSchemaGen, oldValues);
          for(int i=0;i<maxDoc;i++) {
            // TODO: is this still O(blockSize^2)?
            assertEquals(i, oldValues.nextDoc());
            Document oldDoc = reader.document(i);
            Document newDoc = new Document();
            newDoc.add(new NumericDocValuesField("number_" + newSchemaGen, oldValues.longValue()));
            w.addDocument(newDoc);
          }
        }

        w.forceMerge(1);

        w.close();
      }

      @Override
      protected long getCurrentSchemaGen() {
        return currentSchemaGen.get();
      }

      @Override
      protected void checkParallelReader(LeafReader r, LeafReader parR, long schemaGen) throws IOException {
        String fieldName = "number_" + schemaGen;
        if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST: now check parallel number DVs field=" + fieldName + " r=" + r + " parR=" + parR);
        NumericDocValues numbers = parR.getNumericDocValues(fieldName);
        if (numbers == null) {
          return;
        }
        int maxDoc = r.maxDoc();
        boolean failed = false;
        for(int i=0;i<maxDoc;i++) {
          Document oldDoc = r.document(i);
          long value = Long.parseLong(oldDoc.get("text").split(" ")[1]);
          assertEquals(i, numbers.nextDoc());
          if (value != numbers.longValue()) {
            if (DEBUG) System.out.println("FAIL: docID=" + i + " " + oldDoc+ " value=" + value + " number=" + numbers.longValue() + " numbers=" + numbers);
            failed = true;
          } else if (failed) {
            if (DEBUG) System.out.println("OK: docID=" + i + " " + oldDoc+ " value=" + value + " number=" + numbers.longValue());
          }
        }
        assertFalse("FAILED field=" + fieldName + " r=" + r, failed);
      }
    };
  }

  /** Schema change by adding changing how the same "number" DV field is indexed. */
  private ReindexingReader getReindexerSameDVField(Path root, final AtomicLong currentSchemaGen, final AtomicLong mergingSchemaGen) throws IOException {
    return new ReindexingReader(root) {
      @Override
      protected IndexWriterConfig getIndexWriterConfig() throws IOException {
        IndexWriterConfig iwc = newIndexWriterConfig();
        TieredMergePolicy tmp = new TieredMergePolicy();
        // We write tiny docs, so we need tiny floor to avoid O(N^2) merging:
        tmp.setFloorSegmentMB(.01);
        iwc.setMergePolicy(tmp);
        if (TEST_NIGHTLY) {
          // during nightly tests, we might use too many files if we arent careful
          iwc.setUseCompoundFile(true);
        }
        return iwc;
      }

      @Override
      protected Directory openDirectory(Path path) throws IOException {
        MockDirectoryWrapper dir = newMockFSDirectory(path);
        dir.setUseSlowOpenClosers(false);
        dir.setThrottling(Throttling.NEVER);
        return dir;
      }

      @Override
      protected void reindex(long oldSchemaGen, long newSchemaGen, LeafReader reader, Directory parallelDir) throws IOException {
        IndexWriterConfig iwc = newIndexWriterConfig();

        // The order of our docIDs must precisely matching incoming reader:
        iwc.setMergePolicy(new LogByteSizeMergePolicy());
        IndexWriter w = new IndexWriter(parallelDir, iwc);
        int maxDoc = reader.maxDoc();

        if (oldSchemaGen <= 0) {
          // Must slowly parse the stored field into a new doc values field:
          for(int i=0;i<maxDoc;i++) {
            // TODO: is this still O(blockSize^2)?
            Document oldDoc = reader.document(i);
            Document newDoc = new Document();
            long value = Long.parseLong(oldDoc.get("text").split(" ")[1]);
            newDoc.add(new NumericDocValuesField("number", newSchemaGen*value));
            newDoc.add(new LongPoint("number", value));
            w.addDocument(newDoc);
          }
        } else {
          // Just carry over doc values from previous field:
          NumericDocValues oldValues = reader.getNumericDocValues("number");
          assertNotNull("oldSchemaGen=" + oldSchemaGen, oldValues);
          for(int i=0;i<maxDoc;i++) {
            // TODO: is this still O(blockSize^2)?
            Document oldDoc = reader.document(i);
            Document newDoc = new Document();
            assertEquals(i, oldValues.nextDoc());
            newDoc.add(new NumericDocValuesField("number", newSchemaGen*(oldValues.longValue()/oldSchemaGen)));
            w.addDocument(newDoc);
          }
        }

        w.forceMerge(1);

        w.close();
      }

      @Override
      protected long getCurrentSchemaGen() {
        return currentSchemaGen.get();
      }

      @Override
      protected long getMergingSchemaGen() {
        return mergingSchemaGen.get();
      }

      @Override
      protected void checkParallelReader(LeafReader r, LeafReader parR, long schemaGen) throws IOException {
        if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST: now check parallel number DVs r=" + r + " parR=" + parR);
        NumericDocValues numbers = parR.getNumericDocValues("numbers");
        if (numbers == null) {
          return;
        }
        int maxDoc = r.maxDoc();
        boolean failed = false;
        for(int i=0;i<maxDoc;i++) {
          Document oldDoc = r.document(i);
          long value = Long.parseLong(oldDoc.get("text").split(" ")[1]);
          value *= schemaGen;
          assertEquals(i, numbers.nextDoc());
          if (value != numbers.longValue()) {
            System.out.println("FAIL: docID=" + i + " " + oldDoc+ " value=" + value + " number=" + numbers.longValue() + " numbers=" + numbers);
            failed = true;
          } else if (failed) {
            System.out.println("OK: docID=" + i + " " + oldDoc+ " value=" + value + " number=" + numbers.longValue());
          }
        }
        assertFalse("FAILED r=" + r, failed);
      }
    };
  }

  public void testBasicMultipleSchemaGens() throws Exception {

    AtomicLong currentSchemaGen = new AtomicLong();

    // TODO: separate refresh thread, search threads, indexing threads
    Path root = createTempDir();
    ReindexingReader reindexer = getReindexerNewDVFields(root, currentSchemaGen);
    reindexer.commit();

    Document doc = new Document();
    doc.add(newTextField("text", "number " + random().nextLong(), Field.Store.YES));
    reindexer.w.addDocument(doc);

    if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST: refresh @ 1 doc");
    reindexer.mgr.maybeRefresh();
    DirectoryReader r = reindexer.mgr.acquire();
    if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST: got reader=" + r);
    try {
      checkAllNumberDVs(r, "number_" + currentSchemaGen.get(), true, 1);
    } finally {
      reindexer.mgr.release(r);
    }
    //reindexer.printRefCounts();

    currentSchemaGen.incrementAndGet();

    if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST: increment schemaGen");
    if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST: commit");
    reindexer.commit();

    doc = new Document();
    doc.add(newTextField("text", "number " + random().nextLong(), Field.Store.YES));
    reindexer.w.addDocument(doc);

    if (DEBUG) System.out.println("TEST: refresh @ 2 docs");
    reindexer.mgr.maybeRefresh();
    //reindexer.printRefCounts();
    r = reindexer.mgr.acquire();
    if (DEBUG) System.out.println("TEST: got reader=" + r);
    try {
      checkAllNumberDVs(r, "number_" + currentSchemaGen.get(), true, 1);
    } finally {
      reindexer.mgr.release(r);
    }

    if (DEBUG) System.out.println("TEST: forceMerge");
    reindexer.w.forceMerge(1);

    currentSchemaGen.incrementAndGet();

    if (DEBUG) System.out.println("TEST: commit");
    reindexer.commit();

    if (DEBUG) System.out.println("TEST: refresh after forceMerge");
    reindexer.mgr.maybeRefresh();
    r = reindexer.mgr.acquire();
    if (DEBUG) System.out.println("TEST: got reader=" + r);
    try {
      checkAllNumberDVs(r, "number_" + currentSchemaGen.get(), true, 1);
    } finally {
      reindexer.mgr.release(r);
    }

    if (DEBUG) System.out.println("TEST: close writer");
    reindexer.close();
  }

  public void testRandomMultipleSchemaGens() throws Exception {

    AtomicLong currentSchemaGen = new AtomicLong();
    ReindexingReader reindexer = null;

    // TODO: separate refresh thread, search threads, indexing threads
    int numDocs = atLeast(TEST_NIGHTLY ? 20000 : 200);
    int maxID = 0;
    Path root = createTempDir();
    int refreshEveryNumDocs = 100;
    int commitCloseNumDocs = 1000;
    for(int i=0;i<numDocs;i++) {
      if (reindexer == null) {
        reindexer = getReindexerNewDVFields(root, currentSchemaGen);
      }

      Document doc = new Document();
      String id;
      String updateID;
      if (maxID > 0 && random().nextInt(10) == 7) {
        // Replace a doc
        id = "" + random().nextInt(maxID);
        updateID = id;
      } else {
        id = "" + (maxID++);
        updateID = null;
      }
        
      doc.add(newStringField("id", id, Field.Store.NO));
      doc.add(newTextField("text", "number " + random().nextLong(), Field.Store.YES));
      if (updateID == null) {
        reindexer.w.addDocument(doc);
      } else {
        reindexer.w.updateDocument(new Term("id", updateID), doc);
      }
      if (random().nextInt(refreshEveryNumDocs) == 17) {
        if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST TOP: refresh @ " + (i+1) + " docs");
        reindexer.mgr.maybeRefresh();

        DirectoryReader r = reindexer.mgr.acquire();
        if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST TOP: got reader=" + r);
        try {
          checkAllNumberDVs(r, "number_" + currentSchemaGen.get(), true, 1);
        } finally {
          reindexer.mgr.release(r);
        }
        if (DEBUG) reindexer.printRefCounts();
        refreshEveryNumDocs = (int) (1.25 * refreshEveryNumDocs);
      }

      if (random().nextInt(500) == 17) {
        currentSchemaGen.incrementAndGet();
        if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST TOP: advance schemaGen to " + currentSchemaGen);
      }

      if (i > 0 && random().nextInt(10) == 7) {
        // Random delete:
        reindexer.w.deleteDocuments(new Term("id", ""+random().nextInt(i)));
      }

      if (random().nextInt(commitCloseNumDocs) == 17) {
        if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST TOP: commit @ " + (i+1) + " docs");
        reindexer.commit();
        //reindexer.printRefCounts();
        commitCloseNumDocs = (int) (1.25 * commitCloseNumDocs);
      }

      // Sometimes close & reopen writer/manager, to confirm the parallel segments persist:
      if (random().nextInt(commitCloseNumDocs) == 17) {
        if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST TOP: close writer @ " + (i+1) + " docs");
        reindexer.close();
        reindexer = null;
        commitCloseNumDocs = (int) (1.25 * commitCloseNumDocs);
      }
    }

    if (reindexer != null) {
      reindexer.close();
    }
  }

  /** First schema change creates a new "number" DV field off the stored field; subsequent changes just change the value of that number
   *  field for all docs. */
  public void testRandomMultipleSchemaGensSameField() throws Exception {

    AtomicLong currentSchemaGen = new AtomicLong();
    AtomicLong mergingSchemaGen = new AtomicLong();

    ReindexingReader reindexer = null;

    // TODO: separate refresh thread, search threads, indexing threads
    int numDocs = atLeast(TEST_NIGHTLY ? 20000 : 200);
    int maxID = 0;
    Path root = createTempDir();
    int refreshEveryNumDocs = 100;
    int commitCloseNumDocs = 1000;

    for(int i=0;i<numDocs;i++) {
      if (reindexer == null) {
        if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST TOP: open new reader/writer");
        reindexer = getReindexerSameDVField(root, currentSchemaGen, mergingSchemaGen);
      }

      Document doc = new Document();
      String id;
      String updateID;
      if (maxID > 0 && random().nextInt(10) == 7) {
        // Replace a doc
        id = "" + random().nextInt(maxID);
        updateID = id;
      } else {
        id = "" + (maxID++);
        updateID = null;
      }
        
      doc.add(newStringField("id", id, Field.Store.NO));
      doc.add(newTextField("text", "number " + TestUtil.nextInt(random(), -10000, 10000), Field.Store.YES));
      if (updateID == null) {
        reindexer.w.addDocument(doc);
      } else {
        reindexer.w.updateDocument(new Term("id", updateID), doc);
      }
      if (random().nextInt(refreshEveryNumDocs) == 17) {
        if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST TOP: refresh @ " + (i+1) + " docs");
        reindexer.mgr.maybeRefresh();
        DirectoryReader r = reindexer.mgr.acquire();
        if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST TOP: got reader=" + r);
        try {
          checkAllNumberDVs(r, "number", true, (int) currentSchemaGen.get());
        } finally {
          reindexer.mgr.release(r);
        }
        if (DEBUG) reindexer.printRefCounts();
        refreshEveryNumDocs = (int) (1.25 * refreshEveryNumDocs);
      }

      if (random().nextInt(500) == 17) {
        currentSchemaGen.incrementAndGet();
        if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST TOP: advance schemaGen to " + currentSchemaGen);
        if (random().nextBoolean()) {
          mergingSchemaGen.incrementAndGet();
          if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST TOP: advance mergingSchemaGen to " + mergingSchemaGen);
        }
      }

      if (i > 0 && random().nextInt(10) == 7) {
        // Random delete:
        reindexer.w.deleteDocuments(new Term("id", ""+random().nextInt(i)));
      }

      if (random().nextInt(commitCloseNumDocs) == 17) {
        if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST TOP: commit @ " + (i+1) + " docs");
        reindexer.commit();
        //reindexer.printRefCounts();
        commitCloseNumDocs = (int) (1.25 * commitCloseNumDocs);
      }

      // Sometimes close & reopen writer/manager, to confirm the parallel segments persist:
      if (random().nextInt(commitCloseNumDocs) == 17) {
        if (DEBUG) System.out.println(Thread.currentThread().getName() + ": TEST TOP: close writer @ " + (i+1) + " docs");
        reindexer.close();
        reindexer = null;
        commitCloseNumDocs = (int) (1.25 * commitCloseNumDocs);
      }
    }

    if (reindexer != null) {
      reindexer.close();
    }

    // Verify main index never reflects schema changes beyond mergingSchemaGen:
    try (Directory dir = newFSDirectory(root.resolve("index"));
         IndexReader r = DirectoryReader.open(dir)) {
        for (LeafReaderContext ctx : r.leaves()) {
          LeafReader leaf = ctx.reader();
          NumericDocValues numbers = leaf.getNumericDocValues("number");
          if (numbers != null) {
            int maxDoc = leaf.maxDoc();
            for(int i=0;i<maxDoc;i++) {
              Document doc = leaf.document(i);
              long value = Long.parseLong(doc.get("text").split(" ")[1]);
              assertEquals(i, numbers.nextDoc());
              long dvValue = numbers.longValue();
              if (value == 0) {
                assertEquals(0, dvValue);
              } else {
                assertTrue(dvValue % value == 0);
                assertTrue(dvValue / value <= mergingSchemaGen.get());
              }
            }
          }
        }
      }
  }

  public void testBasic() throws Exception {
    Path tempPath = createTempDir();
    ReindexingReader reindexer = getReindexer(tempPath);

    // Start with initial empty commit:
    reindexer.commit();

    Document doc = new Document();
    doc.add(newTextField("text", "number " + random().nextLong(), Field.Store.YES));
    reindexer.w.addDocument(doc);

    if (DEBUG) System.out.println("TEST: refresh @ 1 doc");
    reindexer.mgr.maybeRefresh();
    DirectoryReader r = reindexer.mgr.acquire();
    if (DEBUG) System.out.println("TEST: got reader=" + r);
    try {
      checkAllNumberDVs(r);
      IndexSearcher s = newSearcher(r);
      testNumericDVSort(s);
      testPointRangeQuery(s);
    } finally {
      reindexer.mgr.release(r);
    }
    //reindexer.printRefCounts();

    if (DEBUG) System.out.println("TEST: commit");
    reindexer.commit();

    doc = new Document();
    doc.add(newTextField("text", "number " + random().nextLong(), Field.Store.YES));
    reindexer.w.addDocument(doc);

    if (DEBUG) System.out.println("TEST: refresh @ 2 docs");
    reindexer.mgr.maybeRefresh();
    //reindexer.printRefCounts();
    r = reindexer.mgr.acquire();
    if (DEBUG) System.out.println("TEST: got reader=" + r);
    try {
      checkAllNumberDVs(r);
      IndexSearcher s = newSearcher(r);
      testNumericDVSort(s);
      testPointRangeQuery(s);
    } finally {
      reindexer.mgr.release(r);
    }

    if (DEBUG) System.out.println("TEST: forceMerge");
    reindexer.w.forceMerge(1);

    if (DEBUG) System.out.println("TEST: commit");
    reindexer.commit();

    if (DEBUG) System.out.println("TEST: refresh after forceMerge");
    reindexer.mgr.maybeRefresh();
    r = reindexer.mgr.acquire();
    if (DEBUG) System.out.println("TEST: got reader=" + r);
    try {
      checkAllNumberDVs(r);
      IndexSearcher s = newSearcher(r);
      testNumericDVSort(s);
      testPointRangeQuery(s);
    } finally {
      reindexer.mgr.release(r);
    }

    if (DEBUG) System.out.println("TEST: close writer");
    reindexer.close();
  }

  public void testRandom() throws Exception {
    Path root = createTempDir();
    ReindexingReader reindexer = null;

    // TODO: separate refresh thread, search threads, indexing threads
    int numDocs = atLeast(TEST_NIGHTLY ? 20000 : 200);
    int maxID = 0;
    int refreshEveryNumDocs = 100;
    int commitCloseNumDocs = 1000;
    for(int i=0;i<numDocs;i++) {
      if (reindexer == null) {
        reindexer = getReindexer(root);
      }

      Document doc = new Document();
      String id;
      String updateID;
      if (maxID > 0 && random().nextInt(10) == 7) {
        // Replace a doc
        id = "" + random().nextInt(maxID);
        updateID = id;
      } else {
        id = "" + (maxID++);
        updateID = null;
      }
        
      doc.add(newStringField("id", id, Field.Store.NO));
      doc.add(newTextField("text", "number " + random().nextLong(), Field.Store.YES));
      if (updateID == null) {
        reindexer.w.addDocument(doc);
      } else {
        reindexer.w.updateDocument(new Term("id", updateID), doc);
      }

      if (random().nextInt(refreshEveryNumDocs) == 17) {
        if (DEBUG) System.out.println("TEST: refresh @ " + (i+1) + " docs");
        reindexer.mgr.maybeRefresh();
        DirectoryReader r = reindexer.mgr.acquire();
        if (DEBUG) System.out.println("TEST: got reader=" + r);
        try {
          checkAllNumberDVs(r);
          IndexSearcher s = newSearcher(r);
          testNumericDVSort(s);
          testPointRangeQuery(s);
        } finally {
          reindexer.mgr.release(r);
        }
        refreshEveryNumDocs = (int) (1.25 * refreshEveryNumDocs);
      }

      if (i > 0 && random().nextInt(10) == 7) {
        // Random delete:
        reindexer.w.deleteDocuments(new Term("id", ""+random().nextInt(i)));
      }

      if (random().nextInt(commitCloseNumDocs) == 17) {
        if (DEBUG) System.out.println("TEST: commit @ " + (i+1) + " docs");
        reindexer.commit();
        commitCloseNumDocs = (int) (1.25 * commitCloseNumDocs);
      }

      // Sometimes close & reopen writer/manager, to confirm the parallel segments persist:
      if (random().nextInt(commitCloseNumDocs) == 17) {
        if (DEBUG) System.out.println("TEST: close writer @ " + (i+1) + " docs");
        reindexer.close();
        reindexer = null;
        commitCloseNumDocs = (int) (1.25 * commitCloseNumDocs);
      }
    }
    if (reindexer != null) {
      reindexer.close();
    }
  }

  private static void checkAllNumberDVs(IndexReader r) throws IOException {
    checkAllNumberDVs(r, "number", true, 1);
  }

  private static void checkAllNumberDVs(IndexReader r, String fieldName, boolean doThrow, int multiplier) throws IOException {
    NumericDocValues numbers = MultiDocValues.getNumericValues(r, fieldName);
    int maxDoc = r.maxDoc();
    boolean failed = false;
    long t0 = System.currentTimeMillis();
    for(int i=0;i<maxDoc;i++) {
      Document oldDoc = r.document(i);
      long value = multiplier * Long.parseLong(oldDoc.get("text").split(" ")[1]);
      assertEquals(i, numbers.nextDoc());
      if (value != numbers.longValue()) {
        System.out.println("FAIL: docID=" + i + " " + oldDoc+ " value=" + value + " number=" + numbers.longValue() + " numbers=" + numbers);
        failed = true;
      } else if (failed) {
        System.out.println("OK: docID=" + i + " " + oldDoc+ " value=" + value + " number=" + numbers.longValue());
      }
    }
    if (failed) {
      if (r instanceof LeafReader == false) {
        System.out.println("TEST FAILED; check leaves");
        for(LeafReaderContext ctx : r.leaves()) {
          System.out.println("CHECK LEAF=" + ctx.reader());
          checkAllNumberDVs(ctx.reader(), fieldName, false, 1);
        }
      }
      if (doThrow) {
        assertFalse("FAILED field=" + fieldName + " r=" + r, failed);
      } else {
        System.out.println("FAILED field=" + fieldName + " r=" + r);
      }
    }
  }

  private static void testNumericDVSort(IndexSearcher s) throws IOException {
    // Confirm we can sort by the new DV field:
    TopDocs hits = s.search(new MatchAllDocsQuery(), 100, new Sort(new SortField("number", SortField.Type.LONG)));
    long last = Long.MIN_VALUE;
    for(ScoreDoc scoreDoc : hits.scoreDocs) {
      long value = Long.parseLong(s.doc(scoreDoc.doc).get("text").split(" ")[1]);
      assertTrue(value >= last);
      assertEquals(value, ((Long) ((FieldDoc) scoreDoc).fields[0]).longValue());
      last = value;
    }
  }

  private static void testPointRangeQuery(IndexSearcher s) throws IOException {
    for(int i=0;i<100;i++) {
      // Confirm we can range search by the new indexed (numeric) field:
      long min = random().nextLong();
      long max = random().nextLong();
      if (min > max) {
        long x = min;
        min = max;
        max = x;
      }

      TopDocs hits = s.search(LongPoint.newRangeQuery("number", min, max), 100);
      for(ScoreDoc scoreDoc : hits.scoreDocs) {
        long value = Long.parseLong(s.doc(scoreDoc.doc).get("text").split(" ")[1]);
        assertTrue(value >= min);
        assertTrue(value <= max);
      }

      Arrays.sort(hits.scoreDocs,
                  new Comparator<ScoreDoc>() {
                    @Override
                    public int compare(ScoreDoc a, ScoreDoc b) {
                      return a.doc - b.doc;
                    }
                  });

      NumericDocValues numbers = MultiDocValues.getNumericValues(s.getIndexReader(), "number");
      for(ScoreDoc hit : hits.scoreDocs) {
        if (numbers.docID() < hit.doc) {
          numbers.advance(hit.doc);
        }
        assertEquals(hit.doc, numbers.docID());
        long value = Long.parseLong(s.doc(hit.doc).get("text").split(" ")[1]);
        assertEquals(value, numbers.longValue());
      }
    }
  }

  // TODO: maybe the leading id could be further restricted?  It's from StringHelper.idToString:
  static final Pattern SEG_GEN_SUB_DIR_PATTERN = Pattern.compile("^[a-z0-9]+_([0-9]+)$");

  private static List<Path> segSubDirs(Path segsPath) throws IOException {
    List<Path> result = new ArrayList<>();
    try (DirectoryStream<Path> stream = Files.newDirectoryStream(segsPath)) {
      for (Path path : stream) {
        // Must be form <segIDString>_<longGen>
        if (Files.isDirectory(path) && SEG_GEN_SUB_DIR_PATTERN.matcher(path.getFileName().toString()).matches()) {
          result.add(path);
        }
      }
    }

    return result;
  }

  // TODO: test exceptions
}