Forward data source level lineage for a given datasource

[vishalag001]

Background

Currently in spline there is no way we can explore the forward lineage of a given datasource. Say: what datasets are getting impacted by a single given dataset ? Eg: Dataset A impacts Dataset B & C and Dataset B impacts Dataset D and E. So the forward lineage looks like:
A(Dataset)->someExecutionPlan->B,C(Dataset)->someExecutionPlan->D,E(Dataset)

Question

@wajda
Is it possible to get the lineage using the current Spline data model entities ?
The current model has an edge collection affects and depends which depicts it, but it points to the dataSource collection in its _to filed.

One possible solution

• Create a new collection "impacts" from depends with _from and _to reversed.
• Create a named graph "ForwardLineage" in arango using "impacts" and "affects" edge collection.
• Use a AQL to get the lineage for a given dataSource

FOR a,v,p IN 1..6 OUTBOUND 'dataSource/b6f81921-04c2-4142-b442-ba76ae57d410'
GRAPH 'ForwardLineage'
OPTIONS  {bfs:true}
return p

• This gives forward lineage upto 6 levels.
• Drawback: One duplicate collection "impacts" similar to "affects" is used.

Is the above solution correct or the same can be achieved using the existing data model ?
@pratapmmmec

[wajda]

It is definitely possible to get a forward lineage (or impact) in Spline data model, the feature has just not been exposed through the Consumer API yet. As you correctly noticed the edge affects (which in a certain sense is opposite to depends) can be used, so no additional edge is required.

Basically what you need is to flip the logic of the existing lineage overview query, and traverse the graph forward in time.
Take a look at the Foxx service - https://github.com/AbsaOSS/spline/tree/release/0.7.7/persistence/src/main/resources/foxx/spline

The current implementation traverses the graph along the depends edge starting from a write event to the data sources being read, and from there to all preceding writes visible from the respective of every selected read, and so on (see https://github.com/AbsaOSS/spline/blob/release/0.7.7/persistence/src/main/resources/foxx/spline/services/observed-writes-by-read.js).

Try to switch the edge from depends to affects, and instead of finding visible writes for a read, look for reads for which a given write is visible (meaning all read events connected to the same source that happened after the given write, but before the subsequent overwrite).

[vishalag001]

The AQL to get the forward lineage shall be:

WITH progress, progressOf, executionPlan,affects, depends,dataSource
LET readTime = ${readEvent}.timestamp
FOR wrt IN 2 OUTBOUND ${readEvent} progressOf, affects
	LET allObservedEvents = SLICE(
        (FOR e IN 2 INBOUND wrt depends,progressOf
            FILTER e.timestamp > readTime
            	RETURN e
        ), 0, 1)
    FOR e IN allObservedEvents RETURN e

It traverses the graph along the affects edge starting from the read event of the datasource being written, and from there to all the succeding reads visible from the respective of every selected write.
P.S. - I cannot use the operation collections here because there are a series of preceding operations before it is read by the executionPlan.

On a similar note
The current implementation of the backward lineage makes use of the operation collections. The same lineage can be achieved if we use the affects and depends edges.
AQL for backward lineage:

WITH progress, progressOf, executionPlan, executes, depends, dataSource
LET readTime = ${readEvent}.timestamp
FOR rds IN 2 OUTBOUND ${readEvent} progressOf, depends
    LET allObservedEvents = SLICE(
    	(FOR e IN 1 INBOUND rds affects,progressOf
    		FILTER e.timestamp < readTime
            SORT e.timestamp DESC
            LIMIT 1
            RETURN e
        ), 0, 1)
    FOR e IN allObservedEvents RETURN e

Question

@wajda
Is the above approach and AQL correct for the forward lineage and can the same be applicable for backward lineage (current implementation) ?

[wajda]

Not sure I completely understand your explanation, but the code seems wrong. You basically just pick one random successive read that might not even be related to your write data-wise because your query doesn't take overwrites into account.

[wajda]

Please look at the original query more carefully.

[vishalag001]

If I differentiate between append and overwrite, then I'll have to traverse the graph from the operation collection. A read event after a write event will have a series of operations which are not directly related.
Example: Write Dataset A -> Read Event B(Dataset A).

Now this event can execute an operation which is a projection and then transformation and finally a read.
Event B -> (executes)Operation (projection)->(follows)Operation(transformation)->(follows)Operation(read)->(readsFrom)DataSource(A)
As per the this, the read operation is not directly linked to executionPlan. There are a series of operation in between.
So this gives me the lineage, but the traversal can be long.
However if I just use the affects and depends edges, I could get the forward lineage at a dataSource level (bypassing the operation links):

WITH progress, progressOf, executionPlan,affects, depends,dataSource
LET writeTime = ${writeEvent}.timestamp
FOR wrt IN 2 OUTBOUND ${writeEvent} progressOf, affects
	LET allObservedEvents = SLICE(
        (FOR e IN 2 INBOUND wrt depends,progressOf
            FILTER e.timestamp > writeTime
            	RETURN e
        ), 0, 1)
    FOR e IN allObservedEvents RETURN e

Starting from a writeEvent, I'll traverse the reads which happened after the write event.

@wajda
Does this approach makes sense ?

[wajda]

It does make sense and this is exactly what Spline query does. But your code does something completely different from what you explained.

[vishalag001]

I created a new foxx service which gives the forward lineage for a given write event. I was able to visualise the forward lineage on the lineage overview page.

Attaching the screenshot of the same.

As per the forward lineage, the selected write event is Emp_Sal_Dept_Multiplier which lists Dept_Emp_Sal and Dept_Sal_2 as its succeeding lineage.

The query I used is as follows:

function observedReadsByWrite(writeEvent) {
    return writeEvent && db._query(aql`
        WITH progress, progressOf, executionPlan,affects, depends,dataSource
        LET writeTime = ${writeEvent}.timestamp
        FOR rds IN 2 OUTBOUND ${writeEvent} progressOf, affects
        	LET allObservedEvents = 
                (FOR e IN 2 INBOUND rds depends,progressOf
                    FILTER e.timestamp > writeTime
                    	RETURN e
                )
            FOR e IN allObservedEvents RETURN e
    `).toArray();
}

@wajda Is the above query correct ? Please suggest any changes if required. If required, we can connect over a zoom/team call.
Can this be added as a feature request ? I can contribute to this feature.

[wajda]

Your query returns all subsequent reads from the source, but you only need ones that happen between this write and the subsequent overwrite (potentially by any other job).

[mariobarbosa10]

Hi guys @wajda @vishalag001, Can you explain to me please, where should I put this function code and all the steps necessary to do the configuration? Should it be inside from the observed-writes-by-read.js file? or this should it be a new file called for example observed-read-by-writes.js?

[vishalag001]

Hi @wajda Can I take this as a feature request ? I can contribute to the forward lineage feature. I have created a foxx service which gives the forward lineage for a given write event. You can see the same in the above screenshot attached. Let me know your thoughts. Happy to discuss. Thanks.

[wajda]

Hi Vishal, thank you for offering your contribution.
There is already a ticket for this - #540. Another person has just started working on it, but is currently on vacation. So feel free to send a PR, the responsible person will review it when he's back, and will take it further from that point.

[wajda]

@dk1844 FYI

[vishalag001]

PR - #1092
@wajda