3333

Build # 3333

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/9d73da53df6f47833421481e05581cd4c67ecf6f">9d73da53d">&lt;a href=&quot;https://github.com/apache/carbondata/commit/</a><a class="double-link" href="https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/9d73da53df6f47833421481e05581cd4c67ecf6f">9d73da53d</a><a href="https://github.com/apache/carbondata/commit/9d73da53df6f47833421481e05581cd4c67ecf6f">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/9d73da53df6f47833421481e05581cd4c67ecf6f&quot;&gt;9d73da53d&lt;/a&gt;&lt;a href=&quot;https://github.com/apache/carbondata/commit/9d73da53df6f47833421481e05581cd4c67ecf6f&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&amp;amp;lt;a ... (continued)

Run Details

71 of 71 new or added lines in 7 files covered. (100.0%)

63752 of 80412 relevant lines covered (79.28%)

1.04 hits per line

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

97.1

/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonCompactionExecutor.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.processing.merger;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.carbondata.common.CarbonIterator;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.cache.dictionary.Dictionary;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.datastore.block.TableBlockInfo;
import org.apache.carbondata.core.datastore.block.TaskBlockInfo;
import org.apache.carbondata.core.metadata.blocklet.DataFileFooter;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
import org.apache.carbondata.core.scan.executor.QueryExecutor;
import org.apache.carbondata.core.scan.executor.QueryExecutorFactory;
import org.apache.carbondata.core.scan.executor.exception.QueryExecutionException;
import org.apache.carbondata.core.scan.executor.util.RestructureUtil;
import org.apache.carbondata.core.scan.expression.Expression;
import org.apache.carbondata.core.scan.model.QueryModel;
import org.apache.carbondata.core.scan.model.QueryModelBuilder;
import org.apache.carbondata.core.scan.result.RowBatch;
import org.apache.carbondata.core.scan.result.iterator.ColumnDriftRawResultIterator;
import org.apache.carbondata.core.scan.result.iterator.RawResultIterator;
import org.apache.carbondata.core.scan.wrappers.IntArrayWrapper;
import org.apache.carbondata.core.stats.QueryStatistic;
import org.apache.carbondata.core.stats.QueryStatisticsConstants;
import org.apache.carbondata.core.stats.QueryStatisticsRecorder;
import org.apache.carbondata.core.util.CarbonProperties;
import org.apache.carbondata.core.util.CarbonTimeStatisticsFactory;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.DataTypeConverter;

import org.apache.hadoop.conf.Configuration;
import org.apache.log4j.Logger;

/**
 * Executor class for executing the query on the selected segments to be merged.
 * This will fire a select * query and get the raw result.
 */
public class CarbonCompactionExecutor {

  private static final Logger LOGGER =
      LogServiceFactory.getLogService(CarbonCompactionExecutor.class.getName());
  private final Map<String, List<DataFileFooter>> dataFileMetadataSegMapping;
  private final SegmentProperties destinationSegProperties;
  private final Map<String, TaskBlockInfo> segmentMapping;
  private List<QueryExecutor> queryExecutorList;
  private List<QueryStatisticsRecorder> queryStatisticsRecorders =
      new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
  private CarbonTable carbonTable;
  private QueryModel queryModel;

  /**
   * flag to check whether any restructured block exists in the blocks selected for compaction.
   * Based on this decision will be taken whether complete data has to be sorted again
   */
  private boolean restructuredBlockExists;

  // converter for UTF8String and decimal conversion
  private DataTypeConverter dataTypeConverter;

  /**
   * Constructor
   *
   * @param segmentMapping
   * @param segmentProperties
   * @param carbonTable
   * @param dataFileMetadataSegMapping
   * @param restructuredBlockExists
   */
  public CarbonCompactionExecutor(Map<String, TaskBlockInfo> segmentMapping,
      SegmentProperties segmentProperties, CarbonTable carbonTable,
      Map<String, List<DataFileFooter>> dataFileMetadataSegMapping,
      boolean restructuredBlockExists, DataTypeConverter dataTypeConverter) {
    this.segmentMapping = segmentMapping;
    this.destinationSegProperties = segmentProperties;
    this.carbonTable = carbonTable;
    this.dataFileMetadataSegMapping = dataFileMetadataSegMapping;
    this.restructuredBlockExists = restructuredBlockExists;
    this.queryExecutorList = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
    this.dataTypeConverter = dataTypeConverter;
  }

  /**
   * For processing of the table blocks.
   *
   * @return Map of String with Carbon iterators
   * Map has 2 elements: UNSORTED and SORTED
   * Map(UNSORTED) = List of Iterators which yield sorted data
   * Map(Sorted) = List of Iterators which yield sorted data
   * In Range Column compaction we will have a Filter Expression to process
   */
  public Map<String, List<RawResultIterator>> processTableBlocks(Configuration configuration,
      Expression filterExpr) throws QueryExecutionException, IOException {

    Map<String, List<RawResultIterator>> resultList = new HashMap<>(2);
    resultList.put(CarbonCompactionUtil.UNSORTED_IDX,
        new ArrayList<RawResultIterator>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE));
    resultList.put(CarbonCompactionUtil.SORTED_IDX,
        new ArrayList<RawResultIterator>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE));

    List<TableBlockInfo> tableBlockInfos = null;
    QueryModelBuilder builder = null;
    if (null == filterExpr) {
      builder =
          new QueryModelBuilder(carbonTable).projectAllColumns().dataConverter(dataTypeConverter)
              .enableForcedDetailRawQuery();
    } else {
      builder = new QueryModelBuilder(carbonTable).projectAllColumns().filterExpression(filterExpr)
          .dataConverter(dataTypeConverter).enableForcedDetailRawQuery()
          .convertToRangeFilter(false);
    }
    if (enablePageLevelReaderForCompaction()) {
      builder.enableReadPageByPage();
    }
    queryModel = builder.build();
    // iterate each seg ID
    for (Map.Entry<String, TaskBlockInfo> taskMap : segmentMapping.entrySet()) {
      String segmentId = taskMap.getKey();
      List<DataFileFooter> listMetadata = dataFileMetadataSegMapping.get(segmentId);
      // for each segment get taskblock info
      TaskBlockInfo taskBlockInfo = taskMap.getValue();
      Set<String> taskBlockListMapping = taskBlockInfo.getTaskSet();
      // Check if block needs sorting or not
      boolean sortingRequired =
          !CarbonCompactionUtil.isSortedByCurrentSortColumns(carbonTable, listMetadata.get(0));
      for (String task : taskBlockListMapping) {
        tableBlockInfos = taskBlockInfo.getTableBlockInfoList(task);
        // during update there may be a chance that the cardinality may change within the segment
        // which may lead to failure while converting the row, so get all the blocks present in a
        // task and then split into multiple lists of same column values and create separate
        // RawResultIterator for each tableBlockInfo of same column values. If all the blocks have
        // same column values, then make a single RawResultIterator for all the blocks
        List<List<TableBlockInfo>> listOfTableBlocksBasedOnKeyLength =
            getListOfTableBlocksBasedOnColumnValueSize(tableBlockInfos);
        for (List<TableBlockInfo> tableBlockInfoList : listOfTableBlocksBasedOnKeyLength) {
          Collections.sort(tableBlockInfoList);
          LOGGER.info("for task -" + task + "- in segment id -" + segmentId + "- block size is -"
              + tableBlockInfos.size());
          queryModel.setTableBlockInfos(tableBlockInfoList);
          if (sortingRequired) {
            resultList.get(CarbonCompactionUtil.UNSORTED_IDX).add(
                getRawResultIterator(configuration, segmentId, task, tableBlockInfoList));
          } else {
            resultList.get(CarbonCompactionUtil.SORTED_IDX).add(
                getRawResultIterator(configuration, segmentId, task, tableBlockInfoList));
          }
        }
      }
    }
    return resultList;
  }

  private RawResultIterator getRawResultIterator(Configuration configuration, String segmentId,
      String task, List<TableBlockInfo> tableBlockInfoList)
      throws QueryExecutionException, IOException {
    SegmentProperties sourceSegmentProperties = getSourceSegmentProperties(
        Collections.singletonList(tableBlockInfoList.get(0).getDataFileFooter()));
    boolean hasColumnDrift = carbonTable.hasColumnDrift() &&
        RestructureUtil.hasColumnDriftOnSegment(carbonTable, sourceSegmentProperties);
    if (hasColumnDrift) {
      return new ColumnDriftRawResultIterator(
          executeBlockList(tableBlockInfoList, segmentId, task, configuration),
          sourceSegmentProperties, destinationSegProperties);
    } else {
      return new RawResultIterator(
          executeBlockList(tableBlockInfoList, segmentId, task, configuration),
          sourceSegmentProperties, destinationSegProperties, true);
    }
  }

  /**
   * This method returns the List of TableBlockInfoList, where each listOfTableBlockInfos will have
   * same columnvalues
   * @param tableBlockInfos List of tableBlockInfos present in each task
   */
  private List<List<TableBlockInfo>> getListOfTableBlocksBasedOnColumnValueSize(
      List<TableBlockInfo> tableBlockInfos) {
    List<List<TableBlockInfo>> listOfTableBlockInfoListOnColumnvaluesSize = new ArrayList<>();
    Map<IntArrayWrapper, List<TableBlockInfo>> columnvalueSizeToTableBlockInfoMap = new HashMap<>();
    for (TableBlockInfo tableBlock : tableBlockInfos) {
      // get the columnValueSize for the dataFileFooter
      IntArrayWrapper columnValueSize = new IntArrayWrapper(
          getSourceSegmentProperties(Collections.singletonList(tableBlock.getDataFileFooter()))
              .getColumnsValueSize());
      List<TableBlockInfo> tempBlockInfoList =
          columnvalueSizeToTableBlockInfoMap.get(columnValueSize);
      if (tempBlockInfoList == null) {
        tempBlockInfoList = new ArrayList<>();
        columnvalueSizeToTableBlockInfoMap.put(columnValueSize, tempBlockInfoList);
      }
      tempBlockInfoList.add(tableBlock);
    }
    for (Map.Entry<IntArrayWrapper, List<TableBlockInfo>> taskMap :
        columnvalueSizeToTableBlockInfoMap.entrySet()) {
      listOfTableBlockInfoListOnColumnvaluesSize.add(taskMap.getValue());
    }
    return listOfTableBlockInfoListOnColumnvaluesSize;
  }

  /**
   * This method will create the source segment properties based on restructured block existence
   *
   * @param listMetadata
   * @return
   */
  private SegmentProperties getSourceSegmentProperties(List<DataFileFooter> listMetadata) {
    SegmentProperties sourceSegProperties = null;
    if (restructuredBlockExists) {
      // update cardinality of source segment according to new schema
      Map<String, Integer> columnToCardinalityMap =
          new HashMap<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
      CarbonCompactionUtil
          .addColumnCardinalityToMap(columnToCardinalityMap, listMetadata.get(0).getColumnInTable(),
              listMetadata.get(0).getSegmentInfo().getColumnCardinality());
      List<ColumnSchema> updatedColumnSchemaList =
          new ArrayList<>(listMetadata.get(0).getColumnInTable().size());
      int[] updatedColumnCardinalities = CarbonCompactionUtil
          .updateColumnSchemaAndGetCardinality(columnToCardinalityMap, carbonTable,
              updatedColumnSchemaList);
      sourceSegProperties =
          new SegmentProperties(updatedColumnSchemaList, updatedColumnCardinalities);
    } else {
      sourceSegProperties = new SegmentProperties(listMetadata.get(0).getColumnInTable(),
          listMetadata.get(0).getSegmentInfo().getColumnCardinality());
    }
    return sourceSegProperties;
  }

  /**
   * get executor and execute the query model.
   *
   * @param blockList
   * @return
   */
  private CarbonIterator<RowBatch> executeBlockList(List<TableBlockInfo> blockList,
      String segmentId, String taskId, Configuration configuration)
      throws QueryExecutionException, IOException {
    queryModel.setTableBlockInfos(blockList);
    QueryStatisticsRecorder executorRecorder = CarbonTimeStatisticsFactory
        .createExecutorRecorder(queryModel.getQueryId() + "_" + segmentId + "_" + taskId);
    queryStatisticsRecorders.add(executorRecorder);
    queryModel.setStatisticsRecorder(executorRecorder);
    QueryExecutor queryExecutor = QueryExecutorFactory.getQueryExecutor(queryModel, configuration);
    queryExecutorList.add(queryExecutor);
    return queryExecutor.execute(queryModel);
  }

  /**
   * Below method will be used
   * for cleanup
   */
  public void close(List<RawResultIterator> rawResultIteratorList, long queryStartTime) {
    try {
      // close all the iterators. Iterators might not closed in case of compaction failure
      // or if process is killed
      if (null != rawResultIteratorList) {
        for (RawResultIterator rawResultIterator : rawResultIteratorList) {
          rawResultIterator.close();
        }
      }
      for (QueryExecutor queryExecutor : queryExecutorList) {
        queryExecutor.finish();
      }
      logStatistics(queryStartTime);
    } catch (QueryExecutionException e) {
      LOGGER.error("Problem while close. Ignoring the exception", e);
    }
    clearDictionaryFromQueryModel();
  }

  private void logStatistics(long queryStartTime) {
    if (!queryStatisticsRecorders.isEmpty()) {
      QueryStatistic queryStatistic = new QueryStatistic();
      queryStatistic.addFixedTimeStatistic(QueryStatisticsConstants.EXECUTOR_PART,
          System.currentTimeMillis() - queryStartTime);
      for (QueryStatisticsRecorder recorder : queryStatisticsRecorders) {
        recorder.recordStatistics(queryStatistic);
        // print executor query statistics for each task_id
        recorder.logStatistics();
      }
    }
  }

  /**
   * This method will clear the dictionary access count after its usage is complete so
   * that column can be deleted form LRU cache whenever memory reaches threshold
   */
  private void clearDictionaryFromQueryModel() {
    if (null != queryModel) {
      Map<String, Dictionary> columnToDictionaryMapping = queryModel.getColumnToDictionaryMapping();
      if (null != columnToDictionaryMapping) {
        for (Map.Entry<String, Dictionary> entry : columnToDictionaryMapping.entrySet()) {
          CarbonUtil.clearDictionaryCache(entry.getValue());
        }
      }
    }
  }

  /**
   * Whether to enable page level reader for compaction or not.
   */
  private boolean enablePageLevelReaderForCompaction() {
    String enablePageReaderProperty = CarbonProperties.getInstance()
        .getProperty(CarbonCommonConstants.CARBON_ENABLE_PAGE_LEVEL_READER_IN_COMPACTION,
            CarbonCommonConstants.CARBON_ENABLE_PAGE_LEVEL_READER_IN_COMPACTION_DEFAULT);
    boolean enablePageReader;
    try {
      enablePageReader = Boolean.parseBoolean(enablePageReaderProperty);
    } catch (Exception e) {
      enablePageReader = Boolean.parseBoolean(
          CarbonCommonConstants.CARBON_ENABLE_PAGE_LEVEL_READER_IN_COMPACTION_DEFAULT);
    }
    LOGGER.info("Page level reader is set to: " + enablePageReader);
    return enablePageReader;
  }

}

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.processing.merger;
18
19	import java.io.IOException;
20	import java.util.ArrayList;
21	import java.util.Collections;
22	import java.util.HashMap;
23	import java.util.List;
24	import java.util.Map;
25	import java.util.Set;
26
27	import org.apache.carbondata.common.CarbonIterator;
28	import org.apache.carbondata.common.logging.LogServiceFactory;
29	import org.apache.carbondata.core.cache.dictionary.Dictionary;
30	import org.apache.carbondata.core.constants.CarbonCommonConstants;
31	import org.apache.carbondata.core.datastore.block.SegmentProperties;
32	import org.apache.carbondata.core.datastore.block.TableBlockInfo;
33	import org.apache.carbondata.core.datastore.block.TaskBlockInfo;
34	import org.apache.carbondata.core.metadata.blocklet.DataFileFooter;
35	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
36	import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
37	import org.apache.carbondata.core.scan.executor.QueryExecutor;
38	import org.apache.carbondata.core.scan.executor.QueryExecutorFactory;
39	import org.apache.carbondata.core.scan.executor.exception.QueryExecutionException;
40	import org.apache.carbondata.core.scan.executor.util.RestructureUtil;
41	import org.apache.carbondata.core.scan.expression.Expression;
42	import org.apache.carbondata.core.scan.model.QueryModel;
43	import org.apache.carbondata.core.scan.model.QueryModelBuilder;
44	import org.apache.carbondata.core.scan.result.RowBatch;
45	import org.apache.carbondata.core.scan.result.iterator.ColumnDriftRawResultIterator;
46	import org.apache.carbondata.core.scan.result.iterator.RawResultIterator;
47	import org.apache.carbondata.core.scan.wrappers.IntArrayWrapper;
48	import org.apache.carbondata.core.stats.QueryStatistic;
49	import org.apache.carbondata.core.stats.QueryStatisticsConstants;
50	import org.apache.carbondata.core.stats.QueryStatisticsRecorder;
51	import org.apache.carbondata.core.util.CarbonProperties;
52	import org.apache.carbondata.core.util.CarbonTimeStatisticsFactory;
53	import org.apache.carbondata.core.util.CarbonUtil;
54	import org.apache.carbondata.core.util.DataTypeConverter;
55
56	import org.apache.hadoop.conf.Configuration;
57	import org.apache.log4j.Logger;
58
59	/**
60	* Executor class for executing the query on the selected segments to be merged.
61	* This will fire a select * query and get the raw result.
62	*/
63	public class CarbonCompactionExecutor {
64
65	private static final Logger LOGGER =	1✔
66	LogServiceFactory.getLogService(CarbonCompactionExecutor.class.getName());	1✔
67	private final Map<String, List<DataFileFooter>> dataFileMetadataSegMapping;
68	private final SegmentProperties destinationSegProperties;
69	private final Map<String, TaskBlockInfo> segmentMapping;
70	private List<QueryExecutor> queryExecutorList;
71	private List<QueryStatisticsRecorder> queryStatisticsRecorders =	1✔
72	new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
73	private CarbonTable carbonTable;
74	private QueryModel queryModel;
75
76	/**
77	* flag to check whether any restructured block exists in the blocks selected for compaction.
78	* Based on this decision will be taken whether complete data has to be sorted again
79	*/
80	private boolean restructuredBlockExists;
81
82	// converter for UTF8String and decimal conversion
83	private DataTypeConverter dataTypeConverter;
84
85	/**
86	* Constructor
87	*
88	* @param segmentMapping
89	* @param segmentProperties
90	* @param carbonTable
91	* @param dataFileMetadataSegMapping
92	* @param restructuredBlockExists
93	*/
94	public CarbonCompactionExecutor(Map<String, TaskBlockInfo> segmentMapping,
95	SegmentProperties segmentProperties, CarbonTable carbonTable,
96	Map<String, List<DataFileFooter>> dataFileMetadataSegMapping,
97	boolean restructuredBlockExists, DataTypeConverter dataTypeConverter) {	1✔
98	this.segmentMapping = segmentMapping;	1✔
99	this.destinationSegProperties = segmentProperties;	1✔
100	this.carbonTable = carbonTable;	1✔
101	this.dataFileMetadataSegMapping = dataFileMetadataSegMapping;	1✔
102	this.restructuredBlockExists = restructuredBlockExists;	1✔
103	this.queryExecutorList = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);	1✔
104	this.dataTypeConverter = dataTypeConverter;	1✔
105	}	1✔
106
107	/**
108	* For processing of the table blocks.
109	*
110	* @return Map of String with Carbon iterators
111	* Map has 2 elements: UNSORTED and SORTED
112	* Map(UNSORTED) = List of Iterators which yield sorted data
113	* Map(Sorted) = List of Iterators which yield sorted data
114	* In Range Column compaction we will have a Filter Expression to process
115	*/
116	public Map<String, List<RawResultIterator>> processTableBlocks(Configuration configuration,
117	Expression filterExpr) throws QueryExecutionException, IOException {
118
119	Map<String, List<RawResultIterator>> resultList = new HashMap<>(2);	1✔
120	resultList.put(CarbonCompactionUtil.UNSORTED_IDX,	1✔
121	new ArrayList<RawResultIterator>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE));
122	resultList.put(CarbonCompactionUtil.SORTED_IDX,	1✔
123	new ArrayList<RawResultIterator>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE));
124
125	List<TableBlockInfo> tableBlockInfos = null;	1✔
126	QueryModelBuilder builder = null;	1✔
127	if (null == filterExpr) {	1✔
128	builder =	1✔
129	new QueryModelBuilder(carbonTable).projectAllColumns().dataConverter(dataTypeConverter)	1✔
130	.enableForcedDetailRawQuery();	1✔
131	} else {
132	builder = new QueryModelBuilder(carbonTable).projectAllColumns().filterExpression(filterExpr)	1✔
133	.dataConverter(dataTypeConverter).enableForcedDetailRawQuery()	1✔
134	.convertToRangeFilter(false);	1✔
135	}
136	if (enablePageLevelReaderForCompaction()) {	1✔
137	builder.enableReadPageByPage();	1✔
138	}
139	queryModel = builder.build();	1✔
140	// iterate each seg ID
141	for (Map.Entry<String, TaskBlockInfo> taskMap : segmentMapping.entrySet()) {	1✔
142	String segmentId = taskMap.getKey();	1✔
143	List<DataFileFooter> listMetadata = dataFileMetadataSegMapping.get(segmentId);	1✔
144	// for each segment get taskblock info
145	TaskBlockInfo taskBlockInfo = taskMap.getValue();	1✔
146	Set<String> taskBlockListMapping = taskBlockInfo.getTaskSet();	1✔
147	// Check if block needs sorting or not
148	boolean sortingRequired =	1✔
149	!CarbonCompactionUtil.isSortedByCurrentSortColumns(carbonTable, listMetadata.get(0));	1✔
150	for (String task : taskBlockListMapping) {	1✔
151	tableBlockInfos = taskBlockInfo.getTableBlockInfoList(task);	1✔
152	// during update there may be a chance that the cardinality may change within the segment
153	// which may lead to failure while converting the row, so get all the blocks present in a
154	// task and then split into multiple lists of same column values and create separate
155	// RawResultIterator for each tableBlockInfo of same column values. If all the blocks have
156	// same column values, then make a single RawResultIterator for all the blocks
157	List<List<TableBlockInfo>> listOfTableBlocksBasedOnKeyLength =	1✔
158	getListOfTableBlocksBasedOnColumnValueSize(tableBlockInfos);	1✔
159	for (List<TableBlockInfo> tableBlockInfoList : listOfTableBlocksBasedOnKeyLength) {	1✔
160	Collections.sort(tableBlockInfoList);	1✔
161	LOGGER.info("for task -" + task + "- in segment id -" + segmentId + "- block size is -"	1✔
162	+ tableBlockInfos.size());	1✔
163	queryModel.setTableBlockInfos(tableBlockInfoList);	1✔
164	if (sortingRequired) {	1✔
165	resultList.get(CarbonCompactionUtil.UNSORTED_IDX).add(	1✔
166	getRawResultIterator(configuration, segmentId, task, tableBlockInfoList));	1✔
167	} else {
168	resultList.get(CarbonCompactionUtil.SORTED_IDX).add(	1✔
169	getRawResultIterator(configuration, segmentId, task, tableBlockInfoList));	1✔
170	}
171	}	1✔
172	}	1✔
173	}	1✔
174	return resultList;	1✔
175	}
176
177	private RawResultIterator getRawResultIterator(Configuration configuration, String segmentId,
178	String task, List<TableBlockInfo> tableBlockInfoList)
179	throws QueryExecutionException, IOException {
180	SegmentProperties sourceSegmentProperties = getSourceSegmentProperties(	1✔
181	Collections.singletonList(tableBlockInfoList.get(0).getDataFileFooter()));	1✔
182	boolean hasColumnDrift = carbonTable.hasColumnDrift() &&	1✔
183	RestructureUtil.hasColumnDriftOnSegment(carbonTable, sourceSegmentProperties);	1✔
184	if (hasColumnDrift) {	1✔
185	return new ColumnDriftRawResultIterator(	1✔
186	executeBlockList(tableBlockInfoList, segmentId, task, configuration),	1✔
187	sourceSegmentProperties, destinationSegProperties);
188	} else {
189	return new RawResultIterator(	1✔
190	executeBlockList(tableBlockInfoList, segmentId, task, configuration),	1✔
191	sourceSegmentProperties, destinationSegProperties, true);
192	}
193	}
194
195	/**
196	* This method returns the List of TableBlockInfoList, where each listOfTableBlockInfos will have
197	* same columnvalues
198	* @param tableBlockInfos List of tableBlockInfos present in each task
199	*/
200	private List<List<TableBlockInfo>> getListOfTableBlocksBasedOnColumnValueSize(
201	List<TableBlockInfo> tableBlockInfos) {
202	List<List<TableBlockInfo>> listOfTableBlockInfoListOnColumnvaluesSize = new ArrayList<>();	1✔
203	Map<IntArrayWrapper, List<TableBlockInfo>> columnvalueSizeToTableBlockInfoMap = new HashMap<>();	1✔
204	for (TableBlockInfo tableBlock : tableBlockInfos) {	1✔
205	// get the columnValueSize for the dataFileFooter
206	IntArrayWrapper columnValueSize = new IntArrayWrapper(	1✔
207	getSourceSegmentProperties(Collections.singletonList(tableBlock.getDataFileFooter()))	1✔
208	.getColumnsValueSize());	1✔
209	List<TableBlockInfo> tempBlockInfoList =	1✔
210	columnvalueSizeToTableBlockInfoMap.get(columnValueSize);	1✔
211	if (tempBlockInfoList == null) {	1✔
212	tempBlockInfoList = new ArrayList<>();	1✔
213	columnvalueSizeToTableBlockInfoMap.put(columnValueSize, tempBlockInfoList);	1✔
214	}
215	tempBlockInfoList.add(tableBlock);	1✔
216	}	1✔
217	for (Map.Entry<IntArrayWrapper, List<TableBlockInfo>> taskMap :
218	columnvalueSizeToTableBlockInfoMap.entrySet()) {	1✔
219	listOfTableBlockInfoListOnColumnvaluesSize.add(taskMap.getValue());	1✔
220	}	1✔
221	return listOfTableBlockInfoListOnColumnvaluesSize;	1✔
222	}
223
224	/**
225	* This method will create the source segment properties based on restructured block existence
226	*
227	* @param listMetadata
228	* @return
229	*/
230	private SegmentProperties getSourceSegmentProperties(List<DataFileFooter> listMetadata) {
231	SegmentProperties sourceSegProperties = null;	1✔
232	if (restructuredBlockExists) {	1✔
233	// update cardinality of source segment according to new schema
234	Map<String, Integer> columnToCardinalityMap =	1✔
235	new HashMap<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
236	CarbonCompactionUtil	1✔
237	.addColumnCardinalityToMap(columnToCardinalityMap, listMetadata.get(0).getColumnInTable(),	1✔
238	listMetadata.get(0).getSegmentInfo().getColumnCardinality());	1✔
239	List<ColumnSchema> updatedColumnSchemaList =	1✔
240	new ArrayList<>(listMetadata.get(0).getColumnInTable().size());	1✔
241	int[] updatedColumnCardinalities = CarbonCompactionUtil	1✔
242	.updateColumnSchemaAndGetCardinality(columnToCardinalityMap, carbonTable,	1✔
243	updatedColumnSchemaList);
244	sourceSegProperties =	1✔
245	new SegmentProperties(updatedColumnSchemaList, updatedColumnCardinalities);
246	} else {	1✔
247	sourceSegProperties = new SegmentProperties(listMetadata.get(0).getColumnInTable(),	1✔
248	listMetadata.get(0).getSegmentInfo().getColumnCardinality());	1✔
249	}
250	return sourceSegProperties;	1✔
251	}
252
253	/**
254	* get executor and execute the query model.
255	*
256	* @param blockList
257	* @return
258	*/
259	private CarbonIterator<RowBatch> executeBlockList(List<TableBlockInfo> blockList,
260	String segmentId, String taskId, Configuration configuration)
261	throws QueryExecutionException, IOException {
262	queryModel.setTableBlockInfos(blockList);	1✔
263	QueryStatisticsRecorder executorRecorder = CarbonTimeStatisticsFactory	1✔
264	.createExecutorRecorder(queryModel.getQueryId() + "_" + segmentId + "_" + taskId);	1✔
265	queryStatisticsRecorders.add(executorRecorder);	1✔
266	queryModel.setStatisticsRecorder(executorRecorder);	1✔
267	QueryExecutor queryExecutor = QueryExecutorFactory.getQueryExecutor(queryModel, configuration);	1✔
268	queryExecutorList.add(queryExecutor);	1✔
269	return queryExecutor.execute(queryModel);	1✔
270	}
271
272	/**
273	* Below method will be used
274	* for cleanup
275	*/
276	public void close(List<RawResultIterator> rawResultIteratorList, long queryStartTime) {
277	try {
278	// close all the iterators. Iterators might not closed in case of compaction failure
279	// or if process is killed
280	if (null != rawResultIteratorList) {	1✔
281	for (RawResultIterator rawResultIterator : rawResultIteratorList) {	1✔
282	rawResultIterator.close();	1✔
283	}	1✔
284	}
285	for (QueryExecutor queryExecutor : queryExecutorList) {	1✔
286	queryExecutor.finish();	1✔
287	}	1✔
288	logStatistics(queryStartTime);	1✔
289	} catch (QueryExecutionException e) {	×
290	LOGGER.error("Problem while close. Ignoring the exception", e);	×
291	}	1✔
292	clearDictionaryFromQueryModel();	1✔
293	}	1✔
294
295	private void logStatistics(long queryStartTime) {
296	if (!queryStatisticsRecorders.isEmpty()) {	1✔
297	QueryStatistic queryStatistic = new QueryStatistic();	1✔
298	queryStatistic.addFixedTimeStatistic(QueryStatisticsConstants.EXECUTOR_PART,	1✔
299	System.currentTimeMillis() - queryStartTime);	1✔
300	for (QueryStatisticsRecorder recorder : queryStatisticsRecorders) {	1✔
301	recorder.recordStatistics(queryStatistic);	1✔
302	// print executor query statistics for each task_id
303	recorder.logStatistics();	1✔
304	}	1✔
305	}
306	}	1✔
307
308	/**
309	* This method will clear the dictionary access count after its usage is complete so
310	* that column can be deleted form LRU cache whenever memory reaches threshold
311	*/
312	private void clearDictionaryFromQueryModel() {
313	if (null != queryModel) {	1✔
314	Map<String, Dictionary> columnToDictionaryMapping = queryModel.getColumnToDictionaryMapping();	1✔
315	if (null != columnToDictionaryMapping) {	1✔
316	for (Map.Entry<String, Dictionary> entry : columnToDictionaryMapping.entrySet()) {	1✔
317	CarbonUtil.clearDictionaryCache(entry.getValue());	1✔
318	}	1✔
319	}
320	}
321	}	1✔
322
323	/**
324	* Whether to enable page level reader for compaction or not.
325	*/
326	private boolean enablePageLevelReaderForCompaction() {
327	String enablePageReaderProperty = CarbonProperties.getInstance()	1✔
328	.getProperty(CarbonCommonConstants.CARBON_ENABLE_PAGE_LEVEL_READER_IN_COMPACTION,	1✔
329	CarbonCommonConstants.CARBON_ENABLE_PAGE_LEVEL_READER_IN_COMPACTION_DEFAULT);
330	boolean enablePageReader;
331	try {
332	enablePageReader = Boolean.parseBoolean(enablePageReaderProperty);	1✔
333	} catch (Exception e) {	×
334	enablePageReader = Boolean.parseBoolean(	×
335	CarbonCommonConstants.CARBON_ENABLE_PAGE_LEVEL_READER_IN_COMPACTION_DEFAULT);
336	}	1✔
337	LOGGER.info("Page level reader is set to: " + enablePageReader);	1✔
338	return enablePageReader;	1✔
339	}
340
341	}

apache / carbondata / 3333

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