Kapacitor: Join node with fill null construct invalid line protocol

[desa]

When trying to perform an full outer join and using fill

data1
    |join(data2)
        .as('data1', 'data2')
        .fill('null')
    |influxDBOut()
     ...

fields values that end up null are marshaled into line protocol that is of the form field_x=,field_y=1.

If you instead use a fill with a numerical value, you end up with field type conflicts when writing to the database.

It is likely that fill should allow you to specify specific fill values on a per field basis.

cc @dp1140a

[dp1140a]

Since there are four field typed we need to handle differently where we want to fill with the appropriate "nil" value. Assume we have a 4 fields:
field1 an int
field2 a float
field3 a string
field4 a bool

A possible solution might be the following:
.fill(['field1=int', 'field2=float', 'field3=string', 'field4=bool'])

this would result in line protocol looking like:
myMeasurement,tag1=foo,tag2=bar field1=0,field2=0.0,field3="",field4=false 123456789

The advantage of this approach is that it puts the responsibility on the script writer to let the system know what the field data type is. Otherwise Kapacitor will have to perform a lookup and store that which could be expensive.

[sanderson]

Do we assume that empty values from outer joins will always be filled with null values? Are there use cases where a user would want to use a different fill? The proposed method would also require the user to know all the field keys and data types resulting from the join.

Is the join node able to detect the data types of each field as the join is happening (before translated into line protocol)? If so, I'm wondering if we could define fills for both specific field keys (the ones the user cares about) and data types:

.fill('field1=nil', 'field2=0.0', int=0 float=1.0 string="" bool=false)  

_Ingore syntax, still learning._

But this way, .fill() without any arguments would default to setting all the different data types to their appropriate value. The assumed default would be something like:

.fill(int=0 float=1.0 string="" bool=false)  

So if the user only wants to define fills for specific field keys, they wouldn't have to define the defaults for data types; just the field keys:

.fill('field1=nil', 'field2=0.0')  

If the join process knows the data type of each field and the field key isn't explicitly given a fill, it can fill the "unmentioned" field keys with the appropriate null value.

[desa]

@sanderson I think specifying the field type to fill, like you suggest, would work much better. I'd have to think through if it'd be feasible, at first thought I think it should be.

[dp1140a]

1871

[nathanielc]

This can be solved without making changes to the join node. By using the default node in combination with the fill('null') join property. The default node will set a value for each field that has a null value.

Assume we have 4 fields:
field1 an int
field2 a float
field3 a string
field4 a bool

Then you can do this:

data1
    |join(data2)
        .as('data1', 'data2')
        .fill('null')
    |default()
        .field('field1', 0) // set field1 default as integer 0
        .field('field2', 0.0) // set field2 default as floating point 0.0
        .field('field3', '') // set field3 default as empty string
        .field('field4', false) // set field4 default as boolean false
    |influxDBOut()
     ...

Any field that was filled with null will get updated with its typed default value.

[sriram-rsankar]

@nathanielc I saw issue #1871 is referred in this issue. I am also facing a similar issue and default() node is not solving the issue. The issue is even before the default value is assigned.

I have the following tick script and python script to simulate this issue.

join.tick

var var_inpkts = stream
    |from()
      .database('join')
      .retentionPolicy('autogen')
      .measurement('join')
      .groupBy('name',)
    |eval(lambda: int("inpkts"), lambda: string("name"))
      .as('inpkts', 'name')
      .tags('name')

var var_status = stream
    |from()
      .database('join')
      .retentionPolicy('autogen')
      .measurement('join')
      .groupBy('name',)
    |eval(lambda: string("oper"), lambda: string("name"))
      .as('oper', 'name')
      .tags('name')

var joined_fields = var_inpkts| join(var_status)
    .as('var_inpkts', 'var_status')
    .fill('null')
    .tolerance(1s)
    | default()
       .field('var_status.oper', 'NA')
       .field('var_inpkts.inpkts', '99999')
    | eval(lambda: string("var_status.oper"), lambda: int("var_inpkts.inpkts"))
       .as('oper', 'inpkts')

joined_fields
    |influxDBOut()
      .measurement('join_out')
      .database('join_out')
      .flushInterval(1s)
      .create()

simulator_join.py

#!/usr/bin/env python3
from influxdb import InfluxDBClient
import datetime
import sys
import time

client = InfluxDBClient('localhost', 9092, database='join')
dd = datetime.datetime.utcnow()
point_date = datetime.datetime.fromtimestamp(
    int(dd.strftime('%s'))).strftime(
    '%Y-%m-%dT%H:%M:%SZ')

for x in range(0, 1):
    print(point_date)
    name = 'et' + str(x)
    points = [
        {
            "measurement": "join",
            "time": point_date,
            "tags": {
                "name": name,
            },
            "fields": {
                "oper": "up",
                "inpkts": 200
            }
        }
    ]
    print("Write points: {0}".format(points))
    print(client.write_points(points, retention_policy='autogen'))

Scenario 1: When simulator emits both the fields, fields are joined properly and written to influxdb.

> python3 simulator_join.py
2018-07-27T10:00:37Z
Write points: [{'measurement': 'join', 'time': '2018-07-27T10:00:37Z', 'tags': {'name': 'et0'}, 'fields': {'oper': 'up', 'inpkts': 200}}]
True

stream0 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
stream0 -> from3 [processed="1"];
stream0 -> from1 [processed="1"];

from3 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
from3 -> eval4 [processed="1"];

eval4 [avg_exec_time_ns="0s" errors="0" working_cardinality="1" ];
eval4 -> join6 [processed="1"];

from1 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
from1 -> eval2 [processed="1"];

eval2 [avg_exec_time_ns="0s" errors="0" working_cardinality="1" ];
eval2 -> join6 [processed="1"];

join6 [avg_exec_time_ns="9.138µs" errors="0" working_cardinality="1" ];
join6 -> default7 [processed="1"];

default7 [avg_exec_time_ns="0s" errors="0" fields_defaulted="0" tags_defaulted="0" working_cardinality="0" ];
default7 -> eval8 [processed="1"];

eval8 [avg_exec_time_ns="0s" errors="0" working_cardinality="1" ];
eval8 -> influxdb_out9 [processed="1"];

influxdb_out9 [avg_exec_time_ns="0s" errors="0" points_written="1" working_cardinality="0" write_errors="0" ];

Scenario 2: When the simulator emits only one field, join node does not release the data.
PS: I have commented the script not to emit 'inpkts' packets. Also, redefined the tick script so that processed values are reset to 0

> python3 simulator_join.py
2018-07-27T10:11:27Z
Write points: [{'measurement': 'join', 'time': '2018-07-27T10:11:27Z', 'tags': {'name': 'et0'}, 'fields': {'oper': 'up'}}]
True

From the below output you can see that eval2 has 1 errors because of missing data, which is understandable. eval4 has processed 1 data to join6. But nothing is been processed from join6 to default7.

stream0 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
stream0 -> from3 [processed="1"];
stream0 -> from1 [processed="1"];

from3 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
from3 -> eval4 [processed="1"];

eval4 [avg_exec_time_ns="0s" errors="0" working_cardinality="1" ];
eval4 -> join6 [processed="1"];

from1 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
from1 -> eval2 [processed="1"];

eval2 [avg_exec_time_ns="0s" errors="1" working_cardinality="1" ];
eval2 -> join6 [processed="0"];

join6 [avg_exec_time_ns="0s" errors="0" working_cardinality="1" ];
join6 -> default7 [processed="0"];

default7 [avg_exec_time_ns="0s" errors="0" fields_defaulted="0" tags_defaulted="0" working_cardinality="0" ];
default7 -> eval8 [processed="0"];

eval8 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
eval8 -> influxdb_out9 [processed="0"];

influxdb_out9 [avg_exec_time_ns="0s" errors="0" points_written="0" working_cardinality="0" write_errors="0" ];

Even after sending data three more times, join() node has not released the data.

rsankar-mbp: ~/Desktop/work/iceberg>
> python3 simulator_join.py
2018-07-27T10:14:56Z
Write points: [{'measurement': 'join', 'time': '2018-07-27T10:14:56Z', 'tags': {'name': 'et0'}, 'fields': {'oper': 'up'}}]
True
rsankar-mbp: ~/Desktop/work/iceberg>
> python3 simulator_join.py
2018-07-27T10:14:58Z
Write points: [{'measurement': 'join', 'time': '2018-07-27T10:14:58Z', 'tags': {'name': 'et0'}, 'fields': {'oper': 'up'}}]
True
rsankar-mbp: ~/Desktop/work/iceberg>
> python3 simulator_join.py
2018-07-27T10:15:01Z
Write points: [{'measurement': 'join', 'time': '2018-07-27T10:15:01Z', 'tags': {'name': 'et0'}, 'fields': {'oper': 'up'}}]
True

stream0 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
stream0 -> from3 [processed="4"];
stream0 -> from1 [processed="4"];

from3 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
from3 -> eval4 [processed="4"];

eval4 [avg_exec_time_ns="0s" errors="0" working_cardinality="1" ];
eval4 -> join6 [processed="4"];

from1 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
from1 -> eval2 [processed="4"];

eval2 [avg_exec_time_ns="0s" errors="4" working_cardinality="1" ];
eval2 -> join6 [processed="0"];

join6 [avg_exec_time_ns="0s" errors="0" working_cardinality="1" ];
join6 -> default7 [processed="0"];

default7 [avg_exec_time_ns="0s" errors="0" fields_defaulted="0" tags_defaulted="0" working_cardinality="0" ];
default7 -> eval8 [processed="0"];

eval8 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
eval8 -> influxdb_out9 [processed="0"];

influxdb_out9 [avg_exec_time_ns="0s" errors="0" points_written="0" working_cardinality="0" write_errors="0" ];
````
Scenario : 3 Continuing the scenario 2, now if the script starts sending value for both the fields then join() node emits the first cached node to default() node and default value is assigned to missing data. join() node never emits data if the value is missing for one of the field. 

python3 simulator_join.py
2018-07-27T10:18:21Z
Write points: [{'measurement': 'join', 'time': '2018-07-27T10:18:21Z', 'tags': {'name': 'et0'}, 'fields': {'oper': 'up', 'inpkts': 200}}]
True

stream0 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
stream0 -> from3 [processed="5"];
stream0 -> from1 [processed="5"];

from3 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
from3 -> eval4 [processed="5"];

eval4 [avg_exec_time_ns="0s" errors="0" working_cardinality="1" ];
eval4 -> join6 [processed="5"];

from1 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
from1 -> eval2 [processed="5"];

eval2 [avg_exec_time_ns="0s" errors="4" working_cardinality="1" ];
eval2 -> join6 [processed="1"];

join6 [avg_exec_time_ns="6.578µs" errors="0" working_cardinality="1" ];
join6 -> default7 [processed="1"];

default7 [avg_exec_time_ns="0s" errors="0" fields_defaulted="1" tags_defaulted="0" working_cardinality="0" ];
default7 -> eval8 [processed="1"];

eval8 [avg_exec_time_ns="0s" errors="0" working_cardinality="1" ];
eval8 -> influxdb_out9 [processed="1"];

influxdb_out9 [avg_exec_time_ns="0s" errors="0" points_written="1" working_cardinality="0" write_errors="0" ];

```
default() node assigning the values to missing field(s) can happen only after join node emits the data. Expectation is join() releases previous data as soon as it receives data at a timestamp greater than the tolerance.

The attached patch should fix the issue.

The patch ensures that when data is sent subsequently for fields at a timestamp greater than the tolerance, the previous values are released even if not all of the previous values have come in as part of the next time stamp.

The patch also ensures that when the barrier node is used, that whatever data has come in gets released immediately (post the tolerance window) even if not all of the data has come (without waiting for subsequent values coming in at a later timestamp). 

Please let me know if this make sense and is my fix proper.

join_node_problem_fix.txt

[nathanielc]

This change https://github.com/influxdata/kapacitor/pull/2048 makes it so that Join nodes correctly handle data pauses, which is necessary for outer joins, (i.e fill null).