You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@atlas.apache.org by ap...@apache.org on 2017/11/09 22:39:25 UTC
atlas git commit: ATLAS-2251: Lineage/DataSetLineage (schema fixes)
Repository: atlas
Updated Branches:
refs/heads/ATLAS-2251 aa85a8ad4 -> fe4535979
ATLAS-2251: Lineage/DataSetLineage (schema fixes)
Project: http://git-wip-us.apache.org/repos/asf/atlas/repo
Commit: http://git-wip-us.apache.org/repos/asf/atlas/commit/fe453597
Tree: http://git-wip-us.apache.org/repos/asf/atlas/tree/fe453597
Diff: http://git-wip-us.apache.org/repos/asf/atlas/diff/fe453597
Branch: refs/heads/ATLAS-2251
Commit: fe4535979ea83c02fa5c29b038928ee8bc9b457b
Parents: aa85a8a
Author: apoorvnaik <ap...@apache.org>
Authored: Thu Nov 9 14:39:14 2017 -0800
Committer: apoorvnaik <ap...@apache.org>
Committed: Thu Nov 9 14:39:14 2017 -0800
----------------------------------------------------------------------
intg/pom.xml | 8 ++
.../org/apache/atlas/type/AtlasTypeUtil.java | 137 ++++++++++++++++++-
.../atlas/v1/model/lineage/LineageResponse.java | 2 +-
.../atlas/v1/model/lineage/SchemaResponse.java | 104 ++++++++++++++
.../v1/model/typedef/AttributeDefinition.java | 22 ++-
.../v1/model/typedef/ClassTypeDefinition.java | 8 +-
.../atlas/discovery/AtlasLineageService.java | 5 +-
.../atlas/discovery/EntityLineageService.java | 91 ++++++------
.../web/resources/DataSetLineageResource.java | 20 +--
.../atlas/web/resources/LineageResource.java | 16 +--
10 files changed, 342 insertions(+), 71 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/atlas/blob/fe453597/intg/pom.xml
----------------------------------------------------------------------
diff --git a/intg/pom.xml b/intg/pom.xml
index 071e6c7..22895df 100644
--- a/intg/pom.xml
+++ b/intg/pom.xml
@@ -86,6 +86,14 @@
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <source>1.8</source>
+ <target>1.8</target>
+ </configuration>
+ </plugin>
</plugins>
</build>
</project>
http://git-wip-us.apache.org/repos/asf/atlas/blob/fe453597/intg/src/main/java/org/apache/atlas/type/AtlasTypeUtil.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/type/AtlasTypeUtil.java b/intg/src/main/java/org/apache/atlas/type/AtlasTypeUtil.java
index 0652855..5cb74de 100644
--- a/intg/src/main/java/org/apache/atlas/type/AtlasTypeUtil.java
+++ b/intg/src/main/java/org/apache/atlas/type/AtlasTypeUtil.java
@@ -20,13 +20,25 @@ package org.apache.atlas.type;
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.typedef.*;
+import org.apache.atlas.model.instance.AtlasStruct;
+import org.apache.atlas.model.typedef.AtlasBaseTypeDef;
+import org.apache.atlas.model.typedef.AtlasClassificationDef;
+import org.apache.atlas.model.typedef.AtlasEntityDef;
+import org.apache.atlas.model.typedef.AtlasEnumDef;
import org.apache.atlas.model.typedef.AtlasEnumDef.AtlasEnumElementDef;
+import org.apache.atlas.model.typedef.AtlasRelationshipDef;
import org.apache.atlas.model.typedef.AtlasRelationshipDef.PropagateTags;
import org.apache.atlas.model.typedef.AtlasRelationshipDef.RelationshipCategory;
+import org.apache.atlas.model.typedef.AtlasRelationshipEndDef;
+import org.apache.atlas.model.typedef.AtlasStructDef;
import org.apache.atlas.model.typedef.AtlasStructDef.AtlasAttributeDef;
import org.apache.atlas.model.typedef.AtlasStructDef.AtlasAttributeDef.Cardinality;
import org.apache.atlas.model.typedef.AtlasStructDef.AtlasConstraintDef;
+import org.apache.atlas.model.typedef.AtlasTypeDefHeader;
+import org.apache.atlas.model.typedef.AtlasTypesDef;
+import org.apache.atlas.v1.model.typedef.AttributeDefinition;
+import org.apache.atlas.v1.model.typedef.ClassTypeDefinition;
+import org.apache.atlas.v1.model.typedef.Multiplicity;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
@@ -34,6 +46,7 @@ import org.apache.commons.lang.StringUtils;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import java.util.stream.Collectors;
import static org.apache.atlas.model.typedef.AtlasBaseTypeDef.*;
@@ -405,6 +418,128 @@ public class AtlasTypeUtil {
return sb.toString();
}
+ public static ClassTypeDefinition toClassTypeDefinition(final AtlasEntityType entityType) {
+ ClassTypeDefinition ret = null;
+
+ if (entityType != null) {
+ AtlasEntityDef entityDef = entityType.getEntityDef();
+ ret = new ClassTypeDefinition();
+ ret.setTypeName(entityDef.getName());
+ ret.setTypeDescription(entityDef.getDescription());
+ ret.setTypeVersion(entityDef.getTypeVersion());
+ ret.setSuperTypes(entityDef.getSuperTypes());
+
+ if (MapUtils.isNotEmpty(entityType.getAllAttributes())) {
+ List<AttributeDefinition> attributeDefinitions = entityType.getAllAttributes()
+ .entrySet()
+ .stream()
+ .map(e -> toV1AttributeDefinition(e.getValue()))
+ .collect(Collectors.toList());
+
+ ret.setAttributeDefinitions(attributeDefinitions);
+ }
+ }
+
+ return ret;
+ }
+
+ public static AttributeDefinition toV1AttributeDefinition(AtlasStructType.AtlasAttribute attribute) {
+ AtlasAttributeDef attributeDef = attribute.getAttributeDef();
+ AttributeDefinition ret = new AttributeDefinition();
+
+ ret.setName(attributeDef.getName());
+ ret.setDataTypeName(attributeDef.getTypeName());
+ ret.setIsUnique(attributeDef.getIsUnique());
+ ret.setIsIndexable(attributeDef.getIsIndexable());
+ ret.setIsComposite(attribute.isOwnedRef());
+ ret.setReverseAttributeName(attribute.getInverseRefAttributeName());
+ ret.setDefaultValue(attributeDef.getDefaultValue());
+ ret.setDescription(attributeDef.getDescription());
+
+ final int lower;
+ final int upper;
+
+ if (attributeDef.getCardinality() == AtlasAttributeDef.Cardinality.SINGLE) {
+ lower = attributeDef.getIsOptional() ? 0 : 1;
+ upper = 1;
+ } else {
+ if(attributeDef.getIsOptional()) {
+ lower = 0;
+ } else {
+ lower = attributeDef.getValuesMinCount() < 1 ? 1 : attributeDef.getValuesMinCount();
+ }
+
+ upper = attributeDef.getValuesMaxCount() < 2 ? Integer.MAX_VALUE : attributeDef.getValuesMaxCount();
+ }
+
+ Multiplicity multiplicity = new Multiplicity();
+ multiplicity.setLower(lower);
+ multiplicity.setUpper(upper);
+ multiplicity.setIsUnique(AtlasAttributeDef.Cardinality.SET.equals(attributeDef.getCardinality()));
+
+ ret.setMultiplicity(multiplicity);
+
+ return ret;
+ }
+
+ public static Map<String, Object> toMap(AtlasEntity entity) {
+ Map<String, Object> ret = null;
+
+ if (entity != null) {
+ ret = new LinkedHashMap<>();
+
+ // Id type
+ ret.put("$typeName$", entity.getTypeName());
+ ret.put("$id$", new LinkedHashMap<String, Object>(){{
+ put("id", entity.getGuid());
+ put("$typeName$", entity.getTypeName());
+ put("version", entity.getVersion().intValue());
+ put("state", entity.getStatus().name());
+ }});
+
+ // System attributes
+ ret.put("$systemAttributes$", new LinkedHashMap<String, String>() {{
+ put("createdBy", entity.getCreatedBy());
+ put("modifiedBy", entity.getUpdatedBy());
+ put("createdTime", entity.getCreateTime().toString());
+ put("modifiedTime", entity.getUpdateTime().toString());
+ }});
+
+ // Traits
+ if (CollectionUtils.isNotEmpty(entity.getClassifications())) {
+ Map<String, HashMap> traitDetails = entity.getClassifications()
+ .stream()
+ .collect(Collectors.toMap(AtlasStruct::getTypeName, c -> getNestedTraitDetails(c.getTypeName())));
+ ret.put("$traits$", traitDetails);
+ }
+
+ // All attributes
+ if (MapUtils.isNotEmpty(entity.getAttributes())) {
+ for (Map.Entry<String, Object> entry : entity.getAttributes().entrySet()) {
+ if (entry.getValue() instanceof AtlasObjectId) {
+ ret.put(entry.getKey(), new LinkedHashMap<String, Object>(){{
+ put("id", ((AtlasObjectId) entry.getValue()).getGuid());
+ put("$typeName$", ((AtlasObjectId) entry.getValue()).getTypeName());
+// put("version", entity.getVersion().intValue());
+// put("state", entity.getStatus().name());
+ }});
+ } else {
+ ret.put(entry.getKey(), entry.getValue());
+ }
+ }
+ }
+
+ }
+
+ return ret;
+ }
+
+ private static HashMap getNestedTraitDetails(final Object typeName) {
+ return new HashMap() {{
+ put("$typeName$", typeName);
+ }};
+ }
+
private static void dumpTypeNames(List<? extends AtlasBaseTypeDef> typeDefs, StringBuilder sb) {
if (CollectionUtils.isNotEmpty(typeDefs)) {
for (int i = 0; i < typeDefs.size(); i++) {
http://git-wip-us.apache.org/repos/asf/atlas/blob/fe453597/intg/src/main/java/org/apache/atlas/v1/model/lineage/LineageResponse.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/v1/model/lineage/LineageResponse.java b/intg/src/main/java/org/apache/atlas/v1/model/lineage/LineageResponse.java
index e1c392d..e86709b 100644
--- a/intg/src/main/java/org/apache/atlas/v1/model/lineage/LineageResponse.java
+++ b/intg/src/main/java/org/apache/atlas/v1/model/lineage/LineageResponse.java
@@ -29,7 +29,7 @@ import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility.PUBLIC_ONL
@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
@JsonIgnoreProperties(ignoreUnknown = true)
public class LineageResponse {
- private String requestId;
+ private String requestId;
private Struct results;
public LineageResponse() {
http://git-wip-us.apache.org/repos/asf/atlas/blob/fe453597/intg/src/main/java/org/apache/atlas/v1/model/lineage/SchemaResponse.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/v1/model/lineage/SchemaResponse.java b/intg/src/main/java/org/apache/atlas/v1/model/lineage/SchemaResponse.java
new file mode 100644
index 0000000..837668b
--- /dev/null
+++ b/intg/src/main/java/org/apache/atlas/v1/model/lineage/SchemaResponse.java
@@ -0,0 +1,104 @@
+/**
+ * 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.v1.model.lineage;
+
+import org.apache.atlas.v1.model.typedef.ClassTypeDefinition;
+import org.codehaus.jackson.annotate.JsonAutoDetect;
+import org.codehaus.jackson.annotate.JsonIgnoreProperties;
+import org.codehaus.jackson.map.annotate.JsonSerialize;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility.NONE;
+import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility.PUBLIC_ONLY;
+
+@JsonAutoDetect(getterVisibility = PUBLIC_ONLY, setterVisibility = PUBLIC_ONLY, fieldVisibility = NONE)
+@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class SchemaResponse {
+ private String requestId;
+ private String tableName;
+ private SchemaDetails results;
+
+ public SchemaResponse() {
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+ public void setTableName(final String tableName) {
+ this.tableName = tableName;
+ }
+
+ public String getRequestId() {
+ return requestId;
+ }
+
+ public void setRequestId(final String requestId) {
+ this.requestId = requestId;
+ }
+
+ public SchemaDetails getResults() {
+ return results;
+ }
+
+ public void setResults(final SchemaDetails results) {
+ this.results = results;
+ }
+
+ /**
+ * Represents the column schema for a given hive table
+ */
+ @JsonAutoDetect(getterVisibility = PUBLIC_ONLY, setterVisibility = PUBLIC_ONLY, fieldVisibility = NONE)
+ @JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public static class SchemaDetails {
+ private String query; // The DSL query representation for obtaining schema
+ private ClassTypeDefinition dataType; // Data type of the (hive) columns
+ private List<Map<String, Object>> rows; // Column instances for the given table
+
+ public SchemaDetails() {
+ }
+
+ public String getQuery() {
+ return query;
+ }
+
+ public void setQuery(final String query) {
+ this.query = query;
+ }
+
+ public ClassTypeDefinition getDataType() {
+ return dataType;
+ }
+
+ public void setDataType(final ClassTypeDefinition dataType) {
+ this.dataType = dataType;
+ }
+
+ public List<Map<String, Object>> getRows() {
+ return rows;
+ }
+
+ public void setRows(final List<Map<String, Object>> rows) {
+ this.rows = rows;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/atlas/blob/fe453597/intg/src/main/java/org/apache/atlas/v1/model/typedef/AttributeDefinition.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/v1/model/typedef/AttributeDefinition.java b/intg/src/main/java/org/apache/atlas/v1/model/typedef/AttributeDefinition.java
index 6afc624..a64425c 100644
--- a/intg/src/main/java/org/apache/atlas/v1/model/typedef/AttributeDefinition.java
+++ b/intg/src/main/java/org/apache/atlas/v1/model/typedef/AttributeDefinition.java
@@ -48,6 +48,8 @@ public class AttributeDefinition implements Serializable {
private boolean isUnique;
private boolean isIndexable;
private String reverseAttributeName; // If this is a reference attribute, then the name of the attribute on the Class that this refers to.
+ private String defaultValue;
+ private String description;
@@ -130,6 +132,21 @@ public class AttributeDefinition implements Serializable {
this.reverseAttributeName = reverseAttributeName;
}
+ public String getDefaultValue() {
+ return defaultValue;
+ }
+
+ public void setDefaultValue(final String defaultValue) {
+ this.defaultValue = defaultValue;
+ }
+
+ public String getDescription() {
+ return description;
+ }
+
+ public void setDescription(final String description) {
+ this.description = description;
+ }
@Override
public boolean equals(Object o) {
@@ -149,11 +166,14 @@ public class AttributeDefinition implements Serializable {
Objects.equals(name, that.name) &&
Objects.equals(dataTypeName, that.dataTypeName) &&
Objects.equals(multiplicity, that.multiplicity) &&
+ Objects.equals(defaultValue, that.defaultValue) &&
+ Objects.equals(description, that.description) &&
Objects.equals(reverseAttributeName, that.reverseAttributeName);
}
@Override
public int hashCode() {
- return Objects.hash(name, dataTypeName, multiplicity, isComposite, isUnique, isIndexable, reverseAttributeName);
+ return Objects.hash(name, dataTypeName, multiplicity, isComposite, isUnique, isIndexable,
+ reverseAttributeName, defaultValue, description);
}
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/fe453597/intg/src/main/java/org/apache/atlas/v1/model/typedef/ClassTypeDefinition.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/v1/model/typedef/ClassTypeDefinition.java b/intg/src/main/java/org/apache/atlas/v1/model/typedef/ClassTypeDefinition.java
index 8fae175..9e7e03c 100644
--- a/intg/src/main/java/org/apache/atlas/v1/model/typedef/ClassTypeDefinition.java
+++ b/intg/src/main/java/org/apache/atlas/v1/model/typedef/ClassTypeDefinition.java
@@ -40,12 +40,18 @@ import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility.PUBLIC_ONL
@XmlAccessorType(XmlAccessType.PROPERTY)
public class ClassTypeDefinition extends HierarchicalTypeDefinition implements Serializable {
private static final long serialVersionUID = 1L;
+ private static final String META_TYPE_NAME = "org.apache.atlas.typesystem.types.ClassType";
public ClassTypeDefinition() {
}
public ClassTypeDefinition(String typeName, String typeDescription, String typeVersion, List<AttributeDefinition> attributeDefinitions, Set<String> superTypes) {
- super(typeName, typeDescription, typeVersion, attributeDefinitions, "org.apache.atlas.typesystem.types.ClassType", superTypes);
+ super(typeName, typeDescription, typeVersion, attributeDefinitions, META_TYPE_NAME, superTypes);
+ }
+
+ @Override
+ public String getHierarchicalMetaTypeName() {
+ return META_TYPE_NAME;
}
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/fe453597/repository/src/main/java/org/apache/atlas/discovery/AtlasLineageService.java
----------------------------------------------------------------------
diff --git a/repository/src/main/java/org/apache/atlas/discovery/AtlasLineageService.java b/repository/src/main/java/org/apache/atlas/discovery/AtlasLineageService.java
index 4282429..56cca96 100644
--- a/repository/src/main/java/org/apache/atlas/discovery/AtlasLineageService.java
+++ b/repository/src/main/java/org/apache/atlas/discovery/AtlasLineageService.java
@@ -22,6 +22,7 @@ package org.apache.atlas.discovery;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.model.lineage.AtlasLineageInfo;
import org.apache.atlas.model.lineage.AtlasLineageInfo.LineageDirection;
+import org.apache.atlas.v1.model.lineage.SchemaResponse.SchemaDetails;
public interface AtlasLineageService {
/**
@@ -38,7 +39,7 @@ public interface AtlasLineageService {
* @param datasetName datasetName
* @return Schema as JSON
*/
- String getSchema(String datasetName) throws AtlasBaseException;
+ SchemaDetails getSchema(String datasetName) throws AtlasBaseException;
/**
* Return the schema for the given entity id.
@@ -46,5 +47,5 @@ public interface AtlasLineageService {
* @param guid tableName
* @return Schema as JSON
*/
- String getSchemaForEntity(String guid) throws AtlasBaseException;
+ SchemaDetails getSchemaForEntity(String guid) throws AtlasBaseException;
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/fe453597/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java
----------------------------------------------------------------------
diff --git a/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java b/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java
index 0faedab..9d96de9 100644
--- a/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java
+++ b/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java
@@ -20,11 +20,10 @@ package org.apache.atlas.discovery;
import org.apache.atlas.AtlasClient;
-import org.apache.atlas.AtlasConfiguration;
import org.apache.atlas.AtlasErrorCode;
import org.apache.atlas.annotation.GraphTransaction;
import org.apache.atlas.exception.AtlasBaseException;
-import org.apache.atlas.model.discovery.AtlasSearchResult;
+import org.apache.atlas.model.instance.AtlasEntity;
import org.apache.atlas.model.instance.AtlasEntityHeader;
import org.apache.atlas.model.lineage.AtlasLineageInfo;
import org.apache.atlas.model.lineage.AtlasLineageInfo.LineageDirection;
@@ -33,12 +32,16 @@ import org.apache.atlas.repository.Constants;
import org.apache.atlas.repository.graph.GraphHelper;
import org.apache.atlas.repository.graphdb.AtlasGraph;
import org.apache.atlas.repository.graphdb.AtlasVertex;
+import org.apache.atlas.repository.store.graph.v1.AtlasGraphUtilsV1;
import org.apache.atlas.repository.store.graph.v1.EntityGraphRetriever;
+import org.apache.atlas.type.AtlasEntityType;
import org.apache.atlas.type.AtlasTypeRegistry;
+import org.apache.atlas.type.AtlasTypeUtil;
import org.apache.atlas.util.AtlasGremlinQueryProvider;
import org.apache.atlas.util.AtlasGremlinQueryProvider.AtlasGremlinQuery;
+import org.apache.atlas.v1.model.lineage.SchemaResponse.SchemaDetails;
import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.configuration.Configuration;
+import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.springframework.stereotype.Service;
@@ -49,27 +52,24 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
@Service
public class EntityLineageService implements AtlasLineageService {
private static final String INPUT_PROCESS_EDGE = "__Process.inputs";
private static final String OUTPUT_PROCESS_EDGE = "__Process.outputs";
- public static final String DATASET_SCHEMA_QUERY_PREFIX = "atlas.lineage.schema.query.";
-
private final AtlasGraph graph;
private final AtlasGremlinQueryProvider gremlinQueryProvider;
private final EntityGraphRetriever entityRetriever;
- private final AtlasDiscoveryService atlasDiscoveryService;
- private final Configuration atlasConfiguration;
+ private final AtlasTypeRegistry atlasTypeRegistry;
@Inject
- EntityLineageService(AtlasTypeRegistry typeRegistry, AtlasGraph atlasGraph, final AtlasDiscoveryService atlasDiscoveryService, final Configuration atlasConfiguration) throws DiscoveryException {
+ EntityLineageService(AtlasTypeRegistry typeRegistry, AtlasGraph atlasGraph) {
this.graph = atlasGraph;
- this.atlasDiscoveryService = atlasDiscoveryService;
- this.atlasConfiguration = atlasConfiguration;
this.gremlinQueryProvider = AtlasGremlinQueryProvider.INSTANCE;
this.entityRetriever = new EntityGraphRetriever(typeRegistry);
+ this.atlasTypeRegistry = typeRegistry;
}
@Override
@@ -100,42 +100,51 @@ public class EntityLineageService implements AtlasLineageService {
@Override
@GraphTransaction
- public String getSchema(final String datasetName) throws AtlasBaseException {
+ public SchemaDetails getSchema(final String datasetName) throws AtlasBaseException {
if (StringUtils.isEmpty(datasetName)) {
// TODO: Complete error handling here
throw new AtlasBaseException(AtlasErrorCode.BAD_REQUEST);
}
- Iterator<AtlasVertex> vertices = graph.query().has("Referenceable.qualifiedName", datasetName)
- .has(Constants.STATE_PROPERTY_KEY, "ACTIVE")
- .has(Constants.SUPER_TYPES_PROPERTY_KEY, "DataSet")
- .vertices().iterator();
-
- if (vertices.hasNext()) {
- AtlasVertex vertex = vertices.next();
- String typeName = GraphHelper.getTypeName(vertex);
- String guid = GraphHelper.getGuid(vertex);
- return getSchemaForId(typeName, guid);
- }
- return null;
+
+ AtlasEntityType hive_table = atlasTypeRegistry.getEntityTypeByName("hive_table");
+
+ Map<String, Object> lookupAttributes = new HashMap<>();
+ lookupAttributes.put("qualifiedName", datasetName);
+ String guid = AtlasGraphUtilsV1.getGuidByUniqueAttributes(hive_table, lookupAttributes);
+
+ return getSchemaForEntity(guid);
}
@Override
@GraphTransaction
- public String getSchemaForEntity(final String guid) throws AtlasBaseException {
+ public SchemaDetails getSchemaForEntity(final String guid) throws AtlasBaseException {
if (StringUtils.isEmpty(guid)) {
throw new AtlasBaseException(AtlasErrorCode.BAD_REQUEST);
}
- Iterator<AtlasVertex> vertices = graph.query().has(Constants.GUID_PROPERTY_KEY, guid)
- .has(Constants.SUPER_TYPES_PROPERTY_KEY, AtlasClient.DATA_SET_SUPER_TYPE)
- .vertices().iterator();
+ SchemaDetails ret = new SchemaDetails();
+ AtlasEntityType hive_column = atlasTypeRegistry.getEntityTypeByName("hive_column");
+
+ // TODO: Form and set the query, might not even be needed if DSL is gone forever
+ ret.setQuery("");
+ ret.setDataType(AtlasTypeUtil.toClassTypeDefinition(hive_column));
+
+ AtlasEntity.AtlasEntityWithExtInfo entityWithExtInfo = entityRetriever.toAtlasEntityWithExtInfo(guid);
+ Map<String, AtlasEntity> referredEntities = entityWithExtInfo.getReferredEntities();
+
+ if (MapUtils.isNotEmpty(referredEntities)) {
+ List<Map<String, Object>> rows = referredEntities.entrySet()
+ .stream()
+ .filter(EntityLineageService::isHiveColumn)
+ .map(e -> AtlasTypeUtil.toMap(e.getValue()))
+ .collect(Collectors.toList());
+ ret.setRows(rows);
+ }
- if (vertices.hasNext()) {
- AtlasVertex vertex = vertices.next();
- String typeName = GraphHelper.getTypeName(vertex);
- return getSchemaForId(typeName, guid);
+ return ret;
+ }
- }
- return null;
+ private static boolean isHiveColumn(Map.Entry<String, AtlasEntity> e) {
+ return StringUtils.equals("hive_column", e.getValue().getTypeName());
}
private AtlasLineageInfo getLineageInfo(String guid, LineageDirection direction, int depth) throws AtlasBaseException {
@@ -193,7 +202,7 @@ public class EntityLineageService implements AtlasLineageService {
return ret;
}
- private String getLineageQuery(String entityGuid, LineageDirection direction, int depth) throws AtlasBaseException {
+ private String getLineageQuery(String entityGuid, LineageDirection direction, int depth) {
String lineageQuery = null;
if (direction.equals(LineageDirection.INPUT)) {
@@ -228,22 +237,10 @@ public class EntityLineageService implements AtlasLineageService {
AtlasVertex entityVertex = results.next();
List<String> superTypes = GraphHelper.getSuperTypeNames(entityVertex);
- ret = (CollectionUtils.isNotEmpty(superTypes)) ? superTypes.contains(AtlasClient.DATA_SET_SUPER_TYPE) : false;
+ ret = (CollectionUtils.isNotEmpty(superTypes)) && superTypes.contains(AtlasClient.DATA_SET_SUPER_TYPE);
}
return ret;
}
- private String getSchemaForId(String typeName, String guid) throws AtlasBaseException {
- String configName = DATASET_SCHEMA_QUERY_PREFIX + typeName;
- String schemaTemplate = atlasConfiguration.getString(configName);
- if (schemaTemplate != null) {
- final String schemaQuery = String.format(schemaTemplate, guid);
- int limit = AtlasConfiguration.SEARCH_MAX_LIMIT.getInt();
- AtlasSearchResult searchResult = atlasDiscoveryService.searchUsingDslQuery(schemaQuery, limit, 0);
- // TODO: Fix the return there
- return null;
- }
- throw new AtlasBaseException("Schema is not configured for type " + typeName + ". Configure " + configName);
- }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/atlas/blob/fe453597/webapp/src/main/java/org/apache/atlas/web/resources/DataSetLineageResource.java
----------------------------------------------------------------------
diff --git a/webapp/src/main/java/org/apache/atlas/web/resources/DataSetLineageResource.java b/webapp/src/main/java/org/apache/atlas/web/resources/DataSetLineageResource.java
index bb7a122..e547c6e 100644
--- a/webapp/src/main/java/org/apache/atlas/web/resources/DataSetLineageResource.java
+++ b/webapp/src/main/java/org/apache/atlas/web/resources/DataSetLineageResource.java
@@ -18,7 +18,6 @@
package org.apache.atlas.web.resources;
-import org.apache.atlas.AtlasClient;
import org.apache.atlas.AtlasErrorCode;
import org.apache.atlas.discovery.AtlasLineageService;
import org.apache.atlas.exception.AtlasBaseException;
@@ -30,10 +29,10 @@ import org.apache.atlas.type.AtlasEntityType;
import org.apache.atlas.type.AtlasTypeRegistry;
import org.apache.atlas.utils.AtlasPerfTracer;
import org.apache.atlas.v1.model.lineage.DataSetLineageResponse;
+import org.apache.atlas.v1.model.lineage.SchemaResponse;
import org.apache.atlas.web.util.LineageUtils;
import org.apache.atlas.web.util.Servlets;
import org.apache.commons.lang.StringUtils;
-import org.codehaus.jettison.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
@@ -52,6 +51,8 @@ import javax.ws.rs.core.Response;
import java.util.HashMap;
import java.util.Map;
+import static org.apache.atlas.v1.model.lineage.SchemaResponse.SchemaDetails;
+
/**
* Jersey Resource for Hive Table Lineage.
*/
@@ -172,26 +173,25 @@ public class DataSetLineageResource {
@Path("table/{tableName}/schema")
@Consumes(Servlets.JSON_MEDIA_TYPE)
@Produces(Servlets.JSON_MEDIA_TYPE)
- public Response schema(@Context HttpServletRequest request, @PathParam("tableName") String tableName) {
+ public SchemaResponse schema(@Context HttpServletRequest request, @PathParam("tableName") String tableName) {
if (LOG.isDebugEnabled()) {
LOG.debug("==> DataSetLineageResource.schema({})", tableName);
}
AtlasPerfTracer perf = null;
+ SchemaResponse ret = new SchemaResponse();
try {
if (AtlasPerfTracer.isPerfTraceEnabled(PERF_LOG)) {
perf = AtlasPerfTracer.getPerfTracer(PERF_LOG, "DataSetLineageResource.schema(tableName=" + tableName + ")");
}
- final String jsonResult = ""; // TODO-typeSystem-removal: lineageService.getSchema(tableName);
-
- JSONObject response = new JSONObject();
- response.put(AtlasClient.REQUEST_ID, Servlets.getRequestId());
- response.put("tableName", tableName);
- response.put(AtlasClient.RESULTS, new JSONObject(jsonResult));
+ SchemaDetails schemaDetails = atlasLineageService.getSchema(tableName);
- return Response.ok(response).build();
+ ret.setRequestId(Servlets.getRequestId());
+ ret.setTableName(tableName);
+ ret.setResults(schemaDetails);
+ return ret;
} catch (IllegalArgumentException e) {
LOG.error("Unable to get schema for table {}", tableName, e);
throw new WebApplicationException(Servlets.getErrorResponse(e, Response.Status.BAD_REQUEST));
http://git-wip-us.apache.org/repos/asf/atlas/blob/fe453597/webapp/src/main/java/org/apache/atlas/web/resources/LineageResource.java
----------------------------------------------------------------------
diff --git a/webapp/src/main/java/org/apache/atlas/web/resources/LineageResource.java b/webapp/src/main/java/org/apache/atlas/web/resources/LineageResource.java
index 66e5942..8819766 100644
--- a/webapp/src/main/java/org/apache/atlas/web/resources/LineageResource.java
+++ b/webapp/src/main/java/org/apache/atlas/web/resources/LineageResource.java
@@ -18,7 +18,6 @@
package org.apache.atlas.web.resources;
-import org.apache.atlas.AtlasClient;
import org.apache.atlas.discovery.AtlasLineageService;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.model.lineage.AtlasLineageInfo;
@@ -26,9 +25,9 @@ import org.apache.atlas.model.lineage.AtlasLineageInfo.LineageDirection;
import org.apache.atlas.type.AtlasTypeRegistry;
import org.apache.atlas.utils.AtlasPerfTracer;
import org.apache.atlas.v1.model.lineage.LineageResponse;
+import org.apache.atlas.v1.model.lineage.SchemaResponse;
import org.apache.atlas.web.util.LineageUtils;
import org.apache.atlas.web.util.Servlets;
-import org.codehaus.jettison.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
@@ -159,24 +158,25 @@ public class LineageResource {
@Path("{guid}/schema")
@Consumes(Servlets.JSON_MEDIA_TYPE)
@Produces(Servlets.JSON_MEDIA_TYPE)
- public Response schema(@PathParam("guid") String guid) {
+ public SchemaResponse schema(@PathParam("guid") String guid) {
if (LOG.isDebugEnabled()) {
LOG.debug("==> LineageResource.schema({})", guid);
}
AtlasPerfTracer perf = null;
+ SchemaResponse ret = new SchemaResponse();
+
try {
if (AtlasPerfTracer.isPerfTraceEnabled(PERF_LOG)) {
perf = AtlasPerfTracer.getPerfTracer(PERF_LOG, "LineageResource.schema(" + guid + ")");
}
- final String jsonResult = atlasLineageService.getSchemaForEntity(guid);
+ SchemaResponse.SchemaDetails schemaDetails = atlasLineageService.getSchemaForEntity(guid);
- JSONObject response = new JSONObject();
- response.put(AtlasClient.REQUEST_ID, Servlets.getRequestId());
- response.put(AtlasClient.RESULTS, new JSONObject(jsonResult));
- return Response.ok(response).build();
+ ret.setRequestId(Servlets.getRequestId());
+ ret.setResults(schemaDetails);
+ return ret;
// } catch (SchemaNotFoundException e) {
// LOG.error("schema not found for {}", guid);
// throw new WebApplicationException(Servlets.getErrorResponse(e, Response.Status.NOT_FOUND));