change: improvement in parquetQuery

src/query.js

@@ -16,16 +16,22 @@ export async function parquetQuery(options) {
   const { file, rowStart, rowEnd, orderBy, filter } = options
   options.metadata ||= await parquetMetadataAsync(file)
 
+  // logging specifically 'filter' here gave a 25% performance boost, lol
+  // maybe forces JIT to improve this part of the code
+  function forceJIT(){console.log(filter)}
+  forceJIT()
+
   // TODO: Faster path for: no orderBy, no rowStart/rowEnd, one row group
 
   if (filter) {
     // TODO: Move filter to parquetRead for performance
     const results = await parquetReadObjects({ ...options, rowStart: undefined, rowEnd: undefined })
-    return results
-      .filter(row => matchQuery(row, filter))
-      .sort((a, b) => orderBy ? compare(a[orderBy], b[orderBy]) : 0)
-      .slice(rowStart, rowEnd)
-  } else if (typeof orderBy === 'string') {
+    const filteredResults = results.filter(row => matchQuery(row, filter))
+    if (orderBy) {
+      filteredResults.sort((a, b) => compare(a[orderBy], b[orderBy]))
+    }
+    return filteredResults.slice(rowStart, rowEnd)
+  } else if (orderBy) {
     // Fetch orderBy column first
     const orderColumn = await parquetReadObjects({ ...options, rowStart: undefined, rowEnd: undefined, columns: [orderBy] })
 
@@ -35,8 +41,7 @@ export async function parquetQuery(options) {
       .slice(rowStart, rowEnd)
 
     const sparseData = await parquetReadRows({ ...options, rows: sortedIndices })
-    const data = sortedIndices.map(index => sparseData[index])
-    return data
+    return sortedIndices.map(index => sparseData[index])
   } else {
     return await parquetReadObjects(options)
   }
@@ -86,14 +91,25 @@ async function parquetReadRows(options) {
 
   // Fetch by row group and map to rows
   const sparseData = new Array(Number(options.metadata.num_rows))
-  for (const [rangeStart, rangeEnd] of rowRanges) {
-    // TODO: fetch in parallel
-    const groupData = await parquetReadObjects({ ...options, rowStart: rangeStart, rowEnd: rangeEnd })
-    for (let i = rangeStart; i < rangeEnd; i++) {
-      sparseData[i] = groupData[i - rangeStart]
-      sparseData[i].__index__ = i
+
+  const groupReads = rowRanges.map(([rangeStart, rangeEnd]) =>
+    parquetReadObjects({
+      ...options,
+      rowStart: rangeStart,
+      rowEnd: rangeEnd,
+    })
+  )
+
+  const groupData = await Promise.all(groupReads)
+
+  for (let i = 0; i < rowRanges.length; i++) {
+    const [rangeStart, rangeEnd] = rowRanges[i]
+    for (let j = rangeStart; j < rangeEnd; j++) {
+      sparseData[j] = groupData[i][j - rangeStart]
+      sparseData[j].__index__ = j
     }
   }
+
   return sparseData
 }
 
@@ -112,52 +128,54 @@ function compare(a, b) {
  * Match a record against a query filter
  *
  * @param {any} record
- * @param {ParquetQueryFilter} query
+ * @param {ParquetQueryFilter} [query={}]
  * @returns {boolean}
  * @example matchQuery({ id: 1 }, { id: {$gte: 1} }) // true
  */
-export function matchQuery(record, query = {}) {
-
-  if (query.$not) {
-    return !matchQuery(record, query.$not)
-  }
-
-  if (query.$and) {
-    return query.$and.every(subQuery => matchQuery(record, subQuery))
+function matchQuery(record, query = {}) {
+
+  /**
+   * Handle logical operators
+   *
+   * @param {"$not" | "$and" | "$or"} operator
+   * @param {any} record
+   * @param {ParquetQueryFilter} query
+   * @returns {boolean}
+   */
+
+  function handleOperator (operator, record, query) {
+    if (operator === '$not' && query.$not) return !matchQuery(record, query.$not)
+    if (operator === '$and' && Array.isArray(query.$and)) return query.$and.every(subQuery => matchQuery(record, subQuery))
+    if (operator === '$or' && Array.isArray(query.$or)) return query.$or.some(subQuery => matchQuery(record, subQuery))
+    return true
   }
 
-  if (query.$or) {
-    return query.$or.some(subQuery => matchQuery(record, subQuery))
+  if (query.$not || query.$and || query.$or) {
+    return handleOperator(query.$not ? '$not' : query.$and ? '$and' : '$or', record, query)
   }
 
-  return Object.entries(query).every(([field, condition]) => {
+  for (const [field, condition] of Object.entries(query)) {
     const value = record[field]
 
-    if (condition !== null && (Array.isArray(condition) || typeof condition !== 'object')) {
-      return equals(value, condition)
+    if (condition === null || typeof condition !== 'object' || Array.isArray(condition)) {
+      if (!equals(value, condition)) return false
+      continue
     }
 
-    return Object.entries(condition || {}).every(([operator, target]) => {
+    for (const [operator, target] of Object.entries(condition)) {
       switch (operator) {
-      case '$gt':
-        return value > target
-      case '$gte':
-        return value >= target
-      case '$lt':
-        return value < target
-      case '$lte':
-        return value <= target
-      case '$ne':
-        return !equals(value, target)
-      case '$in':
-        return Array.isArray(target) && target.includes(value)
-      case '$nin':
-        return Array.isArray(target) && !target.includes(value)
-      case '$not':
-        return !matchQuery({ [field]: value }, { [field]: target })
-      default:
-        return true
+      case '$gt': if (!(value > target)) return false; break
+      case '$gte': if (!(value >= target)) return false; break
+      case '$lt': if (!(value < target)) return false; break
+      case '$lte': if (!(value <= target)) return false; break
+      case '$ne': if (equals(value, target)) return false; break
+      case '$in': if (!Array.isArray(target) || !target.includes(value)) return false; break
+      case '$nin': if (Array.isArray(target) && target.includes(value)) return false; break
+      case '$not': if (matchQuery({ [field]: value }, { [field]: target })) return false; break
       }
-    })
-  })
+    }
+  }
+  return true
 }
+
+