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

feat(patch): refactor cll patch #9922

Merged
merged 5 commits into from
Feb 27, 2024
Merged

feat(patch): refactor cll patch #9922

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
236f07d
feat(patch): refactor cll patch
RyanHolstien Feb 26, 2024
86bc906
fixup
RyanHolstien Feb 26, 2024
cef1e2a
Merge branch 'master' into refactor-cll-patch + add quoting fixes
hsheth2 Feb 27, 2024
a33c2dd
fix
hsheth2 Feb 27, 2024
8e684a7
use update golden mechanism
hsheth2 Feb 27, 2024
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
188 changes: 33 additions & 155 deletions ...src/main/java/com/linkedin/metadata/aspect/patch/builder/UpstreamLineagePatchBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,12 +9,11 @@
import com.linkedin.common.urn.DatasetUrn;
import com.linkedin.common.urn.Urn;
import com.linkedin.dataset.DatasetLineageType;
import com.linkedin.dataset.FineGrainedLineageDownstreamType;
import com.linkedin.dataset.FineGrainedLineageUpstreamType;
import com.linkedin.metadata.aspect.patch.PatchOperationType;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import lombok.ToString;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.ImmutableTriple;

@ToString
Expand Down Expand Up @@ -52,117 +51,49 @@ public UpstreamLineagePatchBuilder removeUpstream(@Nonnull DatasetUrn datasetUrn
return this;
}

/**
* Method for adding an upstream FineGrained Dataset
*
* @param datasetUrn dataset to be set as upstream
* @param confidenceScore optional, confidence score for the lineage edge. Defaults to 1.0 for
* full confidence
* @param transformationOperation string operation type that describes the transformation
* operation happening in the lineage edge
* @return this builder
*/
public UpstreamLineagePatchBuilder addFineGrainedUpstreamDataset(
@Nonnull DatasetUrn datasetUrn,
@Nullable Float confidenceScore,
@Nonnull String transformationOperation) {
Float finalConfidenceScore = getConfidenceScoreOrDefault(confidenceScore);

pathValues.add(
ImmutableTriple.of(
PatchOperationType.ADD.getValue(),
FINE_GRAINED_PATH_START
+ transformationOperation
+ "/"
+ "upstreamType"
+ "/"
+ "DATASET"
+ "/"
+ datasetUrn,
instance.numberNode(finalConfidenceScore)));
return this;
}

/**
* Adds a field as a fine grained upstream
*
* @param schemaFieldUrn a schema field to be marked as upstream, format:
* @param upstreamSchemaField a schema field to be marked as upstream, format:
* urn:li:schemaField(DATASET_URN, COLUMN NAME)
* @param confidenceScore optional, confidence score for the lineage edge. Defaults to 1.0 for
* full confidence
* @param transformationOperation string operation type that describes the transformation
* operation happening in the lineage edge
* @param type the upstream lineage type, either Field or Field Set
* @param downstreamSchemaField the downstream schema field this upstream is derived from, format:
* urn:li:schemaField(DATASET_URN, COLUMN NAME)
* @param queryUrn query urn the relationship is derived from
* @return this builder
*/
public UpstreamLineagePatchBuilder addFineGrainedUpstreamField(
@Nonnull Urn schemaFieldUrn,
@Nonnull Urn upstreamSchemaField,
@Nullable Float confidenceScore,
@Nonnull String transformationOperation,
@Nullable FineGrainedLineageUpstreamType type) {
@Nonnull Urn downstreamSchemaField,
@Nullable Urn queryUrn) {
Float finalConfidenceScore = getConfidenceScoreOrDefault(confidenceScore);
String finalType;
if (type == null) {
// Default to set of fields if not explicitly a single field
finalType = FineGrainedLineageUpstreamType.FIELD_SET.toString();
String finalQueryUrn;
if (queryUrn == null || StringUtils.isBlank(queryUrn.toString())) {
finalQueryUrn = "NONE";
} else {
finalType = type.toString();
finalQueryUrn = queryUrn.toString();
}

ObjectNode fineGrainedLineageNode = instance.objectNode();
fineGrainedLineageNode.put("confidenceScore", instance.numberNode(finalConfidenceScore));
pathValues.add(
ImmutableTriple.of(
PatchOperationType.ADD.getValue(),
FINE_GRAINED_PATH_START
+ transformationOperation
+ "/"
+ "upstreamType"
+ downstreamSchemaField
+ "/"
+ finalType
+ finalQueryUrn
+ "/"
+ schemaFieldUrn,
instance.numberNode(finalConfidenceScore)));

return this;
}

/**
* Adds a field as a fine grained downstream
*
* @param schemaFieldUrn a schema field to be marked as downstream, format:
* urn:li:schemaField(DATASET_URN, COLUMN NAME)
* @param confidenceScore optional, confidence score for the lineage edge. Defaults to 1.0 for
* full confidence
* @param transformationOperation string operation type that describes the transformation
* operation happening in the lineage edge
* @param type the downstream lineage type, either Field or Field Set
* @return this builder
*/
public UpstreamLineagePatchBuilder addFineGrainedDownstreamField(
@Nonnull Urn schemaFieldUrn,
@Nullable Float confidenceScore,
@Nonnull String transformationOperation,
@Nullable FineGrainedLineageDownstreamType type) {
Float finalConfidenceScore = getConfidenceScoreOrDefault(confidenceScore);
String finalType;
if (type == null) {
// Default to set of fields if not explicitly a single field
finalType = FineGrainedLineageDownstreamType.FIELD_SET.toString();
} else {
finalType = type.toString();
}
+ upstreamSchemaField,
fineGrainedLineageNode));

pathValues.add(
ImmutableTriple.of(
PatchOperationType.ADD.getValue(),
FINE_GRAINED_PATH_START
+ transformationOperation
+ "/"
+ "downstreamType"
+ "/"
+ finalType
+ "/"
+ schemaFieldUrn,
instance.numberNode(finalConfidenceScore)));
return this;
}

Expand All @@ -180,93 +111,40 @@ private Float getConfidenceScoreOrDefault(@Nullable Float confidenceScore) {
/**
* Removes a field as a fine grained upstream
*
* @param schemaFieldUrn a schema field to be marked as upstream, format:
* @param upstreamSchemaFieldUrn a schema field to be marked as upstream, format:
* urn:li:schemaField(DATASET_URN, COLUMN NAME)
* @param transformationOperation string operation type that describes the transformation
* operation happening in the lineage edge
* @param type the upstream lineage type, either Field or Field Set
* @param downstreamSchemaField the downstream schema field this upstream is derived from, format:
* urn:li:schemaField(DATASET_URN, COLUMN NAME)
* @param queryUrn query urn the relationship is derived from
* @return this builder
*/
public UpstreamLineagePatchBuilder removeFineGrainedUpstreamField(
@Nonnull Urn schemaFieldUrn,
@Nonnull Urn upstreamSchemaFieldUrn,
@Nonnull String transformationOperation,
@Nullable FineGrainedLineageUpstreamType type) {
String finalType;
if (type == null) {
// Default to set of fields if not explicitly a single field
finalType = FineGrainedLineageUpstreamType.FIELD_SET.toString();
} else {
finalType = type.toString();
}

pathValues.add(
ImmutableTriple.of(
PatchOperationType.REMOVE.getValue(),
FINE_GRAINED_PATH_START
+ transformationOperation
+ "/"
+ "upstreamType"
+ "/"
+ finalType
+ "/"
+ schemaFieldUrn,
null));

return this;
}

public UpstreamLineagePatchBuilder removeFineGrainedUpstreamDataset(
@Nonnull DatasetUrn datasetUrn, @Nonnull String transformationOperation) {
@Nonnull Urn downstreamSchemaField,
@Nullable Urn queryUrn) {

pathValues.add(
ImmutableTriple.of(
PatchOperationType.REMOVE.getValue(),
FINE_GRAINED_PATH_START
+ transformationOperation
+ "/"
+ "upstreamType"
+ "/"
+ "DATASET"
+ "/"
+ datasetUrn,
null));
return this;
}

/**
* Adds a field as a fine grained downstream
*
* @param schemaFieldUrn a schema field to be marked as downstream, format:
* urn:li:schemaField(DATASET_URN, COLUMN NAME)
* @param transformationOperation string operation type that describes the transformation
* operation happening in the lineage edge
* @param type the downstream lineage type, either Field or Field Set
* @return this builder
*/
public UpstreamLineagePatchBuilder removeFineGrainedDownstreamField(
@Nonnull Urn schemaFieldUrn,
@Nonnull String transformationOperation,
@Nullable FineGrainedLineageDownstreamType type) {
String finalType;
if (type == null) {
// Default to set of fields if not explicitly a single field
finalType = FineGrainedLineageDownstreamType.FIELD_SET.toString();
String finalQueryUrn;
if (queryUrn == null || StringUtils.isBlank(queryUrn.toString())) {
finalQueryUrn = "NONE";
} else {
finalType = type.toString();
finalQueryUrn = queryUrn.toString();
}

pathValues.add(
ImmutableTriple.of(
PatchOperationType.REMOVE.getValue(),
FINE_GRAINED_PATH_START
+ transformationOperation
+ "/"
+ "downstreamType"
+ downstreamSchemaField
+ "/"
+ finalType
+ finalQueryUrn
+ "/"
+ schemaFieldUrn,
+ upstreamSchemaFieldUrn,
null));

return this;
}

Expand Down
Loading
Loading