HyukjinKwon commented on code in PR #59942:
URL: https://github.com/apache/spark/pull/59942#discussion_r4423325854
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarTable.scala:
##########
@@ -189,11 +220,73 @@ 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]]. The reader gets the
offset that each
+ * micro-batch starts after, the end of the last one, for every trigger, and
the limit on the data
+ * of a micro-batch, its default read limit, all the available data unless it
limits it. With
+ * `Trigger.AvailableNow`, Spark runs micro-batches up to the first latest
offset without a limit
+ * that Spark asks for, rather than wrapping the stream in one that asks for
the latest offset from
+ * the initial offset: in one micro-batch, or within the limit in several.
+ *
+ * @param checkpointLocation the location of the checkpoint of the source,
where the stream keeps
+ * the initial offset, under
`ColumnarMicroBatchStream.InitialOffsetLog`
+ */
+class ColumnarMicroBatchStream(reader: ColumnarStreamReader,
checkpointLocation: String)
+ extends MicroBatchStream with SupportsTriggerAvailableNow {
+
+ // The latest offset without a limit that the reader returned last, which
the progress of the
+ // query reports.
+ @volatile private var lastLatestOffset: Offset = _
+ // The limit on the data of each micro-batch, asked once.
+ private lazy val readLimit = reader.defaultReadLimit()
Review Comment:
Done in 0b94e5b: `ColumnarMicroBatchStream` now stops its reader when
`defaultReadLimit` fails, and rethrows the error, with an error of `stop` added
as suppressed, so any `ColumnarStreamReader` is released when Spark fails the
query before it records its streams. "reject the read limits that a micro-batch
cannot have" now passes `closed_path`, and checks that the stream reader is
closed by the time the query fails, for the unknown kind and for the maximum of
0 rows, and that the library without `default_read_limit` fails before Spark
creates a stream reader.
##########
docs/sql-native-data-sources.md:
##########
@@ -172,17 +207,19 @@ regularly and use timeouts. The task reaper,
`spark.task.reaper.enabled` and
### Compatibility
The function table, `SparkDsLibrary`, and the structs that Spark passes to the
library,
-`SparkDsHost`, `SparkDsCreateInfo`, `SparkDsTaskContext` and
`SparkDsWriteInfo`, start with their
-size, so that they can grow. The version of the interface, 1, is
`SPARK_DS_ABI_VERSION`, which the
-table and the manifest of a package declare. Within a version:
+`SparkDsHost`, `SparkDsTaskContext` and those whose names end with `Info`,
such as
Review Comment:
Done in 0b94e5b: the Compatibility section of the guide now lists
`SparkDsReadLimit` with the structs that start with their size, and says that a
library only sets the fields that fit in the size of a struct that it fills,
such as the `SparkDsReadLimit` of `default_read_limit`. The Compatibility
section of the header says the same, and the guide names `SparkDsReadLimit`
where it describes `default_read_limit`.
--
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]
