dongjoon-hyun commented on code in PR #59947:
URL: https://github.com/apache/spark/pull/59947#discussion_r4420253883
##########
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:
**2. [Low] A rejected package leaves a Spark Connect batch partly applied**
The same happens in `SparkConnectAddArtifactsHandler.flushStagedArtifacts`
([L215-L229](https://github.com/apache/spark/blob/40d50ada443137ce8b2bfa566206ea7673ab55ac/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala#L215-L229)),
which also catches only `ARTIFACT_ALREADY_EXISTS`. Its comment says "Resolve
the complete ordered batch before mutating session state", and it resolves the
Maven dependencies first for that reason. A package that fails `checkAdded` now
fails in the middle of the loop:
- the artifacts before it in the same stream are already in the session,
- the ones after it are not, and
- the client gets the error without the artifact summaries, so it cannot
tell which ones were added.
For example, the Python client sends all the paths of
`spark.addArtifacts("dep.jar", "bad.sparkpkg", "util.jar")` in one stream.
Could the handler check the staged `files/*.sparkpkg` artifacts right after
`prepareArtifacts()`, before it adds any of them? Then this error leaves the
session unchanged. This test adds a single artifact, so a test that adds
several in one call would pin down whichever behavior is chosen.
##########
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:
**3. [Low] The check widens the window between `Files.exists(target)` and
`transferFile`**
`addArtifact` is not synchronized, so two calls that add the same
`files/x.sparkpkg` can both pass `Files.exists(target)` above. Before, the
window up to `transferFile` was short. Now it includes opening the zip and
hashing the whole file, so it grows with the size of the package.
When the second call loses, `Files.move` or `Files.copy` fails with
`FileAlreadyExistsException`. For the same content, that replaces the
idempotent success that `contentEquals` gives. It is not
`ARTIFACT_ALREADY_EXISTS` either, so `addLocalArtifacts` and
`flushStagedArtifacts` do not collect it. This can happen with two threads of
one session that add the same package, or with two Spark Connect clients of one
session.
`NativeDataSourceRegistry.addToArtifacts` already synchronizes on the
artifact manager for this reason. Could the existence check, the package check
and the transfer of a file artifact run under that lock too?
##########
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:
**4. [Low] A local file is checked, and then copied again from the path of
the user**
With `deleteStagedFile = false`, `checkAdded` reads the file of the user,
and `transferFile` then copies it again from the same path. If the file changes
in between, the check passes on the old content and the session gets the new
content without a check. For example, a build may write a new `rs-2.0.sparkpkg`
while a notebook adds it. The lookup then ignores the package with a warning in
the log of the driver, which is the situation this PR prevents.
Copying the file first to a temporary file in the artifact directory of the
session, checking that copy, and then moving it to `target` would check exactly
the bytes that become the artifact. It would also read the file of the user
once instead of twice. The temporary file has another name, so an invalid
package is still never at the path of 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)
Review Comment:
**1. [Low] Rejecting a package stops `addLocalArtifacts` and drops the
`ARTIFACT_ALREADY_EXISTS` errors that it collected**
`checkAdded` throws a `SparkRuntimeException` here, but `addLocalArtifacts`
only catches `ARTIFACT_ALREADY_EXISTS`
([L374-L377](https://github.com/apache/spark/blob/40d50ada443137ce8b2bfa566206ea7673ab55ac/sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala#L374-L377)).
The new error leaves the loop, so:
- the errors already collected in `failedArtifactExceptions` are lost, and
- the artifacts before the package are added, but the ones after it are not.
For example, if `a.jar` was already added with other content,
`spark.addArtifacts(aJar, badPackage, bJar)` only reports the package. The user
fixes the package, calls again, and only then learns about `a.jar`. The
description mentions the artifacts after the rejected package, but not the lost
errors.
`INVALID_NATIVE_DATA_SOURCE_PACKAGE` is a `SparkRuntimeException` too, so
could `addLocalArtifacts` collect it like `ARTIFACT_ALREADY_EXISTS`, and report
all of them with `mergeExceptionsWithSuppressed`? Another option is to check
every package of the call before adding any artifact, so that a rejected
package changes nothing.
##########
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:
**5. [Low] In classic Python, a rejected package also stops the paths listed
before it**
Classic `addArtifacts` adds the `.sparkpkg` paths to the JVM one by one
before it handles any other path ([session.py
L2305-L2311](https://github.com/apache/spark/blob/40d50ada443137ce8b2bfa566206ea7673ab55ac/python/pyspark/sql/session.py#L2305-L2311)).
So a rejected package now also stops the paths before it in the call:
`spark.addArtifacts("a.txt", "bad.sparkpkg", "b.sparkpkg", file=True)` adds
neither `a.txt` nor `b.sparkpkg`.
The description only says that Scala, Java and Spark Connect skip the
artifacts after the rejected package, and this test adds the invalid package
alone. Could the description mention the Python case, or `addArtifacts` keep
the order of the paths?
##########
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:
**6. [Nit] A file that cannot be read is reported as an invalid zip file**
The check skips a file that does not exist, so that it still fails with
`NoSuchFileException`, but it does not skip other I/O errors. For a file
without read permission, `new ZipFile` fails with a `FileNotFoundException`
that says `Permission denied`. The user now gets `INVALID_MANIFEST` with "The
package is not a valid zip file", which suggests a broken package. Before,
`Files.copy` failed with `AccessDeniedException`.
Checking `Files.isReadable` along with `Files.exists` in `addArtifact` would
keep the earlier error. Another option is a reason here that only says "not a
valid zip file" for a `ZipException`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourcePackage.scala:
##########
@@ -132,6 +132,33 @@ 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, or does not
support its version of
+ * the interface. 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 that it
added before. 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:
**10. [Low] A package without a library for the platform of the driver still
leaves the session on a broken data source**
`checkAdded` accepts such a package, so it becomes the package that the
session added last. Every query on its data sources then fails with
`UNSUPPORTED_PLATFORM` until the session adds yet another package under a new
file name. For example, a session that uses `rs-1.0.sparkpkg` and adds
`rs-2.0.sparkpkg`, built only for `osx-aarch64`, cannot read `rs` on a Linux
driver anymore.
The error is visible now, unlike before, but the session cannot go back to
rs-1.0. That is the problem that the description gives for rejecting the other
packages. The driver loads the library of the data source to infer the schema,
so at this point Spark already knows that the package cannot work in this
application. Was there a reason to accept it? If so, could the doc of
`checkAdded` and the docs give it?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala:
##########
@@ -1555,6 +1555,70 @@ 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") {
Review Comment:
**8. [Low] The new Scala tests are canceled in the PR builds**
This test and the new one in `NativeDataSourceConnectSuite` are
`nativeTest`s. They need a C++ compiler and Java 22 or later, since `toolchain`
requires `NativeLibrary.isSupported`. The PR builds use Java 17, so both were
canceled there, and the Python test is the only new test that ran. The Scala
tests only run in the scheduled Java 25 builds, after the merge.
Most of what this PR adds does not load a library:
- the rejection of a newer `abiVersion`, of a missing library, of a file
that is not a zip and of a directory,
- `ARTIFACT_ALREADY_EXISTS` coming first, and `NoSuchFileException` for a
missing file,
- the `pyfiles/` artifact that is not a package, and
- the name of the artifact in the error that a Spark Connect client gets.
These only need `createZip` and `manifestJson`. Could they move to a plain
`test`, so that the PR builds cover them? The parts that run a query can stay
in a `nativeTest`.
##########
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
Review Comment:
**7. [Nit] A rejected in-memory package leaves its temporary file**
When `deleteStagedFile` is true, only the move in `transferFile` removes the
staged file. For `spark.addArtifact(bytes, "x.sparkpkg")`, `addLocalArtifacts`
writes the bytes to a file in a new temporary directory
([L361-L370](https://github.com/apache/spark/blob/40d50ada443137ce8b2bfa566206ea7673ab55ac/sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala#L361-L370)),
and its `finallyBlock` only closes the stream. When `checkAdded` rejects the
package, the file stays until the JVM exits, since `Utils.createTempDir` only
registers the directory for deletion at shutdown. On a long-running driver,
each rejected call keeps a copy of the package on the disk.
Deleting the temporary directory in that `finallyBlock` would cover this,
and the existing `ARTIFACT_ALREADY_EXISTS` case too. Spark Connect is fine: its
handler deletes the staging directory on errors.
##########
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) {
Review Comment:
**11. [Nit] `deleteStagedFile` decides which name the error shows**
`deleteStagedFile` says whether to move or copy the staged file, but here it
also decides whether the staged file is the file of the user. That holds for
the current callers: `LocalFile` copies, and `InMemory` and Spark Connect move.
But nothing ties the two together. A caller that moves a file of the user, or
copies a staged file, would show the wrong name, possibly a path on the server.
`addLocalArtifacts` already knows the kind of storage. Could it pass the
name to show, for example as an `Option[String]` parameter of `addArtifact`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceRegistry.scala:
##########
@@ -146,8 +142,8 @@ object NativeDataSourceRegistry extends Logging {
packages: Seq[NativeDataSourcePackage]): Option[(String,
NativeDataSourcePackage)] = {
val found = packages
.flatMap(pkg =>
pkg.manifest.dataSources.find(_.equalsIgnoreCase(name)).map(_ -> pkg))
- // The same package can be found more than once, for example when it is
in a configured
- // directory and was also added as an artifact.
+ // The same package can be found more than once, for example when it is
listed in
+ // spark.sql.dataSource.native.paths both as a file and through its
directory.
Review Comment:
**9. [Nit] The test "conflicting packages" no longer tests what its comment
says**
This comment now says that `lookupPackage` only sees the configured
packages, which is right: the packages of a session go to `lookupAddedPackage`.
But the end of the test "conflicting packages" ([NativeDataSourceSuite.scala
L1525-L1528](https://github.com/apache/spark/blob/40d50ada443137ce8b2bfa566206ea7673ab55ac/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala#L1525-L1528))
still says that it checks the same package "in the configured paths and in the
artifacts of the session". That case goes through `lookupAddedPackage`, so it
never reaches the deduplication by `sha256` here, and the test still passes
without it.
No test lists the same file both directly and through its directory, which
is the case that this comment now gives. Could the test configure one package
both ways, with a directory that contains only that package, so that it fails
without the deduplication?
##########
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), _))
Review Comment:
**12. [Nit] `checkAdded` computes a SHA-256 checksum that it does not use**
`parse` always hashes the whole file, and `checkAdded` drops the result.
With the copy into the artifacts, and the first lookup, which reads the copy
again in `tryRead`, Spark now reads a package three times when a session adds
it. That can be a lot for a package with libraries for several platforms.
Could `parse` return the manifest, and `tryRead` compute the checksum? Then
`checkAdded` only reads the central directory and the manifest.
##########
docs/sql-native-data-sources.md:
##########
@@ -325,12 +325,15 @@ A session finds the packages in two places, and the
executors get them automatic
The packages of a session take precedence over the installed libraries, and
the package that a
session added last over its other packages, also the configured ones: a
session cannot remove an
-artifact, or add a file with the same name again, so it upgrades a package by
adding the new
-version under another file name, such as `rust_range-1.1.sparkpkg`. Distinct
file names are also
-needed because the sessions that are not isolated share the files they add. A
query fails if more
-than one configured package provides its data source. Spark ignores, and logs,
a package that it
-cannot read or whose version it does not support, unless its manifest lists
the data source that a
-query uses and no other package or installed library provides it.
+artifact, or add another file with the name of one that it added, so it
upgrades a package by adding
+the new version under another file name, such as `rust_range-1.1.sparkpkg`.
Distinct file names are
+also needed because the sessions that are not isolated share the files they
add. A query fails if
+more than one configured package provides its data source. `spark.addArtifact`
fails for a package
+that Spark cannot read or whose version it does not support, so that a session
does not keep using
Review Comment:
**14. [Low] The API docs and the migration guide do not mention the new
failure**
This page now says that `spark.addArtifact` fails for such a package. But
the API docs of `addArtifact` ([SparkSession.scala
L571-L572](https://github.com/apache/spark/blob/40d50ada443137ce8b2bfa566206ea7673ab55ac/sql/api/src/main/scala/org/apache/spark/sql/SparkSession.scala#L571-L572))
and the docstring of `addArtifacts` in Python still only say that a package
makes its data sources available.
As the description notes, a file named `*.sparkpkg` that Spark 4.3 added as
a plain file, with `file=True` or from a Spark Connect client, is now rejected.
The only way to keep the old behavior is
`spark.sql.dataSource.native.enabled=false`, a static configuration that also
turns off all native data sources. A line in the section "Upgrading from Spark
SQL 4.3 to 4.4" of `sql-migration-guide.md` would help those users.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourcePackage.scala:
##########
@@ -132,6 +132,33 @@ 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, or does not
support its version of
+ * the interface. 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 that it
added before. 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 = {
+ parse(file, path) match {
+ case Left(invalid) => throw invalid.error
+ case Right(pkg) if pkg.manifest.abiVersion > NativeLibrary.ABI_VERSION =>
Review Comment:
**13. [Nit] The check of the supported version is now in three places**
`pkg.manifest.abiVersion > NativeLibrary.ABI_VERSION` is here, in `tryRead`
([L182](https://github.com/apache/spark/blob/40d50ada443137ce8b2bfa566206ea7673ab55ac/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourcePackage.scala#L182)),
and negated in `NativeDataSourceRegistry.lookup`
([L67-L70](https://github.com/apache/spark/blob/40d50ada443137ce8b2bfa566206ea7673ab55ac/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceRegistry.scala#L67-L70)).
This PR already moves the error into `unsupportedAbiVersionError` for the same
reason. A helper next to it, such as `isSupported(pkg)`, would keep the check
when a session adds a package and the check at lookup from drifting apart.
--
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]
