Add comparator for UnknownLogicalType (#3292) (#3295)

This was missing, which caused exceptions. This PR adds them
including a test.

It takes the comparator of the underlying physical type, but it
should always compare two null values:

https://github.com/apache/parquet-format/blob/471d33a3c9d46e05b8a189437a9364f959f93f09/LogicalTypes.md?plain=1#L887-L892

Co-authored-by: Fokko Driesprong <fokko@apache.org>

Author: wgtmac

parquet-column/src/main/java/org/apache/parquet/schema/PrimitiveType.java

@@ -119,6 +119,12 @@ public Optional<PrimitiveComparator> visit(
                   LogicalTypeAnnotation.TimestampLogicalTypeAnnotation timestampLogicalType) {
                 return of(PrimitiveComparator.SIGNED_INT64_COMPARATOR);
               }
+
+              @Override
+              public Optional<PrimitiveComparator> visit(
+                  LogicalTypeAnnotation.UnknownLogicalTypeAnnotation unknownLogicalTypeAnnotation) {
+                return of(PrimitiveComparator.SIGNED_INT64_COMPARATOR);
+              }
             })
             .orElseThrow(() -> new ShouldNeverHappenException(
                 "No comparator logic implemented for INT64 logical type: " + logicalType));
@@ -183,6 +189,12 @@ public Optional<PrimitiveComparator> visit(
                 }
                 return empty();
               }
+
+              @Override
+              public Optional<PrimitiveComparator> visit(
+                  LogicalTypeAnnotation.UnknownLogicalTypeAnnotation unknownLogicalTypeAnnotation) {
+                return of(PrimitiveComparator.SIGNED_INT32_COMPARATOR);
+              }
             })
             .orElseThrow(() -> new ShouldNeverHappenException(
                 "No comparator logic implemented for INT32 logical type: " + logicalType));
@@ -283,6 +295,12 @@ public Optional<PrimitiveComparator> visit(
                   LogicalTypeAnnotation.GeographyLogicalTypeAnnotation geographyLogicalType) {
                 return of(PrimitiveComparator.UNSIGNED_LEXICOGRAPHICAL_BINARY_COMPARATOR);
               }
+
+              @Override
+              public Optional<PrimitiveComparator> visit(
+                  LogicalTypeAnnotation.UnknownLogicalTypeAnnotation unknownLogicalTypeAnnotation) {
+                return of(PrimitiveComparator.UNSIGNED_LEXICOGRAPHICAL_BINARY_COMPARATOR);
+              }
             })
             .orElseThrow(() -> new ShouldNeverHappenException(
                 "No comparator logic implemented for BINARY logical type: " + logicalType));
@@ -417,6 +435,12 @@ public Optional<PrimitiveComparator> visit(
                   LogicalTypeAnnotation.Float16LogicalTypeAnnotation float16LogicalType) {
                 return of(PrimitiveComparator.BINARY_AS_FLOAT16_COMPARATOR);
               }
+
+              @Override
+              public Optional<PrimitiveComparator> visit(
+                  LogicalTypeAnnotation.UnknownLogicalTypeAnnotation unknownLogicalTypeAnnotation) {
+                return of(PrimitiveComparator.UNSIGNED_LEXICOGRAPHICAL_BINARY_COMPARATOR);
+              }
             })
             .orElseThrow(() -> new ShouldNeverHappenException(
                 "No comparator logic implemented for FIXED_LEN_BYTE_ARRAY logical type: "

parquet-column/src/test/java/org/apache/parquet/schema/TestPrimitiveComparator.java

@@ -100,6 +100,29 @@ private void testInt32Comparator(PrimitiveComparator<Integer> comparator, Intege
     checkThrowingUnsupportedException(comparator, Integer.TYPE);
   }
 
+  @Test
+  public void testUnknownLogicalTypeComparator() {
+    PrimitiveType.PrimitiveTypeName[] types = new PrimitiveType.PrimitiveTypeName[] {
+      PrimitiveType.PrimitiveTypeName.BOOLEAN,
+      PrimitiveType.PrimitiveTypeName.BINARY,
+      PrimitiveType.PrimitiveTypeName.INT32,
+      PrimitiveType.PrimitiveTypeName.INT64,
+      PrimitiveType.PrimitiveTypeName.FLOAT,
+      PrimitiveType.PrimitiveTypeName.DOUBLE,
+      PrimitiveType.PrimitiveTypeName.INT96,
+      PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY
+    };
+
+    for (PrimitiveType.PrimitiveTypeName type : types) {
+      assertEquals(
+          new PrimitiveType(Type.Repetition.REQUIRED, type, "vo")
+              .withLogicalTypeAnnotation(LogicalTypeAnnotation.unknownType())
+              .comparator()
+              .compare(null, null),
+          0);
+    }
+  }
+
   @Test
   public void testSignedInt64Comparator() {
     testInt64Comparator(