Fixes merge and map collection issues

repository/src/main/java/org/apache/hadoop/metadata/repository/graph/GraphBackedMetadataRepository.java

@@ -36,11 +36,9 @@ import org.apache.hadoop.metadata.typesystem.ITypedReferenceableInstance;
 import org.apache.hadoop.metadata.typesystem.ITypedStruct;
 import org.apache.hadoop.metadata.typesystem.persistence.Id;
 import org.apache.hadoop.metadata.typesystem.persistence.MapIds;
-import org.apache.hadoop.metadata.typesystem.persistence.StructInstance;
 import org.apache.hadoop.metadata.typesystem.types.AttributeInfo;
 import org.apache.hadoop.metadata.typesystem.types.ClassType;
 import org.apache.hadoop.metadata.typesystem.types.DataTypes;
-import org.apache.hadoop.metadata.typesystem.types.EnumType;
 import org.apache.hadoop.metadata.typesystem.types.EnumValue;
 import org.apache.hadoop.metadata.typesystem.types.IDataType;
 import org.apache.hadoop.metadata.typesystem.types.Multiplicity;
@@ -1047,20 +1045,27 @@ public class GraphBackedMetadataRepository implements MetadataRepository {
                 return;
             }
             DataTypes.MapType mapType = (DataTypes.MapType) attributeInfo.dataType();
-            final IDataType elementType = mapType.getValueType();
+            final IDataType keyType = mapType.getKeyType();
+            final IDataType valueType = mapType.getValueType();
 
             HashMap values = new HashMap();
             for (String propertyNameWithSuffix : keys.split(",")) {
-                final String key = propertyNameWithSuffix.substring(
-                        propertyNameWithSuffix.lastIndexOf(".") + 1,
-                        propertyNameWithSuffix.lastIndexOf(":"));
+                final String key = extractKey(propertyNameWithSuffix, keyType);
                 values.put(key, mapVertexToCollectionEntry(instanceVertex, attributeInfo,
-                        elementType, propertyName, propertyNameWithSuffix));
+                        valueType, propertyName, propertyNameWithSuffix));
             }
 
             typedInstance.set(attributeInfo.name, values);
         }
 
+        private String extractKey(String propertyNameWithSuffix, IDataType keyType) {
+            return propertyNameWithSuffix.substring(
+                    propertyNameWithSuffix.lastIndexOf(".") + 1,
+                    keyType.getTypeCategory() == DataTypes.TypeCategory.PRIMITIVE
+                            ? propertyNameWithSuffix.length()
+                            : propertyNameWithSuffix.lastIndexOf(":"));
+        }
+
         private ITypedStruct getStructInstanceFromVertex(Vertex instanceVertex,
                                                          IDataType elemType,
                                                          String attributeName,

repository/src/main/java/org/apache/hadoop/metadata/repository/graph/GraphBackedSearchIndexer.java

@@ -94,8 +94,8 @@ public class GraphBackedSearchIndexer implements SearchIndexer {
         createCompositeAndMixedIndex(Constants.TRAIT_NAMES_INDEX,
                 Constants.TRAIT_NAMES_PROPERTY_KEY, String.class, false, Cardinality.SET);
 
-        //Index for full text search
-        createVertexMixedIndex(Constants.ENTITY_TEXT_PROPERTY_KEY, String.class, Cardinality.SINGLE);
+        // Index for full text search
+        createVertexMixedIndex(Constants.ENTITY_TEXT_PROPERTY_KEY, String.class);
 
         //Indexes for graph backed type system store
         createTypeStoreIndexes();

repository/src/test/java/org/apache/hadoop/metadata/repository/graph/GraphBackedMetadataRepositoryTest.java

@@ -51,6 +51,7 @@ import scala.actors.threadpool.Arrays;
 
 import javax.inject.Inject;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.UUID;
@@ -378,6 +379,21 @@ public class GraphBackedMetadataRepositoryTest {
                         new AttributeDefinition("partitions",
                                 String.format("array<%s>", "partition_type"),
                                 Multiplicity.COLLECTION, true, null),
+                        // map of primitives
+                        new AttributeDefinition("parametersMap",
+                                DataTypes.mapTypeName(DataTypes.STRING_TYPE.getName(),
+                                        DataTypes.STRING_TYPE.getName()),
+                                Multiplicity.COLLECTION, true, null),
+                        // map of classes - todo - enable this
+//                        new AttributeDefinition("columnsMap",
+//                                DataTypes.mapTypeName(DataTypes.STRING_TYPE.getName(),
+//                                        "column_type"),
+//                                Multiplicity.COLLECTION, true, null),
+                        // map of structs   todo - enable this
+//                        new AttributeDefinition("partitionsMap",
+//                                DataTypes.mapTypeName(DataTypes.STRING_TYPE.getName(),
+//                                        "partition_type"),
+//                                Multiplicity.COLLECTION, true, null),
                         // struct reference
                         new AttributeDefinition("serde1",
                                 "serdeType", Multiplicity.REQUIRED, false, null),
@@ -385,7 +401,8 @@ public class GraphBackedMetadataRepositoryTest {
                                 "serdeType", Multiplicity.REQUIRED, false, null),
                         // class reference
                         new AttributeDefinition("database",
-                                DATABASE_TYPE, Multiplicity.REQUIRED, true, null));
+                                DATABASE_TYPE, Multiplicity.REQUIRED, true, null)
+                );
 
         HierarchicalTypeDefinition<TraitType> classificationTypeDefinition =
                 TypesUtil.createTraitTypeDef(CLASSIFICATION,
@@ -427,22 +444,38 @@ public class GraphBackedMetadataRepositoryTest {
         serde2Instance.set("serde", "serde2");
         tableInstance.set("serde2", serde2Instance);
 
+        // HashMap<String, Referenceable> columnsMap = new HashMap<>();
         ArrayList<Referenceable> columns = new ArrayList<>();
         for (int index = 0; index < 5; index++) {
             Referenceable columnInstance = new Referenceable("column_type");
-            columnInstance.set("name", "column_" + index);
+            final String name = "column_" + index;
+            columnInstance.set("name", name);
             columnInstance.set("type", "string");
+
             columns.add(columnInstance);
+            // columnsMap.put(name, columnInstance);
         }
         tableInstance.set("columns", columns);
+        // tableInstance.set("columnsMap", columnsMap);
 
+//        HashMap<String, Struct> partitionsMap = new HashMap<>();
         ArrayList<Struct> partitions = new ArrayList<>();
         for (int index = 0; index < 5; index++) {
             Struct partitionInstance = new Struct("partition_type");
-            partitionInstance.set("name", "partition_" + index);
+            final String name = "partition_" + index;
+            partitionInstance.set("name", name);
+
             partitions.add(partitionInstance);
+//            partitionsMap.put(name, partitionInstance);
         }
         tableInstance.set("partitions", partitions);
+//        tableInstance.set("partitionsMap", partitionsMap);
+
+        HashMap<String, String> parametersMap = new HashMap<>();
+        parametersMap.put("foo", "bar");
+        parametersMap.put("bar", "baz");
+        parametersMap.put("some", "thing");
+        tableInstance.set("parametersMap", parametersMap);
 
         ClassType tableType = typeSystem.getDataType(ClassType.class, TABLE_TYPE);
         return tableType.convert(tableInstance, Multiplicity.REQUIRED);