This is an automated email from the ASF dual-hosted git repository.
claudevdm 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 f29d9af139a init (#39025)
f29d9af139a is described below
commit f29d9af139a3c0ccf1ce33a43485bc140bb55e0c
Author: claudevdm <[email protected]>
AuthorDate: Tue Jun 23 19:53:21 2026 -0400
init (#39025)
---
.../org/apache/beam/sdk/io/parquet/ParquetIO.java | 47 ++++++++++++++++++++++
1 file changed, 47 insertions(+)
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 feaceeeb443..f869286b98b 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
@@ -1033,6 +1033,8 @@ public class ParquetIO {
abstract @Nullable ValueProvider<Integer> getMinRowCountForPageSizeCheck();
+ abstract @Nullable ValueProvider<Integer> getMaxRowCountForPageSizeCheck();
+
abstract @Nullable Class<? extends GenericData> getAvroDataModelClass();
abstract Builder toBuilder();
@@ -1056,6 +1058,9 @@ public class ParquetIO {
abstract Builder setMinRowCountForPageSizeCheck(
ValueProvider<Integer> minRowCountForPageSizeCheck);
+ abstract Builder setMaxRowCountForPageSizeCheck(
+ ValueProvider<Integer> maxRowCountForPageSizeCheck);
+
abstract Builder setAvroDataModelClass(Class<? extends GenericData>
modelClass);
abstract Sink build();
@@ -1130,6 +1135,39 @@ public class ParquetIO {
return
toBuilder().setMinRowCountForPageSizeCheck(minRowCountForPageSizeCheck).build();
}
+ /**
+ * Specify the maximum number of rows to buffer before a page size check
is forced. By default
+ * Parquet estimates the next check from the average row size and may
defer it (up to {@code
+ * 10000} rows); a run of small rows followed by large rows can then let
the column page buffer
+ * overflow {@code Integer.MAX_VALUE} before the deferred check fires.
Setting this (e.g. {@code
+ * 1}) caps the interval so a check -- and flush -- happens at least this
often regardless of
+ * the estimate. Pair it with {@link
#withMinRowCountForPageSizeCheck(int)} to bound the buffer
+ * for tables whose row sizes vary widely.
+ */
+ public Sink withMaxRowCountForPageSizeCheck(int
maxRowCountForPageSizeCheck) {
+ checkArgument(
+ maxRowCountForPageSizeCheck > 0, "maxRowCountForPageSizeCheck must
be positive");
+ return toBuilder()
+ .setMaxRowCountForPageSizeCheck(
+
ValueProvider.StaticValueProvider.of(maxRowCountForPageSizeCheck))
+ .build();
+ }
+
+ /**
+ * Like {@link #withMaxRowCountForPageSizeCheck(int)}, but accepts a
{@link ValueProvider} so
+ * the value can be supplied at runtime (required for classic Dataflow
templates).
+ */
+ public Sink withMaxRowCountForPageSizeCheck(
+ ValueProvider<Integer> maxRowCountForPageSizeCheck) {
+ checkNotNull(maxRowCountForPageSizeCheck, "maxRowCountForPageSizeCheck
can not be null");
+ if (maxRowCountForPageSizeCheck.isAccessible()) {
+ Integer value = maxRowCountForPageSizeCheck.get();
+ checkNotNull(value, "maxRowCountForPageSizeCheck value cannot be
null");
+ checkArgument(value > 0, "maxRowCountForPageSizeCheck must be
positive");
+ }
+ return
toBuilder().setMaxRowCountForPageSizeCheck(maxRowCountForPageSizeCheck).build();
+ }
+
/**
* Define the Avro data model; see {@link
AvroParquetWriter.Builder#withDataModel(GenericData)}.
*/
@@ -1170,6 +1208,15 @@ public class ParquetIO {
}
}
+ ValueProvider<Integer> maxRowCountProvider =
getMaxRowCountForPageSizeCheck();
+ if (maxRowCountProvider != null) {
+ Integer maxRowCount = maxRowCountProvider.get();
+ if (maxRowCount != null) {
+ checkArgument(maxRowCount > 0, "maxRowCountForPageSizeCheck must be
positive");
+ builder = builder.withMaxRowCountForPageSizeCheck(maxRowCount);
+ }
+ }
+
if (modelClass != null) {
try {
builder.withDataModel(buildModelObject(modelClass));
