👋 Hello All, I have a "fun" Scala problem that maybe the community can help with. Essentially I am trying to have a DataStream support CoVariant types so that I can union traits. Ex.

@SerialVersionUID(1L)
abstract class Base {
  val base: Any
}

Generic Wrapper:

@SerialVersionUID(1L)
@TypeInfo(classOf[BaseTypeInfoFactory])
case class AggBase[+T <: Base](
  key: String,
  base: T
) {
  // Required to be a POJO for TypeInfo Serialization
  def this() = this("", null.asInstanceOf[T])
}

class BaseTypeInfoFactory extends TypeInfoFactory[Base[_]] {
  override def createTypeInfo(
    t: Type,
    genericParameters: java.util.Map[String, TypeInformation[_]]
  ): TypeInformation[Base[_]] = {
    val baseType = genericParameters.get("T")
    baseType match {
      case basex if basex.getTypeClass == classOf[BaseX] =>
        val fields = Map(
          "key" -> Types.STRING,
          "base" -> Types.POJO(classOf[BaseX])
        )
        Types.POJO(classOf[AggBase[_]], fields)
      // Add more cases for other supported types
      case _ => throw new Exception("Insight type not supported")
    }
  }
}

Then the datastream would union a bunch of these types

val baseDataStreams: List[DataStream[AggBase[_ <: Base]]] = List(
      basexDataStream,
      baseyDataStream
    )

This obviously doesn't work since DataStream isn't covariant. Thoughts on the best alternative way to approach this. (I've tried to make the Scala DataStream type covariant, that looks like a lift). There are a lot of classes that inherit "Base", so Either or some equivalent isn't ideal if it can be avoided.

Hi @Ryan van Huuksloot. Last time I used covariant types, it was long time ago. What do you want to achieve is to have some common type of the elements in the DataStream? Perhaps some ADT approach like here: https://github.com/flink-extended/flink-scala-api#flink-adt ?

👋 Hi Alexey, I'd not convinced the ADT would work but I would need to try it with the

deriveTypeInformation

. The TypeMapping also looks like it could be useful! I'll fiddle with it a bit today.

yeah, I am not sure if that can really help to achieve the desired logic. Please note, this is an additional library which comes with

deriveTypeInformation

function

No dice with the

deriveTypeInformation

. I did download the lib. That ADT requires you to cast to Event - no? Which means you lose the generic parameter during type casting that I need. The TypeInformation isn't the problem. I am curious more about the TypeMapping. Have you ever used it before? It looks like you can cast to a "generic" type to stay invariant but it holds the type that it should be.

No, I have not used TypeMapping before

Update: Logs - I needed to exclude slf4j Flink-ADT didn't really work because you still can't have a DataStream that has a Base class with multiple final case classes as the individual types. It is only really nice if you want to gen a bunch of TypeInformation for single streams