content: package org.apache.lucene.index;
  
  /**
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  import java.io.IOException;
  import java.io.PrintStream;
  import java.util.List;
  import java.util.ArrayList;
  import java.util.Date;
  import java.util.Comparator;
  import java.util.Collections;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.concurrent.atomic.AtomicLong;
  
  import org.apache.lucene.search.DocIdSet;
  import org.apache.lucene.search.DocIdSetIterator;
  import org.apache.lucene.search.Query;
  import org.apache.lucene.search.QueryWrapperFilter;
  import org.apache.lucene.util.BytesRef;
  
  /* Tracks the stream of {@link BuffereDeletes}.
   * When DocumensWriter flushes, its buffered
   * deletes are appended to this stream.  We later
   * apply these deletes (resolve them to the actual
   * docIDs, per segment) when a merge is started
   * (only to the to-be-merged segments).  We
   * also apply to all segments when NRT reader is pulled,
   * commit/close is called, or when too many deletes are
   * buffered and must be flushed (by RAM usage or by count).
   *
   * Each packet is assigned a generation, and each flushed or
   * merged segment is also assigned a generation, so we can
   * track which BufferedDeletes packets to apply to any given
   * segment. */
  
  class BufferedDeletesStream {
  
    // TODO: maybe linked list?
    private final List<FrozenBufferedDeletes> deletes = new ArrayList<FrozenBufferedDeletes>();
  
    // Starts at 1 so that SegmentInfos that have never had
    // deletes applied (whose bufferedDelGen defaults to 0)
    // will be correct:
    private long nextGen = 1;
  
    // used only by assert
    private Term lastDeleteTerm;
    
    private PrintStream infoStream;
    private final AtomicLong bytesUsed = new AtomicLong();
    private final AtomicInteger numTerms = new AtomicInteger();
    private final int messageID;
  
    public BufferedDeletesStream(int messageID) {
      this.messageID = messageID;
    }
  
    private synchronized void message(String message) {
      if (infoStream != null) {
        infoStream.println("BD " + messageID + " [" + new Date() + "; " + Thread.currentThread().getName() + "]: " + message);
      }
    }
    
    public synchronized void setInfoStream(PrintStream infoStream) {
      this.infoStream = infoStream;
    }
  
    // Appends a new packet of buffered deletes to the stream,
    // setting its generation:
    public synchronized void push(FrozenBufferedDeletes packet) {
      assert packet.any();
      assert checkDeleteStats();    
      assert packet.gen < nextGen;
      deletes.add(packet);
      numTerms.addAndGet(packet.numTermDeletes);
      bytesUsed.addAndGet(packet.bytesUsed);
      if (infoStream != null) {
        message("push deletes " + packet + " delGen=" + packet.gen + " packetCount=" + deletes.size());
      }
      assert checkDeleteStats();    
    }
      
    public synchronized void clear() {
      deletes.clear();
      nextGen = 1;
      numTerms.set(0);
      bytesUsed.set(0);
    }
  
    public boolean any() {
      return bytesUsed.get() != 0;
    }
  
    public int numTerms() {
      return numTerms.get();
    }
  
    public long bytesUsed() {
      return bytesUsed.get();
    }
  
    public static class ApplyDeletesResult {
      // True if any actual deletes took place:
      public final boolean anyDeletes;
  
      // Current gen, for the merged segment:
      public final long gen;
  
      // If non-null, contains segments that are 100% deleted
      public final List<SegmentInfo> allDeleted;
  
      ApplyDeletesResult(boolean anyDeletes, long gen, List<SegmentInfo> allDeleted) {
        this.anyDeletes = anyDeletes;
        this.gen = gen;
        this.allDeleted = allDeleted;
      }
    }
  
    // Sorts SegmentInfos from smallest to biggest bufferedDelGen:
    private static final Comparator<SegmentInfo> sortByDelGen = new Comparator<SegmentInfo>() {
      // @Override -- not until Java 1.6
      public int compare(SegmentInfo si1, SegmentInfo si2) {
        final long cmp = si1.getBufferedDeletesGen() - si2.getBufferedDeletesGen();
        if (cmp > 0) {
          return 1;
        } else if (cmp < 0) {
          return -1;
        } else {
          return 0;
        }
      }
    };
  
    /** Resolves the buffered deleted Term/Query/docIDs, into
     *  actual deleted docIDs in the deletedDocs BitVector for
     *  each SegmentReader. */
    public synchronized ApplyDeletesResult applyDeletes(IndexWriter.ReaderPool readerPool, List<SegmentInfo> infos) throws IOException {
      final long t0 = System.currentTimeMillis();
  
      if (infos.size() == 0) {
        return new ApplyDeletesResult(false, nextGen++, null);
      }
  
      assert checkDeleteStats();
  
      if (!any()) {
        message("applyDeletes: no deletes; skipping");
        return new ApplyDeletesResult(false, nextGen++, null);
      }
  
      if (infoStream != null) {
        message("applyDeletes: infos=" + infos + " packetCount=" + deletes.size());
      }
  
      List<SegmentInfo> infos2 = new ArrayList<SegmentInfo>();
      infos2.addAll(infos);
      Collections.sort(infos2, sortByDelGen);
  
      CoalescedDeletes coalescedDeletes = null;
      boolean anyNewDeletes = false;
  
      int infosIDX = infos2.size()-1;
      int delIDX = deletes.size()-1;
  
      List<SegmentInfo> allDeleted = null;
  
      while (infosIDX >= 0) {
        //System.out.println("BD: cycle delIDX=" + delIDX + " infoIDX=" + infosIDX);
  
        final FrozenBufferedDeletes packet = delIDX >= 0 ? deletes.get(delIDX) : null;
        final SegmentInfo info = infos2.get(infosIDX);
        final long segGen = info.getBufferedDeletesGen();
  
        if (packet != null && segGen < packet.gen) {
          //System.out.println("  coalesce");
          if (coalescedDeletes == null) {
            coalescedDeletes = new CoalescedDeletes();
          }
          coalescedDeletes.update(packet);
          delIDX--;
        } else if (packet != null && segGen == packet.gen) {
          //System.out.println("  eq");
  
          // Lock order: IW -> BD -> RP
          assert readerPool.infoIsLive(info);
          SegmentReader reader = readerPool.get(info, false);
          int delCount = 0;
          final boolean segAllDeletes;
          try {
            if (coalescedDeletes != null) {
              //System.out.println("    del coalesced");
              delCount += applyTermDeletes(coalescedDeletes.termsIterable(), reader);
              delCount += applyQueryDeletes(coalescedDeletes.queriesIterable(), reader);
            }
            //System.out.println("    del exact");
            // Don't delete by Term here; DocumentsWriter
            // already did that on flush:
            delCount += applyQueryDeletes(packet.queriesIterable(), reader);
            segAllDeletes = reader.numDocs() == 0;
          } finally {
            readerPool.release(reader);
          }
          anyNewDeletes |= delCount > 0;
  
          if (segAllDeletes) {
            if (allDeleted == null) {
              allDeleted = new ArrayList<SegmentInfo>();
            }
            allDeleted.add(info);
          }
  
          if (infoStream != null) {
            message("seg=" + info + " segGen=" + segGen + " segDeletes=[" + packet + "]; coalesced deletes=[" + (coalescedDeletes == null ? "null" : coalescedDeletes) + "] delCount=" + delCount + (segAllDeletes ? " 100% deleted" : ""));
          }
  
          if (coalescedDeletes == null) {
            coalescedDeletes = new CoalescedDeletes();
          }
          coalescedDeletes.update(packet);
          delIDX--;
          infosIDX--;
          info.setBufferedDeletesGen(nextGen);
  
        } else {
          //System.out.println("  gt");
  
          if (coalescedDeletes != null) {
            // Lock order: IW -> BD -> RP
            assert readerPool.infoIsLive(info);
            SegmentReader reader = readerPool.get(info, false);
            int delCount = 0;
            final boolean segAllDeletes;
            try {
              delCount += applyTermDeletes(coalescedDeletes.termsIterable(), reader);
              delCount += applyQueryDeletes(coalescedDeletes.queriesIterable(), reader);
              segAllDeletes = reader.numDocs() == 0;
            } finally {
              readerPool.release(reader);
            }
            anyNewDeletes |= delCount > 0;
  
            if (segAllDeletes) {
              if (allDeleted == null) {
                allDeleted = new ArrayList<SegmentInfo>();
              }
              allDeleted.add(info);
            }
  
            if (infoStream != null) {
              message("seg=" + info + " segGen=" + segGen + " coalesced deletes=[" + (coalescedDeletes == null ? "null" : coalescedDeletes) + "] delCount=" + delCount + (segAllDeletes ? " 100% deleted" : ""));
            }
          }
          info.setBufferedDeletesGen(nextGen);
  
          infosIDX--;
        }
      }
  
      assert checkDeleteStats();
      if (infoStream != null) {
        message("applyDeletes took " + (System.currentTimeMillis()-t0) + " msec");
      }
      // assert infos != segmentInfos || !any() : "infos=" + infos + " segmentInfos=" + segmentInfos + " any=" + any;
      
      return new ApplyDeletesResult(anyNewDeletes, nextGen++, allDeleted);
    }
  
    public synchronized long getNextGen() {
      return nextGen++;
    }
  
    // Lock order IW -> BD
    /* Removes any BufferedDeletes that we no longer need to
     * store because all segments in the index have had the
     * deletes applied. */
    public synchronized void prune(SegmentInfos segmentInfos) {
      assert checkDeleteStats();
      long minGen = Long.MAX_VALUE;
      for(SegmentInfo info : segmentInfos) {
        minGen = Math.min(info.getBufferedDeletesGen(), minGen);
      }
  
      if (infoStream != null) {
        message("prune sis=" + segmentInfos + " minGen=" + minGen + " packetCount=" + deletes.size());
      }
  
      final int limit = deletes.size();
      for(int delIDX=0;delIDX<limit;delIDX++) {
        if (deletes.get(delIDX).gen >= minGen) {
          prune(delIDX);
          assert checkDeleteStats();
          return;
        }
      }
  
      // All deletes pruned
      prune(limit);
      assert !any();
      assert checkDeleteStats();
    }
  
    private synchronized void prune(int count) {
      if (count > 0) {
        if (infoStream != null) {
          message("pruneDeletes: prune " + count + " packets; " + (deletes.size() - count) + " packets remain");
        }
        for(int delIDX=0;delIDX<count;delIDX++) {
          final FrozenBufferedDeletes packet = deletes.get(delIDX);
          numTerms.addAndGet(-packet.numTermDeletes);
          assert numTerms.get() >= 0;
          bytesUsed.addAndGet(-packet.bytesUsed);
          assert bytesUsed.get() >= 0;
        }
        deletes.subList(0, count).clear();
      }
    }
  
    // Delete by Term
    private synchronized long applyTermDeletes(Iterable<Term> termsIter, SegmentReader reader) throws IOException {
      long delCount = 0;
          
      assert checkDeleteTerm(null);
      
      final TermDocs docs = reader.termDocs();
  
      for (Term term : termsIter) {
          
        // Since we visit terms sorted, we gain performance
        // by re-using the same TermsEnum and seeking only
        // forwards
        assert checkDeleteTerm(term);
        docs.seek(term);
            
        while (docs.next()) {
          final int docID = docs.doc();
          reader.deleteDocument(docID);
          // TODO: we could/should change
          // reader.deleteDocument to return boolean
          // true if it did in fact delete, because here
          // we could be deleting an already-deleted doc
          // which makes this an upper bound:
          delCount++;
        }
      }
  
      return delCount;
    }
  
    public static class QueryAndLimit {
      public final Query query;
      public final int limit;
      public QueryAndLimit(Query query, int limit) {
        this.query = query;       
        this.limit = limit;
      }
    }
  
    // Delete by query
    private synchronized long applyQueryDeletes(Iterable<QueryAndLimit> queriesIter, SegmentReader reader) throws IOException {
      long delCount = 0;
  
      for (QueryAndLimit ent : queriesIter) {
        Query query = ent.query;
        int limit = ent.limit;
        final DocIdSet docs = new QueryWrapperFilter(query).getDocIdSet(reader);
        if (docs != null) {
          final DocIdSetIterator it = docs.iterator();
          if (it != null) {
            while(true)  {
              int doc = it.nextDoc();
              if (doc >= limit)
                break;
  
              reader.deleteDocument(doc);
              // TODO: we could/should change
              // reader.deleteDocument to return boolean
              // true if it did in fact delete, because here
              // we could be deleting an already-deleted doc
              // which makes this an upper bound:
              delCount++;
            }
          }
        }
      }
  
      return delCount;
    }
  
    // used only by assert
    private boolean checkDeleteTerm(Term term) {
      if (term != null) {
        assert lastDeleteTerm == null || term.compareTo(lastDeleteTerm) > 0: "lastTerm=" + lastDeleteTerm + " vs term=" + term;
      }
      // TODO: we re-use term now in our merged iterable, but we shouldn't clone, instead copy for this assert
      lastDeleteTerm = term == null ? null : new Term(term.field(), term.text());
      return true;
    }
    
    // only for assert
    private boolean checkDeleteStats() {
      int numTerms2 = 0;
      long bytesUsed2 = 0;
      for(FrozenBufferedDeletes packet : deletes) {
        numTerms2 += packet.numTermDeletes;
        bytesUsed2 += packet.bytesUsed;
      }
      assert numTerms2 == numTerms.get(): "numTerms2=" + numTerms2 + " vs " + numTerms.get();
      assert bytesUsed2 == bytesUsed.get(): "bytesUsed2=" + bytesUsed2 + " vs " + bytesUsed;
      return true;
    }
  }
  