Correct edge dictionary to correctly handle unidirectional edge defin…

warehouse/ingest-core/src/main/java/datawave/ingest/mapreduce/handler/edge/ProtobufEdgeDataTypeHandler.java

@@ -74,6 +74,7 @@
 import datawave.ingest.table.config.LoadDateTableConfigHelper;
 import datawave.ingest.time.Now;
 import datawave.marking.MarkingFunctions;
+import datawave.metadata.protobuf.EdgeMetadata;
 import datawave.metadata.protobuf.EdgeMetadata.MetadataValue;
 import datawave.metadata.protobuf.EdgeMetadata.MetadataValue.Metadata;
 import datawave.util.StringUtils;
@@ -936,37 +937,18 @@ protected void registerEventMetadata(Map<Key,Set<Metadata>> eventMetadataRegistr
         // add to the eventMetadataRegistry map
         Key baseKey = createMetadataEdgeKey(edgeValue, edgeValue.getSource(), edgeValue.getSource().getIndexedFieldValue(), edgeValue.getSink(),
                         edgeValue.getSink().getIndexedFieldValue(), this.getVisibility(edgeValue));
-        Key fwdMetaKey = EdgeKey.getMetadataKey(baseKey);
-        Key revMetaKey = EdgeKey.getMetadataKey(EdgeKey.swapSourceSink(EdgeKey.decode(baseKey)).encode());
-
-        Set<Metadata> fwdMetaSet = eventMetadataRegistry.get(fwdMetaKey);
-        if (null == fwdMetaSet) {
-            fwdMetaSet = new HashSet<>();
-            eventMetadataRegistry.put(fwdMetaKey, fwdMetaSet);
-        }
-        Set<Metadata> revMetaSet = eventMetadataRegistry.get(revMetaKey);
-        if (null == revMetaSet) {
-            revMetaSet = new HashSet<>();
-            eventMetadataRegistry.put(revMetaKey, revMetaSet);
-        }
 
-        // Build the Protobuf for the value
-        Metadata.Builder forwardBuilder = Metadata.newBuilder().setSource(edgeValue.getSource().getFieldName()).setSink(edgeValue.getSink().getFieldName())
-                        .setDate(DateHelper.format(new Date(edgeValue.getEventDate())));
-        Metadata.Builder reverseBuilder = Metadata.newBuilder().setDate(DateHelper.format(new Date(edgeValue.getEventDate())))
-                        .setSource(edgeValue.getSink().getFieldName()).setSink(edgeValue.getSource().getFieldName());
-        if (enrichmentFieldName != null) {
-            forwardBuilder.setEnrichment(enrichmentFieldName).setEnrichmentIndex(edgeValue.getEnrichedIndex());
-            reverseBuilder.setEnrichment(enrichmentFieldName).setEnrichmentIndex(edgeValue.getEnrichedIndex());
-        }
+        Key fwdMetaKey = EdgeKey.getMetadataKey(baseKey);
+        addMetadata(eventMetadataRegistry, enrichmentFieldName, edgeValue, jexlPrecondition, fwdMetaKey);
 
-        if (jexlPrecondition != null) {
-            forwardBuilder.setJexlPrecondition(jexlPrecondition);
-            reverseBuilder.setJexlPrecondition(jexlPrecondition);
+        if (isNullOrBidirectional(edgeValue.getEdgeDirection())) {
+            Key revMetaKey = EdgeKey.getMetadataKey(EdgeKey.swapSourceSink(EdgeKey.decode(baseKey)).encode());
+            addMetadata(eventMetadataRegistry, enrichmentFieldName, edgeValue, jexlPrecondition, revMetaKey);
         }
+    }
 
-        fwdMetaSet.add(forwardBuilder.build());
-        revMetaSet.add(reverseBuilder.build());
+    private boolean isNullOrBidirectional(EdgeDirection direction) {
+        return direction == null || direction.equals(EdgeDirection.BIDIRECTIONAL);
     }
 
     protected String getEnrichmentFieldName(EdgeDefinition edgeDef) {
@@ -1184,6 +1166,26 @@ private Key createEdgeKey(EdgeDataBundle edgeValue, VertexValue source, String s
         return builder.build().encode();
     }
 
+    private Set<EdgeMetadata.MetadataValue.Metadata> addMetadata(Map<Key,Set<EdgeMetadata.MetadataValue.Metadata>> metadataRegistry, String enrichmentFieldName,
+                    EdgeDataBundle edgeDataBundle, String jexlPrecondition, Key key) {
+        Set<Metadata> metadata = metadataRegistry.computeIfAbsent(key, k -> new HashSet<>());
+
+        // Build the Protobuf for the value
+        Metadata.Builder builder = Metadata.newBuilder().setSource(edgeDataBundle.getSource().getFieldName()).setSink(edgeDataBundle.getSink().getFieldName())
+                        .setDate(DateHelper.format(new Date(edgeDataBundle.getEventDate())));
+
+        if (enrichmentFieldName != null) {
+            builder.setEnrichment(enrichmentFieldName).setEnrichmentIndex(edgeDataBundle.getEnrichedIndex());
+        }
+        if (jexlPrecondition != null) {
+            builder.setJexlPrecondition(jexlPrecondition);
+        }
+
+        metadata.add(builder.build());
+
+        return metadata;
+    }
+
     protected Key createStatsKey(STATS_TYPE statsType, EdgeDataBundle edgeValue, VertexValue vertex, String value, Text visibility,
                     EdgeKey.DATE_TYPE date_type) {
         String typeName = edgeValue.getDataTypeName();

warehouse/ingest-core/src/test/java/datawave/ingest/mapreduce/handler/edge/EdgeHandlerTestUtil.java

@@ -33,10 +33,12 @@
 public class EdgeHandlerTestUtil {
 
     public static final Text edgeTableName = new Text(TableName.EDGE);
+    public static final Text metaDataTableName = new Text(TableName.METADATA);
     public static final String NB = "\u0000";
 
     public static ListMultimap<String,String[]> edgeKeyResults = ArrayListMultimap.create();
     public static ListMultimap<String,String> edgeValueResults = ArrayListMultimap.create();
+    public static ListMultimap<Key,Value> metaData = ArrayListMultimap.create();
 
     private static Logger log = Logger.getLogger(EdgeHandlerTestUtil.class);
 
@@ -71,6 +73,9 @@ public static void processEvent(Multimap<String,NormalizedContentInterface> even
                         edgeKeys.add(entry.getKey().getKey());
                         edgeValueResults.put(entry.getKey().getKey().getRow().toString().replaceAll(NB, "%00;"), EdgeValue.decode(entry.getValue()).toString());
                     }
+                    if (entry.getKey().getTableName().equals(metaDataTableName)) {
+                        metaData.put(entry.getKey().getKey(), entry.getValue());
+                    }
                     if (!entry.getKey().getTableName().equals(edgeTableName) || entry.getKey().getKey().isDeleted() == edgeDeleteMode) {
                         if (countMap.containsKey(entry.getKey().getTableName())) {
                             countMap.put(entry.getKey().getTableName(), countMap.get(entry.getKey().getTableName()) + 1);