Skip to content

[SPARK-16975][SQL][FOLLOWUP] Do not duplicately check file paths in data sources implementing FileFormat #14627

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

[SPARK-16975][SQL][FOLLOWUP] Do not duplicately check file paths in data sources implementing FileFormat #14627

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
5 changes: 2 additions & 3 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,10 +55,9 @@ class CSVFileFormat extends TextBasedFileFormat with DataSourceRegister {
options: Map[String, String],
files: Seq[FileStatus]): Option[StructType] = {
require(files.nonEmpty, "Cannot infer schema from an empty set of files")
val csvOptions = new CSVOptions(options)

// TODO: Move filtering.
val paths = files.filterNot(_.getPath.getName startsWith "_").map(_.getPath.toString)
val csvOptions = new CSVOptions(options)
val paths = files.map(_.getPath.toString)
val lines: Dataset[String] = readText(sparkSession, csvOptions, paths)
val firstLine: String = findFirstLine(csvOptions, lines)
val firstRow = new CsvReader(csvOptions).parseLine(firstLine)
Expand Down
7 changes: 1 addition & 6 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,13 +51,8 @@ class JsonFileFormat extends TextBasedFileFormat with DataSourceRegister {
val columnNameOfCorruptRecord =
parsedOptions.columnNameOfCorruptRecord
.getOrElse(sparkSession.sessionState.conf.columnNameOfCorruptRecord)
val jsonFiles = files.filterNot { status =>
val name = status.getPath.getName
(name.startsWith("_") && !name.contains("=")) || name.startsWith(".")
}.toArray

val jsonSchema = InferSchema.infer(
createBaseRdd(sparkSession, jsonFiles),
createBaseRdd(sparkSession, files),
columnNameOfCorruptRecord,
parsedOptions)
checkConstraints(jsonSchema)
Expand Down
7 changes: 1 addition & 6 deletions ...src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,12 +241,7 @@ class ParquetFileFormat
commonMetadata: Seq[FileStatus])

private def splitFiles(allFiles: Seq[FileStatus]): FileTypes = {
// Lists `FileStatus`es of all leaf nodes (files) under all base directories.
val leaves = allFiles.filter { f =>
isSummaryFile(f.getPath) ||
!((f.getPath.getName.startsWith("_") && !f.getPath.getName.contains("=")) ||
f.getPath.getName.startsWith("."))
}.toArray.sortBy(_.getPath.toString)
val leaves = allFiles.toArray.sortBy(_.getPath.toString)

FileTypes(
data = leaves.filterNot(f => isSummaryFile(f.getPath)),
Expand Down
17 changes: 17 additions & 0 deletions sql/hive/src/test/scala/org/apache/spark/sql/sources/HadoopFsRelationTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -877,6 +877,23 @@ abstract class HadoopFsRelationTest extends QueryTest with SQLTestUtils with Tes
}
}
}

test("SPARK-16975: Partitioned table with the column having '_' should be read correctly") {
withTempDir { dir =>
val childDir = new File(dir, dataSourceName).getCanonicalPath
val dataDf = spark.range(10).toDF()
val df = dataDf.withColumn("_col", $"id")
df.write.format(dataSourceName).partitionBy("_col").save(childDir)
val reader = spark.read.format(dataSourceName)

// This is needed for SimpleTextHadoopFsRelationSuite as SimpleTextSource needs schema.
if (dataSourceName == classOf[SimpleTextSource].getCanonicalName) {
reader.option("dataSchema", dataDf.schema.json)
}
val readBack = reader.load(childDir)
checkAnswer(df, readBack)
}
}
}

// This class is used to test SPARK-8578. We should not use any custom output committer when
Expand Down