HyukjinKwon commented on code in PR #59932:
URL: https://github.com/apache/spark/pull/59932#discussion_r4418773851
##########
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:
Done in 14a1d3a: the Order of the calls section now says that 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.
##########
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:
Done in 14a1d3a: `ColumnarDataSourceSuite` now has "prune the columns of a
scan once" (with a counter of the `pruneColumns` calls of `TestRangePlanner`)
and "do not abort a write whose commit succeeded" (for `ColumnarBatchWrite` and
`ColumnarStreamingWrite`, with an abort without a commit as the control), which
run on any Java. The native test still checks the side of the library.
##########
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:
Done in 14a1d3a: the Writes item of the guide now uses the suggested text.
##########
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:
Done in 14a1d3a: `checkingOrderOfCalls` waits until the `statusTracker`
reports no running task before it reads the file, and when the test fails, it
adds the recorded violations to its error as suppressed.
##########
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:
Done in 14a1d3a: `initial_offset` now goes through `tracked`, and the option
`fail` has the steps `push_limit`, `prune_columns`, `initial_offset`,
`latest_offset`, `stream_partitions` and `commit_offset`. The new test "call no
other function on a reader whose call failed than its close function" makes
each of them fail, and checks that the stream reader is closed afterwards.
--
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]
