1474

Build # 1474

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/02665da6f239489232f0b7ca81369566bd3845a9">02665da6f<a href="https://github.com/apache/carbondata/commit/02665da6f239489232f0b7ca81369566bd3845a9">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/bf91dfaadd6f6d607176a146466ad4d3bff7e9cb">bf91dfaad</a>">bf91dfaad</a><a href="https://github.com/apache/carbondata/commit/02665da6f239489232f0b7ca81369566bd3845a9"> into 5a0bc6e71">5a0bc6e71</a>

Run Details

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

61877 of 77276 relevant lines covered (80.07%)

1.06 hits per line

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

85.61

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

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

apache / carbondata / 1474

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