Skip to content
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

Allow file names to include = when convert to delta #264

Closed
wants to merge 1 commit into from
Closed

Allow file names to include = when convert to delta #264

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
2 changes: 1 addition & 1 deletion src/main/scala/io/delta/tables/execution/DeltaConvert.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@

package io.delta.tables.execution

import org.apache.spark.sql.delta.commands.{ConvertToDeltaCommand, ConvertToDeltaCommandBase}
import org.apache.spark.sql.delta.commands.ConvertToDeltaCommand
import io.delta.tables.DeltaTable

import org.apache.spark.sql.SparkSession
Expand Down
3 changes: 2 additions & 1 deletion src/main/scala/org/apache/spark/sql/delta/commands/ConvertToDeltaCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -203,8 +203,9 @@ abstract class ConvertToDeltaCommandBase(
val timestampFormatter =
TimestampFormatter(timestampPartitionPattern, java.util.TimeZone.getDefault)
val resolver = conf.resolver
val dir = if (file.isDir) file.getPath else file.getPath.getParent
val (partitionOpt, _) = PartitionUtils.parsePartition(
path,
Copy link
Collaborator

Choose a reason for hiding this comment

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

if path is no longer used, we do not need to have it.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

The path store the absolute path to a file. It is used later to create an AddFile action. But we don't need it to parse the partition in the directory.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Hi @rahulsmahadev please let me know if you have other comments on this pr.

dir,
typeInference = false,
basePaths = Set.empty,
userSpecifiedDataTypes = Map.empty,
Expand Down
25 changes: 25 additions & 0 deletions src/test/scala/org/apache/spark/sql/delta/ConvertToDeltaSuiteBase.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -296,6 +296,31 @@ abstract class ConvertToDeltaSuiteBase extends QueryTest
}
}

test("allow file names to have = character") {
withTempDir { dir =>
val tempDir = dir.getCanonicalPath
writeFiles(tempDir + "/part=1/", Seq(1).toDF("id"))

val sessionHadoopConf = spark.sessionState.newHadoopConf
val fs = new Path(tempDir).getFileSystem(sessionHadoopConf)
def listFileNames: Array[String] =
fs.listStatus(new Path(tempDir + "/part=1/"))
.map(_.getPath)
.filter(path => !path.getName.startsWith("_") && !path.getName.startsWith("."))
.map(_.toUri.toString)

val fileNames = listFileNames
assert(fileNames.size == 1)
fs.rename(new Path(fileNames.head), new Path(fileNames.head
.stripSuffix(".snappy.parquet").concat("-id=1.snappy.parquet")))

val newFileNames = listFileNames
assert(newFileNames.head.endsWith("-id=1.snappy.parquet"))
convertToDelta(s"parquet.`$tempDir`", Some("part string"))
checkAnswer(spark.read.format("delta").load(tempDir), Row(1, "1"))
}
}

test("allow file names to not have .parquet suffix") {
withTempDir { dir =>
val tempDir = dir.getCanonicalPath
Expand Down