GH-373: Implement Jelly-SPARQL

.github/workflows/_build-wheels.yml

@@ -55,9 +55,11 @@ jobs:
           cp -r submodules/protobuf/proto pyjelly/_proto
           rm -rf tests/conformance_tests/rdf
           rm -rf tests/conformance_tests/grpc
+          rm -rf tests/conformance_tests/sparql
           mkdir -p tests/conformance_tests
           cp -r submodules/protobuf/test/rdf tests/conformance_tests/rdf
           cp -r submodules/protobuf/test/grpc tests/conformance_tests/grpc
+          cp -r submodules/protobuf/test/sparql tests/conformance_tests/sparql
 
       - uses: astral-sh/setup-uv@v7
         with:

.github/workflows/ci.yml

@@ -68,9 +68,11 @@ jobs:
         rm -r pyjelly/_proto
         rm -r tests/conformance_tests/rdf
         rm -r tests/conformance_tests/grpc
+        rm -r tests/conformance_tests/sparql
         cp -r submodules/protobuf/proto pyjelly/_proto
         cp -r submodules/protobuf/test/rdf tests/conformance_tests/rdf
         cp -r submodules/protobuf/test/grpc tests/conformance_tests/grpc
+        cp -r submodules/protobuf/test/sparql tests/conformance_tests/sparql
 
     - uses: astral-sh/setup-uv@v7
       with:

README.md

@@ -9,6 +9,7 @@
 ## Features
 
 - **Fast reading and writing** of RDF knowledge graphs in the [Jelly format](http://w3id.org/jelly).
+- Reading and writing **[SPARQL query results](https://w3id.org/jelly/pyjelly/dev/sparql)** in [Jelly-SPARQL](https://w3id.org/jelly/dev/specification/sparql), with rdflib or the generic API.
 - **Standalone [generic API](https://w3id.org/jelly/pyjelly/dev/generic-sink)** with no third-party dependencies, allowing for:
     - Serialization and parsing of statements to and from Jelly files.
     - Parsing and serializing streams of graphs and statements.

benchmarks/README.md

@@ -1,6 +1,6 @@
 # Benchmarks
 
-Throughput and size benchmarks of Jelly-RDF in pyjelly, compared with the RDF formats of rdflib.
+Throughput and size benchmarks of Jelly-RDF and Jelly-SPARQL in pyjelly, compared with the formats of rdflib.
 
 The method follows the [Jelly-JVM benchmarks](https://github.com/Jelly-RDF/jelly-jvm/tree/main/jmh) as closely as Python allows, so that the results can be read the same way.
 
@@ -109,3 +109,38 @@ Before measuring, each case writes the data, reads it back and compares the stat
 - Before each iteration, the garbage collector runs, outside the timing. The data prepared before measuring is frozen with `gc.freeze()`, so the collector doesn't go through it again and again during the measurement.
 - A sweep runs every case once. There are 3 sweeps. The result is the median of the sweeps. Values whose sweeps differ by more than 10% are marked with `*`.
 - Averages over datasets are geometric means. A method that couldn't run on all datasets has the number of datasets it ran on next to its mean.
+
+## Jelly-SPARQL
+
+The `sparql` suite measures SPARQL query results instead of RDF statements, as in the Jelly-SPARQL benchmarks of Jelly-JVM. Pass `--suite sparql` to `run`, `case` and `size`:
+
+```bash
+uv run python -m benchmarks run write --suite sparql
+uv run python -m benchmarks run read --suite sparql
+uv run python -m benchmarks size --suite sparql
+```
+
+Everything else (processes, iterations, sweeps, cores, the round-trip check) is the same as for RDF.
+
+### Data
+
+- The 13 RiverBench datasets, read as the results of `SELECT * WHERE { ?s ?p ?o }` for triples and `SELECT * WHERE { GRAPH ?g { ?s ?p ?o } }` for quads, with `?g` unbound in the default graph. The first 100,000 rows are used.
+- Synthetic result sets, from the presets of Jelly-JVM's generator, ported in `sparql_data.py`.
+- By default, `run` and `size` use all RiverBench datasets and `realistic-mixed`, `wide-20`, `sparse-random-50`, `lit-lang`, `iri-shuffled`. Pass any presets with `--datasets`. Their names are in `PRESETS` in `sparql_data.py`.
+
+### Methods
+
+| Method | Writes with | Reads with |
+|---|---|---|
+| `rdflib-srj` | rdflib's SPARQL JSON results: `Result.serialize(format="json")` | `Result.parse(format="json")` |
+| `rdflib-srx` | rdflib's SPARQL XML results | `Result.parse(format="xml")` |
+| `rdflib-csv` | rdflib's SPARQL CSV results | `Result.parse(format="csv")` |
+| `rdflib-tsv` | - (rdflib can't write TSV results) | `Result.parse(format="tsv")` |
+| `rdflib-jelly` | Jelly-SPARQL through rdflib's plugin API: `Result.serialize(format="jelly")` | `Result.parse(format="jelly")` |
+| `generic-jelly` | Jelly-SPARQL with pyjelly's generic API, from a list of rows | `parse_jelly_sparql` of the generic API |
+
+The rdflib methods write a `Result` whose bindings are a list of dicts, made before measuring. Reading makes a whole `Result`, with its bindings. Jelly uses pyjelly's default options.
+
+Note that CSV is a lossy result set format, so it should not be compared directly with others.
+
+rdflib reads SPARQL TSV results but can't write them, so `rdflib-tsv` is only measured reading. The TSV it reads is written by a helper in `sparql_methods.py`.

benchmarks/__main__.py

@@ -15,7 +15,7 @@
 from pathlib import Path
 
 from benchmarks import data, measure, report
-from benchmarks.methods import METHODS
+from benchmarks.suites import SUITES
 
 RESULTS_DIR = Path(__file__).parent / "results"
 
@@ -24,18 +24,36 @@ def _names(value: str) -> list[str]:
     return [name.strip() for name in value.split(",") if name.strip()]
 
 
+def _add_suite(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "--suite",
+        choices=list(SUITES),
+        default="rdf",
+        help=(
+            "rdf: Jelly-RDF and RDF formats, sparql: Jelly-SPARQL and SPARQL "
+            "result formats (default: %(default)s)"
+        ),
+    )
+
+
 def _add_selection(parser: argparse.ArgumentParser) -> None:
+    _add_suite(parser)
     parser.add_argument(
         "--datasets",
         type=_names,
-        default=list(data.DATASETS),
-        help="comma-separated RiverBench datasets (default: all 13)",
+        help=(
+            "comma-separated datasets (default: the 13 RiverBench datasets, and in "
+            "the sparql suite, also a few synthetic ones)"
+        ),
     )
     parser.add_argument(
         "--methods",
         type=_names,
-        default=list(METHODS),
-        help=f"comma-separated methods (default: all of {', '.join(METHODS)})",
+        help=(
+            "comma-separated methods (default: all of the suite: "
+            + "; ".join(f"{s.name}: {', '.join(s.methods)}" for s in SUITES.values())
+            + ")"
+        ),
     )
     parser.add_argument(
         "--rows",
@@ -69,6 +87,16 @@ def _add_timing(parser: argparse.ArgumentParser) -> None:
     )
 
 
+def _fill_selection(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
+    """Fill in the suite's default datasets and methods, if none are given."""
+    suite = SUITES[args.suite]
+    args.datasets = args.datasets or list(suite.default_datasets)
+    args.methods = args.methods or list(suite.methods)
+    for name in args.methods:
+        if name not in suite.methods:
+            parser.error(f"no method {name} in the {args.suite} suite")
+
+
 def _format_cpus(cpus: list[int]) -> str:
     """Format CPU cores like taskset does: "5-9,15"."""
     ranges: list[list[int]] = []
@@ -112,10 +140,13 @@ def _case(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
         if sys.platform != "linux":
             parser.error("--cpus works only on Linux")
         os.sched_setaffinity(0, args.cpus)
+    if args.method not in SUITES[args.suite].methods:
+        parser.error(f"no method {args.method} in the {args.suite} suite")
     result = measure.run_case(
         args.task,
         args.dataset,
         args.method,
+        suite=args.suite,
         rows=args.rows,
         warmup=args.warmup,
         iterations=args.iterations,
@@ -133,6 +164,7 @@ def _run(args: argparse.Namespace) -> None:
     settings = {
         name: getattr(args, name)
         for name in (
+            "suite",
             "task",
             "datasets",
             "methods",
@@ -156,9 +188,11 @@ def _run(args: argparse.Namespace) -> None:
         seconds=args.time,
         check=args.check,
         cpu_sets=args.cpus,
+        suite=args.suite,
     )
     stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
-    out = args.out or RESULTS_DIR / f"{stamp}-{args.task}.json"
+    prefix = "sparql-" if args.suite == "sparql" else ""
+    out = args.out or RESULTS_DIR / f"{stamp}-{prefix}{args.task}.json"
     measure.save(out, settings, results)
     print(f"\nSaved to {out}\n")
     print(report.throughput_table(args.task, results))
@@ -196,7 +230,8 @@ def main() -> None:
     )
     case.add_argument("task", choices=["write", "read"])
     case.add_argument("dataset")
-    case.add_argument("method", choices=list(METHODS))
+    case.add_argument("method")
+    _add_suite(case)
     case.add_argument("--rows", type=int, default=data.DEFAULT_ROWS)
     case.add_argument(
         "--cpus", type=measure.parse_cpu_list, help="CPU cores to pin the case to"
@@ -215,14 +250,16 @@ def main() -> None:
     show.add_argument("results", type=Path, nargs="+")
 
     args = parser.parse_args()
+    if args.command in ("run", "size"):
+        _fill_selection(args, parser)
     if args.command == "fetch":
         data.fetch(args.datasets)
     elif args.command == "case":
         _case(args, parser)
     elif args.command == "run":
         _run(args)
     elif args.command == "size":
-        print(report.size_tables(args.datasets, args.methods, args.rows))
+        print(report.size_tables(args.datasets, args.methods, args.rows, args.suite))
     elif args.command == "report":
         for path in args.results:
             content = json.loads(path.read_text())

benchmarks/measure.py

@@ -30,8 +30,8 @@
 from typing import Any
 
 import pyjelly.parse.decode
-from benchmarks.data import load
-from benchmarks.methods import METHODS, NullOutput, round_trip
+from benchmarks.methods import NullOutput
+from benchmarks.suites import SUITES
 from pyjelly.options import MAX_VERSION
 
 # The folder that has the `benchmarks` package, where the cases run. With a
@@ -45,9 +45,10 @@ class CaseResult:
     dataset: str
     method: str
     #: Statements done in one operation: every statement the writer writes, or
-    #: every statement the reader found in the bytes
+    #: every statement the reader found in the bytes. In the sparql suite, cells
+    #: (rows times variables).
     statements: int = 0
-    #: Throughput of each iteration, in statements per second
+    #: Throughput of each iteration, in statements (or cells) per second
     warmup: list[float] = field(default_factory=list)
     measured: list[float] = field(default_factory=list)
     #: Mean throughput of the measured iterations, None if the case didn't run
@@ -61,6 +62,8 @@ class CaseResult:
     #: The CPU cores the case ran on
     cpus: list[int] | None = None
     sweep: int = 0
+    #: Suite of the case: "rdf" or "sparql", see suites.py
+    suite: str = "rdf"
 
 
 def _iteration(
@@ -94,21 +97,27 @@ def run_case(  # noqa: PLR0913
     seconds: float,
     check: bool = True,
     profile: Path | None = None,
+    suite: str = "rdf",
 ) -> CaseResult:
     """
     Run one case in this process. See the module docstring.
 
     With `check=False`, the round-trip check is skipped. Do that only when the
     code under test hasn't changed since the check last passed.
     """
-    result = CaseResult(task, dataset, method_name, checked=check)
+    result = CaseResult(task, dataset, method_name, checked=check, suite=suite)
     if hasattr(os, "sched_getaffinity"):
         result.cpus = sorted(os.sched_getaffinity(0))
-    method = METHODS[method_name]
-    data = load(dataset, rows)
+    benchmarks = SUITES[suite]
+    round_trip = benchmarks.round_trip
+    method = benchmarks.methods[method_name]
+    data = benchmarks.load(dataset, rows)
     if not method.supports(data):
         result.skipped = "the format can't hold this kind of statements"
         return result
+    if task == "write" and not getattr(method, "can_write", True):
+        result.skipped = "the library can't write the format"
+        return result
 
     # Setup, not measured: make the terms (and the graph, for writers that need
     # one), check that the method round-trips the data, and make the bytes to read
@@ -140,7 +149,7 @@ def operation() -> None:
         result.statements = method.read(written, quads=data.quads)
 
         def operation() -> int:
-            return method.read(written, quads=data.quads)
+            return int(method.read(written, quads=data.quads))
 
     # Everything made so far lives until the end. Freezing it keeps the garbage
     # collector from going through it again and again during the measurement.
@@ -222,6 +231,7 @@ def _run_case_process(  # noqa: PLR0913
     seconds: float,
     check: bool,
     cpus: Sequence[int] | None,
+    suite: str,
 ) -> CaseResult:
     """Run one case in a new process, pinned to `cpus` if given."""
     command = [
@@ -232,6 +242,7 @@ def _run_case_process(  # noqa: PLR0913
         task,
         dataset,
         method,
+        f"--suite={suite}",
         f"--rows={rows}",
         f"--warmup={warmup}",
         f"--iterations={iterations}",
@@ -250,7 +261,7 @@ def _run_case_process(  # noqa: PLR0913
         # The JSON is the last line, in case a library printed something
         last_line = process.stdout.strip().splitlines()[-1]
         return CaseResult(**json.loads(last_line))
-    result = CaseResult(task, dataset, method, checked=check)
+    result = CaseResult(task, dataset, method, checked=check, suite=suite)
     result.skipped = f"the process failed with code {process.returncode}"
     return result
 
@@ -311,6 +322,7 @@ def run_sweeps(  # noqa: PLR0913
     seconds: float,
     check: bool = True,
     cpu_sets: Sequence[Sequence[int]] = (),
+    suite: str = "rdf",
 ) -> list[CaseResult]:
     """
     Run every case `sweeps` times, each time in a new process.
@@ -340,9 +352,11 @@ def run_share(sweep: int, worker: int) -> None:
                 seconds=seconds,
                 check=check,
                 cpus=sets[worker],
+                suite=suite,
             )
             result.sweep = sweep
-            score = f"{result.score:,.0f} st/s" if result.score else result.skipped
+            unit = SUITES[suite].unit
+            score = f"{result.score:,.0f} {unit}/s" if result.score else result.skipped
             with lock:
                 results.append(result)
                 progress = f"[{len(results)}/{total}] sweep {sweep + 1}"