/*
 * 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.nio.ByteBuffer
import javax.xml.bind.DatatypeConverter

import scala.collection.mutable

import org.apache.avro.Conversions.DecimalConversion
import org.apache.avro.{LogicalTypes, Schema}
import org.apache.avro.generic.GenericData
import org.apache.spark.sql.Row
import org.apache.spark.sql.test.util.QueryTest
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
import org.apache.carbondata.sdk.file.CarbonWriter

/**
 * Test class for Avro supported data types through SDK
 */
class TestNonTransactionalCarbonTableWithAvroDataType extends QueryTest with BeforeAndAfterAll {


  val badRecordAction = CarbonProperties.getInstance()
    .getProperty(CarbonCommonConstants.CARBON_BAD_RECORDS_ACTION)

  var writerPath = new File(this.getClass.getResource("/").getPath
                            +
                            "../." +
                            "./target/SparkCarbonFileFormat/WriterOutput/")
    .getCanonicalPath

  writerPath = writerPath.replace("\\", "/")

  val decimalConversion = new DecimalConversion

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

  override def afterAll(): Unit = {
    sql("DROP TABLE IF EXISTS sdkOutputTable")
    CarbonProperties.getInstance()
      .addProperty(CarbonCommonConstants.CARBON_BAD_RECORDS_ACTION, badRecordAction)
  }

  test("test enum") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        | "fields":
        | [{
        | "name": "id",
        |						"type": {
        |                    "type": "enum",
        |                    "name": "Suit",
        |                    "symbols": ["SPADES", "HEARTS", "DIAMONDS", "CLUBS"]
        |                    }
        |                }
        |            ]
        |}""".stripMargin

    val json1 =
      """{"id":"HEARTS"}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row("HEARTS")))
  }

  test("test enum with struct type") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val mySchema =
      "{ " +
      "  \"name\": \"address\", " +
      "  \"type\": \"record\", " +
      "  \"fields\": [ " +
      "    { " +
      "      \"name\": \"name\", " +
      "      \"type\": \"string\" " +
      "    }, " +
      "    { " +
      "      \"name\": \"age\", " +
      "      \"type\": \"int\" " +
      "    }, " +
      "    { " +
      "       \"name\": \"address\",  \"type\": {" +
      "        \"type\" : \"record\",  \"name\" : \"my_address\"," +
      "         \"fields\" : [" +
      "         {\"name\": \"enumRec\", " +
      "           \"type\": { " +
      "            \"type\": \"enum\", " +
      "             \"name\": \"card\", " +
      "             \"symbols\": [\"SPADES\", \"HEARTS\", \"DIAMONDS\", \"CLUBS\"] " +
      "      } " +
      "}]}" +
      "    } " +
      "  ] " +
      "} "

    val json1 = "{\"name\":\"bob\", \"age\":10, \"address\": {\"enumRec\":\"SPADES\"}}"

    val nn = new org.apache.avro.Schema.Parser().parse(mySchema)
    val record = testUtil.jsonToAvro(json1, mySchema)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row("bob", 10, Row("SPADES"))))
  }

  test("test enum with Array type") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val mySchema =
      """ {
        |      "name": "address",
        |      "type": "record",
        |      "fields": [
        |      {
        |      "name": "name",
        |      "type": "string"
        |      },
        |      {
        |      "name": "age",
        |      "type": "int"
        |      },
        |      {
        |      "name": "address",
        |      "type": {
        |      "type": "array",
        |      "items": {
        |                    "name": "Suit",
        |                    "type": "enum",
        |                    "symbols": ["SPADES", "HEARTS", "DIAMONDS", "CLUBS"]
        |      }}}]
        |  }
      """.stripMargin

    val json: String = """ {"name": "bob","age": 10,"address": ["SPADES", "DIAMONDS"]} """

    val nn = new org.apache.avro.Schema.Parser().parse(mySchema)
    val record = testUtil.jsonToAvro(json, mySchema)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"),
      Seq(Row("bob", 10, mutable.WrappedArray.make(Array("SPADES", "DIAMONDS")))))
  }

  test("test union type long") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |       { "name": "first", "type": ["string", "int", "long"] }
        |     ]
        |}""".stripMargin

    val json1 =
      """{"first":{"long":10345}}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row(Row(null, null, 10345))))
  }

  test("test union type boolean") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |       { "name": "first", "type": ["boolean", "int", "long"] }
        |     ]
        |}""".stripMargin

    val json1 =
      """{"first":{"boolean":true}}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row(Row(true, null, null))))
  }

  test("test union type string") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |       { "name": "first", "type": ["string", "int", "long"] }
        |     ]
        |}""".stripMargin

    val json1 =
      """{"first":{"string":"abc"}}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row(Row("abc", null, null))))
  }

  test("test union type int") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |       { "name": "first", "type": ["string", "int", "long"] }
        |     ]
        |}""".stripMargin

    val json1 =
      """{"first":{"int":10}}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row(Row(null, 10, null))))
  }

  test("test union type with null") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |       { "name": "first", "type": ["null", "int"] }
        |     ]
        |}""".stripMargin

    val json1 =
      """{"first":{"null":null}}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row(Row(null))))
  }

  test("test union type with only type null") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |       { "name": "first", "type": ["null"] }
        |     ]
        |}""".stripMargin

    val json1 =
      """{"first":{"null":null}}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)
    val exception1 = intercept[UnsupportedOperationException] {
      val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
      writer.write(record)
      writer.close()
    }
    assert(exception1.getMessage
      .contains("Carbon do not support Avro UNION with only null type"))
  }

  test("test union type with Enum") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |      {
        |		  "name": "enum_field", "type": [{
        |          "namespace": "org.example.avro",
        |          "name": "EnumField",
        |          "type": "enum",
        |          "symbols": [
        |				"VAL_0",
        |				"VAL_1"
        |			]
        |        },"null"], "default": null
        |	}]
        |}""".stripMargin

    val json1 =
      """{"enum_field":{"org.example.avro.EnumField":"VAL_0"}}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row(Row("VAL_0"))))
  }

  test("test union type with Map") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |      {
        |		  "name": "map_field", "type": [{
        |          "namespace": "org.example.avro",
        |          "name": "mapField",
        |          "type": "map",
        |          "values":"string"
        |        },"int"], "default": null
        |	}]
        |}""".stripMargin

    val json1 =
      """{"map_field":{"map":{"street": "k-lane", "city": "bangalore"}}}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    sql("select * from sdkOutputTable").show(false)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(
      Row(Row(Map("city" -> "bangalore", "street" -> "k-lane"), null))))
  }

  test("test union type") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |      {
        |		  "name": "struct_field", "type": [{
        |          "namespace": "org.example.avro",
        |          "name": "structField",
        |          "type": "array",
        |          "items": { "name" : "name0", "type":"string"}
        |        },"int"], "default": null
        |	}]
        |}""".stripMargin

    val json1 =
      """{"struct_field":{"int":12}}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"),
      Seq(Row(Row(mutable.WrappedArray.make(Array(null)), 12))))
  }

  test("test Struct of Union") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """
        |{"name": "address",
        | "type": "record",
        | "fields": [
        |  { "name": "address",  "type": {
        |    "type" : "record",  "name" : "my_address",
        |        "fields" : [
        |    {"name": "city", "type": ["string","int"]}]}}
        |]}
      """.stripMargin

    val json1 =
      """{"address":{"city":{"int":1}}}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    sql("describe formatted sdkOutputTable").show(false)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row(Row(Row(null, 1)))))
    sql("insert into sdkOutputTable values(named_struct('city', named_struct('city0', 'abc', 'city1', 12)))")
    sql("select address.city.city0 from sdkOutputTable").show(false)
  }

  test("test Union with struct of array") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """
        |{"name": "address",
        | "type": "record",
        | "fields": [
        |  { "name": "address",  "type": {
        |    "type" : "record",  "name" : "my_address",
        |        "fields" : [
        |    {"name": "city", "type": ["string",  {
        |                "type": "array",
        |                "name": "abc_name_0",
        |                "items": {
        |                  "name": "_name_0",
        |                  "type": "record",
        |                  "fields": [
        |                    {
        |                      "name": "app_id",
        |                      "type": [
        |                        "null",
        |                        "string"
        |                      ]
        |                    }
        |                    ]
        |                    }
        |                    }
        |                    ]}]}}
        |]}
      """.stripMargin

    val json1 =
      """{
        |"address":{"city":
        |{"array":[
        |        {
        |          "app_id": {
        |            "string": "abc"
        |          }}]
        |          }}}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"),
      Seq(Row(Row(Row(null, mutable.WrappedArray.make(Array(Row(Row("abc")))))))))
  }

  test("test union type with Array and Struct") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |  "type": "record",
        |  "namespace": "example.avro",
        |  "name": "array_union",
        |  "fields": [
        |    {
        |      "name": "body",
        |      "type": {
        |        "name": "body",
        |        "type": "record",
        |        "fields": [
        |          {
        |            "name": "abc",
        |            "type": [
        |              "int",
        |              {
        |                "type": "array",
        |                "name": "abc_name_0",
        |                "items": {
        |                  "name": "_name_0",
        |                  "type": "record",
        |                  "fields": [
        |                    {
        |                      "name": "app_id",
        |                      "type": [
        |                        "null",
        |                        "string"
        |                      ]
        |                    },
        |                    {
        |                      "name": "app_name",
        |                      "type": [
        |                        "int",
        |                        "float",
        |                        "string"
        |                      ]
        |                    },
        |                    {
        |                      "name": "app_key",
        |                      "type": [
        |                        "null",
        |                        "string"
        |                      ]
        |                    }
        |                  ]
        |                }
        |              }
        |            ]
        |          }
        |        ]
        |      }
        |    }
        |  ]
        |}""".stripMargin

    val json1 =
      """{
        |  "body": {
        |    "abc": {
        |      "array": [
        |        {
        |          "app_id": {
        |            "string": "abc"
        |          },
        |          "app_name": {
        |            "string": "bcd"
        |          },
        |          "app_key": {
        |            "string": "cde"
        |          }
        |        }
        |      ]
        |    }
        |  }
        |}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    sql("describe formatted sdkOutputTable").show(false)
    checkAnswer(sql("select * from sdkOutputTable"),
      Seq(Row(Row(Row(null,
        mutable.WrappedArray.make(Array(Row(Row("abc"), Row(null, null, "bcd"), Row("cde")))))))))
  }

  test("test union type with Decimal") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |      {
        |		  "name": "union_field", "type": [{
        |          "namespace": "org.example.avro",
        |          "name": "dec",
        |          "type": "bytes",
        |         "logicalType": "decimal",
        |                     "precision": 10,
        |                     "scale": 2
        |        },"null"]
        |	}]
        |}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val decimalConversion = new DecimalConversion
    val logicalType = LogicalTypes.decimal(10, 2)
    val decimal = new java.math.BigDecimal("1010").setScale(2)
    //get unscaled 2's complement bytearray
    val bytes =
      decimalConversion.toBytes(decimal, nn.getField("union_field").schema, logicalType)
    val data = DatatypeConverter.printBase64Binary(bytes.array())
    val json1 =
      s"""{"union_field":{"bytes":"$data"}}""".stripMargin
    val record = testUtil.jsonToAvro(json1, schema1)
    val data1 = new String(record.get(0).asInstanceOf[ByteBuffer].array(),
      CarbonCommonConstants.DEFAULT_CHARSET_CLASS)
    val bytes1 = ByteBuffer.wrap(DatatypeConverter.parseBase64Binary(data1))
    val avroRec = new GenericData. Record(nn)
    avroRec.put("union_field", bytes1)


    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(avroRec)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkExistence(sql("select * from sdkOutputTable"), true, "1010.00")
  }

  test("test logical type decimal with struct") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """
        |{"name": "struct_field",
        | "type": "record",
        | "fields": [
        |  { "name": "record1", "type": "string"},
        |  { "name": "record2", "type": "float"},
        |  { "name": "struct_field_decimal",  "type": {
        |    "type" : "record",  "name" : "my_record",
        |        "fields" : [
        |    {"name": "record3", "type": "string"},
        |    {"name": "record4", "type": {"type" : "bytes",
        |                     "logicalType": "decimal",
        |                     "precision": 4,
        |                     "scale": 2
        |                    }}]}}
        |]}
      """.stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)

    val logicalType = LogicalTypes.decimal(4, 2)
    val decimal1 = new java.math.BigDecimal("32").setScale(2)
    //get unscaled 2's complement bytearray
    val bytes =
      decimalConversion.toBytes(decimal1, nn.getField("struct_field_decimal").schema, logicalType)
    val data = DatatypeConverter.printBase64Binary(bytes.array())
    val json1 = s""" {"record1":"bob", "record2":10.24, "struct_field_decimal" : {"record3":"abc", "record4":"$data"}} """
    val record = testUtil.jsonToAvro(json1, schema1)

    val jsonData = new String(record.get(2).asInstanceOf[GenericData.Record].get(1)
      .asInstanceOf[ByteBuffer].array(),
      CarbonCommonConstants.DEFAULT_CHARSET_CLASS)
    val bytesValue = ByteBuffer.wrap(DatatypeConverter.parseBase64Binary(jsonData))
    val mySchema =
      """
        |{"name": "struct_field_decimal",
        | "type": "record",
        | "fields": [
        |  { "name": "record3", "type": "string"},
        |  { "name": "record4", "type": {"type" : "bytes",
        |                     "logicalType": "decimal",
        |                     "precision": 4,
        |                     "scale": 2
        |                    }}
        |]}
      """.stripMargin
    val schema = new org.apache.avro.Schema.Parser().parse(mySchema)
    val genericByteArray = new GenericData.Record(schema)
    genericByteArray.put("record3", "abc")
    genericByteArray.put("record4", bytesValue)
    val avroRec = new GenericData.Record(nn)
    avroRec.put("record1", "bob")
    avroRec.put("record2", 10.24)
    avroRec.put("struct_field_decimal", genericByteArray)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(avroRec)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkExistence(sql("select * from sdkOutputTable"), true, "32.00")
  }

  test("test logical type decimal with Array") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """ {
        |      "name": "address",
        |      "type": "record",
        |      "fields": [
        |      {
        |      "name": "name",
        |      "type": "string"
        |      },
        |      {
        |      "name": "age",
        |      "type": "int"
        |      },
        |      {
        |      "name": "dec_fields",
        |      "type": {
        |      "type": "array",
        |      "items": {
        |      "name": "dec_field",
        |      "type": "bytes",
        |      "logicalType": "decimal",
        |      "precision": 4,
        |      "scale": 1
        |      }}}]
        |  }
      """.stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val logicalType = LogicalTypes.decimal(4, 1)
    val decimal1 = new java.math.BigDecimal("32").setScale(1)
    val decimal2 = new java.math.BigDecimal("42").setScale(1)
    //get unscaled 2's complement bytearray
    val bytes1 =
      decimalConversion.toBytes(decimal1, nn.getField("dec_fields").schema, logicalType)
    val bytes2 =
      decimalConversion.toBytes(decimal2, nn.getField("dec_fields").schema, logicalType)
    val data1 = DatatypeConverter.printBase64Binary(bytes1.array())
    val data2 = DatatypeConverter.printBase64Binary(bytes2.array())
    val json1: String = s""" {"name": "bob","age": 10,"dec_fields":["$data1","$data2"]} """
    val record = testUtil.jsonToAvro(json1, schema1)

    val jsonData1 = new String(record.get(2).asInstanceOf[GenericData.Array[ByteBuffer]].get(0)
      .array(),
      CarbonCommonConstants.DEFAULT_CHARSET_CLASS)
    val jsonData2 = new String(record.get(2).asInstanceOf[GenericData.Array[ByteBuffer]].get(1)
      .array(),
      CarbonCommonConstants.DEFAULT_CHARSET_CLASS)
    val bytesValue1 = ByteBuffer.wrap(DatatypeConverter.parseBase64Binary(jsonData1))
    val bytesValue2 = ByteBuffer.wrap(DatatypeConverter.parseBase64Binary(jsonData2))
    val genericByteArray = new GenericData.Array[ByteBuffer](2,
      Schema.createArray(Schema.create(Schema.Type.BYTES)))
    genericByteArray.add(bytesValue1)
    genericByteArray.add(bytesValue2)
    val avroRec = new GenericData.Record(nn)
    avroRec.put("name", "bob")
    avroRec.put("age", 10)
    avroRec.put("dec_fields", genericByteArray)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(avroRec)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkExistence(sql("select * from sdkOutputTable"), true, "32.0")
  }

  test("test logical type time-millis") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |		{
        |			"name": "id",
        |						"type": {"type" : "int", "logicalType": "time-millis"}
        |		},
        |		{
        |			"name": "course_details",
        |			"type": {
        |				"name": "course_details",
        |				"type": "record",
        |				"fields": [
        |					{
        |						"name": "course_struct_course_time",
        |						"type": {"type" : "int", "logicalType": "time-millis"}
        |					}
        |				]
        |			}
        |		}
        |	]
        |}""".stripMargin

    val json1 =
      """{"id": 172800,"course_details": { "course_struct_course_time":172800}}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)


    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row(172800, Row(172800))))
  }

  test("test logical type time-micros") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |		{
        |			"name": "id",
        |						"type": {"type" : "long", "logicalType": "time-micros"}
        |		},
        |		{
        |			"name": "course_details",
        |			"type": {
        |				"name": "course_details",
        |				"type": "record",
        |				"fields": [
        |					{
        |						"name": "course_struct_course_time",
        |						"type": {"type" : "long", "logicalType": "time-micros"}
        |					}
        |				]
        |			}
        |		}
        |	]
        |}""".stripMargin

    val json1 =
      """{"id": 1728000,"course_details": { "course_struct_course_time":1728000}}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)


    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row(1728000, Row(1728000))))
  }

  test("test logical type decimal through Json") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |		{
        |			"name": "id",
        |						"type": {"type" : "bytes",
        |                     "logicalType": "decimal",
        |                     "precision": 5,
        |                     "scale": 2
        |                    }
        |}
        |	]
        |}""".stripMargin
    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val logicalType = LogicalTypes.decimal(5, 2)
    val decimal = new java.math.BigDecimal("12.8").setScale(2)
    //get unscaled 2's complement bytearray
    val bytes =
      decimalConversion.toBytes(decimal, nn.getField("id").schema, logicalType)
    val data = DatatypeConverter.printBase64Binary(bytes.array())
    val json1 =
      s"""{"id":"$data"}""".stripMargin
    val record = testUtil.jsonToAvro(json1, schema1)
    val data1 = new String(record.get(0).asInstanceOf[ByteBuffer].array(),
      CarbonCommonConstants.DEFAULT_CHARSET_CLASS)
    val bytes1 = ByteBuffer.wrap(DatatypeConverter.parseBase64Binary(data1))
    val avroRec = new GenericData. Record(nn)
    avroRec.put("id", bytes1)
    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(avroRec)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row(decimal)))
  }

  test("test logical type decimal through Json with big decimal value") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |		{
        |			"name": "dec_field",
        |						"type": {"type" : "bytes",
        |                     "logicalType": "decimal",
        |                     "precision": 30,
        |                     "scale": 10
        |                    }
        |}
        |	]
        |}""".stripMargin
    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val logicalType = LogicalTypes.decimal(30, 10)
    val decimal = new java.math.BigDecimal("12672346879023.845789").setScale(10)
    //get unscaled 2's complement bytearray
    val bytes =
      decimalConversion.toBytes(decimal, nn.getField("dec_field").schema, logicalType)
    val data = DatatypeConverter.printBase64Binary(bytes.array())
    val json1 =
      s"""{"dec_field":"$data"}""".stripMargin
    val record = testUtil.jsonToAvro(json1, schema1)
    val data1 = new String(record.get(0).asInstanceOf[ByteBuffer].array(),
      CarbonCommonConstants.DEFAULT_CHARSET_CLASS)
    val bytes1 = ByteBuffer.wrap(DatatypeConverter.parseBase64Binary(data1))
    val avroRec = new GenericData. Record(nn)
    avroRec.put("dec_field", bytes1)
    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(avroRec)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row(decimal)))
  }

  test("test logical type decimal through Json with negative decimal value") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |		{
        |			"name": "dec_field",
        |						"type": {"type" : "bytes",
        |                     "logicalType": "decimal",
        |                     "precision": 30,
        |                     "scale": 6
        |                    }
        |}
        |	]
        |}""".stripMargin
    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val logicalType = LogicalTypes.decimal(30, 6)
    val decimal = new java.math.BigDecimal("-12672346879023.845").setScale(6)
    //get unscaled 2's complement bytearray
    val bytes =
      decimalConversion.toBytes(decimal, nn.getField("dec_field").schema, logicalType)
    val data = DatatypeConverter.printBase64Binary(bytes.array())
    val json1 =
      s"""{"dec_field":"$data"}""".stripMargin
    val record = testUtil.jsonToAvro(json1, schema1)
    val data1 = new String(record.get(0).asInstanceOf[ByteBuffer].array(),
      CarbonCommonConstants.DEFAULT_CHARSET_CLASS)
    val bytes1 = ByteBuffer.wrap(DatatypeConverter.parseBase64Binary(data1))
    val avroRec = new GenericData. Record(nn)
    avroRec.put("dec_field", bytes1)
    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(avroRec)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row(decimal)))
  }

  test("test logical type decimal through Avro") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |		{
        |			"name": "dec_field",
        |						"type": {"type" : "bytes",
        |                     "logicalType": "decimal",
        |                     "precision": 5,
        |                     "scale": 2
        |                    }
        |}
        |	]
        |}""".stripMargin
    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val logicalType = LogicalTypes.decimal(5, 2)
    val decimal = new java.math.BigDecimal("12.8").setScale(2)
    //get unscaled 2's complement bytearray
    val bytes =
      decimalConversion.toBytes(decimal, nn.getField("dec_field").schema, logicalType)
    val data = DatatypeConverter.printBase64Binary(bytes.array())
    val json1 =
      s"""{"dec_field":"$data"}""".stripMargin
    val avroRec = new GenericData. Record(nn)
    avroRec.put("dec_field", bytes)
    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(avroRec)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row(decimal)))
  }

  test("test logical type decimal with data having greater precision than specified precision") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |		{
        |			"name": "dec_field",
        |						"type": {"type" : "bytes",
        |                     "logicalType": "decimal",
        |                     "precision": 5,
        |                     "scale": 2
        |                    }
        |}
        |	]
        |}""".stripMargin
    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val logicalType = LogicalTypes.decimal(5, 2)
    val decimal = new java.math.BigDecimal("1218").setScale(2)
    //get unscaled 2's complement bytearray
    val bytes =
      decimalConversion.toBytes(decimal, nn.getField("dec_field").schema, logicalType)
    val data = DatatypeConverter.printBase64Binary(bytes.array())
    val json1 =
      s"""{"dec_field":"$data"}""".stripMargin
    val avroRec = new GenericData. Record(nn)
    avroRec.put("dec_field", bytes)
    val exception1 = intercept[Exception] {
    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(avroRec)
    writer.close()
    }
    assert(exception1.getMessage
      .contains("Data Loading failed as value Precision 6 is greater than specified Precision 5 in Avro Schema"))
  }

  test("test union with multiple record type") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "test.avro",
        |	"type": "record",
        |	"name": "NewCar2",
        |	"fields": [
        |      {
        |		  "name": "optionalExtra",
        |    "type": ["null",{
        |       "type":"record",
        |       "name":"Stereo",
        |       "fields" :[{
        |       "name":"make",
        |       "type":"string"
        |       },
        |       {
        |       "name":"speakers",
        |       "type":"int"
        |       }]
        |       },{
        |       "type":"record",
        |       "name":"LeatherTrim",
        |       "fields":[{
        |       "name":"colour",
        |       "type":"string"
        |       }]
        |       }],
        |       "default":null
        |       }]
        |
        |}""".stripMargin

    val json1 =
      """{"optionalExtra":{"test.avro.LeatherTrim":{"colour":"ab"}}}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row(Row(Row(null,null),Row("ab")))))
  }

  test("test union with multiple Enum type") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "test.avro",
        |	"type": "record",
        |	"name": "Union_data3",
        |	"fields": [
        |      {
        |		  "name": "enum_record",
        |    "type":
        |    ["long","null","string",
        |    {"type":"enum","name":"t1","symbols":["red","blue","yellow"]},
        |    {"type":"enum","name":"t2","symbols":["sun","mon","tue","wed","thu","fri","sat"]},
        |    "int"
        |    ]}]
        |}""".stripMargin

    val json1 =
      """{"enum_record":{"test.avro.t2":"sun"}}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val record = testUtil.jsonToAvro(json1, schema1)

    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(record)
    writer.close()
    sql(
      s"""CREATE EXTERNAL TABLE sdkOutputTable STORED BY
         |'carbondata' LOCATION
         |'$writerPath' """.stripMargin)
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row(Row(null,null,null,"sun",null))))
  }

  test("test spark file format") {
    sql("drop table if exists sdkOutputTable")
    FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(writerPath))
    val schema1 =
      """{
        |	"namespace": "com.apache.schema",
        |	"type": "record",
        |	"name": "StudentActivity",
        |	"fields": [
        |      {
        |		  "name": "union_field", "type": [{
        |          "namespace": "org.example.avro",
        |          "name": "dec",
        |          "type": "bytes",
        |         "logicalType": "decimal",
        |                     "precision": 10,
        |                     "scale": 2
        |        },"int"]
        |	}]
        |}""".stripMargin

    val nn = new org.apache.avro.Schema.Parser().parse(schema1)
    val decimalConversion = new DecimalConversion
    val logicalType = LogicalTypes.decimal(10, 2)
    val decimal = new java.math.BigDecimal("1010").setScale(2)
    //get unscaled 2's complement bytearray
    val bytes =
      decimalConversion.toBytes(decimal, nn.getField("union_field").schema, logicalType)
    val data = DatatypeConverter.printBase64Binary(bytes.array())
    val json1 =
      s"""{"union_field":{"bytes":"$data"}}""".stripMargin
    val record = testUtil.jsonToAvro(json1, schema1)
    val data1 = new String(record.get(0).asInstanceOf[ByteBuffer].array(),
      CarbonCommonConstants.DEFAULT_CHARSET_CLASS)
    val bytes1 = ByteBuffer.wrap(DatatypeConverter.parseBase64Binary(data1))
    val avroRec = new GenericData. Record(nn)
    avroRec.put("union_field", bytes1)


    val writer = CarbonWriter.builder.outputPath(writerPath).withAvroInput(nn).writtenBy("TestNonTransactionalCarbonTableWithAvroDataType").build()
    writer.write(avroRec)
    writer.close()
    sql(s"create table sdkOutputTable(union_field struct<union_field0:decimal(10,2),union_field1:int>) " +
        s"using carbon options(path='$writerPath')")
    checkAnswer(sql("select * from sdkOutputTable"), Seq(Row(Row(decimal,null))))
  }

}