spark-test commented on code in PR #59942:
URL: https://github.com/apache/spark/pull/59942#discussion_r4420070124
##########
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:
**Nit (P3):** When `reader.defaultReadLimit()` throws here, because
`default_read_limit` fails or returns an unknown kind or a `max_rows` below 1,
the query fails without ever stopping this stream.
`MicroBatchExecution.logicalPlan` asks for the limit while it assigns
`uniqueSources` (MicroBatchExecution.scala:385-413), after `toMicroBatchStream`
has already created the native stream reader (line 301).
`StreamExecution.stopSources` only stops `uniqueSources`, which is still empty
when its assignment throws (StreamExecution.scala:205 and 570-580), so `stop()`
never runs, and `close_stream_reader` waits for the `NativeHandle` Cleaner,
once the failed query is no longer referenced.
The header says that Spark releases a stream reader when the streaming query
stops, so a library whose reader holds connections, threads or files keeps them
past the failed start, one reader per attempt when a driver retries a
misconfigured query. Only the rejection path of a data source with
`SPARK_DS_ADMISSION_CONTROL` is affected, but under every micro-batch trigger,
since `Trigger.Once` and the `Trigger.AvailableNow` fallback still ask for the
limit. The rejection test only checks the error.
Stopping the reader before rethrowing would release it, for any
`ColumnarStreamReader`, for example as `createWriter` in `NativeLibrary` does:
```scala
private lazy val readLimit =
try reader.defaultReadLimit() catch {
case e: Throwable =>
// The query fails before Spark records the streams to stop, so stop
this one here.
try reader.stop() catch { case stopError: Throwable =>
e.addSuppressed(stopError) }
throw e
}
```
**Verification:**
- **Regression:** Start a streaming query on a native source whose
default_read_limit is rejected, both for an unknown kind and for a maximum of 0
rows, with the test library's closed-path marker. Check that the query still
fails with the same NATIVE_DATA_SOURCE_ERROR, and that the stream reader was
closed when the query terminated, without relying on garbage collection.
- **Behavior:** Check that queries with a valid default read limit, and
sources without admission control, still read the same micro-batches and close
their stream reader only when the query stops.
##########
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:
**Nit (P3):** `SparkDsReadLimit` also starts with its size and is passed to
the library, as the argument of `default_read_limit` and through
`SparkDsLatestOffsetInfo.limit`, but its name does not end with `Info`, so this
list leaves it out, and the guide never names it. It is also the one struct
that a library writes into, while the rules below only say that a library reads
the fields that fit in the size of a struct that Spark passes. The header has
the matching rule at the struct: "In default_read_limit, the library sets the
fields that fit in the size that Spark set." Following only the guide, a
library built against a later version-1 header, whose `SparkDsReadLimit` has
grown, could fill its own larger struct into the smaller one that an earlier
Spark allocated.
Could this list include `SparkDsReadLimit`, or say "the structs that Spark
passes to the library" as the header does, and the rules add that
`default_read_limit` only sets the fields that fit in the size that Spark set?
**Verification:**
- **Inspection:** Check that the Compatibility section of the native data
source guide covers every size-prefixed struct that Spark passes to a library,
including SparkDsReadLimit, and states both the read-side and the write-side
size rules consistently with spark_datasource.h.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala:
##########
@@ -911,6 +1162,49 @@ class NativeDataSourceSuite extends QueryTest with
SharedSparkSession {
"msg" -> "create failed with the error code 1."))
}
+ nativeTest("reject a state returned more than once") {
Review Comment:
**Nit (P3):** This covers a state set twice, but no test covers a state that
is never set, which is how a stateless reader or writer now says that it has
none, as the header's optional functions convention says and the guide's Rust
and C++ examples do; no test runs those examples. `NativeLibrary.state` returns
null for zero states, and the callers turn it into an empty state with
`bytesOrEmpty(Option(...))`. But `test_native_datasource.cc` always sends a
state unless `states` says otherwise, and its `read` and `create_data_writer`
fail without one, since they check the kind and the fields kept in it. So a
change that rejected a missing state, or passed null to the executors, would
break every stateless library while this suite still passes.
Could the test library get a stateless mode, in which `partitions`,
`stream_partitions` and `create_writer` never call `set_state`, and `read` and
`create_data_writer` do not need a state?
**Verification:**
- **Regression:** Run a batch read, a micro-batch streaming read and a batch
write against a native library whose partitions, stream_partitions and
create_writer never call set_state. Check that they return the expected rows
and that read and create_data_writer get a state of length 0, so that any
change that rejects or mishandles a missing state fails.
--
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]
