10000

Build # 10000

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/70e0b85629e4ac6613ddcef9c40d9e93310b958c">70e0b8562<a href="https://github.com/apache/carbondata/commit/70e0b85629e4ac6613ddcef9c40d9e93310b958c">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/9ae4520538031147588be171024a7453c1a94a5d">9ae452053</a>">9ae452053</a><a href="https://github.com/apache/carbondata/commit/70e0b85629e4ac6613ddcef9c40d9e93310b958c"> into 88fa067e0">88fa067e0</a>

Run Details

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

61715 of 77606 relevant lines covered (79.52%)

1.05 hits per line

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

56.69

/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonCompactionUtil.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.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.block.TableBlockInfo;
import org.apache.carbondata.core.datastore.block.TaskBlockInfo;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
import org.apache.carbondata.core.metadata.blocklet.DataFileFooter;
import org.apache.carbondata.core.metadata.encoder.Encoding;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonMeasure;
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.path.CarbonTablePath;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.log4j.Logger;

/**
 * Utility Class for the Compaction Flow.
 */
public class CarbonCompactionUtil {

  private static final Logger LOGGER =
      LogServiceFactory.getLogService(CarbonCompactionUtil.class.getName());

  /**
   * To create a mapping of Segment Id and TableBlockInfo.
   *
   * @param tableBlockInfoList
   * @return
   */
  public static Map<String, TaskBlockInfo> createMappingForSegments(
      List<TableBlockInfo> tableBlockInfoList) {

    // stores taskBlockInfo of each segment
    Map<String, TaskBlockInfo> segmentBlockInfoMapping =
        new HashMap<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);


    for (TableBlockInfo info : tableBlockInfoList) {
      String segId = info.getSegmentId();
      // check if segId is already present in map
      TaskBlockInfo taskBlockInfoMapping = segmentBlockInfoMapping.get(segId);
      // extract task ID from file Path.
      String taskNo = CarbonTablePath.DataFileUtil.getTaskNo(info.getFilePath());
      // if taskBlockInfo is not there, then create and add
      if (null == taskBlockInfoMapping) {
        taskBlockInfoMapping = new TaskBlockInfo();
        groupCorrespodingInfoBasedOnTask(info, taskBlockInfoMapping, taskNo);
        // put the taskBlockInfo with respective segment id
        segmentBlockInfoMapping.put(segId, taskBlockInfoMapping);
      } else
      {
        groupCorrespodingInfoBasedOnTask(info, taskBlockInfoMapping, taskNo);
      }
    }
    return segmentBlockInfoMapping;

  }

  /**
   * Grouping the taskNumber and list of TableBlockInfo.
   * @param info
   * @param taskBlockMapping
   * @param taskNo
   */
  private static void groupCorrespodingInfoBasedOnTask(TableBlockInfo info,
      TaskBlockInfo taskBlockMapping, String taskNo) {
    // get the corresponding list from task mapping.
    List<TableBlockInfo> blockLists = taskBlockMapping.getTableBlockInfoList(taskNo);
    if (null != blockLists) {
      blockLists.add(info);
    } else {
      blockLists = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
      blockLists.add(info);
      taskBlockMapping.addTableBlockInfoList(taskNo, blockLists);
    }
  }

  /**
   * To create a mapping of Segment Id and DataFileFooter.
   *
   * @param tableBlockInfoList
   * @return
   */
  public static Map<String, List<DataFileFooter>> createDataFileFooterMappingForSegments(
      List<TableBlockInfo> tableBlockInfoList) throws IOException {

    Map<String, List<DataFileFooter>> segmentBlockInfoMapping = new HashMap<>();
    for (TableBlockInfo blockInfo : tableBlockInfoList) {
      List<DataFileFooter> eachSegmentBlocks = new ArrayList<>();
      String segId = blockInfo.getSegmentId();
      DataFileFooter dataFileMatadata = null;
      // check if segId is already present in map
      List<DataFileFooter> metadataList = segmentBlockInfoMapping.get(segId);
      // check to decide whether to read file footer of carbondata file forcefully. This will help
      // in getting the schema last updated time based on which compaction flow is decided that
      // whether it will go to restructure compaction flow or normal compaction flow.
      // This decision will impact the compaction performance so it needs to be decided carefully
      if (null != blockInfo.getDetailInfo()
          && blockInfo.getDetailInfo().getSchemaUpdatedTimeStamp() == 0L) {
        dataFileMatadata = CarbonUtil.readMetadataFile(blockInfo, true);
      } else {
        dataFileMatadata = CarbonUtil.readMetadataFile(blockInfo);
      }
      if (null == metadataList) {
        // if it is not present
        eachSegmentBlocks.add(dataFileMatadata);
        segmentBlockInfoMapping.put(segId, eachSegmentBlocks);
      } else {
        // if its already present then update the list.
        metadataList.add(dataFileMatadata);
      }
    }
    return segmentBlockInfoMapping;

  }

  /**
   * Check whether the file to indicate the compaction is present or not.
   * @param metaFolderPath
   * @return
   */
  public static boolean isCompactionRequiredForTable(String metaFolderPath) {
    String minorCompactionStatusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR
        + CarbonCommonConstants.minorCompactionRequiredFile;

    String majorCompactionStatusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR
        + CarbonCommonConstants.majorCompactionRequiredFile;
    try {
      if (FileFactory.isFileExist(minorCompactionStatusFile,
          FileFactory.getFileType(minorCompactionStatusFile)) || FileFactory
          .isFileExist(majorCompactionStatusFile,
              FileFactory.getFileType(majorCompactionStatusFile))) {
        return true;
      }
    } catch (IOException e) {
      LOGGER.error("Exception in isFileExist compaction request file " + e.getMessage());
    }
    return false;
  }

  /**
   * Determine the type of the compaction received.
   * @param metaFolderPath
   * @return
   */
  public static CompactionType determineCompactionType(String metaFolderPath) {
    String minorCompactionStatusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR
        + CarbonCommonConstants.minorCompactionRequiredFile;

    String majorCompactionStatusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR
        + CarbonCommonConstants.majorCompactionRequiredFile;
    try {
      if (FileFactory.isFileExist(minorCompactionStatusFile,
          FileFactory.getFileType(minorCompactionStatusFile))) {
        return CompactionType.MINOR;
      }
      if (FileFactory.isFileExist(majorCompactionStatusFile,
          FileFactory.getFileType(majorCompactionStatusFile))) {
        return CompactionType.MAJOR;
      }

    } catch (IOException e) {
      LOGGER.error("Exception in determining the compaction request file " + e.getMessage());
    }
    return CompactionType.MINOR;
  }

  /**
   * Delete the compation request file once the compaction is done.
   * @param metaFolderPath
   * @param compactionType
   * @return
   */
  public static boolean deleteCompactionRequiredFile(String metaFolderPath,
      CompactionType compactionType) {
    String compactionRequiredFile;
    if (compactionType.equals(CompactionType.MINOR)) {
      compactionRequiredFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR
          + CarbonCommonConstants.minorCompactionRequiredFile;
    } else {
      compactionRequiredFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR
          + CarbonCommonConstants.majorCompactionRequiredFile;
    }
    try {
      if (FileFactory
          .isFileExist(compactionRequiredFile, FileFactory.getFileType(compactionRequiredFile))) {
        if (FileFactory
            .getCarbonFile(compactionRequiredFile, FileFactory.getFileType(compactionRequiredFile))
            .delete()) {
          LOGGER.info("Deleted the compaction request file " + compactionRequiredFile);
          return true;
        } else {
          LOGGER.error("Unable to delete the compaction request file " + compactionRequiredFile);
        }
      } else {
        LOGGER.info("Compaction request file is not present. file is : " + compactionRequiredFile);
      }
    } catch (IOException e) {
      LOGGER.error("Exception in deleting the compaction request file " + e.getMessage());
    }
    return false;
  }

  /**
   * Creation of the compaction request if someother compaction is in progress.
   * @param metaFolderPath
   * @param compactionType
   * @return
   */
  public static boolean createCompactionRequiredFile(String metaFolderPath,
      CompactionType compactionType) {
    String statusFile;
    if (CompactionType.MINOR == compactionType) {
      statusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR
          + CarbonCommonConstants.minorCompactionRequiredFile;
    } else {
      statusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR
          + CarbonCommonConstants.majorCompactionRequiredFile;
    }
    try {
      if (!FileFactory.isFileExist(statusFile, FileFactory.getFileType(statusFile))) {
        if (FileFactory.createNewFile(statusFile, FileFactory.getFileType(statusFile))) {
          LOGGER.info("successfully created a compaction required file - " + statusFile);
          return true;
        } else {
          LOGGER.error("Not able to create a compaction required file - " + statusFile);
          return false;
        }
      } else {
        LOGGER.info("Compaction request file : " + statusFile + " already exist.");
      }
    } catch (IOException e) {
      LOGGER.error("Exception in creating the compaction request file " + e.getMessage());
    }
    return false;
  }

  /**
   * This will check if any compaction request has been received for any table.
   *
   * @param carbonTables
   * @return
   */
  public static CarbonTable getNextTableToCompact(CarbonTable[] carbonTables,
      List<CarbonTableIdentifier> skipList) {
    for (CarbonTable ctable : carbonTables) {
      String metadataPath = ctable.getMetadataPath();
      // check for the compaction required file and at the same time exclude the tables which are
      // present in the skip list.
      if (CarbonCompactionUtil.isCompactionRequiredForTable(metadataPath) && !skipList
          .contains(ctable.getCarbonTableIdentifier())) {
        return ctable;
      }
    }
    return null;
  }

  /**
   * This method will add the prepare the max column cardinality map
   *
   * @param columnCardinalityMap
   * @param currentBlockSchema
   * @param currentBlockCardinality
   */
  public static void addColumnCardinalityToMap(Map<String, Integer> columnCardinalityMap,
      List<ColumnSchema> currentBlockSchema, int[] currentBlockCardinality) {
    for (int i = 0; i < currentBlockCardinality.length; i++) {
      // add value to map only if does not exist or new cardinality is > existing value
      String columnUniqueId = currentBlockSchema.get(i).getColumnUniqueId();
      Integer value = columnCardinalityMap.get(columnUniqueId);
      if (null == value) {
        columnCardinalityMap.put(columnUniqueId, currentBlockCardinality[i]);
      } else {
        if (currentBlockCardinality[i] > value) {
          columnCardinalityMap.put(columnUniqueId, currentBlockCardinality[i]);
        }
      }
    }
  }

  /**
   * This method will return the updated cardinality according to the master schema
   *
   * @param columnCardinalityMap
   * @param carbonTable
   * @param updatedColumnSchemaList
   * @return
   */
  public static int[] updateColumnSchemaAndGetCardinality(Map<String, Integer> columnCardinalityMap,
      CarbonTable carbonTable, List<ColumnSchema> updatedColumnSchemaList) {
    List<CarbonDimension> masterDimensions =
        carbonTable.getDimensionByTableName(carbonTable.getTableName());
    List<Integer> updatedCardinalityList = new ArrayList<>(columnCardinalityMap.size());
    for (CarbonDimension dimension : masterDimensions) {
      Integer value = columnCardinalityMap.get(dimension.getColumnId());
      if (null == value) {
        updatedCardinalityList.add(getDimensionDefaultCardinality(dimension));
      } else {
        updatedCardinalityList.add(value);
      }
      updatedColumnSchemaList.add(dimension.getColumnSchema());

      if (dimension.getNumberOfChild() > 0) {
        fillColumnSchemaListForComplexDims(dimension.getListOfChildDimensions(),
            updatedColumnSchemaList, updatedCardinalityList, columnCardinalityMap);
      }
    }
    // add measures to the column schema list
    List<CarbonMeasure> masterSchemaMeasures =
        carbonTable.getMeasureByTableName(carbonTable.getTableName());
    for (CarbonMeasure measure : masterSchemaMeasures) {
      updatedColumnSchemaList.add(measure.getColumnSchema());
    }
    return ArrayUtils
        .toPrimitive(updatedCardinalityList.toArray(new Integer[updatedCardinalityList.size()]));
  }

  /**
   * This method is to get the chile dimensions of the complex dimension and
   * update the cardinality for all complex dimensions
   *
   * @param carbonDimensionsList
   * @param updatedColumnSchemaList
   * @param updatedCardinalityList
   * @param columnCardinalityMap
   */
  private static void fillColumnSchemaListForComplexDims(List<CarbonDimension> carbonDimensionsList,
      List<ColumnSchema> updatedColumnSchemaList, List<Integer> updatedCardinalityList,
      Map<String, Integer> columnCardinalityMap) {
    for (CarbonDimension carbonDimension : carbonDimensionsList) {
      Integer value = columnCardinalityMap.get(carbonDimension.getColumnId());
      if (null == value) {
        updatedCardinalityList.add(getDimensionDefaultCardinality(carbonDimension));
      } else {
        updatedCardinalityList.add(value);
      }
      updatedColumnSchemaList.add(carbonDimension.getColumnSchema());
      List<CarbonDimension> childDims = carbonDimension.getListOfChildDimensions();
      if (null != childDims && childDims.size() > 0) {
        fillColumnSchemaListForComplexDims(childDims, updatedColumnSchemaList,
            updatedCardinalityList, columnCardinalityMap);
      }
    }
  }

  /**
   * This method will return the default cardinality based on dimension type
   *
   * @param dimension
   * @return
   */
  private static int getDimensionDefaultCardinality(CarbonDimension dimension) {
    int cardinality = 0;
    if (dimension.hasEncoding(Encoding.DIRECT_DICTIONARY)) {
      cardinality = Integer.MAX_VALUE;
    } else if (dimension.hasEncoding(Encoding.DICTIONARY)) {
      if (null != dimension.getDefaultValue()) {
        cardinality = CarbonCommonConstants.DICTIONARY_DEFAULT_CARDINALITY + 1;
      } else {
        cardinality = CarbonCommonConstants.DICTIONARY_DEFAULT_CARDINALITY;
      }
    } else {
      cardinality = -1;
    }
    return cardinality;
  }

  /**
   * This method will check for any restructured block in the blocks selected for compaction
   *
   * @param segmentMapping
   * @param dataFileMetadataSegMapping
   * @param tableLastUpdatedTime
   * @return
   */
  public static boolean checkIfAnyRestructuredBlockExists(Map<String, TaskBlockInfo> segmentMapping,
      Map<String, List<DataFileFooter>> dataFileMetadataSegMapping, long tableLastUpdatedTime) {
    boolean restructuredBlockExists = false;
    for (Map.Entry<String, TaskBlockInfo> taskMap : segmentMapping.entrySet()) {
      String segmentId = taskMap.getKey();
      List<DataFileFooter> listMetadata = dataFileMetadataSegMapping.get(segmentId);
      for (DataFileFooter dataFileFooter : listMetadata) {
        // if schema modified timestamp is greater than footer stored schema timestamp,
        // it indicates it is a restructured block
        if (tableLastUpdatedTime > dataFileFooter.getSchemaUpdatedTimeStamp()) {
          restructuredBlockExists = true;
          break;
        }
      }
      if (restructuredBlockExists) {
        break;
      }
    }
    return restructuredBlockExists;
  }
}

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.HashMap;
22	import java.util.List;
23	import java.util.Map;
24
25	import org.apache.carbondata.common.logging.LogServiceFactory;
26	import org.apache.carbondata.core.constants.CarbonCommonConstants;
27	import org.apache.carbondata.core.datastore.block.TableBlockInfo;
28	import org.apache.carbondata.core.datastore.block.TaskBlockInfo;
29	import org.apache.carbondata.core.datastore.impl.FileFactory;
30	import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
31	import org.apache.carbondata.core.metadata.blocklet.DataFileFooter;
32	import org.apache.carbondata.core.metadata.encoder.Encoding;
33	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
34	import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
35	import org.apache.carbondata.core.metadata.schema.table.column.CarbonMeasure;
36	import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
37	import org.apache.carbondata.core.util.CarbonUtil;
38	import org.apache.carbondata.core.util.path.CarbonTablePath;
39
40	import org.apache.commons.lang3.ArrayUtils;
41	import org.apache.log4j.Logger;
42
43	/**
44	* Utility Class for the Compaction Flow.
45	*/
46	public class CarbonCompactionUtil {	×
47
48	private static final Logger LOGGER =	1✔
49	LogServiceFactory.getLogService(CarbonCompactionUtil.class.getName());	1✔
50
51	/**
52	* To create a mapping of Segment Id and TableBlockInfo.
53	*
54	* @param tableBlockInfoList
55	* @return
56	*/
57	public static Map<String, TaskBlockInfo> createMappingForSegments(
58	List<TableBlockInfo> tableBlockInfoList) {
59
60	// stores taskBlockInfo of each segment
61	Map<String, TaskBlockInfo> segmentBlockInfoMapping =	1✔
62	new HashMap<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
63
64
65	for (TableBlockInfo info : tableBlockInfoList) {	1✔
66	String segId = info.getSegmentId();	1✔
67	// check if segId is already present in map
68	TaskBlockInfo taskBlockInfoMapping = segmentBlockInfoMapping.get(segId);	1✔
69	// extract task ID from file Path.
70	String taskNo = CarbonTablePath.DataFileUtil.getTaskNo(info.getFilePath());	1✔
71	// if taskBlockInfo is not there, then create and add
72	if (null == taskBlockInfoMapping) {	1✔
73	taskBlockInfoMapping = new TaskBlockInfo();	1✔
74	groupCorrespodingInfoBasedOnTask(info, taskBlockInfoMapping, taskNo);	1✔
75	// put the taskBlockInfo with respective segment id
76	segmentBlockInfoMapping.put(segId, taskBlockInfoMapping);	1✔
77	} else
78	{
79	groupCorrespodingInfoBasedOnTask(info, taskBlockInfoMapping, taskNo);	1✔
80	}
81	}	1✔
82	return segmentBlockInfoMapping;	1✔
83
84	}
85
86	/**
87	* Grouping the taskNumber and list of TableBlockInfo.
88	* @param info
89	* @param taskBlockMapping
90	* @param taskNo
91	*/
92	private static void groupCorrespodingInfoBasedOnTask(TableBlockInfo info,
93	TaskBlockInfo taskBlockMapping, String taskNo) {
94	// get the corresponding list from task mapping.
95	List<TableBlockInfo> blockLists = taskBlockMapping.getTableBlockInfoList(taskNo);	1✔
96	if (null != blockLists) {	1✔
97	blockLists.add(info);	×
98	} else {
99	blockLists = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);	1✔
100	blockLists.add(info);	1✔
101	taskBlockMapping.addTableBlockInfoList(taskNo, blockLists);	1✔
102	}
103	}	1✔
104
105	/**
106	* To create a mapping of Segment Id and DataFileFooter.
107	*
108	* @param tableBlockInfoList
109	* @return
110	*/
111	public static Map<String, List<DataFileFooter>> createDataFileFooterMappingForSegments(
112	List<TableBlockInfo> tableBlockInfoList) throws IOException {
113
114	Map<String, List<DataFileFooter>> segmentBlockInfoMapping = new HashMap<>();	1✔
115	for (TableBlockInfo blockInfo : tableBlockInfoList) {	1✔
116	List<DataFileFooter> eachSegmentBlocks = new ArrayList<>();	1✔
117	String segId = blockInfo.getSegmentId();	1✔
118	DataFileFooter dataFileMatadata = null;	1✔
119	// check if segId is already present in map
120	List<DataFileFooter> metadataList = segmentBlockInfoMapping.get(segId);	1✔
121	// check to decide whether to read file footer of carbondata file forcefully. This will help
122	// in getting the schema last updated time based on which compaction flow is decided that
123	// whether it will go to restructure compaction flow or normal compaction flow.
124	// This decision will impact the compaction performance so it needs to be decided carefully
125	if (null != blockInfo.getDetailInfo()	1✔
126	&& blockInfo.getDetailInfo().getSchemaUpdatedTimeStamp() == 0L) {	1✔
127	dataFileMatadata = CarbonUtil.readMetadataFile(blockInfo, true);	×
128	} else {
129	dataFileMatadata = CarbonUtil.readMetadataFile(blockInfo);	1✔
130	}
131	if (null == metadataList) {	1✔
132	// if it is not present
133	eachSegmentBlocks.add(dataFileMatadata);	1✔
134	segmentBlockInfoMapping.put(segId, eachSegmentBlocks);	1✔
135	} else {
136	// if its already present then update the list.
137	metadataList.add(dataFileMatadata);	1✔
138	}
139	}	1✔
140	return segmentBlockInfoMapping;	1✔
141
142	}
143
144	/**
145	* Check whether the file to indicate the compaction is present or not.
146	* @param metaFolderPath
147	* @return
148	*/
149	public static boolean isCompactionRequiredForTable(String metaFolderPath) {
150	String minorCompactionStatusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR	×
151	+ CarbonCommonConstants.minorCompactionRequiredFile;
152
153	String majorCompactionStatusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR	×
154	+ CarbonCommonConstants.majorCompactionRequiredFile;
155	try {
156	if (FileFactory.isFileExist(minorCompactionStatusFile,	×
157	FileFactory.getFileType(minorCompactionStatusFile)) \|\| FileFactory	×
158	.isFileExist(majorCompactionStatusFile,	×
159	FileFactory.getFileType(majorCompactionStatusFile))) {	×
160	return true;	×
161	}
162	} catch (IOException e) {	×
163	LOGGER.error("Exception in isFileExist compaction request file " + e.getMessage());	×
164	}	×
165	return false;	×
166	}
167
168	/**
169	* Determine the type of the compaction received.
170	* @param metaFolderPath
171	* @return
172	*/
173	public static CompactionType determineCompactionType(String metaFolderPath) {
174	String minorCompactionStatusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR	×
175	+ CarbonCommonConstants.minorCompactionRequiredFile;
176
177	String majorCompactionStatusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR	×
178	+ CarbonCommonConstants.majorCompactionRequiredFile;
179	try {
180	if (FileFactory.isFileExist(minorCompactionStatusFile,	×
181	FileFactory.getFileType(minorCompactionStatusFile))) {	×
182	return CompactionType.MINOR;	×
183	}
184	if (FileFactory.isFileExist(majorCompactionStatusFile,	×
185	FileFactory.getFileType(majorCompactionStatusFile))) {	×
186	return CompactionType.MAJOR;	×
187	}
188
189	} catch (IOException e) {	×
190	LOGGER.error("Exception in determining the compaction request file " + e.getMessage());	×
191	}	×
192	return CompactionType.MINOR;	×
193	}
194
195	/**
196	* Delete the compation request file once the compaction is done.
197	* @param metaFolderPath
198	* @param compactionType
199	* @return
200	*/
201	public static boolean deleteCompactionRequiredFile(String metaFolderPath,
202	CompactionType compactionType) {
203	String compactionRequiredFile;
204	if (compactionType.equals(CompactionType.MINOR)) {	×
205	compactionRequiredFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR	×
206	+ CarbonCommonConstants.minorCompactionRequiredFile;
207	} else {
208	compactionRequiredFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR	×
209	+ CarbonCommonConstants.majorCompactionRequiredFile;
210	}
211	try {
212	if (FileFactory	×
213	.isFileExist(compactionRequiredFile, FileFactory.getFileType(compactionRequiredFile))) {	×
214	if (FileFactory	×
215	.getCarbonFile(compactionRequiredFile, FileFactory.getFileType(compactionRequiredFile))	×
216	.delete()) {	×
217	LOGGER.info("Deleted the compaction request file " + compactionRequiredFile);	×
218	return true;	×
219	} else {
220	LOGGER.error("Unable to delete the compaction request file " + compactionRequiredFile);	×
221	}
222	} else {
223	LOGGER.info("Compaction request file is not present. file is : " + compactionRequiredFile);	×
224	}
225	} catch (IOException e) {	×
226	LOGGER.error("Exception in deleting the compaction request file " + e.getMessage());	×
227	}	×
228	return false;	×
229	}
230
231	/**
232	* Creation of the compaction request if someother compaction is in progress.
233	* @param metaFolderPath
234	* @param compactionType
235	* @return
236	*/
237	public static boolean createCompactionRequiredFile(String metaFolderPath,
238	CompactionType compactionType) {
239	String statusFile;
240	if (CompactionType.MINOR == compactionType) {	×
241	statusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR	×
242	+ CarbonCommonConstants.minorCompactionRequiredFile;
243	} else {
244	statusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR	×
245	+ CarbonCommonConstants.majorCompactionRequiredFile;
246	}
247	try {
248	if (!FileFactory.isFileExist(statusFile, FileFactory.getFileType(statusFile))) {	×
249	if (FileFactory.createNewFile(statusFile, FileFactory.getFileType(statusFile))) {	×
250	LOGGER.info("successfully created a compaction required file - " + statusFile);	×
251	return true;	×
252	} else {
253	LOGGER.error("Not able to create a compaction required file - " + statusFile);	×
254	return false;	×
255	}
256	} else {
257	LOGGER.info("Compaction request file : " + statusFile + " already exist.");	×
258	}
259	} catch (IOException e) {	×
260	LOGGER.error("Exception in creating the compaction request file " + e.getMessage());	×
261	}	×
262	return false;	×
263	}
264
265	/**
266	* This will check if any compaction request has been received for any table.
267	*
268	* @param carbonTables
269	* @return
270	*/
271	public static CarbonTable getNextTableToCompact(CarbonTable[] carbonTables,
272	List<CarbonTableIdentifier> skipList) {
273	for (CarbonTable ctable : carbonTables) {	×
274	String metadataPath = ctable.getMetadataPath();	×
275	// check for the compaction required file and at the same time exclude the tables which are
276	// present in the skip list.
277	if (CarbonCompactionUtil.isCompactionRequiredForTable(metadataPath) && !skipList	×
278	.contains(ctable.getCarbonTableIdentifier())) {	×
279	return ctable;	×
280	}
281	}
282	return null;	×
283	}
284
285	/**
286	* This method will add the prepare the max column cardinality map
287	*
288	* @param columnCardinalityMap
289	* @param currentBlockSchema
290	* @param currentBlockCardinality
291	*/
292	public static void addColumnCardinalityToMap(Map<String, Integer> columnCardinalityMap,
293	List<ColumnSchema> currentBlockSchema, int[] currentBlockCardinality) {
294	for (int i = 0; i < currentBlockCardinality.length; i++) {	1✔
295	// add value to map only if does not exist or new cardinality is > existing value
296	String columnUniqueId = currentBlockSchema.get(i).getColumnUniqueId();	1✔
297	Integer value = columnCardinalityMap.get(columnUniqueId);	1✔
298	if (null == value) {	1✔
299	columnCardinalityMap.put(columnUniqueId, currentBlockCardinality[i]);	1✔
300	} else {
301	if (currentBlockCardinality[i] > value) {	1✔
302	columnCardinalityMap.put(columnUniqueId, currentBlockCardinality[i]);	×
303	}
304	}
305	}
306	}	1✔
307
308	/**
309	* This method will return the updated cardinality according to the master schema
310	*
311	* @param columnCardinalityMap
312	* @param carbonTable
313	* @param updatedColumnSchemaList
314	* @return
315	*/
316	public static int[] updateColumnSchemaAndGetCardinality(Map<String, Integer> columnCardinalityMap,
317	CarbonTable carbonTable, List<ColumnSchema> updatedColumnSchemaList) {
318	List<CarbonDimension> masterDimensions =	1✔
319	carbonTable.getDimensionByTableName(carbonTable.getTableName());	1✔
320	List<Integer> updatedCardinalityList = new ArrayList<>(columnCardinalityMap.size());	1✔
321	for (CarbonDimension dimension : masterDimensions) {	1✔
322	Integer value = columnCardinalityMap.get(dimension.getColumnId());	1✔
323	if (null == value) {	1✔
324	updatedCardinalityList.add(getDimensionDefaultCardinality(dimension));	1✔
325	} else {
326	updatedCardinalityList.add(value);	1✔
327	}
328	updatedColumnSchemaList.add(dimension.getColumnSchema());	1✔
329
330	if (dimension.getNumberOfChild() > 0) {	1✔
331	fillColumnSchemaListForComplexDims(dimension.getListOfChildDimensions(),	1✔
332	updatedColumnSchemaList, updatedCardinalityList, columnCardinalityMap);
333	}
334	}	1✔
335	// add measures to the column schema list
336	List<CarbonMeasure> masterSchemaMeasures =	1✔
337	carbonTable.getMeasureByTableName(carbonTable.getTableName());	1✔
338	for (CarbonMeasure measure : masterSchemaMeasures) {	1✔
339	updatedColumnSchemaList.add(measure.getColumnSchema());	1✔
340	}	1✔
341	return ArrayUtils	1✔
342	.toPrimitive(updatedCardinalityList.toArray(new Integer[updatedCardinalityList.size()]));	1✔
343	}
344
345	/**
346	* This method is to get the chile dimensions of the complex dimension and
347	* update the cardinality for all complex dimensions
348	*
349	* @param carbonDimensionsList
350	* @param updatedColumnSchemaList
351	* @param updatedCardinalityList
352	* @param columnCardinalityMap
353	*/
354	private static void fillColumnSchemaListForComplexDims(List<CarbonDimension> carbonDimensionsList,
355	List<ColumnSchema> updatedColumnSchemaList, List<Integer> updatedCardinalityList,
356	Map<String, Integer> columnCardinalityMap) {
357	for (CarbonDimension carbonDimension : carbonDimensionsList) {	1✔
358	Integer value = columnCardinalityMap.get(carbonDimension.getColumnId());	1✔
359	if (null == value) {	1✔
360	updatedCardinalityList.add(getDimensionDefaultCardinality(carbonDimension));	×
361	} else {
362	updatedCardinalityList.add(value);	1✔
363	}
364	updatedColumnSchemaList.add(carbonDimension.getColumnSchema());	1✔
365	List<CarbonDimension> childDims = carbonDimension.getListOfChildDimensions();	1✔
366	if (null != childDims && childDims.size() > 0) {	1✔
367	fillColumnSchemaListForComplexDims(childDims, updatedColumnSchemaList,	1✔
368	updatedCardinalityList, columnCardinalityMap);
369	}
370	}	1✔
371	}	1✔
372
373	/**
374	* This method will return the default cardinality based on dimension type
375	*
376	* @param dimension
377	* @return
378	*/
379	private static int getDimensionDefaultCardinality(CarbonDimension dimension) {
380	int cardinality = 0;	1✔
381	if (dimension.hasEncoding(Encoding.DIRECT_DICTIONARY)) {	1✔
382	cardinality = Integer.MAX_VALUE;	×
383	} else if (dimension.hasEncoding(Encoding.DICTIONARY)) {	1✔
384	if (null != dimension.getDefaultValue()) {	×
385	cardinality = CarbonCommonConstants.DICTIONARY_DEFAULT_CARDINALITY + 1;	×
386	} else {
387	cardinality = CarbonCommonConstants.DICTIONARY_DEFAULT_CARDINALITY;	×
388	}
389	} else {
390	cardinality = -1;	1✔
391	}
392	return cardinality;	1✔
393	}
394
395	/**
396	* This method will check for any restructured block in the blocks selected for compaction
397	*
398	* @param segmentMapping
399	* @param dataFileMetadataSegMapping
400	* @param tableLastUpdatedTime
401	* @return
402	*/
403	public static boolean checkIfAnyRestructuredBlockExists(Map<String, TaskBlockInfo> segmentMapping,
404	Map<String, List<DataFileFooter>> dataFileMetadataSegMapping, long tableLastUpdatedTime) {
405	boolean restructuredBlockExists = false;	1✔
406	for (Map.Entry<String, TaskBlockInfo> taskMap : segmentMapping.entrySet()) {	1✔
407	String segmentId = taskMap.getKey();	1✔
408	List<DataFileFooter> listMetadata = dataFileMetadataSegMapping.get(segmentId);	1✔
409	for (DataFileFooter dataFileFooter : listMetadata) {	1✔
410	// if schema modified timestamp is greater than footer stored schema timestamp,
411	// it indicates it is a restructured block
412	if (tableLastUpdatedTime > dataFileFooter.getSchemaUpdatedTimeStamp()) {	1✔
413	restructuredBlockExists = true;	1✔
414	break;	1✔
415	}
416	}	1✔
417	if (restructuredBlockExists) {	1✔
418	break;	1✔
419	}
420	}	1✔
421	return restructuredBlockExists;	1✔
422	}
423	}

apache / carbondata / 10000

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