HyukjinKwon commented on code in PR #59931:
URL: https://github.com/apache/spark/pull/59931#discussion_r4418055000
##########
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.
+ */
+/**
+ * The micro-batch stream of a [[ColumnarDataSource]]. The reader gets the
start of each
+ * micro-batch, the end of the last one, for every trigger: 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 = _
+ // With Trigger.AvailableNow, whether the latest offset of the run is the
first one asked.
+ @volatile private var availableNow = false
+ @volatile private var availableNowEnd: Offset = _
override def initialOffset(): Offset =
ColumnarStreamOffset(reader.initialOffset())
Review Comment:
Done in d597801: the stream keeps the initial offset in a `lazy val`, so
both calls of the first micro-batch get the same start. The header also says
that 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, kept under `checkpoint_location` if it depends on
when it is asked.
##########
sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h:
##########
@@ -304,6 +309,62 @@ typedef struct SparkDsCreateInfo {
int64_t num_options;
} SparkDsCreateInfo;
+/* 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. */
+typedef struct SparkDsInitialOffsetInfo {
+ /* The size of this struct in the header that Spark was built with, as in
SparkDsHost. */
+ int64_t size;
+} SparkDsInitialOffsetInfo;
+
+/* A request for the latest offset of a stream reader, as latest_offset gets
it. */
+typedef struct SparkDsLatestOffsetInfo {
+ /* The size of this struct in the header that Spark was built with, as in
SparkDsHost. */
+ int64_t size;
+ /*
+ * The offset, as JSON, that the next micro-batch starts after: the end of
the last micro-batch,
+ * or the initial offset before the first one, for every trigger. With
Trigger.AvailableNow,
+ * Spark asks for the latest offset only once, when the query starts, and
reads up to it.
Review Comment:
Done in d597801: it now says "only once, before the first micro-batch that
it plans", and the streaming reads of the Lifecycle mention this exception.
##########
sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h:
##########
@@ -414,19 +519,23 @@ typedef struct SparkDsLibrary {
* Creates a reader for a micro-batch streaming scan.
*
* schema: the schema to read: the one the user specified, or the one
returned by schema
+ * info: the location of the checkpoint of the source
* reader: set to the stream reader handle
*/
- int (*create_stream_reader)(void* source, struct ArrowSchema* schema, void**
reader);
+ int (*create_stream_reader)(void* source, struct ArrowSchema* schema,
+ const SparkDsStreamReaderInfo* info, void**
reader);
Review Comment:
Done in d597801: `create_reader` gets a `SparkDsReaderInfo`, `commit` and
`abort` a `SparkDsFinishWriteInfo` (epoch ID, messages, and `rejected` for a
writer whose `create_writer` Spark rejected), and `commit_data_writer` and
`abort_data_writer` a `SparkDsFinishDataWriterInfo`. The Rust and C++ examples
of the docs were rebuilt and run.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala:
##########
@@ -911,6 +979,34 @@ class NativeDataSourceSuite extends QueryTest with
SharedSparkSession {
"msg" -> "create failed with the error code 1."))
}
+ nativeTest("reject a state returned more than once") {
+ def check(action: String, name: String, function: String)(f: SparkSession
=> Unit): Unit = {
+ val e = intercept[Exception](f(newSession(defaultPackage)))
+ checkError(
+ exception = findError(e, "NATIVE_DATA_SOURCE_ERROR"),
+ condition = "NATIVE_DATA_SOURCE_ERROR",
+ parameters = Map("action" -> action, "name" -> name,
+ "msg" -> s"$function returned 2 states instead of one."))
+ }
+ check("plan a scan of", "native_range", "partitions") { session =>
+ range(session, "states" -> "2").collect()
+ }
+ check("plan a scan of", "native_counter", "stream_partitions") { session =>
+ val query = session.readStream.format("native_counter").option("states",
"2").load()
+ .writeStream.format("noop").start()
+ try query.processAllAvailable() finally query.stop()
+ }
+ val dir = Utils.createTempDir()
+ check("plan a write to", "native_sink", "create_writer") { session =>
+ session.range(2).write.format("native_sink").option("path", dir.getPath)
+ .option("states", "2").mode("append").save()
+ }
+ // The rejected writer is aborted, with no messages, so that the library
can undo what
+ // create_writer prepared, and released.
+ assert(Files.readString(new File(dir, "_ABORTED").toPath) == "0")
Review Comment:
Done in d597801: `_ABORTED` now records the NULL messages, the number of
messages and `rejected` (`0,0,1` for a rejected writer). The test also covers
the rejected writer of a streaming write: `_ABORTED`, not `_ABORTED_0`, then
`_CLOSED`.
--
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]
