<commit_before>// Follows http://s3.thinkaurelius.com/docs/titan/current/hadoop-distributed-computing.html

def g

// create a Titan database connection that exists for the life of the mapper
def setup(args) {
    conf = new BaseConfiguration()
    conf.setProperty("storage.backend", "cassandra")
    conf.setProperty("storage.hostname", "127.0.0.1")
    conf.setProperty("storage.cassandra.thrift.frame-size", 1000)
    conf.setProperty("schema.default", null)
    conf.setProperty("storage.index.search.backend", "elasticsearch")
    conf.setProperty("storage.index.search.client-only", false)
    conf.setProperty("storage.index.search.local-mode", true)
    g = TitanFactory.open(conf)
}

// process each FaunusVertex
//  - lookup the vertex in the TitanGraph using the FaunusVertex's numeric ID
//  - for each outgoing edge: collect vertices with the same label, and call findDuplicates
//  - findDuplicates groups edges by incoming vertex, keeps the newest, and removes all others

def map(u, args) {
    def v = g.v(u.id) // the Faunus vertex id is the same as the original Titan vertex id
    def eSameLabel = []
    def lastLabel = null
    def removeMsgs = []
    
    def findDuplicates = { edgeList ->
        eg = [:];
        toRemoveList = [];
        edgeList._().groupBy(eg){it.inV().next()}{it}.iterate();
        
        def keepNewest = { duplicates ->
            newest = duplicates.max{it.created_at};
            msg = "keep ${newest}";
            duplicates.findAll{it != newest}.each{
                msg += "\nremv ${it}";
                it.remove();
            }
            msg
        }
        
        eg.each{ key,val ->
            if(val.size>1){toRemoveList += keepNewest(val)}
        };
        toRemoveList
    }
    
    for (e in v.getEdges()){
        if (e.outV.next() != v) {continue}
        if (e.getLabel() == lastLabel){eSameLabel << e}
        else {
            if(eSameLabel.size>1){removeMsgs += findDuplicates(eSameLabel);}
            eSameLabel = [e]; lastLabel = e.label;
        }
    }
    removeMsgs += findDuplicates(eSameLabel)
    return removeMsgs ? g.commit(); removeMsgs.join("\n") : null
}


// close the Titan database connection
def cleanup(args) {
    g.shutdown()
}


// without Hadoop
def removeDupEdges(v) {
    def eSameLabel = []
    def lastLabel = null
    def removeMsgs = []
    
    def findDuplicates = { edgeList ->
        eg = [:];
        toRemoveList = [];
        edgeList._().groupBy(eg){it.inV().next()}{it}.iterate();
        
        def keepNewest = { duplicates ->
            newest = duplicates.max{it.created_at};
            msg = "keep ${newest}";
            duplicates.findAll{it != newest}.each{
                msg += "\nremv ${it}";<commit_msg>fix commit-per-vertex<commit_after>// Follows http://s3.thinkaurelius.com/docs/titan/current/hadoop-distributed-computing.html

def g

// create a Titan database connection that exists for the life of the mapper
def setup(args) {
    conf = new BaseConfiguration()
    conf.setProperty("storage.backend", "cassandra")
    conf.setProperty("storage.hostname", "127.0.0.1")
    conf.setProperty("storage.cassandra.thrift.frame-size", 1000)
    conf.setProperty("schema.default", null)
    conf.setProperty("storage.index.search.backend", "elasticsearch")
    conf.setProperty("storage.index.search.client-only", false)
    conf.setProperty("storage.index.search.local-mode", true)
    g = TitanFactory.open(conf)
}

// process each FaunusVertex
//  - lookup the vertex in the TitanGraph using the FaunusVertex's numeric ID
//  - for each outgoing edge: collect vertices with the same label, and call findDuplicates
//  - findDuplicates groups edges by incoming vertex, keeps the newest, and removes all others

def map(u, args) {
    def v = g.v(u.id) // the Faunus vertex id is the same as the original Titan vertex id
    def eSameLabel = []
    def lastLabel = null
    def removeMsgs = []
    
    def findDuplicates = { edgeList ->
        eg = [:];
        toRemoveList = [];
        edgeList._().groupBy(eg){it.inV().next()}{it}.iterate();
        
        def keepNewest = { duplicates ->
            newest = duplicates.max{it.created_at};
            msg = "keep ${newest}";
            duplicates.findAll{it != newest}.each{
                msg += "\nremv ${it}";
                it.remove();
            }
            msg
        }
        
        eg.each{ key,val ->
            if(val.size>1){toRemoveList += keepNewest(val)}
        };
        toRemoveList
    }
    
    for (e in v.getEdges()){
        if (e.outV.next() != v) {continue}
        if (e.getLabel() == lastLabel){eSameLabel << e}
        else {
            if(eSameLabel.size>1){removeMsgs += findDuplicates(eSameLabel);}
            eSameLabel = [e]; lastLabel = e.label;
        }
    }
    removeMsgs += findDuplicates(eSameLabel)
    
    if (removeMsgs) {
        g.commit();
        return removeMsgs.join("\n")
    } else {
        return null
    }
}


// close the Titan database connection
def cleanup(args) {
    g.shutdown()
}


// without Hadoop
def removeDupEdges(v) {
    def eSameLabel = []
    def lastLabel = null
    def removeMsgs = []
    
    def findDuplicates = { edgeList ->
        eg = [:];
        toRemoveList = [];
        edgeList._().groupBy(eg){it.inV().next()}{it}.iterate();
        
        def keepNewest = { duplicates ->
            newest = duplicates.max{it.created_at};
            msg = "keep ${newest}";
            duplicates.findAll{it != newest}.each{
                msg += "\nremv ${it}";
{'fraction_non_alphanumeric': 0.13577586206896552, 'fraction_numerical': 0.004669540229885058, 'mean_word_length': 2.268192488262911, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2747574', 'n_tokens_mistral': 1773, 'n_tokens_neox': 1718, 'n_words': 393}