2806

Build # 2806

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/ecceeb2aade71ac15632eef73210c94441ebb4a0">ecceeb2aa">&lt;a href=&quot;https://github.com/apache/carbondata/commit/</a><a class="double-link" href="https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/ecceeb2aade71ac15632eef73210c94441ebb4a0">ecceeb2aa</a><a href="https://github.com/apache/carbondata/commit/ecceeb2aade71ac15632eef73210c94441ebb4a0">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/ecceeb2aade71ac15632eef73210c94441ebb4a0&quot;&gt;ecceeb2aa&lt;/a&gt;&lt;a href=&quot;https://github.com/apache/carbondata/commit/ecceeb2aade71ac15632eef73210c94441ebb4a0&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&amp;amp;lt;a ... (continued)

Run Details

62483 of 78960 relevant lines covered (79.13%)

1.05 hits per line

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

97.84

/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.commons.lang.StringUtils;
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";
  /**
   * support only one column index
   */
  public static final String SELECT_COLUMN_INDEX = "carbon.csvinputformat.select.column.index";
  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);

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

apache / carbondata / 2806

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