Introduce implementation for FileType

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

@@ -188,6 +188,12 @@ protected LogicalTypeAnnotation fromString(List<String> params) {
       protected LogicalTypeAnnotation fromString(List<String> params) {
         return unknownType();
       }
+    },
+    FILE {
+      @Override
+      protected LogicalTypeAnnotation fromString(List<String> params) {
+        return fileType();
+      }
     };
 
     protected abstract LogicalTypeAnnotation fromString(List<String> params);
@@ -378,6 +384,10 @@ public static UnknownLogicalTypeAnnotation unknownType() {
     return UnknownLogicalTypeAnnotation.INSTANCE;
   }
 
+  public static FileLogicalTypeAnnotation fileType() {
+    return FileLogicalTypeAnnotation.INSTANCE;
+  }
+
   public static class StringLogicalTypeAnnotation extends LogicalTypeAnnotation {
     private static final StringLogicalTypeAnnotation INSTANCE = new StringLogicalTypeAnnotation();
 
@@ -1229,6 +1239,93 @@ public boolean equals(Object obj) {
     }
   }
 
+  /**
+   * File logical type annotation. Annotates a group (struct) that represents a reference to a
+   * range of bytes, which may be stored inline in the value or in an external file. Every field is
+   * optional, both in the schema (a writer may omit any field from the group definition) and in the
+   * data (any field that is present has a field repetition type of {@code OPTIONAL}). Fields are
+   * identified by name (case sensitively), not by field order. A group need only define the fields
+   * it uses. The group may contain the following fields:
+   * <ul>
+   *   <li>{@code uri} (STRING): a URI-reference (RFC 3986) that identifies an external file, for
+   *       example {@code s3://bucket/file.jpg}.</li>
+   *   <li>{@code offset} (INT64): start of the byte range within the external file identified by
+   *       {@code uri}; if not set, treated as 0. Must not be negative and may only be set together
+   *       with {@code uri}.</li>
+   *   <li>{@code size} (INT64): byte length of the referenced data within the external file. May be
+   *       omitted for a whole-file external reference, in which case the range runs to the end of
+   *       the referenced file. Must be set whenever {@code offset} is set. Must not be negative.</li>
+   *   <li>{@code content_type} (STRING): the media (MIME) type (RFC 2046) of the resolved bytes;
+   *       when not set, {@code application/octet-stream} is assumed.</li>
+   *   <li>{@code checksum} (STRING): a self-describing integrity token for the resolved bytes, of
+   *       the form {@code <algorithm>:<digest>}.</li>
+   *   <li>{@code inline} (BYTE_ARRAY): the referenced bytes stored inline in the value.</li>
+   * </ul>
+   * No fields with names other than the above are permitted. The schema builder additionally
+   * rejects group definitions that could never produce a valid value: a group must declare at least
+   * one of {@code inline} or {@code uri} (a value resolves to bytes only via inline storage or an
+   * external reference, so a group declaring neither — even if it declares {@code offset} or
+   * {@code size} — can never produce a resolvable value), and a group that declares {@code offset}
+   * must also declare {@code uri} and {@code size} ({@code offset} locates a range within the
+   * external file identified by {@code uri} and is meaningless without it, and always bounds a range
+   * that requires {@code size}). Each declared field must also match its required physical type.
+   * Per-value rules that depend on the data in each row — {@code size} being set whenever
+   * {@code offset} is set, and {@code offset}/{@code size} being non-negative — cannot be enforced
+   * here and are the responsibility of writers and consumers.
+   */
+  public static class FileLogicalTypeAnnotation extends LogicalTypeAnnotation {
+    private static final FileLogicalTypeAnnotation INSTANCE = new FileLogicalTypeAnnotation();
+
+    /** Field name holding the URI-reference of an external file. */
+    public static final String URI_FIELD = "uri";
+
+    /** Field name holding the start of the byte range. */
+    public static final String OFFSET_FIELD = "offset";
+
+    /** Field name holding the byte length of the referenced data. */
+    public static final String SIZE_FIELD = "size";
+
+    /** Field name holding the media (MIME) type of the resolved bytes. */
+    public static final String CONTENT_TYPE_FIELD = "content_type";
+
+    /** Field name holding the integrity token for the resolved bytes. */
+    public static final String CHECKSUM_FIELD = "checksum";
+
+    /** Field name holding the referenced bytes stored inline. */
+    public static final String INLINE_FIELD = "inline";
+
+    /** All recognized field names in a FILE-annotated group. All fields are optional. */
+    public static final Set<String> FIELD_NAMES =
+        Set.of(URI_FIELD, OFFSET_FIELD, SIZE_FIELD, CONTENT_TYPE_FIELD, CHECKSUM_FIELD, INLINE_FIELD);
+
+    private FileLogicalTypeAnnotation() {}
+
+    @Override
+    public OriginalType toOriginalType() {
+      return null;
+    }
+
+    @Override
+    public <T> Optional<T> accept(LogicalTypeAnnotationVisitor<T> logicalTypeAnnotationVisitor) {
+      return logicalTypeAnnotationVisitor.visit(this);
+    }
+
+    @Override
+    LogicalTypeToken getType() {
+      return LogicalTypeToken.FILE;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      return obj instanceof FileLogicalTypeAnnotation;
+    }
+
+    @Override
+    public int hashCode() {
+      return getClass().hashCode();
+    }
+  }
+
   public static class GeometryLogicalTypeAnnotation extends LogicalTypeAnnotation {
     private final String crs;
 
@@ -1434,5 +1531,9 @@ default Optional<T> visit(GeographyLogicalTypeAnnotation geographyLogicalType) {
     default Optional<T> visit(UnknownLogicalTypeAnnotation unknownLogicalTypeAnnotation) {
       return empty();
     }
+
+    default Optional<T> visit(FileLogicalTypeAnnotation fileLogicalType) {
+      return empty();
+    }
   }
 }

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

@@ -821,12 +821,115 @@ public THIS addFields(Type... types) {
     @Override
     protected GroupType build(String name) {
       if (newLogicalTypeSet) {
+        if (logicalTypeAnnotation instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation) {
+          validateFileTypeFields(name, fields);
+        }
         return new GroupType(repetition, name, logicalTypeAnnotation, fields, id);
       } else {
         return new GroupType(repetition, name, getOriginalType(), fields, id);
       }
     }
 
+    private static void validateFileTypeFields(String name, List<Type> fields) {
+      boolean hasUri = false;
+      boolean hasOffset = false;
+      boolean hasSize = false;
+      boolean hasInline = false;
+      for (Type field : fields) {
+        String fieldName = field.getName();
+        if (!LogicalTypeAnnotation.FileLogicalTypeAnnotation.FIELD_NAMES.contains(fieldName)) {
+          throw new IllegalArgumentException("FILE type group '" + name + "' contains unrecognized field '"
+              + fieldName + "'. Valid fields are: "
+              + String.join(", ", LogicalTypeAnnotation.FileLogicalTypeAnnotation.FIELD_NAMES));
+        }
+        Preconditions.checkArgument(
+            field.isPrimitive() && field.getRepetition() == Type.Repetition.OPTIONAL,
+            "FILE type field '%s' must be an optional primitive in group '%s'",
+            fieldName,
+            name);
+        validateFileTypeFieldPhysicalType(name, field.asPrimitiveType());
+        if (LogicalTypeAnnotation.FileLogicalTypeAnnotation.URI_FIELD.equals(fieldName)) {
+          hasUri = true;
+        } else if (LogicalTypeAnnotation.FileLogicalTypeAnnotation.OFFSET_FIELD.equals(fieldName)) {
+          hasOffset = true;
+        } else if (LogicalTypeAnnotation.FileLogicalTypeAnnotation.SIZE_FIELD.equals(fieldName)) {
+          hasSize = true;
+        } else if (LogicalTypeAnnotation.FileLogicalTypeAnnotation.INLINE_FIELD.equals(fieldName)) {
+          hasInline = true;
+        }
+      }
+      // A value resolves to bytes only via inline storage or an external reference, so a group must
+      // declare at least one of `inline` or `uri`. A group declaring neither — even if it declares
+      // `offset` or `size` — can never produce a resolvable value, so reject it at schema-build
+      // time.
+      Preconditions.checkArgument(
+          hasInline || hasUri,
+          "FILE type group '%s' must declare at least one of 'inline' or 'uri'; a value resolves to "
+              + "bytes only via inline storage or an external reference, so a group declaring "
+              + "neither can never produce a valid value",
+          name);
+      // `offset` locates a byte range within the external file identified by `uri`, so it is
+      // meaningless without `uri`. A group that declares `offset` but not `uri` can never produce a
+      // valid value, so reject it at schema-build time.
+      Preconditions.checkArgument(
+          !hasOffset || hasUri,
+          "FILE type group '%s' declares field 'offset' but not 'uri'; 'offset' locates a range "
+              + "within an external file and may only be set together with 'uri'",
+          name);
+      // The spec requires `size` to be set whenever `offset` is set. A group that declares
+      // `offset` but not `size` can never produce a valid value, so reject it at schema-build
+      // time.
+      Preconditions.checkArgument(
+          !hasOffset || hasSize,
+          "FILE type group '%s' declares field 'offset' but not 'size'; 'size' is required whenever 'offset' is set",
+          name);
+      // The remaining spec rules are per-value constraints the schema builder cannot verify because
+      // it sees only which fields are declared, not their values in each row: `size` being set
+      // whenever `offset` is set, and `offset`/`size` being non-negative. Those are the
+      // responsibility of writers and consumers of FILE values.
+    }
+
+    /**
+     * Validates that a declared FILE field uses the physical type required by the spec:
+     * {@code uri}, {@code content_type}, and {@code checksum} are STRING (BINARY), {@code offset}
+     * and {@code size} are INT64, and {@code inline} is BYTE_ARRAY (BINARY).
+     */
+    private static void validateFileTypeFieldPhysicalType(String name, PrimitiveType field) {
+      String fieldName = field.getName();
+      PrimitiveType.PrimitiveTypeName physicalType = field.getPrimitiveTypeName();
+      switch (fieldName) {
+        case LogicalTypeAnnotation.FileLogicalTypeAnnotation.URI_FIELD:
+        case LogicalTypeAnnotation.FileLogicalTypeAnnotation.CONTENT_TYPE_FIELD:
+        case LogicalTypeAnnotation.FileLogicalTypeAnnotation.CHECKSUM_FIELD:
+          Preconditions.checkArgument(
+              physicalType == PrimitiveType.PrimitiveTypeName.BINARY
+                  && field.getLogicalTypeAnnotation()
+                      instanceof LogicalTypeAnnotation.StringLogicalTypeAnnotation,
+              "FILE type field '%s' must be a STRING (BINARY annotated as STRING) in group '%s'",
+              fieldName,
+              name);
+          break;
+        case LogicalTypeAnnotation.FileLogicalTypeAnnotation.OFFSET_FIELD:
+        case LogicalTypeAnnotation.FileLogicalTypeAnnotation.SIZE_FIELD:
+          Preconditions.checkArgument(
+              physicalType == PrimitiveType.PrimitiveTypeName.INT64,
+              "FILE type field '%s' must be an INT64 in group '%s'",
+              fieldName,
+              name);
+          break;
+        case LogicalTypeAnnotation.FileLogicalTypeAnnotation.INLINE_FIELD:
+          Preconditions.checkArgument(
+              physicalType == PrimitiveType.PrimitiveTypeName.BINARY,
+              "FILE type field '%s' must be a BYTE_ARRAY (BINARY) in group '%s'",
+              fieldName,
+              name);
+          break;
+        default:
+          // Unreachable: field names are validated against FIELD_NAMES before this call.
+          break;
+      }
+    }
+
     public MapBuilder<THIS> map(Type.Repetition repetition) {
       return new MapBuilder<>(self()).repetition(repetition);
     }