FelixYBW opened a new issue, #13333: URL: https://github.com/apache/gluten/issues/13333
### Description # Spark PR #59313 (native data sources) and Gluten's Arrow work PR: [apache/spark#59313](https://github.com/apache/spark/pull/59313), by HyukjinKwon, open, targets master (4.4.0). Related: our plan in [arrow_udf_spark42_plan.md](arrow_udf_spark42_plan.md), tracking issue [apache/gluten#13140](https://github.com/apache/gluten/issues/13140). Sources read (2026-10-15): | Repo | Ref | |---|---| | Spark PR | `gh pr diff 59313` | | Spark upstream | `origin/master`, `origin/branch-4.2` | | Spark (our branch) | `felix/SPARK-57468-convention-4.2` | | Gluten | local `main`-based checkout | Everything below comes from reading the code. None of it has been run. --- ## Q1. How is this PR connected to our Arrow work? They are separate efforts. #59313 doesn't depend on SPARK-57468 (apache/spark#59154) and doesn't mention the convention API or `ArrowBatchType`. But it uses the same pieces we do. ### What the PR does - **What it adds:** native data sources, meaning DSv2 sources written in Rust or C++ as a shared library. The library implements JNI functions on `org.apache.spark.sql.datasource.NativeBridge`. - **How data moves:** only as Arrow columnar batches, through the Arrow C Data Interface. It adds `org.apache.arrow:arrow-c-data` as a new Spark dependency. - **Read path:** the library exports an `ArrowArrayStream`. Spark imports it into `ArrowColumnVector`s with `supportColumnarReads = true`. - **Write path:** DSv2 writes are row-based, so `ArrowBatchingDataWriter` turns rows into Arrow with Spark's `ArrowWriter`. It then exports them to the library through C Data. - **Allocator:** children of `ArrowUtils.rootAllocator`. ### Where it connects to our work | Our piece | Overlap in #59313 | |---|---| | Velox <-> Java Arrow over C Data (`LoadArrowData` / `OffloadArrowData`) | The same mechanism (C Data import/export), now inside Spark core. Spark gets `arrow-c-data` itself | | "Arrow write": replacing `ArrowWritableColumnVector` with Spark's `ArrowColumnVector` + `ArrowWriter` | It uses exactly that pair, which supports treating it as Spark's standard Arrow vector API | | Gluten's task allocator as a child of `ArrowUtils.rootAllocator` (`ec3f05360d`) | Its readers and writers allocate from children of the same root, so Gluten could move their buffers with `adopt()` instead of copying them | | SPARK-57468 convention API / `ArrowBatchType` | **Missing.** The scan gives back `ColumnarBatch`es of `ArrowColumnVector`, which Spark and Gluten only see as `VanillaBatchType` | ### What happens with Gluten today - **Read:** `BatchScanExec` over a native source can't be offloaded. Gluten sees vanilla columnar output and goes through rows (`ColumnarToRow -> RowToVeloxColumnar`), even though the data is Arrow from start to finish. If the scan declared `ArrowBatchType`, our bridge would map it to `ArrowJavaBatchType` and insert `ArrowColumnarToVeloxColumnarExec`: zero-copy for fixed-width columns, and no Gluten code for this source. - **Write:** Velox output goes to rows, then `ArrowWriter`, then C Data export. With a columnar write path that requires `ArrowBatchType`, Gluten could pass Arrow straight from `LoadArrowData`. ### Suggested next steps 1. Comment on #59313 (or #59154): with SPARK-57468, native scans can declare `ArrowBatchType`. That makes them a second user of the convention API after `ArrowEvalPythonExec`. 2. Ask that the batch lifetime rules match our `ArrowBatchType` ownership contract: a batch is valid until the next `next()`, and consumers transfer the vectors to keep them. 3. Add "native DSv2 scan/write on `ArrowBatchType`" as a **Should** item in section 4 of `arrow_udf_spark42_plan.md`. --- ## Q2. If the next operator is an Arrow Python UDF, is the data copied to the UDF? Does the UDF take ownership of the C Data? Who releases it, and how? **Short answer:** going to Python is always one copy, because the UDF runs in a separate Python worker process and gets an Arrow IPC stream over a socket. The Python side never sees the C Data structs. The JVM takes ownership of them when it imports them, and the library's `release` callback runs when the last Java buffer that points into them is freed. Plan: `BatchScanExec(NativeDataSourceV2)` -> `ArrowEvalPythonExec`. ### Is data copied? | Step | Copy? | |---|---| | Library -> JVM (`Data.importIntoVectorSchemaRoot`) | **None.** Java `ArrowBuf`s wrap the library's memory | | JVM -> Python worker | **One copy, serialization only.** With `spark.sql.execution.arrow.pythonUDF.columnarInput.enabled` (default `true`, 4.2), `ColumnarArrowPythonInput.writeArrowDirect` sees `ArrowColumnVector`s. It takes their `FieldVector`s and writes them with `VectorUnloader` + `MessageSerializer` into the socket. No `ColumnarToRow`, no `ArrowWriter` | | Python -> JVM (UDF results) | Read from IPC into new vectors in Spark's allocator (a copy) | | Pass-through columns | Not copied. The output batch reuses the input column objects | If the conf is off, or a UDF argument isn't a plain column, it falls back to rows, then `ArrowWriter`, then IPC. That's one more full copy. ### Does the Python UDF take ownership of the C Data? No. Only the JVM takes ownership: - The library fills in an `ArrowArray` that carries its own `release` callback. - `Data.importIntoVectorSchemaRoot` moves that struct into a reference-counted holder. Each imported buffer is an `ArrowBuf` created with `wrapForeignAllocation` and holds one reference to it. - The `array.close()` in `NativePartitionReader.next()` only frees the small struct that Java allocated, not the data. - Python gets bytes over the socket and owns its own pyarrow copy. Python's GC frees it. ### Who releases the C Data, and how? The last Java buffer that refers to an `ArrowArray` drops its reference. That calls the library's `release` callback, which frees the native memory. This happens in one of three places: 1. **On the next batch:** `importIntoVectorSchemaRoot` loads new buffers into the **same** `VectorSchemaRoot`, which clears the old ones. This is the usual case, because the PR imports every batch of a partition into the same vectors. 2. **When the reader closes:** `NativePartitionReader.close()` calls `root.close()`, then `stream.release()` (the stream's own callback), then `allocator.close()`. 3. **With our SPARK-57468 ownership contract**, if the scan declared `ArrowBatchType`: `ArrowBatchOwner.own()` moves the buffers out with `getTransferPair` and leaves the reader's vectors empty. Release happens in `releaseOutput()` after the output batch is used, or in `close()` at the end of the task. The reader's allocator is a child of `ArrowUtils.rootAllocator` with no listener, so `allocatorFor` moves the buffers into the owner's allocator. Memory accounting: imported memory counts against the Arrow allocator limit but not Spark's memory manager, since the reader's allocator has no listener. ### Problem: reused vectors vs. the UDF's pass-through queue - The native reader reuses one `VectorSchemaRoot` and the **same `ReusedArrowColumnVector` objects** for every batch. - The scan reports `VanillaBatchType`, so `arrowBatchInput = false` and no `ArrowBatchOwner` is created. - `ColumnarArrowEvalPythonEvaluatorFactory.evalArrowColumnar` still takes path 1. It holds references to the input columns in `passThruQueue` until the matching UDF result comes back. Its comment assumes "Arrow vectors are independently allocated per batch", which isn't true here. - The Python runner can read input N+1 while it's still waiting for result N (Spark 4.x writes whenever the socket can take more). That import replaces the buffers under the queued column objects. So batch N's UDF results would be paired with batch N+1's pass-through columns. If the batches have the same row count, `assert(passThruRows == numRows)` doesn't fire and the results are silently wrong. Otherwise the assert fails. Two ways to fix it: - **Our contract:** the native scan declares `ArrowBatchType`, which turns on `ArrowBatchOwner`. The transfer makes it safe without a copy. - **Without SPARK-57468:** `ColumnarArrowEvalPythonEvaluatorFactory` treats reused or non-freeable vectors as "not independently allocated". It could use path 2, or transfer the buffers there too. To confirm: a repro test in `ArrowColumnarPythonUDFSuite` with an Arrow source that reuses vectors, plus a pandas UDF and a pass-through column. --- ## Q3. Is that answer about our `SPARK-57468-convention-4.2` branch or about PR #59313? Both. The Q2 answer joins code from two places, and no branch has both today: | Part of the answer | Source | |---|---| | Native reader: C Data import, reused `VectorSchemaRoot`, `ReusedArrowColumnVector`, release via `close()` | **PR #59313** (not on our branch) | | UDF side: `columnarInput.enabled`, `writeArrowDirect`, `evalArrowColumnar` path 1, `passThruQueue` | **SPARK-56350**, already upstream, in both `origin/master` and `origin/branch-4.2` | | `ArrowBatchOwner`, `arrowBatchInput`, `ArrowBatchType`, release case 3 | **Only our `SPARK-57468-convention-4.2` branch.** Not on `origin/master` or `origin/branch-4.2` | What this means: - **The problem shows up with #59313 alone.** #59313 targets master. Master already has the SPARK-56350 path 1 code, which assumes vectors are allocated per batch, and it doesn't have `ArrowBatchOwner`. Our branch isn't involved. - **Our branch is the fix, not the cause.** Our branch alone doesn't hit the problem, because it has no native data source. The ownership transfer only helps once someone makes the native scan declare `ArrowBatchType`, and neither PR does that yet. - **The copy and release answers come from #59313 plus upstream master.** Release cases 1–2 are #59313's own code. The one serialization copy to Python is master's SPARK-56350 path. The wrong-results problem is still unconfirmed. The repro test belongs on **master + #59313**, not on our branch. If it fails, raise it on #59313, with SPARK-57468's ownership contract as one possible fix. --- ## Q4. What's the difference between our Arrow C Data wrapper in the JVM and the PR's? They aren't two wrappers. Both call the same library, `org.apache.arrow:arrow-c-data`, through the same calls (`Data.importIntoVectorSchemaRoot`, `Data.exportVectorSchemaRoot`, `CDataDictionaryProvider`). What differs is how each one uses it. - **Ours:** `ColumnarBatches.load/offload` + `ArrowAbiUtil` in `gluten-arrow`, and `ColumnarBatchJniWrapper` in `cpp/core/jni/JniWrapper.cc`. - **The PR's:** `NativePartitionReader` and `NativeDataWriter` in `execution.datasources.v2.ffi`. ### Read direction (native -> JVM) | | Gluten `ColumnarBatches.load` | PR `NativePartitionReader` | |---|---|---| | Unit | One `ArrowArray` per batch. Java calls `exportToArrow(handle, ...)` for each Velox batch | An `ArrowArrayStream`: the library drives it with `get_schema` / `get_next` | | Schema | Imported and **re-exported on every batch** (`Data.exportSchema(toArrowSchema(cSchema))`), then imported again | Imported **once per stream**, then checked against the expected Spark schema | | Vectors | **A new `VectorSchemaRoot` per batch** (`Data.importVectorSchemaRoot`) | **One reused root for the partition.** Every batch goes into the same vectors | | Column class | `ArrowWritableColumnVector` on main; `RefCountedArrowColumnVector` on our branch. Both reference-counted | `ReusedArrowColumnVector`: `closeIfFreeable()` does nothing, no reference count | | Lifecycle | Copies the input's ref count onto the new columns, closes the input, then `transferVectors(output, input)`, so the **input batch is changed in place** | Batch is valid until the next `next()`. The reader closes everything in `close()` | | Checks | Trusts Velox's output | Rejects offset != 0, children shorter than the parent refers to, dictionary encoding, non-128-bit decimals and list views (Arrow Java would read these wrong without an error) | | Dictionaries | A new `CDataDictionaryProvider` per call, closed right away | Not supported (rejected) | | Freeing C data | When the batch's vectors are closed (ref count reaches 0) | When the next batch replaces the buffers, or when `root.close()` runs | ### Write direction (JVM -> native) | | Gluten `ColumnarBatches.offload` | PR `NativeDataWriter.write` | |---|---|---| | Building the export | `ColumnarBatch` -> `ArrowRecordBatch` (`VectorUnloader`) -> new VSR (`VectorLoader`) -> `exportVectorSchemaRoot`. The data isn't copied, but buffers and the record batch are wrapped again | Wraps the existing `FieldVector`s in a `VectorSchemaRoot` directly, then exports | | Ownership | **Always moved to native**: `ArrowArrayMove` into `ArrowCStructColumnarBatch`, then `ArrowArrayRelease` in its destructor when the `IndicatorVector` ref count reaches 0 | **Lent for the call.** `finally { array.release() }`. The library either uses the data inside `write` or `ArrowArrayMove`s it to keep it, which turns the release into a no-op | | Allocator for the structs | Gluten's task allocator | `ArrowUtils.rootAllocator` directly | ### Overall differences | | Gluten | PR | |---|---|---| | Allocator | Main: a separate per-task `RootAllocator` with a `ManagedAllocationListener`, so **memory is tracked**. Our branch: a child of `ArrowUtils.rootAllocator`, still with a listener | A child of `ArrowUtils.rootAllocator` for each reader, **no listener**, so memory isn't tracked by Spark's memory manager | | JNI interface | Internal and built together with the jar. No versioning | Public and versioned (`abiVersion`), one class loader per library, optional-function rules | | Errors | `JNI_METHOD_START/END` -> `GlutenException` | Every call goes through `call(name, op)` -> `NATIVE_DATA_SOURCE_ERROR` | | Who calls whom | The JVM pulls one batch at a time through a handle | The library is the producer behind a stream | ### What this means for us 1. **The PR's checks are worth copying.** Arrow Java silently misreads arrays with an offset != 0 or short children. We rely on Velox never exporting those. Running the PR's `checkArrays` in debug mode would catch it if that ever changes. 2. **We can drop the per-batch schema round trip.** Importing the schema once per stream removes a JNI call and an export/import on every batch. 3. **Vector reuse is the main difference in meaning.** Gluten makes new vectors for each batch, which is what SPARK-56350's path 1 assumes. The PR reuses them. That's why the pass-through problem in Q2 hits the PR but not `LoadArrowDataExec`. 4. **Write ownership:** the PR's "lend, move if you keep it" model is simpler than our "always move, plus copy the ref count". It matches the G8 lifecycle rewrite in our plan. 5. **Allocator:** on our branch, both sides share `ArrowUtils.rootAllocator`, so `adopt()` / `ArrowBatchOwner` can move a native source's buffers without copying. The PR's untracked allocator is the case where `allocatorFor` moves them into the consumer's allocator. ### Gluten version None -- 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]
