content: /*
   * 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.hive;
  
  import java.io.File;
  import java.sql.ResultSet;
  import java.sql.Statement;
  import java.util.ArrayList;
  
  import org.apache.carbondata.core.constants.CarbonCommonConstants;
  import org.apache.carbondata.core.datastore.chunk.impl.DimensionRawColumnChunk;
  import org.apache.carbondata.core.util.CarbonProperties;
  
  import org.apache.carbondata.core.util.CarbonTestUtil;
  import org.junit.AfterClass;
  import org.junit.BeforeClass;
  import org.junit.Test;
  
  public class HiveCarbonTest extends HiveTestUtils {
  
    private static Statement statement;
  
    // resource directory path
    private static String resourceDirectoryPath = HiveCarbonTest.class.getResource("/").getPath()
            +
            "../." +
            "./src/main/resources/";
  
    // "/csv" subdirectory name
    private static final String CSV = "csv";
  
    // "/complex" subdirectory name
    private static final String COMPLEX = "complex";
  
    // "/text" subdirectory name
    private static final String TEXT = "text";
  
    @BeforeClass
    public static void setup() throws Exception {
      CarbonProperties.getInstance().addProperty(CarbonCommonConstants.ENABLE_OFFHEAP_SORT_DEFAULT, "false");
      CarbonProperties.getInstance().addProperty(CarbonCommonConstants.ENABLE_UNSAFE_SORT, "false");
      CarbonProperties.getInstance().addProperty(CarbonCommonConstants.CARBON_WRITTEN_BY_APPNAME, "hive");
      statement = connection.createStatement();
      statement.execute("drop table if exists hive_carbon_table1");
      statement.execute("drop table if exists hive_carbon_table2");
      statement.execute("drop table if exists hive_carbon_table3");
      statement.execute("drop table if exists hive_carbon_table4");
      statement.execute("drop table if exists hive_carbon_table5");
      statement.execute("drop table if exists hive_table");
      statement.execute("drop table if exists hive_table_complex");
      String csvFilePath = (resourceDirectoryPath + CSV).replace("\\", "/");
      String complexFilePath = (resourceDirectoryPath + COMPLEX).replace("\\", "/");
      statement.execute(String.format("CREATE external TABLE hive_table_complex(arrayField  ARRAY<STRING>, mapField MAP<String, String>, structField STRUCT<city: String, pincode: int>) ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.MultiDelimitSerDe' WITH SERDEPROPERTIES ('field.delim'=',', 'collection.delim'='$', 'mapkey.delim'='@') location '%s' TBLPROPERTIES('external.table.purge'='false')", complexFilePath));
      statement.execute(String.format("CREATE external TABLE hive_table(shortField SMALLINT, intField INT, bigintField BIGINT, doubleField DOUBLE, stringField STRING, timestampField TIMESTAMP, decimalField DECIMAL(18,2), dateField DATE, charField CHAR(5), floatField FLOAT) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' location '%s' TBLPROPERTIES ('external.table.purge'='false')", csvFilePath));
    }
  
    @Test
    public void verifyDataAfterLoad() throws Exception {
      statement.execute("drop table if exists hive_carbon_table4");
      statement.execute("CREATE TABLE hive_carbon_table4(shortField SMALLINT , intField INT, bigintField BIGINT , doubleField DOUBLE, stringField STRING, timestampField TIMESTAMP, decimalField DECIMAL(18,2), dateField DATE, charField CHAR(5), floatField FLOAT) stored by 'org.apache.carbondata.hive.CarbonStorageHandler'");
      statement.execute("insert into hive_carbon_table4 select * from hive_table");
      checkAnswer(statement.executeQuery("select * from hive_carbon_table4"),
              connection.createStatement().executeQuery("select * from hive_table"));
    }
  
    @Test
    public void verifyLocalDictionaryValues() throws Exception {
      statement.execute("drop table if exists hive_carbon_table");
      statement.execute("CREATE TABLE hive_carbon_table(shortField SMALLINT , intField INT, bigintField BIGINT , doubleField DOUBLE, stringField STRING, timestampField TIMESTAMP, decimalField DECIMAL(18,2), dateField DATE, charField CHAR(5), floatField FLOAT) stored by 'org.apache.carbondata.hive.CarbonStorageHandler' TBLPROPERTIES ('local_dictionary_enable'='true','local_dictionary_include'='stringField')");
      statement.execute("insert into hive_carbon_table select * from hive_table");
      File rootPath = new File(HiveTestUtils.class.getResource("/").getPath() + "../../../..");
      String storePath = rootPath.getAbsolutePath() + "/integration/hive/target/warehouse/warehouse/hive_carbon_table/";
      ArrayList<DimensionRawColumnChunk> dimRawChunk = CarbonTestUtil.getDimRawChunk(storePath, 0);
      String[] dictionaryData = new String[]{"hive", "impala", "flink", "spark"};
      assert(CarbonTestUtil.validateDictionary(dimRawChunk.get(0), dictionaryData));
    }
  
    @Test
    public void verifyInsertIntoSelectOperation() throws Exception {
      statement.execute("drop table if exists hive_carbon_table1");
      statement.execute("CREATE TABLE hive_carbon_table1(id INT, name STRING, scale DECIMAL, country STRING, salary DOUBLE) stored by 'org.apache.carbondata.hive.CarbonStorageHandler'");
      statement.execute("INSERT into hive_carbon_table1 SELECT 1, 'RAM', '2.3', 'INDIA', 3500");
      statement.execute("INSERT into hive_carbon_table1 SELECT 2, 'RAJU', '2.4', 'RUSSIA', 3600");
      statement.execute("drop table if exists hive_carbon_table2");
      statement.execute("CREATE TABLE hive_carbon_table2(id INT, name STRING, scale DECIMAL, country STRING, salary DOUBLE) stored by 'org.apache.carbondata.hive.CarbonStorageHandler'");
      statement.execute("INSERT into hive_carbon_table2 SELECT * FROM hive_carbon_table1");
      checkAnswer(statement.executeQuery("SELECT * FROM hive_carbon_table2"),
          connection.createStatement().executeQuery("select * from hive_carbon_table1"));
      statement.execute("drop table if exists hive_carbon_table1");
      statement.execute("drop table if exists hive_carbon_table2");
    }
  
    @Test
    public void verifyEmptyTableSelectQuery() throws Exception {
      statement.execute("drop table if exists hive_carbon_table1");
      statement.execute("CREATE TABLE hive_carbon_table1(id INT, name STRING, scale DECIMAL, country STRING, salary DOUBLE) stored by 'org.apache.carbondata.hive.CarbonStorageHandler'");
      statement.execute("drop table if exists hive_carbon_table2");
      statement.execute("CREATE TABLE hive_carbon_table2(id INT, name STRING, scale DECIMAL, country STRING, salary DOUBLE) stored by 'org.apache.carbondata.hive.CarbonStorageHandler'");
      statement.execute("INSERT into hive_carbon_table2 SELECT * FROM hive_carbon_table1");
      checkAnswer(statement.executeQuery("SELECT * FROM hive_carbon_table2"),
          connection.createStatement().executeQuery("select * from hive_carbon_table1"));
      statement.execute("drop table if exists hive_carbon_table1");
      statement.execute("drop table if exists hive_carbon_table2");
    }
  
    @Test
    public void verifyDataAfterLoadUsingSortColumns() throws Exception {
      statement.execute("drop table if exists hive_carbon_table5");
      statement.execute(
          "CREATE TABLE hive_carbon_table5(shortField SMALLINT , intField INT, bigintField BIGINT , doubleField DOUBLE, stringField STRING, timestampField TIMESTAMP, decimalField DECIMAL(18,2), dateField DATE, charField CHAR(5), floatField FLOAT) stored by 'org.apache.carbondata.hive.CarbonStorageHandler'");
      statement.execute("insert into hive_carbon_table5 select * from hive_table");
      ResultSet resultSet = connection.createStatement()
          .executeQuery("select * from hive_carbon_table5");
      ResultSet hiveResults = connection.createStatement()
          .executeQuery("select * from hive_table");
      checkAnswer(resultSet, hiveResults);
    }
  
    @Test
    public void testCreateAndLoadUsingComplexColumns() throws Exception {
      statement.execute("drop table if exists hive_carbon_table6");
      statement.execute(
          "CREATE TABLE hive_carbon_table6(arrayField  ARRAY<STRING>, mapField MAP<String, String>, structField STRUCT< city: String, pincode: int >) stored by 'org.apache.carbondata.hive.CarbonStorageHandler' TBLPROPERTIES ('complex_delimiter'='$,@')");
      statement.execute(
          "insert into hive_carbon_table6 select * from hive_table_complex");
      ResultSet hiveResult = connection.createStatement().executeQuery("select * from hive_table_complex where mapField['Key1']='Val1'");
      ResultSet carbonResult = connection.createStatement().executeQuery("select * from hive_carbon_table6 where mapField['Key1']='Val1'");
      checkAnswer(carbonResult, hiveResult);
    }
  
    @Test
    public void testArrayType() throws Exception {
      String complexArrayPath = (resourceDirectoryPath + "array").replace("\\", "/");
      statement.execute("drop table if exists hive_table_complexArray");
      statement.execute(String.format("CREATE external TABLE hive_table_complexArray(arrayString  ARRAY<STRING>,"
          + " arrayShort ARRAY<SMALLINT>, arrayInt ARRAY<INT>, arrayLong ARRAY<BIGINT>, arrayFloat ARRAY<FLOAT>,"
          + " arrayDouble ARRAY<DOUBLE>, arrayDecimal ARRAY<DECIMAL(8,2)>, arrayChar ARRAY<CHAR(5)>, "
          + "arrayBoolean ARRAY<BOOLEAN>, arrayVarchar ARRAY<VARCHAR(50)>, arrayByte ARRAY<TINYINT>, arrayDate ARRAY<DATE>)"
          + " ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.MultiDelimitSerDe' WITH SERDEPROPERTIES ('field.delim'=',', 'collection.delim'='$', 'mapkey.delim'='@') location '%s' TBLPROPERTIES('external.table.purge'='false')", complexArrayPath));
  
      statement.execute("drop table if exists hive_carbon_table7");
      statement.execute(
          "CREATE TABLE hive_carbon_table7(arrayString  ARRAY<STRING>, arrayShort ARRAY<SMALLINT>, arrayInt ARRAY<INT>, "
              + "arrayLong ARRAY<BIGINT>, arrayFloat ARRAY<FLOAT>, arrayDouble ARRAY<DOUBLE>, "
              + "arrayDecimal ARRAY<DECIMAL(8,2)>, arrayChar ARRAY<CHAR(5)>, arrayBoolean ARRAY<BOOLEAN>, arrayVarchar ARRAY<VARCHAR(50)>, "
              + "arrayByte ARRAY<TINYINT>, arrayDate ARRAY<DATE>) "
              + "stored by 'org.apache.carbondata.hive.CarbonStorageHandler' TBLPROPERTIES ('complex_delimiter'='$,@')");
  
      statement.execute(
          "insert into hive_carbon_table7 select * from hive_table_complexArray");
  
      ResultSet hiveResult = connection.createStatement().executeQuery("select * from hive_table_complexArray");
      ResultSet carbonResult = connection.createStatement().executeQuery("select * from hive_carbon_table7");
      checkAnswer(carbonResult, hiveResult);
    }
  
  
    @Test
    public void arrayOfTimestamp() throws Exception {
      File rootPath = new File(HiveTestUtils.class.getResource("/").getPath() + "../../../..");
      String tablePath = rootPath.getAbsolutePath() + "/integration/hive/target/hivee";
      statement.execute("drop table if exists hivee");
      statement.execute("CREATE external TABLE hivee(arrayInt ARRAY<timestamp>)"
          + " ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.MultiDelimitSerDe' WITH SERDEPROPERTIES ('field.delim'=',', 'collection.delim'='$', 'mapkey.delim'='@') location '"+ tablePath +"' TBLPROPERTIES('external.table.purge'='false')");
      statement.execute("insert into table hivee values (array(Timestamp('2000-03-12 15:00:00'),Timestamp('2001-04-15 15:58:00'),Timestamp('2002-05-27 15:20:00')))");
      statement.execute("drop table if exists carbonn");
      statement.execute(
          "CREATE TABLE carbonn(timestampField array<timestamp>) "
              + "stored by 'org.apache.carbondata.hive.CarbonStorageHandler' TBLPROPERTIES ('complex_delimiter'='$,@', 'BAD_RECORDS_LOGGER_ENABLE' = 'TRUE')");
      statement.execute("insert into carbonn select * from hivee");
      ResultSet resultSet = connection.createStatement()
          .executeQuery("select * from carbonn");
      ResultSet hiveResults = connection.createStatement()
          .executeQuery("select * from hivee");
     checkAnswer(resultSet, hiveResults);
    }
  
    @Test
    public void testMapType() throws Exception {
      String complexMapPath = (resourceDirectoryPath + "map").replace("\\", "/");
      statement.execute("drop table if exists hive_table_complexMap");
      statement.execute(String.format("CREATE external TABLE hive_table_complexMap(mapField1 MAP<STRING, STRING>, mapField2 MAP<INT, STRING>, mapField3 MAP<DOUBLE, "
          + "FLOAT>,mapField4 MAP<TINYINT, SMALLINT>, mapField5 MAP<DECIMAL(10,2), VARCHAR(50)>, mapField6 MAP<BIGINT,DATE>,mapField7 MAP<CHAR(5),INT>, "
          + "mapField8 MAP<BIGINT,BOOLEAN>)"
          + " ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.MultiDelimitSerDe' WITH SERDEPROPERTIES ('field.delim'=',', 'collection.delim'='$', 'mapkey.delim'='@') location '%s' TBLPROPERTIES('external.table.purge'='false')", complexMapPath));
  
      statement.execute("drop table if exists hive_carbon_table8");
      statement.execute(
          "CREATE TABLE hive_carbon_table8(mapField1 MAP<STRING, STRING>, mapField2 MAP<INT, STRING>, mapField3 MAP<DOUBLE, FLOAT>,mapField4 MAP<TINYINT, SMALLINT>, "
              + "mapField5 MAP<DECIMAL(10,2), VARCHAR(50)>, mapField6 MAP<BIGINT, DATE>,mapField7 MAP<CHAR(5), INT>, mapField8 MAP<BIGINT,BOOLEAN>) "
              + "stored by 'org.apache.carbondata.hive.CarbonStorageHandler' TBLPROPERTIES ('complex_delimiter'='$,@')");
      statement.execute(
          "insert into hive_carbon_table8 select * from hive_table_complexMap");
      ResultSet hiveResult = connection.createStatement().executeQuery("select * from hive_table_complexMap");
      ResultSet carbonResult = connection.createStatement().executeQuery("select * from hive_carbon_table8");
      checkAnswer(carbonResult, hiveResult);
    }
  
    @Test
    public void testStructType() throws Exception {
      String complexStructPath = (resourceDirectoryPath + "struct").replace("\\", "/");
      statement.execute("drop table if exists hive_table_complexSTRUCT");
      statement.execute(String.format("CREATE external TABLE hive_table_complexSTRUCT(structField STRUCT<stringfield: STRING, shortfield: SMALLINT, intfield: INT, "
          + "longfield: BIGINT, floatfield: FLOAT, doublefield: DOUBLE, charfield: CHAR(4), boolfield: BOOLEAN, varcharfield: VARCHAR(50), bytefield: TINYINT, "
          + "datefield: DATE, decimalfield: DECIMAL(8,2)>) "
          + "ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.MultiDelimitSerDe' WITH SERDEPROPERTIES "
          + "('field.delim'=',', 'collection.delim'='$', 'mapkey.delim'='@') location '%s' TBLPROPERTIES('external.table.purge'='false')", complexStructPath));
  
      statement.execute("drop table if exists hive_carbon_table9");
      statement.execute(
          "CREATE TABLE hive_carbon_table9(structField STRUCT<stringfield: STRING, shortfield: SMALLINT, intfield: INT, longfield: BIGINT, floatfield: FLOAT, "
              + "doublefield: DOUBLE, charfield: CHAR(4), boolfield: BOOLEAN, varcharfield: VARCHAR(50), bytefield: TINYINT, datefield: DATE, decimalfield: DECIMAL(8,2)>) "
              + "stored by 'org.apache.carbondata.hive.CarbonStorageHandler' TBLPROPERTIES ('complex_delimiter'='$,@')");
      statement.execute(
          "insert into hive_carbon_table9 select * from hive_table_complexSTRUCT");
      ResultSet hiveResult = connection.createStatement().executeQuery("select * from hive_table_complexSTRUCT");
      ResultSet carbonResult = connection.createStatement().executeQuery("select * from hive_carbon_table9");
      checkAnswer(carbonResult, hiveResult);
    }
  
    @Test
    public void testBinaryDataTypeColumns() throws Exception {
      String dataPath = resourceDirectoryPath + TEXT + "/string.txt";
      statement.execute("drop table if exists mytable_n2");
      statement.execute("drop table if exists hive_carbon_table9");
      statement.execute("drop table if exists mytable_n21");
      statement.execute("CREATE TABLE mytable_n2(key binary, value int) ROW FORMAT DELIMITED FIELDS TERMINATED BY '9'");
      statement.execute("LOAD DATA LOCAL INPATH '" + dataPath + "' INTO TABLE mytable_n2");
      statement.execute("CREATE TABLE mytable_n21(key binary, value int) ROW FORMAT DELIMITED FIELDS TERMINATED BY '9'");
      statement.execute("insert into mytable_n21 select * from mytable_n2");
      statement.execute(
          "CREATE TABLE hive_carbon_table9(key binary, value int) "
              + "stored by 'org.apache.carbondata.hive.CarbonStorageHandler'");
      statement.execute("insert into hive_carbon_table9 select * from mytable_n2");
      ResultSet hiveResult = connection.createStatement().executeQuery("select * from mytable_n2");
      ResultSet carbonResult = connection.createStatement().executeQuery("select * from hive_carbon_table9");
      checkAnswer(carbonResult, hiveResult);
    }
  
    @AfterClass
    public static void tearDown() {
      try {
        connection.close();
        hiveEmbeddedServer2.stop();
      } catch (Exception e) {
        throw new RuntimeException("Unable to close Hive Embedded Server", e);
      }
    }
  }
  