This is definite in the steady-state, but resumption can cause arbitrary orders, depending on the exact moment of resumption.

Noting it here too. This is a bug, there should be no discrepancies in the produced collection no matter how many restart points you insert. We need to store the mz timestamp in the upsert state and use that too when comparing updates

@petrosagg for keys whose partitions are only-increasing, we ARE definite; handling keys whose partitions can go down and handling resumptions requires that we hold the partition of each message in upsert state, which can't be done (currently) unless the partition is part of the output relation, because during hydration we only have the output shard

for keys whose partitions are only-increasing, we ARE definite;

Indeed. I didn't claim there aren't cases we we are definite. We're also definite when we process an empty topic /s But in seriousness, given that there is a way to fix this, we should.

handling resumptions requires that we hold the partition of each message in upsert state,

I don't think so. The timestamp associated with each key in upsert state should be the tuple (mz_timestamp, partition, offset). This is because the (partition, offset) pair is only needed to disambiguate between updates that happen at the same mz timestamp, and so on rehydration we can initialize the upsert state with (resume_ts, 0, 0) without problems because we will only ever process data for timestamps greater than resume_ts.