dongjoon-hyun commented on code in PR #59932:
URL: https://github.com/apache/spark/pull/59932#discussion_r4418406852
##########
sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h:
##########
@@ -56,47 +56,54 @@
*
* - Batch reads. On the driver, Spark creates a reader with create_reader,
pushes operations down
* by calling, each at most once and in this order, push_limit and
prune_columns, plans the
- * partitions with partitions, serializes the state of the reader with
serialize_reader, and
- * closes it. Then it calls read on the executors, in one task for each
partition. If Spark never
- * plans the partitions, for example to only explain the plan of the query,
it closes the reader
- * once the reader is no longer referenced.
- * - Streaming reads. On the driver, Spark creates a stream reader with
create_stream_reader and
- * keeps it until the streaming query stops. For each micro-batch, it asks
for the latest_offset,
- * plans the partitions with stream_partitions and serializes the state with
- * serialize_stream_reader, then calls read on the executors. Offsets are
defined by the library,
- * as compact JSON on a single line: Spark stores each offset as a line of
the offset log of the
- * checkpoint, so it rejects an offset that is not valid JSON or contains a
line break. When a
- * streaming query restarts, from its checkpoint, Spark creates a new stream
reader and may call
- * stream_partitions again for the last range that it planned before, so the
partitions of a
- * range must read the same data each time.
+ * partitions with partitions, which also returns the state of the reader,
and closes it. Then it
+ * calls read on the executors, in one task for each partition, with the
schema to read and the
+ * state of the reader. If Spark never plans the partitions, for example to
only explain the plan
+ * of the query, it closes the reader once the reader is no longer
referenced.
+ * - Streaming reads. On the driver, Spark creates a stream reader with
create_stream_reader, with
+ * the location of the checkpoint of the source, and keeps it until the
streaming query stops. For
+ * each micro-batch, or only before the first one with Trigger.AvailableNow,
it asks for the
+ * latest_offset, with the offset that the micro-batch starts after, plans
the partitions with
+ * stream_partitions, which also returns the state of the stream reader,
then calls read on the
+ * executors. Offsets are defined by the library, as compact JSON on a
single line: Spark stores
+ * each offset as a line of the offset log of the checkpoint, so it rejects
an offset that is not
+ * valid JSON or contains a line break. When a streaming query restarts,
from its checkpoint,
+ * Spark creates a new stream reader and may call stream_partitions again
for the last range that
+ * it planned before, so the partitions of a range must read the same data
each time.
* - Writes. On the driver, right before it starts the tasks of the write (or
of a micro-batch),
- * Spark creates a writer with create_writer, and serializes its state with
serialize_writer.
- * With adaptive query execution, Spark first runs the stages of the data to
write that end with
- * a shuffle, and creates no writer if they fail. The rest of the data, such
as its last stage,
- * is computed in the tasks of the write, so a writer may still be aborted
because computing the
- * data failed: do irreversible work, such as removing the existing data for
an overwrite, in
- * commit. On the executors, each task creates a data writer with
create_data_writer, calls write
- * for each batch of at most spark.sql.execution.arrow.maxRecordsPerBatch
rows, which also ends
- * once it holds spark.sql.execution.arrow.maxBytesPerBatch bytes, and calls
commit_data_writer,
- * or abort_data_writer if the task or the commit fails, before it closes
the data writer. Then
- * the driver calls commit with the messages of the tasks if they all
succeeded, or abort
- * otherwise, including when commit fails, and closes the writer. Several
attempts of a task may
- * write the same partition, after a failure or with speculative execution,
each with a data
- * writer of its own. Spark lets the attempts of a partition commit one at a
time, and passes to
- * commit the message of one committed attempt for each partition, so the
write consists of the
- * data of these attempts only. It aborts the data writers of the other
attempts, except on an
- * executor that is lost, where a data writer is neither committed nor
aborted, and may even have
- * committed before its message reached the driver. A streaming write
commits or aborts each
- * micro-batch (epoch). Spark may commit the same epoch more than once, for
example when a
- * streaming query restarts after a failure, so committing an epoch that is
already committed
- * must have no effect, for exactly-once output. The ID of the query, which
create_writer gets for
- * a streaming write, stays the same when the query restarts from its
checkpoint, so the ID of the
- * query and the epoch ID identify the output of a micro-batch, also when
other queries write to
- * the same destination. A micro-batch streaming query creates a new stream
writer for each
- * micro-batch, and Spark closes the previous stream writer of the query
when the query creates a
- * new one. Spark is not told when a streaming query stops, so it closes the
last stream writer of
- * the query once the writer is no longer referenced, or when the query is
restarted. Streaming
- * writes are micro-batch only: a query with continuous processing fails
before create_writer.
+ * Spark creates a writer with create_writer, which also returns the state
of the writer. If Spark
+ * rejects what create_writer returned, such as a state set more than once,
it aborts the writer,
+ * with the field rejected of SparkDsFinishWriteInfo set, before it closes
it. With adaptive query
+ * execution, Spark first runs the stages of the data to write that end with
a shuffle, and
+ * creates no writer if they fail. The rest of the data, such as its last
stage, is computed in
+ * the tasks of the write, so a writer may still be aborted because
computing the data failed: do
+ * irreversible work, such as removing the existing data for an overwrite,
in commit. On the
+ * executors, each task creates a data writer with create_data_writer, with
the schema of the data
+ * and the state of the writer, calls write for each batch of at most
+ * spark.sql.execution.arrow.maxRecordsPerBatch rows, which also ends once
it holds
+ * spark.sql.execution.arrow.maxBytesPerBatch bytes, and calls
commit_data_writer, or
+ * abort_data_writer if the task or the commit fails before
commit_data_writer succeeded, before
+ * it closes the data writer. Then the driver calls commit with the messages
of the tasks if they
+ * all succeeded, or abort otherwise, including when commit fails, and
closes the writer. Several
Review Comment:
**1. [Low] The data writers of a killed attempt can be called after the
commit of the writer**
The Writes paragraph describes the calls on the data writers first, and then
says "Then the driver calls commit ... and closes the writer". The Order of the
calls section says nothing about the order between the data writers and their
writer. A library could read this as: when the writer's `commit` or `abort`
runs, every data writer of the write has finished.
That does not hold with speculative execution. When attempt A of a partition
commits and the last partition finishes, `runJob` returns, and the driver calls
`commit` and `close_writer`. `TaskSetManager` kills the other attempt B of that
partition with `interruptThread = true`. But B keeps running on its executor
until it sees the interrupt, or until `canCommit` denies it. Only then does
`DataWritingSparkTask` call `abort_data_writer` in its catch block and
`close_data_writer` in its finally block. So B can still call `write`,
`abort_data_writer` and `close_data_writer` after the writer was committed and
closed. The same happens when a job fails: the driver aborts the writer while
the other tasks of the stage are still being killed.
Some libraries would break then. For example, a library that removes a
staging directory in `commit` or `abort`, or that counts the open data writers
there. In local mode, a library might also free state at `close_writer` that
its data writers share with the writer. This PR documents the order that a
library can rely on, so could the header say this too? For example, in Order of
the calls: "The data writers of the task attempts that Spark kills, such as
speculative ones, may still be written to, aborted and closed after their
writer is committed or aborted, and closed."
##########
docs/sql-native-data-sources.md:
##########
@@ -101,32 +103,38 @@ Spark calls the functions of the table as follows:
that the data source does not support before it computes the data to write.
When the user does
not specify a schema, Spark asks for it with `schema`.
- **Batch reads**: on the driver, Spark creates a reader with `create_reader`,
pushes operations
- down with `push_limit` and `prune_columns`, plans the partitions with
`partitions`, and
- serializes the state of the reader with `serialize_reader`. Then it calls
`read` on the
- executors, in one task for each partition, and `read` exports an
`ArrowArrayStream`. Its
- callbacks can be called from any thread, but never concurrently.
-- **Streaming reads**: on the driver, Spark creates a stream reader with
`create_stream_reader`.
- For each micro-batch, it asks for the `latest_offset`, plans the partitions
with
- `stream_partitions`, and calls `read` on the executors. Offsets are defined
by the library, as
- compact JSON on a single line: Spark stores each offset as a line of the
offset log of the
- checkpoint, and rejects an offset that is not valid JSON or contains a line
break, such as
- pretty-printed JSON.
+ down with `push_limit` and `prune_columns`, and plans the partitions with
`partitions`, which
+ also returns the state of the reader. Then it calls `read` on the executors,
in one task for each
+ partition, with a `SparkDsReadInfo` that has the partition, the schema to
read and the state of
+ the reader, and `read` exports an `ArrowArrayStream`. Its callbacks can be
called from any
+ thread, but never concurrently.
+- **Streaming reads**: on the driver, Spark creates a stream reader with
`create_stream_reader`,
+ which gets the location of the checkpoint of the source in a
`SparkDsStreamReaderInfo`. For each
+ micro-batch, it asks for the `latest_offset`, with the offset that the
micro-batch starts after,
+ plans the partitions with `stream_partitions`, which also returns the state
of the stream reader,
+ and calls `read` on the executors. Offsets are defined by the library, as
compact JSON on a
+ single line: Spark stores each offset as a line of the offset log of the
checkpoint, and rejects
+ an offset that is not valid JSON or contains a line break, such as
pretty-printed JSON.
- **Writes**: on the driver, right before the tasks of the write (or of a
micro-batch) start, Spark
creates a writer with `create_writer`, which gets a `SparkDsWriteInfo` with
the mode of the write,
`SPARK_DS_WRITE_APPEND` or `SPARK_DS_WRITE_TRUNCATE`, and the ID of the
query of a streaming
- write. With adaptive query execution, it first runs the shuffle stages of
the data to write, and
- creates no writer if they fail, but the last stage runs in the tasks of the
write, so a writer can
- still be aborted because computing the data failed: do irreversible work,
such as removing the
- existing data for an overwrite, in `commit`. On the executors, each task
creates a data writer
- with `create_data_writer`, writes the batches of its data with `write`, and
calls
- `commit_data_writer`, or `abort_data_writer` if the task or the commit
fails. Then the driver
- calls `commit` with the messages of the tasks, or `abort` if the write
failed. A streaming write
- commits each micro-batch with its epoch ID, and the ID of the query stays
the same when the query
- restarts from its checkpoint: together, they identify the output of a
micro-batch, so that a
- library can skip an epoch that it already committed. A micro-batch streaming
query creates a new
- stream writer for each micro-batch, and Spark closes the previous one when
it creates the next
- one. Because Spark is not told when a streaming query stops, the last stream
writer of a query is
- closed once it is no longer referenced, or when the query is restarted.
+ write, and returns the state of the writer. With adaptive query execution,
it first runs the
+ shuffle stages of the data to write, and creates no writer if they fail, but
the last stage runs
+ in the tasks of the write, so a writer can still be aborted because
computing the data failed: do
+ irreversible work, such as removing the existing data for an overwrite, in
`commit`. On the
+ executors, each task creates a data writer with `create_data_writer`, which
gets the schema of the
+ data and the state of the writer in a `SparkDsDataWriterInfo`, writes the
batches of its data with
+ `write`, and calls `commit_data_writer`, or `abort_data_writer` if the task
or the commit fails.
+ Then the driver calls `commit` with the messages of the tasks, or `abort` if
the write failed.
Review Comment:
**3. [Low] The guide still says that a failed task aborts its data writer**
The header now says two things:
- Spark calls `abort_data_writer` only if the task or the commit fails
before `commit_data_writer` succeeded.
- The writer's `commit` or `abort` cleans up the output of the data writers
whose messages it does not get.
The Writes item of the guide still says "or `abort_data_writer` if the task
or the commit fails", and it does not mention that cleanup. The sentence at
L146-L149 says that Spark never aborts a data writer whose commit succeeded,
but not what a library has to do then. So a reader of the guide could expect
`abort_data_writer` to remove the output of a task that failed after its
commit, and make that output visible in `commit_data_writer`. That output would
then stay next to the output of the attempt that Spark retried. For example:
```suggestion
`write`, and calls `commit_data_writer`, or `abort_data_writer` if the
task or the commit fails
before `commit_data_writer` succeeded. Then the driver calls `commit` with
the messages of the
tasks, or `abort` if the write failed. A data writer whose task failed
after
`commit_data_writer` succeeded is neither aborted nor passed to `commit`,
so `commit` or `abort`
cleans up the output of the data writers whose messages it does not get.
```
##########
sql/core/src/test/resources/native-datasource/test_native_datasource.cc:
##########
@@ -1033,28 +1162,36 @@ int planPartitions(void* handle, SparkDsSink
add_partition, void* sink) {
int64_t stop = std::min(hi, start + step);
sendBytes(add_partition, sink, std::to_string(start) + "," +
std::to_string(stop));
}
+ sendState(reader->state, set_state, sink, serialize(reader->state));
});
}
-int serializeReader(void* handle, SparkDsSink set_state, void* sink) {
- return guarded([&] { sendBytes(set_state, sink,
serialize(fromHandle<Reader>(handle)->state)); });
-}
-
void closeReader(void* handle) { delete fromHandle<Reader>(handle); }
#ifndef READ_ONLY
-int initialOffset(void*, SparkDsSink set_offset, void* sink) {
- return guarded([&] { sendBytes(set_offset, sink, "{\"offset\":0}"); });
+int initialOffset(void*, const SparkDsInitialOffsetInfo* info, SparkDsSink
set_offset,
Review Comment:
**5. [Nit] `initial_offset` is not tracked, and no test makes a stream
reader fail**
`latest_offset`, `stream_partitions` and `commit_offset` now go through
`tracked`. But `initial_offset` is a function of the same stream reader handle,
and it still uses `guarded`. So it does not record a failure, and it does not
report a call after a failed one.
Also, as far as I can see, the option `fail` has no step for the functions
of a stream reader, or for `push_limit` and `prune_columns`. So the `failed`
path of these functions never runs in a test. No test checks that Spark only
closes a stream reader after one of them failed. The new checks would run if
`initial_offset` were tracked too, and if a step such as `latest_offset` were
added, with a streaming test that expects the query to fail.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceTestUtils.scala:
##########
@@ -37,6 +37,29 @@ 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")
+
+ /**
+ * Runs a test that uses the test libraries, and fails it if the libraries
got calls that the
+ * order of the calls of the header rules out. The calls recorded before,
such as by a task of an
+ * earlier test that was still being killed, are dropped.
+ */
+ def checkingOrderOfCalls[T](f: => T): T = {
+ orderViolations.delete()
+ val result = f
Review Comment:
**4. [Nit] `checkingOrderOfCalls` could wait for the tasks of the test**
This follows my round-2 suggestion. Losing a call that comes after the test
was the trade-off that I accepted there. Two smaller gaps remain, though:
- A late call can still fail an unrelated test, when it comes after the next
test deleted the file. An example is `abort_data_writer` from a task of a
failed job that is still being killed.
- When `f` throws, the violations that it recorded are not reported, and the
next test deletes them. A test that fails because of an order violation then
shows only the error that the violation caused.
The check could wait until no task runs before it reads the file. Late calls
would then fail the test that made them, which closes the first gap. For
example, `eventually` could wait until the `statusTracker` of
`SparkContext.getActive` reports no running task in `getExecutorInfos`, as
`checkDataWritersClosed` already waits for the data writers to close. For the
second gap, the violations could be added to the exception of `f` as suppressed.
##########
sql/core/src/test/resources/native-datasource/test_native_datasource.cc:
##########
@@ -727,6 +776,30 @@ void writeFile(const std::string& path, const std::string&
content) {
std::fclose(file);
}
+// Records a call that the order of the calls of the header rules out. Unlike
the files of the
+// options, a failure to record it is ignored: it may be called outside of
`guarded`, as in the
+// close functions, which cannot fail.
Review Comment:
**8. [Nit] Two comments of the test library do not match the code**
- This comment gives the close functions as an example of a call outside of
`guarded`, but no close function calls `orderViolation`. The calls outside of
`guarded` are the checks at the start of `push_limit`, `prune_columns`,
`partitions`, the stream reader functions, `commit`, `abort`, `write`,
`commit_data_writer` and `abort_data_writer`.
- The description of `ORDER_VIOLATIONS_PATH` at L84-L86 still says "which
the suite checks after each test". Three suites now check it at the end of each
test, with `checkingOrderOfCalls`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarWrite.scala:
##########
@@ -82,18 +82,24 @@ class ColumnarBatchWrite(createWriter: () =>
ColumnarWriter, schema: StructType)
extends BatchWrite {
private var writer: ColumnarWriter = _
+ // Whether the writer committed the write, after which it is not aborted,
even if Spark fails
+ // afterwards: a native data source relies on it.
+ private var committed = false
Review Comment:
**6. [Nit] The scaladoc of `ColumnarBatchWrite` says that any later failure
aborts the write**
L78-L79 still say that a failure in the tasks of the write, "as any later
failure, aborts the write". With this flag, a failure after a successful commit
no longer does. Could it say "as any later failure before the commit succeeds"?
Separately, the new sentence of `ColumnarWriter`, "It does not abort a write
whose commit succeeded.", can be read as if the writer does not abort.
`ColumnarStreamWriter` says "Spark does not abort ...", which is clearer.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarWrite.scala:
##########
@@ -115,6 +121,8 @@ class ColumnarStreamingWrite(
extends StreamingWrite {
private var writer: ColumnarStreamWriter = _
+ // The epochs that the writer committed, which are not aborted, as for a
batch write.
+ private val committedEpochs = scala.collection.mutable.Set.empty[Long]
Review Comment:
**7. [Nit] `committedEpochs` holds at most one epoch**
For each micro-batch, `V2Writes` builds a new `MicroBatchWrite(batchId,
write.toStreaming)`, and so a new `ColumnarStreamingWrite`. Continuous
processing fails in `createStreamingWriterFactory`. So this set never has more
than one element. A `private var committed = false`, as in
`ColumnarBatchWrite`, would do the same and keep the two guards alike, without
a mutable set or its fully qualified name.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala:
##########
@@ -706,14 +711,87 @@ class NativeDataSourceSuite extends QueryTest with
SharedSparkSession {
"action" -> "write to",
"name" -> "native_sink",
"msg" -> "commit_data_writer returned 2 values instead of one."))
- // The task failed without aborting the committed data writer, which it
closed, and the driver
- // aborted the write.
- assert(!new File(dir, "_ABORTED_COMMITTED_DATA_WRITER").exists())
+ // The task failed without aborting the committed data writer, which the
test library would
+ // record as a violation of the order of the calls, and closed it. The
driver aborted the write.
checkDataWritersClosed(dir)
assert(new File(dir, "_ABORTED").isFile)
assert(!dir.listFiles().exists(_.getName.startsWith("part-")))
}
+ nativeTest("keep the order of the calls whatever the callers of the adapter
do") {
Review Comment:
**2. [Low] The tests of the new guards only run on Java 22 or later**
Thank you for adding this test. It covers these guards:
- the `committed` flag of `ColumnarBatchWrite`;
- `committedEpochs` of `ColumnarStreamingWrite`;
- the `pruned` flag of `ColumnarScanBuilder`;
- `writeOrCommit` of `NativeDataWriter`.
The first three are in the generic adapter. But the test is a `nativeTest`,
so it is skipped unless there is a C++ compiler and Java 22 or later. The PR
builds run on Java 17 (the default `java` input of `build_and_test.yml`), and
the Java 21 builds skip it too. So only the scheduled Java 25 builds run it. A
change that drops the `!committed` check or the `pruned` guard would pass the
checks of its PR. My round-2 suggestion pointed at the native classes, so this
is partly on me.
The three adapter guards could also be tested in `ColumnarDataSourceSuite`,
which runs on any Java, next to "build and plan a batch scan once".
`TestSinkWriter` records the aborted tables in `TestColumnarSink.aborted`. So a
test could commit and then abort a `ColumnarBatchWrite` and a
`ColumnarStreamingWrite`. It could also prune a `ColumnarScanBuilder` twice,
with a counter of the `pruneColumns` calls added to the planner of
`TestRangeDataSource`. This test would then keep checking the side of the
library.
--
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]
