storage: add regression test for upsert sources with keys that move partitions

[guswynn]

Closes #26965

cc @morsapaes

We only want to commit to the fact that if a keys partition only increases, then we do the right thing.

Motivation

• tests/verification

Checklist

• This PR has adequate test coverage / QA involvement has been duly considered. (trigger-ci for additional test/nightly runs)
• This PR has an associated up-to-date design doc, is a design doc (template), or is sufficiently small to not require a design.
• If this PR evolves an existing $T ⇔ Proto$T mapping (possibly in a backwards-incompatible way), then it is tagged with a T-proto label.
• If this PR will require changes to cloud orchestration or tests, there is a companion cloud PR to account for those changes that is tagged with the release-blocker label (example).
• This PR includes the following user-facing behavior changes:

[petrosagg]

We only want to commit to the fact that if a keys partition only increases, then we do the right thing.

I don't think we want to commit to anything (and I don't think we can even if we wanted to).

If the topic looks like this:

part0: (key1, value1)
part1: (key1, value2)

There is no guarantee that after reclocking the stream will look like this:

(key1, t1, value1)
(key1, t1, value2)

And not like this

(key1, t1, value2)
(key1, t2, value1)

In the first case key1 ends up having the value value2 and in the second case value1

[petrosagg]

There is a detail that I want to verify is done correctly. Upsert should compare the tuple (into_time, from_time) of an incoming update with the stored (into_time, from_time) tuple and not just the from_time. Is this what we store and use from the upsert state?

In other words, the behavior you have documented in the test as only happening in a restart must also happen in the case where the two updates from two different partitions happen to be reclocked to different timestamps. In that case the one with the larger mz timestamp must be the one that wins, no matter what its partition is. Is this how it works?

[guswynn]

@petrosagg We currently sort by FromTime only, except during restarts; We could switch to both, that would resolve your point on #26965, while still supporting usecases that have only-increasing partitions for keys, right?

[petrosagg]

that would resolve your point on #26965,

yes

while still supporting usecases that have only-increasing partitions for keys, right?

If "supporting" means that when a key moves partitions we will always keep the value of the highest partition then no, we can't support that because we can't guarantee that the two updates will land in the same mz timestamp. It's only when they land in the same timestamp that we can promise to keep the highest partition, otherwise the one with the highest mztimestamp will win, and that might be the lower partition.

[petrosagg]

Our correctness promise is correctness property #2, i.e that from_time1 <= from_time2 => mz_time1 <= mz_time2. Two kafka messages from two different partitions are not comparable so there is no guarantee we can offer