apache / carbondata / 389

Build # 389

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/4c1e348a6790dcf318f0895d3849070ed71653fa">4c1e348a6<a href="https://github.com/apache/carbondata/commit/4c1e348a6790dcf318f0895d3849070ed71653fa">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/3debb970430b5443ee113aa04b434e6ba8ac3be8">3debb9704</a>">3debb9704</a><a href="https://github.com/apache/carbondata/commit/4c1e348a6790dcf318f0895d3849070ed71653fa"> into 73a5885a4">73a5885a4</a>

Run Details

13 of 13 new or added lines in 5 files covered. (100.0%)

58200 of 72198 relevant lines covered (80.61%)

1.07 hits per line

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

77.27

/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionResultSortProcessor.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.File;
import java.io.IOException;
import java.util.List;

import org.apache.carbondata.common.logging.LogService;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.datastore.exception.CarbonDataWriterException;
import org.apache.carbondata.core.datastore.row.CarbonRow;
import org.apache.carbondata.core.indexstore.PartitionSpec;
import org.apache.carbondata.core.metadata.SegmentFileStore;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
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.scan.result.iterator.RawResultIterator;
import org.apache.carbondata.core.scan.wrappers.ByteArrayWrapper;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.DataTypeUtil;
import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
import org.apache.carbondata.processing.sort.exception.CarbonSortKeyAndGroupByException;
import org.apache.carbondata.processing.sort.sortdata.SingleThreadFinalSortFilesMerger;
import org.apache.carbondata.processing.sort.sortdata.SortDataRows;
import org.apache.carbondata.processing.sort.sortdata.SortIntermediateFileMerger;
import org.apache.carbondata.processing.sort.sortdata.SortParameters;
import org.apache.carbondata.processing.store.CarbonFactDataHandlerModel;
import org.apache.carbondata.processing.store.CarbonFactHandler;
import org.apache.carbondata.processing.store.CarbonFactHandlerFactory;
import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;

/**
 * This class will process the query result and convert the data
 * into a format compatible for data load
 */
public class CompactionResultSortProcessor extends AbstractResultProcessor {

  /**
   * LOGGER
   */
  private static final LogService LOGGER =
      LogServiceFactory.getLogService(CompactionResultSortProcessor.class.getName());
  /**
   * carbon load model that contains all the required information for load
   */
  private CarbonLoadModel carbonLoadModel;
  /**
   * carbon table
   */
  private CarbonTable carbonTable;
  /**
   * sortDataRows instance for sorting each row read ad writing to sort temp file
   */
  private SortDataRows sortDataRows;
  /**
   * final merger for merge sort
   */
  private SingleThreadFinalSortFilesMerger finalMerger;
  /**
   * data handler VO object
   */
  private CarbonFactHandler dataHandler;
  /**
   * segment properties for getting dimension cardinality and other required information of a block
   */
  private SegmentProperties segmentProperties;
  /**
   * compaction type to decide whether taskID need to be extracted from carbondata files
   */
  private CompactionType compactionType;
  /**
   * boolean mapping for no dictionary columns in schema
   */
  private boolean[] noDictionaryColMapping;
  /**
   * boolean mapping for long string dimension
   */
  private boolean[] isVarcharDimMapping;
  /**
   * agg type defined for measures
   */
  private DataType[] dataTypes;
  /**
   * segment id
   */
  private String segmentId;
  /**
   * temp store location to be sued during data load
   */
  private String[] tempStoreLocation;
  /**
   * table name
   */
  private String tableName;
  /**
   * no dictionary column count in schema
   */
  private int noDictionaryCount;
  /**
   * total count of measures in schema
   */
  private int measureCount;
  /**
   * dimension count excluding complex dimension and no dictionary column count
   */
  private int dimensionColumnCount;
  /**
   * whether the allocated tasks has any record
   */
  private boolean isRecordFound;
  /**
   * intermediate sort merger
   */
  private SortIntermediateFileMerger intermediateFileMerger;

  private PartitionSpec partitionSpec;

  private SortParameters sortParameters;

  public CompactionResultSortProcessor(CarbonLoadModel carbonLoadModel, CarbonTable carbonTable,
      SegmentProperties segmentProperties, CompactionType compactionType, String tableName,
      PartitionSpec partitionSpec) {
    this.carbonLoadModel = carbonLoadModel;
    this.carbonTable = carbonTable;
    this.segmentProperties = segmentProperties;
    this.segmentId = carbonLoadModel.getSegmentId();
    this.compactionType = compactionType;
    this.tableName = tableName;
    this.partitionSpec = partitionSpec;
  }

  /**
   * This method will iterate over the query result and convert it into a format compatible
   * for data loading
   *
   * @param resultIteratorList
   */
  public boolean execute(List<RawResultIterator> resultIteratorList) throws Exception {
    boolean isCompactionSuccess = false;
    try {
      initTempStoreLocation();
      initSortDataRows();
      dataTypes = CarbonDataProcessorUtil.initDataType(carbonTable, tableName, measureCount);
      processResult(resultIteratorList);
      // After delete command, if no records are fetched from one split,
      // below steps are not required to be initialized.
      if (isRecordFound) {
        initializeFinalThreadMergerForMergeSort();
        initDataHandler();
        readAndLoadDataFromSortTempFiles();
      }
      isCompactionSuccess = true;
    } catch (Exception e) {
      throw e;
    } finally {
      if (partitionSpec != null) {
        try {
          SegmentFileStore
              .writeSegmentFile(carbonLoadModel.getTablePath(), carbonLoadModel.getTaskNo(),
                  partitionSpec.getLocation().toString(), carbonLoadModel.getFactTimeStamp() + "",
                  partitionSpec.getPartitions());
        } catch (IOException e) {
          throw e;
        }
      }
      // clear temp files and folders created during compaction
      deleteTempStoreLocation();
    }
    return isCompactionSuccess;
  }

  @Override
  public void close() {
    // close the sorter executor service
    if (null != sortDataRows) {
      sortDataRows.close();
    }
    // close the final merger
    if (null != finalMerger) {
      finalMerger.close();
    }
    // close data handler
    if (null != dataHandler) {
      dataHandler.closeHandler();
    }
  }

  /**
   * This method will clean up the local folders and files created during compaction process
   */
  private void deleteTempStoreLocation() {
    if (null != tempStoreLocation) {
      for (String tempLoc : tempStoreLocation) {
        try {
          CarbonUtil.deleteFoldersAndFiles(new File(tempLoc));
        } catch (IOException | InterruptedException e) {
          LOGGER.error("Problem deleting local folders during compaction: " + e.getMessage());
        }
      }
    }
  }

  /**
   * This method will iterate over the query result and perform row sorting operation
   *
   * @param resultIteratorList
   */
  private void processResult(List<RawResultIterator> resultIteratorList) throws Exception {
    for (RawResultIterator resultIterator : resultIteratorList) {
      if (CompactionType.STREAMING == compactionType) {
        while (resultIterator.hasNext()) {
          // the input iterator of streaming segment is already using raw row
          addRowForSorting(resultIterator.next());
          isRecordFound = true;
        }
      } else {
        while (resultIterator.hasNext()) {
          addRowForSorting(prepareRowObjectForSorting(resultIterator.next()));
          isRecordFound = true;
        }
      }
      resultIterator.close();
    }
    try {
      sortDataRows.startSorting();
    } catch (CarbonSortKeyAndGroupByException e) {
      LOGGER.error(e);
      throw new Exception("Problem loading data during compaction: " + e.getMessage());
    }
  }

  /**
   * This method will prepare the data from raw object that will take part in sorting
   *
   * @param row
   * @return
   */
  private Object[] prepareRowObjectForSorting(Object[] row) {
    ByteArrayWrapper wrapper = (ByteArrayWrapper) row[0];
    // ByteBuffer[] noDictionaryBuffer = new ByteBuffer[noDictionaryCount];
    List<CarbonDimension> dimensions = segmentProperties.getDimensions();
    Object[] preparedRow = new Object[dimensions.size() + measureCount];
    // convert the dictionary from MDKey to surrogate key
    byte[] dictionaryKey = wrapper.getDictionaryKey();
    long[] keyArray = segmentProperties.getDimensionKeyGenerator().getKeyArray(dictionaryKey);
    Object[] dictionaryValues = new Object[dimensionColumnCount + measureCount];
    for (int i = 0; i < keyArray.length; i++) {
      dictionaryValues[i] = Long.valueOf(keyArray[i]).intValue();
    }
    int noDictionaryIndex = 0;
    int dictionaryIndex = 0;
    for (int i = 0; i < dimensions.size(); i++) {
      CarbonDimension dims = dimensions.get(i);
      if (dims.hasEncoding(Encoding.DICTIONARY)) {
        // dictionary
        preparedRow[i] = dictionaryValues[dictionaryIndex++];
      } else {
        // no dictionary dims
        preparedRow[i] = wrapper.getNoDictionaryKeyByIndex(noDictionaryIndex++);
      }
    }
    // fill all the measures
    // measures will always start from 1st index in the row object array
    int measureIndexInRow = 1;
    for (int i = 0; i < measureCount; i++) {
      preparedRow[dimensionColumnCount + i] =
          getConvertedMeasureValue(row[measureIndexInRow++], dataTypes[i]);
    }
    return preparedRow;
  }

  /**
   * This method will convert the spark decimal to java big decimal type
   *
   * @param value
   * @param type
   * @return
   */
  private Object getConvertedMeasureValue(Object value, DataType type) {
    if (DataTypes.isDecimal(type)) {
      if (value != null) {
        value = DataTypeUtil.getDataTypeConverter().convertFromDecimalToBigDecimal(value);
      }
      return value;
    } else {
      return value;
    }
  }

  /**
   * This method will read sort temp files, perform merge sort and add it to store for data loading
   */
  private void readAndLoadDataFromSortTempFiles() throws Exception {
    try {
      intermediateFileMerger.finish();
      finalMerger.startFinalMerge();
      while (finalMerger.hasNext()) {
        Object[] row = finalMerger.next();
        dataHandler.addDataToStore(new CarbonRow(row));
      }
      dataHandler.finish();
    } catch (CarbonDataWriterException e) {
      LOGGER.error(e);
      throw new Exception("Problem loading data during compaction.", e);
    } catch (Exception e) {
      LOGGER.error(e);
      throw new Exception("Problem loading data during compaction.", e);
    } finally {
      if (null != dataHandler) {
        try {
          dataHandler.closeHandler();
        } catch (CarbonDataWriterException e) {
          LOGGER.error(e, "Error in close data handler");
          throw new Exception("Error in close data handler", e);
        }
      }
    }
  }

  /**
   * add row to a temp array which will we written to a sort temp file after sorting
   *
   * @param row
   */
  private void addRowForSorting(Object[] row) throws Exception {
    try {
      sortDataRows.addRow(row);
    } catch (CarbonSortKeyAndGroupByException e) {
      LOGGER.error(e);
      throw new Exception("Row addition for sorting failed during compaction: " + e.getMessage());
    }
  }

  /**
   * create an instance of sort data rows
   */
  private void initSortDataRows() throws Exception {
    measureCount = carbonTable.getMeasureByTableName(tableName).size();
    List<CarbonDimension> dimensions = carbonTable.getDimensionByTableName(tableName);
    noDictionaryColMapping = new boolean[dimensions.size()];
    isVarcharDimMapping = new boolean[dimensions.size()];
    int i = 0;
    for (CarbonDimension dimension : dimensions) {
      if (CarbonUtil.hasEncoding(dimension.getEncoder(), Encoding.DICTIONARY)) {
        i++;
        continue;
      }
      noDictionaryColMapping[i] = true;
      if (dimension.getColumnSchema().getDataType() == DataTypes.VARCHAR) {
        isVarcharDimMapping[i] = true;
      }
      i++;
      noDictionaryCount++;
    }
    dimensionColumnCount = dimensions.size();
    sortParameters = createSortParameters();
    intermediateFileMerger = new SortIntermediateFileMerger(sortParameters);
    // TODO: Now it is only supported onheap merge, but we can have unsafe merge
    // as well by using UnsafeSortDataRows.
    this.sortDataRows = new SortDataRows(sortParameters, intermediateFileMerger);
    try {
      this.sortDataRows.initialize();
    } catch (CarbonSortKeyAndGroupByException e) {
      LOGGER.error(e);
      throw new Exception(
          "Error initializing sort data rows object during compaction: " + e.getMessage());
    }
  }

  /**
   * This method will create the sort parameters VO object
   *
   * @return
   */
  private SortParameters createSortParameters() {
    return SortParameters
        .createSortParameters(carbonTable, carbonLoadModel.getDatabaseName(), tableName,
            dimensionColumnCount, segmentProperties.getComplexDimensions().size(), measureCount,
            noDictionaryCount, segmentId,
            carbonLoadModel.getTaskNo(), noDictionaryColMapping, isVarcharDimMapping, true);
  }

  /**
   * create an instance of finalThread merger which will perform merge sort on all the
   * sort temp files
   */
  private void initializeFinalThreadMergerForMergeSort() {
    boolean[] noDictionarySortColumnMapping = null;
    if (noDictionaryColMapping.length == this.segmentProperties.getNumberOfSortColumns()) {
      noDictionarySortColumnMapping = noDictionaryColMapping;
    } else {
      noDictionarySortColumnMapping = new boolean[this.segmentProperties.getNumberOfSortColumns()];
      System.arraycopy(noDictionaryColMapping, 0,
          noDictionarySortColumnMapping, 0, noDictionarySortColumnMapping.length);
    }
    sortParameters.setNoDictionarySortColumn(noDictionarySortColumnMapping);
    String[] sortTempFileLocation = CarbonDataProcessorUtil.arrayAppend(tempStoreLocation,
        CarbonCommonConstants.FILE_SEPARATOR, CarbonCommonConstants.SORT_TEMP_FILE_LOCATION);
    finalMerger =
        new SingleThreadFinalSortFilesMerger(sortTempFileLocation, tableName, sortParameters);
  }

  /**
   * initialise carbon data writer instance
   */
  private void initDataHandler() throws Exception {
    String carbonStoreLocation;
    if (partitionSpec != null) {
      carbonStoreLocation =
          partitionSpec.getLocation().toString() + CarbonCommonConstants.FILE_SEPARATOR
              + carbonLoadModel.getFactTimeStamp() + ".tmp";
    } else {
      carbonStoreLocation = CarbonDataProcessorUtil
          .createCarbonStoreLocation(carbonLoadModel.getDatabaseName(), tableName,
              carbonLoadModel.getSegmentId());
    }
    CarbonFactDataHandlerModel carbonFactDataHandlerModel = CarbonFactDataHandlerModel
        .getCarbonFactDataHandlerModel(carbonLoadModel, carbonTable, segmentProperties, tableName,
            tempStoreLocation, carbonStoreLocation);
    carbonFactDataHandlerModel.setSegmentId(carbonLoadModel.getSegmentId());
    setDataFileAttributesInModel(carbonLoadModel, compactionType, carbonFactDataHandlerModel);
    dataHandler = CarbonFactHandlerFactory.createCarbonFactHandler(carbonFactDataHandlerModel);
    try {
      dataHandler.initialise();
    } catch (CarbonDataWriterException e) {
      LOGGER.error(e);
      throw new Exception("Problem initialising data handler during compaction: " + e.getMessage());
    }
  }

  /**
   * initialise temporary store location
   */
  private void initTempStoreLocation() {
    tempStoreLocation = CarbonDataProcessorUtil
        .getLocalDataFolderLocation(carbonTable, carbonLoadModel.getTaskNo(),
           segmentId, true, false);
  }
}

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.File;
20	import java.io.IOException;
21	import java.util.List;
22
23	import org.apache.carbondata.common.logging.LogService;
24	import org.apache.carbondata.common.logging.LogServiceFactory;
25	import org.apache.carbondata.core.constants.CarbonCommonConstants;
26	import org.apache.carbondata.core.datastore.block.SegmentProperties;
27	import org.apache.carbondata.core.datastore.exception.CarbonDataWriterException;
28	import org.apache.carbondata.core.datastore.row.CarbonRow;
29	import org.apache.carbondata.core.indexstore.PartitionSpec;
30	import org.apache.carbondata.core.metadata.SegmentFileStore;
31	import org.apache.carbondata.core.metadata.datatype.DataType;
32	import org.apache.carbondata.core.metadata.datatype.DataTypes;
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.scan.result.iterator.RawResultIterator;
37	import org.apache.carbondata.core.scan.wrappers.ByteArrayWrapper;
38	import org.apache.carbondata.core.util.CarbonUtil;
39	import org.apache.carbondata.core.util.DataTypeUtil;
40	import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
41	import org.apache.carbondata.processing.sort.exception.CarbonSortKeyAndGroupByException;
42	import org.apache.carbondata.processing.sort.sortdata.SingleThreadFinalSortFilesMerger;
43	import org.apache.carbondata.processing.sort.sortdata.SortDataRows;
44	import org.apache.carbondata.processing.sort.sortdata.SortIntermediateFileMerger;
45	import org.apache.carbondata.processing.sort.sortdata.SortParameters;
46	import org.apache.carbondata.processing.store.CarbonFactDataHandlerModel;
47	import org.apache.carbondata.processing.store.CarbonFactHandler;
48	import org.apache.carbondata.processing.store.CarbonFactHandlerFactory;
49	import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
50
51	/**
52	* This class will process the query result and convert the data
53	* into a format compatible for data load
54	*/
55	public class CompactionResultSortProcessor extends AbstractResultProcessor {
56
57	/**
58	* LOGGER
59	*/
60	private static final LogService LOGGER =	1✔
61	LogServiceFactory.getLogService(CompactionResultSortProcessor.class.getName());	1✔
62	/**
63	* carbon load model that contains all the required information for load
64	*/
65	private CarbonLoadModel carbonLoadModel;
66	/**
67	* carbon table
68	*/
69	private CarbonTable carbonTable;
70	/**
71	* sortDataRows instance for sorting each row read ad writing to sort temp file
72	*/
73	private SortDataRows sortDataRows;
74	/**
75	* final merger for merge sort
76	*/
77	private SingleThreadFinalSortFilesMerger finalMerger;
78	/**
79	* data handler VO object
80	*/
81	private CarbonFactHandler dataHandler;
82	/**
83	* segment properties for getting dimension cardinality and other required information of a block
84	*/
85	private SegmentProperties segmentProperties;
86	/**
87	* compaction type to decide whether taskID need to be extracted from carbondata files
88	*/
89	private CompactionType compactionType;
90	/**
91	* boolean mapping for no dictionary columns in schema
92	*/
93	private boolean[] noDictionaryColMapping;
94	/**
95	* boolean mapping for long string dimension
96	*/
97	private boolean[] isVarcharDimMapping;
98	/**
99	* agg type defined for measures
100	*/
101	private DataType[] dataTypes;
102	/**
103	* segment id
104	*/
105	private String segmentId;
106	/**
107	* temp store location to be sued during data load
108	*/
109	private String[] tempStoreLocation;
110	/**
111	* table name
112	*/
113	private String tableName;
114	/**
115	* no dictionary column count in schema
116	*/
117	private int noDictionaryCount;
118	/**
119	* total count of measures in schema
120	*/
121	private int measureCount;
122	/**
123	* dimension count excluding complex dimension and no dictionary column count
124	*/
125	private int dimensionColumnCount;
126	/**
127	* whether the allocated tasks has any record
128	*/
129	private boolean isRecordFound;
130	/**
131	* intermediate sort merger
132	*/
133	private SortIntermediateFileMerger intermediateFileMerger;
134
135	private PartitionSpec partitionSpec;
136
137	private SortParameters sortParameters;
138
139	public CompactionResultSortProcessor(CarbonLoadModel carbonLoadModel, CarbonTable carbonTable,
140	SegmentProperties segmentProperties, CompactionType compactionType, String tableName,
141	PartitionSpec partitionSpec) {	1✔
142	this.carbonLoadModel = carbonLoadModel;	1✔
143	this.carbonTable = carbonTable;	1✔
144	this.segmentProperties = segmentProperties;	1✔
145	this.segmentId = carbonLoadModel.getSegmentId();	1✔
146	this.compactionType = compactionType;	1✔
147	this.tableName = tableName;	1✔
148	this.partitionSpec = partitionSpec;	1✔
149	}	1✔
150
151	/**
152	* This method will iterate over the query result and convert it into a format compatible
153	* for data loading
154	*
155	* @param resultIteratorList
156	*/
157	public boolean execute(List<RawResultIterator> resultIteratorList) throws Exception {
158	boolean isCompactionSuccess = false;	1✔
159	try {
160	initTempStoreLocation();	1✔
161	initSortDataRows();	1✔
162	dataTypes = CarbonDataProcessorUtil.initDataType(carbonTable, tableName, measureCount);	1✔
163	processResult(resultIteratorList);	1✔
164	// After delete command, if no records are fetched from one split,
165	// below steps are not required to be initialized.
166	if (isRecordFound) {	1✔
167	initializeFinalThreadMergerForMergeSort();	1✔
168	initDataHandler();	1✔
169	readAndLoadDataFromSortTempFiles();	1✔
170	}
171	isCompactionSuccess = true;	1✔
172	} catch (Exception e) {	×
173	throw e;	×
174	} finally {
175	if (partitionSpec != null) {	1✔
176	try {
177	SegmentFileStore	×
178	.writeSegmentFile(carbonLoadModel.getTablePath(), carbonLoadModel.getTaskNo(),	×
179	partitionSpec.getLocation().toString(), carbonLoadModel.getFactTimeStamp() + "",	×
180	partitionSpec.getPartitions());	×
181	} catch (IOException e) {	×
182	throw e;	×
183	}	×
184	}
185	// clear temp files and folders created during compaction
186	deleteTempStoreLocation();	1✔
187	}	1✔
188	return isCompactionSuccess;	1✔
189	}
190
191	@Override
192	public void close() {
193	// close the sorter executor service
194	if (null != sortDataRows) {	1✔
195	sortDataRows.close();	1✔
196	}
197	// close the final merger
198	if (null != finalMerger) {	1✔
199	finalMerger.close();	1✔
200	}
201	// close data handler
202	if (null != dataHandler) {	1✔
203	dataHandler.closeHandler();	1✔
204	}
205	}	1✔
206
207	/**
208	* This method will clean up the local folders and files created during compaction process
209	*/
210	private void deleteTempStoreLocation() {
211	if (null != tempStoreLocation) {	1✔
212	for (String tempLoc : tempStoreLocation) {	1✔
213	try {
214	CarbonUtil.deleteFoldersAndFiles(new File(tempLoc));	1✔
215	} catch (IOException \| InterruptedException e) {	×
216	LOGGER.error("Problem deleting local folders during compaction: " + e.getMessage());	×
217	}	1✔
218	}
219	}
220	}	1✔
221
222	/**
223	* This method will iterate over the query result and perform row sorting operation
224	*
225	* @param resultIteratorList
226	*/
227	private void processResult(List<RawResultIterator> resultIteratorList) throws Exception {
228	for (RawResultIterator resultIterator : resultIteratorList) {	1✔
229	if (CompactionType.STREAMING == compactionType) {	1✔
230	while (resultIterator.hasNext()) {	1✔
231	// the input iterator of streaming segment is already using raw row
232	addRowForSorting(resultIterator.next());	1✔
233	isRecordFound = true;	1✔
234	}
235	} else {
236	while (resultIterator.hasNext()) {	1✔
237	addRowForSorting(prepareRowObjectForSorting(resultIterator.next()));	1✔
238	isRecordFound = true;	1✔
239	}
240	}
241	resultIterator.close();	1✔
242	}	1✔
243	try {
244	sortDataRows.startSorting();	1✔
245	} catch (CarbonSortKeyAndGroupByException e) {	×
246	LOGGER.error(e);	×
247	throw new Exception("Problem loading data during compaction: " + e.getMessage());	×
248	}	1✔
249	}	1✔
250
251	/**
252	* This method will prepare the data from raw object that will take part in sorting
253	*
254	* @param row
255	* @return
256	*/
257	private Object[] prepareRowObjectForSorting(Object[] row) {
258	ByteArrayWrapper wrapper = (ByteArrayWrapper) row[0];	1✔
259	// ByteBuffer[] noDictionaryBuffer = new ByteBuffer[noDictionaryCount];
260	List<CarbonDimension> dimensions = segmentProperties.getDimensions();	1✔
261	Object[] preparedRow = new Object[dimensions.size() + measureCount];	1✔
262	// convert the dictionary from MDKey to surrogate key
263	byte[] dictionaryKey = wrapper.getDictionaryKey();	1✔
264	long[] keyArray = segmentProperties.getDimensionKeyGenerator().getKeyArray(dictionaryKey);	1✔
265	Object[] dictionaryValues = new Object[dimensionColumnCount + measureCount];	1✔
266	for (int i = 0; i < keyArray.length; i++) {	1✔
267	dictionaryValues[i] = Long.valueOf(keyArray[i]).intValue();	×
268	}
269	int noDictionaryIndex = 0;	1✔
270	int dictionaryIndex = 0;	1✔
271	for (int i = 0; i < dimensions.size(); i++) {	1✔
272	CarbonDimension dims = dimensions.get(i);	1✔
273	if (dims.hasEncoding(Encoding.DICTIONARY)) {	1✔
274	// dictionary
275	preparedRow[i] = dictionaryValues[dictionaryIndex++];	×
276	} else {
277	// no dictionary dims
278	preparedRow[i] = wrapper.getNoDictionaryKeyByIndex(noDictionaryIndex++);	1✔
279	}
280	}
281	// fill all the measures
282	// measures will always start from 1st index in the row object array
283	int measureIndexInRow = 1;	1✔
284	for (int i = 0; i < measureCount; i++) {	1✔
285	preparedRow[dimensionColumnCount + i] =	1✔
286	getConvertedMeasureValue(row[measureIndexInRow++], dataTypes[i]);	1✔
287	}
288	return preparedRow;	1✔
289	}
290
291	/**
292	* This method will convert the spark decimal to java big decimal type
293	*
294	* @param value
295	* @param type
296	* @return
297	*/
298	private Object getConvertedMeasureValue(Object value, DataType type) {
299	if (DataTypes.isDecimal(type)) {	1✔
300	if (value != null) {	1✔
301	value = DataTypeUtil.getDataTypeConverter().convertFromDecimalToBigDecimal(value);	1✔
302	}
303	return value;	1✔
304	} else {
305	return value;	1✔
306	}
307	}
308
309	/**
310	* This method will read sort temp files, perform merge sort and add it to store for data loading
311	*/
312	private void readAndLoadDataFromSortTempFiles() throws Exception {
313	try {
314	intermediateFileMerger.finish();	1✔
315	finalMerger.startFinalMerge();	1✔
316	while (finalMerger.hasNext()) {	1✔
317	Object[] row = finalMerger.next();	1✔
318	dataHandler.addDataToStore(new CarbonRow(row));	1✔
319	}	1✔
320	dataHandler.finish();	1✔
321	} catch (CarbonDataWriterException e) {	×
322	LOGGER.error(e);	×
323	throw new Exception("Problem loading data during compaction.", e);	×
324	} catch (Exception e) {	×
325	LOGGER.error(e);	×
326	throw new Exception("Problem loading data during compaction.", e);	×
327	} finally {
328	if (null != dataHandler) {	1✔
329	try {
330	dataHandler.closeHandler();	1✔
331	} catch (CarbonDataWriterException e) {	×
332	LOGGER.error(e, "Error in close data handler");	×
333	throw new Exception("Error in close data handler", e);	×
334	}	1✔
335	}
336	}	×
337	}	1✔
338
339	/**
340	* add row to a temp array which will we written to a sort temp file after sorting
341	*
342	* @param row
343	*/
344	private void addRowForSorting(Object[] row) throws Exception {
345	try {
346	sortDataRows.addRow(row);	1✔
347	} catch (CarbonSortKeyAndGroupByException e) {	×
348	LOGGER.error(e);	×
349	throw new Exception("Row addition for sorting failed during compaction: " + e.getMessage());	×
350	}	1✔
351	}	1✔
352
353	/**
354	* create an instance of sort data rows
355	*/
356	private void initSortDataRows() throws Exception {
357	measureCount = carbonTable.getMeasureByTableName(tableName).size();	1✔
358	List<CarbonDimension> dimensions = carbonTable.getDimensionByTableName(tableName);	1✔
359	noDictionaryColMapping = new boolean[dimensions.size()];	1✔
360	isVarcharDimMapping = new boolean[dimensions.size()];	1✔
361	int i = 0;	1✔
362	for (CarbonDimension dimension : dimensions) {	1✔
363	if (CarbonUtil.hasEncoding(dimension.getEncoder(), Encoding.DICTIONARY)) {	1✔
364	i++;	1✔
365	continue;	1✔
366	}
367	noDictionaryColMapping[i] = true;	1✔
368	if (dimension.getColumnSchema().getDataType() == DataTypes.VARCHAR) {	1✔
369	isVarcharDimMapping[i] = true;	×
370	}
371	i++;	1✔
372	noDictionaryCount++;	1✔
373	}	1✔
374	dimensionColumnCount = dimensions.size();	1✔
375	sortParameters = createSortParameters();	1✔
376	intermediateFileMerger = new SortIntermediateFileMerger(sortParameters);	1✔
377	// TODO: Now it is only supported onheap merge, but we can have unsafe merge
378	// as well by using UnsafeSortDataRows.
379	this.sortDataRows = new SortDataRows(sortParameters, intermediateFileMerger);	1✔
380	try {
381	this.sortDataRows.initialize();	1✔
382	} catch (CarbonSortKeyAndGroupByException e) {	×
383	LOGGER.error(e);	×
384	throw new Exception(	×
385	"Error initializing sort data rows object during compaction: " + e.getMessage());	×
386	}	1✔
387	}	1✔
388
389	/**
390	* This method will create the sort parameters VO object
391	*
392	* @return
393	*/
394	private SortParameters createSortParameters() {
395	return SortParameters	1✔
396	.createSortParameters(carbonTable, carbonLoadModel.getDatabaseName(), tableName,	1✔
397	dimensionColumnCount, segmentProperties.getComplexDimensions().size(), measureCount,	1✔
398	noDictionaryCount, segmentId,
399	carbonLoadModel.getTaskNo(), noDictionaryColMapping, isVarcharDimMapping, true);	1✔
400	}
401
402	/**
403	* create an instance of finalThread merger which will perform merge sort on all the
404	* sort temp files
405	*/
406	private void initializeFinalThreadMergerForMergeSort() {
407	boolean[] noDictionarySortColumnMapping = null;	1✔
408	if (noDictionaryColMapping.length == this.segmentProperties.getNumberOfSortColumns()) {	1✔
409	noDictionarySortColumnMapping = noDictionaryColMapping;	1✔
410	} else {
411	noDictionarySortColumnMapping = new boolean[this.segmentProperties.getNumberOfSortColumns()];	1✔
412	System.arraycopy(noDictionaryColMapping, 0,	1✔
413	noDictionarySortColumnMapping, 0, noDictionarySortColumnMapping.length);
414	}
415	sortParameters.setNoDictionarySortColumn(noDictionarySortColumnMapping);	1✔
416	String[] sortTempFileLocation = CarbonDataProcessorUtil.arrayAppend(tempStoreLocation,	1✔
417	CarbonCommonConstants.FILE_SEPARATOR, CarbonCommonConstants.SORT_TEMP_FILE_LOCATION);
418	finalMerger =	1✔
419	new SingleThreadFinalSortFilesMerger(sortTempFileLocation, tableName, sortParameters);
420	}	1✔
421
422	/**
423	* initialise carbon data writer instance
424	*/
425	private void initDataHandler() throws Exception {
426	String carbonStoreLocation;
427	if (partitionSpec != null) {	1✔
428	carbonStoreLocation =	×
429	partitionSpec.getLocation().toString() + CarbonCommonConstants.FILE_SEPARATOR	×
430	+ carbonLoadModel.getFactTimeStamp() + ".tmp";	×
431	} else {
432	carbonStoreLocation = CarbonDataProcessorUtil	1✔
433	.createCarbonStoreLocation(carbonLoadModel.getDatabaseName(), tableName,	1✔
434	carbonLoadModel.getSegmentId());	1✔
435	}
436	CarbonFactDataHandlerModel carbonFactDataHandlerModel = CarbonFactDataHandlerModel	1✔
437	.getCarbonFactDataHandlerModel(carbonLoadModel, carbonTable, segmentProperties, tableName,	1✔
438	tempStoreLocation, carbonStoreLocation);
439	carbonFactDataHandlerModel.setSegmentId(carbonLoadModel.getSegmentId());	1✔
440	setDataFileAttributesInModel(carbonLoadModel, compactionType, carbonFactDataHandlerModel);	1✔
441	dataHandler = CarbonFactHandlerFactory.createCarbonFactHandler(carbonFactDataHandlerModel);	1✔
442	try {
443	dataHandler.initialise();	1✔
444	} catch (CarbonDataWriterException e) {	×
445	LOGGER.error(e);	×
446	throw new Exception("Problem initialising data handler during compaction: " + e.getMessage());	×
447	}	1✔
448	}	1✔
449
450	/**
451	* initialise temporary store location
452	*/
453	private void initTempStoreLocation() {
454	tempStoreLocation = CarbonDataProcessorUtil	1✔
455	.getLocalDataFolderLocation(carbonTable, carbonLoadModel.getTaskNo(),	1✔
456	segmentId, true, false);
457	}	1✔
458	}

apache / carbondata / 389

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