<commit_before>package slamdata.engine.physical.mongodb

import collection.immutable.ListMap

import slamdata.engine.fs.Path
import slamdata.engine.{Error}
import slamdata.engine.fp._
import optimize.pipeline._
import WorkflowTask._

import scalaz._
import Scalaz._
import monocle.Macro._
import monocle.syntax._

// FIXME: FIXPOINT all the things!

/**
 * A WorkflowOp is basically an atomic operation, with references to its inputs.
 * After generating a tree of these (actually, a graph, but we’ll get to that),
 * we crush them down into a Workflow. This `crush` gives us a location to
 * optimize our workflow decisions. EG, A sequence of simple ops may be combined
 * into a single pipeline request, but if one of those operations contains JS,
 * we have to execute that outside of a pipeline, possibly reordering the other
 * operations to avoid having two pipelines with a JS operation in the middle.
 *
 * We should also try to implement the optimizations at
 * http://docs.mongodb.org/manual/core/aggregation-pipeline-optimization/ so
 * that we can build others potentially on top of them (including reordering
 * non-pipelines around pipelines, etc.).
 * 
 * Also, this doesn’t yet go far enough – EG, if we have a ProjectOp before a
 * MapReduceOp, it might make sense to collapse the ProjectOp into the mapping
 * function, but we no longer have a handle on the JS to make that happen.
 */
sealed trait WorkflowOp {
  def srcs: List[WorkflowOp]
  def coalesce: WorkflowOp
  def crush: WorkflowTask

  // TODO: Automatically call `coalesce` when an op is created, rather than here
  //       and recursively in every overriden coalesce.
  def finish: Workflow = {
    Workflow(this.coalesce.deleteUnusedFields(Set.empty).crush)
  }

  def vertices: List[WorkflowOp] = this :: srcs.flatMap(_.vertices)

  import ExprOp.{GroupOp => _, _}
  import PipelineOp._
  import WorkflowOp._

  def rewriteRefs(applyVar0: PartialFunction[DocVar, DocVar]): this.type = {
    val applyVar = (f: DocVar) => applyVar0.lift(f).getOrElse(f)

    def applyExprOp(e: ExprOp): ExprOp = e.mapUp {
      case f : DocVar => applyVar(f)
    }

    def applyFieldName(name: BsonField): BsonField = {
      applyVar(DocField(name)).deref.getOrElse(name) // TODO: Delete field if it's transformed away to nothing???
    }

    def applySelector(s: Selector): Selector = s.mapUpFields(PartialFunction(applyFieldName _))

    def applyReshape(shape: Reshape): Reshape = shape match {
      case Reshape.Doc(value) => Reshape.Doc(value.transform {
        case (k, -\/(e)) => -\/(applyExprOp(e))
        case (k, \/-(r)) => \/-(applyReshape(r))
      })

      case Reshape.Arr(value) => Reshape.Arr(value.transform {
        case (k, -\/(e)) => -\/(applyExprOp(e))
        case (k, \/-(r)) => \/-(applyReshape(r))
      })
    }

    def applyGrouped(grouped: Grouped): Grouped = Grouped(grouped.value.transform {
      case (k, groupOp) => applyExprOp(groupOp) match {
        case groupOp : ExprOp.GroupOp => groupOp
        case _ => sys.error("Transformation changed the type -- error!")
      }
    })

    def applyMap[A](m: ListMap[BsonField, A]): ListMap[BsonField, A] = m.map(t => applyFieldName(t._1) -> t._2)

    def applyNel[A](m: NonEmptyList[(BsonField, A)]): NonEmptyList[(BsonField, A)] = m.map(t => applyFieldName(t._1) -> t._2)

    def applyFindQuery(q: FindQuery): FindQuery = {
      q.copy(
        query   = applySelector(q.query),
        max     = q.max.map(applyMap _),
        min     = q.min.map(applyMap _),
        orderby = q.orderby.map(applyNel _)
      )
    }

    (this match {
      case ProjectOp(src, shape)     => ProjectOp(src, applyReshape(shape))
      case GroupOp(src, grouped, by) =>
        GroupOp(src,
          applyGrouped(grouped), by.bimap(applyExprOp _, applyReshape _))
      case MatchOp(src, s)           => MatchOp(src, applySelector(s))
      case RedactOp(src, e)          => RedactOp(src, applyExprOp(e))
      case v @ LimitOp(src, _)       => v
      case v @ SkipOp(src, _)        => v
      case v @ UnwindOp(src, f)      => UnwindOp(src, applyVar(f))
      case v @ SortOp(src, l)        => SortOp(src, applyNel(l))
      // case v @ OutOp(src, _)         => v
      case g : GeoNearOp             =>
        g.copy(
          distanceField = applyFieldName(g.distanceField),
          query = g.query.map(applyFindQuery _))
      case v                       => v
    }).asInstanceOf[this.type]
  }

  final def refs: List[DocVar] = {
    // FIXME: Sorry world
    val vf = new scala.collection.mutable.ListBuffer[DocVar]

    rewriteRefs {
      case v => vf += v; v
    }

    vf.toList
  }

  def collectShapes: (List[PipelineOp.Reshape], WorkflowOp) = this match {
    case ProjectOp(src, shape) =>
      Arrow[Function1].first((x: List[PipelineOp.Reshape]) => shape :: x)(src.collectShapes)
    case _                     => (Nil, this)
  }

  def map(f: WorkflowOp => WorkflowOp): WorkflowOp = this match {
    case _: SourceOp => this
    case p: WPipelineOp => p.reparent(f(p.src))
    case _ => sys.error("FIXME: need to handle branching ops")
  }

  def deleteUnusedFields(usedRefs: Set[DocVar]): WorkflowOp = {
    def getRefs(op: WorkflowOp): Set[DocVar] = (op match {
       // Don't count unwinds (if the var isn't referenced elsewhere, it's effectively unused)
      case UnwindOp(_, _) => Nil
      case _ => op.refs
    }).toSet<commit_msg>Complete WorkflowOp map.<commit_after>package slamdata.engine.physical.mongodb

import collection.immutable.ListMap

import slamdata.engine.fs.Path
import slamdata.engine.{Error}
import slamdata.engine.fp._
import optimize.pipeline._
import WorkflowTask._

import scalaz._
import Scalaz._
import monocle.Macro._
import monocle.syntax._

// FIXME: FIXPOINT all the things!

/**
 * A WorkflowOp is basically an atomic operation, with references to its inputs.
 * After generating a tree of these (actually, a graph, but we’ll get to that),
 * we crush them down into a Workflow. This `crush` gives us a location to
 * optimize our workflow decisions. EG, A sequence of simple ops may be combined
 * into a single pipeline request, but if one of those operations contains JS,
 * we have to execute that outside of a pipeline, possibly reordering the other
 * operations to avoid having two pipelines with a JS operation in the middle.
 *
 * We should also try to implement the optimizations at
 * http://docs.mongodb.org/manual/core/aggregation-pipeline-optimization/ so
 * that we can build others potentially on top of them (including reordering
 * non-pipelines around pipelines, etc.).
 * 
 * Also, this doesn’t yet go far enough – EG, if we have a ProjectOp before a
 * MapReduceOp, it might make sense to collapse the ProjectOp into the mapping
 * function, but we no longer have a handle on the JS to make that happen.
 */
sealed trait WorkflowOp {
  def srcs: List[WorkflowOp]
  def coalesce: WorkflowOp
  def crush: WorkflowTask

  // TODO: Automatically call `coalesce` when an op is created, rather than here
  //       and recursively in every overriden coalesce.
  def finish: Workflow = {
    Workflow(this.coalesce.deleteUnusedFields(Set.empty).crush)
  }

  def vertices: List[WorkflowOp] = this :: srcs.flatMap(_.vertices)

  import ExprOp.{GroupOp => _, _}
  import PipelineOp._
  import WorkflowOp._

  def rewriteRefs(applyVar0: PartialFunction[DocVar, DocVar]): this.type = {
    val applyVar = (f: DocVar) => applyVar0.lift(f).getOrElse(f)

    def applyExprOp(e: ExprOp): ExprOp = e.mapUp {
      case f : DocVar => applyVar(f)
    }

    def applyFieldName(name: BsonField): BsonField = {
      applyVar(DocField(name)).deref.getOrElse(name) // TODO: Delete field if it's transformed away to nothing???
    }

    def applySelector(s: Selector): Selector = s.mapUpFields(PartialFunction(applyFieldName _))

    def applyReshape(shape: Reshape): Reshape = shape match {
      case Reshape.Doc(value) => Reshape.Doc(value.transform {
        case (k, -\/(e)) => -\/(applyExprOp(e))
        case (k, \/-(r)) => \/-(applyReshape(r))
      })

      case Reshape.Arr(value) => Reshape.Arr(value.transform {
        case (k, -\/(e)) => -\/(applyExprOp(e))
        case (k, \/-(r)) => \/-(applyReshape(r))
      })
    }

    def applyGrouped(grouped: Grouped): Grouped = Grouped(grouped.value.transform {
      case (k, groupOp) => applyExprOp(groupOp) match {
        case groupOp : ExprOp.GroupOp => groupOp
        case _ => sys.error("Transformation changed the type -- error!")
      }
    })

    def applyMap[A](m: ListMap[BsonField, A]): ListMap[BsonField, A] = m.map(t => applyFieldName(t._1) -> t._2)

    def applyNel[A](m: NonEmptyList[(BsonField, A)]): NonEmptyList[(BsonField, A)] = m.map(t => applyFieldName(t._1) -> t._2)

    def applyFindQuery(q: FindQuery): FindQuery = {
      q.copy(
        query   = applySelector(q.query),
        max     = q.max.map(applyMap _),
        min     = q.min.map(applyMap _),
        orderby = q.orderby.map(applyNel _)
      )
    }

    (this match {
      case ProjectOp(src, shape)     => ProjectOp(src, applyReshape(shape))
      case GroupOp(src, grouped, by) =>
        GroupOp(src,
          applyGrouped(grouped), by.bimap(applyExprOp _, applyReshape _))
      case MatchOp(src, s)           => MatchOp(src, applySelector(s))
      case RedactOp(src, e)          => RedactOp(src, applyExprOp(e))
      case v @ LimitOp(src, _)       => v
      case v @ SkipOp(src, _)        => v
      case v @ UnwindOp(src, f)      => UnwindOp(src, applyVar(f))
      case v @ SortOp(src, l)        => SortOp(src, applyNel(l))
      // case v @ OutOp(src, _)         => v
      case g : GeoNearOp             =>
        g.copy(
          distanceField = applyFieldName(g.distanceField),
          query = g.query.map(applyFindQuery _))
      case v                       => v
    }).asInstanceOf[this.type]
  }

  final def refs: List[DocVar] = {
    // FIXME: Sorry world
    val vf = new scala.collection.mutable.ListBuffer[DocVar]

    rewriteRefs {
      case v => vf += v; v
    }

    vf.toList
  }

  def collectShapes: (List[PipelineOp.Reshape], WorkflowOp) = this match {
    case ProjectOp(src, shape) =>
      Arrow[Function1].first((x: List[PipelineOp.Reshape]) => shape :: x)(src.collectShapes)
    case _                     => (Nil, this)
  }

  def map(f: WorkflowOp => WorkflowOp): WorkflowOp = this match {
    case _: SourceOp => this
    case p: WPipelineOp => p.reparent(f(p.src))
    case MapReduceOp(src, mr) => MapReduceOp(f(src), mr)
    case FoldLeftOp(srcs) => FoldLeftOp(srcs.map(f))
    case JoinOp(srcs) => JoinOp(srcs.map(f))
  }

  def deleteUnusedFields(usedRefs: Set[DocVar]): WorkflowOp = {
    def getRefs(op: WorkflowOp): Set[DocVar] = (op match {
       // Don't count unwinds (if the var isn't referenced elsewhere, it's effectively unused)
      case UnwindOp(_, _) => Nil
      case _ => op.refs
    }).toSet
{'fraction_non_alphanumeric': 0.1229966114113014, 'fraction_numerical': 0.0012821686967671032, 'mean_word_length': 2.886120996441281, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 79, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, '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': '6888845', 'n_tokens_mistral': 3654, 'n_tokens_neox': 3530, 'n_words': 995}