Benchmarking

benchmark/src/main/scala/benchmark/bench/Bench.scala

@@ -0,0 +1,26 @@
+package benchmark.bench
+
+trait Bench {
+  val RecordCount: Int = 1_000_000
+  val StreamIterations: Int = 1
+  val SearchIterations: Int = 1
+
+  val tasks: List[Task] = List(
+    Task("Insert Records", RecordCount, insertRecords),
+    Task("Stream Records", StreamIterations, streamRecords),
+    Task("Search Each Record", StreamIterations * RecordCount, searchEachRecord),
+    Task("Search All Records", StreamIterations, searchAllRecords)
+  )
+
+  def init(): Unit
+
+  protected def insertRecords(status: StatusCallback): Unit
+
+  protected def streamRecords(status: StatusCallback): Unit
+
+  protected def searchEachRecord(status: StatusCallback): Unit
+
+  protected def searchAllRecords(status: StatusCallback): Unit
+
+  def dispose(): Unit
+}

benchmark/src/main/scala/benchmark/bench/BenchmarkReport.scala

@@ -0,0 +1,9 @@
+package benchmark.bench
+
+import fabric.rw.RW
+
+case class BenchmarkReport(name: String, maxProgress: Double, logs: List[StatusLog])
+
+object BenchmarkReport {
+  implicit val rw: RW[BenchmarkReport] = RW.gen
+}

benchmark/src/main/scala/benchmark/bench/ReportGenerator.scala

@@ -0,0 +1,7 @@
+package benchmark.bench
+
+object ReportGenerator {
+  def main(args: Array[String]): Unit = {
+
+  }
+}

benchmark/src/main/scala/benchmark/bench/Runner.scala

@@ -0,0 +1,58 @@
+package benchmark.bench
+
+import benchmark.bench.impl.LightDBBench
+import fabric.io.JsonFormatter
+import fabric.rw.Convertible
+import lightdb.duckdb.DuckDBIndexer
+import lightdb.h2.H2Indexer
+import lightdb.halo.HaloDBStore
+import lightdb.lucene.LuceneIndexer
+import lightdb.mapdb.MapDBStore
+import lightdb.rocks.RocksDBStore
+import lightdb.sqlite.SQLiteIndexer
+import lightdb.store.{AtomicMapStore, MapStore}
+
+import java.nio.file.{Files, Path}
+
+object Runner {
+  val implementations: Map[String, Bench] = Map(
+    "ldbHaloLucene" -> LightDBBench(HaloDBStore, LuceneIndexer),
+    "ldbMapLucene" -> LightDBBench(MapDBStore, LuceneIndexer),
+    "ldbRocksLucene" -> LightDBBench(RocksDBStore, LuceneIndexer),
+    "ldbAtomicLucene" -> LightDBBench(AtomicMapStore, LuceneIndexer),
+    "ldbMapLucene" -> LightDBBench(MapStore, LuceneIndexer),
+    "ldbHaloSQLite" -> LightDBBench(HaloDBStore, SQLiteIndexer),
+    "ldbHaloH2" -> LightDBBench(HaloDBStore, H2Indexer),
+    "ldbHaloDuck" -> LightDBBench(HaloDBStore, DuckDBIndexer),
+  )
+
+  def main(args: Array[String]): Unit = {
+    args.headOption match {
+      case Some(implName) if implementations.contains(implName) =>
+        val bench = implementations(implName)
+        scribe.info(s"Initializing $implName benchmark...")
+        bench.init()
+        scribe.info(s"Initialized successfully!")
+        val reports = bench.tasks.map { task =>
+          val status = StatusCallback()
+          status.start()
+          scribe.info(s"Executing ${task.name} task...")
+          task.f(status)
+          status.finish()
+          val logs = status.logs
+          scribe.info(s"Completed in ${logs.last.elapsed} seconds")
+          BenchmarkReport(task.name, task.maxProgress, logs)
+        }
+        scribe.info(s"Disposing $implName benchmark...")
+        bench.dispose()
+        scribe.info(s"Disposed!")
+
+        val json = reports.json
+        Files.writeString(Path.of(s"report-$implName.json"), JsonFormatter.Default(json))
+
+        sys.exit(0)
+      case Some(implName) => scribe.error(s"Invalid implementation name: $implName. Valid implementations: ${implementations.keys.mkString(", ")}")
+      case None => scribe.error(s"Exactly one command-line argument must be present to specify the implementation. Valid implementations: ${implementations.keys.mkString(", ")}")
+    }
+  }
+}

benchmark/src/main/scala/benchmark/bench/StatusCallback.scala

@@ -0,0 +1,58 @@
+package benchmark.bench
+
+import com.google.common.util.concurrent.AtomicDouble
+import com.sun.management.OperatingSystemMXBean
+
+import java.lang.management.ManagementFactory
+
+case class StatusCallback(every: Long = 1_000L) {
+  val progress = new AtomicDouble(0.0)
+
+  def logs: List[StatusLog] = _logs.reverse
+
+  private var keepAlive = true
+  private val startTime = System.currentTimeMillis()
+  private var _logs = List.empty[StatusLog]
+
+  def start(): Unit = {
+    val t = new Thread {
+      setDaemon(true)
+
+      override def run(): Unit = {
+        while (keepAlive) {
+          report()
+          Thread.sleep(every)
+        }
+      }
+    }
+    t.start()
+  }
+
+  def finish(): Unit = {
+    keepAlive = false
+    report()
+  }
+
+  private def report(): Unit = {
+    val now = System.currentTimeMillis()
+    val elapsed = (now - startTime) / 1000.0
+    val memory = ManagementFactory.getMemoryMXBean
+    val heap = memory.getHeapMemoryUsage
+    val nonHeap = memory.getNonHeapMemoryUsage
+    val heapUsed = heap.getUsed
+    val nonHeapUsed = nonHeap.getUsed
+    val os = ManagementFactory.getPlatformMXBean(classOf[OperatingSystemMXBean])
+    val cpuLoad = os.getProcessCpuLoad
+    val cpuTime = os.getProcessCpuTime
+    val log = StatusLog(
+      progress = progress.get(),
+      timeStamp = now,
+      elapsed = elapsed,
+      heap = heapUsed,
+      nonHeap = nonHeapUsed,
+      cpuLoad = cpuLoad,
+      cpuTime = cpuTime
+    )
+    _logs = log :: _logs
+  }
+}

benchmark/src/main/scala/benchmark/bench/StatusLog.scala

@@ -0,0 +1,15 @@
+package benchmark.bench
+
+import fabric.rw.RW
+
+case class StatusLog(progress: Double,
+                     timeStamp: Long,
+                     elapsed: Double,
+                     heap: Long,
+                     nonHeap: Long,
+                     cpuLoad: Double,
+                     cpuTime: Long)
+
+object StatusLog {
+  implicit val rw: RW[StatusLog] = RW.gen
+}

benchmark/src/main/scala/benchmark/bench/Task.scala

@@ -0,0 +1,3 @@
+package benchmark.bench
+
+case class Task(name: String, maxProgress: Double = 1.0, f: StatusCallback => Unit)

benchmark/src/main/scala/benchmark/bench/impl/LightDBBench.scala

@@ -0,0 +1,104 @@
+package benchmark.bench.impl
+
+import benchmark.bench.{Bench, StatusCallback}
+import fabric.rw.RW
+import lightdb.document.{Document, DocumentModel}
+import lightdb.index.{Indexed, IndexedCollection, IndexerManager}
+import lightdb.store.StoreManager
+import lightdb.upgrade.DatabaseUpgrade
+import lightdb.util.Unique
+import lightdb.{Id, LightDB}
+import org.apache.commons.io.FileUtils
+
+import java.io.File
+import java.nio.file.Path
+import scala.collection.parallel.CollectionConverters._
+
+case class LightDBBench(sm: StoreManager, im: IndexerManager) extends Bench {
+  override def init(): Unit = {
+    val dbDir = new File("db")
+    FileUtils.deleteDirectory(dbDir)
+    dbDir.mkdirs()
+
+    scribe.info("DB init...")
+    DB.init()
+    scribe.info("Initialized!")
+  }
+
+  override protected def insertRecords(status: StatusCallback): Unit = DB.people.transaction { implicit transaction =>
+    (0 until RecordCount)
+      .foreach { index =>
+        DB.people.set(Person(
+          name = Unique(),
+          age = index
+        ))
+        status.progress.set(index + 1)
+      }
+  }
+
+  override protected def streamRecords(status: StatusCallback): Unit = DB.people.transaction { implicit transaction =>
+    (0 until StreamIterations)
+      .par
+      .foreach { iteration =>
+        val count = DB.people.iterator.size
+        if (count != RecordCount) {
+          scribe.warn(s"RecordCount was not $RecordCount, it was $count")
+        }
+        status.progress.set(iteration + 1)
+      }
+  }
+
+  override protected def searchEachRecord(status: StatusCallback): Unit = DB.people.transaction { implicit transaction =>
+    (0 until StreamIterations)
+      .par
+      .foreach { iteration =>
+        (0 until RecordCount)
+          .par
+          .foreach { index =>
+            val list = DB.people.query.filter(_.age === index).search.docs.list
+            if (list.size != 1) {
+              scribe.warn(s"Unable to find age = $index")
+            }
+            if (list.head.age != index) {
+              scribe.warn(s"${list.head.age} was not $index")
+            }
+            status.progress.set((iteration + 1) * (index + 1))
+          }
+      }
+  }
+
+  override protected def searchAllRecords(status: StatusCallback): Unit = DB.people.transaction { implicit transaction =>
+    (0 until StreamIterations)
+      .par
+      .foreach { iteration =>
+        val count = DB.people.query.search.docs.iterator.foldLeft(0)((count, _) => count + 1)
+        if (count != RecordCount) {
+          scribe.warn(s"RecordCount was not $RecordCount, it was $count")
+        }
+        status.progress.set(iteration + 1)
+      }
+  }
+
+  override def dispose(): Unit = DB.dispose()
+
+  object DB extends LightDB {
+    override lazy val directory: Option[Path] = Some(Path.of(s"db/bench"))
+
+    val people: IndexedCollection[Person, Person.type] = collection("people", Person, im.create[Person, Person.type]())
+
+    override def storeManager: StoreManager = sm
+
+    override def upgrades: List[DatabaseUpgrade] = Nil
+  }
+
+  case class Person(name: String,
+                    age: Int,
+                    _id: Id[Person] = Person.id()) extends Document[Person]
+
+  object Person extends DocumentModel[Person] with Indexed[Person] {
+    implicit val rw: RW[Person] = RW.gen
+
+    val name: I[String] = index.one("name", _.name, store = true)
+    val age: I[Int] = index.one("age", _.age, store = true)
+  }
+}

benchmark/src/main/scala/benchmark/BenchmarkImplementation.scala → benchmark/src/main/scala/benchmark/imdb/BenchmarkImplementation.scala

@@ -1,4 +1,4 @@
-package benchmark
+package benchmark.imdb
 
 import cats.effect.IO
 

benchmark/src/main/scala/benchmark/IMDBBenchmark.scala → benchmark/src/main/scala/benchmark/imdb/IMDBBenchmark.scala

@@ -1,20 +1,19 @@
-package benchmark
+package benchmark.imdb
 
-import cats.effect.unsafe.IORuntime
+import benchmark.IOIterator
 import cats.effect.IO
-
-import java.io.{BufferedOutputStream, BufferedReader, File, FileInputStream, FileOutputStream, FileReader, PrintWriter}
-import java.util.zip.GZIPInputStream
-import scala.io.Source
+import cats.effect.unsafe.IORuntime
 import fs2._
 import fs2.io.file._
 import perfolation._
-import scribe.{Level, Logger}
 
-import java.net.{URI, URL}
-import scala.annotation.tailrec
-import sys.process._
+import java.io._
+import java.net.URI
 import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
+import java.util.zip.GZIPInputStream
+import scala.annotation.tailrec
+import scala.io.Source
+import scala.sys.process._
 
 object IMDBBenchmark { // extends IOApp {
   val limit: Limit = Limit.OneMillion

benchmark/src/main/scala/benchmark/LightDBImplementation.scala → benchmark/src/main/scala/benchmark/imdb/LightDBImplementation.scala

@@ -1,14 +1,13 @@
-package benchmark
+package benchmark.imdb
 
 import cats.effect.IO
 import fabric.rw.{Asable, RW}
 import lightdb.collection.Collection
 import lightdb.document.{Document, DocumentModel}
 import lightdb.halo.HaloDBStore
 import lightdb.index.{Indexed, IndexedCollection}
-import lightdb.lucene.LuceneIndexer
 import lightdb.sqlite.SQLiteIndexer
-import lightdb.store.{AtomicMapStore, StoreManager}
+import lightdb.store.StoreManager
 import lightdb.transaction.Transaction
 import lightdb.upgrade.DatabaseUpgrade
 import lightdb.{Id, LightDB}

benchmark/src/main/scala/benchmark/MariaDBImplementation.scala → benchmark/src/main/scala/benchmark/imdb/MariaDBImplementation.scala

@@ -1,5 +1,6 @@
-package benchmark
+package benchmark.imdb
 
+import benchmark.FlushingBacklog
 import cats.effect.IO
 import cats.effect.unsafe.IORuntime
 

benchmark/src/main/scala/benchmark/MongoDBImplementation.scala → benchmark/src/main/scala/benchmark/imdb/MongoDBImplementation.scala

@@ -1,5 +1,6 @@
-package benchmark
+package benchmark.imdb
 
+import benchmark.FlushingBacklog
 import cats.effect.IO
 import cats.effect.unsafe.IORuntime
 import com.mongodb.client.MongoClients

benchmark/src/main/scala/benchmark/PostgresImplementation.scala → benchmark/src/main/scala/benchmark/imdb/PostgresImplementation.scala

@@ -1,10 +1,10 @@
-package benchmark
-import cats.effect.{IO, Unique}
-import cats.effect.unsafe.IORuntime
+package benchmark.imdb
+
 import benchmark.FlushingBacklog
+import cats.effect.IO
+import cats.effect.unsafe.IORuntime
 
 import java.sql.{Connection, DriverManager, ResultSet}
-import scala.concurrent.{ExecutionContext, Future}
 
 object PostgresImplementation extends BenchmarkImplementation {
   implicit val runtime: IORuntime = IORuntime.global

benchmark/src/main/scala/benchmark/SQLiteImplementation.scala → benchmark/src/main/scala/benchmark/imdb/SQLiteImplementation.scala

@@ -1,5 +1,6 @@
-package benchmark
+package benchmark.imdb
 
+import benchmark.FlushingBacklog
 import cats.effect.IO
 import cats.effect.unsafe.IORuntime
 import lightdb.util.Unique

benchmark/src/main/scala/benchmark/ScarangoImplementation.scala → benchmark/src/main/scala/benchmark/imdb/ScarangoImplementation.scala

@@ -1,12 +1,12 @@
-package benchmark
+package benchmark.imdb
 
+import benchmark.FlushingBacklog
 import cats.effect.IO
 import com.outr.arango.collection.DocumentCollection
 import com.outr.arango.query._
 import com.outr.arango.query.dsl.ref2Wrapped
 import com.outr.arango.{Document, DocumentModel, Field, Graph, Id, Index}
 import fabric.rw.RW
-import benchmark.FlushingBacklog
 
 object ScarangoImplementation extends BenchmarkImplementation {
   override type TitleAka = TitleAkaADB