andygrove opened a new pull request, #5763:
URL: https://github.com/apache/datafusion-comet/pull/5763
## Which issue does this PR close?
Part of #2967 and #1625. Restructures the native write path on Spark 4.0+ 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.
Supersedes #5293, which made the same change but removed the Spark 3.x
writer along
the way. That regression is what held #5293 back, so this version is purely
additive:
Spark 3.4/3.5 keep the existing native writer, unchanged.
## Rationale for this change
Native writes replace the whole `DataWritingCommandExec`, which means
`InsertIntoHadoopFsRelationCommand.run` never runs. Everything that method
does has to
be re-implemented inside `CometNativeWriteExec`: a hardcoded
`SQLHadoopMapReduceCommitProtocol` (so
`spark.sql.sources.commitProtocolClass` is
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 4.0 added the right seam. `V1WritesUtils.getWriteFilesOpt` matches the
`WriteFilesExecBase` **trait** there (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 this is additive rather than a replacement.** 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. So `CometDataWritingCommand` and
`CometNativeWriteExec` stay
exactly as they are and remain the 3.4/3.5 path. `CometExecRule` picks the
path by
Spark version and the two never both fire. The 3.x path goes away with 3.x
support.
## What changes are included in this PR?
Spark 4.0+:
```
Execute InsertIntoHadoopFsRelationCommand <- Spark: SaveMode, catalog,
commitJob, _SUCCESS
+- CometWriteFiles <- Comet: native per-task write
only
+- CometNativeScan ...
```
Spark 3.4/3.5 is unchanged:
```
CometNativeWrite <- Comet: the whole write
+- 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`.
- **Nothing is deleted.** `CometNativeWriteExec`, `CometDataWritingCommand`,
`CometMetricNode.reportNativeWriteOutputMetrics` and the
`EliminateRedundantTransitions` rule for native writes all remain and
serve 3.x.
- 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. The 3.x
writer keeps
inventing its own names.
- Column names, nullability and Parquet field IDs 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 reports `0` there.
- Proto: `ParquetWriter.work_dir` becomes genuinely optional. When it is set
(3.x) the
native writer derives the file name from it as before; when it is unset
(4.0+),
`output_path` is the exact file to write and is used verbatim.
`output_path` was
already unused on the 3.x path, so no field changes meaning for an
existing plan.
- On 4.0+ the opt-in moves to
`spark.comet.operator.WriteFilesExec.allowIncompatible`,
with the old `DataWritingCommandExec` key kept as a deprecated alternative.
`CometConf.isOperatorAllowIncompat` now resolves alternatives; the
planner's by-name
lookup previously bypassed the `ConfigEntry`, so an old key would have
read `true`
from the entry while the planner saw `false`.
- `WriteFilesExec` declines dynamic partition overwrite (it is always a
partitioned
write) and `spark.sql.files.maxRecordsPerFile`, which Spark's own writer
uses to roll
a new file every N rows.
### Fixed along the way
AQE re-plans the write command's child and **re-inserts a `WriteFilesExec`**
above the
node Comet already converted. On the 3.x path that needs an explicit guard in
`CometExecRule` (still present). Leaving `DataWritingCommandExec` in place
on 4.0+
means the situation cannot arise there.
## How are these changes tested?
- `CometParquetWriterSuite`: **41/41** on Spark 4.0 and 4.1: the 33 existing
tests plus
eight new ones for `_SUCCESS` (#2985), Spark-compatible file naming,
`INSERT INTO ... SELECT` visibility (#3521), dynamic-overwrite fallback,
the
`maxRecordsPerFile` fallback (both the write option and the conf,
verifying Spark's
writer rolls 10 files), the schema-only empty-input write (SPARK-23271),
task abort
and retry through an injected failing commit protocol, and the deprecated
opt-in key.
The eight new tests `assume(isSpark40Plus)`.
- `CometParquetWriterSuite` on **Spark 3.4 (32/32 + 9 skipped)** and **3.5
(33/33 + 8
skipped)**: every pre-existing test still passes on the 3.x writer, which
is the point
of keeping it.
- `CometTaskMetricsSuite`: 15/15 on both 3.5 and 4.1. The suite's
native-write test now
picks the version-appropriate opt-in key, so it genuinely exercises the
native path on
both.
- Regression sweep on 4.1: `CometExecSuite` (144),
`CometFallbackInvarianceSuite` (6),
`CometPublicApiSuite` (1).
- Native: new `parquet_writer` unit test asserting the writer uses a
commit-protocol-chosen path verbatim (and that a non-zero partition id
does not leak
into the name). `cargo test -p datafusion-comet parquet_writer` and
`cargo clippy --all-targets --workspace -- -D warnings` pass.
- Compiles and test-compiles against Spark 3.4, 3.5, 4.0, 4.1 and 4.2.
## 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), and the four footer metadata keys (#3427: `legacyINT96`
and
`timeZone` drive rebase decisions on read, so omitting them is a correctness
risk).
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]
