GH-3344: Adaptive compression for v2 page

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

@@ -69,6 +69,7 @@ public class ParquetProperties {
   public static final boolean DEFAULT_SIZE_STATISTICS_ENABLED = true;
 
   public static final boolean DEFAULT_PAGE_WRITE_CHECKSUM_ENABLED = true;
+  public static final double DEFAULT_PAGE_COMPRESS_THRESHOLD = 0.98;
 
   /**
    * @deprecated This shared instance can cause thread safety issues when used by multiple builders concurrently.
@@ -121,6 +122,7 @@ public static WriterVersion fromString(String name) {
   private final int statisticsTruncateLength;
   private final boolean statisticsEnabled;
   private final boolean sizeStatisticsEnabled;
+  private final double pageCompressThreshold;
 
   // The expected NDV (number of distinct values) for each columns
   private final ColumnProperty<Long> bloomFilterNDVs;
@@ -157,6 +159,8 @@ private ParquetProperties(Builder builder) {
     this.statisticsTruncateLength = builder.statisticsTruncateLength;
     this.statisticsEnabled = builder.statisticsEnabled;
     this.sizeStatisticsEnabled = builder.sizeStatisticsEnabled;
+    this.pageCompressThreshold = builder.pageCompressThreshold;
+
     this.bloomFilterNDVs = builder.bloomFilterNDVs.build();
     this.bloomFilterFPPs = builder.bloomFilterFPPs.build();
     this.bloomFilterEnabled = builder.bloomFilterEnabled.build();
@@ -327,6 +331,10 @@ public boolean getPageWriteChecksumEnabled() {
     return pageWriteChecksumEnabled;
   }
 
+  public double pageCompressThreshold() {
+    return pageCompressThreshold;
+  }
+
   public OptionalLong getBloomFilterNDV(ColumnDescriptor column) {
     Long ndv = bloomFilterNDVs.getValue(column);
     return ndv == null ? OptionalLong.empty() : OptionalLong.of(ndv);
@@ -415,7 +423,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'
+        + "Page compress threshold: " + pageCompressThreshold;
     String perColumn = "";
     if (!columnCodecs.toString().equals(Objects.toString(columnCodecs.getDefaultValue()))) {
       perColumn = "Per-column codecs: " + columnCodecs;
@@ -445,6 +454,7 @@ public static class Builder {
     private int statisticsTruncateLength = DEFAULT_STATISTICS_TRUNCATE_LENGTH;
     private boolean statisticsEnabled = DEFAULT_STATISTICS_ENABLED;
     private boolean sizeStatisticsEnabled = DEFAULT_SIZE_STATISTICS_ENABLED;
+    private double pageCompressThreshold = DEFAULT_PAGE_COMPRESS_THRESHOLD;
     private final ColumnProperty.Builder<Long> bloomFilterNDVs;
     private final ColumnProperty.Builder<Double> bloomFilterFPPs;
     private int maxBloomFilterBytes = DEFAULT_MAX_BLOOM_FILTER_BYTES;
@@ -511,6 +521,7 @@ private Builder(ParquetProperties toCopy) {
       this.sizeStatisticsEnabled = toCopy.sizeStatisticsEnabled;
       this.columnCodecs = ColumnProperty.builder(toCopy.columnCodecs);
       this.columnCompressionLevels = ColumnProperty.builder(toCopy.columnCompressionLevels);
+      this.pageCompressThreshold = toCopy.pageCompressThreshold();
     }
 
     /**
@@ -833,6 +844,27 @@ public Builder withCompressionLevel(String columnPath, Integer level) {
       return this;
     }
 
+    /**
+     * Sets the compression threshold for data pages, only effect for V2 pages.
+     *
+     * <p>When the compression ratio (compressed size / uncompressed size) exceeds this threshold,
+     * the uncompressed data will be used instead. For example, with a threshold of 0.98, if
+     * compression saves less than 2% of space, the data will not be compressed.
+     * A value of 0 disables data page compression and a value of 1 only skips compression when it
+     * increases the data size.
+     *
+     * @param threshold the compression ratio threshold, default is {@value #DEFAULT_PAGE_COMPRESS_THRESHOLD}
+     * @return this builder for method chaining
+     */
+    public Builder withPageCompressThreshold(double threshold) {
+      Preconditions.checkArgument(
+          Double.isFinite(threshold) && threshold >= 0.0 && threshold <= 1.0,
+          "Invalid page compression threshold: %s. It must be a finite value between 0.0 and 1.0.",
+          threshold);
+      this.pageCompressThreshold = threshold;
+      return this;
+    }
+
     public ParquetProperties build() {
       ParquetProperties properties = new ParquetProperties(this);
       // we pass a constructed but uninitialized factory to ParquetProperties above as currently

parquet-column/src/test/java/org/apache/parquet/column/TestParquetProperties.java

@@ -122,6 +122,36 @@ public void withCompressionCodec_nullCodec_throwsNullPointerException() {
         NullPointerException.class, () -> ParquetProperties.builder().withCompressionCodec("col_a", null));
   }
 
+  @Test
+  public void pageCompressThreshold_rejectsValuesOutsideRatioRange() {
+    assertThrows(IllegalArgumentException.class, () -> ParquetProperties.builder()
+        .withPageCompressThreshold(-0.01));
+    assertThrows(IllegalArgumentException.class, () -> ParquetProperties.builder()
+        .withPageCompressThreshold(1.01));
+    assertThrows(IllegalArgumentException.class, () -> ParquetProperties.builder()
+        .withPageCompressThreshold(Double.NaN));
+    assertThrows(IllegalArgumentException.class, () -> ParquetProperties.builder()
+        .withPageCompressThreshold(Double.POSITIVE_INFINITY));
+  }
+
+  @Test
+  public void pageCompressThreshold_acceptsRatioBoundaries() {
+    assertEquals(
+        0.0,
+        ParquetProperties.builder()
+            .withPageCompressThreshold(0.0)
+            .build()
+            .pageCompressThreshold(),
+        0.0);
+    assertEquals(
+        1.0,
+        ParquetProperties.builder()
+            .withPageCompressThreshold(1.0)
+            .build()
+            .pageCompressThreshold(),
+        0.0);
+  }
+
   @Test
   public void copyBuilder_preservesColumnCodecAndLevel() {
     ParquetProperties original = ParquetProperties.builder()