sarutak commented on code in PR #59947:
URL: https://github.com/apache/spark/pull/59947#discussion_r4451427971
##########
sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala:
##########
@@ -204,6 +215,100 @@ class ArtifactManager(session: SparkSession) extends
AutoCloseable with Logging
// Convert the normalized string back to a Path object
Paths.get(normalizedPathString).normalize()
}
+
+ /**
+ * Returns true if a file artifact with the same content is at `target`
already, which makes
+ * adding it idempotent, and fails with ARTIFACT_ALREADY_EXISTS if another
file is there: an
+ * artifact is not overwritten.
+ */
+ private def artifactAlreadyAdded(
+ normalizedRemoteRelativePath: Path,
+ serverLocalStagingPath: Path,
+ target: Path): Boolean = {
+ if (Files.exists(target)) {
+ if (Utils.contentEquals(target.toFile, serverLocalStagingPath.toFile)) {
+ return true
+ }
+ throw artifactAlreadyExistsError(normalizedRemoteRelativePath)
+ }
+ false
+ }
+
+ private def artifactAlreadyExistsError(normalizedRemoteRelativePath: Path):
Throwable = {
+ new SparkRuntimeException(
+ "ARTIFACT_ALREADY_EXISTS",
+ Map("normalizedRemoteRelativePath" ->
normalizedRemoteRelativePath.toString)
+ )
+ }
+
+ /**
+ * Transfers a native data source package to `target` if Spark can use it,
see
+ * [[NativeDataSourcePackage.checkAdded]], so that it never is at the path
of an artifact
+ * otherwise. Returns false if the session has the same file already, and
fails with
+ * ARTIFACT_ALREADY_EXISTS if it has another one, as artifactAlreadyAdded
does, or with the error
+ * of the check. The file is compared with the artifact first, so that
adding the same file again
+ * neither copies nor checks it. A file of the user is then copied to a
temporary directory,
+ * outside of the artifacts, which the file server serves and a clone of the
session copies, and
+ * the copy is checked and then moved, so that the session gets the bytes
that are checked. Errors
+ * name the file of the user, or the artifact if Spark staged the file for
it, such as for a Spark
+ * Connect client.
+ */
+ private def addNativeDataSourcePackage(
+ normalizedRemoteRelativePath: Path,
+ serverLocalStagingPath: Path,
+ target: Path,
+ deleteStagedFile: Boolean,
+ sourceName: Option[String]): Boolean = {
+ // Returns true if the same file was added already, and fails with
ARTIFACT_ALREADY_EXISTS if
+ // another one was, as artifactAlreadyAdded does.
+ def alreadyAdded(source: Path): Boolean = {
+ // Utils.contentEquals fails for a directory, which cannot be the same
as a package anyway,
+ // so a directory with the name of an artifact, or an artifact that is a
directory, such
+ // as files/x.sparkpkg after files/x.sparkpkg/y was added, fails as an
existing artifact.
+ if (Files.exists(target) && (Files.isDirectory(source) ||
Files.isDirectory(target))) {
+ throw artifactAlreadyExistsError(normalizedRemoteRelativePath)
+ }
+ artifactAlreadyAdded(normalizedRemoteRelativePath, source, target)
+ }
+ // ARTIFACT_ALREADY_EXISTS comes before the check, which is then not
needed for the same file.
+ if (alreadyAdded(serverLocalStagingPath)) {
+ return false
+ }
+ val copyDir = if (deleteStagedFile) None else Some(Utils.createTempDir())
+ Utils.tryWithSafeFinally {
+ val staged = copyDir.map { dir =>
+ // Fails as for the other artifacts if the file does not exist or
cannot be read.
+ val copy = dir.toPath.resolve(normalizedRemoteRelativePath.getFileName)
+ Files.copy(serverLocalStagingPath, copy)
Review Comment:
Thanks, and thanks for the approval. Fixed in c0924efa51d as you suggested:
the copy is made writable right after `Files.copy`, with a comment about the
read-only attribute on Windows and `File.delete` since Java 25 (JDK-8355954).
One difference from what you wrote: a package that is added is moved from that
copy, so its artifact is now writable even if the file of the user is
read-only, rather than read-only as before this PR. Spark only reads it, and
deletes it with the session, which this also makes possible on Windows with
Java 25. On Linux and macOS the copy keeps the permissions of the file, so a
new test, "add a read-only package as a writable artifact", adds a read-only
package and checks that its artifact has the permission to write (the
permission rather than `Files.isWritable`, which is true for root); it fails
without `setWritable`. The deletion on Windows itself is not tested, which the
description says. The other local artifacts, which `addLocalArtifacts` has
copied
since SPARK-49249, and the copies of configured packages also keep the
read-only attribute; that is not new in this PR, and the session cleanup could
be a follow-up.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala:
##########
@@ -1555,6 +1559,147 @@ class NativeDataSourceSuite extends QueryTest with
SharedSparkSession {
assert(logged.forall(_.contains("for the data source native_range")),
logged)
}
+ nativeTest("reject a package that a session adds when Spark cannot use it") {
+ // The package that a session added last takes precedence, so Spark
rejects a package that it
+ // cannot use when the session adds it, rather than keep using what the
session used before.
+ val session = newSession(defaultPackage)
+ def ids(): DataFrame = range(session, "end" -> "3").select("id")
+ val dir = Utils.createTempDir()
+ def check(pkg: File, subClass: String, params: Map[String, String]): Unit
= {
+ // Also when the same file is added again.
+ (1 to 2).foreach { _ =>
+ val e =
intercept[SparkRuntimeException](session.addArtifact(pkg.getPath))
+ assert(e.getCondition ==
s"INVALID_NATIVE_DATA_SOURCE_PACKAGE.$subClass")
+ assert(e.getMessageParameters.get("path") == pkg.getPath)
+ params.foreach { case (key, value) =>
assert(e.getMessageParameters.get(key) == value) }
+ }
+ // The session does not get it, and keeps using the configured package.
+ assert(session.artifactManager.getNativeDataSourcePackages._1.isEmpty)
+ assert(!session.artifactManager.hasFileArtifact(pkg.getName))
+ assert(!session.sparkContext.listFiles().exists(_.endsWith("/" +
pkg.getName)))
+ checkRows(ids(), Seq(Row(0L), Row(1L), Row(2L)))
+ }
+ check(createPackage(renamedNegatingLibrary, dataSources(""), dir = dir,
+ fileName = "range-2.0.sparkpkg", abiVersion = 2),
+ "UNSUPPORTED_ABI_VERSION", Map("version" -> "2", "supported" -> "1"))
+ check(createZip(dir, "range-2.1.sparkpkg",
+ NativeDataSourcePackage.MANIFEST_NAME -> manifestJson(
+ dataSources(""), Map(NativePlatform.current ->
"lib.so")).getBytes(UTF_8)),
+ "INVALID_MANIFEST", Map("reason" -> "The package does not contain the
library lib.so."))
+ // ZipFile.getEntry("lib") also finds the directory entry "lib/", as for
the manifest below.
+ check(createZip(dir, "range-2.7.sparkpkg",
+ NativeDataSourcePackage.MANIFEST_NAME -> manifestJson(
+ dataSources(""), Map(NativePlatform.current ->
"lib")).getBytes(UTF_8),
+ "lib/" -> Array.emptyByteArray, "lib/x" -> "x".getBytes(UTF_8)),
+ "INVALID_MANIFEST", Map("reason" -> "The package does not contain the
library lib."))
+ check(createZip(dir, "range-2.8.sparkpkg",
+ s"${NativeDataSourcePackage.MANIFEST_NAME}/" -> Array.emptyByteArray),
+ "INVALID_MANIFEST", Map("reason" ->
+ s"The package does not contain the manifest
${NativeDataSourcePackage.MANIFEST_NAME}."))
+ val notAZip = new File(dir, "range-2.2.sparkpkg")
+ Files.writeString(notAZip.toPath, "not a zip")
+ check(notAZip, "INVALID_MANIFEST", Map.empty)
Review Comment:
Fixed in c0924efa51d: `check` now uses `checkError` with `Map("path" ->
pkg.getPath) ++ params`, so every call compares all the parameters, and the
file that is not a zip passes `"reason" -> s"The package is not a valid zip
file: ${intercept[ZipException](new ZipFile(notAZip))}"`. The checks of
`ARTIFACT_ALREADY_EXISTS` use a small `checkExisting` that compares
`normalizedRemoteRelativePath` with `files/range-2.0.sparkpkg` (or
`files/range-5.0.sparkpkg`), also for the error that `addArtifacts` collects,
whose suppressed `INVALID_MANIFEST` is now checked with its `path` and `reason`
too. The check of a staged package with a newer `abiVersion` also compares
`version` and `supported`, and the Spark Connect test checks that the error of
the call that adds several artifacts names `files/other.sparkpkg`. Giving a
file that is not a zip another reason, or naming the file instead of the
artifact in `ARTIFACT_ALREADY_EXISTS`, now fails the test.
##########
python/pyspark/sql/tests/test_artifact.py:
##########
@@ -134,6 +150,15 @@ def new_file(name):
spark.addArtifact("my_source:1.sparkpkg")
finally:
os.chdir(cwd)
+ # A file that is not a valid package is rejected.
+ invalid = os.path.join(d, "invalid.sparkpkg")
+ with open(invalid, "wb") as f:
+ f.write(b"not a zip")
+ with self.assertRaises(SparkRuntimeException) as e:
+ spark.addArtifact(invalid)
+ self.assertEqual(
Review Comment:
Done in c0924efa51d: the test also checks that
`getMessageParameters()["path"]` is `os.path.realpath(invalid)`, with a comment
that the JVM resolves the path to its canonical form.
##########
sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala:
##########
@@ -334,6 +454,14 @@ class ArtifactManager(session: SparkSession) extends
AutoCloseable with Logging
artifacts.foreach { artifact =>
try {
artifact.storage match {
+ case d: Artifact.LocalFile if isCheckedPackage(artifact.path) =>
+ // Its errors name the file of the user.
Review Comment:
Fixed in c0924efa51d: the comment in `addLocalArtifacts`, the doc of the
private `addArtifact` and the doc of `addNativeDataSourcePackage` now say "the
errors of the check", and the doc of `artifactAlreadyAdded` says "an artifact".
The description says that `ARTIFACT_ALREADY_EXISTS` names the artifact, as
before. My reply in the thread of item 11
(https://github.com/apache/spark/pull/59947#discussion_r4439081785) said "whose
errors name the file of the user" too; only the errors of the check do.
##########
sql/core/src/test/scala/org/apache/spark/sql/artifact/ArtifactManagerSuite.scala:
##########
@@ -431,6 +434,30 @@ class ArtifactManagerSuite extends SharedSparkSession {
}
}
+ test("Add local artifacts through the public addArtifact") {
+ // A subclass that overrides addArtifact, for example to audit what a
session adds, sees the
Review Comment:
Fixed in c0924efa51d: the test comment now says "other than a native data
source package that Spark checks, which goes through a private overload", and
the Scaladoc of the public `addArtifact` says that `addLocalArtifacts` calls it
for the artifacts that it adds except a local native data source package that
Spark checks, while `spark.sql.dataSource.native.enabled` is true. The test
uses an `ArrayBuffer` now.
##########
sql/api/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -569,7 +569,10 @@ abstract class SparkSession extends Serializable with
Closeable {
* Ivy repositories local to the client.
*
* Since 4.4.0, it also supports native data source packages: local files
with the extension
- * .sparkpkg, which make the native data sources that they contain available
to the session.
+ * .sparkpkg, which make the native data sources that they contain available
to the session. It
Review Comment:
Fixed in c0924efa51d: the doc of `spark.sql.dataSource.native.enabled` and
its row in the guide now say that, when it is false, `spark.addArtifact` adds a
`.sparkpkg` file as a file of the session, without checking it as a package.
The Scaladoc of the `addArtifact` overloads and of `addArtifacts`, the Python
docstring and the guide paragraph about the check also say it.
--
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]
