sarutak commented on code in PR #59947:
URL: https://github.com/apache/spark/pull/59947#discussion_r4422636724
##########
sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala:
##########
@@ -285,6 +290,19 @@ class ArtifactManager(session: SparkSession) extends
AutoCloseable with Logging
Map("normalizedRemoteRelativePath" ->
normalizedRemoteRelativePath.toString)
)
}
+ // A package is checked before it becomes an artifact of the session:
see checkAdded. A
+ // file that does not exist fails to transfer instead. Errors name the
file of the user, or
+ // the artifact if the file was staged for it, such as by a Spark
Connect client.
+ if (isNativeDataSourcePackage(normalizedRemoteRelativePath) &&
+ session.conf.get(StaticSQLConf.NATIVE_DATA_SOURCE_ENABLED) &&
+ Files.exists(serverLocalStagingPath)) {
+ val path = if (deleteStagedFile) {
+ normalizedRemoteRelativePath.toString
+ } else {
+ serverLocalStagingPath.toString
+ }
+ NativeDataSourcePackage.checkAdded(serverLocalStagingPath.toFile, path)
Review Comment:
Thanks, fixed in 3c422f37c57. `addLocalArtifacts` now collects
`INVALID_NATIVE_DATA_SOURCE_PACKAGE` errors like `ARTIFACT_ALREADY_EXISTS`,
through a shared `ArtifactManager.isRejectedArtifact`, so the call adds the
other artifacts and then throws all the errors with
`mergeExceptionsWithSuppressed`. I kept it consistent with
`ARTIFACT_ALREADY_EXISTS` rather than checking every package of the call first,
since a conflicting artifact does not make the call change nothing either. The
new test calls `addArtifacts` with a file that has the name of an artifact, an
invalid package and a valid one, and checks that the valid one is added and
that the error is `ARTIFACT_ALREADY_EXISTS` with the `INVALID_MANIFEST` error
suppressed. It fails if `addLocalArtifacts` only collects
`ARTIFACT_ALREADY_EXISTS` again.
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/NativeDataSourceConnectSuite.scala:
##########
@@ -92,6 +93,23 @@ class NativeDataSourceConnectSuite extends
SparkConnectServerTest {
}
}
+ nativeTest("reject a package that Spark cannot use when a session adds it") {
+ val invalid = new File(Utils.createTempDir(), "native_range.sparkpkg")
+ Files.writeString(invalid.toPath, "not a zip")
+ withNewSession { session =>
+ // The client gets the error of the server in the message of a
SparkException.
+ val e = intercept[SparkException](session.addArtifact(invalid.getPath))
Review Comment:
Fixed in 3c422f37c57 in the same way: `flushStagedArtifacts` collects these
errors too, so the handler adds the other artifacts of the stream and then
fails with the first error, with the others suppressed in it, as it does for
`ARTIFACT_ALREADY_EXISTS`. As with that error, the client then gets the first
error only, since the server does not send suppressed errors, and no summaries.
I did not move the check before the loop, so a rejected package and a
conflicting artifact behave the same way in a batch. This test now also adds an
invalid package and a valid one in one `addArtifacts` call, and checks that the
session gets the valid one; it fails if the handler only collects
`ARTIFACT_ALREADY_EXISTS`.
##########
sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala:
##########
@@ -285,6 +290,19 @@ class ArtifactManager(session: SparkSession) extends
AutoCloseable with Logging
Map("normalizedRemoteRelativePath" ->
normalizedRemoteRelativePath.toString)
)
}
+ // A package is checked before it becomes an artifact of the session:
see checkAdded. A
+ // file that does not exist fails to transfer instead. Errors name the
file of the user, or
+ // the artifact if the file was staged for it, such as by a Spark
Connect client.
+ if (isNativeDataSourcePackage(normalizedRemoteRelativePath) &&
Review Comment:
Fixed in 3c422f37c57. The package is now checked before the check for an
existing artifact, and its error is thrown after that check, so the existence
check runs right before the transfer again, as before this PR.
`ARTIFACT_ALREADY_EXISTS` still comes first, also for a directory with the name
of an artifact (the test checks both). I did not take the lock of the artifact
manager: `classloader` synchronizes on it, so holding it while copying and
checking a package would make the queries of the session that use its class
loader wait for the add, such as a Spark Connect query that resolves a UDF.
`NativeDataSourceRegistry.addToArtifacts` takes it when it distributes a
configured package, to check whether it copied the package already, and copies
it under the lock only the first time. Two concurrent adds of the same file can
still both pass the existence check, but the window is not wider than before.
One difference: a local package is now moved from its copy, as a staged one
always wa
s, rather than copied, so two adds with different content that race inside
`Files.move` can have the later one replace the file instead of failing with
`FileAlreadyExistsException`; both packages passed the check, so this never
makes an invalid package an artifact.
##########
sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala:
##########
@@ -285,6 +290,19 @@ class ArtifactManager(session: SparkSession) extends
AutoCloseable with Logging
Map("normalizedRemoteRelativePath" ->
normalizedRemoteRelativePath.toString)
)
}
+ // A package is checked before it becomes an artifact of the session:
see checkAdded. A
+ // file that does not exist fails to transfer instead. Errors name the
file of the user, or
+ // the artifact if the file was staged for it, such as by a Spark
Connect client.
+ if (isNativeDataSourcePackage(normalizedRemoteRelativePath) &&
+ session.conf.get(StaticSQLConf.NATIVE_DATA_SOURCE_ENABLED) &&
+ Files.exists(serverLocalStagingPath)) {
+ val path = if (deleteStagedFile) {
+ normalizedRemoteRelativePath.toString
+ } else {
+ serverLocalStagingPath.toString
+ }
+ NativeDataSourcePackage.checkAdded(serverLocalStagingPath.toFile, path)
+ }
transferFile(serverLocalStagingPath, target, deleteSource =
deleteStagedFile)
Review Comment:
Fixed in 3c422f37c57 as you suggested: a local package is first copied to a
temporary directory, Spark checks the copy and then moves it to the path of the
artifact, so the session gets the bytes that were checked, and the file of the
user is read once. I put the copy in a temporary directory rather than in the
directory of the artifacts of the session, since the file server serves that
directory and a clone of the session copies it. A file that does not exist or
cannot be read now fails to copy, with the same `NoSuchFileException` or
`AccessDeniedException` as before. No test covers checking the file of the user
instead of the copy: it would need the file to change during the add.
##########
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:
Right. In 3c422f37c57 the docstring of `addArtifacts` now says that without
Spark Connect the packages are added first, one at a time, so a package that is
rejected also stops the rest of the call, and the PR description says it too. I
kept the order of classic `addArtifacts` as it is: keeping the order of the
paths would mean reworking how it groups the paths by kind, which seems out of
scope here.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourcePackage.scala:
##########
@@ -205,7 +233,7 @@ object NativeDataSourcePackage extends Logging {
}
}
}
- manifest.map(NativeDataSourcePackage(file.getPath, sha256(file), _))
+ manifest.map(NativeDataSourcePackage(path, sha256(file), _))
} catch {
case e: IOException => invalid(s"The package is not a valid zip file:
$e", cause = e)
Review Comment:
Fixed in 3c422f37c57 by the copy in item 4: a local file that cannot be read
fails to copy with `AccessDeniedException`, as it failed to transfer before,
without reaching the check. The test makes a package unreadable and checks that
`addArtifact` fails with `AccessDeniedException`; only that check is skipped
when the user can read any file, as root. A staged file is written by Spark, so
it can be read. A directory named `*.sparkpkg` is also copied (as an empty
directory), so the check now reports it with the reason "The package is a
directory." rather than with the error of `ZipFile`, which would name the copy.
--
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]
