dongjoon-hyun commented on code in PR #58401:
URL: https://github.com/apache/spark/pull/58401#discussion_r3884839704
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/BasicDriverFeatureStep.scala:
##########
@@ -180,12 +180,8 @@ private[spark] class BasicDriverFeatureStep(conf:
KubernetesDriverConf)
}
Review Comment:
Now that the fragment *restoration* below is unconditional, the fragment
*removal* right above (lines 174-180) is still `ARCHIVES`-only, and this PR
removes the reason that asymmetry used to have.
In practice the upload works either way for the other keys: `uploadFileUri`
derives the target name from `fileUri.getPath.split("/").last`, and the local
copy goes through `RawLocalFileSystem.pathToFile` (`path.toUri().getPath()`),
so a fragment on the source `Path` is ignored. But that is quite indirect.
Could we either drop the `if (key == ARCHIVES)` here as well so the whole block
reads uniformly, or keep it and add a short comment explaining why archives
alone need the fragment stripped before upload?
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/BasicDriverFeatureStep.scala:
##########
@@ -180,12 +180,8 @@ private[spark] class BasicDriverFeatureStep(conf:
KubernetesDriverConf)
}
val resolved = KubernetesUtils.uploadAndTransformFileUris(value,
Some(conf.sparkConf))
if (resolved.nonEmpty) {
- val resolvedValue = if (key == ARCHIVES) {
- localUris.zip(resolved).map { case (uri, r) =>
- Utils.getUriBuilder(r).fragment(new
java.net.URI(uri).getFragment).build().toString
- }
- } else {
- resolved
+ val resolvedValue = localUris.zip(resolved).map { case (uri, r) =>
+ Utils.getUriBuilder(r).fragment(new
java.net.URI(uri).getFragment).build().toString
}
Review Comment:
This now sends *every* uploaded JAR/file/py-file through
`jakarta.ws.rs.core.UriBuilder`, including the overwhelmingly common case of no
fragment at all, where the old code passed `resolved` through untouched. I
checked the behavior with the versions Spark ships (jersey 3.1.11 /
jakarta.ws.rs 3.1.0):
```
s3a://b/upload/spark-upload-uuid/some-local-jar.jar -> (unchanged)
s3a://b/upload/spark-upload-uuid/my file.jar -> .../my%20file.jar
s3a://b/upload/spark-upload-uuid/tpl{x}.jar ->
IllegalArgumentException: The template variable 'x' has no value
s3a://b/upload/spark-upload-uuid/kor.jar (non-ASCII) -> percent-encoded
```
So two behavior changes come along for the ride:
1. A jar/file whose name contains `{...}` now fails the submission outright.
`UriBuilder.build()` treats braces as template variables, and
`Utils.resolveURI`'s encoding does not help because `uploadFileUri` builds the
target name from the *decoded* `getPath`. This has been true for `archives`
since SPARK-33615, but this PR spreads it to jars, files and py-files.
2. Names with spaces or non-ASCII characters are now percent-encoded in
`spark.jars` / `spark.files` / `spark.submit.pyFiles`. That is arguably an
improvement, but it is a user-visible change that is not mentioned in the PR
description.
Both go away if we only rebuild the URI when there actually is a fragment,
which also keeps the diff minimal:
```scala
val resolvedValue = localUris.zip(resolved).map { case (uri, r) =>
Option(new java.net.URI(uri).getFragment) match {
case Some(fragment) =>
Utils.getUriBuilder(r).fragment(fragment).build().toString
case None => r
}
}
```
WDYT?
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicDriverFeatureStepSuite.scala:
##########
@@ -404,6 +404,31 @@ class BasicDriverFeatureStepSuite extends SparkFunSuite {
path.startsWith(FILE_UPLOAD_PATH) &&
path.endsWith("some-local-jar.jar")))
}
+ test("SPARK-58969: Local dependency uploads preserve URI fragments") {
+ val fileUploadPath = "s3a://some-bucket/upload-path"
Review Comment:
nit: the neighboring `SPARK-40817` test uses `FILE_UPLOAD_PATH` for the same
value. Matching that naming would keep the two tests easy to read side by side.
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicDriverFeatureStepSuite.scala:
##########
@@ -404,6 +404,31 @@ class BasicDriverFeatureStepSuite extends SparkFunSuite {
path.startsWith(FILE_UPLOAD_PATH) &&
path.endsWith("some-local-jar.jar")))
}
+ test("SPARK-58969: Local dependency uploads preserve URI fragments") {
+ val fileUploadPath = "s3a://some-bucket/upload-path"
+ val sparkConf = new SparkConf()
+ .set(CONTAINER_IMAGE, "spark-driver:latest")
+ .set(JARS, Seq("/tmp/library-random.jar#library.jar"))
+ .set(FILES, Seq("/tmp/query-random.sql#query.sql"))
+ .set(SUBMIT_PYTHON_FILES, Seq("/tmp/module-random.py#module.py"))
+ .set(KUBERNETES_FILE_UPLOAD_PATH, fileUploadPath)
+ .set("spark.hadoop.fs.s3a.impl",
classOf[TestFileSystem].getCanonicalName)
+ .set("spark.hadoop.fs.s3a.impl.disable.cache", "true")
+ val kubernetesConf = KubernetesTestConf.createDriverConf(sparkConf =
sparkConf)
+
+ val properties = new BasicDriverFeatureStep(kubernetesConf)
+ .getAdditionalPodSystemProperties()
+ Seq(
+ (JARS.key, "library-random.jar", "library.jar"),
+ (FILES.key, "query-random.sql", "query.sql"),
+ (SUBMIT_PYTHON_FILES.key, "module-random.py", "module.py")
+ ).foreach { case (key, physicalName, alias) =>
Review Comment:
Could we add `ARCHIVES` to this table? `grep -rn ARCHIVES
resource-managers/kubernetes/core/src/test` currently returns nothing, so the
archive fragment path -- the very code this PR generalizes -- has no coverage
at all today. One extra tuple here would pin down the existing behavior at
basically zero cost.
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicDriverFeatureStepSuite.scala:
##########
@@ -404,6 +404,31 @@ class BasicDriverFeatureStepSuite extends SparkFunSuite {
path.startsWith(FILE_UPLOAD_PATH) &&
path.endsWith("some-local-jar.jar")))
}
+ test("SPARK-58969: Local dependency uploads preserve URI fragments") {
+ val fileUploadPath = "s3a://some-bucket/upload-path"
+ val sparkConf = new SparkConf()
+ .set(CONTAINER_IMAGE, "spark-driver:latest")
+ .set(JARS, Seq("/tmp/library-random.jar#library.jar"))
+ .set(FILES, Seq("/tmp/query-random.sql#query.sql"))
+ .set(SUBMIT_PYTHON_FILES, Seq("/tmp/module-random.py#module.py"))
+ .set(KUBERNETES_FILE_UPLOAD_PATH, fileUploadPath)
+ .set("spark.hadoop.fs.s3a.impl",
classOf[TestFileSystem].getCanonicalName)
+ .set("spark.hadoop.fs.s3a.impl.disable.cache", "true")
+ val kubernetesConf = KubernetesTestConf.createDriverConf(sparkConf =
sparkConf)
+
+ val properties = new BasicDriverFeatureStep(kubernetesConf)
+ .getAdditionalPodSystemProperties()
+ Seq(
+ (JARS.key, "library-random.jar", "library.jar"),
+ (FILES.key, "query-random.sql", "query.sql"),
+ (SUBMIT_PYTHON_FILES.key, "module-random.py", "module.py")
+ ).foreach { case (key, physicalName, alias) =>
+ val uploaded = new java.net.URI(properties(key))
Review Comment:
nit: this file has no `java.net.URI` import yet. Adding `import
java.net.URI` at the top and using the short name here would match the
convention in the rest of the test suites.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
