8707

Build # 8707

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/c3bb352075cc7c94b46d85040f85f958c6e788e1">c3bb35207<a href="https://github.com/apache/carbondata/commit/c3bb352075cc7c94b46d85040f85f958c6e788e1">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/6c7d565403d3eacc73f87106be815d225aa81fa2">6c7d56540</a>">6c7d56540</a><a href="https://github.com/apache/carbondata/commit/c3bb352075cc7c94b46d85040f85f958c6e788e1"> into e3eb03054">e3eb03054</a>

Run Details

62 of 62 new or added lines in 6 files covered. (100.0%)

59829 of 74200 relevant lines covered (80.63%)

1.07 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

69.38

/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapFactory.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.carbondata.core.indexstore.blockletindex;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

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.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datamap.DataMapDistributable;
import org.apache.carbondata.core.datamap.DataMapMeta;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datamap.dev.CacheableDataMap;
import org.apache.carbondata.core.datamap.dev.DataMap;
import org.apache.carbondata.core.datamap.dev.DataMapBuilder;
import org.apache.carbondata.core.datamap.dev.DataMapWriter;
import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMap;
import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMapFactory;
import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.features.TableOperation;
import org.apache.carbondata.core.indexstore.Blocklet;
import org.apache.carbondata.core.indexstore.BlockletDataMapIndexWrapper;
import org.apache.carbondata.core.indexstore.BlockletDetailsFetcher;
import org.apache.carbondata.core.indexstore.ExtendedBlocklet;
import org.apache.carbondata.core.indexstore.PartitionSpec;
import org.apache.carbondata.core.indexstore.SegmentPropertiesFetcher;
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.scan.expression.Expression;
import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.util.BlockletDataMapUtil;
import org.apache.carbondata.core.util.path.CarbonTablePath;
import org.apache.carbondata.events.Event;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;

/**
 * Table map for blocklet
 */
public class BlockletDataMapFactory extends CoarseGrainDataMapFactory
    implements BlockletDetailsFetcher, SegmentPropertiesFetcher, CacheableDataMap {

  private static final Log LOG = LogFactory.getLog(BlockletDataMapFactory.class);
  private static final String NAME = "clustered.btree.blocklet";
  /**
   * variable for cache level BLOCKLET
   */
  public static final String CACHE_LEVEL_BLOCKLET = "BLOCKLET";

  public static final DataMapSchema DATA_MAP_SCHEMA =
      new DataMapSchema(NAME, BlockletDataMapFactory.class.getName());

  private AbsoluteTableIdentifier identifier;

  // segmentId -> list of index file
  private Map<String, Set<TableBlockIndexUniqueIdentifier>> segmentMap = new ConcurrentHashMap<>();

  private Cache<TableBlockIndexUniqueIdentifierWrapper, BlockletDataMapIndexWrapper> cache;

  public BlockletDataMapFactory(CarbonTable carbonTable, DataMapSchema dataMapSchema) {
    super(carbonTable, dataMapSchema);
    this.identifier = carbonTable.getAbsoluteTableIdentifier();
    cache = CacheProvider.getInstance()
        .createCache(CacheType.DRIVER_BLOCKLET_DATAMAP);
  }

  /**
   * create dataMap based on cache level
   *
   * @param carbonTable
   * @return
   */
  public static DataMap createDataMap(CarbonTable carbonTable) {
    boolean cacheLevelBlock = BlockletDataMapUtil.isCacheLevelBlock(carbonTable);
    if (cacheLevelBlock) {
      // case1: when CACHE_LEVEL = BLOCK
      return new BlockDataMap();
    } else {
      // case2: when CACHE_LEVEL = BLOCKLET
      return new BlockletDataMap();
    }
  }

  @Override
  public DataMapWriter createWriter(Segment segment, String shardName,
      SegmentProperties segmentProperties) throws IOException {
    throw new UnsupportedOperationException("not implemented");
  }

  @Override
  public DataMapBuilder createBuilder(Segment segment, String shardName,
      SegmentProperties segmentProperties) throws IOException {
    throw new UnsupportedOperationException("not implemented");
  }

  /**
   * Get the datamap for all segments
   */
  public Map<Segment, List<CoarseGrainDataMap>> getDataMaps(List<Segment> segments)
      throws IOException {
    List<TableBlockIndexUniqueIdentifierWrapper> tableBlockIndexUniqueIdentifierWrappers =
        new ArrayList<>();
    Map<Segment, List<CoarseGrainDataMap>> dataMaps = new HashMap<>();
    Map<String, Segment> segmentMap = new HashMap<>();
    for (Segment segment : segments) {
      segmentMap.put(segment.getSegmentNo(), segment);
      Set<TableBlockIndexUniqueIdentifier> identifiers =
          getTableBlockIndexUniqueIdentifiers(segment);

      for (TableBlockIndexUniqueIdentifier tableBlockIndexUniqueIdentifier : identifiers) {
        tableBlockIndexUniqueIdentifierWrappers.add(
            new TableBlockIndexUniqueIdentifierWrapper(tableBlockIndexUniqueIdentifier,
                this.getCarbonTable(), segment.getConfiguration()));
      }
    }
    List<BlockletDataMapIndexWrapper> blockletDataMapIndexWrappers =
        cache.getAll(tableBlockIndexUniqueIdentifierWrappers);
    for (BlockletDataMapIndexWrapper wrapper : blockletDataMapIndexWrappers) {
      Segment segment = segmentMap.get(wrapper.getSegmentId());
      List<CoarseGrainDataMap> datamapList = dataMaps.get(segment);
      if (null == datamapList) {
        datamapList = new ArrayList<CoarseGrainDataMap>();
      }
      datamapList.addAll(wrapper.getDataMaps());
      dataMaps.put(segment, datamapList);
    }
    return dataMaps;
  }

  @Override public List<CoarseGrainDataMap> getDataMaps(Segment segment) throws IOException {
    List<CoarseGrainDataMap> dataMaps = new ArrayList<>();
    Set<TableBlockIndexUniqueIdentifier> identifiers =
        getTableBlockIndexUniqueIdentifiers(segment);
    List<TableBlockIndexUniqueIdentifierWrapper> tableBlockIndexUniqueIdentifierWrappers =
        new ArrayList<>(identifiers.size());
    for (TableBlockIndexUniqueIdentifier tableBlockIndexUniqueIdentifier : identifiers) {
      tableBlockIndexUniqueIdentifierWrappers.add(
          new TableBlockIndexUniqueIdentifierWrapper(tableBlockIndexUniqueIdentifier,
              this.getCarbonTable()));
    }
    List<BlockletDataMapIndexWrapper> blockletDataMapIndexWrappers =
        cache.getAll(tableBlockIndexUniqueIdentifierWrappers);
    for (BlockletDataMapIndexWrapper wrapper : blockletDataMapIndexWrappers) {
      dataMaps.addAll(wrapper.getDataMaps());
    }
    return dataMaps;
  }

  private Set<TableBlockIndexUniqueIdentifier> getTableBlockIndexUniqueIdentifiers(Segment segment)
      throws IOException {
    Set<TableBlockIndexUniqueIdentifier> tableBlockIndexUniqueIdentifiers =
        segmentMap.get(segment.getSegmentNo());
    if (tableBlockIndexUniqueIdentifiers == null) {
      tableBlockIndexUniqueIdentifiers =
          BlockletDataMapUtil.getTableBlockUniqueIdentifiers(segment);
      if (tableBlockIndexUniqueIdentifiers.size() > 0) {
        segmentMap.put(segment.getSegmentNo(), tableBlockIndexUniqueIdentifiers);
      }
    }
    return tableBlockIndexUniqueIdentifiers;
  }

  /**
   * Get the blocklet detail information based on blockletid, blockid and segmentid. This method is
   * exclusively for BlockletDataMapFactory as detail information is only available in this
   * default datamap.
   */
  @Override
  public List<ExtendedBlocklet> getExtendedBlocklets(List<Blocklet> blocklets, Segment segment)
      throws IOException {
    List<ExtendedBlocklet> detailedBlocklets = new ArrayList<>();
    // If it is already detailed blocklet then type cast and return same
    if (blocklets.size() > 0 && blocklets.get(0) instanceof ExtendedBlocklet) {
      for (Blocklet blocklet : blocklets) {
        detailedBlocklets.add((ExtendedBlocklet) blocklet);
      }
      return detailedBlocklets;
    }
    Set<TableBlockIndexUniqueIdentifier> identifiers =
        getTableBlockIndexUniqueIdentifiers(segment);
    Set<TableBlockIndexUniqueIdentifierWrapper> tableBlockIndexUniqueIdentifierWrappers =
        new HashSet<>(identifiers.size());
    for (TableBlockIndexUniqueIdentifier tableBlockIndexUniqueIdentifier : identifiers) {
      tableBlockIndexUniqueIdentifierWrappers.add(
          new TableBlockIndexUniqueIdentifierWrapper(tableBlockIndexUniqueIdentifier,
              this.getCarbonTable()));
    }
    // Retrieve each blocklets detail information from blocklet datamap
    for (Blocklet blocklet : blocklets) {
      detailedBlocklets.add(getExtendedBlocklet(tableBlockIndexUniqueIdentifierWrappers, blocklet));
    }
    return detailedBlocklets;
  }

  @Override
  public ExtendedBlocklet getExtendedBlocklet(Blocklet blocklet, Segment segment)
      throws IOException {
    if (blocklet instanceof ExtendedBlocklet) {
      return (ExtendedBlocklet) blocklet;
    }
    Set<TableBlockIndexUniqueIdentifier> identifiers =
        getTableBlockIndexUniqueIdentifiers(segment);

    Set<TableBlockIndexUniqueIdentifierWrapper> tableBlockIndexUniqueIdentifierWrappers =
        new HashSet<>(identifiers.size());
    for (TableBlockIndexUniqueIdentifier tableBlockIndexUniqueIdentifier : identifiers) {
      tableBlockIndexUniqueIdentifierWrappers.add(
          new TableBlockIndexUniqueIdentifierWrapper(tableBlockIndexUniqueIdentifier,
              this.getCarbonTable()));
    }
    return getExtendedBlocklet(tableBlockIndexUniqueIdentifierWrappers, blocklet);
  }

  private ExtendedBlocklet getExtendedBlocklet(
      Set<TableBlockIndexUniqueIdentifierWrapper> identifiersWrapper, Blocklet blocklet)
      throws IOException {
    for (TableBlockIndexUniqueIdentifierWrapper identifierWrapper : identifiersWrapper) {
      BlockletDataMapIndexWrapper wrapper = cache.get(identifierWrapper);
      List<BlockDataMap> dataMaps = wrapper.getDataMaps();
      for (DataMap dataMap : dataMaps) {
        if (((BlockDataMap) dataMap)
            .getTableTaskInfo(BlockletDataMapRowIndexes.SUMMARY_INDEX_FILE_NAME)
            .startsWith(blocklet.getFilePath())) {
          return ((BlockDataMap) dataMap).getDetailedBlocklet(blocklet.getBlockletId());
        }
      }
    }
    throw new IOException("Blocklet with blockid " + blocklet.getBlockletId() + " not found ");
  }


  @Override
  public List<DataMapDistributable> toDistributable(Segment segment) {
    List<DataMapDistributable> distributables = new ArrayList<>();
    try {
      Set<TableBlockIndexUniqueIdentifier> tableBlockIndexUniqueIdentifiers =
          getTableBlockIndexUniqueIdentifiers(segment);
      CarbonFile[] carbonIndexFiles = new CarbonFile[tableBlockIndexUniqueIdentifiers.size()];
      int identifierCounter = 0;
      for (TableBlockIndexUniqueIdentifier tableBlockIndexUniqueIdentifier :
          tableBlockIndexUniqueIdentifiers) {
        String indexFilePath = tableBlockIndexUniqueIdentifier.getIndexFilePath();
        String fileName = tableBlockIndexUniqueIdentifier.getIndexFileName();
        carbonIndexFiles[identifierCounter++] = FileFactory
            .getCarbonFile(indexFilePath + CarbonCommonConstants.FILE_SEPARATOR + fileName);
      }
      for (int i = 0; i < carbonIndexFiles.length; i++) {
        Path path = new Path(carbonIndexFiles[i].getPath());
        FileSystem fs = path.getFileSystem(FileFactory.getConfiguration());
        RemoteIterator<LocatedFileStatus> iter = fs.listLocatedStatus(path);
        LocatedFileStatus fileStatus = iter.next();
        String[] location = fileStatus.getBlockLocations()[0].getHosts();
        BlockletDataMapDistributable distributable =
            new BlockletDataMapDistributable(path.toString());
        distributable.setLocations(location);
        distributables.add(distributable);
      }
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
    return distributables;
  }

  @Override
  public void fireEvent(Event event) {

  }

  @Override
  public void clear(Segment segment) {
    Set<TableBlockIndexUniqueIdentifier> blockIndexes = segmentMap.remove(segment.getSegmentNo());
    if (blockIndexes != null) {
      for (TableBlockIndexUniqueIdentifier blockIndex : blockIndexes) {
        TableBlockIndexUniqueIdentifierWrapper blockIndexWrapper =
            new TableBlockIndexUniqueIdentifierWrapper(blockIndex, this.getCarbonTable());
        BlockletDataMapIndexWrapper wrapper = cache.getIfPresent(blockIndexWrapper);
        if (null != wrapper) {
          List<BlockDataMap> dataMaps = wrapper.getDataMaps();
          for (DataMap dataMap : dataMaps) {
            if (dataMap != null) {
              cache.invalidate(blockIndexWrapper);
              dataMap.clear();
            }
          }
        }
      }
    }
  }

  @Override
  public synchronized void clear() {
    if (segmentMap.size() > 0) {
      for (String segmentId : segmentMap.keySet().toArray(new String[segmentMap.size()])) {
        clear(new Segment(segmentId, null, null));
      }
    }
  }

  @Override
  public List<CoarseGrainDataMap> getDataMaps(DataMapDistributable distributable)
      throws IOException {
    BlockletDataMapDistributable mapDistributable = (BlockletDataMapDistributable) distributable;
    List<TableBlockIndexUniqueIdentifierWrapper> identifiersWrapper = new ArrayList<>();
    Path indexPath = new Path(mapDistributable.getFilePath());
    String segmentNo = mapDistributable.getSegment().getSegmentNo();
    if (indexPath.getName().endsWith(CarbonTablePath.INDEX_FILE_EXT)) {
      String parent = indexPath.getParent().toString();
      identifiersWrapper.add(new TableBlockIndexUniqueIdentifierWrapper(
          new TableBlockIndexUniqueIdentifier(parent, indexPath.getName(), null, segmentNo),
          this.getCarbonTable()));
    } else if (indexPath.getName().endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT)) {
      SegmentIndexFileStore fileStore = new SegmentIndexFileStore();
      CarbonFile carbonFile = FileFactory.getCarbonFile(indexPath.toString());
      String parentPath = carbonFile.getParentFile().getAbsolutePath();
      List<String> indexFiles = fileStore.getIndexFilesFromMergeFile(carbonFile.getAbsolutePath());
      for (String indexFile : indexFiles) {
        identifiersWrapper.add(new TableBlockIndexUniqueIdentifierWrapper(
            new TableBlockIndexUniqueIdentifier(parentPath, indexFile, carbonFile.getName(),
                segmentNo), this.getCarbonTable()));
      }
    }
    List<CoarseGrainDataMap> dataMaps = new ArrayList<>();
    try {
      List<BlockletDataMapIndexWrapper> wrappers = cache.getAll(identifiersWrapper);
      for (BlockletDataMapIndexWrapper wrapper : wrappers) {
        dataMaps.addAll(wrapper.getDataMaps());
      }
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
    return dataMaps;
  }

  @Override
  public DataMapMeta getMeta() {
    // TODO: pass SORT_COLUMNS into this class
    return null;
  }

  /**
   * actually DataMapChooser won't call this method since getMeta() return null
   * here return true by default for better code reading
   * since default datamap can handle all expression
   */
  @Override
  public boolean isSupport(Expression expression) {
    return true;
  }

  @Override
  public void deleteDatamapData(Segment segment) throws IOException {

  }

  @Override public void deleteDatamapData() {

  }

  @Override public SegmentProperties getSegmentProperties(Segment segment) throws IOException {
    List<CoarseGrainDataMap> dataMaps = getDataMaps(segment);
    assert (dataMaps.size() > 0);
    CoarseGrainDataMap coarseGrainDataMap = dataMaps.get(0);
    assert (coarseGrainDataMap instanceof BlockDataMap);
    BlockDataMap dataMap = (BlockDataMap) coarseGrainDataMap;
    return dataMap.getSegmentProperties();
  }

  @Override public List<Blocklet> getAllBlocklets(Segment segment, List<PartitionSpec> partitions)
      throws IOException {
    List<Blocklet> blocklets = new ArrayList<>();
    List<CoarseGrainDataMap> dataMaps = getDataMaps(segment);
    for (CoarseGrainDataMap dataMap : dataMaps) {
      blocklets.addAll(
          dataMap.prune((FilterResolverIntf) null, getSegmentProperties(segment), partitions));
    }
    return blocklets;
  }

  @Override public boolean willBecomeStale(TableOperation operation) {
    return false;
  }

  @Override
  public void cache(TableBlockIndexUniqueIdentifierWrapper tableBlockIndexUniqueIdentifierWrapper,
      BlockletDataMapIndexWrapper blockletDataMapIndexWrapper) throws IOException, MemoryException {
    cache.put(tableBlockIndexUniqueIdentifierWrapper, blockletDataMapIndexWrapper);
  }

  @Override
  public List<DataMapDistributable> getAllUncachedDistributables(
      List<DataMapDistributable> distributables) throws IOException {
    List<DataMapDistributable> distributablesToBeLoaded = new ArrayList<>(distributables.size());
    for (DataMapDistributable distributable : distributables) {
      Segment segment = distributable.getSegment();
      Set<TableBlockIndexUniqueIdentifier> tableBlockIndexUniqueIdentifiers =
          getTableBlockIndexUniqueIdentifiers(segment);
      // filter out the tableBlockIndexUniqueIdentifiers based on distributable
      TableBlockIndexUniqueIdentifier validIdentifier = BlockletDataMapUtil
          .filterIdentifiersBasedOnDistributable(tableBlockIndexUniqueIdentifiers,
              (BlockletDataMapDistributable) distributable);
      if (null == cache.getIfPresent(
          new TableBlockIndexUniqueIdentifierWrapper(validIdentifier, this.getCarbonTable()))) {
        ((BlockletDataMapDistributable) distributable)
            .setTableBlockIndexUniqueIdentifier(validIdentifier);
        distributablesToBeLoaded.add(distributable);
      }
    }
    return distributablesToBeLoaded;
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17	package org.apache.carbondata.core.indexstore.blockletindex;
18
19	import java.io.IOException;
20	import java.util.*;
21	import java.util.concurrent.ConcurrentHashMap;
22
23	import org.apache.carbondata.core.cache.Cache;
24	import org.apache.carbondata.core.cache.CacheProvider;
25	import org.apache.carbondata.core.cache.CacheType;
26	import org.apache.carbondata.core.constants.CarbonCommonConstants;
27	import org.apache.carbondata.core.datamap.DataMapDistributable;
28	import org.apache.carbondata.core.datamap.DataMapMeta;
29	import org.apache.carbondata.core.datamap.Segment;
30	import org.apache.carbondata.core.datamap.dev.CacheableDataMap;
31	import org.apache.carbondata.core.datamap.dev.DataMap;
32	import org.apache.carbondata.core.datamap.dev.DataMapBuilder;
33	import org.apache.carbondata.core.datamap.dev.DataMapWriter;
34	import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMap;
35	import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMapFactory;
36	import org.apache.carbondata.core.datastore.block.SegmentProperties;
37	import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
38	import org.apache.carbondata.core.datastore.impl.FileFactory;
39	import org.apache.carbondata.core.features.TableOperation;
40	import org.apache.carbondata.core.indexstore.Blocklet;
41	import org.apache.carbondata.core.indexstore.BlockletDataMapIndexWrapper;
42	import org.apache.carbondata.core.indexstore.BlockletDetailsFetcher;
43	import org.apache.carbondata.core.indexstore.ExtendedBlocklet;
44	import org.apache.carbondata.core.indexstore.PartitionSpec;
45	import org.apache.carbondata.core.indexstore.SegmentPropertiesFetcher;
46	import org.apache.carbondata.core.indexstore.TableBlockIndexUniqueIdentifier;
47	import org.apache.carbondata.core.indexstore.TableBlockIndexUniqueIdentifierWrapper;
48	import org.apache.carbondata.core.memory.MemoryException;
49	import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
50	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
51	import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
52	import org.apache.carbondata.core.scan.expression.Expression;
53	import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
54	import org.apache.carbondata.core.util.BlockletDataMapUtil;
55	import org.apache.carbondata.core.util.path.CarbonTablePath;
56	import org.apache.carbondata.events.Event;
57
58	import org.apache.commons.logging.Log;
59	import org.apache.commons.logging.LogFactory;
60	import org.apache.hadoop.fs.FileSystem;
61	import org.apache.hadoop.fs.LocatedFileStatus;
62	import org.apache.hadoop.fs.Path;
63	import org.apache.hadoop.fs.RemoteIterator;
64
65	/**
66	* Table map for blocklet
67	*/
68	public class BlockletDataMapFactory extends CoarseGrainDataMapFactory	2✔
69	implements BlockletDetailsFetcher, SegmentPropertiesFetcher, CacheableDataMap {
70
71	private static final Log LOG = LogFactory.getLog(BlockletDataMapFactory.class);	2✔
72	private static final String NAME = "clustered.btree.blocklet";
73	/**
74	* variable for cache level BLOCKLET
75	*/
76	public static final String CACHE_LEVEL_BLOCKLET = "BLOCKLET";
77
78	public static final DataMapSchema DATA_MAP_SCHEMA =	2✔
79	new DataMapSchema(NAME, BlockletDataMapFactory.class.getName());	2✔
80
81	private AbsoluteTableIdentifier identifier;
82
83	// segmentId -> list of index file
84	private Map<String, Set<TableBlockIndexUniqueIdentifier>> segmentMap = new ConcurrentHashMap<>();	2✔
85
86	private Cache<TableBlockIndexUniqueIdentifierWrapper, BlockletDataMapIndexWrapper> cache;
87
88	public BlockletDataMapFactory(CarbonTable carbonTable, DataMapSchema dataMapSchema) {
89	super(carbonTable, dataMapSchema);	2✔
90	this.identifier = carbonTable.getAbsoluteTableIdentifier();	2✔
91	cache = CacheProvider.getInstance()	2✔
92	.createCache(CacheType.DRIVER_BLOCKLET_DATAMAP);	2✔
93	}	2✔
94
95	/**
96	* create dataMap based on cache level
97	*
98	* @param carbonTable
99	* @return
100	*/
101	public static DataMap createDataMap(CarbonTable carbonTable) {
102	boolean cacheLevelBlock = BlockletDataMapUtil.isCacheLevelBlock(carbonTable);	1✔
103	if (cacheLevelBlock) {	1✔
104	// case1: when CACHE_LEVEL = BLOCK
105	return new BlockDataMap();	1✔
106	} else {
107	// case2: when CACHE_LEVEL = BLOCKLET
108	return new BlockletDataMap();	1✔
109	}
110	}
111
112	@Override
113	public DataMapWriter createWriter(Segment segment, String shardName,
114	SegmentProperties segmentProperties) throws IOException {
115	throw new UnsupportedOperationException("not implemented");	×
116	}
117
118	@Override
119	public DataMapBuilder createBuilder(Segment segment, String shardName,
120	SegmentProperties segmentProperties) throws IOException {
121	throw new UnsupportedOperationException("not implemented");	×
122	}
123
124	/**
125	* Get the datamap for all segments
126	*/
127	public Map<Segment, List<CoarseGrainDataMap>> getDataMaps(List<Segment> segments)
128	throws IOException {
129	List<TableBlockIndexUniqueIdentifierWrapper> tableBlockIndexUniqueIdentifierWrappers =	1✔
130	new ArrayList<>();
131	Map<Segment, List<CoarseGrainDataMap>> dataMaps = new HashMap<>();	1✔
132	Map<String, Segment> segmentMap = new HashMap<>();	1✔
133	for (Segment segment : segments) {	1✔
134	segmentMap.put(segment.getSegmentNo(), segment);	1✔
135	Set<TableBlockIndexUniqueIdentifier> identifiers =	1✔
136	getTableBlockIndexUniqueIdentifiers(segment);	1✔
137
138	for (TableBlockIndexUniqueIdentifier tableBlockIndexUniqueIdentifier : identifiers) {	1✔
139	tableBlockIndexUniqueIdentifierWrappers.add(	1✔
140	new TableBlockIndexUniqueIdentifierWrapper(tableBlockIndexUniqueIdentifier,
141	this.getCarbonTable(), segment.getConfiguration()));	1✔
142	}	1✔
143	}	1✔
144	List<BlockletDataMapIndexWrapper> blockletDataMapIndexWrappers =	1✔
145	cache.getAll(tableBlockIndexUniqueIdentifierWrappers);	1✔
146	for (BlockletDataMapIndexWrapper wrapper : blockletDataMapIndexWrappers) {	1✔
147	Segment segment = segmentMap.get(wrapper.getSegmentId());	1✔
148	List<CoarseGrainDataMap> datamapList = dataMaps.get(segment);	1✔
149	if (null == datamapList) {	1✔
150	datamapList = new ArrayList<CoarseGrainDataMap>();	1✔
151	}
152	datamapList.addAll(wrapper.getDataMaps());	1✔
153	dataMaps.put(segment, datamapList);	1✔
154	}	1✔
155	return dataMaps;	1✔
156	}
157
158	@Override public List<CoarseGrainDataMap> getDataMaps(Segment segment) throws IOException {
159	List<CoarseGrainDataMap> dataMaps = new ArrayList<>();	1✔
160	Set<TableBlockIndexUniqueIdentifier> identifiers =	1✔
161	getTableBlockIndexUniqueIdentifiers(segment);	1✔
162	List<TableBlockIndexUniqueIdentifierWrapper> tableBlockIndexUniqueIdentifierWrappers =	1✔
163	new ArrayList<>(identifiers.size());	1✔
164	for (TableBlockIndexUniqueIdentifier tableBlockIndexUniqueIdentifier : identifiers) {	1✔
165	tableBlockIndexUniqueIdentifierWrappers.add(	1✔
166	new TableBlockIndexUniqueIdentifierWrapper(tableBlockIndexUniqueIdentifier,
167	this.getCarbonTable()));	1✔
168	}	1✔
169	List<BlockletDataMapIndexWrapper> blockletDataMapIndexWrappers =	1✔
170	cache.getAll(tableBlockIndexUniqueIdentifierWrappers);	1✔
171	for (BlockletDataMapIndexWrapper wrapper : blockletDataMapIndexWrappers) {	1✔
172	dataMaps.addAll(wrapper.getDataMaps());	1✔
173	}	1✔
174	return dataMaps;	1✔
175	}
176
177	private Set<TableBlockIndexUniqueIdentifier> getTableBlockIndexUniqueIdentifiers(Segment segment)
178	throws IOException {
179	Set<TableBlockIndexUniqueIdentifier> tableBlockIndexUniqueIdentifiers =	1✔
180	segmentMap.get(segment.getSegmentNo());	1✔
181	if (tableBlockIndexUniqueIdentifiers == null) {	1✔
182	tableBlockIndexUniqueIdentifiers =	1✔
183	BlockletDataMapUtil.getTableBlockUniqueIdentifiers(segment);	1✔
184	if (tableBlockIndexUniqueIdentifiers.size() > 0) {	1✔
185	segmentMap.put(segment.getSegmentNo(), tableBlockIndexUniqueIdentifiers);	1✔
186	}
187	}
188	return tableBlockIndexUniqueIdentifiers;	1✔
189	}
190
191	/**
192	* Get the blocklet detail information based on blockletid, blockid and segmentid. This method is
193	* exclusively for BlockletDataMapFactory as detail information is only available in this
194	* default datamap.
195	*/
196	@Override
197	public List<ExtendedBlocklet> getExtendedBlocklets(List<Blocklet> blocklets, Segment segment)
198	throws IOException {
199	List<ExtendedBlocklet> detailedBlocklets = new ArrayList<>();	1✔
200	// If it is already detailed blocklet then type cast and return same
201	if (blocklets.size() > 0 && blocklets.get(0) instanceof ExtendedBlocklet) {	1✔
202	for (Blocklet blocklet : blocklets) {	1✔
203	detailedBlocklets.add((ExtendedBlocklet) blocklet);	1✔
204	}	1✔
205	return detailedBlocklets;	1✔
206	}
207	Set<TableBlockIndexUniqueIdentifier> identifiers =	1✔
208	getTableBlockIndexUniqueIdentifiers(segment);	1✔
209	Set<TableBlockIndexUniqueIdentifierWrapper> tableBlockIndexUniqueIdentifierWrappers =	1✔
210	new HashSet<>(identifiers.size());	1✔
211	for (TableBlockIndexUniqueIdentifier tableBlockIndexUniqueIdentifier : identifiers) {	1✔
212	tableBlockIndexUniqueIdentifierWrappers.add(	1✔
213	new TableBlockIndexUniqueIdentifierWrapper(tableBlockIndexUniqueIdentifier,
214	this.getCarbonTable()));	1✔
215	}	1✔
216	// Retrieve each blocklets detail information from blocklet datamap
217	for (Blocklet blocklet : blocklets) {	1✔
218	detailedBlocklets.add(getExtendedBlocklet(tableBlockIndexUniqueIdentifierWrappers, blocklet));	1✔
219	}	1✔
220	return detailedBlocklets;	1✔
221	}
222
223	@Override
224	public ExtendedBlocklet getExtendedBlocklet(Blocklet blocklet, Segment segment)
225	throws IOException {
226	if (blocklet instanceof ExtendedBlocklet) {	1✔
227	return (ExtendedBlocklet) blocklet;	×
228	}
229	Set<TableBlockIndexUniqueIdentifier> identifiers =	1✔
230	getTableBlockIndexUniqueIdentifiers(segment);	1✔
231
232	Set<TableBlockIndexUniqueIdentifierWrapper> tableBlockIndexUniqueIdentifierWrappers =	1✔
233	new HashSet<>(identifiers.size());	1✔
234	for (TableBlockIndexUniqueIdentifier tableBlockIndexUniqueIdentifier : identifiers) {	1✔
235	tableBlockIndexUniqueIdentifierWrappers.add(	1✔
236	new TableBlockIndexUniqueIdentifierWrapper(tableBlockIndexUniqueIdentifier,
237	this.getCarbonTable()));	1✔
238	}	1✔
239	return getExtendedBlocklet(tableBlockIndexUniqueIdentifierWrappers, blocklet);	1✔
240	}
241
242	private ExtendedBlocklet getExtendedBlocklet(
243	Set<TableBlockIndexUniqueIdentifierWrapper> identifiersWrapper, Blocklet blocklet)
244	throws IOException {
245	for (TableBlockIndexUniqueIdentifierWrapper identifierWrapper : identifiersWrapper) {	1✔
246	BlockletDataMapIndexWrapper wrapper = cache.get(identifierWrapper);	1✔
247	List<BlockDataMap> dataMaps = wrapper.getDataMaps();	1✔
248	for (DataMap dataMap : dataMaps) {	1✔
249	if (((BlockDataMap) dataMap)	1✔
250	.getTableTaskInfo(BlockletDataMapRowIndexes.SUMMARY_INDEX_FILE_NAME)	1✔
251	.startsWith(blocklet.getFilePath())) {	1✔
252	return ((BlockDataMap) dataMap).getDetailedBlocklet(blocklet.getBlockletId());	1✔
253	}
254	}	1✔
255	}	×
256	throw new IOException("Blocklet with blockid " + blocklet.getBlockletId() + " not found ");	×
257	}
258
259
260	@Override
261	public List<DataMapDistributable> toDistributable(Segment segment) {
262	List<DataMapDistributable> distributables = new ArrayList<>();	×
263	try {
264	Set<TableBlockIndexUniqueIdentifier> tableBlockIndexUniqueIdentifiers =	×
265	getTableBlockIndexUniqueIdentifiers(segment);	×
266	CarbonFile[] carbonIndexFiles = new CarbonFile[tableBlockIndexUniqueIdentifiers.size()];	×
267	int identifierCounter = 0;	×
268	for (TableBlockIndexUniqueIdentifier tableBlockIndexUniqueIdentifier :
269	tableBlockIndexUniqueIdentifiers) {	×
270	String indexFilePath = tableBlockIndexUniqueIdentifier.getIndexFilePath();	×
271	String fileName = tableBlockIndexUniqueIdentifier.getIndexFileName();	×
272	carbonIndexFiles[identifierCounter++] = FileFactory	×
273	.getCarbonFile(indexFilePath + CarbonCommonConstants.FILE_SEPARATOR + fileName);	×
274	}	×
275	for (int i = 0; i < carbonIndexFiles.length; i++) {	×
276	Path path = new Path(carbonIndexFiles[i].getPath());	×
277	FileSystem fs = path.getFileSystem(FileFactory.getConfiguration());	×
278	RemoteIterator<LocatedFileStatus> iter = fs.listLocatedStatus(path);	×
279	LocatedFileStatus fileStatus = iter.next();	×
280	String[] location = fileStatus.getBlockLocations()[0].getHosts();	×
281	BlockletDataMapDistributable distributable =	×
282	new BlockletDataMapDistributable(path.toString());	×
283	distributable.setLocations(location);	×
284	distributables.add(distributable);	×
285	}
286	} catch (IOException e) {	×
287	throw new RuntimeException(e);	×
288	}	×
289	return distributables;	×
290	}
291
292	@Override
293	public void fireEvent(Event event) {
294
295	}	×
296
297	@Override
298	public void clear(Segment segment) {
299	Set<TableBlockIndexUniqueIdentifier> blockIndexes = segmentMap.remove(segment.getSegmentNo());	1✔
300	if (blockIndexes != null) {	1✔
301	for (TableBlockIndexUniqueIdentifier blockIndex : blockIndexes) {	1✔
302	TableBlockIndexUniqueIdentifierWrapper blockIndexWrapper =	1✔
303	new TableBlockIndexUniqueIdentifierWrapper(blockIndex, this.getCarbonTable());	1✔
304	BlockletDataMapIndexWrapper wrapper = cache.getIfPresent(blockIndexWrapper);	1✔
305	if (null != wrapper) {	1✔
306	List<BlockDataMap> dataMaps = wrapper.getDataMaps();	1✔
307	for (DataMap dataMap : dataMaps) {	1✔
308	if (dataMap != null) {	1✔
309	cache.invalidate(blockIndexWrapper);	1✔
310	dataMap.clear();	1✔
311	}
312	}	1✔
313	}
314	}	1✔
315	}
316	}	1✔
317
318	@Override
319	public synchronized void clear() {
320	if (segmentMap.size() > 0) {	1✔
321	for (String segmentId : segmentMap.keySet().toArray(new String[segmentMap.size()])) {	1✔
322	clear(new Segment(segmentId, null, null));	1✔
323	}
324	}
325	}	1✔
326
327	@Override
328	public List<CoarseGrainDataMap> getDataMaps(DataMapDistributable distributable)
329	throws IOException {
330	BlockletDataMapDistributable mapDistributable = (BlockletDataMapDistributable) distributable;	×
331	List<TableBlockIndexUniqueIdentifierWrapper> identifiersWrapper = new ArrayList<>();	×
332	Path indexPath = new Path(mapDistributable.getFilePath());	×
333	String segmentNo = mapDistributable.getSegment().getSegmentNo();	×
334	if (indexPath.getName().endsWith(CarbonTablePath.INDEX_FILE_EXT)) {	×
335	String parent = indexPath.getParent().toString();	×
336	identifiersWrapper.add(new TableBlockIndexUniqueIdentifierWrapper(	×
337	new TableBlockIndexUniqueIdentifier(parent, indexPath.getName(), null, segmentNo),	×
338	this.getCarbonTable()));	×
339	} else if (indexPath.getName().endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT)) {	×
340	SegmentIndexFileStore fileStore = new SegmentIndexFileStore();	×
341	CarbonFile carbonFile = FileFactory.getCarbonFile(indexPath.toString());	×
342	String parentPath = carbonFile.getParentFile().getAbsolutePath();	×
343	List<String> indexFiles = fileStore.getIndexFilesFromMergeFile(carbonFile.getAbsolutePath());	×
344	for (String indexFile : indexFiles) {	×
345	identifiersWrapper.add(new TableBlockIndexUniqueIdentifierWrapper(	×
346	new TableBlockIndexUniqueIdentifier(parentPath, indexFile, carbonFile.getName(),	×
347	segmentNo), this.getCarbonTable()));	×
348	}	×
349	}
350	List<CoarseGrainDataMap> dataMaps = new ArrayList<>();	×
351	try {
352	List<BlockletDataMapIndexWrapper> wrappers = cache.getAll(identifiersWrapper);	×
353	for (BlockletDataMapIndexWrapper wrapper : wrappers) {	×
354	dataMaps.addAll(wrapper.getDataMaps());	×
355	}	×
356	} catch (IOException e) {	×
357	throw new RuntimeException(e);	×
358	}	×
359	return dataMaps;	×
360	}
361
362	@Override
363	public DataMapMeta getMeta() {
364	// TODO: pass SORT_COLUMNS into this class
365	return null;	×
366	}
367
368	/**
369	* actually DataMapChooser won't call this method since getMeta() return null
370	* here return true by default for better code reading
371	* since default datamap can handle all expression
372	*/
373	@Override
374	public boolean isSupport(Expression expression) {
375	return true;	×
376	}
377
378	@Override
379	public void deleteDatamapData(Segment segment) throws IOException {
380
381	}	×
382
383	@Override public void deleteDatamapData() {
384
385	}	×
386
387	@Override public SegmentProperties getSegmentProperties(Segment segment) throws IOException {
388	List<CoarseGrainDataMap> dataMaps = getDataMaps(segment);	1✔
389	assert (dataMaps.size() > 0);	1✔
390	CoarseGrainDataMap coarseGrainDataMap = dataMaps.get(0);	1✔
391	assert (coarseGrainDataMap instanceof BlockDataMap);	1✔
392	BlockDataMap dataMap = (BlockDataMap) coarseGrainDataMap;	1✔
393	return dataMap.getSegmentProperties();	1✔
394	}
395
396	@Override public List<Blocklet> getAllBlocklets(Segment segment, List<PartitionSpec> partitions)
397	throws IOException {
398	List<Blocklet> blocklets = new ArrayList<>();	1✔
399	List<CoarseGrainDataMap> dataMaps = getDataMaps(segment);	1✔
400	for (CoarseGrainDataMap dataMap : dataMaps) {	1✔
401	blocklets.addAll(	1✔
402	dataMap.prune((FilterResolverIntf) null, getSegmentProperties(segment), partitions));	1✔
403	}	1✔
404	return blocklets;	1✔
405	}
406
407	@Override public boolean willBecomeStale(TableOperation operation) {
408	return false;	×
409	}
410
411	@Override
412	public void cache(TableBlockIndexUniqueIdentifierWrapper tableBlockIndexUniqueIdentifierWrapper,
413	BlockletDataMapIndexWrapper blockletDataMapIndexWrapper) throws IOException, MemoryException {
414	cache.put(tableBlockIndexUniqueIdentifierWrapper, blockletDataMapIndexWrapper);	2✔
415	}	2✔
416
417	@Override
418	public List<DataMapDistributable> getAllUncachedDistributables(
419	List<DataMapDistributable> distributables) throws IOException {
420	List<DataMapDistributable> distributablesToBeLoaded = new ArrayList<>(distributables.size());	2✔
421	for (DataMapDistributable distributable : distributables) {	2✔
422	Segment segment = distributable.getSegment();	2✔
423	Set<TableBlockIndexUniqueIdentifier> tableBlockIndexUniqueIdentifiers =	2✔
424	getTableBlockIndexUniqueIdentifiers(segment);	2✔
425	// filter out the tableBlockIndexUniqueIdentifiers based on distributable
426	TableBlockIndexUniqueIdentifier validIdentifier = BlockletDataMapUtil	2✔
427	.filterIdentifiersBasedOnDistributable(tableBlockIndexUniqueIdentifiers,	2✔
428	(BlockletDataMapDistributable) distributable);
429	if (null == cache.getIfPresent(	2✔
430	new TableBlockIndexUniqueIdentifierWrapper(validIdentifier, this.getCarbonTable()))) {	2✔
431	((BlockletDataMapDistributable) distributable)	2✔
432	.setTableBlockIndexUniqueIdentifier(validIdentifier);	2✔
433	distributablesToBeLoaded.add(distributable);	2✔
434	}
435	}	2✔
436	return distributablesToBeLoaded;	2✔
437	}
438	}

apache / carbondata / 8707

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous