groq - npm Package Compare versions

Comparing version 0.0.7 to 0.0.8

lib/exec/Executor.js

@@ -148,3 +148,4 @@ 'use strict';
         case 'source':
-          const data = scope.dataForSource(operation.id);
+          const data = yield scope.dataForSource(operation.id);
+          (0, _debug2.default)("sourceId will be", operation.id);
           return scope.child({
@@ -189,2 +190,4 @@ sourceId: operation.id,
           return _this2.execMatch(operation, scope);
+        case 'mapJoin':
+          return _this2.execMapJoin(operation, scope);
         default:
@@ -461,6 +464,8 @@ throw new Error(`(exec) Unknown operation ${operation.op}`);
     return _asyncToGenerator(function* () {
-      scope = yield (0, _fetch2.default)(scope, operation, _this12.fetcher);
+      scope = yield (0, _fetch2.default)(scope, operation, _this12.fetcher, _this12);
       const ops = operation.operations.slice();
       const first = ops.shift();
+      (0, _debug2.default)('first op', first);
       let current = yield _this12.exec(first, scope);
+      (0, _debug2.default)('first op result:', current);
       while (ops.length > 0) {
@@ -607,4 +612,32 @@ (0, _debug2.default)('current:', current);
 
+  execMapJoin(operation, scope) {
+    var _this24 = this;
+
+    return _asyncToGenerator(function* () {
+      (0, _debug2.default)('execMapJoin', operation);
+      const source = (0, _scopeTools.asPlainValue)(scope);
+      const parent = scope.parent;
+      const mapValues = (0, _scopeTools.asPlainValue)((yield _this24.exec(operation.pipeline, parent)));
+      (0, _debug2.default)('execMapJoin mapValues', mapValues);
+      (0, _debug2.default)('execMapJoin source', source);
+      const joinResult = mapValues.map(function (id) {
+        return scope.child({
+          value: source.find(function (item) {
+            return item._id == id;
+          })
+        });
+      });
+      (0, _debug2.default)('joinResult', joinResult);
+      return joinResult;
+    })();
+  }
+
   execAccessor(operation, scope) {
     return _asyncToGenerator(function* () {
+      (0, _debug2.default)('execAccessor()', operation.path, scope);
+      if (!scope) {
+        return new _Scope2.default({
+          value: null
+        });
+      }
       return scope.resolveAccessor(operation.path);
@@ -624,6 +657,6 @@ })();
   execFilter(operation, scope) {
-    var _this24 = this;
+    var _this25 = this;
 
     return _asyncToGenerator(function* () {
-      return _this24.exec(operation.filter, scope);
+      return _this25.exec(operation.filter, scope);
     })();
@@ -630,0 +663,0 @@ }

lib/exec/fetch.js

@@ -9,3 +9,3 @@ 'use strict';
 let compile = (() => {
-  var _ref2 = _asyncToGenerator(function* (scope, pipeline) {
+  var _ref2 = _asyncToGenerator(function* (scope, pipeline, executor) {
     const operations = pipeline.operations;
@@ -22,3 +22,5 @@ if (operations.length == 0) {
     const fetchSpec = new FetchSpec(scope, sourceId);
-    operations.slice(1).reverse().forEach(function (operation) {
+    const ops = operations.slice(1);
+    while (ops.length > 0) {
+      const operation = ops.pop();
       (0, _debug2.default)('compiling', operation);
@@ -29,2 +31,5 @@ switch (operation.op) {
           break;
+        case 'mapJoin':
+          const filter = yield generalizeMapJoin(operation, scope, executor);
+          fetchSpec.applyFilter(filter);
         case 'object':
@@ -42,7 +47,7 @@ fetchSpec.project(operation.operations);
         default:
-          throw new Error(`Unknown pipeline operation ${operation.op} when collapsing fetche operation`);
+          throw new Error(`Unknown pipeline operation ${operation.op} when compiling fetch-operation`);
       }
-    });
+    }
 
-    const generalizedFilter = (0, _generalizeJoinFilter2.default)(fetchSpec.filter, scope);
+    const generalizedFilter = yield (0, _generalizeJoinFilter2.default)(fetchSpec.filter, scope);
     const wasJoin = generalizedFilter !== fetchSpec.filter;
@@ -59,6 +64,7 @@
     // to be fetched in order to be executed.
+    (0, _debug2.default)('fetchSpec => ', fetchSpec);
     return fetchSpec;
   });
 
-  return function compile(_x4, _x5) {
+  return function compile(_x5, _x6, _x7) {
     return _ref2.apply(this, arguments);
@@ -68,2 +74,57 @@ };
 
+// Given a map join, generates a filter that will fetch all possible source values for that
+// join.
+
+
+let generalizeMapJoin = (() => {
+  var _ref3 = _asyncToGenerator(function* (operation, scope, executor) {
+    if (!scope) {
+      (0, _debug2.default)("mapJoin over null short circuited");
+      fetchSpec.applyFilter(new _operations.Literal({
+        type: 'boolean',
+        value: false
+      }));
+    }
+    let ids;
+    if (scope.sourceId) {
+      const source = yield scope.dataForSource(scope.sourceId);
+      (0, _debug2.default)('expanding mapJoin', scope, scope.sourceId);
+      const idPromises = source.documents.map((() => {
+        var _ref4 = _asyncToGenerator(function* (document) {
+          const scope = new _Scope2.default({
+            value: document
+          });
+          const resolved = (0, _scopeTools.asPlainValue)((yield executor.exec(operation.pipeline, scope))).filter(Boolean);
+          (0, _debug2.default)('resolved', resolved);
+          if (resolved) {
+            return resolved;
+          }
+        });
+
+        return function (_x11) {
+          return _ref4.apply(this, arguments);
+        };
+      })());
+      const idMap = {};
+      ids = (yield Promise.all(idPromises)).forEach(function (idSet) {
+        return idSet.forEach(function (id) {
+          idMap[id] = true;
+        });
+      });
+      ids = Object.keys(idMap);
+    } else {
+      // TODO
+      throw new Error(`mapJoin on literals not supported yet`);
+    }
+    (0, _debug2.default)('expanded mapJoin', ids);
+    return new _operations.BinaryOperator('in', new _operations.Accessor([new _operations.Attribute('_id')]), new _operations.Literal({
+      value: ids
+    }));
+  });
+
+  return function generalizeMapJoin(_x8, _x9, _x10) {
+    return _ref3.apply(this, arguments);
+  };
+})();
+
 var _operations = require('../plan/operations');
@@ -83,2 +144,4 @@
 
+var _scopeTools = require('./scopeTools');
+
 var _path = require('path');
@@ -105,3 +168,3 @@
 exports.default = (() => {
-  var _ref = _asyncToGenerator(function* (scope, pipeline, fetcher) {
+  var _ref = _asyncToGenerator(function* (scope, pipeline, fetcher, executor) {
     (0, _debug2.default)('fetch()', scope, pipeline);
@@ -115,3 +178,4 @@ const sourceId = extractSourceId(pipeline);
     // We need to actually go get it
-    const fetchSpec = yield compile(scope, pipeline);
+    const responder = scope.claimSource(sourceId);
+    const fetchSpec = yield compile(scope, pipeline, executor);
     const response = yield fetcher(fetchSpec);
@@ -122,9 +186,10 @@ // Cache the result on the scope so it will bind to the source
     scope.cacheRefs(response.refs);
-    return scope.cacheSource(sourceId, {
+    responder.resolve({
       documents: response.results,
       start: response.start || 0
     });
+    return scope;
   });
 
-  function fetch(_x, _x2, _x3) {
+  function fetch(_x, _x2, _x3, _x4) {
     return _ref.apply(this, arguments);
@@ -131,0 +196,0 @@ }

lib/exec/generalizeJoinFilter.js

@@ -6,4 +6,62 @@ 'use strict';
 });
-exports.default = generalizeJoinFilter;
 
+let generalizeEQ = (() => {
+  var _ref2 = _asyncToGenerator(function* (operation, scope) {
+    const lhsIsJoin = isJoinAccessor(operation.lhs);
+    const rhsIsJoin = isJoinAccessor(operation.rhs);
+    if (!lhsIsJoin && !rhsIsJoin) {
+      (0, _debug2.default)("generalizeJoinFilter() (is not a join)");
+      // Not part of the join
+      return operation;
+    }
+
+    if (lhsIsJoin && rhsIsJoin) {
+      throw new Error(`In a join, only the lhs or rhs can reference parent`);
+    }
+    let join;
+    let constant;
+    if (lhsIsJoin) {
+      join = operation.lhs;
+      constant = operation.rhs;
+    } else {
+      join = operation.rhs;
+      constant = operation.lhs;
+    }
+    const joinScopes = yield scope.child({
+      value: {}
+    }).resolveAccessorForAll(join.path);
+
+    if (joinScopes.length == 0) {
+      return {
+        op: 'literal',
+        value: false
+      };
+    }
+    if (joinScopes.length == 1) {
+      return {
+        op: 'eq',
+        lhs: constant,
+        rhs: {
+          op: 'literal',
+          value: joinScopes[0].value
+        }
+      };
+    }
+    return {
+      op: 'in',
+      lhs: constant,
+      rhs: {
+        op: 'literal',
+        value: joinScopes.map(function (item) {
+          return item.value;
+        })
+      }
+    };
+  });
+
+  return function generalizeEQ(_x3, _x4) {
+    return _ref2.apply(this, arguments);
+  };
+})();
+
 var _rewrite = require('../plan/rewrite');
@@ -19,62 +77,23 @@
 
-function generalizeJoinFilter(node, scope) {
-  (0, _debug2.default)('generalizeJoinFilter()', node, scope);
-  return (0, _rewrite2.default)(node, operation => {
-    switch (operation.op) {
-      case 'eq':
-        return generalizeEQ(operation, scope);
-      default:
-        return operation;
-    }
+function _asyncToGenerator(fn) { return function () { var gen = fn.apply(this, arguments); return new Promise(function (resolve, reject) { function step(key, arg) { try { var info = gen[key](arg); var value = info.value; } catch (error) { reject(error); return; } if (info.done) { resolve(value); } else { return Promise.resolve(value).then(function (value) { step("next", value); }, function (err) { step("throw", err); }); } } return step("next"); }); }; }
+
+exports.default = (() => {
+  var _ref = _asyncToGenerator(function* (node, scope) {
+    (0, _debug2.default)('generalizeJoinFilter()', node, scope);
+    return (0, _rewrite2.default)(node, function (operation) {
+      switch (operation.op) {
+        case 'eq':
+          return generalizeEQ(operation, scope);
+        default:
+          return operation;
+      }
+    });
   });
-}
 
-function generalizeEQ(operation, scope) {
-  const lhsIsJoin = isJoinAccessor(operation.lhs);
-  const rhsIsJoin = isJoinAccessor(operation.rhs);
-  if (!lhsIsJoin && !rhsIsJoin) {
-    // Not part of the join
-    return operation;
+  function generalizeJoinFilter(_x, _x2) {
+    return _ref.apply(this, arguments);
   }
-  if (lhsIsJoin && rhsIsJoin) {
-    throw new Error(`In a join, only the lhs or rhs can reference parent`);
-  }
-  let join;
-  let constant;
-  if (lhsIsJoin) {
-    join = operation.lhs;
-    constant = operation.rhs;
-  } else {
-    join = operation.rhs;
-    constant = operation.lhs;
-  }
-  const joinScopes = scope.child({
-    value: {}
-  }).resolveAccessorForAll(join.path);
 
-  if (joinScopes.length == 0) {
-    return {
-      op: 'literal',
-      value: false
-    };
-  }
-  if (joinScopes.length == 1) {
-    return {
-      op: 'eq',
-      lhs: constant,
-      rhs: {
-        op: 'literal',
-        value: joinScopes[0].value
-      }
-    };
-  }
-  return {
-    op: 'in',
-    lhs: constant,
-    rhs: {
-      op: 'literal',
-      value: joinScopes.map(item => item.value)
-    }
-  };
-}
+  return generalizeJoinFilter;
+})();
 
@@ -81,0 +100,0 @@ function isJoinAccessor(operation) {

lib/exec/Scope.js

@@ -13,2 +13,4 @@ 'use strict';
 
+function _asyncToGenerator(fn) { return function () { var gen = fn.apply(this, arguments); return new Promise(function (resolve, reject) { function step(key, arg) { try { var info = gen[key](arg); var value = info.value; } catch (error) { reject(error); return; } if (info.done) { resolve(value); } else { return Promise.resolve(value).then(function (value) { step("next", value); }, function (err) { step("throw", err); }); } } return step("next"); }); }; }
+
 class Scope {
@@ -21,3 +23,4 @@ constructor(options) {
           refs = options.refs,
-          start = options.start;
+          start = options.start,
+          path = options.path;
     // A single source cahce shared between all scopes in the same
@@ -34,2 +37,4 @@ // execution chain.
     this.value = value;
+    // The path of this specific value from source
+    this.path = path || [];
     // A cache of all refs seen in all documents fetched so far, shared between all scopes in
@@ -44,5 +49,9 @@ // the same execution chain
 
-  cacheSource(id, data) {
-    this.sources[id] = data;
-    return this;
+  claimSource(id) {
+    const result = {};
+    this.sources[id] = new Promise((resolve, reject) => {
+      result.resolve = resolve;
+      result.reject = reject;
+    });
+    return result;
   }
@@ -65,2 +74,8 @@
     return this.clone({
+      path: this.path.concat({
+        op: 'subscript',
+        start: start,
+        end: end,
+        first: false
+      }),
       value: this.value.slice(start + collectionStart, end + collectionStart),
@@ -79,2 +94,8 @@ start: 0
     return this.clone({
+      path: this.path.concat({
+        op: 'subscript',
+        start: offset,
+        end: offset + 1,
+        first: true
+      }),
       value: this.value.slice(offset + collectionStart)[0],
@@ -122,2 +143,3 @@ start: 0
       value: this.value,
+      path: this.path,
       sourceId: this.sourceId,
@@ -133,2 +155,3 @@ refs: this.refs,
       sources: this.sources,
+      sourceId: this.sourceId,
       refs: this.refs
@@ -152,5 +175,6 @@ }, options || {}));
               parent: scope,
-              value: scope.value[operation.name]
+              value: scope.value[operation.name],
+              path: scope.path.concat(operation)
             });
-            (0, _debug2.default)(operation.name, '=>', scope.value);
+            (0, _debug2.default)(operation.name, '=>', scope.value, 'path:', scope.path);
             break;
@@ -175,2 +199,26 @@ default:
 
+  // Used in joins to find every possible value for this scope in every other document
+  // from the same source
+  allSiblings() {
+    var _this = this;
+
+    return _asyncToGenerator(function* () {
+      (0, _debug2.default)("sourceId", _this.sourceId, "sources", _this.sources);
+      const source = yield _this.dataForSource(_this.sourceId);
+      if (!source) {
+        return [];
+      }
+      return source.documents.map(function (item) {
+        const scope = _this.clone({
+          path: [],
+          value: item
+        });
+        (0, _debug2.default)('resolving', scope.value, _this.pathAsString());
+        return scope.resolveAccessor(_this.path);
+      }).filter(function (scope) {
+        return !!scope.value;
+      });
+    })();
+  }
+
   // Resolves an accessor for all possible values of that accessor. Used
@@ -180,58 +228,78 @@ // to expand join expressions so that i.e. ^.foo._ref resolves to all
   resolveAccessorForAll(path) {
-    (0, _debug2.default)('resolve accessor (all):', this.value, path);
-    let scope = this;
-    for (let i = 0; i < path.length; i++) {
-      const operation = path[i];
-      switch (operation.op) {
-        case 'parent':
-          scope = scope.parent;
-          (0, _debug2.default)('^ =>', scope.value);
-          break;
-        case 'attribute':
-          if (scope.sourceId) {
-            const source = scope.dataForSource(scope.sourceId).documents;
-            const subPath = path.slice(i);
-            const result = [];
-            source.forEach(item => {
-              const itemScope = scope.clone({
-                value: item
+    var _this2 = this;
+
+    return _asyncToGenerator(function* () {
+      (0, _debug2.default)('resolve accessor (all):', _this2.value, path);
+      let scope = _this2;
+      for (let i = 0; i < path.length; i++) {
+        const operation = path[i];
+        switch (operation.op) {
+          case 'parent':
+            scope = scope.parent;
+            (0, _debug2.default)('^ =>', scope.value);
+            break;
+          case 'attribute':
+            if (scope.sourceId) {
+              const source = (yield scope.dataForSource(scope.sourceId)).documents;
+              const subPath = path.slice(i);
+              const result = [];
+              source.forEach(function (item) {
+                const itemScope = scope.clone({
+                  value: item
+                });
+                const resolvedScope = itemScope.resolveAccessor(subPath);
+                if (resolvedScope.value !== null) {
+                  result.push(resolvedScope);
+                }
               });
-              const resolvedScope = itemScope.resolveAccessor(subPath);
-              if (resolvedScope.value !== null) {
-                result.push(resolvedScope);
-              }
-            });
-            return result;
-          }
+              return result;
+            }
 
-          if (scope.value === null || scope.value === undefined) {
-            return scope.clone({
-              value: null
+            if (scope.value === null || scope.value === undefined) {
+              return scope.clone({
+                value: null
+              });
+            }
+
+            scope = scope.child({
+              parent: scope,
+              value: scope.value[operation.name]
             });
-          }
-
-          scope = scope.child({
-            parent: scope,
-            value: scope.value[operation.name]
-          });
-          (0, _debug2.default)(operation.name, '=>', scope.value);
+            (0, _debug2.default)(operation.name, '=>', scope.value);
+            break;
+          default:
+            throw new Error(`Unkown accessor path element ${operation.op}`);
+        }
+        if (!scope.value) {
           break;
-        default:
-          throw new Error(`Unkown accessor path element ${operation.op}`);
+        }
       }
-      if (!scope.value) {
-        break;
+      // Unresolved attributes should return null, not undefined?
+      if (scope.value === undefined) {
+        scope = scope.clone({
+          value: null
+        });
       }
-    }
-    // Unresolved attributes should return null, not undefined?
-    if (scope.value === undefined) {
-      scope = scope.clone({
-        value: null
-      });
-    }
-    return [scope];
+      return [scope];
+    })();
   }
 
+  pathAsString() {
+    return this.path.map(element => {
+      switch (element.op) {
+        case 'attribute':
+          return element.name;
+        case 'subscript':
+          if (element.first) {
+            return `[${element.start}]`;
+          }
+          return `[${element.start}...${element.end}]`;
+        default:
+          throw new Error(`Unknown path op ${element.op}`);
+      }
+    }).join('.');
+  }
+
   inspect() {
-    return `Scope<${JSON.stringify(this.value)}>`;
+    return `Scope<${this.pathAsString()}: ${JSON.stringify(this.value)}>`;
   }
@@ -238,0 +306,0 @@

lib/plan/Composer.js

@@ -187,2 +187,3 @@ 'use strict';
       filter = new ops.Literal({
+        type: 'boolean',
         value: true
@@ -189,0 +190,0 @@ });

lib/plan/operations.js

@@ -205,2 +205,11 @@ 'use strict';
 
+// A map join takes a pipeline that resolves to an array of id's and fetches those documents returning them in the same order
+// as they resolve in the pipeline.
+class MapJoin {
+  constructor(pipeline) {
+    this.op = 'mapJoin';
+    this.pipeline = pipeline;
+  }
+}
+
 const defaultOperators = {
@@ -293,11 +302,22 @@ andOp(lhs, rhs) {
       (0, _debug2.default)('arrow operator lhs', lhs);
-      return new Pipeline({
-        operations: [new Source(), new Filter(new BinaryOperator('eq', new Accessor([new Parent(), ...lhs.path, new Attribute('_ref')]), new Accessor([new Attribute('_id')]))), new Subscript({
-          start: 0,
-          end: 1,
-          first: true
-        })],
-        alias: (0, _objectExpressionOps.determineUnambiguousAlias)(lhs)
-      });
-      return pipeline;
+      if (lhs.op == 'accessor') {
+        return new Pipeline({
+          operations: [new Source(), new Filter(new BinaryOperator('eq', new Accessor([new Parent(), ...lhs.path, new Attribute('_ref')]), new Accessor([new Attribute('_id')]))), new Subscript({
+            start: 0,
+            end: 1,
+            first: true
+          })],
+          alias: (0, _objectExpressionOps.determineUnambiguousAlias)(lhs)
+        });
+      } else if (lhs.op == 'pipe') {
+        // Rewrite ref[]->  to  mapJoin(ref[]._ref)
+        return new Pipeline({
+          alias: (0, _objectExpressionOps.determineUnambiguousAlias)(lhs),
+          operations: [new Source(), new MapJoin(new Pipeline({
+            operations: [...lhs.operations, new Accessor([new Attribute('_ref')])]
+          }))]
+        });
+      } else {
+        throw new Error(`Unable to apply arrow operator in this context`);
+      }
     default:
@@ -304,0 +324,0 @@ throw new Error(`Unknown postfix operator ${name}`);

notes/joins.md

@@ -0,1 +1,9 @@
+*[^.title == title]
+=>
+
+
+
+
+
+
 ^ = parent, source = parent siblings
@@ -47,2 +55,22 @@
 
-map(ref[]._ref, *[_id in ^^[].ref[]._ref]
+map(ref[]._ref, *[_id in ^^[].ref[]._ref]
+
+
+
+ref[]->name
+=>
+((*[_id in glob(ref[]._ref)]))
+
+
+ref[] | source(_id in glob(^.ref[]._ref)) | [_id == ^._ref]
+
+ref[] | *[^._ref == _id][0]
+
+
+ref[]->name
+
+ref[] | (*[_id in ^.[].ref[]._ref])) | [^._ref == _id][0].name
+
+(*|mapJoin(ref[]))
+
+

notes/todo.md

 * [ ] Refactor boxing. Make general container type? Need to keep track of start-index for collections even after unboxing
+* [ ] a[]->
+* [ ] Support inline join *[_id=="refs"][0].refs[]->{_id, _type}
 
+* [ ] Order by arbitrary expression
 * [ ] Exec pipelines in constraint expressions
@@ -7,3 +10,2 @@ * [ ] Joins
 * [ ] Implement correct array accessors a[].b[].c
-* [ ] a[]->
 * [ ] Optimizer rewrites x in 1...2 to (x >= 1 && x < 2)
@@ -19,1 +21,6 @@ * [ ] Fix string collation in relation to inequality operators
 
+Always just naïvely expand joins, then leave to optimizer to write efficient constraints?
+
+*{"a": *[_createdAt < ^._createdAt>]} => _createdAt < v1 || _createdAt < v2 || _createdAt < v3 ===> _createdAt < max(v1,v2,v3)
+
+ref[] | (*[_id == ^._ref][0]) => ref[] | ((*[_id in all(^._ref)]))

package.json

 {
   "name": "groq",
-  "version": "0.0.7",
+  "version": "0.0.8",
   "description": "A GROQ runtime",
@@ -5,0 +5,0 @@ "main": "lib/index.js",

src/exec/Executor.js

@@ -108,3 +108,4 @@ import Scope from './Scope'
       case 'source':
-        const data = scope.dataForSource(operation.id)
+        const data = await scope.dataForSource(operation.id)
+        debug("sourceId will be", operation.id)
         return scope.child({
@@ -149,2 +150,4 @@ sourceId: operation.id,
         return this.execMatch(operation, scope)
+      case 'mapJoin':
+        return this.execMapJoin(operation, scope)
       default:
@@ -361,6 +364,8 @@ throw new Error(`(exec) Unknown operation ${operation.op}`)
   async execPipeline(operation, scope) {
-    scope = await fetch(scope, operation, this.fetcher)
+    scope = await fetch(scope, operation, this.fetcher, this)
     const ops = operation.operations.slice()
     const first = ops.shift()
+    debug('first op', first)
     let current = await this.exec(first, scope)
+    debug('first op result:', current)
     while (ops.length > 0) {
@@ -450,3 +455,23 @@ debug('current:', current)
 
+  async execMapJoin(operation, scope) {
+    debug('execMapJoin', operation)
+    const source = asPlainValue(scope)
+    const parent = scope.parent
+    const mapValues = asPlainValue((await this.exec(operation.pipeline, parent)))
+    debug('execMapJoin mapValues', mapValues)
+    debug('execMapJoin source', source)
+    const joinResult = mapValues.map(id => scope.child({
+      value: source.find(item => item._id == id)
+    }))
+    debug('joinResult', joinResult)
+    return joinResult
+  }
+
   async execAccessor(operation, scope) {
+    debug('execAccessor()', operation.path, scope)
+    if (!scope) {
+      return new Scope({
+        value: null
+      })
+    }
     return scope.resolveAccessor(operation.path)
@@ -453,0 +478,0 @@ }

src/exec/fetch.js

@@ -1,5 +0,6 @@
-import { BinaryOperator } from '../plan/operations'
+import { BinaryOperator, Literal, Attribute, Accessor } from '../plan/operations'
 import Scope from './Scope'
 import debug from '../debug'
 import generalizeJoinFilter from './generalizeJoinFilter'
+import { asPlainValue } from './scopeTools'
 import { join } from 'path';
@@ -21,3 +22,3 @@
 
-export default async function fetch(scope, pipeline, fetcher) {
+export default async function fetch(scope, pipeline, fetcher, executor) {
   debug('fetch()', scope, pipeline)
@@ -31,3 +32,4 @@ const sourceId = extractSourceId(pipeline)
   // We need to actually go get it
-  const fetchSpec = await compile(scope, pipeline)
+  const responder = scope.claimSource(sourceId)
+  const fetchSpec = await compile(scope, pipeline, executor)
   const response = await fetcher(fetchSpec)
@@ -38,6 +40,7 @@ // Cache the result on the scope so it will bind to the source
   scope.cacheRefs(response.refs)
-  return scope.cacheSource(sourceId, {
+  responder.resolve({
     documents: response.results,
     start: response.start || 0
   })
+  return scope
 }
@@ -109,3 +112,3 @@
 // Given a pipeline, collapses it to a filter, a window and an ordering
-async function compile(scope, pipeline) {
+async function compile(scope, pipeline, executor) {
   const operations = pipeline.operations
@@ -122,3 +125,5 @@ if (operations.length == 0) {
   const fetchSpec = new FetchSpec(scope, sourceId)
-  operations.slice(1).reverse().forEach(operation => {
+  const ops = operations.slice(1)
+  while (ops.length > 0) {
+    const operation = ops.pop()
     debug('compiling', operation)
@@ -129,2 +134,5 @@ switch (operation.op) {
         break
+      case 'mapJoin':
+        const filter = await generalizeMapJoin(operation, scope, executor)
+        fetchSpec.applyFilter(filter)
       case 'object':
@@ -142,7 +150,7 @@ fetchSpec.project(operation.operations)
       default:
-        throw new Error(`Unknown pipeline operation ${operation.op} when collapsing fetche operation`)
+        throw new Error(`Unknown pipeline operation ${operation.op} when compiling fetch-operation`)
     }
-  })
+  }
 
-  const generalizedFilter = generalizeJoinFilter(fetchSpec.filter, scope)
+  const generalizedFilter = await generalizeJoinFilter(fetchSpec.filter, scope)
   const wasJoin = generalizedFilter !== fetchSpec.filter
@@ -159,3 +167,47 @@
   // to be fetched in order to be executed.
+  debug('fetchSpec => ', fetchSpec)
   return fetchSpec
+}
+
+// Given a map join, generates a filter that will fetch all possible source values for that
+// join.
+async function generalizeMapJoin(operation, scope, executor) {
+  if (!scope) {
+    debug("mapJoin over null short circuited")
+    fetchSpec.applyFilter(new Literal({
+      type: 'boolean',
+      value: false
+    }))
+  }
+  let ids
+  if (scope.sourceId) {
+    const source = await scope.dataForSource(scope.sourceId)
+    debug('expanding mapJoin', scope, scope.sourceId)
+    const idPromises = source.documents.map(async document => {
+      const scope = new Scope({
+        value: document
+      })
+      const resolved = asPlainValue(await executor.exec(operation.pipeline, scope)).filter(Boolean)
+      debug('resolved', resolved)
+      if (resolved) {
+        return resolved
+      }
+    })
+    const idMap = {}
+    ids = (await Promise.all(idPromises))
+      .forEach(idSet => idSet.forEach(id => {
+        idMap[id] = true
+      }))
+    ids = Object.keys(idMap)
+  } else {
+    // TODO
+    throw new Error(`mapJoin on literals not supported yet`)
+  }
+  debug('expanded mapJoin', ids)
+  return new BinaryOperator('in',
+    new Accessor([new Attribute('_id')]),
+    new Literal({
+      value: ids
+    })
+  )
 }

src/exec/generalizeJoinFilter.js

 import rewrite from '../plan/rewrite'
 import debug from '../debug'
 
-export default function generalizeJoinFilter(node, scope) {
+export default async function generalizeJoinFilter(node, scope) {
   debug('generalizeJoinFilter()', node, scope)
@@ -16,9 +16,11 @@ return rewrite(node, operation => {
 
-function generalizeEQ(operation, scope) {
+async function generalizeEQ(operation, scope) {
   const lhsIsJoin = isJoinAccessor(operation.lhs)
   const rhsIsJoin = isJoinAccessor(operation.rhs)
   if (!lhsIsJoin && !rhsIsJoin) {
+    debug("generalizeJoinFilter() (is not a join)")
     // Not part of the join
     return operation
   }
+
   if (lhsIsJoin && rhsIsJoin) {
@@ -36,3 +38,3 @@ throw new Error(`In a join, only the lhs or rhs can reference parent`)
   }
-  const joinScopes = scope.child({
+  const joinScopes = await scope.child({
     value: {}
@@ -39,0 +41,0 @@ }).resolveAccessorForAll(join.path)

src/exec/Scope.js

@@ -5,3 +5,3 @@ import debug from '../debug'
   constructor(options) {
-    const {value, parent, sourceId, sources, refs, start} = options
+    const {value, parent, sourceId, sources, refs, start, path} = options
     // A single source cahce shared between all scopes in the same
@@ -17,2 +17,4 @@ // execution chain.
     this.value = value
+    // The path of this specific value from source
+    this.path = path || []
     // A cache of all refs seen in all documents fetched so far, shared between all scopes in
@@ -27,5 +29,9 @@ // the same execution chain
 
-  cacheSource(id, data) {
-    this.sources[id] = data
-    return this
+  claimSource(id) {
+    const result = {}
+    this.sources[id] = new Promise(((resolve, reject) => {
+      result.resolve = resolve
+      result.reject = reject
+    }))
+    return result
   }
@@ -48,2 +54,8 @@
     return this.clone({
+      path: this.path.concat({
+        op: 'subscript',
+        start: start,
+        end: end,
+        first: false
+      }),
       value: this.value.slice(start + collectionStart, end + collectionStart),
@@ -62,2 +74,8 @@ start: 0
     return this.clone({
+      path: this.path.concat({
+        op: 'subscript',
+        start: offset,
+        end: offset + 1,
+        first: true
+      }),
       value: this.value.slice(offset + collectionStart)[0],
@@ -105,2 +123,3 @@ start: 0
       value: this.value,
+      path: this.path,
       sourceId: this.sourceId,
@@ -116,2 +135,3 @@ refs: this.refs,
       sources: this.sources,
+      sourceId: this.sourceId,
       refs: this.refs
@@ -135,5 +155,6 @@ }, options || {}))
               parent: scope,
-              value: scope.value[operation.name]
+              value: scope.value[operation.name],
+              path: scope.path.concat(operation)
             })
-            debug(operation.name, '=>', scope.value)
+            debug(operation.name, '=>', scope.value, 'path:', scope.path)
             break
@@ -149,4 +170,2 @@ default:
 
-
-
     // Unresolved attributes should return null, not undefined?
@@ -161,6 +180,24 @@ if (scope.value === undefined) {
 
+  // Used in joins to find every possible value for this scope in every other document
+  // from the same source
+  async allSiblings() {
+    debug("sourceId", this.sourceId, "sources", this.sources)
+    const source = await this.dataForSource(this.sourceId)
+    if (!source) {
+      return []
+    }
+    return source.documents.map(item => {
+      const scope = this.clone({
+        path: [],
+        value: item
+      })
+      debug('resolving', scope.value, this.pathAsString())
+      return scope.resolveAccessor(this.path)
+    }).filter(scope => !!scope.value)
+  }
+
   // Resolves an accessor for all possible values of that accessor. Used
   // to expand join expressions so that i.e. ^.foo._ref resolves to all
   // foo._refs of the parent expression. Always returns an array of scopes
-  resolveAccessorForAll(path) {
+  async resolveAccessorForAll(path) {
     debug('resolve accessor (all):', this.value, path)
@@ -177,3 +214,3 @@ let scope = this
           if (scope.sourceId) {
-            const source = scope.dataForSource(scope.sourceId).documents
+            const source = (await scope.dataForSource(scope.sourceId)).documents
             const subPath = path.slice(i)
@@ -221,4 +258,20 @@ const result = []
 
+  pathAsString() {
+    return this.path.map(element => {
+      switch (element.op) {
+        case 'attribute':
+          return element.name
+        case 'subscript':
+          if (element.first) {
+            return `[${element.start}]`
+          }
+          return `[${element.start}...${element.end}]`
+        default:
+          throw new Error(`Unknown path op ${element.op}`)
+      }
+    }).join('.')
+  }
+
   inspect() {
-    return `Scope<${JSON.stringify(this.value)}>`
+    return `Scope<${this.pathAsString()}: ${JSON.stringify(this.value)}>`
   }
@@ -225,0 +278,0 @@

src/plan/Composer.js

@@ -179,2 +179,3 @@ import { toObjectOperations } from './objectExpressionOps';
       filter = new ops.Literal({
+        type: 'boolean',
         value: true
@@ -181,0 +182,0 @@ })

src/plan/operations.js

@@ -187,2 +187,11 @@ import { toObjectOperations, determineUnambiguousAlias } from './objectExpressionOps'
 
+// A map join takes a pipeline that resolves to an array of id's and fetches those documents returning them in the same order
+// as they resolve in the pipeline.
+class MapJoin {
+  constructor(pipeline) {
+    this.op = 'mapJoin'
+    this.pipeline = pipeline
+  }
+}
+
 const defaultOperators = {
@@ -275,20 +284,39 @@ andOp(lhs, rhs) {
       debug('arrow operator lhs', lhs)
-      return new Pipeline({
-        operations: [
-          new Source(),
-          new Filter(
-            new BinaryOperator('eq',
-              new Accessor([new Parent(), ...(lhs.path), new Attribute('_ref')]),
-              new Accessor([new Attribute('_id')])
-            )
-          ),
-          new Subscript({
-            start: 0,
-            end: 1,
-            first: true
-          })
-        ],
-        alias: determineUnambiguousAlias(lhs)
-      })
-      return pipeline
+      if (lhs.op == 'accessor') {
+        return new Pipeline({
+          operations: [
+            new Source(),
+            new Filter(
+              new BinaryOperator('eq',
+                new Accessor([new Parent(), ...(lhs.path), new Attribute('_ref')]),
+                new Accessor([new Attribute('_id')])
+              )
+            ),
+            new Subscript({
+              start: 0,
+              end: 1,
+              first: true
+            })
+          ],
+          alias: determineUnambiguousAlias(lhs)
+        })
+      } else if (lhs.op == 'pipe') {
+        // Rewrite ref[]->  to  mapJoin(ref[]._ref)
+        return new Pipeline({
+          alias: determineUnambiguousAlias(lhs),
+          operations: [
+            new Source(),
+            new MapJoin(new Pipeline({
+              operations: [
+                ...(lhs.operations),
+                new Accessor([
+                  new Attribute('_ref')
+                ])
+              ]
+            }))
+          ]
+        })
+      } else {
+        throw new Error(`Unable to apply arrow operator in this context`)
+      }
     default:
@@ -295,0 +323,0 @@ throw new Error(`Unknown postfix operator ${name}`)

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

4028159

increased by0.6%

Number of package files

128

increased by0.79%

Lines of code

65827

increased by0.53%

No dependency changes