11755

Build # 11755

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/d9cab0fc6f0237c130e7821e8a508e3b8b35a220">d9cab0fc6<a href="https://github.com/apache/carbondata/commit/d9cab0fc6f0237c130e7821e8a508e3b8b35a220">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/70a960ec5cddf6b0a6d6dc0b759fec9b13759fb1">70a960ec5</a>">70a960ec5</a><a href="https://github.com/apache/carbondata/commit/d9cab0fc6f0237c130e7821e8a508e3b8b35a220"> into 1baa485d8">1baa485d8</a>

Run Details

22 of 22 new or added lines in 3 files covered. (100.0%)

63997 of 82745 relevant lines covered (77.34%)

1.01 hits per line

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

91.67

/store/sdk/src/main/java/org/apache/carbondata/sdk/file/CarbonReader.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.sdk.file;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.annotations.InterfaceStability;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.util.CarbonProperties;
import org.apache.carbondata.hadoop.CarbonRecordReader;
import org.apache.carbondata.hadoop.util.CarbonVectorizedRecordReader;

import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;


/**
 * Reader for CarbonData file
 */
@InterfaceAudience.User
@InterfaceStability.Evolving
public class CarbonReader<T> {

  private List<RecordReader<Void, T>> readers;

  private RecordReader<Void, T> currentReader;

  private int index;

  private boolean initialise;

  /**
   * save batch rows data
   */
  private Object[] batchRows;

  /**
   * Call {@link #builder(String)} to construct an instance
   */
  CarbonReader(List<RecordReader<Void, T>> readers) {
    this.initialise = true;
    this.readers = readers;
    this.index = 0;
    if (0 == readers.size()) {
      this.currentReader = null;
    } else {
      this.currentReader = readers.get(0);
    }
  }

  /**
   * Return true if has next row
   */
  public boolean hasNext() throws IOException, InterruptedException {
    if (0 == readers.size()) {
      return false;
    }
    validateReader();
    if (currentReader.nextKeyValue()) {
      return true;
    } else {
      for (int i = index; i < readers.size(); i++) {
        if (index == readers.size() - 1) {
          // no more readers
          return false;
        } else {
          // current reader is closed
          currentReader.close();
          // no need to keep a reference to CarbonVectorizedRecordReader,
          // until all the readers are processed.
          // If readers count is very high,
          // we get OOM as GC not happened for any of the content in CarbonVectorizedRecordReader
          readers.set(index, null);
          index++;
          currentReader = readers.get(index);
          boolean result = currentReader.nextKeyValue();
          if (result) {
            return result;
          }
        }
      }
      return false;
    }
  }

  /**
   * Read and return next row object
   */
  public T readNextRow() throws IOException, InterruptedException {
    validateReader();
    return currentReader.getCurrentValue();
  }

  /**
   * Read and return next batch row objects
   */
  public Object[] readNextBatchRow() throws Exception {
    validateReader();
    if (currentReader instanceof CarbonRecordReader) {
      List<Object> batchValue = ((CarbonRecordReader) currentReader).getBatchValue();
      if (batchValue == null) {
        return null;
      } else {
        return batchValue.toArray();
      }
    } else if (currentReader instanceof CarbonVectorizedRecordReader) {
      int batch = Integer.parseInt(CarbonProperties.getInstance()
          .getProperty(CarbonCommonConstants.DETAIL_QUERY_BATCH_SIZE,
              String.valueOf(CarbonCommonConstants.DETAIL_QUERY_BATCH_SIZE_DEFAULT)));
      batchRows = new Object[batch];
      int sum = 0;
      for (int i = 0; i < batch; i++) {
        batchRows[i] = currentReader.getCurrentValue();
        sum++;
        if (i != batch - 1) {
          if (!hasNext()) {
            Object[] lessBatch = new Object[sum];
            for (int j = 0; j < sum; j++) {
              lessBatch[j] = batchRows[j];
            }
            return lessBatch;
          }
        }
      }
      return batchRows;
    } else {
      throw new Exception("Didn't support read next batch row by this reader.");
    }
  }

  /**
   * Return a new {@link CarbonReaderBuilder} instance
   *
   * @param tablePath table store path
   * @param tableName table name
   * @return CarbonReaderBuilder object
   */
  public static CarbonReaderBuilder builder(String tablePath, String tableName) {
    return new CarbonReaderBuilder(tablePath, tableName);
  }

  /**
   * Return a new {@link CarbonReaderBuilder} instance
   *
   * @param inputSplit CarbonInputSplit Object
   * @return CarbonReaderBuilder object
   */
  public static CarbonReaderBuilder builder(InputSplit inputSplit) {
    return new CarbonReaderBuilder(inputSplit);
  }

  /**
   * Return a new {@link CarbonReaderBuilder} instance
   *
   * @param tablePath table path
   * @return CarbonReaderBuilder object
   */
  public static CarbonReaderBuilder builder(String tablePath) {
    UUID uuid = UUID.randomUUID();
    String tableName = "UnknownTable" + uuid;
    return builder(tablePath, tableName);
  }

  /**
   * Return a new {@link CarbonReaderBuilder} instance
   *
   * @return CarbonReaderBuilder object
   */
  public static CarbonReaderBuilder builder() {
    UUID uuid = UUID.randomUUID();
    String tableName = "UnknownTable" + uuid;
    return new CarbonReaderBuilder(tableName);
  }

  /**
   * Breaks the list of CarbonRecordReader in CarbonReader into multiple
   * CarbonReader objects, each iterating through some 'carbondata' files
   * and return that list of CarbonReader objects
   * <p>
   * If the no. of files is greater than maxSplits, then break the
   * CarbonReader into maxSplits splits, with each split iterating
   * through >= 1 file.
   * <p>
   * If the no. of files is less than maxSplits, then return list of
   * CarbonReader with size as the no. of files, with each CarbonReader
   * iterating through exactly one file
   *
   * @param maxSplits: Int
   * @return list of {@link CarbonReader} objects
   */
  public List<CarbonReader> split(int maxSplits) throws IOException {
    validateReader();
    if (maxSplits < 1) {
      throw new RuntimeException(
          this.getClass().getSimpleName() + ".split: maxSplits must be positive");
    }

    List<CarbonReader> carbonReaders = new ArrayList<>();

    if (maxSplits < this.readers.size()) {
      // If maxSplits is less than the no. of files
      // Split the reader into maxSplits splits with each
      // element containing >= 1 CarbonRecordReader objects
      float filesPerSplit = (float) this.readers.size() / maxSplits;
      for (int i = 0; i < maxSplits; ++i) {
        carbonReaders.add(new CarbonReader<>(this.readers.subList(
            (int) Math.ceil(i * filesPerSplit),
            (int) Math.ceil(((i + 1) * filesPerSplit)))));
      }
    } else {
      // If maxSplits is greater than the no. of files
      // Split the reader into <num_files> splits with each
      // element contains exactly 1 CarbonRecordReader object
      for (int i = 0; i < this.readers.size(); ++i) {
        carbonReaders.add(new CarbonReader<>(this.readers.subList(i, i + 1)));
      }
    }

    // This is to disable the use of this CarbonReader object to iterate
    // over the files and forces user to only use the returned splits
    this.initialise = false;

    return carbonReaders;
  }

  /**
   * Close reader
   *
   * @throws IOException
   */
  public void close() throws IOException {
    validateReader();
    CarbonProperties.getInstance()
        .addProperty(CarbonCommonConstants.DETAIL_QUERY_BATCH_SIZE,
            String.valueOf(CarbonCommonConstants.DETAIL_QUERY_BATCH_SIZE_DEFAULT));
    if (null != this.currentReader) {
      this.currentReader.close();
    }
    this.initialise = false;
  }

  /**
   * Validate the reader
   */
  private void validateReader() {
    if (!this.initialise) {
      throw new RuntimeException(this.getClass().getSimpleName() +
          " not initialise, please create it first.");
    }
  }
}

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
18	package org.apache.carbondata.sdk.file;
19
20	import java.io.IOException;
21	import java.util.ArrayList;
22	import java.util.List;
23	import java.util.UUID;
24
25	import org.apache.carbondata.common.annotations.InterfaceAudience;
26	import org.apache.carbondata.common.annotations.InterfaceStability;
27	import org.apache.carbondata.core.constants.CarbonCommonConstants;
28	import org.apache.carbondata.core.util.CarbonProperties;
29	import org.apache.carbondata.hadoop.CarbonRecordReader;
30	import org.apache.carbondata.hadoop.util.CarbonVectorizedRecordReader;
31
32	import org.apache.hadoop.mapreduce.InputSplit;
33	import org.apache.hadoop.mapreduce.RecordReader;
34
35
36	/**
37	* Reader for CarbonData file
38	*/
39	@InterfaceAudience.User
40	@InterfaceStability.Evolving
41	public class CarbonReader<T> {
42
43	private List<RecordReader<Void, T>> readers;
44
45	private RecordReader<Void, T> currentReader;
46
47	private int index;
48
49	private boolean initialise;
50
51	/**
52	* save batch rows data
53	*/
54	private Object[] batchRows;
55
56	/**
57	* Call {@link #builder(String)} to construct an instance
58	*/
59	CarbonReader(List<RecordReader<Void, T>> readers) {	2✔
60	this.initialise = true;	2✔
61	this.readers = readers;	2✔
62	this.index = 0;	2✔
63	if (0 == readers.size()) {	2✔
64	this.currentReader = null;	2✔
65	} else {
66	this.currentReader = readers.get(0);	2✔
67	}
68	}	2✔
69
70	/**
71	* Return true if has next row
72	*/
73	public boolean hasNext() throws IOException, InterruptedException {
74	if (0 == readers.size()) {	2✔
75	return false;	2✔
76	}
77	validateReader();	2✔
78	if (currentReader.nextKeyValue()) {	2✔
79	return true;	2✔
80	} else {
81	for (int i = index; i < readers.size(); i++) {	2✔
82	if (index == readers.size() - 1) {	2✔
83	// no more readers
84	return false;	2✔
85	} else {
86	// current reader is closed
87	currentReader.close();	2✔
88	// no need to keep a reference to CarbonVectorizedRecordReader,
89	// until all the readers are processed.
90	// If readers count is very high,
91	// we get OOM as GC not happened for any of the content in CarbonVectorizedRecordReader
92	readers.set(index, null);	2✔
93	index++;	2✔
94	currentReader = readers.get(index);	2✔
95	boolean result = currentReader.nextKeyValue();	2✔
96	if (result) {	2✔
97	return result;	2✔
98	}
99	}
100	}
101	return false;	×
102	}
103	}
104
105	/**
106	* Read and return next row object
107	*/
108	public T readNextRow() throws IOException, InterruptedException {
109	validateReader();	2✔
110	return currentReader.getCurrentValue();	2✔
111	}
112
113	/**
114	* Read and return next batch row objects
115	*/
116	public Object[] readNextBatchRow() throws Exception {
117	validateReader();	2✔
118	if (currentReader instanceof CarbonRecordReader) {	2✔
119	List<Object> batchValue = ((CarbonRecordReader) currentReader).getBatchValue();	2✔
120	if (batchValue == null) {	2✔
121	return null;	×
122	} else {
123	return batchValue.toArray();	2✔
124	}
125	} else if (currentReader instanceof CarbonVectorizedRecordReader) {	2✔
126	int batch = Integer.parseInt(CarbonProperties.getInstance()	2✔
127	.getProperty(CarbonCommonConstants.DETAIL_QUERY_BATCH_SIZE,	2✔
128	String.valueOf(CarbonCommonConstants.DETAIL_QUERY_BATCH_SIZE_DEFAULT)));	2✔
129	batchRows = new Object[batch];	2✔
130	int sum = 0;	2✔
131	for (int i = 0; i < batch; i++) {	2✔
132	batchRows[i] = currentReader.getCurrentValue();	2✔
133	sum++;	2✔
134	if (i != batch - 1) {	2✔
135	if (!hasNext()) {	2✔
136	Object[] lessBatch = new Object[sum];	2✔
137	for (int j = 0; j < sum; j++) {	2✔
138	lessBatch[j] = batchRows[j];	2✔
139	}
140	return lessBatch;	2✔
141	}
142	}
143	}
144	return batchRows;	2✔
145	} else {
146	throw new Exception("Didn't support read next batch row by this reader.");	×
147	}
148	}
149
150	/**
151	* Return a new {@link CarbonReaderBuilder} instance
152	*
153	* @param tablePath table store path
154	* @param tableName table name
155	* @return CarbonReaderBuilder object
156	*/
157	public static CarbonReaderBuilder builder(String tablePath, String tableName) {
158	return new CarbonReaderBuilder(tablePath, tableName);	2✔
159	}
160
161	/**
162	* Return a new {@link CarbonReaderBuilder} instance
163	*
164	* @param inputSplit CarbonInputSplit Object
165	* @return CarbonReaderBuilder object
166	*/
167	public static CarbonReaderBuilder builder(InputSplit inputSplit) {
168	return new CarbonReaderBuilder(inputSplit);	2✔
169	}
170
171	/**
172	* Return a new {@link CarbonReaderBuilder} instance
173	*
174	* @param tablePath table path
175	* @return CarbonReaderBuilder object
176	*/
177	public static CarbonReaderBuilder builder(String tablePath) {
178	UUID uuid = UUID.randomUUID();	2✔
179	String tableName = "UnknownTable" + uuid;	2✔
180	return builder(tablePath, tableName);	2✔
181	}
182
183	/**
184	* Return a new {@link CarbonReaderBuilder} instance
185	*
186	* @return CarbonReaderBuilder object
187	*/
188	public static CarbonReaderBuilder builder() {
189	UUID uuid = UUID.randomUUID();	2✔
190	String tableName = "UnknownTable" + uuid;	2✔
191	return new CarbonReaderBuilder(tableName);	2✔
192	}
193
194	/**
195	* Breaks the list of CarbonRecordReader in CarbonReader into multiple
196	* CarbonReader objects, each iterating through some 'carbondata' files
197	* and return that list of CarbonReader objects
198	* <p>
199	* If the no. of files is greater than maxSplits, then break the
200	* CarbonReader into maxSplits splits, with each split iterating
201	* through >= 1 file.
202	* <p>
203	* If the no. of files is less than maxSplits, then return list of
204	* CarbonReader with size as the no. of files, with each CarbonReader
205	* iterating through exactly one file
206	*
207	* @param maxSplits: Int
208	* @return list of {@link CarbonReader} objects
209	*/
210	public List<CarbonReader> split(int maxSplits) throws IOException {
211	validateReader();	2✔
212	if (maxSplits < 1) {	2✔
213	throw new RuntimeException(	×
214	this.getClass().getSimpleName() + ".split: maxSplits must be positive");	×
215	}
216
217	List<CarbonReader> carbonReaders = new ArrayList<>();	2✔
218
219	if (maxSplits < this.readers.size()) {	2✔
220	// If maxSplits is less than the no. of files
221	// Split the reader into maxSplits splits with each
222	// element containing >= 1 CarbonRecordReader objects
223	float filesPerSplit = (float) this.readers.size() / maxSplits;	2✔
224	for (int i = 0; i < maxSplits; ++i) {	2✔
225	carbonReaders.add(new CarbonReader<>(this.readers.subList(	2✔
226	(int) Math.ceil(i * filesPerSplit),	2✔
227	(int) Math.ceil(((i + 1) * filesPerSplit)))));	2✔
228	}
229	} else {	2✔
230	// If maxSplits is greater than the no. of files
231	// Split the reader into <num_files> splits with each
232	// element contains exactly 1 CarbonRecordReader object
233	for (int i = 0; i < this.readers.size(); ++i) {	×
234	carbonReaders.add(new CarbonReader<>(this.readers.subList(i, i + 1)));	×
235	}
236	}
237
238	// This is to disable the use of this CarbonReader object to iterate
239	// over the files and forces user to only use the returned splits
240	this.initialise = false;	2✔
241
242	return carbonReaders;	2✔
243	}
244
245	/**
246	* Close reader
247	*
248	* @throws IOException
249	*/
250	public void close() throws IOException {
251	validateReader();	2✔
252	CarbonProperties.getInstance()	2✔
253	.addProperty(CarbonCommonConstants.DETAIL_QUERY_BATCH_SIZE,	2✔
254	String.valueOf(CarbonCommonConstants.DETAIL_QUERY_BATCH_SIZE_DEFAULT));	2✔
255	if (null != this.currentReader) {	2✔
256	this.currentReader.close();	2✔
257	}
258	this.initialise = false;	2✔
259	}	2✔
260
261	/**
262	* Validate the reader
263	*/
264	private void validateReader() {
265	if (!this.initialise) {	2✔
266	throw new RuntimeException(this.getClass().getSimpleName() +	2✔
267	" not initialise, please create it first.");
268	}
269	}	2✔
270	}

apache / carbondata / 11755

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