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,89 @@ 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, elsewhere within the current file,
+   * 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}). The group may contain the following fields, identified
+   * by name:
+   * <ul>
+   *   <li>{@code path} (STRING): an opaque path that identifies an external file, for example a
+   *       URI such as s3://bucket/key. If not set, the value refers to the current file (a
+   *       self-reference).</li>
+   *   <li>{@code offset} (INT64): start of the byte range within the referenced data; if not set,
+   *       treated as 0.</li>
+   *   <li>{@code size} (INT64): byte length of the referenced data. Must be set whenever
+   *       {@code offset} is set or {@code path} is not set; may be omitted only for a whole-file
+   *       external reference, in which case the range runs to the end of the referenced file.</li>
+   *   <li>{@code content_type} (STRING): the media (MIME) type of the resolved bytes.</li>
+   *   <li>{@code checksum} (STRING): an algorithm-tagged integrity token for the resolved bytes,
+   *       of the form {@code <algorithm>:base64(<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 that declares
+   * {@code offset} must also declare {@code size}, and a group must declare at least one of
+   * {@code inline}, {@code path}, or {@code size} (a group without {@code path} or {@code inline}
+   * holds only self-references, which require {@code size}). Per-value rules that depend on the
+   * data in each row — {@code size} being present for a self-reference (null {@code path}) 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 path/URI of an external file. */
+    public static final String PATH_FIELD = "path";
+
+    /** 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(
+        PATH_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 +1527,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,65 @@ 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 hasPath = 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);
+        if (LogicalTypeAnnotation.FileLogicalTypeAnnotation.PATH_FIELD.equals(fieldName)) {
+          hasPath = 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;
+        }
+      }
+      // 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 spec requires `size` to be set whenever `path` is not set (a self-reference). A group
+      // that declares neither `path` nor `inline` can only hold self-references, so it must
+      // declare `size`. More generally, a value can only resolve to bytes via `inline`, `path`,
+      // or `size`, so a group that declares none of these can never produce a valid value.
+      Preconditions.checkArgument(
+          hasInline || hasPath || hasSize,
+          "FILE type group '%s' must declare at least one of 'inline', 'path', or 'size'; a group "
+              + "without 'path' or 'inline' holds only self-references, which require 'size'",
+          name);
+      // The remaining spec rules are per-value constraints that the schema builder cannot verify
+      // because it sees only which fields are declared, not their values in each row: when `path`
+      // is null in a row that value is a self-reference and must carry a non-null `size`, and
+      // `offset`/`size` must be non-negative. Those are the responsibility of writers and
+      // consumers of FILE values.
+    }
+
     public MapBuilder<THIS> map(Type.Repetition repetition) {
       return new MapBuilder<>(self()).repetition(repetition);
     }

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

@@ -528,6 +528,182 @@ public void testVariantLogicalTypeWithShredded() {
     assertEquals(specVersion, ((LogicalTypeAnnotation.VariantLogicalTypeAnnotation) annotation).getSpecVersion());
   }
 
+  @Test
+  public void testFileLogicalTypePathOnly() {
+    String name = "file_field";
+    GroupType file = new GroupType(
+        REQUIRED,
+        name,
+        LogicalTypeAnnotation.fileType(),
+        Types.optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("path"));
+
+    assertEquals(
+        "required group file_field (FILE) {\n"
+            + "  optional binary path (STRING);\n"
+            + "}",
+        file.toString());
+
+    LogicalTypeAnnotation annotation = file.getLogicalTypeAnnotation();
+    assertEquals(LogicalTypeAnnotation.LogicalTypeToken.FILE, annotation.getType());
+    assertNull(annotation.toOriginalType());
+    assertTrue(annotation instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation);
+  }
+
+  @Test
+  public void testFileLogicalTypeAllFields() {
+    String name = "file_field";
+    GroupType file = Types.requiredGroup()
+        .as(LogicalTypeAnnotation.fileType())
+        .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("path")
+        .optional(INT64).named("offset")
+        .optional(INT64).named("size")
+        .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("content_type")
+        .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("checksum")
+        .optional(BINARY).named("inline")
+        .named(name);
+
+    LogicalTypeAnnotation annotation = file.getLogicalTypeAnnotation();
+    assertTrue(annotation instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation);
+    assertEquals(6, file.getFieldCount());
+    assertEquals("path", file.getType("path").getName());
+    assertEquals("offset", file.getType("offset").getName());
+    assertEquals("size", file.getType("size").getName());
+    assertEquals("content_type", file.getType("content_type").getName());
+    assertEquals("checksum", file.getType("checksum").getName());
+    assertEquals("inline", file.getType("inline").getName());
+  }
+
+  @Test
+  public void testFileLogicalTypeInlineOnly() {
+    // Every field is optional, so an inline-only group is valid (spec self-reference / inline case).
+    GroupType file = Types.requiredGroup()
+        .as(LogicalTypeAnnotation.fileType())
+        .optional(BINARY).named("inline")
+        .named("inline_file");
+
+    assertTrue(file.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation);
+    assertEquals(1, file.getFieldCount());
+    assertEquals("inline", file.getType("inline").getName());
+  }
+
+  @Test
+  public void testFileLogicalTypeSelfReference() {
+    // A self-reference omits 'path' and locates bytes within the current file via offset/size.
+    GroupType file = Types.requiredGroup()
+        .as(LogicalTypeAnnotation.fileType())
+        .optional(INT64).named("offset")
+        .optional(INT64).named("size")
+        .named("self_ref_file");
+
+    assertTrue(file.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation);
+    assertEquals(2, file.getFieldCount());
+  }
+
+  @Test
+  public void testFileLogicalTypeSelfReferenceRequiresSize() {
+    // A group without 'path' or 'inline' can only hold self-references, which require 'size'.
+    // Declaring only metadata fields leaves no way to resolve or size the referenced bytes.
+    assertThrows(
+        "FILE type group without 'path'/'inline' must declare 'size'",
+        IllegalArgumentException.class,
+        () -> Types.requiredGroup()
+            .as(LogicalTypeAnnotation.fileType())
+            .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("content_type")
+            .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("checksum")
+            .named("file_metadata_only"));
+  }
+
+  @Test
+  public void testFileLogicalTypeSelfReferenceWithSize() {
+    // A self-reference (no 'path') that declares 'size' is valid.
+    GroupType file = Types.requiredGroup()
+        .as(LogicalTypeAnnotation.fileType())
+        .optional(INT64).named("size")
+        .named("self_ref_with_size");
+
+    assertTrue(file.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation);
+    assertEquals(1, file.getFieldCount());
+  }
+
+  @Test
+  public void testFileLogicalTypeOffsetRequiresSize() {
+    // The spec requires 'size' whenever 'offset' is set, so a group declaring 'offset'
+    // without 'size' can never produce a valid value and is rejected at build time.
+    assertThrows(
+        "FILE type group with 'offset' must also declare 'size'",
+        IllegalArgumentException.class,
+        () -> Types.requiredGroup()
+            .as(LogicalTypeAnnotation.fileType())
+            .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("path")
+            .optional(INT64).named("offset")
+            .named("file_offset_without_size"));
+  }
+
+  @Test
+  public void testFileLogicalTypeOffsetWithSize() {
+    // 'offset' accompanied by 'size' is valid.
+    GroupType file = Types.requiredGroup()
+        .as(LogicalTypeAnnotation.fileType())
+        .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("path")
+        .optional(INT64).named("offset")
+        .optional(INT64).named("size")
+        .named("file_offset_with_size");
+
+    assertTrue(file.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation);
+    assertEquals(3, file.getFieldCount());
+  }
+
+  @Test
+  public void testFileLogicalTypeSizeWithoutOffset() {
+    // 'size' without 'offset' is valid (e.g. a whole-file range starting at 0).
+    GroupType file = Types.requiredGroup()
+        .as(LogicalTypeAnnotation.fileType())
+        .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("path")
+        .optional(INT64).named("size")
+        .named("file_size_without_offset");
+
+    assertTrue(file.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.FileLogicalTypeAnnotation);
+    assertEquals(2, file.getFieldCount());
+  }
+
+  @Test
+  public void testFileLogicalTypeRejectsUnrecognizedField() {
+    assertThrows(
+        "FILE type group must not contain unrecognized field names",
+        IllegalArgumentException.class,
+        () -> Types.requiredGroup()
+            .as(LogicalTypeAnnotation.fileType())
+            .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("path")
+            .optional(BINARY).named("unknown_field")
+            .named("file_with_bad_field"));
+  }
+
+  @Test
+  public void testFileLogicalTypeRejectsRequiredField() {
+    // All FILE fields must have OPTIONAL repetition under the current spec.
+    assertThrows(
+        "FILE type field must be optional",
+        IllegalArgumentException.class,
+        () -> Types.requiredGroup()
+            .as(LogicalTypeAnnotation.fileType())
+            .required(BINARY).as(LogicalTypeAnnotation.stringType()).named("path")
+            .named("file_with_required_path"));
+  }
+
+  @Test
+  public void testFileLogicalTypeRejectsGroupField() {
+    // FILE fields must be primitives, not nested groups.
+    assertThrows(
+        "FILE type field must be primitive",
+        IllegalArgumentException.class,
+        () -> Types.requiredGroup()
+            .as(LogicalTypeAnnotation.fileType())
+            .optionalGroup()
+            .optional(BINARY).named("nested")
+            .named("path")
+            .named("file_with_group_field"));
+  }
+
   /**
    * A convenience method to avoid a large number of @Test(expected=...) tests
    *

parquet-format-structures/src/main/java/org/apache/parquet/format/LogicalTypes.java

@@ -60,4 +60,5 @@ public static LogicalType VARIANT(byte specificationVersion) {
   public static final LogicalType BSON = LogicalType.BSON(new BsonType());
   public static final LogicalType FLOAT16 = LogicalType.FLOAT16(new Float16Type());
   public static final LogicalType UUID = LogicalType.UUID(new UUIDType());
+  public static final LogicalType FILE = LogicalType.FILE(new FileType());
 }