apache / carbondata / 418

Build # 418

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/02115724770629e06ba4cd31108b27b61f85dcf1">021157247<a href="https://github.com/apache/carbondata/commit/02115724770629e06ba4cd31108b27b61f85dcf1">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/6058eafb48acc1a58a164b6c430f0b31abd1d576">6058eafb4</a>">6058eafb4</a><a href="https://github.com/apache/carbondata/commit/02115724770629e06ba4cd31108b27b61f85dcf1"> into 8f1a029b9">8f1a029b9</a>

Run Details

633 of 633 new or added lines in 65 files covered. (100.0%)

58569 of 72573 relevant lines covered (80.7%)

1.07 hits per line

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

90.27

/streaming/src/main/java/org/apache/carbondata/streaming/StreamBlockletReader.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.streaming;

import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;

import org.apache.carbondata.core.datastore.compression.Compressor;
import org.apache.carbondata.core.datastore.compression.CompressorFactory;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.format.BlockletHeader;

/**
 * stream blocklet reader
 */
public class StreamBlockletReader {

  private byte[] buffer;
  private int offset;
  private final byte[] syncMarker;
  private final byte[] syncBuffer;
  private final int syncLen;
  private long pos = 0;
  private final InputStream in;
  private final long limitStart;
  private final long limitEnd;
  private boolean isAlreadySync = false;
  private Compressor compressor;
  private int rowNums = 0;
  private int rowIndex = 0;
  private boolean isHeaderPresent;

  public StreamBlockletReader(byte[] syncMarker, InputStream in, long limit,
      boolean isHeaderPresent, String compressorName) {
    this.syncMarker = syncMarker;
    syncLen = syncMarker.length;
    syncBuffer = new byte[syncLen];
    this.in = in;
    limitStart = limit;
    limitEnd = limitStart + syncLen;
    this.isHeaderPresent = isHeaderPresent;
    this.compressor = CompressorFactory.getInstance().getCompressor(compressorName);
  }

  private void ensureCapacity(int capacity) {
    if (buffer == null || capacity > buffer.length) {
      buffer = new byte[capacity];
    }
  }

  /**
   * find the first position of sync_marker in input stream
   */
  private boolean sync() throws IOException {
    if (!readBytesFromStream(syncBuffer, 0, syncLen)) {
      return false;
    }
    boolean skipHeader = false;
    for (int i = 0; i < limitStart; i++) {
      int j = 0;
      for (; j < syncLen; j++) {
        if (syncMarker[j] != syncBuffer[(i + j) % syncLen]) break;
      }
      if (syncLen == j) {
        if (isHeaderPresent) {
          if (skipHeader) {
            return true;
          } else {
            skipHeader = true;
          }
        } else {
          return true;
        }
      }
      int value = in.read();
      if (-1 == value) {
        return false;
      }
      syncBuffer[i % syncLen] = (byte) value;
      pos++;
    }
    return false;
  }

  public BlockletHeader readBlockletHeader() throws IOException {
    int len = readIntFromStream();
    byte[] b = new byte[len];
    if (!readBytesFromStream(b, 0, len)) {
      throw new EOFException("Failed to read blocklet header");
    }
    BlockletHeader header = CarbonUtil.readBlockletHeader(b);
    rowNums = header.getBlocklet_info().getNum_rows();
    rowIndex = 0;
    return header;
  }

  public void readBlockletData(BlockletHeader header) throws IOException {
    ensureCapacity(header.getBlocklet_length());
    offset = 0;
    int len = readIntFromStream();
    byte[] b = new byte[len];
    if (!readBytesFromStream(b, 0, len)) {
      throw new EOFException("Failed to read blocklet data");
    }
    compressor.rawUncompress(b, buffer);
  }

  public void skipBlockletData(boolean reset) throws IOException {
    int len = readIntFromStream();
    skip(len);
    pos += len;
    if (reset) {
      this.rowNums = 0;
      this.rowIndex = 0;
    }
  }

  private void skip(int len) throws IOException {
    long remaining = len;
    do {
      long skipLen = in.skip(remaining);
      remaining -= skipLen;
    } while (remaining > 0);
  }

  /**
   * find the next blocklet
   */
  public boolean nextBlocklet() throws IOException {
    if (pos >= limitStart) {
      return false;
    }
    if (isAlreadySync) {
      if (!readBytesFromStream(syncBuffer, 0, syncLen)) {
        return false;
      }
    } else {
      isAlreadySync = true;
      if (!sync()) {
        return false;
      }
    }

    return pos < limitEnd;
  }

  public boolean hasNext() throws IOException {
    return rowIndex < rowNums;
  }

  public void nextRow() {
    rowIndex++;
  }

  public int readIntFromStream() throws IOException {
    int ch1 = in.read();
    int ch2 = in.read();
    int ch3 = in.read();
    int ch4 = in.read();
    if ((ch1 | ch2 | ch3 | ch4) < 0) throw new EOFException();
    pos += 4;
    return ((ch1 << 24) + (ch2 << 16) + (ch3 << 8) + (ch4 << 0));
  }

  /**
   * Reads <code>len</code> bytes of data from the input stream into
   * an array of bytes.
   * @return <code>true</code> if reading data successfully, or
   * <code>false</code> if there is no more data because the end of the stream has been reached.
   */
  public boolean readBytesFromStream(byte[] b, int offset, int len) throws IOException {
    int readLen = in.read(b, offset, len);
    if (readLen < 0) {
      return false;
    }
    pos += readLen;
    if (readLen < len) {
      return readBytesFromStream(b, offset + readLen, len - readLen);
    } else {
      return true;
    }
  }

  public boolean readBoolean() {
    return (buffer[offset++]) != 0;
  }

  public short readShort() {
    short v =  (short) ((buffer[offset + 1] & 255) +
        ((buffer[offset]) << 8));
    offset += 2;
    return v;
  }

  public byte[] copy(int len) {
    byte[] b = new byte[len];
    System.arraycopy(buffer, offset, b, 0, len);
    return b;
  }

  public int readInt() {
    int v = ((buffer[offset + 3] & 255) +
        ((buffer[offset + 2] & 255) << 8) +
        ((buffer[offset + 1] & 255) << 16) +
        ((buffer[offset]) << 24));
    offset += 4;
    return v;
  }

  public long readLong() {
    long v = ((long)(buffer[offset + 7] & 255)) +
        ((long) (buffer[offset + 6] & 255) << 8) +
        ((long) (buffer[offset + 5] & 255) << 16) +
        ((long) (buffer[offset + 4] & 255) << 24) +
        ((long) (buffer[offset + 3] & 255) << 32) +
        ((long) (buffer[offset + 2] & 255) << 40) +
        ((long) (buffer[offset + 1] & 255) << 48) +
        ((long) (buffer[offset]) << 56);
    offset += 8;
    return v;
  }

  public double readDouble() {
    return Double.longBitsToDouble(readLong());
  }

  public byte[] readBytes(int len) {
    byte[] b = new byte[len];
    System.arraycopy(buffer, offset, b, 0, len);
    offset += len;
    return b;
  }

  public void skipBytes(int len) {
    offset += len;
  }

  public int getRowNums() {
    return rowNums;
  }

  public void close() {
    CarbonUtil.closeStreams(in);
  }
}

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.streaming;
19
20	import java.io.EOFException;
21	import java.io.IOException;
22	import java.io.InputStream;
23
24	import org.apache.carbondata.core.datastore.compression.Compressor;
25	import org.apache.carbondata.core.datastore.compression.CompressorFactory;
26	import org.apache.carbondata.core.util.CarbonUtil;
27	import org.apache.carbondata.format.BlockletHeader;
28
29	/**
30	* stream blocklet reader
31	*/
32	public class StreamBlockletReader {
33
34	private byte[] buffer;
35	private int offset;
36	private final byte[] syncMarker;
37	private final byte[] syncBuffer;
38	private final int syncLen;
39	private long pos = 0;	1✔
40	private final InputStream in;
41	private final long limitStart;
42	private final long limitEnd;
43	private boolean isAlreadySync = false;	1✔
44	private Compressor compressor;
45	private int rowNums = 0;	1✔
46	private int rowIndex = 0;	1✔
47	private boolean isHeaderPresent;
48
49	public StreamBlockletReader(byte[] syncMarker, InputStream in, long limit,
50	boolean isHeaderPresent, String compressorName) {	1✔
51	this.syncMarker = syncMarker;	1✔
52	syncLen = syncMarker.length;	1✔
53	syncBuffer = new byte[syncLen];	1✔
54	this.in = in;	1✔
55	limitStart = limit;	1✔
56	limitEnd = limitStart + syncLen;	1✔
57	this.isHeaderPresent = isHeaderPresent;	1✔
58	this.compressor = CompressorFactory.getInstance().getCompressor(compressorName);	1✔
59	}	1✔
60
61	private void ensureCapacity(int capacity) {
62	if (buffer == null \|\| capacity > buffer.length) {	1✔
63	buffer = new byte[capacity];	1✔
64	}
65	}	1✔
66
67	/**
68	* find the first position of sync_marker in input stream
69	*/
70	private boolean sync() throws IOException {
71	if (!readBytesFromStream(syncBuffer, 0, syncLen)) {	1✔
72	return false;	×
73	}
74	boolean skipHeader = false;	1✔
75	for (int i = 0; i < limitStart; i++) {	1✔
76	int j = 0;	1✔
77	for (; j < syncLen; j++) {	1✔
78	if (syncMarker[j] != syncBuffer[(i + j) % syncLen]) break;	1✔
79	}
80	if (syncLen == j) {	1✔
81	if (isHeaderPresent) {	1✔
82	if (skipHeader) {	1✔
83	return true;	1✔
84	} else {
85	skipHeader = true;	1✔
86	}
87	} else {
88	return true;	×
89	}
90	}
91	int value = in.read();	1✔
92	if (-1 == value) {	1✔
93	return false;	×
94	}
95	syncBuffer[i % syncLen] = (byte) value;	1✔
96	pos++;	1✔
97	}
98	return false;	×
99	}
100
101	public BlockletHeader readBlockletHeader() throws IOException {
102	int len = readIntFromStream();	1✔
103	byte[] b = new byte[len];	1✔
104	if (!readBytesFromStream(b, 0, len)) {	1✔
105	throw new EOFException("Failed to read blocklet header");	×
106	}
107	BlockletHeader header = CarbonUtil.readBlockletHeader(b);	1✔
108	rowNums = header.getBlocklet_info().getNum_rows();	1✔
109	rowIndex = 0;	1✔
110	return header;	1✔
111	}
112
113	public void readBlockletData(BlockletHeader header) throws IOException {
114	ensureCapacity(header.getBlocklet_length());	1✔
115	offset = 0;	1✔
116	int len = readIntFromStream();	1✔
117	byte[] b = new byte[len];	1✔
118	if (!readBytesFromStream(b, 0, len)) {	1✔
119	throw new EOFException("Failed to read blocklet data");	×
120	}
121	compressor.rawUncompress(b, buffer);	1✔
122	}	1✔
123
124	public void skipBlockletData(boolean reset) throws IOException {
125	int len = readIntFromStream();	1✔
126	skip(len);	1✔
127	pos += len;	1✔
128	if (reset) {	1✔
129	this.rowNums = 0;	1✔
130	this.rowIndex = 0;	1✔
131	}
132	}	1✔
133
134	private void skip(int len) throws IOException {
135	long remaining = len;	1✔
136	do {
137	long skipLen = in.skip(remaining);	1✔
138	remaining -= skipLen;	1✔
139	} while (remaining > 0);	1✔
140	}	1✔
141
142	/**
143	* find the next blocklet
144	*/
145	public boolean nextBlocklet() throws IOException {
146	if (pos >= limitStart) {	1✔
147	return false;	1✔
148	}
149	if (isAlreadySync) {	1✔
150	if (!readBytesFromStream(syncBuffer, 0, syncLen)) {	1✔
151	return false;	×
152	}
153	} else {
154	isAlreadySync = true;	1✔
155	if (!sync()) {	1✔
156	return false;	×
157	}
158	}
159
160	return pos < limitEnd;	1✔
161	}
162
163	public boolean hasNext() throws IOException {
164	return rowIndex < rowNums;	1✔
165	}
166
167	public void nextRow() {
168	rowIndex++;	1✔
169	}	1✔
170
171	public int readIntFromStream() throws IOException {
172	int ch1 = in.read();	1✔
173	int ch2 = in.read();	1✔
174	int ch3 = in.read();	1✔
175	int ch4 = in.read();	1✔
176	if ((ch1 \| ch2 \| ch3 \| ch4) < 0) throw new EOFException();	1✔
177	pos += 4;	1✔
178	return ((ch1 << 24) + (ch2 << 16) + (ch3 << 8) + (ch4 << 0));	1✔
179	}
180
181	/**
182	* Reads <code>len</code> bytes of data from the input stream into
183	* an array of bytes.
184	* @return <code>true</code> if reading data successfully, or
185	* <code>false</code> if there is no more data because the end of the stream has been reached.
186	*/
187	public boolean readBytesFromStream(byte[] b, int offset, int len) throws IOException {
188	int readLen = in.read(b, offset, len);	1✔
189	if (readLen < 0) {	1✔
190	return false;	×
191	}
192	pos += readLen;	1✔
193	if (readLen < len) {	1✔
194	return readBytesFromStream(b, offset + readLen, len - readLen);	×
195	} else {
196	return true;	1✔
197	}
198	}
199
200	public boolean readBoolean() {
201	return (buffer[offset++]) != 0;	×
202	}
203
204	public short readShort() {
205	short v = (short) ((buffer[offset + 1] & 255) +	1✔
206	((buffer[offset]) << 8));
207	offset += 2;	1✔
208	return v;	1✔
209	}
210
211	public byte[] copy(int len) {
212	byte[] b = new byte[len];	1✔
213	System.arraycopy(buffer, offset, b, 0, len);	1✔
214	return b;	1✔
215	}
216
217	public int readInt() {
218	int v = ((buffer[offset + 3] & 255) +	1✔
219	((buffer[offset + 2] & 255) << 8) +
220	((buffer[offset + 1] & 255) << 16) +
221	((buffer[offset]) << 24));
222	offset += 4;	1✔
223	return v;	1✔
224	}
225
226	public long readLong() {
227	long v = ((long)(buffer[offset + 7] & 255)) +	1✔
228	((long) (buffer[offset + 6] & 255) << 8) +
229	((long) (buffer[offset + 5] & 255) << 16) +
230	((long) (buffer[offset + 4] & 255) << 24) +
231	((long) (buffer[offset + 3] & 255) << 32) +
232	((long) (buffer[offset + 2] & 255) << 40) +
233	((long) (buffer[offset + 1] & 255) << 48) +
234	((long) (buffer[offset]) << 56);
235	offset += 8;	1✔
236	return v;	1✔
237	}
238
239	public double readDouble() {
240	return Double.longBitsToDouble(readLong());	1✔
241	}
242
243	public byte[] readBytes(int len) {
244	byte[] b = new byte[len];	1✔
245	System.arraycopy(buffer, offset, b, 0, len);	1✔
246	offset += len;	1✔
247	return b;	1✔
248	}
249
250	public void skipBytes(int len) {
251	offset += len;	1✔
252	}	1✔
253
254	public int getRowNums() {
255	return rowNums;	1✔
256	}
257
258	public void close() {
259	CarbonUtil.closeStreams(in);	1✔
260	}	1✔
261	}

apache / carbondata / 418

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