10130

Build # 10130

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/69a7680311d0ba7e91f23a1553585a86549f988c">69a768031<a href="https://github.com/apache/carbondata/commit/69a7680311d0ba7e91f23a1553585a86549f988c">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/6bd1194e3a4a995089f33a0ffef6b59ab2ef49e2">6bd1194e3</a>">6bd1194e3</a><a href="https://github.com/apache/carbondata/commit/69a7680311d0ba7e91f23a1553585a86549f988c"> into 96ce00758">96ce00758</a>

Run Details

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

61429 of 77713 relevant lines covered (79.05%)

1.05 hits per line

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

56.6

/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.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 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(AbsoluteTableIdentifier identifier, DataMapSchema dataMapSchema,
      DataMapFactory dataMapFactory, BlockletDetailsFetcher blockletDetailsFetcher,
      SegmentPropertiesFetcher segmentPropertiesFetcher) {
    this.identifier = identifier;
    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, identifier));
        }
      }
      blocklets.addAll(addSegmentId(
          blockletDetailsFetcher.getExtendedBlocklets(pruneBlocklets, segment),
          segment.toString()));
    }
    return blocklets;
  }

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

apache / carbondata / 10130

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