TimelyDataflow / TimelyDataflow/differential-dataflow

Wrong `since` frontier for merged batch?

Open
#247 5 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
Rust
Stars
3k
Forks
211
Avg merge
10h 42m
Merged PRs (30d)
34

Description

https://github.com/TimelyDataflow/differential-dataflow/blob/e50c9ce938c9783d7a5eb7fdf25f047f2c6ed145/src/trace/implementations/ord.rs#L227

Consider the full set of updates of e.g. a simple wordcount dataflow (single-dimensional timestamps):

full = { ("hello", (), t=5, +1) }

Suppose we seal two batches: batch1 contains the only update, batch2 is empty.

batch1 = Description { lower: 0,  upper: 10, since: 0 }
         Content { ("hello", (), t=5, +1) }
batch2 = Description { lower: 10, upper: 20, since: 10 }
         Content { }

Suppose the times inside batch1 are advanced to timestamp 15

batch1 = Description { lower: 0,  upper: 10, since: 15 }
         Content { ("hello", (), t=15, +1) }  // times in this batch has been advanced to t=15
batch2 = Description { lower: 10, upper: 20, since: 10 }
         Content { }

Suppose we merge the two batches. According to the code, the resulting batch would have the following description / content

Description { lower: 0, upper: 20, since: 10 }
Content { ("hello", (), t=15, +1) }

The since frontier is taken to be the minimum of the two since frontiers of the batches, so batch2.since = 10 is taken.

However, the property of the batch does not hold anymore.

Consider for example t=12, for which it should hold as it is larger than the since frontier.
There are no accumulated updates before such time, which is not the case if we consider the
original set of updates (full).

To conclude, I think the since frontier of the merged batch should be the maximum of the two original batches (t=15 in this case). And it's easy to see that the property would hold for the case above.

The intuition that justifies this is that if a batch looses the capability to distinguish since a certain timestamp (as batch1 did) also a resulting merged batch will not have such capability.


On a similar note, I believe there is another issue in the work function that performs the merging.
https://github.com/TimelyDataflow/differential-dataflow/blob/e50c9ce938c9783d7a5eb7fdf25f047f2c6ed145/src/trace/implementations/ord.rs#L293

When a frontier (call it compaction_frontier) is supplied, times are advanced to such frontier and compaction of updates is performed. If the compaction_frontier is larger than the since frontier that was determined in the constructor, then the since frontier should be updated to be the compaction_frontier. That is because again we are not able to fulfill the property above (we are loosing the ability to distinguish since that frontier).

So the following snippet should be updated to

// if we are supplied a frontier, we should compact.
if let Some(frontier) = frontier.as_ref() {
    OrdValBatch::advance_builder_from(&mut self.result, frontier, initial_key_pos);
    if self.description.since().iter().all(|t1| frontier.iter().any(|t2| t1.less_equal(t2))) {
        self.description = Description::new(self.description.lower(), self.description.upper(), frontier);
    }
}

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start with the merge logic in src/trace/implementations/ord.rs at the cited lines, then read the batch property in src/trace/description.rs. Check both the merged-batch since frontier and the work function's compaction-frontier handling against the examples in the issue. Done means the resulting descriptions preserve the stated batch property in both cases.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
data-engineering, distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.