Add logical type annotation for UnknownType

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

@@ -146,6 +146,12 @@ protected LogicalTypeAnnotation fromString(List<String> params) {
       protected LogicalTypeAnnotation fromString(List<String> params) {
         return float16Type();
       }
+    },
+    UNKNOWN {
+      @Override
+      protected LogicalTypeAnnotation fromString(List<String> params) {
+        return unknownType();
+      }
     };
 
     protected abstract LogicalTypeAnnotation fromString(List<String> params);
@@ -316,6 +322,10 @@ public static Float16LogicalTypeAnnotation float16Type() {
     return Float16LogicalTypeAnnotation.INSTANCE;
   }
 
+  public static UnknownLogicalTypeAnnotation unknownType() {
+    return UnknownLogicalTypeAnnotation.INSTANCE;
+  }
+
   public static class StringLogicalTypeAnnotation extends LogicalTypeAnnotation {
     private static final StringLogicalTypeAnnotation INSTANCE = new StringLogicalTypeAnnotation();
 
@@ -989,6 +999,33 @@ PrimitiveStringifier valueStringifier(PrimitiveType primitiveType) {
     }
   }
 
+  public static class UnknownLogicalTypeAnnotation extends LogicalTypeAnnotation {
+    private static final UnknownLogicalTypeAnnotation INSTANCE = new UnknownLogicalTypeAnnotation();
+
+    private UnknownLogicalTypeAnnotation() {}
+
+    @Override
+    public OriginalType toOriginalType() {
+      // No OriginalType for UknownType
+      return null;
+    }
+
+    @Override
+    public <T> Optional<T> accept(LogicalTypeAnnotationVisitor<T> logicalTypeAnnotationVisitor) {
+      return logicalTypeAnnotationVisitor.visit(this);
+    }
+
+    @Override
+    LogicalTypeToken getType() {
+      return LogicalTypeToken.UNKNOWN;
+    }
+
+    @Override
+    PrimitiveStringifier valueStringifier(PrimitiveType primitiveType) {
+      return PrimitiveStringifier.UNKNOWN_STRINGIFIER;
+    }
+  }
+
   // This logical type annotation is implemented to support backward compatibility with ConvertedType.
   // The new logical type representation in parquet-format doesn't have any interval type,
   // thus this annotation is mapped to UNKNOWN.
@@ -1162,5 +1199,9 @@ default Optional<T> visit(MapKeyValueTypeAnnotation mapKeyValueLogicalType) {
     default Optional<T> visit(Float16LogicalTypeAnnotation float16LogicalType) {
       return empty();
     }
+
+    default Optional<T> visit(UnknownLogicalTypeAnnotation unknownLogicalTypeAnnotation) {
+      return empty();
+    }
   }
 }

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

@@ -449,4 +449,11 @@ String stringifyNotNull(Binary value) {
       return Float16.toFloatString(value);
     }
   };
+
+  static final PrimitiveStringifier UNKNOWN_STRINGIFIER = new PrimitiveStringifier("UNKNOWN_STRINGIFIER") {
+
+    public String stringify(Binary ignored) {
+      return "UNKNOWN";
+    }
+  };
 }

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

@@ -472,6 +472,12 @@ public Optional<Boolean> visit(
                     LogicalTypeAnnotation.Float16LogicalTypeAnnotation.BYTES, float16LogicalType);
               }
 
+              @Override
+              public Optional<Boolean> visit(
+                  LogicalTypeAnnotation.UnknownLogicalTypeAnnotation unknownLogicalType) {
+                return Optional.of(true);
+              }
+
               @Override
               public Optional<Boolean> visit(
                   LogicalTypeAnnotation.DecimalLogicalTypeAnnotation decimalLogicalType) {

parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java

@@ -515,6 +515,11 @@ public Optional<LogicalType> visit(LogicalTypeAnnotation.Float16LogicalTypeAnnot
       return of(LogicalType.FLOAT16(new Float16Type()));
     }
 
+    @Override
+    public Optional<LogicalType> visit(LogicalTypeAnnotation.UnknownLogicalTypeAnnotation intervalLogicalType) {
+      return of(LogicalType.UNKNOWN(new NullType()));
+    }
+
     @Override
     public Optional<LogicalType> visit(LogicalTypeAnnotation.IntervalLogicalTypeAnnotation intervalLogicalType) {
       return of(LogicalType.UNKNOWN(new NullType()));
@@ -894,7 +899,8 @@ enum SortOrder {
       LogicalTypeAnnotation.StringLogicalTypeAnnotation.class,
       LogicalTypeAnnotation.EnumLogicalTypeAnnotation.class,
       LogicalTypeAnnotation.JsonLogicalTypeAnnotation.class,
-      LogicalTypeAnnotation.Float16LogicalTypeAnnotation.class)));
+      LogicalTypeAnnotation.Float16LogicalTypeAnnotation.class,
+      LogicalTypeAnnotation.UnknownLogicalTypeAnnotation.class)));
 
   /**
    * Returns whether to use signed order min and max with a type. It is safe to
@@ -997,6 +1003,12 @@ public Optional<SortOrder> visit(
               return of(SortOrder.SIGNED);
             }
 
+            @Override
+            public Optional<SortOrder> visit(
+                LogicalTypeAnnotation.UnknownLogicalTypeAnnotation unknownLogicalTypeAnnotation) {
+              return of(SortOrder.UNKNOWN);
+            }
+
             @Override
             public Optional<SortOrder> visit(
                 LogicalTypeAnnotation.DecimalLogicalTypeAnnotation decimalLogicalType) {
@@ -1167,7 +1179,7 @@ LogicalTypeAnnotation getLogicalTypeAnnotation(LogicalType type) {
         IntType integer = type.getINTEGER();
         return LogicalTypeAnnotation.intType(integer.bitWidth, integer.isSigned);
       case UNKNOWN:
-        return null;
+        return LogicalTypeAnnotation.unknownType();
       case TIMESTAMP:
         TimestampType timestamp = type.getTIMESTAMP();
         return LogicalTypeAnnotation.timestampType(timestamp.isAdjustedToUTC, convertTimeUnit(timestamp.unit));