andygrove commented on PR #5763:
URL:
https://github.com/apache/datafusion-comet/pull/5763#issuecomment-5653995121
Everything from this round is addressed in 268fdb9d2. Thanks @sunchao and
@comphead — the two P2s were both real, and the second one had a second half I
had not looked for.
## HDFS filename prefixes (#5763 (comment))
Fixed, and the fix has two layers because one is not enough.
The guard was checking the destination directory when the thing that has to
survive intact is the whole committer-returned path.
`HadoopMapReduceCommitProtocol.getFilename` reads `mapreduce.output.basename`
straight from the task configuration, so caller-controlled text lands in every
file name, and `?`/`#` are worse there than anywhere else in the path: the
native URL parser does not escape them, it treats them as delimiters and
truncates. Every task then writes the same truncated name and they overwrite
each other during commit, which is exactly the 50-rows-of-100 result you
measured. `escapedHdfsDestination` now takes the effective basename alongside
the directory and declines on either.
Planning can only decline what it can predict, though, and the path actually
comes from `FileCommitProtocol.newTaskTempFile` — a replaceable component. So
`NativeWriteUtils.checkNativeWriteDestination` runs the same predicate against
the real path in `executeTask`, before the writer opens anything. That one
throws rather than falling back, which is the right trade at that point: the
alternative is a job that commits successfully with the data somewhere else.
I have no HDFS here, so the regression test exercises the predicate directly
rather than through a write — space, `%`, and non-ASCII basenames declined,
`part` / `out` / `data_v2` / `part-of-it` accepted, a `file:` destination
unaffected by its basename, and the execution-time check accepting a real
staging path and rejecting the same path with a `?` basename. I confirmed it
fails without the fix:
```
- HDFS destinations needing URI escaping are declined at planning *** FAILED
***
NativeWriteUtils.escapedHdfsDestination(plainHdfs, basename).isDefined was
false
expected basename 'part?x' to be declined
```
## Mixed-case compression option (@comphead, Major 1)
Fixed, and you were right that the second half is the dangerous one.
`parseCompressionCodec`, `supportedCompressionCodecs` and the string-to-proto
mapping all moved to `NativeWriteUtils` behind a `CaseInsensitiveMap`, which
also removes the identical bug from the Spark 3.x serde — it mattered less
there only because that writer invents a name with no codec suffix.
I took the stronger variant as well. `CometWriteFilesExec` now re-derives
the codec per task from `CodecConfig.from(taskAttemptContext)`, which is the
same `CodecConfig` the file extension comes from (`ParquetUtils.prepareWrite`
resolves the write's options into `parquet.compression` on the job conf, and
`getFileExtension` reads it back). Name and contents now agree by construction
rather than by two copies of the precedence rule agreeing.
Two tests. The load-bearing one is *parquet write with a mixed-case
unsupported compression codec falls back to Spark*: with the case-sensitive
read restored it fails, because `getSupportLevel` finds `snappy` in the
SQLConf, accepts the write, and Comet writes SNAPPY into a file Spark named
`.lz4raw.parquet`. Worth noting that the other one, *honors a mixed-case
compression option*, **passes** under that same mutation on 4.0+ — the
execution-time re-derivation catches it. That is the defense in depth working,
but it does mean that test alone would not have caught the bug.
## Pinning the `url` encode set (@comphead, Major 2)
Took the Rust-test option. `url_path_rewritten_characters` in
`parquet_support.rs` parses `hdfs://ns/pre<c>post/output` for every printable
ASCII `c` and asserts the rewritten set is exactly `` "#<>?`{} `` (nine
characters, leading space), plus four non-ASCII names that must all be
percent-encoded. The Scala guard's comment now points at it and its comment
points back. The failure message says which constant to update.
## The smaller notes
- **Dead drain.** Right — `batches.foreach(_.close())` in the `else` branch
only runs when `!hasNext` already returned false, so it closed nothing.
Removed; the comment now says `hasNext` already ran the child to completion,
which is also what Spark's `EmptyDirectoryDataWriter` relies on.
- **`DataWritingCommandExec` never tagged on 4.0+.** Fixed as you suggested
— the skip is now conditional on `d.child.isInstanceOf[CometWriteFilesExec]`,
so a write with no `WriteFilesExec` child gets its reason back.
- **Duplicate write metrics.** Dropped all three.
`BasicWriteJobStatsTracker` is authoritative on this path and reports on the
command, and the native `bytes_written` reads 0 on HDFS, so the node was
contributing a second and less trustworthy copy. `CometTaskMetricsSuite` still
passes 15/15, which is the check that matters: its output metrics come from
Spark's tracker, not from the node.
- **`EliminateRedundantTransitions`.** No arm is needed, and you found the
reason before I documented it. There is now a comment on `doExecute` naming
`Columnar.scala`'s `outputsColumnar = write.child.supportsColumnar` for a
`V1WriteCommand` under `plannedWriteEnabled`, and saying that it is
load-bearing and would fail at execution rather than at planning.
- **`installation.md`.** Now says the example is Spark 4.0+ and that on
3.4/3.5 the same message appears on the command.
## Test gaps
- **Mixed-case `compression`** — two tests, above.
- **Partially empty multi-partition write** — *an empty partition writes no
file and still commits*. Hash-partitions into eight with AQE coalescing off and
keeps one id, so at most one partition has rows and at most two files can
appear. Forcing the write branch unconditionally fails it.
- **Third-party `WriteTaskStatsTracker`** — nothing Spark ships lets a V1
write install one, so the test drives `recordRows` directly with a recording
tracker and asserts three callbacks, all with empty rows, plus the warning
naming the tracker's class. `recordRows` became public for this; it is the only
way to reach the documented limitation.
- **#3426** — this one came out differently from what we both expected, and
it corrects a claim in my own description. I added the scenario from Spark's
own `INSERT INTO TABLE - complex type but different names` with a top-level
rename on top, and it does pass on 4.0+ and fail on 3.5. But it fails on 3.5
with *zero rows read back*, i.e. #3521, not with the wrong names — and
reverting `description.dataColumns` to `child.output` does **not** change the
outcome on 4.0+. The reason is that Spark's `castAndRenameQueryOutput` has
already aliased the query's output to the target's column names and cast the
struct to the target's nested field names before Comet sees anything, so the
two sources agree. `dataColumns` is still the right one to read — it is what
Spark treats as authoritative, and it is the only one that stays correct once
partitioned writes land, since `FileFormatWriter` excludes the partition
columns from it — but it is not what fixes #3426. Restructuring is. I have
corrected the code comment and the description accordingly.
## Testing
`CometParquetWriterSuite` is **48/48** on 4.0, 4.1 and 4.2, **36/36 + 12
cancelled** on 3.4 and **38/38 + 10 cancelled** on 3.5. `CometTaskMetricsSuite`
15/15, `CometExecSuite` 144/144, `CometFallbackInvarianceSuite` 6/6,
`CometPublicApiSuite` 1/1, all on 4.1. Native: `cargo test -p datafusion-comet
parquet_writer` and the new `url_path_rewritten_characters` pass, `cargo clippy
--all-targets --workspace -- -D warnings` is clean, and `spotless`,
`scalastyle` and `prettier` are clean.
Still not covered, and I am not claiming otherwise: an end-to-end HDFS
write, speculative concurrent attempts, and any measurement of `recordRows`.
--
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]
