package macrobase.analysis.summary.count;

import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.DoubleSummaryStatistics;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/*
 Maintains probabilistic heavy-hitters:
    - Item counts are overreported.
    - Once we have seen 1/threshold items, size is >= 1/threshold items.

 This is similar to SpaceSaving but with:
   - O(1) update and O(items*log(k)) maintenance
     (normally O(log(k)) update)
   - unlimited space overhead within an epoch (normally O(k))

 Basic idea:
    Store min count from previous epoch
    Item in table? Increment
    Item not in table? Min + count

    End of epoch:
     1.) decay counts
     2.) record new min
     3.) compute 1/thresh highest counts
     4.) discard lower items, updating min if necessary

 */
public class AmortizedMaintenanceCounter extends ApproximateCount {
    private static final Logger log = LoggerFactory.getLogger(AmortizedMaintenanceCounter.class);

    private HashMap<Integer, Double> counts = new HashMap<>();
    private double totalCount = 0;
    private final int maxStableSize;

    private double prevEpochMaxEvicted = 0;

    public AmortizedMaintenanceCounter(int maxStableSize) {
        this.maxStableSize = maxStableSize;
    }

    @Override
    public void multiplyAllCounts(Double by) {
        totalCount *= by;

        log.trace("Decaying; {} items stored", counts.size());

        for (Map.Entry<Integer, Double> entry : counts.entrySet()) {
            double newValue = entry.getValue() * by;
            counts.put(entry.getKey(), newValue);

        }

	if (counts.size() > maxStableSize) {
            List<Map.Entry<Integer, Double>> a = Lists.newArrayList(counts.entrySet());
            a.sort((e1, e2) -> e1.getValue().compareTo(e2.getValue()));

            double prevVal = -1;
            int toRemove = counts.size() - maxStableSize;

            log.trace("Removing {} items from counts", toRemove);

            prevEpochMaxEvicted = Double.MIN_VALUE;

            for (int i = 0; i < toRemove; ++i) {
                Map.Entry<Integer, Double> entry = a.get(i);
                counts.remove(entry.getKey());
                if (entry.getValue() > prevEpochMaxEvicted) {
                    prevEpochMaxEvicted = entry.getValue();
                }
            }
        }

        log.trace("Finished pruning; new size is {}; max evicted is {}",
                  counts.size(),
                  prevEpochMaxEvicted);
    }

    public HashMap<Integer, Double> getCounts() {
        return counts;
    }

    @Override
    public void observe(Integer item, double count) {
        Double value = counts.get(item);
        if (value == null) {
            value = prevEpochMaxEvicted + count;
            totalCount += prevEpochMaxEvicted + count;
        } else {
            value += count;
            totalCount += count;
        }

        counts.put(item, value);
    }

    @Override
    public double getTotalCount() {
        return totalCount;
    }

    @Override
    public double getCount(int item) {
        Double ret = counts.get(item);
        if (ret == null) {
            return 0;
        }

        return ret;
    }
}

@@ -1062,24 +1062,138 @@
 er.class);%0A%0A
+    private double decayFactor = 1;%0A    private static final double DECAY_RESET_THRESHOLD = Double.MAX_VALUE*.9;%0A%0A
     private 
@@ -1545,261 +1545,25 @@
     
-totalCount *= by;%0A%0A        log.trace(%22Decaying; %7B%7D items stored%22, counts.size());%0A%0A        for (Map.Entry%3CInteger, Double%3E entry : counts.entrySet()) %7B%0A            double newValue = entry.getValue() * by;%0A            counts.put(entry.getKey(), newValue)
+decayFactor /= by
 ;%0A%0A 
@@ -1573,12 +1573,8 @@
     
-%7D%0A%0A%09
 if (
@@ -1771,41 +1771,8 @@
 );%0A%0A
-            double prevVal = -1;%0A
     
@@ -2467,24 +2467,52 @@
 tCounts() %7B%0A
+        resetDecayFactor();%0A
         retu
@@ -2521,32 +2521,356 @@
  counts;%0A    %7D%0A%0A
+    private void resetDecayFactor() %7B%0A        log.trace(%22Decaying; %7B%7D items stored%22, counts.size());%0A%0A        for (Map.Entry%3CInteger, Double%3E entry : counts.entrySet()) %7B%0A            double newValue = entry.getValue()/decayFactor;%0A            counts.put(entry.getKey(), newValue);%0A        %7D%0A%0A        decayFactor = 1;%0A    %7D%0A%0A
     @Override%0A  
@@ -2917,24 +2917,55 @@
 le count) %7B%0A
+        count *= decayFactor;%0A%0A
         Doub
@@ -3101,35 +3101,13 @@
  += 
-prevEpochMaxEvicted + count
+value
 ;%0A  
@@ -3221,24 +3221,130 @@
 tem, value);
+%0A%0A        if(value %3E DECAY_RESET_THRESHOLD && decayFactor %3E 1) %7B%0A            resetDecayFactor();%0A        %7D
 %0A    %7D%0A%0A    
@@ -3410,24 +3410,36 @@
 n totalCount
+/decayFactor
 ;%0A    %7D%0A%0A   
@@ -3605,16 +3605,28 @@
 turn ret
+/decayFactor
 ;%0A    %7D%0A

Maintain decay factor in AMC; only divide counts when near MAX_DOUBLE.