|
1 | 1 | # Spark |
2 | 2 |
|
3 | | -The `vortex-spark` connector implements Apache Spark's DataSource V2 API, allowing Spark to read |
4 | | -and write Vortex files as a native data source registered under the format name `vortex`. |
| 3 | +The `vortex-spark` connector is built on Spark's file-source DataSource V2 framework. Shared |
| 4 | +sources compile into Spark 3.5/Scala 2.12, Spark 3.5/Scala 2.13, and Spark 4.0/Scala 2.13 |
| 5 | +artifacts; the Spark 4 artifact is also tested on Spark 4.1. |
5 | 6 |
|
6 | | -## Registration |
| 7 | +## File-source integration |
7 | 8 |
|
8 | | -The connector implements Spark's `TableProvider` and `DataSourceRegister` interfaces. When a |
9 | | -query references the `vortex` format, Spark creates a `VortexTable` that supports both batch |
10 | | -reads (`SupportsRead`) and writes (`SupportsWrite`). Schema inference reads the footer of a |
11 | | -discovered file to extract the Arrow schema and map it to Spark's schema representation. |
| 9 | +`VortexDataSourceV2` extends `FileDataSourceV2`, while `VortexTable`, `VortexScan`, and |
| 10 | +`VortexWrite` use Spark's `FileTable`, `FileScan`, and `FileWrite` abstractions. Spark therefore |
| 11 | +owns file listing, partition discovery and pruning, input bin-packing, output commit, and |
| 12 | +overwrite behavior. `VortexFileFormat` supplies the functional V1 fallback used by catalog |
| 13 | +tables and direct path queries. |
12 | 14 |
|
13 | | -## Multiple Files |
| 15 | +Vortex files are not internally split. A Spark file partition may contain several files, and the |
| 16 | +reader factory opens each `PartitionedFile` in turn. Hive partition values are appended with |
| 17 | +constant column vectors. |
14 | 18 |
|
15 | | -Spark's scan builder enumerates Vortex files by scanning the provided path. If the path is a |
16 | | -directory, native code lists all `.vortex` files within it. Each file becomes an independent |
17 | | -input partition, and Spark's task scheduler distributes partitions across executors in the |
18 | | -cluster. |
| 19 | +## I/O and JNI |
19 | 20 |
|
20 | | -Each partition creates its own file handle and scan state, so there is no shared mutable state |
21 | | -between partitions. This maps naturally to Spark's execution model where each task runs |
22 | | -independently on a separate JVM thread. |
| 21 | +Spark lists paths through Hadoop. Content reads default to pooled Hadoop input streams exposed to |
| 22 | +native Vortex through the JNI `NativeReadable` interface. Writes expose the committer's Hadoop |
| 23 | +task path through `NativeWritable`. Native object-store reads are available as an explicit option, |
| 24 | +but writes always follow the Hadoop commit protocol. |
23 | 25 |
|
24 | | -## Threading Model |
| 26 | +Native arrays cross into Spark through the Arrow C Data Interface. Each partition reader owns its |
| 27 | +native scan, Arrow allocator, and exported batches and closes them at task completion. |
25 | 28 |
|
26 | | -The Spark integration crosses the JNI boundary between Java and Rust. Each Spark partition |
27 | | -reader opens a Vortex file and creates a native scan via JNI. The native side manages its own |
28 | | -async runtime and drives I/O internally, returning results to Java as Arrow-compatible columnar |
29 | | -batches. |
| 29 | +## Pushdown and statistics |
30 | 30 |
|
31 | | -Because each partition reader owns its native resources exclusively, there is no contention |
32 | | -across Spark threads. The JNI boundary is crossed once per batch rather than once per row, |
33 | | -keeping overhead low. A prefetching iterator on the Java side buffers upcoming batches to |
34 | | -overlap I/O with Spark's processing. |
| 31 | +Spark's required schema becomes the Vortex scan projection. Convertible V1 filters become Vortex |
| 32 | +expressions; filters the converter rejects remain Spark residuals. |
35 | 33 |
|
36 | | -## Filter and Projection Pushdown |
| 34 | +The scan accepts `COUNT(*)` aggregation when there are no data filters and grouping uses only |
| 35 | +partition columns. Readers return one footer count per file and Spark performs the final merge. |
| 36 | +Footer row counts from files left after partition pruning are also reported through Spark scan |
| 37 | +statistics. MIN/MAX and `COUNT(column)` need read-side column statistics in the JNI API and are |
| 38 | +not pushed down. |
37 | 39 |
|
38 | | -Projection pushdown is supported through Spark's `SupportsPushDownRequiredColumns` interface. |
39 | | -The scan builder prunes the column list to only those referenced by the query, and the pruned |
40 | | -column set is passed to the native scan via `ScanOptions`. |
| 40 | +## Source layout |
41 | 41 |
|
42 | | -Filter pushdown infrastructure exists in the `ScanOptions` type but is not yet connected to |
43 | | -Spark's `SupportsPushDownFilters` interface. This is planned future work. |
44 | | - |
45 | | -## Data Export |
46 | | - |
47 | | -Native Vortex arrays are exported to Arrow via the C Data Interface, then wrapped in Spark's |
48 | | -columnar batch format using custom `ArrowColumnVector` wrappers. This avoids a copy between |
49 | | -the Rust and JVM heaps -- the Arrow buffers remain in native memory and are accessed from Java |
50 | | -through direct byte buffers. |
51 | | - |
52 | | -## Future Work |
53 | | - |
54 | | -The current integration builds directly on the native file and scan APIs via JNI. Future work |
55 | | -will migrate it to use the [Scan API](/concepts/scanning) `Source` trait, which will provide a |
56 | | -standard interface for file discovery, partitioning, and pushdown. This will also enable |
57 | | -connecting Spark's filter pushdown to Vortex's expression-based filtering. |
| 42 | +Shared Java and Scala live in `java/vortex-spark/common`. Thin projects in `v3.5` and `v4.0` |
| 43 | +select the corresponding Spark and Scala dependencies. Java sources are compiled jointly by |
| 44 | +scalac because the small Scala shims isolate binary differences in Spark's Scala APIs. |
0 commit comments