hive/src/test/java/org/apache/apex/malhar/hive/HiveMockTest.java

/**
 * 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.apex.malhar.hive;

import java.io.File;
import java.io.IOException;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Properties;

import org.junit.Assert;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.apex.malhar.hive.AbstractFSRollingOutputOperator.FilePartitionMapping;
import org.apache.apex.malhar.hive.FSPojoToHiveOperator.FIELD_TYPE;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.thrift.TException;

import io.teknek.hiveunit.HiveTestService;

import static org.apache.apex.malhar.lib.helper.OperatorContextTestHelper.mockOperatorContext;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.serializers.FieldSerializer;

import com.datatorrent.api.Attribute.AttributeMap;
import com.datatorrent.api.Context.OperatorContext;
import com.datatorrent.api.DAG;
import com.datatorrent.api.Operator.ProcessingMode;

public class HiveMockTest extends HiveTestService
{
  public static final String APP_ID = "HiveOperatorTest";

  public static final int OPERATOR_ID = 0;
  public static final int NUM_WINDOWS = 10;
  public static final int BLAST_SIZE = 10;
  public static final int DATABASE_SIZE = NUM_WINDOWS * BLAST_SIZE;
  public static final String tablename = "temp";
  public static final String tablepojo = "temppojo";
  public static final String tablemap = "tempmap";
  public static String delimiterMap = ":";
  public static final String HOST = "localhost";
  public static final String PORT = "10000";
  public static final String DATABASE = "default";
  public static final String HOST_PREFIX = "jdbc:hive://";

  public String testdir;

  private String getDir()
  {
    String filePath = new File("target/hive").getAbsolutePath();
    LOG.info("filepath is {}", filePath);
    return filePath;
  }

  @Override
  public void setUp() throws Exception
  {
    super.setUp();
  }

  private static final int NUMBER_OF_TESTS = 4; // count your tests here
  private int sTestsRun = 0;

  @Override
  public void tearDown() throws Exception
  {
    super.tearDown();
    sTestsRun++;
    //Equivalent of @AfterClass in junit 3.
    if (sTestsRun == NUMBER_OF_TESTS) {
      FileUtils.deleteQuietly(new File(testdir));
    }
  }

  public HiveMockTest() throws IOException, Exception
  {
    super();
    testdir = getDir();
    Properties properties = System.getProperties();
    properties.put("derby.system.home", testdir);
    System.setProperty(HiveConf.ConfVars.METASTOREWAREHOUSE.toString(), testdir);
    new File(testdir).mkdir();
  }

  public static HiveStore createStore(HiveStore hiveStore)
  {
    String host = HOST;
    String port = PORT;

    if (hiveStore == null) {
      hiveStore = new HiveStore();
    }

    StringBuilder sb = new StringBuilder();
    String tempHost = HOST_PREFIX + host + ":" + port;

    LOG.debug("Host name: {}", tempHost);
    LOG.debug("Port: {}", 10000);

    sb.append("user:").append("").append(",");
    sb.append("password:").append("");

    String properties = sb.toString();
    LOG.debug(properties);
    hiveStore.setDatabaseDriver("org.apache.hive.jdbc.HiveDriver");

    hiveStore.setDatabaseUrl("jdbc:hive2://");
    hiveStore.setConnectionProperties(properties);
    return hiveStore;
  }

  public static void hiveInitializeDatabase(HiveStore hiveStore) throws SQLException
  {
    hiveStore.connect();
    Statement stmt = hiveStore.getConnection().createStatement();
    // show tables
    String sql = "show tables";

    LOG.debug(sql);
    ResultSet res = stmt.executeQuery(sql);
    if (res.next()) {
      LOG.debug("tables are {}", res.getString(1));
    }

    stmt.execute("DROP TABLE " + tablename);

    stmt.execute("CREATE TABLE IF NOT EXISTS " + tablename + " (col1 String) PARTITIONED BY(dt STRING) "
        + "ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LINES TERMINATED BY '\n'  \n" + "STORED AS TEXTFILE ");
    /*ResultSet res = stmt.execute("CREATE TABLE IF NOT EXISTS temp4 (col1 map<string,int>,col2 map<string,int>,col3  map<string,int>,col4 map<String,timestamp>, col5 map<string,double>,col6 map<string,double>,col7 map<string,int>,col8 map<string,int>) ROW FORMAT DELIMITED FIELDS TERMINATED BY ','  \n"
     + "COLLECTION ITEMS TERMINATED BY '\n'  \n"
     + "MAP KEYS TERMINATED BY ':'  \n"
     + "LINES TERMINATED BY '\n' "
     + "STORED AS TEXTFILE");*/

    hiveStore.disconnect();
  }

  public static void hiveInitializePOJODatabase(HiveStore hiveStore) throws SQLException
  {
    hiveStore.connect();
    Statement stmt = hiveStore.getConnection().createStatement();
    // show tables
    String sql = "show tables";

    LOG.debug(sql);
    ResultSet res = stmt.executeQuery(sql);
    if (res.next()) {
      LOG.debug("tables are {}", res.getString(1));
    }
    stmt.execute("DROP TABLE " + tablepojo);

    stmt.execute("CREATE TABLE " + tablepojo + " (col1 int) PARTITIONED BY(dt STRING) ROW FORMAT DELIMITED "
        + "FIELDS TERMINATED BY '\t' LINES TERMINATED BY '\n' \n" + "STORED AS TEXTFILE ");
    hiveStore.disconnect();
  }

  public static void hiveInitializeMapDatabase(HiveStore hiveStore) throws SQLException
  {
    hiveStore.connect();
    Statement stmt = hiveStore.getConnection().createStatement();
    // show tables
    String sql = "show tables";

    LOG.debug(sql);
    ResultSet res = stmt.executeQuery(sql);
    if (res.next()) {
      LOG.debug(res.getString(1));
    }

    stmt.execute("DROP TABLE " + tablemap);

    stmt.execute("CREATE TABLE IF NOT EXISTS " + tablemap
        + " (col1 map<string,int>) PARTITIONED BY(dt STRING) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'  \n"
        + "MAP KEYS TERMINATED BY '" + delimiterMap + "' \n" + "STORED AS TEXTFILE ");
    hiveStore.disconnect();
  }

  @Test
  public void testInsertString() throws Exception
  {
    HiveStore hiveStore = createStore(null);
    hiveStore.setFilepath(testdir);
    ArrayList<String> hivePartitionColumns = new ArrayList<String>();
    hivePartitionColumns.add("dt");
    hiveInitializeDatabase(createStore(null));
    HiveOperator hiveOperator = new HiveOperator();
    hiveOperator.setStore(hiveStore);
    hiveOperator.setTablename(tablename);
    hiveOperator.setHivePartitionColumns(hivePartitionColumns);

    FSRollingTestImpl fsRolling = new FSRollingTestImpl();
    fsRolling.setFilePath(testdir);
    short permission = 511;
    fsRolling.setFilePermission(permission);
    fsRolling.setAlwaysWriteToTmp(false);
    fsRolling.setMaxLength(128);
    AttributeMap.DefaultAttributeMap attributeMap = new AttributeMap.DefaultAttributeMap();
    attributeMap.put(OperatorContext.PROCESSING_MODE, ProcessingMode.AT_LEAST_ONCE);
    attributeMap.put(OperatorContext.ACTIVATION_WINDOW_ID, -1L);
    attributeMap.put(DAG.APPLICATION_ID, APP_ID);
    OperatorContext context = mockOperatorContext(OPERATOR_ID, attributeMap);

    fsRolling.setup(context);
    hiveOperator.setup(context);
    FilePartitionMapping mapping1 = new FilePartitionMapping();
    FilePartitionMapping mapping2 = new FilePartitionMapping();
    mapping1.setFilename(APP_ID + "/" + OPERATOR_ID + "/" + "2014-12-10" + "/" + "0-transaction.out.part.0");
    ArrayList<String> partitions1 = new ArrayList<String>();
    partitions1.add("2014-12-10");
    mapping1.setPartition(partitions1);
    ArrayList<String> partitions2 = new ArrayList<String>();
    partitions2.add("2014-12-11");
    mapping2.setFilename(APP_ID + "/" + OPERATOR_ID + "/" + "2014-12-11" + "/" + "0-transaction.out.part.0");
    mapping2.setPartition(partitions2);
    for (int wid = 0, total = 0; wid < NUM_WINDOWS; wid++) {
      fsRolling.beginWindow(wid);
      for (int tupleCounter = 0; tupleCounter < BLAST_SIZE && total < DATABASE_SIZE; tupleCounter++, total++) {
        fsRolling.input.process("2014-12-1" + tupleCounter);
      }
      if (wid == 7) {
        fsRolling.committed(wid - 1);
        hiveOperator.processTuple(mapping1);
        hiveOperator.processTuple(mapping2);
      }

      fsRolling.endWindow();

      if (wid == 6) {
        fsRolling.beforeCheckpoint(wid);
        fsRolling.checkpointed(wid);
      }
    }

    fsRolling.teardown();
    hiveStore.connect();
    client.execute("select * from " + tablename + " where dt='2014-12-10'");
    List<String> recordsInDatePartition1 = client.fetchAll();

    client.execute("select * from " + tablename + " where dt='2014-12-11'");
    List<String> recordsInDatePartition2 = client.fetchAll();
    client.execute("drop table " + tablename);
    hiveStore.disconnect();

    Assert.assertEquals(7, recordsInDatePartition1.size());
    for (int i = 0; i < recordsInDatePartition1.size(); i++) {
      LOG.debug("records in first date partition are {}", recordsInDatePartition1.get(i));
      /*An array containing partition and data is returned as a string record, hence we need to upcast it to an object first
       and then downcast to a string in order to use in Assert.*/
      Object record = recordsInDatePartition1.get(i);
      Object[] records = (Object[])record;
      Assert.assertEquals("2014-12-10", records[1]);
    }
    Assert.assertEquals(7, recordsInDatePartition2.size());
    for (int i = 0; i < recordsInDatePartition2.size(); i++) {
      LOG.debug("records in second date partition are {}", recordsInDatePartition2.get(i));
      Object record = recordsInDatePartition2.get(i);
      Object[] records = (Object[])record;
      Assert.assertEquals("2014-12-11", records[1]);
    }
  }

  @Test
  public void testInsertPOJO() throws Exception
  {
    HiveStore hiveStore = createStore(null);
    hiveStore.setFilepath(testdir);
    ArrayList<String> hivePartitionColumns = new ArrayList<String>();
    hivePartitionColumns.add("dt");
    ArrayList<String> hiveColumns = new ArrayList<String>();
    hiveColumns.add("col1");
    hiveInitializePOJODatabase(createStore(null));
    HiveOperator hiveOperator = new HiveOperator();
    hiveOperator.setStore(hiveStore);
    hiveOperator.setTablename(tablepojo);
    hiveOperator.setHivePartitionColumns(hivePartitionColumns);

    FSPojoToHiveOperator fsRolling = new FSPojoToHiveOperator();
    fsRolling.setFilePath(testdir);
    fsRolling.setHiveColumns(hiveColumns);
    ArrayList<FIELD_TYPE> fieldtypes = new ArrayList<FIELD_TYPE>();
    ArrayList<FIELD_TYPE> partitiontypes = new ArrayList<FIELD_TYPE>();
    fieldtypes.add(FIELD_TYPE.INTEGER);
    partitiontypes.add(FIELD_TYPE.STRING);
    fsRolling.setHiveColumnDataTypes(fieldtypes);
    fsRolling.setHivePartitionColumnDataTypes(partitiontypes);
    //ArrayList<FIELD_TYPE> partitionColumnType = new ArrayList<FIELD_TYPE>();
    //partitionColumnType.add(FIELD_TYPE.STRING);
    fsRolling.setHivePartitionColumns(hivePartitionColumns);
    // fsRolling.setHivePartitionColumnsDataTypes(partitionColumnType);
    ArrayList<String> expressions = new ArrayList<String>();
    expressions.add("getId()");
    ArrayList<String> expressionsPartitions = new ArrayList<String>();

    expressionsPartitions.add("getDate()");
    short permission = 511;
    fsRolling.setFilePermission(permission);
    fsRolling.setAlwaysWriteToTmp(false);
    fsRolling.setMaxLength(128);
    fsRolling.setExpressionsForHiveColumns(expressions);
    fsRolling.setExpressionsForHivePartitionColumns(expressionsPartitions);
    AttributeMap.DefaultAttributeMap attributeMap = new AttributeMap.DefaultAttributeMap();
    attributeMap.put(OperatorContext.PROCESSING_MODE, ProcessingMode.AT_LEAST_ONCE);
    attributeMap.put(OperatorContext.ACTIVATION_WINDOW_ID, -1L);
    attributeMap.put(DAG.APPLICATION_ID, APP_ID);
    OperatorContext context = mockOperatorContext(OPERATOR_ID, attributeMap);

    fsRolling.setup(context);
    hiveOperator.setup(context);
    FilePartitionMapping mapping1 = new FilePartitionMapping();
    FilePartitionMapping mapping2 = new FilePartitionMapping();
    mapping1.setFilename(APP_ID + "/" + OPERATOR_ID + "/" + "2014-12-11" + "/" + "0-transaction.out.part.0");
    ArrayList<String> partitions1 = new ArrayList<String>();
    partitions1.add("2014-12-11");
    mapping1.setPartition(partitions1);
    ArrayList<String> partitions2 = new ArrayList<String>();
    partitions2.add("2014-12-12");
    mapping2.setFilename(APP_ID + "/" + OPERATOR_ID + "/" + "2014-12-12" + "/" + "0-transaction.out.part.0");
    mapping2.setPartition(partitions2);
    for (int wid = 0, total = 0; wid < NUM_WINDOWS; wid++) {
      fsRolling.beginWindow(wid);
      for (int tupleCounter = 1; tupleCounter < BLAST_SIZE && total < DATABASE_SIZE; tupleCounter++, total++) {
        InnerObj innerObj = new InnerObj();
        innerObj.setId(tupleCounter);
        innerObj.setDate("2014-12-1" + tupleCounter);
        fsRolling.input.process(innerObj);
      }
      if (wid == 7) {
        fsRolling.committed(wid - 1);
        hiveOperator.processTuple(mapping1);
        hiveOperator.processTuple(mapping2);
      }

      fsRolling.endWindow();

      if (wid == 6) {
        fsRolling.beforeCheckpoint(wid);
        fsRolling.checkpointed(wid);
      }
    }

    fsRolling.teardown();
    hiveStore.connect();
    client.execute("select * from " + tablepojo + " where dt='2014-12-11'");
    List<String> recordsInDatePartition1 = client.fetchAll();

    client.execute("select * from " + tablepojo + " where dt='2014-12-12'");
    List<String> recordsInDatePartition2 = client.fetchAll();
    client.execute("drop table " + tablepojo);
    hiveStore.disconnect();

    Assert.assertEquals(7, recordsInDatePartition1.size());
    for (int i = 0; i < recordsInDatePartition1.size(); i++) {
      LOG.debug("records in first date partition are {}", recordsInDatePartition1.get(i));
      /*An array containing partition and data is returned as a string record, hence we need to upcast it to an object first
       and then downcast to a string in order to use in Assert.*/
      Object record = recordsInDatePartition1.get(i);
      Object[] records = (Object[])record;
      Assert.assertEquals(1, records[0]);
      Assert.assertEquals("2014-12-11", records[1]);
    }
    Assert.assertEquals(7, recordsInDatePartition2.size());
    for (int i = 0; i < recordsInDatePartition2.size(); i++) {
      LOG.debug("records in second date partition are {}", recordsInDatePartition2.get(i));
      Object record = recordsInDatePartition2.get(i);
      Object[] records = (Object[])record;
      Assert.assertEquals(2, records[0]);
      Assert.assertEquals("2014-12-12", records[1]);
    }
  }

  @Test
  public void testHiveInsertMapOperator() throws SQLException, TException
  {
    HiveStore hiveStore = createStore(null);
    hiveStore.setFilepath(testdir);
    ArrayList<String> hivePartitionColumns = new ArrayList<String>();
    hivePartitionColumns.add("dt");
    hiveInitializeMapDatabase(createStore(null));
    HiveOperator hiveOperator = new HiveOperator();
    hiveOperator.setStore(hiveStore);
    hiveOperator.setTablename(tablemap);
    hiveOperator.setHivePartitionColumns(hivePartitionColumns);

    FSRollingMapTestImpl fsRolling = new FSRollingMapTestImpl();
    fsRolling.setFilePath(testdir);
    short permission = 511;
    fsRolling.setFilePermission(permission);
    fsRolling.setAlwaysWriteToTmp(false);
    fsRolling.setMaxLength(128);
    AttributeMap.DefaultAttributeMap attributeMap = new AttributeMap.DefaultAttributeMap();
    attributeMap.put(OperatorContext.PROCESSING_MODE, ProcessingMode.AT_LEAST_ONCE);
    attributeMap.put(OperatorContext.ACTIVATION_WINDOW_ID, -1L);
    attributeMap.put(DAG.APPLICATION_ID, APP_ID);
    OperatorContext context = mockOperatorContext(OPERATOR_ID, attributeMap);

    fsRolling.setup(context);
    hiveOperator.setup(context);
    HashMap<String, Object> map = new HashMap<String, Object>();
    FilePartitionMapping mapping1 = new FilePartitionMapping();
    FilePartitionMapping mapping2 = new FilePartitionMapping();
    ArrayList<String> partitions1 = new ArrayList<String>();
    partitions1.add("2014-12-10");
    mapping1.setFilename(APP_ID + "/" + OPERATOR_ID + "/" + "2014-12-10" + "/" + "0-transaction.out.part.0");
    mapping1.setPartition(partitions1);
    ArrayList<String> partitions2 = new ArrayList<String>();
    partitions2.add("2014-12-11");
    mapping2.setFilename(APP_ID + "/" + OPERATOR_ID + "/" + "2014-12-11" + "/" + "0-transaction.out.part.0");
    mapping2.setPartition(partitions2);
    for (int wid = 0; wid < NUM_WINDOWS; wid++) {
      fsRolling.beginWindow(wid);
      for (int tupleCounter = 0; tupleCounter < BLAST_SIZE; tupleCounter++) {
        map.put(2014 - 12 - 10 + "", 2014 - 12 - 10);
        fsRolling.input.put(map);
        map.clear();
      }

      if (wid == 7) {
        fsRolling.committed(wid - 1);
        hiveOperator.processTuple(mapping1);
        hiveOperator.processTuple(mapping2);
      }

      fsRolling.endWindow();
    }

    fsRolling.teardown();

    hiveStore.connect();

    client.execute("select * from " + tablemap + " where dt='2014-12-10'");
    List<String> recordsInDatePartition1 = client.fetchAll();

    client.execute("drop table " + tablemap);
    hiveStore.disconnect();

    Assert.assertEquals(13, recordsInDatePartition1.size());
    for (int i = 0; i < recordsInDatePartition1.size(); i++) {
      LOG.debug("records in first date partition are {}", recordsInDatePartition1.get(i));
      /*An array containing partition and data is returned as a string record, hence we need to upcast it to an object first
       and then downcast to a string in order to use in Assert.*/
      Object record = recordsInDatePartition1.get(i);
      Object[] records = (Object[])record;
      Assert.assertEquals("2014-12-10", records[1]);
    }

  }

  @Test
  public void testHDFSHiveCheckpoint() throws SQLException, TException
  {
    hiveInitializeDatabase(createStore(null));
    HiveStore hiveStore = createStore(null);
    hiveStore.setFilepath(testdir);
    HiveOperator outputOperator = new HiveOperator();
    HiveOperator newOp;
    outputOperator.setStore(hiveStore);
    ArrayList<String> hivePartitionColumns = new ArrayList<String>();
    hivePartitionColumns.add("dt");
    FSRollingTestImpl fsRolling = new FSRollingTestImpl();
    hiveInitializeDatabase(createStore(null));
    outputOperator.setHivePartitionColumns(hivePartitionColumns);
    outputOperator.setTablename(tablename);
    fsRolling.setFilePath(testdir);
    short persmission = 511;
    fsRolling.setFilePermission(persmission);
    fsRolling.setAlwaysWriteToTmp(false);
    fsRolling.setMaxLength(128);
    AttributeMap.DefaultAttributeMap attributeMap = new AttributeMap.DefaultAttributeMap();
    attributeMap.put(OperatorContext.PROCESSING_MODE, ProcessingMode.AT_LEAST_ONCE);
    attributeMap.put(OperatorContext.ACTIVATION_WINDOW_ID, -1L);
    attributeMap.put(DAG.APPLICATION_ID, APP_ID);
    OperatorContext context = mockOperatorContext(OPERATOR_ID, attributeMap);

    fsRolling.setup(context);

    FilePartitionMapping mapping1 = new FilePartitionMapping();
    FilePartitionMapping mapping2 = new FilePartitionMapping();
    FilePartitionMapping mapping3 = new FilePartitionMapping();
    outputOperator.setup(context);
    mapping1.setFilename(APP_ID + "/" + OPERATOR_ID + "/" + "2014-12-10" + "/" + "0-transaction.out.part.0");
    ArrayList<String> partitions1 = new ArrayList<String>();
    partitions1.add("2014-12-10");
    mapping1.setPartition(partitions1);
    mapping2.setFilename(APP_ID + "/" + OPERATOR_ID + "/" + "2014-12-11" + "/" + "0-transaction.out.part.0");
    ArrayList<String> partitions2 = new ArrayList<String>();
    partitions2.add("2014-12-11");
    mapping2.setPartition(partitions2);
    ArrayList<String> partitions3 = new ArrayList<String>();
    partitions3.add("2014-12-12");
    mapping3.setFilename(APP_ID + "/" + OPERATOR_ID + "/" + "2014-12-12" + "/" + "0-transaction.out.part.0");
    mapping3.setPartition(partitions3);
    for (int wid = 0, total = 0; wid < NUM_WINDOWS; wid++) {
      fsRolling.beginWindow(wid);
      for (int tupleCounter = 0; tupleCounter < BLAST_SIZE && total < DATABASE_SIZE; tupleCounter++, total++) {
        fsRolling.input.process("2014-12-1" + tupleCounter);
      }
      if (wid == 7) {
        fsRolling.committed(wid - 1);
        outputOperator.processTuple(mapping1);
        outputOperator.processTuple(mapping2);
      }

      fsRolling.endWindow();

      if ((wid == 6) || (wid == 9)) {
        fsRolling.beforeCheckpoint(wid);
        fsRolling.checkpointed(wid);
      }

      if (wid == 9) {
        Kryo kryo = new Kryo();
        FieldSerializer<HiveOperator> f1 = (FieldSerializer<HiveOperator>)kryo.getSerializer(HiveOperator.class);
        FieldSerializer<HiveStore> f2 = (FieldSerializer<HiveStore>)kryo.getSerializer(HiveStore.class);

        f1.setCopyTransient(false);
        f2.setCopyTransient(false);
        newOp = kryo.copy(outputOperator);
        outputOperator.teardown();
        newOp.setup(context);

        newOp.beginWindow(7);
        newOp.processTuple(mapping3);

        newOp.endWindow();
        newOp.teardown();
        break;
      }

    }

    hiveStore.connect();

    client.execute("select * from " + tablename + " where dt='2014-12-10'");
    List<String> recordsInDatePartition1 = client.fetchAll();

    client.execute("select * from " + tablename + " where dt='2014-12-11'");
    List<String> recordsInDatePartition2 = client.fetchAll();

    client.execute("select * from " + tablename + " where dt='2014-12-12'");
    List<String> recordsInDatePartition3 = client.fetchAll();

    client.execute("drop table " + tablename);
    hiveStore.disconnect();

    Assert.assertEquals(7, recordsInDatePartition1.size());
    for (int i = 0; i < recordsInDatePartition1.size(); i++) {
      LOG.debug("records in first date partition are {}", recordsInDatePartition1.get(i));
      /*An array containing partition and data is returned as a string record, hence we need to upcast it to an object first
       and then downcast to a string in order to use in Assert.*/
      Object record = recordsInDatePartition1.get(i);
      Object[] records = (Object[])record;
      Assert.assertEquals("2014-12-10", records[1]);
    }
    Assert.assertEquals(7, recordsInDatePartition2.size());
    for (int i = 0; i < recordsInDatePartition2.size(); i++) {
      LOG.debug("records in second date partition are {}", recordsInDatePartition2.get(i));
      Object record = recordsInDatePartition2.get(i);
      Object[] records = (Object[])record;
      Assert.assertEquals("2014-12-11", records[1]);
    }
    Assert.assertEquals(10, recordsInDatePartition3.size());
    for (int i = 0; i < recordsInDatePartition3.size(); i++) {
      LOG.debug("records in second date partition are {}", recordsInDatePartition3.get(i));
      Object record = recordsInDatePartition3.get(i);
      Object[] records = (Object[])record;
      Assert.assertEquals("2014-12-12", records[1]);
    }

  }

  public class InnerObj
  {
    public InnerObj()
    {
    }

    private int id;
    private String date;

    public String getDate()
    {
      return date;
    }

    public void setDate(String date)
    {
      this.date = date;
    }

    public int getId()
    {
      return id;
    }

    public void setId(int id)
    {
      this.id = id;
    }

  }

  private static final Logger LOG = LoggerFactory.getLogger(HiveMockTest.class);

}