checkpoints: remove transient dependency between chains

[ilongin]

Currently each chain's checkpoint hash includes the hash of the last checkpoint in the current job. This means if a script has two independent chains:

dc.read_storage("s3://a/").filter(...).save("ds1")
dc.read_storage("s3://b/").map(...).save("ds2")

ds2's hash depends on ds1's checkpoint. Changing ds1's logic, reordering the chains, or wrapping one in an if invalidates ds2's checkpoint even though its own inputs and steps haven't changed.

This transient dependency was meant to handle cases like read_dataset("d1") depending on a previously saved d1, but that dependency is already captured naturally — read_dataset("d1") includes the dataset name/version in its QueryStep.hash().

Proposal: remove _last_checkpoint_hash from the hash calculation. Each chain's hash should depend solely on its own starting data and steps.

Benefits:

• Easier to reason about — no surprising invalidations from unrelated chains
• Robust to code reordering, conditionals, and script restructuring
• Enables global checkpoints — if the same chain with the same inputs ran in another job months ago, the checkpoint can be reused

[ilongin]

@shcheklein @dmpetrov This is not as easy task as it should be. Here are the issues:

Removing transient dependencies means two chains with identical flow in the same job produce the same UDF hash. This causes collisions in checkpoints and output tables.

Example:

try:
    dc.read_storage("s3://a/").map(udf).save("ds1")  # crashes mid-UDF
except:
    pass
dc.read_storage("s3://a/").map(udf).save("ds2")  # crashes mid-UDF

Without transient hash calculation there is no way to establish identity between chains across runs. Transient hashing acted as an implicit ordering key — without it, if a user reorders, adds, or removes chains between runs, there is no mechanism to match a chain in the current run to its counterpart in the previous run. This makes partial checkpoint recovery unreliable.

Previously this was impossible because each chain's hash included the previous checkpoint's hash, making every chain position-aware and uniquely identifiable within a job.

A possible solution is adding a per-job chain sequence number to the hash. This is a weaker dependency than transient hashes — with transient deps, modifying chain 1's logic (different filter, different UDF) changes chain 1's checkpoint hash which changes chain 2's hash, invalidating chain 2 even though it's untouched. With a sequence number, chain 2 is always "#2" regardless of chain 1's content — only inserting, removing, or reordering chains before it would change its hash. So you depend on position, not on the content of previous chains.

However, this still reintroduces ordering dependency and doesn't solve original "if" issue in #1629 — reordering or adding chains changes sequence numbers, which invalidates checkpoints. The fundamental issue remains: you need ordering to distinguish identical chains, but ordering makes chains dependent on each other. This is one of the reasons I introduced transient checkpoint calculation in the first place btw (not the only one though).

What are your ideas to fix this? ATM I don't really have a nice solution except this ordering number in hash.

[shcheklein]

Good question.

If it is exactly the same code, should it be just reusing the checkpoint created in the first loop? I think this is fine, no?

Users might even want this logic - run the same code in the loop - if it fils it picks it up from the checkpoint, until it is done. WDYT?

[ilongin]

Good question.

If it is exactly the same code, should it be just reusing the checkpoint created in the first loop? I think this is fine, no?

Users might even want this logic - run the same code in the loop - if it fils it picks it up from the checkpoint, until it is done. WDYT?

I think "just reuse" approach has issues even for normal save checkpoints. Consider:

for i in range(3):                                                                                               
    dc.read_storage("s3://a/").filter(...).save("my_dataset")                                              

Same chain, same save name → same hash. Without transient deps, second and third calls would reuse the checkpoint from the first call. But user explicitly wants 3 versions of my_dataset. I think if user explicitly calls .save(), we should create the dataset. The idea of checkpoints is to skip what was created in a previous run, not to deduplicate within the same run.

But there is an even bigger issue with UDF checkpoints. For UDF partial recovery, we need to find the exact chain from the previous run, not just any chain with a matching hash. We have two types of UDF checkpoint hashes: a "full" hash that includes the UDF function code (used to skip completed UDFs), and a "partial" hash that excludes the function code (used to continue from partial results). The partial hash only includes the chain flow + output schema — this way if a UDF crashes, the user can fix the bug and continue from where it stopped.

But because the partial hash doesn't include the function code, two chains with the same flow and output schema but completely different UDF implementations share the same partial hash. Without some ordering mechanism, we could pick up partial results from a chain that ran a different function entirely.

[shcheklein]

But user explicitly wants 3 versions of my_dataset.

this is not obvious - same inputs, same code -> same output (reused). I think that make sense. E.g. if you'd like to retry exactly same chain 3 times if it fails in the middle.

Without some ordering mechanism, we could pick up partial results from a chain that ran a different function entirely.

but relying on other chains and order is also seems to be too complicated, causes surprises, fragile on its own way (in real life any small change at the top would lead to a complete invalidation) 🤔

I think what we need is still have it isolated to a single chain, but also find some behavior with least surprises.

1. If we call function by their names in mappers, etc - then we can distinguish by it, right?
2. If we use lambdas - may be then any change invalidates it (again, no surprises + clear way how to avoid it)

any other cases?

[ilongin]

I think what we need is still have it isolated to a single chain, but also find some behavior with least surprises.

1. If we call function by their names in mappers, etc - then we can distinguish by it, right?
2. If we use lambdas - may be then any change invalidates it (again, no surprises + clear way how to avoid it)

any other cases?

I think your idea with including function name is the right way, yes. We could say that if any 2 chains have same steps before UDF and then same function name in UDF that this means they can safely reuse partial UDF results. If that's in our "contract", and I think it's reasonable, then this will work fine.

Added PR #1653