/*
    * 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.commons.collections4.bloomfilter;
  
  import java.util.Arrays;
  import java.util.Objects;
  import java.util.function.IntPredicate;
  import java.util.function.LongPredicate;
  import java.util.stream.IntStream;
  
  /**
   * A counting Bloom filter using an int array to track cells for each enabled bit.
   *
   * <p>Any operation that results in negative counts or integer overflow of
   * counts will mark this filter as invalid. This transition is not reversible.
   * The operation is completed in full, no exception is raised and the state is
   * set to invalid. This allows the cells for the filter immediately prior to the
   * operation that created the invalid state to be recovered. See the documentation
   * in {@link #isValid()} for details.</p>
   *
   * <p>All the operations in the filter assume the cells are currently valid,
   * for example {@code cardinality} or {@code contains} operations. Behavior of an invalid
   * filter is undefined. It will no longer function identically to a standard
   * Bloom filter that is the merge of all the Bloom filters that have been added
   * to and not later subtracted from the counting Bloom filter.</p>
   *
   * <p>The maximum supported number of items that can be stored in the filter is
   * limited by the maximum array size combined with the {@link Shape}. For
   * example an implementation using a {@link Shape} with a false-positive
   * probability of 1e-6 and {@link Integer#MAX_VALUE} bits can reversibly store
   * approximately 75 million items using 20 hash functions per item with a memory
   * consumption of approximately 8 GB.
   *
   * @see Shape
   * @see CellProducer
   * @since 4.5
   */
  public final class ArrayCountingBloomFilter implements CountingBloomFilter {
  
      /**
       * The shape of this Bloom filter.
       */
      private final Shape shape;
  
      /**
       * The cell for each bit index in the filter.
       */
      private final int[] cells;
  
      /**
       * The state flag. This is a bitwise {@code OR} of the entire history of all updated
       * cells. If negative then a negative cell or integer overflow has occurred on
       * one or more cells in the history of the filter and the state is invalid.
       *
       * <p>Maintenance of this state flag is branch-free for improved performance. It
       * eliminates a conditional check for a negative cell during remove/subtract
       * operations and a conditional check for integer overflow during merge/add
       * operations.</p>
       *
       * <p>Note: Integer overflow is unlikely in realistic usage scenarios. A cell
       * that overflows indicates that the number of items in the filter exceeds the
       * maximum possible size (number of bits) of any Bloom filter constrained by
       * integer indices. At this point the filter is most likely full (all bits are
       * non-zero) and thus useless.</p>
       *
       * <p>Negative cells are a concern if the filter is used incorrectly by
       * removing an item that was never added. It is expected that a user of a
       * counting Bloom filter will not perform this action as it is a mistake.
       * Enabling an explicit recovery path for negative or overflow cells is a major
       * performance burden not deemed necessary for the unlikely scenarios when an
       * invalid state is created. Maintenance of the state flag is a concession to
       * flag improper use that should not have a major performance impact.</p>
       */
      private int state;
  
      private ArrayCountingBloomFilter(final ArrayCountingBloomFilter source) {
          this.shape = source.shape;
          this.state = source.state;
          this.cells = source.cells.clone();
      }
  
      /**
       * Constructs an empty counting Bloom filter with the specified shape.
       *
       * @param shape the shape of the filter
      */
     public ArrayCountingBloomFilter(final Shape shape) {
         Objects.requireNonNull(shape, "shape");
         this.shape = shape;
         cells = new int[shape.getNumberOfBits()];
     }
 
     @Override
     public boolean add(final CellProducer other) {
         Objects.requireNonNull(other, "other");
         other.forEachCell(this::add);
         return isValid();
     }
 
     /**
      * Add to the cell for the bit index.
      *
      * @param idx the index
      * @param addend the amount to add
      * @return {@code true} always.
      */
     private boolean add(final int idx, final int addend) {
         try {
             final int updated = cells[idx] + addend;
             state |= updated;
             cells[idx] = updated;
             return true;
         } catch (final IndexOutOfBoundsException e) {
             throw new IllegalArgumentException(
                     String.format("Filter only accepts values in the [0,%d) range", getShape().getNumberOfBits()), e);
         }
     }
 
     @Override
     public int[] asIndexArray() {
         return IntStream.range(0, cells.length).filter(i -> cells[i] > 0).toArray();
     }
 
     @Override
     public int cardinality() {
         return (int) IntStream.range(0, cells.length).filter(i -> cells[i] > 0).count();
     }
 
     @Override
     public int characteristics() {
         return SPARSE;
     }
 
     @Override
     public void clear() {
         Arrays.fill(cells, 0);
     }
 
     @Override
     public boolean contains(final BitMapProducer bitMapProducer) {
         return contains(IndexProducer.fromBitMapProducer(bitMapProducer));
     }
 
     @Override
     public boolean contains(final IndexProducer indexProducer) {
         return indexProducer.forEachIndex(idx -> this.cells[idx] != 0);
     }
 
     @Override
     public ArrayCountingBloomFilter copy() {
         return new ArrayCountingBloomFilter(this);
     }
 
     @Override
     public boolean forEachBitMap(final LongPredicate consumer) {
         Objects.requireNonNull(consumer, "consumer");
         final int blocksm1 = BitMap.numberOfBitMaps(cells.length) - 1;
         int i = 0;
         long value;
         // must break final block separate as the number of bits may not fall on the long boundary
         for (int j = 0; j < blocksm1; j++) {
             value = 0;
             for (int k = 0; k < Long.SIZE; k++) {
                 if (cells[i++] != 0) {
                     value |= BitMap.getLongBit(k);
                 }
             }
             if (!consumer.test(value)) {
                 return false;
             }
         }
         // Final block
         value = 0;
         for (int k = 0; i < cells.length; k++) {
             if (cells[i++] != 0) {
                 value |= BitMap.getLongBit(k);
             }
         }
         return consumer.test(value);
     }
 
     @Override
     public boolean forEachCell(final CellProducer.CellConsumer consumer) {
         Objects.requireNonNull(consumer, "consumer");
         for (int i = 0; i < cells.length; i++) {
             if (cells[i] != 0 && !consumer.test(i, cells[i])) {
                 return false;
             }
         }
         return true;
     }
 
     @Override
     public boolean forEachIndex(final IntPredicate consumer) {
         Objects.requireNonNull(consumer, "consumer");
         for (int i = 0; i < cells.length; i++) {
             if (cells[i] != 0 && !consumer.test(i)) {
                 return false;
             }
         }
         return true;
     }
 
     @Override
     public int getMaxCell() {
         return Integer.MAX_VALUE;
     }
 
     @Override
     public int getMaxInsert(final CellProducer cellProducer) {
         final int[] max = {Integer.MAX_VALUE};
         cellProducer.forEachCell( (x, y) -> {
             final int count = cells[x] / y;
             if (count < max[0]) {
                 max[0] = count;
             }
             return max[0] > 0;
         });
         return max[0];
     }
 
     @Override
     public Shape getShape() {
         return shape;
     }
 
     /**
      * {@inheritDoc}
      *
      * <p><em>Implementation note</em>
      *
      * <p>The state transition to invalid is permanent.</p>
      *
      * <p>This implementation does not correct negative cells to zero or integer
      * overflow cells to {@link Integer#MAX_VALUE}. Thus the operation that
      * generated invalid cells can be reversed by using the complement of the
      * original operation with the same Bloom filter. This will restore the cells
      * to the state prior to the invalid operation. Cells can then be extracted
      * using {@link #forEachCell(CellConsumer)}.</p>
      */
     @Override
     public boolean isValid() {
         return state >= 0;
     }
 
     @Override
     public boolean subtract(final CellProducer other) {
         Objects.requireNonNull(other, "other");
         other.forEachCell(this::subtract);
         return isValid();
     }
 
     /**
      * Subtract from the cell for the bit index.
      *
      * @param idx the index
      * @param subtrahend the amount to subtract
      * @return {@code true} always.
      */
     private boolean subtract(final int idx, final int subtrahend) {
         try {
             final int updated = cells[idx] - subtrahend;
             state |= updated;
             cells[idx] = updated;
             return true;
         } catch (final IndexOutOfBoundsException e) {
             throw new IllegalArgumentException(
                     String.format("Filter only accepts values in the [0,%d) range", getShape().getNumberOfBits()), e);
         }
     }
 }