2793

Build # 2793

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/2be0440a1660ae37f793ed3eb52c8b7e9df96656">2be0440a1<a href="https://github.com/apache/carbondata/commit/2be0440a1660ae37f793ed3eb52c8b7e9df96656">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/6efd8fb5b333442e54b39f3944f62fc53302c0c9">6efd8fb5b</a>">6efd8fb5b</a><a href="https://github.com/apache/carbondata/commit/2be0440a1660ae37f793ed3eb52c8b7e9df96656"> into 2ecf30ca0">2ecf30ca0</a>

Run Details

65 of 65 new or added lines in 8 files covered. (100.0%)

62465 of 78942 relevant lines covered (79.13%)

1.05 hits per line

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

57.01

/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.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.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.expression.Expression;
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;
  }


  /**
   * Pass the valid segments and prune the datamap using filter expression
   *
   * @param segments
   * @param filterExp
   * @return
   */
  public List<ExtendedBlocklet> prune(List<Segment> segments, Expression filterExp,
      List<PartitionSpec> partitions) throws IOException {
    List<ExtendedBlocklet> blocklets = new ArrayList<>();
    SegmentProperties segmentProperties;
    Map<Segment, List<DataMap>> dataMaps = dataMapFactory.getDataMaps(segments);
    for (Segment segment : segments) {
      List<Blocklet> pruneBlocklets = new ArrayList<>();
      // if filter is not passed then return all the blocklets
      if (filterExp == null) {
        pruneBlocklets = blockletDetailsFetcher.getAllBlocklets(segment, partitions);
      } else {
        segmentProperties = segmentPropertiesFetcher.getSegmentProperties(segment);
        for (DataMap dataMap : dataMaps.get(segment)) {
          pruneBlocklets.addAll(dataMap
              .prune(filterExp, segmentProperties, partitions, table));
        }
      }
      blocklets.addAll(addSegmentId(
          blockletDetailsFetcher.getExtendedBlocklets(pruneBlocklets, segment),
          segment.toString()));
    }
    return blocklets;
  }

  public CarbonTable getTable() {
    return table;
  }

  /**
   * Pass the valid segments and prune the datamap using filter expression
   *
   * @param segments
   * @param filterExp
   * @return
   */
  public List<ExtendedBlocklet> prune(List<Segment> segments, final FilterResolverIntf filterExp,
      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
    if (filterExp == null) {
      // if filter is not passed, then return all the blocklets.
      return pruneWithoutFilter(segments, partitions, blocklets);
    }
    // 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 = 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.
      return pruneWithFilter(segments, filterExp, partitions, blocklets, dataMaps);
    }
    // handle by multi-thread
    return pruneWithFilterMultiThread(segments, filterExp, partitions, blocklets, dataMaps,
        totalFiles);
  }

  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.toString()));
    }
    return blocklets;
  }

  private List<ExtendedBlocklet> pruneWithFilter(List<Segment> segments,
      FilterResolverIntf filterExp, 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);
      for (DataMap dataMap : dataMaps.get(segment)) {
        pruneBlocklets.addAll(dataMap.prune(filterExp, segmentProperties, partitions));
      }
      blocklets.addAll(
          addSegmentId(blockletDetailsFetcher.getExtendedBlocklets(pruneBlocklets, segment),
              segment.toString()));
    }
    return blocklets;
  }

  private List<ExtendedBlocklet> pruneWithFilterMultiThread(List<Segment> segments,
      final FilterResolverIntf filterExp, 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 = getNumOfThreadsForPruning();
    LOG.info(
        "Number of threads selected for multi-thread block pruning is " + numOfThreadsForPruning
            + ". total files: " + totalFiles + ". total segments: " + segments.size());
    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 ");
    }
    List<Future<Void>> results = new ArrayList<>(numOfThreadsForPruning);
    final Map<Segment, List<Blocklet>> 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<Blocklet> pruneBlocklets = new ArrayList<>();
            List<DataMap> dataMapList = dataMaps.get(segmentDataMapGroup.getSegment());
            for (int i = segmentDataMapGroup.getFromIndex();
                 i <= segmentDataMapGroup.getToIndex(); i++) {
              pruneBlocklets.addAll(dataMapList.get(i).prune(filterExp,
                  segmentPropertiesFetcher.getSegmentProperties(segmentDataMapGroup.getSegment()),
                  partitions));
            }
            synchronized (prunedBlockletMap) {
              List<Blocklet> 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<Blocklet>> entry : prunedBlockletMap.entrySet()) {
      try {
        blocklets.addAll(addSegmentId(
            blockletDetailsFetcher.getExtendedBlocklets(entry.getValue(), entry.getKey()),
            entry.getKey().toString()));
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
    }
    return blocklets;
  }

  private int getNumOfThreadsForPruning() {
    int numOfThreadsForPruning = Integer.parseInt(CarbonProperties.getInstance()
        .getProperty(CarbonCommonConstants.CARBON_MAX_DRIVER_THREADS_FOR_BLOCK_PRUNING,
            CarbonCommonConstants.CARBON_MAX_DRIVER_THREADS_FOR_BLOCK_PRUNING_DEFAULT));
    if (numOfThreadsForPruning > Integer
        .parseInt(CarbonCommonConstants.CARBON_MAX_DRIVER_THREADS_FOR_BLOCK_PRUNING_DEFAULT)
        || numOfThreadsForPruning < 1) {
      LOG.info("Invalid value for carbon.max.driver.threads.for.block.pruning, value :"
          + numOfThreadsForPruning + " .using the default threads : "
          + CarbonCommonConstants.CARBON_MAX_DRIVER_THREADS_FOR_BLOCK_PRUNING_DEFAULT);
      numOfThreadsForPruning = Integer
          .parseInt(CarbonCommonConstants.CARBON_MAX_DRIVER_THREADS_FOR_BLOCK_PRUNING_DEFAULT);
    }
    return numOfThreadsForPruning;
  }

  private List<ExtendedBlocklet> addSegmentId(List<ExtendedBlocklet> pruneBlocklets,
      String segmentId) {
    for (ExtendedBlocklet blocklet : pruneBlocklets) {
      blocklet.setSegmentId(segmentId);
    }
    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) {
      List<DataMapDistributable> list =
          dataMapFactory.toDistributable(segment);
      for (DataMapDistributable distributable : list) {
        distributable.setDataMapSchema(dataMapSchema);
        distributable.setSegment(segment);
        distributable.setTablePath(identifier.getTablePath());
      }
      distributables.addAll(list);
    }
    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.setSegmentId(distributable.getSegment().toString());
      detailedBlocklets.add(detailedBlocklet);
    }
    return detailedBlocklets;
  }

  /**
   * Clear only the datamaps of the segments
   * @param segments
   */
  public void clear(List<Segment> segments) {
    for (Segment segment: segments) {
      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();
  }

  public DataMapSchema getDataMapSchema() {
    return dataMapSchema;
  }

  public DataMapFactory getDataMapFactory() {
    return dataMapFactory;
  }

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

  /**
   * Method to prune the segments based on task min/max values
   *
   * @param segments
   * @param filterExp
   * @return
   * @throws IOException
   */
  public List<Segment> pruneSegments(List<Segment> segments, FilterResolverIntf filterExp)
      throws IOException {
    List<Segment> prunedSegments = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
    for (Segment segment : segments) {
      List<DataMap> dataMaps = dataMapFactory.getDataMaps(segment);
      for (DataMap dataMap : dataMaps) {
        if (dataMap.isScanRequired(filterExp)) {
          // If any one task in a given segment contains the data that means the segment need to
          // be scanned and we need to validate further data maps in the same segment
          prunedSegments.add(segment);
          break;
        }
      }
    }
    return prunedSegments;
  }
}

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

apache / carbondata / 2793

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