New release

README.md

@@ -88,6 +88,14 @@ pip install jsoniq
 
 *Important note*: since the jsoniq package depends on pyspark 4, Java 17 or Java 21 is a requirement. If another version of Java is installed, the execution of a Python program attempting to create a RumbleSession will lead to an error message on stderr that contains explanations.
 
+By default, session creation uses Spark bundled with PySpark. Both `getOrCreate()` and `create()` temporarily unset `SPARK_HOME` and restore its original value (or absence) after startup, even if startup fails. To use an external Spark installation configured through `SPARK_HOME`, opt out before creating the session:
+
+```python
+rumble = RumbleSession.builder.withBundledSpark(False).getOrCreate()
+```
+
+Use `withBundledSpark()` or `withBundledSpark(True)` to enable the default again. This setting cannot change the Spark installation of an existing session or JVM. Startup calls through this library are serialized while the environment is changed; unrelated threads should avoid changing `SPARK_HOME` or starting Spark at the same time.
+
 ## Sample code
 
 We will make more documentation available as we go. In the meantime, you will find a sample, commented code below that should just run
@@ -386,7 +394,7 @@ Even more queries can be found [here](https://colab.research.google.com/github/R
 
 # Latest updates
 
-## Version 3.0.0
+## Version 3.0.1
 - Upgraded to RumbleDB 3.0.0 and its immutable configuration and external bindings APIs.
 - There is a breaking change in how configuration parameters are set. New:
 ```python
@@ -398,6 +406,7 @@ rumble.getRumbleConf().set("debug.showErrorInfo", True)
 - Fixed object conversion and binding query results as DataFrames. Keyword bindings are scoped to a query and restore persistent bindings even when query compilation fails.
 - New rumble.xquery() call and %%xquery magic.
 - In the %%jsoniq magic, the new -s parameter outputs the results as a single string following the W3C Serialization 3.1 specification. If using `rumble.xquery()` or `rumble.jsoniq()` call, the same can be achieved by chaining a `.serialize() ` call returning a string.  All standard serialization methods (xml, json, xhtml, html, text, adaptive) are available. For %%xquery, this is the default behavior.
+- By default, the Spark bundled with pyspark is used. Use .withBundledSpark(False) when creating the session to use a different spark installation that SPARK_HOME points to (for advanced users only).
 
 ## Version 2.1.9
 - Fixed a bug in the inferred conversion to DataFrames of output involving arrays of objects.

src/jsoniq/session.py

@@ -6,9 +6,12 @@
 import platform
 import os
 import re
+from threading import RLock
 import pandas as pd
 from importlib.resources import files, as_file
 
+_spark_creation_lock = RLock()
+
 with as_file(files("jsoniq.jars").joinpath("rumbledb-3.0.0.jar")) as jar_path:
     if (os.name == 'nt'):
         jar_path_str = str(jar_path)
@@ -78,23 +81,48 @@ def __init__(self):
                 sys.stderr.write("[Error] Could not determine Java version. Please ensure Java is installed and JAVA_HOME is properly set.\n")
                 sys.exit(43)
             self._sparkbuilder = SparkSession.builder.config("spark.jars", jar_path_str)
+            self._use_bundled_spark = True
             self._appendable_keys = {
                 "spark.jars.packages",
                 "spark.sql.extensions",
             }
 
+        def withBundledSpark(self, enabled=True):
+            """Use PySpark's bundled Spark by default; pass False to respect SPARK_HOME.
+
+            This only affects startup of a new JVM. Existing sessions are reused.
+            """
+            self._use_bundled_spark = enabled
+            return self
+
+        def _create_session(self, method):
+            # Serialize this library's startup calls while changing the process environment.
+            with _spark_creation_lock:
+                if not self._use_bundled_spark:
+                    return RumbleSession(getattr(self._sparkbuilder, method)())
+                previous_spark_home = os.environ.pop("SPARK_HOME", None)
+                try:
+                    return RumbleSession(getattr(self._sparkbuilder, method)())
+                finally:
+                    if previous_spark_home is None:
+                        os.environ.pop("SPARK_HOME", None)
+                    else:
+                        os.environ["SPARK_HOME"] = previous_spark_home
+
         def getOrCreate(self):
             if RumbleSession._rumbleSession is None:
                 try:
-                    RumbleSession._rumbleSession = RumbleSession(self._sparkbuilder.getOrCreate())
+                    RumbleSession._rumbleSession = self._create_session("getOrCreate")
                 except FileNotFoundError as e:
-                    if not os.environ.get('SPARK_HOME') is None:
+                    if not self._use_bundled_spark and not os.environ.get('SPARK_HOME') is None:
                         sys.stderr.write("[Error] SPARK_HOME environment variable may not be set properly. Please check that it points to a valid path to a Spark 4.0 directory, or maybe the easiest would be to delete the environment variable SPARK_HOME completely to fall back to the installation of Spark 4.0 packaged with pyspark.\n")
                         sys.stderr.write(f"Current value of SPARK_HOME: {os.environ.get('SPARK_HOME')}\n")
                         sys.exit(43)
                     else:
                         raise e
                 except TypeError as e:
+                    if self._use_bundled_spark:
+                        raise
                     spark_version = get_spark_version()
                     if not os.environ.get('SPARK_HOME') is None and spark_version is None:
                         sys.stderr.write("[Error] Could not determine Spark version. The SPARK_HOME environment variable may not be set properly. Please check that it points to a valid path to a Spark 4.0 directory, or maybe the easiest would be to delete the environment variable SPARK_HOME completely to fall back to the installation of Spark 4.0 packaged with pyspark.\n")
@@ -111,7 +139,7 @@ def getOrCreate(self):
             return RumbleSession._rumbleSession
 
         def create(self):
-            RumbleSession._rumbleSession = RumbleSession(self._sparkbuilder.create())
+            RumbleSession._rumbleSession = self._create_session("create")
             return RumbleSession._rumbleSession
 
         def remote(self, spark_url):

tests/test_bundled_spark.py

@@ -0,0 +1,117 @@
+import os
+from unittest.mock import Mock, patch
+
+import pytest
+
+from jsoniq import RumbleSession
+
+
+@pytest.fixture
+def builder(monkeypatch):
+    monkeypatch.setattr(RumbleSession, "_rumbleSession", None)
+    spark_builder = Mock()
+    with patch("jsoniq.session.os.popen") as popen, patch("jsoniq.session.SparkSession") as spark:
+        popen.return_value.read.return_value = 'openjdk version "17.0.1"'
+        spark.builder.config.return_value = spark_builder
+        return RumbleSession.Builder()
+
+
+def set_spark_home(monkeypatch, value):
+    if value is None:
+        monkeypatch.delenv("SPARK_HOME", raising=False)
+    else:
+        monkeypatch.setenv("SPARK_HOME", value)
+
+
+@pytest.mark.parametrize("method", ["getOrCreate", "create"])
+@pytest.mark.parametrize("original", [None, "", "/external/spark"])
+def test_bundled_spark_restores_environment_after_startup(builder, monkeypatch, method, original):
+    set_spark_home(monkeypatch, original)
+    spark_session = Mock()
+
+    def start():
+        assert "SPARK_HOME" not in os.environ
+        # Also restore absence if startup itself changes the environment.
+        os.environ["SPARK_HOME"] = "/startup/spark"
+        return spark_session
+
+    getattr(builder._sparkbuilder, method).side_effect = start
+    result = getattr(builder, method)()
+    assert result._sparksession is spark_session
+    assert os.environ.get("SPARK_HOME") == original
+    assert ("SPARK_HOME" in os.environ) == (original is not None)
+
+
+@pytest.mark.parametrize("method", ["getOrCreate", "create"])
+@pytest.mark.parametrize("original", [None, "", "/external/spark"])
+@pytest.mark.parametrize("error_type", [RuntimeError, FileNotFoundError, TypeError, SystemExit])
+def test_bundled_spark_restores_environment_after_failure(builder, monkeypatch, method, original, error_type):
+    set_spark_home(monkeypatch, original)
+    failure = error_type("Startup failed")
+
+    def start():
+        assert "SPARK_HOME" not in os.environ
+        raise failure
+
+    getattr(builder._sparkbuilder, method).side_effect = start
+    with pytest.raises(error_type) as error:
+        getattr(builder, method)()
+    assert error.value is failure
+    assert os.environ.get("SPARK_HOME") == original
+    assert ("SPARK_HOME" in os.environ) == (original is not None)
+    assert RumbleSession._rumbleSession is None
+
+
+@pytest.mark.parametrize("method", ["getOrCreate", "create"])
+def test_external_spark_opt_out_preserves_environment(builder, monkeypatch, method):
+    monkeypatch.setenv("SPARK_HOME", "/external/spark")
+
+    def start():
+        assert os.environ["SPARK_HOME"] == "/external/spark"
+        return Mock()
+
+    getattr(builder._sparkbuilder, method).side_effect = start
+    assert builder.withBundledSpark(False) is builder
+    getattr(builder, method)()
+    assert os.environ["SPARK_HOME"] == "/external/spark"
+
+
+def test_get_or_create_reuses_existing_session(builder, monkeypatch):
+    existing = Mock()
+    monkeypatch.setattr(RumbleSession, "_rumbleSession", existing)
+    monkeypatch.setenv("SPARK_HOME", "/external/spark")
+    assert builder.getOrCreate() is existing
+    builder._sparkbuilder.getOrCreate.assert_not_called()
+    assert os.environ["SPARK_HOME"] == "/external/spark"
+
+
+def test_bundled_spark_can_be_reenabled(builder, monkeypatch):
+    monkeypatch.setenv("SPARK_HOME", "/external/spark")
+    builder.withBundledSpark(False)
+    assert builder.withBundledSpark() is builder
+
+    def start():
+        assert "SPARK_HOME" not in os.environ
+        return Mock()
+
+    builder._sparkbuilder.getOrCreate.side_effect = start
+    builder.getOrCreate()
+    assert os.environ["SPARK_HOME"] == "/external/spark"
+
+
+def test_bundled_spark_starts_with_invalid_external_home(monkeypatch, tmp_path):
+    monkeypatch.setattr(RumbleSession, "_rumbleSession", None)
+    invalid_home = str(tmp_path / "missing-spark")
+    monkeypatch.setenv("SPARK_HOME", invalid_home)
+    session = None
+    try:
+        session = (
+            RumbleSession.Builder().master("local[2]")
+            .config("spark.ui.enabled", "false").getOrCreate()
+        )
+        assert session.version == "4.0.3"
+        assert session.jsoniq("1 + 1").json() == (2,)
+        assert os.environ["SPARK_HOME"] == invalid_home
+    finally:
+        if session is not None:
+            session.stop()