From 73847f9a5655abd84b612b3bb0f7893359a7009d Mon Sep 17 00:00:00 2001 From: Ghislain Fourny Date: Wed, 30 Sep 2026 17:11:09 +0200 Subject: [PATCH 1/2] Now offer option between pyspark or SPARK_HOME. --- README.md | 11 ++++++++++- src/jsoniq/session.py | 34 +++++++++++++++++++++++++++++++--- 2 files changed, 41 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 3d1b7b9..dddf19f 100644 --- a/README.md +++ b/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. diff --git a/src/jsoniq/session.py b/src/jsoniq/session.py index 5b525da..36c7a5c 100644 --- a/src/jsoniq/session.py +++ b/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): From 1850b3e6d5edbb2b62a2f34bfdb8dd8193e968c7 Mon Sep 17 00:00:00 2001 From: Ghislain Fourny Date: Wed, 30 Sep 2026 17:11:24 +0200 Subject: [PATCH 2/2] Add test. --- tests/test_bundled_spark.py | 117 ++++++++++++++++++++++++++++++++++++ 1 file changed, 117 insertions(+) create mode 100644 tests/test_bundled_spark.py diff --git a/tests/test_bundled_spark.py b/tests/test_bundled_spark.py new file mode 100644 index 0000000..951c372 --- /dev/null +++ b/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()