sarutak commented on code in PR #59947:
URL: https://github.com/apache/spark/pull/59947#discussion_r4422638341
##########
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:
Fixed in 3c422f37c57: the `finallyBlock` of the in-memory case now deletes
the temporary directory, which also covers `ARTIFACT_ALREADY_EXISTS`. No test
checks it: the directory is created under `java.io.tmpdir`, which the other
tests of the suite share.
##########
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:
They do run in the PR builds: the job "Java 25 build with Maven" in
`build_and_test.yml` has a step "Test native data sources", which runs
`NativeDataSourceSuite`, `NativeDataSourceClusterSuite` and
`NativeDataSourceConnectSuite` on Java 25 with the C++ compiler of the runner.
On the previous head, 40d50ada443, it ran "reject a package that a session adds
when Spark cannot use it" and "reject a package that Spark cannot use when a
session adds it", and both passed
(https://github.com/sarutak/spark/actions/runs/40969344226/job/123798271572).
So I kept them as `nativeTest`s, like the other tests of these suites.
##########
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:
Fixed in 3c422f37c57. The test now lists a package in
`spark.sql.dataSource.native.paths` both as a file and through its directory,
which contains only that package, and the comment says so. It fails without the
deduplication by `sha256` in `lookupPackage`. The part that adds a configured
package as an artifact too is kept, with a comment that says that the same
package, configured and also added by the session, is not a conflict either.
(That the package of the session takes precedence is tested by "the package
that a session added last replaces the other packages".)
##########
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:
Agreed, there was no good reason to accept it: the driver loads the library
to plan any query on its data sources (`capabilities` at analysis, then the
planner), so such a package can never be used from that driver, and the session
could not go back to the package it added before. In 3c422f37c57 `checkAdded`
also rejects a package without a library for `NativePlatform.current`, with
`INVALID_NATIVE_DATA_SOURCE_PACKAGE.UNSUPPORTED_PLATFORM`, the error that a
query got before. The check still does not load the library, so a library that
fails to load fails the queries that use it, which the doc of `checkAdded` and
the docs now say. The executors are not affected: they check their own platform
when they load the library, as before.
##########
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:
Fixed in 3c422f37c57. `addLocalArtifacts` passes the name of the file of the
user to a private overload of `addArtifact`, as `sourceName: Option[String]`,
and the errors name the artifact when it is not given. The public `addArtifact`
keeps its signature and passes `None`.
##########
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:
Fixed in 3c422f37c57. `readManifest` now reads the manifest and checks the
entries of the libraries, and `parse` uses it and then computes the checksum
for the lookup. `checkAdded` only calls `readManifest`, so it does not read the
whole file. With the copy from item 4, adding a local package reads it once, by
the copy, instead of twice; a staged package, as from a Spark Connect client,
is only moved. The first lookup still hashes the artifact in `tryRead`, which
caches the result.
##########
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:
Fixed in 3c422f37c57: `NativeDataSourcePackage.isSupported(manifest)` is
used by `checkAdded`, `tryRead` and `NativeDataSourceRegistry.lookup`.
--
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]
