dongjoon-hyun commented on code in PR #59947:
URL: https://github.com/apache/spark/pull/59947#discussion_r4433063716
##########
sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala:
##########
@@ -339,7 +451,8 @@ class ArtifactManager(session: SparkSession) extends
AutoCloseable with Logging
artifact.path,
d.path,
fragment = None,
- deleteStagedFile = false)
+ deleteStagedFile = false,
+ sourceName = Some(d.path.toString))
Review Comment:
**11. [Nit] A local file no longer goes through the public `addArtifact`**
This follows item 11 of my first review
([thread](https://github.com/apache/spark/pull/59947#discussion_r4420253942)).
Before this PR, `addLocalArtifacts` called the public `addArtifact` for a
`LocalFile` too. Now a `LocalFile` goes to the private overload, while an
in-memory artifact (L462) and `SparkConnectAddArtifactsHandler` still call the
public one. `ArtifactManager` is a public class with `protected` members, such
as `artifactRootPath`, so a subclass that overrides the public `addArtifact`,
for example to audit what a session adds, would no longer see the jars, files
and packages that `spark.addArtifact(path)` adds, and nothing would tell it.
If such a subclass is not a case to keep, this is fine. Otherwise, could
`addLocalArtifacts` use the overload only for a native data source package, the
only artifact whose errors use `sourceName`, and the public `addArtifact` for
the others?
##########
docs/sql-native-data-sources.md:
##########
@@ -325,12 +325,26 @@ 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 file that
+is not a valid package, or for a package whose version Spark does not support,
or that has no
+library for the platform of the driver, which loads the library to plan a
query, instead of adding a
+package that the session could not use: the session would ignore it and
silently keep using the
+package or library that it used before, or fail every query on its data
sources. After the failure,
+the session keeps the packages that it had, and a call that adds several
artifacts in Scala or Java
+still adds the others. In Python, a rejected package stops the rest of the
call: without Spark
+Connect, `spark.addArtifacts` adds the packages first, one at a time, so only
the packages before it
+are added; with Spark Connect, it adds the paths one at a time, so the paths
before the package are
+added. Spark does not load the library when it checks a package, so a library
that fails to load
+fails the queries that use it. Nor does it check that the driver can run
native data sources at all,
+which requires Java 22 or later and a platform where the C data interface of
Arrow Java loads:
+there, the add succeeds, and the queries fail with
`NATIVE_DATA_SOURCE_UNSUPPORTED_JAVA_VERSION` or
Review Comment:
**10. [Low] On Linux with musl, the add fails, unlike what the docs say**
This sentence, which came with item 5
([thread](https://github.com/apache/spark/pull/59947#discussion_r4424226338)),
says that where the C data interface of Arrow Java does not load, the add
succeeds and the queries fail with `NATIVE_DATA_SOURCE_UNSUPPORTED_PLATFORM`.
That holds for the version of Java, but usually not for the platform:
`checkAdded` rejects a package without a library for `NativePlatform.current`,
and on these platforms a package usually has none.
- On Linux with musl, `NativePlatform.current` is `linux-<arch>-musl`, such
as `linux-x86_64-musl` ([NativeDataSourcePackage.scala
L379-L383](https://github.com/apache/spark/blob/465dbfc06f5ec1d2abae1784d7c6817690404895/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourcePackage.scala#L379-L383)),
where, as L47-L50 say, a query fails with
`NATIVE_DATA_SOURCE_UNSUPPORTED_PLATFORM`. So a package built for
`linux-x86_64` fails at the add with
`INVALID_NATIVE_DATA_SOURCE_PACKAGE.UNSUPPORTED_PLATFORM`, not at the query.
- On Linux on ppc64le or s390x, the same happens unless the package has a
library for `linux-ppc64le` or `linux-s390x`.
So there, the add only succeeds for a package with a library for the
platform of the driver. The doc of `checkAdded`
(NativeDataSourcePackage.scala:144-146) and the description say the same as
this sentence. Could all three say that the add succeeds there only if the
package has a library for that platform?
##########
sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala:
##########
@@ -204,6 +210,96 @@ 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
+ * compared with the artifact first, so that adding the same file again does
not copy it. It is
+ * then 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 = {
+ def exists(source: Path): Boolean = {
+ // Utils.contentEquals fails for a directory, which cannot be the same
as a package anyway,
+ // so a directory with the name of an artifact, or an artifact that is a
directory, such
+ // as files/x.sparkpkg after files/x.sparkpkg/y was added, fails as an
existing artifact.
+ if (Files.exists(target) && (Files.isDirectory(source) ||
Files.isDirectory(target))) {
+ throw artifactAlreadyExistsError(normalizedRemoteRelativePath)
+ }
+ artifactExists(normalizedRemoteRelativePath, source, target)
+ }
+ if (!deleteStagedFile && exists(serverLocalStagingPath)) {
Review Comment:
**12. [Nit] A staged package could also be compared with the artifact first**
This follows items 6 and 8
([thread](https://github.com/apache/spark/pull/59947#discussion_r4424226345),
[thread](https://github.com/apache/spark/pull/59947#discussion_r4424226366)).
Since item 6, a local file is compared with the artifact here, before the
check, and again right before the transfer. So for a local file,
`ARTIFACT_ALREADY_EXISTS` comes first through this comparison, and the `Try`
only matters when another add of the same name races with this one. A staged
file could do the same:
1. `if (exists(serverLocalStagingPath)) return false` for both kinds,
2. `checkAdded` on `staged`, without `Try`,
3. `exists(staged)` right before the transfer, as now.
The `Try` that I suggested in item 8, `check.get`, the comment at L273-L274
and the `!deleteStagedFile &&` would then go. A Spark Connect client that sends
a package again would also not have the zip opened for a check whose result is
dropped. The only difference is an add that races with another add of the same
name: the error of its check would come before `ARTIFACT_ALREADY_EXISTS`. If
that is fine, could both kinds share the first comparison?
##########
sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala:
##########
@@ -204,6 +210,96 @@ 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
+ * compared with the artifact first, so that adding the same file again does
not copy it. It is
+ * then 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 = {
+ def exists(source: Path): Boolean = {
Review Comment:
**13. [Nit] `exists` and `artifactExists` read as checks, but throw**
Both return a `Boolean`, but both also throw `ARTIFACT_ALREADY_EXISTS` when
another file is at `target` (L226, L259). At the call sites, `!deleteStagedFile
&& exists(serverLocalStagingPath)` (L263) and `if (exists(staged))` (L277),
nothing shows that the condition can throw, or, at L277, that its error comes
before the error of the check, which is the order that L273-L274 describe.
Could the names, or a short comment at these calls, say that they fail for
another file?
--
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]
