/*
 * 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.spark.testsuite.createTable

import java.io.File
import java.util

import org.apache.avro
import org.apache.commons.io.FileUtils
import org.apache.spark.sql.Row
import org.apache.spark.sql.test.util.QueryTest
import org.junit.Assert
import org.scalatest.BeforeAndAfterAll

import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.util.{CarbonProperties, CarbonTestUtil}
import org.apache.carbondata.sdk.file.CarbonWriter

class TestNonTransactionalCarbonTableWithComplexType extends QueryTest with BeforeAndAfterAll {

  var writerPath = new File(this.getClass.getResource("/").getPath +
                            "../../target/SparkCarbonFileFormat/WriterOutput/").getCanonicalPath
  // getCanonicalPath gives path with \, but the code expects /.
  writerPath = writerPath.replace("\\", "/")

  private def cleanTestData() = {
    FileUtils.deleteDirectory(new File(writerPath))
  }

  override def beforeAll(): Unit = {
    CarbonProperties.getInstance()
      .addProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
        CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT)
    sql("DROP TABLE IF EXISTS sdkOutputTable")
  }

  override def afterAll(): Unit = {
    sql("DROP TABLE IF EXISTS sdkOutputTable")
  }

  private def WriteFilesWithAvroWriter(rows: Int,
      mySchema: String,
      json: String,
      isLocalDictionary: Boolean): Unit = {
    // conversion to GenericData.Record
    val nn = new avro.Schema.Parser().parse(mySchema)
    val record = testUtil.jsonToAvro(json, mySchema)
    try {
      val writer = if (isLocalDictionary) {
        CarbonWriter.builder
          .outputPath(writerPath).enableLocalDictionary(true)
          .localDictionaryThreshold(2000)
          .uniqueIdentifier(System.currentTimeMillis())
          .withAvroInput(nn)
          .writtenBy("TestNonTransactionalCarbonTableWithComplexType")
          .build()
      } else {
        CarbonWriter.builder
          .outputPath(writerPath)
          .uniqueIdentifier(System.currentTimeMillis())
          .withAvroInput(nn)
          .writtenBy("TestNonTransactionalCarbonTableWithComplexType")
          .build()
      }
      var i = 0
      while (i < rows) {
        writer.write(record)
        i = i + 1
      }
      writer.close()
    }
    catch {
      case e: Exception =>
        e.printStackTrace()
        Assert.fail(e.getMessage)
    }
  }

  // test multi level -- 4 levels [array of array of array of struct]
  def buildAvroTestDataMultiLevel4(rows: Int,
      options: util.Map[String, String],
      isLocalDictionary: Boolean): Any = {
    FileUtils.deleteDirectory(new File(writerPath))

    val mySchema = """ {
                      |  "name": "address",
                      |  "type": "record",
                      |  "fields": [
                      |    {
                      |      "name": "name",
                      |      "type": "string"
                      |    },
                      |    {
                      |      "name": "age",
                      |      "type": "int"
                      |    },
                      |    {
                      |      "name": "BuildNum",
                      |      "type": {
                      |        "type": "array",
                      |        "items": {
                      |          "name": "FloorNum",
                      |          "type": "array",
                      |          "items": {
                      |            "name": "doorNum",
                      |            "type": "array",
                      |            "items": {
                      |              "name": "my_address",
                      |              "type": "record",
                      |              "fields": [
                      |                {
                      |                  "name": "street",
                      |                  "type": "string"
                      |                },
                      |                {
                      |                  "name": "city",
                      |                  "type": "string"
                      |                },
                      |               {
                      |                  "name": "Temperature",
                      |                  "type": "double"
                      |                },
                      |               {
                      |                  "name": "WindSpeed",
                      |                  "type": "string"
                      |                },
                      |               {
                      |                  "name": "year",
                      |                  "type": "string"
                      |                }
                      |              ]
                      |            }
                      |          }
                      |        }
                      |      }
                      |    }
                      |  ]
                      |} """.stripMargin

    val json =
      """ {
        |  "name": "bob",
        |  "age": 10,
        |  "BuildNum": [
        |    [
        |      [
        |        {"street":"abc", "city":"city1", "Temperature":12.6,
        |         "WindSpeed":"1234.56", "year":"2018-05-10"},
        |        {"street":"def", "city":"city2", "Temperature":13.6,
        |         "WindSpeed":"1234.56", "year":"2018-05-10"},
        |        {"street":"cfg", "city":"city3", "Temperature":14.6,
        |         "WindSpeed":"1234.56", "year":"2018-05-10"}
        |      ],
        |      [
        |         {"street":"abc1", "city":"city3", "Temperature":12.6,
        |          "WindSpeed":"1234.56", "year":"2018-05-10"},
        |         {"street":"def1", "city":"city4", "Temperature":12.6,
        |          "WindSpeed":"1234.56", "year":"2018-05-10"},
        |         {"street":"cfg1", "city":"city5", "Temperature":12.6,
        |          "WindSpeed":"1234.56", "year":"2018-05-10"}
        |      ]
        |    ],
        |    [
        |      [
        |         {"street":"abc2", "city":"cityx", "Temperature":12.6,
        |          "WindSpeed":"1234.56", "year":"2018-05-10"},
        |         {"street":"abc3", "city":"cityy", "Temperature":12.6,
        |          "WindSpeed":"1234.56", "year":"2018-05-10"},
        |         {"street":"abc4", "city":"cityz", "Temperature":12.6,
        |          "WindSpeed":"1234.56", "year":"2018-05-10"}
        |      ],
        |      [
        |         {"street":"a1bc", "city":"cityA", "Temperature":12.6,
        |          "WindSpeed":"1234.56", "year":"2018-05-10"},
        |         {"street":"a1bc", "city":"cityB", "Temperature":12.6,
        |          "WindSpeed":"1234.56", "year":"2018-05-10"},
        |         {"street":"a1bc", "city":"cityc", "Temperature":12.6,
        |          "WindSpeed":"1234.56", "year":"2018-05-10"}
        |      ]
        |    ]
        |  ]
        |} """.stripMargin

    WriteFilesWithAvroWriter(rows, mySchema, json, isLocalDictionary)
  }

  def buildAvroTestDataMultiLevel4Type(isLocalDictionary: Boolean): Any = {
    FileUtils.deleteDirectory(new File(writerPath))
    buildAvroTestDataMultiLevel4(3, null, isLocalDictionary)
  }

  // test multi level -- 4 levels [array of array of array of struct]
  test("test multi level support : array of array of array of struct") {
    buildAvroTestDataMultiLevel4Type(false)
    assert(new File(writerPath).exists())
    sql("DROP TABLE IF EXISTS sdkOutputTable")
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED AS carbondata LOCATION
         |'$writerPath' """.stripMargin)

    sql("select * from sdkOutputTable").collect()

    // TODO: Add a validation

    sql("DROP TABLE sdkOutputTable")
    // drop table should not delete the files
    cleanTestData()
  }

  test("test local dictionary for complex datatype") {
    buildAvroTestDataMultiLevel4Type(true)
    assert(new File(writerPath).exists())
    sql("DROP TABLE IF EXISTS localComplex")
    sql(
      s"""CREATE EXTERNAL TABLE localComplex STORED AS carbondata LOCATION
         |'$writerPath' """.stripMargin)
    assert(FileFactory.getCarbonFile(writerPath).exists())
    assert(CarbonTestUtil.checkForLocalDictionary(CarbonTestUtil.getDimRawChunk(writerPath, 0)))
    sql("describe formatted localComplex").collect
    val descLoc = sql("describe formatted localComplex").collect
    descLoc.find(_.get(0).toString.contains("Local Dictionary Enabled")) match {
      case Some(row) => assert(row.get(1).toString.contains("true"))
      case None => assert(false)
    }

    // TODO: Add a validation

    sql("DROP TABLE localComplex")
    // drop table should not delete the files
    cleanTestData()
  }

  test("test multi level support : array of array of array of with Double data type") {
    cleanTestData()
    val mySchema = """ {
                      |  "name": "address",
                      |  "type": "record",
                      |  "fields": [
                      |    {
                      |      "name": "name",
                      |      "type": "string"
                      |    },
                      |    {
                      |      "name": "age",
                      |      "type": "int"
                      |    },
                      |    {
                      |   "name" :"my_address",
                      |   "type" :{
                      |              "name": "my_address",
                      |              "type": "record",
                      |              "fields": [
                      |               {
                      |                  "name": "Temperaturetest",
                      |                  "type": "double"
                      |                }
                      |              ]
                      |       }
                      |      }
                      |  ]
                      |} """.stripMargin

    val jsonvalue =
      """{
        |"name" :"babu",
        |"age" :12,
        |"my_address" :{ "Temperaturetest" :123 }
        |}
      """.stripMargin
    val pschema = org.apache.avro.Schema.parse(mySchema)
    val records = testUtil.jsonToAvro(jsonvalue, mySchema)
    val writer = CarbonWriter.builder()
      .outputPath(writerPath)
      .withAvroInput(pschema)
      .writtenBy("TestNonTransactionalCarbonTableWithComplexType")
      .build()
    writer.write(records)
    writer.close()
    sql("DROP TABLE IF EXISTS sdkOutputTable")
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED AS carbondata LOCATION
         |'$writerPath' """.stripMargin)

    sql("select * from sdkOutputTable").collect()

    // TODO: Add a validation

    sql("DROP TABLE sdkOutputTable")
    // drop table should not delete the files
    cleanTestData()
  }

  // test multi level -- 4 levels [array of array of array of struct]
  test("test ComplexDataType projection for array of array of array of struct") {
    buildAvroTestDataMultiLevel4Type(false)
    assert(new File(writerPath).exists())
    sql("DROP TABLE IF EXISTS sdkOutputTable")
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED AS carbondata LOCATION
         |'$writerPath' """.stripMargin)

    checkAnswer(sql("select BuildNum[0][0][0].street from sdkOutputTable"),
      Seq(Row("abc"), Row("abc"), Row("abc")))
    checkAnswer(sql("select BuildNum[1][0][0].street from sdkOutputTable"),
      Seq(Row("abc2"), Row("abc2"), Row("abc2")))

    sql("DROP TABLE sdkOutputTable")
    // drop table should not delete the files
    cleanTestData()
  }

  def buildAvroTestDataMultiLevel6Type(isLocalDictionary: Boolean): Any = {
    FileUtils.deleteDirectory(new File(writerPath))
    buildAvroTestDataMultiLevel6(1, null, isLocalDictionary)
  }

  // test multi level -- 6 levels
  def buildAvroTestDataMultiLevel6(rows: Int,
      options: util.Map[String, String],
      isLocalDictionary: Boolean): Any = {
    FileUtils.deleteDirectory(new File(writerPath))

    val mySchema =
      """ {
        |"type": "record",
        |  "name": "UserInfo",
        |  "namespace": "com.apache.schema.schemalevel6_struct",
        |  "fields": [
        |    {
        |      "name": "username",
        |      "type": "string",
        |      "default": "NONE"
        |    },
        |    {
        |      "name": "age",
        |      "type": "int",
        |      "default": -1
        |    },
        |    {
        |      "name": "phone",
        |      "type": "string",
        |      "default": "NONE"
        |    },
        |    {
        |      "name": "housenum",
        |      "type": "string",
        |      "default": "NONE"
        |    },
        |    {
        |      "name": "address",
        |      "type": {
        |        "type": "record",
        |        "name": "Mailing_Address",
        |        "fields": [
        |          {
        |            "name": "Address_Detail",
        |            "type": {
        |              "type": "record",
        |              "name": "Address_Detail",
        |              "fields": [
        |                {
        |                  "name": "Building_Detail",
        |                  "type": {
        |                    "type": "record",
        |                    "name": "Building_Address",
        |                    "fields": [
        |                      {
        |                        "name": "Society_name",
        |                        "type": "string"
        |                      },
        |                      {
        |                        "name": "building_no",
        |                        "type": "string"
        |                      },
        |                      {
        |                        "name": "house_no",
        |                        "type": "int"
        |                      },
        |                      {
        |                        "name": "Building_Type",
        |                        "type": {
        |                          "type": "record",
        |                          "name": "Building_Type",
        |                          "fields": [
        |                            {
        |                              "name":"Buildingname",
        |                              "type":"string"
        |                            },
        |                            {
        |                              "name":"buildingArea",
        |                              "type":"int"
        |                            },
        |                            {
        |                              "name":"Building_Criteria",
        |                              "type":{
        |                                "type":"record",
        |                                "name":"BuildDet",
        |                                "fields":[
        |                                  {
        |                                    "name":"f1",
        |                                    "type":"int"
        |                                  },
        |                                  {
        |                                    "name":"f2",
        |                                    "type":"string"
        |                                  },
        |                                  {
        |                                    "name":"BuildDetInner",
        |                                    "type":
        |                                      {
        |                                        "type":"record",
        |                                        "name":"BuildInner",
        |                                        "fields":[
        |                                            {
        |                                              "name": "duplex",
        |                                              "type": "boolean"
        |                                            },
        |                                            {
        |                                              "name": "Price",
        |                                              "type": "int"
        |                                            },
        |                                            {
        |                                              "name": "TotalCost",
        |                                              "type": "int"
        |                                            },
        |                                            {
        |                                              "name": "Floor",
        |                                              "type": "int"
        |                                            },
        |                                            {
        |                                              "name": "PhoneNo",
        |                                              "type": "long"
        |                                            },
        |                                            {
        |                                              "name": "value",
        |                                              "type": "string"
        |                                            }
        |                                        ]
        |                                      }
        |                                  }
        |                                ]
        |                              }
        |                            }
        |                          ]
        |                        }
        |                      }
        |                    ]
        |                  }
        |                }
        |              ]
        |            }
        |          }
        |        ]
        |      }
        |    }
        |  ]
        |} """.stripMargin

    val json =
      """ {
        |"username": "DON",
        |"age": 21,
        |"phone": "9888",
        |"housenum": "44",
        |"address": {
        |"Address_Detail": {
        |"Building_Detail": {
        |"Society_name": "TTTT",
        |"building_no": "5",
        |"house_no": 78,
        |"Building_Type": {
        |"Buildingname": "Amaranthus",
        |"buildingArea": 34,
        |"Building_Criteria": {
        |"f1": 23,
        |"f2": "RRR",
        |"BuildDetInner": {
        |"duplex": true,
        |"Price": 3434,
        |"TotalCost": 7777,
        |"Floor": 4,
        |"PhoneNo": 5656,
        |"value":"Value"
        |}
        |}
        |}
        |}
        |}
        |}
        |} """.stripMargin

    WriteFilesWithAvroWriter(rows, mySchema, json, isLocalDictionary)
  }


  test("test ComplexDataType projection for struct of struct -6 levels") {
    buildAvroTestDataMultiLevel6Type(false)
    assert(new File(writerPath).exists())
    sql("DROP TABLE IF EXISTS sdkOutputTable")
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED AS carbondata LOCATION
         |'$writerPath' """.stripMargin)

    // scalastyle:off lineLength
    checkAnswer(sql("select * from sdkOutputTable"),
      Seq(Row("DON", 21, "9888", "44", Row(Row(Row("TTTT", "5", 78, Row("Amaranthus", 34,
        Row(23, "RRR", Row(true, 3434, 7777, 4, 5656, "Value")))))))))
    checkAnswer(sql("select address from sdkOutputTable"),
      Seq(Row(Row(Row(Row("TTTT",
        "5",
        78,
        Row("Amaranthus", 34, Row(23, "RRR", Row(true, 3434, 7777, 4, 5656, "Value")))))))))
    checkAnswer(sql("select address.Address_Detail from sdkOutputTable"),
      Seq(Row(Row(Row("TTTT",
        "5",
        78,
        Row("Amaranthus", 34, Row(23, "RRR", Row(true, 3434, 7777, 4, 5656, "Value"))))))))
    checkAnswer(sql("select address.Address_Detail.Building_Detail from sdkOutputTable"),
      Seq(Row(Row("TTTT",
        "5",
        78,
        Row("Amaranthus", 34, Row(23, "RRR", Row(true, 3434, 7777, 4, 5656, "Value")))))))
    checkAnswer(sql(
      "select address.Address_Detail.Building_Detail.Building_Type from sdkOutputTable"),
      Seq(Row(Row("Amaranthus", 34, Row(23, "RRR", Row(true, 3434, 7777, 4, 5656, "Value"))))))
    checkAnswer(sql(
      "select address.Address_Detail.Building_Detail.Building_Type.Building_Criteria from " +
      "sdkOutputTable"), Seq(Row(Row(23, "RRR", Row(true, 3434, 7777, 4, 5656, "Value")))))
    checkAnswer(sql(
      "select address.Address_Detail.Building_Detail.Building_Type.Building_Criteria" +
      ".BuildDetInner.duplex from sdkOutputTable"), Seq(Row(true)))
    checkAnswer(sql(
      "select address.Address_Detail.Building_Detail.Building_Type.Building_Criteria" +
      ".BuildDetInner.price from sdkOutputTable"), Seq(Row(3434)))
    checkAnswer(sql(
      "select address.Address_Detail.Building_Detail.Building_Type.Building_Criteria" +
      ".BuildDetInner.totalcost from sdkOutputTable"), Seq(Row(7777)))
    checkAnswer(sql(
      "select address.Address_Detail.Building_Detail.Building_Type.Building_Criteria" +
      ".BuildDetInner.floor from sdkOutputTable"), Seq(Row(4)))
    checkAnswer(sql(
      "select address.Address_Detail.Building_Detail.Building_Type.Building_Criteria" +
      ".BuildDetInner.phoneNo from sdkOutputTable"), Seq(Row(5656)))
    checkAnswer(sql(
      "select address.Address_Detail.Building_Detail.Building_Type.Building_Criteria" +
      ".BuildDetInner.value from sdkOutputTable"), Seq(Row("Value")))
    checkAnswer(sql("select address,address.Address_Detail from sdkOutputTable"),
      Seq(Row(Row(Row(Row("TTTT",
        "5",
        78,
        Row("Amaranthus", 34, Row(23, "RRR", Row(true, 3434, 7777, 4, 5656, "Value"))))))
        ,
        Row(Row("TTTT",
          "5",
          78,
          Row("Amaranthus", 34, Row(23, "RRR", Row(true, 3434, 7777, 4, 5656, "Value"))))))))
    checkAnswer(sql(
      "select address.Address_Detail.Building_Detail.Building_Type,address.Address_Detail" +
      ".Building_Detail.Building_Type.Building_Criteria from sdkOutputTable"),
      Seq(Row(Row("Amaranthus", 34, Row(23, "RRR", Row(true, 3434, 7777, 4, 5656, "Value"))),
        Row(23, "RRR", Row(true, 3434, 7777, 4, 5656, "Value")))))
    checkAnswer(sql(
      "select address.Address_Detail,address.Address_Detail.Building_Detail.Building_Type" +
      ".Building_Criteria from sdkOutputTable"),
      Seq(Row(Row(Row("TTTT",
        "5",
        78,
        Row("Amaranthus", 34, Row(23, "RRR", Row(true, 3434, 7777, 4, 5656, "Value"))))),
        Row(23, "RRR", Row(true, 3434, 7777, 4, 5656, "Value")))))
    checkAnswer(sql(
      "select address.Address_Detail,address.Address_Detail.Building_Detail.Society_name,address" +
      ".Address_Detail.Building_Detail.Building_Type.Building_Criteria.f1 from sdkOutputTable"),
      Seq(Row(Row(Row("TTTT",
        "5",
        78,
        Row("Amaranthus", 34, Row(23, "RRR", Row(true, 3434, 7777, 4, 5656, "Value"))))),
        "TTTT",
        23)))
    checkAnswer(sql(
      "select address.Address_Detail.Building_Detail.Society_name,address.Address_Detail" +
      ".Building_Detail.building_no from sdkOutputTable"),
      Seq(Row("TTTT", "5")))
    sql("select address.Address_Detail.Building_Detail.Society_name," +
        "address.Address_Detail.Building_Detail.building_no from sdkOutputTable " +
        "where address.Address_Detail.Building_Detail.Society_name ='TTTT'").collect()
    sql("DROP TABLE sdkOutputTable")
    // scalastyle:on lineLength
    cleanTestData()
  }
}
