You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@atlas.apache.org by am...@apache.org on 2019/11/06 22:26:55 UTC
[atlas] branch branch-2.0 updated (b572eaa -> f78f2f3)
This is an automated email from the ASF dual-hosted git repository.
amestry pushed a change to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/atlas.git.
from b572eaa ATLAS-3488 : Update Simple Authentication(file-based) password with ShaPasswordEncoder with Salt (Addendum patch #1).
new 1c078ed ATLAS-3416 Import API: delete non-exported hive_table for table level replication
new f78f2f3 ATLAS-3451 Export API connected export mismatch for lineage of entities from different DBs
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../atlas/model/impexp/ExportImportAuditEntry.java | 1 +
.../atlas/repository/impexp/AuditsWriter.java | 22 ++
.../atlas/repository/impexp/ImportService.java | 40 ++-
.../impexp/RelationshipAttributesExtractor.java | 79 ++++--
.../impexp/TableReplicationRequestProcessor.java | 181 ++++++++++++++
.../atlas/repository/impexp/ImportServiceTest.java | 83 ++++++-
.../RelationshipAttributesExtractorTest.java | 65 ++++-
.../TableReplicationRequestProcessorTest.java | 146 +++++++++++
.../resources/json/entities/column-lineage.json | 89 +++++++
.../src/test/resources/json/entities/db1.json | 27 ++
.../src/test/resources/json/entities/db2.json | 27 ++
.../resources/json/entities/table-lineage.json | 240 ++++++++++++++++++
.../src/test/resources/json/entities/table1.json | 276 +++++++++++++++++++++
.../src/test/resources/json/entities/table2.json | 276 +++++++++++++++++++++
14 files changed, 1510 insertions(+), 42 deletions(-)
create mode 100644 repository/src/main/java/org/apache/atlas/repository/impexp/TableReplicationRequestProcessor.java
create mode 100644 repository/src/test/java/org/apache/atlas/repository/impexp/TableReplicationRequestProcessorTest.java
create mode 100644 repository/src/test/resources/json/entities/column-lineage.json
create mode 100644 repository/src/test/resources/json/entities/db1.json
create mode 100644 repository/src/test/resources/json/entities/db2.json
create mode 100644 repository/src/test/resources/json/entities/table-lineage.json
create mode 100644 repository/src/test/resources/json/entities/table1.json
create mode 100644 repository/src/test/resources/json/entities/table2.json
[atlas] 01/02: ATLAS-3416 Import API: delete non-exported
hive_table for table level replication
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amestry pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/atlas.git
commit 1c078edc1fa0a6befff232b6f75fc0b4ae330ce4
Author: nikhilbonte <ni...@freestoneinfotech.com>
AuthorDate: Thu Aug 29 12:47:20 2019 +0530
ATLAS-3416 Import API: delete non-exported hive_table for table level replication
Signed-off-by: nixonrodrigues <ni...@apache.org>
---
.../atlas/model/impexp/ExportImportAuditEntry.java | 1 +
.../atlas/repository/impexp/AuditsWriter.java | 22 +++
.../atlas/repository/impexp/ImportService.java | 40 ++++-
.../impexp/TableReplicationRequestProcessor.java | 181 +++++++++++++++++++++
.../atlas/repository/impexp/ImportServiceTest.java | 83 +++++++++-
.../TableReplicationRequestProcessorTest.java | 146 +++++++++++++++++
6 files changed, 461 insertions(+), 12 deletions(-)
diff --git a/intg/src/main/java/org/apache/atlas/model/impexp/ExportImportAuditEntry.java b/intg/src/main/java/org/apache/atlas/model/impexp/ExportImportAuditEntry.java
index a199c6e..90f4296 100644
--- a/intg/src/main/java/org/apache/atlas/model/impexp/ExportImportAuditEntry.java
+++ b/intg/src/main/java/org/apache/atlas/model/impexp/ExportImportAuditEntry.java
@@ -35,6 +35,7 @@ public class ExportImportAuditEntry extends AtlasBaseModelObject implements Seri
private static final long serialVersionUID = 1L;
public static final String OPERATION_EXPORT = "EXPORT";
public static final String OPERATION_IMPORT = "IMPORT";
+ public static final String OPERATION_IMPORT_DELETE_REPL = "IMPORT_DELETE_REPL";
private String userName;
private String operation;
diff --git a/repository/src/main/java/org/apache/atlas/repository/impexp/AuditsWriter.java b/repository/src/main/java/org/apache/atlas/repository/impexp/AuditsWriter.java
index f2d36ed..55990f7 100644
--- a/repository/src/main/java/org/apache/atlas/repository/impexp/AuditsWriter.java
+++ b/repository/src/main/java/org/apache/atlas/repository/impexp/AuditsWriter.java
@@ -45,6 +45,7 @@ import org.springframework.util.CollectionUtils;
import javax.inject.Inject;
import java.util.Collections;
import java.util.List;
+import java.util.Set;
@Component
public class AuditsWriter {
@@ -68,6 +69,10 @@ public class AuditsWriter {
this.auditService = auditService;
}
+ public AtlasServerService getAtlasServerService() {
+ return atlasServerService;
+ }
+
public void write(String userName, AtlasExportResult result,
long startTime, long endTime,
List<String> entityCreationOrder) throws AtlasBaseException {
@@ -80,6 +85,12 @@ public class AuditsWriter {
auditForImport.add(userName, result, startTime, endTime, entityCreationOrder);
}
+ public void write(String userName, String sourceCluster,
+ long startTime, long endTime,
+ Set<String> entityCreationOrder) throws AtlasBaseException {
+ auditForImport.add(userName, sourceCluster, startTime, endTime, entityCreationOrder);
+ }
+
private void updateReplicationAttribute(boolean isReplicationSet,
String serverName, String serverFullName,
List<String> exportedGuids,
@@ -238,5 +249,16 @@ public class AuditsWriter {
updateReplicationAttribute(replicationOptionState, sourceServerName, sourceServerFullName, entityGuids,
Constants.ATTR_NAME_REPLICATED_FROM, result.getExportResult().getChangeMarker());
}
+
+ public void add(String userName, String sourceCluster, long startTime,
+ long endTime, Set<String> entityGuids) throws AtlasBaseException {
+
+ sourceServerName = getServerNameFromFullName(sourceCluster);
+ auditService.add(userName,
+ sourceServerName, getCurrentClusterName(),
+ ExportImportAuditEntry.OPERATION_IMPORT_DELETE_REPL,
+ AtlasType.toJson(entityGuids), startTime, endTime, !entityGuids.isEmpty());
+
+ }
}
}
diff --git a/repository/src/main/java/org/apache/atlas/repository/impexp/ImportService.java b/repository/src/main/java/org/apache/atlas/repository/impexp/ImportService.java
index df49ae1..27001e3 100644
--- a/repository/src/main/java/org/apache/atlas/repository/impexp/ImportService.java
+++ b/repository/src/main/java/org/apache/atlas/repository/impexp/ImportService.java
@@ -24,8 +24,10 @@ import org.apache.atlas.RequestContext;
import org.apache.atlas.entitytransform.BaseEntityHandler;
import org.apache.atlas.entitytransform.TransformerContext;
import org.apache.atlas.exception.AtlasBaseException;
+import org.apache.atlas.model.impexp.AtlasExportRequest;
import org.apache.atlas.model.impexp.AtlasImportRequest;
import org.apache.atlas.model.impexp.AtlasImportResult;
+import org.apache.atlas.model.instance.AtlasObjectId;
import org.apache.atlas.model.typedef.AtlasTypesDef;
import org.apache.atlas.repository.store.graph.BulkImporter;
import org.apache.atlas.repository.store.graph.v2.EntityImportStream;
@@ -54,24 +56,28 @@ import static org.apache.atlas.model.impexp.AtlasImportRequest.TRANSFORMS_KEY;
public class ImportService {
private static final Logger LOG = LoggerFactory.getLogger(ImportService.class);
+ private static final String ATLAS_TYPE_HIVE_TABLE = "hive_table";
private final AtlasTypeDefStore typeDefStore;
private final AtlasTypeRegistry typeRegistry;
private final BulkImporter bulkImporter;
private final AuditsWriter auditsWriter;
private final ImportTransformsShaper importTransformsShaper;
+ private TableReplicationRequestProcessor tableReplicationRequestProcessor;
+
private long startTimestamp;
private long endTimestamp;
@Inject
public ImportService(AtlasTypeDefStore typeDefStore, AtlasTypeRegistry typeRegistry, BulkImporter bulkImporter,
- AuditsWriter auditsWriter,
- ImportTransformsShaper importTransformsShaper) {
+ AuditsWriter auditsWriter, ImportTransformsShaper importTransformsShaper,
+ TableReplicationRequestProcessor tableReplicationRequestProcessor) {
this.typeDefStore = typeDefStore;
this.typeRegistry = typeRegistry;
this.bulkImporter = bulkImporter;
this.auditsWriter = auditsWriter;
this.importTransformsShaper = importTransformsShaper;
+ this.tableReplicationRequestProcessor = tableReplicationRequestProcessor;
}
public AtlasImportResult run(InputStream inputStream, String userName,
@@ -109,7 +115,11 @@ public class ImportService {
startTimestamp = System.currentTimeMillis();
processTypes(source.getTypesDef(), result);
setStartPosition(request, source);
+
processEntities(userName, source, result);
+
+ processReplicationDeletion(source.getExportResult().getRequest(), request);
+
} catch (AtlasBaseException excp) {
LOG.error("import(user={}, from={}): failed", userName, requestingIP, excp);
@@ -228,6 +238,12 @@ public class ImportService {
auditsWriter.write(userName, result, startTimestamp, endTimestamp, importSource.getCreationOrder());
}
+ private void processReplicationDeletion(AtlasExportRequest exportRequest, AtlasImportRequest importRequest) throws AtlasBaseException {
+ if (checkHiveTableIncrementalSkipLineage(importRequest, exportRequest)) {
+ tableReplicationRequestProcessor.process(exportRequest, importRequest);
+ }
+ }
+
private int getDuration(long endTime, long startTime) {
return (int) (endTime - startTime);
}
@@ -239,9 +255,25 @@ public class ImportService {
}
return new ZipSourceWithBackingDirectory(inputStream, configuredTemporaryDirectory);
- }
- catch (IOException ex) {
+ } catch (IOException ex) {
throw new AtlasBaseException(ex);
}
}
+
+ @VisibleForTesting
+ boolean checkHiveTableIncrementalSkipLineage(AtlasImportRequest importRequest, AtlasExportRequest exportRequest) {
+ if (CollectionUtils.isEmpty(exportRequest.getItemsToExport())) {
+ return false;
+ }
+
+ for (AtlasObjectId itemToExport : exportRequest.getItemsToExport()) {
+ if (!itemToExport.getTypeName().equalsIgnoreCase(ATLAS_TYPE_HIVE_TABLE)){
+ return false;
+ }
+ }
+
+ return importRequest.isReplicationOptionSet() && exportRequest.isReplicationOptionSet() &&
+ exportRequest.getFetchTypeOptionValue().equalsIgnoreCase(AtlasExportRequest.FETCH_TYPE_INCREMENTAL) &&
+ exportRequest.getSkipLineageOptionValue();
+ }
}
diff --git a/repository/src/main/java/org/apache/atlas/repository/impexp/TableReplicationRequestProcessor.java b/repository/src/main/java/org/apache/atlas/repository/impexp/TableReplicationRequestProcessor.java
new file mode 100644
index 0000000..d5807a5
--- /dev/null
+++ b/repository/src/main/java/org/apache/atlas/repository/impexp/TableReplicationRequestProcessor.java
@@ -0,0 +1,181 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.atlas.repository.impexp;
+
+import org.apache.atlas.authorize.AtlasAuthorizationUtils;
+import org.apache.atlas.discovery.AtlasDiscoveryService;
+import org.apache.atlas.exception.AtlasBaseException;
+import org.apache.atlas.model.discovery.AtlasSearchResult;
+import org.apache.atlas.model.discovery.SearchParameters;
+import org.apache.atlas.model.impexp.AtlasExportRequest;
+import org.apache.atlas.model.impexp.AtlasImportRequest;
+import org.apache.atlas.model.instance.AtlasEntityHeader;
+import org.apache.atlas.model.instance.AtlasObjectId;
+import org.apache.atlas.repository.store.graph.AtlasEntityStore;
+import org.apache.atlas.type.AtlasTypeRegistry;
+import org.apache.commons.collections.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import javax.inject.Inject;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.Map;
+import java.util.ArrayList;
+import java.util.HashSet;
+
+@Component
+public class TableReplicationRequestProcessor {
+ private static final Logger LOG = LoggerFactory.getLogger(TableReplicationRequestProcessor.class);
+
+ private static final String QUERY_DB_NAME_EQUALS= "where db.name='%s'";
+ private static final String ATTR_NAME_KEY = "name";
+ private static final String TYPE_HIVE_TABLE = "hive_table";
+ private static final String ATTR_QUALIFIED_NAME_KEY = "qualifiedName";
+ private static final String REPLICATED_TAG_NAME = "%s_replicated";
+
+ private long startTstamp;
+ private long endTstamp;
+ private AuditsWriter auditsWriter;
+ private AtlasEntityStore entityStore;
+ private AtlasTypeRegistry typeRegistry;
+ private AtlasDiscoveryService discoveryService;
+
+ @Inject
+ public TableReplicationRequestProcessor(AuditsWriter auditsWriter, AtlasEntityStore entityStore,
+ AtlasDiscoveryService atlasDiscoveryService, AtlasTypeRegistry typeRegistry) {
+ this.auditsWriter = auditsWriter;
+ this.entityStore = entityStore;
+ this.typeRegistry = typeRegistry;
+ this.discoveryService = atlasDiscoveryService;
+ }
+
+ public void process(AtlasExportRequest exportRequest, AtlasImportRequest importRequest) throws AtlasBaseException {
+ startTstamp = System.currentTimeMillis();
+ LOG.info("process: deleting entities with type hive_table which are not imported.");
+ String sourceCluster = importRequest.getOptionKeyReplicatedFrom();
+
+ List<String> qualifiedNames = getQualifiedNamesFromRequest(exportRequest);
+
+ List<String> safeGUIDs = getEntitiesFromQualifiedNames(qualifiedNames);
+
+ String dbName = getDbName(safeGUIDs.get(0));
+
+ Set<String> guidsToDelete = getGuidsToDelete(dbName, safeGUIDs, sourceCluster);
+
+ deleteTables(sourceCluster, guidsToDelete);
+ }
+
+ private List<String> getQualifiedNamesFromRequest(AtlasExportRequest exportRequest){
+ List<String> qualifiedNames = new ArrayList<>();
+
+ for (AtlasObjectId objectId : exportRequest.getItemsToExport()) {
+ qualifiedNames.add(objectId.getUniqueAttributes().get(ATTR_QUALIFIED_NAME_KEY).toString());
+ }
+ return qualifiedNames;
+ }
+
+ private List<String> getEntitiesFromQualifiedNames(List<String> qualifiedNames) throws AtlasBaseException {
+
+ List<String> safeGUIDs = new ArrayList<>();
+ for(String qualifiedName : qualifiedNames) {
+ String guid = getGuidByUniqueAttributes(Collections.singletonMap(ATTR_QUALIFIED_NAME_KEY, qualifiedName));
+ safeGUIDs.add(guid);
+ }
+ return safeGUIDs;
+ }
+
+ private String getGuidByUniqueAttributes(Map<String, Object> uniqueAttributes) throws AtlasBaseException {
+ return entityStore.getGuidByUniqueAttributes(typeRegistry.getEntityTypeByName(TYPE_HIVE_TABLE), uniqueAttributes);
+ }
+
+ private String getDbName(String tableGuid) throws AtlasBaseException {
+ String dbGuid = AuditsWriter.ReplKeyGuidFinder.get(typeRegistry, entityStore, tableGuid);
+ return (String) entityStore.getById(dbGuid).getEntity().getAttribute(ATTR_NAME_KEY);
+ }
+
+ private Set<String> getGuidsToDelete(String dbName, List<String> excludeGUIDs, String sourceCluster) throws AtlasBaseException {
+
+ SearchParameters parameters = getSearchParameters(dbName, sourceCluster);
+ Set<String> unsafeGUIDs = new HashSet<>();
+
+ final int max = 10000;
+ int fetchedSize = 0;
+ int i = 0;
+ parameters.setLimit(max);
+
+ while (fetchedSize == (max * i)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("i={}, fetchedSize={}, unsafeGUIDs.size()={}", i, fetchedSize, unsafeGUIDs.size());
+ }
+
+ int offset = max * i;
+ parameters.setOffset(offset);
+
+ AtlasSearchResult searchResult = discoveryService.searchWithParameters(parameters);
+
+ if (CollectionUtils.isEmpty(searchResult.getEntities())) {
+ break;
+ }
+
+ for (AtlasEntityHeader entityHeader : searchResult.getEntities()) {
+ String guid = entityHeader.getGuid();
+ if (!excludeGUIDs.contains(guid)) {
+ unsafeGUIDs.add(guid);
+ }
+ }
+ fetchedSize = searchResult.getEntities().size();
+ i++;
+ }
+ return unsafeGUIDs;
+ }
+
+ private SearchParameters getSearchParameters(String dbName, String sourceCluster) {
+ String query = String.format(QUERY_DB_NAME_EQUALS, dbName);
+
+ SearchParameters parameters = new SearchParameters();
+ parameters.setExcludeDeletedEntities(false);
+ parameters.setTypeName(TYPE_HIVE_TABLE);
+ parameters.setExcludeDeletedEntities(true);
+
+ parameters.setClassification(String.format(REPLICATED_TAG_NAME, sourceCluster));
+ parameters.setAttributes(new HashSet<String>(){{ add(AtlasImportRequest.OPTION_KEY_REPLICATED_FROM); }});
+ parameters.setQuery(query);
+
+ return parameters;
+ }
+
+ private void deleteTables(String sourceCluster, Set<String> guidsToDelete) throws AtlasBaseException {
+ if (!CollectionUtils.isEmpty(guidsToDelete)) {
+ entityStore.deleteByIds(new ArrayList<>(guidsToDelete));
+
+ endTstamp = System.currentTimeMillis();
+ createAuditEntry(sourceCluster, guidsToDelete);
+ }
+ }
+
+ private void createAuditEntry(String sourceCluster, Set<String> guidsToDelete) throws AtlasBaseException {
+ auditsWriter.write(AtlasAuthorizationUtils.getCurrentUserName(), sourceCluster, startTstamp, endTstamp, guidsToDelete);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Deleted entities => {}", guidsToDelete);
+ }
+ }
+}
diff --git a/repository/src/test/java/org/apache/atlas/repository/impexp/ImportServiceTest.java b/repository/src/test/java/org/apache/atlas/repository/impexp/ImportServiceTest.java
index 71dd44f..1cb0140 100644
--- a/repository/src/test/java/org/apache/atlas/repository/impexp/ImportServiceTest.java
+++ b/repository/src/test/java/org/apache/atlas/repository/impexp/ImportServiceTest.java
@@ -24,9 +24,11 @@ import org.apache.atlas.TestModules;
import org.apache.atlas.TestUtilsV2;
import org.apache.atlas.discovery.EntityDiscoveryService;
import org.apache.atlas.exception.AtlasBaseException;
+import org.apache.atlas.model.impexp.AtlasExportRequest;
import org.apache.atlas.model.impexp.AtlasImportRequest;
import org.apache.atlas.model.instance.AtlasEntity;
import org.apache.atlas.model.instance.AtlasEntityHeader;
+import org.apache.atlas.model.instance.AtlasObjectId;
import org.apache.atlas.model.instance.AtlasRelatedObjectId;
import org.apache.atlas.model.instance.AtlasRelationship;
import org.apache.atlas.model.instance.EntityMutationResponse;
@@ -41,10 +43,7 @@ import org.apache.atlas.store.AtlasTypeDefStore;
import org.apache.atlas.type.AtlasClassificationType;
import org.apache.atlas.type.AtlasTypeRegistry;
import org.apache.commons.lang.StringUtils;
-import org.apache.tinkerpop.shaded.kryo.io.Input;
import org.mockito.stubbing.Answer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.testng.ITestContext;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterTest;
@@ -55,6 +54,7 @@ import org.testng.annotations.Test;
import java.io.IOException;
import java.io.InputStream;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -62,11 +62,17 @@ import java.util.Map;
import static org.apache.atlas.graph.GraphSandboxUtil.useLocalSolr;
import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.getDefaultImportRequest;
import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.getZipSource;
+import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.getInputStreamFrom;
import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.loadModelFromJson;
import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.loadModelFromResourcesJson;
import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.runAndVerifyQuickStart_v1_Import;
import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.runImportWithNoParameters;
import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.runImportWithParameters;
+import static org.apache.atlas.model.impexp.AtlasExportRequest.OPTION_FETCH_TYPE;
+import static org.apache.atlas.model.impexp.AtlasExportRequest.OPTION_KEY_REPLICATED_TO;
+import static org.apache.atlas.model.impexp.AtlasExportRequest.OPTION_SKIP_LINEAGE;
+import static org.apache.atlas.model.impexp.AtlasExportRequest.FETCH_TYPE_FULL;
+import static org.apache.atlas.model.impexp.AtlasExportRequest.FETCH_TYPE_INCREMENTAL;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
@@ -398,7 +404,7 @@ public class ImportServiceTest extends ExportImportTestBase {
@Test
public void importServiceProcessesIOException() {
- ImportService importService = new ImportService(typeDefStore, typeRegistry, null, null,null);
+ ImportService importService = new ImportService(typeDefStore, typeRegistry, null,null, null,null);
AtlasImportRequest req = mock(AtlasImportRequest.class);
Answer<Map> answer = invocationOnMock -> {
@@ -453,8 +459,8 @@ public class ImportServiceTest extends ExportImportTestBase {
@Test(dataProvider = "salesNewTypeAttrs-next")
public void transformUpdatesForSubTypesAddsToExistingTransforms(InputStream inputStream) throws IOException, AtlasBaseException {
- loadBaseModel();
- loadHiveModel();
+ loadBaseModel();
+ loadHiveModel();
String transformJSON = "{ \"Asset\": { \"qualifiedName\":[ \"replace:@cl1:@cl2\" ] }, \"hive_table\": { \"qualifiedName\":[ \"lowercase\" ] } }";
ZipSource zipSource = new ZipSource(inputStream);
@@ -467,10 +473,71 @@ public class ImportServiceTest extends ExportImportTestBase {
assertEquals(importTransforms.getTransforms().get("hive_table").get("qualifiedName").size(), 2);
}
-
@Test(expectedExceptions = AtlasBaseException.class)
public void importEmptyZip() throws IOException, AtlasBaseException {
- new ZipSource((InputStream) getZipSource("empty.zip")[0][0]);
+ new ZipSource(getInputStreamFrom("empty.zip"));
+ }
+
+ @Test
+ public void testCheckHiveTableIncrementalSkipLineage() {
+ AtlasImportRequest importRequest;
+ AtlasExportRequest exportRequest;
+
+ importRequest = getImportRequest("cl1");
+ exportRequest = getExportRequest(FETCH_TYPE_INCREMENTAL, "cl2", true, getItemsToExport("hive_table", "hive_table"));
+ assertTrue(importService.checkHiveTableIncrementalSkipLineage(importRequest, exportRequest));
+
+ exportRequest = getExportRequest(FETCH_TYPE_INCREMENTAL, "cl2", true, getItemsToExport("hive_table", "hive_db", "hive_table"));
+ assertFalse(importService.checkHiveTableIncrementalSkipLineage(importRequest, exportRequest));
+
+ exportRequest = getExportRequest(FETCH_TYPE_FULL, "cl2", true, getItemsToExport("hive_table", "hive_table"));
+ assertFalse(importService.checkHiveTableIncrementalSkipLineage(importRequest, exportRequest));
+
+ exportRequest = getExportRequest(FETCH_TYPE_FULL, "", true, getItemsToExport("hive_table", "hive_table"));
+ assertFalse(importService.checkHiveTableIncrementalSkipLineage(importRequest, exportRequest));
+
+ importRequest = getImportRequest("");
+ exportRequest = getExportRequest(FETCH_TYPE_INCREMENTAL, "cl2", true, getItemsToExport("hive_table", "hive_table"));
+ assertFalse(importService.checkHiveTableIncrementalSkipLineage(importRequest, exportRequest));
+ }
+
+ private AtlasImportRequest getImportRequest(String replicatedFrom){
+ AtlasImportRequest importRequest = getDefaultImportRequest();
+
+ if (!StringUtils.isEmpty(replicatedFrom)) {
+ importRequest.setOption(AtlasImportRequest.OPTION_KEY_REPLICATED_FROM, replicatedFrom);
+ }
+ return importRequest;
+ }
+
+ private AtlasExportRequest getExportRequest(String fetchType, String replicatedTo, boolean skipLineage, List<AtlasObjectId> itemsToExport){
+ AtlasExportRequest request = new AtlasExportRequest();
+
+ request.setOptions(getOptionsMap(fetchType, replicatedTo, skipLineage));
+ request.setItemsToExport(itemsToExport);
+ return request;
+ }
+
+ private List<AtlasObjectId> getItemsToExport(String... typeNames){
+ List<AtlasObjectId> itemsToExport = new ArrayList<>();
+ for (String typeName : typeNames) {
+ itemsToExport.add(new AtlasObjectId(typeName, "qualifiedName", "db.table@cluster"));
+ }
+ return itemsToExport;
+ }
+
+ private Map<String, Object> getOptionsMap(String fetchType, String replicatedTo, boolean skipLineage){
+ Map<String, Object> options = new HashMap<>();
+
+ if (!StringUtils.isEmpty(fetchType)) {
+ options.put(OPTION_FETCH_TYPE, fetchType);
+ }
+ if (!StringUtils.isEmpty(replicatedTo)) {
+ options.put(OPTION_KEY_REPLICATED_TO, replicatedTo);
+ }
+ options.put(OPTION_SKIP_LINEAGE, skipLineage);
+
+ return options;
}
@Test(dataProvider = "dup_col_data")
diff --git a/repository/src/test/java/org/apache/atlas/repository/impexp/TableReplicationRequestProcessorTest.java b/repository/src/test/java/org/apache/atlas/repository/impexp/TableReplicationRequestProcessorTest.java
new file mode 100644
index 0000000..c9bb11c
--- /dev/null
+++ b/repository/src/test/java/org/apache/atlas/repository/impexp/TableReplicationRequestProcessorTest.java
@@ -0,0 +1,146 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.atlas.repository.impexp;
+
+import com.google.inject.Inject;
+import org.apache.atlas.RequestContext;
+import org.apache.atlas.TestModules;
+import org.apache.atlas.TestUtilsV2;
+import org.apache.atlas.exception.AtlasBaseException;
+import org.apache.atlas.model.impexp.AtlasImportRequest;
+import org.apache.atlas.model.impexp.ExportImportAuditEntry;
+import org.apache.atlas.repository.graph.AtlasGraphProvider;
+import org.apache.atlas.repository.store.graph.AtlasEntityStore;
+import org.apache.atlas.runner.LocalSolrRunner;
+import org.apache.atlas.store.AtlasTypeDefStore;
+import org.apache.atlas.type.AtlasType;
+import org.apache.atlas.type.AtlasTypeRegistry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.ITestContext;
+import org.testng.SkipException;
+import org.testng.annotations.Guice;
+import org.testng.annotations.Test;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeTest;
+import org.testng.annotations.DataProvider;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.List;
+
+import static org.apache.atlas.graph.GraphSandboxUtil.useLocalSolr;
+import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.getDefaultImportRequest;
+import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.getZipSource;
+import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.getInputStreamFrom;
+import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.runImportWithParameters;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+
+
+@Guice(modules = TestModules.TestOnlyModule.class)
+public class TableReplicationRequestProcessorTest extends ExportImportTestBase {
+ private static final Logger LOG = LoggerFactory.getLogger(TableReplicationRequestProcessorTest.class);
+
+ private static final String ENTITY_GUID_REPLICATED = "718a6d12-35a8-4731-aff8-3a64637a43a3";
+ private static final String ENTITY_GUID_NOT_REPLICATED_1 = "e19e5683-d9ae-436a-af1e-0873582d0f1e";
+ private static final String ENTITY_GUID_NOT_REPLICATED_2 = "2e28ae34-576e-4a8b-be48-cf5f925d7b15";
+ private static final String REPL_FROM = "cl1";
+ private static final String REPL_TRANSFORMER = "[{\"conditions\":{\"__entity\":\"topLevel: \"}," +
+ "\"action\":{\"__entity\":\"ADD_CLASSIFICATION: cl1_replicated\"}}," +
+ "{\"action\":{\"__entity.replicatedTo\":\"CLEAR:\",\"__entity.replicatedFrom\":\"CLEAR:\"}}," +
+ "{\"conditions\":{\"hive_db.clusterName\":\"EQUALS: cl1\"},\"action\":{\"hive_db.clusterName\":\"SET: cl2\"}}," +
+ "{\"conditions\":{\"hive_db.location\":\"STARTS_WITH_IGNORE_CASE: file:///\"}," +
+ "\"action\":{\"hive_db.location\":\"REPLACE_PREFIX: = :file:///=file:///\"}}," +
+ "{\"conditions\":{\"hive_storagedesc.location\":\"STARTS_WITH_IGNORE_CASE: file:///\"}," +
+ "\"action\":{\"hive_storagedesc.location\":\"REPLACE_PREFIX: = :file:///=file:///\"}}]";
+
+ @Inject
+ private ImportService importService;
+
+ @Inject
+ private AtlasTypeRegistry typeRegistry;
+
+ @Inject
+ private AtlasEntityStore entityStore;
+
+ @Inject
+ private ExportImportAuditService auditService;
+
+ @Inject
+ private AtlasTypeDefStore typeDefStore;
+
+ @BeforeTest
+ public void setupTest() throws IOException, AtlasBaseException {
+ RequestContext.clear();
+ RequestContext.get().setUser(TestUtilsV2.TEST_USER, null);
+ basicSetup(typeDefStore, typeRegistry);
+ }
+
+ @AfterClass
+ public void clear() throws Exception {
+ AtlasGraphProvider.cleanup();
+
+ if (useLocalSolr()) {
+ LocalSolrRunner.stop();
+ }
+ }
+
+ @DataProvider(name = "source1")
+ public static Object[][] getData1(ITestContext context) throws IOException, AtlasBaseException {
+ return getZipSource("repl_exp_1.zip");
+ }
+
+ public static InputStream getData2() {
+ return getInputStreamFrom("repl_exp_2.zip");
+ }
+
+ @Test(dataProvider = "source1")
+ public void importWithIsReplTrue(InputStream zipSource) throws AtlasBaseException, IOException {
+ AtlasImportRequest atlasImportRequest = getDefaultImportRequest();
+
+ atlasImportRequest.setOption("replicatedFrom", REPL_FROM);
+ atlasImportRequest.setOption("transformers", REPL_TRANSFORMER);
+
+ runImportWithParameters(importService, atlasImportRequest, zipSource);
+
+ runImportWithParameters(importService, atlasImportRequest, getData2());
+
+ assertAuditEntry();
+ }
+
+ private void assertAuditEntry() {
+ pauseForIndexCreation();
+ List<ExportImportAuditEntry> result;
+ try {
+ result = auditService.get("", "IMPORT_DELETE_REPL", "", "", "", 10, 0);
+ } catch (Exception e) {
+ throw new SkipException("audit entries not retrieved.");
+ }
+
+ assertNotNull(result);
+ assertTrue(result.size() > 0);
+
+ List<String> deletedGuids = AtlasType.fromJson(result.get(0).getResultSummary(), List.class);
+ assertNotNull(deletedGuids);
+ assertFalse(deletedGuids.contains(ENTITY_GUID_REPLICATED));
+ assertTrue(deletedGuids.contains(ENTITY_GUID_NOT_REPLICATED_1));
+ assertTrue(deletedGuids.contains(ENTITY_GUID_NOT_REPLICATED_2));
+ }
+}
[atlas] 02/02: ATLAS-3451 Export API connected export mismatch for
lineage of entities from different DBs
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amestry pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/atlas.git
commit f78f2f32d66b024b6b133483e46dbe578f3d37bd
Author: nikhilbonte <ni...@freestoneinfotech.com>
AuthorDate: Mon Oct 7 19:20:10 2019 +0530
ATLAS-3451 Export API connected export mismatch for lineage of entities from different DBs
Signed-off-by: nixonrodrigues <ni...@apache.org>
---
.../impexp/RelationshipAttributesExtractor.java | 79 ++++--
.../RelationshipAttributesExtractorTest.java | 65 ++++-
.../resources/json/entities/column-lineage.json | 89 +++++++
.../src/test/resources/json/entities/db1.json | 27 ++
.../src/test/resources/json/entities/db2.json | 27 ++
.../resources/json/entities/table-lineage.json | 240 ++++++++++++++++++
.../src/test/resources/json/entities/table1.json | 276 +++++++++++++++++++++
.../src/test/resources/json/entities/table2.json | 276 +++++++++++++++++++++
8 files changed, 1049 insertions(+), 30 deletions(-)
diff --git a/repository/src/main/java/org/apache/atlas/repository/impexp/RelationshipAttributesExtractor.java b/repository/src/main/java/org/apache/atlas/repository/impexp/RelationshipAttributesExtractor.java
index d609071..d8b0a20 100644
--- a/repository/src/main/java/org/apache/atlas/repository/impexp/RelationshipAttributesExtractor.java
+++ b/repository/src/main/java/org/apache/atlas/repository/impexp/RelationshipAttributesExtractor.java
@@ -19,7 +19,9 @@ package org.apache.atlas.repository.impexp;
import org.apache.atlas.model.instance.AtlasEntity;
import org.apache.atlas.model.instance.AtlasRelatedObjectId;
+import org.apache.atlas.model.typedef.AtlasBaseTypeDef;
import org.apache.atlas.model.typedef.AtlasEntityDef;
+import org.apache.atlas.repository.impexp.ExportService.TraversalDirection;
import org.apache.atlas.type.AtlasTypeRegistry;
import org.apache.atlas.type.AtlasTypeUtil;
import org.slf4j.Logger;
@@ -29,6 +31,12 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
+import static org.apache.atlas.repository.impexp.ExportService.ExportContext;
+import static org.apache.atlas.repository.impexp.ExportService.TraversalDirection.BOTH;
+import static org.apache.atlas.repository.impexp.ExportService.TraversalDirection.INWARD;
+import static org.apache.atlas.repository.impexp.ExportService.TraversalDirection.OUTWARD;
+import static org.apache.atlas.repository.impexp.ExportService.TraversalDirection.UNKNOWN;
+
public class RelationshipAttributesExtractor implements ExtractStrategy {
private static final Logger LOG = LoggerFactory.getLogger(RelationshipAttributesExtractor.class);
@@ -40,7 +48,7 @@ public class RelationshipAttributesExtractor implements ExtractStrategy {
}
@Override
- public void fullFetch(AtlasEntity entity, ExportService.ExportContext context) {
+ public void fullFetch(AtlasEntity entity, ExportContext context) {
if (LOG.isDebugEnabled()) {
LOG.debug("==> fullFetch({}): guidsToProcess {}", AtlasTypeUtil.getAtlasObjectId(entity), context.guidsToProcess.size());
}
@@ -53,7 +61,7 @@ public class RelationshipAttributesExtractor implements ExtractStrategy {
if (context.skipLineage && isLineage) {
continue;
}
- context.addToBeProcessed(isLineage, ar.getGuid(), ExportService.TraversalDirection.BOTH);
+ context.addToBeProcessed(isLineage, ar.getGuid(), BOTH);
}
if (LOG.isDebugEnabled()) {
@@ -62,31 +70,24 @@ public class RelationshipAttributesExtractor implements ExtractStrategy {
}
@Override
- public void connectedFetch(AtlasEntity entity, ExportService.ExportContext context) {
+ public void connectedFetch(AtlasEntity entity, ExportContext context) {
if (LOG.isDebugEnabled()) {
- LOG.debug("==> connectedFetch({}): guidsToProcess {} isSkipConnectedFetch :{}", AtlasTypeUtil.getAtlasObjectId(entity), context.guidsToProcess.size(), context.isSkipConnectedFetch);
+ LOG.debug("==> connectedFetch({}): guidsToProcess {}", AtlasTypeUtil.getAtlasObjectId(entity), context.guidsToProcess.size());
}
- List<AtlasRelatedObjectId> atlasRelatedObjectIdList = getRelatedObjectIds(entity);
- for (AtlasRelatedObjectId ar : atlasRelatedObjectIdList) {
- boolean isLineage = isLineageType(ar.getTypeName());
+ ExportService.TraversalDirection direction = context.guidDirection.get(entity.getGuid());
- if (context.skipLineage && isLineage) {
- continue;
+ if (direction == null || direction == UNKNOWN) {
+ addToBeProcessed(entity, context, OUTWARD, INWARD);
+ } else {
+ if (isLineageType(entity.getTypeName())) {
+ direction = OUTWARD;
}
- if (!context.isSkipConnectedFetch || isLineage) {
- context.addToBeProcessed(isLineage, ar.getGuid(), ExportService.TraversalDirection.BOTH);
- }
- }
-
- if(isLineageType(entity.getTypeName())){
- context.isSkipConnectedFetch = false;
- }else{
- context.isSkipConnectedFetch = true;
+ addToBeProcessed(entity, context, direction);
}
if (LOG.isDebugEnabled()) {
- LOG.debug("==> connectedFetch({}): guidsToProcess {}, isSkipConnectedFetch :{}", AtlasTypeUtil.getAtlasObjectId(entity), context.guidsToProcess.size(), context.isSkipConnectedFetch);
+ LOG.debug("==> connectedFetch({}): guidsToProcess {}", AtlasTypeUtil.getAtlasObjectId(entity), context.guidsToProcess.size());
}
}
@@ -94,9 +95,46 @@ public class RelationshipAttributesExtractor implements ExtractStrategy {
public void close() {
}
+ private void addToBeProcessed(AtlasEntity entity, ExportContext context, TraversalDirection... directions) {
+ if (directions == null || directions.length == 0) {
+ return;
+ }
+
+ boolean isLineageEntity = isLineageType(entity.getTypeName());
+ List<AtlasRelatedObjectId> relatedObjectIds = getRelatedObjectIds(entity);
+
+ for (TraversalDirection direction : directions) {
+ for (AtlasRelatedObjectId id : relatedObjectIds) {
+ String guid = id.getGuid();
+ TraversalDirection currentDirection = context.guidDirection.get(guid);
+ boolean isLineageId = isLineageType(id.getTypeName());
+ TraversalDirection edgeDirection = getRelationshipEdgeDirection(id, entity.getTypeName());
+
+ if (context.skipLineage && isLineageId) continue;
+
+ if (!isLineageEntity && direction != edgeDirection ||
+ isLineageEntity && direction == edgeDirection)
+ continue;
+
+ if (currentDirection == null) {
+ context.addToBeProcessed(isLineageId, guid, direction);
+
+ } else if (currentDirection == OUTWARD && direction == INWARD) {
+ context.guidsProcessed.remove(guid);
+ context.addToBeProcessed(isLineageId, guid, direction);
+ }
+ }
+ }
+ }
+
+ private TraversalDirection getRelationshipEdgeDirection(AtlasRelatedObjectId relatedObjectId, String entityTypeName) {
+ boolean isOutEdge = typeRegistry.getRelationshipDefByName(relatedObjectId.getRelationshipType()).getEndDef1().getType().equals(entityTypeName);
+ return isOutEdge ? OUTWARD : INWARD;
+ }
+
private boolean isLineageType(String typeName) {
AtlasEntityDef entityDef = typeRegistry.getEntityDefByName(typeName);
- return entityDef.getSuperTypes().contains("Process");
+ return entityDef.getSuperTypes().contains(AtlasBaseTypeDef.ATLAS_TYPE_PROCESS);
}
private List<AtlasRelatedObjectId> getRelatedObjectIds(AtlasEntity entity) {
@@ -109,7 +147,6 @@ public class RelationshipAttributesExtractor implements ExtractStrategy {
relatedObjectIds.addAll((List) o);
}
}
-
return relatedObjectIds;
}
}
diff --git a/repository/src/test/java/org/apache/atlas/repository/impexp/RelationshipAttributesExtractorTest.java b/repository/src/test/java/org/apache/atlas/repository/impexp/RelationshipAttributesExtractorTest.java
index 920fc28..a1b512f 100644
--- a/repository/src/test/java/org/apache/atlas/repository/impexp/RelationshipAttributesExtractorTest.java
+++ b/repository/src/test/java/org/apache/atlas/repository/impexp/RelationshipAttributesExtractorTest.java
@@ -26,6 +26,7 @@ import org.apache.atlas.model.impexp.AtlasExportResult;
import org.apache.atlas.model.instance.AtlasEntity;
import org.apache.atlas.model.instance.AtlasObjectId;
import org.apache.atlas.repository.graph.AtlasGraphProvider;
+import org.apache.atlas.repository.store.graph.v2.AtlasEntityStoreV2;
import org.apache.atlas.runner.LocalSolrRunner;
import org.apache.atlas.store.AtlasTypeDefStore;
import org.apache.atlas.type.AtlasTypeRegistry;
@@ -48,26 +49,45 @@ import java.util.ArrayList;
import java.util.HashMap;
import static org.apache.atlas.graph.GraphSandboxUtil.useLocalSolr;
-import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.*;
-import static org.testng.Assert.*;
+import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.getZipSource;
+import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.loadModelFromJson;
+import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.runImportWithNoParameters;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
@Guice(modules = TestModules.TestOnlyModule.class)
-public class RelationshipAttributesExtractorTest {
+public class RelationshipAttributesExtractorTest extends ExportImportTestBase {
- private static final String EXPORT_FULL = "full";
+ private static final String EXPORT_FULL = "full";
private static final String EXPORT_CONNECTED = "connected";
- private static final String QUALIFIED_NAME_DB = "db_test_1@02052019";
- private static final String QUALIFIED_NAME_TABLE_LINEAGE = "db_test_1.test_tbl_ctas_2@02052019";
+ private static final String ENTITIES_SUB_DIR = "entities";
+
+ private static final String QUALIFIED_NAME_DB = "db_test_1@02052019";
+ private static final String QUALIFIED_NAME_TABLE_LINEAGE = "db_test_1.test_tbl_ctas_2@02052019";
private static final String QUALIFIED_NAME_TABLE_NON_LINEAGE = "db_test_1.test_tbl_1@02052019";
- private static final String GUID_DB = "f0b72ab4-7452-4e42-ac74-2aee7728cce4";
- private static final String GUID_TABLE_1 = "4d5adf00-2c9b-4877-ad23-c41fd7319150";
- private static final String GUID_TABLE_2 = "8d0b834c-61ce-42d8-8f66-6fa51c36bccb";
+ private static final String GUID_DB = "f0b72ab4-7452-4e42-ac74-2aee7728cce4";
+ private static final String GUID_TABLE_1 = "4d5adf00-2c9b-4877-ad23-c41fd7319150";
+ private static final String GUID_TABLE_2 = "8d0b834c-61ce-42d8-8f66-6fa51c36bccb";
private static final String GUID_TABLE_CTAS_2 = "eaec545b-3ac7-4e1b-a497-bd4a2b6434a2";
private static final String GUID_HIVE_PROCESS = "bd3138b2-f29e-4226-b859-de25eaa1c18b";
+ private static final String DB1 = "db1";
+ private static final String DB2 = "db2";
+ private static final String TBL1 = "table1";
+ private static final String TBL2 = "table2";
+ private static final String HIVE_PROCESS = "table-lineage";
+ private static final String HIVE_COLUMN_LINEAGE = "column-lineage";
+
+ private static final String GUID_DB1 = "1c4e939e-ff6b-4229-92a4-b60c00deb547";
+ private static final String GUID_DB2 = "77c3bccf-ca3f-42e7-b2dd-f5a35f63eea6";
+ private static final String GUID_TBL1 = "3f6c02be-61e8-4dae-a7b8-cc37f289ce6e";
+ private static final String GUID_TBL2 = "b8cbc39f-4467-429b-a7fe-4ba2c28cceca";
+ private static final String GUID_PROCESS = "caf7f40a-b334-4f9e-9bf2-f24ce43db47f";
+ private static final String GUID_COLUMN_LINEAGE = "d4cf482b-423c-4c88-9bd1-701477ed6fd8";
+
@Inject
private ImportService importService;
@@ -80,6 +100,9 @@ public class RelationshipAttributesExtractorTest {
@Inject
private ExportService exportService;
+ @Inject
+ private AtlasEntityStoreV2 entityStore;
+
@BeforeClass
public void setup() throws IOException, AtlasBaseException {
loadBaseModel();
@@ -183,6 +206,22 @@ public class RelationshipAttributesExtractorTest {
verifyTableWithoutLineageSkipLineageConn(source);
}
+ @Test
+ public void interDbLineageConnectedExportTest() throws Exception {
+ setupInterDbLineageData();
+
+ ZipSource source = runExport(getExportRequestForHiveTable("db_1.table_1@cl1", EXPORT_CONNECTED, false));
+ assertInterDbLineageConnectedExport(source);
+ }
+
+ private void setupInterDbLineageData() {
+ RequestContext.get().setImportInProgress(true);
+ createEntities(entityStore, ENTITIES_SUB_DIR, new String[]{DB1, DB2, TBL1, TBL2, HIVE_PROCESS, HIVE_COLUMN_LINEAGE});
+ final String[] entityGuids = {GUID_DB1, GUID_DB2, GUID_TBL1, GUID_TBL2, GUID_PROCESS, GUID_COLUMN_LINEAGE};
+ verifyCreatedEntities(entityStore, entityGuids, 6);
+ RequestContext.get().setImportInProgress(false);
+ }
+
private void loadHiveModel() throws IOException, AtlasBaseException {
loadModelFromJson("1000-Hadoop/1030-hive_model.json", typeDefStore, typeRegistry);
}
@@ -334,6 +373,14 @@ public class RelationshipAttributesExtractorTest {
verifyExpectedEntities(getFileNames(zipSource), GUID_DB, GUID_TABLE_1);
}
+ private void assertInterDbLineageConnectedExport(ZipSource zipSource) {
+ assertNotNull(zipSource.getCreationOrder());
+ assertEquals(zipSource.getCreationOrder().size(), 5);
+
+ assertTrue(zipSource.getCreationOrder().contains(GUID_PROCESS));
+ verifyExpectedEntities(getFileNames(zipSource), GUID_DB1, GUID_DB2, GUID_TBL1, GUID_TBL2, GUID_PROCESS);
+ }
+
private void verifyExpectedEntities(List<String> fileNames, String... guids){
assertEquals(fileNames.size(), guids.length);
for (String guid : guids) {
diff --git a/repository/src/test/resources/json/entities/column-lineage.json b/repository/src/test/resources/json/entities/column-lineage.json
new file mode 100644
index 0000000..a98122d
--- /dev/null
+++ b/repository/src/test/resources/json/entities/column-lineage.json
@@ -0,0 +1,89 @@
+{
+ "entity": {
+ "attributes": {
+ "depenendencyType": "SIMPLE",
+ "description": null,
+ "expression": null,
+ "inputs": [
+ {
+ "guid": "9f6a3c7d-2b93-485b-b7cd-47775c50d6fb",
+ "typeName": "hive_column",
+ "uniqueAttributes": {
+ "qualifiedName": "db_1.table_1.id@cl1"
+ }
+ }
+ ],
+ "name": "db_2.table_2@cl1:1569932823000:id",
+ "outputs": [
+ {
+ "guid": "d3ba7485-6b77-4f32-b3c9-e3d45d0949c0",
+ "typeName": "hive_column",
+ "uniqueAttributes": {
+ "qualifiedName": "db_2.table_2.id@cl1"
+ }
+ }
+ ],
+ "owner": null,
+ "qualifiedName": "db_2.table_2@cl1:1569932823000:id",
+ "query": {
+ "guid": "9827e05b-7f56-4eeb-8ce5-2ebc2f1bd001",
+ "typeName": "hive_process"
+ },
+ "replicatedFrom": null,
+ "replicatedTo": null
+ },
+ "createTime": 1570087822414,
+ "createdBy": "hive",
+ "guid": "d4cf482b-423c-4c88-9bd1-701477ed6fd8",
+ "isIncomplete": false,
+ "relationshipAttributes": {
+ "inputs": [
+ {
+ "displayText": "id",
+ "entityStatus": "ACTIVE",
+ "guid": "9f6a3c7d-2b93-485b-b7cd-47775c50d6fb",
+ "relationshipAttributes": {
+ "typeName": "dataset_process_inputs"
+ },
+ "relationshipGuid": "4cce3e95-f780-4d8f-ab41-74d64dc91e23",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "dataset_process_inputs",
+ "typeName": "hive_column"
+ }
+ ],
+ "meanings": [],
+ "outputs": [
+ {
+ "displayText": "id",
+ "entityStatus": "ACTIVE",
+ "guid": "d3ba7485-6b77-4f32-b3c9-e3d45d0949c0",
+ "relationshipAttributes": {
+ "typeName": "process_dataset_outputs"
+ },
+ "relationshipGuid": "b0054651-7003-4de9-a9f6-49ef7c0b0616",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "process_dataset_outputs",
+ "typeName": "hive_column"
+ }
+ ],
+ "query": {
+ "displayText": "create table db_2.table_2 as select * from db_1.table_1",
+ "entityStatus": "ACTIVE",
+ "guid": "9827e05b-7f56-4eeb-8ce5-2ebc2f1bd001",
+ "relationshipAttributes": {
+ "typeName": "hive_process_column_lineage"
+ },
+ "relationshipGuid": "ad56ea7f-8556-421d-8979-7eaf79dd0050",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_process_column_lineage",
+ "typeName": "hive_process"
+ }
+ },
+ "status": "ACTIVE",
+ "typeName": "hive_column_lineage",
+ "updateTime": 1570087822414,
+ "updatedBy": "hive",
+ "version": 0
+ },
+ "referredEntities": {}
+}
\ No newline at end of file
diff --git a/repository/src/test/resources/json/entities/db1.json b/repository/src/test/resources/json/entities/db1.json
new file mode 100644
index 0000000..c1a670f
--- /dev/null
+++ b/repository/src/test/resources/json/entities/db1.json
@@ -0,0 +1,27 @@
+{
+ "entity": {
+ "attributes": {
+ "clusterName": "cl1",
+ "description": null,
+ "location": "hdfs://localhost:8020/warehouse/tablespace/managed/hive/db_1.db",
+ "name": "db_1",
+ "owner": "hive",
+ "ownerType": "USER",
+ "parameters": {},
+ "qualifiedName": "db_1@cl1",
+ "replicatedFrom": null,
+ "replicatedTo": null
+ },
+ "createTime": 1570444587942,
+ "createdBy": "hive",
+ "guid": "1c4e939e-ff6b-4229-92a4-b60c00deb547",
+ "isIncomplete": false,
+ "relationshipAttributes": {},
+ "status": "ACTIVE",
+ "typeName": "hive_db",
+ "updateTime": 1570444591401,
+ "updatedBy": "hive",
+ "version": 0
+ },
+ "referredEntities": {}
+}
\ No newline at end of file
diff --git a/repository/src/test/resources/json/entities/db2.json b/repository/src/test/resources/json/entities/db2.json
new file mode 100644
index 0000000..d71c707
--- /dev/null
+++ b/repository/src/test/resources/json/entities/db2.json
@@ -0,0 +1,27 @@
+{
+ "entity": {
+ "attributes": {
+ "clusterName": "cl1",
+ "description": null,
+ "location": "hdfs://localhost:8020/warehouse/tablespace/managed/hive/db_2.db",
+ "name": "db_2",
+ "owner": "hive",
+ "ownerType": "USER",
+ "parameters": {},
+ "qualifiedName": "db_2@cl1",
+ "replicatedFrom": null,
+ "replicatedTo": null
+ },
+ "createTime": 1570444593617,
+ "createdBy": "hive",
+ "guid": "77c3bccf-ca3f-42e7-b2dd-f5a35f63eea6",
+ "isIncomplete": false,
+ "relationshipAttributes": {},
+ "status": "ACTIVE",
+ "typeName": "hive_db",
+ "updateTime": 1570444602640,
+ "updatedBy": "hive",
+ "version": 0
+ },
+ "referredEntities": {}
+}
\ No newline at end of file
diff --git a/repository/src/test/resources/json/entities/table-lineage.json b/repository/src/test/resources/json/entities/table-lineage.json
new file mode 100644
index 0000000..40834bc
--- /dev/null
+++ b/repository/src/test/resources/json/entities/table-lineage.json
@@ -0,0 +1,240 @@
+{
+ "entity": {
+ "attributes": {
+ "clusterName": null,
+ "description": null,
+ "endTime": 0,
+ "inputs": [
+ {
+ "guid": "3f6c02be-61e8-4dae-a7b8-cc37f289ce6e",
+ "typeName": "hive_table",
+ "uniqueAttributes": {
+ "qualifiedName": "db_1.table_1@cl1"
+ }
+ }
+ ],
+ "name": "create table db_2.table_2 as select * from db_1.table_1",
+ "operationType": "CREATETABLE_AS_SELECT",
+ "outputs": [
+ {
+ "guid": "b8cbc39f-4467-429b-a7fe-4ba2c28cceca",
+ "typeName": "hive_table",
+ "uniqueAttributes": {
+ "qualifiedName": "db_2.table_2@cl1"
+ }
+ }
+ ],
+ "owner": null,
+ "qualifiedName": "db_2.table_2@cl1:1570444602000",
+ "queryGraph": null,
+ "queryId": "",
+ "queryPlan": "Not Supported",
+ "queryText": "",
+ "recentQueries": [
+ "create table db_2.table_2 as select * from db_1.table_1"
+ ],
+ "replicatedFrom": null,
+ "replicatedTo": null,
+ "startTime": 0,
+ "userName": ""
+ },
+ "createTime": 1570444602640,
+ "createdBy": "hive",
+ "guid": "caf7f40a-b334-4f9e-9bf2-f24ce43db47f",
+ "isIncomplete": false,
+ "relationshipAttributes": {
+ "columnLineages": [
+ {
+ "displayText": "db_2.table_2@cl1:1570444602000:id",
+ "entityStatus": "ACTIVE",
+ "guid": "52b4662b-999d-4f63-a977-89bc9ea8ff46",
+ "relationshipAttributes": {
+ "typeName": "hive_process_column_lineage"
+ },
+ "relationshipGuid": "26c9dec4-f3c9-4a17-82df-7156d68568f4",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_process_column_lineage",
+ "typeName": "hive_column_lineage"
+ }
+ ],
+ "inputs": [
+ {
+ "displayText": "table_1",
+ "entityStatus": "ACTIVE",
+ "guid": "3f6c02be-61e8-4dae-a7b8-cc37f289ce6e",
+ "relationshipAttributes": {
+ "typeName": "dataset_process_inputs"
+ },
+ "relationshipGuid": "69a92de6-2ff0-48bf-821a-5be9b3605aea",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "dataset_process_inputs",
+ "typeName": "hive_table"
+ }
+ ],
+ "meanings": [],
+ "outputs": [
+ {
+ "displayText": "table_2",
+ "entityStatus": "ACTIVE",
+ "guid": "b8cbc39f-4467-429b-a7fe-4ba2c28cceca",
+ "relationshipAttributes": {
+ "typeName": "process_dataset_outputs"
+ },
+ "relationshipGuid": "d38c4409-65c4-4ba4-99bb-c655d8be8197",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "process_dataset_outputs",
+ "typeName": "hive_table"
+ }
+ ],
+ "processExecutions": [
+ {
+ "displayText": "create table db_2.table_2 as select * from db_1.table_1:1570444589240",
+ "entityStatus": "ACTIVE",
+ "guid": "10a024a4-cb38-450b-a601-0106ffb758ad",
+ "relationshipAttributes": {
+ "typeName": "hive_process_process_executions"
+ },
+ "relationshipGuid": "e40f8127-3415-4c3e-ac7e-424d3349f116",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_process_process_executions",
+ "typeName": "hive_process_execution"
+ }
+ ]
+ },
+ "status": "ACTIVE",
+ "typeName": "hive_process",
+ "updateTime": 1570444602640,
+ "updatedBy": "hive",
+ "version": 0
+ },
+ "referredEntities": {
+ "10a024a4-cb38-450b-a601-0106ffb758ad": {
+ "attributes": {
+ "description": null,
+ "endTime": 1570444602623,
+ "hostName": "nb-atl-2409-2.openstacklocal",
+ "name": "create table db_2.table_2 as select * from db_1.table_1:1570444589240",
+ "owner": null,
+ "qualifiedName": "db_2.table_2@cl1:1570444602000:1570444589240:1570444602623",
+ "queryGraph": null,
+ "queryId": "hive_20191007103629_e6f69952-08e4-4b0c-a3b8-faf42d819ff3",
+ "queryPlan": "Not Supported",
+ "queryText": "create table db_2.table_2 as select * from db_1.table_1",
+ "replicatedFrom": null,
+ "replicatedTo": null,
+ "startTime": 1570444589240,
+ "userName": "hive"
+ },
+ "createTime": 1570444602640,
+ "createdBy": "hive",
+ "guid": "10a024a4-cb38-450b-a601-0106ffb758ad",
+ "isIncomplete": false,
+ "relationshipAttributes": {
+ "meanings": [],
+ "process": {
+ "displayText": "create table db_2.table_2 as select * from db_1.table_1",
+ "entityStatus": "ACTIVE",
+ "guid": "caf7f40a-b334-4f9e-9bf2-f24ce43db47f",
+ "relationshipAttributes": {
+ "typeName": "hive_process_process_executions"
+ },
+ "relationshipGuid": "e40f8127-3415-4c3e-ac7e-424d3349f116",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_process_process_executions",
+ "typeName": "hive_process"
+ }
+ },
+ "status": "ACTIVE",
+ "typeName": "hive_process_execution",
+ "updateTime": 1570444602640,
+ "updatedBy": "hive",
+ "version": 0
+ },
+ "52b4662b-999d-4f63-a977-89bc9ea8ff46": {
+ "attributes": {
+ "depenendencyType": "SIMPLE",
+ "description": null,
+ "expression": null,
+ "inputs": [
+ {
+ "guid": "9f6a3c7d-2b93-485b-b7cd-47775c50d6fb",
+ "typeName": "hive_column",
+ "uniqueAttributes": {
+ "qualifiedName": "db_1.table_1.id@cl1"
+ }
+ }
+ ],
+ "name": "db_2.table_2@cl1:1570444602000:id",
+ "outputs": [
+ {
+ "guid": "d3ba7485-6b77-4f32-b3c9-e3d45d0949c0",
+ "typeName": "hive_column",
+ "uniqueAttributes": {
+ "qualifiedName": "db_2.table_2.id@cl1"
+ }
+ }
+ ],
+ "owner": null,
+ "qualifiedName": "db_2.table_2@cl1:1570444602000:id",
+ "query": {
+ "guid": "caf7f40a-b334-4f9e-9bf2-f24ce43db47f",
+ "typeName": "hive_process"
+ },
+ "replicatedFrom": null,
+ "replicatedTo": null
+ },
+ "createTime": 1570444602640,
+ "createdBy": "hive",
+ "guid": "52b4662b-999d-4f63-a977-89bc9ea8ff46",
+ "isIncomplete": false,
+ "relationshipAttributes": {
+ "inputs": [
+ {
+ "displayText": "id",
+ "entityStatus": "ACTIVE",
+ "guid": "9f6a3c7d-2b93-485b-b7cd-47775c50d6fb",
+ "relationshipAttributes": {
+ "typeName": "dataset_process_inputs"
+ },
+ "relationshipGuid": "82b2e578-e707-4c62-9e70-e94bfbe5caf2",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "dataset_process_inputs",
+ "typeName": "hive_column"
+ }
+ ],
+ "meanings": [],
+ "outputs": [
+ {
+ "displayText": "id",
+ "entityStatus": "ACTIVE",
+ "guid": "d3ba7485-6b77-4f32-b3c9-e3d45d0949c0",
+ "relationshipAttributes": {
+ "typeName": "process_dataset_outputs"
+ },
+ "relationshipGuid": "4c42c8a1-a0c5-4125-ace3-8fc59312b6dd",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "process_dataset_outputs",
+ "typeName": "hive_column"
+ }
+ ],
+ "query": {
+ "displayText": "create table db_2.table_2 as select * from db_1.table_1",
+ "entityStatus": "ACTIVE",
+ "guid": "caf7f40a-b334-4f9e-9bf2-f24ce43db47f",
+ "relationshipAttributes": {
+ "typeName": "hive_process_column_lineage"
+ },
+ "relationshipGuid": "26c9dec4-f3c9-4a17-82df-7156d68568f4",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_process_column_lineage",
+ "typeName": "hive_process"
+ }
+ },
+ "status": "ACTIVE",
+ "typeName": "hive_column_lineage",
+ "updateTime": 1570444602640,
+ "updatedBy": "hive",
+ "version": 0
+ }
+ }
+}
\ No newline at end of file
diff --git a/repository/src/test/resources/json/entities/table1.json b/repository/src/test/resources/json/entities/table1.json
new file mode 100644
index 0000000..a1411c1
--- /dev/null
+++ b/repository/src/test/resources/json/entities/table1.json
@@ -0,0 +1,276 @@
+{
+ "entity": {
+ "attributes": {
+ "aliases": null,
+ "columns": [
+ {
+ "guid": "9f6a3c7d-2b93-485b-b7cd-47775c50d6fb",
+ "typeName": "hive_column"
+ }
+ ],
+ "comment": null,
+ "createTime": 1570444588000,
+ "db": {
+ "guid": "1c4e939e-ff6b-4229-92a4-b60c00deb547",
+ "typeName": "hive_db"
+ },
+ "description": null,
+ "lastAccessTime": 1570444588000,
+ "name": "table_1",
+ "owner": "hive",
+ "parameters": {
+ "COLUMN_STATS_ACCURATE": "{\"BASIC_STATS\":\"true\",\"COLUMN_STATS\":{\"id\":\"true\"}}",
+ "bucketing_version": "2",
+ "numFiles": "0",
+ "numRows": "0",
+ "rawDataSize": "0",
+ "totalSize": "0",
+ "transactional": "true",
+ "transactional_properties": "default",
+ "transient_lastDdlTime": "1570444588"
+ },
+ "partitionKeys": [],
+ "qualifiedName": "db_1.table_1@cl1",
+ "replicatedFrom": null,
+ "replicatedTo": null,
+ "retention": 0,
+ "sd": {
+ "guid": "52e8353a-36e5-475f-8626-908a07230dfd",
+ "typeName": "hive_storagedesc"
+ },
+ "tableType": "MANAGED_TABLE",
+ "temporary": false,
+ "viewExpandedText": null,
+ "viewOriginalText": null
+ },
+ "createTime": 1570444591401,
+ "createdBy": "hive",
+ "guid": "3f6c02be-61e8-4dae-a7b8-cc37f289ce6e",
+ "isIncomplete": false,
+ "relationshipAttributes": {
+ "columns": [
+ {
+ "displayText": "id",
+ "entityStatus": "ACTIVE",
+ "guid": "9f6a3c7d-2b93-485b-b7cd-47775c50d6fb",
+ "relationshipAttributes": {
+ "typeName": "hive_table_columns"
+ },
+ "relationshipGuid": "f9e10d9a-0021-485e-8f49-b015b2223faf",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_table_columns",
+ "typeName": "hive_column"
+ }
+ ],
+ "db": {
+ "displayText": "db_1",
+ "entityStatus": "ACTIVE",
+ "guid": "1c4e939e-ff6b-4229-92a4-b60c00deb547",
+ "relationshipAttributes": {
+ "typeName": "hive_table_db"
+ },
+ "relationshipGuid": "1108f124-2dc9-4406-b063-a9751205308d",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_table_db",
+ "typeName": "hive_db"
+ },
+ "ddlQueries": [
+ {
+ "displayText": "db_1.table_1@cl1:1570444588003",
+ "entityStatus": "ACTIVE",
+ "guid": "942802d2-0310-40f3-ab0a-8e35a6e43b53",
+ "relationshipAttributes": {
+ "typeName": "hive_table_ddl_queries"
+ },
+ "relationshipGuid": "30b3a990-c362-4044-9e78-5a7be7f35307",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_table_ddl_queries",
+ "typeName": "hive_table_ddl"
+ }
+ ],
+ "inputToProcesses": [
+ {
+ "displayText": "create table db_2.table_2 as select * from db_1.table_1",
+ "entityStatus": "ACTIVE",
+ "guid": "caf7f40a-b334-4f9e-9bf2-f24ce43db47f",
+ "relationshipAttributes": {
+ "typeName": "dataset_process_inputs"
+ },
+ "relationshipGuid": "69a92de6-2ff0-48bf-821a-5be9b3605aea",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "dataset_process_inputs",
+ "typeName": "hive_process"
+ }
+ ],
+ "meanings": [],
+ "outputFromProcesses": [],
+ "partitionKeys": [],
+ "schema": [],
+ "sd": {
+ "displayText": "db_1.table_1@cl1_storage",
+ "entityStatus": "ACTIVE",
+ "guid": "52e8353a-36e5-475f-8626-908a07230dfd",
+ "relationshipAttributes": {
+ "typeName": "hive_table_storagedesc"
+ },
+ "relationshipGuid": "5a40781f-edf1-4854-ba53-d94829039f4a",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_table_storagedesc",
+ "typeName": "hive_storagedesc"
+ }
+ },
+ "status": "ACTIVE",
+ "typeName": "hive_table",
+ "updateTime": 1570444602640,
+ "updatedBy": "hive",
+ "version": 0
+ },
+ "referredEntities": {
+ "52e8353a-36e5-475f-8626-908a07230dfd": {
+ "attributes": {
+ "bucketCols": null,
+ "compressed": false,
+ "inputFormat": "org.apache.hadoop.hive.ql.io.orc.OrcInputFormat",
+ "location": "hdfs://localhost:8020/warehouse/tablespace/managed/hive/db_1.db/table_1",
+ "numBuckets": -1,
+ "outputFormat": "org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat",
+ "parameters": {},
+ "qualifiedName": "db_1.table_1@cl1_storage",
+ "replicatedFrom": null,
+ "replicatedTo": null,
+ "serdeInfo": {
+ "attributes": {
+ "name": null,
+ "parameters": {
+ "serialization.format": "1"
+ },
+ "serializationLib": "org.apache.hadoop.hive.ql.io.orc.OrcSerde"
+ },
+ "typeName": "hive_serde"
+ },
+ "sortCols": [],
+ "storedAsSubDirectories": false,
+ "table": {
+ "guid": "3f6c02be-61e8-4dae-a7b8-cc37f289ce6e",
+ "typeName": "hive_table"
+ }
+ },
+ "createTime": 1570444591401,
+ "createdBy": "hive",
+ "guid": "52e8353a-36e5-475f-8626-908a07230dfd",
+ "isIncomplete": false,
+ "relationshipAttributes": {
+ "meanings": [],
+ "table": {
+ "displayText": "table_1",
+ "entityStatus": "ACTIVE",
+ "guid": "3f6c02be-61e8-4dae-a7b8-cc37f289ce6e",
+ "relationshipAttributes": {
+ "typeName": "hive_table_storagedesc"
+ },
+ "relationshipGuid": "5a40781f-edf1-4854-ba53-d94829039f4a",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_table_storagedesc",
+ "typeName": "hive_table"
+ }
+ },
+ "status": "ACTIVE",
+ "typeName": "hive_storagedesc",
+ "updateTime": 1570444591401,
+ "updatedBy": "hive",
+ "version": 0
+ },
+ "942802d2-0310-40f3-ab0a-8e35a6e43b53": {
+ "attributes": {
+ "execTime": 1570444588003,
+ "qualifiedName": "db_1.table_1@cl1:1570444588003",
+ "queryText": "create table db_1.table_1(id int)",
+ "replicatedFrom": null,
+ "replicatedTo": null,
+ "serviceType": "hive",
+ "userName": "hive"
+ },
+ "createTime": 1570444591401,
+ "createdBy": "hive",
+ "guid": "942802d2-0310-40f3-ab0a-8e35a6e43b53",
+ "isIncomplete": false,
+ "relationshipAttributes": {
+ "meanings": [],
+ "table": {
+ "displayText": "table_1",
+ "entityStatus": "ACTIVE",
+ "guid": "3f6c02be-61e8-4dae-a7b8-cc37f289ce6e",
+ "relationshipAttributes": {
+ "typeName": "hive_table_ddl_queries"
+ },
+ "relationshipGuid": "30b3a990-c362-4044-9e78-5a7be7f35307",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_table_ddl_queries",
+ "typeName": "hive_table"
+ }
+ },
+ "status": "ACTIVE",
+ "typeName": "hive_table_ddl",
+ "updateTime": 1570444591401,
+ "updatedBy": "hive",
+ "version": 0
+ },
+ "9f6a3c7d-2b93-485b-b7cd-47775c50d6fb": {
+ "attributes": {
+ "comment": null,
+ "description": null,
+ "name": "id",
+ "owner": "hive",
+ "position": 0,
+ "qualifiedName": "db_1.table_1.id@cl1",
+ "replicatedFrom": null,
+ "replicatedTo": null,
+ "table": {
+ "guid": "3f6c02be-61e8-4dae-a7b8-cc37f289ce6e",
+ "typeName": "hive_table"
+ },
+ "type": "int"
+ },
+ "createTime": 1570444591401,
+ "createdBy": "hive",
+ "guid": "9f6a3c7d-2b93-485b-b7cd-47775c50d6fb",
+ "isIncomplete": false,
+ "relationshipAttributes": {
+ "inputToProcesses": [
+ {
+ "displayText": "db_2.table_2@cl1:1570444602000:id",
+ "entityStatus": "ACTIVE",
+ "guid": "52b4662b-999d-4f63-a977-89bc9ea8ff46",
+ "relationshipAttributes": {
+ "typeName": "dataset_process_inputs"
+ },
+ "relationshipGuid": "82b2e578-e707-4c62-9e70-e94bfbe5caf2",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "dataset_process_inputs",
+ "typeName": "hive_column_lineage"
+ }
+ ],
+ "meanings": [],
+ "outputFromProcesses": [],
+ "schema": [],
+ "table": {
+ "displayText": "table_1",
+ "entityStatus": "ACTIVE",
+ "guid": "3f6c02be-61e8-4dae-a7b8-cc37f289ce6e",
+ "relationshipAttributes": {
+ "typeName": "hive_table_columns"
+ },
+ "relationshipGuid": "f9e10d9a-0021-485e-8f49-b015b2223faf",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_table_columns",
+ "typeName": "hive_table"
+ }
+ },
+ "status": "ACTIVE",
+ "typeName": "hive_column",
+ "updateTime": 1570444602640,
+ "updatedBy": "hive",
+ "version": 0
+ }
+ }
+}
\ No newline at end of file
diff --git a/repository/src/test/resources/json/entities/table2.json b/repository/src/test/resources/json/entities/table2.json
new file mode 100644
index 0000000..5374d86
--- /dev/null
+++ b/repository/src/test/resources/json/entities/table2.json
@@ -0,0 +1,276 @@
+{
+ "entity": {
+ "attributes": {
+ "aliases": null,
+ "columns": [
+ {
+ "guid": "d3ba7485-6b77-4f32-b3c9-e3d45d0949c0",
+ "typeName": "hive_column"
+ }
+ ],
+ "comment": null,
+ "createTime": 1570444602000,
+ "db": {
+ "guid": "77c3bccf-ca3f-42e7-b2dd-f5a35f63eea6",
+ "typeName": "hive_db"
+ },
+ "description": null,
+ "lastAccessTime": 1570444602000,
+ "name": "table_2",
+ "owner": "hive",
+ "parameters": {
+ "COLUMN_STATS_ACCURATE": "{\"BASIC_STATS\":\"true\",\"COLUMN_STATS\":{\"id\":\"true\"}}",
+ "bucketing_version": "2",
+ "numFiles": "0",
+ "numRows": "0",
+ "rawDataSize": "0",
+ "totalSize": "0",
+ "transactional": "true",
+ "transactional_properties": "default",
+ "transient_lastDdlTime": "1570444602"
+ },
+ "partitionKeys": [],
+ "qualifiedName": "db_2.table_2@cl1",
+ "replicatedFrom": null,
+ "replicatedTo": null,
+ "retention": 0,
+ "sd": {
+ "guid": "403607a6-2092-43a8-8816-9d128592004f",
+ "typeName": "hive_storagedesc"
+ },
+ "tableType": "MANAGED_TABLE",
+ "temporary": false,
+ "viewExpandedText": null,
+ "viewOriginalText": null
+ },
+ "createTime": 1570444602640,
+ "createdBy": "hive",
+ "guid": "b8cbc39f-4467-429b-a7fe-4ba2c28cceca",
+ "isIncomplete": false,
+ "relationshipAttributes": {
+ "columns": [
+ {
+ "displayText": "id",
+ "entityStatus": "ACTIVE",
+ "guid": "d3ba7485-6b77-4f32-b3c9-e3d45d0949c0",
+ "relationshipAttributes": {
+ "typeName": "hive_table_columns"
+ },
+ "relationshipGuid": "a7c41412-35c9-408c-894b-aa58005b46ae",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_table_columns",
+ "typeName": "hive_column"
+ }
+ ],
+ "db": {
+ "displayText": "db_2",
+ "entityStatus": "ACTIVE",
+ "guid": "77c3bccf-ca3f-42e7-b2dd-f5a35f63eea6",
+ "relationshipAttributes": {
+ "typeName": "hive_table_db"
+ },
+ "relationshipGuid": "a6cf6e79-2614-4284-b3e8-48d1d261c2e9",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_table_db",
+ "typeName": "hive_db"
+ },
+ "ddlQueries": [
+ {
+ "displayText": "db_2.table_2@cl1:1570444589240",
+ "entityStatus": "ACTIVE",
+ "guid": "eb5d6056-f5a3-46c4-948f-dae2f9481076",
+ "relationshipAttributes": {
+ "typeName": "hive_table_ddl_queries"
+ },
+ "relationshipGuid": "e807ef3b-d7f0-4e1f-b358-57525790e376",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_table_ddl_queries",
+ "typeName": "hive_table_ddl"
+ }
+ ],
+ "inputToProcesses": [],
+ "meanings": [],
+ "outputFromProcesses": [
+ {
+ "displayText": "create table db_2.table_2 as select * from db_1.table_1",
+ "entityStatus": "ACTIVE",
+ "guid": "caf7f40a-b334-4f9e-9bf2-f24ce43db47f",
+ "relationshipAttributes": {
+ "typeName": "process_dataset_outputs"
+ },
+ "relationshipGuid": "d38c4409-65c4-4ba4-99bb-c655d8be8197",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "process_dataset_outputs",
+ "typeName": "hive_process"
+ }
+ ],
+ "partitionKeys": [],
+ "schema": [],
+ "sd": {
+ "displayText": "db_2.table_2@cl1_storage",
+ "entityStatus": "ACTIVE",
+ "guid": "403607a6-2092-43a8-8816-9d128592004f",
+ "relationshipAttributes": {
+ "typeName": "hive_table_storagedesc"
+ },
+ "relationshipGuid": "66633928-9605-428a-8b1e-270ebbaec62b",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_table_storagedesc",
+ "typeName": "hive_storagedesc"
+ }
+ },
+ "status": "ACTIVE",
+ "typeName": "hive_table",
+ "updateTime": 1570444602640,
+ "updatedBy": "hive",
+ "version": 0
+ },
+ "referredEntities": {
+ "403607a6-2092-43a8-8816-9d128592004f": {
+ "attributes": {
+ "bucketCols": null,
+ "compressed": false,
+ "inputFormat": "org.apache.hadoop.hive.ql.io.orc.OrcInputFormat",
+ "location": "hdfs://localhost:8020/warehouse/tablespace/managed/hive/db_2.db/table_2",
+ "numBuckets": -1,
+ "outputFormat": "org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat",
+ "parameters": {},
+ "qualifiedName": "db_2.table_2@cl1_storage",
+ "replicatedFrom": null,
+ "replicatedTo": null,
+ "serdeInfo": {
+ "attributes": {
+ "name": null,
+ "parameters": {
+ "serialization.format": "1"
+ },
+ "serializationLib": "org.apache.hadoop.hive.ql.io.orc.OrcSerde"
+ },
+ "typeName": "hive_serde"
+ },
+ "sortCols": [],
+ "storedAsSubDirectories": false,
+ "table": {
+ "guid": "b8cbc39f-4467-429b-a7fe-4ba2c28cceca",
+ "typeName": "hive_table"
+ }
+ },
+ "createTime": 1570444602640,
+ "createdBy": "hive",
+ "guid": "403607a6-2092-43a8-8816-9d128592004f",
+ "isIncomplete": false,
+ "relationshipAttributes": {
+ "meanings": [],
+ "table": {
+ "displayText": "table_2",
+ "entityStatus": "ACTIVE",
+ "guid": "b8cbc39f-4467-429b-a7fe-4ba2c28cceca",
+ "relationshipAttributes": {
+ "typeName": "hive_table_storagedesc"
+ },
+ "relationshipGuid": "66633928-9605-428a-8b1e-270ebbaec62b",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_table_storagedesc",
+ "typeName": "hive_table"
+ }
+ },
+ "status": "ACTIVE",
+ "typeName": "hive_storagedesc",
+ "updateTime": 1570444602640,
+ "updatedBy": "hive",
+ "version": 0
+ },
+ "d3ba7485-6b77-4f32-b3c9-e3d45d0949c0": {
+ "attributes": {
+ "comment": null,
+ "description": null,
+ "name": "id",
+ "owner": "hive",
+ "position": 0,
+ "qualifiedName": "db_2.table_2.id@cl1",
+ "replicatedFrom": null,
+ "replicatedTo": null,
+ "table": {
+ "guid": "b8cbc39f-4467-429b-a7fe-4ba2c28cceca",
+ "typeName": "hive_table"
+ },
+ "type": "int"
+ },
+ "createTime": 1570444602640,
+ "createdBy": "hive",
+ "guid": "d3ba7485-6b77-4f32-b3c9-e3d45d0949c0",
+ "isIncomplete": false,
+ "relationshipAttributes": {
+ "inputToProcesses": [],
+ "meanings": [],
+ "outputFromProcesses": [
+ {
+ "displayText": "db_2.table_2@cl1:1570444602000:id",
+ "entityStatus": "ACTIVE",
+ "guid": "52b4662b-999d-4f63-a977-89bc9ea8ff46",
+ "relationshipAttributes": {
+ "typeName": "process_dataset_outputs"
+ },
+ "relationshipGuid": "4c42c8a1-a0c5-4125-ace3-8fc59312b6dd",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "process_dataset_outputs",
+ "typeName": "hive_column_lineage"
+ }
+ ],
+ "schema": [],
+ "table": {
+ "displayText": "table_2",
+ "entityStatus": "ACTIVE",
+ "guid": "b8cbc39f-4467-429b-a7fe-4ba2c28cceca",
+ "relationshipAttributes": {
+ "typeName": "hive_table_columns"
+ },
+ "relationshipGuid": "a7c41412-35c9-408c-894b-aa58005b46ae",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_table_columns",
+ "typeName": "hive_table"
+ }
+ },
+ "status": "ACTIVE",
+ "typeName": "hive_column",
+ "updateTime": 1570444602640,
+ "updatedBy": "hive",
+ "version": 0
+ },
+ "eb5d6056-f5a3-46c4-948f-dae2f9481076": {
+ "attributes": {
+ "execTime": 1570444589240,
+ "qualifiedName": "db_2.table_2@cl1:1570444589240",
+ "queryText": "create table db_2.table_2 as select * from db_1.table_1",
+ "replicatedFrom": null,
+ "replicatedTo": null,
+ "serviceType": "hive",
+ "userName": "hive"
+ },
+ "createTime": 1570444602640,
+ "createdBy": "hive",
+ "guid": "eb5d6056-f5a3-46c4-948f-dae2f9481076",
+ "isIncomplete": false,
+ "relationshipAttributes": {
+ "meanings": [],
+ "table": {
+ "displayText": "table_2",
+ "entityStatus": "ACTIVE",
+ "guid": "b8cbc39f-4467-429b-a7fe-4ba2c28cceca",
+ "relationshipAttributes": {
+ "typeName": "hive_table_ddl_queries"
+ },
+ "relationshipGuid": "e807ef3b-d7f0-4e1f-b358-57525790e376",
+ "relationshipStatus": "ACTIVE",
+ "relationshipType": "hive_table_ddl_queries",
+ "typeName": "hive_table"
+ }
+ },
+ "status": "ACTIVE",
+ "typeName": "hive_table_ddl",
+ "updateTime": 1570444602640,
+ "updatedBy": "hive",
+ "version": 0
+ }
+ }
+}
\ No newline at end of file