zhouyuan commented on PR #59424: URL: https://github.com/apache/spark/pull/59424#issuecomment-6196064644
Moving my comment from #59313 here, updated for the C header. Thanks for this. The plain C ABI also makes native data sources a good fit for native execution engines running inside Spark, such as [Apache Gluten](https://github.com/apache/incubator-gluten) (Velox) and Comet (DataFusion): as the description says, other hosts can load the same libraries. The library reads the data, and the engine runs the rest of the query natively. What's missing is on the Spark side. **Today**, a query on a native data source with Gluten enabled looks like this: ``` read -> ArrowArrayStream -> Arrow Java -> ArrowColumnVector -> ColumnarToRow -> RowToVeloxColumnar -> Velox operators ``` That's two extra copies, so turning on a native engine can make these queries slower. With a small hook, the data could stay columnar in native memory from the source to the shuffle: ``` read -> ArrowArrayStream -> Velox (import) -> filter / agg / join / shuffle ``` A source library written once would then run on vanilla Spark and get native downstream execution under Gluten or Comet, without per-source connector work in each engine. **Two small requests:** 1. **A developer API to open a partition as an Arrow C stream.** The library location, reader state and partitions are private to `execution.datasources.v2.ffi`, so an external engine can't call `read` for a planned scan today. Something like: ```scala @DeveloperApi trait SupportsArrowStreamRead { this: PartitionReaderFactory => /** Initializes the ArrowArrayStream at `streamAddress` with the data of `partition`. * The caller owns the stream and validates the arrays it reads. */ def openArrowStream(partition: InputPartition, streamAddress: Long): Unit } ``` `ColumnarPartitionReaderFactory` would implement it by calling `read` with Spark's own `SparkDsTaskContext`, which is freed when the caller releases the stream. Library loading, the checksum checks and `spark_ds_init` stay in Spark, and the vanilla path stays exactly as it is. 2. **A threading contract for the stream in the header.** The doc of `read` says Spark calls the stream "from the thread of the task that reads the partition", so library authors may rely on it. A native engine may call `get_next` from its own driver threads. Could version 1 guarantee instead that the stream may be called from any thread, though never concurrently? The functions of `SparkDsHost` and `SparkDsTaskContext` can already be called from any thread, so this would only extend that to the stream. It's free now, but hard to add once libraries depend on the current wording. **Optional, could be a follow-up:** a columnar write entry point (the internal `DataWriter[ColumnarBatch]` that takes Arrow batches), so a native engine can write to a native sink without going through rows. If this direction sounds reasonable, I'm happy to help review. -- 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]
