The Three-Part `make()` — Long Computations Without Long Transactions¶

A standard make() runs fetch → compute → insert inside one database transaction. That is fine for quick derivations, but when the compute step takes minutes to hours, holding a transaction open that long blocks other writers, risks transaction timeouts, and pins resources. The three-part (tripartite) make() splits the work into make_fetch, make_compute, and make_insert so the slow part runs with no transaction held — while DataJoint still guarantees referential integrity by re-fetching and hash-verifying the inputs before it inserts.

In [1]:

Copied!





import datajoint as dj
import numpy as np

schema = dj.Schema("tripartite_demo")
schema.drop(prompt=False)          # reset so the example is re-runnable
schema = dj.Schema("tripartite_demo")
import datajoint as dj
import numpy as np

schema = dj.Schema("tripartite_demo")
schema.drop(prompt=False)          # reset so the example is re-runnable
schema = dj.Schema("tripartite_demo")

[2026-07-18 19:34:08] DataJoint 2.3.1 connected to root@mysql:3306

An upstream source table¶

Recording is a Manual table holding one 1-D waveform per recording (an in-store <blob>). It is the declared input — everything the computation reads comes from here.

In [2]:

Copied!





@schema
class Recording(dj.Manual):
    definition = """
    recording_id     : int32
    ---
    sampling_rate_hz : float64
    signal           : <blob>      # 1-D waveform, float64 samples
    """

rng = np.random.default_rng(0)
Recording.insert(
    {
        "recording_id": i,
        "sampling_rate_hz": 1000.0,
        "signal": (rng.standard_normal(5000) * (i + 1)).astype(np.float64),
    }
    for i in range(3)
)
Recording()
@schema
class Recording(dj.Manual):
    definition = """
    recording_id     : int32
    ---
    sampling_rate_hz : float64
    signal           : <blob>      # 1-D waveform, float64 samples
    """

rng = np.random.default_rng(0)
Recording.insert(
    {
        "recording_id": i,
        "sampling_rate_hz": 1000.0,
        "signal": (rng.standard_normal(5000) * (i + 1)).astype(np.float64),
    }
    for i in range(3)
)
Recording()

Out[2]:

recording_id	sampling_rate_hz	signal
0	1000.0	<blob>
1	1000.0	<blob>
2	1000.0	<blob>

Total: 3

The computed table, in three phases¶

RecordingStats derives amplitude statistics for each recording and, in a Part table, the RMS of four equal-length windows.

make_fetch(key) reads only the declared upstream input and returns it as a tuple. It runs outside the transaction and is re-run inside it; its output is hash-verified, so it must be reproducible and must not write.
make_compute(key, signal) is the slow part. It touches no database (no transaction is open) and depends only on what make_fetch returned. Because its result is inserted once and never re-verified, it may even be stochastic.
make_insert(key, master_row, window_rows) runs inside a brief transaction and writes only to self and its Part table, atomically.

In [3]:

Copied!





@schema
class RecordingStats(dj.Computed):
    definition = """
    -> Recording
    ---
    n_samples : int32
    mean_amp  : float64
    rms_amp   : float64
    peak_amp  : float64
    """

    class Window(dj.Part):
        definition = """
        -> master
        window_id  : int32
        ---
        window_rms : float64
        """

    def make_fetch(self, key, **kwargs):
        # Phase 1 — read the declared upstream input; return a tuple.
        signal = (Recording & key).fetch1("signal")
        return (signal,)

    def make_compute(self, key, signal):
        # Phase 2 — no DB access; a long computation would live here.
        master_row = {
            **key,
            "n_samples": int(signal.size),
            "mean_amp": float(signal.mean()),
            "rms_amp": float(np.sqrt(np.mean(signal ** 2))),
            "peak_amp": float(np.max(np.abs(signal))),
        }
        window_rows = [
            {**key, "window_id": i, "window_rms": float(np.sqrt(np.mean(w ** 2)))}
            for i, w in enumerate(np.array_split(signal, 4))
        ]
        return (master_row, window_rows)   # tuple -> unpacked into make_insert

    def make_insert(self, key, master_row, window_rows):
        # Phase 3 — atomic write to self + Part, inside the transaction.
        self.insert1(master_row)
        self.Window.insert(window_rows)
@schema
class RecordingStats(dj.Computed):
    definition = """
    -> Recording
    ---
    n_samples : int32
    mean_amp  : float64
    rms_amp   : float64
    peak_amp  : float64
    """

    class Window(dj.Part):
        definition = """
        -> master
        window_id  : int32
        ---
        window_rms : float64
        """

    def make_fetch(self, key, **kwargs):
        # Phase 1 — read the declared upstream input; return a tuple.
        signal = (Recording & key).fetch1("signal")
        return (signal,)

    def make_compute(self, key, signal):
        # Phase 2 — no DB access; a long computation would live here.
        master_row = {
            **key,
            "n_samples": int(signal.size),
            "mean_amp": float(signal.mean()),
            "rms_amp": float(np.sqrt(np.mean(signal ** 2))),
            "peak_amp": float(np.max(np.abs(signal))),
        }
        window_rows = [
            {**key, "window_id": i, "window_rms": float(np.sqrt(np.mean(w ** 2)))}
            for i, w in enumerate(np.array_split(signal, 4))
        ]
        return (master_row, window_rows)   # tuple -> unpacked into make_insert

    def make_insert(self, key, master_row, window_rows):
        # Phase 3 — atomic write to self + Part, inside the transaction.
        self.insert1(master_row)
        self.Window.insert(window_rows)

Run it¶

populate() orchestrates all three phases per key: it calls make_fetch and make_compute with no transaction open, then opens a transaction, re-runs make_fetch, verifies the inputs are byte-identical, and finally calls make_insert and commits. If an upstream row changed mid-computation, the verification fails with a DataJointError and nothing is inserted.

In [4]:

Copied!

RecordingStats.populate(display_progress=True)
RecordingStats()
RecordingStats.populate(display_progress=True)
RecordingStats()

RecordingStats:   0%|          | 0/3 [00:00<?, ?it/s]

RecordingStats:  67%|██████▋   | 2/3 [00:00<00:00, 18.40it/s]

RecordingStats: 100%|██████████| 3/3 [00:00<00:00, 20.65it/s]

Out[4]:

recording_id	n_samples	mean_amp	rms_amp	peak_amp
0	5000	-0.004532247621656686	0.9954048273054507	3.899421730054339
1	5000	0.03431204343517783	2.001563146013397	6.963674475871197
2	5000	0.029421682585778063	3.0056591656476797	12.069475942671023

Total: 3

In [5]:

Copied!

RecordingStats.Window & "recording_id = 0"
RecordingStats.Window & "recording_id = 0"

Out[5]:

recording_id	window_id	window_rms
0	0	0.9817028961918416
0	1	1.013580648563692
0	2	0.9919901565252676
0	3	0.9940786199867229

Total: 4

Testing fetch and compute directly¶

Because make_fetch performs no writes and make_compute touches no database, you can call them directly to inspect intermediate results — something you cannot do with a standard make(). The insert step is exercised only through populate().

The Three-Part make() — Long Computations Without Long Transactions¶

An upstream source table¶

The computed table, in three phases¶

Run it¶

Testing fetch and compute directly¶

The Three-Part `make()` — Long Computations Without Long Transactions¶