Merge pull request #23 from ringcentral/allow_null_values_only_for_op…

…tion_type

Allow null values only for option type

build.sbt

@@ -25,16 +25,17 @@ lazy val root = (project in file("."))
   .configs(IntegrationTest)
   .settings(
     Defaults.itSettings,
+    IntegrationTest / fork := true,
     libraryDependencies ++= Seq(
       "org.typelevel"       %% "cats-effect"                    % "3.3.5",
-      "co.fs2"              %% "fs2-core"                       % "3.2.4",
-      "com.datastax.oss"     % "java-driver-core"               % "4.13.0",
-      "com.chuusai"         %% "shapeless"                      % "2.3.7"
+      "co.fs2"              %% "fs2-core"                       % "3.2.5",
+      "com.datastax.oss"     % "java-driver-core"               % "4.14.0",
+      "com.chuusai"         %% "shapeless"                      % "2.3.8"
     ) ++ Seq(
-      "com.disneystreaming" %% "weaver-cats"                    % "0.7.6"  % "it,test",
+      "com.disneystreaming" %% "weaver-cats"                    % "0.7.10"  % "it,test",
       "org.testcontainers"   % "testcontainers"                 % "1.16.3" % "it",
-      "com.dimafeng"        %% "testcontainers-scala-cassandra" % "0.40.0" % "it",
-      "ch.qos.logback"       % "logback-classic"                % "1.2.10" % "it,test"
+      "com.dimafeng"        %% "testcontainers-scala-cassandra" % "0.40.2" % "it",
+      "ch.qos.logback"       % "logback-classic"                % "1.2.11" % "it,test"
     ) ++ (scalaBinaryVersion.value match {
       case v if v.startsWith("2.13") =>
         Seq.empty
@@ -84,4 +85,4 @@ Compile / compile / scalacOptions ++= Seq(
     case other                     => sys.error(s"Unsupported scala version: $other")
   })
 
-testFrameworks += new TestFramework("weaver.framework.CatsEffect")
+testFrameworks := Seq(new TestFramework("weaver.framework.CatsEffect"))

src/it/resources/migration/1__test_tables.cql

@@ -1,12 +1,14 @@
 create table test_data(
     id bigint,
     data text,
+    count int,
+    dataset frozen<set<int>>,
     PRIMARY KEY (id)
 );
 
-insert into test_data (id, data) values (0, null);
-insert into test_data (id, data) values (1, 'one');
-insert into test_data (id, data) values (2, 'two');
+insert into test_data (id, data, count, dataset) values (0, null, null, null);
+insert into test_data (id, data, count, dataset) values (1, 'one', 10, {});
+insert into test_data (id, data, count, dataset) values (2, 'two', 20, {201});
 insert into test_data (id, data) values (3, 'three');
 
 create table test_data_multiple_keys(

src/it/scala/com/ringcentral/cassandra4io/CassandraSessionSuite.scala

@@ -10,9 +10,6 @@ import weaver._
 
 trait CassandraSessionSuite { self: IOSuite with CassandraTestsSharedInstances =>
 
-  private def getError[T](either: Either[Throwable, T]): Throwable =
-    either.swap.getOrElse(new RuntimeException("Either is right defined"))
-
   implicit def toStatement(s: String): SimpleStatement = SimpleStatement.newInstance(s)
 
   test("CassandraSession.connect be referentially transparent") { _ =>

src/it/scala/com/ringcentral/cassandra4io/CassandraTestsSharedInstances.scala

@@ -11,10 +11,14 @@ import org.testcontainers.utility.DockerImageName
 import weaver.IOSuite
 
 import java.net.InetSocketAddress
+import java.time.Duration
 import scala.io.BufferedSource
+import org.slf4j.LoggerFactory
 
 trait CassandraTestsSharedInstances { self: IOSuite =>
 
+  val logger = LoggerFactory.getLogger(self.getClass)
+
   val keyspace  = "cassandra4io"
   val container = CassandraContainer(DockerImageName.parse("cassandra:3.11.11"))
 
@@ -24,10 +28,16 @@ trait CassandraTestsSharedInstances { self: IOSuite =>
       _         <- session.execute(s"use $keyspace")
       source    <- migrationSource
       migrations = splitToMigrations(source)
+      _         <- IO(logger.info("start cassandra migration for tests"))
       _         <- migrations.toList.traverse_ { migration =>
-                     val st = SimpleStatement.newInstance(migration)
-                     session.execute(st)
+                     val st = SimpleStatement.newInstance(migration).setTimeout(Duration.ofSeconds(4))
+                     session.execute(st).onError { error =>
+                       IO {
+                         logger.error(s"Error in execution migration $migration", error)
+                       }
+                     }
                    }
+      _         <- IO(logger.info("cassandra migration done"))
     } yield ()
   }
 
@@ -70,4 +80,7 @@ trait CassandraTestsSharedInstances { self: IOSuite =>
       .mkString("")
     s1.split(';').toList.map(_.strip())
   }
+
+  def getError[T](either: Either[Throwable, T]): Throwable =
+    either.swap.getOrElse(new RuntimeException("Either is right defined"))
 }

src/it/scala/com/ringcentral/cassandra4io/cql/CqlSuite.scala

@@ -9,19 +9,39 @@ import weaver._
 
 import java.time.{ Duration, LocalDate, LocalTime }
 import java.util.UUID
+import java.util.concurrent.atomic.AtomicInteger
 
-trait CqlSuite { self: IOSuite with CassandraTestsSharedInstances =>
+trait CqlSuite {
+  self: IOSuite with CassandraTestsSharedInstances =>
 
   case class Data(id: Long, data: String)
 
+  case class OptData(id: Long, data: Option[String])
+
   case class BasicInfo(weight: Double, height: String, datapoints: Set[Int])
+
   object BasicInfo {
     implicit val cqlReads: Reads[BasicInfo]   = FromUdtValue.deriveReads[BasicInfo]
     implicit val cqlBinder: Binder[BasicInfo] = ToUdtValue.deriveBinder[BasicInfo]
   }
 
   case class PersonAttribute(personId: Int, info: BasicInfo)
 
+  object PersonAttribute {
+    val idxCounter = new AtomicInteger(0)
+  }
+
+  case class PersonAttributeOpt(personId: Int, info: Option[BasicInfo])
+
+  case class OptBasicInfo(weight: Option[Double], height: Option[String], datapoints: Option[Set[Int]])
+
+  object OptBasicInfo {
+    implicit val cqlReads: Reads[OptBasicInfo]   = FromUdtValue.deriveReads[OptBasicInfo]
+    implicit val cqlBinder: Binder[OptBasicInfo] = ToUdtValue.deriveBinder[OptBasicInfo]
+  }
+
+  case class PersonAttributeUdtOpt(personId: Int, info: OptBasicInfo)
+
   case class CollectionTestRow(
     id: Int,
     maptest: Map[String, UUID],
@@ -34,6 +54,7 @@ trait CqlSuite { self: IOSuite with CassandraTestsSharedInstances =>
   case class ExampleNestedType(a: Int, b: String, c: Option[ExampleType])
 
   case class ExampleCollectionNestedUdtType(a: Int, b: Map[Int, Set[Set[Set[Set[ExampleNestedType]]]]])
+
   object ExampleCollectionNestedUdtType {
     implicit val binderExampleCollectionNestedUdtType: Binder[ExampleCollectionNestedUdtType] =
       ToUdtValue.deriveBinder[ExampleCollectionNestedUdtType]
@@ -43,6 +64,7 @@ trait CqlSuite { self: IOSuite with CassandraTestsSharedInstances =>
   }
 
   case class ExampleNestedPrimitiveType(a: Int, b: Map[Int, Set[Set[Set[Set[Int]]]]])
+
   object ExampleNestedPrimitiveType {
     implicit val binderExampleNestedPrimitiveType: Binder[ExampleNestedPrimitiveType] =
       ToUdtValue.deriveBinder[ExampleNestedPrimitiveType]
@@ -68,12 +90,12 @@ trait CqlSuite { self: IOSuite with CassandraTestsSharedInstances =>
 
   test("interpolated select template should return tuples from migration") { session =>
     for {
-      prepared <- cqlt"select id, data FROM cassandra4io.test_data WHERE id in ${Put[List[Long]]}"
-                    .as[(Long, String)]
+      prepared <- cqlt"select id, data, dataset FROM cassandra4io.test_data WHERE id in ${Put[List[Long]]}"
+                    .as[(Long, String, Option[Set[Int]])]
                     .prepare(session)
       query     = prepared(List[Long](1, 2, 3))
       results  <- query.select.compile.toList
-    } yield expect(results == Seq((1, "one"), (2, "two"), (3, "three")))
+    } yield expect(results == Seq((1, "one", Some(Set.empty)), (2, "two", Some(Set(201))), (3, "three", None)))
   }
 
   test("interpolated select template should return tuples from migration with multiple binding") { session =>
@@ -116,6 +138,7 @@ trait CqlSuite { self: IOSuite with CassandraTestsSharedInstances =>
       cql"select data FROM cassandra4io.test_data WHERE id in $ids"
         .as[String]
         .config(_.setConsistencyLevel(ConsistencyLevel.ALL))
+
     for {
       results <- getDataByIds(List(1, 2, 3)).select(session).compile.toList
     } yield expect(results == Seq("one", "two", "three"))
@@ -124,6 +147,7 @@ trait CqlSuite { self: IOSuite with CassandraTestsSharedInstances =>
   test("interpolated select should return tuples from migration") { session =>
     def getAllByIds(ids: List[Long]) =
       cql"select id, data FROM cassandra4io.test_data WHERE id in $ids".as[(Long, String)]
+
     for {
       results <- getAllByIds(List(1, 2, 3)).config(_.setQueryTimestamp(0L)).select(session).compile.toList
     } yield expect(results == Seq((1, "one"), (2, "two"), (3, "three")))
@@ -132,6 +156,7 @@ trait CqlSuite { self: IOSuite with CassandraTestsSharedInstances =>
   test("interpolated select should return tuples from migration with multiple binding") { session =>
     def getAllByIds(id1: Long, id2: Int) =
       cql"select data FROM cassandra4io.test_data_multiple_keys WHERE id1 = $id1 and id2 = $id2".as[String]
+
     for {
       results <- getAllByIds(1, 2).select(session).compile.toList
     } yield expect(results == Seq("one-two"))
@@ -141,6 +166,7 @@ trait CqlSuite { self: IOSuite with CassandraTestsSharedInstances =>
     def getAllByIds(id1: Long, id2: Int) =
       cql"""select data FROM cassandra4io.test_data_multiple_keys
            |WHERE id1 = $id1 and id2 = $id2""".stripMargin.as[String]
+
     for {
       results <- getAllByIds(1, 2).select(session).compile.toList
     } yield expect(results == Seq("one-two"))
@@ -149,6 +175,7 @@ trait CqlSuite { self: IOSuite with CassandraTestsSharedInstances =>
   test("interpolated select should return data case class from migration") { session =>
     def getIds(ids: List[Long]) =
       cql"select id, data FROM cassandra4io.test_data WHERE id in $ids".as[Data]
+
     for {
       results <- getIds(List(1, 2, 3)).select(session).compile.toList
     } yield expect(results == Seq(Data(1, "one"), Data(2, "two"), Data(3, "three")))
@@ -157,7 +184,8 @@ trait CqlSuite { self: IOSuite with CassandraTestsSharedInstances =>
   test(
     "interpolated inserts and selects should produce UDTs and return data case classes when nested case classes are used"
   ) { session =>
-    val data   = PersonAttribute(1, BasicInfo(180.0, "tall", Set(1, 2, 3, 4, 5)))
+    val data   =
+      PersonAttribute(PersonAttribute.idxCounter.incrementAndGet(), BasicInfo(180.0, "tall", Set(1, 2, 3, 4, 5)))
     val insert =
       cql"INSERT INTO cassandra4io.person_attributes (person_id, info) VALUES (${data.personId}, ${data.info})"
         .execute(session)
@@ -297,7 +325,8 @@ trait CqlSuite { self: IOSuite with CassandraTestsSharedInstances =>
   }
 
   test("cqlConst allows you to interpolate on what is usually not possible with cql strings") { session =>
-    val data         = PersonAttribute(2, BasicInfo(180.0, "tall", Set(1, 2, 3, 4, 5)))
+    val data         =
+      PersonAttribute(PersonAttribute.idxCounter.incrementAndGet(), BasicInfo(180.0, "tall", Set(1, 2, 3, 4, 5)))
     val keyspaceName = "cassandra4io"
     val tableName    = "person_attributes"
     val selectFrom   = cql"SELECT person_id, info FROM "
@@ -316,4 +345,151 @@ trait CqlSuite { self: IOSuite with CassandraTestsSharedInstances =>
       result <- (selectFrom ++ keyspace ++ table ++ where(data.personId)).as[PersonAttribute].selectFirst(session)
     } yield expect(result.isDefined && result.get == data)
   }
+
+  // handle NULL values
+  test("decoding from null should return None for Option[String]") { session =>
+    for {
+      result <- cql"select data FROM cassandra4io.test_data WHERE id = 0".as[Option[String]].selectFirst(session)
+    } yield expect(result.isDefined && result.get.isEmpty)
+  }
+
+  test("decoding from null should raise error for String(non-primitive)") { session =>
+    for {
+      result <-
+        cql"select data FROM cassandra4io.test_data WHERE id = 0".as[String].selectFirst(session).attempt
+    } yield expect(result.isLeft) && expect(
+      getError(result).isInstanceOf[UnexpectedNullValue]
+    )
+  }
+
+  test("decoding from null should raise error for Int(primitive)") { session =>
+    for {
+      result <-
+        cql"select count FROM cassandra4io.test_data WHERE id = 0".as[String].selectFirst(session).attempt
+    } yield expect(result.isLeft) && expect(
+      getError(result).isInstanceOf[UnexpectedNullValue]
+    )
+  }
+
+  test("decoding from null should raise error for Set(collection)") { session =>
+    for {
+      result <-
+        cql"select dataset FROM cassandra4io.test_data WHERE id = 0".as[Set[Int]].selectFirst(session).attempt
+    } yield expect(result.isLeft) && expect(
+      getError(result).isInstanceOf[UnexpectedNullValue]
+    )
+  }
+
+  test("decoding from null should return None for Option[String] field in case class") { session =>
+    for {
+      row <- cql"select id, data FROM cassandra4io.test_data WHERE id = 0".as[OptData].selectFirst(session)
+    } yield expect(row.isDefined && row.get.data.isEmpty)
+  }
+
+  test("decoding from null should raise error String field in case class") { session =>
+    for {
+      result <- cql"select id, data FROM cassandra4io.test_data WHERE id = 0".as[Data].selectFirst(session).attempt
+    } yield expect(result.isLeft) && expect(getError(result).isInstanceOf[UnexpectedNullValue])
+  }
+
+  // handle NULL values for udt columns
+
+  test("decoding from null at udt column should return None for Option type") { session =>
+    val data = PersonAttributeOpt(PersonAttribute.idxCounter.incrementAndGet(), None)
+
+    for {
+      _      <- cql"INSERT INTO cassandra4io.person_attributes (person_id, info) VALUES (${data.personId}, ${data.info})"
+                  .execute(session)
+      result <- cql"SELECT person_id, info FROM cassandra4io.person_attributes WHERE person_id = ${data.personId}"
+                  .as[PersonAttributeOpt]
+                  .select(session)
+                  .compile
+                  .toList
+    } yield expect(result.length == 1 && result.head == data)
+  }
+
+  test("decoding from null at udt column should raise Error for non Option type") { session =>
+    val data = PersonAttributeOpt(PersonAttribute.idxCounter.incrementAndGet(), None)
+
+    for {
+      _      <- cql"INSERT INTO cassandra4io.person_attributes (person_id, info) VALUES (${data.personId}, ${data.info})"
+                  .execute(session)
+      result <- cql"SELECT person_id, info FROM cassandra4io.person_attributes WHERE person_id = ${data.personId}"
+                  .as[PersonAttribute]
+                  .selectFirst(session)
+                  .attempt
+    } yield expect(result.isLeft) && expect(getError(result).isInstanceOf[UnexpectedNullValue])
+  }
+
+  // handle NULL inside udt
+
+  test("decoding from null at udt field should return None for Option type") { session =>
+    val data = PersonAttributeUdtOpt(
+      PersonAttribute.idxCounter.incrementAndGet(),
+      OptBasicInfo(None, None, None)
+    )
+
+    for {
+      _      <- cql"INSERT INTO cassandra4io.person_attributes (person_id, info) VALUES (${data.personId}, ${data.info})"
+                  .execute(session)
+      result <- cql"SELECT person_id, info FROM cassandra4io.person_attributes WHERE person_id = ${data.personId}"
+                  .as[PersonAttributeUdtOpt]
+                  .selectFirst(session)
+    } yield expect(result.contains(data))
+  }
+
+  test("decoding from null at udt field should raise error for String(non-primitive)") { session =>
+    val data =
+      PersonAttributeUdtOpt(
+        PersonAttribute.idxCounter.incrementAndGet(),
+        OptBasicInfo(Some(160.0), None, Some(Set(1)))
+      )
+    for {
+      _      <-
+        cql"INSERT INTO cassandra4io.person_attributes (person_id, info) VALUES (${data.personId}, ${data.info})"
+          .execute(session)
+      result <-
+        cql"SELECT person_id, info FROM cassandra4io.person_attributes WHERE person_id = ${data.personId}"
+          .as[PersonAttribute]
+          .selectFirst(session)
+          .attempt
+    } yield expect(result.isLeft) && expect(getError(result).isInstanceOf[UnexpectedNullValue])
+  }
+
+  test("decoding from null at udt field should raise error for Double(primitive)") { session =>
+    val data =
+      PersonAttributeUdtOpt(
+        PersonAttribute.idxCounter.incrementAndGet(),
+        OptBasicInfo(None, Some("tall"), Some(Set(1)))
+      )
+    for {
+      _      <-
+        cql"INSERT INTO cassandra4io.person_attributes (person_id, info) VALUES (${data.personId}, ${data.info})"
+          .execute(session)
+      result <-
+        cql"SELECT person_id, info FROM cassandra4io.person_attributes WHERE person_id = ${data.personId}"
+          .as[PersonAttribute]
+          .selectFirst(session)
+          .attempt
+    } yield expect(result.isLeft) && expect(getError(result).isInstanceOf[UnexpectedNullValue])
+  }
+
+  test("decoding from null at udt field should raise error for Set(collection)") { session =>
+    val data =
+      PersonAttributeUdtOpt(
+        PersonAttribute.idxCounter.incrementAndGet(),
+        OptBasicInfo(Some(180.0), Some("tall"), None)
+      )
+    for {
+      _      <-
+        cql"INSERT INTO cassandra4io.person_attributes (person_id, info) VALUES (${data.personId}, ${data.info})"
+          .execute(session)
+      result <-
+        cql"SELECT person_id, info FROM cassandra4io.person_attributes WHERE person_id = ${data.personId}"
+          .as[PersonAttribute]
+          .selectFirst(session)
+          .attempt
+    } yield expect(result.isLeft) && expect(getError(result).isInstanceOf[UnexpectedNullValue])
+  }
+
 }