<commit_before>package ch.unibas.dmi.dbis.adam.query.scanner

import ch.unibas.dmi.dbis.adam.datatypes.feature.Feature._
import ch.unibas.dmi.dbis.adam.entity.Tuple.TupleID
import ch.unibas.dmi.dbis.adam.entity.{Entity, Tuple}
import ch.unibas.dmi.dbis.adam.main.SparkStartup
import ch.unibas.dmi.dbis.adam.query.Result
import ch.unibas.dmi.dbis.adam.query.query.NearestNeighbourQuery
import org.apache.spark.sql.{DataFrame, Row}

import scala.collection.immutable.HashSet
import scala.collection.mutable.ListBuffer

/**
  * adamtwo
  *
  * Scanner for the feature data.
  *
  * <NAME>
  * August 2015
  */
object FeatureScanner {
  /**
    * Scans the feature data based on a nearest neighbour query.
    *
    * @param entity
    * @param query
    * @param filter if filter is defined, we pre-filter for the features
    * @return
    */
  def apply(entity: Entity, query: NearestNeighbourQuery, filter: Option[HashSet[TupleID]]): DataFrame = {
    val data = if (filter.isDefined) {
      //scan based on tuples filtered in index
      SparkStartup.sc.setLocalProperty("spark.scheduler.pool", "feature")
      SparkStartup.sc.setJobGroup(query.queryID.getOrElse(""), entity.entityname, true)
      entity.filter(filter.get).collect()
    } else {
      //sequential scan
      SparkStartup.sc.setLocalProperty("spark.scheduler.pool", "slow")
      SparkStartup.sc.setJobGroup(query.queryID.getOrElse(""), entity.entityname, true)
      entity.rdd.map(row => (row: Tuple)).collect()
    }

    val it = data.par.iterator

    //compute distance for candidates
    val ls = ListBuffer[Result]()
    while (it.hasNext) {
      val tuple = it.next
      val f: FeatureVector = tuple.feature
      ls += Result(query.distance(query.q, f), tuple.tid)
    }

    //kNN
    val result: ListBuffer[Result] = ls.sortBy(_.distance).take(query.k)

    //to DF
    val rdd = SparkStartup.sc.parallelize(result).map(res => Row(res.distance, res.tid))
    SparkStartup.sqlContext.createDataFrame(rdd, Result.resultSchema)
  }
}
<commit_msg>sequential scan goes explicitly only on feature data, joining with metadata happens only after<commit_after>package ch.unibas.dmi.dbis.adam.query.scanner

import ch.unibas.dmi.dbis.adam.datatypes.feature.Feature._
import ch.unibas.dmi.dbis.adam.entity.Tuple.TupleID
import ch.unibas.dmi.dbis.adam.entity.{Entity, Tuple}
import ch.unibas.dmi.dbis.adam.main.SparkStartup
import ch.unibas.dmi.dbis.adam.query.Result
import ch.unibas.dmi.dbis.adam.query.query.NearestNeighbourQuery
import org.apache.spark.sql.{DataFrame, Row}

import scala.collection.immutable.HashSet
import scala.collection.mutable.ListBuffer

/**
  * adamtwo
  *
  * Scanner for the feature data.
  *
  * <NAME>
  * August 2015
  */
object FeatureScanner {
  /**
    * Scans the feature data based on a nearest neighbour query.
    *
    * @param entity
    * @param query
    * @param filter if filter is defined, we pre-filter for the features
    * @return
    */
  def apply(entity: Entity, query: NearestNeighbourQuery, filter: Option[HashSet[TupleID]]): DataFrame = {
    val data = if (filter.isDefined) {
      //scan based on tuples filtered in index
      SparkStartup.sc.setLocalProperty("spark.scheduler.pool", "feature")
      SparkStartup.sc.setJobGroup(query.queryID.getOrElse(""), entity.entityname, true)
      entity.filter(filter.get).collect()
    } else {
      //sequential scan
      SparkStartup.sc.setLocalProperty("spark.scheduler.pool", "slow")
      SparkStartup.sc.setJobGroup(query.queryID.getOrElse(""), entity.entityname, true)
      entity.getFeaturedata.map(row => (row: Tuple)).collect()
    }

    val it = data.par.iterator

    //compute distance for candidates
    val ls = ListBuffer[Result]()
    while (it.hasNext) {
      val tuple = it.next
      val f: FeatureVector = tuple.feature
      ls += Result(query.distance(query.q, f), tuple.tid)
    }

    //kNN
    val result: ListBuffer[Result] = ls.sortBy(_.distance).take(query.k)

    //to DF
    val rdd = SparkStartup.sc.parallelize(result).map(res => Row(res.distance, res.tid))
    SparkStartup.sqlContext.createDataFrame(rdd, Result.resultSchema)
  }
}

{'fraction_non_alphanumeric': 0.12602409638554216, 'fraction_numerical': 0.0019277108433734939, 'mean_word_length': 3.4443254817987152, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 9, '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': '7663235', 'n_tokens_mistral': 1453, 'n_tokens_neox': 1405, 'n_words': 277}