Merged with latest main branch

Author: kumar-zlai

.github/workflows/test_scala_no_spark.yaml

@@ -48,6 +48,7 @@ jobs:
 
       - name: Run Flink tests
         run: |
+          export SBT_OPTS="-Xmx24G -Xms4G --add-opens=java.base/sun.nio.ch=ALL-UNNAMED"
           sbt "++ 2.12.18 flink/test"
 
       - name: Run Aggregator tests
@@ -58,6 +59,10 @@ jobs:
         run: |
           sbt "++ 2.12.18 online/test"
 
+      - name: Run Catalyst Util tests
+        run: |
+          sbt "++ 2.12.18 online/testOnly -- -n catalystUtilHiveUdfTest"
+
       - name: Run api tests
         run: |
           sbt "++ 2.12.18 api/test"

.plugin-versions

@@ -1,5 +1,7 @@
 asdf-plugin-manager  https://github.com/asdf-community/asdf-plugin-manager.git  b5862c1
+gcloud               https://github.com/jthegedus/asdf-gcloud.git               00cdf06
 java                 https://github.com/halcyon/asdf-java.git                   0ec69b2
 python               https://github.com/danhper/asdf-python.git                 a3a0185
 sbt                  https://github.com/lerencao/asdf-sbt                       53c9f4b
 scala                https://github.com/asdf-community/asdf-scala.git           0533444
+thrift               https://github.com/alisaifee/asdf-thrift.git               fecdd6c

.tool-versions

@@ -1,7 +1,8 @@
-java corretto-11.0.25.9.1
-scala 2.12.20
+java corretto-17.0.9.8.1
+scala 2.12.18
 asdf-plugin-manager 1.4.0
 sbt 1.8.2
 python
     3.7.17
     3.11.0
+gcloud 504.0.1

aggregator/src/test/scala/ai/chronon/aggregator/test/ApproxDistinctTest.scala

@@ -17,10 +17,10 @@
 package ai.chronon.aggregator.test
 
 import ai.chronon.aggregator.base.ApproxDistinctCount
-import junit.framework.TestCase
 import org.junit.Assert._
+import org.scalatest.flatspec.AnyFlatSpec
 
-class ApproxDistinctTest extends TestCase {
+class ApproxDistinctTest extends AnyFlatSpec {
   def testErrorBound(uniques: Int, errorBound: Int, lgK: Int): Unit = {
     val uniqueElems = 1 to uniques
     val duplicates = uniqueElems ++ uniqueElems ++ uniqueElems
@@ -50,13 +50,13 @@ class ApproxDistinctTest extends TestCase {
     assertTrue(Math.abs(estimated - uniques) < errorBound)
   }
 
-  def testErrorBounds(): Unit = {
+  it should "error bounds" in {
     testErrorBound(uniques = 100, errorBound = 1, lgK = 10)
     testErrorBound(uniques = 1000, errorBound = 20, lgK = 10)
     testErrorBound(uniques = 10000, errorBound = 300, lgK = 10)
   }
 
-  def testMergingErrorBounds(): Unit = {
+  it should "merging error bounds" in {
     testMergingErrorBound(uniques = 100, errorBound = 1, lgK = 10, merges = 10)
     testMergingErrorBound(uniques = 1000, errorBound = 20, lgK = 10, merges = 4)
     testMergingErrorBound(uniques = 10000, errorBound = 400, lgK = 10, merges = 100)

aggregator/src/test/scala/ai/chronon/aggregator/test/ApproxHistogramTest.scala

@@ -2,14 +2,14 @@ package ai.chronon.aggregator.test
 
 import ai.chronon.aggregator.base.ApproxHistogram
 import ai.chronon.aggregator.base.ApproxHistogramIr
-import junit.framework.TestCase
 import org.junit.Assert._
+import org.scalatest.flatspec.AnyFlatSpec
 
 import java.util
 import scala.jdk.CollectionConverters._
 
-class ApproxHistogramTest extends TestCase {
-  def testHistogram(): Unit = {
+class ApproxHistogramTest extends AnyFlatSpec {
+  it should "histogram" in {
     val approxHistogram = new ApproxHistogram[String](3)
     val counts = (1L to 3).map(i => i.toString -> i).toMap
     val ir = makeIr(approxHistogram, counts)
@@ -19,7 +19,7 @@ class ApproxHistogramTest extends TestCase {
     assertEquals(toHashMap(counts), approxHistogram.finalize(ir))
   }
 
-  def testSketch(): Unit = {
+  it should "sketch" in {
     val approxHistogram = new ApproxHistogram[String](3)
     val counts = (1L to 4).map(i => i.toString -> i).toMap
     val expected = counts.toSeq.sortBy(_._2).reverse.take(3).toMap
@@ -30,7 +30,7 @@ class ApproxHistogramTest extends TestCase {
     assertEquals(toHashMap(expected), approxHistogram.finalize(ir))
   }
 
-  def testMergeSketches(): Unit = {
+  it should "merge sketches" in {
     val approxHistogram = new ApproxHistogram[String](3)
     val counts1: Map[String, Long] = Map("5" -> 5L, "4" -> 4, "2" -> 2, "1" -> 1)
     val counts2: Map[String, Long] = Map("6" -> 6L, "4" -> 4, "2" -> 2, "1" -> 1)
@@ -52,7 +52,7 @@ class ApproxHistogramTest extends TestCase {
     assertTrue(ir.histogram.isEmpty)
   }
 
-  def testMergeHistograms(): Unit = {
+  it should "merge histograms" in {
     val approxHistogram = new ApproxHistogram[String](3)
     val counts1: Map[String, Long] = Map("4" -> 4L, "2" -> 2)
     val counts2: Map[String, Long] = Map("3" -> 3L, "2" -> 2)
@@ -74,7 +74,7 @@ class ApproxHistogramTest extends TestCase {
     assertTrue(ir.sketch.isEmpty)
   }
 
-  def testMergeHistogramsToSketch(): Unit = {
+  it should "merge histograms to sketch" in {
     val approxHistogram = new ApproxHistogram[String](3)
     val counts1: Map[String, Long] = Map("4" -> 4L, "3" -> 3)
     val counts2: Map[String, Long] = Map("2" -> 2L, "1" -> 1)
@@ -97,7 +97,7 @@ class ApproxHistogramTest extends TestCase {
     assertTrue(ir.histogram.isEmpty)
   }
 
-  def testMergeSketchAndHistogram(): Unit = {
+  it should "merge sketch and histogram" in {
     val approxHistogram = new ApproxHistogram[String](3)
     val counts1: Map[String, Long] = Map("5" -> 5L, "3" -> 3, "2" -> 2, "1" -> 1)
     val counts2: Map[String, Long] = Map("2" -> 2L)
@@ -119,7 +119,7 @@ class ApproxHistogramTest extends TestCase {
     assert(ir.histogram.isEmpty)
   }
 
-  def testNormalizeHistogram(): Unit = {
+  it should "normalize histogram" in {
     val approxHistogram = new ApproxHistogram[String](3)
     val counts = (1L to 3).map(i => i.toString -> i).toMap
     val ir = makeIr(approxHistogram, counts)
@@ -129,7 +129,7 @@ class ApproxHistogramTest extends TestCase {
     assertEquals(ir, normalized)
   }
 
-  def testNormalizeSketch(): Unit = {
+  it should "normalize sketch" in {
     val approxHistogram = new ApproxHistogram[String](3)
     val counts = (1L to 4).map(i => i.toString -> i).toMap
     val expected = counts.toSeq.sortBy(_._2).reverse.take(3).toMap

aggregator/src/test/scala/ai/chronon/aggregator/test/ApproxPercentilesTest.scala

@@ -18,15 +18,15 @@ package ai.chronon.aggregator.test
 
 import ai.chronon.aggregator.base.ApproxPercentiles
 import ai.chronon.aggregator.row.StatsGenerator
-import junit.framework.TestCase
 import org.apache.datasketches.kll.KllFloatsSketch
 import org.junit.Assert._
+import org.scalatest.flatspec.AnyFlatSpec
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 
 import scala.util.Random
 
-class ApproxPercentilesTest extends TestCase {
+class ApproxPercentilesTest extends AnyFlatSpec {
   @transient lazy val logger: Logger = LoggerFactory.getLogger(getClass)
 
   def basicImplTestHelper(nums: Int, slide: Int, k: Int, percentiles: Array[Double], errorPercent: Float): Unit = {
@@ -56,7 +56,7 @@ class ApproxPercentilesTest extends TestCase {
     diffs.foreach(diff => assertTrue(diff < errorMargin))
   }
 
-  def testBasicPercentiles: Unit = {
+  it should "basic percentiles: unit = {" in {
     val percentiles_tested: Int = 31
     val percentiles: Array[Double] = (0 to percentiles_tested).toArray.map(i => i * 1.0 / percentiles_tested)
     basicImplTestHelper(3000, 5, 100, percentiles, errorPercent = 4)
@@ -74,7 +74,7 @@ class ApproxPercentilesTest extends TestCase {
     drift
   }
 
-  def testPSIDrifts(): Unit = {
+  it should "psi drifts" in {
     assertTrue(
       getPSIDrift(
         Array(1, 1, 1, 2, 2, 3, 3, 4, 4, 5, 5, 6, 6, 7, 7).map(_.toFloat),

aggregator/src/test/scala/ai/chronon/aggregator/test/FrequentItemsTest.scala

@@ -4,14 +4,14 @@ import ai.chronon.aggregator.base.FrequentItemType
 import ai.chronon.aggregator.base.FrequentItems
 import ai.chronon.aggregator.base.FrequentItemsFriendly
 import ai.chronon.aggregator.base.ItemsSketchIR
-import junit.framework.TestCase
 import org.junit.Assert._
+import org.scalatest.flatspec.AnyFlatSpec
 
 import java.util
 import scala.jdk.CollectionConverters._
 
-class FrequentItemsTest extends TestCase {
-  def testNonPowerOfTwoAndTruncate(): Unit = {
+class FrequentItemsTest extends AnyFlatSpec {
+  it should "non power of two and truncate" in {
     val size = 3
     val items = new FrequentItems[String](size)
     val ir = items.prepare("4")
@@ -32,7 +32,7 @@ class FrequentItemsTest extends TestCase {
     )), result)
   }
 
-  def testLessItemsThanSize(): Unit = {
+  it should "less items than size" in {
     val size = 10
     val items = new FrequentItems[java.lang.Long](size)
     val ir = items.prepare(3)
@@ -52,7 +52,7 @@ class FrequentItemsTest extends TestCase {
     )), result)
   }
 
-  def testZeroSize(): Unit = {
+  it should "zero size" in {
     val size = 0
     val items = new FrequentItems[java.lang.Double](size)
     val ir = items.prepare(3.0)
@@ -68,7 +68,7 @@ class FrequentItemsTest extends TestCase {
     assertEquals(new util.HashMap[String, Double](), result)
   }
 
-  def testSketchSizes(): Unit = {
+  it should "sketch sizes" in {
     val expectedSketchSizes =
       Map(
         -1 -> 2,
@@ -87,7 +87,7 @@ class FrequentItemsTest extends TestCase {
     assertEquals(expectedSketchSizes, actualSketchSizes)
   }
 
-  def testNormalization(): Unit = {
+  it should "normalization" in {
     val testValues = (1 to 4)
       .map(i => i -> i)
       .toMap
@@ -118,7 +118,7 @@ class FrequentItemsTest extends TestCase {
     assertEquals(expectedStringValues, actualStringValues)
   }
 
-   def testBulkMerge(): Unit = {
+  it should "bulk merge" in {
      val sketch = new FrequentItems[String](3)
 
      val irs = Seq(

aggregator/src/test/scala/ai/chronon/aggregator/test/MinHeapTest.scala

@@ -17,14 +17,14 @@
 package ai.chronon.aggregator.test
 
 import ai.chronon.aggregator.base.MinHeap
-import junit.framework.TestCase
 import org.junit.Assert._
+import org.scalatest.flatspec.AnyFlatSpec
 
 import java.util
 import scala.collection.JavaConverters._
 
-class MinHeapTest extends TestCase {
-  def testInserts(): Unit = {
+class MinHeapTest extends AnyFlatSpec {
+  it should "inserts" in {
     val mh = new MinHeap[Int](maxSize = 4, Ordering.Int)
 
     def make_container = new util.ArrayList[Int](4)

aggregator/src/test/scala/ai/chronon/aggregator/test/MomentTest.scala

@@ -1,12 +1,12 @@
 package ai.chronon.aggregator.test
 
 import ai.chronon.aggregator.base._
-import junit.framework.TestCase
 import org.apache.commons.math3.stat.descriptive.moment.{Kurtosis => ApacheKurtosis}
 import org.apache.commons.math3.stat.descriptive.moment.{Skewness => ApacheSkew}
 import org.junit.Assert._
+import org.scalatest.flatspec.AnyFlatSpec
 
-class MomentTest extends TestCase {
+class MomentTest extends AnyFlatSpec {
   def makeAgg(aggregator: MomentAggregator, values: Seq[Double]): (MomentAggregator, MomentsIR) = {
     var ir = aggregator.prepare(values.head)
 
@@ -36,32 +36,32 @@ class MomentTest extends TestCase {
     assertEquals(expected(v1 ++ v2), agg.finalize(ir), 0.1)
   }
 
-  def testUpdate(): Unit = {
+  it should "update" in {
     val values = Seq(1.1, 2.2, 3.3, 4.4, 5.5)
     assertUpdate(new Skew(), values, expectedSkew)
     assertUpdate(new Kurtosis(), values, expectedKurtosis)
   }
 
-  def testInsufficientSizes(): Unit = {
+  it should "insufficient sizes" in {
     val values = Seq(1.1, 2.2, 3.3, 4.4)
     assertUpdate(new Skew(), values.take(2), _ => Double.NaN)
     assertUpdate(new Kurtosis(), values.take(3), _ => Double.NaN)
   }
 
-  def testNoVariance(): Unit = {
+  it should "no variance" in {
     val values = Seq(1.0, 1.0, 1.0, 1.0)
     assertUpdate(new Skew(), values, _ => Double.NaN)
     assertUpdate(new Kurtosis(), values, _ => Double.NaN)
   }
 
-  def testMerge(): Unit = {
+  it should "merge" in {
     val values1 = Seq(1.1, 2.2, 3.3)
     val values2 = Seq(4.4, 5.5)
     assertMerge(new Kurtosis(), values1, values2, expectedKurtosis)
     assertMerge(new Skew(), values1, values2, expectedSkew)
   }
 
-  def testNormalize(): Unit = {
+  it should "normalize" in {
     val values = Seq(1.0, 2.0, 3.0, 4.0, 5.0)
     val (agg, ir) = makeAgg(new Kurtosis, values)
 

aggregator/src/test/scala/ai/chronon/aggregator/test/RowAggregatorTest.scala

@@ -18,8 +18,8 @@ package ai.chronon.aggregator.test
 
 import ai.chronon.aggregator.row.RowAggregator
 import ai.chronon.api._
-import junit.framework.TestCase
 import org.junit.Assert._
+import org.scalatest.flatspec.AnyFlatSpec
 
 import java.util
 import scala.collection.JavaConverters._
@@ -48,8 +48,8 @@ object TestRow {
   def apply(inputsArray: Any*): TestRow = new TestRow(inputsArray: _*)()
 }
 
-class RowAggregatorTest extends TestCase {
-  def testUpdate(): Unit = {
+class RowAggregatorTest extends AnyFlatSpec {
+  it should "update" in {
     val rows = List(
       TestRow(1L, 4, 5.0f, "A", Seq(5, 3, 4), Seq("D", "A", "B", "A"), Map("A" -> 1, "B" -> 2)),
       TestRow(2L, 3, 4.0f, "B", Seq(6, null), Seq(), null),

aggregator/src/test/scala/ai/chronon/aggregator/test/SawtoothAggregatorTest.scala

@@ -22,8 +22,8 @@ import ai.chronon.aggregator.windowing._
 import ai.chronon.api.Extensions.AggregationOps
 import ai.chronon.api._
 import com.google.gson.Gson
-import junit.framework.TestCase
 import org.junit.Assert._
+import org.scalatest.flatspec.AnyFlatSpec
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 
@@ -46,9 +46,9 @@ class Timer {
   }
 }
 
-class SawtoothAggregatorTest extends TestCase {
+class SawtoothAggregatorTest extends AnyFlatSpec {
 
-  def testTailAccuracy(): Unit = {
+  it should "tail accuracy" in {
     val timer = new Timer
     val queries = CStream.genTimestamps(new Window(30, TimeUnit.DAYS), 10000, 5 * 60 * 1000)
 
@@ -119,7 +119,7 @@ class SawtoothAggregatorTest extends TestCase {
     }
   }
 
-  def testRealTimeAccuracy(): Unit = {
+  it should "real time accuracy" in {
     val timer = new Timer
     val queries = CStream.genTimestamps(new Window(1, TimeUnit.DAYS), 1000)
     val columns = Seq(Column("ts", LongType, 180),

aggregator/src/test/scala/ai/chronon/aggregator/test/SawtoothOnlineAggregatorTest.scala

@@ -24,17 +24,17 @@ import ai.chronon.api.Extensions.WindowOps
 import ai.chronon.api.Extensions.WindowUtils
 import ai.chronon.api._
 import com.google.gson.Gson
-import junit.framework.TestCase
 import org.junit.Assert.assertEquals
+import org.scalatest.flatspec.AnyFlatSpec
 
 import java.time.Instant
 import java.time.ZoneOffset
 import java.time.format.DateTimeFormatter
 import java.util.Locale
 
-class SawtoothOnlineAggregatorTest extends TestCase {
+class SawtoothOnlineAggregatorTest extends AnyFlatSpec {
 
-  def testConsistency(): Unit = {
+  it should "consistency" in {
     val queryEndTs = TsUtils.round(System.currentTimeMillis(), WindowUtils.Day.millis)
     val batchEndTs = queryEndTs - WindowUtils.Day.millis
     val queries = CStream.genTimestamps(new Window(1, TimeUnit.DAYS), 1000)