andygrove commented on code in PR #6940:
URL: https://github.com/apache/datafusion-comet/pull/6940#discussion_r4314196211
##########
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:
I am keeping the scope. Non-magic S3A writes stay on Spark until streaming a
file through S3A has been benchmarked against uploading it natively (#6944), so
the reason is scope, not correctness, and your point that the old wording no
longer describes it is right. The doc comment on `unsupportedFileSystem` and
the last test of `ParquetWriteWithMagicCommitterSuite` now say that.
(65e2bfdfe1)
##########
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:
Dropped. Spark's job configuration always has a name from
`core-default.xml`, and `S3ACommitterFactory` reads the job's first, so a
bucket's own name never decides it. The doc comment of `usesS3AMagicCommitter`
says so. (65e2bfdfe1)
##########
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:
Replaced. The `Configuration(false)` bucket assertions are gone with the
fallback. A `new Configuration()` case now asserts that the default name is
`file`, that a bucket's `magic` does not win over it, and that a global `magic`
name does, including against a bucket's `file`. (65e2bfdfe1)
##########
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:
Done. The full explanation is now in the `HadoopFileOutput` class doc, and
the other places refer to it: both exec scaladocs, `writeFile`, the
`NativeWriteUtils` docs, `parquet_writer.rs`, and the test comments.
`operators.md` keeps only the user-facing sentence. The proto comment now says
that `CometWriteFilesExec.writeFile` sets the flag, and that both write
operators call it. (65e2bfdfe1)
--
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]
