andygrove opened a new pull request, #5293:
URL: https://github.com/apache/datafusion-comet/pull/5293
## Which issue does this PR close?
Part of #2967 and #1625. Restructures the native write path so the following
can be
fixed at all, and closes the ones that were purely symptoms of the old
design:
Closes #2985 (no `_SUCCESS` file)
Closes #3521 (`INSERT INTO ... SELECT` invisible to subsequent reads)
Closes #3426 (complex type with different names)
Unblocks (not fixed here, but no longer require re-implementing Spark's write
framework inside Comet): #2957, #2970, #3015, #3041, #3193, #3194, #3417,
#3428.
## Rationale for this change
Native writes replaced the whole `DataWritingCommandExec`, which means
`InsertIntoHadoopFsRelationCommand.run` never ran. Everything that method
does had to
be re-implemented inside `CometNativeWriteExec`: a hardcoded
`SQLHadoopMapReduceCommitProtocol` (so
`spark.sql.sources.commitProtocolClass` was
ignored), `dynamicPartitionOverwrite` pinned to `false`, a hand-ported copy
of the
SaveMode logic, a bespoke commit-message accumulator, and its own
`commitJob` call.
Most of the open native-writer issues are symptoms of that one decision
rather than
independent defects. Fixing them one at a time against the old design means
writing a
second, worse `FileFormatWriter` inside Comet.
Spark already has the right seam. On Spark 4.0+,
`V1WritesUtils.getWriteFilesOpt`
matches the `WriteFilesExecBase` **trait** (introduced in 4.0 precisely for
this), so a
Comet node that extends it gets driven through
`FileFormatWriter.executeWrite` → `SparkPlan.executeWrite` →
`doExecuteWrite`, and
Spark keeps ownership of everything above the per-task write.
**Why Spark 4.0+ only.** On 3.4/3.5 `getWriteFilesOpt` matches the concrete
`WriteFilesExec` case class. A Comet node there would not be found,
`writeFilesOpt`
would be `None`, and Spark would silently take `FileFormatWriter`'s
non-planned,
row-based branch — ignoring `doExecuteWrite` entirely. The only way in on
3.x is to
inherit from a case class, which brings `copy`/`equals` hazards; that isn't
worth
carrying, so native writes now require 4.0+ and report a fallback reason on
3.x.
## What changes are included in this PR?
```
Execute InsertIntoHadoopFsRelationCommand <- Spark: SaveMode, catalog,
commitJob, _SUCCESS
+- CometWriteFiles <- Comet: native per-task write
only
+- CometNativeScan ...
```
- **New** `CometWriteFilesExec` overriding `doExecuteWrite`, mirroring
`FileFormatWriter.executeTask` for the parts Comet must do itself: build
the
`TaskAttemptContext`, ask the commit protocol for a path, run the native
writer,
drive the stats trackers, commit or abort. Plus the `CometWriteFiles`
serde and a
two-line `ShimCometWriteFilesExec` in `spark-4.x` / `spark-3.x`.
- **Deleted** `CometNativeWriteExec.scala` and
`CometDataWritingCommand.scala` (594
lines). Net −549 lines.
- File paths come from `FileCommitProtocol.newTaskTempFile` and are used
**verbatim**,
so names match Spark's `part-<id>-<uuid>-c000.<codec>.parquet` and
committers that
track individual files (S3A magic, streaming manifest) work. Previously
only
`.getParent` was kept and the native writer invented its own names.
- Column names come from `WriteJobDescription.dataColumns` rather than the
query
output, so `INSERT INTO t SELECT a+1` writes the target column's name
(#3426).
- Byte/row counts come from `BasicWriteTaskStatsTracker`, which stats files
through the
`FileSystem` API and is therefore correct on HDFS. The native writer's
`std::fs::metadata` call reported `0` there. `CometMetricNode`'s
now-redundant
`reportNativeWriteOutputMetrics` is removed.
- Proto: `work_dir` / `job_id` / `task_attempt_id` removed (reserved);
`output_path` is
now the exact file to write, set per task.
- Opt-in moves to `spark.comet.operator.WriteFilesExec.allowIncompatible`,
with the old
`DataWritingCommandExec` key kept via `withAlternative`.
### Fixed along the way
AQE re-plans the write command's child and **re-inserts a `WriteFilesExec`**
above the
node Comet already converted. Without a guard that produced nested native
writes — the
data written twice, and the inner node's empty output read as a zero-column
schema.
`CometExecRule` now collapses the redundant node. The
`basic parquet write with repartition` test catches this.
## How are these changes tested?
- `CometParquetWriterSuite`: **33/33** on Spark 4.1 — the 30 existing tests
plus three
new regression tests for `_SUCCESS` (#2985), Spark-compatible file naming,
and
`INSERT INTO ... SELECT` visibility (#3521). Suite-level
`assume(isSpark40Plus)`.
- `CometTaskMetricsSuite`: 6/6. Note this suite's native-write test was
pinned to the
old operator's config key, so it would have silently fallen back to
Spark's writer
and still passed; it is repointed and now genuinely exercises the native
path.
- Regression sweep: `CometExecSuite` (143), `CometExpressionSuite` +
`CometAggregateSuite` + `CometFuzzTestSuite` (273) — all pass.
- Compiles against Spark 3.4, 3.5, 4.0 and 4.1. `cargo check` and the native
`parquet_writer` unit tests pass.
## Known limitation
`WriteTaskStatsTracker.newRow(filePath, row)` is a per-row callback. Comet
has columnar
batches, so rather than materializing every row just to hand it straight
back,
`recordRows` passes `InternalRow.empty` and feeds only the count. That is
exactly right
for `BasicWriteTaskStatsTracker`, which ignores the row argument, but a
third-party
tracker inspecting row contents would see empty rows — so that case logs a
warning
rather than silently reporting wrong statistics. A plan-time guard isn't
possible
because `statsTrackers` only exists at execution time.
## Follow-ups
Independent of this change and the next highest-value work, since the Spark
default is
affected: full `WriterProperties` (block/page size, dictionary, writer
version), INT96
timestamps (#3425 — `spark.sql.parquet.outputTimestampType` defaults to
`INT96` and we
write INT64 micros), the four footer metadata keys (#3427 — `legacyINT96`
and `timeZone`
drive rebase decisions on read, so omitting them is a correctness risk),
field IDs, and
Catalyst nullability. Then partitioned (#3193) → bucketed (#3194) → object
stores.
--
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]
