apache / carbondata / 182

Build # 182

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/0146cb83bcee838bcf1667b7432b6b0ce25275e5">0146cb83b<a href="https://github.com/apache/carbondata/commit/0146cb83bcee838bcf1667b7432b6b0ce25275e5">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/88da31643d7e33b2b76b9f4bbce3a222089236a4">88da31643</a>">88da31643</a><a href="https://github.com/apache/carbondata/commit/0146cb83bcee838bcf1667b7432b6b0ce25275e5"> into c8cc92b6d">c8cc92b6d</a>

Run Details

5 of 5 new or added lines in 1 file covered. (100.0%)

65019 of 83724 relevant lines covered (77.66%)

1.02 hits per line

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

56.04

/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.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.datamap;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datamap.dev.BlockletSerializer;
import org.apache.carbondata.core.datamap.dev.DataMap;
import org.apache.carbondata.core.datamap.dev.DataMapFactory;
import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMap;
import org.apache.carbondata.core.datamap.dev.fgdatamap.FineGrainBlocklet;
import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.indexstore.Blocklet;
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.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.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.util.CarbonProperties;
import org.apache.carbondata.events.Event;
import org.apache.carbondata.events.OperationContext;
import org.apache.carbondata.events.OperationEventListener;

import org.apache.log4j.Logger;

/**
 * Index at the table level, user can add any number of DataMap for one table, by
 * {@code
 *   CREATE DATAMAP dm ON TABLE table
 *   USING 'class name of DataMapFactory implementation'
 * }
 * Depends on the filter condition it can prune the data (blocklet or row level).
 */
@InterfaceAudience.Internal
public final class TableDataMap extends OperationEventListener {

  private CarbonTable table;

  private AbsoluteTableIdentifier identifier;

  private DataMapSchema dataMapSchema;

  private DataMapFactory dataMapFactory;

  private BlockletDetailsFetcher blockletDetailsFetcher;

  private SegmentPropertiesFetcher segmentPropertiesFetcher;

  private static final Logger LOG =
      LogServiceFactory.getLogService(TableDataMap.class.getName());

  /**
   * It is called to initialize and load the required table datamap metadata.
   */
  TableDataMap(CarbonTable table, DataMapSchema dataMapSchema,
      DataMapFactory dataMapFactory, BlockletDetailsFetcher blockletDetailsFetcher,
      SegmentPropertiesFetcher segmentPropertiesFetcher) {
    this.identifier = table.getAbsoluteTableIdentifier();
    this.table = table;
    this.dataMapSchema = dataMapSchema;
    this.dataMapFactory = dataMapFactory;
    this.blockletDetailsFetcher = blockletDetailsFetcher;
    this.segmentPropertiesFetcher = segmentPropertiesFetcher;
  }

  public BlockletDetailsFetcher getBlockletDetailsFetcher() {
    return blockletDetailsFetcher;
  }

  public CarbonTable getTable() {
    return table;
  }

  /**
   * Pass the valid segments and prune the datamap using filter expression
   *
   * @param segments
   * @param filter
   * @return
   */
  public List<ExtendedBlocklet> prune(List<Segment> segments, final DataMapFilter filter,
      final List<PartitionSpec> partitions) throws IOException {
    final List<ExtendedBlocklet> blocklets = new ArrayList<>();
    final Map<Segment, List<DataMap>> dataMaps = dataMapFactory.getDataMaps(segments);
    // for non-filter queries
    // for filter queries
    int totalFiles = 0;
    int datamapsCount = 0;
    for (Segment segment : segments) {
      for (DataMap dataMap: dataMaps.get(segment)) {
        totalFiles += dataMap.getNumberOfEntries();
        datamapsCount++;
      }
    }
    int numOfThreadsForPruning = CarbonProperties.getNumOfThreadsForPruning();
    if (numOfThreadsForPruning == 1 || datamapsCount < numOfThreadsForPruning || totalFiles
        < CarbonCommonConstants.CARBON_DRIVER_PRUNING_MULTI_THREAD_ENABLE_FILES_COUNT) {
      // use multi-thread, only if the files are more than 0.1 million.
      // As 0.1 million files block pruning can take only 1 second.
      // Doing multi-thread for smaller values is not recommended as
      // driver should have minimum threads opened to support multiple concurrent queries.
      if (filter.isEmpty()) {
        // if filter is not passed, then return all the blocklets.
        return pruneWithoutFilter(segments, partitions, blocklets);
      }
      return pruneWithFilter(segments, filter, partitions, blocklets, dataMaps);
    }
    // handle by multi-thread
    List<ExtendedBlocklet> extendedBlocklets = pruneMultiThread(
        segments, filter, partitions, blocklets, dataMaps, totalFiles);
    return extendedBlocklets;
  }

  private List<ExtendedBlocklet> pruneWithoutFilter(List<Segment> segments,
      List<PartitionSpec> partitions, List<ExtendedBlocklet> blocklets) throws IOException {
    for (Segment segment : segments) {
      List<Blocklet> allBlocklets = blockletDetailsFetcher.getAllBlocklets(segment, partitions);
      blocklets.addAll(
          addSegmentId(blockletDetailsFetcher.getExtendedBlocklets(allBlocklets, segment),
              segment));
    }
    return blocklets;
  }

  private List<ExtendedBlocklet> pruneWithFilter(List<Segment> segments, DataMapFilter filter,
      List<PartitionSpec> partitions, List<ExtendedBlocklet> blocklets,
      Map<Segment, List<DataMap>> dataMaps) throws IOException {
    for (Segment segment : segments) {
      List<Blocklet> pruneBlocklets = new ArrayList<>();
      SegmentProperties segmentProperties = segmentPropertiesFetcher.getSegmentProperties(segment);
      if (filter.isResolvedOnSegment(segmentProperties)) {
        for (DataMap dataMap : dataMaps.get(segment)) {
          pruneBlocklets.addAll(
              dataMap.prune(filter.getResolver(), segmentProperties, partitions));
        }
      } else {
        for (DataMap dataMap : dataMaps.get(segment)) {
          pruneBlocklets.addAll(
              dataMap.prune(filter.getExpression(), segmentProperties, partitions, table));
        }
      }
      blocklets.addAll(
          addSegmentId(blockletDetailsFetcher.getExtendedBlocklets(pruneBlocklets, segment),
              segment));
    }
    return blocklets;
  }

  private List<ExtendedBlocklet> pruneMultiThread(List<Segment> segments,
      final DataMapFilter filter, final List<PartitionSpec> partitions,
      List<ExtendedBlocklet> blocklets, final Map<Segment, List<DataMap>> dataMaps,
      int totalFiles) {
    /*
     *********************************************************************************
     * Below is the example of how this part of code works.
     * consider a scenario of having 5 segments, 10 datamaps in each segment,
     * and each datamap has one record. So total 50 records.
     *
     * Datamaps in each segment looks like below.
     * s0 [0-9], s1 [0-9], s2 [0-9], s3[0-9], s4[0-9]
     *
     * If number of threads are 4. so filesPerEachThread = 50/4 = 12 files per each thread.
     *
     * SegmentDataMapGroup look like below: [SegmentId, fromIndex, toIndex]
     * In each segment only those datamaps are processed between fromIndex and toIndex.
     *
     * Final result will be: (4 list created as numOfThreadsForPruning is 4)
     * Thread1 list: s0 [0-9], s1 [0-1]  : 12 files
     * Thread2 list: s1 [2-9], s2 [0-3]  : 12 files
     * Thread3 list: s2 [4-9], s3 [0-5]  : 12 files
     * Thread4 list: s3 [6-9], s4 [0-9]  : 14 files
     * so each thread will process almost equal number of records.
     *
     *********************************************************************************
     */

    int numOfThreadsForPruning = CarbonProperties.getNumOfThreadsForPruning();
    int filesPerEachThread = totalFiles / numOfThreadsForPruning;
    int prev;
    int filesCount = 0;
    int processedFileCount = 0;
    List<List<SegmentDataMapGroup>> datamapListForEachThread =
        new ArrayList<>(numOfThreadsForPruning);
    List<SegmentDataMapGroup> segmentDataMapGroupList = new ArrayList<>();
    for (Segment segment : segments) {
      List<DataMap> eachSegmentDataMapList = dataMaps.get(segment);
      prev = 0;
      for (int i = 0; i < eachSegmentDataMapList.size(); i++) {
        DataMap dataMap = eachSegmentDataMapList.get(i);
        filesCount += dataMap.getNumberOfEntries();
        if (filesCount >= filesPerEachThread) {
          if (datamapListForEachThread.size() != numOfThreadsForPruning - 1) {
            // not the last segmentList
            segmentDataMapGroupList.add(new SegmentDataMapGroup(segment, prev, i));
            // save the last value to process in next thread
            prev = i + 1;
            datamapListForEachThread.add(segmentDataMapGroupList);
            segmentDataMapGroupList = new ArrayList<>();
            processedFileCount += filesCount;
            filesCount = 0;
          } else {
            // add remaining in the end
            processedFileCount += filesCount;
            filesCount = 0;
          }
        }
      }
      if (prev == 0 || prev != eachSegmentDataMapList.size()) {
        // if prev == 0. Add a segment's all datamaps
        // eachSegmentDataMapList.size() != prev, adding the last remaining datamaps of this segment
        segmentDataMapGroupList
            .add(new SegmentDataMapGroup(segment, prev, eachSegmentDataMapList.size() - 1));
      }
    }
    // adding the last segmentList data
    datamapListForEachThread.add(segmentDataMapGroupList);
    processedFileCount += filesCount;
    if (processedFileCount != totalFiles) {
      // this should not happen
      throw new RuntimeException(" not all the files processed ");
    }
    if (datamapListForEachThread.size() < numOfThreadsForPruning) {
      // If the total datamaps fitted in lesser number of threads than numOfThreadsForPruning.
      // Launch only that many threads where datamaps are fitted while grouping.
      LOG.info("Datamaps is distributed in " + datamapListForEachThread.size() + " threads");
      numOfThreadsForPruning = datamapListForEachThread.size();
    }
    LOG.info(
        "Number of threads selected for multi-thread block pruning is " + numOfThreadsForPruning
            + ". total files: " + totalFiles + ". total segments: " + segments.size());
    List<Future<Void>> results = new ArrayList<>(numOfThreadsForPruning);
    final Map<Segment, List<ExtendedBlocklet>> prunedBlockletMap =
        new ConcurrentHashMap<>(segments.size());
    final ExecutorService executorService = Executors.newFixedThreadPool(numOfThreadsForPruning);
    final String threadName = Thread.currentThread().getName();
    for (int i = 0; i < numOfThreadsForPruning; i++) {
      final List<SegmentDataMapGroup> segmentDataMapGroups = datamapListForEachThread.get(i);
      results.add(executorService.submit(new Callable<Void>() {
        @Override public Void call() throws IOException {
          Thread.currentThread().setName(threadName);
          for (SegmentDataMapGroup segmentDataMapGroup : segmentDataMapGroups) {
            List<ExtendedBlocklet> pruneBlocklets = new ArrayList<>();
            List<DataMap> dataMapList = dataMaps.get(segmentDataMapGroup.getSegment());
            SegmentProperties segmentProperties =
                segmentPropertiesFetcher.getSegmentPropertiesFromDataMap(dataMapList.get(0));
            Segment segment = segmentDataMapGroup.getSegment();
            if (filter.isResolvedOnSegment(segmentProperties)) {
              for (int i = segmentDataMapGroup.getFromIndex();
                   i <= segmentDataMapGroup.getToIndex(); i++) {
                List<Blocklet> dmPruneBlocklets = dataMapList.get(i).prune(
                    filter.getResolver(), segmentProperties, partitions);
                pruneBlocklets.addAll(addSegmentId(
                    blockletDetailsFetcher.getExtendedBlocklets(dmPruneBlocklets, segment),
                    segment));
              }
            } else {
              for (int i = segmentDataMapGroup.getFromIndex();
                   i <= segmentDataMapGroup.getToIndex(); i++) {
                List<Blocklet> dmPruneBlocklets = dataMapList.get(i).prune(
                    filter.getExpression(), segmentProperties, partitions, table);
                pruneBlocklets.addAll(addSegmentId(
                    blockletDetailsFetcher.getExtendedBlocklets(dmPruneBlocklets, segment),
                    segment));
              }
            }
            synchronized (prunedBlockletMap) {
              List<ExtendedBlocklet> pruneBlockletsExisting =
                  prunedBlockletMap.get(segmentDataMapGroup.getSegment());
              if (pruneBlockletsExisting != null) {
                pruneBlockletsExisting.addAll(pruneBlocklets);
              } else {
                prunedBlockletMap.put(segmentDataMapGroup.getSegment(), pruneBlocklets);
              }
            }
          }
          return null;
        }
      }));
    }
    executorService.shutdown();
    try {
      executorService.awaitTermination(2, TimeUnit.HOURS);
    } catch (InterruptedException e) {
      LOG.error("Error in pruning datamap in multi-thread: " + e.getMessage());
    }
    // check for error
    for (Future<Void> result : results) {
      try {
        result.get();
      } catch (InterruptedException | ExecutionException e) {
        throw new RuntimeException(e);
      }
    }
    for (Map.Entry<Segment, List<ExtendedBlocklet>> entry : prunedBlockletMap.entrySet()) {
      blocklets.addAll(entry.getValue());
    }
    return blocklets;
  }

  private List<ExtendedBlocklet> addSegmentId(List<ExtendedBlocklet> pruneBlocklets,
      Segment segment) {
    for (ExtendedBlocklet blocklet : pruneBlocklets) {
      blocklet.setSegment(segment);
    }
    return pruneBlocklets;
  }
  /**
   * This is used for making the datamap distributable.
   * It takes the valid segments and returns all the datamaps as distributable objects so that
   * it can be distributed across machines.
   *
   * @return
   */
  public List<DataMapDistributable> toDistributable(List<Segment> segments) throws IOException {
    List<DataMapDistributable> distributables = new ArrayList<>();
    for (Segment segment : segments) {
      distributables.addAll(dataMapFactory.toDistributable(segment));
    }
    return distributables;
  }

  /**
   * This method returns all the datamaps corresponding to the distributable object
   *
   * @param distributable
   * @return
   * @throws IOException
   */
  public List<DataMap> getTableDataMaps(DataMapDistributable distributable) throws IOException {
    return dataMapFactory.getDataMaps(distributable);
  }

  /**
   * This method is used from any machine after it is distributed. It takes the distributable object
   * to prune the filters.
   *
   * @param distributable
   * @param filterExp
   * @return
   */
  public List<ExtendedBlocklet> prune(List<DataMap> dataMaps, DataMapDistributable distributable,
      FilterResolverIntf filterExp, List<PartitionSpec> partitions) throws IOException {
    List<ExtendedBlocklet> detailedBlocklets = new ArrayList<>();
    List<Blocklet> blocklets = new ArrayList<>();
    for (DataMap dataMap : dataMaps) {
      blocklets.addAll(dataMap.prune(filterExp,
          segmentPropertiesFetcher.getSegmentProperties(distributable.getSegment()),
          partitions));
    }
    BlockletSerializer serializer = new BlockletSerializer();
    String writePath =
        identifier.getTablePath() + CarbonCommonConstants.FILE_SEPARATOR + dataMapSchema
            .getDataMapName();
    if (dataMapFactory.getDataMapLevel() == DataMapLevel.FG) {
      FileFactory.mkdirs(writePath, FileFactory.getFileType(writePath));
    }
    for (Blocklet blocklet : blocklets) {
      ExtendedBlocklet detailedBlocklet = blockletDetailsFetcher
          .getExtendedBlocklet(blocklet, distributable.getSegment());
      if (dataMapFactory.getDataMapLevel() == DataMapLevel.FG) {
        String blockletwritePath =
            writePath + CarbonCommonConstants.FILE_SEPARATOR + System.nanoTime();
        detailedBlocklet.setDataMapWriterPath(blockletwritePath);
        serializer.serializeBlocklet((FineGrainBlocklet) blocklet, blockletwritePath);
      }
      detailedBlocklet.setSegment(distributable.getSegment());
      detailedBlocklets.add(detailedBlocklet);
    }
    return detailedBlocklets;
  }

  /**
   * Clear only the datamaps of the segments
   * @param segmentIds list of segmentIds to be cleared from cache.
   */
  public void clear(List<String> segmentIds) {
    for (String segment: segmentIds) {
      dataMapFactory.clear(segment);
    }
  }

  /**
   * Clears all datamap
   */
  public void clear() {
    if (null != dataMapFactory) {
      dataMapFactory.clear();
    }
  }

  /**
   * delete only the datamaps of the segments
   */
  public void deleteDatamapData(List<Segment> segments) throws IOException {
    for (Segment segment: segments) {
      dataMapFactory.deleteDatamapData(segment);
    }
  }
  /**
   * delete datamap data if any
   */
  public void deleteDatamapData() {
    dataMapFactory.deleteDatamapData();
  }

  /**
   * delete datamap data for a segment if any
   */
  public void deleteSegmentDatamapData(String segmentNo) throws IOException {
    dataMapFactory.deleteSegmentDatamapData(segmentNo);
  }

  public DataMapSchema getDataMapSchema() {
    return dataMapSchema;
  }

  public DataMapFactory getDataMapFactory() {
    return dataMapFactory;
  }

  @Override public void onEvent(Event event, OperationContext opContext) throws Exception {
    dataMapFactory.fireEvent(event);
  }

  /**
   * Prune the datamap of the given segments and return the Map of blocklet path and row count
   *
   * @param segments
   * @param partitions
   * @return
   * @throws IOException
   */
  public Map<String, Long> getBlockRowCount(List<Segment> segments,
      final List<PartitionSpec> partitions, TableDataMap defaultDataMap)
      throws IOException {
    Map<String, Long> blockletToRowCountMap = new HashMap<>();
    for (Segment segment : segments) {
      List<CoarseGrainDataMap> dataMaps = defaultDataMap.getDataMapFactory().getDataMaps(segment);
      for (CoarseGrainDataMap dataMap : dataMaps) {
        dataMap.getRowCountForEachBlock(segment, partitions, blockletToRowCountMap);
      }
    }
    return blockletToRowCountMap;
  }

  /**
   * Prune the datamap of the given segments and return the Map of blocklet path and row count
   *
   * @param segments
   * @param partitions
   * @return
   * @throws IOException
   */
  public long getRowCount(List<Segment> segments, final List<PartitionSpec> partitions,
      TableDataMap defaultDataMap) throws IOException {
    long totalRowCount = 0L;
    for (Segment segment : segments) {
      List<CoarseGrainDataMap> dataMaps = defaultDataMap.getDataMapFactory().getDataMaps(segment);
      for (CoarseGrainDataMap dataMap : dataMaps) {
        totalRowCount += dataMap.getRowCount(segment, partitions);
      }
    }
    return totalRowCount;
  }

}

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.datamap;
18
19	import java.io.IOException;
20	import java.util.ArrayList;
21	import java.util.HashMap;
22	import java.util.List;
23	import java.util.Map;
24	import java.util.concurrent.Callable;
25	import java.util.concurrent.ConcurrentHashMap;
26	import java.util.concurrent.ExecutionException;
27	import java.util.concurrent.ExecutorService;
28	import java.util.concurrent.Executors;
29	import java.util.concurrent.Future;
30	import java.util.concurrent.TimeUnit;
31
32	import org.apache.carbondata.common.annotations.InterfaceAudience;
33	import org.apache.carbondata.common.logging.LogServiceFactory;
34	import org.apache.carbondata.core.constants.CarbonCommonConstants;
35	import org.apache.carbondata.core.datamap.dev.BlockletSerializer;
36	import org.apache.carbondata.core.datamap.dev.DataMap;
37	import org.apache.carbondata.core.datamap.dev.DataMapFactory;
38	import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMap;
39	import org.apache.carbondata.core.datamap.dev.fgdatamap.FineGrainBlocklet;
40	import org.apache.carbondata.core.datastore.block.SegmentProperties;
41	import org.apache.carbondata.core.datastore.impl.FileFactory;
42	import org.apache.carbondata.core.indexstore.Blocklet;
43	import org.apache.carbondata.core.indexstore.BlockletDetailsFetcher;
44	import org.apache.carbondata.core.indexstore.ExtendedBlocklet;
45	import org.apache.carbondata.core.indexstore.PartitionSpec;
46	import org.apache.carbondata.core.indexstore.SegmentPropertiesFetcher;
47	import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
48	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
49	import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
50	import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
51	import org.apache.carbondata.core.util.CarbonProperties;
52	import org.apache.carbondata.events.Event;
53	import org.apache.carbondata.events.OperationContext;
54	import org.apache.carbondata.events.OperationEventListener;
55
56	import org.apache.log4j.Logger;
57
58	/**
59	* Index at the table level, user can add any number of DataMap for one table, by
60	* {@code
61	* CREATE DATAMAP dm ON TABLE table
62	* USING 'class name of DataMapFactory implementation'
63	* }
64	* Depends on the filter condition it can prune the data (blocklet or row level).
65	*/
66	@InterfaceAudience.Internal
67	public final class TableDataMap extends OperationEventListener {
68
69	private CarbonTable table;
70
71	private AbsoluteTableIdentifier identifier;
72
73	private DataMapSchema dataMapSchema;
74
75	private DataMapFactory dataMapFactory;
76
77	private BlockletDetailsFetcher blockletDetailsFetcher;
78
79	private SegmentPropertiesFetcher segmentPropertiesFetcher;
80
81	private static final Logger LOG =	1✔
82	LogServiceFactory.getLogService(TableDataMap.class.getName());	1✔
83
84	/**
85	* It is called to initialize and load the required table datamap metadata.
86	*/
87	TableDataMap(CarbonTable table, DataMapSchema dataMapSchema,
88	DataMapFactory dataMapFactory, BlockletDetailsFetcher blockletDetailsFetcher,
89	SegmentPropertiesFetcher segmentPropertiesFetcher) {	1✔
90	this.identifier = table.getAbsoluteTableIdentifier();	1✔
91	this.table = table;	1✔
92	this.dataMapSchema = dataMapSchema;	1✔
93	this.dataMapFactory = dataMapFactory;	1✔
94	this.blockletDetailsFetcher = blockletDetailsFetcher;	1✔
95	this.segmentPropertiesFetcher = segmentPropertiesFetcher;	1✔
96	}	1✔
97
98	public BlockletDetailsFetcher getBlockletDetailsFetcher() {
99	return blockletDetailsFetcher;	1✔
100	}
101
102	public CarbonTable getTable() {
103	return table;	1✔
104	}
105
106	/**
107	* Pass the valid segments and prune the datamap using filter expression
108	*
109	* @param segments
110	* @param filter
111	* @return
112	*/
113	public List<ExtendedBlocklet> prune(List<Segment> segments, final DataMapFilter filter,
114	final List<PartitionSpec> partitions) throws IOException {
115	final List<ExtendedBlocklet> blocklets = new ArrayList<>();	1✔
116	final Map<Segment, List<DataMap>> dataMaps = dataMapFactory.getDataMaps(segments);	1✔
117	// for non-filter queries
118	// for filter queries
119	int totalFiles = 0;	1✔
120	int datamapsCount = 0;	1✔
121	for (Segment segment : segments) {	1✔
122	for (DataMap dataMap: dataMaps.get(segment)) {	1✔
123	totalFiles += dataMap.getNumberOfEntries();	1✔
124	datamapsCount++;	1✔
125	}	1✔
126	}	1✔
127	int numOfThreadsForPruning = CarbonProperties.getNumOfThreadsForPruning();	1✔
128	if (numOfThreadsForPruning == 1 \|\| datamapsCount < numOfThreadsForPruning \|\| totalFiles	1✔
129	< CarbonCommonConstants.CARBON_DRIVER_PRUNING_MULTI_THREAD_ENABLE_FILES_COUNT) {
130	// use multi-thread, only if the files are more than 0.1 million.
131	// As 0.1 million files block pruning can take only 1 second.
132	// Doing multi-thread for smaller values is not recommended as
133	// driver should have minimum threads opened to support multiple concurrent queries.
134	if (filter.isEmpty()) {	1✔
135	// if filter is not passed, then return all the blocklets.
136	return pruneWithoutFilter(segments, partitions, blocklets);	1✔
137	}
138	return pruneWithFilter(segments, filter, partitions, blocklets, dataMaps);	1✔
139	}
140	// handle by multi-thread
141	List<ExtendedBlocklet> extendedBlocklets = pruneMultiThread(	×
142	segments, filter, partitions, blocklets, dataMaps, totalFiles);
143	return extendedBlocklets;	×
144	}
145
146	private List<ExtendedBlocklet> pruneWithoutFilter(List<Segment> segments,
147	List<PartitionSpec> partitions, List<ExtendedBlocklet> blocklets) throws IOException {
148	for (Segment segment : segments) {	1✔
149	List<Blocklet> allBlocklets = blockletDetailsFetcher.getAllBlocklets(segment, partitions);	1✔
150	blocklets.addAll(	1✔
151	addSegmentId(blockletDetailsFetcher.getExtendedBlocklets(allBlocklets, segment),	1✔
152	segment));
153	}	1✔
154	return blocklets;	1✔
155	}
156
157	private List<ExtendedBlocklet> pruneWithFilter(List<Segment> segments, DataMapFilter filter,
158	List<PartitionSpec> partitions, List<ExtendedBlocklet> blocklets,
159	Map<Segment, List<DataMap>> dataMaps) throws IOException {
160	for (Segment segment : segments) {	1✔
161	List<Blocklet> pruneBlocklets = new ArrayList<>();	1✔
162	SegmentProperties segmentProperties = segmentPropertiesFetcher.getSegmentProperties(segment);	1✔
163	if (filter.isResolvedOnSegment(segmentProperties)) {	1✔
164	for (DataMap dataMap : dataMaps.get(segment)) {	1✔
165	pruneBlocklets.addAll(	1✔
166	dataMap.prune(filter.getResolver(), segmentProperties, partitions));	1✔
167	}	1✔
168	} else {
169	for (DataMap dataMap : dataMaps.get(segment)) {	1✔
170	pruneBlocklets.addAll(	1✔
171	dataMap.prune(filter.getExpression(), segmentProperties, partitions, table));	1✔
172	}	1✔
173	}
174	blocklets.addAll(	1✔
175	addSegmentId(blockletDetailsFetcher.getExtendedBlocklets(pruneBlocklets, segment),	1✔
176	segment));
177	}	1✔
178	return blocklets;	1✔
179	}
180
181	private List<ExtendedBlocklet> pruneMultiThread(List<Segment> segments,
182	final DataMapFilter filter, final List<PartitionSpec> partitions,
183	List<ExtendedBlocklet> blocklets, final Map<Segment, List<DataMap>> dataMaps,
184	int totalFiles) {
185	/*
186	*********************************************************************************
187	* Below is the example of how this part of code works.
188	* consider a scenario of having 5 segments, 10 datamaps in each segment,
189	* and each datamap has one record. So total 50 records.
190	*
191	* Datamaps in each segment looks like below.
192	* s0 [0-9], s1 [0-9], s2 [0-9], s3[0-9], s4[0-9]
193	*
194	* If number of threads are 4. so filesPerEachThread = 50/4 = 12 files per each thread.
195	*
196	* SegmentDataMapGroup look like below: [SegmentId, fromIndex, toIndex]
197	* In each segment only those datamaps are processed between fromIndex and toIndex.
198	*
199	* Final result will be: (4 list created as numOfThreadsForPruning is 4)
200	* Thread1 list: s0 [0-9], s1 [0-1] : 12 files
201	* Thread2 list: s1 [2-9], s2 [0-3] : 12 files
202	* Thread3 list: s2 [4-9], s3 [0-5] : 12 files
203	* Thread4 list: s3 [6-9], s4 [0-9] : 14 files
204	* so each thread will process almost equal number of records.
205	*
206	*********************************************************************************
207	*/
208
209	int numOfThreadsForPruning = CarbonProperties.getNumOfThreadsForPruning();	×
210	int filesPerEachThread = totalFiles / numOfThreadsForPruning;	×
211	int prev;
212	int filesCount = 0;	×
213	int processedFileCount = 0;	×
214	List<List<SegmentDataMapGroup>> datamapListForEachThread =	×
215	new ArrayList<>(numOfThreadsForPruning);
216	List<SegmentDataMapGroup> segmentDataMapGroupList = new ArrayList<>();	×
217	for (Segment segment : segments) {	×
218	List<DataMap> eachSegmentDataMapList = dataMaps.get(segment);	×
219	prev = 0;	×
220	for (int i = 0; i < eachSegmentDataMapList.size(); i++) {	×
221	DataMap dataMap = eachSegmentDataMapList.get(i);	×
222	filesCount += dataMap.getNumberOfEntries();	×
223	if (filesCount >= filesPerEachThread) {	×
224	if (datamapListForEachThread.size() != numOfThreadsForPruning - 1) {	×
225	// not the last segmentList
226	segmentDataMapGroupList.add(new SegmentDataMapGroup(segment, prev, i));	×
227	// save the last value to process in next thread
228	prev = i + 1;	×
229	datamapListForEachThread.add(segmentDataMapGroupList);	×
230	segmentDataMapGroupList = new ArrayList<>();	×
231	processedFileCount += filesCount;	×
232	filesCount = 0;	×
233	} else {
234	// add remaining in the end
235	processedFileCount += filesCount;	×
236	filesCount = 0;	×
237	}
238	}
239	}
240	if (prev == 0 \|\| prev != eachSegmentDataMapList.size()) {	×
241	// if prev == 0. Add a segment's all datamaps
242	// eachSegmentDataMapList.size() != prev, adding the last remaining datamaps of this segment
243	segmentDataMapGroupList	×
244	.add(new SegmentDataMapGroup(segment, prev, eachSegmentDataMapList.size() - 1));	×
245	}
246	}	×
247	// adding the last segmentList data
248	datamapListForEachThread.add(segmentDataMapGroupList);	×
249	processedFileCount += filesCount;	×
250	if (processedFileCount != totalFiles) {	×
251	// this should not happen
252	throw new RuntimeException(" not all the files processed ");	×
253	}
254	if (datamapListForEachThread.size() < numOfThreadsForPruning) {	×
255	// If the total datamaps fitted in lesser number of threads than numOfThreadsForPruning.
256	// Launch only that many threads where datamaps are fitted while grouping.
257	LOG.info("Datamaps is distributed in " + datamapListForEachThread.size() + " threads");	×
258	numOfThreadsForPruning = datamapListForEachThread.size();	×
259	}
260	LOG.info(	×
261	"Number of threads selected for multi-thread block pruning is " + numOfThreadsForPruning
262	+ ". total files: " + totalFiles + ". total segments: " + segments.size());	×
263	List<Future<Void>> results = new ArrayList<>(numOfThreadsForPruning);	×
264	final Map<Segment, List<ExtendedBlocklet>> prunedBlockletMap =	×
265	new ConcurrentHashMap<>(segments.size());	×
266	final ExecutorService executorService = Executors.newFixedThreadPool(numOfThreadsForPruning);	×
267	final String threadName = Thread.currentThread().getName();	×
268	for (int i = 0; i < numOfThreadsForPruning; i++) {	×
269	final List<SegmentDataMapGroup> segmentDataMapGroups = datamapListForEachThread.get(i);	×
270	results.add(executorService.submit(new Callable<Void>() {	×
271	@Override public Void call() throws IOException {
272	Thread.currentThread().setName(threadName);	×
273	for (SegmentDataMapGroup segmentDataMapGroup : segmentDataMapGroups) {	×
274	List<ExtendedBlocklet> pruneBlocklets = new ArrayList<>();	×
275	List<DataMap> dataMapList = dataMaps.get(segmentDataMapGroup.getSegment());	×
276	SegmentProperties segmentProperties =	×
277	segmentPropertiesFetcher.getSegmentPropertiesFromDataMap(dataMapList.get(0));	×
278	Segment segment = segmentDataMapGroup.getSegment();	×
279	if (filter.isResolvedOnSegment(segmentProperties)) {	×
280	for (int i = segmentDataMapGroup.getFromIndex();	×
281	i <= segmentDataMapGroup.getToIndex(); i++) {	×
282	List<Blocklet> dmPruneBlocklets = dataMapList.get(i).prune(	×
283	filter.getResolver(), segmentProperties, partitions);	×
284	pruneBlocklets.addAll(addSegmentId(	×
285	blockletDetailsFetcher.getExtendedBlocklets(dmPruneBlocklets, segment),	×
286	segment));
287	}
288	} else {
289	for (int i = segmentDataMapGroup.getFromIndex();	×
290	i <= segmentDataMapGroup.getToIndex(); i++) {	×
291	List<Blocklet> dmPruneBlocklets = dataMapList.get(i).prune(	×
292	filter.getExpression(), segmentProperties, partitions, table);	×
293	pruneBlocklets.addAll(addSegmentId(	×
294	blockletDetailsFetcher.getExtendedBlocklets(dmPruneBlocklets, segment),	×
295	segment));
296	}
297	}
298	synchronized (prunedBlockletMap) {	×
299	List<ExtendedBlocklet> pruneBlockletsExisting =	×
300	prunedBlockletMap.get(segmentDataMapGroup.getSegment());	×
301	if (pruneBlockletsExisting != null) {	×
302	pruneBlockletsExisting.addAll(pruneBlocklets);	×
303	} else {
304	prunedBlockletMap.put(segmentDataMapGroup.getSegment(), pruneBlocklets);	×
305	}
306	}	×
307	}	×
308	return null;	×
309	}
310	}));
311	}
312	executorService.shutdown();	×
313	try {
314	executorService.awaitTermination(2, TimeUnit.HOURS);	×
315	} catch (InterruptedException e) {	×
316	LOG.error("Error in pruning datamap in multi-thread: " + e.getMessage());	×
317	}	×
318	// check for error
319	for (Future<Void> result : results) {	×
320	try {
321	result.get();	×
322	} catch (InterruptedException \| ExecutionException e) {	×
323	throw new RuntimeException(e);	×
324	}	×
325	}	×
326	for (Map.Entry<Segment, List<ExtendedBlocklet>> entry : prunedBlockletMap.entrySet()) {	×
327	blocklets.addAll(entry.getValue());	×
328	}	×
329	return blocklets;	×
330	}
331
332	private List<ExtendedBlocklet> addSegmentId(List<ExtendedBlocklet> pruneBlocklets,
333	Segment segment) {
334	for (ExtendedBlocklet blocklet : pruneBlocklets) {	1✔
335	blocklet.setSegment(segment);	1✔
336	}	1✔
337	return pruneBlocklets;	1✔
338	}
339	/**
340	* This is used for making the datamap distributable.
341	* It takes the valid segments and returns all the datamaps as distributable objects so that
342	* it can be distributed across machines.
343	*
344	* @return
345	*/
346	public List<DataMapDistributable> toDistributable(List<Segment> segments) throws IOException {
347	List<DataMapDistributable> distributables = new ArrayList<>();	1✔
348	for (Segment segment : segments) {	1✔
349	distributables.addAll(dataMapFactory.toDistributable(segment));	1✔
350	}	1✔
351	return distributables;	1✔
352	}
353
354	/**
355	* This method returns all the datamaps corresponding to the distributable object
356	*
357	* @param distributable
358	* @return
359	* @throws IOException
360	*/
361	public List<DataMap> getTableDataMaps(DataMapDistributable distributable) throws IOException {
362	return dataMapFactory.getDataMaps(distributable);	1✔
363	}
364
365	/**
366	* This method is used from any machine after it is distributed. It takes the distributable object
367	* to prune the filters.
368	*
369	* @param distributable
370	* @param filterExp
371	* @return
372	*/
373	public List<ExtendedBlocklet> prune(List<DataMap> dataMaps, DataMapDistributable distributable,
374	FilterResolverIntf filterExp, List<PartitionSpec> partitions) throws IOException {
375	List<ExtendedBlocklet> detailedBlocklets = new ArrayList<>();	1✔
376	List<Blocklet> blocklets = new ArrayList<>();	1✔
377	for (DataMap dataMap : dataMaps) {	1✔
378	blocklets.addAll(dataMap.prune(filterExp,	1✔
379	segmentPropertiesFetcher.getSegmentProperties(distributable.getSegment()),	1✔
380	partitions));
381	}	1✔
382	BlockletSerializer serializer = new BlockletSerializer();	1✔
383	String writePath =	1✔
384	identifier.getTablePath() + CarbonCommonConstants.FILE_SEPARATOR + dataMapSchema	1✔
385	.getDataMapName();	1✔
386	if (dataMapFactory.getDataMapLevel() == DataMapLevel.FG) {	1✔
387	FileFactory.mkdirs(writePath, FileFactory.getFileType(writePath));	1✔
388	}
389	for (Blocklet blocklet : blocklets) {	1✔
390	ExtendedBlocklet detailedBlocklet = blockletDetailsFetcher	1✔
391	.getExtendedBlocklet(blocklet, distributable.getSegment());	1✔
392	if (dataMapFactory.getDataMapLevel() == DataMapLevel.FG) {	1✔
393	String blockletwritePath =	1✔
394	writePath + CarbonCommonConstants.FILE_SEPARATOR + System.nanoTime();	1✔
395	detailedBlocklet.setDataMapWriterPath(blockletwritePath);	1✔
396	serializer.serializeBlocklet((FineGrainBlocklet) blocklet, blockletwritePath);	1✔
397	}
398	detailedBlocklet.setSegment(distributable.getSegment());	1✔
399	detailedBlocklets.add(detailedBlocklet);	1✔
400	}	1✔
401	return detailedBlocklets;	1✔
402	}
403
404	/**
405	* Clear only the datamaps of the segments
406	* @param segmentIds list of segmentIds to be cleared from cache.
407	*/
408	public void clear(List<String> segmentIds) {
409	for (String segment: segmentIds) {	1✔
410	dataMapFactory.clear(segment);	1✔
411	}	1✔
412	}	1✔
413
414	/**
415	* Clears all datamap
416	*/
417	public void clear() {
418	if (null != dataMapFactory) {	1✔
419	dataMapFactory.clear();	1✔
420	}
421	}	1✔
422
423	/**
424	* delete only the datamaps of the segments
425	*/
426	public void deleteDatamapData(List<Segment> segments) throws IOException {
427	for (Segment segment: segments) {	1✔
428	dataMapFactory.deleteDatamapData(segment);	1✔
429	}	1✔
430	}	1✔
431	/**
432	* delete datamap data if any
433	*/
434	public void deleteDatamapData() {
435	dataMapFactory.deleteDatamapData();	1✔
436	}	1✔
437
438	/**
439	* delete datamap data for a segment if any
440	*/
441	public void deleteSegmentDatamapData(String segmentNo) throws IOException {
442	dataMapFactory.deleteSegmentDatamapData(segmentNo);	1✔
443	}	1✔
444
445	public DataMapSchema getDataMapSchema() {
446	return dataMapSchema;	1✔
447	}
448
449	public DataMapFactory getDataMapFactory() {
450	return dataMapFactory;	1✔
451	}
452
453	@Override public void onEvent(Event event, OperationContext opContext) throws Exception {
454	dataMapFactory.fireEvent(event);	×
455	}	×
456
457	/**
458	* Prune the datamap of the given segments and return the Map of blocklet path and row count
459	*
460	* @param segments
461	* @param partitions
462	* @return
463	* @throws IOException
464	*/
465	public Map<String, Long> getBlockRowCount(List<Segment> segments,
466	final List<PartitionSpec> partitions, TableDataMap defaultDataMap)
467	throws IOException {
468	Map<String, Long> blockletToRowCountMap = new HashMap<>();	1✔
469	for (Segment segment : segments) {	1✔
470	List<CoarseGrainDataMap> dataMaps = defaultDataMap.getDataMapFactory().getDataMaps(segment);	1✔
471	for (CoarseGrainDataMap dataMap : dataMaps) {	1✔
472	dataMap.getRowCountForEachBlock(segment, partitions, blockletToRowCountMap);	1✔
473	}	1✔
474	}	1✔
475	return blockletToRowCountMap;	1✔
476	}
477
478	/**
479	* Prune the datamap of the given segments and return the Map of blocklet path and row count
480	*
481	* @param segments
482	* @param partitions
483	* @return
484	* @throws IOException
485	*/
486	public long getRowCount(List<Segment> segments, final List<PartitionSpec> partitions,
487	TableDataMap defaultDataMap) throws IOException {
488	long totalRowCount = 0L;	1✔
489	for (Segment segment : segments) {	1✔
490	List<CoarseGrainDataMap> dataMaps = defaultDataMap.getDataMapFactory().getDataMaps(segment);	1✔
491	for (CoarseGrainDataMap dataMap : dataMaps) {	1✔
492	totalRowCount += dataMap.getRowCount(segment, partitions);	1✔
493	}	1✔
494	}	1✔
495	return totalRowCount;	1✔
496	}
497
498	}

apache / carbondata / 182

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