sunchao commented on code in PR #5763:
URL: https://github.com/apache/datafusion-comet/pull/5763#discussion_r4000202635
##########
spark/src/main/scala/org/apache/comet/serde/operator/NativeWriteUtils.scala:
##########
@@ -51,4 +67,166 @@ object NativeWriteUtils {
.setScan(scan.build())
.build())
}
+
+ /**
+ * ASCII characters the native URL parser rewrites inside a path. Determined
against the locked
+ * `url` 2.5 crate by parsing `hdfs://ns/pre<c>post/output` for every
printable ASCII `c` and
+ * comparing `url.path()` with the input: these nine are rewritten and the
rest survive,
+ * including `%`, `[`, `\`, `]`, `^` and `|`. Control characters and DEL are
handled separately
+ * in [[needsNativeUrlEscaping]] rather than listed here.
+ *
+ * `native/core/src/parquet/parquet_support.rs` has a
`url_path_rewritten_characters` test that
+ * fails if a `url` upgrade changes this set, so the two cannot drift apart
silently.
+ *
+ * `?` and `#` are the worst of the nine: they are not escaped but treated
as delimiters, so the
+ * native path is *truncated* there rather than merely spelled differently.
+ */
+ private val nativeUrlEscapedAscii: Set[Char] =
+ Set(' ', '"', '#', '<', '>', '?', '`', '{', '}')
+
+ /**
+ * Whether the native URL parser would rewrite `s`, so that the name it
creates on HDFS differs
+ * from the Hadoop filename Spark commits.
+ *
+ * `percent_encoding`'s `should_percent_encode` is `!byte.is_ascii() ||
set.contains(byte)`, so
+ * every non-ASCII byte is escaped regardless of the encode set. That is the
case Java's URI
+ * comparison cannot see, because `java.net.URI` leaves non-ASCII path
characters alone: a name
+ * holding U+00E9 comes back identically from `getRawPath` and `getPath`,
while the native
+ * parser produces `caf%C3%A9`.
+ */
+ private def needsNativeUrlEscaping(s: String): Boolean =
+ s.exists(c => c < ' ' || c > '~' || nativeUrlEscapedAscii.contains(c))
+
+ /**
+ * Whether the native writer and Spark would spell `path` differently, and
how.
+ *
+ * The native side receives `Path.toString`, which is not a URI string, and
reaches HDFS through
+ * `create_hdfs_object_store`, which hands `url.path()` -- now escaped by
the Rust parser -- to
+ * `object_store::path::Path::parse`. So the native writer creates a
directory literally called
+ * `dir%20with%20space`, or `caf%C3%A9` for a name holding U+00E9. Spark's
committer, meanwhile,
+ * works with the unescaped Hadoop `Path` and commits `dir with space`, or
that same U+00E9 name
+ * unescaped. Job commit then succeeds while the data sits somewhere else,
which is worse than
+ * not accelerating the write.
+ *
+ * Two conditions, because neither covers the other:
+ *
+ * - the native parser escapes a character, which is the direct statement
of the divergence
+ * and the only condition that catches non-ASCII names;
+ * - `java.net.URI` had to escape something, which catches a literal `%`
in the Hadoop name.
+ * The native parser leaves `%` alone, so `50%off` reaches `Path::parse`
as an invalid
+ * escape rather than as a rewritten name.
+ *
+ * Local `file:` destinations are unaffected and deliberately not gated
here: they go through a
+ * different object-store constructor that keeps the string verbatim.
+ */
+ private def hdfsPathDivergence(path: String): Option[String] = {
+ if (!path.startsWith("hdfs:")) return None
+ val uri = new Path(path).toUri
+ val raw = uri.getRawPath
+ val decoded = uri.getPath
+ val javaEscaped = raw != null && decoded != null && raw != decoded
+ // Checked against the string handed to the native writer, which is `path`
itself.
+ if (javaEscaped || needsNativeUrlEscaping(path)) {
+ Some(if (decoded != null) decoded else path)
+ } else {
+ None
+ }
+ }
+
+ /**
+ * A fallback reason when a write to `outputPath` would land somewhere the
committer is not
+ * looking, or `None` when the write can proceed.
+ *
+ * Two things go into every committed file name, and the native writer has
to reproduce both
+ * byte for byte (see [[hdfsPathDivergence]] for why it may not):
+ *
+ * - the destination directory, and
+ * - `fileNamePrefix`, the basename every file name is built from. On
Spark 4.0+ that is
+ * `mapreduce.output.basename`, which
`HadoopMapReduceCommitProtocol.getFilename`
+ * interpolates into `<basename>-<split>-<jobId>`; on 3.x Comet names
the files itself and
+ * the basename is always the literal `part`. A basename holding `?` or
`#` is the dangerous
+ * one: the native URL parser truncates there, so *every* task writes a
file with the same
+ * truncated name and they overwrite each other during commit.
+ */
+ def escapedHdfsDestination(outputPath: String, fileNamePrefix: String):
Option[String] = {
+ if (!outputPath.startsWith("hdfs:")) return None
+ hdfsPathDivergence(outputPath)
+ .map(shown =>
+ "HDFS output paths needing URI escaping are not supported: the native
writer would " +
+ s"write to the escaped path while Spark commits the unescaped one
($shown)")
+ .orElse {
+ if (needsNativeUrlEscaping(fileNamePrefix)) {
Review Comment:
[P2] Fall back for percent-bearing HDFS basenames during planning
Could we apply the Java URI-escaping check to `fileNamePrefix` as well?
`needsNativeUrlEscaping` deliberately excludes `%`, so
`mapreduce.output.basename=part%foo` or `part%25` is accepted here. Once Spark
includes it in the task filename, `hdfsPathDivergence` detects `getRawPath !=
getPath` and `checkNativeWriteDestination` aborts the job instead of falling
back.
I reproduced both cases with a native scan over 100 rows in two partitions
on Spark 4.1.3 and a real MiniDFS cluster, using native and JVM code built from
`5f5397f5`. The Spark-writer controls preserve all 100 IDs. With native writing
enabled, both plans contain `CometWriteFiles` and fail with
`UnsupportedOperationException` at the complete-path guard. The ordinary `part`
control writes natively, and the fixed `?`/`#` cases fall back successfully.
The basename is already available during planning. Please reject its
Java-escaped forms here too, and add `%` basename cases to the admission and
fallback regression coverage. The existing basename test currently covers `?`,
`#`, spaces and Unicode, but no literal percent.
--
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]
