Use a checkpoint to reuse a result

To reuse an expensive result without writing a table yourself, set SAIL_EXECUTION__CHECKPOINT__PATH to a directory before the server starts and call df.checkpoint(); the returned DataFrame is not recomputed by later actions.

Check targets: pysail==0.7.3 with pyspark-client==4.1.3

The Python assertions below check this behavior. Details attributed to external documentation are outside those assertions. Newer releases may behave differently.

Run the check

Follow the test setup, then run from the repository root (sail-examples/). The harness applies the configuration declared in this check.

Sail

uv run pytest 'tests/test_facts.py::test_fact[checkpoint-keeps-results[sail]]'

Python code

checkpoint-keeps-results.py37 lines

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37FACT = {
    "id": "checkpoint-keeps-results",
    "title": "Use a checkpoint to reuse a result",
    "category": "configure",
    "text": """To reuse an expensive result without writing a table yourself, set
        `SAIL_EXECUTION__CHECKPOINT__PATH` to a directory before the server starts and call
        `df.checkpoint()`; the returned DataFrame is not recomputed by later actions.""",
    "engines": ["sail"],
    "order": 4,
    "env": {"SAIL_EXECUTION__CHECKPOINT__PATH": "{tmp}"},
}

import tempfile
from pathlib import Path

from pyspark.sql import functions as F


def check(spark, engine):
    log = Path(tempfile.mkdtemp()) / "calls.log"
    log_path = str(log)

    @F.udf("long")
    def counted(x):
        with open(log_path, "a") as f:
            f.write("x\n")
        return x

    rows = 20
    saved = spark.range(rows).select(counted("id").alias("v")).checkpoint()
    after_checkpoint = len(log.read_text().splitlines())
    saved.collect()
    saved.collect()
    total = len(log.read_text().splitlines())
    assert after_checkpoint == rows, f"checkpoint should compute once ({rows}), got {after_checkpoint}"
    assert total == rows, f"later actions recomputed the UDF: {total} calls"
    return {"udf_calls": total, "rows": rows}