HyukjinKwon commented on code in PR #59931:
URL: https://github.com/apache/spark/pull/59931#discussion_r4418603848
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala:
##########
@@ -735,7 +735,7 @@ class NativeDataSourceSuite extends QueryTest with
SharedSparkSession {
"name" -> "native_sink",
"msg" -> "injected failure in write"))
// Spark aborted the first epoch, with a null message for its task, which
did not commit.
- assert(Files.readString(new File(dir, "_ABORTED_0").toPath) == "1")
+ assert(Files.readString(new File(dir, "_ABORTED_0").toPath) == "1,1,0")
Review Comment:
Done in 2379f0c: the failed batch writes now compare the content of
`_ABORTED`: "2,2,0" when the writes or the commits of the two data writers
fail, "0,2,0" for the abort after a failed `commit`, and "1,1,0" for the data
writer that committed with an invalid message.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/v2/ffi/NativeLibrary.java:
##########
@@ -750,43 +950,46 @@ void write(long dataWriter, long arrayAddress, long
schemaAddress) throws Throwa
*/
byte[] commitDataWriter(long dataWriter, Runnable committed) throws
Throwable {
MethodHandle f = function(Function.COMMIT_DATA_WRITER);
- try (Sink sink = new Sink()) {
- check((int) f.invokeExact(dataWriter, Ffm.SINK, sink.id),
Function.COMMIT_DATA_WRITER);
+ try (Memory memory = new Memory(); Sink sink = new Sink()) {
+ long info = memory.zeroed(FINISH_DATA_WRITER_INFO_SIZE);
+ Platform.putLong(null, info, FINISH_DATA_WRITER_INFO_SIZE);
+ check((int) f.invokeExact(dataWriter, info, Ffm.SINK, sink.id),
+ Function.COMMIT_DATA_WRITER);
committed.run();
byte[] message = single(sink, Function.COMMIT_DATA_WRITER);
return message == null ? new byte[0] : message;
}
}
void abortDataWriter(long dataWriter) throws Throwable {
- check((int) function(Function.ABORT_DATA_WRITER).invokeExact(dataWriter),
- Function.ABORT_DATA_WRITER);
+ try (Memory memory = new Memory()) {
+ long info = memory.zeroed(FINISH_DATA_WRITER_INFO_SIZE);
+ Platform.putLong(null, info, FINISH_DATA_WRITER_INFO_SIZE);
Review Comment:
Done in 2379f0c: `SparkDsReaderInfo`, `SparkDsPartitionsInfo`,
`SparkDsInitialOffsetInfo` and `SparkDsFinishDataWriterInfo` are now allocated
once, and never freed, so `abortDataWriter` is a single call again.
--
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]
