influxdata / influxdata/kapacitor
Disable join buffering / full outer join
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
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- 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