|
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 | +Spark's file listing keeps every file that is not hidden. The scan, the statistics, and both |
| 16 | +reader factories skip files without the `.vortex` extension, so other files in a dataset directory |
| 17 | +are not read. |
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 | +Vortex files are not internally split. A Spark file partition may contain several files, and the |
| 20 | +reader factory opens each `PartitionedFile` in turn. Hive partition values are appended with |
| 21 | +constant column vectors. |
19 | 22 |
|
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. |
| 23 | +## I/O and JNI |
23 | 24 |
|
24 | | -## Threading Model |
| 25 | +Spark lists paths through Hadoop. Content reads go through pooled Hadoop input streams exposed to |
| 26 | +native Vortex through the JNI `NativeReadable` interface. Writes expose the committer's Hadoop |
| 27 | +task path through `NativeWritable`. Vortex's own object-store clients are not used, so the |
| 28 | +connector sees the same schemes and credentials as Spark's file index and commit protocol. |
25 | 29 |
|
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. |
| 30 | +Native arrays cross into Spark through the Arrow C Data Interface. Each partition reader owns its |
| 31 | +native scan, Arrow allocator, and exported batches and closes them at task completion. |
30 | 32 |
|
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. |
| 33 | +## Pushdown and statistics |
35 | 34 |
|
36 | | -## Filter and Projection Pushdown |
| 35 | +Schema inference merges every footer, so the dataset schema is the union of the top-level fields its |
| 36 | +files carry. The partition reader projects only the fields the file it opened actually holds and |
| 37 | +fills the rest with constant null vectors, and it converts filters against those same fields, so a |
| 38 | +filter on a column the file lacks stays a Spark residual rather than reaching the native scan. |
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 | +Spark's required schema becomes the Vortex scan projection. Convertible V1 filters become Vortex |
| 41 | +expressions; filters the converter rejects remain Spark residuals. |
41 | 42 |
|
42 | | -Filter pushdown is supported through `SupportsPushDownV2Filters`. The scan builder converts each |
43 | | -predicate it recognizes into a Vortex expression and keeps the rest for Spark to evaluate after the |
44 | | -scan; the converted expression reaches native code as the `ScanOptions` filter. |
| 43 | +The scan accepts `COUNT(*)` aggregation when there are no data filters and grouping uses only |
| 44 | +partition columns, and only from footers that state their row count exactly. Readers return one |
| 45 | +footer count per file and Spark performs the final merge; the scan then reports one row per file as |
| 46 | +its statistics rather than reading those footers again on the driver. Footer row counts from files |
| 47 | +left after partition pruning are reported through Spark scan statistics when no aggregate is |
| 48 | +pushed. MIN/MAX and `COUNT(column)` need read-side column statistics in the JNI API and are |
| 49 | +not pushed down. |
45 | 50 |
|
46 | | -## Data Export |
| 51 | +## Source layout |
47 | 52 |
|
48 | | -Native Vortex arrays are exported to Arrow via the C Data Interface, then wrapped in Spark's |
49 | | -columnar batch format using custom `ArrowColumnVector` wrappers. This avoids a copy between |
50 | | -the Rust and JVM heaps -- the Arrow buffers remain in native memory and are accessed from Java |
51 | | -through direct byte buffers. |
52 | | - |
53 | | -## Future Work |
54 | | - |
55 | | -The current integration builds directly on the native file and scan APIs via JNI. Future work |
56 | | -will migrate it to use the [Scan API](/concepts/scanning) `Source` trait, which will provide a |
57 | | -standard interface for file discovery, partitioning, and pushdown. |
| 53 | +Shared Java and Scala live in `java/vortex-spark/common`. Thin projects in `v3.5` and `v4.0` |
| 54 | +select the corresponding Spark and Scala dependencies. The Scala shims isolate the binary |
| 55 | +differences in Spark's Scala APIs; nothing in Java depends on them, so javac compiles the Java |
| 56 | +sources and ErrorProne and Nopen keep covering them. |
0 commit comments