[SPARK-60101][SQL] Support native data sources implemented in Rust or C++

LICENSE-binary

@@ -301,6 +301,7 @@ joda-time:joda-time
 net.sf.opencsv:opencsv
 net.sf.supercsv:super-csv
 net.sf.jpam:jpam
+org.apache.arrow:arrow-c-data
 org.apache.arrow:arrow-compression
 org.apache.arrow:arrow-format
 org.apache.arrow:arrow-memory-core

common/utils/src/main/resources/error/error-conditions.json

@@ -4792,6 +4792,57 @@
     ],
     "sqlState" : "42000"
   },
+  "INVALID_NATIVE_DATA_SOURCE_LIBRARY" : {
+    "message" : [
+      "Invalid native data source library <path>:"
+    ],
+    "subClass" : {
+      "CANNOT_LOAD" : {
+        "message" : [
+          "<reason>"
+        ]
+      },
+      "REPLACED" : {
+        "message" : [
+          "The library was replaced after Spark loaded it: Spark loaded the build with the SHA-256 checksum <loadedChecksum>, but the file now has the checksum <checksum>. A JVM cannot load another build of a library from the same path, so restart the Spark application to use the new build."
+        ]
+      },
+      "UNSUPPORTED_ABI_VERSION" : {
+        "message" : [
+          "It requires version <version> of the native data source interface, but Spark supports version <supported>."
+        ]
+      }
+    },
+    "sqlState" : "22023"
+  },
+  "INVALID_NATIVE_DATA_SOURCE_PACKAGE" : {
+    "message" : [
+      "Invalid native data source package <path>:"
+    ],
+    "subClass" : {
+      "INVALID_LIBRARY" : {
+        "message" : [
+          "Cannot load the library <library>: <reason>"
+        ]
+      },
+      "INVALID_MANIFEST" : {
+        "message" : [
+          "<reason>"
+        ]
+      },
+      "UNSUPPORTED_ABI_VERSION" : {
+        "message" : [
+          "It requires version <version> of the native data source interface, but Spark supports version <supported>."
+        ]
+      },
+      "UNSUPPORTED_PLATFORM" : {
+        "message" : [
+          "It has no library for the platform <platform>. It has libraries for: <platforms>."
+        ]
+      }
+    },
+    "sqlState" : "22023"
+  },
   "INVALID_NON_DETERMINISTIC_EXPRESSIONS" : {
     "message" : [
       "The operator expects a deterministic expression, but the actual expression is <sqlExprs>."
@@ -6262,6 +6313,30 @@
     ],
     "sqlState" : "0A000"
   },
+  "NATIVE_DATA_SOURCE_ERROR" : {
+    "message" : [
+      "Failed to <action> native data source <name>: <msg>"
+    ],
+    "sqlState" : "38000"
+  },
+  "NATIVE_DATA_SOURCE_LIBRARY_NOT_FOUND" : {
+    "message" : [
+      "Cannot find the native data source library <fileName> with the SHA-256 checksum <checksum>, which is installed at <path> on the driver. Install the same build of the library on every node, at the same path or in the native library path."
+    ],
+    "sqlState" : "42K02"
+  },
+  "NATIVE_DATA_SOURCE_PACKAGE_CONFLICT" : {
+    "message" : [
+      "More than one native data source package provides the data source <name>: <packages>. Remove all but one of them."
+    ],
+    "sqlState" : "42710"
+  },
+  "NATIVE_DATA_SOURCE_PACKAGE_NOT_FOUND" : {
+    "message" : [
+      "Cannot find the native data source package <fileName> with the SHA-256 checksum <checksum>. Add the package with `spark.addArtifact`, or install the same package under `spark.sql.dataSource.native.paths` on every node."
+    ],
+    "sqlState" : "42K02"
+  },
   "NEAREST_BY_JOIN" : {
     "message" : [
       "Invalid nearest-by join."

dev/deps/spark-deps-hadoop-3-hive-2.3

@@ -21,6 +21,7 @@ antlr4-runtime/4.13.1//antlr4-runtime-4.13.1.jar
 aopalliance-repackaged/3.0.6//aopalliance-repackaged-3.0.6.jar
 arpack/3.2.0//arpack-3.2.0.jar
 arpack_combined_all/0.1//arpack_combined_all-0.1.jar
+arrow-c-data/19.0.0//arrow-c-data-19.0.0.jar
 arrow-compression/19.0.0//arrow-compression-19.0.0.jar
 arrow-format/19.0.0//arrow-format-19.0.0.jar
 arrow-memory-core/19.0.0//arrow-memory-core-19.0.0.jar

docs/_data/menu-sql.yaml

@@ -75,6 +75,19 @@
       url: sql-v2-data-sources.html#streaming
     - text: Further Reading
       url: sql-v2-data-sources.html#further-reading
+- text: Native Data Sources
+  url: sql-native-data-sources.html
+  subitems:
+    - text: Overview
+      url: sql-native-data-sources.html#overview
+    - text: How It Works
+      url: sql-native-data-sources.html#how-it-works
+    - text: Packages
+      url: sql-native-data-sources.html#packages
+    - text: Example in Rust
+      url: sql-native-data-sources.html#example-in-rust
+    - text: Example in C++
+      url: sql-native-data-sources.html#example-in-c
 - text: Performance Tuning
   url: sql-performance-tuning.html
   subitems:

docs/sql-v2-data-sources.md

@@ -40,7 +40,7 @@ Compared to the earlier Data Source V1 API, DSV2 offers:
 
 | Feature | Description |
 |---------|-------------|
-| **Java API** | The connector interfaces are pure Java (`org.apache.spark.sql.connector`), removing the Scala dependency that DSV1 required. A [Python Data Source API](api/python/reference/pyspark.sql/api/pyspark.sql.datasource.DataSource.html) (`pyspark.sql.datasource`) is also available as a wrapper for lightweight connectors written entirely in Python. |
+| **Java API** | The connector interfaces are pure Java (`org.apache.spark.sql.connector`), removing the Scala dependency that DSV1 required. A [Python Data Source API](api/python/reference/pyspark.sql/api/pyspark.sql.datasource.DataSource.html) (`pyspark.sql.datasource`) is also available as a wrapper for lightweight connectors written entirely in Python. [Native data sources](sql-native-data-sources.html) can be written entirely in native code, such as Rust or C++. |
 | **Catalog integration** | Connectors can expose namespaces, tables, views, and functions natively through Spark SQL. |
 | **Operator pushdown** | Connectors can accept pushed-down filters, required columns, aggregates, limits, offsets, and more. |
 | **Report partitioning and ordering** | Connectors can report the physical layout of data so that Spark can avoid unnecessary shuffles and sorts. |
@@ -473,5 +473,7 @@ A `Table` that declares `STREAMING_WRITE` supports streaming writes through
 - [Data Sources](sql-data-sources.html) for the user-facing guide to built-in data sources (DSV1).
 - [Python Data Source API](api/python/reference/pyspark.sql/api/pyspark.sql.datasource.DataSource.html)
   for writing lightweight connectors entirely in Python.
+- [Native Data Sources](sql-native-data-sources.html) for writing connectors entirely in native
+  code, such as Rust or C++.
 - [Storage Partition Join](sql-performance-tuning.html#storage-partition-join) for how DSV2
   partitioning reporting enables join optimizations.

pom.xml

@@ -2570,6 +2570,25 @@
           </exclusion>
         </exclusions>
       </dependency>
+      <dependency>
+        <groupId>org.apache.arrow</groupId>
+        <artifactId>arrow-c-data</artifactId>
+        <version>${arrow.version}</version>
+        <exclusions>
+          <exclusion>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
+          </exclusion>
+          <exclusion>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-core</artifactId>
+          </exclusion>
+          <exclusion>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-common</artifactId>
+          </exclusion>
+        </exclusions>
+      </dependency>
       <dependency>
         <groupId>org.apache.arrow</groupId>
         <artifactId>arrow-compression</artifactId>

python/pyspark/sql/connect/client/artifact.py

@@ -237,7 +237,8 @@ def _parse_artifacts(
                     name = f"{name}#{parsed.fragment}"
 
                 artifact = new_archive_artifact(name, LocalFile(local_path))
-            elif file:
+            elif file or name.endswith(".sparkpkg"):
+                # Native data source packages are added as files.
                 artifact = new_file_artifact(name, LocalFile(local_path))
             elif name.endswith(".jar"):
                 artifact = new_jar_artifact(name, LocalFile(local_path))

python/pyspark/sql/session.py

@@ -41,6 +41,7 @@
     no_type_check,
     overload,
 )
+from urllib.parse import urlparse
 
 from pyspark.conf import SparkConf
 from pyspark.errors import PySparkRuntimeError, PySparkTypeError, PySparkValueError
@@ -2267,10 +2268,14 @@ def addArtifacts(
         .. versionchanged:: 4.0.0
             Supports Spark Classic.
 
+        .. versionchanged:: 4.4.0
+            Supports native data source packages (``.sparkpkg`` files).
+
         Parameters
         ----------
         *path : tuple of str
-            Artifact's URIs to add.
+            Artifact's URIs to add. A native data source package (a ``.sparkpkg`` file) makes
+            the native data sources it contains available to the session.
         pyfile : bool
             Whether to add them as Python dependencies such as .py, .egg, .zip or .jar files.
             The pyfiles are directly inserted into the path when executing Python functions
@@ -2292,6 +2297,18 @@ def addArtifacts(
                     "condition": "True together",
                 },
             )
+        # Native data source packages are artifacts of the session in the JVM, whatever the
+        # flags, as with Spark Connect. The JVM resolves file URIs, such as
+        # file:///tmp/a.sparkpkg, but resolves a relative path against its own working directory,
+        # so it gets an absolute path instead. Anything else is a path, such as C:/a.sparkpkg on
+        # Windows, or a file name with a colon, such as a:1.sparkpkg.
+        for p in path:
+            if p.endswith(".sparkpkg"):
+                if urlparse(p).scheme == "file":
+                    self._jsparkSession.addArtifact(p)
+                else:
+                    self._jsparkSession.addArtifact(os.path.abspath(p))
+        path = tuple(p for p in path if not p.endswith(".sparkpkg"))
         for p in path:
             normalized_path = os.path.abspath(p)
             target_dir = os.path.join(

python/pyspark/sql/tests/connect/client/test_artifact.py

@@ -447,6 +447,21 @@ def test_cache_artifact(self):
         self.assertEqual(actualHash, expected_hash)
         self.assertEqual(self.artifact_manager.is_cached_artifact(expected_hash), True)
 
+    def test_native_data_source_package(self):
+        with tempfile.TemporaryDirectory(prefix="test_native_data_source_package") as d:
+            package_path = os.path.join(d, "my_source.sparkpkg")
+            with open(package_path, "wb") as f:
+                f.write(b"package")
+            requests = list(
+                self.artifact_manager._create_requests(
+                    package_path, pyfile=False, archive=False, file=False
+                )
+            )
+            self.assertEqual(len(requests), 1)
+            artifacts = requests[0].batch.artifacts
+            self.assertEqual(len(artifacts), 1)
+            self.assertEqual(artifacts[0].name, os.path.join("files", "my_source.sparkpkg"))
+
     def test_add_not_existing_artifact(self):
         with tempfile.TemporaryDirectory(prefix="test_add_not_existing_artifact") as d:
             with self.assertRaises(FileNotFoundError):

python/pyspark/sql/tests/test_artifact.py

@@ -16,6 +16,7 @@
 #
 import os
 import tempfile
+from pathlib import Path
 
 from pyspark.errors import PySparkRuntimeError
 from pyspark.sql.functions import assert_true, lit, udf
@@ -105,6 +106,47 @@ def test_artifacts_cannot_be_overwritten(self):
                 messageParameters={"normalized_path": pyfile_path},
             )
 
+    def test_add_native_data_source_package(self):
+        # A new session, so that the packages do not affect the other tests.
+        spark = self.spark.newSession()
+        with tempfile.TemporaryDirectory(prefix="test_add_native_data_source_package") as d:
+
+            def new_file(name):
+                file_path = os.path.join(d, name)
+                with open(file_path, "wb") as f:
+                    f.write(name.encode("utf-8"))
+                return file_path
+
+            spark.addArtifact(new_file("my_source.sparkpkg"))
+            # Packages are added to the session whatever the flags, as with Spark Connect, also
+            # together with other files.
+            spark.addArtifact(new_file("my_file_source.sparkpkg"), file=True)
+            spark.addArtifacts(new_file("my_other_source.sparkpkg"), new_file("other.txt"))
+            # A URI, and paths relative to the working directory of Python, which is not the
+            # one of the JVM, also with a colon, which is not the scheme of a URI.
+            spark.addArtifact(Path(new_file("my_uri_source.sparkpkg")).as_uri())
+            new_file("my_relative_source.sparkpkg")
+            new_file("my_source:1.sparkpkg")
+            cwd = os.getcwd()
+            os.chdir(d)
+            try:
+                spark.addArtifact("my_relative_source.sparkpkg")
+                spark.addArtifact("my_source:1.sparkpkg")
+            finally:
+                os.chdir(cwd)
+            packages = spark._jsparkSession.artifactManager().getNativeDataSourcePackages()._1()
+            self.assertEqual(
+                sorted(packages.apply(i).getName() for i in range(packages.size())),
+                [
+                    "my_file_source.sparkpkg",
+                    "my_other_source.sparkpkg",
+                    "my_relative_source.sparkpkg",
+                    "my_source.sparkpkg",
+                    "my_source:1.sparkpkg",
+                    "my_uri_source.sparkpkg",
+                ],
+            )
+
     def test_add_zipped_package(self):
         self.check_add_zipped_package(self.spark)
 

sql/api/src/main/scala/org/apache/spark/sql/Artifact.scala

@@ -38,6 +38,10 @@ private[sql] object Artifact {
   val CLASS_PREFIX: Path = Paths.get("classes")
   val JAR_PREFIX: Path = Paths.get("jars")
   val CACHE_PREFIX: Path = Paths.get("cache")
+  val FILE_PREFIX: Path = Paths.get("files")
+
+  /** The file extension of native data source packages, which are added as file artifacts. */
+  val NATIVE_DATA_SOURCE_PACKAGE_EXTENSION = ".sparkpkg"
 
   def newArtifactFromExtension(
       fileName: String,
@@ -48,13 +52,15 @@ private[sql] object Artifact {
         newJarArtifact(targetFilePath, storage)
       case cf if cf.endsWith(".class") =>
         newClassArtifact(targetFilePath, storage)
+      case pkg if pkg.endsWith(NATIVE_DATA_SOURCE_PACKAGE_EXTENSION) =>
+        newFileArtifact(targetFilePath, storage)
       case other =>
         throw new UnsupportedOperationException(s"Unsupported file format: $other")
     }
   }
 
   def parseArtifacts(uri: URI): Seq[Artifact] = {
-    // Currently only local files with extensions .jar and .class are supported.
+    // Currently only local files with extensions .jar, .class and .sparkpkg are supported.
     uri.getScheme match {
       case "file" =>
         val path = Paths.get(uri)
@@ -80,6 +86,10 @@ private[sql] object Artifact {
     newArtifact(CLASS_PREFIX, ".class", targetFilePath, storage)
   }
 
+  def newFileArtifact(targetFilePath: Path, storage: LocalData): Artifact = {
+    newArtifact(FILE_PREFIX, "", targetFilePath, storage)
+  }
+
   def newCacheArtifact(id: String, storage: LocalData): Artifact = {
     newArtifact(CACHE_PREFIX, "", Paths.get(id), storage)
   }

sql/api/src/main/scala/org/apache/spark/sql/SparkSession.scala

@@ -568,6 +568,9 @@ abstract class SparkSession extends Serializable with Closeable {
    * Ivy settings. Otherwise, Ivy URIs are resolved by the server, which does not search Maven or
    * Ivy repositories local to the client.
    *
+   * Since 4.4.0, it also supports native data source packages: local files with the extension
+   * .sparkpkg, which make the native data sources that they contain available to the session.
+   *
    * @since 4.0.0
    */
   @Experimental
@@ -583,6 +586,9 @@ abstract class SparkSession extends Serializable with Closeable {
    * Ivy settings. Otherwise, Ivy URIs are resolved by the server, which does not search Maven or
    * Ivy repositories local to the client.
    *
+   * Since 4.4.0, it also supports native data source packages: local files with the extension
+   * .sparkpkg, which make the native data sources that they contain available to the session.
+   *
    * @since 4.0.0
    */
   @Experimental
@@ -593,7 +599,9 @@ abstract class SparkSession extends Serializable with Closeable {
    * specified by `target` under the session's working directory of that particular file
    * extension.
    *
-   * Supported target file extensions are .jar and .class.
+   * Supported target file extensions are .jar and .class, and since 4.4.0, .sparkpkg for native
+   * data source packages, which make the native data sources that they contain available to the
+   * session.
    *
    * ==Example==
    * {{{
@@ -613,7 +621,9 @@ abstract class SparkSession extends Serializable with Closeable {
    * Add a single artifact to the session while preserving the directory structure specified by
    * `target` under the session's working directory of that particular file extension.
    *
-   * Supported target file extensions are .jar and .class.
+   * Supported target file extensions are .jar and .class, and since 4.4.0, .sparkpkg for native
+   * data source packages, which make the native data sources that they contain available to the
+   * session.
    *
    * ==Example==
    * {{{
@@ -639,6 +649,9 @@ abstract class SparkSession extends Serializable with Closeable {
    * Ivy settings. Otherwise, Ivy URIs are resolved by the server, which does not search Maven or
    * Ivy repositories local to the client.
    *
+   * Since 4.4.0, it also supports native data source packages: local files with the extension
+   * .sparkpkg, which make the native data sources that they contain available to the session.
+   *
    * @since 4.0.0
    */
   @Experimental