[FLINK-10116] [types] Fix getTotalFields() implementation of some Typ…

…eInfos.

This closes apache#6569.

flink-core/src/main/java/org/apache/flink/api/common/typeinfo/NothingTypeInfo.java

@@ -53,7 +53,7 @@ public int getArity() {
  @Override
  @PublicEvolving
  public int getTotalFields() {
- return 0;
+ return 1;
  }
 
  @Override

flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java

@@ -111,6 +111,7 @@ public abstract class TypeInformation<T> implements Serializable {
  * fields, in the case of composite types. In the example below, the OuterType type has three
  * fields in total.
  *
+ * <p>The total number of fields must be at least 1.
  *
  * @return The number of fields in this type, including its sub-fields (for composite types)
  */

flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeInformationTestBase.java

@@ -95,6 +95,16 @@ public void testSerialization() {
  }
  }
 
+ @Test
+ public void testGetTotalFields() {
+ final T[] testData = getTestData();
+ for (T typeInfo : testData) {
+ assertTrue(
+ "Number of total fields must be at least 1",
+ typeInfo.getTotalFields() > 0);
+ }
+ }
+
  private static class UnrelatedTypeInfo extends TypeInformation<Object> {
 
  @Override

flink-core/src/test/java/org/apache/flink/api/java/typeutils/MissingTypeInfoTest.java

@@ -41,4 +41,9 @@ protected MissingTypeInfo[] getTestData() {
  public void testSerialization() {
  // this class is not intended to be serialized
  }
+
+ @Override
+ public void testGetTotalFields() {
+ // getTotalFields is not meant to be called
+ }
 }

flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeInfoFactoryTest.java

@@ -387,7 +387,7 @@ public int getArity() {
 
  @Override
  public int getTotalFields() {
- return 0;
+ return 1;
  }
 
  @Override

flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/VoidNamespaceTypeInfo.java

@@ -52,7 +52,7 @@ public int getArity() {
 
  @Override
  public int getTotalFields() {
- return 0;
+ return 1;
  }
 
  @Override

flink-runtime/src/main/java/org/apache/flink/runtime/state/VoidNamespaceTypeInfo.java

@@ -55,7 +55,7 @@ public int getArity() {
  @Override
  @PublicEvolving
  public int getTotalFields() {
- return 0;
+ return 1;
  }
 
  @Override

flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/ScalaNothingTypeInfo.scala

@@ -32,7 +32,7 @@ class ScalaNothingTypeInfo extends TypeInformation[Nothing] {
  @PublicEvolving
  override def getArity: Int = 0
  @PublicEvolving
- override def getTotalFields: Int = 0
+ override def getTotalFields: Int = 1
  @PublicEvolving
  override def getTypeClass: Class[Nothing] = classOf[Nothing]
  @PublicEvolving

flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/UnitTypeInfo.scala

@@ -31,7 +31,7 @@ class UnitTypeInfo extends TypeInformation[Unit] {
  @PublicEvolving
  override def getArity(): Int = 0
  @PublicEvolving
- override def getTotalFields(): Int = 0
+ override def getTotalFields(): Int = 1
  @PublicEvolving
  override def getTypeClass(): Class[Unit] = classOf[Unit]
  @PublicEvolving