parthchandra commented on code in PR #6940: URL: https://github.com/apache/datafusion-comet/pull/6940#discussion_r4342237703
########## spark/src/test/java/org/apache/comet/hadoop/fs/RecordingFileSystem.java: ########## @@ -0,0 +1,175 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.comet.hadoop.fs; + +import java.io.IOException; +import java.io.OutputStream; +import java.net.URI; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.apache.hadoop.fs.Abortable; +import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.RawLocalFileSystem; +import org.apache.hadoop.fs.StreamCapabilities; +import org.apache.hadoop.util.Progressable; + +/** + * A local-disk-backed FileSystem under its own scheme that records every file created through it + * and how each file's stream ended, and can fail writes on demand. Its streams can abort, as S3A's + * do, and an aborted file is deleted, as an aborted S3A upload never becomes an object. + * + * <p>It lets a test check that a file was created through Hadoop's FileSystem API without a remote + * store: Comet's native writer creates only local and HDFS files itself. The recordings are static, + * since FileSystem instances are cached and the tests run in local mode. + */ +public class RecordingFileSystem extends RawLocalFileSystem { + + public static final String SCHEME = "comet-recording"; + + public static final String PREFIX = SCHEME + "://recording"; + + /** The message of the {@link IOException} every write throws while {@link #failWrites} is set. */ + public static final String WRITE_FAILURE = "injected output stream failure"; + + /** The files created through this FileSystem, in order. */ + public static final List<String> created = new CopyOnWriteArrayList<>(); + + /** The files whose stream was closed. */ + public static final List<String> closed = new CopyOnWriteArrayList<>(); + + /** The files whose stream was aborted. */ + public static final List<String> aborted = new CopyOnWriteArrayList<>(); + + public static volatile boolean failWrites = false; + + public static void reset() { + created.clear(); + closed.clear(); + aborted.clear(); + failWrites = false; + } + + public RecordingFileSystem() { + // Avoid `URI scheme is not "file"` error on + // RawLocalFileSystem$DeprecatedRawLocalFileStatus.getOwner + RawLocalFileSystem.useStatIfAvailable(); + } + + @Override + public String getScheme() { + return SCHEME; + } + + @Override + public URI getUri() { + return URI.create(PREFIX); + } + + @Override + public FSDataOutputStream create( + Path f, + boolean overwrite, + int bufferSize, + short replication, + long blockSize, + Progressable progress) + throws IOException { + return record(f, super.create(f, overwrite, bufferSize, replication, blockSize, progress)); + } + + private FSDataOutputStream record(Path path, FSDataOutputStream out) { + created.add(path.toString()); + return new FSDataOutputStream(new RecordingStream(path, out), statistics); + } + + private final class RecordingStream extends OutputStream + implements Abortable, StreamCapabilities { + private final Path path; + private final OutputStream out; + private boolean finished = false; + + RecordingStream(Path path, OutputStream out) { + this.path = path; + this.out = out; + } + + @Override + public void write(int b) throws IOException { + checkWritable(); + out.write(b); + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + checkWritable(); + out.write(b, off, len); + } + + private void checkWritable() throws IOException { + if (failWrites) { + throw new IOException(WRITE_FAILURE); + } + } + + @Override + public void close() throws IOException { Review Comment: Can we add a `failsClose` switch and a test? (With S3A, the last part upload and the .pending. files are written in `close()` so this might be where we will likely get most failures) We would expect the task to fail, `abortTask` to run and no part files to be left behind. ########## spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala: ########## @@ -1997,6 +2002,175 @@ 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 Review Comment: Could the output directory in `withRecordingFileSystem` have a name with a space, `%`, `#`, `?` and a non-ASCII character? The old object_store route had to decline paths like these because the JVM and native sides spelled them differently. This route builds a Hadoop `Path` from the string the committer returned, so I expect it to work. A test would show it and would stop a later change to `HadoopFileOutput.create` from breaking it. ########## spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala: ########## @@ -1997,6 +2002,175 @@ 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 is created through + // Hadoop's FileSystem API (see HadoopFileOutput) 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") { Review Comment: Could we add a test for a task that is retried on this route? For example, the first attempt fails after its file is created and the second attempt succeeds. We would expect one aborted file, one closed file, each row once in the output, and nothing left over. ########## spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala: ########## @@ -1997,6 +2002,175 @@ 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 is created through + // Hadoop's FileSystem API (see HadoopFileOutput) 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) => Review Comment: This writes rather small files. The 1MB chunking, the reused java array only get exercised when the file is bigger tahn 1 MB. Could we add a test with a file of a few MiB? For example, use many rows of random non-ASCII strings with `compression=none`, and check that the data reads back. It would also help to let `RecordingFileSystem` fail the Nth write. ########## spark/src/test/java/org/apache/comet/hadoop/fs/RecordingFileSystem.java: ########## @@ -0,0 +1,175 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.comet.hadoop.fs; + +import java.io.IOException; +import java.io.OutputStream; +import java.net.URI; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.apache.hadoop.fs.Abortable; +import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.RawLocalFileSystem; +import org.apache.hadoop.fs.StreamCapabilities; +import org.apache.hadoop.util.Progressable; + +/** + * A local-disk-backed FileSystem under its own scheme that records every file created through it + * and how each file's stream ended, and can fail writes on demand. Its streams can abort, as S3A's + * do, and an aborted file is deleted, as an aborted S3A upload never becomes an object. + * + * <p>It lets a test check that a file was created through Hadoop's FileSystem API without a remote + * store: Comet's native writer creates only local and HDFS files itself. The recordings are static, + * since FileSystem instances are cached and the tests run in local mode. + */ +public class RecordingFileSystem extends RawLocalFileSystem { + + public static final String SCHEME = "comet-recording"; + + public static final String PREFIX = SCHEME + "://recording"; + + /** The message of the {@link IOException} every write throws while {@link #failWrites} is set. */ + public static final String WRITE_FAILURE = "injected output stream failure"; + + /** The files created through this FileSystem, in order. */ + public static final List<String> created = new CopyOnWriteArrayList<>(); + + /** The files whose stream was closed. */ + public static final List<String> closed = new CopyOnWriteArrayList<>(); + + /** The files whose stream was aborted. */ + public static final List<String> aborted = new CopyOnWriteArrayList<>(); + + public static volatile boolean failWrites = false; + + public static void reset() { + created.clear(); + closed.clear(); + aborted.clear(); + failWrites = false; + } + + public RecordingFileSystem() { + // Avoid `URI scheme is not "file"` error on + // RawLocalFileSystem$DeprecatedRawLocalFileStatus.getOwner + RawLocalFileSystem.useStatIfAvailable(); + } + + @Override + public String getScheme() { + return SCHEME; + } + + @Override + public URI getUri() { + return URI.create(PREFIX); + } + + @Override + public FSDataOutputStream create( + Path f, + boolean overwrite, + int bufferSize, + short replication, + long blockSize, + Progressable progress) + throws IOException { + return record(f, super.create(f, overwrite, bufferSize, replication, blockSize, progress)); + } + + private FSDataOutputStream record(Path path, FSDataOutputStream out) { + created.add(path.toString()); + return new FSDataOutputStream(new RecordingStream(path, out), statistics); + } + + private final class RecordingStream extends OutputStream + implements Abortable, StreamCapabilities { + private final Path path; + private final OutputStream out; + private boolean finished = false; + + RecordingStream(Path path, OutputStream out) { + this.path = path; + this.out = out; + } + + @Override + public void write(int b) throws IOException { + checkWritable(); + out.write(b); + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + checkWritable(); + out.write(b, off, len); + } + + private void checkWritable() throws IOException { + if (failWrites) { + throw new IOException(WRITE_FAILURE); + } + } + + @Override + public void close() throws IOException { Review Comment: Also, `hasCapability` always has the stream as abortable. Can we also have a switch to turn that off? ########## spark/src/main/scala/org/apache/spark/sql/comet/CometWriteFilesExec.scala: ########## @@ -318,25 +327,77 @@ object CometWriteFilesExec extends Logging { } }) - val execIterator = CometExec.getCometIterator( + writeFile( + filePath, + hadoopConf, + taskWrite.nativeOp.toBuilder.setParquetWriter(parquetWriter).build(), CometArrowStream.inputObjects(trackedBatches, taskWrite.childSchema, taskWrite.nodeName), taskWrite.dataColumnNames.length, - taskOp, taskWrite.nativeMetrics, taskWrite.numPartitions, - partitionId, - broadcastedHadoopConfForEncryption = None, - encryptedFilePaths = Seq.empty) - - // `close()` propagates teardown failures, including from the final native metrics update, so - // a plain `finally` would let a cleanup error replace the write error that caused it. - Utils.tryWithSafeFinally { - // The native writer emits no batches; draining performs the write. - while (execIterator.hasNext) { - execIterator.next().close() + partitionId) + } + + /** + * Writes the task file at `filePath` by running `nativeOp`, a native write plan whose Parquet + * writer is set up for the task except for the file, over `inputs`. Both native writers write + * their files with this: CometWriteFilesExec on Spark 4.0+ and CometNativeWriteExec on 3.x. + * + * When the native writer cannot create the file itself (see + * [[NativeWriteUtils.nativeWriterCreatesFile]]), it is created here first, through Hadoop's + * FileSystem API with the task's configuration, as [[HadoopFileOutput]] explains, and the + * plan's writer writes the file's bytes to it. The file is closed once the plan has written all + * of it, and aborted if the plan fails. + */ + private[comet] def writeFile( + filePath: String, + hadoopConf: Configuration, + nativeOp: Operator, + inputs: Array[Object], + numOutputCols: Int, + nativeMetrics: CometMetricNode, + numPartitions: Int, + partitionId: Int): Unit = { + def run(output: Option[HadoopFileOutput]): Unit = { + val writer = nativeOp.getParquetWriter.toBuilder + .setOutputPath(filePath) + .setJvmOutputStream(output.isDefined) + .build() + val execIterator = new CometExecIterator( + CometExec.newIterId, + inputs, + numOutputCols, + CometExec.serializeNativePlan(nativeOp.toBuilder.setParquetWriter(writer).build()), + nativeMetrics, + numPartitions, + partitionId, + outputStream = output) + + // `close()` propagates teardown failures, including from the final native metrics update, + // so a plain `finally` would let a cleanup error replace the write error that caused it. + Utils.tryWithSafeFinally { + // The native writer emits no batches; draining performs the write. + while (execIterator.hasNext) { + execIterator.next().close() + } + } { + execIterator.close() + } + } + + if (NativeWriteUtils.nativeWriterCreatesFile(filePath)) { + run(None) + } else { + val file = HadoopFileOutput.create(filePath, hadoopConf) Review Comment: Spark's writer wraps a failed task write in `TASK_WRITE_FAILED` ("Task failed while writing rows to <path>"). It lets `FetchFailedException` through. In the default setup it turns `FileAlreadyExistsException` into `TaskOutputFileAlreadyExistException`, which tells Spark not to retry the task. See `FileFormatDataWriter.enrichWriteError`, and `FileFormatWriter.executeTask` on 3.x. `executeTask` here rethrows the raw error. This route makes the gap easier to hit. `HadoopFileOutput.create` uses `overwrite = false`, so it can now throw `FileAlreadyExistsException`. Spark would stop the job right away, and Comet will retry the task up to `spark.task.maxFailures` times. The test comment at `CometParquetWriterSuite.scala:2057` says the failure surfaces as it would in Spark's writer. That is not true for the error class. Is this intended? If it is out of scope here, could you open an issue and link it from `executeTask`? Otherwise a small wrapper with the same three cases would match Spark. ########## spark/src/test/scala/org/apache/comet/parquet/ParquetWriteWithMagicCommitterSuite.scala: ########## @@ -0,0 +1,228 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.comet.parquet + +import java.io.File + +import scala.jdk.CollectionConverters._ +import scala.util.Try + +import org.apache.hadoop.fs.Path +import org.apache.hadoop.fs.s3a.commit.files.SuccessData +import org.apache.parquet.hadoop.ParquetOutputCommitter +import org.apache.spark.SparkConf +import org.apache.spark.sql.DataFrame +import org.apache.spark.sql.comet.CometNativeWriteExec +import org.apache.spark.sql.execution.SparkPlan +import org.apache.spark.sql.execution.command.DataWritingCommandExec +import org.apache.spark.sql.functions.col +import org.apache.spark.sql.internal.SQLConf + +import org.apache.comet.CometS3TestBase +import org.apache.comet.serde.operator.NativeWriteUtils + +import software.amazon.awssdk.services.s3.model.ListMultipartUploadsRequest + +/** + * Native Parquet writes to S3 that S3A's magic committer commits, against MinIO. + * + * A manual suite like the other MinIO suites, since it needs Docker. Every session here gets the + * magic committer the way Spark configures it when spark-hadoop-cloud is on the classpath: by + * default on Spark 4.1+ (SPARK-47618), and for a bucket that asks for it on earlier versions. The + * shared test configuration turns that off for every other suite. + */ +class ParquetWriteWithMagicCommitterSuite + extends CometParquetWriterTestBase + with CometS3TestBase { + + import testImplicits._ + + override protected val testBucketName = "magic-committer" + + override protected def sparkConf: SparkConf = { + val conf = super.sparkConf + conf.set("spark.hadoop.fs.s3a.committer.magic.enabled", "true") + conf.set(s"spark.hadoop.fs.s3a.bucket.$testBucketName.committer.magic.enabled", "true") + conf + } + + /** The multipart uploads still pending in `bucket`, as their object keys. */ + private def pendingUploads(bucket: String): Seq[String] = withS3Client { client => + client + .listMultipartUploads(ListMultipartUploadsRequest.builder().bucket(bucket).build()) + .uploads() + .asScala + .map(_.key()) + .toSeq + } + + /** `rows` rows read natively from a local copy, so the write's input is a Comet scan. */ + private def source(rows: Int, dir: File): DataFrame = + materializeAsCometSource( + (1 to rows).map(i => (i, s"str_$i")).toDF("id", "name"), + new File(dir, "source.parquet").getAbsolutePath) + + /** + * The files and bytes the write reports: from Spark's stats tracker on Spark 4.0+, which leaves + * the write command in the plan, and from the native writer on 3.x, which replaces it. + */ + private def writtenFilesAndBytes(plan: SparkPlan): (Long, Long) = + plan + .collectFirst { + case write: DataWritingCommandExec => + (write.cmd.metrics("numFiles").value, write.cmd.metrics("numOutputBytes").value) + case write: CometNativeWriteExec => + (write.metrics("files_written").value, write.metrics("bytes_written").value) + } + .getOrElse(fail(s"No write in the plan:\n${plan.treeString}")) + + /** + * Writes `rows` rows in `files` tasks natively to `table` under `confs`, and checks that the + * magic committer committed them: its `_SUCCESS` manifest names it and the files, no `__magic` + * directory or pending upload is left behind, the write reports each file's real size, and the + * data reads back. On Spark 4.0+ that size comes from the marker S3A writes when it closes a + * magic file, and on 3.x from what the native writer counted. + */ + private def checkMagicCommittedWrite(table: String, rows: Int, files: Int)( + confs: (String, String)*): Unit = + withTempPath { dir => + val df = source(rows, dir).repartition(files) + withNativeWriter { + withSQLConf(confs: _*) { + val plan = captureWritePlan(p => df.write.parquet(p), table) + assertHasCometNativeWriteExec(plan) + + val tablePath = new Path(table) + val fs = tablePath.getFileSystem(spark.sessionState.newHadoopConf()) + val manifest = SuccessData.load(fs, new Path(tablePath, "_SUCCESS")) + assert(manifest.getCommitter == "magic", manifest) + assert(manifest.getFilenames.size == files, manifest) + + val entries = fs.listStatus(tablePath) + assert( + entries.forall(!_.getPath.getName.startsWith("__magic")), + entries.map(_.getPath).mkString(", ")) + val parts = entries.filter(_.getPath.getName.startsWith("part-")) + assert(parts.length == files, parts.map(_.getPath).mkString(", ")) + assert(pendingUploads(tablePath.toUri.getHost).isEmpty) + assert(writtenFilesAndBytes(plan) == ((files.toLong, parts.map(_.getLen).sum))) + } + } + checkAnswer(spark.read.parquet(table), df) + } + + test("a native write is committed by S3A's magic committer") { + // Spark's own magic committer configuration, rather than one this suite spells out. + assert( + spark.conf.get(SQLConf.FILE_COMMIT_PROTOCOL_CLASS.key) == + NativeWriteUtils.PathOutputCommitProtocolClass) + assert( + spark.conf.get(SQLConf.PARQUET_OUTPUT_COMMITTER_CLASS.key) == + NativeWriteUtils.BindingParquetOutputCommitterClass) + checkMagicCommittedWrite(s"s3a://$testBucketName/committed", rows = 10000, files = 4)() + } + + test("a native write is committed with the magic committer tracking commits in memory") { + assume( + Try( + Class.forName( + "org.apache.hadoop.fs.s3a.commit.magic.InMemoryMagicCommitTracker")).isSuccess, + "Requires S3A's in-memory commit tracking, which Hadoop 3.3.4 does not have") + // Task commit then reads the pending uploads from S3A's in-process tracker, not from the + // `.pending` files: a native writer that only reproduced those files would lose its data + // here. Its own bucket, since S3A reads the setting from the file system instance it caches + // per bucket. + val bucket = "magic-in-memory" + createBucketIfNotExists(bucket) + checkMagicCommittedWrite(s"s3a://$bucket/committed", rows = 1000, files = 2)( + "fs.s3a.committer.magic.track.commits.in.memory.enabled" -> "true") + } + + // Spark 4.1 configures both bindings at once. Either one alone asks the committer factory for + // the magic committer: PathOutputCommitProtocol ignores the Parquet committer class, and Spark's + // default protocol instantiates BindingParquetOutputCommitter, which hands out the magic + // committer's paths. + Seq( + "PathOutputCommitProtocol" -> (SQLConf.PARQUET_OUTPUT_COMMITTER_CLASS.key -> + classOf[ParquetOutputCommitter].getName), + "BindingParquetOutputCommitter" -> (SQLConf.FILE_COMMIT_PROTOCOL_CLASS.key -> + SQLConf.FILE_COMMIT_PROTOCOL_CLASS.defaultValueString)).foreach { + case (binding, otherBindingUnset) => + test(s"a native write is committed by the magic committer through $binding alone") { + checkMagicCommittedWrite(s"s3a://$testBucketName/$binding", rows = 1000, files = 2)( + otherBindingUnset) + } + } + + test("a failed native write leaves no pending upload behind") { Review Comment: This test fails in the child projection on the first batch, so the file is created but no bytes reach the stream. It covers an upload that was started and never fed. Could you add cases that show rollback with real S3A? First, a source big enough to upload at least one part before the failure. For example, set `fs.s3a.multipart.size` to 5M and write more than 5 MiB. Then the abort has to cancel a part that was already uploaded. Second, a commit protocol based on `PathOutputCommitProtocol` that throws in `commitTask` before calling `super`. The file is already closed in that case, so only S3A's `abortTask` can clean up. Third, the same for `commitJob`, which on 3.x is Comet's own `abortJob` call. After each case I would expect no visible data, no `__magic` directory and no pending upload. -- 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]
