sunchao commented on code in PR #5368:
URL: https://github.com/apache/datafusion-comet/pull/5368#discussion_r3884203202
##########
spark/src/test/resources/pyspark/benchmark_pyarrow_udf.py:
##########
@@ -84,6 +84,8 @@ def _build_spark() -> SparkSession:
.config("spark.plugins", "org.apache.spark.CometPlugin")
.config("spark.comet.enabled", "true")
.config("spark.comet.exec.enabled", "true")
+ # Keep Comet's scan and execution rules active with Spark's default
shuffle manager.
+ .config("spark.comet.shuffle.enabled", "false")
Review Comment:
Updated in 5dd4782c. The benchmark now configures `CometShuffleManager` and
leaves Comet shuffle enabled at its default, instead of disabling shuffle to
keep the extension active with Spark's default manager.
Smoke-tested all three workloads with both `mapInArrow` and `mapInPandas`,
in vanilla and optimized modes, on Spark 4.0.4. All 12 combinations completed
and the existing plan assertions passed. This was a small functional run, not a
new performance measurement.
##########
spark/src/main/scala/org/apache/comet/rules/EliminateRedundantTransitions.scala:
##########
@@ -112,10 +112,10 @@ case class EliminateRedundantTransitions(session:
SparkSession)
// 4.1+ matches the renamed `MapInArrowExec`.
//
// Falls back to vanilla Spark when
`spark.sql.execution.arrow.useLargeVarTypes` is enabled:
- // CometArrowPythonRunnerBase.copyVector does raw `setBytes` on each
Arrow buffer, but Comet's
- // source string/binary vectors always use 4-byte offsets while the
destination root is
- // allocated with 8-byte offsets when this conf is on. The buffer counts
match but the
- // offset width does not, so a direct memcpy would corrupt the offsets.
+ // Native Comet string/binary vectors use 4-byte offsets. The IPC schema
follows these
Review Comment:
The old comment described raw buffer copying in `copyVector`, which this PR
removes. I updated the explanation for direct serialization: the IPC schema
matches the source buffers, so the stream is valid, but it would expose
`string` / `binary` instead of the large input types requested by
`useLargeVarTypes=true`.
Only the explanation changed here; the existing fallback behavior is
unchanged.
##########
docs/source/user-guide/latest/pyarrow-udfs.md:
##########
@@ -203,18 +203,14 @@ on the unoptimized path.
session time zone such a UDF can diverge from the unoptimized path. Set
`spark.comet.exec.pyarrowUDF.enabled=false` for those UDFs.
- `spark.sql.execution.arrow.useLargeVarTypes=true` is not supported. With
this conf enabled,
- Spark widens `StringType` and `BinaryType` to Arrow's 8-byte-offset variants
in the
- destination IPC root, while Comet's source vectors always use 4-byte
offsets. The buffer-copy
- path cannot bridge that mismatch, so `EliminateRedundantTransitions` skips
the rewrite and
- vanilla Spark handles the operation.
-- Each batch is copied twice on the JVM side: once from Comet's vectors into
Spark's
- destination IPC root (per-buffer `setBytes`), and a second time inside the
IPC writer when
- `VectorUnloader` / `MessageSerializer.serialize` walks the root and writes
bytes to the
- pipe to the Python worker. The pipe write is structural (Spark's transport
to Python is
- fork + pipe + Arrow IPC, so the buffer bytes must reach the pipe at least
once); dropping
- the first copy by serialising directly from Comet's vectors is tracked in
- [#4294](https://github.com/apache/datafusion-comet/issues/4294). Even after
that,
- true zero-copy at the JVM boundary is blocked because Comet's source
`FieldVector`s are
- imported from native via Arrow C Data Interface (their buffers route
`release` through FFI),
- while Spark's destination IPC root is a child of `ArrowUtils.rootAllocator`.
The two
- reference managers cannot share buffers via `TransferPair`.
+ Spark supplies `large_string` and `large_binary` input columns with 8-byte
offsets. Native
Review Comment:
`large_string` corresponds to Arrow's `LargeUtf8` / `LargeStringArray`, but
no new native mapping is needed for this PR because the large-type
configuration still falls back to Spark.
Supporting it on the accelerated path would require producing or converting
inputs to actual 64-bit-offset arrays, not just changing the schema label.
Comet already handles large string/binary output from Python on the JVM side;
that is separate from widening the inputs sent to Python. I have left that
additional support out of this PR and kept the fallback.
--
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]
