Repository: influxdata/influxdb-client-python

Issue:
using Pandas DataFrame can corrupt data
The code in [dataframe_serializer.py](https://github.com/influxdata/influxdb-client-python/blob/91dcafb5e1c7b7854cf225afba87f5bdfe00b3a1/influxdb_client/client/write/dataframe_serializer.py) could use some improvement.

- the code is hard to understand. It operates by building up a string expression and then using `eval`, which is hard to understand, error-prone, and has potential for security flaws.
- when there are null values in the data, it passes the encoded points through a regular-expression-based translation phase which can corrupt data.
- the line-protocol encoding code amounts to an independent encoder to the encoder in [influxdb_client/client/write/point.py](https://github.com/influxdata/influxdb-client-python/blob/91dcafb5e1c7b7854cf225afba87f5bdfe00b3a1/influxdb_client/client/write/point.py), which means there's room for independent encoding bugs.
- the code has undocumented (and probably unwanted side-effects) on the data frame that's being encoded. Default tags will be added directly to the `data_frame` object rather than being added only to the encoded line-protocol points.
- Any `DataFrame` index values will be converted to timestamps regardless of whether that's appropriate or not. It would be better if that was something to explicitly opt into (converting the default `RangeIndex` index into time values is almost never going to be correct - a better default would probably to omit the time stamps and let the server add them).

Here is some example code that demonstrates corruption of data:

```
import pandas as pd
from influxdb_client.client.write_api import WriteOptions, WriteApi, PointSettings
from influxdb_client.client.write.point import Point
from influxdb_client.client.write.dataframe_serializer import data_frame_to_list_of_points

frame = pd.DataFrame(
	data=[
	    ["coyote_creek", 1.0, "a"],
	    ["coyote_creek", None, "b"],
	    ["coyote_creek", 3.0, "c"],
	    ["coyote_creek", 4.0, "d"],
	],
	index=[1, 2, 3, 4],
	columns=["location", "level water_level", "str"],
)
ps=data_frame_to_list_of_points(frame, PointSettings(), data_frame_measurement_name='h2o_feet', data_frame_tag_columns=['location'])
for p in ps:
   print(p)
```

This prints:

```
h2o_feet,location=coyote_creek level\ water_level=1.0,str="comma separated string" 1
h2o_feet,location=coyote_creek level\ water_level=nan,str="b" 2
h2o_feet,location=coyote_creek level\ water_level=3.0,str="cff" 3
h2o_feet,location=coyote_creek level\ water_level=4.0,str="d" 4
```

Note that two string values are corrupted and the null value still remains because the regexp code has not taken account of the escaped key.

I would expect it to print this instead:

```
h2o_feet,location=coyote_creek level\ water_level=1.0,str="comma, separated, string" 1
h2o_feet,location=coyote_creek str="b" 2
h2o_feet,location=coyote_creek level\ water_level=3.0,str="clocation=nanff" 3
h2o_feet,location=coyote_creek level\ water_level=4.0,str="d" 4
```


Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/influxdata/influxdb-client-python/pull/183?src=pr&el=h1) Report
> Merging [#183](https://codecov.io/gh/influxdata/influxdb-client-python/pull/183?src=pr&el=desc) (f4be9f4) into [master](https://codecov.io/gh/influxdata/influxdb-client-python/commit/91dcafb5e1c7b7854cf225afba87f5bdfe00b3a1?el=desc) (91dcafb) will **decrease** coverage by `0.05%`.
> The diff coverage is `95.91%`.

[![Impacted file tree graph](https://codecov.io/gh/influxdata/influxdb-client-python/pull/183/graphs/tree.svg?width=650&height=150&src=pr&token=PcyupQSPFt)](https://codecov.io/gh/influxdata/influxdb-client-python/pull/183?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #183      +/-   ##
==========================================
- Coverage   89.72%   89.67%   -0.06%     
==========================================
  Files          26       26              
  Lines        1957     1966       +9     
==========================================
+ Hits         1756     1763       +7     
- Misses        201      203       +2     
```


| [Impacted Files](https://codecov.io/gh/influxdata/influxdb-client-python/pull/183?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...fluxdb\_client/client/write/dataframe\_serializer.py](https://codecov.io/gh/influxdata/influxdb-client-python/pull/183/diff?src=pr&el=tree#diff-aW5mbHV4ZGJfY2xpZW50L2NsaWVudC93cml0ZS9kYXRhZnJhbWVfc2VyaWFsaXplci5weQ==) | `94.87% <95.65%> (-2.23%)` | :arrow_down: |
| [influxdb\_client/client/write/point.py](https://codecov.io/gh/influxdata/influxdb-client-python/pull/183/diff?src=pr&el=tree#diff-aW5mbHV4ZGJfY2xpZW50L2NsaWVudC93cml0ZS9wb2ludC5weQ==) | `98.36% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/influxdata/influxdb-client-python/pull/183?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/influxdata/influxdb-client-python/pull/183?src=pr&el=footer). Last update [91dcafb...f4be9f4](https://codecov.io/gh/influxdata/influxdb-client-python/pull/183?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/.gitignore b/.gitignore
index 59b4156..3be131d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -114,4 +114,3 @@ sandbox
 # OpenAPI-generator
 /.openapi-generator*
 **/writer.pickle
-/tests/data_frame_file.csv
diff --git a/CHANGELOG.md b/CHANGELOG.md
index db72fe7..50cd774 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,13 +6,16 @@
 ### CI
 1. [#179](https://github.com/influxdata/influxdb-client-python/pull/179): Updated default docker image to v2.0.3
 
+### Bug Fixes
+1. [#183](https://github.com/influxdata/influxdb-client-python/pull/183): Fixes to DataFrame writing.
+
 ## 1.13.0 [2020-12-04]
 
 ### Features
 1. [#171](https://github.com/influxdata/influxdb-client-python/pull/171): CSV parser is able to parse export from UI
 
 ### Bug Fixes
-1. [#170](https://github.com/influxdata/influxdb-client-python/pull/170): Skip DataFrame rows without data - all fields are nan. 
+1. [#170](https://github.com/influxdata/influxdb-client-python/pull/170): Skip DataFrame rows without data - all fields are nan.
 
 ### CI
 1. [#175](https://github.com/influxdata/influxdb-client-python/pull/175): Updated default docker image to v2.0.2
@@ -25,7 +28,7 @@
 1. [#161](https://github.com/influxdata/influxdb-client-python/pull/161): Added logging message for retries
 
 ### Bug Fixes
-1. [#164](https://github.com/influxdata/influxdb-client-python/pull/164): Excluded tests from packaging 
+1. [#164](https://github.com/influxdata/influxdb-client-python/pull/164): Excluded tests from packaging
 
 ## 1.11.0 [2020-10-02]
 
@@ -38,16 +41,16 @@
 1. [#156](https://github.com/influxdata/influxdb-client-python/pull/156): Removed labels in organization API, removed Pkg* structure and package service
 
 ### Bug Fixes
-1. [#154](https://github.com/influxdata/influxdb-client-python/pull/154): Fixed escaping string fields in DataFrame serialization 
+1. [#154](https://github.com/influxdata/influxdb-client-python/pull/154): Fixed escaping string fields in DataFrame serialization
 
 ## 1.10.0 [2020-08-14]
 
 ### Features
-1. [#140](https://github.com/influxdata/influxdb-client-python/pull/140): Added exponential backoff strategy for batching writes, Allowed to configure default retry strategy. Default value for `retry_interval` is 5_000 milliseconds.   
+1. [#140](https://github.com/influxdata/influxdb-client-python/pull/140): Added exponential backoff strategy for batching writes, Allowed to configure default retry strategy. Default value for `retry_interval` is 5_000 milliseconds.
 1. [#136](https://github.com/influxdata/influxdb-client-python/pull/136): Allows users to skip of verifying SSL certificate
 1. [#143](https://github.com/influxdata/influxdb-client-python/pull/143): Skip of verifying SSL certificate could be configured via config file or environment properties
 1. [#141](https://github.com/influxdata/influxdb-client-python/pull/141): Added possibility to use datetime nanoseconds precision by `pandas.Timestamp`
-1. [#145](https://github.com/influxdata/influxdb-client-python/pull/145): Api generator was moved to influxdb-clients-apigen  
+1. [#145](https://github.com/influxdata/influxdb-client-python/pull/145): Api generator was moved to influxdb-clients-apigen
 
 ## 1.9.0 [2020-07-17]
 
@@ -58,10 +61,10 @@
 1. [#132](https://github.com/influxdata/influxdb-client-python/pull/132): Use microseconds resolutions for data points
 
 ### Bug Fixes
-1. [#117](https://github.com/influxdata/influxdb-client-python/pull/117): Fixed appending default tags for single Point 
-1. [#115](https://github.com/influxdata/influxdb-client-python/pull/115): Fixed serialization of `\n`, `\r` and `\t` to Line Protocol, `=` is valid sign for measurement name 
+1. [#117](https://github.com/influxdata/influxdb-client-python/pull/117): Fixed appending default tags for single Point
+1. [#115](https://github.com/influxdata/influxdb-client-python/pull/115): Fixed serialization of `\n`, `\r` and `\t` to Line Protocol, `=` is valid sign for measurement name
 1. [#118](https://github.com/influxdata/influxdb-client-python/issues/118): Fixed serialization of DataFrame with empty (NaN) values
-1. [#130](https://github.com/influxdata/influxdb-client-python/pull/130): Use `Retry-After` header value for Retryable error codes 
+1. [#130](https://github.com/influxdata/influxdb-client-python/pull/130): Use `Retry-After` header value for Retryable error codes
 
 ## 1.8.0 [2020-06-19]
 
@@ -153,13 +156,13 @@
 1. [#27](https://github.com/influxdata/influxdb-client-python/issues/27): Added possibility to write bytes type of data
 1. [#30](https://github.com/influxdata/influxdb-client-python/issues/30): Added support for streaming a query response
 1. [#35](https://github.com/influxdata/influxdb-client-python/pull/35): FluxRecord supports dictionary-style access
-1. [#31](https://github.com/influxdata/influxdb-client-python/issues/31): Added support for delete metrics  
+1. [#31](https://github.com/influxdata/influxdb-client-python/issues/31): Added support for delete metrics
 
 ### API
 1. [#28](https://github.com/bonitoo-io/influxdb-client-python/pull/28): Updated swagger to latest version
 
 ### Bug Fixes
-1. [#19](https://github.com/bonitoo-io/influxdb-client-python/pull/19): Removed strict checking of enum values 
+1. [#19](https://github.com/bonitoo-io/influxdb-client-python/pull/19): Removed strict checking of enum values
 
 ### Documentation
 1. [#22](https://github.com/bonitoo-io/influxdb-client-python/issues/22): Documented how to connect to InfluxCloud
@@ -168,8 +171,8 @@
 
 ### Features
 1. [#2](https://github.com/bonitoo-io/influxdb-client-python/issues/2): The write client is able to write data in batches (configuration: `batch_size`, `flush_interval`, `jitter_interval`, `retry_interval`)
-1. [#5](https://github.com/bonitoo-io/influxdb-client-python/issues/5): Added support for gzip compression of query response and write body 
- 
+1. [#5](https://github.com/bonitoo-io/influxdb-client-python/issues/5): Added support for gzip compression of query response and write body
+
 ### API
 1. [#10](https://github.com/bonitoo-io/influxdb-client-python/pull/10): Updated swagger to latest version
 
@@ -178,5 +181,5 @@
 1. [#7](https://github.com/bonitoo-io/influxdb-client-python/issues/7): Drop NaN and infinity values from fields when writing to InfluxDB
 
 ### CI
-1. [#11](https://github.com/bonitoo-io/influxdb-client-python/pull/11): Switch CI to CircleCI 
-1. [#12](https://github.com/bonitoo-io/influxdb-client-python/pull/12): CI generate code coverage report on CircleCI 
+1. [#11](https://github.com/bonitoo-io/influxdb-client-python/pull/11): Switch CI to CircleCI
+1. [#12](https://github.com/bonitoo-io/influxdb-client-python/pull/12): CI generate code coverage report on CircleCI
diff --git a/README.rst b/README.rst
index 8f7c2a8..dc4af4b 100644
--- a/README.rst
+++ b/README.rst
@@ -46,7 +46,7 @@ InfluxDB 2.0 client features
     - into csv, raw data, `flux_table <https://github.com/influxdata/influxdb-client-python/blob/master/influxdb_client/client/flux_table.py#L5>`_ structure, `Pandas DataFrame <https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.html>`_
     - `How to queries <#queries>`_
 - Writing data using
-    - `Line Protocol <https://docs.influxdata.com/influxdb/latest/write_protocols/line_protocol_tutorial>`_
+    - `Line Protocol <https://docs.influxdata.com/influxdb/latest/reference/syntax/line-protocol>`_
     - `Data Point <https://github.com/influxdata/influxdb-client-python/blob/master/influxdb_client/client/write/point.py#L16>`__
     - `RxPY <https://rxpy.readthedocs.io/en/latest/>`__ Observable
     - `Pandas DataFrame <https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.html>`_
@@ -908,7 +908,7 @@ The following forward compatible APIs are available:
 =======================================================  ====================================================================================================  =======
  API                                                     Endpoint                                                                                              Description
 =======================================================  ====================================================================================================  =======
-`query_api.py <influxdb_client/client/query_api.py>`_    `/api/v2/query <https://docs.influxdata.com/influxdb/latest/tools/api/#api-v2-query-http-endpoint>`_  Query data in InfluxDB 1.8.0+ using the InfluxDB 2.0 API and `Flux <https://docs.influxdata.com/flux/latest/>`_ (endpoint should be enabled by `flux-enabled option <https://docs.influxdata.com/influxdb/latest/administration/config/#flux-enabled-false>`_)
+`query_api.py <influxdb_client/client/query_api.py>`_    `/api/v2/query <https://docs.influxdata.com/influxdb/latest/tools/api/#api-v2-query-http-endpoint>`_  Query data in InfluxDB 1.8.0+ using the InfluxDB 2.0 API and `Flux <https://docs.influxdata.com/flux/latest/>`_ (endpoint should be enabled by `flux-enabled option <https://docs.influxdata.com/influxdb/v1.8/administration/config/#flux-enabled-false>`_)
 `write_api.py <influxdb_client/client/write_api.py>`_    `/api/v2/write <https://docs.influxdata.com/influxdb/latest/tools/api/#api-v2-write-http-endpoint>`_  Write data to InfluxDB 1.8.0+ using the InfluxDB 2.0 API
 `health() <influxdb_client/client/influxdb_client.py>`_  `/health <https://docs.influxdata.com/influxdb/latest/tools/api/#health-http-endpointt>`_             Check the health of your InfluxDB instance
 =======================================================  ====================================================================================================  =======
diff --git a/influxdb_client/client/write/dataframe_serializer.py b/influxdb_client/client/write/dataframe_serializer.py
index 6abbddc..ee1d9ba 100644
--- a/influxdb_client/client/write/dataframe_serializer.py
+++ b/influxdb_client/client/write/dataframe_serializer.py
@@ -5,115 +5,199 @@ Much of the code here is inspired by that in the aioinflux packet found here: ht
 """
 
 import re
-from functools import reduce
-from itertools import chain
+import math
 
 from influxdb_client.client.write.point import _ESCAPE_KEY, _ESCAPE_STRING, _ESCAPE_MEASUREMENT
 
 
-def _replace(data_frame):
-    from ...extras import np
-
-    # string columns
-    obj_cols = {k for k, v in dict(data_frame.dtypes).items() if v is np.dtype('O')}
-
-    # number columns
-    other_cols = set(data_frame.columns) - obj_cols
-
-    obj_nans = (f'{k}=nan' for k in obj_cols)
-    other_nans = (f'{k}=nani?' for k in other_cols)
-
-    replacements = [
-        ('|'.join(chain(obj_nans, other_nans)), ''),
-        (',{2,}', ','),
-        ('|'.join([', ,', ', ', ' ,']), ' '),
-    ]
-
-    return replacements
-
-
 def _itertuples(data_frame):
     cols = [data_frame.iloc[:, k] for k in range(len(data_frame.columns))]
     return zip(data_frame.index, *cols)
 
 
-def _is_nan(x):
-    return x != x
+def _not_nan(x):
+    return x == x
 
 
 def _any_not_nan(p, indexes):
-    return any(map(lambda inx: not _is_nan(p[inx]), indexes))
+    return any(map(lambda x: _not_nan(p[x]), indexes))
 
 
 def data_frame_to_list_of_points(data_frame, point_settings, **kwargs):
     """Serialize DataFrame into LineProtocols."""
+    # This function is hard to understand but for good reason:
+    # the approach used here is considerably more efficient
+    # than the alternatives.
+    #
+    # We build up a Python expression that efficiently converts a data point
+    # tuple into line-protocol entry, and then evaluate the expression
+    # as a lambda so that we can call it. This avoids the overhead of
+    # invoking a function on every data value - we only have one function
+    # call per row instead. The expression consists of exactly
+    # one f-string, so we build up the parts of it as segments
+    # that are concatenated together to make the full f-string inside
+    # the lambda.
+    #
+    # Things are made a little more complex because fields and tags with NaN
+    # values and empty tags are omitted from the generated line-protocol
+    # output.
+    #
+    # As an example, say we have a data frame with two value columns:
+    #        a float
+    #        b int
+    #
+    # This will generate a lambda expression to be evaluated that looks like
+    # this:
+    #
+    #     lambda p: f"""{measurement_name} {keys[0]}={p[1]},{keys[1]}={p[2]}i {p[0].value}"""
+    #
+    # This lambda is then executed for each row p.
+    #
+    # When NaNs are present, the expression looks like this (split
+    # across two lines to satisfy the code-style checker)
+    #
+    #    lambda p: f"""{measurement_name} {"" if math.isnan(p[1])
+    #    else f"{keys[0]}={p[1]}"},{keys[1]}={p[2]}i {p[0].value}"""
+    #
+    # When there's a NaN value in column a, we'll end up with a comma at the start of the
+    # fields, so we run a regexp substitution after generating the line-protocol entries
+    # to remove this.
+    #
+    # We're careful to run these potentially costly extra steps only when NaN values actually
+    # exist in the data.
+
     from ...extras import pd, np
     if not isinstance(data_frame, pd.DataFrame):
         raise TypeError('Must be DataFrame, but type was: {0}.'
                         .format(type(data_frame)))
 
-    if 'data_frame_measurement_name' not in kwargs:
+    data_frame_measurement_name = kwargs.get('data_frame_measurement_name')
+    if data_frame_measurement_name is None:
         raise TypeError('"data_frame_measurement_name" is a Required Argument')
 
+    data_frame = data_frame.copy(deep=False)
     if isinstance(data_frame.index, pd.PeriodIndex):
         data_frame.index = data_frame.index.to_timestamp()
     else:
+        # TODO: this is almost certainly not what you want
+        # when the index is the default RangeIndex.
+        # Instead, it would probably be better to leave
+        # out the timestamp unless a time column is explicitly
+        # enabled.
         data_frame.index = pd.to_datetime(data_frame.index)
 
     if data_frame.index.tzinfo is None:
         data_frame.index = data_frame.index.tz_localize('UTC')
 
-    measurement_name = str(kwargs.get('data_frame_measurement_name')).translate(_ESCAPE_MEASUREMENT)
     data_frame_tag_columns = kwargs.get('data_frame_tag_columns')
     data_frame_tag_columns = set(data_frame_tag_columns or [])
 
+    # keys holds a list of string keys.
+    keys = []
+    # tags holds a list of tag f-string segments ordered alphabetically by tag key.
     tags = []
+    # fields holds a list of field f-string segments  ordered alphebetically by field key
     fields = []
-    fields_indexes = []
-    keys = []
+    # field_indexes holds the index into each row of all the fields.
+    field_indexes = []
 
     if point_settings.defaultTags:
         for key, value in point_settings.defaultTags.items():
-            data_frame[key] = value
-            data_frame_tag_columns.add(key)
-
-    for index, (key, value) in enumerate(data_frame.dtypes.items()):
+            # Avoid overwriting existing data if there's a column
+            # that already exists with the default tag's name.
+            # Note: when a new column is added, the old DataFrame
+            # that we've made a shallow copy of is unaffected.
+            # TODO: when there are NaN or empty values in
+            # the column, we could make a deep copy of the
+            # data and fill in those values with the default tag value.
+            if key not in data_frame.columns:
+                data_frame[key] = value
+                data_frame_tag_columns.add(key)
+
+    # Get a list of all the columns sorted by field/tag key.
+    # We want to iterate through the columns in sorted order
+    # so that we know when we're on the first field so we
+    # can know whether a comma is needed for that
+    # field.
+    columns = sorted(enumerate(data_frame.dtypes.items()), key=lambda col: col[1][0])
+
+    # null_columns has a bool value for each column holding
+    # whether that column contains any null (NaN or None) values.
+    null_columns = data_frame.isnull().any()
+
+    # Iterate through the columns building up the expression for each column.
+    for index, (key, value) in columns:
         key = str(key)
+        key_format = f'{{keys[{len(keys)}]}}'
         keys.append(key.translate(_ESCAPE_KEY))
-        key_format = f'{{keys[{index}]}}'
+        # The field index is one more than the column index because the
+        # time index is at column zero in the finally zipped-together
+        # result columns.
+        field_index = index + 1
+        val_format = f'p[{field_index}]'
 
-        index_value = index + 1
         if key in data_frame_tag_columns:
-            tags.append({'key': key, 'value': f"{key_format}={{str(p[{index_value}]).translate(_ESCAPE_KEY)}}"})
-        elif issubclass(value.type, np.integer):
-            fields.append(f"{key_format}={{p[{index_value}]}}i")
-            fields_indexes.append(index_value)
-        elif issubclass(value.type, (np.float, np.bool_)):
-            fields.append(f"{key_format}={{p[{index_value}]}}")
-            fields_indexes.append(index_value)
+            # This column is a tag column.
+            if null_columns[index]:
+                key_value = f"""{{
+                    '' if {val_format} == '' or type({val_format}) == float and math.isnan({val_format}) else
+                    f',{key_format}={{str({val_format}).translate(_ESCAPE_STRING)}}'
+                }}"""
+            else:
+                key_value = f',{key_format}={{str({val_format}).translate(_ESCAPE_KEY)}}'
+            tags.append(key_value)
+            continue
+
+        # This column is a field column.
+        # Note: no comma separator is needed for the first field.
+        # It's important to omit it because when the first
+        # field column has no nulls, we don't run the comma-removal
+        # regexp substitution step.
+        sep = '' if len(field_indexes) == 0 else ','
+        if issubclass(value.type, np.integer):
+            field_value = f"{sep}{key_format}={{{val_format}}}i"
+        elif issubclass(value.type, np.bool_):
+            field_value = f'{sep}{key_format}={{{val_format}}}'
+        elif issubclass(value.type, np.float):
+            if null_columns[index]:
+                field_value = f"""{{"" if math.isnan({val_format}) else f"{sep}{key_format}={{{val_format}}}"}}"""
+            else:
+                field_value = f'{sep}{key_format}={{{val_format}}}'
         else:
-            fields.append(f"{key_format}=\"{{str(p[{index_value}]).translate(_ESCAPE_STRING)}}\"")
-            fields_indexes.append(index_value)
-
-    tags.sort(key=lambda x: x['key'])
-    tags = ','.join(map(lambda y: y['value'], tags))
-
-    fmt = ('{measurement_name}', f'{"," if tags else ""}', tags,
-           ' ', ','.join(fields), ' {p[0].value}')
-    f = eval("lambda p: f'{}'".format(''.join(fmt)),
-             {'measurement_name': measurement_name, '_ESCAPE_KEY': _ESCAPE_KEY, '_ESCAPE_STRING': _ESCAPE_STRING,
-              'keys': keys})
+            if null_columns[index]:
+                field_value = f"""{{
+                    '' if type({val_format}) == float and math.isnan({val_format}) else
+                    f'{sep}{key_format}="{{str({val_format}).translate(_ESCAPE_STRING)}}"'
+                }}"""
+            else:
+                field_value = f'''{sep}{key_format}="{{str({val_format}).translate(_ESCAPE_STRING)}}"'''
+        field_indexes.append(field_index)
+        fields.append(field_value)
+
+    measurement_name = str(data_frame_measurement_name).translate(_ESCAPE_MEASUREMENT)
+
+    tags = ''.join(tags)
+    fields = ''.join(fields)
+    timestamp = '{p[0].value}'
+
+    f = eval(f'lambda p: f"""{{measurement_name}}{tags} {fields} {timestamp}"""', {
+        'measurement_name': measurement_name,
+        '_ESCAPE_KEY': _ESCAPE_KEY,
+        '_ESCAPE_STRING': _ESCAPE_STRING,
+        'keys': keys,
+        'math': math,
+    })
 
     for k, v in dict(data_frame.dtypes).items():
         if k in data_frame_tag_columns:
             data_frame[k].replace('', np.nan, inplace=True)
 
-    isnull = data_frame.isnull().any(axis=1)
-
-    if isnull.any():
-        rep = _replace(data_frame)
-        lp = (reduce(lambda a, b: re.sub(*b, a), rep, f(p))
-              for p in filter(lambda x: _any_not_nan(x, fields_indexes), _itertuples(data_frame)))
+    first_field_maybe_null = null_columns[field_indexes[0] - 1]
+    if first_field_maybe_null:
+        # When the first field is null (None/NaN), we'll have
+        # a spurious leading comma which needs to be removed.
+        lp = (re.sub('^((\\ |[^ ])* ),', '\\1', f(p))
+              for p in filter(lambda x: _any_not_nan(x, field_indexes), _itertuples(data_frame)))
         return list(lp)
     else:
         return list(map(f, _itertuples(data_frame)))
diff --git a/influxdb_client/client/write/point.py b/influxdb_client/client/write/point.py
index a09099c..f862d1d 100644
--- a/influxdb_client/client/write/point.py
+++ b/influxdb_client/client/write/point.py
@@ -14,10 +14,32 @@ from influxdb_client.client.util.date_utils import get_date_helper
 from influxdb_client.domain.write_precision import WritePrecision
 
 EPOCH = UTC.localize(datetime.utcfromtimestamp(0))
+
 DEFAULT_WRITE_PRECISION = WritePrecision.NS
-_ESCAPE_MEASUREMENT = str.maketrans({'\\': '\\\\', ',': r'\,', ' ': r'\ ', '\n': '\\n', '\t': '\\t', '\r': '\\r'})
-_ESCAPE_KEY = str.maketrans({'\\': '\\\\', ',': r'\,', ' ': r'\ ', '=': r'\=', '\n': '\\n', '\t': '\\t', '\r': '\\r'})
-_ESCAPE_STRING = str.maketrans({'\"': r"\"", "\\": r"\\"})
+
+_ESCAPE_MEASUREMENT = str.maketrans({
+    '\\': r'\\',      # Note: this is wrong. Backslashes are not escaped like this in measurements.
+    ',': r'\,',
+    ' ': r'\ ',
+    '\n': r'\n',
+    '\t': r'\t',
+    '\r': r'\r',
+})
+
+_ESCAPE_KEY = str.maketrans({
+    '\\': r'\\',      # Note: this is wrong. Backslashes are not escaped like this in keys.
+    ',': r'\,',
+    '=': r'\=',
+    ' ': r'\ ',
+    '\n': r'\n',
+    '\t': r'\t',
+    '\r': r'\r',
+})
+
+_ESCAPE_STRING = str.maketrans({
+    '"': r'\"',
+    '\\': r'\\',
+})
 
 
 class Point(object):


Test patch:
diff --git a/tests/test_WriteApiDataFrame.py b/tests/test_WriteApiDataFrame.py
index 63de0c8..4f01202 100644
--- a/tests/test_WriteApiDataFrame.py
+++ b/tests/test_WriteApiDataFrame.py
@@ -23,42 +23,6 @@ class DataFrameWriteTest(BaseTest):
         super().tearDown()
         self._write_client.__del__()
 
-    @unittest.skip('Test big file')
-    def test_write_data_frame(self):
-        import random
-        from influxdb_client.extras import pd
-
-        if not os.path.isfile("data_frame_file.csv"):
-            with open('data_frame_file.csv', mode='w+') as csv_file:
-                _writer = csv.writer(csv_file, delimiter=',', quotechar='"', quoting=csv.QUOTE_MINIMAL)
-                _writer.writerow(['time', 'col1', 'col2', 'col3', 'col4', 'col5', 'col6', 'col7', 'col8'])
-
-                for i in range(1, 1500000):
-                    choice = ['test_a', 'test_b', 'test_c']
-                    _writer.writerow([i, random.choice(choice), 'test', random.random(), random.random(),
-                                      random.random(), random.random(), random.random(), random.random()])
-
-            csv_file.close()
-
-        with open('data_frame_file.csv', mode='rb') as csv_file:
-
-            data_frame = pd.read_csv(csv_file, index_col='time')
-            print(data_frame)
-
-            print('Writing...')
-
-            start = time.time()
-
-            self._write_client.write("my-bucket", "my-org", record=data_frame,
-                                     data_frame_measurement_name='h2o_feet',
-                                     data_frame_tag_columns=['location'])
-
-            self._write_client.__del__()
-
-            print("Time elapsed: ", (time.time() - start))
-
-        csv_file.close()
-
     def test_write_num_py(self):
         from influxdb_client.extras import pd, np
 
@@ -90,17 +54,40 @@ class DataFrameWriteTest(BaseTest):
 
 
 class DataSerializerTest(unittest.TestCase):
+    @unittest.skip('Test big data')
+    def test_convert_data_frame(self):
+        from influxdb_client.extras import pd, np
+
+        num_rows=1500000
+        col_data={
+            'time': np.arange(0, num_rows, 1, dtype=int),
+            'col1': np.random.choice(['test_a', 'test_b', 'test_c'], size=(num_rows,)),
+        }
+        for n in range(2, 9):
+            col_data[f'col{n}'] = np.random.rand(num_rows)
+
+        data_frame = pd.DataFrame(data=col_data)
+        print(data_frame)
+
+        start = time.time()
+        data_frame_to_list_of_points(data_frame, PointSettings(),
+            data_frame_measurement_name='h2o_feet',
+            data_frame_tag_columns=['location'])
+
+        print("Time elapsed: ", (time.time() - start))
 
     def test_write_nan(self):
         from influxdb_client.extras import pd, np
 
         now = pd.Timestamp('2020-04-05 00:00+00:00')
 
-        data_frame = pd.DataFrame(data=[[3.1955, np.nan, 20.514305, np.nan],
+        data_frame = pd.DataFrame(data=[
+                                        [3.1955, np.nan, 20.514305, np.nan],
                                         [5.7310, np.nan, 23.328710, np.nan],
                                         [np.nan, 3.138664, np.nan, 20.755026],
                                         [5.7310, 5.139563, 23.328710, 19.791240],
-                                        [np.nan, np.nan, np.nan, np.nan]],
+                                        [np.nan, np.nan, np.nan, np.nan],
+                                  ],
                                   index=[now, now + timedelta(minutes=30), now + timedelta(minutes=60),
                                          now + timedelta(minutes=90), now + timedelta(minutes=120)],
                                   columns=["actual_kw_price", "forecast_kw_price", "actual_general_use",
@@ -110,14 +97,14 @@ class DataSerializerTest(unittest.TestCase):
                                               data_frame_measurement_name='measurement')
 
         self.assertEqual(4, len(points))
-        self.assertEqual("measurement actual_kw_price=3.1955,actual_general_use=20.514305 1586044800000000000",
+        self.assertEqual("measurement actual_general_use=20.514305,actual_kw_price=3.1955 1586044800000000000",
                          points[0])
-        self.assertEqual("measurement actual_kw_price=5.731,actual_general_use=23.32871 1586046600000000000",
+        self.assertEqual("measurement actual_general_use=23.32871,actual_kw_price=5.731 1586046600000000000",
                          points[1])
-        self.assertEqual("measurement forecast_kw_price=3.138664,forecast_general_use=20.755026 1586048400000000000",
+        self.assertEqual("measurement forecast_general_use=20.755026,forecast_kw_price=3.138664 1586048400000000000",
                          points[2])
-        self.assertEqual("measurement actual_kw_price=5.731,forecast_kw_price=5.139563,actual_general_use=23.32871,"
-                         "forecast_general_use=19.79124 1586050200000000000",
+        self.assertEqual("measurement actual_general_use=23.32871,actual_kw_price=5.731,forecast_general_use=19.79124"
+                         ",forecast_kw_price=5.139563 1586050200000000000",
                          points[3])
 
     def test_write_tag_nan(self):
@@ -125,10 +112,12 @@ class DataSerializerTest(unittest.TestCase):
 
         now = pd.Timestamp('2020-04-05 00:00+00:00')
 
-        data_frame = pd.DataFrame(data=[["", 3.1955, 20.514305],
+        data_frame = pd.DataFrame(data=[
+                                        ["", 3.1955, 20.514305],
                                         ['', 5.7310, 23.328710],
                                         [np.nan, 5.7310, 23.328710],
-                                        ["tag", 3.138664, 20.755026]],
+                                        ["tag", 3.138664, 20.755026],
+                                  ],
                                   index=[now, now + timedelta(minutes=30),
                                          now + timedelta(minutes=60), now + timedelta(minutes=90)],
                                   columns=["tag", "actual_kw_price", "forecast_kw_price"])
@@ -148,12 +137,59 @@ class DataSerializerTest(unittest.TestCase):
         self.assertEqual("measurement,tag=tag actual_kw_price=3.138664,forecast_kw_price=20.755026 1586050200000000000",
                          points[3])
 
+    def test_write_object_field_nan(self):
+        from influxdb_client.extras import pd, np
+
+        now = pd.Timestamp('2020-04-05 00:00+00:00')
+
+        data_frame = pd.DataFrame(data=[
+                                        ["foo", 1],
+                                        [np.nan, 2],
+                                  ],
+                                  index=[now, now + timedelta(minutes=30)],
+                                  columns=["obj", "val"])
+
+        points = data_frame_to_list_of_points(data_frame=data_frame,
+                                              point_settings=PointSettings(),
+                                              data_frame_measurement_name='measurement')
+
+        self.assertEqual(2, len(points))
+        self.assertEqual("measurement obj=\"foo\",val=1i 1586044800000000000",
+                         points[0])
+        self.assertEqual("measurement val=2i 1586046600000000000",
+                         points[1])
+
+    def test_write_field_bool(self):
+        from influxdb_client.extras import pd, np
+
+        now = pd.Timestamp('2020-04-05 00:00+00:00')
+
+        data_frame = pd.DataFrame(data=[
+                                        [True],
+                                        [False],
+                                  ],
+                                  index=[now, now + timedelta(minutes=30)],
+                                  columns=["status"])
+
+        points = data_frame_to_list_of_points(data_frame=data_frame,
+                                              point_settings=PointSettings(),
+                                              data_frame_measurement_name='measurement')
+
+        self.assertEqual(2, len(points))
+        self.assertEqual("measurement status=True 1586044800000000000",
+                         points[0])
+        self.assertEqual("measurement status=False 1586046600000000000",
+                         points[1])
+
     def test_escaping_measurement(self):
         from influxdb_client.extras import pd, np
 
         now = pd.Timestamp('2020-04-05 00:00+00:00')
 
-        data_frame = pd.DataFrame(data=[["coyote_creek", np.int64(100.5)], ["coyote_creek", np.int64(200)]],
+        data_frame = pd.DataFrame(data=[
+                                        ["coyote_creek", np.int64(100.5)],
+                                        ["coyote_creek", np.int64(200)],
+                                  ],
                                   index=[now + timedelta(hours=1), now + timedelta(hours=2)],
                                   columns=["location", "water_level"])
 
@@ -235,3 +271,55 @@ class DataSerializerTest(unittest.TestCase):
         self.assertEqual(2, len(points))
         self.assertEqual("test,d=foo\\ bar b=\"hello world\",c=1i 1586041200000000000", points[0])
         self.assertEqual("test,d=bar\\ foo b=\"goodbye cruel world\",c=2i 1586044800000000000", points[1])
+
+    def test_with_default_tags(self):
+        from influxdb_client.extras import pd, np
+        now = pd.Timestamp('2020-04-05 00:00+00:00')
+        data_frame = pd.DataFrame(data={
+                                      'value': [1, 2],
+                                      't1': ['a1', 'a2'],
+                                      't3': ['c1', 'c2'],
+                                },
+                                index=[now + timedelta(hours=1), now + timedelta(hours=2)])
+        original_data = data_frame.copy()
+
+        points = data_frame_to_list_of_points(data_frame=data_frame,
+                                              point_settings=PointSettings(t2='every'),
+                                              data_frame_measurement_name='h2o',
+                                              data_frame_tag_columns={"t1", "t3"})
+
+        self.assertEqual(2, len(points))
+        self.assertEqual("h2o,t1=a1,t2=every,t3=c1 value=1i 1586048400000000000", points[0])
+        self.assertEqual("h2o,t1=a2,t2=every,t3=c2 value=2i 1586052000000000000", points[1])
+
+        # Check that the data frame hasn't been changed (an earlier version did change it)
+        self.assertEqual(True, (data_frame == original_data).all(axis = None), f'data changed; old:\n{original_data}\nnew:\n{data_frame}')
+
+        # Check that the default tags won't override actual column data.
+        # This is arguably incorrect behavior, but it's how it works currently.
+        points = data_frame_to_list_of_points(data_frame=data_frame,
+                                              point_settings=PointSettings(t1='every'),
+                                              data_frame_measurement_name='h2o',
+                                              data_frame_tag_columns={"t1", "t3"})
+
+        self.assertEqual(2, len(points))
+        self.assertEqual("h2o,t1=a1,t3=c1 value=1i 1586048400000000000", points[0])
+        self.assertEqual("h2o,t1=a2,t3=c2 value=2i 1586052000000000000", points[1])
+
+        self.assertEqual(True, (data_frame == original_data).all(axis = None), f'data changed; old:\n{original_data}\nnew:\n{data_frame}')
+
+    def test_with_period_index(self):
+        from influxdb_client.extras import pd, np
+        now = pd.Timestamp('2020-04-05 00:00+00:00')
+        data_frame = pd.DataFrame(data={
+                                      'value': [1, 2],
+                                },
+                                index=pd.period_range(start='2020-04-05 01:00+00:00', freq='H', periods=2))
+
+        points = data_frame_to_list_of_points(data_frame=data_frame,
+                                              point_settings=PointSettings(),
+                                              data_frame_measurement_name='h2o')
+
+        self.assertEqual(2, len(points))
+        self.assertEqual("h2o value=1i 1586048400000000000", points[0])
+        self.assertEqual("h2o value=2i 1586052000000000000", points[1])