[FEATURE] Skip reprocessing of entire rows based on finer-grain lineage information

[georgeh0]

Currently, when source row + the flow logic doesn't change from last time, we skip reprocessing.

The flow logic is determined based on the entire flow spec. This is too coarse grain. Not all changes should affect (e.g. exclude_pattern changes from sources, intermediate field ordering / name changes). We want to detect this with finer granularity based on lineage information.

---

❤️ Contributors, please refer to 📙Contributing Guide.
Unless the PR can be sent immediately (e.g. just a few lines of code), we recommend you to leave a comment on the issue like I'm working on it or Can I work on this issue? to avoid duplicating work. Our Discord server is always open and friendly.

[Davda-James]

@georgeh0 was working on this, can you assign it to me, so can raise pr for review ?

[badmonster0]

@Davda-James assigned thanks a lot for taking a look!

[georgeh0]

Want to share more context and thought on this.

The goal for this issue is, we have a new approach compute a fingerprint to replace the existing logic_fingerprint. With this, when both the input data and this new logic_fingerprint have no change, we can safely skip reprocessing.

This is the approach in my mind to compute the new logic_fingerprint:

• For each op, there's a behavior fingerprint
  • For sources, use the default fingerprint as its behavior fingerprint by default (i.e. spec has no influence on its fingerprint, as source spec mainly specify where to find the source data, and the flow output can be solely determined by "source value + logic fingerprint post source value", so usually source spec doesn't matter for this purpose)
  • For functions, use Fingerprint(spec) as its behavior fingerprint by default
  • For targets, use Fingerprint(spec) as its behavior fingerprint by default
  • It's possible to override the way to compute behavior fingerprint on these factories (e.g. they can exclude certain spec fields for functions / targets, and include certain fields for source)
• For each field in the schema, there's a lineage fingerprint. Once the fingerprint hasn't change, it means the field value will not change if the source data has no change.
  • For a field directly imported from the source, we simply use the source's behavior fingerprint.
  • For a field as output of a function, we use Fingerprint(behavior fingerprint, behavior version, lineage fingerprint of input fields). For input fields, put positional arguments in order, then keyword arguments in alphabetical order (together with their argument name)
    • An input field may be a sub portion of another field (e.g. in a flow, the output of SplitRecursively function may be put in a field called chunks, and chunks[*].text is used as input for the EmbedText function). In this case, we can build lineage fingerprint for the sub field by approach such as Fingerprint(fingerprint of chunks, "row", "field:text").
    • For a collector, compute lineage fingerprint of input fields, similar to functions.
    • For an export op, we use Fingerprint(behavior fingerprint, lineage fingerprint of its collector) as its fingerprint .
• We use fingerprint of all export op to build logic_fingerprint of the entire flow.