fix: bound UDF and pattern caches with LRU eviction

The per-thread CometUDF instance cache (CometUdfBridge.INSTANCES) and
the per-instance Pattern cache in RegExpLikeUDF were unbounded, so a
workload that registers many UDF classes or evaluates many distinct
regex patterns would retain entries for the executor's JVM lifetime.
Both caches now use access-order LinkedHashMap with removeEldestEntry
bounds (64 UDF instances, 128 patterns).

Author: andygrove

common/src/main/java/org/apache/comet/udf/CometUdfBridge.java

@@ -19,7 +19,8 @@
 
 package org.apache.comet.udf;
 
-import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
 
 import org.apache.arrow.c.ArrowArray;
 import org.apache.arrow.c.ArrowSchema;
@@ -35,14 +36,23 @@
  */
 public class CometUdfBridge {
 
-  // Per-thread cache of UDF instances keyed by class name. Comet native
-  // execution threads (Tokio/DataFusion worker pool) are reused across
-  // tasks within an executor, so the effective lifetime of cached entries
-  // is the worker thread (i.e. the executor JVM). This is fine for
+  // Per-thread, bounded LRU of UDF instances keyed by class name. Comet
+  // native execution threads (Tokio/DataFusion worker pool) are reused
+  // across tasks within an executor, so the effective lifetime of cached
+  // entries is the worker thread (i.e. the executor JVM). This is fine for
   // stateless UDFs like RegExpLikeUDF; future stateful UDFs would need
   // explicit per-task isolation.
-  private static final ThreadLocal<HashMap<String, CometUDF>> INSTANCES =
-      ThreadLocal.withInitial(HashMap::new);
+  private static final int CACHE_CAPACITY = 64;
+
+  private static final ThreadLocal<LinkedHashMap<String, CometUDF>> INSTANCES =
+      ThreadLocal.withInitial(
+          () ->
+              new LinkedHashMap<String, CometUDF>(CACHE_CAPACITY, 0.75f, true) {
+                @Override
+                protected boolean removeEldestEntry(Map.Entry<String, CometUDF> eldest) {
+                  return size() > CACHE_CAPACITY;
+                }
+              });
 
   /**
    * Called from native via JNI.
@@ -59,7 +69,7 @@ public static void evaluate(
       long[] inputSchemaPtrs,
       long outArrayPtr,
       long outSchemaPtr) {
-    HashMap<String, CometUDF> cache = INSTANCES.get();
+    LinkedHashMap<String, CometUDF> cache = INSTANCES.get();
     CometUDF udf = cache.get(udfClassName);
     if (udf == null) {
       try {

common/src/main/scala/org/apache/comet/udf/RegExpLikeUDF.scala

@@ -20,10 +20,9 @@
 package org.apache.comet.udf
 
 import java.nio.charset.StandardCharsets
+import java.util
 import java.util.regex.Pattern
 
-import scala.collection.mutable
-
 import org.apache.arrow.vector.{BitVector, ValueVector, VarCharVector}
 
 import org.apache.comet.CometArrowAllocator
@@ -39,7 +38,13 @@ import org.apache.comet.CometArrowAllocator
  */
 class RegExpLikeUDF extends CometUDF {
 
-  private val patternCache = mutable.Map.empty[String, Pattern]
+  // Bounded LRU so a workload with many distinct patterns does not retain
+  // Pattern objects for the executor's lifetime.
+  private val patternCache =
+    new util.LinkedHashMap[String, Pattern](RegExpLikeUDF.PatternCacheCapacity, 0.75f, true) {
+      override def removeEldestEntry(eldest: util.Map.Entry[String, Pattern]): Boolean =
+        size() > RegExpLikeUDF.PatternCacheCapacity
+    }
 
   override def evaluate(inputs: Array[ValueVector]): ValueVector = {
     require(inputs.length == 2, s"RegExpLikeUDF expects 2 inputs, got ${inputs.length}")
@@ -50,7 +55,15 @@ class RegExpLikeUDF extends CometUDF {
       "RegExpLikeUDF requires a non-null scalar pattern")
 
     val patternStr = new String(patternVec.get(0), StandardCharsets.UTF_8)
-    val pattern = patternCache.getOrElseUpdate(patternStr, Pattern.compile(patternStr))
+    val pattern = {
+      val cached = patternCache.get(patternStr)
+      if (cached != null) cached
+      else {
+        val compiled = Pattern.compile(patternStr)
+        patternCache.put(patternStr, compiled)
+        compiled
+      }
+    }
 
     val n = subject.getValueCount
     val out = new BitVector("rlike_result", CometArrowAllocator)
@@ -70,3 +83,7 @@ class RegExpLikeUDF extends CometUDF {
     out
   }
 }
+
+object RegExpLikeUDF {
+  private val PatternCacheCapacity: Int = 128
+}