MySQL databases are not performant
I'm running an application "Powered by Fiware" that is using Cygnus for persistence. A few months ago, when I implemented it, I came across the problem of slow database performance. Looking at the structure of each created table, I found out that cygnus creates by default each column as text:
```
+-------------------+------------+------+-----+---------+-------+
| Field             | Type       | Null | Key | Default | Extra |
+-------------------+------------+------+-----+---------+-------+
| recvTimeTs        | mediumtext | YES  |     | NULL    |       |
| recvTime          | text       | YES  |     | NULL    |       |
| fiwareServicePath | text       | YES  |     | NULL    |       |
| entityId          | text       | YES  |     | NULL    |       |
| entityType        | text       | YES  |     | NULL    |       |
| attrName          | text       | YES  |     | NULL    |       |
| attrType          | text       | YES  |     | NULL    |       |
| attrValue         | text       | YES  |     | NULL    |       |
| attrMd            | text       | YES  |     | NULL    |       |
+-------------------+------------+------+-----+---------+-------+
```
At that moment, I thought to myself that it had to be configurable, so I left it like that and changed every table "by hand", changing each column to a suitable type.
Last week, I added some entities to the platform, tested the DB and it responded quickly. After a week of data insertions, the DB grew and now it is not usable anymore. It has no more than 200.000 records but it is really slow. 
Digging a little bit deeper, I looked in the source code and found out that, actually, each column is created with type `text`:
https://github.com/telefonicaid/fiware-cygnus/blob/424fb851e2b90df989a0341f8f2c4d5f68e14aec/cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/sinks/NGSIMySQLSink.java#L401

```java
        public String getFieldsForCreate() {
            String fieldsForCreate = "(";
            boolean first = true;
            Iterator<String> it = aggregation.keySet().iterator();
            
            while (it.hasNext()) {
                if (first) {
                    fieldsForCreate += (String) it.next() + " text";
                    first = false;
                } else {
                    fieldsForCreate += "," + (String) it.next() + " text";
                } // if else
            } // while
            
            return fieldsForCreate + ")";
        } // getFieldsForCreate
```
In my opinion, this is a **big** TODO
diff --git a/CHANGES_NEXT_RELEASE b/CHANGES_NEXT_RELEASE
index 86524f25..473ae82d 100644
--- a/CHANGES_NEXT_RELEASE
+++ b/CHANGES_NEXT_RELEASE
@@ -1,3 +1,4 @@
+[cygnus-ngsi][MySQL] Insert right SQL types values (Number, Null), non stringify (#1661)
 [doc] Add Install Section in README.md (#1576)
 [cygnus-ngsi] Fix error response code from 501 to 404 in case of invalid API (#1630)
 [cygnus-ngsi][ArcGis] Add ArcGis sink (#1672)
diff --git a/cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/sinks/NGSIMySQLSink.java b/cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/sinks/NGSIMySQLSink.java
index 6e7afe17..964bab9f 100644
--- a/cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/sinks/NGSIMySQLSink.java
+++ b/cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/sinks/NGSIMySQLSink.java
@@ -57,7 +57,8 @@ public class NGSIMySQLSink extends NGSISink {
     private static final String DEFAULT_HOST = "localhost";
     private static final String DEFAULT_USER_NAME = "root";
     private static final int DEFAULT_MAX_POOL_SIZE = 3;
-    
+    private static final String DEFAULT_ATTR_NATIVE_TYPES = "false";
+
     private static final CygnusLogger LOGGER = new CygnusLogger(NGSIMySQLSink.class);
     private String mysqlHost;
     private String mysqlPort;
@@ -66,7 +67,8 @@ public class NGSIMySQLSink extends NGSISink {
     private int maxPoolSize;
     private boolean rowAttrPersistence;
     private MySQLBackendImpl persistenceBackend;
-    
+    private boolean attrNativeTypes;
+
     /**
      * Constructor.
      */
@@ -130,6 +132,16 @@ public class NGSIMySQLSink extends NGSISink {
     protected void setPersistenceBackend(MySQLBackendImpl persistenceBackend) {
         this.persistenceBackend = persistenceBackend;
     } // setPersistenceBackend
+
+
+   /**
+     * Returns if the attribute value will be native or stringfy. It will be stringfy due to backward compatibility
+     * purposes.
+     * @return True if the attribute value will be native, false otherwise
+     */
+    protected boolean getNativeAttrTypes() {
+        return attrNativeTypes;
+    } // attrNativeTypes
     
     @Override
     public void configure(Context context) {
@@ -166,7 +178,17 @@ public class NGSIMySQLSink extends NGSISink {
             LOGGER.warn("[" + this.getName() + "] Invalid configuration (attr_persistence="
                 + persistence + ") must be 'row' or 'column'");
         }  // if else
-        
+
+        String attrNativeTypesStr = context.getString("attr_native_types", DEFAULT_ATTR_NATIVE_TYPES);
+        if (attrNativeTypesStr.equals("true") || attrNativeTypesStr.equals("false")) {
+            attrNativeTypes = Boolean.valueOf(attrNativeTypesStr);
+            LOGGER.debug("[" + this.getName() + "] Reading configuration (attr_native_types=" + attrNativeTypesStr + ")");
+        } else {
+            invalidConfiguration = true;
+            LOGGER.debug("[" + this.getName() + "] Invalid configuration (attr_native_types="
+                + attrNativeTypesStr + ") -- Must be 'true' or 'false'");
+        } // if else
+
         super.configure(context);
     } // configure
 
@@ -384,11 +406,18 @@ public class NGSIMySQLSink extends NGSISink {
             
                 while (it.hasNext()) {
                     ArrayList<String> values = (ArrayList<String>) aggregation.get((String) it.next());
+                    String valueType = (String) it.next();
+                    String value = values.get(i);
+                    if (value == null) { // TBD check also if value == "" ?
+                        value = "NULL";
+                    } else if (!(attrNativeTypes && valueType.equals("Number"))) {
+                        value = "'" + value + "'";
+                    } // if valueType is Number then value is value without ' '
                     if (first) {
-                        valuesForInsert += "'" + values.get(i) + "'";
+                        valuesForInsert += value;
                         first = false;
                     } else {
-                        valuesForInsert += ",'" + values.get(i) + "'";
+                        valuesForInsert += "," + value;
                     } // if else
                 } // while
                 
diff --git a/doc/cygnus-ngsi/flume_extensions_catalogue/ngsi_mysql_sink.md b/doc/cygnus-ngsi/flume_extensions_catalogue/ngsi_mysql_sink.md
index d7d1f258..15a40aea 100644
--- a/doc/cygnus-ngsi/flume_extensions_catalogue/ngsi_mysql_sink.md
+++ b/doc/cygnus-ngsi/flume_extensions_catalogue/ngsi_mysql_sink.md
@@ -220,6 +220,7 @@ If `attr_persistence=colum` then `NGSIMySQLSink` will persist the data within th
 | persistence\_policy.max_records | no | -1 | Maximum number of records allowed for a table before it is capped. `-1` disables this policy. |
 | persistence\_policy.expiration_time | no | -1 | Maximum number of seconds a record is maintained in a table before expiration. `-1` disables this policy. |
 | persistence\_policy.checking_time | no | 3600 | Frequency (in seconds) at which the sink checks for record expiration. |
+| attr\_native\_types | no | false | if the attribute value will be native <i>true</i> or stringfy or <i>false</i>. |
 
 A configuration example could be:
 
@@ -239,6 +240,7 @@ A configuration example could be:
     cygnus-ngsi.sinks.mysql-sink.mysql_password = mypassword
     cygnus-ngsi.sinks.mysql-sink.mysql_maxPoolSize = 3
     cygnus-ngsi.sinks.mysql-sink.attr_persistence = row
+    cygnus-ngsi.sinks.mysql-sink.attr_native_types = false
     cygnus-ngsi.sinks.mysql-sink.batch_size = 100
     cygnus-ngsi.sinks.mysql-sink.batch_timeout = 30
     cygnus-ngsi.sinks.mysql-sink.batch_ttl = 10
diff --git a/doc/cygnus-ngsi/installation_and_administration_guide/install_with_docker.md b/doc/cygnus-ngsi/installation_and_administration_guide/install_with_docker.md
index 5c166e95..5000a86b 100644
--- a/doc/cygnus-ngsi/installation_and_administration_guide/install_with_docker.md
+++ b/doc/cygnus-ngsi/installation_and_administration_guide/install_with_docker.md
@@ -190,6 +190,7 @@ As seen above, the default configuration distributed with the image is tied to c
     * CYGNUS_MYSQL_ENABLE_LOWERCASE: true if lower case is wanted to forced in all the element names, false otherwise.
     * CYGNUS_MYSQL_DATA_MODEL: select the data_model: dm-by-service-path, dm-by-entity or dm-by-entity-type.
     * CYGNUS_MYSQL_ATTR_PERSISTENCE: how the attributes are stored, either per row either per column (row, column).
+    * CYGNUS_MYSQL_ATTR_NATIVE_TYPES: how the attribute are stored, using native type (true) or stringfy (false, by default).
     * CYGNUS_MYSQL_BATCH_SIZE: number of notifications to be included within a processing batch.
     * CYGNUS_MYSQL_BATCH_TIMEOUT: timeout for batch accumulation in seconds.
     * CYGNUS_MYSQL_BATCH_TTL: number of retries upon persistence error.
diff --git a/docker/cygnus-ngsi/agent.conf b/docker/cygnus-ngsi/agent.conf
index 9ac20759..702ffbeb 100644
--- a/docker/cygnus-ngsi/agent.conf
+++ b/docker/cygnus-ngsi/agent.conf
@@ -46,6 +46,7 @@ cygnus-ngsi.sinks.mysql-sink.mysql_port = 3306
 cygnus-ngsi.sinks.mysql-sink.mysql_username =
 cygnus-ngsi.sinks.mysql-sink.mysql_password =
 #cygnus-ngsi.sinks.mysql-sink.attr_persistence = row
+#cygnus-ngsi.sinks.mysql-sink.attr_native_types = false
 #cygnus-ngsi.sinks.mysql-sink.batch_size = 1
 #cygnus-ngsi.sinks.mysql-sink.batch_timeout = 30
 #cygnus-ngsi.sinks.mysql-sink.batch_ttl = 10
['doc/cygnus-ngsi/flume_extensions_catalogue/ngsi_mysql_sink.md', 'doc/cygnus-ngsi/installation_and_administration_guide/install_with_docker.md', 'cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/sinks/NGSIMySQLSink.java', 'docker/cygnus-ngsi/agent.conf', 'CHANGES_NEXT_RELEASE']