From 38fcf6b7f1df851b760c5511d70443cb5a9ea9f0 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 3 Jul 2026 10:59:44 +0000 Subject: [PATCH] Add docs on how to use Spark vortex datasource Fix minor usage annoyances w.r.t. managed tables and direct file queries --- README.md | 2 +- docs/user-guide/spark.md | 123 +++++++++++- java/README.md | 4 +- java/vortex-spark/README.md | 115 ++++++++++++ .../java/dev/vortex/spark/VortexCatalog.java | 119 ++++++++++++ .../dev/vortex/spark/VortexDataSourceV2.java | 21 ++- .../vortex/spark/VortexSessionCatalog.java | 79 ++++++++ .../spark/VortexSessionCatalogTest.java | 113 +++++++++++ .../java/dev/vortex/spark/VortexSqlTest.java | 176 ++++++++++++++++++ 9 files changed, 737 insertions(+), 15 deletions(-) create mode 100644 java/vortex-spark/README.md create mode 100644 java/vortex-spark/src/main/java/dev/vortex/spark/VortexCatalog.java create mode 100644 java/vortex-spark/src/main/java/dev/vortex/spark/VortexSessionCatalog.java create mode 100644 java/vortex-spark/src/test/java/dev/vortex/spark/VortexSessionCatalogTest.java create mode 100644 java/vortex-spark/src/test/java/dev/vortex/spark/VortexSqlTest.java diff --git a/README.md b/README.md index 8279e9651ca..893a6a1cb47 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ [![CodSpeed Badge](https://img.shields.io/endpoint?url=https://codspeed.io/badge.json)](https://codspeed.io/vortex-data/vortex) [![Crates.io](https://img.shields.io/crates/v/vortex.svg)](https://crates.io/crates/vortex) [![PyPI - Version](https://img.shields.io/pypi/v/vortex-data)](https://pypi.org/project/vortex-data/) -[![Maven - Version](https://img.shields.io/maven-central/v/dev.vortex/vortex-spark)](https://central.sonatype.com/artifact/dev.vortex/vortex-spark) +[![Maven - Version](https://img.shields.io/maven-central/v/dev.vortex/vortex-spark_2.13)](https://central.sonatype.com/artifact/dev.vortex/vortex-spark_2.13) [![codecov](https://codecov.io/github/vortex-data/vortex/graph/badge.svg)](https://codecov.io/github/vortex-data/vortex) [![Cite](https://img.shields.io/badge/cite-CITATION.cff-blue)](CITATION.cff) diff --git a/docs/user-guide/spark.md b/docs/user-guide/spark.md index 676c11f332b..5b1f9b44b8d 100644 --- a/docs/user-guide/spark.md +++ b/docs/user-guide/spark.md @@ -1,15 +1,52 @@ # Spark Vortex provides a Spark DataSource V2 connector for reading and writing Vortex files. The -connector is published to Maven Central as `dev.vortex:vortex-spark`. +connector is published to Maven Central in two flavors: -## Installation +- `dev.vortex:vortex-spark_2.13` for Spark 4.x (Scala 2.13) +- `dev.vortex:vortex-spark_2.12` for Spark 3.5.x (Scala 2.12) -Add the dependency to your build. The connector is built against Spark 4.x with Scala 2.13. +Use the `all` classifier JAR (e.g. `vortex-spark_2.13-0.78.0-all.jar`). It is self-contained: +it bundles the Vortex JNI bindings, native libraries for Linux (x86_64 and aarch64) and macOS +(aarch64), and relocates its Arrow, Guava, and Jackson dependencies to avoid classpath +conflicts with Spark. The thin (unclassified) JAR does not work on its own because it +references relocated classes that only ship in the `all` JAR. + +## Getting Vortex into Spark + +For `spark-shell`, `spark-submit`, or `pyspark`, pass the `all` JAR with `--jars`. Spark +accepts either a local path or a URL, so you can point directly at Maven Central: + +```shell +spark-shell --jars https://repo1.maven.org/maven2/dev/vortex/vortex-spark_2.13/0.78.0/vortex-spark_2.13-0.78.0-all.jar +``` + +Or equivalently when building a session programmatically, e.g. in PySpark: + +```python +spark = ( + SparkSession.builder + .config("spark.jars", "/path/to/vortex-spark_2.13-0.78.0-all.jar") + .getOrCreate() +) +``` + +```{note} +`--packages dev.vortex:vortex-spark_2.13:0.78.0` does not work: `--packages` cannot select +the `all` classifier and resolves the thin JAR, which fails at runtime with +`NoClassDefFoundError: dev/vortex/relocated/...`. +``` + +Once the JAR is on the classpath, the connector registers itself automatically under the +format name `vortex` — no session configuration is required. + +## Installation as a Build Dependency + +To depend on the connector from a JVM project, add the `all` classifier to the dependency: ````{tab} Gradle (Kotlin) ```kotlin -implementation("dev.vortex:vortex-spark:") +implementation("dev.vortex:vortex-spark_2.13:0.78.0:all") ``` ```` @@ -17,18 +54,18 @@ implementation("dev.vortex:vortex-spark:") ```xml dev.vortex - vortex-spark - ${vortex.version} + vortex-spark_2.13 + 0.78.0 + all ``` ```` -The connector ships as a shadow JAR that relocates its Arrow, Guava, and Protobuf dependencies -to avoid classpath conflicts with Spark. - ## Reading Vortex Files -Use the `vortex` format to read a single file or a directory of Vortex files: +Paths may be local filesystem paths (`/path/to/data`) or URLs (`file:///path/to/data`, +`s3://bucket/path/to/data`). Use the `vortex` format to read a single file or a directory of +Vortex files: ```java Dataset df = spark.read() @@ -65,6 +102,72 @@ Each Spark partition produces one output file named `part-{partitionId}-{taskId} The connector supports all standard Spark save modes: `Overwrite`, `Append`, `Ignore`, and `ErrorIfExists`. +## Spark SQL + +The connector can also be used from pure SQL. To query existing Vortex files, register them +as a temporary view: + +```sql +CREATE TEMPORARY VIEW people +USING vortex +OPTIONS (path '/path/to/data'); + +SELECT name, age FROM people WHERE age > 30; +``` + +Tables can be created with `USING vortex`, then written to and read back with plain SQL. +With a `LOCATION` clause the table is external, backed by the files at that path; without +one the table is managed, and Spark stores its data under the warehouse directory (and +deletes it on `DROP TABLE`): + +```sql +CREATE TABLE student (id INT, name STRING, age INT) +USING vortex; + +INSERT INTO student VALUES (1, 'Alice', 20), (2, 'Bob', 21); + +SELECT * FROM student; +``` + +`CREATE TABLE ... AS SELECT` works the same way: + +```sql +CREATE TABLE adults +USING vortex +AS SELECT * FROM people WHERE age >= 18; +``` + +```{note} +On Spark 3.5, `CREATE TABLE ... USING vortex` additionally requires replacing the session +catalog, because Spark 3.5's built-in catalog cannot read tables backed by a DataSource +V2-only connector: + + spark.sql.catalog.spark_catalog=dev.vortex.spark.VortexSessionCatalog + +The extension delegates everything to the built-in session catalog (including the Hive +metastore, if configured) and only changes how `vortex` tables are resolved; tables of other +providers are untouched. It is not needed on Spark 4, though setting it is harmless. +``` + +## Direct File Queries + +Spark's built-in ``SELECT * FROM format.`path` `` syntax only works for built-in file +formats, so the connector ships a path-based catalog that provides the equivalent for +Vortex. Register it in the session configuration under the name `vortex`: + +```shell +spark-sql --conf spark.sql.catalog.vortex=dev.vortex.spark.VortexCatalog +``` + +Then query a Vortex file, or a directory of Vortex files, directly by path — no view or +table required: + +```sql +SELECT * FROM vortex.`/path/to/data`; + +INSERT INTO vortex.`/path/to/data` VALUES (1, 'Alice', 20); +``` + ## Supported Types | Spark Type | Vortex Type | diff --git a/java/README.md b/java/README.md index d9554420998..f45f2474f11 100644 --- a/java/README.md +++ b/java/README.md @@ -3,7 +3,9 @@ We provide two interfaces for working with Vortex from Java: - `vortex-java` - a low-level interface JNI for working with Vortex files and arrays on cloud and local storage -- `vortex-spark` - A Spark connector for working with datasets of Vortex files +- `vortex-spark` - A Spark connector for working with datasets of Vortex files. See + [vortex-spark/README.md](vortex-spark/README.md) for how to load the connector into Spark + and query Vortex files from the DataFrame API or Spark SQL. ## Publishing diff --git a/java/vortex-spark/README.md b/java/vortex-spark/README.md new file mode 100644 index 00000000000..c6e8c9345bd --- /dev/null +++ b/java/vortex-spark/README.md @@ -0,0 +1,115 @@ +# vortex-spark + +A Spark DataSource V2 connector for reading and writing [Vortex](https://vortex.dev) files. +It registers itself under the format name `vortex` and supports both the DataFrame API and +Spark SQL. + +Two flavors are published to Maven Central: + +| Artifact | Spark | Scala | +|-------------------------------|-----------|-------| +| `dev.vortex:vortex-spark_2.13` | Spark 4.x | 2.13 | +| `dev.vortex:vortex-spark_2.12` | Spark 3.5.x | 2.12 | + +Use the `all` classifier JAR (e.g. `vortex-spark_2.13-0.78.0-all.jar`). It is self-contained: +it bundles the Vortex JNI bindings, native libraries for Linux (x86_64 and aarch64) and macOS +(aarch64), and relocates its Arrow, Guava, and Jackson dependencies to avoid classpath +conflicts with Spark. The thin (unclassified) JAR does not work on its own because it +references relocated classes that only ship in the `all` JAR. + +## Getting Vortex into Spark + +Pass the `all` JAR to `spark-shell`, `spark-submit`, or `pyspark` with `--jars`. Spark accepts +either a local path or a URL, so you can point directly at Maven Central: + +```shell +spark-shell --jars https://repo1.maven.org/maven2/dev/vortex/vortex-spark_2.13/0.78.0/vortex-spark_2.13-0.78.0-all.jar +``` + +Or configure it on the session builder, e.g. in PySpark: + +```python +spark = ( + SparkSession.builder + .config("spark.jars", "/path/to/vortex-spark_2.13-0.78.0-all.jar") + .getOrCreate() +) +``` + +Note that `--packages dev.vortex:vortex-spark_2.13:0.78.0` does not work: `--packages` cannot +select the `all` classifier and resolves the thin JAR, which fails at runtime with +`NoClassDefFoundError: dev/vortex/relocated/...`. + +To depend on the connector from a JVM project instead, add the `all` classifier to the +dependency: + +```kotlin +implementation("dev.vortex:vortex-spark_2.13:0.78.0:all") +``` + +## Usage + +Paths may be local filesystem paths (`/path/to/data`) or URLs (`file:///path/to/data`, +`s3://bucket/path/to/data`). + +### DataFrame API + +```java +// Write +df.write() + .format("vortex") + .option("path", "/path/to/output") + .mode(SaveMode.Overwrite) + .save(); + +// Read a single file or a directory of .vortex files +Dataset df = spark.read() + .format("vortex") + .option("path", "/path/to/output") + .load(); +``` + +### Spark SQL + +```sql +-- Query existing Vortex files through a temporary view +CREATE TEMPORARY VIEW people +USING vortex +OPTIONS (path '/path/to/data'); + +SELECT name, age FROM people WHERE age > 30; + +-- Create a table and write to it. With a LOCATION clause the table is external, +-- backed by the files at that path; without one it is managed by Spark. +CREATE TABLE student (id INT, name STRING, age INT) +USING vortex; + +INSERT INTO student VALUES (1, 'Alice', 20), (2, 'Bob', 21); + +SELECT * FROM student; +``` + +On Spark 3.5, `CREATE TABLE ... USING vortex` additionally requires replacing the session +catalog with `spark.sql.catalog.spark_catalog=dev.vortex.spark.VortexSessionCatalog`, +because Spark 3.5's built-in catalog cannot read tables backed by a DataSource V2-only +connector. The extension delegates everything else to the built-in session catalog and +leaves tables of other providers untouched; it is not needed on Spark 4. + +### Direct file queries + +Spark's built-in ``SELECT * FROM format.`path` `` syntax only works for built-in file +formats, so the connector ships a path-based catalog that provides the equivalent for +Vortex. Register it under the name `vortex` with this session config: + +``` +spark.sql.catalog.vortex=dev.vortex.spark.VortexCatalog +``` + +Then query (or insert into) Vortex files directly by path: + +```sql +SELECT * FROM vortex.`/path/to/data`; +``` + +See the [Spark user guide](https://docs.vortex.dev/user-guide/spark.html) for the full +documentation, including supported types, write options, and S3 configuration. diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/VortexCatalog.java b/java/vortex-spark/src/main/java/dev/vortex/spark/VortexCatalog.java new file mode 100644 index 00000000000..3d496a4cee0 --- /dev/null +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/VortexCatalog.java @@ -0,0 +1,119 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +package dev.vortex.spark; + +import java.util.Map; +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException; +import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.connector.catalog.Table; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.connector.catalog.TableChange; +import org.apache.spark.sql.connector.expressions.Transform; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; + +/** + * A path-based Spark catalog for querying Vortex files directly from SQL. + * + *

Spark only supports {@code SELECT * FROM format.`path`} syntax for built-in file formats, so this catalog provides + * the equivalent for Vortex. Register it under the name {@code vortex}: + * + * 

spark.sql.catalog.vortex=dev.vortex.spark.VortexCatalog
+ * + *

then query a Vortex file, or a directory of Vortex files, directly by path: + * + * 

SELECT * FROM vortex.`/path/to/data`;
+ * + *

The table identifier must look like a path — contain a {@code /} — and resolves to the same table a + * {@code spark.read.format("vortex")} load of that path would produce, so reads, writes ({@code INSERT INTO + * vortex.`/path/to/data`}), and pushdown all behave identically. The catalog holds no state and supports no DDL. + */ +public final class VortexCatalog implements TableCatalog { + private static final String PATH_KEY = "path"; + + private String name = "vortex"; + + /** + * Creates a new catalog instance. + * + *

This no-argument constructor is required for Spark to instantiate the catalog through reflection from the + * {@code spark.sql.catalog.} configuration. + */ + public VortexCatalog() {} + + @Override + public void initialize(String name, CaseInsensitiveStringMap options) { + this.name = name; + } + + @Override + public String name() { + return name; + } + + /** + * Returns no identifiers: this catalog holds no state, tables are addressed by path. + * + * @param namespace the namespace to list, ignored + * @return an empty array + */ + @Override + public Identifier[] listTables(String[] namespace) { + return new Identifier[0]; + } + + /** + * Loads the Vortex file or directory of Vortex files at the path given by the identifier name. + * + * @param ident identifier whose name is a filesystem path or URL, e.g. {@code vortex.`/path/to/data`} + * @return a table backed by the Vortex files at the path + * @throws NoSuchTableException if the identifier does not look like a path, or the path cannot be read + */ + @SuppressWarnings("deprecation") + @Override + public Table loadTable(Identifier ident) throws NoSuchTableException { + String path = ident.name(); + if (ident.namespace().length != 0 || !path.contains("/")) { + throw new NoSuchTableException(ident); + } + var options = new CaseInsensitiveStringMap(Map.of(PATH_KEY, path)); + var provider = new VortexDataSourceV2(); + StructType schema; + Transform[] partitioning; + try { + schema = provider.inferSchema(options); + partitioning = provider.inferPartitioning(options); + } catch (RuntimeException e) { + // Missing or unreadable paths surface as "table not found" to SQL users. + throw new NoSuchTableException(ident); + } + return provider.getTable(schema, partitioning, Map.of(PATH_KEY, path)); + } + + /** Unsupported: tables are addressed by path, create them by writing data with the {@code vortex} format. */ + @Override + public Table createTable( + Identifier ident, StructType schema, Transform[] partitions, Map properties) { + throw new UnsupportedOperationException( + "VortexCatalog does not support CREATE TABLE, write data to the path instead"); + } + + /** Unsupported: this catalog holds no table metadata to alter. */ + @Override + public Table alterTable(Identifier ident, TableChange... changes) { + throw new UnsupportedOperationException("VortexCatalog does not support ALTER TABLE"); + } + + /** Unsupported: this catalog never drops data, returns false. */ + @Override + public boolean dropTable(Identifier ident) { + return false; + } + + /** Unsupported: this catalog holds no table metadata to rename. */ + @Override + public void renameTable(Identifier oldIdent, Identifier newIdent) { + throw new UnsupportedOperationException("VortexCatalog does not support RENAME TABLE"); + } +} diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/VortexDataSourceV2.java b/java/vortex-spark/src/main/java/dev/vortex/spark/VortexDataSourceV2.java index b3d7d637504..a5ce4e3b1ce 100644 --- a/java/vortex-spark/src/main/java/dev/vortex/spark/VortexDataSourceV2.java +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/VortexDataSourceV2.java @@ -126,10 +126,14 @@ public StructType inferSchema(CaseInsensitiveStringMap options) { * explicit partitioning. Returning identity transforms here lets downstream components (notably * {@link dev.vortex.spark.read.VortexScanBuilder}) tell which schema columns are encoded in the directory layout * rather than stored inside the Vortex files, which matters for predicate pushdown. + * + *

The options may contain no path at all: when a managed table is created ({@code CREATE TABLE ... USING vortex} + * without a {@code LOCATION} clause), Spark's session catalog calls this before it has assigned the table's + * warehouse location. No transforms are inferred in that case. */ @Override public Transform[] inferPartitioning(CaseInsensitiveStringMap options) { - var paths = getPaths(options); + var paths = getPathsOrEmpty(options); if (paths.isEmpty()) { return new Transform[0]; } @@ -157,16 +161,20 @@ public Transform[] inferPartitioning(CaseInsensitiveStringMap options) { *

This method creates a VortexWritableTable that can be used to both read from and write to Vortex files. The * partitioning parameter is currently ignored. * + *

The properties may contain no path at all: when a managed table is created ({@code CREATE TABLE ... USING + * vortex} without a {@code LOCATION} clause), Spark's session catalog validates the table before it has assigned + * the table's warehouse location. The returned table then has no paths; once the table is loaded for reads or + * writes, Spark always supplies the resolved table location as the {@code path} property. + * * @param schema the table schema * @param partitioning table partitioning transforms * @param properties the table properties containing file paths and other options * @return a VortexTable instance for reading and writing data - * @throws RuntimeException if required path properties are missing */ @Override public Table getTable(StructType schema, Transform[] partitioning, Map properties) { var uncased = new CaseInsensitiveStringMap(properties); - ImmutableList paths = getPaths(uncased); + ImmutableList paths = getPathsOrEmpty(uncased); return new VortexTable(paths, schema, buildDataSourceOptions(properties), partitioning); } @@ -210,6 +218,13 @@ private Map buildDataSourceOptions(Map propertie return options.build(); } + private static ImmutableList getPathsOrEmpty(CaseInsensitiveStringMap uncased) { + if (!uncased.containsKey(PATH_KEY) && !uncased.containsKey(PATHS_KEY)) { + return ImmutableList.of(); + } + return getPaths(uncased); + } + private static ImmutableList getPaths(CaseInsensitiveStringMap uncased) { if (uncased.containsKey(PATH_KEY)) { return ImmutableList.of(uncased.get(PATH_KEY)); diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/VortexSessionCatalog.java b/java/vortex-spark/src/main/java/dev/vortex/spark/VortexSessionCatalog.java new file mode 100644 index 00000000000..1d546880cc1 --- /dev/null +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/VortexSessionCatalog.java @@ -0,0 +1,79 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +package dev.vortex.spark; + +import java.util.Map; +import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException; +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException; +import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException; +import org.apache.spark.sql.connector.catalog.DelegatingCatalogExtension; +import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.connector.catalog.Table; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.connector.expressions.Transform; +import org.apache.spark.sql.types.StructType; + +/** + * A session catalog extension that resolves {@code USING vortex} tables through the Vortex DataSource V2 connector. + * + *

Spark 3.5's built-in session catalog resolves the tables it stores through the V1 {@code DataSource} path, which + * rejects DataSource-V2-only connectors like Vortex, so {@code CREATE TABLE ... USING vortex} tables cannot be read + * back. (Spark 4 resolves them through the V2 provider directly and needs none of this.) Registering this extension as + * the session catalog fixes that on Spark 3.5: + * + * 

spark.sql.catalog.spark_catalog=dev.vortex.spark.VortexSessionCatalog
+ * + *

All operations are delegated to the built-in session catalog — table metadata lives wherever it normally would, + * including the Hive metastore — but any table whose provider is {@code vortex} is loaded as a Vortex DataSource V2 + * table, backed by the files at the table's location. Tables of every other provider are untouched. + */ +public final class VortexSessionCatalog extends DelegatingCatalogExtension { + + /** + * Creates a new session catalog extension. + * + *

This no-argument constructor is required for Spark to instantiate the catalog through reflection from the + * {@code spark.sql.catalog.spark_catalog} configuration. + */ + public VortexSessionCatalog() {} + + @Override + public Table loadTable(Identifier ident) throws NoSuchTableException { + return asVortexTableIfVortex(super.loadTable(ident)); + } + + /** + * Creates the table in the delegate session catalog, then returns it resolved through the Vortex connector when its + * provider is {@code vortex}. + * + *

The conversion matters for {@code CREATE TABLE ... AS SELECT}: Spark writes the query result into the table + * returned here, which must therefore support V2 writes. + */ + @SuppressWarnings("deprecation") + @Override + public Table createTable( + Identifier ident, StructType schema, Transform[] partitions, Map properties) + throws TableAlreadyExistsException, NoSuchNamespaceException { + return asVortexTableIfVortex(super.createTable(ident, schema, partitions, properties)); + } + + /** + * Rebuilds a session-catalog table as a Vortex DataSource V2 table when its provider is {@code vortex} and it has a + * location; returns every other table unchanged. The schema and partitioning stored in the catalog are used as-is, + * no file needs to be opened. + */ + @SuppressWarnings("deprecation") + private static Table asVortexTableIfVortex(Table table) { + Map properties = table.properties(); + VortexDataSourceV2 provider = new VortexDataSourceV2(); + if (!provider.shortName().equalsIgnoreCase(properties.get(TableCatalog.PROP_PROVIDER))) { + return table; + } + String location = properties.get(TableCatalog.PROP_LOCATION); + if (location == null) { + return table; + } + return provider.getTable(table.schema(), table.partitioning(), Map.of("path", location)); + } +} diff --git a/java/vortex-spark/src/test/java/dev/vortex/spark/VortexSessionCatalogTest.java b/java/vortex-spark/src/test/java/dev/vortex/spark/VortexSessionCatalogTest.java new file mode 100644 index 00000000000..b31d674d884 --- /dev/null +++ b/java/vortex-spark/src/test/java/dev/vortex/spark/VortexSessionCatalogTest.java @@ -0,0 +1,113 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +package dev.vortex.spark; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.api.io.TempDir; + +/** + * Integration tests for {@link VortexSessionCatalog}, the session catalog extension that makes {@code CREATE TABLE ... + * USING vortex} work on Spark 3.5 as well as Spark 4. + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public final class VortexSessionCatalogTest { + + private SparkSession spark; + private Path warehouseDir; + + @TempDir + Path tempDir; + + @BeforeAll + public void setUp() throws IOException { + warehouseDir = Files.createTempDirectory("vortex-warehouse"); + spark = SparkSession.builder() + .appName("VortexSessionCatalogTest") + .master("local[2]") + .config("spark.driver.host", "127.0.0.1") + .config("spark.sql.warehouse.dir", warehouseDir.toUri().toString()) + .config("spark.sql.catalog.spark_catalog", VortexSessionCatalog.class.getName()) + .config("spark.ui.enabled", "false") + .getOrCreate(); + } + + @AfterAll + public void tearDown() { + if (spark != null) { + spark.stop(); + } + } + + @Test + @DisplayName("Managed table lifecycle: CREATE, SELECT while empty, INSERT, INSERT OVERWRITE, DROP") + public void testManagedTableLifecycle() { + spark.sql("CREATE TABLE managed_students (id INT, name STRING, age INT) USING vortex"); + + assertEquals(0, spark.sql("SELECT * FROM managed_students").count(), "New managed table should be empty"); + + spark.sql("INSERT INTO managed_students VALUES (1, 'Alice', 20), (2, 'Bob', 21)"); + List rows = spark.sql("SELECT name FROM managed_students WHERE age > 20 ORDER BY name") + .collectAsList(); + assertEquals(1, rows.size()); + assertEquals("Bob", rows.get(0).getString(0)); + + spark.sql("INSERT OVERWRITE managed_students VALUES (3, 'Carol', 22)"); + assertEquals(1, spark.sql("SELECT * FROM managed_students").count(), "Overwrite should replace all rows"); + + Path tableDir = warehouseDir.resolve("managed_students"); + assertTrue(Files.exists(tableDir), "Managed table data should live under the warehouse dir"); + spark.sql("DROP TABLE managed_students"); + assertFalse(Files.exists(tableDir), "Dropping a managed table should remove its data"); + } + + @Test + @DisplayName("CREATE TABLE AS SELECT without a LOCATION clause") + public void testCreateManagedTableAsSelect() { + spark.sql("CREATE TABLE ctas_source (id INT, name STRING) USING vortex"); + spark.sql("INSERT INTO ctas_source VALUES (1, 'Alice'), (2, 'Bob')"); + + spark.sql("CREATE TABLE ctas_target USING vortex AS SELECT * FROM ctas_source WHERE id > 1"); + List rows = spark.sql("SELECT name FROM ctas_target").collectAsList(); + assertEquals(1, rows.size()); + assertEquals("Bob", rows.get(0).getString(0)); + + spark.sql("DROP TABLE ctas_target"); + spark.sql("DROP TABLE ctas_source"); + } + + @Test + @DisplayName("External table with a LOCATION clause") + public void testExternalTable() { + Path location = tempDir.resolve("ext_students"); + spark.sql( + String.format("CREATE TABLE ext_students (id INT, name STRING) USING vortex LOCATION '%s'", location)); + spark.sql("INSERT INTO ext_students VALUES (1, 'Alice')"); + assertEquals(1, spark.sql("SELECT * FROM ext_students").count()); + spark.sql("DROP TABLE ext_students"); + } + + @Test + @DisplayName("Tables of other providers pass through the extension untouched") + public void testOtherProviderPassthrough() { + spark.sql("CREATE TABLE pq_table (id INT) USING parquet"); + spark.sql("INSERT INTO pq_table VALUES (7)"); + assertEquals( + 7, spark.sql("SELECT * FROM pq_table").collectAsList().get(0).getInt(0)); + spark.sql("DROP TABLE pq_table"); + } +} diff --git a/java/vortex-spark/src/test/java/dev/vortex/spark/VortexSqlTest.java b/java/vortex-spark/src/test/java/dev/vortex/spark/VortexSqlTest.java new file mode 100644 index 00000000000..1a9b21ef779 --- /dev/null +++ b/java/vortex-spark/src/test/java/dev/vortex/spark/VortexSqlTest.java @@ -0,0 +1,176 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +package dev.vortex.spark; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Arrays; +import java.util.List; +import org.apache.spark.sql.AnalysisException; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.RowFactory; +import org.apache.spark.sql.SaveMode; +import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.api.io.TempDir; + +/** + * Integration tests for Spark SQL access to Vortex: managed tables created without a {@code LOCATION} clause, and + * direct file queries through {@link VortexCatalog}. + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public final class VortexSqlTest { + + private SparkSession spark; + private Path warehouseDir; + + @TempDir + Path tempDir; + + @BeforeAll + public void setUp() throws IOException { + warehouseDir = Files.createTempDirectory("vortex-warehouse"); + spark = SparkSession.builder() + .appName("VortexSqlTest") + .master("local[2]") + .config("spark.driver.host", "127.0.0.1") + .config("spark.sql.warehouse.dir", warehouseDir.toUri().toString()) + .config("spark.sql.catalog.vortex", VortexCatalog.class.getName()) + .config("spark.ui.enabled", "false") + .getOrCreate(); + } + + @AfterAll + public void tearDown() { + if (spark != null) { + spark.stop(); + } + } + + /** + * Spark 3.5's built-in session catalog cannot read tables backed by a DataSource-V2-only provider: its + * {@code FindDataSourceTable} rule falls back to the V1 {@code DataSource} path, which rejects the provider with + * "vortex is not a valid Spark SQL Data Source". Spark 4 resolves such tables through the provider directly. On + * Spark 3.5 the {@link VortexSessionCatalog} extension provides the same support — see + * {@link VortexSessionCatalogTest}, which runs these scenarios on both versions. + */ + private void assumeSupportsSqlTables() { + assumeTrue( + spark.version().startsWith("4."), + "CREATE TABLE ... USING vortex requires Spark 4 or the VortexSessionCatalog extension"); + } + + @Test + @DisplayName("Managed table lifecycle: CREATE, SELECT while empty, INSERT, INSERT OVERWRITE, DROP") + public void testManagedTableLifecycle() { + assumeSupportsSqlTables(); + spark.sql("CREATE TABLE managed_students (id INT, name STRING, age INT) USING vortex"); + + assertEquals(0, spark.sql("SELECT * FROM managed_students").count(), "New managed table should be empty"); + + spark.sql("INSERT INTO managed_students VALUES (1, 'Alice', 20), (2, 'Bob', 21)"); + List rows = spark.sql("SELECT name FROM managed_students WHERE age > 20 ORDER BY name") + .collectAsList(); + assertEquals(1, rows.size()); + assertEquals("Bob", rows.get(0).getString(0)); + + spark.sql("INSERT OVERWRITE managed_students VALUES (3, 'Carol', 22)"); + assertEquals(1, spark.sql("SELECT * FROM managed_students").count(), "Overwrite should replace all rows"); + + Path tableDir = warehouseDir.resolve("managed_students"); + assertTrue(Files.exists(tableDir), "Managed table data should live under the warehouse dir"); + spark.sql("DROP TABLE managed_students"); + assertFalse(Files.exists(tableDir), "Dropping a managed table should remove its data"); + } + + @Test + @DisplayName("CREATE TABLE AS SELECT without a LOCATION clause") + public void testCreateManagedTableAsSelect() { + assumeSupportsSqlTables(); + spark.sql("CREATE TABLE ctas_source (id INT, name STRING) USING vortex"); + spark.sql("INSERT INTO ctas_source VALUES (1, 'Alice'), (2, 'Bob')"); + + spark.sql("CREATE TABLE ctas_target USING vortex AS SELECT * FROM ctas_source WHERE id > 1"); + List rows = spark.sql("SELECT name FROM ctas_target").collectAsList(); + assertEquals(1, rows.size()); + assertEquals("Bob", rows.get(0).getString(0)); + + spark.sql("DROP TABLE ctas_target"); + spark.sql("DROP TABLE ctas_source"); + } + + @Test + @DisplayName("Reading the vortex format without a path option still fails") + public void testReadWithoutPathStillThrows() { + assertThrows( + IllegalArgumentException.class, + () -> spark.read().format("vortex").load(), + "A read with no path should not silently return an empty DataFrame"); + } + + @Test + @DisplayName("Direct file query through the vortex catalog") + public void testDirectPathQuery() { + Path dataDir = tempDir.resolve("direct_query"); + writeTestData(dataDir); + + List rows = spark.sql(String.format("SELECT name FROM vortex.`%s` WHERE age > 30 ORDER BY name", dataDir)) + .collectAsList(); + assertEquals(2, rows.size()); + assertEquals("Alice", rows.get(0).getString(0)); + assertEquals("Carol", rows.get(1).getString(0)); + } + + @Test + @DisplayName("INSERT INTO a direct path through the vortex catalog") + public void testDirectPathInsert() { + Path dataDir = tempDir.resolve("direct_insert"); + writeTestData(dataDir); + + spark.sql(String.format("INSERT INTO vortex.`%s` VALUES (4, 'Dave', 50)", dataDir)); + assertEquals( + 4, + spark.sql(String.format("SELECT * FROM vortex.`%s`", dataDir)).count()); + } + + @Test + @DisplayName("Direct queries of missing paths and non-path names fail as table-not-found") + public void testDirectPathNotFound() { + assertThrows(AnalysisException.class, () -> spark.sql( + String.format("SELECT * FROM vortex.`%s`", tempDir.resolve("no_such_dir"))) + .collect()); + assertThrows(AnalysisException.class, () -> spark.sql("SELECT * FROM vortex.not_a_path") + .collect()); + } + + private void writeTestData(Path dataDir) { + StructType schema = DataTypes.createStructType(new StructField[] { + DataTypes.createStructField("id", DataTypes.IntegerType, false), + DataTypes.createStructField("name", DataTypes.StringType, false), + DataTypes.createStructField("age", DataTypes.IntegerType, false) + }); + Dataset df = spark.createDataFrame( + Arrays.asList( + RowFactory.create(1, "Alice", 34), + RowFactory.create(2, "Bob", 27), + RowFactory.create(3, "Carol", 45)), + schema); + df.write().format("vortex").mode(SaveMode.Overwrite).save(dataDir.toString()); + } +}