1500

pending completion

Build # 1500

Build Type

Pull #380

travis-ci-com

Committed by

web-flow

Commit Message

Merge 318dfa2ba into 73931d76a

Pull Request Pull Request #380: Bring back join

Run Details

4609 of 6448 branches covered (71.48%)

Branch coverage included in aggregate %.

156 of 182 new or added lines in 9 files covered. (85.71%)

67 existing lines in 1 file now uncovered.

10504 of 12984 relevant lines covered (80.9%)

48273.19 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.86

/lib/runtime/stream_table_ops.ts

// -*- mode: typescript; indent-tabs-mode: nil; js-basic-offset: 4 -*-
//
// This file is part of ThingTalk
//
// Copyright 2018-2019 The Board of Trustees of the Leland Stanford Junior University
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: Giovanni Campagna <gcampagn@cs.stanford.edu>

import assert from 'assert';
import AsyncQueue from 'consumer-queue';

import { combineOutputTypes } from './output_type_ops';
import { equality } from './primitive_ops';
import type { ExecEnvironment } from './exec_environment';

// Library helpers used by the compiled TT code

interface MonitorTupleLike {
    __timestamp : number;
}

function tupleEquals<T, K extends keyof T>(a : T, b : T, keys : K[]) : boolean {
    for (const key of keys) {
        if (!equality(a[key], b[key]))
            return false;
    }
    return true;
}

export function isNewTuple<T extends MonitorTupleLike, K extends keyof T>(state : T[]|null,
                                                                          tuple : T,
                                                                          keys : K[]) : boolean {
    if (state === null)
        return true;

    let tlast, tprevious;
    for (let i = state.length-1; i >= 0; i--) {
        if (tlast === undefined)
            tlast = state[i].__timestamp;
        else if (tprevious === undefined && state[i].__timestamp < tlast)
            tprevious = state[i].__timestamp;
        else if (tprevious !== undefined && state[i].__timestamp < tprevious)
            break;
    }
    if (tuple.__timestamp === tlast)
        tlast = tprevious;
    if (tlast === undefined)
        return true;

    for (let i = 0; i < state.length; i++) {
        if (state[i].__timestamp !== tlast)
            continue;
        if (tupleEquals(state[i], tuple, keys))
            return false;
    }
    return true;
}

export function addTuple<T extends MonitorTupleLike>(state : T[]|null, tuple : T) : T[] {
    if (state === null)
        return [tuple];
    state.push(tuple);

    // trim the state to
    let tlast, tprevious;
    let i;
    for (i = state.length-1; i >= 0; i--) {
        if (tlast === undefined)
            tlast = state[i].__timestamp;
        else if (tprevious === undefined && state[i].__timestamp < tlast)
            tprevious = state[i].__timestamp;
        else if (tprevious !== undefined && state[i].__timestamp < tprevious)
            break;
    }
    if (i >= 0) {
        assert(tprevious !== undefined);
        assert(state[i].__timestamp < tprevious);
        state = state.slice(i+1);
    }

    return state;
}

type ResultT<T> = [string, T];

type EmitFunction<T> = (type : string, value : T) => void;
type Stream<T> = (emit : EmitFunction<T>) => Promise<void>;

export function streamUnion<T>(lhs : Stream<T>, rhs : Stream<T>) : AsyncIterator<ResultT<T>> {
    const queue = new AsyncQueue<IteratorResult<ResultT<T>, void>>();

    let currentLeft : ResultT<T>|null = null;
    let currentRight : ResultT<T>|null = null;
    let doneLeft = false;
    let doneRight = false;
    function emit() {
        if (currentLeft === null || currentRight === null)
            return;
        const [leftType, leftValue] = currentLeft;
        const [rightType, rightValue] = currentRight;
        const newValue = {} as T;
        Object.assign(newValue, leftValue);
        Object.assign(newValue, rightValue);
        const newType = combineOutputTypes(leftType, rightType);
        queue.push({ value: [newType, newValue], done: false });
    }
    function checkDone() {
        if (doneLeft && doneRight)
            queue.push({ value: undefined, done: true });
    }

    lhs((...v) => {
        currentLeft = v;
        emit();
    }).then(() => {
        doneLeft = true;
        checkDone();
    }).catch((err) => queue.cancelWait(err));

    rhs((...v) => {
        currentRight = v;
        emit();
    }).then(() => {
        doneRight = true;
        checkDone();
    }).catch((err) => queue.cancelWait(err));

    return queue;
}

function accumulateStream<T>(stream : Stream<T>) : Promise<Array<ResultT<T>>> {
    const into : Array<ResultT<T>> = [];

    return stream((type : string, value : T) => {
        into.push([type, value]);
    }).then(() => into);
}

class DelayedIterator<T> implements AsyncIterator<T> {
    private _promise : Promise<Iterator<T>>;
    private _iterator : Iterator<T>|null;

    constructor(promise : Promise<Iterator<T>>) {
        this._promise = promise;
        this._iterator = null;
    }

    next() : Promise<IteratorResult<T>> {
        if (this._iterator !== null)
            return Promise.resolve(this._iterator.next());
        return this._promise.then((iterator) => {
            this._iterator = iterator;
            return this._iterator.next();
        });
    }
}

export function tableJoin<T>(lhs : Stream<T>, rhs : Stream<T>) : AsyncIterator<ResultT<T>, void> {
    return new DelayedIterator(Promise.all([
        accumulateStream(lhs),
        accumulateStream(rhs)
    ]).then(([left, right]) => {
        return (function*() : Generator<ResultT<T>, void> {
            for (const l of left) {
                for (const r of right) {
                    const [leftType, leftValue] = l;
                    const [rightType, rightValue] = r;
                    const newValue : Record<string, any> = {};
                    for (const [key, value] of Object.entries(leftValue))
                        newValue[`first.${key}`] = value;
                    for (const [key, value] of Object.entries(rightValue))
                        newValue[`second.${key}`] = value;
                    Object.assign(newValue, leftValue);
                    Object.assign(newValue, rightValue);
                    const newType = combineOutputTypes(leftType, rightType);
                    yield [newType, newValue as T];
                }
            }
        })();
    }));
}

export function tableCrossJoin<T>(lhs : Stream<T>, rhs : Stream<T>) : AsyncIterator<ResultT<T>, void> {
    return new DelayedIterator(Promise.all([
        accumulateStream(lhs),
        accumulateStream(rhs)
    ]).then(([left, right]) => {
        return (function*() : Generator<ResultT<T>, void> {
            for (const l of left) {
                for (const r of right) {
                    const [leftType, leftValue] = l;
                    const [rightType, rightValue] = r;
                    const newValue = {} as T;
                    Object.assign(newValue, leftValue);
                    Object.assign(newValue, rightValue);
                    const newType = combineOutputTypes(leftType, rightType);
                    yield [newType, newValue];
                }
            }
        })();
    }));
}

type StreamFunction<T> = (env : ExecEnvironment, emit : EmitFunction<T>, ...args : any[]) => Promise<void>;

export function invokeStreamVarRef<T>(env : ExecEnvironment,
                                      varref : StreamFunction<T>,
                                      ...args : any[]) : AsyncQueue<IteratorResult<ResultT<T>, void>> {
    const queue = new AsyncQueue<IteratorResult<ResultT<T>, void>>();

    function emit(type : string, value : T) {
        queue.push({ value: [type, value], done: false });
    }
    varref(env, emit, ...args).then(() => {
        queue.push({ value: undefined, done: true });
    }).catch((err) => {
        queue.cancelWait(err);
    });

    return queue;
}

interface IterableOrAsyncIterable<T> {
    [Symbol.iterator] ?: () => Iterator<T>;
    [Symbol.asyncIterator] ?: () => AsyncIterator<T>;
}

export function getAsyncIterator<T>(obj : IterableOrAsyncIterable<T>) : Iterator<T>|AsyncIterator<T> {
    const getAsync = obj[Symbol.asyncIterator];
    if (typeof getAsync === 'function')
        return getAsync.call(obj);
    return obj[Symbol.iterator]!();
}

1	// -- mode: typescript; indent-tabs-mode: nil; js-basic-offset: 4 --
2	//
3	// This file is part of ThingTalk
4	//
5	// Copyright 2018-2019 The Board of Trustees of the Leland Stanford Junior University
6	//
7	// Licensed under the Apache License, Version 2.0 (the "License");
8	// you may not use this file except in compliance with the License.
9	// You may obtain a copy of the License at
10	//
11	// http://www.apache.org/licenses/LICENSE-2.0
12	//
13	// Unless required by applicable law or agreed to in writing, software
14	// distributed under the License is distributed on an "AS IS" BASIS,
15	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16	// See the License for the specific language governing permissions and
17	// limitations under the License.
18	//
19	// Author: Giovanni Campagna <gcampagn@cs.stanford.edu>
20
21	import assert from 'assert';	2✔
22	import AsyncQueue from 'consumer-queue';	2✔
23
24	import { combineOutputTypes } from './output_type_ops';	2✔
25	import { equality } from './primitive_ops';	2✔
26	import type { ExecEnvironment } from './exec_environment';
27
28	// Library helpers used by the compiled TT code
29
30	interface MonitorTupleLike {
31	__timestamp : number;
32	}
33
34	function tupleEquals<T, K extends keyof T>(a : T, b : T, keys : K[]) : boolean {
35	for (const key of keys) {	62✔
36	if (!equality(a[key], b[key]))	98✔
37	return false;	46✔
38	}
39	return true;	16✔
40	}
41
42	export function isNewTuple<T extends MonitorTupleLike, K extends keyof T>(state : T[]\|null,	2✔
43	tuple : T,
44	keys : K[]) : boolean {
45	if (state === null)	100✔
46	return true;	34✔
47
48	let tlast, tprevious;
49	for (let i = state.length-1; i >= 0; i--) {	66✔
50	if (tlast === undefined)	120✔
51	tlast = state[i].__timestamp;	66✔
52	else if (tprevious === undefined && state[i].__timestamp < tlast)	54✔
53	tprevious = state[i].__timestamp;	28✔
54	else if (tprevious !== undefined && state[i].__timestamp < tprevious)	26!
55	break;	×
56	}
57	if (tuple.__timestamp === tlast)	66✔
58	tlast = tprevious;	26✔
59	if (tlast === undefined)	66✔
60	return true;	16✔
61
62	for (let i = 0; i < state.length; i++) {	50✔
63	if (state[i].__timestamp !== tlast)	92✔
64	continue;	30✔
65	if (tupleEquals(state[i], tuple, keys))	62✔
66	return false;	16✔
67	}
68	return true;	34✔
69	}
70
71	export function addTuple<T extends MonitorTupleLike>(state : T[]\|null, tuple : T) : T[] {	2✔
72	if (state === null)	100✔
73	return [tuple];	34✔
74	state.push(tuple);	66✔
75
76	// trim the state to
77	let tlast, tprevious;
78	let i;
79	for (i = state.length-1; i >= 0; i--) {	66✔
80	if (tlast === undefined)	180✔
81	tlast = state[i].__timestamp;	66✔
82	else if (tprevious === undefined && state[i].__timestamp < tlast)	114✔
83	tprevious = state[i].__timestamp;	50✔
84	else if (tprevious !== undefined && state[i].__timestamp < tprevious)	64✔
85	break;	18✔
86	}
87	if (i >= 0) {	66✔
88	assert(tprevious !== undefined);	18✔
89	assert(state[i].__timestamp < tprevious);	18✔
90	state = state.slice(i+1);	18✔
91	}
92
93	return state;	66✔
94	}
95
96	type ResultT<T> = [string, T];
97
98	type EmitFunction<T> = (type : string, value : T) => void;
99	type Stream<T> = (emit : EmitFunction<T>) => Promise<void>;
100
101	export function streamUnion<T>(lhs : Stream<T>, rhs : Stream<T>) : AsyncIterator<ResultT<T>> {	2✔
102	const queue = new AsyncQueue<IteratorResult<ResultT<T>, void>>();	2✔
103
104	let currentLeft : ResultT<T>\|null = null;	2✔
105	let currentRight : ResultT<T>\|null = null;	2✔
106	let doneLeft = false;	2✔
107	let doneRight = false;	2✔
108	function emit() {
109	if (currentLeft === null \|\| currentRight === null)	14✔
110	return;	2✔
111	const [leftType, leftValue] = currentLeft;	12✔
112	const [rightType, rightValue] = currentRight;	12✔
113	const newValue = {} as T;	12✔
114	Object.assign(newValue, leftValue);	12✔
115	Object.assign(newValue, rightValue);	12✔
116	const newType = combineOutputTypes(leftType, rightType);	12✔
117	queue.push({ value: [newType, newValue], done: false });	12✔
118	}
119	function checkDone() {
120	if (doneLeft && doneRight)	4✔
121	queue.push({ value: undefined, done: true });	2✔
122	}
123
124	lhs((...v) => {	2✔
125	currentLeft = v;	8✔
126	emit();	8✔
127	}).then(() => {
128	doneLeft = true;	2✔
129	checkDone();	2✔
130	}).catch((err) => queue.cancelWait(err));	×
131
132	rhs((...v) => {	2✔
133	currentRight = v;	6✔
134	emit();	6✔
135	}).then(() => {
136	doneRight = true;	2✔
137	checkDone();	2✔
138	}).catch((err) => queue.cancelWait(err));	×
139
140	return queue;	2✔
141	}
142
143	function accumulateStream<T>(stream : Stream<T>) : Promise<Array<ResultT<T>>> {
144	const into : Array<ResultT<T>> = [];	12✔
145
146	return stream((type : string, value : T) => {	12✔
147	into.push([type, value]);	20✔
148	}).then(() => into);	12✔
149	}
150
151	class DelayedIterator<T> implements AsyncIterator<T> {
152	private _promise : Promise<Iterator<T>>;
153	private _iterator : Iterator<T>\|null;
154
155	constructor(promise : Promise<Iterator<T>>) {
156	this._promise = promise;	6✔
157	this._iterator = null;	6✔
158	}
159
160	next() : Promise<IteratorResult<T>> {
161	if (this._iterator !== null)	32✔
162	return Promise.resolve(this._iterator.next());	26✔
163	return this._promise.then((iterator) => {	6✔
164	this._iterator = iterator;	6✔
165	return this._iterator.next();	6✔
166	});
167	}
168	}
169
170	export function tableJoin<T>(lhs : Stream<T>, rhs : Stream<T>) : AsyncIterator<ResultT<T>, void> {	2✔
NEW 171	return new DelayedIterator(Promise.all([	×
172	accumulateStream(lhs),
173	accumulateStream(rhs)
174	]).then(([left, right]) => {
NEW 175	return (function*() : Generator<ResultT<T>, void> {	×
NEW 176	for (const l of left) {	×
NEW 177	for (const r of right) {	×
NEW 178	const [leftType, leftValue] = l;	×
NEW 179	const [rightType, rightValue] = r;	×
NEW 180	const newValue : Record<string, any> = {};	×
NEW 181	for (const [key, value] of Object.entries(leftValue))	×
NEW 182	newValue[`first.${key}`] = value;	×
NEW 183	for (const [key, value] of Object.entries(rightValue))	×
NEW 184	newValue[`second.${key}`] = value;	×
NEW 185	Object.assign(newValue, leftValue);	×
NEW 186	Object.assign(newValue, rightValue);	×
NEW 187	const newType = combineOutputTypes(leftType, rightType);	×
NEW 188	yield [newType, newValue as T];	×
189	}
190	}
191	})();
192	}));
193	}
194
195	export function tableCrossJoin<T>(lhs : Stream<T>, rhs : Stream<T>) : AsyncIterator<ResultT<T>, void> {	2✔
196	return new DelayedIterator(Promise.all([	6✔
197	accumulateStream(lhs),
198	accumulateStream(rhs)
199	]).then(([left, right]) => {
200	return (function*() : Generator<ResultT<T>, void> {	6✔
201	for (const l of left) {	6✔
202	for (const r of right) {	12✔
203	const [leftType, leftValue] = l;	26✔
204	const [rightType, rightValue] = r;	26✔
205	const newValue = {} as T;	26✔
206	Object.assign(newValue, leftValue);	26✔
207	Object.assign(newValue, rightValue);	26✔
208	const newType = combineOutputTypes(leftType, rightType);	26✔
209	yield [newType, newValue];	26✔
210	}
211	}
212	})();
213	}));
214	}
215
216	type StreamFunction<T> = (env : ExecEnvironment, emit : EmitFunction<T>, ...args : any[]) => Promise<void>;
217
218	export function invokeStreamVarRef<T>(env : ExecEnvironment,	2✔
219	varref : StreamFunction<T>,
220	...args : any[]) : AsyncQueue<IteratorResult<ResultT<T>, void>> {
221	const queue = new AsyncQueue<IteratorResult<ResultT<T>, void>>();	48✔
222
223	function emit(type : string, value : T) {
224	queue.push({ value: [type, value], done: false });	×
225	}
226	varref(env, emit, ...args).then(() => {	48✔
227	queue.push({ value: undefined, done: true });	48✔
228	}).catch((err) => {
229	queue.cancelWait(err);	×
230	});
231
232	return queue;	48✔
233	}
234
235	interface IterableOrAsyncIterable<T> {
236	[Symbol.iterator] ?: () => Iterator<T>;
237	[Symbol.asyncIterator] ?: () => AsyncIterator<T>;
238	}
239
240	export function getAsyncIterator<T>(obj : IterableOrAsyncIterable<T>) : Iterator<T>\|AsyncIterator<T> {	2✔
241	const getAsync = obj[Symbol.asyncIterator];	240✔
242	if (typeof getAsync === 'function')	240✔
243	return getAsync.call(obj);	214✔
244	return obj[Symbol.iterator]!();	26✔
245	}

stanford-oval / thingtalk / 1500

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous