review: Fix checks on arity and totalFields

Author: arnaud-daroussin

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/typeinfo/CaseClassTypeInfo.scala

@@ -63,6 +63,9 @@ class CaseClassTypeInfo[T <: Product](
     Pattern.compile(REGEX_NESTED_FIELDS_WILDCARD)
   private val PATTERN_INT_FIELD: Pattern = Pattern.compile(REGEX_INT_FIELD)
 
+  override def getTotalFields: Int =
+    if (super.getTotalFields == 0) 1 else super.getTotalFields // The total number of fields must be at least 1.
+
   @PublicEvolving
   def getFieldIndices(fields: Array[String]): Array[Int] = {
     fields map { x => fieldNames.indexOf(x) }

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/typeinfo/CollectionTypeInformation.scala

@@ -19,5 +19,5 @@ case class CollectionTypeInformation[T: ClassTag](serializer: TypeSerializer[T])
   override def isKeyType: Boolean     = false
   override def getTotalFields: Int    = 1
   override def getTypeClass: Class[T] = clazz
-  override def getArity: Int          = 1
+  override def getArity: Int          = 0
 }

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/typeinfo/CoproductTypeInformation.scala

@@ -16,5 +16,5 @@ case class CoproductTypeInformation[T](c: Class[T], ser: TypeSerializer[T]) exte
   override def isKeyType: Boolean     = false
   override def getTotalFields: Int    = 1
   override def getTypeClass: Class[T] = c
-  override def getArity: Int          = 1
+  override def getArity: Int          = 0
 }

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/typeinfo/UnitTypeInformation.scala

@@ -13,11 +13,11 @@ class UnitTypeInformation extends TypeInformation[Unit] {
   def createSerializer(config: ExecutionConfig): TypeSerializer[Unit] = new UnitSerializer()
 
   override def isKeyType: Boolean          = false
-  override def getTotalFields: Int         = 0
+  override def getTotalFields: Int         = 1 // The total number of fields must be at least 1.
   override def isTupleType: Boolean        = false
   override def canEqual(obj: Any): Boolean = obj.isInstanceOf[UnitTypeInformation]
   override def getTypeClass: Class[Unit]   = classOf[Unit]
-  override def getArity: Int               = 0
+  override def getArity: Int               = 1 // Basic type has an arity of 1
   override def isBasicType: Boolean        = true
 
   override def toString: String = "{}"

modules/flink-common-api/src/test/scala/org/apache/flinkx/api/TestUtils.scala

@@ -1,6 +1,5 @@
 package org.apache.flinkx.api
 
-import org.apache.flink.api.common.serialization.SerializerConfigImpl
 import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
 import org.apache.flink.api.common.typeutils.{TypeSerializer, TypeSerializerSnapshot}
 import org.apache.flink.api.java.typeutils.TupleTypeInfoBase
@@ -154,9 +153,15 @@ trait TestUtils extends Matchers with Inspectors {
 
   private def isNotBitmapField(field: Field): Boolean = !field.getName.startsWith("bitmap$")
 
+  private def isNotOverriddenField(field: Field, childFields: Array[Field]): Boolean =
+    !childFields.exists(_.getName == field.getName)
+
   @tailrec
   private def getAllFields[T](typeClass: Class[T], fields: Array[Field] = Array.empty): Array[Field] = {
-    val allFields  = fields ++ typeClass.getDeclaredFields.filter(isInstanceField).filter(isNotBitmapField)
+    val allFields  = fields ++ typeClass.getDeclaredFields
+      .filter(isInstanceField)
+      .filter(isNotBitmapField)
+      .filter(isNotOverriddenField(_, fields))
     val superClass = typeClass.getSuperclass
     if (superClass == null || superClass == classOf[Object]) {
       allFields
@@ -195,9 +200,13 @@ trait TestUtils extends Matchers with Inspectors {
   }
 
   def checkTotalFields[T](info: TypeInformation[T]): Assertion = {
-    val typeClass   = info.getTypeClass
-    val totalFields = getTotalFields(typeClass)
-    withClue("getTotalFields:")(info.getTotalFields shouldBe totalFields)
+    val typeClass = info.getTypeClass
+    if (typeClass.getTypeParameters.isEmpty) {
+      val totalFields = getTotalFields(typeClass)
+      withClue("getTotalFields:")(info.getTotalFields shouldBe totalFields)
+    } else {
+      succeed // cannot check getTotalFields for generic types.
+    }
   }
 
   def checkKeyType[T](info: TypeInformation[T]): Assertion = info match {
@@ -230,8 +239,8 @@ trait TestUtils extends Matchers with Inspectors {
     }
     checkBasicType(infoToCheck)
     checkTupleType(infoToCheck)
-//    checkArity(infoToCheck)
-//    checkTotalFields(infoToCheck)
+    checkArity(infoToCheck)
+    checkTotalFields(infoToCheck)
     checkKeyType(infoToCheck)
     checkToString(infoToCheck)
     checkEquals(infoToCheck)