(beam) branch master updated: Make MinRowCountForPageSizeCheck available with ValueProvider (#38114)

[cv...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

cvandermerwe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new 47031efee65 Make MinRowCountForPageSizeCheck available with ValueProvider (#38114)
47031efee65 is described below

commit 47031efee655761fbd8c19d5f1ddf5d404f870e6
Author: claudevdm <33...@users.noreply.github.com>
AuthorDate: Wed Apr 8 20:11:03 2026 -0400

    Make MinRowCountForPageSizeCheck available with ValueProvider (#38114)
    
    * Make MinRowCountForPageSizeCheck available with ValueProvider for classic templates
    
    * comments
    
    * comments
---
 .../org/apache/beam/sdk/io/parquet/ParquetIO.java  | 36 +++++++++++++++++++---
 1 file changed, 31 insertions(+), 5 deletions(-)

diff --git a/sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetIO.java b/sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetIO.java
index d365cd5f7c7..feaceeeb443 100644
--- a/sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetIO.java
+++ b/sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetIO.java
@@ -1010,7 +1010,6 @@ public class ParquetIO {
         .setPageSize(ParquetWriter.DEFAULT_PAGE_SIZE)
         .setEnableDictionary(ParquetWriter.DEFAULT_IS_DICTIONARY_ENABLED)
         .setEnableBloomFilter(ParquetProperties.DEFAULT_BLOOM_FILTER_ENABLED)
-        .setMinRowCountForPageSizeCheck(ParquetProperties.DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK)
         .build();
   }
 
@@ -1032,7 +1031,7 @@ public class ParquetIO {
 
     abstract boolean getEnableBloomFilter();
 
-    abstract int getMinRowCountForPageSizeCheck();
+    abstract @Nullable ValueProvider<Integer> getMinRowCountForPageSizeCheck();
 
     abstract @Nullable Class<? extends GenericData> getAvroDataModelClass();
 
@@ -1054,7 +1053,8 @@ public class ParquetIO {
 
       abstract Builder setEnableBloomFilter(boolean enableBloomFilter);
 
-      abstract Builder setMinRowCountForPageSizeCheck(int minRowCountForPageSizeCheck);
+      abstract Builder setMinRowCountForPageSizeCheck(
+          ValueProvider<Integer> minRowCountForPageSizeCheck);
 
       abstract Builder setAvroDataModelClass(Class<? extends GenericData> modelClass);
 
@@ -1109,6 +1109,24 @@ public class ParquetIO {
     public Sink withMinRowCountForPageSizeCheck(int minRowCountForPageSizeCheck) {
       checkArgument(
           minRowCountForPageSizeCheck > 0, "minRowCountForPageSizeCheck must be positive");
+      return toBuilder()
+          .setMinRowCountForPageSizeCheck(
+              ValueProvider.StaticValueProvider.of(minRowCountForPageSizeCheck))
+          .build();
+    }
+
+    /**
+     * Like {@link #withMinRowCountForPageSizeCheck(int)}, but accepts a {@link ValueProvider} so
+     * the value can be supplied at runtime (required for classic Dataflow templates).
+     */
+    public Sink withMinRowCountForPageSizeCheck(
+        ValueProvider<Integer> minRowCountForPageSizeCheck) {
+      checkNotNull(minRowCountForPageSizeCheck, "minRowCountForPageSizeCheck can not be null");
+      if (minRowCountForPageSizeCheck.isAccessible()) {
+        Integer value = minRowCountForPageSizeCheck.get();
+        checkNotNull(value, "minRowCountForPageSizeCheck value cannot be null");
+        checkArgument(value > 0, "minRowCountForPageSizeCheck must be positive");
+      }
       return toBuilder().setMinRowCountForPageSizeCheck(minRowCountForPageSizeCheck).build();
     }
 
@@ -1141,8 +1159,16 @@ public class ParquetIO {
               .withRowGroupSize(getRowGroupSize())
               .withPageSize(getPageSize())
               .withDictionaryEncoding(getEnableDictionary())
-              .withBloomFilterEnabled(getEnableBloomFilter())
-              .withMinRowCountForPageSizeCheck(getMinRowCountForPageSizeCheck());
+              .withBloomFilterEnabled(getEnableBloomFilter());
+
+      ValueProvider<Integer> minRowCountProvider = getMinRowCountForPageSizeCheck();
+      if (minRowCountProvider != null) {
+        Integer minRowCount = minRowCountProvider.get();
+        if (minRowCount != null) {
+          checkArgument(minRowCount > 0, "minRowCountForPageSizeCheck must be positive");
+          builder = builder.withMinRowCountForPageSizeCheck(minRowCount);
+        }
+      }
 
       if (modelClass != null) {
         try {