spark-test commented on code in PR #59947:
URL: https://github.com/apache/spark/pull/59947#discussion_r4420328181
##########
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 {
Review Comment:
**Nit (P3):** `checkAdded` goes through `parse`, which for a readable
manifest builds `NativeDataSourcePackage(path, sha256(file), _)` (line 236), so
every add of a `files/*.sparkpkg` artifact reads and hashes the whole package.
`checkAdded` never uses that checksum: it only looks at `manifest.abiVersion`,
and `unsupportedAbiVersionError` only reads the path and the manifest. The
validation itself only needs the zip's central directory, the manifest entry
and the `getEntry` lookups.
The cost scales with the package size, and a package bundles one library per
platform. On the Spark Connect path the staged file is then moved into place,
so this discarded hash is the only full read of the file during the add, and it
runs on the handler thread. The first lookup reads and hashes the artifact
again in `tryRead` before caching it. Query results and query-time work are not
affected; this is extra latency on `spark.addArtifact`.
One way to avoid it: move the zip, manifest and library-entry validation out
of `parse` into a helper that returns the manifest or the invalid-package
result, with the same `IOException` handling. `parse` keeps building the
package and its checksum for the lookup path, and `checkAdded` calls the helper
directly and builds the unsupported-version error from `path` and the manifest,
without constructing a package.
**Verification:**
- **Inspection:** The add-time check reads only the zip's central directory,
the manifest entry and the library-entry lookups, and no code path from it
computes the package checksum.
- **Regression:** Adding an unreadable package, a package missing a listed
library, and a package with a newer abiVersion still fails with the same error
conditions, path and parameters. A valid package is still added, listed and
used. Lookup-time errors for unusable configured packages are unchanged.
##########
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:
**Nit (P3):** This reads as if the failure itself stops the session from
using the package it added before. After a rejected add, though, the session
keeps using exactly the packages it had. `checkAdded` throws before the file is
transferred or registered, and the new `NativeDataSourceSuite` test asserts
after each rejection that queries still use the earlier package ("keeps using
the configured package"). What the failure changes is that the user learns the
new package was not added, instead of the session silently staying on the old
one while the add appears to succeed. The `checkAdded` Scaladoc is closer: it
is ignoring the package that would keep the session on the package that it
added before. As written, a reader could expect a failed upgrade to make the
session fall back to a configured package or an installed library.
For example: "`spark.addArtifact` fails for a package that Spark cannot read
or whose version it does not support, instead of adding it while the session
silently keeps using the package that it added before. After the failure, the
session keeps using the packages that it had."
**Verification:**
- **Inspection:** The reworded guide sentence matches the add-time behavior:
a rejected package never becomes an artifact, the session keeps the packages it
had, and the failure is what tells the user that the upgrade did not happen, as
the checkAdded Scaladoc already says.
##########
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))
+
assert(e.getMessage.contains("[INVALID_NATIVE_DATA_SOURCE_PACKAGE.INVALID_MANIFEST]"),
e)
+ // It names the artifact, rather than the file in which the server
staged it.
+ assert(e.getMessage.contains(s"package
files${File.separator}native_range.sparkpkg:"), e)
Review Comment:
**Nit (P3):** This assertion covers the artifact naming only for
`INVALID_MANIFEST`, whose path comes straight from `invalidManifestError(path,
...)`. A staged `UNSUPPORTED_ABI_VERSION` rejection takes its path one step
later, from the package that `parse` builds: `NativeDataSourcePackage(path,
sha256(file), _)` at line 236 of `NativeDataSourcePackage.scala`, which this PR
changed from `file.getPath`. No test exercises that step with a staged file.
The `NativeDataSourceSuite` unsupported-version case adds a local file, where
`path` and `file.getPath` are the same string, and the lookup-time cases read
configured packages by their canonical path. So if line 236 were reverted, a
Spark Connect client, or a classic `addArtifact(bytes, "x.sparkpkg")`, would
get the server's staging path for a package with a newer `abiVersion`, and CI
would still pass.
The mutation that the PR description mentions, naming the staged file in
`ArtifactManager`, is caught by this test, because both errors share that path.
The gap is only the separate step in `parse`. A staged add of a package with
`abiVersion = 2` that asserts the error names `files/<name>` would close it,
either here or as an in-memory add in `NativeDataSourceSuite`.
**Verification:**
- **Regression:** A test fails if the unsupported-version error for a staged
add takes its path from the staged file instead of the artifact path. For
example, it fails when the package that parse builds is named after the file it
reads.
--
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]
