liuneng1994 commented on code in PR #13378:
URL: https://github.com/apache/gluten/pull/13378#discussion_r4345084371
##########
cpp/bolt/memory/BoltColumnarBatch.cc:
##########
@@ -98,8 +98,11 @@ std::shared_ptr<BoltColumnarBatch> BoltColumnarBatch::from(
if (cb->getType() == "bolt") {
return std::dynamic_pointer_cast<BoltColumnarBatch>(cb);
}
- auto vp = bolt::importFromArrowAsOwner(
- *cb->exportArrowSchema(), *cb->exportArrowArray(),
ArrowUtils::getBridgeOptions(), pool);
+ auto options = ArrowUtils::getBridgeOptions();
+ // Arrow NullType columns contain no buffers. The IPC import path
materializes
+ // them as native null constants before inspecting a validity buffer.
+ options.exportToArrowIPC = true;
Review Comment:
`exportToArrowIPC = true` is applied to every non-Bolt Arrow batch, not only
to schemas containing `NullType`. This flag can change how buffers/offsets are
interpreted for all types, so the blast radius is larger than the bug being
fixed. Could it be scoped to the `NullType` case, or is there a test that runs
a non-NullType batch through this path? If it is intentionally global, please
state why it is safe for all Arrow inputs.
##########
backends-bolt/src/main/scala/org/apache/spark/sql/execution/SparkWriteFilesCommitProtocol.scala:
##########
@@ -86,7 +88,9 @@ class SparkWriteFilesCommitProtocol(
// Note that %05d does not truncate the split number, so if we have more
than 100000 tasks,
// the file name is fine and won't overflow.
val split = taskAttemptContext.getTaskAttemptID.getTaskID.getId
- val fileName =
f"${spec.prefix}part-$split%05d-${UUID.randomUUID().toString()}${spec.suffix}"
+ val basename =
taskAttemptContext.getConfiguration.get("mapreduce.output.basename", "part")
Review Comment:
`mapreduce.output.basename` is a Hadoop `FileOutputFormat` convention.
Please add a test comparing this against vanilla Spark for (a) the default, (b)
an explicit basename, and (c) an empty/odd basename, so the generated path
cannot escape the staging dir or diverge from Spark's naming.
##########
backends-bolt/src/main/scala/org/apache/spark/api/python/ColumnarArrowEvalPythonExec.scala:
##########
@@ -151,52 +151,29 @@ class ColumnarArrowPythonRunner(
ColumnarArrowPythonRunner.this.writeUdf(dataOut, argMetas)
}
+ private val inputWriter = new BoltArrowPythonInputWriter(
+ inputIterator,
+ schema,
+ timeZoneId,
+ () => ArrowBufferAllocators.contextInstance())
+
+ context.addTaskCompletionListener[Unit] {
+ _ =>
+ try {
+ inputWriter.close()
+ } catch {
+ case scala.util.control.NonFatal(_) =>
Review Comment:
Swallowing `NonFatal` makes a failed `close()` invisible; if `writer.end()`
throws, the Python side may see a truncated stream with no diagnostic. Please
at least log a warning here (and consider rethrowing when the task is already
failing).
--
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]
