<commit_before>package ingraph.model.fplan

import ingraph.model.expr._
import ingraph.model.jplan
import ingraph.model.treenodes.{GenericBinaryNode, GenericLeafNode, GenericUnaryNode}
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, SortOrder}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan

trait FNode extends LogicalPlan {
  def extraAttributes: Seq[ResolvableName]
  def jnode: jplan.JNode
  def internalSchema: Seq[ResolvableName]
  override def children: Seq[FNode]
  override def output: Seq[Attribute] = Seq()
}

abstract class LeafFNode extends GenericLeafNode[FNode] with FNode {}
abstract class UnaryFNode extends GenericUnaryNode[FNode] with FNode {
  override def internalSchema: Seq[ResolvableName] = child.internalSchema
}
abstract class BinaryFNode extends GenericBinaryNode[FNode] with FNode {}

trait JoinLike extends BinaryFNode {
  def internalCommon: Seq[ResolvableName] = left.internalSchema.filter(x => right.internalSchema.map(_.resolvedName).contains(x.resolvedName))
  lazy val leftMask:  Seq[Int] = SchemaMapper.schemaToIndices(this, left)
  lazy val rightMask: Seq[Int] = SchemaMapper.schemaToIndices(this, right)
}
trait EquiJoinLike extends JoinLike {
  override def internalSchema: Seq[ResolvableName] =
    left.internalSchema ++ right.internalSchema.filter(x => !left.internalSchema.map(_.resolvedName).contains(x.resolvedName))
}

// leaf nodes
case class GetVertices(extraAttributes: Seq[ResolvableName], jnode: jplan.GetVertices) extends LeafFNode {
  override def internalSchema: Seq[ResolvableName] = jnode.output ++ extraAttributes
}

case class GetEdges(extraAttributes: Seq[ResolvableName], jnode: jplan.GetEdges) extends LeafFNode {
  override def internalSchema: Seq[ResolvableName] = jnode.output ++ extraAttributes
}

case class Dual(jnode: jplan.Dual) extends LeafFNode {
  override def internalSchema = Seq()
  override def extraAttributes: Seq[ResolvableName] = Seq()
}

// unary nodes
case class AllDifferent(extraAttributes: Seq[ResolvableName],
                        jnode: jplan.AllDifferent,
                        child: FNode
                        ) extends UnaryFNode {

}

case class DuplicateElimination(extraAttributes: Seq[ResolvableName],
                                jnode: jplan.DuplicateElimination,
                                child: FNode
                               ) extends UnaryFNode {}

case class Production(extraAttributes: Seq[ResolvableName],
                      jnode: jplan.Production,
                      child: FNode
                     ) extends UnaryFNode {
  override def output = jnode.output
}

case class Projection(extraAttributes: Seq[ResolvableName],
                      jnode: jplan.Projection,
                      child: FNode
                     ) extends UnaryFNode {
  override def internalSchema: Seq[ResolvableName] = jnode.projectList ++ extraAttributes
  lazy val projectionTuple: Seq[Expression] =
    jnode.projectList.map(_.child).map(SchemaMapper.transformExpression(_, child.internalSchema)) ++
      extraAttributes.map(SchemaMapper.transformExpression(_, child.internalSchema))
}

case class Grouping(extraAttributes: Seq[ResolvableName],
                      jnode: jplan.Grouping,
                      child: FNode
                     ) extends UnaryFNode {
  override def internalSchema: Seq[ResolvableName] = jnode.projectList ++ extraAttributes
  lazy val aggregationCriteria: Seq[Expression] =
    jnode.aggregationCriteria.map(SchemaMapper.transformExpression(_, child.internalSchema))<commit_msg>Add 'outputNames' method to ProductionNode<commit_after>package ingraph.model.fplan

import ingraph.model.expr._
import ingraph.model.jplan
import ingraph.model.treenodes.{GenericBinaryNode, GenericLeafNode, GenericUnaryNode}
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, SortOrder}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan

trait FNode extends LogicalPlan {
  def extraAttributes: Seq[ResolvableName]
  def jnode: jplan.JNode
  def internalSchema: Seq[ResolvableName]
  override def children: Seq[FNode]
  override def output: Seq[Attribute] = Seq()
}

abstract class LeafFNode extends GenericLeafNode[FNode] with FNode {}
abstract class UnaryFNode extends GenericUnaryNode[FNode] with FNode {
  override def internalSchema: Seq[ResolvableName] = child.internalSchema
}
abstract class BinaryFNode extends GenericBinaryNode[FNode] with FNode {}

trait JoinLike extends BinaryFNode {
  def internalCommon: Seq[ResolvableName] = left.internalSchema.filter(x => right.internalSchema.map(_.resolvedName).contains(x.resolvedName))
  lazy val leftMask:  Seq[Int] = SchemaMapper.schemaToIndices(this, left)
  lazy val rightMask: Seq[Int] = SchemaMapper.schemaToIndices(this, right)
}
trait EquiJoinLike extends JoinLike {
  override def internalSchema: Seq[ResolvableName] =
    left.internalSchema ++ right.internalSchema.filter(x => !left.internalSchema.map(_.resolvedName).contains(x.resolvedName))
}

// leaf nodes
case class GetVertices(extraAttributes: Seq[ResolvableName], jnode: jplan.GetVertices) extends LeafFNode {
  override def internalSchema: Seq[ResolvableName] = jnode.output ++ extraAttributes
}

case class GetEdges(extraAttributes: Seq[ResolvableName], jnode: jplan.GetEdges) extends LeafFNode {
  override def internalSchema: Seq[ResolvableName] = jnode.output ++ extraAttributes
}

case class Dual(jnode: jplan.Dual) extends LeafFNode {
  override def internalSchema = Seq()
  override def extraAttributes: Seq[ResolvableName] = Seq()
}

// unary nodes
case class AllDifferent(extraAttributes: Seq[ResolvableName],
                        jnode: jplan.AllDifferent,
                        child: FNode
                        ) extends UnaryFNode {

}

case class DuplicateElimination(extraAttributes: Seq[ResolvableName],
                                jnode: jplan.DuplicateElimination,
                                child: FNode
                               ) extends UnaryFNode {}

case class Production(extraAttributes: Seq[ResolvableName],
                      jnode: jplan.Production,
                      child: FNode
                     ) extends UnaryFNode {
  override def output = jnode.output
  def outputNames: Iterable[String] = output.map(_.resolvedName.get.resolvedName)
}

case class Projection(extraAttributes: Seq[ResolvableName],
                      jnode: jplan.Projection,
                      child: FNode
                     ) extends UnaryFNode {
  override def internalSchema: Seq[ResolvableName] = jnode.projectList ++ extraAttributes
  lazy val projectionTuple: Seq[Expression] =
    jnode.projectList.map(_.child).map(SchemaMapper.transformExpression(_, child.internalSchema)) ++
      extraAttributes.map(SchemaMapper.transformExpression(_, child.internalSchema))
}

case class Grouping(extraAttributes: Seq[ResolvableName],
                      jnode: jplan.Grouping,
                      child: FNode
                     ) extends UnaryFNode {
  override def internalSchema: Seq[ResolvableName] = jnode.projectList ++ extraAttributes
  lazy val aggregationCriteria: Seq[Expression] =
    jnode.aggregationCriteria.map(SchemaMapper.transformExpression(_, child.internalSchema))
{'fraction_non_alphanumeric': 0.08423950447350309, 'fraction_numerical': 0.0, 'mean_word_length': 3.6104060913705585, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '364883', 'n_tokens_mistral': 2086, 'n_tokens_neox': 2043, 'n_words': 453}