sarutak commented on code in PR #59914:
URL: https://github.com/apache/spark/pull/59914#discussion_r4418425463
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSource.scala:
##########
@@ -468,6 +469,13 @@ object NativeWriterCommitMessage {
/**
* Reads a partition of a native data source: each array of the Arrow stream
exported by the
* library is checked, and imported into the Arrow vectors that back the
returned batch.
+ *
+ * A reader can be closed while another thread is in [[next]]: some operators
read their input on
+ * a thread of their own that the task does not wait for, such as the stdin
writer thread of
+ * `PipedRDD`, while `DataSourceRDD` closes the reader once the task
completes. The library must
+ * not get the release of the stream during get_next, and the task must stay
valid until then, so
+ * [[close]] then leaves the release to [[next]], once get_next returns,
rather than wait for it
+ * and keep the task from completing. [[next]] returns false once the reader
is closed.
Review Comment:
Fixed in d4d5c202ca8 with the suggested wording: a `next()` that the close
overlaps returns false or throws the error of its read, and a `next()` called
after the close returns false without calling the library.
##########
sql/core/src/test/resources/native-datasource/test_native_datasource.cc:
##########
@@ -790,24 +812,52 @@ int streamGetNext(ArrowArrayStream* stream, ArrowArray*
out) {
state->last_error = "the task was killed";
return ECANCELED;
}
- if (state->next >= state->end) {
- out->release = nullptr; // The end of the stream.
- return 0;
+ if (!state->slow_read.empty()) {
+ // The flag covers the slow read, for streamRelease to tell whether it
comes during get_next.
+ state->in_get_next = true;
+ int result = EIO;
+ try {
+ std::string reading = state->slow_read + ".reading";
+ struct stat info;
+ if (stat(reading.c_str(), &info) != 0) {
+ writeFile(reading, "");
+ std::string go = state->slow_read + ".continue";
+ for (int i = 0; i < 6000 && stat(go.c_str(), &info) != 0; i++) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ }
+ }
+ if (state->fail_slow_read) {
Review Comment:
Fixed in d4d5c202ca8. After the wait, the `slow_read` stream calls
`is_interrupted` for its task context and writes the answer to
`<slow_read>.interrupted`. Both cross-thread tests assert that it is `false`,
and the thread test also asserts it for the failing read. The stream calls the
`is_interrupted` pointer that it kept when `read` was called, rather than
reading it from the struct, and passes the task address, which Spark only looks
up: a context that Spark freed is no longer registered, so the answer is `true`
without reading freed memory. A first version that read the pointer from the
struct crashed the forked test JVM under the mutant below, instead of failing.
Verified with the mutant you described: when `close()` calls `task.close()`
before it checks `reading`, and still defers the stream release, both tests
fail on this assertion, with `"[tru]e" did not equal "[fals]e"`.
--
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]
