Extract type information constants

Author: arnaud-daroussin

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

@@ -0,0 +1,16 @@
+package org.apache.flinkx
+
+import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
+
+package object api {
+
+  /** Basic type has an arity of 1. See [[BasicTypeInfo#getArity()]] */
+  private[api] val BasicTypeArity: Int = 1
+
+  /** Basic type has 1 field. See [[BasicTypeInfo#getTotalFields()]] */
+  private[api] val BasicTypeTotalFields: Int = 1
+
+  /** Documentation of [[TypeInformation#getTotalFields()]] states the total number of fields must be at least 1. */
+  private[api] val MinimumTotalFields: Int = 1
+
+}

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

@@ -30,6 +30,7 @@ import org.apache.flink.api.common.typeutils.CompositeType.{
 }
 import org.apache.flink.api.common.typeutils._
 import org.apache.flink.api.java.typeutils.TupleTypeInfoBase
+import org.apache.flinkx.api.MinimumTotalFields
 
 import java.util.regex.{Matcher, Pattern}
 import scala.annotation.tailrec
@@ -64,7 +65,7 @@ class CaseClassTypeInfo[T <: Product](
   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.
+    if (super.getTotalFields == 0) MinimumTotalFields else super.getTotalFields
 
   @PublicEvolving
   def getFieldIndices(fields: Array[String]): Array[Int] = {

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

@@ -4,6 +4,7 @@ import org.apache.flink.api.common.ExecutionConfig
 import org.apache.flink.api.common.serialization.SerializerConfig
 import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.common.typeutils.TypeSerializer
+import org.apache.flinkx.api.MinimumTotalFields
 
 import scala.reflect.{ClassTag, classTag}
 
@@ -17,7 +18,7 @@ case class CollectionTypeInformation[T: ClassTag](serializer: TypeSerializer[T])
   override def isBasicType: Boolean   = false
   override def isTupleType: Boolean   = false
   override def isKeyType: Boolean     = false
-  override def getTotalFields: Int    = 1
+  override def getTotalFields: Int    = MinimumTotalFields
   override def getTypeClass: Class[T] = clazz
   override def getArity: Int          = 0
 }

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

@@ -4,6 +4,7 @@ import org.apache.flink.api.common.ExecutionConfig
 import org.apache.flink.api.common.serialization.SerializerConfig
 import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.common.typeutils.TypeSerializer
+import org.apache.flinkx.api.MinimumTotalFields
 
 case class CoproductTypeInformation[T](c: Class[T], ser: TypeSerializer[T]) extends TypeInformation[T] {
 
@@ -14,7 +15,7 @@ case class CoproductTypeInformation[T](c: Class[T], ser: TypeSerializer[T]) exte
   override def isBasicType: Boolean   = false
   override def isTupleType: Boolean   = false
   override def isKeyType: Boolean     = false
-  override def getTotalFields: Int    = 1
+  override def getTotalFields: Int    = MinimumTotalFields
   override def getTypeClass: Class[T] = c
   override def getArity: Int          = 0
 }

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

@@ -5,6 +5,7 @@ import org.apache.flink.api.common.serialization.SerializerConfig
 import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.common.typeutils.TypeSerializer
 import org.apache.flinkx.api.serializer.UnitSerializer
+import org.apache.flinkx.api.{BasicTypeArity, BasicTypeTotalFields}
 
 class UnitTypeInformation extends TypeInformation[Unit] {
 
@@ -13,11 +14,11 @@ class UnitTypeInformation extends TypeInformation[Unit] {
   def createSerializer(config: ExecutionConfig): TypeSerializer[Unit] = new UnitSerializer()
 
   override def isKeyType: Boolean          = false
-  override def getTotalFields: Int         = 1 // The total number of fields must be at least 1.
+  override def getTotalFields: Int         = BasicTypeTotalFields
   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               = 1 // Basic type has an arity of 1
+  override def getArity: Int               = BasicTypeArity
   override def isBasicType: Boolean        = true
 
   override def toString: String = "{}"