/*
 * 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.carbondata.examples

import java.io.File
import java.sql.{DriverManager, ResultSet, Statement}

import org.apache.hadoop.fs.Path
import org.apache.hadoop.fs.permission.{FsAction, FsPermission}

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.examples.util.ExampleUtils
import org.apache.carbondata.hive.test.server.HiveEmbeddedServer2

// scalastyle:off println
object HiveExample {

  private val driverName: String = "org.apache.hive.jdbc.HiveDriver"

  val rootPath = new File(this.getClass.getResource("/").getPath
                          + "../../../..").getCanonicalPath
  private val targetLoc = s"$rootPath/examples/spark2/target"
  val metaStoreLoc = s"$targetLoc/metastore_db"
  val storeLocation = s"$targetLoc/store"
  val logger = LogServiceFactory.getLogService(this.getClass.getCanonicalName)


  def main(args: Array[String]) {
    createCarbonTable(storeLocation)
    readFromHive
    System.exit(0)
  }

  def createCarbonTable(store: String): Unit = {

    val carbonSession = ExampleUtils.createCarbonSession("HiveExample")

    carbonSession.sql("""DROP TABLE IF EXISTS HIVE_CARBON_EXAMPLE""".stripMargin)

    carbonSession.sql(
      s"""
         | CREATE TABLE HIVE_CARBON_EXAMPLE
         | (ID int,NAME string,SALARY double)
         | STORED BY 'carbondata'
       """.stripMargin)

    var inputPath = FileFactory
      .getUpdatedFilePath(s"$rootPath/examples/spark2/src/main/resources/sample.csv")

    carbonSession.sql(
      s"""
         | LOAD DATA LOCAL INPATH '$inputPath'
         | INTO TABLE HIVE_CARBON_EXAMPLE
       """.stripMargin)

    carbonSession.sql(
      s"""
         | LOAD DATA LOCAL INPATH '$inputPath'
         | INTO TABLE HIVE_CARBON_EXAMPLE
       """.stripMargin)

    carbonSession.sql("SELECT * FROM HIVE_CARBON_EXAMPLE").show()

    carbonSession.sql("DROP TABLE IF EXISTS TEST_BOUNDARY")

    carbonSession
      .sql(
        s"""CREATE TABLE TEST_BOUNDARY (c1_int int,c2_Bigint Bigint,c3_Decimal Decimal(38,30),
           |c4_double double,c5_string string,c6_Timestamp Timestamp,c7_Datatype_Desc string)
           |STORED BY 'org.apache.carbondata.format' """.stripMargin)

    inputPath = FileFactory
      .getUpdatedFilePath(s"$rootPath/examples/spark2/src/main/resources/Test_Data1.csv")

    carbonSession
      .sql(
        s"LOAD DATA INPATH '$inputPath' INTO table TEST_BOUNDARY OPTIONS('DELIMITER'=','," +
        "'QUOTECHAR'='\"', 'BAD_RECORDS_ACTION'='FORCE','FILEHEADER'='c1_int,c2_Bigint," +
        "c3_Decimal,c4_double,c5_string,c6_Timestamp,c7_Datatype_Desc')")

    carbonSession.sql("""DROP TABLE IF EXISTS complexMap""".stripMargin)

    carbonSession.sql("create table complexMap(name map<string,string>) stored by 'carbondata'")

    carbonSession
      .sql(
        "insert into complexMap values(map('Manish','Nalla','Shardul','Singh','Vishal','Kumar'," +
        "'EmptyVal','','NullVal', 'null'))")

    carbonSession.close()

    // delete the already existing lock on metastore so that new derby instance
    // for HiveServer can run on the same metastore
    checkAndDeleteDBLock

  }

  def checkAndDeleteDBLock: Unit = {
    val dbLockPath = FileFactory.getUpdatedFilePath(s"$metaStoreLoc/db.lck")
    val dbexLockPath = FileFactory.getUpdatedFilePath(s"$metaStoreLoc/dbex.lck")
    if(FileFactory.isFileExist(dbLockPath)) {
      FileFactory.deleteFile(dbLockPath)
    }
    if(FileFactory.isFileExist(dbexLockPath)) {
      FileFactory.deleteFile(dbexLockPath)
    }
  }


  def readFromHive: Unit = {
    try {
      Class.forName(driverName)
    }
    catch {
      case classNotFoundException: ClassNotFoundException =>
        classNotFoundException.printStackTrace()
    }

    // make HDFS writable
    val path = new Path(targetLoc)
    val fileSys = path.getFileSystem(FileFactory.getConfiguration)
    fileSys.setPermission(path, new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL))

    val hiveEmbeddedServer2 = new HiveEmbeddedServer2()
    hiveEmbeddedServer2.start(targetLoc)
    val port = hiveEmbeddedServer2.getFreePort
    val connection = DriverManager.getConnection(s"jdbc:hive2://localhost:$port/default", "", "")
    val statement: Statement = connection.createStatement

    logger.info(s"============HIVE CLI IS STARTED ON PORT $port ==============")

    val resultSet: ResultSet = statement.executeQuery("SELECT * FROM HIVE_CARBON_EXAMPLE")

    var rowsFetched = 0
    var resultId = ""
    var resultName = ""
    var resultSalary = ""

    while (resultSet.next) {
      if (rowsFetched == 0) {
        println("+---+" + "+-------+" + "+--------------+")
        println("| ID|" + "| NAME |" + "| SALARY        |")

        println("+---+" + "+-------+" + "+--------------+")

        resultId = resultSet.getString("id")
        resultName = resultSet.getString("name")
        resultSalary = resultSet.getString("salary")

        println(s"| $resultId |" + s"| $resultName |" + s"| $resultSalary  |")
        println("+---+" + "+-------+" + "+--------------+")
      }
      else {
        resultId = resultSet.getString("ID")
        resultName = resultSet.getString("NAME")
        resultSalary = resultSet.getString("SALARY")

        println(s"| $resultId |" + s"| $resultName |" + s"| $resultSalary   |")
        println("+---+" + "+-------+" + "+--------------+")
      }
      rowsFetched = rowsFetched + 1
    }
    println(s"******Total Number Of Rows Fetched ****** $rowsFetched")
    assert(rowsFetched == 4)

    logger.info("Fetching the Individual Columns ")

    // fetching the separate columns
    var individualColRowsFetched = 0

    val resultIndividualCol = statement.executeQuery("SELECT NAME FROM HIVE_CARBON_EXAMPLE")

    while (resultIndividualCol.next) {
      if (individualColRowsFetched == 0) {
        println("+--------------+")
        println("| NAME         |")

        println("+---++---------+")

        resultName = resultIndividualCol.getString("name")

        println(s"| $resultName    |")
        println("+---+" + "+---------+")
      }
      else {
        resultName = resultIndividualCol.getString("NAME")

        println(s"| $resultName      |")
        println("+---+" + "+---------+")
      }
      individualColRowsFetched = individualColRowsFetched + 1
    }
    println(" ********** Total Rows Fetched When Quering The Individual Columns **********" +
      s"$individualColRowsFetched")
    assert(individualColRowsFetched == 4)

    logger.info("Fetching the Out Of Order Columns ")

    val resultOutOfOrderCol = statement
      .executeQuery("SELECT SALARY,ID,NAME FROM HIVE_CARBON_EXAMPLE")
    var outOfOrderColFetched = 0

    while (resultOutOfOrderCol.next()) {
      if (outOfOrderColFetched == 0) {
        println("+---+" + "+-------+" + "+--------------+")
        println("| Salary|" + "| ID |" + "| NAME        |")

        println("+---+" + "+-------+" + "+--------------+")

        resultId = resultOutOfOrderCol.getString("id")
        resultName = resultOutOfOrderCol.getString("name")
        resultSalary = resultOutOfOrderCol.getString("salary")

        println(s"| $resultSalary |" + s"| $resultId |" + s"| $resultName  |")
        println("+---+" + "+-------+" + "+--------------+")
      }
      else {
        resultId = resultOutOfOrderCol.getString("ID")
        resultName = resultOutOfOrderCol.getString("NAME")
        resultSalary = resultOutOfOrderCol.getString("SALARY")

        println(s"| $resultSalary |" + s"| $resultId |" + s"| $resultName   |")
        println("+---+" + "+-------+" + "+--------------+")
      }
      outOfOrderColFetched = outOfOrderColFetched + 1
    }
    println(" ********** Total Rows Fetched When Quering The Out Of Order Columns **********" +
      s"$outOfOrderColFetched")
    assert(outOfOrderColFetched == 4)

    val resultAggQuery = statement
      .executeQuery(
        "SELECT min(c3_Decimal) as min, max(c3_Decimal) as max, " +
        "sum(c3_Decimal) as sum FROM TEST_BOUNDARY")

    var resultAggQueryFetched = 0

    var resultMin = ""
    var resultMax = ""
    var resultSum = ""

    while (resultAggQuery.next) {
      if (resultAggQueryFetched == 0) {
        println("+-----+" + "+-------------------+" + "+--------------------------------+")
        println("| min |" + "| max               |" + "| sum                            |")

        println("+-----+" + "+-------------------+" + "+--------------------------------+")

        resultMin = resultAggQuery.getString("min")
        resultMax = resultAggQuery.getString("max")
        resultSum = resultAggQuery.getString("sum")

        println(s"| $resultMin   |" + s"| $resultMax               |" + s"| $resultSum|")
        println("+-----+" + "+-------------------+" + "+--------------------------------+")
      }
      resultAggQueryFetched = resultAggQueryFetched + 1
    }
    println(" ********** Total Rows Fetched When Aggregate Query **********" +
            s"$resultAggQueryFetched")
    assert(resultAggQueryFetched == 1)

    val resultComplexQuery = statement
      .executeQuery(
        "SELECT name FROM complexMap")

    var resultComplex = 0

    var name = ""

    while (resultComplexQuery.next) {
      if (resultComplex == 0) {
        println("+------------------------------------------------------------------------------" +
                "------+")
        println("| name                                                                          " +
                "     |")

        println("+-------------------------------------------------------------------------------" +
                "-----+")

        name = resultComplexQuery.getString("name")

        println(s"|$name|")
        println("+-------------------------------------------------------------------------------" +
                "-----+")      }
      resultComplex = resultComplex + 1
    }
    println(" ********** Total Rows Fetched When Complex Query **********" +
            s"$resultComplex")
    assert(resultComplex == 1)

    hiveEmbeddedServer2.stop()
  }
}