LuciferYang commented on PR #59313: URL: https://github.com/apache/spark/pull/59313#issuecomment-6195888020
> Thanks for this, it's a really nice design. Since the PR is still a draft, I'd like to raise a use case from [Apache Gluten](https://github.com/apache/incubator-gluten), the native execution engine for Spark built on Velox. Other native engines such as Comet (DataFusion) are in the same position. > > **Why it matters.** Native data sources and native execution engines fit together well. The library reads the data, and the engine runs the rest of the query natively. Today, a query on a native data source with Gluten enabled looks like this: > > ``` > lib.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: > > ``` > lib.read -> ArrowArrayStream -> Velox (import) -> filter / agg / join / shuffle > ``` > > No C ABI change is needed for this. The JNI `read` call only initializes the `ArrowArrayStream`. After that, `get_next` and `release` are plain C function pointers that a native engine can call directly. 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`, and each library is loaded through its own copy of `NativeBridge`, so an external engine can't do this 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, and the vanilla path would stay exactly as it is. > 2. **A threading contract for the stream in the `NativeBridge` docs.** The docs currently say Spark calls the stream "from the thread of the task that reads the partition". Could ABI v1 instead guarantee that `get_next`/`release` don't use JNI or the JVM, may be called from a thread other than the one that called `read` (never concurrently), and report errors through `get_last_error`? 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. @zhouyuan should have commented on https://github.com/apache/spark/pull/59424. -- 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]
