Add threshold-driven inline vs self-reference storage for FILE values

parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java

@@ -68,6 +68,13 @@ public class ParquetProperties {
   public static final boolean DEFAULT_STATISTICS_ENABLED = true;
   public static final boolean DEFAULT_SIZE_STATISTICS_ENABLED = true;
 
+  /**
+   * Payload size at or below which a {@code FILE} value is stored inline rather than as a
+   * self-reference. Defaults to the page size: a payload that would fill a page on its own is
+   * better kept out of the column chunk.
+   */
+  public static final int DEFAULT_FILE_SELF_REFERENCE_THRESHOLD = DEFAULT_PAGE_SIZE;
+
   public static final boolean DEFAULT_PAGE_WRITE_CHECKSUM_ENABLED = true;
 
   /**
@@ -138,6 +145,7 @@ public static WriterVersion fromString(String name) {
   private final ColumnProperty<Boolean> sizeStatistics;
   private final ColumnProperty<CompressionCodecName> columnCodecs;
   private final ColumnProperty<Integer> columnCompressionLevels;
+  private final int fileSelfReferenceThreshold;
 
   private ParquetProperties(Builder builder) {
     this.pageSizeThreshold = builder.pageSize;
@@ -172,6 +180,7 @@ private ParquetProperties(Builder builder) {
     this.sizeStatistics = builder.sizeStatistics.build();
     this.columnCodecs = builder.columnCodecs.build();
     this.columnCompressionLevels = builder.columnCompressionLevels.build();
+    this.fileSelfReferenceThreshold = builder.fileSelfReferenceThreshold;
   }
 
   public static Builder builder() {
@@ -345,6 +354,14 @@ public int getMaxBloomFilterBytes() {
     return maxBloomFilterBytes;
   }
 
+  /**
+   * @return the payload size at or below which a {@code FILE} value is stored inline rather than as
+   *     a self-reference
+   */
+  public int getFileSelfReferenceThreshold() {
+    return fileSelfReferenceThreshold;
+  }
+
   public boolean getAdaptiveBloomFilterEnabled(ColumnDescriptor column) {
     return adaptiveBloomFilterEnabled.getValue(column);
   }
@@ -415,7 +432,8 @@ public String toString() {
         + "Page row count limit to " + getPageRowCountLimit() + '\n'
         + "Writing page checksums is: " + (getPageWriteChecksumEnabled() ? "on" : "off") + '\n'
         + "Statistics enabled: " + statisticsEnabled + '\n'
-        + "Size statistics enabled: " + sizeStatisticsEnabled;
+        + "Size statistics enabled: " + sizeStatisticsEnabled + '\n'
+        + "FILE self-reference threshold is: " + getFileSelfReferenceThreshold();
     String perColumn = "";
     if (!columnCodecs.toString().equals(Objects.toString(columnCodecs.getDefaultValue()))) {
       perColumn = "Per-column codecs: " + columnCodecs;
@@ -460,6 +478,7 @@ public static class Builder {
     private final ColumnProperty.Builder<Boolean> sizeStatistics;
     private final ColumnProperty.Builder<CompressionCodecName> columnCodecs;
     private final ColumnProperty.Builder<Integer> columnCompressionLevels;
+    private int fileSelfReferenceThreshold = DEFAULT_FILE_SELF_REFERENCE_THRESHOLD;
 
     private Builder() {
       enableDict = ColumnProperty.<Boolean>builder().withDefaultValue(DEFAULT_IS_DICTIONARY_ENABLED);
@@ -511,6 +530,7 @@ private Builder(ParquetProperties toCopy) {
       this.sizeStatisticsEnabled = toCopy.sizeStatisticsEnabled;
       this.columnCodecs = ColumnProperty.builder(toCopy.columnCodecs);
       this.columnCompressionLevels = ColumnProperty.builder(toCopy.columnCompressionLevels);
+      this.fileSelfReferenceThreshold = toCopy.fileSelfReferenceThreshold;
     }
 
     /**
@@ -657,6 +677,27 @@ public Builder withStatisticsTruncateLength(int length) {
       return this;
     }
 
+    /**
+     * Set the payload size at or below which a {@code FILE} value is stored inline rather than as a
+     * self-reference.
+     *
+     * <p>Small payloads are cheaper to keep in the column chunk, where they are read as part of the
+     * ordinary page stream. Large ones are better stored out of line as self-references, so that
+     * reading the surrounding columns does not pull the payload bytes along with them. Set to 0 to
+     * store every payload as a self-reference, or to {@link Integer#MAX_VALUE} to always inline.
+     *
+     * @param fileSelfReferenceThreshold the inline size limit in bytes; must not be negative
+     * @return this builder for method chaining
+     */
+    public Builder withFileSelfReferenceThreshold(int fileSelfReferenceThreshold) {
+      Preconditions.checkArgument(
+          fileSelfReferenceThreshold >= 0,
+          "Invalid FILE self-reference threshold (negative): %s",
+          fileSelfReferenceThreshold);
+      this.fileSelfReferenceThreshold = fileSelfReferenceThreshold;
+      return this;
+    }
+
     /**
      * Set max Bloom filter bytes for related columns.
      *

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

@@ -1302,8 +1302,8 @@ public static class FileLogicalTypeAnnotation extends LogicalTypeAnnotation {
     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);
+    public static final Set<String> FIELD_NAMES =
+        Set.of(URI_FIELD, OFFSET_FIELD, SIZE_FIELD, CONTENT_TYPE_FIELD, CHECKSUM_FIELD, INLINE_FIELD);
 
     private FileLogicalTypeAnnotation() {}
 

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

@@ -883,7 +883,9 @@ private static void validateFileTypeFields(String name, List<Type> fields) {
       // `uri` is not required to declare `inline`: `offset`/`size` there describe an external
       // ranged reference, and although the per-value `uri` could be left unset in some rows, the
       // schema is treated as an external-reference schema and the `inline` requirement is not
-      // imposed.
+      // imposed. A writer must therefore not emit a self-reference under such a schema, since there
+      // would be no `inline` column chunk to inherit compression and encryption from; that is
+      // enforced on the write path rather than here.
       Preconditions.checkArgument(
           !(hasOffset && !hasUri) || hasInline,
           "FILE type group '%s' declares field 'offset' but neither 'uri' nor 'inline'; a schema "

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

@@ -560,9 +560,7 @@ public void testFileLogicalTypeUriOnly() {
         Types.optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("uri"));
 
     assertThat(file.toString())
-        .isEqualTo("required group file_field (FILE) {\n"
-            + "  optional binary uri (STRING);\n"
-            + "}");
+        .isEqualTo("required group file_field (FILE) {\n" + "  optional binary uri (STRING);\n" + "}");
 
     LogicalTypeAnnotation annotation = file.getLogicalTypeAnnotation();
     assertThat(annotation.getType()).isEqualTo(LogicalTypeAnnotation.LogicalTypeToken.FILE);
@@ -575,12 +573,21 @@ public void testFileLogicalTypeAllFields() {
     String name = "file_field";
     GroupType file = Types.requiredGroup()
         .as(LogicalTypeAnnotation.fileType())
-        .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("uri")
-        .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")
+        .optional(BINARY)
+        .as(LogicalTypeAnnotation.stringType())
+        .named("uri")
+        .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();
@@ -599,7 +606,8 @@ public void testFileLogicalTypeInlineOnly() {
     // Every field is optional, so an inline-only group is valid (spec inline case).
     GroupType file = Types.requiredGroup()
         .as(LogicalTypeAnnotation.fileType())
-        .optional(BINARY).named("inline")
+        .optional(BINARY)
+        .named("inline")
         .named("inline_file");
 
     assertThat(file.getLogicalTypeAnnotation()).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class);
@@ -614,9 +622,12 @@ public void testFileLogicalTypeSelfReference() {
     // reference point.
     GroupType file = Types.requiredGroup()
         .as(LogicalTypeAnnotation.fileType())
-        .optional(INT64).named("offset")
-        .optional(INT64).named("size")
-        .optional(BINARY).named("inline")
+        .optional(INT64)
+        .named("offset")
+        .optional(INT64)
+        .named("size")
+        .optional(BINARY)
+        .named("inline")
         .named("self_ref_file");
 
     assertThat(file.getLogicalTypeAnnotation()).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class);
@@ -630,8 +641,10 @@ public void testFileLogicalTypeOffsetRequiresInline() {
     // neither 'uri' nor 'inline' is rejected at build time.
     assertThatThrownBy(() -> Types.requiredGroup()
             .as(LogicalTypeAnnotation.fileType())
-            .optional(INT64).named("offset")
-            .optional(INT64).named("size")
+            .optional(INT64)
+            .named("offset")
+            .optional(INT64)
+            .named("size")
             .named("self_ref_without_inline"))
         .isInstanceOf(IllegalArgumentException.class);
   }
@@ -643,9 +656,13 @@ public void testFileLogicalTypeExternalRangedReferenceWithoutInline() {
     // schema and is not required to declare 'inline', even though it declares 'offset'.
     GroupType file = Types.requiredGroup()
         .as(LogicalTypeAnnotation.fileType())
-        .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("uri")
-        .optional(INT64).named("offset")
-        .optional(INT64).named("size")
+        .optional(BINARY)
+        .as(LogicalTypeAnnotation.stringType())
+        .named("uri")
+        .optional(INT64)
+        .named("offset")
+        .optional(INT64)
+        .named("size")
         .named("external_ranged_file");
 
     assertThat(file.getLogicalTypeAnnotation()).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class);
@@ -658,8 +675,12 @@ public void testFileLogicalTypeMetadataOnlyRejected() {
     // 'offset'. A group declaring only metadata fields can never produce a resolvable value.
     assertThatThrownBy(() -> Types.requiredGroup()
             .as(LogicalTypeAnnotation.fileType())
-            .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("content_type")
-            .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("checksum")
+            .optional(BINARY)
+            .as(LogicalTypeAnnotation.stringType())
+            .named("content_type")
+            .optional(BINARY)
+            .as(LogicalTypeAnnotation.stringType())
+            .named("checksum")
             .named("file_metadata_only"))
         .isInstanceOf(IllegalArgumentException.class);
   }
@@ -670,7 +691,8 @@ public void testFileLogicalTypeSizeOnlyRejected() {
     // rejected: it declares no locator ('inline', 'uri', or 'offset').
     assertThatThrownBy(() -> Types.requiredGroup()
             .as(LogicalTypeAnnotation.fileType())
-            .optional(INT64).named("size")
+            .optional(INT64)
+            .named("size")
             .named("file_size_only"))
         .isInstanceOf(IllegalArgumentException.class);
   }
@@ -681,8 +703,11 @@ public void testFileLogicalTypeOffsetRequiresSize() {
     // without 'size' can never produce a valid value and is rejected at build time.
     assertThatThrownBy(() -> Types.requiredGroup()
             .as(LogicalTypeAnnotation.fileType())
-            .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("uri")
-            .optional(INT64).named("offset")
+            .optional(BINARY)
+            .as(LogicalTypeAnnotation.stringType())
+            .named("uri")
+            .optional(INT64)
+            .named("offset")
             .named("file_offset_without_size"))
         .isInstanceOf(IllegalArgumentException.class);
   }
@@ -692,9 +717,13 @@ public void testFileLogicalTypeOffsetWithSize() {
     // 'offset' accompanied by 'size' is valid.
     GroupType file = Types.requiredGroup()
         .as(LogicalTypeAnnotation.fileType())
-        .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("uri")
-        .optional(INT64).named("offset")
-        .optional(INT64).named("size")
+        .optional(BINARY)
+        .as(LogicalTypeAnnotation.stringType())
+        .named("uri")
+        .optional(INT64)
+        .named("offset")
+        .optional(INT64)
+        .named("size")
         .named("file_offset_with_size");
 
     assertThat(file.getLogicalTypeAnnotation()).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class);
@@ -706,8 +735,11 @@ public void testFileLogicalTypeSizeWithoutOffset() {
     // 'uri' + 'size' (without 'offset') is valid: an external reference to '[0, size)'.
     GroupType file = Types.requiredGroup()
         .as(LogicalTypeAnnotation.fileType())
-        .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("uri")
-        .optional(INT64).named("size")
+        .optional(BINARY)
+        .as(LogicalTypeAnnotation.stringType())
+        .named("uri")
+        .optional(INT64)
+        .named("size")
         .named("file_size_without_offset");
 
     assertThat(file.getLogicalTypeAnnotation()).isInstanceOf(LogicalTypeAnnotation.FileLogicalTypeAnnotation.class);
@@ -718,8 +750,11 @@ public void testFileLogicalTypeSizeWithoutOffset() {
   public void testFileLogicalTypeRejectsUnrecognizedField() {
     assertThatThrownBy(() -> Types.requiredGroup()
             .as(LogicalTypeAnnotation.fileType())
-            .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("uri")
-            .optional(BINARY).named("unknown_field")
+            .optional(BINARY)
+            .as(LogicalTypeAnnotation.stringType())
+            .named("uri")
+            .optional(BINARY)
+            .named("unknown_field")
             .named("file_with_bad_field"))
         .isInstanceOf(IllegalArgumentException.class);
   }
@@ -729,7 +764,9 @@ public void testFileLogicalTypeRejectsRequiredField() {
     // All FILE fields must have OPTIONAL repetition under the current spec.
     assertThatThrownBy(() -> Types.requiredGroup()
             .as(LogicalTypeAnnotation.fileType())
-            .required(BINARY).as(LogicalTypeAnnotation.stringType()).named("uri")
+            .required(BINARY)
+            .as(LogicalTypeAnnotation.stringType())
+            .named("uri")
             .named("file_with_required_uri"))
         .isInstanceOf(IllegalArgumentException.class);
   }
@@ -740,7 +777,8 @@ public void testFileLogicalTypeRejectsGroupField() {
     assertThatThrownBy(() -> Types.requiredGroup()
             .as(LogicalTypeAnnotation.fileType())
             .optionalGroup()
-            .optional(BINARY).named("nested")
+            .optional(BINARY)
+            .named("nested")
             .named("uri")
             .named("file_with_group_field"))
         .isInstanceOf(IllegalArgumentException.class);
@@ -751,7 +789,8 @@ public void testFileLogicalTypeRejectsWrongStringPhysicalType() {
     // 'uri' must be a STRING (BINARY annotated as STRING); an INT64 is rejected.
     assertThatThrownBy(() -> Types.requiredGroup()
             .as(LogicalTypeAnnotation.fileType())
-            .optional(INT64).named("uri")
+            .optional(INT64)
+            .named("uri")
             .named("file_uri_wrong_type"))
         .isInstanceOf(IllegalArgumentException.class);
   }
@@ -761,7 +800,8 @@ public void testFileLogicalTypeRejectsUnannotatedStringField() {
     // A STRING field must carry the STRING logical annotation; plain BINARY is rejected.
     assertThatThrownBy(() -> Types.requiredGroup()
             .as(LogicalTypeAnnotation.fileType())
-            .optional(BINARY).named("uri")
+            .optional(BINARY)
+            .named("uri")
             .named("file_uri_unannotated"))
         .isInstanceOf(IllegalArgumentException.class);
   }
@@ -771,8 +811,11 @@ public void testFileLogicalTypeRejectsWrongInt64PhysicalType() {
     // 'offset' and 'size' must be INT64; an INT32 is rejected.
     assertThatThrownBy(() -> Types.requiredGroup()
             .as(LogicalTypeAnnotation.fileType())
-            .optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("uri")
-            .optional(INT32).named("size")
+            .optional(BINARY)
+            .as(LogicalTypeAnnotation.stringType())
+            .named("uri")
+            .optional(INT32)
+            .named("size")
             .named("file_size_wrong_type"))
         .isInstanceOf(IllegalArgumentException.class);
   }
@@ -782,7 +825,8 @@ public void testFileLogicalTypeRejectsWrongInlinePhysicalType() {
     // 'inline' must be a BYTE_ARRAY (BINARY); an INT64 is rejected.
     assertThatThrownBy(() -> Types.requiredGroup()
             .as(LogicalTypeAnnotation.fileType())
-            .optional(INT64).named("inline")
+            .optional(INT64)
+            .named("inline")
             .named("file_inline_wrong_type"))
         .isInstanceOf(IllegalArgumentException.class);
   }