HyukjinKwon commented on PR #59424:
URL: https://github.com/apache/spark/pull/59424#issuecomment-6204554351
@gengliangwang Thank you for both suggestions, and for the Rust SDK
prototype in #59520.
1. **Planning on the driver, reading on the executors**: done in
534dfca15723, as you suggested. `ColumnarReader` is replaced by a driver-only
`ColumnarBatchPlanner`, which gets the pushdown, and whose `plan()` returns
`ColumnarPlannedBatch(partitions, readerFactory)`. For a native data source,
`NativeBatchPlanner` holds the native reader: `plan()` calls `partitions`, then
`serialize_reader`, and closes the reader in `finally`. The serializable
`NativePartitionReaderFactory` only has the location of the library, the name
of the data source, the read schema and the state of the reader.
`ColumnarScanBuilder.build()` builds its scan once, and the scan takes over the
planner that the operations were pushed down to, with no closure over the
builder. A scan whose operations were not pushed down only creates a planner if
its batch is planned, so a streaming scan never creates one.
`ColumnarBatchScan` plans once, when Spark first asks for the partitions or the
reader factory, and keeps th
e result or the failure. The `Cleaner` still releases the reader of a scan
that is never planned. Tests: `build and plan a batch scan once` in
`ColumnarDataSourceSuite`, and `a scan whose planning failed fails again with
the same error` in `NativeDataSourceSuite`, now through `ColumnarBatchScan`.
2. **Capabilities in `Table.capabilities()`**: done for the writes.
`ColumnarTable.capabilities()` now asks the data source, once per table, with
the options of the table: for a native data source, Spark creates a data source
with them and reads its capabilities. The table reports `BATCH_WRITE` and
`TRUNCATE` as the data source declares them, so Data Source V2 rejects a batch
write or overwrite that the source does not support while it analyzes the
query, with `UNSUPPORTED_FEATURE.TABLE_OPERATION`, before it computes the data
to write. Before, with adaptive execution, the shuffle stages of the data to
write ran first.
The table always reports `BATCH_READ`, `MICRO_BATCH_READ` and
`STREAMING_WRITE`. When a `TableProvider`'s table lacks one of them, Data
Source V2 does not reject the query: it falls back to Data Source V1, in
`DataFrameReader.load`, `DataStreamReader.load` and `DataStreamWriter.start`.
The query then fails with `<format> is not a valid Spark SQL Data Source.` or
`Data source <format> does not support streamed reading.` (legacy errors),
which do not say that the data source does not declare the operation. So those
operations fail when Spark plans them instead, still on the driver and before
any task, with `DATA_SOURCE_BATCH_SCAN_NOT_SUPPORTED`,
`DATA_SOURCE_MICRO_BATCH_SCAN_NOT_SUPPORTED` or
`DATA_SOURCE_STREAMING_WRITE_NOT_SUPPORTED`.
On the semantics: the table has the options of the table, which for a
table in a catalog are those of `CREATE TABLE`, and the options of a write can
override them. So, as you said, the checks of each operation remain, with its
own options. The header and the docs say so.
The test `operations that a data source does not support` covers both: a
batch write whose data would fail in a shuffle stage now fails while the query
is analyzed, and a write to a table in a catalog whose options remove a
capability of the table fails when it is planned.
--
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]
