gengliangwang commented on PR #59313:
URL: https://github.com/apache/spark/pull/59313#issuecomment-6198311277
I have two suggestions on the DSv2 adapter:
1. **Separate driver-side planning from executor-side reading.**
[`ColumnarReader`](https://github.com/apache/spark/blob/e3ff80cc6bba0bbdd63cf260bc3878f772401fc8/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarDataSource.scala#L62-L90)
currently combines pushdown negotiation and partition planning with
serialization and `read(partition)` on executors. `NativeDataSourceReader`
switches between these roles using a transient native handle and mutable
`readerState`. Also, `ColumnarScanBuilder.build()` passes closures that capture
the mutable builder, alongside the scan's schema and pushdown metadata.
Could we keep `ColumnarTable -> ColumnarScanBuilder -> ColumnarScan`, but
introduce a driver-only `BatchReadPlanner` whose `plan()` returns
`PlannedBatch(partitions, readerFactory)`? The executor factory would contain
only the library location, source name, read schema, and serialized reader
state. `build()` could establish stable scan metadata and explicit planner
ownership, with defined behavior for repeated calls.
This can preserve the native ABI and the current lazy partition-planning
point: call `partitions`, then `serializeReader`, and close the native reader
in `finally`. The batch adapter would cache the planned result, and we should
retain the existing caching of planning failures as well. Batch planner
creation should remain lazy for streaming-only sources. Plans that never reach
partition planning would still need the Cleaner fallback, since DSv2 has no
general scan-close hook. The benefit is a clearer driver/executor boundary and
resource ownership.
2. **Have `Table.capabilities()` reflect the individual source's supported
operations.**
[`ColumnarTable.capabilities()`](https://github.com/apache/spark/blob/e3ff80cc6bba0bbdd63cf260bc3878f772401fc8/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarTable.scala#L55-L56)
currently advertises batch/streaming reads and writes plus truncate for every
source, including read-only libraries. Unsupported operations are then
discovered when the corresponding native creation method is called.
Could the adapter receive a source-specific capability set, so Spark can
reject unsupported operations earlier? For native sources, this needs a
reliable discovery contract, such as capability metadata or an explicit
capability query, covering both packaged and installed libraries. That
mechanism is a separate design choice: a new native query would need an ABI
compatibility plan, while the planner/factory split above can remain entirely
internal.
--
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]
