Make MinRowCountForPageSizeCheck available with ValueProvider (#38114)

sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetIO.java

@@ -1010,7 +1010,6 @@ public static Sink sink(Schema schema) {
         .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 abstract static class Sink implements FileIO.Sink<GenericRecord> {
 
     abstract boolean getEnableBloomFilter();
 
-    abstract int getMinRowCountForPageSizeCheck();
+    abstract @Nullable ValueProvider<Integer> getMinRowCountForPageSizeCheck();
 
     abstract @Nullable Class<? extends GenericData> getAvroDataModelClass();
 
@@ -1054,7 +1053,8 @@ abstract static class Builder {
 
       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 Sink withBloomFilterEnabled(boolean enableBloomFilter) {
     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 void open(WritableByteChannel channel) throws IOException {
               .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 {