package org.apache.lucene.facet.taxonomy.directory;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.lucene.analysis.TokenStream;
import org.apache.lucene.analysis.tokenattributes.CharTermAttribute;
import org.apache.lucene.analysis.tokenattributes.PositionIncrementAttribute;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.document.FieldType;
import org.apache.lucene.document.StringField;
import org.apache.lucene.document.TextField;
import org.apache.lucene.facet.FacetsConfig;
import org.apache.lucene.facet.taxonomy.FacetLabel;
import org.apache.lucene.facet.taxonomy.TaxonomyReader;
import org.apache.lucene.facet.taxonomy.TaxonomyWriter;
import org.apache.lucene.facet.taxonomy.writercache.TaxonomyWriterCache;
import org.apache.lucene.facet.taxonomy.writercache.Cl2oTaxonomyWriterCache;
import org.apache.lucene.facet.taxonomy.writercache.LruTaxonomyWriterCache;
import org.apache.lucene.index.AtomicReader;
import org.apache.lucene.index.AtomicReaderContext;
import org.apache.lucene.index.CorruptIndexException; // javadocs
import org.apache.lucene.index.DirectoryReader;
import org.apache.lucene.index.DocsEnum;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.IndexWriterConfig.OpenMode;
import org.apache.lucene.index.IndexWriterConfig;
import org.apache.lucene.index.LogByteSizeMergePolicy;
import org.apache.lucene.index.ReaderManager;
import org.apache.lucene.index.SegmentInfos;
import org.apache.lucene.index.Terms;
import org.apache.lucene.index.TermsEnum;
import org.apache.lucene.index.TieredMergePolicy;
import org.apache.lucene.store.AlreadyClosedException;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.LockObtainFailedException; // javadocs
import org.apache.lucene.store.NativeFSLockFactory;
import org.apache.lucene.store.SimpleFSLockFactory;
import org.apache.lucene.util.BytesRef;
import org.apache.lucene.util.Version;

/*
 * 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.
 */

/**
 * {@link TaxonomyWriter} which uses a {@link Directory} to store the taxonomy
 * information on disk, and keeps an additional in-memory cache of some or all
 * categories.
 * <p>
 * In addition to the permanently-stored information in the {@link Directory},
 * efficiency dictates that we also keep an in-memory cache of <B>recently
 * seen</B> or <B>all</B> categories, so that we do not need to go back to disk
 * for every category addition to see which ordinal this category already has,
 * if any. A {@link TaxonomyWriterCache} object determines the specific caching
 * algorithm used.
 * <p>
 * This class offers some hooks for extending classes to control the
 * {@link IndexWriter} instance that is used. See {@link #openIndexWriter}.
 * 
 * @lucene.experimental
 */
public class DirectoryTaxonomyWriter implements TaxonomyWriter {
  
  /**
   * Property name of user commit data that contains the index epoch. The epoch
   * changes whenever the taxonomy is recreated (i.e. opened with
   * {@link OpenMode#CREATE}.
   * <p>
   * Applications should not use this property in their commit data because it
   * will be overridden by this taxonomy writer.
   */
  public static final String INDEX_EPOCH = "index.epoch";
  
  private final Directory dir;
  private final IndexWriter indexWriter;
  private final TaxonomyWriterCache cache;
  private final AtomicInteger cacheMisses = new AtomicInteger(0);
  
  // Records the taxonomy index epoch, updated on replaceTaxonomy as well.
  private long indexEpoch;

  private SinglePositionTokenStream parentStream = new SinglePositionTokenStream(Consts.PAYLOAD_PARENT);
  private Field parentStreamField;
  private Field fullPathField;
  private int cacheMissesUntilFill = 11;
  private boolean shouldFillCache = true;
  
  // even though lazily initialized, not volatile so that access to it is
  // faster. we keep a volatile boolean init instead.
  private ReaderManager readerManager;
  private volatile boolean initializedReaderManager = false;
  private volatile boolean shouldRefreshReaderManager;
  
  /**
   * We call the cache "complete" if we know that every category in our
   * taxonomy is in the cache. When the cache is <B>not</B> complete, and
   * we can't find a category in the cache, we still need to look for it
   * in the on-disk index; Therefore when the cache is not complete, we
   * need to open a "reader" to the taxonomy index.
   * The cache becomes incomplete if it was never filled with the existing
   * categories, or if a put() to the cache ever returned true (meaning
   * that some of the cached data was cleared).
   */
  private volatile boolean cacheIsComplete;
  private volatile boolean isClosed = false;
  private volatile TaxonomyIndexArrays taxoArrays;
  private volatile int nextID;

  /** Reads the commit data from a Directory. */
  private static Map<String, String> readCommitData(Directory dir) throws IOException {
    SegmentInfos infos = new SegmentInfos();
    infos.read(dir);
    return infos.getUserData();
  }
  
  /**
   * Forcibly unlocks the taxonomy in the named directory.
   * <P>
   * Caution: this should only be used by failure recovery code, when it is
   * known that no other process nor thread is in fact currently accessing
   * this taxonomy.
   * <P>
   * This method is unnecessary if your {@link Directory} uses a
   * {@link NativeFSLockFactory} instead of the default
   * {@link SimpleFSLockFactory}. When the "native" lock is used, a lock
   * does not stay behind forever when the process using it dies. 
   */
  public static void unlock(Directory directory) throws IOException {
    IndexWriter.unlock(directory);
  }

  /**
   * Construct a Taxonomy writer.
   * 
   * @param directory
   *    The {@link Directory} in which to store the taxonomy. Note that
   *    the taxonomy is written directly to that directory (not to a
   *    subdirectory of it).
   * @param openMode
   *    Specifies how to open a taxonomy for writing: <code>APPEND</code>
   *    means open an existing index for append (failing if the index does
   *    not yet exist). <code>CREATE</code> means create a new index (first
   *    deleting the old one if it already existed).
   *    <code>APPEND_OR_CREATE</code> appends to an existing index if there
   *    is one, otherwise it creates a new index.
   * @param cache
   *    A {@link TaxonomyWriterCache} implementation which determines
   *    the in-memory caching policy. See for example
   *    {@link LruTaxonomyWriterCache} and {@link Cl2oTaxonomyWriterCache}.
   *    If null or missing, {@link #defaultTaxonomyWriterCache()} is used.
   * @throws CorruptIndexException
   *     if the taxonomy is corrupted.
   * @throws LockObtainFailedException
   *     if the taxonomy is locked by another writer. If it is known
   *     that no other concurrent writer is active, the lock might
   *     have been left around by an old dead process, and should be
   *     removed using {@link #unlock(Directory)}.
   * @throws IOException
   *     if another error occurred.
   */
  public DirectoryTaxonomyWriter(Directory directory, OpenMode openMode,
      TaxonomyWriterCache cache) throws IOException {

    dir = directory;
    IndexWriterConfig config = createIndexWriterConfig(openMode);
    indexWriter = openIndexWriter(dir, config);

    // verify (to some extent) that merge policy in effect would preserve category docids 
    assert !(indexWriter.getConfig().getMergePolicy() instanceof TieredMergePolicy) : 
      "for preserving category docids, merging none-adjacent segments is not allowed";
    
    // after we opened the writer, and the index is locked, it's safe to check
    // the commit data and read the index epoch
    openMode = config.getOpenMode();
    if (!DirectoryReader.indexExists(directory)) {
      indexEpoch = 1;
    } else {
      String epochStr = null;
      Map<String, String> commitData = readCommitData(directory);
      if (commitData != null) {
        epochStr = commitData.get(INDEX_EPOCH);
      }
      // no commit data, or no epoch in it means an old taxonomy, so set its epoch to 1, for lack
      // of a better value.
      indexEpoch = epochStr == null ? 1 : Long.parseLong(epochStr, 16);
    }
    
    if (openMode == OpenMode.CREATE) {
      ++indexEpoch;
    }
    
    FieldType ft = new FieldType(TextField.TYPE_NOT_STORED);
    ft.setOmitNorms(true);
    parentStreamField = new Field(Consts.FIELD_PAYLOADS, parentStream, ft);
    fullPathField = new StringField(Consts.FULL, "", Field.Store.YES);

    nextID = indexWriter.maxDoc();

    if (cache == null) {
      cache = defaultTaxonomyWriterCache();
    }
    this.cache = cache;

    if (nextID == 0) {
      cacheIsComplete = true;
      // Make sure that the taxonomy always contain the root category
      // with category id 0.
      addCategory(new FacetLabel());
    } else {
      // There are some categories on the disk, which we have not yet
      // read into the cache, and therefore the cache is incomplete.
      // We choose not to read all the categories into the cache now,
      // to avoid terrible performance when a taxonomy index is opened
      // to add just a single category. We will do it later, after we
      // notice a few cache misses.
      cacheIsComplete = false;
    }
  }

  /**
   * Open internal index writer, which contains the taxonomy data.
   * <p>
   * Extensions may provide their own {@link IndexWriter} implementation or instance. 
   * <br><b>NOTE:</b> the instance this method returns will be closed upon calling
   * to {@link #close()}.
   * <br><b>NOTE:</b> the merge policy in effect must not merge none adjacent segments. See
   * comment in {@link #createIndexWriterConfig(IndexWriterConfig.OpenMode)} for the logic behind this.
   *  
   * @see #createIndexWriterConfig(IndexWriterConfig.OpenMode)
   * 
   * @param directory
   *          the {@link Directory} on top of which an {@link IndexWriter}
   *          should be opened.
   * @param config
   *          configuration for the internal index writer.
   */
  protected IndexWriter openIndexWriter(Directory directory, IndexWriterConfig config)
      throws IOException {
    return new IndexWriter(directory, config);
  }

  /**
   * Create the {@link IndexWriterConfig} that would be used for opening the internal index writer.
   * <br>Extensions can configure the {@link IndexWriter} as they see fit,
   * including setting a {@link org.apache.lucene.index.MergeScheduler merge-scheduler}, or
   * {@link org.apache.lucene.index.IndexDeletionPolicy deletion-policy}, different RAM size
   * etc.<br>
   * <br><b>NOTE:</b> internal docids of the configured index must not be altered.
   * For that, categories are never deleted from the taxonomy index.
   * In addition, merge policy in effect must not merge none adjacent segments.
   * 
   * @see #openIndexWriter(Directory, IndexWriterConfig)
   * 
   * @param openMode see {@link OpenMode}
   */
  protected IndexWriterConfig createIndexWriterConfig(OpenMode openMode) {
    // TODO: should we use a more optimized Codec, e.g. Pulsing (or write custom)?
    // The taxonomy has a unique structure, where each term is associated with one document

    // :Post-Release-Update-Version.LUCENE_XY:
    // Make sure we use a MergePolicy which always merges adjacent segments and thus
    // keeps the doc IDs ordered as well (this is crucial for the taxonomy index).
    return new IndexWriterConfig(Version.LUCENE_50, null).setOpenMode(openMode).setMergePolicy(
        new LogByteSizeMergePolicy());
  }
  
  /** Opens a {@link ReaderManager} from the internal {@link IndexWriter}. */
  private void initReaderManager() throws IOException {
    if (!initializedReaderManager) {
      synchronized (this) {
        // verify that the taxo-writer hasn't been closed on us.
        ensureOpen();
        if (!initializedReaderManager) {
          readerManager = new ReaderManager(indexWriter, false);
          shouldRefreshReaderManager = false;
          initializedReaderManager = true;
        }
      }
    }
  }

  /**
   * Creates a new instance with a default cache as defined by
   * {@link #defaultTaxonomyWriterCache()}.
   */
  public DirectoryTaxonomyWriter(Directory directory, OpenMode openMode)
  throws IOException {
    this(directory, openMode, defaultTaxonomyWriterCache());
  }

  /**
   * Defines the default {@link TaxonomyWriterCache} to use in constructors
   * which do not specify one.
   * <P>  
   * The current default is {@link Cl2oTaxonomyWriterCache} constructed
   * with the parameters (1024, 0.15f, 3), i.e., the entire taxonomy is
   * cached in memory while building it.
   */
  public static TaxonomyWriterCache defaultTaxonomyWriterCache() {
    return new Cl2oTaxonomyWriterCache(1024, 0.15f, 3);
  }

  /** Create this with {@code OpenMode.CREATE_OR_APPEND}. */
  public DirectoryTaxonomyWriter(Directory d) throws IOException {
    this(d, OpenMode.CREATE_OR_APPEND);
  }

  /**
   * Frees used resources as well as closes the underlying {@link IndexWriter},
   * which commits whatever changes made to it to the underlying
   * {@link Directory}.
   */
  @Override
  public synchronized void close() throws IOException {
    if (!isClosed) {
      commit();
      doClose();
    }
  }
  
  private void doClose() throws IOException {
    indexWriter.close();
    isClosed = true;
    closeResources();
  }

  /**
   * A hook for extending classes to close additional resources that were used.
   * The default implementation closes the {@link IndexReader} as well as the
   * {@link TaxonomyWriterCache} instances that were used. <br>
   * <b>NOTE:</b> if you override this method, you should include a
   * <code>super.closeResources()</code> call in your implementation.
   */
  protected synchronized void closeResources() throws IOException {
    if (initializedReaderManager) {
      readerManager.close();
      readerManager = null;
      initializedReaderManager = false;
    }
    if (cache != null) {
      cache.close();
    }
  }

  /**
   * Look up the given category in the cache and/or the on-disk storage,
   * returning the category's ordinal, or a negative number in case the
   * category does not yet exist in the taxonomy.
   */
  protected synchronized int findCategory(FacetLabel categoryPath) throws IOException {
    // If we can find the category in the cache, or we know the cache is
    // complete, we can return the response directly from it
    int res = cache.get(categoryPath);
    if (res >= 0 || cacheIsComplete) {
      return res;
    }

    cacheMisses.incrementAndGet();
    // After a few cache misses, it makes sense to read all the categories
    // from disk and into the cache. The reason not to do this on the first
    // cache miss (or even when opening the writer) is that it will
    // significantly slow down the case when a taxonomy is opened just to
    // add one category. The idea only spending a long time on reading
    // after enough time was spent on cache misses is known as an "online
    // algorithm".
    perhapsFillCache();
    res = cache.get(categoryPath);
    if (res >= 0 || cacheIsComplete) {
      // if after filling the cache from the info on disk, the category is in it
      // or the cache is complete, return whatever cache.get returned.
      return res;
    }

    // if we get here, it means the category is not in the cache, and it is not
    // complete, and therefore we must look for the category on disk.
    
    // We need to get an answer from the on-disk index.
    initReaderManager();

    int doc = -1;
    DirectoryReader reader = readerManager.acquire();
    try {
      final BytesRef catTerm = new BytesRef(FacetsConfig.pathToString(categoryPath.components, categoryPath.length));
      TermsEnum termsEnum = null; // reuse
      DocsEnum docs = null; // reuse
      for (AtomicReaderContext ctx : reader.leaves()) {
        Terms terms = ctx.reader().terms(Consts.FULL);
        if (terms != null) {
          termsEnum = terms.iterator(termsEnum);
          if (termsEnum.seekExact(catTerm)) {
            // liveDocs=null because the taxonomy has no deletes
            docs = termsEnum.docs(null, docs, 0 /* freqs not required */);
            // if the term was found, we know it has exactly one document.
            doc = docs.nextDoc() + ctx.docBase;
            break;
          }
        }
      }
    } finally {
      readerManager.release(reader);
    }
    if (doc > 0) {
      addToCache(categoryPath, doc);
    }
    return doc;
  }

  @Override
  public int addCategory(FacetLabel categoryPath) throws IOException {
    ensureOpen();
    // check the cache outside the synchronized block. this results in better
    // concurrency when categories are there.
    int res = cache.get(categoryPath);
    if (res < 0) {
      // the category is not in the cache - following code cannot be executed in parallel.
      synchronized (this) {
        res = findCategory(categoryPath);
        if (res < 0) {
          // This is a new category, and we need to insert it into the index
          // (and the cache). Actually, we might also need to add some of
          // the category's ancestors before we can add the category itself
          // (while keeping the invariant that a parent is always added to
          // the taxonomy before its child). internalAddCategory() does all
          // this recursively
          res = internalAddCategory(categoryPath);
        }
      }
    }
    return res;
  }

  /**
   * Add a new category into the index (and the cache), and return its new
   * ordinal.
   * <p>
   * Actually, we might also need to add some of the category's ancestors
   * before we can add the category itself (while keeping the invariant that a
   * parent is always added to the taxonomy before its child). We do this by
   * recursion.
   */
  private int internalAddCategory(FacetLabel cp) throws IOException {
    // Find our parent's ordinal (recursively adding the parent category
    // to the taxonomy if it's not already there). Then add the parent
    // ordinal as payloads (rather than a stored field; payloads can be
    // more efficiently read into memory in bulk by LuceneTaxonomyReader)
    int parent;
    if (cp.length > 1) {
      FacetLabel parentPath = cp.subpath(cp.length - 1);
      parent = findCategory(parentPath);
      if (parent < 0) {
        parent = internalAddCategory(parentPath);
      }
    } else if (cp.length == 1) {
      parent = TaxonomyReader.ROOT_ORDINAL;
    } else {
      parent = TaxonomyReader.INVALID_ORDINAL;
    }
    int id = addCategoryDocument(cp, parent);

    return id;
  }

  /**
   * Verifies that this instance wasn't closed, or throws
   * {@link AlreadyClosedException} if it is.
   */
  protected final void ensureOpen() {
    if (isClosed) {
      throw new AlreadyClosedException("The taxonomy writer has already been closed");
    }
  }
  
  /**
   * Note that the methods calling addCategoryDocument() are synchornized, so
   * this method is effectively synchronized as well.
   */
  private int addCategoryDocument(FacetLabel categoryPath, int parent) throws IOException {
    // Before Lucene 2.9, position increments >=0 were supported, so we
    // added 1 to parent to allow the parent -1 (the parent of the root).
    // Unfortunately, starting with Lucene 2.9, after LUCENE-1542, this is
    // no longer enough, since 0 is not encoded consistently either (see
    // comment in SinglePositionTokenStream). But because we must be
    // backward-compatible with existing indexes, we can't just fix what
    // we write here (e.g., to write parent+2), and need to do a workaround
    // in the reader (which knows that anyway only category 0 has a parent
    // -1).    
    parentStream.set(Math.max(parent + 1, 1));
    Document d = new Document();
    d.add(parentStreamField);

    fullPathField.setStringValue(FacetsConfig.pathToString(categoryPath.components, categoryPath.length));
    d.add(fullPathField);

    // Note that we do no pass an Analyzer here because the fields that are
    // added to the Document are untokenized or contains their own TokenStream.
    // Therefore the IndexWriter's Analyzer has no effect.
    indexWriter.addDocument(d);
    int id = nextID++;

    // added a category document, mark that ReaderManager is not up-to-date
    shouldRefreshReaderManager = true;
    
    // also add to the parent array
    taxoArrays = getTaxoArrays().add(id, parent);

    // NOTE: this line must be executed last, or else the cache gets updated
    // before the parents array (LUCENE-4596)
    addToCache(categoryPath, id);

    return id;
  }

  private static class SinglePositionTokenStream extends TokenStream {
    private CharTermAttribute termAtt;
    private PositionIncrementAttribute posIncrAtt;
    private boolean returned;
    private int val;
    private final String word;

    public SinglePositionTokenStream(String word) {
      termAtt = addAttribute(CharTermAttribute.class);
      posIncrAtt = addAttribute(PositionIncrementAttribute.class);
      this.word = word;
      returned = true;
    }

    /**
     * Set the value we want to keep, as the position increment.
     * Note that when TermPositions.nextPosition() is later used to
     * retrieve this value, val-1 will be returned, not val.
     * <P>
     * IMPORTANT NOTE: Before Lucene 2.9, val>=0 were safe (for val==0,
     * the retrieved position would be -1). But starting with Lucene 2.9,
     * this unfortunately changed, and only val>0 are safe. val=0 can
     * still be used, but don't count on the value you retrieve later
     * (it could be 0 or -1, depending on circumstances or versions).
     * This change is described in Lucene's JIRA: LUCENE-1542. 
     */
    public void set(int val) {
      this.val = val;
      returned = false;
    }

    @Override
    public boolean incrementToken() throws IOException {
      if (returned) {
        return false;
      }
      clearAttributes();
      posIncrAtt.setPositionIncrement(val);
      termAtt.setEmpty();
      termAtt.append(word);
      returned = true;
      return true;
    }
  }

  private void addToCache(FacetLabel categoryPath, int id) throws IOException {
    if (cache.put(categoryPath, id)) {
      // If cache.put() returned true, it means the cache was limited in
      // size, became full, and parts of it had to be evicted. It is
      // possible that a relatively-new category that isn't yet visible
      // to our 'reader' was evicted, and therefore we must now refresh 
      // the reader.
      refreshReaderManager();
      cacheIsComplete = false;
    }
  }

  private synchronized void refreshReaderManager() throws IOException {
    // this method is synchronized since it cannot happen concurrently with
    // addCategoryDocument -- when this method returns, we must know that the
    // reader manager's state is current. also, it sets shouldRefresh to false, 
    // and this cannot overlap with addCatDoc too.
    // NOTE: since this method is sync'ed, it can call maybeRefresh, instead of
    // maybeRefreshBlocking. If ever this is changed, make sure to change the
    // call too.
    if (shouldRefreshReaderManager && initializedReaderManager) {
      readerManager.maybeRefresh();
      shouldRefreshReaderManager = false;
    }
  }
  
  @Override
  public synchronized void commit() throws IOException {
    ensureOpen();
    // LUCENE-4972: if we always call setCommitData, we create empty commits
    String epochStr = indexWriter.getCommitData().get(INDEX_EPOCH);
    if (epochStr == null || Long.parseLong(epochStr, 16) != indexEpoch) {
      indexWriter.setCommitData(combinedCommitData(indexWriter.getCommitData()));
    }
    indexWriter.commit();
  }

  /** Combine original user data with the taxonomy epoch. */
  private Map<String,String> combinedCommitData(Map<String,String> commitData) {
    Map<String,String> m = new HashMap<String, String>();
    if (commitData != null) {
      m.putAll(commitData);
    }
    m.put(INDEX_EPOCH, Long.toString(indexEpoch, 16));
    return m;
  }
  
  @Override
  public void setCommitData(Map<String,String> commitUserData) {
    indexWriter.setCommitData(combinedCommitData(commitUserData));
  }
  
  @Override
  public Map<String,String> getCommitData() {
    return combinedCommitData(indexWriter.getCommitData());
  }
  
  /**
   * prepare most of the work needed for a two-phase commit.
   * See {@link IndexWriter#prepareCommit}.
   */
  @Override
  public synchronized void prepareCommit() throws IOException {
    ensureOpen();
    // LUCENE-4972: if we always call setCommitData, we create empty commits
    String epochStr = indexWriter.getCommitData().get(INDEX_EPOCH);
    if (epochStr == null || Long.parseLong(epochStr, 16) != indexEpoch) {
      indexWriter.setCommitData(combinedCommitData(indexWriter.getCommitData()));
    }
    indexWriter.prepareCommit();
  }
  
  @Override
  public int getSize() {
    ensureOpen();
    return nextID;
  }
  
  /**
   * Set the number of cache misses before an attempt is made to read the entire
   * taxonomy into the in-memory cache.
   * <p>
   * This taxonomy writer holds an in-memory cache of recently seen categories
   * to speed up operation. On each cache-miss, the on-disk index needs to be
   * consulted. When an existing taxonomy is opened, a lot of slow disk reads
   * like that are needed until the cache is filled, so it is more efficient to
   * read the entire taxonomy into memory at once. We do this complete read
   * after a certain number (defined by this method) of cache misses.
   * <p>
   * If the number is set to {@code 0}, the entire taxonomy is read into the
   * cache on first use, without fetching individual categories first.
   * <p>
   * NOTE: it is assumed that this method is called immediately after the
   * taxonomy writer has been created.
   */
  public void setCacheMissesUntilFill(int i) {
    ensureOpen();
    cacheMissesUntilFill = i;
  }
  
  // we need to guarantee that if several threads call this concurrently, only
  // one executes it, and after it returns, the cache is updated and is either
  // complete or not.
  private synchronized void perhapsFillCache() throws IOException {
    if (cacheMisses.get() < cacheMissesUntilFill) {
      return;
    }
    
    if (!shouldFillCache) {
      // we already filled the cache once, there's no need to re-fill it
      return;
    }
    shouldFillCache = false;
    
    initReaderManager();

    boolean aborted = false;
    DirectoryReader reader = readerManager.acquire();
    try {
      TermsEnum termsEnum = null;
      DocsEnum docsEnum = null;
      for (AtomicReaderContext ctx : reader.leaves()) {
        Terms terms = ctx.reader().terms(Consts.FULL);
        if (terms != null) { // cannot really happen, but be on the safe side
          termsEnum = terms.iterator(termsEnum);
          while (termsEnum.next() != null) {
            if (!cache.isFull()) {
              BytesRef t = termsEnum.term();
              // Since we guarantee uniqueness of categories, each term has exactly
              // one document. Also, since we do not allow removing categories (and
              // hence documents), there are no deletions in the index. Therefore, it
              // is sufficient to call next(), and then doc(), exactly once with no
              // 'validation' checks.
              FacetLabel cp = new FacetLabel(FacetsConfig.stringToPath(t.utf8ToString()));
              docsEnum = termsEnum.docs(null, docsEnum, DocsEnum.FLAG_NONE);
              boolean res = cache.put(cp, docsEnum.nextDoc() + ctx.docBase);
              assert !res : "entries should not have been evicted from the cache";
            } else {
              // the cache is full and the next put() will evict entries from it, therefore abort the iteration.
              aborted = true;
              break;
            }
          }
        }
        if (aborted) {
          break;
        }
      }
    } finally {
      readerManager.release(reader);
    }

    cacheIsComplete = !aborted;
    if (cacheIsComplete) {
      synchronized (this) {
        // everything is in the cache, so no need to keep readerManager open.
        // this block is executed in a sync block so that it works well with
        // initReaderManager called in parallel.
        readerManager.close();
        readerManager = null;
        initializedReaderManager = false;
      }
    }
  }

  private TaxonomyIndexArrays getTaxoArrays() throws IOException {
    if (taxoArrays == null) {
      synchronized (this) {
        if (taxoArrays == null) {
          initReaderManager();
          DirectoryReader reader = readerManager.acquire();
          try {
            // according to Java Concurrency, this might perform better on some
            // JVMs, since the object initialization doesn't happen on the
            // volatile member.
            TaxonomyIndexArrays tmpArrays = new TaxonomyIndexArrays(reader);
            taxoArrays = tmpArrays;
          } finally {
            readerManager.release(reader);
          }
        }
      }
    }
    return taxoArrays;
  }
  
  @Override
  public int getParent(int ordinal) throws IOException {
    ensureOpen();
    // Note: the following if() just enforces that a user can never ask
    // for the parent of a nonexistant category - even if the parent array
    // was allocated bigger than it really needs to be.
    if (ordinal >= nextID) {
      throw new ArrayIndexOutOfBoundsException("requested ordinal is bigger than the largest ordinal in the taxonomy");
    }
    
    int[] parents = getTaxoArrays().parents();
    assert ordinal < parents.length : "requested ordinal (" + ordinal + "); parents.length (" + parents.length + ") !";
    return parents[ordinal];
  }
  
  /**
   * Takes the categories from the given taxonomy directory, and adds the
   * missing ones to this taxonomy. Additionally, it fills the given
   * {@link OrdinalMap} with a mapping from the original ordinal to the new
   * ordinal.
   */
  public void addTaxonomy(Directory taxoDir, OrdinalMap map) throws IOException {
    ensureOpen();
    DirectoryReader r = DirectoryReader.open(taxoDir);
    try {
      final int size = r.numDocs();
      final OrdinalMap ordinalMap = map;
      ordinalMap.setSize(size);
      int base = 0;
      TermsEnum te = null;
      DocsEnum docs = null;
      for (final AtomicReaderContext ctx : r.leaves()) {
        final AtomicReader ar = ctx.reader();
        final Terms terms = ar.terms(Consts.FULL);
        te = terms.iterator(te);
        while (te.next() != null) {
          FacetLabel cp = new FacetLabel(FacetsConfig.stringToPath(te.term().utf8ToString()));
          final int ordinal = addCategory(cp);
          docs = te.docs(null, docs, DocsEnum.FLAG_NONE);
          ordinalMap.addMapping(docs.nextDoc() + base, ordinal);
        }
        base += ar.maxDoc(); // no deletions, so we're ok
      }
      ordinalMap.addDone();
    } finally {
      r.close();
    }
  }

  /**
   * Mapping from old ordinal to new ordinals, used when merging indexes 
   * wit separate taxonomies.
   * <p> 
   * addToTaxonomies() merges one or more taxonomies into the given taxonomy
   * (this). An OrdinalMap is filled for each of the added taxonomies,
   * containing the new ordinal (in the merged taxonomy) of each of the
   * categories in the old taxonomy.
   * <P>  
   * There exist two implementations of OrdinalMap: MemoryOrdinalMap and
   * DiskOrdinalMap. As their names suggest, the former keeps the map in
   * memory and the latter in a temporary disk file. Because these maps will
   * later be needed one by one (to remap the counting lists), not all at the
   * same time, it is recommended to put the first taxonomy's map in memory,
   * and all the rest on disk (later to be automatically read into memory one
   * by one, when needed).
   */
  public static interface OrdinalMap {
    /**
     * Set the size of the map. This MUST be called before addMapping().
     * It is assumed (but not verified) that addMapping() will then be
     * called exactly 'size' times, with different origOrdinals between 0
     * and size-1.  
     */
    public void setSize(int size) throws IOException;

    /** Record a mapping. */
    public void addMapping(int origOrdinal, int newOrdinal) throws IOException;

    /**
     * Call addDone() to say that all addMapping() have been done.
     * In some implementations this might free some resources. 
     */
    public void addDone() throws IOException;

    /**
     * Return the map from the taxonomy's original (consecutive) ordinals
     * to the new taxonomy's ordinals. If the map has to be read from disk
     * and ordered appropriately, it is done when getMap() is called.
     * getMap() should only be called once, and only when the map is actually
     * needed. Calling it will also free all resources that the map might
     * be holding (such as temporary disk space), other than the returned int[].
     */
    public int[] getMap() throws IOException;
  }

  /**
   * {@link OrdinalMap} maintained in memory
   */
  public static final class MemoryOrdinalMap implements OrdinalMap {
    int[] map;

    /** Sole constructor. */
    public MemoryOrdinalMap() {
    }

    @Override
    public void setSize(int taxonomySize) {
      map = new int[taxonomySize];
    }
    @Override
    public void addMapping(int origOrdinal, int newOrdinal) {
      map[origOrdinal] = newOrdinal;
    }
    @Override
    public void addDone() { /* nothing to do */ }
    @Override
    public int[] getMap() {
      return map;
    }
  }

  /**
   * {@link OrdinalMap} maintained on file system
   */
  public static final class DiskOrdinalMap implements OrdinalMap {
    File tmpfile;
    DataOutputStream out;

    /** Sole constructor. */
    public DiskOrdinalMap(File tmpfile) throws FileNotFoundException {
      this.tmpfile = tmpfile;
      out = new DataOutputStream(new BufferedOutputStream(
          new FileOutputStream(tmpfile)));
    }

    @Override
    public void addMapping(int origOrdinal, int newOrdinal) throws IOException {
      out.writeInt(origOrdinal);
      out.writeInt(newOrdinal);
    }

    @Override
    public void setSize(int taxonomySize) throws IOException {
      out.writeInt(taxonomySize);
    }

    @Override
    public void addDone() throws IOException {
      if (out!=null) {
        out.close();
        out = null;
      }
    }

    int[] map = null;

    @Override
    public int[] getMap() throws IOException {
      if (map!=null) {
        return map;
      }
      addDone(); // in case this wasn't previously called
      DataInputStream in = new DataInputStream(new BufferedInputStream(
          new FileInputStream(tmpfile)));
      map = new int[in.readInt()];
      // NOTE: The current code assumes here that the map is complete,
      // i.e., every ordinal gets one and exactly one value. Otherwise,
      // we may run into an EOF here, or vice versa, not read everything.
      for (int i=0; i<map.length; i++) {
        int origordinal = in.readInt();
        int newordinal = in.readInt();
        map[origordinal] = newordinal;
      }
      in.close();
      // Delete the temporary file, which is no longer needed.
      if (!tmpfile.delete()) {
        tmpfile.deleteOnExit();
      }
      return map;
    }
  }

  /**
   * Rollback changes to the taxonomy writer and closes the instance. Following
   * this method the instance becomes unusable (calling any of its API methods
   * will yield an {@link AlreadyClosedException}).
   */
  @Override
  public synchronized void rollback() throws IOException {
    ensureOpen();
    indexWriter.rollback();
    doClose();
  }
  
  /**
   * Replaces the current taxonomy with the given one. This method should
   * generally be called in conjunction with
   * {@link IndexWriter#addIndexes(Directory...)} to replace both the taxonomy
   * as well as the search index content.
   */
  public synchronized void replaceTaxonomy(Directory taxoDir) throws IOException {
    // replace the taxonomy by doing IW optimized operations
    indexWriter.deleteAll();
    indexWriter.addIndexes(taxoDir);
    shouldRefreshReaderManager = true;
    initReaderManager(); // ensure that it's initialized
    refreshReaderManager();
    nextID = indexWriter.maxDoc();
    taxoArrays = null; // must nullify so that it's re-computed next time it's needed
    
    // need to clear the cache, so that addCategory won't accidentally return
    // old categories that are in the cache.
    cache.clear();
    cacheIsComplete = false;
    shouldFillCache = true;
    cacheMisses.set(0);
    
    // update indexEpoch as a taxonomy replace is just like it has be recreated
    ++indexEpoch;
  }

  /** Returns the {@link Directory} of this taxonomy writer. */
  public Directory getDirectory() {
    return dir;
  }
  
  /**
   * Used by {@link DirectoryTaxonomyReader} to support NRT.
   * <p>
   * <b>NOTE:</b> you should not use the obtained {@link IndexWriter} in any
   * way, other than opening an IndexReader on it, or otherwise, the taxonomy
   * index may become corrupt!
   */
  final IndexWriter getInternalIndexWriter() {
    return indexWriter;
  }
  
  /** Expert: returns current index epoch, if this is a
   * near-real-time reader.  Used by {@link
   * DirectoryTaxonomyReader} to support NRT. 
   *
   * @lucene.internal */
  public final long getTaxonomyEpoch() {
    return indexEpoch;
  }
}