fix: Delete completed tasks' data files when an Iceberg write job fails (#5663) * fix: delete data files left by a failed native Iceberg write task iceberg-java's writer abort deletes the files a failed task attempt wrote; the native path left them for remove_orphan_files. Close the gap in both places a task can fail. Inside the native writer, a TrackingLocationGenerator records every location handed to a file writer, since iceberg-rust's writers keep finalized files private until close and have no abort hook. The task deletes the recorded locations when a write fails, and an AbortOnDrop guard does the same when the task future is dropped without ever seeing an error, which is what happens when the JVM input iterator throws: executePlan returns that error from its JNI batch pull and the JVM releases the plan. After the native writer has returned, CometIcebergWriteExec registers a task failure listener that deletes the decoded manifest's files through the table FileIO, via a new best-effort IcebergReflection helper. Both deletions log failures rather than raising them, so the original task failure is the one Spark reports. Closes #5618 * fix: keep the Iceberg abort guard armed until deletion finishes * fix: delete completed tasks' data files when an Iceberg write job fails IcebergCommitExec collected task commit messages with executeCollect, which only returns once every task has succeeded, so a job failure left the committer with no messages and the data files of the tasks that had completed stayed in the table's data location. Collect each task's message as it finishes through a runJob result handler, as Spark's own V2 write does, and abort with the completed messages. Iceberg's SparkWrite.abort only deletes files after a cleanable commit failure and skips cleanup before any commit was attempted, so the committer then deletes the completed tasks' data files itself through the table FileIO; nothing can reference them at that point. The failing task's own files are handled by task-level cleanup. Closes #5277 --------- Co-authored-by: test <a@b.c>
Apache DataFusion Comet is a high-performance accelerator for Apache Spark. Comet keeps Spark queries Arrow-native end-to-end: operators, expressions, shuffle, and broadcast all stay in Apache Arrow columnar format, avoiding the per-row overhead of Spark's row-based engine. Within the Arrow-native pipeline, operators and expressions execute as Rust code (via the Apache DataFusion query engine) or as JVM code that operates directly on Arrow batches. Comet integrates with the Spark ecosystem without requiring any code changes.
Comet provides a ~2x speedup for TPC-DS @ SF 1000 (1TB), resulting in ~50% cost savings.
That 2x speedup gives you a choice: finish the same Spark workload in half the time on the cluster you already have, or match your current Spark performance on roughly half the resources. Either way, the gain translates directly into lower cloud bills, reduced on-prem capacity, and lower energy usage, with no changes to your existing Spark SQL, DataFrame, or PySpark code. Comet runs on commodity hardware: no GPUs, FPGAs, or other specialized accelerators are required, so the savings come from better utilization of the infrastructure you already run on.
See the Comet Benchmarking Guide for more details.
Comet accelerates Spark workloads by replacing Spark operators and expressions with high-performance implementations that process Apache Arrow columnar data directly. Most operators are powered by native Rust execution built on Apache DataFusion, while others run efficiently in the JVM on Arrow batches. This unified columnar execution model keeps processing within the Comet engine end-to-end, reducing overhead and delivering faster, more efficient query execution without reverting to Spark's traditional row-based engine.
FILTER (WHERE ...) clausesFor the authoritative lists, see the supported expressions and supported operators pages.
Comet is designed as a drop-in accelerator for Apache Spark, allowing you to integrate Comet into your existing Spark deployments and workflows seamlessly. With no code changes required, you can immediately harness the benefits of Comet's acceleration capabilities without disrupting your Spark applications.
Comet supports Apache Spark 3.4, 3.5, 4.0, and 4.1, and provides experimental support for Spark 4.2. See the installation guide for the detailed version, Java, and Scala compatibility matrix.
Install Comet by adding the jar for your Spark and Scala version to the Spark classpath and enabling the plugin. A typical configuration looks like:
export COMET_JAR=/path/to/comet-spark-spark3.5_2.12-<version>.jar $SPARK_HOME/bin/spark-shell \ --jars $COMET_JAR \ --conf spark.driver.extraClassPath=$COMET_JAR \ --conf spark.executor.extraClassPath=$COMET_JAR \ --conf spark.plugins=org.apache.spark.CometPlugin \ --conf spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager \ --conf spark.comet.explain.fallback.enabled=true \ --conf spark.memory.offHeap.enabled=true \ --conf spark.memory.offHeap.size=4g
For full installation instructions, published jar downloads, and configuration reference, see the installation guide and the configuration reference.
Join the DataFusion Slack and Discord channels to connect with other users, ask questions, and share your experiences with Comet.
We welcome contributions from the community to help improve and enhance Apache DataFusion Comet. Whether it's fixing bugs, adding new features, writing documentation, or optimizing performance, your contributions are invaluable in shaping the future of Comet. Check out our contributor guide to get started.
Apache DataFusion Comet is licensed under the Apache License 2.0. See the LICENSE.txt file for details.