dongjoon-hyun commented on code in PR #59931:
URL: https://github.com/apache/spark/pull/59931#discussion_r4418344125
##########
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:
**2. [Low] A library may not be able to keep its initial offset under
`checkpoint_location`**
This asks a library whose initial offset depends on when it is asked to keep
it under `checkpoint_location`, but
[`SparkDsStreamReaderInfo`](https://github.com/apache/spark/blob/d597801d15f4c8ff773f46917f797c7ff466cb41/sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h#L329-L337)
says that its file system may be one that the library cannot access, such as
s3a or hdfs. A library that starts from the latest data, with a checkpoint on
S3 and no S3 client of its own, then cannot return the same offset when the
query restarts before its first micro-batch is committed, and Spark plans that
micro-batch again from a newer start up to the end in the offset log.
Spark can already write there: `ColumnarMicroBatchStream` gets the location
in `toMicroBatchStream`, and could keep the first initial offset in a metadata
log, as
[`KafkaMicroBatchStream.getOrCreateInitialPartitionOffsets`](https://github.com/apache/spark/blob/d597801d15f4c8ff773f46917f797c7ff466cb41/connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala#L366-L395)
does for Kafka, under a name that the header reserves for Spark. Every library
would then get a stable initial offset without accessing the checkpoint itself,
and the header would not need this requirement. If that is out of scope for
version 1, could the header at least say that a library can only meet it when
it can access the file system of the checkpoint?
##########
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:
**3. [Low] No test fails if the initial offset is asked twice again**
The `lazy val` fixes the two calls of the first micro-batch, but
[`initialOffset`](https://github.com/apache/spark/blob/d597801d15f4c8ff773f46917f797c7ff466cb41/sql/core/src/test/resources/native-datasource/test_native_datasource.cc#L1081-L1087)
of the test library ignores its reader and returns the same offset each time,
so the suites pass with or without it. Could `native_counter` record its calls
of `initial_offset`, for example by appending a line to a file given by an
option, as `starts_path` does for `latest_offset`, and a test check that a
query that starts without a checkpoint calls it once?
##########
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:
**4. [Nit] The guide does not have the two new rules of the streaming reads**
The header now says that with `Trigger.AvailableNow`, Spark asks for the
`latest_offset` only once, before the first micro-batch that it plans, and that
`initial_offset` must return the same offset each time. This paragraph still
says that Spark asks for the `latest_offset` for each micro-batch, and does not
mention `initial_offset`. Since a library that only reads the guide could
compute its initial offset from the latest data each time it is asked, could it
mention both?
##########
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:
**1. [Low] `rejected` leaves 7 bytes of padding at the end of the struct,
where a later field would fit without changing `size`**
`SparkDsFinishWriteInfo` ends with a `bool` at offset 40, and its size is 48
because the struct is aligned to 8 bytes. A `bool` or an `int32_t` appended
later within version 1 would be at offset 41 or 44, inside these 48 bytes, so
`size` would stay 48. A library built against that header checks that `size`
covers the new field, which also passes with the 48 that Spark 4.4.0 passes,
and then reads a padding byte. That only works because `NativeLibrary` zeroes
the struct, which the header does not promise, and only for a field whose zero
means that it is absent.
The other structs that Spark passes all end with an 8-byte field, so this is
the only one. Could `rejected` be an `int64_t`, 0 or 1, so that each appended
field grows `size`? Otherwise, the Compatibility section could say that Spark
zeroes the struct up to `size`, and that a field appended later must mean its
default when it is 0.
##########
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:
**5. [Nit] The Scaladoc still says that the run ends at the latest offset
that the first micro-batch got**
As the header now says, Spark asks for the latest offset before the first
micro-batch that it plans, which is not the first micro-batch of the run when
the query restarts with a micro-batch that was not committed: Spark runs that
one again without asking. Maybe "up to the first latest offset that Spark asks
for", as the comment of `availableNowEnd` says.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala:
##########
@@ -735,7 +735,7 @@ class NativeDataSourceSuite extends QueryTest with
SharedSparkSession {
"name" -> "native_sink",
"msg" -> "injected failure in write"))
// Spark aborted the first epoch, with a null message for its task, which
did not commit.
- assert(Files.readString(new File(dir, "_ABORTED_0").toPath) == "1")
+ assert(Files.readString(new File(dir, "_ABORTED_0").toPath) == "1,1,0")
Review Comment:
**6. [Nit] The tests of failed batch writes do not check the new content of
`_ABORTED`**
Only this streaming test and the rejected writers compare `_ABORTED` with
the number of NULL messages, the number of messages and `rejected`. [The batch
writes that
fail](https://github.com/apache/spark/blob/d597801d15f4c8ff773f46917f797c7ff466cb41/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala#L617-L715)
only check that `_ABORTED` is a file, so a batch abort that set `rejected` or
lost its messages would pass them. Could they compare the content too, for
example `"2,2,0"` when the two data writers fail to commit? The abort after a
failed `commit` matters most, since all its messages are set and the library
must remove the data that they point to: its first and last numbers would be 0.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/v2/ffi/NativeLibrary.java:
##########
@@ -750,43 +950,46 @@ void write(long dataWriter, long arrayAddress, long
schemaAddress) throws Throwa
*/
byte[] commitDataWriter(long dataWriter, Runnable committed) throws
Throwable {
MethodHandle f = function(Function.COMMIT_DATA_WRITER);
- try (Sink sink = new Sink()) {
- check((int) f.invokeExact(dataWriter, Ffm.SINK, sink.id),
Function.COMMIT_DATA_WRITER);
+ try (Memory memory = new Memory(); Sink sink = new Sink()) {
+ long info = memory.zeroed(FINISH_DATA_WRITER_INFO_SIZE);
+ Platform.putLong(null, info, FINISH_DATA_WRITER_INFO_SIZE);
+ check((int) f.invokeExact(dataWriter, info, Ffm.SINK, sink.id),
+ Function.COMMIT_DATA_WRITER);
committed.run();
byte[] message = single(sink, Function.COMMIT_DATA_WRITER);
return message == null ? new byte[0] : message;
}
}
void abortDataWriter(long dataWriter) throws Throwable {
- check((int) function(Function.ABORT_DATA_WRITER).invokeExact(dataWriter),
- Function.ABORT_DATA_WRITER);
+ try (Memory memory = new Memory()) {
+ long info = memory.zeroed(FINISH_DATA_WRITER_INFO_SIZE);
+ Platform.putLong(null, info, FINISH_DATA_WRITER_INFO_SIZE);
Review Comment:
**7. [Nit] The structs that only hold their size are allocated for each
call**
`SparkDsReaderInfo`, `SparkDsPartitionsInfo`, `SparkDsInitialOffsetInfo` and
`SparkDsFinishDataWriterInfo` only hold their own size, which does not change,
but each call allocates, zeroes and frees one, and `commit_data_writer` and
`abort_data_writer` now do it for each task. Until fields are added to them, a
single struct of each kind, allocated once and never freed as `Ffm.host` does
for `SparkDsHost`, would do, since the library gets a `const` pointer, and
`abortDataWriter` would stay a single call. This is minor next to the cost of
the call itself.
--
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]
