apache / apache/datafusion-comet

Native Sort-Merge Writer for Iceberg ClusteredWriter Path

Open
#3,595 1 comment 0 reactions 1 assignee Claimed by @peterxcli View on GitHub
enhancement
Dominant language
Scala
Stars
1.3k
Forks
373
Avg merge
2d 4h
Merged PRs (30d)
198

Description

### What is the problem the feature request solves?

Implement a fused sort-merge writer in native Rust for Comet's Iceberg write path, eliminating JNI round-trips when writing partitioned/sorted data to Iceberg tables.

#### Current Comet Writer Behavior

```rust
// parquet_writer.rs - Current implementation
while let Some(batch) = stream.try_next().await {
writer.write(&batch)?; // All batches go to ONE file
}
writer.close()?;
```
- No sorting capability
- No partition detection
- Single file per Spark partition
- ClusteredWriter logic (from iceberg-rust) is not utilized

#### Related Work

- iceberg-rust ClusteredWriter: iceberg-rust/crates/iceberg/src/writer/partitioning/clustered_writer.rs
- DataFusion External Sort: datafusion/physical-plan/src/sorts/sort.rs
- Comet Native Sort: CometSortExec in operators.scala

### Describe the potential solution

Instead of:
1. Spark: "Execute CometSort" -> JNI -> Native sort -> JNI -> Arrow batches to Spark
2. Spark: "Execute CometWrite" -> JNI -> Native write

It would be:
1. Spark: "Execute CometSortWrite" -> JNI -> Native (Sort -> Write) -> JNI -> metadata only

- Design protobuf schema for ClusteredParquetWriter operator
- Implement ClusteredParquetWriterExec in Rust
- Integrate DataFusion's external sort with spill support
- Add partition extraction and validation logic
- Create Scala-side CometClusteredWriteExec operator
- Add integration with Spark's FileCommitProtocol
- Benchmark against current Spark sort + Comet write path
- Add tests for partition ordering validation

### Additional context

_No response_

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.