Implement LIKE operator

common/expression-cnf.js

@@ -1,28 +1,17 @@
 "use strict";
 
-function collectRanges(exp, ranges = new Map()) {
-  if (["AND", "OR", "NOT"].includes(exp[0]))
-    for (let i = 1; i < exp.length; ++i) collectRanges(exp[i], ranges);
-  else if ([">", ">=", "<", "<=", "=", "<>"].includes(exp[0]))
-    if (!Array.isArray(exp[2])) {
-      const key = `${JSON.stringify(exp[1])}:${typeof exp[2]}`;
-      let s = ranges.get(key);
-      if (!s) ranges.set(key, (s = new Set()));
-      s.add(exp[2]);
-    } else if (!Array.isArray(exp[1])) {
-      const key = `${JSON.stringify(exp[2])}:${typeof exp[1]}`;
-      let s = ranges.get(key);
-      if (!s) ranges.set(key, (s = new Set()));
-      s.add(exp[2]);
-    }
-
-  return ranges;
-}
+const expressionParser = require("./expression-parser");
 
 function or(cnf1, cnf2) {
   if (cnf1.length === 0) return cnf2;
   else if (cnf2.length === 0) return cnf1;
 
+  if (cnf1.length > cnf2.length) {
+    let c = cnf1;
+    cnf1 = cnf2;
+    cnf2 = c;
+  }
+
   const res = [];
   for (let i = 0; i < cnf1.length; ++i)
     for (let j = i; j < cnf2.length; ++j) res.push(cnf1[i].concat(cnf2[j]));
@@ -31,7 +20,39 @@ function or(cnf1, cnf2) {
 }
 
 function booleanCnf(exp) {
-  const ranges = collectRanges(exp);
+  const ranges = new Map();
+  const likePatterns = new Map();
+  expressionParser.map(exp, e => {
+    if (!Array.isArray(e)) return e;
+    if (e[0] === "LIKE" || e[0] === "NOT LIKE") {
+      if (
+        typeof e[2] === "string" &&
+        (e.length >= 4 || (e[3] != null && !Array.isArray(e[3])))
+      ) {
+        const key = JSON.stringify(e[1]);
+        let s = likePatterns.get(key);
+        if (!s) likePatterns.set(key, (s = new Map()));
+        s.set(
+          `${e[3] || ""}:${e[2]}`,
+          expressionParser.parseLikePattern(e[2], e[3])
+        );
+      }
+    } else if ([">", ">=", "<", "<=", "=", "<>"].includes(e[0])) {
+      if (!Array.isArray(e[2])) {
+        const key = `${JSON.stringify(e[1])}:${typeof e[2]}`;
+        let s = ranges.get(key);
+        if (!s) ranges.set(key, (s = new Set()));
+        s.add(e[2]);
+      } else if (!Array.isArray(e[1])) {
+        const key = `${JSON.stringify(e[2])}:${typeof e[1]}`;
+        let s = ranges.get(key);
+        if (!s) ranges.set(key, (s = new Set()));
+        s.add(e[2]);
+      }
+    }
+    return e;
+  });
+
   for (let [k, v] of ranges)
     if (k.endsWith(":number"))
       ranges.set(k, Array.from(v).sort((a, b) => a - b));
@@ -59,16 +80,24 @@ function booleanCnf(exp) {
     const vars = [];
 
     if (from == null) vars.push(`${lhs}:--${range[0]}`);
-    if (to == null) vars.push(`${lhs}:${range[range.length - 1]}--`);
-    if (fromInclusive) vars.push(`${lhs}:${from}`);
-    if (toInclusive) vars.push(`${lhs}:${to}`);
 
-    for (let i = 1; i < range.length; ++i)
-      if ((from == null || range[i] > from) && (to == null || range[i] <= to)) {
-        vars.push(`${lhs}:${range[i - 1]}`);
-        if (range[i] !== to) vars.push(`${lhs}:${range[i]}`);
+    for (let i = 0; i < range.length; ++i) {
+      if (
+        i > 0 &&
+        (from == null || range[i] > from) &&
+        (to == null || range[i] <= to)
+      )
         vars.push(`${lhs}:${range[i - 1]}--${range[i]}`);
-      }
+
+      if (
+        (range[i] > from && range[i] < to) ||
+        (range[i] === from && fromInclusive) ||
+        (range[i] === to && toInclusive)
+      )
+        vars.push(`${lhs}:${range[i]}`);
+    }
+
+    if (to == null) vars.push(`${lhs}:${range[range.length - 1]}--`);
 
     return vars.map(key => {
       let f = expressions.get(key);
@@ -82,6 +111,21 @@ function booleanCnf(exp) {
     });
   }
 
+  function getLikeVariables(lhs, pat, esc) {
+    const vars = [];
+    let pats = likePatterns.get(lhs);
+    pat = pats.get(`${esc || ""}:${pat}`);
+    for (let p of pats.values())
+      if (pat === p || likePatternIncludes(pat, p))
+        vars.push(`${lhs}:like:${JSON.stringify(p)}`);
+
+    return vars.map(key => {
+      let f = expressions.get(key);
+      if (!f) expressions.set(key, (f = expressions.size + 1));
+      return f;
+    });
+  }
+
   const comparisonOperators = {
     "=": v => [v, true, v, true],
     ">": v => [v, false, null, null],
@@ -151,6 +195,20 @@ function booleanCnf(exp) {
         if (negate) return [[0 - v]];
         else return [[v]];
       }
+    } else if (
+      (op === "LIKE" || op === "NOT LIKE") &&
+      !Array.isArray(clause[2]) &&
+      clause[2] != null &&
+      (clause.length >= 4 || clause[3] != null)
+    ) {
+      if (op === "NOT LIKE") negate = !negate;
+      const vars = getLikeVariables(
+        JSON.stringify(clause[1]),
+        clause[2],
+        clause[3]
+      );
+      if (negate) return vars.map(x => [0 - x]);
+      else return [vars];
     } else {
       const v = getVariable(JSON.stringify(clause), "");
       if (negate) return [[0 - v]];
@@ -169,4 +227,27 @@ function booleanCnf(exp) {
   return { vars: expressions.size, clauses: cnf };
 }
 
+function likePatternIncludes(pat1, pat2, idx1 = 0, idx2 = 0) {
+  while (idx1 < pat1.length && idx2 < pat2.length) {
+    if (pat1[idx1] === "\\%") {
+      for (let i = idx2; i <= pat2.length; ++i)
+        if (likePatternIncludes(pat1, pat2, idx1 + 1, i)) return true;
+      return false;
+    } else if (pat2[idx2] === "\\%") {
+      if (pat1[idx1] !== "\\%") return false;
+    } else if (pat1[idx1] === "\\_") {
+      // Ignore
+    } else if (pat2[idx2] === "\\_") {
+      return false;
+    } else if (pat2[idx2] !== pat1[idx1]) {
+      return false;
+    }
+    ++idx1;
+    ++idx2;
+  }
+
+  if (idx1 === pat1.length && idx2 === pat2.length) return true;
+  return false;
+}
+
 exports.booleanCnf = booleanCnf;

common/expression-parser.js

@@ -65,6 +65,21 @@ const lang = parsimmon.createLanguage({
       )
       .skip(parsimmon.optWhitespace);
   },
+  LikeOperator: function() {
+    return parsimmon
+      .alt(
+        parsimmon
+          .regexp(/like/i)
+          .result("LIKE")
+          .desc("LIKE"),
+        parsimmon
+          .regexp(/not\s+like/i)
+          .result("NOT LIKE")
+          .desc("NOT LIKE")
+      )
+      .notFollowedBy(parsimmon.regexp(/[a-zA-Z0-9_]/))
+      .skip(parsimmon.optWhitespace);
+  },
   IsNullOperator: function() {
     return parsimmon
       .alt(
@@ -213,12 +228,31 @@ const lang = parsimmon.createLanguage({
   },
   Comparison: function(r) {
     return parsimmon.alt(
-      parsimmon.seqMap(r.Parameter, r.IsNullOperator, (p, o) => [o, p]),
+      parsimmon.seqMap(r.ValueExpression, r.IsNullOperator, (p, o) => [o, p]),
       parsimmon.seqMap(
         r.ValueExpression,
         r.ComparisonOperator,
         r.ValueExpression,
         (p, o, v) => [o, p, v]
+      ),
+      parsimmon.seqMap(
+        r.ValueExpression,
+        r.LikeOperator,
+        r.ValueExpression.skip(
+          parsimmon
+            .regexp(/escape/i)
+            .result("ESCAPE")
+            .skip(parsimmon.whitespace)
+            .desc("ESCAPE")
+        ),
+        r.ValueExpression,
+        (a, b, c, d) => [b, a, c, d]
+      ),
+      parsimmon.seqMap(
+        r.ValueExpression,
+        r.LikeOperator,
+        r.ValueExpression,
+        (a, b, c) => [b, a, c]
       )
     );
   },
@@ -242,7 +276,7 @@ function parse(str) {
   return lang.Expression.tryParse(str);
 }
 
-function stringify(exp) {
+function stringify(exp, level = 0) {
   if (!Array.isArray(exp)) return JSON.stringify(exp);
 
   const opLevels = {
@@ -250,11 +284,15 @@ function stringify(exp) {
     AND: 11,
     NOT: 12,
     "=": 20,
-    "<>": 21,
-    ">": 22,
-    ">=": 23,
-    "<": 24,
-    "<=": 25,
+    "<>": 20,
+    ">": 20,
+    ">=": 20,
+    "<": 20,
+    "<=": 20,
+    LIKE: 20,
+    "NOT LIKE": 20,
+    "IS NULL": 20,
+    "IS NOT NULL": 20,
     "||": 30,
     "+": 31,
     "-": 31,
@@ -264,45 +302,79 @@ function stringify(exp) {
 
   const op = exp[0].toUpperCase();
 
+  function wrap(e) {
+    if (opLevels[op] <= level) return `(${e})`;
+    else return e;
+  }
+
   if (op === "FUNC") {
-    return `${exp[1]}(${exp
-      .slice(2)
-      .map(e => stringify(e))
-      .join(", ")})`;
+    return wrap(
+      `${exp[1]}(${exp
+        .slice(2)
+        .map(e => stringify(e))
+        .join(", ")})`
+    );
   } else if (op === "PARAM") {
-    if (typeof exp[1] === "string") return exp[1];
+    if (typeof exp[1] === "string") return wrap(exp[1]);
     else if (Array.isArray(exp[1]) && exp[1][0] === "||")
-      return exp[1]
-        .slice(1)
-        .map(p => {
-          if (typeof p === "string") return p;
-          else return `{${stringify(p)}}`;
-        })
-        .join("");
-    else return `{${stringify(exp[1])}}`;
+      return wrap(
+        exp[1]
+          .slice(1)
+          .map(p => {
+            if (typeof p === "string") return p;
+            else return `{${stringify(p)}}`;
+          })
+          .join("")
+      );
+    else return wrap(`{${stringify(exp[1])}}`);
   } else if (op === "IS NULL" || op === "IS NOT NULL") {
-    return `${stringify(exp[1])} ${op}`;
+    return wrap(`${stringify(exp[1], opLevels[op])} ${op}`);
+  } else if (op === "LIKE" || op === "NOT LIKE") {
+    if (exp[3])
+      return wrap(
+        `${stringify(exp[1], opLevels[op])} ${op} ${stringify(
+          exp[2],
+          opLevels[op]
+        )} ESCAPE ${stringify(exp[3], opLevels[op])}`
+      );
+    else
+      return wrap(
+        `${stringify(exp[1], opLevels[op])} ${op} ${stringify(
+          exp[2],
+          opLevels[op]
+        )}`
+      );
   } else if (op in opLevels) {
     const parts = exp.slice(1).map((e, i) => {
-      if (
-        Array.isArray(e) &&
-        (opLevels[e[0]] < opLevels[exp[0]] ||
-          (opLevels[e[0]] < opLevels[exp[0]] && i > 0) ||
-          (opLevels[exp[0]] >= 20 &&
-            opLevels[e[0]] >= 20 &&
-            opLevels[e[0]] < 30))
-      )
-        return `(${stringify(e)})`;
-      else return stringify(e);
+      return stringify(e, opLevels[exp[0]] + Math.min(i - 1, 0));
     });
 
-    if (op === "NOT") return `${op} ${parts[0]}`;
-    else return parts.join(` ${op} `);
+    if (op === "NOT") return wrap(`${op} ${parts[0]}`);
+    else return wrap(parts.join(` ${op} `));
   } else {
     throw new Error(`Unrecognized operator ${exp[0]}`);
   }
 }
 
+function parseLikePattern(pat, esc) {
+  let chars = pat.split("");
+
+  for (let i = 0; i < chars.length; ++i) {
+    let c = chars[i];
+    if (c === esc) {
+      chars[i] = chars[i + 1] || "";
+      chars[i + 1] = "";
+    } else if (c === "_") {
+      chars[i] = "\\_";
+    } else if (c === "%") {
+      chars[i] = "\\%";
+      while (chars[i + 1] === "%") chars[++i] = "";
+    }
+  }
+  return chars.filter(c => c);
+}
+
 exports.parse = parse;
 exports.stringify = stringify;
 exports.map = map;
+exports.parseLikePattern = parseLikePattern;

common/expression.js

@@ -6,6 +6,8 @@ const satSolver = require("./sat-solver");
 
 const isArray = Array.isArray;
 
+const regExpCache = new WeakMap();
+
 function* permute(arr) {
   if (arr.length <= 1) {
     for (let i = 0; i < arr[0]; ++i) yield [i];
@@ -35,6 +37,14 @@ function reduce(exp, callback) {
 }
 
 function evaluate(exp, obj, now) {
+  function getRegExp(pat, esc) {
+    const k = `${esc || ""}:${pat}`;
+    let c = regExpCache.get(exp);
+    if (!c) regExpCache.set(exp, (c = {}));
+    if (!c[k]) c[k] = likePatternToRegExp(pat, esc);
+    return c[k];
+  }
+
   return expressionParser.map(exp, e => {
     if (!isArray(e)) return e;
 
@@ -45,7 +55,7 @@ function evaluate(exp, obj, now) {
         if (e[2] == null) return null;
         if (!isArray(e[2])) return `${e[2]}`.toUpperCase();
       } else if (e[1] === "LOWER") {
-        if (e[12] == null) return null;
+        if (e[2] == null) return null;
         if (!isArray(e[2])) return `${e[2]}`.toLowerCase();
       }
     } else if (e[0] === "PARAM") {
@@ -77,6 +87,19 @@ function evaluate(exp, obj, now) {
       if (isArray(e[1])) return e;
       else if (e[1] != null) return true;
       else return null;
+    } else if (e[0] === "LIKE") {
+      if (isArray(e[1]) || isArray(e[2]) || isArray(e[3])) return e;
+      else if (e[1] == null || e[2] == null || (e.length >= 4 && e[3] == null))
+        return null;
+
+      const r = getRegExp(e[2], e[3]);
+      return r.test(e[1]);
+    } else if (e[0] === "NOT LIKE") {
+      if (isArray(e[1]) || isArray(e[2]) || isArray(e[3])) return e;
+      else if (e[1] == null || e[2] == null || (e.length >= 4 && e[3] == null))
+        return null;
+      const r = getRegExp(e[2], e[3]);
+      return !r.test(e[1]);
     } else if (e[0] === "=") {
       if (isArray(e[1]) || isArray(e[2])) return e;
       if (e[1] == null || e[2] == null) return null;
@@ -183,6 +206,36 @@ function subset(exp1, exp2) {
   return !satSolver.naiveDpll(clauses, vars);
 }
 
+function likePatternToRegExp(pat, esc = "", flags = "") {
+  const convChars = {
+    "-": "\\-",
+    "/": "\\/",
+    "\\": "\\/",
+    "^": "\\^",
+    $: "\\$",
+    "*": "\\*",
+    "+": "\\+",
+    "?": "\\?",
+    ".": "\\.",
+    "(": "\\(",
+    ")": "\\)",
+    "|": "\\|",
+    "[": "\\[",
+    "]": "\\]",
+    "{": "\\{",
+    "}": "\\}",
+    "\\%": ".*",
+    "\\_": "."
+  };
+  let chars = expressionParser.parseLikePattern(pat, esc);
+  if (!chars.length) return new RegExp("", flags);
+  chars = chars.map(c => convChars[c] || c);
+  chars[0] = chars[0] === ".*" ? "" : "^" + chars[0];
+  const l = chars.length - 1;
+  chars[l] = [".*", ""].includes(chars[l]) ? "" : chars[l] + "$";
+  return new RegExp(chars.join(""), flags);
+}
+
 exports.evaluate = evaluate;
 exports.and = and;
 exports.or = or;
@@ -190,3 +243,4 @@ exports.not = not;
 exports.subset = subset;
 exports.parse = expressionParser.parse;
 exports.stringify = expressionParser.stringify;
+exports.likePatternToRegExp = likePatternToRegExp;

server/mongodb-functions.js

@@ -1,6 +1,7 @@
 "use strict";
 const mongodb = require("mongodb");
 const expressionParser = require("../common/expression-parser");
+const expression = require("../common/expression");
 
 function processDeviceFilter(filter) {
   return expressionParser.map(filter, exp => {
@@ -130,6 +131,31 @@ function filterToMongoQuery(filter, negate = false, res = {}) {
   } else if (op === "IS NOT NULL") {
     const param = filter[1][1];
     res[param] = { $exists: !negate };
+  } else if (op === "LIKE" || op === "NOT LIKE") {
+    if (op === "NOT LIKE") negate = !negate;
+    let param;
+    let flags;
+    if (filter[1][0] === "FUNC" && filter[1][1] === "UPPER") {
+      if (filter[2] !== filter[2].toUpperCase())
+        throw new Error(
+          "Cannot compare UPPER() against non upper case pattern"
+        );
+      param = filter[1][2][1];
+      flags = "i";
+    } else if (filter[1][0] === "FUNC" && filter[1][1] === "LOWER") {
+      if (filter[2] !== filter[2].toLowerCase())
+        throw new Error(
+          "Cannot compare LOWER() against non lower case pattern"
+        );
+      param = filter[1][2][1];
+      flags = "i";
+    } else {
+      param = filter[1][1];
+      flags = "";
+    }
+    const r = expression.likePatternToRegExp(filter[2], filter[3], flags);
+    if (negate) res[param] = { $not: r };
+    else res[param] = r;
   } else {
     throw new Error(`Unrecognized operator ${op}`);
   }