Add names to joins

[HarukiMoriarty]

Adds caller-provided names to the core free join operators.

What I changed:

• join_traces and join_with_tactic now take name: &str.
• Existing inherent convenience methods retain their current signatures and forward the existing "Join" default.
• Updated direct callers to preserve current names.

This lets FlowLog build its own named, rule-level wrapper around the core join operators without expanding Differential Dataflow’s inherent API. #770

[frankmcsherry]

This is great in principle, but I think I probably need to get some thoughts together on how to minimize the number of methods floating around. You should definitely be able to specify a name, but also there isn't really a semver constraint on breaks here, so we can also just have the "root" method accept a new argument and ditch all of the _named variants.

I'm happy to pick this up if you aren't in love with changing the PR around to guess what makes sense to me, but I'm also happy to do some more back and forth. There are other changes coming along, in the "chunk/tactics" flavor, which .. probably won't change your mind here (about names) but may shake up which methods you end up wanting to use (we're moving away from cursors internally, towards operators that have fewer but chonkier boundary crossings).

[HarukiMoriarty]

Thanks, that makes sense! I’m happy to revise this.

I get the general idea, avoid lots of _named methods and put the name on the main APIs, but could you share what API shape you have in mind, especially with the chunk/tactics changes coming? I’d rather follow your direction than guess.

Happy to update the PR, or happy for you to pick it up.

[frankmcsherry]

For example, in reduce.rs, both reduce_trace and reduce_with_tactic both just take a name: &str argument. There's one "core" method to support (well, two I guess; but not with/without names). It should be easy for others to write wrappers that put a name opinion in there (e.g. fn reduce_abelian is defined outside of reduce.rs, and is just an opinion about the reduction logic, when the diff is Abelian).

Don't sweat the tactics versions yet; I think they are already in place for join, and .. flat_map is tbd but in the case of chunks it probably ends up being a simple flat map from batches-as-list-of-chunks to the chunks in the batches.

Mostly, I think it's a matter of avoiding the product of _named and not in the methods, because I'll probably just delete them and rename (keeping the name option, but removing the _named suffix). Any of these are good though; it can land and will evolve before the next release, but the ability to specify a name will remain.

[HarukiMoriarty]

I have a version for #770 following my understanding of your suggestion:

• join_core, flat_map, arrange, arrange_by_key, and arrange_by_self now take name: &str.
• Their _named counterparts are removed.
• Existing callers pass their previous default names.

Does this non-backward-compatible direction match what you had in mind?

[frankmcsherry]

Ah, so skimming this and a few thoughts:

1. Looks a bit noisier than I expected, which is my bad.
2. I think the intended (if not well communicated) intent was that the core operators, the free-standing, non-inherent methods, certainly want to come with names, and everything else (e.g. Collection inherents, Arranged inherents) can have or not have the variants. If these are the entry points, and there need to be named options here, let's discuss.
3. Materialize, and most other times I/we work with DD, I create a new "superficial" layer that provides the inherent methods we want to support (e.g. MZ's arrangement builder needs some accounting that the conventional builder doesn't introduce). This is a great moment to have a point of view on where and whether names are mandatory (e.g. maybe yes for join, maybe not for concat).

A proposal: let's think about adding names to the free functions, leaving the inherent methods unmodified (where acceptable). If it turns out that the inherent methods are the problem, let's discuss that independently. As I suspect you've noticed, all of the inherents are usually 1-2 lines of wrapper around the core free methods, and are/can be super easy to adapt locally. I'm totally happy to pick this up if/when you tire of rewriting things / having to guess about what the next round of feedback might be (the names for free functions is great; everything else is borderline aesthetics for demo code).

Directionally, the "chunk" work seems likely to demote "one DD Collection type", on account of the way you say "filter" to a row-at-a-time collection is different from how you say it to a columnar layout, is different from how you say it to an interpreted substrate (each of which exist now). I expect the Collection types and their opinions about their own ergonomic vocabulary to flap around and probably become "examples" of DD rather than core DD; we'll be using some of them and not others. Whether flat_map takes a name is important, but reconciling that with filter and join_function and others may end up devolving to the module that handles vec-based row-at-a-time logic.

I would definitely recommend that for FlowLog (and anything else) that it's worth having your own Collection wrapper or replacement, that gives you the agency to define the set of methods that exist, and to take DD out of the critical path. The logic exists as free functions so that you can do this, except (as you observe) when the core methods don't support taking a name. I guess by that standard, the PR should add the name: &str argument to the core free join methods, but perhaps leave it off of flat_map and other inherent methods, because they can all be added by wrappers.

[HarukiMoriarty]

Ah, I see what you mean now, indeed, this is much clearer.

I’ll keep the name on join_traces / join_with_tactic, leave the inherent convenience methods alone, and have FlowLog provide its own rule-level wrapper where it needs that surface.

[HarukiMoriarty]

e.g. if I understand it correctly, I can customized an operator like this...?

pub fn flowlog_flat_map_named<'scope, T, D, R, I, L>(
      collection: differential_dataflow::Collection<'scope, T, D, R>,
      name: &str,
      mut logic: L,
  ) -> differential_dataflow::Collection<'scope, T, I::Item, R>
  where
      T: timely::progress::Timestamp + Clone + 'static,
      D: differential_dataflow::Data,
      R: differential_dataflow::difference::Semigroup,
      I: IntoIterator<Item: Clone + 'static>,
      L: FnMut(D) -> I + 'static,
  {
      use differential_dataflow::AsCollection;
      use timely::dataflow::channels::pact::Pipeline;
      use timely::dataflow::operators::generic::Operator;

      collection
          .inner
          .unary(Pipeline, name, move |_, _| {
              move |input, output| {
                  input.for_each_time(|time, data| {
                      let data = data.flat_map(|batch| batch.drain(..));
                      let data = data.flat_map(|(datum, time, diff)| {
                          logic(datum)
                              .into_iter()
                              .map(move |output| (output, time.clone(), diff.clone()))
                      });
                      output.session(&time).give_iterator(data);
                  });
              }
          })
          .as_collection()
  }

[frankmcsherry]

Looks great! Your example looks clean, and generally you should be able to copy/paste other operator definitions if they are more to your liking. Happy to merge this, though, and if there are other rough edges we can pick them up then!