Skip to content

[SPARK-26990][SQL]FileIndex: use user specified field names if possible #23894

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
gengliangwang wants to merge 2 commits into from
Closed

[SPARK-26990][SQL]FileIndex: use user specified field names if possible #23894

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
18c3dd4
SPARK-26990: use user specified field names if possible
gengliangwang Feb 26, 2019
17f7852
revise code
gengliangwang Feb 26, 2019
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
13 changes: 8 additions & 5 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,15 +111,16 @@ object PartitioningUtils {
caseSensitive: Boolean,
validatePartitionColumns: Boolean,
timeZone: TimeZone): PartitionSpec = {
val userSpecifiedDataTypes = if (userSpecifiedSchema.isDefined) {
val (userSpecifiedDataTypes, userSpecifiedNames) = if (userSpecifiedSchema.isDefined) {
Copy link
Copy Markdown
Contributor

@mgaido91 mgaido91 Feb 26, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

just a nit: we can maybe separate the userSpecifiedNames map creation in order to improve code readability. Moreover we need it only if !caseSensitive. So what about:

val userSpecifiedNames = userSpecifiedSchema.filterNot(_ => caseSensitive).map { schema => CaseInsensitiveMap(...) }.getOrElse(Map.empty[String, String])

Copy link
Copy Markdown
Member Author

@gengliangwang gengliangwang Feb 26, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks! I have revised the code.

val nameToDataType = userSpecifiedSchema.get.fields.map(f => f.name -> f.dataType).toMap
val nameToName = userSpecifiedSchema.get.fields.map(f => f.name -> f.name).toMap
if (!caseSensitive) {
CaseInsensitiveMap(nameToDataType)
(CaseInsensitiveMap(nameToDataType), CaseInsensitiveMap(nameToName))
} else {
nameToDataType
(nameToDataType, nameToName)
}
} else {
Map.empty[String, DataType]
(Map.empty[String, DataType], Map.empty[String, String])
}

val dateFormatter = DateFormatter()
Expand Down Expand Up @@ -170,7 +171,9 @@ object PartitioningUtils {
columnNames.zip(literals).map { case (name, Literal(_, dataType)) =>
// We always assume partition columns are nullable since we've no idea whether null values
// will be appended in the future.
StructField(name, userSpecifiedDataTypes.getOrElse(name, dataType), nullable = true)
val resultName = userSpecifiedNames.getOrElse(name, name)
val resultDataType = userSpecifiedDataTypes.getOrElse(name, dataType)
StructField(resultName, resultDataType, nullable = true)
}
}

Expand Down
15 changes: 15 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileIndexSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,21 @@ class FileIndexSuite extends SharedSQLContext {
}
}

test("SPARK-26990: use user specified field names if possible") {
withTempDir { dir =>
val partitionDirectory = new File(dir, "a=foo")
partitionDirectory.mkdir()
val file = new File(partitionDirectory, "text.txt")
stringToFile(file, "text")
val path = new Path(dir.getCanonicalPath)
val schema = StructType(Seq(StructField("A", StringType, false)))
withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
Copy link
Copy Markdown
Member

@gatorsmile gatorsmile Feb 26, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also test the behavior when the conf is on.

Copy link
Copy Markdown
Member Author

@gengliangwang gengliangwang Feb 27, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When the conf is on, the schema will be a instead of A. Both 2.4 and the current code has the same result.
Not sure if we should throw exceptions on this.

val fileIndex = new InMemoryFileIndex(spark, Seq(path), Map.empty, Some(schema))
assert(fileIndex.partitionSchema.length == 1 && fileIndex.partitionSchema.head.name == "A")
}
}
}

test("SPARK-26230: if case sensitive, validate partitions with original column names") {
withTempDir { dir =>
val partitionDirectory = new File(dir, "a=1")
Expand Down