<commit_before>package org.vitrivr.adampro.query.handler.internal

import org.vitrivr.adampro.config.AttributeNames
import org.vitrivr.adampro.entity.Entity
import org.vitrivr.adampro.entity.Entity._
import org.vitrivr.adampro.main.AdamContext
import org.vitrivr.adampro.query.handler.generic.{ExpressionDetails, QueryEvaluationOptions, QueryExpression}
import org.vitrivr.adampro.query.query.BooleanQuery
import org.vitrivr.adampro.utils.Logging
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions._
import org.vitrivr.adampro.helpers.tracker.OperationTracker

import scala.collection.mutable

/**
  * adamtwo
  *
  * <NAME>
  * May 2016
  */
object BooleanFilterExpression extends Logging {

  /**
    *
    * @param entity entity
    * @param bq     boolean query
    */
  case class BooleanFilterScanExpression(private val entity: Entity)(private val bq: BooleanQuery, id: Option[String] = None)(filterExpr: Option[QueryExpression] = None)(@transient implicit val ac: AdamContext) extends QueryExpression(id) {
    override val info = ExpressionDetails(Some(entity.entityname), Some("Table Boolean-Scan Expression"), id, None)
    _children ++= filterExpr.map(Seq(_)).getOrElse(Seq())

    def this(entityname: EntityName)(bq: BooleanQuery, id: Option[String])(filterExpr: Option[QueryExpression])(implicit ac: AdamContext) {
      this(Entity.load(entityname).get)(bq, id)(filterExpr)(ac)
    }

    override protected def run(options : Option[QueryEvaluationOptions], filter: Option[DataFrame] = None)(tracker : OperationTracker)(implicit ac: AdamContext): Option[DataFrame] = {
      log.debug("run boolean filter operation on entity")

      ac.sc.setJobGroup(id.getOrElse(""), "boolean filter scan", interruptOnCancel = true)

      var df =  entity.getData(predicates = bq.where)

      var ids = mutable.Set[Any]()

      if (filter.isDefined) {
        ids ++= filter.get.select(entity.pk.name).collect().map(_.getAs[Any](entity.pk.name))
      }

      if (filterExpr.isDefined) {
        filterExpr.get.filter = filter
        ids ++= filterExpr.get.evaluate(options)(tracker).get.select(entity.pk.name).collect().map(_.getAs[Any](entity.pk.name))
      }

      if (ids.nonEmpty) {
        val idsBc = ac.sc.broadcast(ids)
        df = df.map(d => {
          val rdd = d.rdd.filter(x => idsBc.value.contains(x.getAs[Any](entity.pk.name)))
          ac.sqlContext.createDataFrame(rdd, d.schema)
        })
        tracker.addBroadcast(idsBc)
      }

      val res = df.map(BooleanFilterExpression.filter(_, bq))
      res
    }

    override def equals(that: Any): Boolean =
      that match {
        case that: BooleanFilterScanExpression => this.entity.equals(that.entity) && this.bq.equals(that.bq)
        case _ => false
      }

    override def hashCode(): Int = {
      val prime = 31
      var result = 1
      result = prime * result + entity.hashCode
      result = prime * result + bq.hashCode
      result
    }
  }

  /**
    *
    * @param bq boolean query
    */
  case class BooleanFilterAdHocExpression(expr: QueryExpression, bq: BooleanQuery, id: Option[String] = None)(implicit ac: AdamContext) extends QueryExpression(id) {
    override val info = ExpressionDetails(None, Some("Ad-Hoc Boolean-Scan Expression"), id, None)
    _children ++= Seq(expr)

    override protected def run(options : Option[QueryEvaluationOptions], filter: Option[DataFrame] = None)(tracker : OperationTracker)(implicit ac: AdamContext): Option[DataFrame] = {
      log.debug("run boolean filter operation on data")

      ac.sc.setJobGroup(id.getOrElse(""), "boolean filter scan", interruptOnCancel = true)<commit_msg>Boolean filtering hopefully implemented more efficiently<commit_after>package org.vitrivr.adampro.query.handler.internal

import org.vitrivr.adampro.config.AttributeNames
import org.vitrivr.adampro.entity.Entity
import org.vitrivr.adampro.entity.Entity._
import org.vitrivr.adampro.main.AdamContext
import org.vitrivr.adampro.query.handler.generic.{ExpressionDetails, QueryEvaluationOptions, QueryExpression}
import org.vitrivr.adampro.query.query.BooleanQuery
import org.vitrivr.adampro.utils.Logging
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions._
import org.vitrivr.adampro.helpers.tracker.OperationTracker

import scala.collection.mutable

/**
  * adamtwo
  *
  * <NAME>
  * May 2016
  */
object BooleanFilterExpression extends Logging {

  /**
    *
    * @param entity entity
    * @param bq     boolean query
    */
  case class BooleanFilterScanExpression(private val entity: Entity)(private val bq: BooleanQuery, id: Option[String] = None)(filterExpr: Option[QueryExpression] = None)(@transient implicit val ac: AdamContext) extends QueryExpression(id) {
    override val info = ExpressionDetails(Some(entity.entityname), Some("Table Boolean-Scan Expression"), id, None)
    _children ++= filterExpr.map(Seq(_)).getOrElse(Seq())

    def this(entityname: EntityName)(bq: BooleanQuery, id: Option[String])(filterExpr: Option[QueryExpression])(implicit ac: AdamContext) {
      this(Entity.load(entityname).get)(bq, id)(filterExpr)(ac)
    }

    override protected def run(options : Option[QueryEvaluationOptions], filter: Option[DataFrame] = None)(tracker : OperationTracker)(implicit ac: AdamContext): Option[DataFrame] = {
      log.debug("run boolean filter operation on entity")

      ac.sc.setJobGroup(id.getOrElse(""), "boolean filter scan", interruptOnCancel = true)

      var df =  entity.getData(predicates = bq.where)

      /*var ids = mutable.Set[Any]()

      if (filter.isDefined) {
        ids ++= filter.get.select(entity.pk.name).collect().map(_.getAs[Any](entity.pk.name))
      }

      if (filterExpr.isDefined) {
        filterExpr.get.filter = filter
        ids ++= filterExpr.get.evaluate(options)(tracker).get.select(entity.pk.name).collect().map(_.getAs[Any](entity.pk.name))
      }

      if (ids.nonEmpty) {
        val idsBc = ac.sc.broadcast(ids)
        df = df.map(d => {
          val rdd = d.rdd.filter(x => idsBc.value.contains(x.getAs[Any](entity.pk.name)))
          ac.sqlContext.createDataFrame(rdd, d.schema)
        })
        tracker.addBroadcast(idsBc)

        val res = df.map(BooleanFilterExpression.filter(_, bq))
        res
      }*/

      val ids = if (filter.isDefined && filterExpr.isDefined) {
        Some(filter.get.select(entity.pk.name).intersect(filterExpr.get.evaluate(options)(tracker).get.select(entity.pk.name)))
      } else if(filter.isDefined){
        Some(filter.get.select(entity.pk.name))
      } else if(filterExpr.isDefined){
        Some(filterExpr.get.evaluate(options)(tracker).get.select(entity.pk.name))
      } else {
        None
      }

     val res = if(ids.isDefined){
        import org.apache.spark.sql.functions.broadcast
        df.map(_.join(broadcast(ids.get), Seq(entity.pk.name.toString)))
      } else {
        df
      }

      res
    }

    override def equals(that: Any): Boolean =
      that match {
        case that: BooleanFilterScanExpression => this.entity.equals(that.entity) && this.bq.equals(that.bq)
        case _ => false
      }

    override def hashCode(): Int = {
      val prime = 31
      var result = 1
      result = prime * result + entity.hashCode
      result = prime * result + bq.hashCode
      result
    }
  }

  /**
    *
    * @param bq boolean query
    */
  case class BooleanFilterAdHocExpression(expr: QueryExpression, bq: BooleanQuery, id: Option[String] = None)(implicit ac: AdamContext) extends QueryExpression(id) {
    override val info = ExpressionDetails(None, Some("Ad-Hoc Boolean-Scan Expression"), id, None)
    _children ++= Seq(expr)

    override protected def run(options : Option[QueryEvaluationOptions], filter: Option[DataFrame] = None)(tracker : OperationTracker)(implicit ac: AdamContext): Option[DataFrame] = {
      log.debug("run boolean filter operation on data")

      ac.sc.setJobGroup(id.getOrElse(""), "boolean filter scan", interruptOnCancel = true)
{'fraction_non_alphanumeric': 0.1264051961029228, 'fraction_numerical': 0.0017486884836372721, 'mean_word_length': 3.5780445969125214, 'pattern_counts': {'":': 0, '<': 5, '<?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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4674284', 'n_tokens_mistral': 2520, 'n_tokens_neox': 2415, 'n_words': 484}