package org.apache.lucene.index;

import org.apache.lucene.document.NumericDocValuesField;
import org.apache.lucene.index.DocValuesUpdate.NumericDocValuesUpdate;
import org.apache.lucene.search.DocIdSetIterator;
import org.apache.lucene.util.FixedBitSet;
import org.apache.lucene.util.InPlaceMergeSorter;
import org.apache.lucene.util.packed.PackedInts;
import org.apache.lucene.util.packed.PagedGrowableWriter;
import org.apache.lucene.util.packed.PagedMutable;

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

/**
 * A {@link DocValuesFieldUpdates} which holds updates of documents, of a single
 * {@link NumericDocValuesField}.
 * 
 * @lucene.experimental
 */
class NumericDocValuesFieldUpdates extends DocValuesFieldUpdates {
  
  final static class Iterator extends DocValuesFieldUpdates.Iterator {
    private final int size;
    private final PagedGrowableWriter values;
    private final FixedBitSet docsWithField;
    private final PagedMutable docs;
    private long idx = 0; // long so we don't overflow if size == Integer.MAX_VALUE
    private int doc = -1;
    private Long value = null;
    
    Iterator(int size, PagedGrowableWriter values, FixedBitSet docsWithField, PagedMutable docs) {
      this.size = size;
      this.values = values;
      this.docsWithField = docsWithField;
      this.docs = docs;
    }
    
    @Override
    Long value() {
      return value;
    }
    
    @Override
    int nextDoc() {
      if (idx >= size) {
        value = null;
        return doc = DocIdSetIterator.NO_MORE_DOCS;
      }
      doc = (int) docs.get(idx);
      ++idx;
      while (idx < size && docs.get(idx) == doc) {
        ++idx;
      }
      if (!docsWithField.get((int) (idx - 1))) {
        value = null;
      } else {
        // idx points to the "next" element
        value = Long.valueOf(values.get(idx - 1));
      }
      return doc;
    }
    
    @Override
    int doc() {
      return doc;
    }
    
    @Override
    void reset() {
      doc = -1;
      value = null;
      idx = 0;
    }
  }

  private FixedBitSet docsWithField;
  private PagedMutable docs;
  private PagedGrowableWriter values;
  private int size;
  
  public NumericDocValuesFieldUpdates(String field, int maxDoc) {
    super(field, Type.NUMERIC);
    docsWithField = new FixedBitSet(64);
    docs = new PagedMutable(1, 1024, PackedInts.bitsRequired(maxDoc - 1), PackedInts.COMPACT);
    values = new PagedGrowableWriter(1, 1024, 1, PackedInts.FAST);
    size = 0;
  }
  
  @Override
  public void add(int doc, Object value) {
    // TODO: if the Sorter interface changes to take long indexes, we can remove that limitation
    if (size == Integer.MAX_VALUE) {
      throw new IllegalStateException("cannot support more than Integer.MAX_VALUE doc/value entries");
    }

    Long val = (Long) value;
    if (val == null) {
      val = NumericDocValuesUpdate.MISSING;
    }
    
    // grow the structures to have room for more elements
    if (docs.size() == size) {
      docs = docs.grow(size + 1);
      values = values.grow(size + 1);
      docsWithField = FixedBitSet.ensureCapacity(docsWithField, (int) docs.size());
    }
    
    if (val != NumericDocValuesUpdate.MISSING) {
      // only mark the document as having a value in that field if the value wasn't set to null (MISSING)
      docsWithField.set(size);
    }
    
    docs.set(size, doc);
    values.set(size, val.longValue());
    ++size;
  }
  
  @Override
  public Iterator iterator() {
    final PagedMutable docs = this.docs;
    final PagedGrowableWriter values = this.values;
    final FixedBitSet docsWithField = this.docsWithField;
    new InPlaceMergeSorter() {
      @Override
      protected void swap(int i, int j) {
        long tmpDoc = docs.get(j);
        docs.set(j, docs.get(i));
        docs.set(i, tmpDoc);
        
        long tmpVal = values.get(j);
        values.set(j, values.get(i));
        values.set(i, tmpVal);
        
        boolean tmpBool = docsWithField.get(j);
        if (docsWithField.get(i)) {
          docsWithField.set(j);
        } else {
          docsWithField.clear(j);
        }
        if (tmpBool) {
          docsWithField.set(i);
        } else {
          docsWithField.clear(i);
        }
      }
      
      @Override
      protected int compare(int i, int j) {
        int x = (int) docs.get(i);
        int y = (int) docs.get(j);
        return (x < y) ? -1 : ((x == y) ? 0 : 1);
      }
    }.sort(0, size);
    
    return new Iterator(size, values, docsWithField, docs);
  }
  
  @Override
  public void merge(DocValuesFieldUpdates other) {
    assert other instanceof NumericDocValuesFieldUpdates;
    NumericDocValuesFieldUpdates otherUpdates = (NumericDocValuesFieldUpdates) other;
    if (size  + otherUpdates.size > Integer.MAX_VALUE) {
      throw new IllegalStateException(
          "cannot support more than Integer.MAX_VALUE doc/value entries; size="
              + size + " other.size=" + otherUpdates.size);
    }
    docs = docs.grow(size + otherUpdates.size);
    values = values.grow(size + otherUpdates.size);
    docsWithField = FixedBitSet.ensureCapacity(docsWithField, (int) docs.size());
    for (int i = 0; i < otherUpdates.size; i++) {
      int doc = (int) otherUpdates.docs.get(i);
      if (otherUpdates.docsWithField.get(i)) {
        docsWithField.set(size);
      }
      docs.set(size, doc);
      values.set(size, otherUpdates.values.get(i));
      ++size;
    }
  }

  @Override
  public boolean any() {
    return size > 0;
  }

}
