Merge pull request #324 from Cryptonomic/aggregation_and_datetime_fixes

Aggregation and DateTime fixes

doc/MetadataAndQuery.md

@@ -4,7 +4,8 @@ Description of endpoints with example usages. Probably all of those request will
 
 ## Query interface
 
-Query interface is using `POST` for passing the query
+Query interface is using `POST` for passing the query. Keep in mind that `DateTime` fields are represented as Unix timestamps in milliseconds.
+
 
 #### Example query
 ```

src/main/scala/tech/cryptonomic/conseil/generic/chain/DataTypes.scala

@@ -1,5 +1,8 @@
 package tech.cryptonomic.conseil.generic.chain
 
+import java.time.format.DateTimeFormatter.ISO_INSTANT
+import java.time.{Instant, ZoneOffset}
+
 import tech.cryptonomic.conseil.generic.chain.DataTypes.AggregationType.AggregationType
 import tech.cryptonomic.conseil.generic.chain.DataTypes.OperationType.OperationType
 import tech.cryptonomic.conseil.generic.chain.DataTypes.OrderDirection.OrderDirection
@@ -19,15 +22,44 @@ object DataTypes {
   import cats.implicits._
   import io.scalaland.chimney.dsl._
 
+
   /** Type representing Map[String, Option[Any]] for query response */
   type QueryResponse = Map[String, Option[Any]]
   /** Method checks if type can be aggregated */
-  lazy val canBeAggregated: DataType => Boolean = Set(DataType.Decimal, DataType.Int, DataType.LargeInt, DataType.DateTime)
+  lazy val canBeAggregated: DataType => AggregationType => Boolean = {
+    dataType => {
+      case AggregationType.count => true
+      case AggregationType.max | AggregationType.min => Set(DataType.Decimal, DataType.Int, DataType.LargeInt, DataType.DateTime)(dataType)
+      case AggregationType.avg | AggregationType.sum => Set(DataType.Decimal, DataType.Int, DataType.LargeInt)(dataType)
+    }
+  }
   /** Default value of limit parameter */
   val defaultLimitValue: Int = 10000
   /** Max value of limit parameter */
   val maxLimitValue: Int = 100000
 
+  /** Replaces timestamp represented as Long in predicates with one understood by the SQL */
+  private def replaceTimestampInPredicates(entity: String, query: Query, tezosPlatformDiscoveryOperations: TezosPlatformDiscoveryOperations)
+    (implicit executionContext: ExecutionContext): Future[Query] = {
+    query.predicates.map { predicate =>
+      tezosPlatformDiscoveryOperations.getTableAttributesWithoutUpdatingCache(entity).map { maybeAttributes =>
+        maybeAttributes.flatMap { attributes =>
+          attributes.find(_.name == predicate.field).map {
+            case attribute if attribute.dataType == DataType.DateTime =>
+              predicate.copy(set = predicate.set.map(x => formatToIso(x.toString.toLong)))
+            case _ => predicate
+          }
+        }.toList
+      }
+    }.sequence.map(pred => query.copy(predicates = pred.flatten))
+  }
+
+  /** Method formatting millis to ISO format */
+  def formatToIso(epochMillis: Long): String =
+    Instant.ofEpochMilli(epochMillis)
+      .atZone(ZoneOffset.UTC)
+      .format(ISO_INSTANT)
+
   /** Helper method for finding fields used in query that don't exist in the database */
   private def findNonExistingFields(query: Query, entity: String, tezosPlatformDiscovery: TezosPlatformDiscoveryOperations)
     (implicit ec: ExecutionContext): Future[List[QueryValidationError]] = {
@@ -58,7 +90,7 @@ object DataTypes {
           attributesOpt.flatMap { attributes =>
             attributes
               .find(_.name == aggregation.field)
-              .map(attribute => canBeAggregated(attribute.dataType) -> aggregation.field)
+              .map(attribute => canBeAggregated(attribute.dataType)(aggregation.function) -> aggregation.field)
           }
         }
       }
@@ -147,12 +179,14 @@ object DataTypes {
       for {
         invalidNonExistingFields <- nonExistingFields
         invalidAggregationFieldForTypes <- invalidTypeAggregationField
+        updatedQuery <- replaceTimestampInPredicates(entity, query, tezosPlatformDiscovery)
       } yield {
         invalidNonExistingFields ::: invalidAggregationFieldForTypes match {
-          case Nil => Right(query)
+          case Nil => Right(updatedQuery)
           case wrongFields => Left(wrongFields)
         }
       }
+
     }
   }
 

src/test/scala/tech/cryptonomic/conseil/tezos/DataTest.scala

@@ -6,7 +6,6 @@ import akka.http.scaladsl.testkit.ScalatestRouteTest
 import org.scalamock.scalatest.MockFactory
 import org.scalatest.concurrent.ScalaFutures
 import org.scalatest.{Matchers, WordSpec}
-import tech.cryptonomic.conseil.config.Newest
 import tech.cryptonomic.conseil.config.Platforms.{PlatformsConfiguration, Tezos, TezosConfiguration, TezosNodeConfiguration}
 import tech.cryptonomic.conseil.generic.chain.DataTypes.{Query, QueryResponse}
 import tech.cryptonomic.conseil.generic.chain.{DataOperations, DataPlatform}
@@ -100,6 +99,7 @@ class DataTest extends WordSpec with Matchers with ScalatestRouteTest with Scala
 
     "return a correct response with OK status code with POST" in {
       (tezosPlatformDiscoveryOperationsStub.isAttributeValid _).when(*, *).returns(Future.successful(true))
+      (tezosPlatformDiscoveryOperationsStub.getTableAttributesWithoutUpdatingCache _).when(*).returns(Future.successful(None))
       val postRequest = HttpRequest(
         HttpMethods.POST,
         uri = "/v2/data/tezos/alphanet/accounts",
@@ -115,6 +115,7 @@ class DataTest extends WordSpec with Matchers with ScalatestRouteTest with Scala
 
     "return 404 NotFound status code for request for the not supported platform with POST" in {
       (tezosPlatformDiscoveryOperationsStub.isAttributeValid _).when(*, *).returns(Future.successful(true))
+      (tezosPlatformDiscoveryOperationsStub.getTableAttributesWithoutUpdatingCache _).when(*).returns(Future.successful(None))
       val postRequest = HttpRequest(
         HttpMethods.POST,
         uri = "/v2/data/notSupportedPlatform/alphanet/accounts",
@@ -126,6 +127,7 @@ class DataTest extends WordSpec with Matchers with ScalatestRouteTest with Scala
 
     "return 404 NotFound status code for request for the not supported network with POST" in {
       (tezosPlatformDiscoveryOperationsStub.isAttributeValid _).when(*, *).returns(Future.successful(true))
+      (tezosPlatformDiscoveryOperationsStub.getTableAttributesWithoutUpdatingCache _).when(*).returns(Future.successful(None))
       val postRequest = HttpRequest(
         HttpMethods.POST,
         uri = "/v2/data/tezos/notSupportedNetwork/accounts",

src/test/scala/tech/cryptonomic/conseil/tezos/DataTypesTest.scala

@@ -50,7 +50,16 @@ import scala.concurrent.ExecutionContext.Implicits.global
     }
 
     "validate correct predicate field" in {
+      val attribute = Attribute(
+        name = "valid",
+        displayName = "valid",
+        dataType = DataType.Int,
+        cardinality = None,
+        keyType = KeyType.NonKey,
+        entity = "test"
+      )
       (tpdo.isAttributeValid _).when("test", "valid").returns(Future.successful(true))
+      (tpdo.getTableAttributesWithoutUpdatingCache _).when("test").returns(Future.successful(Some(List(attribute))))
 
       val query = ApiQuery(
         fields = None,
@@ -67,7 +76,16 @@ import scala.concurrent.ExecutionContext.Implicits.global
     }
 
     "return error with incorrect predicate fields" in {
+      val attribute = Attribute(
+        name = "valid",
+        displayName = "valid",
+        dataType = DataType.Int,
+        cardinality = None,
+        keyType = KeyType.NonKey,
+        entity = "test"
+      )
       (tpdo.isAttributeValid _).when("test", "invalid").returns(Future.successful(false))
+      (tpdo.getTableAttributesWithoutUpdatingCache _).when("test").returns(Future.successful(Some(List(attribute))))
 
       val query = ApiQuery(
         fields = None,
@@ -197,6 +215,60 @@ import scala.concurrent.ExecutionContext.Implicits.global
 
       result.futureValue.left.get should contain theSameElementsAs List(InvalidAggregationField("invalid"), InvalidAggregationFieldForType("invalid"))
     }
+
+    "successfully validates aggregation for any dataType when COUNT function is used" in {
+      val attribute = Attribute(
+        name = "valid",
+        displayName = "Valid",
+        dataType = DataType.String, // only COUNT function can be used on types other than numeric and DateTime
+        cardinality = None,
+        keyType = KeyType.NonKey,
+        entity = "test"
+      )
+
+      (tpdo.isAttributeValid _).when("test", "valid").returns(Future.successful(true))
+      (tpdo.getTableAttributesWithoutUpdatingCache _).when("test").returns(Future.successful(Some(List(attribute))))
+
+      val query = ApiQuery(
+        fields = Some(List("valid")),
+        predicates = None,
+        orderBy = None,
+        limit = None,
+        output = None,
+        aggregation = Some(Aggregation(field = "valid", function = AggregationType.count))
+      )
+
+      val result = query.validate("test", tpdo)
+
+      result.futureValue.right.get shouldBe Query(fields = List("valid"), aggregation = Some(Aggregation(field = "valid", function = AggregationType.count)))
+    }
+
+    "correctly transform predicate DateTime field as Long into ISO" in {
+      val attribute = Attribute(
+        name = "valid",
+        displayName = "Valid",
+        dataType = DataType.DateTime, // only COUNT function can be used on types other than numeric and DateTime
+        cardinality = None,
+        keyType = KeyType.NonKey,
+        entity = "test"
+      )
+
+      (tpdo.isAttributeValid _).when("test", "valid").returns(Future.successful(true))
+      (tpdo.getTableAttributesWithoutUpdatingCache _).when("test").returns(Future.successful(Some(List(attribute))))
+
+      val query = ApiQuery(
+        fields = None,
+        predicates = Some(List(Predicate(field = "valid", operation = OperationType.in, set = List(123456789000L)))),
+        orderBy = None,
+        limit = None,
+        output = None,
+        aggregation = None
+      )
+
+      val result = query.validate("test", tpdo)
+
+      result.futureValue.right.get shouldBe Query(predicates = List(Predicate(field = "valid", operation = OperationType.in, set = List("1973-11-29T21:33:09Z"))))
+    }
   }
 
 }