Don't cache nested classes with generic type parameters (#243)

* Don't cache nested classes with generic type parameters

* Don't use debug variable in tests

Author: DSlug

modules/flink-1-api/src/test/scala-3/org/apache/flinkx/api/GenericCaseClassScala3Test.scala

@@ -1,8 +1,9 @@
 package org.apache.flinkx.api
 
 import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flinkx.api.serializer.{CoproductSerializer, ListCCSerializer, ScalaCaseClassSerializer}
 import org.apache.flinkx.api.serializers.*
-import org.apache.flinkx.api.typeinfo.ProductTypeInformation
+import org.apache.flinkx.api.typeinfo.{CoproductTypeInformation, ProductTypeInformation}
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should
 
@@ -45,6 +46,28 @@ class GenericCaseClassScala3Test extends AnyFlatSpec with should.Matchers {
     aBasketInfo.asInstanceOf[ProductTypeInformation[A]].getFieldTypes()(0) should be theSameInstanceAs aInfo
   }
 
+  "Nested generics" should "be resolved correctly" in {
+    val intTypeInfo: TypeInformation[Option[Option[Int]]] = generateTypeInfo[Int]
+    val stringTypeInfo: TypeInformation[Option[Option[String]]] = generateTypeInfo[String]
+
+    intTypeInfo shouldNot be theSameInstanceAs stringTypeInfo
+  }
+
+  it should "work with multiple type parameters" in {
+    val intTypeInfo  = generateEitherTypeInfo[Int]
+    val boolTypeInfo = generateEitherTypeInfo[Boolean]
+
+    intTypeInfo shouldNot be theSameInstanceAs boolTypeInfo
+
+  }
+
+  def generateTypeInfo[A: TypeInformation]: TypeInformation[Option[Option[A]]] = {
+    deriveTypeInformation
+  }
+
+  def generateEitherTypeInfo[A: TypeInformation]: TypeInformation[Either[Option[A], Int]] = {
+    deriveTypeInformation
+  }
 }
 
 object GenericCaseClassScala3Test {

modules/flink-common-api/src/main/scala-3/org/apache/flinkx/api/TypeTagMacro.scala

@@ -7,15 +7,19 @@ object TypeTagMacro:
   def gen[A: Type](using q: Quotes): Expr[TypeTag[A]] =
     import q.reflect.*
 
-    val A            = TypeRepr.of[A]
-    val symA         = A.typeSymbol
-    val flagsA       = symA.flags
-    val isModuleExpr = Expr(flagsA.is(Flags.Module))
-    val isCachableExpr = Expr(A match {
-      // this type is not cachable if one of its type args is abstract
-      case a: AppliedType => !a.args.exists { t => t.typeSymbol.isAbstractType }
-      case _              => true
-    })
+    def check(r: TypeRepr): Boolean =
+      r match {
+        case a: AppliedType =>
+          !a.args.exists { t => t.typeSymbol.isAbstractType } && a.args.forall { t => check(t) }
+        case _ => true
+      }
+
+    val A              = TypeRepr.of[A]
+    val symA           = A.typeSymbol
+    val flagsA         = symA.flags
+    val isModuleExpr   = Expr(flagsA.is(Flags.Module))
+    val isCachableExpr = Expr(check(A))
+
     val toStringExpr = Expr(A.show)
 
     '{