Skip to content

[SQL][Minor] A few minor tweaks to CSV reader. #10919

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
wants to merge 1 commit into from
Closed

[SQL][Minor] A few minor tweaks to CSV reader. #10919

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
21 changes: 8 additions & 13 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVInferSchema.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,16 +30,15 @@ import org.apache.spark.sql.catalyst.analysis.HiveTypeCoercion
import org.apache.spark.sql.types._


private[sql] object CSVInferSchema {
private[csv] object CSVInferSchema {

/**
* Similar to the JSON schema inference
* 1. Infer type of each row
* 2. Merge row types to find common type
* 3. Replace any null types with string type
* TODO(hossein): Can we reuse JSON schema inference? [SPARK-12670]
*/
def apply(
def infer(
tokenRdd: RDD[Array[String]],
header: Array[String],
nullValue: String = ""): StructType = {
Expand All @@ -65,10 +64,7 @@ private[sql] object CSVInferSchema {
rowSoFar
}

private[csv] def mergeRowTypes(
first: Array[DataType],
second: Array[DataType]): Array[DataType] = {

def mergeRowTypes(first: Array[DataType], second: Array[DataType]): Array[DataType] = {
first.zipAll(second, NullType, NullType).map { case ((a, b)) =>
val tpe = findTightestCommonType(a, b).getOrElse(StringType)
tpe match {
Expand All @@ -82,8 +78,7 @@ private[sql] object CSVInferSchema {
* Infer type of string field. Given known type Double, and a string "1", there is no
* point checking if it is an Int, as the final type must be Double or higher.
*/
private[csv] def inferField(
typeSoFar: DataType, field: String, nullValue: String = ""): DataType = {
def inferField(typeSoFar: DataType, field: String, nullValue: String = ""): DataType = {
if (field == null || field.isEmpty || field == nullValue) {
typeSoFar
} else {
Expand Down Expand Up @@ -155,7 +150,8 @@ private[sql] object CSVInferSchema {
}
}

object CSVTypeCast {

private[csv] object CSVTypeCast {

/**
* Casts given string datum to specified type.
Expand All @@ -167,7 +163,7 @@ object CSVTypeCast {
* @param datum string value
* @param castType SparkSQL type
*/
private[csv] def castTo(
def castTo(
datum: String,
castType: DataType,
nullable: Boolean = true,
Expand Down Expand Up @@ -201,10 +197,9 @@ object CSVTypeCast {
* Helper method that converts string representation of a character to actual character.
* It handles some Java escaped strings and throws exception if given string is longer than one
* character.
*
*/
@throws[IllegalArgumentException]
private[csv] def toChar(str: String): Char = {
def toChar(str: String): Char = {
if (str.charAt(0) == '\\') {
str.charAt(1)
match {
Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVRelation.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,7 +139,7 @@ private[csv] class CSVRelation(

val parsedRdd = tokenRdd(header, paths)
if (params.inferSchemaFlag) {
CSVInferSchema(parsedRdd, header, params.nullValue)
CSVInferSchema.infer(parsedRdd, header, params.nullValue)
} else {
// By default fields are assumed to be StringType
val schemaFields = header.map { fieldName =>
Expand Down