BufferUnderflowException for altered custom type in set

[Matzz]

Hi,
I have two interesting and related issues.

Issue 1:

We have a table with a custom type in a set column. This custom type was altered. One optional column was added. Cqlsh and native Cassandra driver works fine in that case. Result sets for rows inserted before ALTER TYPE returns null for a new column.
However, phantom throws BufferUnderflowException for such rows. Rows inserted after ALTER TYPE works correctly.

Stack trace:

com.datastax.driver.core.exceptions.InvalidTypeException: Not enough bytes to deserialize a tuple
at xyz.Entities$$anon$1.deserialize(Test.scala:17)
at xyz.Entities$$anon$1.deserialize(Test.scala:17)
at com.outworkers.phantom.builder.primitives.Primitive$$anon$6.deserialize(Primitive.scala:651)
at com.outworkers.phantom.builder.primitives.Primitives$$anon$1.$anonfun$deserialize$1(Primitives.scala:155)
at com.outworkers.phantom.builder.primitives.Primitives$$anon$1.$anonfun$deserialize$1$adapted(Primitives.scala:153)
at scala.collection.immutable.Range.foreach(Range.scala:158)
at com.outworkers.phantom.builder.primitives.Primitives$$anon$1.deserialize(Primitives.scala:153)
at com.outworkers.phantom.builder.primitives.Primitives$$anon$1.deserialize(Primitives.scala:120)
at com.outworkers.phantom.builder.primitives.Primitive.$anonfun$fromRow$1(Primitive.scala:131)
at com.outworkers.phantom.builder.primitives.Primitive.$anonfun$nullCheck$4(Primitive.scala:105)
at scala.util.Try$.apply(Try.scala:213)
at com.outworkers.phantom.builder.primitives.Primitive.nullCheck(Primitive.scala:105)
at com.outworkers.phantom.builder.primitives.Primitive.fromRow(Primitive.scala:131)
at com.outworkers.phantom.column.PrimitiveColumn.parse(PrimitiveColumn.scala:38)
at com.outworkers.phantom.column.Column.apply(Column.scala:30)
at xyz.Entities$anon$macro$1$1.fromRow(Test.scala:22)
at xyz.Entities$anon$macro$1$1.fromRow(Test.scala:22)
at com.outworkers.phantom.CassandraTable.fromRow(CassandraTable.scala:65)
at com.outworkers.phantom.SelectTable.$anonfun$select$1(SelectTable.scala:26)
at com.outworkers.phantom.builder.query.SelectQuery.fromRow(SelectQuery.scala:58)
at com.outworkers.phantom.ops.SelectQueryOps.fromRow(SelectQueryOps.scala:100)
at com.outworkers.phantom.builder.query.execution.ResultQueryInterface.$anonfun$singleResult$1(ResultQueryInterface.scala:34)
at scala.Option.map(Option.scala:230)
at com.outworkers.phantom.builder.query.execution.ResultQueryInterface.singleResult(ResultQueryInterface.scala:34)
at com.outworkers.phantom.builder.query.execution.ResultQueryInterface.$anonfun$singleFetch$1(ResultQueryInterface.scala:57)
at scala.util.Success.$anonfun$map$1(Try.scala:255)
at scala.util.Success.map(Try.scala:213)
at scala.concurrent.Future.$anonfun$map$1(Future.scala:292)
at scala.concurrent.impl.Promise.liftedTree1$1(Promise.scala:33)
at scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:33)
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Cause: java.nio.BufferUnderflowException:
at java.nio.Buffer.nextGetIndex(Buffer.java:506)
at java.nio.HeapByteBuffer.getInt(HeapByteBuffer.java:361)
at xyz.Entities$$anon$1.$anonfun$deserialize$1(Test.scala:17)
at scala.Option.flatMap(Option.scala:271)
at xyz.Entities$$anon$1.deserialize(Test.scala:17)
at xyz.Entities$$anon$1.deserialize(Test.scala:17)
at com.outworkers.phantom.builder.primitives.Primitive$$anon$6.deserialize(Primitive.scala:651)
at com.outworkers.phantom.builder.primitives.Primitives$$anon$1.$anonfun$deserialize$1(Primitives.scala:155)
at com.outworkers.phantom.builder.primitives.Primitives$$anon$1.$anonfun$deserialize$1$adapted(Primitives.scala:153)
at scala.collection.immutable.Range.foreach(Range.scala:158)
at com.outworkers.phantom.builder.primitives.Primitives$$anon$1.deserialize(Primitives.scala:153)
at com.outworkers.phantom.builder.primitives.Primitives$$anon$1.deserialize(Primitives.scala:120)
at com.outworkers.phantom.builder.primitives.Primitive.$anonfun$fromRow$1(Primitive.scala:131)
at com.outworkers.phantom.builder.primitives.Primitive.$anonfun$nullCheck$4(Primitive.scala:105)
at scala.util.Try$.apply(Try.scala:213)
at com.outworkers.phantom.builder.primitives.Primitive.nullCheck(Primitive.scala:105)
at com.outworkers.phantom.builder.primitives.Primitive.fromRow(Primitive.scala:131)
at com.outworkers.phantom.column.PrimitiveColumn.parse(PrimitiveColumn.scala:38)
at com.outworkers.phantom.column.Column.apply(Column.scala:30)
at xyz.Entities$anon$macro$1$1.fromRow(Test.scala:22)
at xyz.Entities$anon$macro$1$1.fromRow(Test.scala:22)
at com.outworkers.phantom.CassandraTable.fromRow(CassandraTable.scala:65)
at com.outworkers.phantom.SelectTable.$anonfun$select$1(SelectTable.scala:26)
at com.outworkers.phantom.builder.query.SelectQuery.fromRow(SelectQuery.scala:58)
at com.outworkers.phantom.ops.SelectQueryOps.fromRow(SelectQueryOps.scala:100)
at com.outworkers.phantom.builder.query.execution.ResultQueryInterface.$anonfun$singleResult$1(ResultQueryInterface.scala:34)
at scala.Option.map(Option.scala:230)
at com.outworkers.phantom.builder.query.execution.ResultQueryInterface.singleResult(ResultQueryInterface.scala:34)
at com.outworkers.phantom.builder.query.execution.ResultQueryInterface.$anonfun$singleFetch$1(ResultQueryInterface.scala:57)
at scala.util.Success.$anonfun$map$1(Try.scala:255)
at scala.util.Success.map(Try.scala:213)
at scala.concurrent.Future.$anonfun$map$1(Future.scala:292)
at scala.concurrent.impl.Promise.liftedTree1$1(Promise.scala:33)
at scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:33)
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

Here is the min example which throws above exception

package com.onzo.cassandra.repository

import com.onzo.cassandra.CassandraSupport
import com.outworkers.phantom.connectors.ContactPoint
import com.outworkers.phantom.dsl._
import org.scalatest.WordSpec

import scala.collection.JavaConverters._
import scala.concurrent.Await
import scala.concurrent.duration._

case class SomeCustomType(a: String, b: Option[String])
case class Entity(id: String, value: String, value2: Set[SomeCustomType])

object Entities {
  implicit val customTypePrimitive = Primitive.derive[SomeCustomType, (String, Option[String])](
    c => (c.a, c.b))(
    SomeCustomType.tupled
  )
}
import Entities._
abstract class Entities extends Table[Entities, Entity] {
  object id extends Col[String] with PartitionKey
  object value extends Col[String]
  object value2 extends Col[Set[SomeCustomType]]
}

class MyDb(override val connector: CassandraConnection) extends Database[MyDb](connector) {
  object entities extends Entities with Connector

  def firstRow() = {
    entities
      .select
      .one()
  }
}

class Test
    extends WordSpec
    with CassandraSupport {
  "UDT" should {
    "work" in {
      val connection = ContactPoint.embedded.noHeartbeat.keySpace(keyspace)
      val session = connection.session

      // Initial type and table definition
      session.execute(s"CREATE TYPE ${keyspace}.some_custom_type (a text);")
      session.execute(s"CREATE TABLE ${keyspace}.entities ( id text PRIMARY KEY, value text, value2 set<frozen<some_custom_type>>);")
      // Insert before modifying some_custom_type
      session.execute(s"INSERT INTO ${keyspace}.entities (id, value, value2) VALUES('test', 'abc', {{a: 'AAA'}});")
      // Add column to some_custom_type
      session.execute(s"ALTER TYPE ${keyspace}.some_custom_type ADD b text;")

      //Query row using raw cassandra driver
      val rawRow = session.execute(s"select * from ${keyspace}.entities limit 1").asScala.head
      println(s"Raw row: $rawRow")
      //Prints - Raw row: Row[test, abc, [{a:'AAA',b:NULL}]]

      val myDb = new MyDb(connection)
      val phantomRow = Await.result(myDb.firstRow(), 5.seconds)
      //Below fails if insert is before alter type but works if insert is after alter type
      println(s"Phantom row: $phantomRow")
      // Expected result - Phantom row: Some(Entity(test,abc,Set(SomeCustomType(AAA,None))))
    }
  }

}

Issue 2.

When I change Col[Set[SomeCustomType]] to SetColumn[SomeCustomType] I don't even get a exception. Instead of that phantom produce an empty Set (Phantom row: Some(Entity(test,abc,Set()))). It is very dangerous because due to lack of proper exception, you may not notice any problems on production for months. While the returned data is incorrect.