@uwdata/mosaic-core - npm Package Compare versions

Comparing version 0.1.0 to 0.2.0

package.json

 {
   "name": "@uwdata/mosaic-core",
-  "version": "0.1.0",
+  "version": "0.2.0",
   "description": "Scalable and extensible linked data views.",
@@ -31,7 +31,7 @@ "keywords": [
   "dependencies": {
-    "@duckdb/duckdb-wasm": "^1.20.0",
-    "@uwdata/mosaic-sql": "^0.1.0",
-    "apache-arrow": "^11.0.0"
+    "@duckdb/duckdb-wasm": "^1.25.0",
+    "@uwdata/mosaic-sql": "^0.2.0",
+    "apache-arrow": "^12.0.0"
   },
-  "gitHead": "a7967c35349bdf7f00abb113ce1dd9abb233cd62"
+  "gitHead": "e53cd914c807f99aabe78dcbe618dd9543e2f438"
 }

src/Catalog.js

@@ -52,4 +52,16 @@ import { jsType } from './util/js-type.js';
 
-    const result = await this.mc.query(summarize(colInfo, stats));
+    const result = await this.mc.query(
+      summarize(colInfo, stats),
+      { persist: true }
+    );
     const info = { ...colInfo, ...(Array.from(result)[0]) };
+
+    // coerce bigint to number
+    for (const key in info) {
+      const value = info[key];
+      if (typeof value === 'bigint') {
+        info[key] = Number(value);
+      }
+    }
+
     return info;
@@ -56,0 +68,0 @@ }

src/Coordinator.js

@@ -1,5 +0,5 @@
-import { socketClient } from './clients/socket.js';
+import { socketConnector } from './connectors/socket.js';
 import { Catalog } from './Catalog.js';
 import { FilterGroup } from './FilterGroup.js';
-import { QueryCache, voidCache } from './QueryCache.js';
+import { QueryManager, Priority } from './QueryManager.js';
 import { voidLogger } from './util/void-logger.js';
@@ -19,18 +19,22 @@
 export class Coordinator {
-  constructor(db = socketClient(), options = {}) {
+  constructor(db = socketConnector(), options = {}) {
     this.catalog = new Catalog(this);
+    this.manager = options.manager || QueryManager();
     this.logger(options.logger || console);
     this.configure(options);
-    this.databaseClient(db);
+    this.databaseConnector(db);
     this.clear();
+    this._recorders = [];
   }
 
   logger(logger) {
-    return arguments.length
-      ? (this._logger = logger || voidLogger())
-      : this._logger;
+    if (arguments.length) {
+      this._logger = logger || voidLogger();
+      this.manager.logger(this._logger);
+    }
+    return this._logger;
   }
 
   configure({ cache = true, indexes = true }) {
-    this.cache = cache ? new QueryCache() : voidCache();
+    this.manager.cache(cache);
     this.indexes = indexes;
@@ -46,54 +50,54 @@ }
     }
-    if (cache) this.cache.clear();
+    if (cache) this.manager.cache().clear();
     if (catalog) this.catalog.clear();
   }
 
-  databaseClient(db) {
-    if (arguments.length > 0) {
-      this.db = db;
-    }
-    return this.db;
+  databaseConnector(db) {
+    return this.manager.connector(db);
   }
 
-  async exec(sql) {
-    try {
-      await this.db.query({ type: 'exec', sql });
-    } catch (err) {
-      this._logger.error(err);
-    }
+  // -- Query Management ----
+
+  cancel(requests) {
+    this.manager.cancel(requests);
   }
 
-  query(query, { type = 'arrow', cache = true } = {}) {
-    const sql = String(query);
-    const t0 = performance.now();
-    const cached = this.cache.get(sql);
-    if (cached) {
-      this._logger.debug('Cache');
-      return cached;
-    } else {
-      const request = this.db.query({ type, sql });
-      const result = cache ? this.cache.set(sql, request) : request;
-      result.then(() => this._logger.debug(`Query: ${performance.now() - t0}`));
-      return result;
-    }
+  exec(query, { priority = Priority.Normal } = {}) {
+    return this.manager.request({ type: 'exec', query }, priority);
   }
 
-  async updateClient(client, query) {
-    let result;
-    try {
-      client.queryPending();
-      result = await this.query(query);
-    } catch (err) {
-      this._logger.error(err);
-      client.queryError(err);
-      return;
-    }
-    try {
-      client.queryResult(result).update();
-    } catch (err) {
-      this._logger.error(err);
-    }
+  query(query, {
+    type = 'arrow',
+    cache = true,
+    priority = Priority.Normal,
+    ...options
+  } = {}) {
+    return this.manager.request({ type, query, cache, options }, priority);
   }
 
-  async requestQuery(client, query) {
+  prefetch(query, options = {}) {
+    return this.query(query, { ...options, cache: true, priority: Priority.Low });
+  }
+
+  createBundle(name, queries, priority = Priority.Low) {
+    const options = { name, queries };
+    return this.manager.request({ type: 'create-bundle', options }, priority);
+  }
+
+  loadBundle(name, priority = Priority.High) {
+    const options = { name };
+    return this.manager.request({ type: 'load-bundle', options }, priority);
+  }
+
+  // -- Client Management ----
+
+  updateClient(client, query, priority = Priority.Normal) {
+    client.queryPending();
+    return this.query(query, { priority }).then(
+      data => client.queryResult(data).update(),
+      err => { client.queryError(err); this._logger.error(err); }
+    );
+  }
+
+  requestQuery(client, query) {
     this.filterGroups.get(client.filterBy)?.reset();
@@ -116,3 +120,3 @@ return query
     if (fields?.length) {
-      client.fieldStats(await catalog.queryFields(fields));
+      client.fieldInfo(await catalog.queryFields(fields));
     }
@@ -119,0 +123,0 @@

src/DataTileIndexer.js

@@ -1,4 +0,3 @@
-import { Query, expr, and, isBetween, asColumn, epoch_ms } from '@uwdata/mosaic-sql';
+import { Query, and, asColumn, epoch_ms, isBetween, sql } from '@uwdata/mosaic-sql';
 import { fnv_hash } from './util/hash.js';
-import { skipClient } from './util/skip-client.js';
 
@@ -33,11 +32,18 @@ const identity = x => x;
 
+  clear() {
+    if (this.indices) {
+      this.mc.cancel(Array.from(this.indices.values(), index => index.result));
+      this.indices = null;
+    }
+  }
+
   index(clients, active) {
     if (this.clients !== clients) {
       // test client views for compatibility
-      const cols = Array.from(clients).map(getIndexColumns);
+      const cols = Array.from(clients, getIndexColumns);
       const from = cols[0]?.from;
       this.enabled = cols.every(c => c && c.from === from);
       this.clients = clients;
-      this.indices = null;
       this.activeView = null;
+      this.clear();
     }
@@ -48,4 +54,6 @@ if (!this.enabled) return false; // client views are not indexable
     const { source } = active;
+    if (source && source === this.activeView?.source) return true; // we're good!
+
+    this.clear();
     if (!source) return false; // nothing to work with
-    if (source === this.activeView?.source) return true; // we're good!
     const activeView = this.activeView = getActiveView(active);
@@ -56,10 +64,9 @@ if (!activeView) return false; // active selection clause not compatible
 
-    // create a selection with the active client removed
-    const sel = this.selection.clone().update({ source });
+    // create a selection with the active source removed
+    const sel = this.selection.remove(source);
 
     // generate data tile indices
     const indices = this.indices = new Map;
-    const promises = [];
     for (const client of clients) {
-      if (sel.cross && skipClient(client, active)) continue;
+      if (sel.skip(client, active)) continue;
       const index = getIndexColumns(client);
@@ -82,7 +89,5 @@
       const table = `tile_index_${id}`;
-      indices.set(client, { table, ...index });
-      promises.push(createIndex(this.mc, table, sql));
+      const result = createIndex(this.mc, table, sql);
+      indices.set(client, { table, result, ...index });
     }
-
-    return promises;
   }
@@ -107,8 +112,8 @@
     const { table, dims, aggr } = index;
-    return this.mc.updateClient(client, Query
+    const query = Query
       .select(dims, aggr)
       .from(table)
       .groupby(dims)
-      .where(filter)
-    );
+      .where(filter);
+    return this.mc.updateClient(client, query);
   }
@@ -121,18 +126,18 @@ }
   if (!schema || !columns) return null;
-  const { type, scales } = schema;
+  const { type, scales, pixelSize = 1 } = schema;
   let predicate;
 
   if (type === 'interval' && scales) {
-    const bins = scales.map(s => binInterval(s));
+    const bins = scales.map(s => binInterval(s, pixelSize));
     if (bins.some(b => b == null)) return null; // unsupported scale type
 
     if (bins.length === 1) {
-      predicate = p => p ? isBetween('active0', p.value.map(bins[0])) : [];
-      columns = { active0: bins[0](clause.predicate.expr) };
+      predicate = p => p ? isBetween('active0', p.range.map(bins[0])) : [];
+      columns = { active0: bins[0](clause.predicate.field) };
     } else {
       predicate = p => p
-        ? and(p.value.map(({ value }, i) => isBetween(`active${i}`, value.map(bins[i]))))
+        ? and(p.children.map(({ range }, i) => isBetween(`active${i}`, range.map(bins[i]))))
         : [];
       columns = Object.fromEntries(
-        clause.predicate.value.map((p, i) => [`active${i}`, bins[i](p.expr)])
+        clause.predicate.children.map((p, i) => [`active${i}`, bins[i](p.field)])
       );
@@ -150,5 +155,5 @@ }
 
-function binInterval(scale) {
+function binInterval(scale, pixelSize) {
   const { type, domain, range } = scale;
-  let lift, sql;
+  let lift, toSql;
 
@@ -158,7 +163,7 @@ switch (type) {
       lift = identity;
-      sql = asColumn;
+      toSql = asColumn;
       break;
     case 'log':
       lift = Math.log;
-      sql = c => `LN(${asColumn(c)})`;
+      toSql = c => sql`LN(${asColumn(c)})`;
       break;
@@ -168,7 +173,7 @@ case 'symlog':
       lift = x => Math.sign(x) * Math.log1p(Math.abs(x));
-      sql = c => (c = asColumn(c), `SIGN(${c}) * LN(1 + ABS(${c}))`);
+      toSql = c => (c = asColumn(c), sql`SIGN(${c}) * LN(1 + ABS(${c}))`);
       break;
     case 'sqrt':
       lift = Math.sqrt;
-      sql = c => `SQRT(${asColumn(c)})`;
+      toSql = c => sql`SQRT(${asColumn(c)})`;
       break;
@@ -178,24 +183,18 @@ case 'utc':
       lift = x => +x;
-      sql = c => c instanceof Date ? +c : epoch_ms(asColumn(c));
+      toSql = c => c instanceof Date ? +c : epoch_ms(asColumn(c));
       break;
   }
-  return lift ? binFunction(domain, range, lift, sql) : null;
+  return lift ? binFunction(domain, range, pixelSize, lift, toSql) : null;
 }
 
-function binFunction(domain, range, lift, sql) {
+function binFunction(domain, range, pixelSize, lift, toSql) {
   const lo = lift(Math.min(domain[0], domain[1]));
   const hi = lift(Math.max(domain[0], domain[1]));
-  const a = Math.abs(lift(range[1]) - lift(range[0])) / (hi - lo);
-  return value => expr(
-    `FLOOR(${a}::DOUBLE * (${sql(value)} - ${lo}::DOUBLE))`,
-    asColumn(value).columns
-  );
+  const a = (Math.abs(lift(range[1]) - lift(range[0])) / (hi - lo)) / pixelSize;
+  const s = pixelSize === 1 ? '' : `${pixelSize}::INTEGER * `;
+  return value => sql`${s}FLOOR(${a}::DOUBLE * (${toSql(value)} - ${lo}::DOUBLE))::INTEGER`;
 }
 
-async function createIndex(mc, table, query) {
-  try {
-    await mc.exec(`CREATE TEMP TABLE IF NOT EXISTS ${table} AS ${query}`);
-  } catch (err) {
-    mc.logger().error(err);
-  }
+function createIndex(mc, table, query) {
+  return mc.exec(`CREATE TEMP TABLE IF NOT EXISTS ${table} AS ${query}`);
 }
@@ -212,4 +211,4 @@
 
-  let aggr = [];
-  let dims = [];
+  const aggr = [];
+  const dims = [];
   let count;
@@ -221,13 +220,13 @@
       case 'SUM':
-        aggr.push({ [as]: expr(`SUM("${as}")::DOUBLE`) });
+        aggr.push({ [as]: sql`SUM("${as}")::DOUBLE` });
         break;
       case 'AVG':
         count = '_count_';
-        aggr.push({ [as]: expr(`(SUM("${as}" * ${count}) / SUM(${count}))::DOUBLE`) });
+        aggr.push({ [as]: sql`(SUM("${as}" * ${count}) / SUM(${count}))::DOUBLE` });
         break;
       case 'MAX':
-        aggr.push({ [as]: expr(`MAX("${as}")`) });
+        aggr.push({ [as]: sql`MAX("${as}")` });
         break;
       case 'MIN':
-        aggr.push({ [as]: expr(`MIN("${as}")`) });
+        aggr.push({ [as]: sql`MIN("${as}")` });
         break;
@@ -243,3 +242,3 @@ default:
     dims,
-    count: count ? { [count]: expr('COUNT(*)') } : {},
+    count: count ? { [count]: sql`COUNT(*)` } : {},
     from
@@ -260,3 +259,3 @@ };
   // handle set operations / subqueries
-  let base = getBaseTable(subq[0]);
+  const base = getBaseTable(subq[0]);
   for (let i = 1; i < subq.length; ++i) {
@@ -263,0 +262,0 @@ const from = getBaseTable(subq[i]);

src/FilterGroup.js

 import { DataTileIndexer } from './DataTileIndexer.js';
-import { throttle } from './util/throttle.js';
 
@@ -12,6 +11,4 @@ export class FilterGroup {
     const { value, activate } = this.handlers = {
-      value: throttle(() => this.update()),
-      activate: clause => {
-        this.indexer?.index(this.clients, clause);
-      }
+      value: () => this.update(),
+      activate: clause => this.indexer?.index(this.clients, clause)
     };
@@ -44,3 +41,3 @@ selection.addEventListener('value', value);
 
-  async update() {
+  update() {
     const { mc, indexer, clients, selection } = this;
@@ -47,0 +44,0 @@ return indexer?.index(clients)

src/index.js

@@ -5,7 +5,10 @@ export { MosaicClient } from './MosaicClient.js';
 export { Param, isParam } from './Param.js';
+export { Priority } from './QueryManager.js';
+
+export { restConnector } from './connectors/rest.js';
+export { socketConnector } from './connectors/socket.js';
+export { wasmConnector } from './connectors/wasm.js';
+
 export { distinct } from './util/distinct.js';
-export { sqlFrom } from './util/sql-from.js';
+export { synchronizer } from './util/synchronizer.js';
 export { throttle } from './util/throttle.js';
-export { restClient } from './clients/rest.js';
-export { socketClient } from './clients/socket.js';
-export { wasmClient } from './clients/wasm.js';

src/Param.js

@@ -0,3 +1,9 @@
+import { AsyncDispatch } from './util/AsyncDispatch.js';
 import { distinct } from './util/distinct.js';
 
+/**
+ * Test if a value is a Param instance.
+ * @param {*} x The value to test.
+ * @returns {boolean} True if the input is a Param, false otherwise.
+ */
 export function isParam(x) {
@@ -7,8 +13,22 @@ return x instanceof Param;
 
-export class Param {
+/**
+ * Represents a dynamic parameter that dispatches updates
+ * upon parameter changes.
+ */
+export class Param extends AsyncDispatch {
+
+  /**
+   * Create a new Param instance.
+   * @param {*} value The initial value of the Param.
+   */
   constructor(value) {
+    super();
     this._value = value;
-    this._listeners = new Map;
   }
 
+  /**
+   * Create a new Param instance with the given initial value.
+   * @param {*} value The initial value of the Param.
+   * @returns {Param} The new Param instance.
+   */
   static value(value) {
@@ -18,2 +38,22 @@ return new Param(value);
 
+  /**
+   * Create a new Param instance over an array of initial values,
+   * which may contain nested Params.
+   * @param {*} values The initial values of the Param.
+   * @returns {Param} The new Param instance.
+   */
+  static array(values) {
+    if (values.some(v => isParam(v))) {
+      const p = new Param();
+      const update = () => p.update(values.map(v => isParam(v) ? v.value : v));
+      update();
+      values.forEach(v => isParam(v) ? v.addEventListener('value', update) : 0);
+      return p;
+    }
+    return new Param(values);
+  }
+
+  /**
+   * The current value of the Param.
+   */
   get value() {
@@ -23,27 +63,33 @@ return this._value;
 
+  /**
+   * Update the Param value
+   * @param {*} value The new value of the Param.
+   * @param {object} [options] The update options.
+   * @param {boolean} [options.force] A boolean flag indicating if the Param
+   *  should emit a 'value' event even if the internal value is unchanged.
+   * @returns {this} This Param instance.
+   */
   update(value, { force } = {}) {
-    const changed = distinct(this._value, value);
-    if (changed) this._value = value;
-    if (changed || force) this.emit('value', this.value);
+    const shouldEmit = distinct(this._value, value) || force;
+    if (shouldEmit) {
+      this.emit('value', value);
+    } else {
+      this.cancel('value');
+    }
     return this;
   }
 
-  addEventListener(type, callback) {
-    let list = this._listeners.get(type) || [];
-    if (!list.includes(callback)) {
-      list = list.concat(callback);
+  /**
+   * Upon value-typed updates, sets the current value to the input value
+   * immediately prior to the event value being emitted to listeners.
+   * @param {string} type The event type.
+   * @param {*} value The input event value.
+   * @returns {*} The input event value.
+   */
+  willEmit(type, value) {
+    if (type === 'value') {
+      this._value = value;
     }
-    this._listeners.set(type, list);
+    return value;
   }
-
-  removeEventListener(type, callback) {
-    const list = this._listeners.get(type);
-    if (list?.length) {
-      this._listeners.set(type, list.filter(x => x !== callback));
-    }
-  }
-
-  emit(type, event) {
-    this._listeners.get(type)?.forEach(l => l(event));
-  }
 }

src/Selection.js

 import { or } from '@uwdata/mosaic-sql';
 import { Param } from './Param.js';
-import { skipClient } from './util/skip-client.js';
 
+/**
+ * Test if a value is a Selection instance.
+ * @param {*} x The value to test.
+ * @returns {boolean} True if the input is a Selection, false otherwise.
+ */
 export function isSelection(x) {
@@ -9,43 +13,108 @@ return x instanceof Selection;
 
+/**
+ * Represents a dynamic set of query filter predicates.
+ */
 export class Selection extends Param {
 
-  static intersect() {
-    return new Selection();
+  /**
+   * Create a new Selection instance with an
+   * intersect (conjunction) resolution strategy.
+   * @param {object} [options] The selection options.
+   * @param {boolean} [options.cross=false] Boolean flag indicating
+   *  cross-filtered resolution. If true, selection clauses will not
+   *  be applied to the clients they are associated with.
+   * @returns {Selection} The new Selection instance.
+   */
+  static intersect({ cross = false } = {}) {
+    return new Selection(new SelectionResolver({ cross }));
   }
 
-  static crossfilter() {
-    return new Selection({ cross: true });
+  /**
+   * Create a new Selection instance with a
+   * union (disjunction) resolution strategy.
+   * @param {object} [options] The selection options.
+   * @param {boolean} [options.cross=false] Boolean flag indicating
+   *  cross-filtered resolution. If true, selection clauses will not
+   *  be applied to the clients they are associated with.
+   * @returns {Selection} The new Selection instance.
+   */
+  static union({ cross = false } = {}) {
+    return new Selection(new SelectionResolver({ cross, union: true }));
   }
 
-  static union() {
-    return new Selection({ union: true });
+  /**
+   * Create a new Selection instance with a singular resolution strategy
+   * that keeps only the most recent selection clause.
+   * @param {object} [options] The selection options.
+   * @param {boolean} [options.cross=false] Boolean flag indicating
+   *  cross-filtered resolution. If true, selection clauses will not
+   *  be applied to the clients they are associated with.
+   * @returns {Selection} The new Selection instance.
+   */
+  static single({ cross = false } = {}) {
+    return new Selection(new SelectionResolver({ cross, single: true }));
   }
 
-  static single() {
-    return new Selection({ single: true });
+  /**
+   * Create a new Selection instance with a
+   * cross-filtered intersect resolution strategy.
+   * @returns {Selection} The new Selection instance.
+   */
+  static crossfilter() {
+    return new Selection(new SelectionResolver({ cross: true }));
   }
 
-  constructor({ union, cross, single } = {}) {
+  /**
+   * Create a new Selection instance.
+   * @param {SelectionResolver} resolver The selection resolution
+   *  strategy to apply.
+   */
+  constructor(resolver = new SelectionResolver()) {
     super([]);
-    this.active = null;
-    this.union = !!union;
-    this.cross = !!cross;
-    this.single = !!single;
+    this._resolved = this._value;
+    this._resolver = resolver;
   }
 
+  /**
+   * Create a cloned copy of this Selection instance.
+   * @returns {this} A clone of this selection.
+   */
   clone() {
-    const s = new Selection();
-    s.active = this.active;
-    s.union = this.union;
-    s.cross = this.cross;
-    s._value = this._value;
+    const s = new Selection(this._resolver);
+    s._value = s._resolved = this._value;
     return s;
   }
 
+  /**
+   * Create a clone of this Selection with clauses corresponding
+   * to provided source removed.
+   * @param {*} source The clause source to remove.
+   * @returns {this} A cloned and updated Selection.
+   */
+  remove(source) {
+    const s = this.clone();
+    s._value = s._resolved = s._resolver.resolve(this._resolved, { source });
+    s._value.active = { source };
+    return s;
+  }
+
+  /**
+   * The current active (most recently updated) selection clause.
+   */
+  get active() {
+    return this.clauses.active;
+  }
+
+  /**
+   * The value corresponding to the current active selection clause.
+   * This method ensures compatibility where a normal Param is expected.
+   */
   get value() {
-    // return value of most recently added clause
-    const { clauses } = this;
-    return clauses[clauses.length - 1]?.value;
+    // return value of the active clause
+    return this.active?.value;
   }
 
+  /**
+   * The current array of selection clauses.
+   */
   get clauses() {
@@ -55,2 +124,6 @@ return super.value;
 
+  /**
+   * Emit an activate event with the given selection clause.
+   * @param {*} clause The clause repesenting the potential activation.
+   */
   activate(clause) {
@@ -60,26 +133,149 @@ this.emit('activate', clause);
 
+  /**
+   * Update the selection with a new selection clause.
+   * @param {*} clause The selection clause to add.
+   * @returns {this} This Selection instance.
+   */
   update(clause) {
+    // we maintain an up-to-date list of all resolved clauses
+    // this ensures consistent clause state across unemitted event values
+    this._resolved = this._resolver.resolve(this._resolved, clause, true);
+    this._resolved.active = clause;
+    return super.update(this._resolved);
+  }
+
+  /**
+   * Upon value-typed updates, sets the current clause list to the
+   * input value and returns the active clause value.
+   * @param {string} type The event type.
+   * @param {*} value The input event value.
+   * @returns {*} For value-typed events, returns the active clause
+   *  values. Otherwise returns the input event value as-is.
+   */
+  willEmit(type, value) {
+    if (type === 'value') {
+      this._value = value;
+      return this.value;
+    }
+    return value;
+  }
+
+  /**
+   * Upon value-typed updates, returns a dispatch queue filter function.
+   * The return value depends on the selection resolution strategy.
+   * @param {string} type The event type.
+   * @param {*} value The input event value.
+   * @returns {*} For value-typed events, returns a dispatch queue filter
+   *  function. Otherwise returns null.
+   */
+  emitQueueFilter(type, value) {
+    return type === 'value'
+      ? this._resolver.queueFilter(value)
+      : null;
+  }
+
+  /**
+   * Indicates if a selection clause should not be applied to a given client.
+   * The return value depends on the selection resolution strategy.
+   * @param {*} client The selection clause.
+   * @param {*} clause The client to test.
+   * @returns True if the client should be skipped, false otherwise.
+   */
+  skip(client, clause) {
+    return this._resolver.skip(client, clause);
+  }
+
+  /**
+   * Return a selection query predicate for the given client.
+   * @param {*} client The client whose data may be filtered.
+   * @returns {*} The query predicate for filtering client data,
+   *  based on the current state of this selection.
+   */
+  predicate(client) {
+    const { clauses } = this;
+    return this._resolver.predicate(clauses, clauses.active, client);
+  }
+}
+
+/**
+ * Implements selection clause resolution strategies.
+ */
+export class SelectionResolver {
+
+  /**
+   * Create a new selection resolved instance.
+   * @param {object} [options] The resolution strategy options.
+   * @param {boolean} [options.union=false] Boolean flag to indicate a union strategy.
+   *  If false, an intersection strategy is used.
+   * @param {boolean} [options.cross=false] Boolean flag to indicate cross-filtering.
+   * @param {boolean} [options.single=false] Boolean flag to indicate single clauses only.
+   */
+  constructor({ union, cross, single } = {}) {
+    this.union = !!union;
+    this.cross = !!cross;
+    this.single = !!single;
+  }
+
+  /**
+   * Resolve a list of selection clauses according to the resolution strategy.
+   * @param {*[]} clauseList An array of selection clauses.
+   * @param {*} clause A new selection clause to add.
+   * @returns {*[]} An updated array of selection clauses.
+   */
+  resolve(clauseList, clause, reset = false) {
     const { source, predicate } = clause;
-    this.active = clause;
-    const clauses = this.single ? [] : this.clauses.filter(c => source !== c.source);
+    const filtered = clauseList.filter(c => source !== c.source);
+    const clauses = this.single ? [] : filtered;
+    if (this.single && reset) filtered.forEach(c => c.source?.reset?.());
     if (predicate) clauses.push(clause);
-    return super.update(clauses);
+    return clauses;
   }
 
-  predicate(client) {
-    const { active, clauses, cross, union } = this;
+  /**
+   * Indicates if a selection clause should not be applied to a given client.
+   * The return value depends on the resolution strategy.
+   * @param {*} client The selection clause.
+   * @param {*} clause The client to test.
+   * @returns True if the client should be skipped, false otherwise.
+   */
+  skip(client, clause) {
+    return this.cross && clause?.clients?.has(client);
+  }
 
+  /**
+   * Return a selection query predicate for the given client.
+   * @param {*[]} clauseList An array of selection clauses.
+   * @param {*} active The current active selection clause.
+   * @param {*} client The client whose data may be filtered.
+   * @returns {*} The query predicate for filtering client data,
+   *  based on the current state of this selection.
+   */
+  predicate(clauseList, active, client) {
+    const { union } = this;
+
     // do nothing if cross-filtering and client is currently active
-    if (cross && skipClient(client, active)) return undefined;
+    if (this.skip(client, active)) return undefined;
 
     // remove client-specific predicates if cross-filtering
-    const list = (cross
-      ? clauses.filter(clause => !skipClient(client, clause))
-      : clauses
-    ).map(s => s.predicate);
+    const predicates = clauseList
+      .filter(clause => !this.skip(client, clause))
+      .map(clause => clause.predicate);
 
     // return appropriate conjunction or disjunction
     // an array of predicates is implicitly conjunctive
-    return union && list.length > 1 ? or(list) : list;
+    return union && predicates.length > 1 ? or(predicates) : predicates;
   }
+
+  /**
+   * Returns a filter function for queued selection updates.
+   * @param {*} value The new event value that will be enqueued.
+   * @returns {(value: *) => boolean|null} A dispatch queue filter
+   *  function, or null if all unemitted event values should be filtered.
+   */
+  queueFilter(value) {
+    if (this.cross) {
+      const source = value.active?.source;
+      return clauses => clauses.active?.source !== source;
+    }
+  }
 }

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

1341023

increased by6.27%

Number of package files

27

increased by8%

Lines of code

28659

increased by7.98%

Dependency changes

• + Added@types/node@18.14.5(transitive)

• + Added@uwdata/mosaic-sql@0.2.0(transitive)

• + Addedapache-arrow@12.0.1(transitive)

• + Addedflatbuffers@23.3.3(transitive)

• - Removed@types/flatbuffers@1.10.3(transitive)

• - Removed@types/node@18.7.23(transitive)

• - Removed@uwdata/mosaic-sql@0.1.0(transitive)

• - Removedapache-arrow@11.0.0(transitive)

• - Removedflatbuffers@2.0.4(transitive)

• Updated@duckdb/duckdb-wasm@^1.25.0

• Updated@uwdata/mosaic-sql@^0.2.0

• Updatedapache-arrow@^12.0.0