/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.hadoop.hbase.spark

import java.net.InetSocketAddress
import java.util
import java.util.UUID
import javax.management.openmbean.KeyAlreadyExistsException

import org.apache.yetus.audience.InterfaceAudience;
import org.apache.hadoop.hbase.fs.HFileSystem
import org.apache.hadoop.hbase._
import org.apache.hadoop.hbase.io.compress.Compression
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding
import org.apache.hadoop.hbase.io.hfile.{HFile, CacheConfig, HFileContextBuilder, HFileWriterImpl}
import org.apache.hadoop.hbase.regionserver.{HStore, HStoreFile, StoreFileWriter, BloomType}
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.mapred.JobConf
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.spark.HBaseRDDFunctions._
import org.apache.hadoop.hbase.client._
import scala.reflect.ClassTag
import org.apache.spark.{SerializableWritable, SparkContext}
import org.apache.hadoop.hbase.mapreduce.{TableMapReduceUtil,
TableInputFormat, IdentityTableMapper}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.mapreduce.Job
import org.apache.spark.streaming.dstream.DStream
import java.io._
import org.apache.hadoop.security.UserGroupInformation
import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod
import org.apache.hadoop.fs.{Path, FileAlreadyExistsException, FileSystem}
import scala.collection.mutable

/**
  * HBaseContext is a façade for HBase operations
  * like bulk put, get, increment, delete, and scan
  *
  * HBaseContext will take the responsibilities
  * of disseminating the configuration information
  * to the working and managing the life cycle of Connections.
 */
@InterfaceAudience.Public
class HBaseContext(@transient val sc: SparkContext,
                   @transient val config: Configuration,
                   val tmpHdfsConfgFile: String = null)
  extends Serializable with Logging {

  @transient var tmpHdfsConfiguration:Configuration = config
  @transient var appliedCredentials = false
  @transient val job = Job.getInstance(config)
  TableMapReduceUtil.initCredentials(job)
  val broadcastedConf = sc.broadcast(new SerializableWritable(config))

  LatestHBaseContextCache.latest = this

  if (tmpHdfsConfgFile != null && config != null) {
    val fs = FileSystem.newInstance(config)
    val tmpPath = new Path(tmpHdfsConfgFile)
    if (!fs.exists(tmpPath)) {
      val outputStream = fs.create(tmpPath)
      config.write(outputStream)
      outputStream.close()
    } else {
      logWarning("tmpHdfsConfigDir " + tmpHdfsConfgFile + " exist!!")
    }
  }

  /**
   * A simple enrichment of the traditional Spark RDD foreachPartition.
   * This function differs from the original in that it offers the
   * developer access to a already connected Connection object
   *
   * Note: Do not close the Connection object.  All Connection
   * management is handled outside this method
   *
   * @param rdd  Original RDD with data to iterate over
   * @param f    Function to be given a iterator to iterate through
   *             the RDD values and a Connection object to interact
   *             with HBase
   */
  def foreachPartition[T](rdd: RDD[T],
                          f: (Iterator[T], Connection) => Unit):Unit = {
    rdd.foreachPartition(
      it => hbaseForeachPartition(broadcastedConf, it, f))
  }

  /**
   * A simple enrichment of the traditional Spark Streaming dStream foreach
   * This function differs from the original in that it offers the
   * developer access to a already connected Connection object
   *
   * Note: Do not close the Connection object.  All Connection
   * management is handled outside this method
   *
   * @param dstream  Original DStream with data to iterate over
   * @param f        Function to be given a iterator to iterate through
   *                 the DStream values and a Connection object to
   *                 interact with HBase
   */
  def foreachPartition[T](dstream: DStream[T],
                    f: (Iterator[T], Connection) => Unit):Unit = {
    dstream.foreachRDD((rdd, time) => {
      foreachPartition(rdd, f)
    })
  }

  /**
   * A simple enrichment of the traditional Spark RDD mapPartition.
   * This function differs from the original in that it offers the
   * developer access to a already connected Connection object
   *
   * Note: Do not close the Connection object.  All Connection
   * management is handled outside this method
   *
   * @param rdd  Original RDD with data to iterate over
   * @param mp   Function to be given a iterator to iterate through
   *             the RDD values and a Connection object to interact
   *             with HBase
   * @return     Returns a new RDD generated by the user definition
   *             function just like normal mapPartition
   */
  def mapPartitions[T, R: ClassTag](rdd: RDD[T],
                                   mp: (Iterator[T], Connection) => Iterator[R]): RDD[R] = {

    rdd.mapPartitions[R](it => hbaseMapPartition[T, R](broadcastedConf,
      it,
      mp))

  }

  /**
   * A simple enrichment of the traditional Spark Streaming DStream
   * foreachPartition.
   *
   * This function differs from the original in that it offers the
   * developer access to a already connected Connection object
   *
   * Note: Do not close the Connection object.  All Connection
   * management is handled outside this method
   *
   * Note: Make sure to partition correctly to avoid memory issue when
   *       getting data from HBase
   *
   * @param dstream  Original DStream with data to iterate over
   * @param f       Function to be given a iterator to iterate through
   *                 the DStream values and a Connection object to
   *                 interact with HBase
   * @return         Returns a new DStream generated by the user
   *                 definition function just like normal mapPartition
   */
  def streamForeachPartition[T](dstream: DStream[T],
                                f: (Iterator[T], Connection) => Unit): Unit = {

    dstream.foreachRDD(rdd => this.foreachPartition(rdd, f))
  }

  /**
   * A simple enrichment of the traditional Spark Streaming DStream
   * mapPartition.
   *
   * This function differs from the original in that it offers the
   * developer access to a already connected Connection object
   *
   * Note: Do not close the Connection object.  All Connection
   * management is handled outside this method
   *
   * Note: Make sure to partition correctly to avoid memory issue when
   *       getting data from HBase
   *
   * @param dstream  Original DStream with data to iterate over
   * @param f       Function to be given a iterator to iterate through
   *                 the DStream values and a Connection object to
   *                 interact with HBase
   * @return         Returns a new DStream generated by the user
   *                 definition function just like normal mapPartition
   */
  def streamMapPartitions[T, U: ClassTag](dstream: DStream[T],
                                f: (Iterator[T], Connection) => Iterator[U]):
  DStream[U] = {
    dstream.mapPartitions(it => hbaseMapPartition[T, U](
      broadcastedConf,
      it,
      f))
  }

  /**
   * A simple abstraction over the HBaseContext.foreachPartition method.
   *
   * It allow addition support for a user to take RDD
   * and generate puts and send them to HBase.
   * The complexity of managing the Connection is
   * removed from the developer
   *
   * @param rdd       Original RDD with data to iterate over
   * @param tableName The name of the table to put into
   * @param f         Function to convert a value in the RDD to a HBase Put
   */
  def bulkPut[T](rdd: RDD[T], tableName: TableName, f: (T) => Put) {

    val tName = tableName.getName
    rdd.foreachPartition(
      it => hbaseForeachPartition[T](
        broadcastedConf,
        it,
        (iterator, connection) => {
          val m = connection.getBufferedMutator(TableName.valueOf(tName))
          iterator.foreach(T => m.mutate(f(T)))
          m.flush()
          m.close()
        }))
  }

  def applyCreds[T] (){
    if (!appliedCredentials) {
      appliedCredentials = true

      @transient val ugi = UserGroupInformation.getCurrentUser
      // specify that this is a proxy user
      ugi.setAuthenticationMethod(AuthenticationMethod.PROXY)
    }
  }

  /**
   * A simple abstraction over the HBaseContext.streamMapPartition method.
   *
   * It allow addition support for a user to take a DStream and
   * generate puts and send them to HBase.
   *
   * The complexity of managing the Connection is
   * removed from the developer
   *
   * @param dstream    Original DStream with data to iterate over
   * @param tableName  The name of the table to put into
   * @param f          Function to convert a value in
   *                   the DStream to a HBase Put
   */
  def streamBulkPut[T](dstream: DStream[T],
                       tableName: TableName,
                       f: (T) => Put) = {
    val tName = tableName.getName
    dstream.foreachRDD((rdd, time) => {
      bulkPut(rdd, TableName.valueOf(tName), f)
    })
  }

  /**
   * A simple abstraction over the HBaseContext.foreachPartition method.
   *
   * It allow addition support for a user to take a RDD and generate delete
   * and send them to HBase.  The complexity of managing the Connection is
   * removed from the developer
   *
   * @param rdd       Original RDD with data to iterate over
   * @param tableName The name of the table to delete from
   * @param f         Function to convert a value in the RDD to a
   *                  HBase Deletes
   * @param batchSize       The number of delete to batch before sending to HBase
   */
  def bulkDelete[T](rdd: RDD[T], tableName: TableName,
                    f: (T) => Delete, batchSize: Integer) {
    bulkMutation(rdd, tableName, f, batchSize)
  }

  /**
   * A simple abstraction over the HBaseContext.streamBulkMutation method.
   *
   * It allow addition support for a user to take a DStream and
   * generate Delete and send them to HBase.
   *
   * The complexity of managing the Connection is
   * removed from the developer
   *
   * @param dstream    Original DStream with data to iterate over
   * @param tableName  The name of the table to delete from
   * @param f          function to convert a value in the DStream to a
   *                   HBase Delete
   * @param batchSize        The number of deletes to batch before sending to HBase
   */
  def streamBulkDelete[T](dstream: DStream[T],
                          tableName: TableName,
                          f: (T) => Delete,
                          batchSize: Integer) = {
    streamBulkMutation(dstream, tableName, f, batchSize)
  }

  /**
   *  Under lining function to support all bulk mutations
   *
   *  May be opened up if requested
   */
  private def bulkMutation[T](rdd: RDD[T], tableName: TableName,
                              f: (T) => Mutation, batchSize: Integer) {

    val tName = tableName.getName
    rdd.foreachPartition(
      it => hbaseForeachPartition[T](
        broadcastedConf,
        it,
        (iterator, connection) => {
          val table = connection.getTable(TableName.valueOf(tName))
          val mutationList = new java.util.ArrayList[Mutation]
          iterator.foreach(T => {
            mutationList.add(f(T))
            if (mutationList.size >= batchSize) {
              table.batch(mutationList, null)
              mutationList.clear()
            }
          })
          if (mutationList.size() > 0) {
            table.batch(mutationList, null)
            mutationList.clear()
          }
          table.close()
        }))
  }

  /**
   *  Under lining function to support all bulk streaming mutations
   *
   *  May be opened up if requested
   */
  private def streamBulkMutation[T](dstream: DStream[T],
                                    tableName: TableName,
                                    f: (T) => Mutation,
                                    batchSize: Integer) = {
    val tName = tableName.getName
    dstream.foreachRDD((rdd, time) => {
      bulkMutation(rdd, TableName.valueOf(tName), f, batchSize)
    })
  }

  /**
   * A simple abstraction over the HBaseContext.mapPartition method.
   *
   * It allow addition support for a user to take a RDD and generates a
   * new RDD based on Gets and the results they bring back from HBase
   *
   * @param rdd     Original RDD with data to iterate over
   * @param tableName        The name of the table to get from
   * @param makeGet    function to convert a value in the RDD to a
   *                   HBase Get
   * @param convertResult This will convert the HBase Result object to
   *                   what ever the user wants to put in the resulting
   *                   RDD
   * return            new RDD that is created by the Get to HBase
   */
  def bulkGet[T, U: ClassTag](tableName: TableName,
                    batchSize: Integer,
                    rdd: RDD[T],
                    makeGet: (T) => Get,
                    convertResult: (Result) => U): RDD[U] = {

    val getMapPartition = new GetMapPartition(tableName,
      batchSize,
      makeGet,
      convertResult)

    rdd.mapPartitions[U](it =>
      hbaseMapPartition[T, U](
        broadcastedConf,
        it,
        getMapPartition.run))
  }

  /**
   * A simple abstraction over the HBaseContext.streamMap method.
   *
   * It allow addition support for a user to take a DStream and
   * generates a new DStream based on Gets and the results
   * they bring back from HBase
   *
   * @param tableName     The name of the table to get from
   * @param batchSize     The number of Gets to be sent in a single batch
   * @param dStream       Original DStream with data to iterate over
   * @param makeGet       Function to convert a value in the DStream to a
   *                      HBase Get
   * @param convertResult This will convert the HBase Result object to
   *                      what ever the user wants to put in the resulting
   *                      DStream
   * @return              A new DStream that is created by the Get to HBase
   */
  def streamBulkGet[T, U: ClassTag](tableName: TableName,
                                    batchSize: Integer,
                                    dStream: DStream[T],
                                    makeGet: (T) => Get,
                                    convertResult: (Result) => U): DStream[U] = {

    val getMapPartition = new GetMapPartition(tableName,
      batchSize,
      makeGet,
      convertResult)

    dStream.mapPartitions[U](it => hbaseMapPartition[T, U](
      broadcastedConf,
      it,
      getMapPartition.run))
  }

  /**
   * This function will use the native HBase TableInputFormat with the
   * given scan object to generate a new RDD
   *
   *  @param tableName the name of the table to scan
   *  @param scan      the HBase scan object to use to read data from HBase
   *  @param f         function to convert a Result object from HBase into
   *                   what the user wants in the final generated RDD
   *  @return          new RDD with results from scan
   */
  def hbaseRDD[U: ClassTag](tableName: TableName, scan: Scan,
                            f: ((ImmutableBytesWritable, Result)) => U): RDD[U] = {

    val job: Job = Job.getInstance(getConf(broadcastedConf))

    TableMapReduceUtil.initCredentials(job)
    TableMapReduceUtil.initTableMapperJob(tableName, scan,
      classOf[IdentityTableMapper], null, null, job)

    val jconf = new JobConf(job.getConfiguration)
    val jobCreds = jconf.getCredentials()
    UserGroupInformation.setConfiguration(sc.hadoopConfiguration)
    jobCreds.mergeAll(UserGroupInformation.getCurrentUser().getCredentials())

    new NewHBaseRDD(sc,
      classOf[TableInputFormat],
      classOf[ImmutableBytesWritable],
      classOf[Result],
      job.getConfiguration,
      this).map(f)
  }

  /**
   * A overloaded version of HBaseContext hbaseRDD that defines the
   * type of the resulting RDD
   *
   *  @param tableName the name of the table to scan
   *  @param scans     the HBase scan object to use to read data from HBase
   *  @return          New RDD with results from scan
   *
   */
  def hbaseRDD(tableName: TableName, scans: Scan):
  RDD[(ImmutableBytesWritable, Result)] = {

    hbaseRDD[(ImmutableBytesWritable, Result)](
      tableName,
      scans,
      (r: (ImmutableBytesWritable, Result)) => r)
  }

  /**
   *  underlining wrapper all foreach functions in HBaseContext
   */
  private def hbaseForeachPartition[T](configBroadcast:
                                       Broadcast[SerializableWritable[Configuration]],
                                        it: Iterator[T],
                                        f: (Iterator[T], Connection) => Unit) = {

    val config = getConf(configBroadcast)

    applyCreds
    // specify that this is a proxy user
    val smartConn = HBaseConnectionCache.getConnection(config)
    f(it, smartConn.connection)
    smartConn.close()
  }

  private def getConf(configBroadcast: Broadcast[SerializableWritable[Configuration]]):
  Configuration = {

    if (tmpHdfsConfiguration == null && tmpHdfsConfgFile != null) {
      val fs = FileSystem.newInstance(sc.hadoopConfiguration)
      val inputStream = fs.open(new Path(tmpHdfsConfgFile))
      tmpHdfsConfiguration = new Configuration(false)
      tmpHdfsConfiguration.readFields(inputStream)
      inputStream.close()
    }

    if (tmpHdfsConfiguration == null) {
      try {
        tmpHdfsConfiguration = configBroadcast.value.value
      } catch {
        case ex: Exception => logError("Unable to getConfig from broadcast", ex)
      }
    }
    tmpHdfsConfiguration
  }

  /**
   *  underlining wrapper all mapPartition functions in HBaseContext
   *
   */
  private def hbaseMapPartition[K, U](
                                       configBroadcast:
                                       Broadcast[SerializableWritable[Configuration]],
                                       it: Iterator[K],
                                       mp: (Iterator[K], Connection) =>
                                         Iterator[U]): Iterator[U] = {

    val config = getConf(configBroadcast)
    applyCreds

    val smartConn = HBaseConnectionCache.getConnection(config)
    val res = mp(it, smartConn.connection)
    smartConn.close()
    res
  }

  /**
   *  underlining wrapper all get mapPartition functions in HBaseContext
   */
  private class GetMapPartition[T, U](tableName: TableName,
                                      batchSize: Integer,
                                      makeGet: (T) => Get,
                                      convertResult: (Result) => U)
    extends Serializable {

    val tName = tableName.getName

    def run(iterator: Iterator[T], connection: Connection): Iterator[U] = {
      val table = connection.getTable(TableName.valueOf(tName))

      val gets = new java.util.ArrayList[Get]()
      var res = List[U]()

      while (iterator.hasNext) {
        gets.add(makeGet(iterator.next()))

        if (gets.size() == batchSize) {
          val results = table.get(gets)
          res = res ++ results.map(convertResult)
          gets.clear()
        }
      }
      if (gets.size() > 0) {
        val results = table.get(gets)
        res = res ++ results.map(convertResult)
        gets.clear()
      }
      table.close()
      res.iterator
    }
  }

  /**
   * Produces a ClassTag[T], which is actually just a casted ClassTag[AnyRef].
   *
   * This method is used to keep ClassTags out of the external Java API, as
   * the Java compiler cannot produce them automatically. While this
   * ClassTag-faking does please the compiler, it can cause problems at runtime
   * if the Scala API relies on ClassTags for correctness.
   *
   * Often, though, a ClassTag[AnyRef] will not lead to incorrect behavior,
   * just worse performance or security issues.
   * For instance, an Array of AnyRef can hold any type T, but may lose primitive
   * specialization.
   */
  private[spark]
  def fakeClassTag[T]: ClassTag[T] = ClassTag.AnyRef.asInstanceOf[ClassTag[T]]

  /**
   * Spark Implementation of HBase Bulk load for wide rows or when
   * values are not already combined at the time of the map process
   *
   * This will take the content from an existing RDD then sort and shuffle
   * it with respect to region splits.  The result of that sort and shuffle
   * will be written to HFiles.
   *
   * After this function is executed the user will have to call
   * LoadIncrementalHFiles.doBulkLoad(...) to move the files into HBase
   *
   * Also note this version of bulk load is different from past versions in
   * that it includes the qualifier as part of the sort process. The
   * reason for this is to be able to support rows will very large number
   * of columns.
   *
   * @param rdd                            The RDD we are bulk loading from
   * @param tableName                      The HBase table we are loading into
   * @param flatMap                        A flapMap function that will make every
   *                                       row in the RDD
   *                                       into N cells for the bulk load
   * @param stagingDir                     The location on the FileSystem to bulk load into
   * @param familyHFileWriteOptionsMap     Options that will define how the HFile for a
   *                                       column family is written
   * @param compactionExclude              Compaction excluded for the HFiles
   * @param maxSize                        Max size for the HFiles before they roll
   * @param nowTimeStamp                   Version timestamp
   * @tparam T                             The Type of values in the original RDD
   */
  def bulkLoad[T](rdd:RDD[T],
                  tableName: TableName,
                  flatMap: (T) => Iterator[(KeyFamilyQualifier, Array[Byte])],
                  stagingDir:String,
                  familyHFileWriteOptionsMap:
                  util.Map[Array[Byte], FamilyHFileWriteOptions] =
                  new util.HashMap[Array[Byte], FamilyHFileWriteOptions],
                  compactionExclude: Boolean = false,
                  maxSize:Long = HConstants.DEFAULT_MAX_FILE_SIZE,
                  nowTimeStamp:Long = System.currentTimeMillis()):
  Unit = {
    val stagingPath = new Path(stagingDir)
    val fs = stagingPath.getFileSystem(config)
    if (fs.exists(stagingPath)) {
      throw new FileAlreadyExistsException("Path " + stagingDir + " already exists")
    }
    val conn = HBaseConnectionCache.getConnection(config)
    try {
      val regionLocator = conn.getRegionLocator(tableName)
      val startKeys = regionLocator.getStartKeys
      if (startKeys.length == 0) {
        logInfo("Table " + tableName.toString + " was not found")
      }
      val defaultCompressionStr = config.get("hfile.compression",
        Compression.Algorithm.NONE.getName)
      val hfileCompression = HFileWriterImpl
        .compressionByName(defaultCompressionStr)
      val tableRawName = tableName.getName

      val familyHFileWriteOptionsMapInternal =
        new util.HashMap[ByteArrayWrapper, FamilyHFileWriteOptions]

      val entrySetIt = familyHFileWriteOptionsMap.entrySet().iterator()

      while (entrySetIt.hasNext) {
        val entry = entrySetIt.next()
        familyHFileWriteOptionsMapInternal.put(new ByteArrayWrapper(entry.getKey), entry.getValue)
      }

      val regionSplitPartitioner =
        new BulkLoadPartitioner(startKeys)

      //This is where all the magic happens
      //Here we are going to do the following things
      // 1. FlapMap every row in the RDD into key column value tuples
      // 2. Then we are going to repartition sort and shuffle
      // 3. Finally we are going to write out our HFiles
      rdd.flatMap( r => flatMap(r)).
        repartitionAndSortWithinPartitions(regionSplitPartitioner).
        hbaseForeachPartition(this, (it, conn) => {

          val conf = broadcastedConf.value.value
          val fs = FileSystem.get(conf)
          val writerMap = new mutable.HashMap[ByteArrayWrapper, WriterLength]
          var previousRow:Array[Byte] = HConstants.EMPTY_BYTE_ARRAY
          var rollOverRequested = false
          val localTableName = TableName.valueOf(tableRawName)

          //Here is where we finally iterate through the data in this partition of the
          //RDD that has been sorted and partitioned
          it.foreach{ case (keyFamilyQualifier, cellValue:Array[Byte]) =>

            val wl = writeValueToHFile(keyFamilyQualifier.rowKey,
              keyFamilyQualifier.family,
              keyFamilyQualifier.qualifier,
              cellValue,
              nowTimeStamp,
              fs,
              conn,
              localTableName,
              conf,
              familyHFileWriteOptionsMapInternal,
              hfileCompression,
              writerMap,
              stagingDir)

            rollOverRequested = rollOverRequested || wl.written > maxSize

            //This will only roll if we have at least one column family file that is
            //bigger then maxSize and we have finished a given row key
            if (rollOverRequested && Bytes.compareTo(previousRow, keyFamilyQualifier.rowKey) != 0) {
              rollWriters(fs, writerMap,
                regionSplitPartitioner,
                previousRow,
                compactionExclude)
              rollOverRequested = false
            }

            previousRow = keyFamilyQualifier.rowKey
          }
          //We have finished all the data so lets close up the writers
          rollWriters(fs, writerMap,
            regionSplitPartitioner,
            previousRow,
            compactionExclude)
          rollOverRequested = false
        })
    } finally {
      if(null != conn) conn.close()
    }
  }

  /**
   * Spark Implementation of HBase Bulk load for short rows some where less then
   * a 1000 columns.  This bulk load should be faster for tables will thinner
   * rows then the other spark implementation of bulk load that puts only one
   * value into a record going into a shuffle
   *
   * This will take the content from an existing RDD then sort and shuffle
   * it with respect to region splits.  The result of that sort and shuffle
   * will be written to HFiles.
   *
   * After this function is executed the user will have to call
   * LoadIncrementalHFiles.doBulkLoad(...) to move the files into HBase
   *
   * In this implementation, only the rowKey is given to the shuffle as the key
   * and all the columns are already linked to the RowKey before the shuffle
   * stage.  The sorting of the qualifier is done in memory out side of the
   * shuffle stage
   *
   * Also make sure that incoming RDDs only have one record for every row key.
   *
   * @param rdd                            The RDD we are bulk loading from
   * @param tableName                      The HBase table we are loading into
   * @param mapFunction                    A function that will convert the RDD records to
   *                                       the key value format used for the shuffle to prep
   *                                       for writing to the bulk loaded HFiles
   * @param stagingDir                     The location on the FileSystem to bulk load into
   * @param familyHFileWriteOptionsMap     Options that will define how the HFile for a
   *                                       column family is written
   * @param compactionExclude              Compaction excluded for the HFiles
   * @param maxSize                        Max size for the HFiles before they roll
   * @tparam T                             The Type of values in the original RDD
   */
  def bulkLoadThinRows[T](rdd:RDD[T],
                  tableName: TableName,
                  mapFunction: (T) =>
                    (ByteArrayWrapper, FamiliesQualifiersValues),
                  stagingDir:String,
                  familyHFileWriteOptionsMap:
                  util.Map[Array[Byte], FamilyHFileWriteOptions] =
                  new util.HashMap[Array[Byte], FamilyHFileWriteOptions],
                  compactionExclude: Boolean = false,
                  maxSize:Long = HConstants.DEFAULT_MAX_FILE_SIZE):
  Unit = {
    val stagingPath = new Path(stagingDir)
    val fs = stagingPath.getFileSystem(config)
    if (fs.exists(stagingPath)) {
      throw new FileAlreadyExistsException("Path " + stagingDir + " already exists")
    }
    val conn = HBaseConnectionCache.getConnection(config)
    try {
      val regionLocator = conn.getRegionLocator(tableName)
      val startKeys = regionLocator.getStartKeys
      if (startKeys.length == 0) {
        logInfo("Table " + tableName.toString + " was not found")
      }
      val defaultCompressionStr = config.get("hfile.compression",
        Compression.Algorithm.NONE.getName)
      val defaultCompression = HFileWriterImpl
        .compressionByName(defaultCompressionStr)
      val nowTimeStamp = System.currentTimeMillis()
      val tableRawName = tableName.getName

      val familyHFileWriteOptionsMapInternal =
        new util.HashMap[ByteArrayWrapper, FamilyHFileWriteOptions]

      val entrySetIt = familyHFileWriteOptionsMap.entrySet().iterator()

      while (entrySetIt.hasNext) {
        val entry = entrySetIt.next()
        familyHFileWriteOptionsMapInternal.put(new ByteArrayWrapper(entry.getKey), entry.getValue)
      }

      val regionSplitPartitioner =
        new BulkLoadPartitioner(startKeys)

      //This is where all the magic happens
      //Here we are going to do the following things
      // 1. FlapMap every row in the RDD into key column value tuples
      // 2. Then we are going to repartition sort and shuffle
      // 3. Finally we are going to write out our HFiles
      rdd.map( r => mapFunction(r)).
        repartitionAndSortWithinPartitions(regionSplitPartitioner).
        hbaseForeachPartition(this, (it, conn) => {

          val conf = broadcastedConf.value.value
          val fs = FileSystem.get(conf)
          val writerMap = new mutable.HashMap[ByteArrayWrapper, WriterLength]
          var previousRow:Array[Byte] = HConstants.EMPTY_BYTE_ARRAY
          var rollOverRequested = false
          val localTableName = TableName.valueOf(tableRawName)

          //Here is where we finally iterate through the data in this partition of the
          //RDD that has been sorted and partitioned
          it.foreach{ case (rowKey:ByteArrayWrapper,
          familiesQualifiersValues:FamiliesQualifiersValues) =>


            if (Bytes.compareTo(previousRow, rowKey.value) == 0) {
              throw new KeyAlreadyExistsException("The following key was sent to the " +
                "HFile load more then one: " + Bytes.toString(previousRow))
            }

            //The family map is a tree map so the families will be sorted
            val familyIt = familiesQualifiersValues.familyMap.entrySet().iterator()
            while (familyIt.hasNext) {
              val familyEntry = familyIt.next()

              val family = familyEntry.getKey.value

              val qualifierIt = familyEntry.getValue.entrySet().iterator()

              //The qualifier map is a tree map so the families will be sorted
              while (qualifierIt.hasNext) {

                val qualifierEntry = qualifierIt.next()
                val qualifier = qualifierEntry.getKey
                val cellValue = qualifierEntry.getValue

                writeValueToHFile(rowKey.value,
                  family,
                  qualifier.value, // qualifier
                  cellValue, // value
                  nowTimeStamp,
                  fs,
                  conn,
                  localTableName,
                  conf,
                  familyHFileWriteOptionsMapInternal,
                  defaultCompression,
                  writerMap,
                  stagingDir)

                previousRow = rowKey.value
              }

              writerMap.values.foreach( wl => {
                rollOverRequested = rollOverRequested || wl.written > maxSize

                //This will only roll if we have at least one column family file that is
                //bigger then maxSize and we have finished a given row key
                if (rollOverRequested) {
                  rollWriters(fs, writerMap,
                    regionSplitPartitioner,
                    previousRow,
                    compactionExclude)
                  rollOverRequested = false
                }
              })
            }
          }

          //This will get a writer for the column family
          //If there is no writer for a given column family then
          //it will get created here.
          //We have finished all the data so lets close up the writers
          rollWriters(fs, writerMap,
            regionSplitPartitioner,
            previousRow,
            compactionExclude)
          rollOverRequested = false
        })
    } finally {
      if(null != conn) conn.close()
    }
  }

  /**
   *  This will return a new HFile writer when requested
   *
   * @param family       column family
   * @param conf         configuration to connect to HBase
   * @param favoredNodes nodes that we would like to write too
   * @param fs           FileSystem object where we will be writing the HFiles to
   * @return WriterLength object
   */
  private def getNewHFileWriter(family: Array[Byte], conf: Configuration,
                   favoredNodes: Array[InetSocketAddress],
                   fs:FileSystem,
                   familydir:Path,
                   familyHFileWriteOptionsMapInternal:
                   util.HashMap[ByteArrayWrapper, FamilyHFileWriteOptions],
                   defaultCompression:Compression.Algorithm): WriterLength = {


    var familyOptions = familyHFileWriteOptionsMapInternal.get(new ByteArrayWrapper(family))

    if (familyOptions == null) {
      familyOptions = new FamilyHFileWriteOptions(defaultCompression.toString,
        BloomType.NONE.toString, HConstants.DEFAULT_BLOCKSIZE, DataBlockEncoding.NONE.toString)
      familyHFileWriteOptionsMapInternal.put(new ByteArrayWrapper(family), familyOptions)
    }

    val tempConf = new Configuration(conf)
    tempConf.setFloat(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, 0.0f)
    val contextBuilder = new HFileContextBuilder()
      .withCompression(Algorithm.valueOf(familyOptions.compression))
      .withChecksumType(HStore.getChecksumType(conf))
      .withBytesPerCheckSum(HStore.getBytesPerChecksum(conf))
      .withBlockSize(familyOptions.blockSize)

    if (HFile.getFormatVersion(conf) >= HFile.MIN_FORMAT_VERSION_WITH_TAGS) {
      contextBuilder.withIncludesTags(true)
    }

    contextBuilder.withDataBlockEncoding(DataBlockEncoding.
      valueOf(familyOptions.dataBlockEncoding))
    val hFileContext = contextBuilder.build()

    //Add a '_' to the file name because this is a unfinished file.  A rename will happen
    // to remove the '_' when the file is closed.
    new WriterLength(0,
      new StoreFileWriter.Builder(conf, new CacheConfig(tempConf), new HFileSystem(fs))
        .withBloomType(BloomType.valueOf(familyOptions.bloomType))
        .withComparator(CellComparator.getInstance()).withFileContext(hFileContext)
        .withFilePath(new Path(familydir, "_" + UUID.randomUUID.toString.replaceAll("-", "")))
        .withFavoredNodes(favoredNodes).build())

  }

  /**
   * Encompasses the logic to write a value to an HFile
   *
   * @param rowKey                             The RowKey for the record
   * @param family                             HBase column family for the record
   * @param qualifier                          HBase column qualifier for the record
   * @param cellValue                          HBase cell value
   * @param nowTimeStamp                       The cell time stamp
   * @param fs                                 Connection to the FileSystem for the HFile
   * @param conn                               Connection to HBaes
   * @param tableName                          HBase TableName object
   * @param conf                               Configuration to be used when making a new HFile
   * @param familyHFileWriteOptionsMapInternal Extra configs for the HFile
   * @param hfileCompression                   The compression codec for the new HFile
   * @param writerMap                          HashMap of existing writers and their offsets
   * @param stagingDir                         The staging directory on the FileSystem to store
   *                                           the HFiles
   * @return                                   The writer for the given HFile that was writen
   *                                           too
   */
  private def writeValueToHFile(rowKey: Array[Byte],
                        family: Array[Byte],
                        qualifier: Array[Byte],
                        cellValue:Array[Byte],
                        nowTimeStamp: Long,
                        fs: FileSystem,
                        conn: Connection,
                        tableName: TableName,
                        conf: Configuration,
                        familyHFileWriteOptionsMapInternal:
                        util.HashMap[ByteArrayWrapper, FamilyHFileWriteOptions],
                        hfileCompression:Compression.Algorithm,
                        writerMap:mutable.HashMap[ByteArrayWrapper, WriterLength],
                        stagingDir: String
                         ): WriterLength = {

    val wl = writerMap.getOrElseUpdate(new ByteArrayWrapper(family), {
      val familyDir = new Path(stagingDir, Bytes.toString(family))

      fs.mkdirs(familyDir)

      val loc:HRegionLocation = {
        try {
          val locator =
            conn.getRegionLocator(tableName)
          locator.getRegionLocation(rowKey)
        } catch {
          case e: Throwable =>
            logWarning("there's something wrong when locating rowkey: " +
              Bytes.toString(rowKey))
            null
        }
      }
      if (null == loc) {
        if (log.isTraceEnabled) {
          logTrace("failed to get region location, so use default writer: " +
            Bytes.toString(rowKey))
        }
        getNewHFileWriter(family = family,
          conf = conf,
          favoredNodes = null,
          fs = fs,
          familydir = familyDir,
          familyHFileWriteOptionsMapInternal,
          hfileCompression)
      } else {
        if (log.isDebugEnabled) {
          logDebug("first rowkey: [" + Bytes.toString(rowKey) + "]")
        }
        val initialIsa =
          new InetSocketAddress(loc.getHostname, loc.getPort)
        if (initialIsa.isUnresolved) {
          if (log.isTraceEnabled) {
            logTrace("failed to resolve bind address: " + loc.getHostname + ":"
              + loc.getPort + ", so use default writer")
          }
          getNewHFileWriter(family,
            conf,
            null,
            fs,
            familyDir,
            familyHFileWriteOptionsMapInternal,
            hfileCompression)
        } else {
          if(log.isDebugEnabled) {
            logDebug("use favored nodes writer: " + initialIsa.getHostString)
          }
          getNewHFileWriter(family,
            conf,
            Array[InetSocketAddress](initialIsa),
            fs,
            familyDir,
            familyHFileWriteOptionsMapInternal,
            hfileCompression)
        }
      }
    })

    val keyValue =new KeyValue(rowKey,
      family,
      qualifier,
      nowTimeStamp,cellValue)

    wl.writer.append(keyValue)
    wl.written += keyValue.getLength

    wl
  }

  /**
   * This will roll all Writers
   * @param fs                     Hadoop FileSystem object
   * @param writerMap              HashMap that contains all the writers
   * @param regionSplitPartitioner The partitioner with knowledge of how the
   *                               Region's are split by row key
   * @param previousRow            The last row to fill the HFile ending range metadata
   * @param compactionExclude      The exclude compaction metadata flag for the HFile
   */
  private def rollWriters(fs:FileSystem,
                          writerMap:mutable.HashMap[ByteArrayWrapper, WriterLength],
                  regionSplitPartitioner: BulkLoadPartitioner,
                  previousRow: Array[Byte],
                  compactionExclude: Boolean): Unit = {
    writerMap.values.foreach( wl => {
      if (wl.writer != null) {
        logDebug("Writer=" + wl.writer.getPath +
          (if (wl.written == 0) "" else ", wrote=" + wl.written))
        closeHFileWriter(fs, wl.writer,
          regionSplitPartitioner,
          previousRow,
          compactionExclude)
      }
    })
    writerMap.clear()

  }

  /**
   * Function to close an HFile
   * @param fs                     Hadoop FileSystem object
   * @param w                      HFile Writer
   * @param regionSplitPartitioner The partitioner with knowledge of how the
   *                               Region's are split by row key
   * @param previousRow            The last row to fill the HFile ending range metadata
   * @param compactionExclude      The exclude compaction metadata flag for the HFile
   */
  private def closeHFileWriter(fs:FileSystem,
                               w: StoreFileWriter,
                               regionSplitPartitioner: BulkLoadPartitioner,
                               previousRow: Array[Byte],
                               compactionExclude: Boolean): Unit = {
    if (w != null) {
      w.appendFileInfo(HStoreFile.BULKLOAD_TIME_KEY,
        Bytes.toBytes(System.currentTimeMillis()))
      w.appendFileInfo(HStoreFile.BULKLOAD_TASK_KEY,
        Bytes.toBytes(regionSplitPartitioner.getPartition(previousRow)))
      w.appendFileInfo(HStoreFile.MAJOR_COMPACTION_KEY,
        Bytes.toBytes(true))
      w.appendFileInfo(HStoreFile.EXCLUDE_FROM_MINOR_COMPACTION_KEY,
        Bytes.toBytes(compactionExclude))
      w.appendTrackedTimestampsToMetadata()
      w.close()

      val srcPath = w.getPath

      //In the new path you will see that we are using substring.  This is to
      // remove the '_' character in front of the HFile name.  '_' is a character
      // that will tell HBase that this file shouldn't be included in the bulk load
      // This feature is to protect for unfinished HFiles being submitted to HBase
      val newPath = new Path(w.getPath.getParent, w.getPath.getName.substring(1))
      if (!fs.rename(srcPath, newPath)) {
        throw new IOException("Unable to rename '" + srcPath +
          "' to " + newPath)
      }
    }
  }

  /**
   * This is a wrapper class around StoreFileWriter.  The reason for the
   * wrapper is to keep the length of the file along side the writer
   *
   * @param written The writer to be wrapped
   * @param writer  The number of bytes written to the writer
   */
  class WriterLength(var written:Long, val writer:StoreFileWriter)
}

@InterfaceAudience.Private
object LatestHBaseContextCache {
  var latest:HBaseContext = null
}