<commit_before>package hu.bme.mit.ire.nodes.binary

import scala.Vector
import scala.collection.immutable.VectorBuilder
import scala.collection.mutable

import hu.bme.mit.ire.SingleForwarder
import hu.bme.mit.ire.datatypes.Mask
import hu.bme.mit.ire.datatypes.Path
import hu.bme.mit.ire.datatypes.Tuple
import hu.bme.mit.ire.messages.ChangeSet
import hu.bme.mit.ire.messages.ReteMessage
import hu.bme.mit.ire.util.SizeCounter
import hu.bme.mit.ire.util.TestUtil.tuple
import scala.collection.mutable.HashSet
import hu.bme.mit.ire.util.BufferMultimap


class TransitiveClosureJoinNode(override val next: (ReteMessage) => Unit,
                                override val primaryTupleWidth: Int,
                                override val secondaryTupleWidth: Int,
                                override val primaryMask: Mask,
                                override val secondaryMask: Mask,
                                var minHops: Long = 0,
                                var maxHops: Long = Long.MaxValue
                                ) extends JoinNodeBase with SingleForwarder {

  minHops = if (minHops >= 0) minHops else 0
  maxHops = if (maxHops >= 1) maxHops else Long.MaxValue

  val sourceLookup = new mutable.HashMap[Any, Tuple]
  val targetLookup = new mutable.HashMap[Any, Tuple]

  def composeTuple(sourceId: Long, path: Path, targetId: Long): Tuple = {
    tuple(sourceId) ++ sourceLookup.getOrElse(sourceId, tuple()) ++
    tuple(path) ++ // TODO: there might be extra attributes for the path
    tuple(targetId) ++ targetLookup.getOrElse(targetId, tuple())
  }

  val sourceVerticesIndexer = new HashSet[Long]

  val reachableVertices = new mutable.HashMap[Long, mutable.HashMap[Long, mutable.MutableList[PathData]]]
  val reachableFrom = new mutable.HashMap[Long, mutable.HashSet[Long]]

  override def onSizeRequest(): Long = SizeCounter.count(reachableVertices.values, reachableFrom.values)

  override def onPrimary(changeSet: ChangeSet): Unit = {
    for (tuple <- changeSet.positive)
      sourceLookup.put(tuple(0), tuple.drop(1))
    for (tuple <- changeSet.negative)
      sourceLookup.remove(tuple(0))

    val positiveTuples = changeSet.positive.flatMap(pathsFromSourceVertex)
    val negativeTuples = changeSet.negative.flatMap(pathsFromSourceVertex)

    changeSet.positive.map(inputTuple => inputTuple(0).asInstanceOf[Number].longValue).foreach(sourceVerticesIndexer.add(_))
    changeSet.negative.map(inputTuple => inputTuple(0).asInstanceOf[Number].longValue).foreach(sourceVerticesIndexer.remove(_))

    forward(ChangeSet(
      positive = positiveTuples,
      negative = negativeTuples
    ))
  }

  override def onSecondary(changeSet: ChangeSet): Unit = {
    // in the current implementation we can assume that the secondary input is a GetEdges operator
    for (tuple <- changeSet.positive)
      targetLookup.put(tuple(2), tuple.drop(3))
    for (tuple <- changeSet.negative)
      targetLookup.remove(tuple(2))

    forward(ChangeSet(
      positive = changeSet.positive.flatMap(newPathsWithEdge),
      negative = changeSet.negative.flatMap(removePathsWithEdge)
    ))
  }

  private def newPathsWithEdge(inputTuple: Tuple): Vector[Tuple] = {
    val sourceId = inputTuple(0).asInstanceOf[Number].longValue
    val targetId = inputTuple(2).asInstanceOf[Number].longValue
    val edgeId = inputTuple(1).asInstanceOf[Number].longValue
    val edgeAndTargetExtraVariables = inputTuple.drop(3)

    var newPathsBuilder = new VectorBuilder[Tuple]

    if (!reachableVertices.contains(sourceId)) {
      reachableVertices(sourceId) = new mutable.HashMap
    }

    if (!reachableFrom.contains(targetId)) {
      reachableFrom(targetId) = new mutable.HashSet
    }

    val pathsToSource = pathsToSourceVertex(sourceId)
    val pathsFromTarget = pathsFromTargetVertex(targetId)
    for(pathToSourceVertex <- pathsToSource; pathFromTargetVertex <- pathsFromTarget){
      if(pathToSourceVertex._2.path.size + pathFromTargetVertex._2.path.size + 1 <= maxHops && areDistinctPaths(pathToSourceVertex._2.pathHash, pathFromTargetVertex._2.pathHash)){
        var pathBuilder = new VectorBuilder[Long]
        pathBuilder ++= pathToSourceVertex._2.path
        pathBuilder += edgeId
        pathBuilder ++= pathFromTargetVertex._2.path
        val path = pathBuilder.result<commit_msg>Remove unused variable<commit_after>package hu.bme.mit.ire.nodes.binary

import scala.Vector
import scala.collection.immutable.VectorBuilder
import scala.collection.mutable

import hu.bme.mit.ire.SingleForwarder
import hu.bme.mit.ire.datatypes.Mask
import hu.bme.mit.ire.datatypes.Path
import hu.bme.mit.ire.datatypes.Tuple
import hu.bme.mit.ire.messages.ChangeSet
import hu.bme.mit.ire.messages.ReteMessage
import hu.bme.mit.ire.util.SizeCounter
import hu.bme.mit.ire.util.TestUtil.tuple
import scala.collection.mutable.HashSet
import hu.bme.mit.ire.util.BufferMultimap


class TransitiveClosureJoinNode(override val next: (ReteMessage) => Unit,
                                override val primaryTupleWidth: Int,
                                override val secondaryTupleWidth: Int,
                                override val primaryMask: Mask,
                                override val secondaryMask: Mask,
                                var minHops: Long = 0,
                                var maxHops: Long = Long.MaxValue
                                ) extends JoinNodeBase with SingleForwarder {

  minHops = if (minHops >= 0) minHops else 0
  maxHops = if (maxHops >= 1) maxHops else Long.MaxValue

  val sourceLookup = new mutable.HashMap[Any, Tuple]
  val targetLookup = new mutable.HashMap[Any, Tuple]

  def composeTuple(sourceId: Long, path: Path, targetId: Long): Tuple = {
    tuple(sourceId) ++ sourceLookup.getOrElse(sourceId, tuple()) ++
    tuple(path) ++ // TODO: there might be extra attributes for the path
    tuple(targetId) ++ targetLookup.getOrElse(targetId, tuple())
  }

  val sourceVerticesIndexer = new HashSet[Long]

  val reachableVertices = new mutable.HashMap[Long, mutable.HashMap[Long, mutable.MutableList[PathData]]]
  val reachableFrom = new mutable.HashMap[Long, mutable.HashSet[Long]]

  override def onSizeRequest(): Long = SizeCounter.count(reachableVertices.values, reachableFrom.values)

  override def onPrimary(changeSet: ChangeSet): Unit = {
    for (tuple <- changeSet.positive)
      sourceLookup.put(tuple(0), tuple.drop(1))
    for (tuple <- changeSet.negative)
      sourceLookup.remove(tuple(0))

    val positiveTuples = changeSet.positive.flatMap(pathsFromSourceVertex)
    val negativeTuples = changeSet.negative.flatMap(pathsFromSourceVertex)

    changeSet.positive.map(inputTuple => inputTuple(0).asInstanceOf[Number].longValue).foreach(sourceVerticesIndexer.add(_))
    changeSet.negative.map(inputTuple => inputTuple(0).asInstanceOf[Number].longValue).foreach(sourceVerticesIndexer.remove(_))

    forward(ChangeSet(
      positive = positiveTuples,
      negative = negativeTuples
    ))
  }

  override def onSecondary(changeSet: ChangeSet): Unit = {
    // in the current implementation we can assume that the secondary input is a GetEdges operator
    for (tuple <- changeSet.positive)
      targetLookup.put(tuple(2), tuple.drop(3))
    for (tuple <- changeSet.negative)
      targetLookup.remove(tuple(2))

    forward(ChangeSet(
      positive = changeSet.positive.flatMap(newPathsWithEdge),
      negative = changeSet.negative.flatMap(removePathsWithEdge)
    ))
  }

  private def newPathsWithEdge(inputTuple: Tuple): Vector[Tuple] = {
    val sourceId = inputTuple(0).asInstanceOf[Number].longValue
    val targetId = inputTuple(2).asInstanceOf[Number].longValue
    val edgeId = inputTuple(1).asInstanceOf[Number].longValue

    var newPathsBuilder = new VectorBuilder[Tuple]

    if (!reachableVertices.contains(sourceId)) {
      reachableVertices(sourceId) = new mutable.HashMap
    }

    if (!reachableFrom.contains(targetId)) {
      reachableFrom(targetId) = new mutable.HashSet
    }

    val pathsToSource = pathsToSourceVertex(sourceId)
    val pathsFromTarget = pathsFromTargetVertex(targetId)
    for(pathToSourceVertex <- pathsToSource; pathFromTargetVertex <- pathsFromTarget){
      if(pathToSourceVertex._2.path.size + pathFromTargetVertex._2.path.size + 1 <= maxHops && areDistinctPaths(pathToSourceVertex._2.pathHash, pathFromTargetVertex._2.pathHash)){
        var pathBuilder = new VectorBuilder[Long]
        pathBuilder ++= pathToSourceVertex._2.path
        pathBuilder += edgeId
        pathBuilder ++= pathFromTargetVertex._2.path
        val path = pathBuilder.result
{'fraction_non_alphanumeric': 0.09809390980939098, 'fraction_numerical': 0.005230125523012552, 'mean_word_length': 3.519432773109244, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7156418', 'n_tokens_mistral': 2599, 'n_tokens_neox': 2500, 'n_words': 506}