1677

Build # 1677

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/701f0ba5221c22a89a4d3ef6904421ad423f5de2">701f0ba52<a href="https://github.com/apache/carbondata/commit/701f0ba5221c22a89a4d3ef6904421ad423f5de2">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/25163c09393332e4f53eb88b7059b8d3f09a4388">25163c093</a>">25163c093</a><a href="https://github.com/apache/carbondata/commit/701f0ba5221c22a89a4d3ef6904421ad423f5de2"> into c5930527a">c5930527a</a>

Run Details

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

61284 of 77021 relevant lines covered (79.57%)

1.06 hits per line

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

97.79

/processing/src/main/java/org/apache/carbondata/processing/loading/csvinput/CSVInputFormat.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.processing.loading.csvinput;

import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import java.nio.charset.Charset;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.util.CarbonProperties;

import com.univocity.parsers.csv.CsvParser;
import com.univocity.parsers.csv.CsvParserSettings;
import org.apache.commons.io.input.BOMInputStream;
import org.apache.commons.lang.BooleanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.Seekable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.CodecPool;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionInputStream;
import org.apache.hadoop.io.compress.Decompressor;
import org.apache.hadoop.io.compress.SplitCompressionInputStream;
import org.apache.hadoop.io.compress.SplittableCompressionCodec;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.util.LineReader;
import org.apache.log4j.Logger;

/**
 * An {@link org.apache.hadoop.mapreduce.InputFormat} for csv files.  Files are broken into lines.
 * Values are the line of csv files.
 */
public class CSVInputFormat extends FileInputFormat<NullWritable, StringArrayWritable> {

  public static final String DELIMITER = "carbon.csvinputformat.delimiter";
  public static final String DELIMITER_DEFAULT = ",";
  public static final String COMMENT = "carbon.csvinputformat.comment";
  public static final String COMMENT_DEFAULT = "#";
  public static final String QUOTE = "carbon.csvinputformat.quote";
  public static final String QUOTE_DEFAULT = "\"";
  public static final String ESCAPE = "carbon.csvinputformat.escape";
  public static final String ESCAPE_DEFAULT = "\\";
  public static final String HEADER_PRESENT = "carbon.csvinputformat.header.present";
  public static final boolean HEADER_PRESENT_DEFAULT = false;
  public static final String SKIP_EMPTY_LINE = "carbon.csvinputformat.skip.empty.line";
  public static final String READ_BUFFER_SIZE = "carbon.csvinputformat.read.buffer.size";
  public static final String READ_BUFFER_SIZE_DEFAULT = "65536";
  public static final String MAX_COLUMNS = "carbon.csvinputformat.max.columns";
  public static final String NUMBER_OF_COLUMNS = "carbon.csvinputformat.number.of.columns";
  public static final int DEFAULT_MAX_NUMBER_OF_COLUMNS_FOR_PARSING = 2000;
  public static final int THRESHOLD_MAX_NUMBER_OF_COLUMNS_FOR_PARSING = 20000;

  private static final Logger LOGGER =
      LogServiceFactory.getLogService(CSVInputFormat.class.toString());


  @Override
  public RecordReader<NullWritable, StringArrayWritable> createRecordReader(InputSplit inputSplit,
      TaskAttemptContext context) throws IOException, InterruptedException {
    return new CSVRecordReader();
  }

  @Override
  protected boolean isSplitable(JobContext context, Path file) {
    final CompressionCodec codec = new CompressionCodecFactory(context.getConfiguration())
        .getCodec(file);
    if (null == codec) {
      return true;
    }
    return codec instanceof SplittableCompressionCodec;
  }

  /**
   * Sets the comment char to configuration. Default it is #.
   * @param configuration
   * @param commentChar
   */
  public static void setCommentCharacter(Configuration configuration, String commentChar) {
    if (commentChar != null && !commentChar.isEmpty()) {
      configuration.set(COMMENT, commentChar);
    }
  }

  /**
   * Sets the delimiter to configuration. Default it is ','
   * @param configuration
   * @param delimiter
   */
  public static void setCSVDelimiter(Configuration configuration, String delimiter) {
    if (delimiter != null && !delimiter.isEmpty()) {
      configuration.set(DELIMITER, delimiter);
    }
  }

  /**
   * Sets the skipEmptyLine to configuration. Default it is false
   *
   * @param configuration
   * @param skipEmptyLine
   */
  public static void setSkipEmptyLine(Configuration configuration, String skipEmptyLine) {
    if (skipEmptyLine != null && !skipEmptyLine.isEmpty()) {
      configuration.set(SKIP_EMPTY_LINE, skipEmptyLine);
    } else {
      try {
        BooleanUtils.toBoolean(CarbonProperties.getInstance()
            .getProperty(CarbonCommonConstants.CARBON_SKIP_EMPTY_LINE),"true", "false");
        configuration.set(SKIP_EMPTY_LINE, CarbonProperties.getInstance()
            .getProperty(CarbonCommonConstants.CARBON_SKIP_EMPTY_LINE));
      } catch (Exception e) {
        configuration.set(SKIP_EMPTY_LINE, CarbonCommonConstants.CARBON_SKIP_EMPTY_LINE_DEFAULT);
      }
    }
  }

  /**
   * Sets the escape character to configuration. Default it is \
   * @param configuration
   * @param escapeCharacter
   */
  public static void setEscapeCharacter(Configuration configuration, String escapeCharacter) {
    if (escapeCharacter != null && !escapeCharacter.isEmpty()) {
      configuration.set(ESCAPE, escapeCharacter);
    }
  }

  /**
   * Whether header needs to read from csv or not. By default it is false.
   * @param configuration
   * @param headerExtractEnable
   */
  public static void setHeaderExtractionEnabled(Configuration configuration,
      boolean headerExtractEnable) {
    configuration.set(HEADER_PRESENT, String.valueOf(headerExtractEnable));
  }

  /**
   * Sets the quote character to configuration. Default it is "
   * @param configuration
   * @param quoteCharacter
   */
  public static void setQuoteCharacter(Configuration configuration, String quoteCharacter) {
    if (quoteCharacter != null && !quoteCharacter.isEmpty()) {
      configuration.set(QUOTE, quoteCharacter);
    }
  }

  /**
   * Sets the read buffer size to configuration.
   * @param configuration
   * @param bufferSize
   */
  public static void setReadBufferSize(Configuration configuration, String bufferSize) {
    if (bufferSize != null && !bufferSize.isEmpty()) {
      configuration.set(READ_BUFFER_SIZE, bufferSize);
    }
  }

  public static void setMaxColumns(Configuration configuration, String maxColumns) {
    if (maxColumns != null) {
      configuration.set(MAX_COLUMNS, maxColumns);
    }
  }

  public static void setNumberOfColumns(Configuration configuration, String numberOfColumns) {
    configuration.set(NUMBER_OF_COLUMNS, numberOfColumns);
  }

  public static CsvParserSettings extractCsvParserSettings(Configuration job) {
    CsvParserSettings parserSettings = new CsvParserSettings();
    parserSettings.getFormat().setDelimiter(job.get(DELIMITER, DELIMITER_DEFAULT).charAt(0));
    parserSettings.getFormat().setComment(job.get(COMMENT, COMMENT_DEFAULT).charAt(0));
    parserSettings.setLineSeparatorDetectionEnabled(true);
    parserSettings.setNullValue("");
    parserSettings.setEmptyValue("");
    parserSettings.setIgnoreLeadingWhitespaces(false);
    parserSettings.setIgnoreTrailingWhitespaces(false);
    parserSettings.setSkipEmptyLines(
        Boolean.valueOf(job.get(SKIP_EMPTY_LINE,
            CarbonCommonConstants.CARBON_SKIP_EMPTY_LINE_DEFAULT)));
    // todo: will verify whether there is a performance degrade using -1 here
    // parserSettings.setMaxCharsPerColumn(CarbonCommonConstants.MAX_CHARS_PER_COLUMN_DEFAULT);
    parserSettings.setMaxCharsPerColumn(CarbonCommonConstants.MAX_CHARS_PER_COLUMN_INFINITY);
    String maxColumns = job.get(MAX_COLUMNS, "" + DEFAULT_MAX_NUMBER_OF_COLUMNS_FOR_PARSING);
    parserSettings.setMaxColumns(Integer.parseInt(maxColumns));
    parserSettings.getFormat().setQuote(job.get(QUOTE, QUOTE_DEFAULT).charAt(0));
    parserSettings.getFormat().setQuoteEscape(job.get(ESCAPE, ESCAPE_DEFAULT).charAt(0));
    // setting the content length to to limit the length of displayed contents being parsed/written
    // in the exception message when an error occurs.
    parserSettings.setErrorContentLength(CarbonCommonConstants.CARBON_ERROR_CONTENT_LENGTH);
    return parserSettings;
  }

  /**
   * Treats value as line in file. Key is null.
   */
  public static class CSVRecordReader extends RecordReader<NullWritable, StringArrayWritable> {

    private long start;
    private long end;
    private BoundedInputStream boundedInputStream;
    private Reader reader;
    private CsvParser csvParser;
    private StringArrayWritable value;
    private String[] columns;
    private Seekable filePosition;
    private boolean isCompressedInput;
    private Decompressor decompressor;

    @Override
    public void initialize(InputSplit inputSplit, TaskAttemptContext context)
        throws IOException, InterruptedException {
      FileSplit split = (FileSplit) inputSplit;
      start = split.getStart();
      end = start + split.getLength();
      Path file = split.getPath();
      Configuration job = context.getConfiguration();
      CompressionCodec codec = (new CompressionCodecFactory(job)).getCodec(file);
      FileSystem fs = file.getFileSystem(job);
      int bufferSize = Integer.parseInt(job.get(READ_BUFFER_SIZE, READ_BUFFER_SIZE_DEFAULT));
      FSDataInputStream fileIn = fs.open(file, bufferSize);
      InputStream inputStream;
      if (codec != null) {
        isCompressedInput = true;
        decompressor = CodecPool.getDecompressor(codec);
        if (codec instanceof SplittableCompressionCodec) {
          SplitCompressionInputStream scIn = ((SplittableCompressionCodec) codec)
              .createInputStream(fileIn, decompressor, start, end, SplittableCompressionCodec
                  .READ_MODE.BYBLOCK);
          start = scIn.getAdjustedStart();
          end = scIn.getAdjustedEnd();
          if (start != 0) {
            LineReader lineReader = new LineReader(scIn, 1);
            start += lineReader.readLine(new Text(), 0);
          }
          filePosition = scIn;
          inputStream = scIn;
        } else {
          CompressionInputStream cIn = codec.createInputStream(fileIn, decompressor);
          filePosition = cIn;
          inputStream = cIn;
        }
      } else {
        fileIn.seek(start);
        if (start != 0) {
          LineReader lineReader = new LineReader(fileIn, 1);
          start += lineReader.readLine(new Text(), 0);
        }
        boundedInputStream = new BoundedInputStream(fileIn, end - start);
        filePosition = fileIn;
        inputStream = boundedInputStream;
      }

      //Wrap input stream with BOMInputStream to skip UTF-8 BOM characters
      reader = new InputStreamReader(new BOMInputStream(inputStream),
          Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET));

      CsvParserSettings settings = extractCsvParserSettings(job);
      if (start == 0) {
        settings.setHeaderExtractionEnabled(job.getBoolean(HEADER_PRESENT,
            HEADER_PRESENT_DEFAULT));
      }
      csvParser = new CsvParser(settings);
      csvParser.beginParsing(reader);
    }

    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
      if (csvParser == null) {
        return false;
      }
      columns = csvParser.parseNext();
      if (columns == null) {
        value = null;
        return false;
      }
      if (value == null) {
        value = new StringArrayWritable();
      }
      value.set(columns);
      return true;
    }

    @Override
    public NullWritable getCurrentKey() throws IOException, InterruptedException {
      return NullWritable.get();
    }

    @Override
    public StringArrayWritable getCurrentValue() throws IOException, InterruptedException {
      return value;
    }

    private long getPos() throws IOException {
      long retVal = start;
      if (null != boundedInputStream) {
        retVal = end - boundedInputStream.getRemaining();
      } else if (isCompressedInput && null != filePosition) {
        retVal = filePosition.getPos();
      }
      return retVal;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
      return start == end ? 0.0F : Math.min(1.0F, (float) (getPos() -
          start) / (float) (end - start));
    }

    @Override
    public void close() throws IOException {
      try {
        if (reader != null) {
          reader.close();
        }
        if (boundedInputStream != null) {
          boundedInputStream.close();
        }
        if (null != csvParser) {
          csvParser.stopParsing();
        }
      } finally {
        reader = null;
        boundedInputStream = null;
        csvParser = null;
        filePosition = null;
        value = null;
        if (decompressor != null) {
          CodecPool.returnDecompressor(decompressor);
          decompressor = null;
        }
      }
    }
  }
}

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.processing.loading.csvinput;
18
19	import java.io.IOException;
20	import java.io.InputStream;
21	import java.io.InputStreamReader;
22	import java.io.Reader;
23	import java.nio.charset.Charset;
24
25	import org.apache.carbondata.common.logging.LogServiceFactory;
26	import org.apache.carbondata.core.constants.CarbonCommonConstants;
27	import org.apache.carbondata.core.util.CarbonProperties;
28
29	import com.univocity.parsers.csv.CsvParser;
30	import com.univocity.parsers.csv.CsvParserSettings;
31	import org.apache.commons.io.input.BOMInputStream;
32	import org.apache.commons.lang.BooleanUtils;
33	import org.apache.hadoop.conf.Configuration;
34	import org.apache.hadoop.fs.FSDataInputStream;
35	import org.apache.hadoop.fs.FileSystem;
36	import org.apache.hadoop.fs.Path;
37	import org.apache.hadoop.fs.Seekable;
38	import org.apache.hadoop.io.NullWritable;
39	import org.apache.hadoop.io.Text;
40	import org.apache.hadoop.io.compress.CodecPool;
41	import org.apache.hadoop.io.compress.CompressionCodec;
42	import org.apache.hadoop.io.compress.CompressionCodecFactory;
43	import org.apache.hadoop.io.compress.CompressionInputStream;
44	import org.apache.hadoop.io.compress.Decompressor;
45	import org.apache.hadoop.io.compress.SplitCompressionInputStream;
46	import org.apache.hadoop.io.compress.SplittableCompressionCodec;
47	import org.apache.hadoop.mapreduce.InputSplit;
48	import org.apache.hadoop.mapreduce.JobContext;
49	import org.apache.hadoop.mapreduce.RecordReader;
50	import org.apache.hadoop.mapreduce.TaskAttemptContext;
51	import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
52	import org.apache.hadoop.mapreduce.lib.input.FileSplit;
53	import org.apache.hadoop.util.LineReader;
54	import org.apache.log4j.Logger;
55
56	/**
57	* An {@link org.apache.hadoop.mapreduce.InputFormat} for csv files. Files are broken into lines.
58	* Values are the line of csv files.
59	*/
60	public class CSVInputFormat extends FileInputFormat<NullWritable, StringArrayWritable> {	2✔
61
62	public static final String DELIMITER = "carbon.csvinputformat.delimiter";
63	public static final String DELIMITER_DEFAULT = ",";
64	public static final String COMMENT = "carbon.csvinputformat.comment";
65	public static final String COMMENT_DEFAULT = "#";
66	public static final String QUOTE = "carbon.csvinputformat.quote";
67	public static final String QUOTE_DEFAULT = "\"";
68	public static final String ESCAPE = "carbon.csvinputformat.escape";
69	public static final String ESCAPE_DEFAULT = "\\";
70	public static final String HEADER_PRESENT = "carbon.csvinputformat.header.present";
71	public static final boolean HEADER_PRESENT_DEFAULT = false;
72	public static final String SKIP_EMPTY_LINE = "carbon.csvinputformat.skip.empty.line";
73	public static final String READ_BUFFER_SIZE = "carbon.csvinputformat.read.buffer.size";
74	public static final String READ_BUFFER_SIZE_DEFAULT = "65536";
75	public static final String MAX_COLUMNS = "carbon.csvinputformat.max.columns";
76	public static final String NUMBER_OF_COLUMNS = "carbon.csvinputformat.number.of.columns";
77	public static final int DEFAULT_MAX_NUMBER_OF_COLUMNS_FOR_PARSING = 2000;
78	public static final int THRESHOLD_MAX_NUMBER_OF_COLUMNS_FOR_PARSING = 20000;
79
80	private static final Logger LOGGER =	2✔
81	LogServiceFactory.getLogService(CSVInputFormat.class.toString());	2✔
82
83
84	@Override
85	public RecordReader<NullWritable, StringArrayWritable> createRecordReader(InputSplit inputSplit,
86	TaskAttemptContext context) throws IOException, InterruptedException {
87	return new CSVRecordReader();	2✔
88	}
89
90	@Override
91	protected boolean isSplitable(JobContext context, Path file) {
92	final CompressionCodec codec = new CompressionCodecFactory(context.getConfiguration())	2✔
93	.getCodec(file);	2✔
94	if (null == codec) {	2✔
95	return true;	2✔
96	}
97	return codec instanceof SplittableCompressionCodec;	2✔
98	}
99
100	/**
101	* Sets the comment char to configuration. Default it is #.
102	* @param configuration
103	* @param commentChar
104	*/
105	public static void setCommentCharacter(Configuration configuration, String commentChar) {
106	if (commentChar != null && !commentChar.isEmpty()) {	1✔
107	configuration.set(COMMENT, commentChar);	1✔
108	}
109	}	1✔
110
111	/**
112	* Sets the delimiter to configuration. Default it is ','
113	* @param configuration
114	* @param delimiter
115	*/
116	public static void setCSVDelimiter(Configuration configuration, String delimiter) {
117	if (delimiter != null && !delimiter.isEmpty()) {	1✔
118	configuration.set(DELIMITER, delimiter);	1✔
119	}
120	}	1✔
121
122	/**
123	* Sets the skipEmptyLine to configuration. Default it is false
124	*
125	* @param configuration
126	* @param skipEmptyLine
127	*/
128	public static void setSkipEmptyLine(Configuration configuration, String skipEmptyLine) {
129	if (skipEmptyLine != null && !skipEmptyLine.isEmpty()) {	1✔
130	configuration.set(SKIP_EMPTY_LINE, skipEmptyLine);	1✔
131	} else {
132	try {
133	BooleanUtils.toBoolean(CarbonProperties.getInstance()	1✔
134	.getProperty(CarbonCommonConstants.CARBON_SKIP_EMPTY_LINE),"true", "false");	1✔
135	configuration.set(SKIP_EMPTY_LINE, CarbonProperties.getInstance()	1✔
136	.getProperty(CarbonCommonConstants.CARBON_SKIP_EMPTY_LINE));	1✔
137	} catch (Exception e) {	1✔
138	configuration.set(SKIP_EMPTY_LINE, CarbonCommonConstants.CARBON_SKIP_EMPTY_LINE_DEFAULT);	1✔
139	}	1✔
140	}
141	}	1✔
142
143	/**
144	* Sets the escape character to configuration. Default it is \
145	* @param configuration
146	* @param escapeCharacter
147	*/
148	public static void setEscapeCharacter(Configuration configuration, String escapeCharacter) {
149	if (escapeCharacter != null && !escapeCharacter.isEmpty()) {	1✔
150	configuration.set(ESCAPE, escapeCharacter);	1✔
151	}
152	}	1✔
153
154	/**
155	* Whether header needs to read from csv or not. By default it is false.
156	* @param configuration
157	* @param headerExtractEnable
158	*/
159	public static void setHeaderExtractionEnabled(Configuration configuration,
160	boolean headerExtractEnable) {
161	configuration.set(HEADER_PRESENT, String.valueOf(headerExtractEnable));	1✔
162	}	1✔
163
164	/**
165	* Sets the quote character to configuration. Default it is "
166	* @param configuration
167	* @param quoteCharacter
168	*/
169	public static void setQuoteCharacter(Configuration configuration, String quoteCharacter) {
170	if (quoteCharacter != null && !quoteCharacter.isEmpty()) {	1✔
171	configuration.set(QUOTE, quoteCharacter);	1✔
172	}
173	}	1✔
174
175	/**
176	* Sets the read buffer size to configuration.
177	* @param configuration
178	* @param bufferSize
179	*/
180	public static void setReadBufferSize(Configuration configuration, String bufferSize) {
181	if (bufferSize != null && !bufferSize.isEmpty()) {	1✔
182	configuration.set(READ_BUFFER_SIZE, bufferSize);	1✔
183	}
184	}	1✔
185
186	public static void setMaxColumns(Configuration configuration, String maxColumns) {
187	if (maxColumns != null) {	1✔
188	configuration.set(MAX_COLUMNS, maxColumns);	1✔
189	}
190	}	1✔
191
192	public static void setNumberOfColumns(Configuration configuration, String numberOfColumns) {
193	configuration.set(NUMBER_OF_COLUMNS, numberOfColumns);	1✔
194	}	1✔
195
196	public static CsvParserSettings extractCsvParserSettings(Configuration job) {
197	CsvParserSettings parserSettings = new CsvParserSettings();	2✔
198	parserSettings.getFormat().setDelimiter(job.get(DELIMITER, DELIMITER_DEFAULT).charAt(0));	2✔
199	parserSettings.getFormat().setComment(job.get(COMMENT, COMMENT_DEFAULT).charAt(0));	2✔
200	parserSettings.setLineSeparatorDetectionEnabled(true);	2✔
201	parserSettings.setNullValue("");	2✔
202	parserSettings.setEmptyValue("");	2✔
203	parserSettings.setIgnoreLeadingWhitespaces(false);	2✔
204	parserSettings.setIgnoreTrailingWhitespaces(false);	2✔
205	parserSettings.setSkipEmptyLines(	2✔
206	Boolean.valueOf(job.get(SKIP_EMPTY_LINE,	2✔
207	CarbonCommonConstants.CARBON_SKIP_EMPTY_LINE_DEFAULT)));
208	// todo: will verify whether there is a performance degrade using -1 here
209	// parserSettings.setMaxCharsPerColumn(CarbonCommonConstants.MAX_CHARS_PER_COLUMN_DEFAULT);
210	parserSettings.setMaxCharsPerColumn(CarbonCommonConstants.MAX_CHARS_PER_COLUMN_INFINITY);	2✔
211	String maxColumns = job.get(MAX_COLUMNS, "" + DEFAULT_MAX_NUMBER_OF_COLUMNS_FOR_PARSING);	2✔
212	parserSettings.setMaxColumns(Integer.parseInt(maxColumns));	2✔
213	parserSettings.getFormat().setQuote(job.get(QUOTE, QUOTE_DEFAULT).charAt(0));	2✔
214	parserSettings.getFormat().setQuoteEscape(job.get(ESCAPE, ESCAPE_DEFAULT).charAt(0));	2✔
215	// setting the content length to to limit the length of displayed contents being parsed/written
216	// in the exception message when an error occurs.
217	parserSettings.setErrorContentLength(CarbonCommonConstants.CARBON_ERROR_CONTENT_LENGTH);	2✔
218	return parserSettings;	2✔
219	}
220
221	/**
222	* Treats value as line in file. Key is null.
223	*/
224	public static class CSVRecordReader extends RecordReader<NullWritable, StringArrayWritable> {	2✔
225
226	private long start;
227	private long end;
228	private BoundedInputStream boundedInputStream;
229	private Reader reader;
230	private CsvParser csvParser;
231	private StringArrayWritable value;
232	private String[] columns;
233	private Seekable filePosition;
234	private boolean isCompressedInput;
235	private Decompressor decompressor;
236
237	@Override
238	public void initialize(InputSplit inputSplit, TaskAttemptContext context)
239	throws IOException, InterruptedException {
240	FileSplit split = (FileSplit) inputSplit;	2✔
241	start = split.getStart();	2✔
242	end = start + split.getLength();	2✔
243	Path file = split.getPath();	2✔
244	Configuration job = context.getConfiguration();	2✔
245	CompressionCodec codec = (new CompressionCodecFactory(job)).getCodec(file);	2✔
246	FileSystem fs = file.getFileSystem(job);	2✔
247	int bufferSize = Integer.parseInt(job.get(READ_BUFFER_SIZE, READ_BUFFER_SIZE_DEFAULT));	2✔
248	FSDataInputStream fileIn = fs.open(file, bufferSize);	2✔
249	InputStream inputStream;
250	if (codec != null) {	2✔
251	isCompressedInput = true;	2✔
252	decompressor = CodecPool.getDecompressor(codec);	2✔
253	if (codec instanceof SplittableCompressionCodec) {	2✔
254	SplitCompressionInputStream scIn = ((SplittableCompressionCodec) codec)	2✔
255	.createInputStream(fileIn, decompressor, start, end, SplittableCompressionCodec	2✔
256	.READ_MODE.BYBLOCK);
257	start = scIn.getAdjustedStart();	2✔
258	end = scIn.getAdjustedEnd();	2✔
259	if (start != 0) {	2✔
260	LineReader lineReader = new LineReader(scIn, 1);	×
261	start += lineReader.readLine(new Text(), 0);	×
262	}
263	filePosition = scIn;	2✔
264	inputStream = scIn;	2✔
265	} else {	2✔
266	CompressionInputStream cIn = codec.createInputStream(fileIn, decompressor);	2✔
267	filePosition = cIn;	2✔
268	inputStream = cIn;	2✔
269	}	2✔
270	} else {
271	fileIn.seek(start);	2✔
272	if (start != 0) {	2✔
273	LineReader lineReader = new LineReader(fileIn, 1);	1✔
274	start += lineReader.readLine(new Text(), 0);	1✔
275	}
276	boundedInputStream = new BoundedInputStream(fileIn, end - start);	2✔
277	filePosition = fileIn;	2✔
278	inputStream = boundedInputStream;	2✔
279	}
280
281	//Wrap input stream with BOMInputStream to skip UTF-8 BOM characters
282	reader = new InputStreamReader(new BOMInputStream(inputStream),	2✔
283	Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET));	2✔
284
285	CsvParserSettings settings = extractCsvParserSettings(job);	2✔
286	if (start == 0) {	2✔
287	settings.setHeaderExtractionEnabled(job.getBoolean(HEADER_PRESENT,	2✔
288	HEADER_PRESENT_DEFAULT));
289	}
290	csvParser = new CsvParser(settings);	2✔
291	csvParser.beginParsing(reader);	2✔
292	}	2✔
293
294	@Override
295	public boolean nextKeyValue() throws IOException, InterruptedException {
296	if (csvParser == null) {	2✔
297	return false;	1✔
298	}
299	columns = csvParser.parseNext();	2✔
300	if (columns == null) {	2✔
301	value = null;	2✔
302	return false;	2✔
303	}
304	if (value == null) {	2✔
305	value = new StringArrayWritable();	2✔
306	}
307	value.set(columns);	2✔
308	return true;	2✔
309	}
310
311	@Override
312	public NullWritable getCurrentKey() throws IOException, InterruptedException {
313	return NullWritable.get();	2✔
314	}
315
316	@Override
317	public StringArrayWritable getCurrentValue() throws IOException, InterruptedException {
318	return value;	2✔
319	}
320
321	private long getPos() throws IOException {
322	long retVal = start;	2✔
323	if (null != boundedInputStream) {	2✔
324	retVal = end - boundedInputStream.getRemaining();	2✔
325	} else if (isCompressedInput && null != filePosition) {	2✔
326	retVal = filePosition.getPos();	2✔
327	}
328	return retVal;	2✔
329	}
330
331	@Override
332	public float getProgress() throws IOException, InterruptedException {
333	return start == end ? 0.0F : Math.min(1.0F, (float) (getPos() -	2✔
334	start) / (float) (end - start));
335	}
336
337	@Override
338	public void close() throws IOException {
339	try {
340	if (reader != null) {	2✔
341	reader.close();	2✔
342	}
343	if (boundedInputStream != null) {	2✔
344	boundedInputStream.close();	2✔
345	}
346	if (null != csvParser) {	2✔
347	csvParser.stopParsing();	2✔
348	}
349	} finally {
350	reader = null;	2✔
351	boundedInputStream = null;	2✔
352	csvParser = null;	2✔
353	filePosition = null;	2✔
354	value = null;	2✔
355	if (decompressor != null) {	2✔
356	CodecPool.returnDecompressor(decompressor);	2✔
357	decompressor = null;	2✔
358	}
359	}	×
360	}	2✔
361	}
362	}

apache / carbondata / 1677

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