Repository: atlas Updated Branches: refs/heads/branch-0.8 7b131c06a -> be6a4ffa0
ATLAS-2891: updated hook notification processing with option to ignore potentially incorrect hive_column_lineage - #2 Project: http://git-wip-us.apache.org/repos/asf/atlas/repo Commit: http://git-wip-us.apache.org/repos/asf/atlas/commit/be6a4ffa Tree: http://git-wip-us.apache.org/repos/asf/atlas/tree/be6a4ffa Diff: http://git-wip-us.apache.org/repos/asf/atlas/diff/be6a4ffa Branch: refs/heads/branch-0.8 Commit: be6a4ffa053cf5e8eed431ac07a104a31e55c1e2 Parents: 7b131c0 Author: Madhan Neethiraj <[email protected]> Authored: Wed Oct 10 16:46:18 2018 -0700 Committer: Madhan Neethiraj <[email protected]> Committed: Wed Oct 10 16:46:18 2018 -0700 ---------------------------------------------------------------------- .../org/apache/atlas/hive/hook/events/CreateHiveProcess.java | 2 +- .../org/apache/atlas/notification/NotificationHookConsumer.java | 5 ++++- 2 files changed, 5 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/atlas/blob/be6a4ffa/addons/hive-bridge/src/main/java/org/apache/atlas/hive/hook/events/CreateHiveProcess.java ---------------------------------------------------------------------- diff --git a/addons/hive-bridge/src/main/java/org/apache/atlas/hive/hook/events/CreateHiveProcess.java b/addons/hive-bridge/src/main/java/org/apache/atlas/hive/hook/events/CreateHiveProcess.java index 9b9852e..12a65bd 100644 --- a/addons/hive-bridge/src/main/java/org/apache/atlas/hive/hook/events/CreateHiveProcess.java +++ b/addons/hive-bridge/src/main/java/org/apache/atlas/hive/hook/events/CreateHiveProcess.java @@ -188,7 +188,7 @@ public class CreateHiveProcess extends BaseHiveEvent { columnLineages.add(columnLineageProcess); } - boolean skipColumnLineage = context.getSkipHiveColumnLineageHive20633() && isSameInputsSize && lineageInputsSize > context.getSkipHiveColumnLineageHive20633InputsThreshold(); + boolean skipColumnLineage = context.getSkipHiveColumnLineageHive20633() && columnLineages.size() > 1 && isSameInputsSize && lineageInputsSize > context.getSkipHiveColumnLineageHive20633InputsThreshold(); if (!skipColumnLineage) { for (AtlasEntity columnLineage : columnLineages) { http://git-wip-us.apache.org/repos/asf/atlas/blob/be6a4ffa/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java ---------------------------------------------------------------------- diff --git a/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java b/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java index 6b34aa6..753ea9e 100644 --- a/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java +++ b/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java @@ -652,12 +652,15 @@ public class NotificationHookConsumer implements Service, ActiveStateChangeHandl if (entities != null && entities.getEntities() != null) { boolean isSameInputsSize = true; int lineageInputsSize = -1; + int lineageCount = 0; // find if all hive_column_lineage entities have same number of inputs, which is likely to be caused by HIVE-20633 that results in incorrect lineage in some cases for (ListIterator<AtlasEntity> iter = entities.getEntities().listIterator(); iter.hasNext(); ) { AtlasEntity entity = iter.next(); if (StringUtils.equals(entity.getTypeName(), TYPE_HIVE_COLUMN_LINEAGE)) { + lineageCount++; + Object objInputs = entity.getAttribute(ATTRIBUTE_INPUTS); if (objInputs instanceof Collection) { @@ -674,7 +677,7 @@ public class NotificationHookConsumer implements Service, ActiveStateChangeHandl } } - if (isSameInputsSize && lineageInputsSize > skipHiveColumnLineageHive20633InputsThreshold) { + if (isSameInputsSize && lineageCount > 1 && lineageInputsSize > skipHiveColumnLineageHive20633InputsThreshold) { int numRemovedEntities = 0; for (ListIterator<AtlasEntity> iter = entities.getEntities().listIterator(); iter.hasNext(); ) {
