dongjoon-hyun commented on code in PR #59932:
URL: https://github.com/apache/spark/pull/59932#discussion_r4417878342
##########
sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h:
##########
@@ -115,8 +118,14 @@
* released once they are no longer referenced (see Lifecycle above). A
reader or writer handle
* must not depend on the data source handle it was created from, because
Spark closes the data
* source handle right after creating it.
- * - Memory. What Spark passes to a function, such as strings, byte arrays,
SparkDsCreateInfo and
- * SparkDsWriteInfo, is only valid during the call, except for SparkDsHost
and
+ * - Order of the calls. After a function on a handle fails, Spark calls no
other function on the
+ * handle than its close function, except that it aborts a writer whose
commit failed and a data
+ * writer whose write or commit failed. It does not abort a writer or a data
writer whose commit
+ * succeeded. It calls each function of a batch reader at most once. It
calls commit and abort at
Review Comment:
**1. [Medium] The Lifecycle section still says that every other attempt is
aborted**
This rule matches the code: `NativeDataWriter` sets `finished` as soon as
the library commits, so a data writer whose `commit_data_writer` succeeded is
never aborted, even if its task fails afterwards. The Lifecycle section,
though, still says that Spark "aborts the data writers of the other attempts,
except on an executor that is lost" (L90-L91). For an attempt that committed
and whose task then failed on a live executor, the two sentences now disagree.
That case is reachable today. In "do not abort a data writer that committed
with an invalid message", the library commits, `NativeLibrary.commitDataWriter`
then rejects the two messages, and the task fails without `abort_data_writer`.
The same happens when anything after `dataWriter.commit()` in
`DataWritingSparkTask` throws. `OutputCommitCoordinator.taskCompleted` then
clears the authorized committer of the partition, so another attempt can
commit, and the driver's `commit` gets only the message of that attempt. The
output of the first attempt is neither aborted nor passed to `commit`: the test
library leaves its file in `_temporary`. A library that follows the Lifecycle
text, and expects every attempt but those on lost executors to be either
referenced by `commit` or aborted, keeps that output forever.
Since version 1 freezes this text, could the Lifecycle sentence name this
case too? For example: "It aborts the data writers of the other attempts,
except one whose commit_data_writer succeeded before its task failed, and one
on an executor that is lost, ...". Libraries would then know that the writer's
`commit` or `abort` has to clean up the output of the data writers whose
messages it does not get.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala:
##########
@@ -1114,6 +1236,30 @@ class NativeDataSourceSuite extends QueryTest with
SharedSparkSession {
message.contains("a message from native code") &&
message.contains(System.mapLibraryName("test_native_datasource"))
})
+ // The library does not log a warning when Spark logs errors only.
+ val errorAppender = new LogAppender("native data source error logs")
+ withLogAppender(errorAppender, loggerNames =
Seq(classOf[NativeLibrary].getName),
+ level = Some(org.apache.logging.log4j.Level.ERROR)) {
+ range(session, "log" -> "a message that Spark does not log")
+ }
+ assert(!errorAppender.loggingEvents.exists(
+ _.getMessage.getFormattedMessage.contains("a message that Spark does not
log")))
Review Comment:
**2. [Low] The ERROR-level check passes even without the check in `hostLog`**
This is the test that I suggested in [item
6](https://github.com/apache/spark/pull/59932#discussion_r4417356131) of the
first round, and my suggestion was wrong, sorry. With the logger of
`NativeLibrary` at ERROR, `LOGGER.warn` in `NativeLibrary.log` already drops
the message. So the appender gets no event whether or not `hostLog` asked
`is_log_enabled`, and removing `if (hasIsLogEnabled() &&
!host->is_log_enabled(level)) return;` does not fail this assertion.
Only the library knows that it skipped the call. For example, an option such
as `log_calls_path` could name a file to which `hostLog` appends the level
right before it calls `host->log`, and the test could check that the file stays
empty at ERROR. The `log_levels_path` check below already covers
`is_log_enabled` itself, so this is only about the check in `hostLog`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala:
##########
@@ -100,6 +100,19 @@ class NativeDataSourceSuite extends QueryTest with
SharedSparkSession {
session
}
+ // The test libraries record the calls that the order of the calls of the
header rules out.
+ override def afterEach(): Unit = {
+ try {
+ if (orderViolations.exists()) {
+ val violations = Files.readString(orderViolations.toPath)
+ orderViolations.delete()
+ fail(s"The test library got calls that the header rules
out:\n$violations")
Review Comment:
**4. [Low] A violation is not reported as a failure of the test that caused
it**
`BeforeAndAfterEach.runTest` runs `afterEach` as an after-effect of the
status of the test (`Status.withAfterEffect`). So when this `fail` throws, the
test has already been reported as succeeded, and the exception is reported as
an abort of the whole suite. Its message does not say which test made the
calls. A call that comes after the test returns also lands in the next test,
for example `abort_data_writer` from a task that is still being killed after a
failed job.
Checking the file at the end of `nativeTest`, after `f` returns, would fail
the test that broke the order, with its name in the report. Deleting the file
when `nativeTest` starts would also keep a late call from failing an unrelated
test. The test that caused the late call would then miss it, but that seems
acceptable.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceTestUtils.scala:
##########
@@ -37,6 +37,12 @@ object NativeDataSourceTestUtils {
private val libraries = new ConcurrentHashMap[String, File]()
+ /**
+ * The file to which the test libraries append the calls that the order of
the calls of the
+ * header rules out, one per line. It does not exist as long as there are
none.
+ */
+ lazy val orderViolations: File = new File(buildDir, "order_violations")
Review Comment:
**5. [Low] The other suites that use the test library do not check the
violations**
Every library that `compile` builds appends to this one file, but only
`NativeDataSourceSuite` reads it. `NativeDataSourceClusterSuite` is in the same
module, so it usually runs in the same test JVM. It writes and streams to
`native_sink` with libraries from the same `compile`, and its executors, which
run in JVMs of their own, append to the same absolute path. A violation there
does not fail that suite. It stays in the file until the next test of
`NativeDataSourceSuite`, which then fails for it.
`NativeDataSourceConnectSuite`, in the Connect server module, never reads the
file at all.
A helper in `NativeDataSourceTestUtils` that each of these suites calls
after each test would report a violation in the suite that made it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSource.scala:
##########
@@ -376,35 +385,49 @@ object NativeDataSourceStreamReader {
JsonMapper.builder().enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS).build()
}
-/** Writes to a native data source for a batch write. */
+/**
+ * Writes to a native data source for a batch write.
+ *
+ * @param schema the schema of the data to write
+ * @param timeZoneId the time zone of the timestamps of the schema that the
writer got
+ */
class NativeDataSourceWriter(
location: NativeLibraryLocation,
name: String,
+ schema: StructType,
+ timeZoneId: String,
@transient private val handle: NativeHandle,
writerState: Array[Byte])
extends ColumnarWriter {
import NativeCalls._
@transient private lazy val library = NativeLibraries.get(location)
+ // Whether the library committed the write, after which it is not aborted,
as the header says,
+ // even if Spark fails afterwards.
+ @transient private var committed = false
Review Comment:
**3. [Low] No test reaches the new guards**
The checks added in 8b94b3b keep the order of the calls true if a caller of
the adapter changes, but no test takes their paths. No test aborts a writer
after a successful commit, writes or commits a data writer after a failed call,
or prunes a scan builder twice. So removing `if (!committed)` from either
writer, the `failed` check of `writeOrCommit`, or the `pruned` guard of
`ColumnarScanBuilder` would not fail any test. The violations file would catch
the calls that reach the library then, but only in a test that makes them. A
few direct tests would cover the guards, for example:
- a writer from `NativeDataSource.writer`, and one from `streamWriter`,
committed and then aborted, with no `_ABORTED` file and no violation;
- a `NativeDataWriter` whose `write` fails with the option `fail` set to
`write`, then `commit`, which fails with the internal error without calling the
library;
- `ColumnarScanBuilder.pruneColumns` called twice, with the reader asked
once, and the internal error for a column outside the read schema.
##########
sql/core/src/test/resources/native-datasource/test_native_datasource.cc:
##########
@@ -957,47 +1036,63 @@ int createReader(void* handle, struct ArrowSchema*
schema, void** out) {
for (const auto& column : columns) findColumn(column);
if (returnsZero(source->options, "create_reader")) return;
auto* reader = new Reader();
+ reader->state["kind"] = std::to_string(SPARK_DS_READ_BATCH);
reader->state["columns"] = join(columns);
reader->state["lo"] = "0";
reader->state["hi"] = get(source->options, "end", "10");
reader->state["partitions"] = get(source->options, "partitions", "2");
reader->state["batch_size"] = get(source->options, "batch_size", "1000");
for (const char* option :
- {"fail", "slice", "short", "null_rows", "wait_for_interrupt",
"task_path"}) {
+ {"fail", "slice", "short", "null_rows", "wait_for_interrupt",
"task_path", "states"}) {
reader->state[option] = get(source->options, option, "");
}
*out = reader;
});
}
+// Passes the state to set_state once, or as many times as the option "states"
says.
+void sendState(const Options& options, SparkDsSink set_state, void* sink,
+ const std::string& state) {
+ std::string times = get(options, "states", "");
+ for (int64_t i = 0; i < (times.empty() ? 1 : std::stoll(times)); i++) {
+ sendBytes(set_state, sink, state);
+ }
+}
+
void closeSource(void* handle) { delete fromHandle<DataSource>(handle); }
#ifndef READ_ONLY
-int createStreamReader(void* handle, struct ArrowSchema* schema, void** out) {
+int createStreamReader(void* handle, struct ArrowSchema* schema,
+ const SparkDsStreamReaderInfo* info, void** out) {
return guarded([&] {
auto* source = fromHandle<DataSource>(handle);
if (source->name != name("native_counter")) {
throw std::runtime_error(source->name + " does not support streaming
reads");
}
auto columns = join(importColumns(schema));
+ if (info->checkpoint_location == nullptr) throw std::runtime_error("no
checkpoint location");
+ std::string checkpoint_path = get(source->options, "checkpoint_path", "");
+ if (!checkpoint_path.empty()) writeFile(checkpoint_path,
info->checkpoint_location);
if (returnsZero(source->options, "create_stream_reader")) return;
auto* reader = new Reader();
+ reader->state["kind"] = std::to_string(SPARK_DS_READ_MICRO_BATCH);
reader->state["columns"] = columns;
reader->state["max_offset"] = get(source->options, "max_offset", "10");
reader->state["step"] = get(source->options, "step",
reader->state["max_offset"]);
- reader->state["current"] = "0";
reader->state["batch_size"] = get(source->options, "batch_size", "1000");
- reader->state["commit_path"] = get(source->options, "commit_path", "");
- reader->state["bad_offset"] = get(source->options, "bad_offset", "");
- reader->state["closed_path"] = get(source->options, "closed_path", "");
+ for (const char* option :
+ {"commit_path", "bad_offset", "closed_path", "starts_path",
"states"}) {
+ reader->state[option] = get(source->options, option, "");
+ }
*out = reader;
});
}
int pushLimit(void* handle, int32_t limit, bool* applied) {
- return guarded([&] {
- auto* reader = fromHandle<Reader>(handle);
+ auto* reader = fromHandle<Reader>(handle);
+ if (reader->failed) orderViolation("push_limit after a failed call");
+ return tracked(&reader->failed, [&] {
// Spark calls push_limit at most once.
if (reader->state.count("limit") > 0) throw std::runtime_error("pushLimit
was called twice");
Review Comment:
**6. [Nit] Stream readers and a second `push_limit` are not recorded as
violations**
The rule that only the close function follows a failed call also covers
stream readers. But `latestOffset`, `streamPartitions` and `commitOffset` do
not go through `tracked`, so a call after a failed one would go unnoticed.
Here, a second `push_limit` throws, which only shows up as a
`NATIVE_DATA_SOURCE_ERROR` that a test checking only that the query fails could
hide. `prune_columns` and `partitions` instead record "called twice" in the
violations file. Recording this case the same way would make the three
functions of a batch reader consistent.
##########
docs/sql-native-data-sources.md:
##########
@@ -135,20 +141,24 @@ function reports an error by returning a non-zero value,
with the message return
`last_error`, and Spark reports it as a `NATIVE_DATA_SOURCE_ERROR`. A C++
exception or a Rust panic
must not cross the interface. Each handle that a `create` function returns is
released exactly
once, by the `close` function of its kind. The close functions cannot fail,
and they can run after
-the query, so do the work that can fail, such as flushing data, in `commit`.
Strings are UTF-8.
+the query, so do the work that can fail, such as flushing data, in `commit`.
After a function on
+a handle fails, Spark only closes the handle, except that it aborts a writer
or a data writer whose
+commit or write failed, and it never aborts one whose commit succeeded: the
header lists the order
+of the calls that a library can rely on. Strings are UTF-8.
Review Comment:
**8. [Nit] The guide could read as if a writer had a write**
"it aborts a writer or a data writer whose commit or write failed" can be
read as a writer being aborted after a failed write, although only a data
writer has `write`. The header splits it more precisely ("a writer whose commit
failed and a data writer whose write or commit failed"), and the guide could do
the same:
```suggestion
the query, so do the work that can fail, such as flushing data, in `commit`.
After a function on
a handle fails, Spark only closes the handle, except that it aborts a writer
whose commit failed
and a data writer whose write or commit failed, and it never aborts one
whose commit succeeded:
the header lists the order of the calls that a library can rely on. Strings
are UTF-8.
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSource.scala:
##########
@@ -426,18 +451,26 @@ class NativeDataSourceStreamWriter(
partitionId: Int,
taskId: Long,
epochId: Long): DataWriter[ColumnarBatch] = {
- new NativeDataWriter(location, name, writerState, partitionId, taskId,
epochId)
+ new NativeDataWriter(
+ location, name, schema, timeZoneId, writerState, partitionId, taskId,
epochId)
}
+ // Whether the library committed the micro-batch, after which it is not
aborted, as the header
+ // says, even if Spark fails afterwards.
+ @transient private var committed = false
Review Comment:
**7. [Nit] The guard for a committed write could live in
`ColumnarBatchWrite` and `ColumnarStreamingWrite`**
The same field and comment are now in both native writers, while the guard
for pruning went into the generic `ColumnarScanBuilder`, next to `limitResult`.
Spark reaches the `commit` and `abort` of a `ColumnarWriter` and of a
`ColumnarStreamWriter` only through `ColumnarBatchWrite` and
`ColumnarStreamingWrite`. A flag there, set when `writer.commit` returns and
checked before `writer.abort`, would keep the rule in one place, for any
`ColumnarDataSource`.
If the field stays here, a smaller point: it sits between `createWriter` and
`commit`, while `NativeDataSourceWriter` declares it with its other fields,
after `library`.
##########
sql/core/src/test/resources/native-datasource/test_native_datasource.cc:
##########
@@ -1290,10 +1462,13 @@ int commitDataWriter(void* handle, SparkDsSink
set_message, void* sink) {
int abortDataWriter(void* handle) {
if (usedAfterClose(handle)) return 0;
+ auto* writer = fromHandle<DataWriter>(handle);
+ if (writer->abort_called) orderViolation("abort_data_writer called twice");
+ writer->abort_called = true;
return guarded([&] {
- auto* writer = fromHandle<DataWriter>(handle);
// Spark does not abort a data writer that committed: records it if it
does.
if (writer->committed) {
+ orderViolation("abort_data_writer after a successful commit");
writeFile(writer->state["path"] + "/_ABORTED_COMMITTED_DATA_WRITER", "");
Review Comment:
**9. [Nit] An abort after a successful commit is recorded twice**
This branch now records the call in the violations file, which fails the
test. It also still writes `_ABORTED_COMMITTED_DATA_WRITER`, which "do not
abort a data writer that committed with an invalid message" checks. One of the
two is enough: dropping the marker and its assertion would leave a single
mechanism.
Separately, `orderViolation` repeats the `fopen`/`fputs`/`fclose` append
code of `starts_path` in `latestOffset`. The two also differ: `orderViolation`
drops a violation silently when the file cannot be opened, while `latestOffset`
fails. A small `appendLine` helper could share the code.
--
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]
