apache / carbondata / 25

Build # 25

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/00d619ab7336d88b168b57620d3e723fe2b4be87">00d619ab7<a href="https://github.com/apache/carbondata/commit/00d619ab7336d88b168b57620d3e723fe2b4be87">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/7440c38b5f3b5d2eec292b70bce838829b489d11">7440c38b5</a>">7440c38b5</a><a href="https://github.com/apache/carbondata/commit/00d619ab7336d88b168b57620d3e723fe2b4be87"> into 1fb1f19f2">1fb1f19f2</a>

Run Details

212 of 212 new or added lines in 66 files covered. (100.0%)

57377 of 71135 relevant lines covered (80.66%)

1.07 hits per line

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

82.98

/core/src/main/java/org/apache/carbondata/core/datastore/impl/DFSFileReaderImpl.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.core.datastore.impl;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.FileReader;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class DFSFileReaderImpl implements FileReader {
  /**
   * cache to hold filename and its stream
   */
  private Map<String, FSDataInputStream> fileNameAndStreamCache;

  private boolean readPageByPage;

  private Configuration configuration;

  public DFSFileReaderImpl(Configuration configuration) {
    this.configuration = configuration;
    this.fileNameAndStreamCache =
        new HashMap<String, FSDataInputStream>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
  }

  @Override public byte[] readByteArray(String filePath, long offset, int length)
      throws IOException {
    FSDataInputStream fileChannel = updateCache(filePath);
    return read(fileChannel, length, offset);
  }

  /**
   * This method will be used to check whether stream is already present in
   * cache or not for filepath if not present then create it and then add to
   * cache, other wise get from cache
   *
   * @param filePath fully qualified file path
   * @return channel
   */
  private FSDataInputStream updateCache(String filePath) throws IOException {
    FSDataInputStream fileChannel = fileNameAndStreamCache.get(filePath);
    if (null == fileChannel) {
      Path pt = new Path(filePath);
      FileSystem fs = pt.getFileSystem(configuration);
      fileChannel = fs.open(pt);
      fileNameAndStreamCache.put(filePath, fileChannel);
    }
    return fileChannel;
  }

  /**
   * This method will be used to read from file based on number of bytes to be read and position
   *
   * @param channel file channel
   * @param size    number of bytes
   * @param offset  position
   * @return byte buffer
   */
  private byte[] read(FSDataInputStream channel, int size, long offset) throws IOException {
    byte[] byteBffer = new byte[size];
    channel.seek(offset);
    channel.readFully(byteBffer);
    return byteBffer;
  }

  /**
   * This method will be used to read from file based on number of bytes to be read and position
   *
   * @param channel file channel
   * @param size    number of bytes
   * @return byte buffer
   */
  private byte[] read(FSDataInputStream channel, int size) throws IOException {
    byte[] byteBffer = new byte[size];
    channel.readFully(byteBffer);
    return byteBffer;
  }

  @Override public int readInt(String filePath, long offset) throws IOException {
    FSDataInputStream fileChannel = updateCache(filePath);
    fileChannel.seek(offset);
    return fileChannel.readInt();
  }

  @Override public long readDouble(String filePath, long offset) throws IOException {
    FSDataInputStream fileChannel = updateCache(filePath);
    fileChannel.seek(offset);
    return fileChannel.readLong();
  }

  @Override public void finish() throws IOException {
    for (Entry<String, FSDataInputStream> entry : fileNameAndStreamCache.entrySet()) {
      FSDataInputStream channel = entry.getValue();
      if (null != channel) {
        channel.close();
      }
    }
  }

  @Override public byte[] readByteArray(String filePath, int length) throws IOException {
    FSDataInputStream fileChannel = updateCache(filePath);
    return read(fileChannel, length);
  }

  @Override public long readLong(String filePath, long offset) throws IOException {
    FSDataInputStream fileChannel = updateCache(filePath);
    fileChannel.seek(offset);
    return fileChannel.readLong();
  }

  @Override public int readInt(String filePath) throws IOException {
    FSDataInputStream fileChannel = updateCache(filePath);
    return fileChannel.readInt();
  }

  @Override public ByteBuffer readByteBuffer(String filePath, long offset, int length)
      throws IOException {
    byte[] readByteArray = readByteArray(filePath, offset, length);
    ByteBuffer byteBuffer = ByteBuffer.wrap(readByteArray);
    byteBuffer.rewind();
    return byteBuffer;
  }

  @Override public void setReadPageByPage(boolean isReadPageByPage) {
    this.readPageByPage = isReadPageByPage;
  }

  @Override public boolean isReadPageByPage() {
    return readPageByPage;
  }

  public Map<String, FSDataInputStream> getFileNameAndStreamCache() {
    return fileNameAndStreamCache;
  }

}

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.core.datastore.impl;
18
19	import java.io.IOException;
20	import java.nio.ByteBuffer;
21	import java.util.HashMap;
22	import java.util.Map;
23	import java.util.Map.Entry;
24
25	import org.apache.carbondata.core.constants.CarbonCommonConstants;
26	import org.apache.carbondata.core.datastore.FileReader;
27
28	import org.apache.hadoop.conf.Configuration;
29	import org.apache.hadoop.fs.FSDataInputStream;
30	import org.apache.hadoop.fs.FileSystem;
31	import org.apache.hadoop.fs.Path;
32
33	public class DFSFileReaderImpl implements FileReader {
34	/**
35	* cache to hold filename and its stream
36	*/
37	private Map<String, FSDataInputStream> fileNameAndStreamCache;
38
39	private boolean readPageByPage;
40
41	private Configuration configuration;
42
43	public DFSFileReaderImpl(Configuration configuration) {	2✔
44	this.configuration = configuration;	2✔
45	this.fileNameAndStreamCache =	2✔
46	new HashMap<String, FSDataInputStream>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
47	}	2✔
48
49	@Override public byte[] readByteArray(String filePath, long offset, int length)
50	throws IOException {
51	FSDataInputStream fileChannel = updateCache(filePath);	2✔
52	return read(fileChannel, length, offset);	2✔
53	}
54
55	/**
56	* This method will be used to check whether stream is already present in
57	* cache or not for filepath if not present then create it and then add to
58	* cache, other wise get from cache
59	*
60	* @param filePath fully qualified file path
61	* @return channel
62	*/
63	private FSDataInputStream updateCache(String filePath) throws IOException {
64	FSDataInputStream fileChannel = fileNameAndStreamCache.get(filePath);	2✔
65	if (null == fileChannel) {	2✔
66	Path pt = new Path(filePath);	2✔
67	FileSystem fs = pt.getFileSystem(configuration);	2✔
68	fileChannel = fs.open(pt);	2✔
69	fileNameAndStreamCache.put(filePath, fileChannel);	2✔
70	}
71	return fileChannel;	2✔
72	}
73
74	/**
75	* This method will be used to read from file based on number of bytes to be read and position
76	*
77	* @param channel file channel
78	* @param size number of bytes
79	* @param offset position
80	* @return byte buffer
81	*/
82	private byte[] read(FSDataInputStream channel, int size, long offset) throws IOException {
83	byte[] byteBffer = new byte[size];	2✔
84	channel.seek(offset);	2✔
85	channel.readFully(byteBffer);	2✔
86	return byteBffer;	2✔
87	}
88
89	/**
90	* This method will be used to read from file based on number of bytes to be read and position
91	*
92	* @param channel file channel
93	* @param size number of bytes
94	* @return byte buffer
95	*/
96	private byte[] read(FSDataInputStream channel, int size) throws IOException {
97	byte[] byteBffer = new byte[size];	2✔
98	channel.readFully(byteBffer);	2✔
99	return byteBffer;	2✔
100	}
101
102	@Override public int readInt(String filePath, long offset) throws IOException {
103	FSDataInputStream fileChannel = updateCache(filePath);	2✔
104	fileChannel.seek(offset);	2✔
105	return fileChannel.readInt();	2✔
106	}
107
108	@Override public long readDouble(String filePath, long offset) throws IOException {
109	FSDataInputStream fileChannel = updateCache(filePath);	2✔
110	fileChannel.seek(offset);	2✔
111	return fileChannel.readLong();	2✔
112	}
113
114	@Override public void finish() throws IOException {
115	for (Entry<String, FSDataInputStream> entry : fileNameAndStreamCache.entrySet()) {	2✔
116	FSDataInputStream channel = entry.getValue();	2✔
117	if (null != channel) {	2✔
118	channel.close();	2✔
119	}
120	}	2✔
121	}	2✔
122
123	@Override public byte[] readByteArray(String filePath, int length) throws IOException {
124	FSDataInputStream fileChannel = updateCache(filePath);	2✔
125	return read(fileChannel, length);	2✔
126	}
127
128	@Override public long readLong(String filePath, long offset) throws IOException {
129	FSDataInputStream fileChannel = updateCache(filePath);	2✔
130	fileChannel.seek(offset);	2✔
131	return fileChannel.readLong();	2✔
132	}
133
134	@Override public int readInt(String filePath) throws IOException {
135	FSDataInputStream fileChannel = updateCache(filePath);	2✔
136	return fileChannel.readInt();	2✔
137	}
138
139	@Override public ByteBuffer readByteBuffer(String filePath, long offset, int length)
140	throws IOException {
141	byte[] readByteArray = readByteArray(filePath, offset, length);	×
142	ByteBuffer byteBuffer = ByteBuffer.wrap(readByteArray);	×
143	byteBuffer.rewind();	×
144	return byteBuffer;	×
145	}
146
147	@Override public void setReadPageByPage(boolean isReadPageByPage) {
148	this.readPageByPage = isReadPageByPage;	×
149	}	×
150
151	@Override public boolean isReadPageByPage() {
152	return readPageByPage;	×
153	}
154
155	public Map<String, FSDataInputStream> getFileNameAndStreamCache() {
156	return fileNameAndStreamCache;	×
157	}
158
159	}

apache / carbondata / 25

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