/*
* 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 id 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.standardpartition

import scala.collection.JavaConverters._

import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.hive.CarbonRelation
import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, CarbonEnv, Row}
import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.BeforeAndAfterAll

import org.apache.carbondata.core.datastore.impl.FileFactory

class StandardPartitionWithPreaggregateTestCase extends QueryTest with BeforeAndAfterAll {

  val testData = s"$resourcesPath/sample.csv"

  override def beforeAll(): Unit = {
    sql("drop database if exists partition_preaggregate cascade")
    sql("create database partition_preaggregate")
    sql("use partition_preaggregate")
    sql(
      """
        | CREATE TABLE par(id INT, name STRING, age INT) PARTITIONED BY(city STRING)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql(
      """
        | CREATE TABLE maintable(id int, name string, city string) partitioned by (age int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql(s"LOAD DATA LOCAL INPATH '$testData' into table maintable")
  }

  override def afterAll(): Unit = {
    sql("drop database if exists partition_preaggregate cascade")
    sql("use default")
  }

  // Create aggregate table on partition with partition column in aggregation only.
  test("test preaggregate table creation on partition table with partition col as aggregation") {
    sql("create datamap p1 on table par using 'preaggregate' as select id, sum(city) from par group by id")
    assert(!CarbonEnv.getCarbonTable(Some("partition_preaggregate"), "par_p1")(sqlContext.sparkSession).isHivePartitionTable)
  }

  // Create aggregate table on partition with partition column in projection and aggregation only.
  test("test preaggregate table creation on partition table with partition col as projection") {
    sql("create datamap p2 on table par using 'preaggregate' as select id, city, min(city) from par group by id,city ")
    assert(CarbonEnv.getCarbonTable(Some("partition_preaggregate"), "par_p2")(sqlContext.sparkSession).isHivePartitionTable)
  }

  // Create aggregate table on partition with partition column as group by.
  test("test preaggregate table creation on partition table with partition col as group by") {
    sql("create datamap p3 on table par using 'preaggregate' as select id, max(city) from par group by id,city ")
    assert(CarbonEnv.getCarbonTable(Some("partition_preaggregate"), "par_p3")(sqlContext.sparkSession).isHivePartitionTable)
  }

  // Create aggregate table on partition without partition column.
  test("test preaggregate table creation on partition table without partition column") {
    sql("create datamap p4 on table par using 'preaggregate' as select name, count(id) from par group by name ")
    assert(!CarbonEnv.getCarbonTable(Some("partition_preaggregate"), "par_p4")(sqlContext.sparkSession).isHivePartitionTable)
  }
  
  test("test data correction in aggregate table when partition column is used") {
    sql("create datamap p1 on table maintable using 'preaggregate' as select id, sum(age) from maintable group by id, age")
    checkAnswer(sql("select * from maintable_p1"),
      Seq(Row(1,31,31),
        Row(2,27,27),
        Row(3,70,35),
        Row(4,26,26),
        Row(4,29,29)))
    preAggTableValidator(sql("select id, sum(age) from maintable group by id, age").queryExecution.analyzed, "maintable_p1")
    sql("drop datamap p1 on table maintable")
  }

  test("test data correction in aggregate table when partition column is not used") {
    sql("create datamap p2 on table maintable using 'preaggregate' as select id, max(age) from maintable group by id")
    checkAnswer(sql("select * from maintable_p2"),
      Seq(Row(1,31),
        Row(2,27),
        Row(3,35),
        Row(4,29)))
    preAggTableValidator(sql("select id, max(age) from maintable group by id").queryExecution.analyzed, "maintable_p2")
    sql("drop datamap p2 on table maintable")
  }

  test("test data correction with insert overwrite") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month,day")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert overwrite table partitionone values('v',2014,1,1)")
    checkAnswer(sql("select * from partitionone"), Seq(Row("v",2014,1,1)))
    checkAnswer(sql("select * from partitionone_p1"), Seq(Row("v",2014,2014,1,1)))
  }

  test("test data correction with insert overwrite on different value") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month,day")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert overwrite table partitionone values('v',2015,1,1)")
    checkAnswer(sql("select * from partitionone"), Seq(Row("k",2014,1,1), Row("v",2015,1,1)))
    checkAnswer(sql("select * from partitionone_p1"), Seq(Row("k",2014,2014,1,1), Row("v",2015,2015,1,1)))
  }

  test("test to check column ordering in parent and child table") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, month, year,day")
    val parentTable = CarbonEnv.getCarbonTable(Some("partition_preaggregate"), "partitionone")(sqlContext.sparkSession)
    val childTable = CarbonEnv.getCarbonTable(Some("partition_preaggregate"), "partitionone_p1")(sqlContext.sparkSession)
    val parentPartitionColumns = parentTable.getPartitionInfo.getColumnSchemaList
    val childPartitionColumns = childTable.getPartitionInfo.getColumnSchemaList
    assert(parentPartitionColumns.asScala.zip(childPartitionColumns.asScala).forall {
      case (a,b) =>
        a.getColumnName.equalsIgnoreCase(b.getParentColumnTableRelations.get(0).getColumnName)
    })
  }
  
  test("test data after minor compaction on partition table with pre-aggregate") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month,day")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2014,1,2)")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2015,2,1)")
    sql("alter table partitionone compact 'minor'")
    val showSegments = sql("show segments for table partitionone").collect().map{a=> (a.get(0), a.get(1))}
    assert(showSegments.count(_._2 == "Success") == 1)
    assert(showSegments.count(_._2 == "Compacted") == 4)
  }

  test("test data after major compaction on partition table with pre-aggregate") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month,day")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2014,1,2)")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2015,2,1)")
    sql("insert into partitionone values('k',2015,2,1)")
    sql("alter table partitionone compact 'major'")
    val showSegments = sql("show segments for table partitionone").collect().map{a=> (a.get(0), a.get(1))}
    assert(showSegments.count(_._2 == "Success") == 1)
    assert(showSegments.count(_._2 == "Compacted") == 5)
  }

  test("test drop partition 1") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2014,1,2)")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2015,2,1)")
    sql("insert into partitionone values('k',2015,2,1)")
    sql("alter table partitionone drop partition(day=1)")
    checkAnswer(sql("select * from partitionone"), Seq(Row("k",2014,1,2)))
    checkAnswer(sql("select * from partitionone_p1"), Seq(Row("k",2014,2014,1,2)))
  }

  test("test drop partition 2") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2014,1,2)")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2015,2,3)")
    sql("insert into partitionone values('k',2015,2,1)")
    sql("alter table partitionone drop partition(day=1)")
    checkAnswer(sql("select * from partitionone"), Seq(Row("k",2014,1,2), Row("k",2015,2,3)))
    checkAnswer(sql("select * from partitionone_p1"), Seq(Row("k",2014,2014,1,2), Row("k",2015,2015,2,3)))
  }

  test("test drop partition directory") {
    sql("drop table if exists droppartition")
    sql(
      """
        | CREATE TABLE if not exists droppartition (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql("insert into droppartition values('k',2014,1,1)")
    sql("insert into droppartition values('k',2015,2,3)")
    sql("alter table droppartition drop partition(year=2015,month=2,day=3)")
    sql("clean files for table droppartition")
    val table = CarbonEnv.getCarbonTable(Option("partition_preaggregate"), "droppartition")(sqlContext.sparkSession)
    val tablePath = table.getTablePath
    val carbonFiles = FileFactory.getCarbonFile(tablePath).listFiles().filter{
      file => file.getName.equalsIgnoreCase("year=2015")
    }
    assert(carbonFiles.length == 0)
  }

  test("test data with filter query") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2014,1,2)")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2015,2,3)")
    sql("insert into partitionone values('k',2015,2,1)")
    sql("alter table partitionone drop partition(day=1)")
    checkAnswer(sql("select empname, sum(year) from partitionone where day=3 group by empname, year, month, day"), Seq(Row("k",2015)))
    checkAnswer(sql("select * from partitionone_p1"), Seq(Row("k",2014,2014,1,2), Row("k",2015,2015,2,3)))
  }

  test("test drop partition 3") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2014,1,2)")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2015,2,3)")
    sql("insert into partitionone values('k',2015,2,1)")
    sql("alter table partitionone drop partition(day=1,month=1)")
    checkAnswer(sql("select * from partitionone"), Seq(Row("k",2014,1,2), Row("k",2015, 2,3), Row("k",2015, 2,1)))
    checkAnswer(sql("select * from partitionone_p1"), Seq(Row("k",2014,2014,1,2), Row("k",2015,2015,2,3), Row("k",2015,2015,2,1)))
  }

  test("test drop partition 4") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2014,1,2)")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2015,2,3)")
    sql("insert into partitionone values('k',2015,2,1)")
    sql("alter table partitionone drop partition(year=2014,day=1)")
    checkAnswer(sql("select * from partitionone"), Seq(Row("k",2014,1,2), Row("k",2015, 2,3), Row("k",2015, 2,1)))
    checkAnswer(sql("select * from partitionone_p1"), Seq(Row("k",2014,2014,1,2), Row("k",2015,2015, 2,3), Row("k",2015,2015, 2,1)))
  }

  test("test drop partition 5") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2014,1,2)")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2015,2,3)")
    sql("insert into partitionone values('k',2015,2,1)")
    sql("alter table partitionone drop partition(year=2014,month=1, day=1)")

    checkAnswer(sql("select * from partitionone"), Seq(Row("k",2014,1,2), Row("k",2015, 2,3), Row("k",2015, 2,1)))
    checkAnswer(sql("select * from partitionone_p1"), Seq(Row("k",2014,2014,1,2), Row("k",2015,2015,2,3), Row("k",2015,2015,2,1)))
    sql("show partitions partitionone_p1").show()
  }

  test("test drop partition 6") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2014,1,2)")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2015,2,3)")
    sql("insert into partitionone values('k',2015,2,1)")
    sql("alter table partitionone drop partition(year=2014,month=1, day=1)")
    checkAnswer(sql("select * from partitionone"), Seq(Row("k",2014,1,2), Row("k",2015, 2,3), Row("k",2015, 2,1)))
    checkAnswer(sql("select * from partitionone_p1"), Seq(Row("k",2014,2014,1,2), Row("k",2015,2015,2,3), Row("k",2015,2015,2,1)))
  }

  test("test drop partition 7") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql(
      "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, day")
    sql(
      "create datamap p2 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, month")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2014,1,2)")
    val exceptionMessage = intercept[Exception] {
      sql("alter table partitionone drop partition(year=2014,month=1, day=1)")
    }.getMessage
    assert(exceptionMessage.contains("Cannot drop partition as one of the partition"))
    assert(exceptionMessage.contains("p2"))
    assert(exceptionMessage.contains("p1"))
  }

  test("test drop partition 8") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql(
      "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
    sql(
      "create datamap p2 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, month, day")
    sql(
      "create datamap p3 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2014,1,2)")
    val exceptionMessage = intercept[Exception] {
      sql("alter table partitionone drop partition(year=2014,month=1, day=1)")
    }.getMessage
    assert(exceptionMessage.contains("Cannot drop partition as one of the partition"))
    assert(exceptionMessage.contains("p2"))
    assert(exceptionMessage.contains("p3"))
    assert(!exceptionMessage.contains("p1"))
  }

  test("test drop partition 9") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql(
      "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2014,1,2)")
    val exceptionMessage = intercept[Exception] {
      sql("alter table partitionone drop partition(year=2014,month=1, day=1)")
    }.getMessage
    assert(exceptionMessage.contains("Cannot drop partition as one of the partition"))
    assert(exceptionMessage.contains("p1"))
  }

  test("test drop partition 10") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql(
      "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname")
    sql(
      "create datamap p2 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
    sql("insert into partitionone values('k',2014,1,1)")
    sql("insert into partitionone values('k',2014,1,2)")
    val exceptionMessage = intercept[Exception] {
      sql("alter table partitionone drop partition(year=2014,month=1, day=1)")
    }.getMessage
    assert(exceptionMessage.contains("Cannot drop partition as one of the partition"))
    assert(exceptionMessage.contains("p1"))
    sql("drop datamap p1 on table partitionone")
    sql("alter table partitionone drop partition(year=2014,month=1, day=1)")
  }

  test("test drop partition 11") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql(
      "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year")
    sql("insert into partitionone values('k',2014,1,1)")
    val exceptionMessage = intercept[Exception] {
      sql("alter table partitionone_p1 drop partition(partitionone_year=1)")
    }.getMessage
    assert(exceptionMessage.contains("Cannot drop partition directly on aggregate table"))
  }

  test("test drop partition 12") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql(
      "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname")
    sql("insert into partitionone values('k',2014,1,1)")
    val exceptionMessage = intercept[Exception] {
      sql("alter table partitionone_p1 drop partition(year=2014)")
    }.getMessage
    assert(exceptionMessage.contains("operation failed for partition_preaggregate.partitionone_p1: Not a partitioned table"))
  }

  test("test add partition on aggregate table") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql(
      "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname")
    assert(intercept[Exception] {
      sql("alter table partitionone_p1 add partition(c=1)")
    }.getMessage.equals("Cannot add partition directly on non partitioned table"))
  }

  test("test if alter rename is blocked on partition table with preaggregate") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String, id int)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql(
      "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname")
    intercept[Exception] {
      sql("alter table partitionone rename to p")
    }
  }

  test("test dropping partition which has already been deleted") {
    sql("drop table if exists partitiontable")
    sql("create table partitiontable(id int,name string) partitioned by (email string) " +
      "stored by 'carbondata' tblproperties('sort_scope'='global_sort')")
    sql("insert into table partitiontable select 1,'huawei','abc'")
    sql("create datamap ag1 on table partitiontable using 'preaggregate' as select count(email),id" +
      " from partitiontable group by id")
    sql("create datamap ag2 on table partitiontable using 'preaggregate' as select sum(email),name" +
      " from partitiontable group by name")
    sql("create datamap ag3 on table partitiontable using 'preaggregate' as select max(email),name" +
      " from partitiontable group by name")
    sql("create datamap ag4 on table partitiontable using 'preaggregate' as select min(email),name" +
      " from partitiontable group by name")
    sql("create datamap ag5 on table partitiontable using 'preaggregate' as select avg(email),name" +
      " from partitiontable group by name")
    sql("alter table partitiontable add partition (email='def')")
    sql("insert into table partitiontable select 1,'huawei','def'")
    sql("drop datamap ag1 on table partitiontable")
    sql("drop datamap ag2 on table partitiontable")
    sql("drop datamap ag3 on table partitiontable")
    sql("drop datamap ag4 on table partitiontable")
    sql("drop datamap ag5 on table partitiontable")
    sql("alter table partitiontable drop partition(email='def')")
    assert(intercept[Exception] {
      sql("alter table partitiontable drop partition(email='def')")
    }.getMessage.contains("No partition is dropped. One partition spec 'Map(email -> def)' does not exist in table 'partitiontable' database 'partition_preaggregate'"))
  }

  test("test Pre-Aggregate table creation with count(*) on Partition table") {
    sql("drop table if exists partitiontable")
    sql("create table partitiontable(id int,name string) partitioned by (email string) " +
      "stored by 'carbondata' tblproperties('sort_scope'='global_sort')")
    sql("insert into table partitiontable select 1,'huawei','abc'")
    sql("create datamap ag1 on table partitiontable using 'preaggregate' as select count(*),id" +
      " from partitiontable group by id")
    sql("insert into table partitiontable select 1,'huawei','def'")
    assert(sql("show datamap on table partitiontable").collect().head.get(0).toString.equalsIgnoreCase("ag1"))
    sql("drop datamap ag1 on table partitiontable")
  }
  
  test("test blocking partitioning of Pre-Aggregate table") {
    sql("drop table if exists updatetime_8")
    sql("create table updatetime_8" +
      "(countryid smallint,hs_len smallint,minstartdate string,startdate string,newdate string,minnewdate string) partitioned by (imex smallint) stored by 'carbondata' tblproperties('sort_scope'='global_sort','sort_columns'='countryid,imex,hs_len,minstartdate,startdate,newdate,minnewdate','table_blocksize'='256')")
    sql("create datamap ag on table updatetime_8 using 'preaggregate' dmproperties('partitioning'='false') as select imex,sum(hs_len) from updatetime_8 group by imex")
    val carbonTable = CarbonEnv.getCarbonTable(Some("partition_preaggregate"), "updatetime_8_ag")(sqlContext.sparkSession)
    assert(!carbonTable.isHivePartitionTable)
    sql("drop table if exists updatetime_8")
  }

  test("Test data updation after compaction on Partition with Pre-Aggregate tables") {
    sql("drop table if exists partitionallcompaction")
    sql(
      "create table partitionallcompaction(empno int,empname String,designation String," +
      "workgroupcategory int,workgroupcategoryname String,deptno int,projectjoindate timestamp," +
      "projectenddate date,attendance int,utilization int,salary int) partitioned by (deptname " +
      "String,doj timestamp,projectcode int) stored  by 'carbondata' tblproperties" +
      "('sort_scope'='global_sort')")
    sql(
      "create datamap sensor_1 on table partitionallcompaction using 'preaggregate' as select " +
      "sum(salary),doj, deptname,projectcode from partitionallcompaction group by doj," +
      "deptname,projectcode")
    sql(
      s"""LOAD DATA local inpath '$resourcesPath/data.csv' OVERWRITE INTO TABLE
         |partitionallcompaction OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""".stripMargin)
    sql(
      s"""LOAD DATA local inpath '$resourcesPath/data.csv' OVERWRITE INTO TABLE
         |partitionallcompaction OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""".stripMargin)
    sql(
      s"""LOAD DATA local inpath '$resourcesPath/data.csv' OVERWRITE INTO TABLE
         |partitionallcompaction PARTITION(deptname='Learning', doj, projectcode) OPTIONS
         |('DELIMITER'= ',', 'QUOTECHAR'= '"') """.stripMargin)
    sql(
      s"""LOAD DATA local inpath '$resourcesPath/data.csv' OVERWRITE INTO TABLE
         |partitionallcompaction PARTITION(deptname='configManagement', doj, projectcode) OPTIONS
         |('DELIMITER'= ',', 'QUOTECHAR'= '"')""".stripMargin)
    sql(
      s"""LOAD DATA local inpath '$resourcesPath/data.csv' OVERWRITE INTO TABLE
         |partitionallcompaction PARTITION(deptname='network', doj, projectcode) OPTIONS
         |('DELIMITER'= ',', 'QUOTECHAR'= '"')""".stripMargin)
    sql(
      s"""LOAD DATA local inpath '$resourcesPath/data.csv' OVERWRITE INTO TABLE
         |partitionallcompaction PARTITION(deptname='protocol', doj, projectcode) OPTIONS
         |('DELIMITER'= ',', 'QUOTECHAR'= '"')""".stripMargin)
    sql(
      s"""LOAD DATA local inpath '$resourcesPath/data.csv' OVERWRITE INTO TABLE
         |partitionallcompaction PARTITION(deptname='security', doj, projectcode) OPTIONS
         |('DELIMITER'= ',', 'QUOTECHAR'= '"')""".stripMargin)
    sql("ALTER TABLE partitionallcompaction COMPACT 'MINOR'").collect()
    checkAnswer(sql("select count(empno) from partitionallcompaction where empno=14"),
      Seq(Row(5)))
  }

  test("Test data updation in Aggregate query after compaction on Partitioned table with Pre-Aggregate table") {
    sql("drop table if exists updatetime_8")
    sql("create table updatetime_8" +
      "(countryid smallint,hs_len smallint,minstartdate string,startdate string,newdate string,minnewdate string) partitioned by (imex smallint) stored by 'carbondata' tblproperties('sort_scope'='global_sort','sort_columns'='countryid,imex,hs_len,minstartdate,startdate,newdate,minnewdate','table_blocksize'='256')")
    sql("create datamap ag on table updatetime_8 using 'preaggregate' as select sum(hs_len) from updatetime_8 group by imex")
    sql("insert into updatetime_8 select 21,20,'fbv','gbv','wvsw','vwr',23")
    sql("insert into updatetime_8 select 21,20,'fbv','gbv','wvsw','vwr',24")
    sql("insert into updatetime_8 select 21,20,'fbv','gbv','wvsw','vwr',23")
    sql("insert into updatetime_8 select 21,21,'fbv','gbv','wvsw','vwr',24")
    sql("insert into updatetime_8 select 21,21,'fbv','gbv','wvsw','vwr',24")
    sql("insert into updatetime_8 select 21,21,'fbv','gbv','wvsw','vwr',24")
    sql("insert into updatetime_8 select 21,21,'fbv','gbv','wvsw','vwr',25")
    sql("insert into updatetime_8 select 21,21,'fbv','gbv','wvsw','vwr',25")
    sql("alter table updatetime_8 compact 'minor'")
    sql("alter table updatetime_8 compact 'minor'")
    checkAnswer(sql("select sum(hs_len) from updatetime_8 group by imex"),Seq(Row(40),Row(42),Row(83)))
  }

  test("check partitioning for child tables with various combinations") {
    sql("drop table if exists partitionone")
    sql(
      """
        | CREATE TABLE if not exists partitionone (empname String, id int)
        | PARTITIONED BY (year int, month int,day int)
        | STORED BY 'org.apache.carbondata.format'
      """.stripMargin)
    sql(
      "create datamap p7 on table partitionone using 'preaggregate' as select empname, sum(year), sum(day) from partitionone group by empname, year, day")
    sql(
      "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname")
    sql(
      "create datamap p2 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year")
    sql(
      "create datamap p3 on table partitionone using 'preaggregate' as select empname, sum(year), sum(month) from partitionone group by empname, year, month")
    sql(
      "create datamap p4 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
    sql(
      "create datamap p5 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, month")
    sql(
      "create datamap p6 on table partitionone using 'preaggregate' as select empname, sum(year), sum(month) from partitionone group by empname, month, day")
    assert(!CarbonEnv.getCarbonTable(Some("partition_preaggregate"),"partitionone_p1")(sqlContext.sparkSession).isHivePartitionTable)
    assert(CarbonEnv.getCarbonTable(Some("partition_preaggregate"),"partitionone_p2")(sqlContext.sparkSession).getPartitionInfo.getColumnSchemaList.size() == 1)
    assert(CarbonEnv.getCarbonTable(Some("partition_preaggregate"),"partitionone_p3")(sqlContext.sparkSession).getPartitionInfo.getColumnSchemaList.size == 2)
    assert(CarbonEnv.getCarbonTable(Some("partition_preaggregate"),"partitionone_p4")(sqlContext.sparkSession).getPartitionInfo.getColumnSchemaList.size == 3)
    assert(!CarbonEnv.getCarbonTable(Some("partition_preaggregate"),"partitionone_p5")(sqlContext.sparkSession).isHivePartitionTable)
    assert(!CarbonEnv.getCarbonTable(Some("partition_preaggregate"),"partitionone_p6")(sqlContext.sparkSession).isHivePartitionTable)
    assert(!CarbonEnv.getCarbonTable(Some("partition_preaggregate"),"partitionone_p7")(sqlContext.sparkSession).isHivePartitionTable)
  }

  test("test partition at last column") {
    sql("drop table if exists partitionone")
    sql("create table partitionone(a int,b int) partitioned by (c int) stored by 'carbondata'")
    sql("insert into partitionone values(1,2,3)")
    sql("drop datamap if exists dm1")
    sql("create datamap dm1 on table partitionone using 'preaggregate' as select c,sum(b) from partitionone group by c")
    checkAnswer(sql("select c,sum(b) from partitionone group by c"), Seq(Row(3,2)))
    sql("drop table if exists partitionone")
  }

  def preAggTableValidator(plan: LogicalPlan, actualTableName: String) : Unit = {
    var isValidPlan = false
    plan.transform {
      // first check if any preaTable1 scala function is applied it is present is in plan
      // then call is from create preaTable1regate table class so no need to transform the query plan
      case ca:CarbonRelation =>
        if (ca.isInstanceOf[CarbonDatasourceHadoopRelation]) {
          val relation = ca.asInstanceOf[CarbonDatasourceHadoopRelation]
          if(relation.carbonTable.getTableName.equalsIgnoreCase(actualTableName)) {
            isValidPlan = true
          }
        }
        ca
      case logicalRelation:LogicalRelation =>
        if(logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation]) {
          val relation = logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]
          if(relation.carbonTable.getTableName.equalsIgnoreCase(actualTableName)) {
            isValidPlan = true
          }
        }
        logicalRelation
    }
    assert(isValidPlan)
  }

}