stanford-oval / thingtalk / 807

pending completion

Build # 807

Build Type

Pull #184

travis-ci

Committed by

web-flow

Commit Message

Fix compiler undefined devices bug

Pull Request Pull Request #184: Fix compiler undefined devices bug

Run Details

4049 of 5269 branches covered (76.85%)

Branch coverage included in aggregate %.

12 of 13 new or added lines in 1 file covered. (92.31%)

13 existing lines in 1 file now uncovered.

7717 of 9132 relevant lines covered (84.51%)

758.26 hits per line

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

78.83

/lib/compiler/ast-to-ops.js

// -*- mode: js; indent-tabs-mode: nil; js-basic-offset: 4 -*-
//
// This file is part of ThingTalk
//
// Copyright 2015-2018 The Board of Trustees of the Leland Stanford Junior University
//
// Author: Giovanni Campagna <gcampagn@cs.stanford.edu>
//
// See COPYING for details
"use strict";

const Ast = require('../ast');
const { NotImplementedError } = require('../errors');
const { getScalarExpressionName } = require('../utils');

const { PointWiseOp, StreamOp, TableOp, RuleOp } = require('./ops');
const ReduceOp = require('./reduceop');

function sameDevice(lhs, rhs) {
    if (lhs.isBuiltin && rhs.isBuiltin)
        return true;
    if (lhs.isBuiltin || rhs.isBuiltin)
        return false;
    if (lhs.kind !== rhs.kind)
        return false;
    if (lhs.id !== rhs.id)
        return false;
    if (lhs.principal !== rhs.principal)
        return false;
    return true;
}

// compile a table that is being monitored to a stream
function compileMonitorTableToOps(table) {
    if (table.isVarRef ||
        table.isWindow ||
        table.isTimeSeries ||
        table.isHistory ||
        table.isSequence ||
        table.isAlias)
        throw new NotImplementedError(table);

    if (table.isInvocation) {
        // subscribe is optimistic, we still need EdgeNew
        return new StreamOp.EdgeNew(
            new StreamOp.InvokeSubscribe(table.invocation, Ast.BooleanExpression.True, table),
            table
        );
    } else if (table.isFilter) {
        return new StreamOp.Filter(
            compileMonitorTableToOps(table.table),
            table.filter,
            table
        );
    } else if (table.isProjection) {
        // note the "edge new" operation here, because
        // the projection might cause fewer values to
        // be new
        return new StreamOp.EdgeNew(
            new StreamOp.Map(
                compileMonitorTableToOps(table.table),
                new PointWiseOp.Projection(table.args),
                table
            ),
            table
        );
    } else if (table.isSort || table.isIndex || table.isSlice) {
        // sort, index and slice have no effect on monitor
        //
        // XXX is this correct?
        return compileMonitorTableToOps(table);
    } else if (table.isCompute) {
        // note the "edge new" operation here, because
        // the projection might cause fewer values to
        // be new
        return new StreamOp.EdgeNew(
            new StreamOp.Map(
                compileMonitorTableToOps(table.table),
                new PointWiseOp.Compute(table.expression),
                table
            ),
            table
        );
    } else if (table.isAggregation) {
        // for an aggregation, we subscribe to the inner table
        // (ie react to all changes), then when the table changes
        // we fetch it completely again and compute the aggregation
        // note the "edge new" operation here, because
        // the aggregation might cause fewer values to
        // be new
        return new StreamOp.EdgeNew(new StreamOp.InvokeTable(
            compileMonitorTableToOps(table.table),
            compileTableToOps(table, []),
            table
        ), table);
    } else if (table.isJoin) {
        if (table.in_params.length === 0) {
            // if there is no parameter passing, we can individually monitor
            // the two tables and return the union
            return new StreamOp.EdgeNew(new StreamOp.Union(
                compileMonitorTableToOps(table.lhs),
                compileMonitorTableToOps(table.rhs),
                table),
                table
            );
        } else {
            // otherwise we need to subscribe to the left hand side, and
            // every time it fires, create/update a subscription to the
            // right hand side
            // this is VERY MESSY
            // so it's not implemented
            throw new NotImplementedError(table);
        }
    } else {
        throw new TypeError();
    }
}

// compile a TT stream to a stream op and zero or more
// tableops
function compileStreamToOps(stream) {
    if (stream.isAlias)
        throw new NotImplementedError(stream);

    if (stream.isVarRef) {
        return new StreamOp.InvokeVarRef(stream.name, stream.in_params, stream);
    } else if (stream.isTimer) {
        return new StreamOp.Timer(stream.base, stream.interval, stream.frequency, stream);
    } else if (stream.isAtTimer) {
        return new StreamOp.AtTimer(stream.time, stream.expiration_date, stream);
    } else if (stream.isMonitor) {
        return compileMonitorTableToOps(stream.table);
    } else if (stream.isEdgeNew) {
        let op = compileStreamToOps(stream.stream);
        return new StreamOp.EdgeNew(op, op.ast);
    } else if (stream.isEdgeFilter) {
        let op = compileStreamToOps(stream.stream);
        return new StreamOp.EdgeFilter(op, stream.filter, op.ast);
    } else if (stream.isFilter) {
        let op = compileStreamToOps(stream.stream);
        return new StreamOp.Filter(op, stream.filter, op.ast);
    } else if (stream.isProjection) {
        let op = compileStreamToOps(stream.stream);
        return new StreamOp.Map(op, new PointWiseOp.Projection(stream.args), op.ast);
    } else if (stream.isCompute) {
        let op = compileStreamToOps(stream.stream);
        return new StreamOp.Map(op, new PointWiseOp.Compute(stream.expression,
            stream.alias || getScalarExpressionName(stream.expression)), op.ast);
    } else if (stream.isJoin) {
        let streamOp = compileStreamToOps(stream.stream);
        let tableOp = compileTableToOps(stream.table, stream.in_params);
        return new StreamOp.Join(streamOp, tableOp, stream);
    } else {
        throw new TypeError();
    }
}

function compileTableToOps(table, extra_in_params) {
    if (table.isWindow ||
        table.isTimeSeries ||
        table.isHistory ||
        table.isSequence ||
        table.isAlias)
        throw new NotImplementedError(table);

    if (table.isVarRef) {
        const compiled = new TableOp.InvokeVarRef(table.name, table.in_params.concat(extra_in_params), table);
        compiled.device = null;
        compiled.handle_thingtalk = false;
        return compiled;
    } else if (table.isResultRef) {
        return new TableOp.ReadResult(table.kind + ':' + table.channel, table.index, table.schema, table);
    } else if (table.isInvocation) {
        const device = table.invocation.selector;
        const handle_thingtalk = table.schema.annotations['handle_thingtalk'] ? table.schema.annotations['handle_thingtalk'].value : false;
        return new TableOp.InvokeGet(
            table.invocation,
            extra_in_params,
            Ast.BooleanExpression.True,
            device,
            handle_thingtalk,
            table
        );
    } else if (table.isFilter) {
        const compiled = compileTableToOps(table.table, extra_in_params);
        return new TableOp.Filter(
            compiled,
            table.filter,
            compiled.device,
            compiled.handle_thingtalk,
            table
        );
    } else if (table.isProjection) {
        const compiled = compileTableToOps(table.table, extra_in_params);
        return new TableOp.Map(
            compiled,
            new PointWiseOp.Projection(table.args),
            compiled.device,
            compiled.handle_thingtalk,
            table
        );
    } else if (table.isCompute) {
        const compiled = compileTableToOps(table.table, extra_in_params);
        return new TableOp.Map(
            compiled,
            new PointWiseOp.Compute(table.expression,
                table.alias || getScalarExpressionName(table.expression)),
            compiled.device,
            compiled.handle_thingtalk,
            table
        );
    } else if (table.isAggregation) {
        let reduceop;
        if (table.operator === 'count' && table.field === '*')
            reduceop = new ReduceOp.Count;
        else if (table.operator === 'count')
            reduceop = new ReduceOp.CountDistinct(table.field);
        else if (table.operator === 'avg')
            reduceop = new ReduceOp.Average(table.field, table.schema.out[table.field]);
        else
            reduceop = new ReduceOp.SimpleAggregation(table.operator, table.field, table.schema.out[table.field]);

        const compiled = compileTableToOps(table.table, extra_in_params);
        return new TableOp.Reduce(
            compiled,
            reduceop,
            compiled.device,
            compiled.handle_thingtalk,
            table
        );
    } else if (table.isIndex && table.indices.length === 1 && table.indices[0].isNumber && table.table.isSort) {
        // convert sort followed by a single index into argminmax
        let reduceop;
        if (table.indices[0].value === 1 || table.indices[0].value === -1) {
            // common case of simple argmin/argmax
            let argminmaxop;
            if ((table.indices[0].value === 1 && table.table.direction === 'asc') ||
                (table.indices[0].value === -1 && table.table.direction === 'desc'))
                argminmaxop = 'argmin';
            else
                argminmaxop = 'argmax';
            reduceop = new ReduceOp.SimpleArgMinMax(argminmaxop, table.table.field);
        } else {
            let argminmaxop;
            if (table.table.direction === 'asc')
                argminmaxop = 'argmin';
            else
                argminmaxop = 'argmax';
            reduceop = new ReduceOp.ComplexArgMinMax(argminmaxop, table.table.field, table.indices[0], Ast.Value.Number(1));
        }

        const compiled = compileTableToOps(table.table.table, extra_in_params);
        return new TableOp.Reduce(
            compiled,
            reduceop,
            compiled.device,
            compiled.handle_thingtalk,
            table
        );
    } else if (table.isSlice && table.table.isSort) {
        // convert sort followed by a single slice into argminmax
        let argminmaxop;
        if (table.table.direction === 'asc')
            argminmaxop = 'argmin';
        else
            argminmaxop = 'argmax';
        let reduceop = new ReduceOp.ComplexArgMinMax(argminmaxop, table.table.field, table.base, table.limit);

        const compiled = compileTableToOps(table.table.table, extra_in_params);
        return new TableOp.Reduce(
            compiled,
            reduceop,
            compiled.device,
            compiled.handle_thingtalk,
            table
        );
    } else if (table.isSort) {
        const compiled = compileTableToOps(table.table, extra_in_params);
        return new TableOp.Reduce(
            compiled,
            new ReduceOp.Sort(table.field, table.direction),
            compiled.device,
            compiled.handle_thingtalk,
            table
        );
    } else if (table.isIndex && table.indices.length === 1 && table.indices[0].isNumber && table.indices[0].value > 0) {
        const compiled = compileTableToOps(table.table, extra_in_params);
        return new TableOp.Reduce(
            compiled,
            new ReduceOp.SimpleIndex(table.indices[0]),
            compiled.device,
            compiled.handle_thingtalk,
            table
        );
    } else if (table.isIndex) {
        const compiled = compileTableToOps(table.table, extra_in_params);
        return new TableOp.Reduce(
            compiled,
            new ReduceOp.ComplexIndex(table.indices),
            compiled.device,
            compiled.handle_thingtalk,
            table
        );
    } else if (table.isSlice) {
        const compiled = compileTableToOps(table.table, extra_in_params);
        return new TableOp.Reduce(
            compiled,
            new ReduceOp.Slice(table.base, table.limit),
            compiled.device,
            compiled.handle_thingtalk,
            table
        );
    } else if (table.isArgMinMax) {
        let reduceop;
        if (table.base.isNumber && table.base.value === 1 &&
            table.limit.isNumber && table.limit.value === 1)
            reduceop = new ReduceOp.SimpleArgMinMax(table.operator, table.field);
        else
            reduceop = new ReduceOp.ComplexArgMinMax(table.operator, table.field, table.base, table.limit);

        const compiled = compileTableToOps(table.table, extra_in_params);
        return new TableOp.Reduce(
            compiled,
            reduceop,
            compiled.device,
            compiled.handle_thingtalk,
            table
        );
    } else if (table.isJoin) {
        if (table.in_params.length === 0) {
            const lhs = compileTableToOps(table.lhs, extra_in_params);
            const rhs = compileTableToOps(table.rhs, extra_in_params);
            const invocation = sameDevice(lhs.device, rhs.device) ? lhs.device : null;
            const handle_thingtalk = sameDevice(lhs.device, rhs.device) ? lhs.handle_thingtalk && rhs.handle_thingtalk : false;

            return new TableOp.CrossJoin(lhs, rhs, invocation, handle_thingtalk, table);
        } else {
            let lhs_in_params = [];
            let rhs_in_params = [];
            for (let in_param of extra_in_params) {
                if (in_param.name in table.lhs.schema.inReq ||
                    in_param.name in table.lhs.schema.inOpt)
                    lhs_in_params.push(in_param);
                if (in_param.name in table.rhs.schema.inReq ||
                    in_param.name in table.rhs.schema.inOpt)
                    rhs_in_params.push(in_param);
            }

            const lhs = compileTableToOps(table.lhs, lhs_in_params);
            const rhs = compileTableToOps(table.rhs, rhs_in_params.concat(table.in_params));
            const device = sameDevice(lhs.device, rhs.device) ? lhs.device : null;
            const handle_thingtalk = sameDevice(lhs.device, rhs.device) ? lhs.handle_thingtalk && rhs.handle_thingtalk : false;

            return new TableOp.NestedLoopJoin(lhs, rhs, device, handle_thingtalk, table);
        }
    } else {
        throw new TypeError();
    }
}

function optimizeStreamOp(streamop) {
    return streamop;
}
/*function optimizeTableOp(tableop) {
    return tableop;
}*/

function optimizeLoop(what, optimizer) {
    let optimized = optimizer(what);
    if (optimized !== what)
        return optimizeLoop(optimized, optimizer);
    else
        return optimized;
}

function getDefaultProjection(statement) {
    let stream;
    if (statement.isRule)
        stream = statement.stream;
    else if (statement.table)
        stream = statement.table;
    if (stream && stream.schema && stream.schema.default_projection)
        return stream.schema.default_projection;
    return [];
}

// compile a rule/command statement to a RuleOp
function compileStatementToOp(statement) {
    let default_projection = getDefaultProjection(statement);
    let args = [];
    if (default_projection.length > 0) {
        statement.actions.forEach((action) => {
            if (action.isInvocation && action.invocation.selector.isBuiltin) {
                args = args.concat(default_projection);
            } else {
                action.invocation.in_params.forEach((p) => {
                    if (p.value.isVarRef)
                        args.push(p.value.name);
                });
            }
        });
    }
    args = [...new Set(args)];
    let streamop;
    if (statement.isRule) {
        streamop = compileStreamToOps(statement.stream);
        if (args.length > 0) {
            let schema = statement.stream.schema;
            schema = schema.filterArguments((a) => a.is_input || args.includes(a.name));
            streamop = new StreamOp.Map(
                streamop,
                new PointWiseOp.Projection(args),
                new Ast.Stream.Projection(null, statement.stream, args, schema)
            );
        }
    } else if (statement.table) {
        let tableop = compileTableToOps(statement.table, []);
        if (args.length > 0) {
            let schema = statement.table.schema.clone();
            schema = schema.filterArguments((a) => a.is_input || args.includes(a.name));
            let newtable = new Ast.Table.Projection(null, statement.table, args, schema);
            tableop = new TableOp.Map(
                tableop,
                new PointWiseOp.Projection(args),
                tableop.device,
                tableop.handle_thingtalk,
                newtable
            );
            streamop = new StreamOp.Join(StreamOp.Now, tableop, newtable);
        } else {
            streamop = new StreamOp.Join(StreamOp.Now, tableop, statement.table);
        }
    } else {
        streamop = StreamOp.Now;
    }

    streamop = optimizeLoop(streamop, optimizeStreamOp);

    return new RuleOp(streamop, statement.actions, statement);
}

module.exports = {
    compileStatementToOp,
    compileStreamToOps,
    compileTableToOps
};

1	// -- mode: js; indent-tabs-mode: nil; js-basic-offset: 4 --
2	//
3	// This file is part of ThingTalk
4	//
5	// Copyright 2015-2018 The Board of Trustees of the Leland Stanford Junior University
6	//
7	// Author: Giovanni Campagna <gcampagn@cs.stanford.edu>
8	//
9	// See COPYING for details
10	"use strict";
11
12	const Ast = require('../ast');	1✔
13	const { NotImplementedError } = require('../errors');	1✔
14	const { getScalarExpressionName } = require('../utils');	1✔
15
16	const { PointWiseOp, StreamOp, TableOp, RuleOp } = require('./ops');	1✔
17	const ReduceOp = require('./reduceop');	1✔
18
19	function sameDevice(lhs, rhs) {
20	if (lhs.isBuiltin && rhs.isBuiltin)	16!
21	return true;	×
22	if (lhs.isBuiltin \|\| rhs.isBuiltin)	16!
23	return false;	×
24	if (lhs.kind !== rhs.kind)	16✔
25	return false;	10✔
26	if (lhs.id !== rhs.id)	6!
27	return false;	×
28	if (lhs.principal !== rhs.principal)	6!
29	return false;	×
30	return true;	6✔
31	}
32
33	// compile a table that is being monitored to a stream
34	function compileMonitorTableToOps(table) {
35	if (table.isVarRef \|\|	58!
36	table.isWindow \|\|
37	table.isTimeSeries \|\|
38	table.isHistory \|\|
39	table.isSequence \|\|
40	table.isAlias)
41	throw new NotImplementedError(table);	×
42
43	if (table.isInvocation) {	58✔
44	// subscribe is optimistic, we still need EdgeNew
45	return new StreamOp.EdgeNew(	39✔
46	new StreamOp.InvokeSubscribe(table.invocation, Ast.BooleanExpression.True, table),
47	table
48	);
49	} else if (table.isFilter) {	19✔
50	return new StreamOp.Filter(	16✔
51	compileMonitorTableToOps(table.table),
52	table.filter,
53	table
54	);
55	} else if (table.isProjection) {	3!
56	// note the "edge new" operation here, because
57	// the projection might cause fewer values to
58	// be new
59	return new StreamOp.EdgeNew(	×
60	new StreamOp.Map(
61	compileMonitorTableToOps(table.table),
62	new PointWiseOp.Projection(table.args),
63	table
64	),
65	table
66	);
67	} else if (table.isSort \|\| table.isIndex \|\| table.isSlice) {	3!
68	// sort, index and slice have no effect on monitor
69	//
70	// XXX is this correct?
71	return compileMonitorTableToOps(table);	×
72	} else if (table.isCompute) {	3!
73	// note the "edge new" operation here, because
74	// the projection might cause fewer values to
75	// be new
76	return new StreamOp.EdgeNew(	×
77	new StreamOp.Map(
78	compileMonitorTableToOps(table.table),
79	new PointWiseOp.Compute(table.expression),
80	table
81	),
82	table
83	);
84	} else if (table.isAggregation) {	3✔
85	// for an aggregation, we subscribe to the inner table
86	// (ie react to all changes), then when the table changes
87	// we fetch it completely again and compute the aggregation
88	// note the "edge new" operation here, because
89	// the aggregation might cause fewer values to
90	// be new
91	return new StreamOp.EdgeNew(new StreamOp.InvokeTable(	1✔
92	compileMonitorTableToOps(table.table),
93	compileTableToOps(table, []),
94	table
95	), table);
96	} else if (table.isJoin) {	2!
97	if (table.in_params.length === 0) {	2!
98	// if there is no parameter passing, we can individually monitor
99	// the two tables and return the union
100	return new StreamOp.EdgeNew(new StreamOp.Union(	2✔
101	compileMonitorTableToOps(table.lhs),
102	compileMonitorTableToOps(table.rhs),
103	table),
104	table
105	);
106	} else {
107	// otherwise we need to subscribe to the left hand side, and
108	// every time it fires, create/update a subscription to the
109	// right hand side
110	// this is VERY MESSY
111	// so it's not implemented
112	throw new NotImplementedError(table);	×
113	}
114	} else {
115	throw new TypeError();	×
116	}
117	}
118
119	// compile a TT stream to a stream op and zero or more
120	// tableops
121	function compileStreamToOps(stream) {
122	if (stream.isAlias)	84!
123	throw new NotImplementedError(stream);	×
124
125	if (stream.isVarRef) {	84✔
126	return new StreamOp.InvokeVarRef(stream.name, stream.in_params, stream);	1✔
127	} else if (stream.isTimer) {	83✔
128	return new StreamOp.Timer(stream.base, stream.interval, stream.frequency, stream);	16✔
129	} else if (stream.isAtTimer) {	67✔
130	return new StreamOp.AtTimer(stream.time, stream.expiration_date, stream);	7✔
131	} else if (stream.isMonitor) {	60✔
132	return compileMonitorTableToOps(stream.table);	37✔
133	} else if (stream.isEdgeNew) {	23!
134	let op = compileStreamToOps(stream.stream);	×
135	return new StreamOp.EdgeNew(op, op.ast);	×
136	} else if (stream.isEdgeFilter) {	23!
137	let op = compileStreamToOps(stream.stream);	×
138	return new StreamOp.EdgeFilter(op, stream.filter, op.ast);	×
139	} else if (stream.isFilter) {	23✔
140	let op = compileStreamToOps(stream.stream);	3✔
141	return new StreamOp.Filter(op, stream.filter, op.ast);	3✔
142	} else if (stream.isProjection) {	20!
143	let op = compileStreamToOps(stream.stream);	×
144	return new StreamOp.Map(op, new PointWiseOp.Projection(stream.args), op.ast);	×
145	} else if (stream.isCompute) {	20!
146	let op = compileStreamToOps(stream.stream);	×
147	return new StreamOp.Map(op, new PointWiseOp.Compute(stream.expression,	×
148	stream.alias \|\| getScalarExpressionName(stream.expression)), op.ast);	×
149	} else if (stream.isJoin) {	20!
150	let streamOp = compileStreamToOps(stream.stream);	20✔
151	let tableOp = compileTableToOps(stream.table, stream.in_params);	20✔
152	return new StreamOp.Join(streamOp, tableOp, stream);	20✔
153	} else {
154	throw new TypeError();	×
155	}
156	}
157
158	function compileTableToOps(table, extra_in_params) {
159	if (table.isWindow \|\|	221!
160	table.isTimeSeries \|\|
161	table.isHistory \|\|
162	table.isSequence \|\|
163	table.isAlias)
164	throw new NotImplementedError(table);	×
165
166	if (table.isVarRef) {	221✔
167	const compiled = new TableOp.InvokeVarRef(table.name, table.in_params.concat(extra_in_params), table);	18✔
168	compiled.device = null;	18✔
169	compiled.handle_thingtalk = false;	18✔
170	return compiled;	18✔
171	} else if (table.isResultRef) {	203✔
172	return new TableOp.ReadResult(table.kind + ':' + table.channel, table.index, table.schema, table);	8✔
173	} else if (table.isInvocation) {	195✔
174	const device = table.invocation.selector;	120✔
175	const handle_thingtalk = table.schema.annotations['handle_thingtalk'] ? table.schema.annotations['handle_thingtalk'].value : false;	120✔
176	return new TableOp.InvokeGet(	120✔
177	table.invocation,
178	extra_in_params,
179	Ast.BooleanExpression.True,
180	device,
181	handle_thingtalk,
182	table
183	);
184	} else if (table.isFilter) {	75✔
185	const compiled = compileTableToOps(table.table, extra_in_params);	24✔
186	return new TableOp.Filter(	24✔
187	compiled,
188	table.filter,
189	compiled.device,
190	compiled.handle_thingtalk,
191	table
192	);
193	} else if (table.isProjection) {	51✔
194	const compiled = compileTableToOps(table.table, extra_in_params);	1✔
195	return new TableOp.Map(	1✔
196	compiled,
197	new PointWiseOp.Projection(table.args),
198	compiled.device,
199	compiled.handle_thingtalk,
200	table
201	);
202	} else if (table.isCompute) {	50✔
203	const compiled = compileTableToOps(table.table, extra_in_params);	4✔
204	return new TableOp.Map(	4✔
205	compiled,
206	new PointWiseOp.Compute(table.expression,
207	table.alias \|\| getScalarExpressionName(table.expression)),	8✔
208	compiled.device,
209	compiled.handle_thingtalk,
210	table
211	);
212	} else if (table.isAggregation) {	46✔
213	let reduceop;
214	if (table.operator === 'count' && table.field === '*')	12✔
215	reduceop = new ReduceOp.Count;	4✔
216	else if (table.operator === 'count')	8✔
217	reduceop = new ReduceOp.CountDistinct(table.field);	2✔
218	else if (table.operator === 'avg')	6✔
219	reduceop = new ReduceOp.Average(table.field, table.schema.out[table.field]);	2✔
220	else
221	reduceop = new ReduceOp.SimpleAggregation(table.operator, table.field, table.schema.out[table.field]);	4✔
222
223	const compiled = compileTableToOps(table.table, extra_in_params);	12✔
224	return new TableOp.Reduce(	12✔
225	compiled,
226	reduceop,
227	compiled.device,
228	compiled.handle_thingtalk,
229	table
230	);
231	} else if (table.isIndex && table.indices.length === 1 && table.indices[0].isNumber && table.table.isSort) {	34✔
232	// convert sort followed by a single index into argminmax
233	let reduceop;
234	if (table.indices[0].value === 1 \|\| table.indices[0].value === -1) {	9✔
235	// common case of simple argmin/argmax
236	let argminmaxop;
237	if ((table.indices[0].value === 1 && table.table.direction === 'asc') \|\|	5✔
238	(table.indices[0].value === -1 && table.table.direction === 'desc'))
239	argminmaxop = 'argmin';	3✔
240	else
241	argminmaxop = 'argmax';	2✔
242	reduceop = new ReduceOp.SimpleArgMinMax(argminmaxop, table.table.field);	5✔
243	} else {
244	let argminmaxop;
245	if (table.table.direction === 'asc')	4✔
246	argminmaxop = 'argmin';	3✔
247	else
248	argminmaxop = 'argmax';	1✔
249	reduceop = new ReduceOp.ComplexArgMinMax(argminmaxop, table.table.field, table.indices[0], Ast.Value.Number(1));	4✔
250	}
251
252	const compiled = compileTableToOps(table.table.table, extra_in_params);	9✔
253	return new TableOp.Reduce(	9✔
254	compiled,
255	reduceop,
256	compiled.device,
257	compiled.handle_thingtalk,
258	table
259	);
260	} else if (table.isSlice && table.table.isSort) {	25✔
261	// convert sort followed by a single slice into argminmax
262	let argminmaxop;
263	if (table.table.direction === 'asc')	3!
264	argminmaxop = 'argmin';	3✔
265	else
UNCOV 266	argminmaxop = 'argmax';	×
267	let reduceop = new ReduceOp.ComplexArgMinMax(argminmaxop, table.table.field, table.base, table.limit);	3✔
268
269	const compiled = compileTableToOps(table.table.table, extra_in_params);	3✔
270	return new TableOp.Reduce(	3✔
271	compiled,
272	reduceop,
273	compiled.device,
274	compiled.handle_thingtalk,
275	table
276	);
277	} else if (table.isSort) {	22✔
278	const compiled = compileTableToOps(table.table, extra_in_params);	4✔
279	return new TableOp.Reduce(	4✔
280	compiled,
281	new ReduceOp.Sort(table.field, table.direction),
282	compiled.device,
283	compiled.handle_thingtalk,
284	table
285	);
286	} else if (table.isIndex && table.indices.length === 1 && table.indices[0].isNumber && table.indices[0].value > 0) {	18✔
287	const compiled = compileTableToOps(table.table, extra_in_params);	4✔
288	return new TableOp.Reduce(	4✔
289	compiled,
290	new ReduceOp.SimpleIndex(table.indices[0]),
291	compiled.device,
292	compiled.handle_thingtalk,
293	table
294	);
295	} else if (table.isIndex) {	14✔
296	const compiled = compileTableToOps(table.table, extra_in_params);	4✔
297	return new TableOp.Reduce(	4✔
298	compiled,
299	new ReduceOp.ComplexIndex(table.indices),
300	compiled.device,
301	compiled.handle_thingtalk,
302	table
303	);
304	} else if (table.isSlice) {	10✔
305	const compiled = compileTableToOps(table.table, extra_in_params);	2✔
306	return new TableOp.Reduce(	2✔
307	compiled,
308	new ReduceOp.Slice(table.base, table.limit),
309	compiled.device,
310	compiled.handle_thingtalk,
311	table
312	);
313	} else if (table.isArgMinMax) {	8!
314	let reduceop;
UNCOV 315	if (table.base.isNumber && table.base.value === 1 &&	×
316	table.limit.isNumber && table.limit.value === 1)
UNCOV 317	reduceop = new ReduceOp.SimpleArgMinMax(table.operator, table.field);	×
318	else
UNCOV 319	reduceop = new ReduceOp.ComplexArgMinMax(table.operator, table.field, table.base, table.limit);	×
320
UNCOV 321	const compiled = compileTableToOps(table.table, extra_in_params);	×
322	return new TableOp.Reduce(	×
323	compiled,
324	reduceop,
325	compiled.device,
326	compiled.handle_thingtalk,
327	table
328	);
329	} else if (table.isJoin) {	8!
330	if (table.in_params.length === 0) {	8✔
331	const lhs = compileTableToOps(table.lhs, extra_in_params);	4✔
332	const rhs = compileTableToOps(table.rhs, extra_in_params);	4✔
333	const invocation = sameDevice(lhs.device, rhs.device) ? lhs.device : null;	4✔
334	const handle_thingtalk = sameDevice(lhs.device, rhs.device) ? lhs.handle_thingtalk && rhs.handle_thingtalk : false;	4!
335
336	return new TableOp.CrossJoin(lhs, rhs, invocation, handle_thingtalk, table);	4✔
337	} else {
338	let lhs_in_params = [];	4✔
339	let rhs_in_params = [];	4✔
340	for (let in_param of extra_in_params) {	4✔
NEW 341	if (in_param.name in table.lhs.schema.inReq \|\|	×
342	in_param.name in table.lhs.schema.inOpt)
UNCOV 343	lhs_in_params.push(in_param);	×
UNCOV 344	if (in_param.name in table.rhs.schema.inReq \|\|	×
345	in_param.name in table.rhs.schema.inOpt)
UNCOV 346	rhs_in_params.push(in_param);	×
347	}
348
349	const lhs = compileTableToOps(table.lhs, lhs_in_params);	4✔
350	const rhs = compileTableToOps(table.rhs, rhs_in_params.concat(table.in_params));	4✔
351	const device = sameDevice(lhs.device, rhs.device) ? lhs.device : null;	4!
352	const handle_thingtalk = sameDevice(lhs.device, rhs.device) ? lhs.handle_thingtalk && rhs.handle_thingtalk : false;	4!
353
354	return new TableOp.NestedLoopJoin(lhs, rhs, device, handle_thingtalk, table);	4✔
355	}
356	} else {
UNCOV 357	throw new TypeError();	×
358	}
359	}
360
361	function optimizeStreamOp(streamop) {
362	return streamop;	204✔
363	}
364	/*function optimizeTableOp(tableop) {
365	return tableop;
366	}*/
367
368	function optimizeLoop(what, optimizer) {
369	let optimized = optimizer(what);	204✔
370	if (optimized !== what)	204!
UNCOV 371	return optimizeLoop(optimized, optimizer);	×
372	else
373	return optimized;	204✔
374	}
375
376	function getDefaultProjection(statement) {
377	let stream;
378	if (statement.isRule)	204✔
379	stream = statement.stream;	60✔
380	else if (statement.table)	144✔
381	stream = statement.table;	107✔
382	if (stream && stream.schema && stream.schema.default_projection)	204✔
383	return stream.schema.default_projection;	167✔
384	return [];	37✔
385	}
386
387	// compile a rule/command statement to a RuleOp
388	function compileStatementToOp(statement) {
389	let default_projection = getDefaultProjection(statement);	204✔
390	let args = [];	204✔
391	if (default_projection.length > 0) {	204✔
392	statement.actions.forEach((action) => {	4✔
393	if (action.isInvocation && action.invocation.selector.isBuiltin) {	4✔
394	args = args.concat(default_projection);	2✔
395	} else {
396	action.invocation.in_params.forEach((p) => {	2✔
397	if (p.value.isVarRef)	2!
398	args.push(p.value.name);	2✔
399	});
400	}
401	});
402	}
403	args = [...new Set(args)];	204✔
404	let streamop;
405	if (statement.isRule) {	204✔
406	streamop = compileStreamToOps(statement.stream);	60✔
407	if (args.length > 0) {	60!
UNCOV 408	let schema = statement.stream.schema;	×
UNCOV 409	schema = schema.filterArguments((a) => a.is_input \|\| args.includes(a.name));	×
UNCOV 410	streamop = new StreamOp.Map(	×
411	streamop,
412	new PointWiseOp.Projection(args),
413	new Ast.Stream.Projection(null, statement.stream, args, schema)
414	);
415	}
416	} else if (statement.table) {	144✔
417	let tableop = compileTableToOps(statement.table, []);	107✔
418	if (args.length > 0) {	107✔
419	let schema = statement.table.schema.clone();	4✔
420	schema = schema.filterArguments((a) => a.is_input \|\| args.includes(a.name));	272✔
421	let newtable = new Ast.Table.Projection(null, statement.table, args, schema);	4✔
422	tableop = new TableOp.Map(	4✔
423	tableop,
424	new PointWiseOp.Projection(args),
425	tableop.device,
426	tableop.handle_thingtalk,
427	newtable
428	);
429	streamop = new StreamOp.Join(StreamOp.Now, tableop, newtable);	4✔
430	} else {
431	streamop = new StreamOp.Join(StreamOp.Now, tableop, statement.table);	103✔
432	}
433	} else {
434	streamop = StreamOp.Now;	37✔
435	}
436
437	streamop = optimizeLoop(streamop, optimizeStreamOp);	204✔
438
439	return new RuleOp(streamop, statement.actions, statement);	204✔
440	}
441
442	module.exports = {	1✔
443	compileStatementToOp,
444	compileStreamToOps,
445	compileTableToOps
446	};

stanford-oval / thingtalk / 807

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