HyukjinKwon opened a new pull request, #59942: URL: https://github.com/apache/spark/pull/59942
### What changes were proposed in this pull request? Admission control for micro-batch streaming reads of native data sources, appended within version 1 of `spark_datasource.h`. The guide of SPARK-60101 (#59424) lists it as a limitation. This PR is based on #59931 (SPARK-60723), whose commits it includes until that PR is merged. - **Capability and default limit.** A data source with the new capability `SPARK_DS_ADMISSION_CONTROL` returns the limit on the data of each micro-batch with `default_read_limit`, appended to `SparkDsLibrary`, in a size-prefixed `SparkDsReadLimit`: `SPARK_DS_READ_MAX_ROWS` with a maximum number of rows, or `SPARK_DS_READ_ALL_AVAILABLE`, typically from an option of the data source. Spark asks once, when the query starts. It fails the query for an unknown kind of limit, a maximum below one row, or a library that declares the capability without the function. - **Limited latest offsets.** `SparkDsLatestOffsetInfo` gets the limit, `limit_kind` and `max_rows`: the latest offset is then the end of the data of the next micro-batch, within the limit. Spark passes the default limit, or `SPARK_DS_READ_ALL_AVAILABLE` for `Trigger.Once`, as for the other sources. - **`Trigger.AvailableNow`.** Spark first asks for the latest offset without a limit, as before, and then passes it as `end` with each limited request, so the library never returns an offset past it. The run then reads the data that was available when it started, in micro-batches within the limit. Spark compares no offsets itself, since they are defined by the library. - **Progress.** With a limit, the progress of the query reports the latest offset without the limit, as Spark's other limited sources do. On the Spark side, `ColumnarStreamReader` gets `defaultReadLimit()` and the limit and the end in `latestOffset`, which `ColumnarMicroBatchStream` maps from `getDefaultReadLimit` and `latestOffset(start, limit)`. The header, the FFM bindings, the C++ test library and the guide are updated: admission control leaves the Limitations of the guide, and the capability table lists it. ### Why are the changes needed? Without a limit, every micro-batch reads all the data that is available, which can make the first micro-batch of a large source, or a backlog after a restart, too large. Spark's own sources offer such limits, for example `maxOffsetsPerTrigger` for Kafka and `maxFilesPerTrigger` for files. ### Does this PR introduce _any_ user-facing change? No for released versions: native data sources are new in the unreleased 4.4.0. Data sources without the new capability, and libraries built against the header before these additions, keep reading all the available data. ### How was this patch tested? New tests in `NativeDataSourceSuite`: - "limit the rows of each micro-batch with admission control" checks micro-batches of 5, 5 and 2 rows, and the reported latest offset without the limit. It also checks `Trigger.AvailableNow` within the limit, and a single micro-batch without the capability. - "reject the read limits that a micro-batch cannot have" covers an unknown kind of limit and a maximum of 0 rows. It also uses a library whose table ends before `default_read_limit`, as one built against the earlier header, that declares the capability. The existing suites pass, and the C++ example of the guide builds against the new header. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) This pull request and its description were written by Isaac. -- 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]
