4024

Build # 4024

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/5012ccccb946998aa3a78c6e54c386f0334f7436">5012ccccb<a href="https://github.com/apache/carbondata/commit/5012ccccb946998aa3a78c6e54c386f0334f7436">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/11df84b210a88a3c078db218a10152752a5a8e97">11df84b21</a>">11df84b21</a><a href="https://github.com/apache/carbondata/commit/5012ccccb946998aa3a78c6e54c386f0334f7436"> into ebe78dca1">ebe78dca1</a>

Run Details

19 of 19 new or added lines in 1 file covered. (100.0%)

65010 of 83686 relevant lines covered (77.68%)

1.02 hits per line

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

85.93

/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.ArrayList;
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> unsortedResultIteratorList,
      List<RawResultIterator> sortedResultIteratorList) throws Exception {
    List<RawResultIterator> finalIteratorList = new ArrayList<>(unsortedResultIteratorList);
    finalIteratorList.addAll(sortedResultIteratorList);

    initRecordHolderHeap(finalIteratorList);
    boolean mergeStatus = false;
    int index = 0;
    boolean isDataPresent = false;
    try {

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

apache / carbondata / 4024

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