HyukjinKwon commented on code in PR #59931:
URL: https://github.com/apache/spark/pull/59931#discussion_r4418056529
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarTable.scala:
##########
@@ -189,11 +189,51 @@ class ColumnarBatchScan(planner: ColumnarBatchPlanner)
extends Batch {
case class ColumnarStreamOffset(json: String) extends Offset
-class ColumnarMicroBatchStream(reader: ColumnarStreamReader) extends
MicroBatchStream {
+/**
+ * The micro-batch stream of a [[ColumnarDataSource]]. It supports admission
control only so that
+ * the data source gets the offset that each micro-batch starts after: its
default read limit reads
+ * all the available data, which Spark then always asks for.
+ */
Review Comment:
Done in d597801: a single Scaladoc remains, and the comments of the two
fields are as suggested.
##########
sql/core/src/test/resources/native-datasource/test_native_datasource.cc:
##########
@@ -1033,28 +1067,38 @@ 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,
+ void* sink) {
+ return guarded([&] {
+ checkSize("SparkDsInitialOffsetInfo", info->size,
sizeof(SparkDsInitialOffsetInfo));
+ sendBytes(set_offset, sink, "{\"offset\":0}");
+ });
}
-// Each call makes `step` more rows available, up to `max_offset`.
-int latestOffset(void* handle, SparkDsSink set_offset, void* sink) {
+// Makes `step` more rows available after the start of the micro-batch, up to
`max_offset`.
+int latestOffset(void* handle, const SparkDsLatestOffsetInfo* info,
SparkDsSink set_offset,
+ void* sink) {
return guarded([&] {
- Options& state = fromHandle<Reader>(handle)->state;
- int64_t latest = std::min(getLong(state, "max_offset", 10),
- getLong(state, "current", 0) + getLong(state,
"step", 10));
- state["current"] = std::to_string(latest);
+ const Options& state = fromHandle<Reader>(handle)->state;
+ if (info->start == nullptr) throw std::runtime_error("no start offset");
Review Comment:
Done in d597801: every function of the test library that gets a struct of
Spark calls `checkSize`, also `readPartition` and `createDataWriter` instead of
their inline checks.
##########
docs/sql-native-data-sources.md:
##########
@@ -101,32 +101,36 @@ 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. A streaming write commits each micro-batch with its epoch ID, and
the ID of the query
Review Comment:
Done in d597801: the Writes paragraph of the guide now says that a rejected
writer is aborted, with `rejected` set and an epoch ID of -1, before
`close_writer`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSource.scala:
##########
@@ -185,28 +193,32 @@ class NativeDataSource(
}
}
- /** Creates a writer for a batch write, or for a streaming write with the ID
of its query. */
+ /**
+ * Creates a writer for a batch write, or for a streaming write with the ID
of its query, with
+ * the state that its data writers need.
+ */
private def createWriter(
library: NativeLibrary,
schema: StructType,
+ timeZoneId: String,
Review Comment:
Done in d597801: `createWriter` no longer takes the time zone. The writers
also give it to `ArrowBatchingOptions` (`ColumnarWriter.timeZoneId`), so the
arrays of `write` and the schema of `create_writer` have a single source.
--
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]
