fix: use ConcurrentHashMap for pattern cache in regexp UDFs

The bridge now caches a single shared CometUDF instance per class across
native worker threads, so UDF instance state must be thread-safe. Replace
the per-instance LinkedHashMap LRU (which mutates on get when accessOrder
is true) with ConcurrentHashMap.

Author: andygrove

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

@@ -20,7 +20,7 @@
 package org.apache.comet.udf
 
 import java.nio.charset.StandardCharsets
-import java.util
+import java.util.concurrent.ConcurrentHashMap
 import java.util.regex.Pattern
 
 import org.apache.arrow.vector.{IntVector, ValueVector, VarCharVector}
@@ -44,14 +44,7 @@ import org.apache.comet.CometArrowAllocator
  */
 class RegExpExtractAllUDF extends CometUDF {
 
-  private val patternCache =
-    new util.LinkedHashMap[String, Pattern](
-      RegExpExtractAllUDF.PatternCacheCapacity,
-      0.75f,
-      true) {
-      override def removeEldestEntry(eldest: util.Map.Entry[String, Pattern]): Boolean =
-        size() > RegExpExtractAllUDF.PatternCacheCapacity
-    }
+  private val patternCache = new ConcurrentHashMap[String, Pattern]()
 
   override def evaluate(inputs: Array[ValueVector]): ValueVector = {
     require(inputs.length == 3, s"RegExpExtractAllUDF expects 3 inputs, got ${inputs.length}")
@@ -117,7 +110,3 @@ class RegExpExtractAllUDF extends CometUDF {
     out
   }
 }
-
-object RegExpExtractAllUDF {
-  private val PatternCacheCapacity: Int = 128
-}

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

@@ -20,7 +20,7 @@
 package org.apache.comet.udf
 
 import java.nio.charset.StandardCharsets
-import java.util
+import java.util.concurrent.ConcurrentHashMap
 import java.util.regex.Pattern
 
 import org.apache.arrow.vector.{IntVector, ValueVector, VarCharVector}
@@ -42,11 +42,7 @@ import org.apache.comet.CometArrowAllocator
  */
 class RegExpExtractUDF extends CometUDF {
 
-  private val patternCache =
-    new util.LinkedHashMap[String, Pattern](RegExpExtractUDF.PatternCacheCapacity, 0.75f, true) {
-      override def removeEldestEntry(eldest: util.Map.Entry[String, Pattern]): Boolean =
-        size() > RegExpExtractUDF.PatternCacheCapacity
-    }
+  private val patternCache = new ConcurrentHashMap[String, Pattern]()
 
   override def evaluate(inputs: Array[ValueVector]): ValueVector = {
     require(inputs.length == 3, s"RegExpExtractUDF expects 3 inputs, got ${inputs.length}")
@@ -100,7 +96,3 @@ class RegExpExtractUDF extends CometUDF {
     out
   }
 }
-
-object RegExpExtractUDF {
-  private val PatternCacheCapacity: Int = 128
-}

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

@@ -20,7 +20,7 @@
 package org.apache.comet.udf
 
 import java.nio.charset.StandardCharsets
-import java.util
+import java.util.concurrent.ConcurrentHashMap
 import java.util.regex.Pattern
 
 import org.apache.arrow.vector.{IntVector, ValueVector, VarCharVector}
@@ -42,11 +42,7 @@ import org.apache.comet.CometArrowAllocator
  */
 class RegExpInStrUDF extends CometUDF {
 
-  private val patternCache =
-    new util.LinkedHashMap[String, Pattern](RegExpInStrUDF.PatternCacheCapacity, 0.75f, true) {
-      override def removeEldestEntry(eldest: util.Map.Entry[String, Pattern]): Boolean =
-        size() > RegExpInStrUDF.PatternCacheCapacity
-    }
+  private val patternCache = new ConcurrentHashMap[String, Pattern]()
 
   override def evaluate(inputs: Array[ValueVector]): ValueVector = {
     require(inputs.length == 3, s"RegExpInStrUDF expects 3 inputs, got ${inputs.length}")
@@ -96,7 +92,3 @@ class RegExpInStrUDF extends CometUDF {
     out
   }
 }
-
-object RegExpInStrUDF {
-  private val PatternCacheCapacity: Int = 128
-}

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

@@ -20,7 +20,7 @@
 package org.apache.comet.udf
 
 import java.nio.charset.StandardCharsets
-import java.util
+import java.util.concurrent.ConcurrentHashMap
 import java.util.regex.Pattern
 
 import org.apache.arrow.vector.{BitVector, ValueVector, VarCharVector}
@@ -38,13 +38,7 @@ import org.apache.comet.CometArrowAllocator
  */
 class RegExpLikeUDF extends CometUDF {
 
-  // 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
-    }
+  private val patternCache = new ConcurrentHashMap[String, Pattern]()
 
   override def evaluate(inputs: Array[ValueVector]): ValueVector = {
     require(inputs.length == 2, s"RegExpLikeUDF expects 2 inputs, got ${inputs.length}")
@@ -83,7 +77,3 @@ class RegExpLikeUDF extends CometUDF {
     out
   }
 }
-
-object RegExpLikeUDF {
-  private val PatternCacheCapacity: Int = 128
-}

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

@@ -20,7 +20,7 @@
 package org.apache.comet.udf
 
 import java.nio.charset.StandardCharsets
-import java.util
+import java.util.concurrent.ConcurrentHashMap
 import java.util.regex.Pattern
 
 import org.apache.arrow.vector.{ValueVector, VarCharVector}
@@ -41,11 +41,7 @@ import org.apache.comet.CometArrowAllocator
  */
 class RegExpReplaceUDF extends CometUDF {
 
-  private val patternCache =
-    new util.LinkedHashMap[String, Pattern](RegExpReplaceUDF.PatternCacheCapacity, 0.75f, true) {
-      override def removeEldestEntry(eldest: util.Map.Entry[String, Pattern]): Boolean =
-        size() > RegExpReplaceUDF.PatternCacheCapacity
-    }
+  private val patternCache = new ConcurrentHashMap[String, Pattern]()
 
   override def evaluate(inputs: Array[ValueVector]): ValueVector = {
     require(inputs.length == 3, s"RegExpReplaceUDF expects 3 inputs, got ${inputs.length}")
@@ -90,7 +86,3 @@ class RegExpReplaceUDF extends CometUDF {
     out
   }
 }
-
-object RegExpReplaceUDF {
-  private val PatternCacheCapacity: Int = 128
-}

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

@@ -20,7 +20,7 @@
 package org.apache.comet.udf
 
 import java.nio.charset.StandardCharsets
-import java.util
+import java.util.concurrent.ConcurrentHashMap
 import java.util.regex.Pattern
 
 import org.apache.arrow.vector.ValueVector
@@ -44,11 +44,7 @@ import org.apache.comet.CometArrowAllocator
  */
 class StringSplitUDF extends CometUDF {
 
-  private val patternCache =
-    new util.LinkedHashMap[String, Pattern](StringSplitUDF.PatternCacheCapacity, 0.75f, true) {
-      override def removeEldestEntry(eldest: util.Map.Entry[String, Pattern]): Boolean =
-        size() > StringSplitUDF.PatternCacheCapacity
-    }
+  private val patternCache = new ConcurrentHashMap[String, Pattern]()
 
   override def evaluate(inputs: Array[ValueVector]): ValueVector = {
     require(inputs.length == 3, s"StringSplitUDF expects 3 inputs, got ${inputs.length}")
@@ -106,7 +102,3 @@ class StringSplitUDF extends CometUDF {
     out
   }
 }
-
-object StringSplitUDF {
-  private val PatternCacheCapacity: Int = 128
-}