Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fix ETL Jobs in Open Lineage Connector #5851

Merged
merged 4 commits into from
Nov 3, 2021
Merged

Fix ETL Jobs in Open Lineage Connector #5851

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
347425f
Fix ETL Jobs in Open Lineage Connector
Oct 28, 2021
df009ad
Send empty qualified name and GUID for condensed vertex in OLS
Oct 28, 2021
ed085bb
Merge branch 'master' into fix-etl-jobs-lineage
lcpopa Oct 29, 2021
476fd8d
Merge branch 'master' into fix-etl-jobs-lineage
Nov 3, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions ...ncedaemonconnectors/openlineageconnectors/janusconnector/graph/LineageGraphConnector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -826,10 +826,10 @@ public LineageResponse lineage(Scope scope, String guid, String displayNameMustC
lineageVerticesAndEdges = helper.endToEnd(guid, includeProcesses);
break;
case ULTIMATE_SOURCE:
lineageVerticesAndEdges = helper.ultimateSource(guid, includeProcesses);
lineageVerticesAndEdges = helper.ultimateSource(guid);
break;
case ULTIMATE_DESTINATION:
lineageVerticesAndEdges = helper.ultimateDestination(guid, includeProcesses);
lineageVerticesAndEdges = helper.ultimateDestination(guid);
break;
case VERTICAL:
lineageVerticesAndEdges = helper.verticalLineage(guid);
Expand Down
16 changes: 8 additions & 8 deletions ...monconnectors/openlineageconnectors/janusconnector/graph/LineageGraphConnectorHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.bothE;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.hasLabel;
Expand Down Expand Up @@ -90,6 +89,7 @@
import static org.odpi.openmetadata.openconnectors.governancedaemonconnectors.openlineageconnectors.janusconnector.utils.GraphConstants.EDGE_LABEL_TRANSLATION;
import static org.odpi.openmetadata.openconnectors.governancedaemonconnectors.openlineageconnectors.janusconnector.utils.GraphConstants.NODE_LABEL_CONDENSED;
import static org.odpi.openmetadata.openconnectors.governancedaemonconnectors.openlineageconnectors.janusconnector.utils.GraphConstants.NODE_LABEL_SUB_PROCESS;
import static org.odpi.openmetadata.openconnectors.governancedaemonconnectors.openlineageconnectors.janusconnector.utils.GraphConstants.PROCESS_NODES;
import static org.odpi.openmetadata.openconnectors.governancedaemonconnectors.openlineageconnectors.janusconnector.utils.GraphConstants.PROPERTY_KEY_ADDITIONAL_PROPERTIES;
import static org.odpi.openmetadata.openconnectors.governancedaemonconnectors.openlineageconnectors.janusconnector.utils.GraphConstants.PROPERTY_KEY_DISPLAY_NAME;
import static org.odpi.openmetadata.openconnectors.governancedaemonconnectors.openlineageconnectors.janusconnector.utils.GraphConstants.PROPERTY_KEY_ENTITY_GUID;
Expand Down Expand Up @@ -143,11 +143,10 @@ public LineageGraphConnectorHelper(GraphTraversalSource graphTraversalSource, bo
* directly used
*
* @param guid queried entity
* @param includeProcesses include processes
*
* @return graph in an Open Lineage specific format
*/
public Optional<LineageVerticesAndEdges> ultimateSource(String guid, boolean includeProcesses) {
public Optional<LineageVerticesAndEdges> ultimateSource(String guid) {
Vertex queriedVertex = g.V().has(PROPERTY_KEY_ENTITY_GUID, guid).next();

List<Vertex> sourcesList;
Expand Down Expand Up @@ -201,11 +200,10 @@ private List<Vertex> querySources(String guid, String edgeLabel) {
* directly used
*
* @param guid queried entity
* @param includeProcesses include processes
*
* @return graph in an Open Lineage specific format
*/
public Optional<LineageVerticesAndEdges> ultimateDestination(String guid, boolean includeProcesses) {
public Optional<LineageVerticesAndEdges> ultimateDestination(String guid) {
Vertex queriedVertex = g.V().has(PROPERTY_KEY_ENTITY_GUID, guid).next();

List<Vertex> destinationsList;
Expand Down Expand Up @@ -507,6 +505,8 @@ private Set<LineageEdge> getCondensedLineageEdges(Set<LineageVertex> lineageVert
private LineageVertex getCondensedVertex(String condensationType) {
LineageVertex condensedVertex = new LineageVertex(getCondensedNodeId(condensationType), NODE_LABEL_CONDENSED);
condensedVertex.setDisplayName(CONDENSED_NODE_DISPLAY_NAME);
condensedVertex.setQualifiedName("");
condensedVertex.setGuid("");
return condensedVertex;
}

Expand Down Expand Up @@ -679,7 +679,7 @@ private Set<LineageVertex> getLineageVertices(List<Vertex> vertexList) {
private void condenseProcesses(boolean includeProcesses, Set<LineageVertex> lineageVertices, Set<LineageEdge> lineageEdges) {
if (!includeProcesses) {
Set<LineageVertex> verticesToRemove = lineageVertices.stream()
.filter(this::isSubProcess)
.filter(this::isProcessOrSubprocessNode)
.collect(Collectors.toSet());
Set<String> verticesToRemoveIDs = verticesToRemove.stream().map(LineageVertex::getNodeID).collect(Collectors.toSet());
Set<LineageEdge> edgesToRemove = lineageEdges.stream().filter(edge ->
Expand Down Expand Up @@ -708,8 +708,8 @@ private List<LineageEdge> createEdgesThatReplaceProcesses(Set<LineageEdge> linea
return edgesToReplaceProcesses;
}

private boolean isSubProcess(LineageVertex vertex) {
return vertex.getNodeType().equalsIgnoreCase(NODE_LABEL_SUB_PROCESS);
private boolean isProcessOrSubprocessNode(LineageVertex vertex) {
return PROCESS_NODES.stream().anyMatch(vertex.getNodeType()::equalsIgnoreCase);
}

private boolean isInVertexesToRemove(Set<String> verticesToRemoveNames, LineageEdge edge) {
Expand Down
4 changes: 3 additions & 1 deletion ...governancedaemonconnectors/openlineageconnectors/janusconnector/utils/GraphConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,10 @@
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;

import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;

public class GraphConstants {
Expand Down Expand Up @@ -59,10 +61,10 @@ private GraphConstants() {
public static final String SOURCE_CONDENSATION = "source";
public static final String DESTINATION_CONDENSATION = "destination";

public static final String NODE_LABEL_GLOSSARYTERM = "glossaryTerm";
public static final String NODE_LABEL_CONDENSED = "condensedNode";
public static final String NODE_LABEL_PROCESS = "process";
public static final String NODE_LABEL_SUB_PROCESS = "subProcess";
public static final List<String> PROCESS_NODES = Arrays.asList(NODE_LABEL_PROCESS, NODE_LABEL_SUB_PROCESS);

// GLOSSARY TERM RELATED EDGE LABELS
public static final String EDGE_LABEL_SEMANTIC_ASSIGNMENT = "SemanticAssignment";
Expand Down
12 changes: 6 additions & 6 deletions ...rs/openlineageconnectors/janusconnector/lineagegraph/LineageGraphConnectorHelperTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ void ultimateSourceColumnLevel() {
expectedNodeIDs.add(queriedNodeID);
expectedNodeIDs.add(PROPERTY_VALUE_NODE_ID_CONDENSED_SOURCE);

LineageVerticesAndEdges lineageVerticesAndEdges = lineageGraphConnectorHelper.ultimateSource(queriedNodeID, true).get();
LineageVerticesAndEdges lineageVerticesAndEdges = lineageGraphConnectorHelper.ultimateSource(queriedNodeID).get();
Set<LineageVertex> lineageVertices = lineageVerticesAndEdges.getLineageVertices();
validateResponse(expectedNodeIDs, lineageVertices);
}
Expand All @@ -75,7 +75,7 @@ void ultimateDestinationColumnLevel() {
expectedNodeIDs.add(queriedNodeID);
expectedNodeIDs.add(PROPERTY_VALUE_NODE_ID_CONDENSED_DESTINATION);

LineageVerticesAndEdges lineageVerticesAndEdges = lineageGraphConnectorHelper.ultimateDestination(queriedNodeID, true).get();
LineageVerticesAndEdges lineageVerticesAndEdges = lineageGraphConnectorHelper.ultimateDestination(queriedNodeID).get();
Set<LineageVertex> lineageVertices = lineageVerticesAndEdges.getLineageVertices();

validateResponse(expectedNodeIDs, lineageVertices);
Expand Down Expand Up @@ -130,7 +130,7 @@ void ultimateSourceTableLevel() {
expectedNodeIDs.add(PROPERTY_VALUE_NODE_ID_CONDENSED_SOURCE);
expectedNodeIDs.add(queriedNodeID);

LineageVerticesAndEdges lineageVerticesAndEdges = lineageGraphConnectorHelper.ultimateSource(queriedNodeID, true).get();
LineageVerticesAndEdges lineageVerticesAndEdges = lineageGraphConnectorHelper.ultimateSource(queriedNodeID).get();
Set<LineageVertex> lineageVertices = lineageVerticesAndEdges.getLineageVertices();
validateResponse(expectedNodeIDs, lineageVertices);
}
Expand All @@ -143,7 +143,7 @@ void ultimateDestinationTableLevel() {
expectedNodeIDs.add(PROPERTY_VALUE_NODE_ID_CONDENSED_DESTINATION);
expectedNodeIDs.add(queriedNodeID);

LineageVerticesAndEdges lineageVerticesAndEdges = lineageGraphConnectorHelper.ultimateDestination(queriedNodeID, true).get();
LineageVerticesAndEdges lineageVerticesAndEdges = lineageGraphConnectorHelper.ultimateDestination(queriedNodeID).get();
Set<LineageVertex> lineageVertices = lineageVerticesAndEdges.getLineageVertices();

validateResponse(expectedNodeIDs, lineageVertices);
Expand Down Expand Up @@ -173,7 +173,7 @@ void ultimateSourceTableLevelViaLineageMapping() {
expectedNodeIDs.add(PROPERTY_VALUE_NODE_ID_CONDENSED_SOURCE);
expectedNodeIDs.add(queriedNodeID);

LineageVerticesAndEdges lineageVerticesAndEdges = lineageGraphConnectorHelper.ultimateSource(queriedNodeID, true).get();
LineageVerticesAndEdges lineageVerticesAndEdges = lineageGraphConnectorHelper.ultimateSource(queriedNodeID).get();
Set<LineageVertex> lineageVertices = lineageVerticesAndEdges.getLineageVertices();
validateResponse(expectedNodeIDs, lineageVertices);
}
Expand All @@ -186,7 +186,7 @@ void ultimateDestinationTableLevelViaLineageMapping() {
expectedNodeIDs.add("t20");
expectedNodeIDs.add(queriedNodeID);

LineageVerticesAndEdges lineageVerticesAndEdges = lineageGraphConnectorHelper.ultimateDestination(queriedNodeID, true).get();
LineageVerticesAndEdges lineageVerticesAndEdges = lineageGraphConnectorHelper.ultimateDestination(queriedNodeID).get();
Set<LineageVertex> lineageVertices = lineageVerticesAndEdges.getLineageVertices();

validateResponse(expectedNodeIDs, lineageVertices);
Expand Down