viirya commented on code in PR #6940:
URL: https://github.com/apache/datafusion-comet/pull/6940#discussion_r4304262258
##########
spark/src/main/scala/org/apache/comet/serde/operator/NativeWriteUtils.scala:
##########
@@ -187,6 +189,119 @@ object NativeWriteUtils {
"spark.comet.write.parquet.enabled=false to write this table with
Spark.")
}
+ /**
+ * Whether the native writer creates the file at `filePath` itself, which it
does for local and
+ * HDFS files. Any other file is created through Hadoop's FileSystem API, as
Spark's own writer
+ * creates it, and the native writer only writes its bytes. That is always
correct, whatever the
+ * file system and whatever the committer expects of it, which is why this
is decided per task
+ * on the path the commit protocol actually returned rather than at planning.
+ */
+ def nativeWriterCreatesFile(filePath: String): Boolean =
+ filePath.startsWith("file:") || filePath.startsWith("hdfs:")
+
+ /**
+ * A fallback reason when the native writer cannot write to `outputPath`'s
file system, or
+ * `None` when it can. Local and HDFS files are created natively. A write
that S3A's magic
+ * committer commits is supported too, since each of its files is created
through S3A (see
+ * `CometWriteFilesExec.writeFile`), which is the only way to give the
committer the pending
+ * upload it completes. Other S3 writes stay on Spark until the native
writer reproduces what
+ * S3A does at create time.
+ *
+ * @param hadoopConf
+ * the write's Hadoop configuration, only built for a destination the
native writer does not
+ * create files in
+ */
+ def unsupportedFileSystem(
+ outputPath: String,
+ commitProtocolClass: String,
+ hadoopConf: => Configuration): Option[String] =
+ if (nativeWriterCreatesFile(outputPath) ||
+ usesS3AMagicCommitter(outputPath, commitProtocolClass, hadoopConf)) {
+ None
+ } else {
+ Some("Supported output filesystems: local, HDFS, and S3 with the S3A
magic committer")
+ }
+
+ /** Spark's commit protocol for Hadoop's `PathOutputCommitter`s, in
spark-hadoop-cloud. */
+ val PathOutputCommitProtocolClass: String =
+ "org.apache.spark.internal.io.cloud.PathOutputCommitProtocol"
+
+ /**
+ * Spark's Parquet output committer that delegates to a
`PathOutputCommitter`, in
+ * spark-hadoop-cloud.
+ */
+ val BindingParquetOutputCommitterClass: String =
+ "org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter"
+
+ /** S3A's committer factory that always creates the magic committer. */
+ private val MagicCommitterFactoryClass =
+ "org.apache.hadoop.fs.s3a.commit.magic.MagicS3GuardCommitterFactory"
+
+ /** S3A's committer factory that creates the committer
`fs.s3a.committer.name` names. */
+ private val S3ACommitterFactoryClass =
"org.apache.hadoop.fs.s3a.commit.S3ACommitterFactory"
+
+ /**
+ * Whether S3A's magic committer commits a write to `outputPath`.
+ *
+ * Spark reaches Hadoop's committer factories through one of two bindings in
spark-hadoop-cloud.
+ * Spark 4.1 configures both whenever that module is on the classpath
(SPARK-47618), and earlier
+ * versions for a bucket that sets
`fs.s3a.bucket.<bucket>.committer.magic.enabled`:
+ *
+ * - the commit protocol `PathOutputCommitProtocol`, or
+ * - `BindingParquetOutputCommitter` as
`spark.sql.parquet.output.committer.class`, under a
+ * commit protocol that instantiates that class, as Spark's default one
does.
+ *
+ * Either asks Hadoop's `PathOutputCommitterFactory` for the committer. Its
factory is
+ * `mapreduce.outputcommitter.factory.class` when set, else
+ * `mapreduce.outputcommitter.factory.scheme.<scheme>`. The committer is the
magic committer
+ * when that factory is `MagicS3GuardCommitterFactory`, or when it is
`S3ACommitterFactory` and
+ * the committer it is asked for is `magic`: `fs.s3a.committer.name` in the
job's configuration,
+ * or failing that in the file system's, where
`fs.s3a.bucket.<bucket>.committer.name` replaces
+ * it.
+ *
+ * Only planning asks, to decide whether a write is worth running natively.
A wrong answer
+ * cannot misplace data: each task writes the path the committer actually
returns, and creates
+ * any file that is not local or on HDFS through Hadoop, see
`CometWriteFilesExec.writeFile`.
+ *
+ * @param commitProtocolClass
+ * the write's `spark.sql.sources.commitProtocolClass`
+ * @param hadoopConf
+ * the write's Hadoop configuration, which carries
`spark.sql.parquet.output.committer.class`
+ * as Spark's Parquet writer reads it
+ */
+ def usesS3AMagicCommitter(
+ outputPath: String,
+ commitProtocolClass: String,
+ hadoopConf: Configuration): Boolean = {
+ val outputCommitterClass =
+ hadoopConf.getTrimmed(SQLConf.PARQUET_OUTPUT_COMMITTER_CLASS.key, "")
+ val asksCommitterFactory = commitProtocolClass ==
PathOutputCommitProtocolClass ||
+ (commitProtocolClass ==
SQLConf.FILE_COMMIT_PROTOCOL_CLASS.defaultValueString &&
+ outputCommitterClass == BindingParquetOutputCommitterClass)
+ val uri = new Path(outputPath).toUri
+ if (!asksCommitterFactory || uri.getScheme == null) {
+ return false
+ }
+
+ val factory = Seq(
+ PathOutputCommitterFactory.COMMITTER_FACTORY_CLASS,
+
PathOutputCommitterFactory.COMMITTER_FACTORY_SCHEME_PATTERN.format(uri.getScheme))
+ .map(key => hadoopConf.getTrimmed(key, ""))
+ .find(_.nonEmpty)
+ factory match {
+ case Some(MagicCommitterFactoryClass) => true
+ case Some(S3ACommitterFactoryClass) =>
+ // S3ACommitterFactory takes the name from the job's configuration
when it has one, and
Review Comment:
`core-default.xml` sets `fs.s3a.committer.name=file` in both 3.3.4 and
3.4.1, so with a defaults-loaded configuration this lookup always finds a
job-level name, and the bucket fallback never applies. That matches
`S3ACommitterFactory`, which also reads the task conf first, so it isn't wrong.
But Spark never reaches this branch. Could it be dropped?
##########
spark/src/main/scala/org/apache/comet/serde/operator/NativeWriteUtils.scala:
##########
@@ -187,6 +189,119 @@ object NativeWriteUtils {
"spark.comet.write.parquet.enabled=false to write this table with
Spark.")
}
+ /**
+ * Whether the native writer creates the file at `filePath` itself, which it
does for local and
+ * HDFS files. Any other file is created through Hadoop's FileSystem API, as
Spark's own writer
+ * creates it, and the native writer only writes its bytes. That is always
correct, whatever the
+ * file system and whatever the committer expects of it, which is why this
is decided per task
+ * on the path the commit protocol actually returned rather than at planning.
+ */
+ def nativeWriterCreatesFile(filePath: String): Boolean =
+ filePath.startsWith("file:") || filePath.startsWith("hdfs:")
+
+ /**
+ * A fallback reason when the native writer cannot write to `outputPath`'s
file system, or
+ * `None` when it can. Local and HDFS files are created natively. A write
that S3A's magic
+ * committer commits is supported too, since each of its files is created
through S3A (see
+ * `CometWriteFilesExec.writeFile`), which is the only way to give the
committer the pending
+ * upload it completes. Other S3 writes stay on Spark until the native
writer reproduces what
Review Comment:
The per-task routing makes every S3A-served write correct. Staging
committers hand out local `file:` paths, which the native writer creates
itself, and the `file` committer's `_temporary` paths are created through
Hadoop. So this gate now only decides scope, and "until the native writer
reproduces what S3A does at create time" no longer describes why other S3
writes fall back. The same reason appears in the last test of
`ParquetWriteWithMagicCommitterSuite`.
Since #6355 hasn't landed, the alternative for a non-magic S3A write today
is Spark's writer, not `object_store`. Would admitting every S3A destination be
the simpler choice? It would drop `usesS3AMagicCommitter` and its test, which
mirror Hadoop's committer resolution and could drift as Hadoop changes. If
you'd rather keep the scope, could the comments here and in the suite say the
real reason (benchmarks against #6355)?
##########
spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala:
##########
@@ -1997,6 +2002,177 @@ class CometParquetWriterSuite extends
CometParquetWriterTestBase {
}
}
+ /**
+ * Runs `write` with a source of `rows` rows in `partitions` partitions,
read natively, and the
+ * path to write it to, while RecordingFileSystemCommitProtocol puts the
write's task files on
+ * [[RecordingFileSystem]], whose recordings start empty and are cleared
afterwards.
+ */
+ private def withRecordingFileSystem(rows: Int, partitions: Int)(
+ write: (DataFrame, String) => Unit): Unit =
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+ val sourcePath = new File(dir, "source.parquet").getAbsolutePath
+ withNativeWriter {
+ val df = materializeAsCometSource(
+ (1 to rows).map(i => (i, s"str_$i")).toDF("id",
"name").repartition(partitions),
+ sourcePath)
+ RecordingFileSystem.reset()
+ try {
+ withSQLConf(
+ SQLConf.FILE_COMMIT_PROTOCOL_CLASS.key ->
+ classOf[RecordingFileSystemCommitProtocol].getName,
+ s"fs.${RecordingFileSystem.SCHEME}.impl" ->
classOf[RecordingFileSystem].getName) {
+ write(df, outputPath)
+ }
+ } finally {
+ RecordingFileSystem.reset()
+ }
+ }
+ }
+
+ test("a task file the native writer cannot create is created through
Hadoop") {
+ // The native writer creates only local and HDFS files. Every other file,
such as the one
+ // S3A's magic committer hands out under its `__magic` directory, has to
be created through
+ // Hadoop's FileSystem API so the FileSystem can do whatever it does at
create time, and the
+ // native writer only supplies the bytes.
RecordingFileSystemCommitProtocol puts task files on
+ // a scheme the native writer cannot open at all, so the write can only
succeed that way.
+ withRecordingFileSystem(rows = 1000, partitions = 3) { (df, outputPath) =>
+ val plan = captureWritePlan(p => df.write.parquet(p), outputPath)
+ assertHasCometNativeWriteExec(plan)
+
+ val created = RecordingFileSystem.created.asScala.toSeq
+ assert(created.size == 3, s"Expected one file per task, got $created")
+ assert(created.forall(_.startsWith(RecordingFileSystem.PREFIX)), created)
+ // Closed, which is where a FileSystem such as S3A completes or records
the upload, and
+ // only once the native writer had written all of it.
+ assert(RecordingFileSystem.closed.asScala.toSet == created.toSet)
+ assert(RecordingFileSystem.aborted.isEmpty)
+
+ checkAnswer(spark.read.parquet(outputPath), df)
+ assert(listPartFileNames(outputPath).size == 3)
+ assert(new File(outputPath, "_SUCCESS").exists())
+ }
+ }
+
+ test("a failed write through Hadoop aborts the file and fails with what its
stream threw") {
+ // A file the JVM created is discarded when the native write fails:
aborted, so that S3A
+ // abandons the upload rather than completing a truncated file. The native
writer fails the
+ // task with the exception the stream threw, which the JVM rethrows as it
was, as a failure in
+ // Spark's own writer would surface.
+ withRecordingFileSystem(rows = 100, partitions = 1) { (df, outputPath) =>
+ RecordingFileSystem.failWrites = true
+ val (plan, error) = captureFailedWritePlan(df.write.parquet(outputPath))
+ assertHasCometNativeWriteExec(plan)
+ assert(
+ causeChain(error).exists {
+ case e: IOException => e.getMessage ==
RecordingFileSystem.WRITE_FAILURE
+ case _ => false
+ },
+ s"Expected the stream's own IOException, got: $error")
+
+ assert(RecordingFileSystem.created.size == 1)
+ assert(RecordingFileSystem.aborted.asScala ==
RecordingFileSystem.created.asScala)
+ assert(RecordingFileSystem.closed.isEmpty)
+ assert(
+ listPartFileNames(outputPath).isEmpty,
+ s"A failed write must not leave part files:
${listPartFileNames(outputPath)}")
+ assert(!new File(outputPath, "_temporary").exists())
+ }
+ }
+
+ test("S3 writes plan natively when S3A's magic committer commits them") {
+ // Planning admits an S3 destination only when the magic committer will
commit it, as
+ // `PathOutputCommitterFactory` and `S3ACommitterFactory` resolve the
committer. Its tasks
+ // then have S3A create each file. Other S3 writes stay on Spark's writer.
+ val pathOutput = NativeWriteUtils.PathOutputCommitProtocolClass
+ val defaultProtocol = SQLConf.FILE_COMMIT_PROTOCOL_CLASS.defaultValueString
+ val bindingCommitter =
+ SQLConf.PARQUET_OUTPUT_COMMITTER_CLASS.key ->
NativeWriteUtils.BindingParquetOutputCommitterClass
+ val s3aFactory = S3ACommitterFactory.CLASSNAME
+ val magicFactory = MagicS3GuardCommitterFactory.CLASSNAME
+ // Main code names spark-hadoop-cloud's classes rather than linking them.
+ assert(pathOutput == classOf[PathOutputCommitProtocol].getName)
+ assert(
+ NativeWriteUtils.BindingParquetOutputCommitterClass ==
+ classOf[BindingParquetOutputCommitter].getName)
+ val s3aSchemeFactory = "mapreduce.outputcommitter.factory.scheme.s3a" ->
s3aFactory
+ val magicName = "fs.s3a.committer.name" -> "magic"
+
+ def usesMagic(protocol: String, settings: (String, String)*)(
+ path: String = "s3a://bucket/warehouse/table"): Boolean = {
+ val conf = new Configuration(false)
+ settings.foreach { case (key, value) => conf.set(key, value) }
+ NativeWriteUtils.usesS3AMagicCommitter(path, protocol, conf)
+ }
+
+ // What Spark 4.1 configures when spark-hadoop-cloud is on the classpath
(SPARK-47618).
+ assert(usesMagic(pathOutput, s3aSchemeFactory, magicName,
bindingCommitter)())
+ // Either binding asks the committer factory on its own.
+ assert(usesMagic(pathOutput, s3aSchemeFactory, magicName)())
+ assert(usesMagic(defaultProtocol, s3aSchemeFactory, magicName,
bindingCommitter)())
+ // Without one, the committer is Spark's own FileOutputCommitter, whatever
S3A's settings.
+ assert(!usesMagic(defaultProtocol, s3aSchemeFactory, magicName)())
+ // A protocol Comet does not know may never instantiate the binding
committer.
+ assert(
+ !usesMagic("com.example.CommitProtocol", s3aSchemeFactory, magicName,
bindingCommitter)())
+
+ // The global factory key wins over the destination scheme's, in either
direction.
+ val globalFactory = "mapreduce.outputcommitter.factory.class"
+ assert(usesMagic(pathOutput, globalFactory -> magicFactory)())
+ assert(
+ !usesMagic(
+ pathOutput,
+ globalFactory ->
"org.apache.hadoop.mapreduce.lib.output.FileOutputCommitterFactory",
+ s3aSchemeFactory,
+ magicName)())
+ // The scheme key is the destination's own, for any scheme S3A serves.
+ assert(!usesMagic(pathOutput, s3aSchemeFactory,
magicName)("s3://bucket/warehouse/table"))
+ assert(
+ usesMagic(
+ pathOutput,
+ "mapreduce.outputcommitter.factory.scheme.s3" -> s3aFactory,
+ magicName)("s3://bucket/warehouse/table"))
+ // No factory means FileOutputCommitterFactory.
+ assert(!usesMagic(pathOutput, magicName)())
+
+ // MagicS3GuardCommitterFactory creates the magic committer whatever the
name says, while
+ // S3ACommitterFactory creates the one named, matched as exactly as S3A
matches it.
+ assert(
+ usesMagic(pathOutput, "mapreduce.outputcommitter.factory.scheme.s3a" ->
magicFactory)())
+ Seq("file", "directory", "partitioned", "staging", "Magic", "").foreach {
name =>
+ assert(
+ !usesMagic(pathOutput, s3aSchemeFactory, "fs.s3a.committer.name" ->
name)(),
+ s"committer name '$name'")
+ }
+ // A bucket's committer name applies when the job sets none of its own,
and a global name in
Review Comment:
These cases build `new Configuration(false)`, so "a bucket's committer name
applies when the job sets none" describes a setup Spark never has. Spark's
Hadoop conf carries `fs.s3a.committer.name=file` from `core-default.xml` and,
once `JobConf` has loaded, the `s3a` scheme factory from `mapred-default.xml`.
Could you add a case with `new Configuration()` that pins the behavior Spark
actually sees, or drop these assertions together with the fallback?
##########
native/proto/src/proto/operator.proto:
##########
@@ -1016,6 +1016,13 @@ message ParquetWriter {
// Catalyst's target schema, including top-level nullability and Parquet
field IDs.
// Nested collection field IDs are carried by the corresponding DataType
messages.
repeated SparkStructField output_schema = 9;
+ // Whether the JVM has already created `output_path` through Hadoop's
FileSystem API and
+ // registered its output stream with the plan (Native.setOutputStream). The
native writer then
+ // writes the file's bytes to that stream instead of opening the path
itself, and the JVM closes
+ // the file afterwards. Set per task by CometWriteFilesExec and
CometNativeWriteExec for every
Review Comment:
Only `CometWriteFilesExec.writeFile` sets this flag. `CometNativeWriteExec`
reaches it through that helper. More generally, the explanation of why the
magic committer needs S3A to create the file now appears in about eight places:
here, `HadoopFileOutput`, both exec scaladocs, `writeFile`, `NativeWriteUtils`,
`parquet_writer.rs` and `operators.md`. Could it live once, in
`HadoopFileOutput` or `writeFile`, with the other places referring to it? This
sentence shows the copies are already drifting.
--
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]
