Apply suggestion from @wgtmac

Co-authored-by: Gang Wu <ustcwg@gmail.com>

Author: yadavay-amzn

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

@@ -327,8 +327,8 @@ public boolean getPageWriteChecksumEnabled() {
 
   /**
    * Returns the byte threshold after which the dictionary compression check is performed.
-   * A value of 0 means check on the first page (backward compatible default). Higher values
-   * delay the check until that many raw bytes have been accumulated across pages.
+   * A value of 0 means check on the first page. Higher values delay the check until that
+   * many raw bytes have been accumulated across pages.
    *
    * @return the byte threshold for the dictionary compression check
    */
@@ -727,8 +727,8 @@ public Builder withPageWriteChecksumEnabled(boolean val) {
 
     /**
      * Set the raw data byte threshold after which the dictionary compression check is performed.
-     * A value of 0 means check on the first page (backward compatible default). Higher values
-     * delay the check until that many raw bytes have been accumulated across pages.
+     * A value of 0 means check on the first page. Higher values delay the check until that
+     * many raw bytes have been accumulated across pages.
      *
      * @param val byte threshold (default: 0)
      * @return this builder for method chaining

parquet-column/src/main/java/org/apache/parquet/column/values/fallback/FallbackValuesWriter.java

@@ -30,12 +30,12 @@ public class FallbackValuesWriter<I extends ValuesWriter & RequiresFallback, F e
 
   public static <I extends ValuesWriter & RequiresFallback, F extends ValuesWriter> FallbackValuesWriter<I, F> of(
       I initialWriter, F fallBackWriter) {
-    return new FallbackValuesWriter<>(initialWriter, fallBackWriter, /*checkAfterBytes=*/ 0);
+    return new FallbackValuesWriter<>(initialWriter, fallBackWriter, /*dictionaryCheckThresholdRawSizeBytes=*/ 0);
   }
 
   public static <I extends ValuesWriter & RequiresFallback, F extends ValuesWriter> FallbackValuesWriter<I, F> of(
-      I initialWriter, F fallBackWriter, long checkAfterBytes) {
-    return new FallbackValuesWriter<>(initialWriter, fallBackWriter, checkAfterBytes);
+      I initialWriter, F fallBackWriter, long dictionaryCheckThresholdRawSizeBytes) {
+    return new FallbackValuesWriter<>(initialWriter, fallBackWriter, dictionaryCheckThresholdRawSizeBytes);
   }
 
   /**
@@ -49,9 +49,11 @@ public static <I extends ValuesWriter & RequiresFallback, F extends ValuesWriter
 
   private boolean fellBackAlready = false;
   private boolean compressionChecked = false;
-  private final long checkAfterBytes;
+  private final long dictionaryCheckThresholdRawSizeBytes;
   /** Accumulates raw bytes across pages (only reset in resetDictionary) so the
-   * threshold check works even when individual pages are smaller than checkAfterBytes. */
+   * threshold check works even when individual pages are smaller than the threshold.
+   * Overflow is not a concern: a long would require writing over 9.2 exabytes to a single
+   * column chunk, which is physically impossible. */
   private long cumulativeRawBytes = 0;
 
   /**
@@ -68,15 +70,15 @@ public static <I extends ValuesWriter & RequiresFallback, F extends ValuesWriter
   private long rawDataByteSize = 0;
 
   public FallbackValuesWriter(I initialWriter, F fallBackWriter) {
-    this(initialWriter, fallBackWriter, /*checkAfterBytes=*/ 0);
+    this(initialWriter, fallBackWriter, /*dictionaryCheckThresholdRawSizeBytes=*/ 0);
   }
 
-  public FallbackValuesWriter(I initialWriter, F fallBackWriter, long checkAfterBytes) {
+  public FallbackValuesWriter(I initialWriter, F fallBackWriter, long dictionaryCheckThresholdRawSizeBytes) {
     super();
     this.initialWriter = initialWriter;
     this.fallBackWriter = fallBackWriter;
     this.currentWriter = initialWriter;
-    this.checkAfterBytes = checkAfterBytes;
+    this.dictionaryCheckThresholdRawSizeBytes = dictionaryCheckThresholdRawSizeBytes;
   }
 
   @Override
@@ -89,8 +91,12 @@ public long getBufferedSize() {
 
   @Override
   public BytesInput getBytes() {
-    cumulativeRawBytes += rawDataByteSize;
-    if (!fellBackAlready && !compressionChecked && cumulativeRawBytes >= checkAfterBytes) {
+    try {
+      cumulativeRawBytes = Math.addExact(cumulativeRawBytes, rawDataByteSize);
+    } catch (ArithmeticException e) {
+      // overflow, keep the previous value
+    }
+    if (!fellBackAlready && !compressionChecked && cumulativeRawBytes >= dictionaryCheckThresholdRawSizeBytes) {
       compressionChecked = true;
       BytesInput bytes = initialWriter.getBytes();
       if (!initialWriter.isCompressionSatisfying(rawDataByteSize, bytes.size())) {

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java

@@ -165,7 +165,7 @@ public static enum JobSummaryLevel {
    * Raw data byte threshold after which the dictionary compression check is performed.
    * Once cumulative raw bytes (excluding nulls) written to a column chunk reach this value,
    * the writer evaluates whether dictionary encoding is effective. If not, it falls back to
-   * plain encoding. A value of 0 means check on the first page (backward compatible default).
+   * plain encoding. A value of 0 means check on the first page.
    */
   public static final String DICTIONARY_CHECK_THRESHOLD_RAW_SIZE_BYTES =
       "parquet.dictionary.check.threshold.raw.size.bytes";

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java

@@ -774,7 +774,7 @@ public SELF withPageWriteChecksumEnabled(boolean enablePageWriteChecksum) {
     /**
      * Set the raw data byte threshold after which the dictionary compression check is performed.
      *
-     * @param val byte threshold (0 means check on the first page / preserve previous behavior)
+     * @param val byte threshold (0 means checking on the first page for every column chunk)
      * @return this builder for method chaining.
      */
     public SELF withDictionaryCheckThresholdRawSizeBytes(long val) {