9776

Build # 9776

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/44af05b343bd63bdd6665e2f14f2075bc03bd01d">44af05b34<a href="https://github.com/apache/carbondata/commit/44af05b343bd63bdd6665e2f14f2075bc03bd01d">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/cce1811fa518545b6b31246efd5668048e7d24ea">cce1811fa</a>">cce1811fa</a><a href="https://github.com/apache/carbondata/commit/44af05b343bd63bdd6665e2f14f2075bc03bd01d"> into c5de10cb1">c5de10cb1</a>

Run Details

61525 of 77284 relevant lines covered (79.61%)

1.06 hits per line

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

90.28

/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.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) {
    if (readers.size() == 0) {
      throw new IllegalArgumentException("no reader");
    }
    this.initialise = true;
    this.readers = readers;
    this.index = 0;
    this.currentReader = readers.get(0);
  }

  /**
   * Return true if has next row
   */
  public boolean hasNext() throws IOException, InterruptedException {
    validateReader();
    if (currentReader.nextKeyValue()) {
      return true;
    } else {
      if (index == readers.size() - 1) {
        // no more readers
        return false;
      } else {
        index++;
        // current reader is closed
        currentReader.close();
        currentReader = readers.get(index);
        return currentReader.nextKeyValue();
      }
    }
  }

  /**
   * 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
   * Default value of table name is table + tablePath + time
   *
   * @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);
  }

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

apache / carbondata / 9776

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