gengliangwang commented on PR #59424: URL: https://github.com/apache/spark/pull/59424#issuecomment-6481066910
For tracking, here are the rest of my suggestions from the Rust SDK side (#59520), which can be follow-ups, after [the three above](https://github.com/apache/spark/pull/59424#issuecomment-6480975990). Checked at `1f40484`. 1. **Give the stream reader the inputs of `toMicroBatchStream` and of admission control.** [`ColumnarTable.toMicroBatchStream`](https://github.com/apache/spark/blob/1f4048404befa88ac1b956e982711388bf694520/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarTable.scala#L167-L168) drops `checkpointLocation`, since [`create_stream_reader`](https://github.com/apache/spark/blob/1f4048404befa88ac1b956e982711388bf694520/sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h#L419) only takes the schema, and [`latest_offset`](https://github.com/apache/spark/blob/1f4048404befa88ac1b956e982711388bf694520/sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h#L492) has no place for the start offset and the `ReadLimit` of `SupportsAdmissionControl.latestOffset`, so a source cannot limit the size of a micro-batch (#59520 leaves admission control out for this reason). These functions take positional arguments, so they can only get new inputs through new functions, unless they take size-prefixed stru cts, like `create` and `create_writer`, before 4.4.0. 2. **Prune the columns of streaming reads.** [`toMicroBatchStream`](https://github.com/apache/spark/blob/1f4048404befa88ac1b956e982711388bf694520/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarTable.scala#L167-L168) creates the stream reader with `fullSchema`, and [`prune_columns`](https://github.com/apache/spark/blob/1f4048404befa88ac1b956e982711388bf694520/sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h#L456) only takes a batch reader, so a streaming read always reads every column. It could pass the pruned schema to `create_stream_reader`, or call a pruning function on stream readers, with a capability. The same would apply to predicates when they come back. 3. **Make the order of the calls on each handle a rule of the header.** Spark already keeps a failed planning and stops a stream after a failure, but the header does not promise it, so the Rust SDK keeps the first failure of a stream reader and returns it from every later call ([`StreamReader::call`](https://github.com/apache/spark/blob/19351be1babb11b5ea97b2d7a72dc31374f439ac/sql/native/rust/src/bridge.rs#L559-L568)), and tracks each writer through [`Phase`](https://github.com/apache/spark/blob/19351be1babb11b5ea97b2d7a72dc31374f439ac/sql/native/rust/src/write.rs#L37-L42) to guard against repeated commits and aborts. Two rules would let SDKs drop most of this: after a function on a handle fails, Spark only calls its abort or close function; and Spark calls `commit_data_writer` and `abort_data_writer` at most once for each data writer, and never `write` after either. This only changes comments. 4. **Tell the library which log levels are enabled.** [`SparkDsHost.log`](https://github.com/apache/spark/blob/1f4048404befa88ac1b956e982711388bf694520/sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h#L286) does not say which levels Spark logs, so the Rust SDK [passes `Info` and more severe messages](https://github.com/apache/spark/blob/19351be1babb11b5ea97b2d7a72dc31374f439ac/sql/native/rust/src/host.rs#L89) unless the library sets another level: debug messages are lost when the log of Spark enables them, and formatted for nothing otherwise. A function appended to `SparkDsHost`, such as `bool (*is_log_enabled)(int32_t level)`, would let an SDK follow the level of the log of Spark, also when it changes at runtime. -- 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]
