(WiP) Remediating issues seen when profiling filters.

plugin/virtualized-table-plugin/src/main/scala/org/finos/vuu/plugin/virtualized/table/VirtualizedTableKeys.scala

@@ -36,7 +36,5 @@ case class VirtualizedTableKeys(window: MovingWindow[String], dataSize: Int) ext
   }
 
   override def iterator: Iterator[String] = window.iterator
-
-  override def intersect(keys: Iterable[String]): TablePrimaryKeys = ???
 
 }

toolbox/src/main/scala/org/finos/toolbox/collection/set/ImmutableArraySet.scala

@@ -38,7 +38,7 @@ object ImmutableArraySet {
     }
   }
 
-  def from[T <: Object](set: Set[T])(implicit c: ClassTag[T]): ImmutableArraySet[T] = {
+  def from[T <: Object](set: scala.collection.Set[T])(implicit c: ClassTag[T]): ImmutableArraySet[T] = {
     if (set.isEmpty) {
       empty()
     } else {

toolbox/src/main/scala/org/finos/toolbox/collection/set/VectorImmutableArraySet.scala

@@ -30,8 +30,8 @@ object VectorImmutableArraySet {
     VectorImmutableArraySet(builder.result(), seen.toSet)
   }
 
-  def from[T <: Object : ClassTag](set: Set[T]): ImmutableArraySet[T] = {
-    VectorImmutableArraySet(set.toVector, set)
+  def from[T <: Object : ClassTag](set: scala.collection.Set[T]): ImmutableArraySet[T] = {
+    VectorImmutableArraySet(set.toVector, set.toSet)
   }
 
   def empty[T <: Object : ClassTag](): ImmutableArraySet[T] = {

vuu/src/main/scala/org/finos/vuu/core/filter/FilterClause.scala

@@ -1,6 +1,7 @@
 package org.finos.vuu.core.filter
 
-import org.finos.toolbox.collection.array.ImmutableArray
+import org.finos.toolbox.collection.array.{ImmutableArray, VectorImmutableArray}
+import org.finos.toolbox.collection.set.ImmutableArraySet
 import org.finos.vuu.core.filter.FilterClause.joinResults
 import org.finos.vuu.core.index.*
 import org.finos.vuu.core.table.column.{Error, Result}
@@ -47,14 +48,28 @@ sealed trait RowFilterClause extends FilterClause {
     else Error(s"Column `$columnName` not found.")
 
   protected def hitIndex[T](primaryKeys: TablePrimaryKeys, value: T,
-                            indexLookup: T => ImmutableArray[String], firstInChain: Boolean): TablePrimaryKeys = {
+                            indexLookup: T => ImmutableArraySet[String], firstInChain: Boolean): TablePrimaryKeys = {
       val results  = indexLookup.apply(value)
       if (results.isEmpty) {
         EmptyTablePrimaryKeys
       } else if (firstInChain) {
-        InMemTablePrimaryKeys(results)
+        InMemTablePrimaryKeys(results.toImmutableArray)
       } else {
-        primaryKeys.intersect(results)
+
+        val keyLength = primaryKeys.length
+        val builder = Vector.newBuilder[String]
+        builder.sizeHint(results.length)
+
+        var i = 0
+        while (i < keyLength) {
+          val key = primaryKeys.get(i)
+          if (results.contains(key)) {
+            builder += key
+          }
+          i += 1
+        }
+
+        InMemTablePrimaryKeys(VectorImmutableArray.from(builder.result()))
       }
   }
 

vuu/src/main/scala/org/finos/vuu/core/filter/Filters.scala

@@ -1,12 +1,32 @@
 package org.finos.vuu.core.filter
 
+import com.typesafe.scalalogging.StrictLogging
 import org.finos.vuu.core.table.{EmptyTablePrimaryKeys, TablePrimaryKeys}
 import org.finos.vuu.viewport.{RowSource, ViewPortColumns}
 
 trait Filter {
   def doFilter(source: RowSource, primaryKeys: TablePrimaryKeys, firstInChain: Boolean): TablePrimaryKeys
 }
 
+case class CompoundFilter(filters: Filter*) extends Filter with StrictLogging {
+
+  override def doFilter(source: RowSource, primaryKeys: TablePrimaryKeys, firstInChain: Boolean): TablePrimaryKeys = {
+    logger.trace(s"Starting filter with ${primaryKeys.length} rows")
+    if (primaryKeys.isEmpty || filters.isEmpty) return primaryKeys
+
+    var currentKeys = primaryKeys
+    val filterIterator = filters.iterator
+
+    while (filterIterator.hasNext && currentKeys.nonEmpty) {
+      val filter = filterIterator.next()
+      val stillFirst = firstInChain && (currentKeys eq primaryKeys)
+      currentKeys = filter.doFilter(source, currentKeys, stillFirst)
+    }
+
+    currentKeys
+  }
+}
+
 trait ViewPortFilter {
   def doFilter(source: RowSource, primaryKeys: TablePrimaryKeys, vpColumns:ViewPortColumns, firstInChain: Boolean): TablePrimaryKeys
 }
@@ -21,23 +41,22 @@ object FilterOutEverythingFilter extends ViewPortFilter {
                         vpColumns: ViewPortColumns, firstInChain: Boolean): TablePrimaryKeys = EmptyTablePrimaryKeys
 }
 
-case class CompoundFilter(filters: ViewPortFilter*) extends ViewPortFilter {
+case class CompoundViewPortFilter(filters: ViewPortFilter*) extends ViewPortFilter with StrictLogging {
 
   override def doFilter(source: RowSource, primaryKeys: TablePrimaryKeys, 
                         vpColumns: ViewPortColumns, firstInChain: Boolean): TablePrimaryKeys = {
-    if (primaryKeys.isEmpty) {
-      primaryKeys
-    } else {
-      filters.foldLeft(primaryKeys) {
-        (remainingKeys, filter) => {
-          if (remainingKeys.isEmpty) {
-            remainingKeys
-          } else {
-            val stillFirstInChain = firstInChain && remainingKeys == primaryKeys
-            filter.doFilter(source, remainingKeys, vpColumns, stillFirstInChain)
-          }
-        }
-      }
+    logger.trace(s"Starting filter with ${primaryKeys.length} rows")
+    if (primaryKeys.isEmpty || filters.isEmpty) return primaryKeys
+
+    var currentKeys = primaryKeys
+    val filterIterator = filters.iterator
+
+    while (filterIterator.hasNext && currentKeys.nonEmpty) {
+      val filter = filterIterator.next()      
+      val stillFirst = firstInChain && (currentKeys eq primaryKeys)
+      currentKeys = filter.doFilter(source, currentKeys, vpColumns, stillFirst)
     }
+
+    currentKeys
   }
 }

vuu/src/main/scala/org/finos/vuu/core/filter/type/AntlrBasedFilter.scala

@@ -1,12 +1,12 @@
 package org.finos.vuu.core.filter.`type`
 
-import com.typesafe.scalalogging.LazyLogging
+import com.typesafe.scalalogging.StrictLogging
 import org.finos.vuu.core.filter.{FilterClause, ViewPortFilter}
 import org.finos.vuu.core.table.column.{Error, Success}
 import org.finos.vuu.core.table.{EmptyTablePrimaryKeys, TablePrimaryKeys}
 import org.finos.vuu.viewport.{RowSource, ViewPortColumns}
 
-case class AntlrBasedFilter(clause: FilterClause) extends ViewPortFilter with LazyLogging {
+case class AntlrBasedFilter(clause: FilterClause) extends ViewPortFilter with StrictLogging {
 
   override def doFilter(source: RowSource, rowKeys: TablePrimaryKeys, vpColumns: ViewPortColumns, firstInChain: Boolean): TablePrimaryKeys = {
     logger.trace(s"Starting filter with ${rowKeys.length}")

vuu/src/main/scala/org/finos/vuu/core/filter/type/BaseFilter.scala

@@ -1,6 +1,6 @@
 package org.finos.vuu.core.filter.`type`
 
-import org.finos.vuu.core.filter.ViewPortFilter
+import org.finos.vuu.core.filter.{CompoundFilter, ViewPortFilter}
 import org.finos.vuu.core.table.TablePrimaryKeys
 import org.finos.vuu.core.table.datatype.EpochTimestamp
 import org.finos.vuu.viewport.{RowSource, ViewPortColumns}
@@ -11,7 +11,7 @@ object BaseFilter {
 
   def apply(permissionFilter: PermissionFilter, frozenTime: Option[EpochTimestamp]): BaseFilter = {
     frozenTime match {
-      case Some(value) => PermissionAndFrozenTimeFilter(permissionFilter, FrozenTimeFilter(value))
+      case Some(value) => PermissionAndFrozenTimeFilter(permissionFilter, value)
       case None => OnlyPermissionFilter(permissionFilter)
     }
   }
@@ -27,20 +27,13 @@ private case class OnlyPermissionFilter(permissionFilter: PermissionFilter) exte
 
 }
 
-private case class PermissionAndFrozenTimeFilter(permissionFilter: PermissionFilter, frozenTimeFilter: FrozenTimeFilter) extends BaseFilter {
-
-  override def doFilter(source: RowSource, primaryKeys: TablePrimaryKeys, vpColumns: ViewPortColumns, firstInChain: Boolean): TablePrimaryKeys = {
-    if (primaryKeys.isEmpty) {
-      primaryKeys
-    } else {
-      val remainingKeys = permissionFilter.doFilter(source, primaryKeys, firstInChain)
-      if (remainingKeys.isEmpty) {
-        remainingKeys
-      } else {
-        val stillFirstInChain = firstInChain && remainingKeys == primaryKeys
-        frozenTimeFilter.doFilter(source, remainingKeys, stillFirstInChain)
-      }
-    }
+private case class PermissionAndFrozenTimeFilter(permissionFilter: PermissionFilter, frozenTime: EpochTimestamp) extends BaseFilter {
+
+  private val internalFilter = CompoundFilter(permissionFilter, FrozenTimeFilter(frozenTime))
+
+  override def doFilter(source: RowSource, primaryKeys: TablePrimaryKeys,
+                        vpColumns: ViewPortColumns, firstInChain: Boolean): TablePrimaryKeys = {
+    internalFilter.doFilter(source, primaryKeys, firstInChain)
   }
 
 }

vuu/src/main/scala/org/finos/vuu/core/filter/type/FrozenTimeFilter.scala

@@ -1,9 +1,9 @@
 package org.finos.vuu.core.filter.`type`
 
 import com.typesafe.scalalogging.LazyLogging
-import org.finos.toolbox.collection.array.ImmutableArray
+import org.finos.toolbox.collection.array.{ImmutableArray, VectorImmutableArray}
 import org.finos.vuu.core.filter.Filter
-import org.finos.vuu.core.index.{EpochTimestampIndexedField, LongIndexedField}
+import org.finos.vuu.core.index.EpochTimestampIndexedField
 import org.finos.vuu.core.table.datatype.EpochTimestamp
 import org.finos.vuu.core.table.{DefaultColumn, EmptyTablePrimaryKeys, TablePrimaryKeys}
 import org.finos.vuu.feature.inmem.InMemTablePrimaryKeys
@@ -13,9 +13,10 @@ case class FrozenTimeFilter(frozenTime: EpochTimestamp) extends Filter with Lazy
 
   override def doFilter(source: RowSource, primaryKeys: TablePrimaryKeys, firstInChain: Boolean): TablePrimaryKeys = {
     logger.trace(s"Starting filter with ${primaryKeys.length}")
+    if (primaryKeys.isEmpty) return primaryKeys
 
     val column = source.asTable.columnForName(DefaultColumn.CreatedTime.name)
-    if (column == null || primaryKeys.isEmpty) {
+    if (column == null) {
       EmptyTablePrimaryKeys
     } else {
       source.asTable.indexForColumn(column) match {
@@ -32,18 +33,41 @@ case class FrozenTimeFilter(frozenTime: EpochTimestamp) extends Filter with Lazy
     if (results.isEmpty) {
       EmptyTablePrimaryKeys
     } else if (firstInChain) {
-      InMemTablePrimaryKeys(results)
+      InMemTablePrimaryKeys(results.toImmutableArray)
     } else {
-      primaryKeys.intersect(results)
+      val keyLength = primaryKeys.length
+      val builder = Vector.newBuilder[String]
+      builder.sizeHint(Math.min(keyLength, results.length))
+
+      var i = 0
+      while (i < keyLength) {
+        val key = primaryKeys.get(i)
+        if (results.contains(key)) {
+          builder += key
+        }
+        i += 1
+      }
+
+      InMemTablePrimaryKeys(VectorImmutableArray.from(builder.result()))
     }
   }
 
-  private def filterAll(source: RowSource, rowKeys: TablePrimaryKeys): TablePrimaryKeys = {
-    val filtered = rowKeys.filter(key => {
-      val vuuCreatedTimestamp = source.pullRow(key).get(DefaultColumn.CreatedTime.name)
-      vuuCreatedTimestamp != null && vuuCreatedTimestamp.asInstanceOf[EpochTimestamp] < frozenTime
-    })
-
-    InMemTablePrimaryKeys(ImmutableArray.from[String](filtered.toArray))
+  private def filterAll(source: RowSource, primaryKeys: TablePrimaryKeys): TablePrimaryKeys = {
+    val length = primaryKeys.length
+    val builder = Vector.newBuilder[String]
+    builder.sizeHint(length)
+
+    var i = 0
+    while (i < length) {
+      val key = primaryKeys.get(i)
+      val row = source.pullRow(key)
+      val value = row.get(DefaultColumn.CreatedTime.name)
+      if (value != null && value.asInstanceOf[EpochTimestamp] < frozenTime) {
+        builder += key
+      }
+      i += 1
+    }
+
+    InMemTablePrimaryKeys(ImmutableArray.from[String](builder.result()))
   }
 }