sarutak commented on code in PR #59947:
URL: https://github.com/apache/spark/pull/59947#discussion_r4429231709
##########
sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala:
##########
@@ -204,6 +210,75 @@ 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 if another file is there: an artifact is
not overwritten.
+ */
+ private def artifactExists(
+ 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. A file
of the user is
+ * first 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 = {
+ val copyDir = if (deleteStagedFile) None else
Some(Utils.createTempDir().toPath)
+ Utils.tryWithSafeFinally {
+ val staged = copyDir.map { dir =>
+ // Fails as for the other artifacts if the file does not exist or
cannot be read.
+ Files.copy(serverLocalStagingPath,
dir.resolve(normalizedRemoteRelativePath.getFileName))
Review Comment:
Fixed in 6456e6aeb22 as you suggested: for a local file, Spark first
compares the file of the user with the artifact, and returns if it is the same,
before it creates the temporary directory or copies anything. The comparison on
the copy still runs right before the transfer. As a side effect, a file that
does not exist, with the name of an artifact, fails with
`ARTIFACT_ALREADY_EXISTS` again, as before this PR, which the test now checks.
Since a local file now fails with `ARTIFACT_ALREADY_EXISTS` before Spark checks
it, the test also adds invalid bytes with the name of an artifact, so that
reporting the error of the check after the comparison is still tested for a
staged package.
##########
sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala:
##########
@@ -204,6 +210,75 @@ 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 if another file is there: an artifact is
not overwritten.
+ */
+ private def artifactExists(
+ 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. A file
of the user is
+ * first 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 = {
+ val copyDir = if (deleteStagedFile) None else
Some(Utils.createTempDir().toPath)
+ Utils.tryWithSafeFinally {
+ val staged = copyDir.map { dir =>
+ // Fails as for the other artifacts if the file does not exist or
cannot be read.
+ Files.copy(serverLocalStagingPath,
dir.resolve(normalizedRemoteRelativePath.getFileName))
+ }.getOrElse(serverLocalStagingPath)
+ // The package is checked before the check for an existing artifact,
which then runs right
+ // before the transfer, but its error is reported after that check.
+ val rejection = Try(NativeDataSourcePackage.checkAdded(
+ staged.toFile,
sourceName.getOrElse(normalizedRemoteRelativePath.toString))).failed
+ // Utils.contentEquals fails for a directory, which cannot be the same
as the artifact
+ // anyway, so a directory with the name of an artifact fails as one.
+ if (Files.isDirectory(staged) && Files.exists(target)) {
Review Comment:
I kept it in the path of packages in 6456e6aeb22, and folded it into the
comparison that runs before the copy and before the transfer. It now also
covers an artifact that is a directory, such as `files/x.sparkpkg` after
`files/x.sparkpkg/y` was added, which the test checks. I did not move it into
`artifactExists`, since that changes the behavior of a released API for jars,
Python files and archives: a directory with the name of an existing artifact
would fail with `ARTIFACT_ALREADY_EXISTS`, which `addArtifacts` collects,
instead of `IllegalArgumentException`, which stops it. It would also fix
directories only halfway for those kinds: a directory without an artifact of
its name is still copied as an empty directory. Rejecting directories for every
kind of artifact seems worth a JIRA of its own; I can file it if you agree.
##########
sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala:
##########
@@ -204,6 +210,75 @@ 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 if another file is there: an artifact is
not overwritten.
+ */
+ private def artifactExists(
+ 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. A file
of the user is
+ * first 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 = {
+ val copyDir = if (deleteStagedFile) None else
Some(Utils.createTempDir().toPath)
+ Utils.tryWithSafeFinally {
+ val staged = copyDir.map { dir =>
+ // Fails as for the other artifacts if the file does not exist or
cannot be read.
+ Files.copy(serverLocalStagingPath,
dir.resolve(normalizedRemoteRelativePath.getFileName))
+ }.getOrElse(serverLocalStagingPath)
+ // The package is checked before the check for an existing artifact,
which then runs right
+ // before the transfer, but its error is reported after that check.
+ val rejection = Try(NativeDataSourcePackage.checkAdded(
Review Comment:
Done in 6456e6aeb22: `val check = Try(...)`, and `check.get` right before
the transfer.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourcePackage.scala:
##########
@@ -177,17 +228,40 @@ object NativeDataSourcePackage extends Logging {
packages.asMap().keySet().asScala.exists(_.path == canonicalPath)
}
- private def invalidManifestError(file: File, reason: String, cause:
Throwable = null) = {
+ private def invalidManifestError(path: String, reason: String, cause:
Throwable = null) = {
QueryExecutionErrors.invalidNativeDataSourcePackageError(
- file.getPath, "INVALID_MANIFEST", Map("reason" -> reason), cause)
+ path, "INVALID_MANIFEST", Map("reason" -> reason), cause)
}
- private def parse(file: File): ReadResult = {
+ /** Reads the package in the given file, which errors call `path`. */
+ private def parse(file: File, path: String): ReadResult = {
Review Comment:
Done in 6456e6aeb22: `parse(file)` uses `file.getPath`, and only
`readManifest` takes the name that errors give.
##########
python/pyspark/sql/session.py:
##########
@@ -2275,7 +2275,12 @@ def addArtifacts(
----------
*path : tuple of str
Artifact's URIs to add. A native data source package (a
``.sparkpkg`` file) makes
- the native data sources it contains available to the session.
+ the native data sources it contains available to the session. It
fails for a
+ package that Spark cannot read, whose version Spark does not
support, or that has
+ no library for the platform of the driver; the session then does
not get the
+ package.
+ Without Spark Connect, the packages are added first, one at a
time, so a package
Review Comment:
Decision: fixed in 6456e6aeb22, by rewording the docs, without changing how
the Python client sends artifacts. The docstring of `addArtifacts` (also used
by the Spark Connect session) now says: "A package that is rejected also stops
the rest of the call: without Spark Connect, the packages are added first, one
at a time, so only the packages before it are added; with Spark Connect, the
paths are added one at a time, so the paths before it are added." The guide now
says that a call that adds several artifacts in Scala or Java still adds the
others, and that in Python a rejected package stops the rest of the call in
both modes, with what each mode adds before it, and the PR description says the
same.
Verification: `SparkConnectClient.add_artifacts` loops over the paths and
sends one `AddArtifacts` stream per path, and the rejection is not retried, so
the first rejected path raises before the later paths are sent; only the Scala
client sends all the artifacts of a call in one stream, which
`flushStagedArtifacts` collects.
The other three findings of this review are in the threads of
dongjoon-hyun's second review, and are addressed there in 6456e6aeb22: the
re-add test (4424226329), the docs about a file that cannot be read
(4424226312), and the cleanup and the remaining window after an interrupt
(4424226285).
##########
python/pyspark/sql/tests/test_artifact.py:
##########
@@ -134,6 +147,15 @@ def new_file(name):
spark.addArtifact("my_source:1.sparkpkg")
finally:
os.chdir(cwd)
+ # A package that Spark cannot read 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)
Review Comment:
Follow-up in 6456e6aeb22: as spark-test pointed out in
https://github.com/apache/spark/pull/59947#discussion_r4426310101, the Python
Spark Connect client also stops at a rejected package, since it sends one
`AddArtifacts` stream per path. The docstring of `addArtifacts` and the guide
now say what each mode adds before a rejected package: without Spark Connect
only the packages before it, with Spark Connect the paths before 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]
