package edu.neu.ccs.pyramid.elasticsearch;

import java.io.Serializable;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.IntStream;

/**
 * Created by chengli on 1/20/15.
 */
public class DuplicateDetector implements Serializable{
    private static final long serialVersionUID = 1L;
    private transient ESIndex esIndex;
    Set<String> allDuplicates;

    public DuplicateDetector(ESIndex esIndex) {
        this.esIndex = esIndex;
        this.allDuplicates = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());
    }

    public void addDuplicates(String id1, String id2){
        allDuplicates.add(id1);
        allDuplicates.add(id2);
    }

    public void detect(){
        Map<Integer, Set<String>> lengthToIds = new ConcurrentHashMap<>();
        int numDocs = esIndex.getNumDocs();
        IntStream.range(0,numDocs).parallel()
                .forEach(i -> {
                    Map<Integer,String> termVector = esIndex.getTermVectorFromIndex(""+i);
                    if (!lengthToIds.containsKey(termVector.size())){
                        lengthToIds.put(termVector.size(),Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>()));
                    }
                    lengthToIds.get(termVector.size()).add(""+i);
                });
        lengthToIds.entrySet().stream().parallel().map(Map.Entry::getValue)
                .forEach(this::check);

    }

    private void check(Set<String> candidates){
        List<String> list = new ArrayList<>(candidates);


        int size = candidates.size();
        if (size==1){
            return;
        }
        for (int i=0;i<size-1;i++){
            for (int j=i+i;j<size;j++){
                String id1 = list.get(i);
                String id2 = list.get(j);
                Map<Integer,String> termVector1 = esIndex.getTermVectorFromIndex(id1);
                Map<Integer,String> termVector2 = esIndex.getTermVectorFromIndex(id2);
                if (termVector1.equals(termVector2)){
                    addDuplicates(id1,id2);
                }
            }
        }
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        for (String str: allDuplicates){
            sb.append(str);
            sb.append(",");
        }
        return sb.toString();
    }
}

@@ -391,16 +391,97 @@
 cates;%0A%0A
+    public Set%3CString%3E getAllDuplicates() %7B%0A        return allDuplicates;%0A    %7D%0A%0A
     publ
@@ -649,32 +649,60 @@
 ean%3E());%0A    %7D%0A%0A
+    //todo keep unique ones%0A
     public void 
@@ -871,21 +871,19 @@
 tring%3E%3E 
-lengt
+has
 hToIds =
@@ -1148,17 +1148,69 @@
    i
-f (!lengt
+nt hash = termVector.hashCode();%0A                    if (!has
 hToI
@@ -1224,33 +1224,20 @@
 ainsKey(
-termVector.size()
+hash
 ))%7B%0A    
@@ -1252,29 +1252,27 @@
             
-lengt
+has
 hToIds.put(t
@@ -1274,26 +1274,14 @@
 put(
-termVector.size(),
+hash, 
 Coll
@@ -1384,29 +1384,27 @@
             
-lengt
+has
 hToIds.get(t
@@ -1406,25 +1406,12 @@
 get(
-termVector.size()
+hash
 ).ad
@@ -1451,13 +1451,11 @@
     
-lengt
+has
 hToI
@@ -2343,39 +2343,55 @@
     
-for (String str: allDuplicates)
+allDuplicates.stream().sorted().forEach(str -%3E 
 %7B%0A  
@@ -2445,32 +2445,34 @@
 (%22,%22);%0A        %7D
+);
 %0A        return 

change hashing