Merge pull request #1920 from telefonicaid/fix/attribute_type_key

Fix/attribute type key (workaround)
telefonicaid · Sep 2, 2020 · c886ae8 · c886ae8
2 parents 49bd2fc + ab16f1c
commit c886ae8
Show file tree

Hide file tree

Showing 5 changed files with 9 additions and 6 deletions.
diff --git a/CHANGES_NEXT_RELEASE b/CHANGES_NEXT_RELEASE
@@ -1,3 +1,4 @@
+[cygnus-ngsi][ColumnAggregator] Fix attribute type key to make it unique. (#1904)
 [cygnus-ngsi][PostgisSink, PostgreSQLSink] Implement records expiration for Postgis and PostgreSQL sinks (#1915)
 [cygnus-ngsi][ColumnAggregator] Handle metadata as json. This is to upgrade json-like sinks behaviour. (#1902)
 [cygnus-ngsi][CKANSink] Fix bug about bad service name used by CKAN orgName with a name mapping (#1910)

diff --git a/...ngsi/src/main/java/com/telefonica/iot/cygnus/aggregation/NGSIGenericColumnAggregator.java b/...ngsi/src/main/java/com/telefonica/iot/cygnus/aggregation/NGSIGenericColumnAggregator.java
@@ -68,7 +68,7 @@ public void initialize(NGSIEvent event) {
             String attrName = contextAttribute.getName();
             aggregation.put(attrName, new ArrayList<JsonElement>());
             aggregation.put(attrName + "_md", new ArrayList<JsonElement>());
-            aggregation.put(attrName + "_type", new ArrayList<JsonElement>());
+            aggregation.put(attrName + NGSIConstants.AUTOGENERATED_ATTR_TYPE, new ArrayList<JsonElement>());
         } // for
         setAggregation(aggregation);
     } // initialize
@@ -127,7 +127,7 @@ public void aggregate(NGSIEvent event) {
             if (aggregation.containsKey(attrName)) {
                 aggregation.get(attrName).add(attrValue);
                 aggregation.get(attrName + "_md").add(jsonAttrMetadata);
-                aggregation.get(attrName + "_type").add(new JsonPrimitive(attrType));
+                aggregation.get(attrName + NGSIConstants.AUTOGENERATED_ATTR_TYPE).add(new JsonPrimitive(attrType));
             } else {
                 ArrayList<JsonElement> values = new ArrayList<JsonElement>(Collections.nCopies(numPreviousValues, null));
                 values.add(attrValue);
@@ -137,7 +137,7 @@ public void aggregate(NGSIEvent event) {
                 aggregation.put(attrName + "_md", valuesMd);
                 ArrayList<JsonElement> valuesType = new ArrayList<JsonElement>(Collections.nCopies(numPreviousValues, null));
                 valuesType.add(new JsonPrimitive(attrType));
-                aggregation.put(attrName + "_type", valuesType);
+                aggregation.put(attrName + NGSIConstants.AUTOGENERATED_ATTR_TYPE, valuesType);
             } // if else
         } // for
         // Iterate on all the aggregations, checking for not updated attributes; add an empty value if missing

diff --git a/cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/sinks/NGSIHDFSSink.java b/cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/sinks/NGSIHDFSSink.java
@@ -683,7 +683,7 @@ protected NGSIGenericAggregator processCSVFields (NGSIGenericAggregator genericA
                         JsonElement metadata = genericAggregator.getAggregation().get(attributeName + "_md").get(i);
                         if (genericAggregator.isAttrMetadataStore() && metadata != null && !metadata.toString().isEmpty() && !metadata.toString().contains("[]")) {
                             String attrMdFileName = buildAttrMdFilePath(genericAggregator.getService(), genericAggregator.getServicePathForNaming(), genericAggregator.getEntityForNaming(), attributeName,
-                                    genericAggregator.getAggregation().get(attributeName + "_type").get(i).toString());
+                                    genericAggregator.getAggregation().get(attributeName + NGSIConstants.AUTOGENERATED_ATTR_TYPE).get(i).toString());
                             String printableAttrMdFileName = "hdfs:///user/" + username + "/" + attrMdFileName;
                             line += csvSeparator + printableAttrMdFileName;
                             if (metadata.isJsonPrimitive()) {

diff --git a/cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/utils/NGSIConstants.java b/cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/utils/NGSIConstants.java
@@ -51,6 +51,8 @@ private NGSIConstants() {
     public static final String ATTR_VALUE          = "attrValue";
     public static final String ATTR_MD             = "attrMd";
     public static final String ATTR_MD_FILE        = "attrMdFile";
+
+    public static final String AUTOGENERATED_ATTR_TYPE           = "_attr_internal_type";
 
     // FIWARE service and FIWARE service path specific constants
     public static final int SERVICE_HEADER_MAX_LEN      = 50;

diff --git a/cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/utils/NGSIUtils.java b/cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/utils/NGSIUtils.java
@@ -376,7 +376,7 @@ public static LinkedHashMap<String, ArrayList<JsonElement>> linkedHashMapWithout
         Iterator<String> it = aggregation.keySet().iterator();
         while (it.hasNext()) {
             String entry = (String) it.next();
-            if ((!attrMetadataStore && (entry.contains("_md") || entry.contains("_MD") || entry.equals(NGSIConstants.ATTR_MD)) || entry.contains("_type")) || (entry.equals(NGSIConstants.RECV_TIME_TS+"C"))) {
+            if ((!attrMetadataStore && (entry.contains("_md") || entry.contains("_MD") || entry.equals(NGSIConstants.ATTR_MD)) || entry.contains(NGSIConstants.AUTOGENERATED_ATTR_TYPE)) || (entry.equals(NGSIConstants.RECV_TIME_TS+"C"))) {
                 keysToCrop.add(entry);
             }
         }
@@ -400,7 +400,7 @@ public static ArrayList<String> attributeNames(LinkedHashMap<String, ArrayList<J
         attributeNames.remove(NGSIConstants.ENTITY_TYPE);
         ArrayList <String> cropedList = (ArrayList<String>) attributeNames.clone();
         for (String key : cropedList) {
-            if (key.contains("_md") || key.contains("_type")) {
+            if (key.contains("_md") || key.contains(NGSIConstants.AUTOGENERATED_ATTR_TYPE)) {
                 attributeNames.remove(key);
             }
         }