/*
 * 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.alterTable

import java.io.{ByteArrayOutputStream, PrintStream}

import org.apache.spark.sql.{CarbonEnv, 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.util.CarbonProperties
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.tool.CarbonCli

class TestAlterTableSortColumnsProperty extends QueryTest with BeforeAndAfterAll {

  override def beforeAll(): Unit = {
    CarbonProperties.getInstance().addProperty(CarbonCommonConstants.CARBON_DATE_FORMAT,
      "yyyy-MM-dd")
    CarbonProperties.getInstance().addProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
      "yyyy-MM-dd HH:mm:ss")
    CarbonProperties.getInstance()
      .addProperty(CarbonCommonConstants.ENABLE_QUERY_STATISTICS, "false")
    dropTable()
    prepareTable()
  }

  override def afterAll(): Unit = {
    dropTable()
    CarbonProperties.getInstance().addProperty(CarbonCommonConstants.CARBON_DATE_FORMAT,
      CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT)
    CarbonProperties.getInstance().addProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
      CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT)
    CarbonProperties.getInstance()
      .addProperty(CarbonCommonConstants.ENABLE_QUERY_STATISTICS,
        CarbonCommonConstants.ENABLE_QUERY_STATISTICS_DEFAULT)
  }

  private def prepareTable(): Unit = {
    createTable(
      "alter_sc_base",
      Map("sort_scope"->"local_sort", "sort_columns"->"stringField")
    )
    createTable(
      "alter_sc_base_complex",
      Map("sort_scope"->"local_sort", "sort_columns"->"stringField"),
      true
    )
    createTable(
      "alter_sc_validate",
      Map.empty,
      true
    )
    createTable(
      "alter_sc_iud",
      Map.empty
    )
    createTable(
      "alter_sc_iud_complex",
      Map.empty,
      true
    )
    createTable(
      "alter_sc_long_string",
      Map("LONG_STRING_COLUMNS"->"stringField"),
      true
    )
    createTable(
      "alter_sc_insert",
      Map("sort_scope"->"local_sort", "sort_columns"->"stringField")
    )
    loadData("alter_sc_insert")
    createTable(
      "alter_sc_insert_complex",
      Map("sort_scope"->"local_sort", "sort_columns"->"stringField"),
      true
    )
    loadData("alter_sc_insert_complex")
    createTable(
      "alter_sc_range_column",
      Map("sort_scope"->"local_sort", "sort_columns"->"stringField", "range_column"->"smallIntField")
    )
    createTable(
      "alter_sc_range_column_base",
      Map("sort_scope"->"local_sort", "sort_columns"->"stringField")
    )

    Array("alter_sc_add_column", "alter_sc_add_column_base").foreach { tableName =>
      sql(
        s"""create table $tableName(
           | smallIntField smallInt,
           | intField int,
           | bigIntField bigint,
           | floatField float,
           | doubleField double,
           | timestampField timestamp,
           | dateField date,
           | stringField string
           | )
           | stored as carbondata
      """.stripMargin)
    }
    // decimalField decimal(25, 4),

    createTable(
      "alter_sc_bloom",
      Map("sort_scope"->"local_sort", "sort_columns"->"stringField")
    )
    createBloomDataMap("alter_sc_bloom", "alter_sc_bloom_dm1")
    createTable(
      "alter_sc_bloom_base",
      Map("sort_scope"->"local_sort", "sort_columns"->"stringField")
    )
    createBloomDataMap("alter_sc_bloom_base", "alter_sc_bloom_base_dm1")
    createTable(
      "alter_sc_agg",
      Map("sort_scope"->"local_sort", "sort_columns"->"intField")
    )
    createTable(
      "alter_sc_agg_base",
      Map("sort_scope"->"local_sort", "sort_columns"->"intField")
    )

    createTable(
      "alter_sc_cli",
      Map.empty
    )
  }

  private def dropTable(): Unit = {
    sql(s"drop table if exists alter_sc_base")
    sql(s"drop table if exists alter_sc_base_complex")
    sql(s"drop table if exists alter_sc_validate")
    sql(s"drop table if exists alter_sc_iud")
    sql(s"drop table if exists alter_sc_iud_complex")
    sql(s"drop table if exists alter_sc_long_string")
    sql(s"drop table if exists alter_sc_insert")
    sql(s"drop table if exists alter_sc_insert_complex")
    sql(s"drop table if exists alter_sc_range_column")
    sql(s"drop table if exists alter_sc_range_column_base")
    sql(s"drop table if exists alter_sc_add_column")
    sql(s"drop table if exists alter_sc_add_column_base")
    sql(s"drop table if exists alter_sc_bloom")
    sql(s"drop table if exists alter_sc_bloom_base")
    sql(s"drop table if exists alter_sc_agg")
    sql(s"drop table if exists alter_sc_agg_base")
    sql(s"drop table if exists alter_sc_cli")
  }

  private def createTable(
      tableName: String,
      tblProperties: Map[String, String] = Map.empty,
      withComplex: Boolean = false
  ): Unit = {
    val complexSql =
      if (withComplex) {
        ", arrayField array<string>, structField struct<col1:string, col2:string, col3:string>"
      } else {
        ""
      }
    val tblPropertiesSql =
      if (tblProperties.isEmpty) {
        ""
      } else {
        val propertiesString =
          tblProperties
            .map { entry =>
              s"'${ entry._1 }'='${ entry._2 }'"
            }
            .mkString(",")
        s"tblproperties($propertiesString)"
      }

    sql(
      s"""create table $tableName(
         | smallIntField smallInt,
         | intField int,
         | bigIntField bigint,
         | floatField float,
         | doubleField double,
         | timestampField timestamp,
         | dateField date,
         | stringField string,
         | varcharField varchar(10),
         | charField char(10)
         | $complexSql
         | )
         | stored as carbondata
         | $tblPropertiesSql
      """.stripMargin)
    // decimalField decimal(25, 4),
  }

  private def createBloomDataMap(tableName: String, dataMapName: String): Unit = {
    sql(
      s"""
         | CREATE DATAMAP $dataMapName ON TABLE $tableName
         | USING 'bloomfilter'
         | DMPROPERTIES(
         | 'INDEX_COLUMNS'='smallIntField,floatField,timestampField,dateField,stringField',
         | 'BLOOM_SIZE'='6400',
         | 'BLOOM_FPP'='0.001',
         | 'BLOOM_COMPRESS'='TRUE')
       """.stripMargin)
  }

  private def loadData(tableNames: String*): Unit = {
    tableNames.foreach { tableName =>
      sql(
        s"""load data local inpath '$resourcesPath/sort_columns'
           | into table $tableName
           | options ('global_sort_partitions'='2', 'COMPLEX_DELIMITER_LEVEL_1'='$$', 'COMPLEX_DELIMITER_LEVEL_2'=':')
      """.stripMargin)
    }
  }

  private def insertData(insertTable: String, tableNames: String*): Unit = {
    tableNames.foreach { tableName =>
      sql(
        s"""insert into table $tableName select * from $insertTable
      """.stripMargin)
    }
  }

  test("validate sort_scope and sort_columns") {
    // invalid combination
    var ex = intercept[RuntimeException] {
      sql("alter table alter_sc_validate set tblproperties('sort_scope'='local_sort')")
    }
    assert(ex.getMessage.contains("Cannot set SORT_SCOPE as local_sort when table has no SORT_COLUMNS"))

    ex = intercept[RuntimeException] {
      sql("alter table alter_sc_validate set tblproperties('sort_scope'='global_sort')")
    }
    assert(ex.getMessage.contains("Cannot set SORT_SCOPE as global_sort when table has no SORT_COLUMNS"))

    ex = intercept[RuntimeException] {
      sql("alter table alter_sc_validate set tblproperties('sort_scope'='local_sort', 'sort_columns'='')")
    }
    assert(ex.getMessage.contains("Cannot set SORT_COLUMNS as empty when setting SORT_SCOPE as local_sort"))

    ex = intercept[RuntimeException] {
      sql("alter table alter_sc_validate set tblproperties('sort_scope'='global_sort', 'sort_columns'=' ')")
    }
    assert(ex.getMessage.contains("Cannot set SORT_COLUMNS as empty when setting SORT_SCOPE as global_sort"))

    sql("alter table alter_sc_validate set tblproperties('sort_columns'='stringField', 'sort_scope'='local_sort')")
    ex = intercept[RuntimeException] {
      sql("alter table alter_sc_validate set tblproperties('sort_columns'=' ')")
    }
    assert(ex.getMessage.contains("Cannot set SORT_COLUMNS as empty when SORT_SCOPE is LOCAL_SORT"))

    sql("alter table alter_sc_validate set tblproperties('sort_scope'='global_sort')")
    ex = intercept[RuntimeException] {
      sql("alter table alter_sc_validate set tblproperties('sort_columns'='')")
    }
    assert(ex.getMessage.contains("Cannot set SORT_COLUMNS as empty when SORT_SCOPE is GLOBAL_SORT"))

    // wrong/duplicate sort_columns
    ex = intercept[RuntimeException] {
      sql("alter table alter_sc_validate set tblproperties('sort_columns'=' stringField1 , intField')")
    }
    assert(ex.getMessage.contains("stringField1 does not exist in table"))

    ex = intercept[RuntimeException] {
      sql("alter table alter_sc_validate set tblproperties('sort_columns'=' stringField1 , intField, stringField1')")
    }
    assert(ex.getMessage.contains("SORT_COLUMNS Either having duplicate columns : stringField1 or it contains illegal argumnet"))

    ex = intercept[RuntimeException] {
      sql("alter table alter_sc_validate set tblproperties('sort_columns'=' stringField , intField, stringField')")
    }
    assert(ex.getMessage.contains("SORT_COLUMNS Either having duplicate columns : stringField or it contains illegal argumnet"))

    // not supported data type
//    ex = intercept[RuntimeException] {
//      sql("alter table alter_sc_validate set tblproperties('sort_columns'='decimalField')")
//    }
//    assert(ex.getMessage.contains("sort_columns is unsupported for DECIMAL data type column: decimalField"))

    ex = intercept[RuntimeException] {
      sql("alter table alter_sc_validate set tblproperties('sort_columns'='doubleField')")
    }
    assert(ex.getMessage.contains("sort_columns is unsupported for DOUBLE datatype column: doubleField"))

    ex = intercept[RuntimeException] {
      sql("alter table alter_sc_validate set tblproperties('sort_columns'='arrayField')")
    }
    assert(ex.getMessage.contains("sort_columns is unsupported for ARRAY datatype column: arrayField"))

    ex = intercept[RuntimeException] {
      sql("alter table alter_sc_validate set tblproperties('sort_columns'='structField')")
    }
    assert(ex.getMessage.contains("sort_columns is unsupported for STRUCT datatype column: structField"))

    ex = intercept[RuntimeException] {
      sql("alter table alter_sc_validate set tblproperties('sort_columns'='structField.col1')")
    }
    assert(ex.getMessage.contains("sort_columns: structField.col1 does not exist in table"))
  }

  test("long string column") {
    val ex = intercept[RuntimeException] {
      sql("alter table alter_sc_long_string set tblproperties('sort_columns'='intField, stringField')")
    }
    assert(ex.getMessage.contains("sort_columns is unsupported for long string datatype column: stringField"))
  }

  test("describe formatted") {
    // valid combination
    sql("alter table alter_sc_validate set tblproperties('sort_scope'='no_sort', 'sort_columns'='')")
    checkExistence(sql("describe formatted alter_sc_validate"), true, "NO_SORT")

    sql("alter table alter_sc_validate set tblproperties('sort_scope'='no_sort', 'sort_columns'='bigIntField,stringField')")
    checkExistence(sql("describe formatted alter_sc_validate"), true, "no_sort", "bigIntField, stringField".toLowerCase())

    sql("alter table alter_sc_validate set tblproperties('sort_scope'='local_sort', 'sort_columns'='stringField,bigIntField')")
    checkExistence(sql("describe formatted alter_sc_validate"), true, "local_sort", "stringField, bigIntField".toLowerCase())

    // global dictionary or direct dictionary
    sql("alter table alter_sc_validate set tblproperties('sort_scope'='global_sort', 'sort_columns'=' charField , bigIntField , timestampField ')")
    checkExistence(sql("describe formatted alter_sc_validate"), true, "global_sort", "charField, bigIntField, timestampField".toLowerCase())

    // supported data type
    sql("alter table alter_sc_validate set tblproperties('sort_scope'='local_sort', 'sort_columns'='smallIntField, intField, bigIntField, timestampField, dateField, stringField, varcharField, charField')")
    checkExistence(sql("describe formatted alter_sc_validate"), true, "local_sort", "smallIntField, intField, bigIntField, timestampField, dateField, stringField, varcharField, charField".toLowerCase())
  }

  test("IUD and Query") {
    testIUDAndQuery("alter_sc_iud", "alter_sc_base", "alter_sc_insert")
  }

  test("IUD and Query with complex data type") {
    testIUDAndQuery("alter_sc_iud_complex", "alter_sc_base_complex", "alter_sc_insert_complex")
  }

  private def testIUDAndQuery(tableName: String, baseTableName: String, insertTableName: String): Unit = {
    loadData(tableName, baseTableName)
    checkAnswer(sql(s"select * from $tableName order by floatField"), sql(s"select * from $baseTableName order by floatField"))

    // alter table to change SORT_SCOPE and SORT_COLUMNS
    sql(s"alter table $tableName set tblproperties('sort_scope'='global_sort', 'sort_columns'='charField')")
    loadData(tableName, baseTableName)
    checkAnswer(sql(s"select * from $tableName order by floatField"), sql(s"select * from $baseTableName order by floatField"))

    // alter table to local_sort with new SORT_COLUMNS
    sql(s"alter table $tableName set tblproperties('sort_scope'='local_sort', 'sort_columns'='timestampField, intField, stringField')")
    loadData(tableName, baseTableName)
    checkAnswer(sql(s"select * from $tableName order by floatField"), sql(s"select * from $baseTableName order by floatField"))

    // alter table to revert SORT_COLUMNS
    sql(s"alter table $tableName set tblproperties('sort_columns'='stringField, intField, timestampField')")
    loadData(tableName, baseTableName)
    checkAnswer(sql(s"select * from $tableName order by floatField"), sql(s"select * from $baseTableName order by floatField"))

    // alter table to change SORT_COLUMNS
    sql(s"alter table $tableName set tblproperties('sort_columns'='intField, stringField')")
    loadData(tableName, baseTableName)
    checkAnswer(sql(s"select * from $tableName order by floatField"), sql(s"select * from $baseTableName order by floatField"))

    // alter table to change SORT_SCOPE
    sql(s"alter table $tableName set tblproperties('sort_scope'='local_sort', 'sort_columns'='charField, smallIntField')")
    loadData(tableName, baseTableName)
    checkAnswer(sql(s"select * from $tableName order by floatField"), sql(s"select * from $baseTableName order by floatField"))

    // set input segments
    (0 to 5).foreach { segment =>
      sql(s"set carbon.input.segments.default.$tableName=$segment").collect()
      sql(s"set carbon.input.segments.default.$baseTableName=$segment").collect()
      checkAnswer(sql(s"select count(*) from $tableName"), sql(s"select count(*) from $baseTableName"))
      checkAnswer(sql(s"select * from $tableName order by floatField"), sql(s"select * from $baseTableName order by floatField"))
      checkAnswer(sql(s"select * from $tableName where smallIntField = 2 order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 order by floatField"))
      checkAnswer(sql(s"select * from $tableName where intField >= 2 order by floatField"), sql(s"select * from $baseTableName where intField >= 2 order by floatField"))
      checkAnswer(sql(s"select * from $tableName where smallIntField = 2 or intField >= 2 order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 or intField >= 2 order by floatField"))
    }
    sql(s"set carbon.input.segments.default.$tableName=*").collect()
    sql(s"set carbon.input.segments.default.$baseTableName=*").collect()

    // query
    checkAnswer(sql(s"select count(*) from $tableName"), sql(s"select count(*) from $baseTableName"))
    checkAnswer(sql(s"select * from $tableName order by floatField"), sql(s"select * from $baseTableName order by floatField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 order by floatField"))
    checkAnswer(sql(s"select * from $tableName where intField >= 2 order by floatField"), sql(s"select * from $baseTableName where intField >= 2 order by floatField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 or intField >= 2 order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 or intField >= 2 order by floatField"))

    // delete
    sql(s"delete from $tableName where smallIntField = 2")
    sql(s"delete from $baseTableName where smallIntField = 2")
    checkAnswer(sql(s"select * from $tableName order by floatField"), sql(s"select * from $baseTableName order by floatField"))

    // compaction for column drift
    sql(s"alter table $tableName set tblproperties('sort_scope'='local_sort', 'sort_columns'='charField, intField')")
    // [Segment info]:
    //   | sorted | dimension order(sort_columns is in [])                                                     | measure order
    // -------------------------------------------------------------------------------------------------------------------------------------------------------------------------
    // 0 | false  | timestampField, dateField, stringField, varcharField, charField                            | smallIntField, intField, bigIntField, floatField, doubleField
    // 1 | true   | [charField], timestampField, dateField, stringField, varcharField                          | smallIntField, intField, bigIntField, floatField, doubleField
    // 2 | false  | [timestampField, intField, stringField], charField, dateField, varcharField                | smallIntField, bigIntField, floatField, doubleField
    // 3 | false  | [stringField, intField, timestampField], charField, dateField, varcharField                | smallIntField, bigIntField, floatField, doubleField
    // 4 | false  | [intField, stringField], timestampField, charField, dateField, varcharField                | smallIntField, bigIntField, floatField, doubleField
    // 5 | true   | [charField, smallIntField], intField, stringField, timestampField, dateField, varcharField | bigIntField, floatField, doubleField
    // Column drift happened, intField and smallIntField became dimension.
    // The order of columns also changed.
    //
    // [Table info]:
    //            | dimension order(sort_columns is in [])                                                     | measure order
    // --------------------------------------------------------------------------------------------------------------------------------------------------------------------------
    // table      | [charField], smallIntField, intField, stringField, timestampField, dateField, varcharField | bigIntField, floatField, doubleField
    sql(s"alter table $tableName compact 'minor'")
    sql(s"alter table $baseTableName compact 'minor'")
    checkAnswer(sql(s"select count(*) from $tableName"), sql(s"select count(*) from $baseTableName"))
    checkAnswer(sql(s"select * from $tableName order by floatField"), sql(s"select * from $baseTableName order by floatField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 order by floatField"))
    checkAnswer(sql(s"select * from $tableName where intField >= 2 order by floatField"), sql(s"select * from $baseTableName where intField >= 2 order by floatField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 or intField >= 2 order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 or intField >= 2 order by floatField"))

    sql(s"delete from $tableName")
    checkAnswer(sql(s"select count(*) from $tableName"), Seq(Row(0)))
    sql(s"delete from $baseTableName")
    checkAnswer(sql(s"select count(*) from $baseTableName"), Seq(Row(0)))

    // insert & load data
    sql(s"alter table $tableName set tblproperties('sort_scope'='global_sort', 'sort_columns'='timestampField')")
    insertData(insertTableName, tableName, baseTableName)
    loadData(tableName, baseTableName)
    checkAnswer(sql(s"select * from $tableName order by floatField"), sql(s"select * from $baseTableName order by floatField"))

    sql(s"alter table $tableName set tblproperties('sort_scope'='no_sort', 'sort_columns'='')")
    insertData(insertTableName, tableName, baseTableName)
    loadData(tableName, baseTableName)
    checkAnswer(sql(s"select * from $tableName order by floatField"), sql(s"select * from $baseTableName order by floatField"))

    sql(s"alter table $tableName set tblproperties('sort_scope'='local_sort', 'sort_columns'='charField, bigIntField, smallIntField')")
    insertData(insertTableName, tableName, baseTableName)
    loadData(tableName, baseTableName)
    checkAnswer(sql(s"select * from $tableName order by floatField"), sql(s"select * from $baseTableName order by floatField"))

    // update
    sql(s"update $tableName set (smallIntField, intField, bigIntField, floatField, doubleField) = (smallIntField + 3, intField + 3, bigIntField + 3, floatField + 3, doubleField + 3) where smallIntField = 2").collect()
    sql(s"update $baseTableName set (smallIntField, intField, bigIntField, floatField, doubleField) = (smallIntField + 3, intField + 3, bigIntField + 3, floatField + 3, doubleField + 3) where smallIntField = 2").collect()
    checkAnswer(sql(s"select * from $tableName order by floatField"), sql(s"select * from $baseTableName order by floatField"))

    // query
    checkAnswer(sql(s"select count(*) from $tableName"), sql(s"select count(*) from $baseTableName"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 order by floatField"))
    checkAnswer(sql(s"select * from $tableName where intField >= 2 order by floatField"), sql(s"select * from $baseTableName where intField >= 2 order by floatField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 or intField >= 2 order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 or intField >= 2 order by floatField"))

    // set input segments
    (6 to 11).foreach { segment =>
      sql(s"set carbon.input.segments.default.$tableName=$segment").collect()
      sql(s"set carbon.input.segments.default.$baseTableName=$segment").collect()
      checkAnswer(sql(s"select * from $tableName where smallIntField = 2 order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 order by floatField"))
    }
    sql(s"set carbon.input.segments.default.$tableName=*").collect()
    sql(s"set carbon.input.segments.default.$baseTableName=*").collect()

    // no_sort compaction flow for column drift
    sql(s"alter table $tableName set tblproperties('sort_scope'='no_sort', 'sort_columns'='charField, intField')")
    // sort_scope become no_sort
    sql(s"alter table $tableName compact 'minor'")
    sql(s"alter table $baseTableName compact 'minor'")
    checkAnswer(sql(s"select count(*) from $tableName"), sql(s"select count(*) from $baseTableName"))
    checkAnswer(sql(s"select * from $tableName order by floatField"), sql(s"select * from $baseTableName order by floatField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 order by floatField"))
    checkAnswer(sql(s"select * from $tableName where intField >= 2 order by floatField"), sql(s"select * from $baseTableName where intField >= 2 order by floatField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 or intField >= 2 order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 or intField >= 2 order by floatField"))
  }

  test("range column") {
    val tableName = "alter_sc_range_column"
    val baseTableName = "alter_sc_range_column_base"
    loadData(tableName, baseTableName)
    sql(s"alter table $tableName set tblproperties('sort_scope'='local_sort', 'sort_columns'='smallIntField, charField')")
    loadData(tableName, baseTableName)

    checkAnswer(sql(s"select count(*) from $tableName"), sql(s"select count(*) from $baseTableName"))
    checkAnswer(sql(s"select * from $tableName order by floatField"), sql(s"select * from $baseTableName order by floatField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 order by floatField"))
  }

  test("add/drop column for sort_columns") {
    val tableName = "alter_sc_add_column"
    val baseTableName = "alter_sc_add_column_base"
    loadData(tableName, baseTableName)
    sql(s"alter table $tableName set tblproperties('sort_scope'='local_sort', 'sort_columns'='smallIntField, stringField')")
    loadData(tableName, baseTableName)
    // add column
    sql(s"alter table $tableName add columns( varcharField varchar(10), charField char(10))")
    sql(s"alter table $baseTableName add columns( varcharField varchar(10), charField char(10))")
    loadData(tableName, baseTableName)

    checkAnswer(sql(s"select count(*) from $tableName"), sql(s"select count(*) from $baseTableName"))
    checkAnswer(sql(s"select * from $tableName order by floatField, charField"), sql(s"select * from $baseTableName order by floatField, charField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 order by floatField, charField"), sql(s"select * from $baseTableName where smallIntField = 2 order by floatField, charField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 and charField is null order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 and charField is null order by floatField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 and charField is not null order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 and charField is not null order by floatField"))

    // add new column to sort_columns
    sql(s"alter table $tableName set tblproperties('sort_scope'='local_sort', 'sort_columns'='smallIntField, charField')")
    loadData(tableName, baseTableName)
    checkAnswer(sql(s"select count(*) from $tableName"), sql(s"select count(*) from $baseTableName"))
    checkAnswer(sql(s"select * from $tableName order by floatField, charField"), sql(s"select * from $baseTableName order by floatField, charField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 order by floatField, charField"), sql(s"select * from $baseTableName where smallIntField = 2 order by floatField, charField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 and charField is null order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 and charField is null order by floatField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 and charField is not null order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 and charField is not null order by floatField"))

    // drop column of old sort_columns
    sql(s"alter table $tableName drop columns(stringField)")
    sql(s"alter table $baseTableName drop columns(stringField)")
    loadData(tableName, baseTableName)
    checkAnswer(sql(s"select count(*) from $tableName"), sql(s"select count(*) from $baseTableName"))
    checkAnswer(sql(s"select * from $tableName order by floatField, charField"), sql(s"select * from $baseTableName order by floatField, charField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 order by floatField, charField"), sql(s"select * from $baseTableName where smallIntField = 2 order by floatField, charField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 and charField is null order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 and charField is null order by floatField"))
    checkAnswer(sql(s"select * from $tableName where smallIntField = 2 and charField is not null order by floatField"), sql(s"select * from $baseTableName where smallIntField = 2 and charField is not null order by floatField"))
  }

  test("bloom filter") {
    CarbonProperties.getInstance()
      .addProperty(CarbonCommonConstants.ENABLE_QUERY_STATISTICS, "true")
    val tableName = "alter_sc_bloom"
    val dataMapName = "alter_sc_bloom_dm1"
    val baseTableName = "alter_sc_bloom_base"
    loadData(tableName, baseTableName)
    checkExistence(sql(s"SHOW DATAMAP ON TABLE $tableName"), true, "bloomfilter", dataMapName)
    checkExistence(sql(s"EXPLAIN SELECT * FROM $tableName WHERE smallIntField = 3"), true, "bloomfilter", dataMapName)
    checkAnswer(sql(s"select * from $tableName where smallIntField = 3 order by floatField"), sql(s"select * from $baseTableName where smallIntField = 3 order by floatField"))

    sql(s"alter table $tableName set tblproperties('sort_scope'='global_sort', 'sort_columns'='smallIntField, charField')")
    loadData(tableName, baseTableName)
    checkExistence(sql(s"EXPLAIN SELECT * FROM $tableName WHERE smallIntField = 3"), true, "bloomfilter", dataMapName)
    checkAnswer(sql(s"select * from $tableName where smallIntField = 3 order by floatField"), sql(s"select * from $baseTableName where smallIntField = 3 order by floatField"))
  }

  test("carboncli -cmd sort_columns -p <segment folder>") {
    val tableName = "alter_sc_cli"
    // no_sort
    loadData(tableName)
    sql(s"alter table $tableName set tblproperties('sort_scope'='global_sort', 'sort_columns'='charField, timestampField')")
    // global_sort
    loadData(tableName)
    sql(s"alter table $tableName set tblproperties('sort_scope'='local_sort', 'sort_columns'='intField, stringField')")
    // local_sort
    loadData(tableName)
    // update table to generate one more index in each segment
    sql(s"update $tableName set (smallIntField, intField, bigIntField, floatField, doubleField) = (smallIntField + 3, intField + 3, bigIntField + 3, floatField + 3, doubleField + 3) where smallIntField = 2").collect()

    val table = CarbonEnv.getCarbonTable(Option("default"), tableName)(sqlContext.sparkSession)
    val tablePath = table.getTablePath
    (0 to 2).foreach { segmentId =>
      val segmentPath = CarbonTablePath.getSegmentPath(tablePath, segmentId.toString)
      val args: Array[String] = Array("-cmd", "sort_columns", "-p", segmentPath)
      val out: ByteArrayOutputStream = new ByteArrayOutputStream
      val stream: PrintStream = new PrintStream(out)
      CarbonCli.run(args, stream)
      val output: String = new String(out.toByteArray)
      if (segmentId == 2) {
        assertResult(s"Input Folder: $segmentPath\nsorted by intfield,stringfield\n")(output)
      } else {
        assertResult(s"Input Folder: $segmentPath\nunsorted\n")(output)
      }
    }
  }

}