/**

   * Join two datasets to approximately find all pairs of rows whose distance are smaller than

   * the threshold. If the [[outputCol]] is missing, the method will transform the data; if the

   * [[outputCol]] exists, it will use the [[outputCol]]. This allows caching of the transformed

   * data when necessary.

   *

   * @param datasetA One of the datasets to join.

   * @param datasetB Another dataset to join.

   * @param threshold The threshold for the distance of row pairs.

   * @param distCol Output column for storing the distance between each pair of rows.

   * @return A joined dataset containing pairs of rows. The original rows are in columns

   *         "datasetA" and "datasetB", and a column "distCol" is added to show the distance

   *         between each pair.

   */

  def approxSimilarityJoin(

      datasetA: Dataset[_],

      datasetB: Dataset[_],

      threshold: Double,

      distCol: String): Dataset[_] = {

    val leftColName = "datasetA"

    val rightColName = "datasetB"

    val explodeCols = Seq("entry", "hashValue")

    val explodedA = processDataset(datasetA, leftColName, explodeCols)

    // If this is a self join, we need to recreate the inputCol of datasetB to avoid ambiguity.

    // TODO: Remove recreateCol logic once SPARK-17154 is resolved.

    val explodedB = if (datasetA != datasetB) {

      processDataset(datasetB, rightColName, explodeCols)

    } else {

      val recreatedB = recreateCol(datasetB, $(inputCol), s"${$(inputCol)}#${Random.nextString(5)}")

      processDataset(recreatedB, rightColName, explodeCols)

    }

    // Do a hash join on where the exploded hash values are equal.

    val joinedDataset = explodedA.join(explodedB, explodeCols)

      .drop(explodeCols: _*).distinct()

    // Add a new column to store the distance of the two rows.

    val distUDF = udf((x: Vector, y: Vector) => keyDistance(x, y), DataTypes.DoubleType)

    val joinedDatasetWithDist = joinedDataset.select(col("*"),

      distUDF(col(s"$leftColName.${$(inputCol)}"), col(s"$rightColName.${$(inputCol)}")).as(distCol)

    )

    // Filter the joined datasets where the distance are smaller than the threshold.

    joinedDatasetWithDist.filter(col(distCol) < threshold)

  }

/**

 * :: Experimental ::

 *

 * Model produced by [[MinHashLSH]], where multiple hash functions are stored. Each hash function

 * is picked from the following family of hash functions, where a_i and b_i are randomly chosen

 * integers less than prime:

 *    `h_i(x) = ((x \cdot a_i + b_i) \mod prime)`

 *

 * This hash family is approximately min-wise independent according to the reference.

 *

 * Reference:

 * Tom Bohman, Colin Cooper, and Alan Frieze. "Min-wise independent linear permutations."

 * Electronic Journal of Combinatorics 7 (2000): R26.

 *

 * @param randCoefficients Pairs of random coefficients. Each pair is used by one hash function.

 */

@Experimental

@Since("2.1.0")

class MinHashLSHModel private[ml](

    override val uid: String,

    private[ml] val randCoefficients: Array[(Int, Int)])

  extends LSHModel[MinHashLSHModel] {

  /** @group setParam */

  @Since("2.4.0")

  override def setInputCol(value: String): this.type = super.set(inputCol, value)

  /** @group setParam */

  @Since("2.4.0")

  override def setOutputCol(value: String): this.type = super.set(outputCol, value)

  @Since("2.1.0")

  override protected[ml] def hashFunction(elems: Vector): Array[Vector] = {

    require(elems.numNonzeros > 0, "Must have at least 1 non zero entry.")

    val elemsList = elems.toSparse.indices.toList

    val hashValues = randCoefficients.map { case (a, b) =>

      elemsList.map { elem: Int =>

        ((1L + elem) * a + b) % MinHashLSH.HASH_PRIME

      }.min.toDouble

    }

    // TODO: Output vectors of dimension numHashFunctions in SPARK-18450

    hashValues.map(Vectors.dense(_))

  }

  @Since("2.1.0")

  override protected[ml] def keyDistance(x: Vector, y: Vector): Double = {

    val xSet = x.toSparse.indices.toSet

    val ySet = y.toSparse.indices.toSet

    val intersectionSize = xSet.intersect(ySet).size.toDouble

    val unionSize = xSet.size + ySet.size - intersectionSize

    assert(unionSize > 0, "The union of two input sets must have at least 1 elements")

    1 - intersectionSize / unionSize

  }

  @Since("2.1.0")

  override protected[ml] def hashDistance(x: Seq[Vector], y: Seq[Vector]): Double = {

    // Since it's generated by hashing, it will be a pair of dense vectors.

    // TODO: This hashDistance function requires more discussion in SPARK-18454

    x.zip(y).map(vectorPair =>

      vectorPair._1.toArray.zip(vectorPair._2.toArray).count(pair => pair._1 != pair._2)

    ).min

  }

  @Since("2.1.0")

  override def copy(extra: ParamMap): MinHashLSHModel = {

    val copied = new MinHashLSHModel(uid, randCoefficients).setParent(parent)

    copyValues(copied, extra)

  }

  @Since("2.1.0")

  override def write: MLWriter = new MinHashLSHModel.MinHashLSHModelWriter(this)

}