<commit_before>package ch.unibas.dmi.dbis.adam.index.structures.spectrallsh

import breeze.linalg._
import ch.unibas.dmi.dbis.adam.datatypes.Feature._
import ch.unibas.dmi.dbis.adam.datatypes.MovableFeature
import ch.unibas.dmi.dbis.adam.datatypes.bitString.BitString
import ch.unibas.dmi.dbis.adam.index.Index._
import ch.unibas.dmi.dbis.adam.index.{Index, IndexMetaStorage, IndexMetaStorageBuilder, IndexTuple}
import ch.unibas.dmi.dbis.adam.table.Table._
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.DataFrame

import scala.collection.immutable.BitSet

/**
 * adamtwo
 *
 * <NAME>
 * August 2015
 */
class SpectralLSHIndex(val indexname: IndexName, val tablename: TableName, protected val indexdata: DataFrame, private val indexMetaData: SpectralLSHIndexMetaData)
  extends Index {

  /**
   *
   * @return
   */
  override protected lazy val indextuples : RDD[IndexTuple] = {
    indexdata
      .map{ tuple =>
      IndexTuple(tuple.getLong(0), tuple.getAs[BitString[_]](1)) }
  }

  /**
   *
   * @param q
   * @param options
   * @return
   */
  override def scan(q: WorkingVector, options: Map[String, String]): BitSet = {
    val k = options("k").toInt
    val numOfQueries = options.getOrElse("numOfQ", "3").toInt

    import MovableFeature.conv_feature2MovableFeature
    val originalQuery = SpectralLSHUtils.hashFeature(q, indexMetaData)
    val queries = List.fill(numOfQueries)(SpectralLSHUtils.hashFeature(q.move(indexMetaData.radius), indexMetaData)) ::: List(originalQuery)

    val ids = indextuples.map { tuple =>
        var i = 0
        var score = 0
        while(i < queries.length){
          val query = queries(i)
          score += tuple.bits.intersectionCount(query)
        }

      (tuple.tid, score)
    }.takeOrdered(k * 3)(Ordering[Int].reverse.on(x => x._2)).map(_._1).toSeq

    BitSet(ids.map(_.toInt):_*)
  }

  /**
   *
   * @param metaBuilder
   */
  override private[index] def prepareMeta(metaBuilder : IndexMetaStorageBuilder) : Unit = {
    //metaBuilder.put("pca", indexMetaData.pca.toDenseMatrix)
    metaBuilder.put("pca_mat", indexMetaData.pca.toDenseMatrix.toArray)<commit_msg>staged results (per partition) for SLSH<commit_after>package ch.unibas.dmi.dbis.adam.index.structures.spectrallsh

import breeze.linalg._
import ch.unibas.dmi.dbis.adam.datatypes.Feature._
import ch.unibas.dmi.dbis.adam.datatypes.MovableFeature
import ch.unibas.dmi.dbis.adam.datatypes.bitString.BitString
import ch.unibas.dmi.dbis.adam.index.Index._
import ch.unibas.dmi.dbis.adam.index.{Index, IndexMetaStorage, IndexMetaStorageBuilder, IndexTuple}
import ch.unibas.dmi.dbis.adam.table.Table._
import com.google.common.collect.MinMaxPriorityQueue
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.DataFrame

import scala.collection.immutable.BitSet
import scala.collection.mutable.ListBuffer

/**
 * adamtwo
 *
 * <NAME>
 * August 2015
 */
class SpectralLSHIndex(val indexname: IndexName, val tablename: TableName, protected val indexdata: DataFrame, private val indexMetaData: SpectralLSHIndexMetaData)
  extends Index {

  /**
   *
   * @return
   */
  override protected lazy val indextuples : RDD[IndexTuple] = {
    indexdata
      .map{ tuple =>
      IndexTuple(tuple.getLong(0), tuple.getAs[BitString[_]](1)) }
  }

  /**
   *
   * @param q
   * @param options
   * @return
   */
  override def scan(q: WorkingVector, options: Map[String, String]): BitSet = {
    val k = options("k").toInt
    val numOfQueries = options.getOrElse("numOfQ", "3").toInt

    import MovableFeature.conv_feature2MovableFeature
    val originalQuery = SpectralLSHUtils.hashFeature(q, indexMetaData)
    val queries = (List.fill(numOfQueries)(SpectralLSHUtils.hashFeature(q.move(indexMetaData.radius), indexMetaData)) ::: List(originalQuery)).par

    val ids = indextuples
      .mapPartitions(tuplesIt => {
      //TODO: put in separate class, cf. VA
        @transient var ls =  ListBuffer[Long]()
        @transient var queue = MinMaxPriorityQueue.orderedBy(scala.math.Ordering.Int).maximumSize(k).create[Int]
        @transient var max =  Float.MaxValue

        while(tuplesIt.hasNext){
          val tuple = tuplesIt.next()

          var i = 0
          var score = 0
          while(i < queries.length){
            val query = queries(i)
            score += tuple.bits.intersectionCount(query)
            i += 1
          }

          if(score < max || queue.size < k){
            ls += tuple.tid
            queue.add(score)
            max = queue.peekLast()
          }
        }

        ls.toList.iterator
      }).collect()

    BitSet(ids.map(_.toInt):_*)
  }

  /**
   *
   * @param metaBuilder
   */
  override private[index] def prepareMeta(metaBuilder : IndexMetaStorageBuilder) : Unit = {
    //metaBuilder.put("pca", indexMetaData.pca.toDenseMatrix)
    metaBuilder.put("pca_mat", indexMetaData.pca.toDenseMatrix.toArray)
{'fraction_non_alphanumeric': 0.12379783098015142, 'fraction_numerical': 0.004910988336402701, 'mean_word_length': 3.314210061782877, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 10, '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': '684951', 'n_tokens_mistral': 1690, 'n_tokens_neox': 1644, 'n_words': 287}