Skip to content

[SPARK-30530][SQL][FOLLOW-UP] Remove unnecessary codes and fix comments accordingly in UnivocityParser #27287

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
HyukjinKwon wants to merge 3 commits into from
Closed

[SPARK-30530][SQL][FOLLOW-UP] Remove unnecessary codes and fix comments accordingly in UnivocityParser #27287

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
bde0b64
Cleanup
HyukjinKwon Jan 20, 2020
e2cf12a
Address comments
HyukjinKwon Jan 20, 2020
c2c5ab0
Add a comment
HyukjinKwon Jan 20, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 12 additions & 21 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,11 +67,10 @@ class UnivocityParser(
// their positions in the data schema.
private val parsedSchema = if (options.columnPruning) requiredSchema else dataSchema

val tokenizer = {
val tokenizer: CsvParser = {
val parserSetting = options.asParserSettings
// When to-be-parsed schema is shorter than the to-be-read data schema, we let Univocity CSV
// parser select a sequence of fields for reading by their positions.
// if (options.columnPruning && requiredSchema.length < dataSchema.length) {
if (parsedSchema.length < dataSchema.length) {
parserSetting.selectIndexes(tokenIndexArr: _*)
}
Expand Down Expand Up @@ -203,7 +202,11 @@ class UnivocityParser(
}
}

private val doParse = if (options.columnPruning && requiredSchema.isEmpty) {
/**
* Parses a single CSV string and turns it into either one resulting row or no row (if the
* the record is malformed).
*/
val parse: String => Option[InternalRow] = if (options.columnPruning && requiredSchema.isEmpty) {
Comment thread
HyukjinKwon marked this conversation as resolved.
Outdated
// If `columnPruning` enabled and partition attributes scanned only,
// `schema` gets empty.
(_: String) => Some(InternalRow.empty)
Expand All @@ -212,12 +215,6 @@ class UnivocityParser(
(input: String) => convert(tokenizer.parseLine(input))
}

/**
* Parses a single CSV string and turns it into either one resulting row or no row (if the
* the record is malformed).
*/
def parse(input: String): Option[InternalRow] = doParse(input)

private val getToken = if (options.columnPruning) {
(tokens: Array[String], index: Int) => tokens(index)
} else {
Expand All @@ -232,18 +229,13 @@ class UnivocityParser(
new RuntimeException("Malformed CSV record"))
}

var checkedTokens = tokens
var badRecordException: Option[Throwable] = None
Comment thread
HyukjinKwon marked this conversation as resolved.
Outdated

if (tokens.length != parsedSchema.length) {
// If the number of tokens doesn't match the schema, we should treat it as a malformed record.
// However, we still have chance to parse some of the tokens, by adding extra null tokens in
// the tail if the number is smaller, or by dropping extra tokens if the number is larger.
checkedTokens = if (parsedSchema.length > tokens.length) {
tokens ++ new Array[String](parsedSchema.length - tokens.length)
} else {
tokens.take(parsedSchema.length)
}
// However, we still have chance to parse some of the tokens. It continues to parses the
// tokens normally and sets null when `ArrayIndexOutOfBoundsException` occurs for missing
// tokens.
badRecordException = Some(new RuntimeException("Malformed CSV record"))
}
// When the length of the returned tokens is identical to the length of the parsed schema,
Expand All @@ -255,15 +247,14 @@ class UnivocityParser(
var skipRow = false
while (i < requiredSchema.length) {
try {
if (!skipRow) {
if (skipRow) {
row.setNullAt(i)
} else {
row(i) = valueConverters(i).apply(getToken(tokens, i))
if (csvFilters.skipRow(row, i)) {
skipRow = true
}
}
if (skipRow) {
row.setNullAt(i)
}
} catch {
case NonFatal(e) =>
badRecordException = badRecordException.orElse(Some(e))
Expand Down