10727

Build # 10727

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/023cb4dc95481c796f8ef3beda2c43d3089e7ad5">023cb4dc9<a href="https://github.com/apache/carbondata/commit/023cb4dc95481c796f8ef3beda2c43d3089e7ad5">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/25942c506bcaff552d033f5dd177d3f8d22e341b">25942c506</a>">25942c506</a><a href="https://github.com/apache/carbondata/commit/023cb4dc95481c796f8ef3beda2c43d3089e7ad5"> into aa0702091">aa0702091</a>

Run Details

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

62445 of 78823 relevant lines covered (79.22%)

1.05 hits per line

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

58.02

/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.BlockletInfo;
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());


  /**
   * The processTableBlocks returns a Map<String, List<RawResultIterator>>.
   *
   * This Map has two K,V Pairs. The element with Key=UNSORTED_IDX
   * is the list of all the iterators which yield unsorted data.
   *
   * This Map has two K,V Pairs. The element with Key=SORTED_IDX
   * is the list of all the iterators which yield sorted data.
   */
  public static final String UNSORTED_IDX = "UNSORTED_IDX";
  public static final String SORTED_IDX = "SORTED_IDX";

  /**
   * 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, boolean isSortedTable) 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
      final BlockletInfo blockletInfo = blockInfo.getDetailInfo().getBlockletInfo();
      if (null != blockInfo.getDetailInfo() && (
          blockInfo.getDetailInfo().getSchemaUpdatedTimeStamp() == 0L || null == blockletInfo
              || null == blockletInfo.isSorted() || !blockletInfo.isSorted())) {
        dataFileMatadata = CarbonUtil.readMetadataFile(blockInfo, true);
        if (null == dataFileMatadata.isSorted()) {
          dataFileMatadata.setSorted(isSortedTable);
        }
      } else {
        dataFileMatadata = CarbonUtil.readMetadataFile(blockInfo);
      }
      blockInfo.setDataFileFooter(dataFileMatadata);
      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) {

    for (Map.Entry<String, TaskBlockInfo> segmentEntry : segmentMapping.entrySet()) {
      String segmentId = segmentEntry.getKey();
      List<DataFileFooter> listMetadata = dataFileMetadataSegMapping.get(segmentId);

      if (isRestructured(listMetadata, tableLastUpdatedTime)) {
        return true;
      }
    }

    return false;
  }

  /**
   * Returns if any element in the list of DataFileFooter
   * is restructured or not.
   *
   * @param listMetadata
   * @param tableLastUpdatedTime
   * @return
   */
  public static boolean isRestructured(List<DataFileFooter> listMetadata,
      long tableLastUpdatedTime) {
    /*
     * TODO: only in case of add and drop this variable should be true
     */
    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()) {
        return true;
      }
    }
    return false;
  }

  /**
   * Returns if the DataFileFooter containing carbondata file contains
   * sorted data or not.
   *
   * @param footer
   * @return
   * @throws IOException
   */
  public static boolean isSorted(DataFileFooter footer) throws IOException {
    return footer.isSorted();
  }

}

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

apache / carbondata / 10727

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