sensorbee-join

!README.md

build_sensorbee
./sensorbee run &
./sensorbee t c test

./sensorbee shell -t test < 01_select.bql &
./sensorbee shell -t test < 02_resume.bql

Expected

{"data1":"hello","data2":"world","id1":1,"id2":1}
{"data1":"hi",   "data2":"there","id1":2,"id2":2}
{"data1":"hello","data2":"world","id1":3,"id2":3}
{"data1":"hi",   "data2":"there","id1":4,"id2":4}
{"data1":"hello","data2":"world","id1":5,"id2":5}
{"data1":"hi",   "data2":"there","id1":6,"id2":6}
{"data1":"hello","data2":"world","id1":7,"id2":7}
{"data1":"hi",   "data2":"there","id1":8,"id2":8}
{"data1":"hello","data2":"world","id1":9,"id2":9}
{"data1":"hi",   "data2":"there","id1":10,"id2":10}

Actual

{"data1":"hi",   "data2":"world","id1":10,"id2":1}
{"data1":"hi",   "data2":"there","id1":10,"id2":2}
{"data1":"hi",   "data2":"world","id1":10,"id2":3}
{"data1":"hi",   "data2":"there","id1":10,"id2":4}
{"data1":"hi",   "data2":"world","id1":10,"id2":5}
{"data1":"hi",   "data2":"there","id1":10,"id2":6}
{"data1":"hi",   "data2":"world","id1":10,"id2":7}
{"data1":"hi",   "data2":"there","id1":10,"id2":8}
{"data1":"hi",   "data2":"world","id1":10,"id2":9}
{"data1":"hi",   "data2":"there","id1":10,"id2":10}

01_select-merge.bql

DROP SOURCE test1;
DROP SOURCE test2;
CREATE PAUSED SOURCE test1 TYPE file WITH path = "test1.jsonl", rewindable = true;
CREATE PAUSED SOURCE test2 TYPE file WITH path = "test2.jsonl", rewindable = true;

CREATE STREAM merged AS SELECT RSTREAM * FROM test1 [RANGE 1 TUPLES] UNION ALL SELECT RSTREAM * FROM test2 [RANGE 1 TUPLES];
SELECT RSTREAM * FROM merged [RANGE 1 TUPLES];

01_select-wait.bql

DROP SOURCE test1;
DROP SOURCE test2;
CREATE PAUSED SOURCE test1 TYPE file WITH path = "test1.jsonl", rewindable = true;
CREATE PAUSED SOURCE test2 TYPE file WITH path = "test2.jsonl", rewindable = true;

SELECT RSTREAM * FROM test1 [RANGE 1 TUPLES, BUFFER SIZE 1, WAIT IF FULL], test2 [RANGE 1 tuples, BUFFER SIZE 1, WAIT IF FULL];

01_select-where.bql

DROP SOURCE test1;
DROP SOURCE test2;
CREATE PAUSED SOURCE test1 TYPE file WITH path = "test1.jsonl", rewindable = true;
CREATE PAUSED SOURCE test2 TYPE file WITH path = "test2.jsonl", rewindable = true;

SELECT RSTREAM * FROM test1 [RANGE 1 TUPLES], test2 [RANGE 1 tuples] WHERE test1:id1 = test2:id2;

01_select.bql

DROP SOURCE test1;
DROP SOURCE test2;
CREATE PAUSED SOURCE test1 TYPE file WITH path = "test1.jsonl", rewindable = true;
CREATE PAUSED SOURCE test2 TYPE file WITH path = "test2.jsonl", rewindable = true;

SELECT RSTREAM * FROM test1 [RANGE 1 TUPLES], test2 [RANGE 1 tuples];

02_resume.bql

RESUME SOURCE test1;
RESUME SOURCE test2;

03_rewind.bql

REWIND SOURCE test1;
REWIND SOURCE test2;

build.yaml

sensorbee_main.go

package main

import (
	"github.com/codegangsta/cli"
	_ "gopkg.in/sensorbee/sensorbee.v0/bql/udf/builtin"
	"gopkg.in/sensorbee/sensorbee.v0/cmd/lib/exp"
	"gopkg.in/sensorbee/sensorbee.v0/cmd/lib/run"
	"gopkg.in/sensorbee/sensorbee.v0/cmd/lib/runfile"
	"gopkg.in/sensorbee/sensorbee.v0/cmd/lib/shell"
	"gopkg.in/sensorbee/sensorbee.v0/cmd/lib/topology"
	"gopkg.in/sensorbee/sensorbee.v0/version"
	"os"
	"time"
)

func init() {
	// TODO
	time.Local = time.UTC
}

func main() {
	app := cli.NewApp()
	app.Name = "sensorbee"
	app.Usage = "SensorBee built with build_sensorbee 0.5.1"
	app.Version = version.Version
	app.Commands = []cli.Command{
		run.SetUp(),
		shell.SetUp(),
		topology.SetUp(),
		exp.SetUp(),
		runfile.SetUp(),
	}
	if err := app.Run(os.Args); err != nil {
		os.Exit(1)
	}
}

test1.jsonl

{"id1": 1, "data1": "hello"}
{"id1": 2, "data1": "hi"}
{"id1": 3, "data1": "hello"}
{"id1": 4, "data1": "hi"}
{"id1": 5, "data1": "hello"}
{"id1": 6, "data1": "hi"}
{"id1": 7, "data1": "hello"}
{"id1": 8, "data1": "hi"}
{"id1": 9, "data1": "hello"}
{"id1": 10, "data1": "hi"}

test2.jsonl

{"id2": 1, "data2": "world"}
{"id2": 2, "data2": "there"}
{"id2": 3, "data2": "world"}
{"id2": 4, "data2": "there"}
{"id2": 5, "data2": "world"}
{"id2": 6, "data2": "there"}
{"id2": 7, "data2": "world"}
{"id2": 8, "data2": "there"}
{"id2": 9, "data2": "world"}
{"id2": 10, "data2": "there"}