HyukjinKwon commented on code in PR #59931:
URL: https://github.com/apache/spark/pull/59931#discussion_r4418602053
##########
sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h:
##########
@@ -350,16 +451,62 @@ typedef struct SparkDsWriteInfo {
const char* query_id;
} SparkDsWriteInfo;
+/* The data writer of a task, as create_data_writer gets it, on an executor. */
+typedef struct SparkDsDataWriterInfo {
+ /* The size of this struct in the header that Spark was built with, as in
SparkDsHost. */
+ int64_t size;
+ /*
+ * The schema of the data to write, as create_writer got it. Each array that
write gets has this
+ * schema, so the library keeps it, or what it needs of it, to read the
arrays.
+ */
+ struct ArrowSchema* schema;
+ /* The state that create_writer returned, or empty. */
+ const uint8_t* writer_state;
+ int64_t writer_state_length;
+ /* The ID of the micro-batch for a streaming write, or -1 for a batch write.
*/
+ int64_t epoch_id;
+} SparkDsDataWriterInfo;
+
+/* The end of a write, as commit and abort get it. */
+typedef struct SparkDsFinishWriteInfo {
+ /* The size of this struct in the header that Spark was built with, as in
SparkDsHost. */
+ int64_t size;
+ /*
+ * The ID of the micro-batch for a streaming write, or -1 for a batch write,
and for a writer
+ * whose create_writer Spark rejected, before it knew the micro-batch.
+ */
+ int64_t epoch_id;
+ /*
+ * The num_messages messages returned by commit_data_writer of each task.
For abort, a message
+ * is NULL if its task did not commit.
+ */
+ const uint8_t* const* messages;
+ const int64_t* message_lengths;
+ int64_t num_messages;
+ /*
+ * For abort, whether Spark rejected what create_writer returned, such as a
state set more than
+ * once: no task ran, so there are no messages.
+ */
+ bool rejected;
Review Comment:
Done in 2379f0c: `rejected` is now an `int64_t`, 1 or 0, so the struct ends
with an 8-byte field like the others and a field appended later grows `size`.
The header, the guide and the test library say it is set to 1.
##########
sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h:
##########
@@ -304,6 +310,74 @@ typedef struct SparkDsCreateInfo {
int64_t num_options;
} SparkDsCreateInfo;
+/* A batch scan, as create_reader gets it. */
+typedef struct SparkDsReaderInfo {
+ /* The size of this struct in the header that Spark was built with, as in
SparkDsHost. */
+ int64_t size;
+} SparkDsReaderInfo;
+
+/* A request for the partitions of a batch scan, as partitions gets it. */
+typedef struct SparkDsPartitionsInfo {
+ /* The size of this struct in the header that Spark was built with, as in
SparkDsHost. */
+ int64_t size;
+} SparkDsPartitionsInfo;
+
+/* A micro-batch streaming scan, as create_stream_reader gets it. */
+typedef struct SparkDsStreamReaderInfo {
+ /* The size of this struct in the header that Spark was built with, as in
SparkDsHost. */
+ int64_t size;
+ /*
+ * The location of a directory in the checkpoint of the streaming query that
is reserved for this
+ * source, such as file:/tmp/checkpoint/sources/0, where the library may
keep what it needs to
+ * recover from a failure. It stays the same when the query restarts from
its checkpoint. It is a
+ * fully qualified Hadoop path, whose characters are not escaped as in a
URI: a space stays a
+ * space, and % stays %. Its file system may be one that the library cannot
access, such as s3a
+ * or hdfs.
+ */
+ const char* checkpoint_location;
+} SparkDsStreamReaderInfo;
+
+/*
+ * A request for the initial offset of a stream reader, as initial_offset gets
it. Spark may call
+ * initial_offset more than once for a query, for example when it restarts
before its first
+ * micro-batch is committed, so it must return the same offset each time: a
library whose initial
+ * offset depends on when it is asked, such as the latest data, keeps it under
checkpoint_location.
Review Comment:
Done in 2379f0c: `ColumnarMicroBatchStream` now keeps the first initial
offset in an `HDFSMetadataLog` under `_spark_initial_offset` in the checkpoint
location of the source, and reuses it on restart, as `KafkaMicroBatchStream`
does. The header and the guide drop the requirement on the library, and reserve
the names that start with `_spark` under `checkpoint_location` for Spark.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarTable.scala:
##########
@@ -189,11 +189,52 @@ 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 reader gets the offset that each micro-batch starts after, the end of
the last one, for
+ * every trigger: its default read limit reads all the available data. With
+ * `Trigger.AvailableNow`, Spark runs micro-batches up to the latest offset
that the first
+ * micro-batch got, rather than wrapping the stream in one that asks for the
latest offset from
+ * the initial offset.
+ */
+class ColumnarMicroBatchStream(reader: ColumnarStreamReader)
+ extends MicroBatchStream with SupportsTriggerAvailableNow {
+
+ // The offset that the reader returned last, which the progress of the query
reports.
+ @volatile private var lastLatestOffset: Offset = _
+ // Whether the trigger is Trigger.AvailableNow.
+ @volatile private var availableNow = false
+ // The end of the run with Trigger.AvailableNow: the first latest offset
asked.
+ @volatile private var availableNowEnd: Offset = _
+ // Spark asks for the initial offset for the start of the first latest
offset, and again to plan
+ // the first micro-batch, which both get the same offset.
+ private lazy val initial = ColumnarStreamOffset(reader.initialOffset())
Review Comment:
Done in 2379f0c: with the option `initial_offsets_path`, `native_counter`
records each call of `initial_offset` and returns the number of calls before it
as the offset. The new test "ask for the initial offset once, and keep it in
the checkpoint" checks that a query without a checkpoint asks once, and that a
restart before the first micro-batch is committed reads from the kept offset
without asking again.
##########
docs/sql-native-data-sources.md:
##########
@@ -101,32 +101,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,
Review Comment:
Done in 2379f0c: the guide now says that with `Trigger.AvailableNow` Spark
asks for the `latest_offset` only once, before the first micro-batch that it
plans, and that Spark asks for `initial_offset` only when the checkpoint does
not have it yet and keeps it there.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarTable.scala:
##########
@@ -189,11 +189,52 @@ 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 reader gets the offset that each micro-batch starts after, the end of
the last one, for
+ * every trigger: its default read limit reads all the available data. With
+ * `Trigger.AvailableNow`, Spark runs micro-batches up to the latest offset
that the first
+ * micro-batch got, rather than wrapping the stream in one that asks for the
latest offset from
Review Comment:
Done in 2379f0c: the Scaladoc now says "up to the first latest offset that
Spark asks for".
--
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]
