DC picks up checkpoint only on 3rd run, instead of 2nd

[dmpetrov]

It doesn't pick up checkpoint in 2nd run.

import os
import datachain as dc

first_run = os.environ.get("FIRST_RUN", "0") == "1"

def score_number(num: int) -> float:
    if first_run:
        if num > 5:
            raise ValueError("test error")
    print("processing", num)
    return float(num**2) / 1000.0

if first_run:
    (
        dc.read_values(num=list(range(1, 10)))
        .save("filtered_nums")
    )

result = (
    # dc.read_values(num=list(range(1, 10)))
    dc.read_dataset("filtered_nums")
    .settings(batch_size=1)          # process one row at a time for fine-grained checkpointing
    .map(score=score_number)
    .save("scored_nums")
)

Output:

$ FIRST_RUN=1 python process_data.py
processing 1
processing 2
processing 3
processing 4
processing 5
Traceback (most recent call last):
   [stack trace]

$ python process_data.py
processing 1  # this should not be here
processing 2 # this should not be here
processing 3 # this should not be here
processing 4 # this should not be here
processing 5 # this should not be here
processing 6
processing 7
processing 8
processing 9

PS: Funny part - it DOES pick it up in 3rd run if you still fail in the 2nd run.

[dmpetrov]

Also, it looks like a checkpoint will be always recomputed after dc.read_values(num=list(range(1, 10))).
Is there a way to not to recompute it?

[ilongin]

The checkpoint hash is a cumulative chain — each step's hash depends on all previous steps. On the first run the pipeline is read_values → save → read_dataset → map, but on the second run it's just read_dataset → map (since data is already saved). The completely different chain produces different hashes, so run 2 can't find run 1's checkpoints. Run 3 matches run 2's chain, so it works.

Regarding read_values producing a different hash every time — yes, that's a known limitation. The reason it's not implemented yet is that computing a deterministic hash would require hashing all input data, which can be arbitrarily large (e.g. millions of rows). It's doable with streaming hash (SHA256 etc.) but adds latency at chain construction time, and there are issues with deterministic serialization of arbitrary Python objects as well. If you want I can do this or we can put note in documentation about read_values (this is what I should've done before though)

[shcheklein]

@ilongin how about read_csv, read_database, etc?

How about read_storage(update=True)?

does it means checkpoints are essentially disabled for all those ^^?

[dmpetrov]

The checkpoint hash is a cumulative chain — each step's hash depends on all previous steps. On the first run the pipeline is read_values → save → read_dataset → map

I understand why it's happening after filter(), map() etc but read_dataset() should reset that. It's a terminal data source, not a continuation of the previos pipeline.

Once a dataset is materialized via .save(), its identity should be defined by the dataset name@version itself (or an internal, random uuid), now it's whole lineage.

Otherwise we get overly strict checkpoint invalidation which limits their usage without reason.

[dmpetrov]

I played a bit more with a use case: fail --> fix problem and re-read previously created dataset --> rerun. It doesn't work as expected - it recomputes everything from scratch.

So, we're missing the opportunity to reuse run 1st checkpoint even though it could be reused. It has to be implemented.

Another missing opportunity:

1. dc.read_csv("data.csv").filter(dc.C("score") > 0.5).save("filtered_data")
2. filtered.map(value=lambda score: score * 2, output=float).save("transformed_data")

Both of these should reuse their checkpoints if file wasn't changed. It looks like @shcheklein is referencing to this issue.

[dmpetrov]

I'd add that the read_csv() checkpoint is very important. It give me an ability to run the same scrip without commenting out the dataset creation and re-reading it again read_dataset("filtered"). That's basically the core checkpoint use case 🙂 - fix an issue and re-run the script.

With the current approach we are pushing user to modify the code according to "cumulative chain" logic 😅 that user should not even know about.

[ilongin]

@shcheklein Here's how "entrypoints" work with checkpoints:

Entry point	Checkpoints work?	Reason
read_dataset()	✅ Yes	Hash based on dataset name + version (deterministic)
read_storage()	✅ Yes	Reads from listing dataset with stable name/version
read_csv()	✅ Yes and ❌ No	Built on read_storage() if pointing to directory, otherwise using read_records which doesn't work with Checkpoints yet
read_parquet()	✅ Yes	Built on read_storage()
read_json()	✅ Yes	Built on read_storage()
read_storage(update=True)	✅ Yes	For listings we only include listing dataset name
read_values()	❌ No	Creates temp dataset with random name each run
read_records()	❌ No	Same as read_values() — random temp dataset name
read_pandas()	❌ No	Built on read_values()
read_hf()	❌ No	Built on read_values()
read_database()	❌ No	Built on read_records()

The root cause for the ❌ group is read_records() — it creates a temp dataset via session.generate_temp_dataset_name() which includes a random UUID. The starting hash is derived from this random dataset name, so it's different every run. Computing a deterministic hash here is not straightforward — it would require hashing potentially large input data (e.g. millions of rows), with trade-offs around latency and deterministic serialization of arbitrary Python objects. I'll create a separate issue to track this.

For read_storage(update=True) — this should work with checkpoints. The starting hash only uses the listing dataset name (not the version), because listing is lazy — it happens during step calculation, and hash calculation needs to be possible without applying steps (this was one of my "hard" requirements to make hash calculation lightweight and fast). So update=True bumping the version doesn't affect the hash. Though arguably this might be wrong — if a user passes update=True they likely expect new data in the listing, so downstream steps should probably re-run rather than reuse a checkpoint from stale input. WDYT?

There are actually two separate issues here:

1. read_records()-based methods not producing deterministic hashes — the entry points listed above that go through read_records() can't participate in checkpoint reuse. This is a known limitation, not yet implemented due to the trade-offs mentioned above.

2. Starting methods not breaking the checkpoint hash chain — this is what @dmpetrov describes. Currently, if a script does read_values(...).save("A") followed by read_dataset("A").map(...).save("B"), the hash for B depends on the entire chain including how A was created. On the next run, if the script only does read_dataset("A").map(...).save("B"), the hash is different because the chain structure changed.

On (2) — the current behavior is intentionally conservative: the cumulative hash ensures that if someone modifies the code in the first chain (which produces the input to the second chain), the second chain's checkpoint is invalidated and it gets re-computed. If we make read_dataset() break the hash chain (so it only depends on dataset name+version), then changing the first chain's code would still produce the same downstream checkpoint hash — even though the input data changed. The user would likely expect the downstream chain to re-run in that case, but it wouldn't.

LMK if you still think the above is not a concern and I can change the behavior so that read_dataset() breaks the hash chain.

[dmpetrov]

read_values() ❌ No Creates temp dataset with random name each run

Ok. Not a problem for now. Let't forget for now read_values() and assume that the data source is from read_csv("data.csv").

LMK if you still think the above is not a concern and I can change the behavior so that read_dataset() breaks the hash chain.

It's not enough just to break the hash chains. We need to find a way when it work as user would expect re-compute work:

1. It picks up the checkpoint if user just fix a bug in score_number without modifying the script.
2. it DOES recompute everything if data.csv file was changed.

read_csv() ✅ Yes and ❌ No Built on read_storage() if pointing to directory, otherwise using read_records which doesn't work with Checkpoints yet

A better script example:

import os
import datachain as dc

filtered = (
    dc.read_csv("data.csv")
    .filter(dc.C("score") > 0.5)
    .save("filtered_data")
)

print("filtered dataset", filtered.version)

transformed = (
    filtered
    .map(value=lambda score: score * 2, output=float)
    .save("transformed_data")
)

transformed = dc.read_dataset("transformed_data")
print("transformed dataaset", transformed.version)

run=0
#to_fail = os.environ.get("FAIL", "") != ""

def my_func(value) -> float:
    global run
    run += 1
    #if run > 5:
    #    raise ValueError("")
    print("run", run)
    return value*5

result = (
    transformed
    .settings(batch_size=1)
    .map(total=my_func)
    .save("final_results")
)

result.show()