I will mark this as draft to add a few more tests.

@mihaibudiu will this happen through the graph or just close to input sources?

@mihaibudiu , from the description, this only partially fixes #5815. We'd like index reuse to apply to other operators and to PK indexes as well. These could be separate future PRs, but we shouldn't close #5815 after this PR.

Before merging this, we should check that this generates expected DBSP dataflows with shared integrals.

inputs do not appear in this pattern

@ryzhyk does the above profile fragment prove that indexes are shared?

@ryzhyk does the above profile fragment prove that indexes are shared?

This fragment only contains one join, so no.

Here is the full graph:

This is the source SQL

CREATE LOCAL VIEW V1 AS SELECT
                    o.order_id,
                    o.amount,
                    c.name
                FROM orders1 AS o
                LEFT JOIN customers AS c
                    ON o.customer_id = c.customer_id;
                
                CREATE LOCAL VIEW V2 AS SELECT
                    o.order_id,
                    o.amount,
                    c.first
                FROM orders2 AS o
                LEFT JOIN customers AS c
                    ON o.customer_id = c.customer_id;
                
                CREATE VIEW V AS (SELECT * FROM V1) UNION ALL (SELECT * FROM V2);

Here is the full graph:

yep, this one shows reuse!

This PR won't be the last word even for joins. Today the optimization runs early in the pipeline, but after other optimizations are executed, more opportunities for sharing may surface, which the code here does not support. For example, the inputs could be map_index_flatmap operators. So we will need to revisit this.

I think this will need to be rebased on #5752

@mihaibudiu , how does this handle GC? If one or both joins consuming the stream are able to GC it, the effective waterline should be the greatest lower bound of the two.

This is done before GC. It may inhibit GC, I haven't checked. But none of our tests was affected so far.

Or, more likely, GC may undo this optimization

Or, more likely, GC may undo this optimization

We really need a test for this. Undoing the optimization is fine in this case, but we have to make sure it doesn't do unsound stuff like apply one or both of the individual bounds. Likewise, disabling GC altogether in this case would also be a serious issue.

Indeed, when using LATENESS what happens is that first the index is shared, then later it is unshared by inserting a noop, and LATENESS is applied separately after each index operator.

We don't have an optimization to compute the min of two waterlines, that's TBD. But the result is correct.

After sharing indexes:

Before lateness computation starts: notice the noop nodes inserted

After lateness inserts GC operators:

Final graph:

Without lateness the first graph is essentially the output graph as well, with shared indexes.