Merge pull request #152 from crobox/feature/first_last_value

Added support for first_ and lastValue
crobox · Nov 1, 2023 · 0ca44db · 0ca44db
2 parents 1bf451e + b647ba6
commit 0ca44db
Show file tree

Hide file tree

Showing 4 changed files with 56 additions and 14 deletions.
diff --git a/dsl/src/it/scala/com/crobox/clickhouse/dsl/column/AggregationFunctionsIT.scala b/dsl/src/it/scala/com/crobox/clickhouse/dsl/column/AggregationFunctionsIT.scala
@@ -1,7 +1,7 @@
 package com.crobox.clickhouse.dsl.column
 
-import com.crobox.clickhouse.dsl.{TableColumn, forEach, quantiles, ref, select, sum, uniq, uniqExact}
-import com.crobox.clickhouse.{ClickhouseClientSpec, DslITSpec}
+import com.crobox.clickhouse.DslITSpec
+import com.crobox.clickhouse.dsl._
 import spray.json.DefaultJsonProtocol._
 import spray.json.RootJsonFormat
 
@@ -13,8 +13,9 @@ class AggregationFunctionsIT extends DslITSpec {
   private val delta   = 2
   override val table1Entries: Seq[Table1Entry] =
     Seq.fill(entries)(Table1Entry(UUID.randomUUID(), numbers = Seq(1, 2, 3)))
-  override val table2Entries: Seq[Table2Entry] =
-    Seq.fill(entries)(Table2Entry(UUID.randomUUID(), randomString, randomInt, randomString, None))
+  override val table2Entries: Seq[Table2Entry] = {
+    (1 to entries).map(i => Table2Entry(UUID.randomUUID(), i + "_" + randomString, randomInt, randomString, None))
+  }
 
   "Combinators" should "apply for aggregations" in {
     case class Result(columnResult: String) {
@@ -58,4 +59,23 @@ class AggregationFunctionsIT extends DslITSpec {
     queryResult should contain theSameElementsAs Seq(entries, entries * 2, entries * 3)
   }
 
+  it should "firstValue in aggregate" in {
+    val resultRows =
+      chExecutor
+        .execute[StringResult](select(firstValue(col1) as "result").from(TwoTestTable))
+        .futureValue
+        .rows
+    resultRows.length shouldBe 1
+    resultRows.map(_.result).head.startsWith("1_") should be(true)
+  }
+
+  it should "lastValue in aggregate" in {
+    val resultRows =
+      chExecutor
+        .execute[StringResult](select(lastValue(col1) as "result").from(TwoTestTable))
+        .futureValue
+        .rows
+    resultRows.length shouldBe 1
+    resultRows.map(_.result).head.startsWith(s"${entries}_") should be(true)
+  }
 }
diff --git a/dsl/src/main/scala/com.crobox.clickhouse/dsl/column/AggregationFunctions.scala b/dsl/src/main/scala/com.crobox.clickhouse/dsl/column/AggregationFunctions.scala
@@ -32,6 +32,10 @@ trait AggregationFunctions {
 
   case class Max[V](tableColumn: TableColumn[V]) extends AggregateFunction[V](tableColumn)
 
+  case class FirstValue[V](tableColumn: TableColumn[V]) extends AggregateFunction[V](tableColumn)
+
+  case class LastValue[V](tableColumn: TableColumn[V]) extends AggregateFunction[V](tableColumn)
+
   case class TimeSeries(tableColumn: TableColumn[Long], interval: MultiInterval)
       extends AggregateFunction[Long](tableColumn)
 
@@ -45,6 +49,10 @@ trait AggregationFunctions {
 
   def max[V](tableColumn: TableColumn[V]): Max[V] = Max(tableColumn)
 
+  def firstValue[V](tableColumn: TableColumn[V]): FirstValue[V] = FirstValue(tableColumn)
+
+  def lastValue[V](tableColumn: TableColumn[V]): LastValue[V] = LastValue(tableColumn)
+
   /**
    * This function will push back the timestamp represented by tableColumn to the start of this interval,
    * this happens deterministically.

diff --git a/dsl/src/main/scala/com.crobox.clickhouse/dsl/language/AggregationFunctionTokenizer.scala b/dsl/src/main/scala/com.crobox.clickhouse/dsl/language/AggregationFunctionTokenizer.scala
@@ -35,28 +35,30 @@ trait AggregationFunctionTokenizer { this: ClickhouseTokenizerModule =>
       agg: AggregateFunction[_]
   )(implicit ctx: TokenizeContext): (String, String) =
     agg match {
-      case Avg(column)   => ("avg", tokenizeColumn(column))
-      case Count(column) => ("count", tokenizeColumn(column.getOrElse(EmptyColumn)))
+      case AnyResult(column, modifier) => (s"any${tokenizeAnyModifier(modifier)}", tokenizeColumn(column))
+      case Avg(column)                 => ("avg", tokenizeColumn(column))
+      case Count(column)               => ("count", tokenizeColumn(column.getOrElse(EmptyColumn)))
+      case FirstValue(column)          => ("first_value", tokenizeColumn(column))
+      case GroupArray(tableColumn, maxValues) =>
+        ("groupArray", s"${maxValues.map(_.toString + ")(").getOrElse("")}${tokenizeColumn(tableColumn)}")
+      case GroupUniqArray(tableColumn) => ("groupUniqArray", tokenizeColumn(tableColumn))
+      case LastValue(column)           => ("last_value", tokenizeColumn(column))
       case Median(column, level, modifier) =>
         val (modifierName, modifierValue) = tokenizeLevelModifier(modifier)
         (s"median$modifierName", s"$level)(${tokenizeColumn(column)}${modifierValue.map("," + _).getOrElse("")}")
+      case Min(tableColumn) => ("min", tokenizeColumn(tableColumn))
+      case Max(tableColumn) => ("max", tokenizeColumn(tableColumn))
       case Quantile(column, level, modifier) =>
         val (modifierName, modifierValue) = tokenizeLevelModifier(modifier)
         (s"quantile$modifierName", s"$level)(${tokenizeColumn(column)}${modifierValue.map("," + _).getOrElse("")})")
       case Quantiles(column, levels, modifier) =>
         val (modifierName, modifierValue) = tokenizeLevelModifier(modifier)
         (s"quantiles$modifierName",
          s"${levels.mkString(",")})(${tokenizeColumn(column)}${modifierValue.map("," + _).getOrElse("")}")
+      case Sum(column, modifier) => (s"sum${tokenizeSumModifier(modifier)}", tokenizeColumn(column))
+      case SumMap(key, value)    => (s"sumMap", tokenizeColumns(Seq(key, value)))
       case Uniq(columns, modifier) =>
         (s"uniq${tokenizeUniqModifier(modifier)}", columns.map(tokenizeColumn).mkString(","))
-      case Sum(column, modifier)       => (s"sum${tokenizeSumModifier(modifier)}", tokenizeColumn(column))
-      case SumMap(key, value)          => (s"sumMap", tokenizeColumns(Seq(key, value)))
-      case AnyResult(column, modifier) => (s"any${tokenizeAnyModifier(modifier)}", tokenizeColumn(column))
-      case Min(tableColumn)            => ("min", tokenizeColumn(tableColumn))
-      case Max(tableColumn)            => ("max", tokenizeColumn(tableColumn))
-      case GroupUniqArray(tableColumn) => ("groupUniqArray", tokenizeColumn(tableColumn))
-      case GroupArray(tableColumn, maxValues) =>
-        ("groupArray", s"${maxValues.map(_.toString + ")(").getOrElse("")}${tokenizeColumn(tableColumn)}")
       case f: AggregateFunction[_] =>
         throw new IllegalArgumentException(s"Cannot use $f aggregated function with combinator")
     }

diff --git a/dsl/src/test/scala/com/crobox/clickhouse/dsl/language/AggregationFunctionTokenizerTest.scala b/dsl/src/test/scala/com/crobox/clickhouse/dsl/language/AggregationFunctionTokenizerTest.scala
@@ -10,4 +10,16 @@ class AggregationFunctionTokenizerTest extends DslTestSpec {
       "SELECT groupArray(column_1)[1] AS p"
     )
   }
+
+  it should "firstValue in groupArray" in {
+    toSQL(select(firstValue(groupArray(col1)) as "p"), false) should matchSQL(
+      "SELECT first_value(groupArray(column_1)) AS p"
+    )
+  }
+
+  it should "lastValue in groupArray" in {
+    toSQL(select(lastValue(groupArray(col1)) as "p"), false) should matchSQL(
+      "SELECT last_value(groupArray(column_1)) AS p"
+    )
+  }
 }