Skip to content

fix: prevent duplicate lineage cleanup and remove blocking refresh in PATCH /api/v1/tables/{id} #26702

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

Open
kojikokojiko wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: prevent duplicate lineage cleanup and remove blocking refresh in PATCH /api/v1/tables/{id} #26702

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
59 changes: 40 additions & 19 deletions openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/EntityRepository.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6036,6 +6036,7 @@ public class EntityUpdater {
@Setter private Set<String> patchedFields;
private final List<Runnable> deferredReactOperations = new ArrayList<>();
private boolean deferredReactExecuted;
private boolean deferOpsEnabled = true;

// Store the original FQN at construction time, before any modifications or revert.
// This is needed because during change consolidation, revert() reassigns 'original' to
Expand Down Expand Up @@ -6121,7 +6122,7 @@ public EntityUpdater(
}

protected final void deferReactOperation(Runnable operation) {
if (operation != null) {
if (operation != null && deferOpsEnabled) {
deferredReactOperations.add(operation);
}
}
Expand Down Expand Up @@ -6311,15 +6312,25 @@ public T getUpdated() {

private void incrementalChange() {
changeDescription = new ChangeDescription();
updateInternal(false);
deferOpsEnabled = false;
try {
updateInternal(false);
} finally {
deferOpsEnabled = true;
}
incrementalChangeDescription = changeDescription;
incrementalChangeDescription.setPreviousVersion(original.getVersion());
updated.setIncrementalChangeDescription(incrementalChangeDescription);
}

private void incrementalChangeForImport() {
changeDescription = new ChangeDescription();
updateInternalForImport(false);
deferOpsEnabled = false;
try {
updateInternalForImport(false);
} finally {
deferOpsEnabled = true;
}
incrementalChangeDescription = changeDescription;
incrementalChangeDescription.setPreviousVersion(original.getVersion());
updated.setIncrementalChangeDescription(incrementalChangeDescription);
Expand Down Expand Up @@ -6383,14 +6394,19 @@ private void revert() {
previous.getVersion());
changeDescription = new ChangeDescription();
updated = previous;
updateInternal(true);
LOG.info(
"In session change consolidation. Reverting to previous version {} completed",
previous.getVersion());

// Now go from original to updated
updated = updatedOld;
updateInternal();
deferOpsEnabled = false;
try {
updateInternal(true);
LOG.info(
"In session change consolidation. Reverting to previous version {} completed",
previous.getVersion());

// Now go from original to updated
updated = updatedOld;
updateInternal();
} finally {
deferOpsEnabled = true;
}

// Finally, go from previous to the latest updated entity to consolidate changes
original = previous;
Expand All @@ -6410,14 +6426,19 @@ private void revertForImport() {
previous.getVersion());
changeDescription = new ChangeDescription();
updated = previous;
updateInternalForImport(true);
LOG.info(
"In session change consolidation. Reverting to previous version {} completed",
previous.getVersion());

// Now go from original to updated
updated = updatedOld;
updateInternalForImport();
deferOpsEnabled = false;
try {
updateInternalForImport(true);
LOG.info(
"In session change consolidation. Reverting to previous version {} completed",
previous.getVersion());

// Now go from original to updated
updated = updatedOld;
updateInternalForImport();
} finally {
deferOpsEnabled = true;
}

// Finally, go from previous to the latest updated entity to consolidate changes
original = previous;
Expand Down
6 changes: 2 additions & 4 deletions ...c/main/java/org/openmetadata/service/search/elasticsearch/ElasticSearchEntityManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -814,8 +814,7 @@ public void updateColumnsInUpstreamLineage(
s ->
s.source(ss -> ss.scriptString(UPDATE_COLUMN_LINEAGE_SCRIPT))
.lang(ScriptLanguage.Painless)
.params(params))
.refresh(true));
.params(params)));

LOG.info(
"Successfully updated columns in upstream lineage for index: {}, updated: {}",
Expand Down Expand Up @@ -864,8 +863,7 @@ public void deleteColumnsInUpstreamLineage(String indexName, List<String> delete
s ->
s.source(ss -> ss.scriptString(DELETE_COLUMN_LINEAGE_SCRIPT))
.lang(ScriptLanguage.Painless)
.params(params))
.refresh(true));
.params(params)));

LOG.info(
"Successfully deleted columns from upstream lineage for index: {}, updated: {}",
Expand Down
6 changes: 2 additions & 4 deletions ...ice/src/main/java/org/openmetadata/service/search/opensearch/OpenSearchEntityManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -883,8 +883,7 @@ public void updateColumnsInUpstreamLineage(
os.org.opensearch.client.opensearch._types
.BuiltinScriptLanguage.Painless))
.source(UPDATE_COLUMN_LINEAGE_SCRIPT)
.params(params)))
.refresh(Refresh.True));
.params(params))));

LOG.info(
"Successfully updated columns in upstream lineage for index: {}, updated: {}",
Expand Down Expand Up @@ -938,8 +937,7 @@ public void deleteColumnsInUpstreamLineage(String indexName, List<String> delete
os.org.opensearch.client.opensearch._types
.BuiltinScriptLanguage.Painless))
.source(DELETE_COLUMN_LINEAGE_SCRIPT)
.params(params)))
.refresh(Refresh.True));
.params(params))));

LOG.info(
"Successfully deleted columns from upstream lineage for index: {}, updated: {}",
Expand Down
Loading