1179

Build # 1179

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/7458b63281cb948d4664300e55c22f52851dcce8">7458b6328<a href="https://github.com/apache/carbondata/commit/7458b63281cb948d4664300e55c22f52851dcce8">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/682141366b4f9805c92f85b6438a36c223a9f037">682141366</a>">682141366</a><a href="https://github.com/apache/carbondata/commit/7458b63281cb948d4664300e55c22f52851dcce8"> into fa9a4eeeb">fa9a4eeeb</a>

Run Details

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

60558 of 75124 relevant lines covered (80.61%)

1.06 hits per line

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

85.5

/processing/src/main/java/org/apache/carbondata/processing/merger/RowResultMergerProcessor.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.AbstractQueue;
import java.util.Comparator;
import java.util.List;
import java.util.PriorityQueue;

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.datastore.row.WriteStepRowUtil;
import org.apache.carbondata.core.indexstore.PartitionSpec;
import org.apache.carbondata.core.keygenerator.KeyGenException;
import org.apache.carbondata.core.metadata.CarbonMetadata;
import org.apache.carbondata.core.metadata.SegmentFileStore;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.core.scan.result.iterator.RawResultIterator;
import org.apache.carbondata.core.scan.wrappers.ByteArrayWrapper;
import org.apache.carbondata.core.util.ByteUtil;
import org.apache.carbondata.processing.exception.SliceMergerException;
import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
import org.apache.carbondata.processing.store.CarbonFactDataHandlerColumnar;
import org.apache.carbondata.processing.store.CarbonFactDataHandlerModel;
import org.apache.carbondata.processing.store.CarbonFactHandler;
import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;

import org.apache.log4j.Logger;

/**
 * This is the Merger class responsible for the merging of the segments.
 */
public class RowResultMergerProcessor extends AbstractResultProcessor {

  private CarbonFactHandler dataHandler;
  private SegmentProperties segprop;
  private CarbonLoadModel loadModel;
  private PartitionSpec partitionSpec;

  CarbonColumn[] noDicAndComplexColumns;
  /**
   * record holder heap
   */
  private AbstractQueue<RawResultIterator> recordHolderHeap;

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

  public RowResultMergerProcessor(String databaseName,
      String tableName, SegmentProperties segProp, String[] tempStoreLocation,
      CarbonLoadModel loadModel, CompactionType compactionType, PartitionSpec partitionSpec)
      throws IOException {
    this.segprop = segProp;
    this.partitionSpec = partitionSpec;
    this.loadModel = loadModel;
    CarbonDataProcessorUtil.createLocations(tempStoreLocation);

    CarbonTable carbonTable = CarbonMetadata.getInstance().getCarbonTable(databaseName, tableName);
    String carbonStoreLocation;
    if (partitionSpec != null) {
      carbonStoreLocation =
          partitionSpec.getLocation().toString() + CarbonCommonConstants.FILE_SEPARATOR + loadModel
              .getFactTimeStamp() + ".tmp";
    } else {
      carbonStoreLocation = CarbonDataProcessorUtil.createCarbonStoreLocation(
          loadModel.getDatabaseName(), tableName, loadModel.getSegmentId());
    }
    CarbonFactDataHandlerModel carbonFactDataHandlerModel = CarbonFactDataHandlerModel
        .getCarbonFactDataHandlerModel(loadModel, carbonTable, segProp, tableName,
            tempStoreLocation, carbonStoreLocation);
    setDataFileAttributesInModel(loadModel, compactionType, carbonFactDataHandlerModel);
    carbonFactDataHandlerModel.setCompactionFlow(true);
    carbonFactDataHandlerModel.setSegmentId(loadModel.getSegmentId());
    this.noDicAndComplexColumns = carbonFactDataHandlerModel.getNoDictAndComplexColumns();
    dataHandler = new CarbonFactDataHandlerColumnar(carbonFactDataHandlerModel);
  }

  private void initRecordHolderHeap(List<RawResultIterator> rawResultIteratorList) {
    // create the List of RawResultIterator.
    recordHolderHeap = new PriorityQueue<RawResultIterator>(rawResultIteratorList.size(),
        new RowResultMergerProcessor.CarbonMdkeyComparator());
  }

  /**
   * Merge function
   *
   */
  public boolean execute(List<RawResultIterator> resultIteratorList) throws Exception {
    initRecordHolderHeap(resultIteratorList);
    boolean mergeStatus = false;
    int index = 0;
    boolean isDataPresent = false;
    try {

      // add all iterators to the queue
      for (RawResultIterator leaftTupleIterator : resultIteratorList) {
        this.recordHolderHeap.add(leaftTupleIterator);
        index++;
      }
      RawResultIterator iterator = null;
      while (index > 1) {
        // iterator the top record
        iterator = this.recordHolderHeap.poll();
        Object[] convertedRow = iterator.next();
        if (null == convertedRow) {
          index--;
          iterator.close();
          continue;
        }
        if (!isDataPresent) {
          dataHandler.initialise();
          isDataPresent = true;
        }
        // get the mdkey
        addRow(convertedRow);
        // if there is no record in the leaf and all then decrement the
        // index
        if (!iterator.hasNext()) {
          index--;
          iterator.close();
          continue;
        }
        // add record to heap
        this.recordHolderHeap.add(iterator);
      }
      // if record holder is not empty then iterator the slice holder from
      // heap
      iterator = this.recordHolderHeap.poll();
      if (null != iterator) {
        while (true) {
          Object[] convertedRow = iterator.next();
          if (null == convertedRow) {
            iterator.close();
            break;
          }
          // do it only once
          if (!isDataPresent) {
            dataHandler.initialise();
            isDataPresent = true;
          }
          addRow(convertedRow);
          // check if leaf contains no record
          if (!iterator.hasNext()) {
            break;
          }
        }
      }
      if (isDataPresent)
      {
        this.dataHandler.finish();
      }
      mergeStatus = true;
    } catch (Exception e) {
      mergeStatus = false;
      LOGGER.error(e.getLocalizedMessage(), e);
      throw e;
    } finally {
      try {
        if (isDataPresent) {
          this.dataHandler.closeHandler();
        }
        if (partitionSpec != null) {
          SegmentFileStore.writeSegmentFile(loadModel.getTablePath(), loadModel.getTaskNo(),
              partitionSpec.getLocation().toString(), loadModel.getFactTimeStamp() + "",
              partitionSpec.getPartitions());
        }
      } catch (CarbonDataWriterException | IOException e) {
        mergeStatus = false;
        throw e;
      }
    }

    return mergeStatus;
  }

  @Override
  public void close() {
    // close data handler
    if (null != dataHandler) {
      dataHandler.closeHandler();
    }
  }

  /**
   * Below method will be used to add sorted row
   *
   * @throws SliceMergerException
   */
  private void addRow(Object[] carbonTuple) throws SliceMergerException {
    CarbonRow row = WriteStepRowUtil.fromMergerRow(carbonTuple, segprop, noDicAndComplexColumns);
    try {
      this.dataHandler.addDataToStore(row);
    } catch (CarbonDataWriterException e) {
      throw new SliceMergerException("Problem in merging the slice", e);
    }
  }

  /**
   * Comparator class for comparing 2 raw row result.
   */
  private class CarbonMdkeyComparator implements Comparator<RawResultIterator> {
    int[] columnValueSizes = segprop.getEachDimColumnValueSize();
    public CarbonMdkeyComparator() {
      initSortColumns();
    }

    private void initSortColumns() {
      int numberOfSortColumns = segprop.getNumberOfSortColumns();
      if (numberOfSortColumns != columnValueSizes.length) {
        int[] sortColumnValueSizes = new int[numberOfSortColumns];
        System.arraycopy(columnValueSizes, 0, sortColumnValueSizes, 0, numberOfSortColumns);
        this.columnValueSizes = sortColumnValueSizes;
      }
    }

    @Override public int compare(RawResultIterator o1, RawResultIterator o2) {

      Object[] row1 = new Object[0];
      Object[] row2 = new Object[0];
      try {
        row1 = o1.fetchConverted();
        row2 = o2.fetchConverted();
      } catch (KeyGenException e) {
        LOGGER.error(e.getMessage());
      }
      if (null == row1 || null == row2) {
        return 0;
      }
      ByteArrayWrapper key1 = (ByteArrayWrapper) row1[0];
      ByteArrayWrapper key2 = (ByteArrayWrapper) row2[0];
      int compareResult = 0;
      int dictionaryKeyOffset = 0;
      byte[] dimCols1 = key1.getDictionaryKey();
      byte[] dimCols2 = key2.getDictionaryKey();
      int noDicIndex = 0;
      for (int eachColumnValueSize : columnValueSizes) {
        // case of dictionary cols
        if (eachColumnValueSize > 0) {

          compareResult = ByteUtil.UnsafeComparer.INSTANCE
              .compareTo(dimCols1, dictionaryKeyOffset, eachColumnValueSize, dimCols2,
                  dictionaryKeyOffset, eachColumnValueSize);
          dictionaryKeyOffset += eachColumnValueSize;
        } else { // case of no dictionary

          byte[] noDictionaryDim1 = key1.getNoDictionaryKeyByIndex(noDicIndex);
          byte[] noDictionaryDim2 = key2.getNoDictionaryKeyByIndex(noDicIndex);
          compareResult =
              ByteUtil.UnsafeComparer.INSTANCE.compareTo(noDictionaryDim1, noDictionaryDim2);
          noDicIndex++;

        }
        if (0 != compareResult) {
          return compareResult;
        }
      }
      return 0;
    }
  }

}

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.AbstractQueue;
21	import java.util.Comparator;
22	import java.util.List;
23	import java.util.PriorityQueue;
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.SegmentProperties;
28	import org.apache.carbondata.core.datastore.exception.CarbonDataWriterException;
29	import org.apache.carbondata.core.datastore.row.CarbonRow;
30	import org.apache.carbondata.core.datastore.row.WriteStepRowUtil;
31	import org.apache.carbondata.core.indexstore.PartitionSpec;
32	import org.apache.carbondata.core.keygenerator.KeyGenException;
33	import org.apache.carbondata.core.metadata.CarbonMetadata;
34	import org.apache.carbondata.core.metadata.SegmentFileStore;
35	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
36	import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
37	import org.apache.carbondata.core.scan.result.iterator.RawResultIterator;
38	import org.apache.carbondata.core.scan.wrappers.ByteArrayWrapper;
39	import org.apache.carbondata.core.util.ByteUtil;
40	import org.apache.carbondata.processing.exception.SliceMergerException;
41	import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
42	import org.apache.carbondata.processing.store.CarbonFactDataHandlerColumnar;
43	import org.apache.carbondata.processing.store.CarbonFactDataHandlerModel;
44	import org.apache.carbondata.processing.store.CarbonFactHandler;
45	import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
46
47	import org.apache.log4j.Logger;
48
49	/**
50	* This is the Merger class responsible for the merging of the segments.
51	*/
52	public class RowResultMergerProcessor extends AbstractResultProcessor {
53
54	private CarbonFactHandler dataHandler;
55	private SegmentProperties segprop;
56	private CarbonLoadModel loadModel;
57	private PartitionSpec partitionSpec;
58
59	CarbonColumn[] noDicAndComplexColumns;
60	/**
61	* record holder heap
62	*/
63	private AbstractQueue<RawResultIterator> recordHolderHeap;
64
65	private static final Logger LOGGER =	1✔
66	LogServiceFactory.getLogService(RowResultMergerProcessor.class.getName());	1✔
67
68	public RowResultMergerProcessor(String databaseName,
69	String tableName, SegmentProperties segProp, String[] tempStoreLocation,
70	CarbonLoadModel loadModel, CompactionType compactionType, PartitionSpec partitionSpec)
71	throws IOException {	1✔
72	this.segprop = segProp;	1✔
73	this.partitionSpec = partitionSpec;	1✔
74	this.loadModel = loadModel;	1✔
75	CarbonDataProcessorUtil.createLocations(tempStoreLocation);	1✔
76
77	CarbonTable carbonTable = CarbonMetadata.getInstance().getCarbonTable(databaseName, tableName);	1✔
78	String carbonStoreLocation;
79	if (partitionSpec != null) {	1✔
80	carbonStoreLocation =	1✔
81	partitionSpec.getLocation().toString() + CarbonCommonConstants.FILE_SEPARATOR + loadModel	1✔
82	.getFactTimeStamp() + ".tmp";	1✔
83	} else {
84	carbonStoreLocation = CarbonDataProcessorUtil.createCarbonStoreLocation(	1✔
85	loadModel.getDatabaseName(), tableName, loadModel.getSegmentId());	1✔
86	}
87	CarbonFactDataHandlerModel carbonFactDataHandlerModel = CarbonFactDataHandlerModel	1✔
88	.getCarbonFactDataHandlerModel(loadModel, carbonTable, segProp, tableName,	1✔
89	tempStoreLocation, carbonStoreLocation);
90	setDataFileAttributesInModel(loadModel, compactionType, carbonFactDataHandlerModel);	1✔
91	carbonFactDataHandlerModel.setCompactionFlow(true);	1✔
92	carbonFactDataHandlerModel.setSegmentId(loadModel.getSegmentId());	1✔
93	this.noDicAndComplexColumns = carbonFactDataHandlerModel.getNoDictAndComplexColumns();	1✔
94	dataHandler = new CarbonFactDataHandlerColumnar(carbonFactDataHandlerModel);	1✔
95	}	1✔
96
97	private void initRecordHolderHeap(List<RawResultIterator> rawResultIteratorList) {
98	// create the List of RawResultIterator.
99	recordHolderHeap = new PriorityQueue<RawResultIterator>(rawResultIteratorList.size(),	1✔
100	new RowResultMergerProcessor.CarbonMdkeyComparator());
101	}	1✔
102
103	/**
104	* Merge function
105	*
106	*/
107	public boolean execute(List<RawResultIterator> resultIteratorList) throws Exception {
108	initRecordHolderHeap(resultIteratorList);	1✔
109	boolean mergeStatus = false;	1✔
110	int index = 0;	1✔
111	boolean isDataPresent = false;	1✔
112	try {
113
114	// add all iterators to the queue
115	for (RawResultIterator leaftTupleIterator : resultIteratorList) {	1✔
116	this.recordHolderHeap.add(leaftTupleIterator);	1✔
117	index++;	1✔
118	}	1✔
119	RawResultIterator iterator = null;	1✔
120	while (index > 1) {	1✔
121	// iterator the top record
122	iterator = this.recordHolderHeap.poll();	1✔
123	Object[] convertedRow = iterator.next();	1✔
124	if (null == convertedRow) {	1✔
125	index--;	×
126	iterator.close();	×
127	continue;	×
128	}
129	if (!isDataPresent) {	1✔
130	dataHandler.initialise();	1✔
131	isDataPresent = true;	1✔
132	}
133	// get the mdkey
134	addRow(convertedRow);	1✔
135	// if there is no record in the leaf and all then decrement the
136	// index
137	if (!iterator.hasNext()) {	1✔
138	index--;	1✔
139	iterator.close();	1✔
140	continue;	1✔
141	}
142	// add record to heap
143	this.recordHolderHeap.add(iterator);	1✔
144	}	1✔
145	// if record holder is not empty then iterator the slice holder from
146	// heap
147	iterator = this.recordHolderHeap.poll();	1✔
148	if (null != iterator) {	1✔
149	while (true) {
150	Object[] convertedRow = iterator.next();	1✔
151	if (null == convertedRow) {	1✔
152	iterator.close();	×
153	break;	×
154	}
155	// do it only once
156	if (!isDataPresent) {	1✔
157	dataHandler.initialise();	1✔
158	isDataPresent = true;	1✔
159	}
160	addRow(convertedRow);	1✔
161	// check if leaf contains no record
162	if (!iterator.hasNext()) {	1✔
163	break;	1✔
164	}
165	}	1✔
166	}
167	if (isDataPresent)	1✔
168	{
169	this.dataHandler.finish();	1✔
170	}
171	mergeStatus = true;	1✔
172	} catch (Exception e) {	×
173	mergeStatus = false;	×
174	LOGGER.error(e.getLocalizedMessage(), e);	×
175	throw e;	×
176	} finally {
177	try {	×
178	if (isDataPresent) {	1✔
179	this.dataHandler.closeHandler();	1✔
180	}
181	if (partitionSpec != null) {	1✔
182	SegmentFileStore.writeSegmentFile(loadModel.getTablePath(), loadModel.getTaskNo(),	1✔
183	partitionSpec.getLocation().toString(), loadModel.getFactTimeStamp() + "",	1✔
184	partitionSpec.getPartitions());	1✔
185	}
186	} catch (CarbonDataWriterException \| IOException e) {	×
187	mergeStatus = false;	×
188	throw e;	×
189	}	1✔
190	}	×
191
192	return mergeStatus;	1✔
193	}
194
195	@Override
196	public void close() {
197	// close data handler
198	if (null != dataHandler) {	1✔
199	dataHandler.closeHandler();	1✔
200	}
201	}	1✔
202
203	/**
204	* Below method will be used to add sorted row
205	*
206	* @throws SliceMergerException
207	*/
208	private void addRow(Object[] carbonTuple) throws SliceMergerException {
209	CarbonRow row = WriteStepRowUtil.fromMergerRow(carbonTuple, segprop, noDicAndComplexColumns);	1✔
210	try {
211	this.dataHandler.addDataToStore(row);	1✔
212	} catch (CarbonDataWriterException e) {	×
213	throw new SliceMergerException("Problem in merging the slice", e);	×
214	}	1✔
215	}	1✔
216
217	/**
218	* Comparator class for comparing 2 raw row result.
219	*/
220	private class CarbonMdkeyComparator implements Comparator<RawResultIterator> {
221	int[] columnValueSizes = segprop.getEachDimColumnValueSize();	1✔
222	public CarbonMdkeyComparator() {	1✔
223	initSortColumns();	1✔
224	}	1✔
225
226	private void initSortColumns() {
227	int numberOfSortColumns = segprop.getNumberOfSortColumns();	1✔
228	if (numberOfSortColumns != columnValueSizes.length) {	1✔
229	int[] sortColumnValueSizes = new int[numberOfSortColumns];	1✔
230	System.arraycopy(columnValueSizes, 0, sortColumnValueSizes, 0, numberOfSortColumns);	1✔
231	this.columnValueSizes = sortColumnValueSizes;	1✔
232	}
233	}	1✔
234
235	@Override public int compare(RawResultIterator o1, RawResultIterator o2) {
236
237	Object[] row1 = new Object[0];	1✔
238	Object[] row2 = new Object[0];	1✔
239	try {
240	row1 = o1.fetchConverted();	1✔
241	row2 = o2.fetchConverted();	1✔
242	} catch (KeyGenException e) {	×
243	LOGGER.error(e.getMessage());	×
244	}	1✔
245	if (null == row1 \|\| null == row2) {	1✔
246	return 0;	×
247	}
248	ByteArrayWrapper key1 = (ByteArrayWrapper) row1[0];	1✔
249	ByteArrayWrapper key2 = (ByteArrayWrapper) row2[0];	1✔
250	int compareResult = 0;	1✔
251	int dictionaryKeyOffset = 0;	1✔
252	byte[] dimCols1 = key1.getDictionaryKey();	1✔
253	byte[] dimCols2 = key2.getDictionaryKey();	1✔
254	int noDicIndex = 0;	1✔
255	for (int eachColumnValueSize : columnValueSizes) {	1✔
256	// case of dictionary cols
257	if (eachColumnValueSize > 0) {	1✔
258
259	compareResult = ByteUtil.UnsafeComparer.INSTANCE	1✔
260	.compareTo(dimCols1, dictionaryKeyOffset, eachColumnValueSize, dimCols2,	1✔
261	dictionaryKeyOffset, eachColumnValueSize);
262	dictionaryKeyOffset += eachColumnValueSize;	1✔
263	} else { // case of no dictionary
264
265	byte[] noDictionaryDim1 = key1.getNoDictionaryKeyByIndex(noDicIndex);	1✔
266	byte[] noDictionaryDim2 = key2.getNoDictionaryKeyByIndex(noDicIndex);	1✔
267	compareResult =	1✔
268	ByteUtil.UnsafeComparer.INSTANCE.compareTo(noDictionaryDim1, noDictionaryDim2);	1✔
269	noDicIndex++;	1✔
270
271	}
272	if (0 != compareResult) {	1✔
273	return compareResult;	1✔
274	}
275	}
276	return 0;	1✔
277	}
278	}
279
280	}

apache / carbondata / 1179

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