spark-test commented on code in PR #59931:
URL: https://github.com/apache/spark/pull/59931#discussion_r4420518881
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSource.scala:
##########
@@ -780,6 +822,7 @@ class NativePartitionReader(
throw error(name, "read", s"The batch has an offset of ${batch.offset},
but Spark only " +
"reads arrays with an offset of 0. Copy sliced arrays before exporting
them.")
}
+ checkBuffers(batch, "The batch")
Review Comment:
**Nit (P3):** Only the column-level call of `checkBuffers` has a test:
`"short" -> "buffers"` clears the pointer of column `m` only
(`test_native_datasource.cc:475-477`). The batch struct always gets a valid
pointer from `initArray`, and `validity()` returns 0 for a NULL one, so
deleting this line leaves the suite green, and a batch with a NULL `buffers`
pointer would reach the Arrow import instead of failing with "The batch has 1
buffers, but a NULL pointer to them." An option of the test library that clears
the pointer of the batch struct, next to `"children"` in `exportBatch`, and one
more `check` in "reject the Arrow data that Spark cannot read" would cover it.
**Verification:**
- **Regression:** Reading a batch whose struct array declares buffers but
has a NULL pointer to them fails with NATIVE_DATA_SOURCE_ERROR naming the
batch. This fails if the batch-level check is removed.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSource.scala:
##########
@@ -350,17 +359,16 @@ class NativeDataSourceStreamReader(
}
override def partitions(start: String, end: String): Array[InputPartition] =
{
- val partitions = call(name, "plan a scan of") {
+ val planned = call(name, "plan a scan of") {
library.streamPartitions(handle.get, start, end)
}
- readerState = bytesOrEmpty(
- call(name, "plan a scan
of")(optional(library.serializeStreamReader(handle.get))))
- NativeInputPartition.from(partitions)
+ readerState = bytesOrEmpty(Option(planned.state))
Review Comment:
**Non-blocking (P2):** Each `stream_partitions` call replaces this single
`readerState`, and `read` passes it with every partition. The header now
promises more: `read` gets "the state that partitions or stream_partitions
returned with the partition" (`spark_datasource.h:421`), and `set_state`
returns the state "to read these partitions" (`:655-656`).
The two differ when one micro-batch plans the stream twice, as a streaming
self-union does. `DeduplicateRelations` keeps both branches of a `Union` on one
`StreamingRelationV2` (projection-based deduplication), so
`MicroBatchExecution` maps them to one stream, and so to one
`NativeDataSourceStreamReader`. Each branch still becomes its own
`MicroBatchScanExec`, which calls `planInputPartitions(start, end)` for the
same range, and `UnionExec` builds both RDDs before the job runs. The task
binary then serializes the reader after the second call, and the partitions of
the first call are read with the state of the second.
The test library returns the same reader-wide state from every call, so no
test sees this. A library whose state is specific to the call, such as a
snapshot, or a file listing that its partitions index into, can fail or read
data that does not match its partitions. The batch path is fine:
`NativeBatchPlanner` binds the state of its `partitions` call into the reader
factory of its own plan.
**Recommended change:** Make NativeDataSourceStreamReader plan each range
once: keep, in transient fields that are not serialized to the executors, the
start and end offsets of its last stream_partitions call and the partitions
that it returned. When Spark plans the same range again, return those
partitions without calling stream_partitions again, and replace readerState
only when a new range is planned. Add native streaming tests in which the test
library returns a different state from each stream_partitions call and records
the state that each read gets, for a query that unions a native source with
itself.
**Why this works:** All scans of a source in one micro-batch plan the same
range. With a memo of the last range, they all read the partitions of a single
stream_partitions call, so readerState holds exactly the state returned with
those partitions. A new range replaces both together.
**Scope:** Memoize the last planned range in the native stream reader, and
add a self-union regression test with call-specific states.
**Compatibility:** read keeps getting the state that partitions or
stream_partitions returned with its partition, as the version 1 header states.
**Risks:** If a future engine path planned two different ranges of one
stream before the tasks of the first were serialized, a memo of only the last
range would again pair the earlier partitions with the later state. A memo that
is not transient would serialize the partitions of the last range into the task
binary of each micro-batch, with the reader.
**Constraints:** Keep the rule that the partitions of a range read the same
data each time. A restarted query, with a new stream reader, still calls
stream_partitions again for the last range that it planned. Leave the batch
path unchanged: its planner already binds its state into the reader factory of
its plan.
**Exclusions:**
- Moving the state into each NativeInputPartition, which would change how
the state reaches the tasks.
**Success:** In a streaming self-union of a native source, Spark calls
stream_partitions once for the range of each micro-batch, and every partition
of both branches is read with the state that this call returned. Streaming
reads without a self-union keep their stream_partitions calls, start offsets
and rows.
**Verification:**
- **Regression:** A streaming query that unions a native micro-batch source
with itself runs against a library that returns a different state from each
stream_partitions call. Every partition is read with the state returned with
it, and the library sees one stream_partitions call per micro-batch.
- **Behavior:** Native streaming reads, restarts from a checkpoint and
Trigger.AvailableNow runs keep their start offsets, their rows and the ranges
that stream_partitions plans.
##########
sql/core/src/test/resources/native-datasource/test_native_datasource.cc:
##########
@@ -1033,58 +1125,86 @@ int planPartitions(void* handle, SparkDsSink
add_partition, void* sink) {
int64_t stop = std::min(hi, start + step);
sendBytes(add_partition, sink, std::to_string(start) + "," +
std::to_string(stop));
}
+ sendState(reader->state, set_state, sink, serialize(reader->state));
});
}
-int serializeReader(void* handle, SparkDsSink set_state, void* sink) {
- return guarded([&] { sendBytes(set_state, sink,
serialize(fromHandle<Reader>(handle)->state)); });
-}
-
void closeReader(void* handle) { delete fromHandle<Reader>(handle); }
#ifndef READ_ONLY
-int initialOffset(void*, SparkDsSink set_offset, void* sink) {
- return guarded([&] { sendBytes(set_offset, sink, "{\"offset\":0}"); });
+// Sends the offset, or, with the option `bad_option`, no offset for "none", a
pretty-printed one
+// for "multiline", or one that is not JSON for "invalid".
+void sendOffset(const Options& state, const std::string& bad_option,
SparkDsSink set_offset,
+ void* sink, int64_t offset) {
+ std::string bad_offset = get(state, bad_option, "");
+ if (bad_offset == "none") return;
+ if (bad_offset == "multiline") {
+ // Pretty-printed JSON, as serde_json::to_string_pretty writes it.
+ sendBytes(set_offset, sink, "{\n \"offset\": " + std::to_string(offset) +
"\n}");
+ } else if (bad_offset == "invalid") {
+ sendBytes(set_offset, sink, "offset " + std::to_string(offset));
+ } else {
+ sendBytes(set_offset, sink, "{\"offset\":" + std::to_string(offset) + "}");
+ }
}
-// Each call makes `step` more rows available, up to `max_offset`.
-int latestOffset(void* handle, SparkDsSink set_offset, void* sink) {
+int initialOffset(void* handle, const SparkDsInitialOffsetInfo* info,
SparkDsSink set_offset,
+ void* sink) {
return guarded([&] {
- Options& state = fromHandle<Reader>(handle)->state;
- int64_t latest = std::min(getLong(state, "max_offset", 10),
- getLong(state, "current", 0) + getLong(state,
"step", 10));
- state["current"] = std::to_string(latest);
- std::string bad_offset = get(state, "bad_offset", "");
- if (bad_offset == "none") return;
- if (bad_offset == "multiline") {
- // Pretty-printed JSON, as serde_json::to_string_pretty writes it.
- sendBytes(set_offset, sink, "{\n \"offset\": " + std::to_string(latest)
+ "\n}");
- } else if (bad_offset == "invalid") {
- sendBytes(set_offset, sink, "offset " + std::to_string(latest));
- } else {
- sendBytes(set_offset, sink, "{\"offset\":" + std::to_string(latest) +
"}");
+ checkSize("SparkDsInitialOffsetInfo", info->size,
sizeof(SparkDsInitialOffsetInfo));
+ const Options& state = fromHandle<Reader>(handle)->state;
+ int64_t offset = 0;
+ std::string path = get(state, "initial_offsets_path", "");
+ if (!path.empty()) {
+ FILE* file = std::fopen(path.c_str(), "a+");
Review Comment:
**Nit (P3):** The count depends on where reading starts in an `"a+"` stream,
which C leaves implementation-defined (C11 7.21.3p1). glibc starts at the
beginning of the file, but the BSD-derived libcs, including macOS, start at its
end. There, the loop sees EOF at once and every call returns `{"offset":0}`, so
the option no longer "returns the number of calls before it", and in "ask for
the initial offset once, and keep it in the checkpoint" only the line count
still tells a reused offset from one asked again. A `std::rewind(file);` before
the loop makes the count portable: `fputs` still appends, and since it follows
a read that reached EOF, it needs no positioning call.
**Verification:**
- **Behavior:** With initial_offsets_path, successive initial_offset calls
return the offsets 0, 1 and 2, whatever initial read position the platform
gives append-mode streams.
--
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]
