ATLAS-2327: updated V1 to V2 conversion with addition of validation of attribute values (#2)

repository/src/main/java/org/apache/atlas/repository/converters/AtlasAbstractFormatConverter.java

@@ -19,9 +19,14 @@ package org.apache.atlas.repository.converters;
 
 
 import org.apache.atlas.model.TypeCategory;
+import org.apache.atlas.type.AtlasType;
 import org.apache.atlas.type.AtlasTypeRegistry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public abstract class AtlasAbstractFormatConverter implements AtlasFormatConverter {
+    private static final Logger LOG = LoggerFactory.getLogger(AtlasAbstractFormatConverter.class);
+
 
     protected final AtlasFormatConverters converterRegistry;
     protected final AtlasTypeRegistry     typeRegistry;
@@ -34,6 +39,18 @@ public abstract class AtlasAbstractFormatConverter implements AtlasFormatConvert
     }
 
     @Override
+    public boolean isValidValueV1(Object v1Obj, AtlasType type) {
+        boolean ret = type.isValidValue(v1Obj);
+
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("AtlasAbstractFormatConverter.isValidValueV1(type={}, value={}): {}", (v1Obj != null ? v1Obj.getClass().getCanonicalName() : null), v1Obj, ret);
+        }
+
+        return ret;
+    }
+
+
+    @Override
     public TypeCategory getTypeCategory() {
         return typeCategory;
     }

repository/src/main/java/org/apache/atlas/repository/converters/AtlasArrayFormatConverter.java

@@ -24,6 +24,10 @@ import org.apache.atlas.model.TypeCategory;
 import org.apache.atlas.type.AtlasArrayType;
 import org.apache.atlas.type.AtlasType;
 import org.apache.atlas.type.AtlasTypeRegistry;
+import org.apache.atlas.v1.model.instance.Id;
+import org.apache.atlas.v1.model.instance.Referenceable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -32,12 +36,56 @@ import java.util.List;
 import java.util.Set;
 
 public class AtlasArrayFormatConverter extends AtlasAbstractFormatConverter {
+    private static final Logger LOG = LoggerFactory.getLogger(AtlasArrayFormatConverter.class);
 
     public AtlasArrayFormatConverter(AtlasFormatConverters registry, AtlasTypeRegistry typeRegistry) {
         super(registry, typeRegistry, TypeCategory.ARRAY);
     }
 
     @Override
+    public boolean isValidValueV1(Object v1Obj, AtlasType type) {
+        boolean ret = false;
+
+        if (v1Obj == null) {
+            return true;
+        } if (type instanceof AtlasArrayType) {
+            AtlasArrayType       arrType       = (AtlasArrayType) type;
+            AtlasType            elemType      = arrType.getElementType();
+            AtlasFormatConverter elemConverter = null;
+
+            try {
+                elemConverter = converterRegistry.getConverter(elemType.getTypeCategory());
+            } catch (AtlasBaseException excp) {
+                LOG.warn("failed to get element converter. type={}", type.getTypeName(), excp);
+
+                ret = false;
+            }
+
+            if (elemConverter != null) {
+                if (v1Obj instanceof Collection) {
+                    ret = true; // for empty array
+
+                    for (Object v1Elem : (Collection) v1Obj) {
+                        ret = elemConverter.isValidValueV1(v1Elem, elemType);
+
+                        if (!ret) {
+                            break;
+                        }
+                    }
+                } else {
+                    ret = elemConverter.isValidValueV1(v1Obj, elemType);
+                }
+            }
+        }
+
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("AtlasArrayFormatConverter.isValidValueV1(type={}, value={}): {}", (v1Obj != null ? v1Obj.getClass().getCanonicalName() : null), v1Obj, ret);
+        }
+
+        return ret;
+    }
+
+    @Override
     public Collection fromV1ToV2(Object v1Obj, AtlasType type, ConverterContext ctx) throws AtlasBaseException {
         Collection ret = null;
 

repository/src/main/java/org/apache/atlas/repository/converters/AtlasEntityFormatConverter.java

@@ -50,6 +50,17 @@ public class AtlasEntityFormatConverter extends AtlasStructFormatConverter {
     }
 
     @Override
+    public boolean isValidValueV1(Object v1Obj, AtlasType type) {
+        boolean ret = (v1Obj == null) || v1Obj instanceof Id || v1Obj instanceof Referenceable;
+
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("AtlasEntityFormatConverter.isValidValueV1(type={}, value={}): {}", (v1Obj != null ? v1Obj.getClass().getCanonicalName() : null), v1Obj, ret);
+        }
+
+        return ret;
+    }
+
+    @Override
     public AtlasEntity fromV1ToV2(Object v1Obj, AtlasType type, ConverterContext context) throws AtlasBaseException {
         AtlasEntity entity = null;
 

repository/src/main/java/org/apache/atlas/repository/converters/AtlasEnumFormatConverter.java

@@ -25,16 +25,61 @@ import org.apache.atlas.v1.model.typedef.EnumTypeDefinition.EnumValue;
 import org.apache.atlas.type.AtlasEnumType;
 import org.apache.atlas.type.AtlasType;
 import org.apache.atlas.type.AtlasTypeRegistry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Map;
 
 public class AtlasEnumFormatConverter extends AtlasAbstractFormatConverter {
+    private static final Logger LOG = LoggerFactory.getLogger(AtlasEnumFormatConverter.class);
+
 
     public AtlasEnumFormatConverter(AtlasFormatConverters registry, AtlasTypeRegistry typeRegistry) {
         super(registry, typeRegistry, TypeCategory.ENUM);
     }
 
     @Override
+    public boolean isValidValueV1(Object v1Obj, AtlasType type) {
+        boolean ret = false;
+
+        if (v1Obj == null) {
+            ret = true;
+        } else if (type instanceof AtlasEnumType) {
+            final AtlasEnumType enumType = (AtlasEnumType) type;
+
+            if (v1Obj instanceof EnumValue) {
+                Object enumValue = ((EnumValue)v1Obj).getValue();
+
+                if (enumValue != null) {
+                    ret = enumType.getEnumDef().hasElement(enumValue.toString());
+                }
+            } else if (v1Obj instanceof Map) {
+                Object enumValue = ((Map)v1Obj).get("value");
+
+                if (enumValue != null) {
+                    ret = enumType.getEnumDef().hasElement(enumValue.toString());
+                } else {
+                    Object enumOrdinal = ((Map)v1Obj).get("ordinal");
+
+                    if (enumOrdinal != null) {
+                        ret = enumType.getEnumElementDef((Number) enumOrdinal) != null;
+                    }
+                }
+            } else if (v1Obj instanceof Number) {
+                ret = enumType.getEnumElementDef((Number) v1Obj) != null;
+            } else {
+                ret = enumType.getEnumElementDef(v1Obj.toString()) != null;
+            }
+        }
+
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("AtlasEnumFormatConverter.isValidValueV1(type={}, value={}): {}", (v1Obj != null ? v1Obj.getClass().getCanonicalName() : null), v1Obj, ret);
+        }
+
+        return ret;
+    }
+
+    @Override
     public Object fromV1ToV2(Object v1Obj, AtlasType type, ConverterContext ctx) throws AtlasBaseException {
         String ret = null;
 

repository/src/main/java/org/apache/atlas/repository/converters/AtlasFormatConverter.java

@@ -26,6 +26,8 @@ import org.apache.atlas.type.AtlasType;
 
 
 public interface AtlasFormatConverter {
+    boolean isValidValueV1(Object v1Ob, AtlasType typej);
+
     Object fromV1ToV2(Object v1Obj, AtlasType type, ConverterContext context) throws AtlasBaseException;
 
     Object fromV2ToV1(Object v2Obj, AtlasType type, ConverterContext context) throws AtlasBaseException;

repository/src/main/java/org/apache/atlas/repository/converters/AtlasMapFormatConverter.java

@@ -21,20 +21,71 @@ package org.apache.atlas.repository.converters;
 import org.apache.atlas.AtlasErrorCode;
 import org.apache.atlas.exception.AtlasBaseException;
 import org.apache.atlas.model.TypeCategory;
+import org.apache.atlas.type.AtlasArrayType;
 import org.apache.atlas.type.AtlasMapType;
 import org.apache.atlas.type.AtlasType;
 import org.apache.atlas.type.AtlasTypeRegistry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 
 public class AtlasMapFormatConverter extends AtlasAbstractFormatConverter {
+    private static final Logger LOG = LoggerFactory.getLogger(AtlasMapFormatConverter.class);
+
 
     public AtlasMapFormatConverter(AtlasFormatConverters registry, AtlasTypeRegistry typeRegistry) {
         super(registry, typeRegistry, TypeCategory.MAP);
     }
 
     @Override
+    public boolean isValidValueV1(Object v1Obj, AtlasType type) {
+        boolean ret = false;
+
+        if (v1Obj == null) {
+            return true;
+        } if (type instanceof AtlasMapType && v1Obj instanceof Map) {
+            AtlasMapType         mapType        = (AtlasMapType) type;
+            AtlasType            keyType        = mapType.getKeyType();
+            AtlasType            valueType      = mapType.getValueType();
+            AtlasFormatConverter keyConverter   = null;
+            AtlasFormatConverter valueConverter = null;
+            Map                  v1Map          = (Map)v1Obj;
+
+            try {
+                keyConverter   = converterRegistry.getConverter(keyType.getTypeCategory());
+                valueConverter = converterRegistry.getConverter(valueType.getTypeCategory());
+            } catch (AtlasBaseException excp) {
+                LOG.warn("failed to get key/value converter. type={}", type.getTypeName(), excp);
+
+                ret = false;
+            }
+
+            if (keyConverter != null && valueConverter != null) {
+                ret = true; // for empty map
+
+                for (Object key : v1Map.keySet()) {
+                    Object value = v1Map.get(key);
+
+                    ret = keyConverter.isValidValueV1(key, keyType) && valueConverter.isValidValueV1(value, valueType);
+
+                    if (!ret) {
+                        break;
+                    }
+                }
+            }
+        }
+
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("AtlasArrayFormatConverter.isValidValueV1(type={}, value={}): {}", (v1Obj != null ? v1Obj.getClass().getCanonicalName() : null), v1Obj, ret);
+        }
+
+        return ret;
+    }
+
+    @Override
     public Map fromV1ToV2(Object v1Obj, AtlasType type, ConverterContext ctx) throws AtlasBaseException {
         Map ret = null;
 

repository/src/main/java/org/apache/atlas/repository/converters/AtlasObjectIdConverter.java

@@ -23,17 +23,21 @@ import org.apache.atlas.exception.AtlasBaseException;
 import org.apache.atlas.model.TypeCategory;
 import org.apache.atlas.model.instance.AtlasEntity;
 import org.apache.atlas.model.instance.AtlasObjectId;
-import org.apache.atlas.v1.model.instance.Id;
-import org.apache.atlas.v1.model.instance.Referenceable;
 import org.apache.atlas.type.AtlasEntityType;
 import org.apache.atlas.type.AtlasType;
 import org.apache.atlas.type.AtlasTypeRegistry;
+import org.apache.atlas.v1.model.instance.Id;
+import org.apache.atlas.v1.model.instance.Referenceable;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Map;
 
 public class AtlasObjectIdConverter extends  AtlasAbstractFormatConverter {
+    private static final Logger LOG = LoggerFactory.getLogger(AtlasObjectIdConverter.class);
+
 
     public AtlasObjectIdConverter(AtlasFormatConverters registry, AtlasTypeRegistry typeRegistry) {
         this(registry, typeRegistry, TypeCategory.OBJECT_ID_TYPE);
@@ -44,6 +48,17 @@ public class AtlasObjectIdConverter extends  AtlasAbstractFormatConverter {
     }
 
     @Override
+    public boolean isValidValueV1(Object v1Obj, AtlasType type) {
+        boolean ret = (v1Obj == null) || v1Obj instanceof Id || v1Obj instanceof Referenceable;
+
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("AtlasObjectIdConverter.isValidValueV1(type={}, value={}): {}", (v1Obj != null ? v1Obj.getClass().getCanonicalName() : null), v1Obj, ret);
+        }
+
+        return ret;
+    }
+
+    @Override
     public Object fromV1ToV2(Object v1Obj, AtlasType type, AtlasFormatConverter.ConverterContext converterContext) throws AtlasBaseException {
         Object ret = null;
 

repository/src/main/java/org/apache/atlas/repository/converters/AtlasStructFormatConverter.java

@@ -52,6 +52,17 @@ public class AtlasStructFormatConverter extends AtlasAbstractFormatConverter {
     }
 
     @Override
+    public boolean isValidValueV1(Object v1Obj, AtlasType type) {
+        boolean ret = (v1Obj == null) || v1Obj instanceof Map || v1Obj instanceof Struct;
+
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("AtlasStructFormatConverter.isValidValueV1(type={}, value={}): {}", (v1Obj != null ? v1Obj.getClass().getCanonicalName() : null), v1Obj, ret);
+        }
+
+        return ret;
+    }
+
+    @Override
     public Object fromV1ToV2(Object v1Obj, AtlasType type, ConverterContext converterContext) throws AtlasBaseException {
         AtlasStruct ret = null;
 
@@ -248,7 +259,7 @@ public class AtlasStructFormatConverter extends AtlasAbstractFormatConverter {
                 AtlasFormatConverter attrConverter = converterRegistry.getConverter(attrType.getTypeCategory());
                 Object               v1Value       = attributes.get(attrName);
 
-                if (attrType.isValidValue(v1Value)) {
+                if (attrConverter.isValidValueV1(v1Value, attrType)) {
                     Object v2Value = attrConverter.fromV1ToV2(v1Value, attrType, context);
 
                     ret.put(attrName, v2Value);