1460

Build # 1460

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/5791b3dc0e733286e03ebb97f699df41791168c4">5791b3dc0<a href="https://github.com/apache/carbondata/commit/5791b3dc0e733286e03ebb97f699df41791168c4">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/1f090ced37743f18f9d4d7cbcbdc6d9575b92df2">1f090ced3</a>">1f090ced3</a><a href="https://github.com/apache/carbondata/commit/5791b3dc0e733286e03ebb97f699df41791168c4"> into f51f5cde1">f51f5cde1</a>

Run Details

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

61810 of 77202 relevant lines covered (80.06%)

1.06 hits per line

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

89.8

/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.util.CarbonTaskInfo;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.ThreadLocalTaskInfo;

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;

  /**
   * 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);
    CarbonTaskInfo carbonTaskInfo = new CarbonTaskInfo();
    carbonTaskInfo.setTaskId(CarbonUtil.generateUUID());
    ThreadLocalTaskInfo.setCarbonTaskInfo(carbonTaskInfo);
  }

  /**
   * 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();
  }

  /**
   * 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();
    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.util.CarbonTaskInfo;
28	import org.apache.carbondata.core.util.CarbonUtil;
29	import org.apache.carbondata.core.util.ThreadLocalTaskInfo;
30
31	import org.apache.hadoop.mapreduce.RecordReader;
32
33
34	/**
35	* Reader for carbondata file
36	*/
37	@InterfaceAudience.User
38	@InterfaceStability.Evolving
39	public class CarbonReader<T> {
40
41	private List<RecordReader<Void, T>> readers;
42
43	private RecordReader<Void, T> currentReader;
44
45	private int index;
46
47	private boolean initialise;
48
49	/**
50	* Call {@link #builder(String)} to construct an instance
51	*/
52	CarbonReader(List<RecordReader<Void, T>> readers) {	2✔
53	if (readers.size() == 0) {	2✔
54	throw new IllegalArgumentException("no reader");	×
55	}
56	this.initialise = true;	2✔
57	this.readers = readers;	2✔
58	this.index = 0;	2✔
59	this.currentReader = readers.get(0);	2✔
60	CarbonTaskInfo carbonTaskInfo = new CarbonTaskInfo();	2✔
61	carbonTaskInfo.setTaskId(CarbonUtil.generateUUID());	2✔
62	ThreadLocalTaskInfo.setCarbonTaskInfo(carbonTaskInfo);	2✔
63	}	2✔
64
65	/**
66	* Return true if has next row
67	*/
68	public boolean hasNext() throws IOException, InterruptedException {
69	validateReader();	2✔
70	if (currentReader.nextKeyValue()) {	2✔
71	return true;	2✔
72	} else {
73	if (index == readers.size() - 1) {	2✔
74	// no more readers
75	return false;	2✔
76	} else {
77	index++;	2✔
78	// current reader is closed
79	currentReader.close();	2✔
80	currentReader = readers.get(index);	2✔
81	return currentReader.nextKeyValue();	2✔
82	}
83	}
84	}
85
86	/**
87	* Read and return next row object
88	*/
89	public T readNextRow() throws IOException, InterruptedException {
90	validateReader();	2✔
91	return currentReader.getCurrentValue();	2✔
92	}
93
94	/**
95	* Return a new {@link CarbonReaderBuilder} instance
96	*
97	* @param tablePath table store path
98	* @param tableName table name
99	* @return CarbonReaderBuilder object
100	*/
101	public static CarbonReaderBuilder builder(String tablePath, String tableName) {
102	return new CarbonReaderBuilder(tablePath, tableName);	2✔
103	}
104
105	/**
106	* Return a new {@link CarbonReaderBuilder} instance
107	* Default value of table name is table + tablePath + time
108	*
109	* @param tablePath table path
110	* @return CarbonReaderBuilder object
111	*/
112	public static CarbonReaderBuilder builder(String tablePath) {
113	UUID uuid = UUID.randomUUID();	2✔
114	String tableName = "UnknownTable" + uuid;	2✔
115	return builder(tablePath, tableName);	2✔
116	}
117
118	/**
119	* Breaks the list of CarbonRecordReader in CarbonReader into multiple
120	* CarbonReader objects, each iterating through some 'carbondata' files
121	* and return that list of CarbonReader objects
122	*
123	* If the no. of files is greater than maxSplits, then break the
124	* CarbonReader into maxSplits splits, with each split iterating
125	* through >= 1 file.
126	*
127	* If the no. of files is less than maxSplits, then return list of
128	* CarbonReader with size as the no. of files, with each CarbonReader
129	* iterating through exactly one file
130	*
131	* @param maxSplits: Int
132	* @return list of {@link CarbonReader} objects
133	*/
134	public List<CarbonReader> split(int maxSplits) throws IOException {
135	validateReader();	2✔
136	if (maxSplits < 1) {	2✔
137	throw new RuntimeException(	×
138	this.getClass().getSimpleName() + ".split: maxSplits must be positive");	×
139	}
140
141	List<CarbonReader> carbonReaders = new ArrayList<>();	2✔
142
143	if (maxSplits < this.readers.size()) {	2✔
144	// If maxSplits is less than the no. of files
145	// Split the reader into maxSplits splits with each
146	// element containing >= 1 CarbonRecordReader objects
147	float filesPerSplit = (float) this.readers.size() / maxSplits;	2✔
148	for (int i = 0; i < maxSplits; ++i) {	2✔
149	carbonReaders.add(new CarbonReader<>(this.readers.subList(	2✔
150	(int) Math.ceil(i * filesPerSplit),	2✔
151	(int) Math.ceil(((i + 1) * filesPerSplit)))));	2✔
152	}
153	} else {	2✔
154	// If maxSplits is greater than the no. of files
155	// Split the reader into <num_files> splits with each
156	// element contains exactly 1 CarbonRecordReader object
157	for (int i = 0; i < this.readers.size(); ++i) {	×
158	carbonReaders.add(new CarbonReader<>(this.readers.subList(i, i + 1)));	×
159	}
160	}
161
162	// This is to disable the use of this CarbonReader object to iterate
163	// over the files and forces user to only use the returned splits
164	this.initialise = false;	2✔
165
166	return carbonReaders;	2✔
167	}
168
169	/**
170	* Close reader
171	*
172	* @throws IOException
173	*/
174	public void close() throws IOException {
175	validateReader();	2✔
176	this.currentReader.close();	2✔
177	this.initialise = false;	2✔
178	}	2✔
179
180	/**
181	* Validate the reader
182	*/
183	private void validateReader() {
184	if (!this.initialise) {	2✔
185	throw new RuntimeException(this.getClass().getSimpleName() +	2✔
186	" not initialise, please create it first.");
187	}
188	}	2✔
189	}

apache / carbondata / 1460

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