dongjoon-hyun commented on code in PR #59947:
URL: https://github.com/apache/spark/pull/59947#discussion_r4424226319
##########
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:
**3. [Low] A directory entry passes the check that the package contains its
library**
`readManifest` checks each library with `zip.getEntry(_) == null`
([L274](https://github.com/apache/spark/blob/3c422f37c57f6e24ca24140360fde1fdd96eaf6c/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourcePackage.scala#L274)),
but `ZipFile.getEntry(name)` also finds the entry `name/`. I checked this with
`jshell` on Java 25: in a zip with the entry `linux-x86_64/`,
`getEntry("linux-x86_64")` returns that directory entry, and its stream has 0
bytes.
So a manifest with `"linux-x86_64": "linux-x86_64"`, a path that is easy to
write by mistake, passes `checkAdded`. `extract` then writes an empty file, and
every query on the data sources of the package fails with `INVALID_LIBRARY`.
The session cannot go back to the package that it used before until it adds
another one under a new file name, which is the case that this PR rejects for a
missing library.
The line is not new, but `checkAdded` now relies on it.
`Option(zip.getEntry(library)).forall(_.isDirectory)` would treat a directory
entry as missing.
##########
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:
**4. [Low] No test adds a valid package twice**
This loop repeats the add only for rejected packages, and the other tests
add a valid package once to each session. So no test reaches the branch of
`addNativeDataSourcePackage` that returns false
([L268-L269](https://github.com/apache/spark/blob/3c422f37c57f6e24ca24140360fde1fdd96eaf6c/sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala#L268-L269)).
That path changed for a local package, which is now copied first, and the copy
is compared with the artifact.
Could this test add a valid package again, from its file and from the same
bytes, and check that both calls succeed and that `getNativeDataSourcePackages`
lists it once?
##########
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:
**2. [Low] A package that cannot be read still stops `addArtifacts`, unlike
what the docs say**
This catch now collects a rejected package, but the `AccessDeniedException`
or `NoSuchFileException` of the copy
([L257](https://github.com/apache/spark/blob/3c422f37c57f6e24ca24140360fde1fdd96eaf6c/sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala#L257))
is not a `SparkRuntimeException`, so it leaves the loop. For example, with
`spark.addArtifacts(notAZip, unreadable, valid)`:
- `valid` is not added, and
- the `INVALID_MANIFEST` error of `notAZip` is lost.
Other artifacts behave the same way, so the behavior itself may be fine. But
the new docs promise more:
- The [Scaladoc of
`addArtifacts`](https://github.com/apache/spark/blob/3c422f37c57f6e24ca24140360fde1fdd96eaf6c/sql/api/src/main/scala/org/apache/spark/sql/SparkSession.scala#L661-L664)
says "A package that Spark cannot read ... is rejected: the session does not
get it, and the call fails after it adds the other artifacts".
- The
[guide](https://github.com/apache/spark/blob/3c422f37c57f6e24ca24140360fde1fdd96eaf6c/docs/sql-native-data-sources.md?plain=1#L331-L336)
says that `spark.addArtifact` fails for a package "that Spark cannot read",
and that a call "still adds the others".
Could the docs say that a file that does not exist or cannot be read stops
the call, or could `addLocalArtifacts` also collect these I/O errors for a
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)) {
+ 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:
**1. [Low] An interrupt while the copy is deleted leaves the package
unregistered**
`Utils.deleteRecursively` runs `rm -rf` on Unix and rethrows the
`InterruptedException` of `process.waitFor()` ([JavaUtils.java
L340-L344](https://github.com/apache/spark/blob/3c422f37c57f6e24ca24140360fde1fdd96eaf6c/common/utils-java/src/main/java/org/apache/spark/network/util/JavaUtils.java#L340-L344)).
`tryLogNonFatalError` only catches non-fatal errors, so this exception leaves
the `finally` block, unlike what the comment above says.
At this point `transferFile` has moved the package to `target`, but
`addFile` and `sparkContextRelativePaths.add` have not run. If the thread is
interrupted here:
1. `spark.addArtifact` fails with `InterruptedException`, and the package
stays at the path of the artifact without being registered.
2. Adding the same file again finds the same content at `target`, so
`artifactExists` returns true and the call succeeds without registering
anything.
3. The session does not get the package under that file name, and no error
says so.
I found this by reading the code and did not reproduce it, since it depends
on when the interrupt comes.
Could this use `Utils.deleteQuietly` instead? It deletes in Java, so it
neither starts a process nor throws. It leaves the path registered in
`ShutdownHookManager`, so `removeShutdownDeleteDir` would be needed too.
The in-memory case at
[L445](https://github.com/apache/spark/blob/3c422f37c57f6e24ca24140360fde1fdd96eaf6c/sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala#L445)
makes the same call. There the artifact is already registered, but the
exception still replaces the result and stops the loop. That call also starts
one `rm` process for each in-memory artifact of any kind, which is new in this
PR.
##########
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:
**6. [Nit] Adding the same package again copies the whole file**
The copy runs before the check for an existing artifact. So when a notebook
runs again and adds the same large package, Spark copies it, checks the copy,
compares the copy with the artifact, and deletes the copy. Before this PR, that
call only compared the two files.
`artifactExists` could run on the file of the user first, so that the call
ends there if the artifact exists. It would still run again on the copy, right
before the transfer. With item 7, this also covers a directory.
##########
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:
**8. [Nit] `Try(...).failed` could be a plain `Try`**
This turns the result of the check into a `Try[Throwable]` with `.failed`,
and rethrows it later with `foreach`. Keeping `val check = Try(...)` and
calling `check.get` before the transfer does the same and is easier to read. It
also avoids the `UnsupportedOperationException`, with its stack trace, that
`Success.failed` creates for every valid package.
##########
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:
**5. [Nit] A package is still accepted where no native data source can run**
On a driver with Java before 22, or on a platform where the C data interface
of Arrow Java does not load, `spark.addArtifact` succeeds. Every query on the
data sources of the package then fails with
`NATIVE_DATA_SOURCE_UNSUPPORTED_JAVA_VERSION` or
`NATIVE_DATA_SOURCE_UNSUPPORTED_PLATFORM`.
This does less harm than the cases that this PR rejects: no package can run
there, so the session does not lose a working one, and the error is clear.
`test_add_native_data_source_package` also relies on it in the Java 17 PR
builds. So keeping it seems fine, but the title says "when Spark cannot use
it". Could the doc of `checkAdded` and the guide mention this case, so that a
reader does not expect it to be rejected?
##########
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:
**9. [Nit] `parse` does not need `path`**
Its only caller is `parse(canonical, canonical.getPath)`
([L201](https://github.com/apache/spark/blob/3c422f37c57f6e24ca24140360fde1fdd96eaf6c/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourcePackage.scala#L201)).
Only `readManifest` needs a name that can differ from the file, for
`checkAdded`. `parse(file)` could use `file.getPath`, which also shows that the
package never gets the name of another file.
##########
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:
**7. [Nit] Only packages get this check for a directory**
`artifactExists` is shared, but the directory case is handled only here. A
directory named `a.jar`, added when `jars/a.jar` exists, still fails with the
`IllegalArgumentException` of `Utils.contentEquals`. Moving this check into
`artifactExists` would give every artifact that goes through it
`ARTIFACT_ALREADY_EXISTS`, and remove the special case.
--
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]
