influxdata / influxdata/kapacitor

Disable join buffering / full outer join

Open
#1,704 2 comments 4 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
Go
Stars
2.4k
Forks
479
Avg merge
4d 16h
Merged PRs (30d)
4

Description

It would be nice to disable the buffering that the join node does, and get the behavior seen in SQL databases when using a full outer join.

Currently the `join` node buffers the output point until it receives an input point with a timestamp outside the tolerance of the buffered output point. This behavior causes unwanted delay in data processing (such as alerting).

The documentation also says this for the `fill()` property:
> null - fill missing points with null, full outer join.

This is incorrect. In SQL databases, a full outer join emits a row for every permutation of the joined tables.
For example:
```
# select * from t1;
i | v
---+---
0 | a
1 | b
2 | c

# select * from t2;
i | v
---+----
1 | b
2 | c
2 | cc
3 | d

# select t1.i, t1.v, t2.v from t1 full outer join t2 on t1.i = t2.i;
i | v | v
---+---+----
0 | a |
1 | b | b
2 | c | c
2 | c | cc
| | d
```

Kapacitor instead behaves like this (assuming `i` is a timestamp):
```
i | v | v
---+---+----
0 | a |
1 | b | b
2 | c | c
2 | | cc -- missing value from t1
| | d
```

If kapacitor behaved like a traditional SQL full outer join, it would simply emit each permutation as it is received (using the buffered point from other nodes if able), not wait for a new point from every node, and not wait for a point outside the tolerance before emitting.

 

And for some explicit definitions of expected behavior:
```
test,host=a v=1i 1000000000
test,host=b v=2i 1000000000
test,host=a v=2i 2000000000
test,host=a v=3i 3000000000
test,host=b v=4i 3000000000
```
```
var hosta = stream|from().measurement('test').where(lambda: "host"=='a')
var hostb = stream|from().measurement('test').where(lambda: "host"=='b')

hosta|join(hostb).as('hosta','hostb').tolerance(2s).fill('null')|log()
```
Expected output:
```
[test:log5] 2017/12/01 15:18:22 I! {"Name":"test","Database":"","RetentionPolicy":"","Group":"","Dimensions":{"ByName":false,"TagNames":null},"Tags":{},"Fields":{"hosta.v":1,"hostb.v":2},"Time":"1970-01-01T00:00:01Z"}

[test:log5] 2017/12/01 15:18:22 I! {"Name":"test","Database":"","RetentionPolicy":"","Group":"","Dimensions":{"ByName":false,"TagNames":null},"Tags":{},"Fields":{"hosta.v":2,"hostb.v":2},"Time":"1970-01-01T00:00:02Z"}

[test:log5] 2017/12/01 15:18:22 I! {"Name":"test","Database":"","RetentionPolicy":"","Group":"","Dimensions":{"ByName":false,"TagNames":null},"Tags":{},"Fields":{"hosta.v":3,"hostb.v":4},"Time":"1970-01-01T00:00:03Z"}

```
Actual output:
```
[test:log5] 2017/12/01 15:18:22 I! {"Name":"test","Database":"","RetentionPolicy":"","Group":"","Dimensions":{"ByName":false,"TagNames":null},"Tags":{},"Fields":{"hosta.v":1,"hostb.v":2},"Time":"1970-01-01T00:00:02Z"}

[test:log5] 2017/12/01 15:18:22 I! {"Name":"test","Database":"","RetentionPolicy":"","Group":"","Dimensions":{"ByName":false,"TagNames":null},"Tags":{},"Fields":{"hosta.v":2,"hostb.v":null},"Time":"1970-01-01T00:00:02Z"}
```

 

This I think is important to address as one of the common use cases with the `join` node is with any influxql node. With influxql nodes, you cannot add the computed value as a new field. You instead have to use `join()`. And every time you use `join()` you increment the buffering. And when your data arrives at an infrequent interval, this can cause massive delays.

Ref #1249

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

The issue concerns Kapacitor's join node and its fill('null') behavior; begin by tracing the join-node implementation and documentation, then reproduce the supplied host=a/host=b stream with a 2s tolerance. Compare the output with the stated SQL-like expected rows, including duplicate timestamps and unmatched points. Done means the described buffering delay and join semantics are addressed, and the fill() documentation matches the behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
go
Domain
stream-processing
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.