sunchao commented on code in PR #5531:
URL: https://github.com/apache/datafusion-comet/pull/5531#discussion_r3883250776
##########
native/core/src/execution/operators/shuffle_scan.rs:
##########
@@ -173,7 +190,27 @@ impl ShuffleScanExec {
// Decode the compressed IPC data
let mut timer = decode_time.timer();
- let batch = read_ipc_compressed(slice)?;
+ let decoded = if requires_validation {
+ read_ipc_compressed_validated(slice)
+ } else {
+ read_ipc_compressed(slice)
+ };
+ let batch = match decoded.and_then(|batch|
check_column_count(batch, data_types.len()))
Review Comment:
[P2] Validate remote logical types before schema casting
The remote validation path checks Arrow buffer validity and column count,
but does not compare the decoded logical types with `data_types`. A corrupt IPC
schema can therefore reach `cast_and_stamp_schema`, which can silently change
values instead of reporting a fetch failure.
Reproduced at `7467fb296f1ac229ba499cea663d1e1cec1ca1c5` with Spark
3.5.9/JDK 17 and the published Celeborn 0.6.3 and 0.7.0 clients, using a native
Celeborn handle with stage reruns enabled, `spark.shuffle.compress=false`, no
Spark I/O encryption, and Celeborn integrity checking disabled
(`celeborn.client.shuffle.integrityCheck.enabled=false`, also its default):
1. Generate a native `NONE` frame using `ShuffleBlockWriter` for
`Field("value", Int32, false)` and rows `[-1, 0, 1]`.
2. Flip only the IPC schema's `Int.is_signed` bit from true to false
(`frame[135]: 1 -> 0` in this 540-byte frame). The native length, field count,
and array buffers remain unchanged.
3. Read that committed mapper batch through the new manager/reader route,
stock `ShuffleClientImpl.readPartition(..., needDecompress=false)`, and native
`ShuffleScan` with the original Spark `IntegerType` declaration.
Arrow accepts the resulting `UInt32` array, this count check succeeds, and
the later cast turns `4294967295` into null. The corrupted frame produces
`[null, 0, 1]` with no exception and zero `reportShuffleFetchFailure` calls.
The JVM `read()` route instead throws `UnsupportedOperationException:
Unsupported data type: Int(32, false)` during import, also without a
`FetchFailedException` or failure report. Both routes should reject this
incompatible logical schema and invoke the remote failure handler before
casting/importing it.
The pristine-frame and corrupt-stale-attempt controls both preserve `[-1, 0,
1]`; the existing bad-codec control correctly reports a fetch failure. The
exact-base (`a223ba14e9f1445b5d315962ae9005f7ec45b187`) manager rejects native
Celeborn reads before fetching. The local cast itself is unchanged: the
introduced exposure is the new remote reader reaching it with
`requires_validation=true` but no logical-type check. Compatible dictionary
representations and nested-nullability normalization still need to remain
supported.
These controls used actual Comet native/JNI execution and stock client reads
over loopback transport, with controlled worker/lifecycle responses; they are
not deployed-cluster or planner-enablement tests.
--
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]
