<commit_before>package ducttape.hyperdag.walker

import collection._
import ducttape.hyperdag.PhantomHyperDag
import ducttape.hyperdag.UnpackedVertex
import ducttape.util.MultiSet
import ducttape.hyperdag.PackedVertex
import grizzled.slf4j.Logging
import ducttape.hyperdag.meta.PhantomMetaHyperDag
import ducttape.hyperdag.meta.UnpackedChainedMetaVertex
import ducttape.hyperdag.meta.UnpackedMetaVertex
import ducttape.hyperdag.HyperEdge
import scala.annotation.tailrec

// NOTE: Root/terminal phantom vertices behave very different than internal phantom vertices
//       perhaps they deserve different names or special treatment otherwise?
// TODO: Create ChainedUnpackedVertex and return those instead
class UnpackedPhantomMetaDagWalker[V,M,H,E,D,F](
        val dag: PhantomMetaHyperDag[V,M,H,E],
        selectionFilter: SelectionFilter[D] = new DefaultSelectionFilter[D],
        hedgeFilter: HyperEdgeFilter[H,E] = new DefaultHyperEdgeFilter[H,E],
        constraintFilter: ConstraintFilter[Option[V],D,F] = new DefaultConstraintFilter[Option[V],D,F],
        vertexFilter: MetaVertexFilter[Option[V],H,E,D] = new DefaultMetaVertexFilter[Option[V],H,E,D],
        comboTransformer: ComboTransformer[H,E,D] = new DefaultComboTransformer[H,E,D],
        toD: H => D = new DefaultToD[H])
        (implicit ordering: Ordering[D])
  extends Walker[UnpackedChainedMetaVertex[V,H,E,D]] with Logging {
  
  object MetaVertexFilterAdapter extends MetaVertexFilter[Option[V],H,E,D] {
    override def apply(v: UnpackedMetaVertex[Option[V],H,E,D]) = v.packed.value match {
      case Some(_) => vertexFilter(v)
      case None => true
    }
  }
  
  val delegate = new UnpackedMetaDagWalker[Option[V],M,H,E,D,F](
    dag.delegate, selectionFilter, hedgeFilter, constraintFilter, MetaVertexFilterAdapter, comboTransformer, toD)
    
  // we must be able to recover the phantom-antecedents of non-phantom vertices
  // so that we can properly populate their state maps
  // unfortunately, this kills space complexity, since we must hold on to these until
  // all members of a packed vertex have been unpacked
  // TODO: Right now, we don't dectect when all members of a packed vertex have
  // been unpacked so that we can reclaim space. Could we refcount them?
  private val unpackedMap = new mutable.HashMap[(PackedVertex[Option[V]],Seq[D]), UnpackedMetaVertex[Option[V],H,E,D]]
  
  override def take(): Option[UnpackedChainedMetaVertex[V,H,E,D]] = recursiveTake()
  
  private def getOnly[A](seq: Seq[A]): A = seq match {
    case Seq(only) => only
    case _ => throw new RuntimeException("phantom chains can only have single parents; " +
          "expected exactly one element in: " + seq)
  }

  @tailrec
  private def followPhantomChain(v: UnpackedMetaVertex[Option[V],H,E,D], edge: E, parentReal: Seq[D])
                                : Seq[(E, Seq[D])] = {
    trace("Follow phantom chain at " + v)
    v.packed.value match {
      case Some(_) => Seq( (edge, parentReal) )
      case None => {
        trace("Zip: " + v.edges.zip(v.parentRealizations))

        // use active hyperedges to find parents
        val parents: Seq[UnpackedMetaVertex[Option[V],H,E,D]]
          = v.edges.zip(v.parentRealizations).
              flatMap { case (hyperedge, parentReals) => {
                dag.delegate.sources(hyperedge).zip(parentReals).map { case (parent, parentReal) =>
                  // parent: PackedVertex[Option[V]]
                  trace("Resolving hyperedge %s: parent is %s".format(hyperedge, parent))
                  try {
                    val uv: UnpackedMetaVertex[Option[V],H,E,D] = unpackedMap( (parent, parentReal.sorted(ordering)) )
                    uv
                  } catch {
                    case e => {
                      debug(unpackedMap.keySet.toSeq.map(_.toString).sorted.mkString("\n"))
                      throw e
                    }
                  }                  
                }
              }
            }
        trace("Parents of %s are: %s".format(v, parents))
        parents match {
          case Seq() => Seq( (edge, parentReal) ) // this is a root phantom vertex
          case Seq(singleUnpackedV) => {
            
            
            // TODO: How do we expect epsilon vertices to get handled?
            
            
            
            
            
            
            val myHyperEdge = getOnly(v.edges)
            val myParentReals = getOnly(v.parentRealizations)
            val myEdge = getOnly(myHyperEdge.e)
            val myParentReal = getOnly(myParentReals)
            followPhantomChain(singleUnpackedV, myEdge, myParentReal)
          }
          case _ => throw new RuntimeException("Found more than one parent of a phantom vertex: %s => %s".format(v,parents))
        }
      }
    }
  }
  
  @tailrec
  private def recursiveTake(): Option[UnpackedChainedMetaVertex[V,H,E,D]] = delegate.take() match {
    case None => None
    case Some(umv) => umv.packed.value match {
      case Some(packed) => {
        // we can have phantom vertex chains of arbitrary length
        // epsilons are already removed by our delegate
        import ducttape.util.Collections._
        val parentInfo: Seq[(Seq[E], Seq[Seq[D]])]
          = zip3(umv.edges, umv.parentRealizations, umv.edges).
            map { case (hyperedge, parentReals, hyperEdge) =>
              val munged: Seq[(E, Seq[D])] = zip3(dag.delegate.sources(hyperedge), parentReals, hyperedge.e) flatMap {
                case (parent, parentReal, edge) => {
                  trace("Begin backtracing phantom chain for " + parent)
                  val unpackedV = unpackedMap( (parent, parentReal.sorted(ordering)) )
                  val leafParents: Seq[(E, Seq[D])] = followPhantomChain(unpackedV, edge, parentReal)
                  leafParents
                }
              }<commit_msg>Potentially fixed issue with phantom vertices being able to have multiple parents (flatMap for the win)<commit_after>package ducttape.hyperdag.walker

import collection._
import ducttape.hyperdag.PhantomHyperDag
import ducttape.hyperdag.UnpackedVertex
import ducttape.util.MultiSet
import ducttape.hyperdag.PackedVertex
import grizzled.slf4j.Logging
import ducttape.hyperdag.meta.PhantomMetaHyperDag
import ducttape.hyperdag.meta.UnpackedChainedMetaVertex
import ducttape.hyperdag.meta.UnpackedMetaVertex
import ducttape.hyperdag.HyperEdge
import scala.annotation.tailrec

// NOTE: Root/terminal phantom vertices behave very different than internal phantom vertices
//       perhaps they deserve different names or special treatment otherwise?
// TODO: Create ChainedUnpackedVertex and return those instead
class UnpackedPhantomMetaDagWalker[V,M,H,E,D,F](
        val dag: PhantomMetaHyperDag[V,M,H,E],
        selectionFilter: SelectionFilter[D] = new DefaultSelectionFilter[D],
        hedgeFilter: HyperEdgeFilter[H,E] = new DefaultHyperEdgeFilter[H,E],
        constraintFilter: ConstraintFilter[Option[V],D,F] = new DefaultConstraintFilter[Option[V],D,F],
        vertexFilter: MetaVertexFilter[Option[V],H,E,D] = new DefaultMetaVertexFilter[Option[V],H,E,D],
        comboTransformer: ComboTransformer[H,E,D] = new DefaultComboTransformer[H,E,D],
        toD: H => D = new DefaultToD[H])
        (implicit ordering: Ordering[D])
  extends Walker[UnpackedChainedMetaVertex[V,H,E,D]] with Logging {
  
  object MetaVertexFilterAdapter extends MetaVertexFilter[Option[V],H,E,D] {
    override def apply(v: UnpackedMetaVertex[Option[V],H,E,D]) = v.packed.value match {
      case Some(_) => vertexFilter(v)
      case None => true
    }
  }
  
  val delegate = new UnpackedMetaDagWalker[Option[V],M,H,E,D,F](
    dag.delegate, selectionFilter, hedgeFilter, constraintFilter, MetaVertexFilterAdapter, comboTransformer, toD)
    
  // we must be able to recover the phantom-antecedents of non-phantom vertices
  // so that we can properly populate their state maps
  // unfortunately, this kills space complexity, since we must hold on to these until
  // all members of a packed vertex have been unpacked
  // TODO: Right now, we don't dectect when all members of a packed vertex have
  // been unpacked so that we can reclaim space. Could we refcount them?
  private val unpackedMap = new mutable.HashMap[(PackedVertex[Option[V]],Seq[D]), UnpackedMetaVertex[Option[V],H,E,D]]
  
  override def take(): Option[UnpackedChainedMetaVertex[V,H,E,D]] = recursiveTake()
  
  private def getOnly[A](seq: Seq[A]): A = seq match {
    case Seq(only) => only
    case _ => throw new RuntimeException("phantom chains can only have single parents; " +
          "expected exactly one element in: " + seq)
  }

  // note: cannot be @tailrec
  private def followPhantomChain(v: UnpackedMetaVertex[Option[V],H,E,D], edge: E, parentReal: Seq[D])
                                : Seq[(E, Seq[D])] = {
    trace("Follow phantom chain at " + v)
    v.packed.value match {
      case Some(_) => Seq( (edge, parentReal) )
      case None => {
        trace("Zip: " + v.edges.zip(v.parentRealizations))

        // use active hyperedges to find parents
        val parents: Seq[(E,UnpackedMetaVertex[Option[V],H,E,D])]
          = v.edges.zip(v.parentRealizations).
              flatMap { case (hyperedge, parentReals) => {
                import ducttape.util.Collections._
                zip3(dag.delegate.sources(hyperedge), hyperedge.e, parentReals).map { case (parent, e, parentReal) =>
                  // parent: PackedVertex[Option[V]]
                  trace("Resolving hyperedge %s: parent is %s".format(hyperedge, parent))
                  val uv: UnpackedMetaVertex[Option[V],H,E,D] = unpackedMap( (parent, parentReal.sorted(ordering)) )
                  (e, uv)
                }
              }
            }
        trace("Parents of %s are: %s".format(v, parents))
        parents match {
          case Seq() => Seq( (edge, parentReal) ) // this is a leaf/terminal phantom vertex
          case _ => {
            parents.flatMap { case (e, unpackedV) =>
              followPhantomChain(unpackedV, e, unpackedV.realization)
            }
          }
        }
      }
    }
  }
  
  @tailrec
  private def recursiveTake(): Option[UnpackedChainedMetaVertex[V,H,E,D]] = delegate.take() match {
    case None => None
    case Some(umv) => umv.packed.value match {
      case Some(packed) => {
        // we can have phantom vertex chains of arbitrary length
        // epsilons are already removed by our delegate
        import ducttape.util.Collections._
        val parentInfo: Seq[(Seq[E], Seq[Seq[D]])]
          = zip3(umv.edges, umv.parentRealizations, umv.edges).
            map { case (hyperedge, parentReals, hyperEdge) =>
              val munged: Seq[(E, Seq[D])] = zip3(dag.delegate.sources(hyperedge), parentReals, hyperedge.e) flatMap {
                case (parent, parentReal, edge) => {
                  trace("Begin backtracing phantom chain for " + parent)
                  val unpackedV = unpackedMap( (parent, parentReal.sorted(ordering)) )
                  val leafParents: Seq[(E, Seq[D])] = followPhantomChain(unpackedV, edge, parentReal)
                  leafParents
                }
              }
{'fraction_non_alphanumeric': 0.10715242432360032, 'fraction_numerical': 0.0006250558085543353, 'mean_word_length': 2.750837240455459, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3142728', 'n_tokens_mistral': 3432, 'n_tokens_neox': 3306, 'n_words': 835}