sarutak commented on code in PR #59947:
URL: https://github.com/apache/spark/pull/59947#discussion_r4429230724
##########
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)) {
+ throw artifactAlreadyExistsError(normalizedRemoteRelativePath)
+ }
+ if (artifactExists(normalizedRemoteRelativePath, staged, target)) {
+ false
+ } else {
+ rejection.foreach(e => throw e)
+ transferFile(staged, target, deleteSource = true)
+ true
+ }
+ } {
+ // Only deletes the temporary copy, so failing to delete it must not
replace the result or
+ // the error of the add. Utils.createTempDir deletes it at shutdown
anyway.
+ copyDir.foreach(dir =>
Utils.tryLogNonFatalError(Utils.deleteRecursively(dir.toFile)))
Review Comment:
Thanks, fixed in 6456e6aeb22. Both temporary directories, the one of the
copy of a local package and the one of the file of an in-memory artifact, are
now deleted with `Utils.deleteQuietly`, which deletes in Java and neither
starts a process nor throws, and then removed from `ShutdownHookManager` only
if they are gone, so a directory that could not be deleted is still deleted at
shutdown; the add logs a warning then. This also removes the `rm` process for
each in-memory artifact.
This removes the failure point that this PR added between the transfer and
the registration, but not the window as a whole. `deleteQuietly` keeps the
interrupt status, and `SparkContext.addFile` then fetches the file from the
file server of the driver through an interruptible channel, so an interrupt, or
an I/O error there, can still leave the file at the path of the artifact,
recorded by the `SparkContext` (whose `addFile` records it before it fetches
it) but not by the session, and a retry then returns early. That has been the
case for file, Python file and archive artifacts, which `SparkContext` fetches
when it adds them, since SPARK-45486, which made an add whose content is at the
path already return without registering it. I would rather fix it separately,
since it touches those kinds of artifacts too, for example by making that early
return add the artifact to the session if it is not there yet (`addFile` would
then do nothing, since it recorded the file already, so the copy
of the file on the driver would need a look too). No test covers the
interrupt, since its timing cannot be controlled, and the local runs on macOS
take the Java path of `deleteRecursively` anyway.
##########
sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala:
##########
@@ -348,13 +440,16 @@ class ArtifactManager(session: SparkSession) extends
AutoCloseable with Logging
d.stream.transferTo(outStream)
addArtifact(artifact.path, tempFile, fragment = None)
}(finallyBlock = {
- outStream.close()
+ // addArtifact moves the file only when it adds the artifact, so
delete what is left.
+ Utils.tryWithSafeFinally(outStream.close()) {
+
Utils.tryLogNonFatalError(Utils.deleteRecursively(tempDir.toFile))
+ }
})
case _ =>
throw SparkException.internalError(s"Unsupported artifact storage:
${artifact.storage}")
}
} catch {
- case e: SparkRuntimeException if e.getCondition ==
"ARTIFACT_ALREADY_EXISTS" =>
+ case e: SparkRuntimeException if ArtifactManager.isRejectedArtifact(e)
=>
Review Comment:
Right, the docs promised more. In 6456e6aeb22 I kept the behavior, which is
the same as for the other artifacts, and fixed the docs: the Scaladoc of
`addArtifacts` and the guide now say that a file that does not exist or cannot
be read stops the call, as for the other artifacts, and the description says
that it stops the call (except a file with the name of an artifact: with item 6
it is compared with the artifact first again, so a missing one, or an
unreadable one of another size, fails with `ARTIFACT_ALREADY_EXISTS`, which is
collected). I also replaced "a package that Spark cannot read", which suggested
exactly this case, with "a file that is not a valid package" in the Scaladoc of
the `addArtifact` overloads, the Python docstring and the guide. As for the
other artifacts, such an I/O error also replaces the errors that the call
collected before it, such as the `INVALID_MANIFEST` of `notAZip` in your
example. I did not collect these I/O errors: packages would then be the only a
rtifacts that continue past a missing file, and a Spark Connect client reads
the file itself, so the server never sees a missing one.
##########
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 = {
+ readManifest(file, path).flatMap { manifest =>
+ try {
+ Right(NativeDataSourcePackage(path, sha256(file), manifest))
+ } catch {
+ case e: IOException => Left(InvalidNativeDataSourcePackage(
+ invalidManifestError(path, s"The package cannot be read: $e", e),
manifest.dataSources))
+ }
+ }
+ }
+
+ /**
+ * Reads the manifest of the package in the given file, which errors call
`path`, and checks
+ * that the package contains the libraries that it lists. Only reads the
central directory of
Review Comment:
Good catch, thanks for checking it. Fixed in 6456e6aeb22: `readManifest` now
treats a library whose entry is missing or is a directory entry as missing,
with `Option(zip.getEntry(library)).forall(_.isDirectory)`. The rejection test
adds a package whose manifest maps the platform to `lib`, with an explicit
`lib/` directory entry (`ZipOutputStream` does not write one by itself), and
expects `INVALID_MANIFEST` with "The package does not contain the library
lib."; it fails with the old `== null` check. The lookup of the manifest had
the same problem: `getEntry` also finds a directory entry
`spark-native-datasource.json/`, which was then rejected with the misleading
reason "The manifest must be a JSON object."; it is now treated as a missing
manifest, which the test also checks. The guide now says that `libraries` maps
each platform to the path of the library file. Since `readManifest` also reads
the configured packages, such a configured package is now invalid when it is
read too, rat
her than failing with `INVALID_LIBRARY` when a query loads the empty file; the
description says so.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala:
##########
@@ -1555,6 +1559,112 @@ 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.
Review Comment:
Added in 6456e6aeb22: after the valid `range-2.0.sparkpkg` is added, the
test adds it again from its file and from the same bytes, checks that both
calls succeed, that `getNativeDataSourcePackages` lists it once, and that
queries still use it. With item 6, the two re-adds take different paths: the
one from the file ends at the new comparison of the file of the user, before
any copy, and the one from bytes, which is staged and not copied, reaches the
comparison right before the transfer, the branch you pointed to. A mutant that
transfers the package even when the comparison before the transfer finds the
same content fails the re-add from bytes. Dropping the comparison before the
copy fails the test at a file that does not exist with the name of an artifact;
skipping only its early return would just copy the file and compare it again,
which no test can see, and the description lists it as not covered.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourcePackage.scala:
##########
@@ -132,6 +132,57 @@ object NativeDataSourcePackage extends Logging {
/** Reads the package in the given file. The result is cached until the file
changes. */
def read(file: File): NativeDataSourcePackage = tryRead(file).fold(p =>
throw p.error, identity)
+ /**
+ * Checks a package that a session adds with `spark.addArtifact`, in the
given file, before the
+ * session gets it: fails if Spark cannot read the package, does not support
its version of the
+ * interface, or if it has no library for the platform of the driver, which
loads the library to
+ * plan a query. The package that a session added last takes precedence, and
a session cannot
+ * remove an artifact, so Spark rejects such a package rather than ignore
it, as it ignores the
+ * configured ones, which would keep the session on the package or library
that it used before,
+ * or add a package that fails every query on its data sources. The library
is not loaded, so a
+ * library that fails to load fails the queries that use it. The file is not
the artifact
+ * itself, so the result is not cached, and errors call it `path`.
+ */
+ def checkAdded(file: File, path: String): Unit = {
Review Comment:
Agreed to keep it. In 6456e6aeb22 the doc of `checkAdded`, the guide and the
description say that the add does not check that the driver can run native data
sources at all (Java 22 or later, and a platform where the C data interface of
Arrow Java loads), and that there the add succeeds and the queries fail; the
guide names `NATIVE_DATA_SOURCE_UNSUPPORTED_JAVA_VERSION` and
`NATIVE_DATA_SOURCE_UNSUPPORTED_PLATFORM`.
--
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]
