

package org.apache.carbondata.core.indexstore.blockletindex;

import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;

import org.apache.carbondata.core.cache.Cache;
import org.apache.carbondata.core.cache.CacheProvider;
import org.apache.carbondata.core.cache.CacheType;
import org.apache.carbondata.core.datamap.DataMapDistributable;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.indexstore.BlockletDataMapIndexWrapper;
import org.apache.carbondata.core.indexstore.TableBlockIndexUniqueIdentifier;
import org.apache.carbondata.core.indexstore.TableBlockIndexUniqueIdentifierWrapper;
import org.apache.carbondata.core.memory.MemoryException;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
import org.apache.carbondata.core.metadata.schema.table.TableInfo;
import org.apache.carbondata.core.readcommitter.TableStatusReadCommittedScope;

import mockit.Deencapsulation;
import mockit.Mock;
import mockit.MockUp;
import org.apache.hadoop.conf.Configuration;
import org.junit.Before;
import org.junit.Test;

public class TestBlockletDataMapFactory {

  private CarbonTable carbonTable;

  private AbsoluteTableIdentifier absoluteTableIdentifier;

  private TableInfo tableInfo;

  private BlockletDataMapFactory blockletDataMapFactory;

  private TableBlockIndexUniqueIdentifier tableBlockIndexUniqueIdentifier;

  private TableBlockIndexUniqueIdentifierWrapper tableBlockIndexUniqueIdentifierWrapper;

  private Cache<TableBlockIndexUniqueIdentifierWrapper, BlockletDataMapIndexWrapper> cache;

  @Before public void setUp()
      throws ClassNotFoundException, IllegalAccessException, InvocationTargetException,
      InstantiationException {
    tableInfo = new TableInfo();
    Constructor<?> constructor =
        Class.forName("org.apache.carbondata.core.metadata.schema.table.CarbonTable")
            .getDeclaredConstructors()[0];
    constructor.setAccessible(true);
    carbonTable = (CarbonTable) constructor.newInstance();
    absoluteTableIdentifier = AbsoluteTableIdentifier
        .from("/opt/store/default/carbon_table/", "default", "carbon_table",
            UUID.randomUUID().toString());
    Deencapsulation.setField(tableInfo, "identifier", absoluteTableIdentifier);
    Deencapsulation.setField(carbonTable, "tableInfo", tableInfo);
    new MockUp<CarbonTable>() {
      @Mock
      public AbsoluteTableIdentifier getAbsoluteTableIdentifier(){
        return absoluteTableIdentifier;
      }
    };
    blockletDataMapFactory = new BlockletDataMapFactory(carbonTable, new DataMapSchema());
    Deencapsulation.setField(blockletDataMapFactory, "cache",
        CacheProvider.getInstance().createCache(CacheType.DRIVER_BLOCKLET_DATAMAP));
    tableBlockIndexUniqueIdentifier =
        new TableBlockIndexUniqueIdentifier("/opt/store/default/carbon_table/Fact/Part0/Segment_0",
            "0_batchno0-0-1521012756709.carbonindex", null, "0");
    tableBlockIndexUniqueIdentifierWrapper =
        new TableBlockIndexUniqueIdentifierWrapper(tableBlockIndexUniqueIdentifier, carbonTable);
    cache = CacheProvider.getInstance().createCache(CacheType.DRIVER_BLOCKLET_DATAMAP);
  }

  @Test public void addDataMapToCache()
      throws IOException, MemoryException, NoSuchMethodException, InvocationTargetException,
      IllegalAccessException {
    List<BlockDataMap> dataMaps = new ArrayList<>();
    Method method = BlockletDataMapFactory.class
        .getDeclaredMethod("cache", TableBlockIndexUniqueIdentifierWrapper.class,
            BlockletDataMapIndexWrapper.class);
    method.setAccessible(true);
    method.invoke(blockletDataMapFactory, tableBlockIndexUniqueIdentifierWrapper,
        new BlockletDataMapIndexWrapper(tableBlockIndexUniqueIdentifier.getSegmentId(), dataMaps,
            tableBlockIndexUniqueIdentifierWrapper.getConfiguration()));
    BlockletDataMapIndexWrapper result = cache.getIfPresent(tableBlockIndexUniqueIdentifierWrapper);
    assert null != result;
  }

  @Test public void getValidDistributables() throws IOException {
    BlockletDataMapDistributable blockletDataMapDistributable = new BlockletDataMapDistributable(
        "/opt/store/default/carbon_table/Fact/Part0/Segment_0/0_batchno0-0-1521012756709.carbonindex");
    Segment segment = new Segment("0", null, new TableStatusReadCommittedScope(carbonTable
        .getAbsoluteTableIdentifier(), new Configuration(false)));
    blockletDataMapDistributable.setSegment(segment);
    BlockletDataMapDistributable blockletDataMapDistributable1 = new BlockletDataMapDistributable(
        "/opt/store/default/carbon_table/Fact/Part0/Segment_0/0_batchno0-0-1521012756701.carbonindex");
    blockletDataMapDistributable1.setSegment(segment);
    List<DataMapDistributable> dataMapDistributables = new ArrayList<>(2);
    dataMapDistributables.add(blockletDataMapDistributable);
    dataMapDistributables.add(blockletDataMapDistributable1);
    new MockUp<BlockletDataMapFactory>() {
      @Mock Set<TableBlockIndexUniqueIdentifier> getTableBlockIndexUniqueIdentifiers(
          Segment segment) {
        TableBlockIndexUniqueIdentifier tableBlockIndexUniqueIdentifier1 =
            new TableBlockIndexUniqueIdentifier(
                "/opt/store/default/carbon_table/Fact/Part0/Segment_0",
                "0_batchno0-0-1521012756701.carbonindex", null, "0");
        Set<TableBlockIndexUniqueIdentifier> tableBlockIndexUniqueIdentifiers = new HashSet<>(3);
        tableBlockIndexUniqueIdentifiers.add(tableBlockIndexUniqueIdentifier);
        tableBlockIndexUniqueIdentifiers.add(tableBlockIndexUniqueIdentifier1);
        return tableBlockIndexUniqueIdentifiers;
      }
    };
    List<DataMapDistributable> validDistributables =
        blockletDataMapFactory.getAllUncachedDistributables(dataMapDistributables);
    assert 1 == validDistributables.size();
  }
}