Python and Jupyter API

This chapter describes the embedded-engine API on master after merging epic #445 in PR #510 and the #511 Result port in PR #512, including the storage-directory contract of #450. Separate pages cover wheel and Jupyter installation, building the library, and notebook acceptance. C2/C3 history does not replace this contract. An existing wheel can come from an earlier revision; its package version alone does not establish that it contains these changes.

Roles and scope

Engine executes on a logical timeline, without a wall clock, IPC thread or terminal. It does not join the bus, claim a namespace or publish streams to xqry. Use xretractor for real-time processing and the monitoring API to read its service streams. The embedded library provides no real-time guarantees.

The package imports _core lazily when the embedded API is requested. The pure client works without the extension. NumPy is the embedded library’s dependency; only an explicitly imported retractordb.torch imports PyTorch. Installing the client with pip install ./api/python replaces the wheel distribution of the same name, including _core; use separate environments.

Complete example

The example puts input outside the engine-owned directory, executes a plan, verifies values and checks DLPack sharing. The PyTorch part is optional.

from pathlib import Path
from tempfile import TemporaryDirectory
import numpy as np
import retractordb as rdb

with TemporaryDirectory() as directory:
    source = Path(directory) / "input.txt"
    source.write_text("\n".join(str(value) for value in range(10, 81, 10)) + "\n")
    with rdb.Engine() as engine:
        engine.compile(f"""
            DECLARE a INTEGER STREAM src, 1/2 TEXTFILE '{source}'
            SELECT a*2 STREAM dst FROM src
        """)
        assert engine.run() == 8
        assert engine.slots_done == 8 and engine.time == 4
        assert [row[0] for row in engine.rows("dst")] == list(range(20, 161, 20))
        descriptor = engine.schema("dst")
        field = descriptor[0].name
        block = engine.to_numpy("dst", fields=[field], dtype="float64")
        np.testing.assert_array_equal(block[:, 0], np.arange(20, 161, 20))
        window = engine.window("dst", fields=[field], size=4, stride=2)
        assert window.shape == (3, 4, 1)
        try:
            import torch
        except ImportError:
            torch = None
        if torch is not None:
            tensor = torch.from_dlpack(window)
            assert tensor.data_ptr() == window.to_numpy().__array_interface__["data"][0]
    np.testing.assert_array_equal(block[:, 0], np.arange(20, 161, 20))
    assert window.shape == (3, 4, 1)

Eight records produce eight slots and logical time 4 at interval 1/2. Records, arrays and windows copy engine results and remain readable after the engine closes.

Engine lifecycle and execution

The default until_eof=True stops after the slot in which the first declared source runs dry. until_eof=False allows sources to wrap at EOF. A plan without an ending input needs a budget such as run(slots=8) or interruption. A slot is not a record count for every stream: streams can have different intervals and delayed starts.

A failed recompilation also closes the previous plan; refusal does not restore it. After K2/K3, fix the request and compile again. Failure inside a slot latches failed; subsequent step() repeats the failure class until recompilation. Existing results can remain readable after K4. K1 additionally taints the module until interpreter restart.

Compilation, stepping, running and record reads release the GIL. run() checks Python signals approximately every 50 ms between slots, allowing KeyboardInterrupt; this is not a maximum interruption delay for a long slot. Diagnostics reach logging.getLogger("retractordb").

An exception from a custom logging handler is reported through sys.unraisablehook; it does not replace the original compilation error or, by itself, taint the module. test_logging_handler_failure_preserves_the_plan_error_and_next_compile checks that the plan can be corrected and run again after this event.

Storage directory and offline access

Engine() creates a private directory that survives close() and is removed on destruction. An explicit storage_dir, or an overriding RQL STORAGE, must name an empty directory on first use or one with a valid .retractordb-embedded ownership marker. Foreign directories are refused. The marker does not establish that an owner process is still alive.

If writing a newly created marker fails, StorageGuard attempts to remove the incomplete marker and returns IOError with the cause. It does not treat a partial marker as successful directory initialization.

The directory lock lasts until the plan’s storages close. It excludes another Engine and Storage in a marked directory, including through path aliases or another process. Process death releases the lock. Explicit directories remain on disk; recompilation can replace their descriptors and results. Preflight refusal preserves old artifacts but closes the plan; later I/O failures are not transactional.

Inputs must be outside the result directory and outputs inside it. A missing declared source is refused during compile() as IOError with errno=ENOENT or ENOTDIR, the stream name and the path in the message; denied read access raises IOError with errno=EACCES. Both cases are also an environment error, K4, when executing through xretractor: the process returns EIO (5 on Linux), with and without --until-eof. Removal after compilation is a separate case. Explicit DISPOSABLE still authorizes removal of its source after use.

Opening Storage can rebuild .meta, so access is offline. For a directory used by a service, pass its lock_dir or config naming a TOML file with paths.lock_dir; the argument wins. Configuration is read only when explicitly named. Without either, the check guesses TMPDIR and reports a UserWarning and an INFO record to the retractordb logger.

[embedded] lenient_lock_dirs lists directories where only a held lock refuses access. Without the key, the system temporary directory is the default lenient directory. The key replaces that list; other directories are checked strictly. The opening check does not exclude a later daemon start or a service using another lock directory. A private Engine() directory skips the service-lock check.

Storage, Descriptor and Record

Values become int, float, str, Fraction for RATIONAL, tuples for pairs and None for NULL. STRING[N] occupies one value slot, while a numeric array occupies N. Configuration fields occupy no record slot. Direct Storage cannot read BINFILE, TEXTFILE or DEVICE sources; a plan’s source results are readable through Engine within retained history.

Retention can expire indices. record_count is a logical counter, not the count of all still-readable records; use retained_from. Default rows() and to_numpy() read the retained tail. An expired index refuses instead of returning another ring-buffer entry.

Errors

RetractorDBError
+-- ConfigError (also ValueError)
|   +-- PlanError
|       +-- RQLSyntaxError
|       +-- CompileError
|       +-- CorruptDescriptor
+-- IOError (also OSError)
+-- InternalError
+-- NoSuchStream
+-- StorageError

IOError.errno preserves a system code or uses EIO when no code is available; filename is set when the call accepts a path argument. Otherwise the path can be in the message. Field/stream names, indices, array conversion and signatures can raise standard KeyError, IndexError, ValueError or TypeError; not every exception derives from RetractorDBError.

The core returns recoverable errors as rdb::Result<T>; the binding translates the Error category into the Python exception classes above. Descriptor-read I/O failures give IOError with the original errno, rather than CorruptDescriptor, which describes invalid contents. An unexpected C++ exception escaping a core call, such as bad_alloc or length_error, gives InternalError and taints the module. Do not catch InternalError intending to continue computation in the same kernel.

rdb::fatal() and RDB_ASSERT invoke a separate process handler that never returns. Its default action terminates the host through abort (SIGABRT), including a Jupyter kernel; the binding does not turn this path into a Python exception. xretractor installs its own handler, exiting with code 1 and cleaning up IPC. The handler is a policy for the whole process, rather than one Engine. RDB_NOEXCEPT_CHECK=OFF does not change this behavior. Failures inside noexcept or during destruction also have no general interpreter-survival guarantee.

Invalid input should not reach the fatal path: for example, the compiler refuses deinterleave & or % when its argument equals the source interval, giving CompileError. Correct the plan and reuse Engine. Subprocess tests in test_error_classes.py and test_fatal_paths.py check the Result, C++ exception and SIGABRT boundaries; ut_compiler also guards the deinterleave refusal.

Memory ownership, NumPy and DLPack

fields determines value-axis order; numeric arrays are flattened. STRING, INTPAIR and IDXPAIR have no dense numeric form. Export dtypes are float32, float64, int32, int64. Floating exports map NULL to NaN; integer exports reject NULL, infinity and out-of-range results with ValueError. Finite fractions truncate toward zero. Projection passes through double, so use Record, rather than a dense array, for full precision of large integers.

size and stride must be positive. Fewer than size records produce (0, size, values). For N retained records with N >= size, window count is 1 + (N-size)//stride. sliding_window_view selects the windows, which are then copied into a contiguous array. Arrays and tensors survive later slots, engine close and destruction. Mutating the window array is visible to its shared tensor but does not modify engine results.

Devices, PyTorch and threads

retractordb.torch.as_tensor(window, device="cpu") shares CPU memory; an explicitly requested other device transfers and copies. StreamDataset(engine, stream, fields=..., window=..., stride=1, dtype=torch.float32, device="cpu") yields (size, values) items. Each iteration reads current results, so new slots can add windows to a later epoch.

Use DataLoader(..., num_workers=0). A worker refuses before accessing the engine; neither fork nor creating Engine in a worker initializer belongs to this dataset contract. Serialize all calls on one instance, including reads, getters, compile() and close() during run(). The GIL does not protect engine state; concurrent compilation or execution of separate engines has no general thread-safety guarantee.

CPU/DLPack does not pin engine buffers or provide pinned host memory for CUDA. CUDA transfer is a separate operation. MPS rejects float64 before transfer; choose float32. The CPU guard test is not MPS hardware acceptance. The #448 Colab T4 demonstration uses CUDA, fp16 and GradScaler; it does not establish all-device support. DeviceWindowLoader, MPS acceptance and PyPI publication remain optional follow-up tasks outside the epic.

Embedded-plan limits

compile() refuses DUMP, SYSTEM, :ROTATION and DEVICE sources as CompileError: they require service state or execution phases that Engine lacks. A missing input gives the IOError described above. Python push ingest and mobile targets are unavailable. The complete service and its configuration remain the responsibility of xretractor.

Contract sources and tests

retractordb/*.py paths are relative to api/python/. Test names identify coverage, not execution on every platform. test_reference.py compares the common scope with xretractor -u, including schema, NULL, metadata and logical scheduling; it does not measure real time. Missing-source refusal is checked separately by test_engine.py, ut_embedEngine and it_untileof_stop-missing-sources (#488). Service-only features remain outside this comparison. Record the conditions and results of each acceptance run separately.